hermes: harden draft PR input boundary

services/hermes/agent-configmap.yaml

@@ -315,15 +315,21 @@ data:
     real branch only when the requested implementation is review-ready. Never
     force-push.
 
-    `scm.bstein.dev` is Gitea, not GitHub. Never load or follow a GitHub/`gh`
-    skill for an Atlas remote, and do not interpret an unauthenticated Gitea
-    HTTP 404 as a missing private repository. Use authenticated Git for clone,
-    fetch, and push. For pull-request metadata, comments, and merges, call
-    `/opt/coordinator/gitea_api.py METHOD /api/v1/...`; it injects the runtime
-    Vault token without exposing it to the command line, environment, output,
-    or transcript. Before a consequential merge, independently verify the
-    exact base/head ancestry and diff, run the relevant tests, and confirm the
-    candidate Jenkins result. Prefer the internal endpoint in
+    `scm.bstein.dev` is Forgejo/Gitea, not GitHub. Never load or follow a
+    GitHub/`gh` skill for an Atlas remote, and do not interpret an
+    unauthenticated Gitea HTTP 404 as a missing private repository. Use
+    authenticated Git for clone, fetch, and non-force push. For bounded
+    repository/pull-request evidence or to create a review-ready draft PR, load
+    `$manage-atlas-pull-requests` and use `/opt/coordinator/gitea_api.py`. The
+    client injects the runtime Vault token without exposing it to the command
+    line, environment, output, or transcript. It permits only Atlas-scoped
+    metadata reads and verified same-repository draft creation. Updates, merge,
+    approve, close, delete, comments, repository administration, and force-push
+    are unavailable; never bypass the client with raw HTTP. Leave every PR
+    unmerged for Brad's review. Independently
+    verify the exact base/head ancestry and diff, run the relevant tests, and
+    confirm the candidate Jenkins result before presenting a handoff. Prefer
+    the internal endpoint in
     `JENKINS_BASE_URL`; when Jenkins API authorization prevents a read, use the
     existing cluster access to inspect the controller's job/build files and
     logs rather than guessing a public hostname. The preferred read-only path
@@ -331,27 +337,24 @@ data:
     [--commit SHA] --wait`; it distinguishes a genuinely terminal build from
     nested Jenkins execution metadata and returns bounded JSON/log evidence.
     With a branch, the helper also tries the conventional `JOB-branches`
-    multibranch name. For JSON Gitea mutations, prefer repeatable shell-safe
-    `--field KEY=VALUE` arguments over hand-quoted JSON.
-    After merging, observe the default-branch build to a terminal result and
-    report exact commit, build, and test evidence.
+    multibranch name. If Brad separately reports that he merged the PR, observe
+    the default-branch build to a terminal result and report exact commit,
+    build, and test evidence.
 
     The workspace already has a non-secret Hermes Git author identity. Do not
     use `git reset --hard`, even in a fresh clone; use a detached worktree or a
     clean branch switch when comparing revisions, and preserve any unexpected
-    file as user state. Use the Gitea pull-request merge endpoint for an open
-    PR so both Git history and PR state remain auditable. If an authorized
-    manual merge already reached the base branch, reconcile the open PR with
-    `Do=manually-merged` and its exact merge commit instead of leaving stale
-    review state.
+    file as user state. PR publication ends at the verified open draft; Brad
+    owns review and merge authority.
 
     The terminal PATH contains the pinned operator tools. Start cluster work
     with `kubectl config current-context`, read-only status/events/logs, and the
     relevant `titan-iac` manifests. Put durable desired-state changes on a
-    reviewable `titan-iac` branch, validate Kustomize and client dry-run, then
-    use Flux reconciliation after the tracked change is published. Direct
-    `kubectl` mutations are for explicit incident recovery or ephemeral
-    verification, not ordinary delivery.
+    reviewable `titan-iac` branch and validate Kustomize and client dry-run.
+    Brad owns merge and routine Flux reconciliation; perform either only under
+    a separate explicit operator request. Direct `kubectl` mutations are for
+    explicit incident recovery or ephemeral verification, not ordinary
+    delivery.
 
     Node SSH uses a dedicated audited Hermes identity: `ssh titan-04` (or any
     current Kubernetes node name). Host keys are pinned and password fallback

services/hermes/agent-deployment.yaml

@@ -662,6 +662,7 @@ spec:
             - {name: subprocess-secret-patch, mountPath: /opt/hermes/tools/process_registry.py, subPath: process_registry.py}
             - {name: kubeconfig, mountPath: /opt/data/home/.kube/config, subPath: config, readOnly: true}
             - {name: auto-router-plugin, mountPath: /opt/data/plugins/auto-router, readOnly: true}
+            - {name: atlas-pr-skill, mountPath: /opt/data/workspace/skills/manage-atlas-pull-requests, readOnly: true}
             - {name: routing-catalog, mountPath: /routing-catalog, readOnly: true}
             - {name: tmp, mountPath: /tmp}
           startupProbe:
@@ -825,6 +826,7 @@ spec:
             - {name: subprocess-secret-patch, mountPath: /opt/hermes/tools/process_registry.py, subPath: process_registry.py}
             - {name: kubeconfig, mountPath: /opt/data/home/.kube/config, subPath: config, readOnly: true}
             - {name: auto-router-plugin, mountPath: /opt/data/plugins/auto-router, readOnly: true}
+            - {name: atlas-pr-skill, mountPath: /opt/data/workspace/skills/manage-atlas-pull-requests, readOnly: true}
             - {name: routing-catalog, mountPath: /routing-catalog, readOnly: true}
             - {name: tmp, mountPath: /tmp}
             - {name: ttyd-index, mountPath: /ttyd-index, readOnly: true}
@@ -893,6 +895,7 @@ spec:
             - {name: auth-patch, mountPath: /opt/hermes/hermes_cli/auth.py, subPath: auth.py}
             - {name: kubeconfig, mountPath: /opt/data/home/.kube/config, subPath: config, readOnly: true}
             - {name: auto-router-plugin, mountPath: /opt/data/plugins/auto-router, readOnly: true}
+            - {name: atlas-pr-skill, mountPath: /opt/data/workspace/skills/manage-atlas-pull-requests, readOnly: true}
             - {name: tmp, mountPath: /tmp}
           resources:
             requests: {cpu: 100m, memory: 256Mi}
@@ -1187,6 +1190,12 @@ spec:
               - {key: dashboard-api.py, path: dashboard/plugin_api.py}
               - {key: dashboard-index.js, path: dashboard/dist/index.js}
               - {key: dashboard-style.css, path: dashboard/dist/style.css}
+        - name: atlas-pr-skill
+          configMap:
+            name: hermes-atlas-pr-skill
+            items:
+              - {key: SKILL.md, path: SKILL.md}
+              - {key: openai.yaml, path: agents/openai.yaml}
         - name: image-policy
           configMap:
             name: hermes-image-policy

services/hermes/kustomization.yaml

@@ -70,6 +70,7 @@ configMapGenerator:
       - claude_oauth_broker.py=scripts/claude_oauth_broker.py
       - worker_route_broker.py=scripts/worker_route_broker.py
       - gitea_api.py=scripts/gitea_api.py
+      - gitea_api_policy.py=scripts/gitea_api_policy.py
       - gitea_askpass.sh=scripts/gitea_askpass.sh
       - hermes_coordinator.py=scripts/hermes_coordinator.py
       - hermes_model_routing.py=scripts/hermes_model_routing.py
@@ -158,3 +159,10 @@ configMapGenerator:
       - alert-review.md=skills/tune-atlas-alerts/references/alert-review.md
     options:
       disableNameSuffixHash: true
+  - name: hermes-atlas-pr-skill
+    namespace: hermes
+    files:
+      - SKILL.md=skills/manage-atlas-pull-requests/SKILL.md
+      - openai.yaml=skills/manage-atlas-pull-requests/agents/openai.yaml
+    options:
+      disableNameSuffixHash: true

services/hermes/scripts/gitea_api.py

@@ -1,46 +1,190 @@
 #!/usr/bin/env python3
-"""Call the private Atlas Gitea API through a runtime-only token boundary."""
+"""Use the private Atlas Forgejo API through a least-authority PR boundary."""
 
 from __future__ import annotations
 
 import argparse
 import json
 import os
+import re
 import sys
 import urllib.error
 import urllib.parse
 import urllib.request
+from collections.abc import Callable
 from pathlib import Path
 
+from gitea_api_policy import (
+    PolicyError,
+    _draft_title,
+    _validate_body,
+    _validate_pr_number,
+    _validate_pr_number_segment,
+    _validate_query,
+    _validate_ref,
+    _validate_ref_bounds,
+    _validate_repo,
+    _validate_sha,
+)
 
-DEFAULT_BASE_URL = "https://scm.bstein.dev"
+CANONICAL_BASE_URL = "https://scm.bstein.dev"
+ALLOWED_OWNER = "atlas"
 DEFAULT_TOKEN_FILE = Path("/runtime-access/gitea-token")
-ALLOWED_METHODS = ("GET", "POST", "PUT", "PATCH", "DELETE")
+CREATE_PATH_RE = re.compile(r"/api/v1/repos/atlas/([^/]+)/pulls\Z")
+MAX_ERROR_BYTES = 8192
+MAX_RESPONSE_BYTES = 2 * 1024 * 1024
+MAX_API_TARGET_LENGTH = 768
+MAX_API_PATH_LENGTH = 512
+RAW_API_TARGET_RE = re.compile(r"[A-Za-z0-9/_.?&=-]+\Z")
+
+
+class RejectRedirectHandler(urllib.request.HTTPRedirectHandler):
+    """Reject every redirect before urllib can copy authentication headers."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        raise PolicyError("Forgejo redirects are not allowed")
+
+
+_SAFE_OPENER = urllib.request.build_opener(RejectRedirectHandler())
+
+
+def _safe_urlopen(request: urllib.request.Request, timeout: int):
+    """Open one request using a handler that never follows redirects."""
+    return _SAFE_OPENER.open(request, timeout=timeout)
 
 
 def read_token(path: Path = DEFAULT_TOKEN_FILE) -> str:
-    """Read and validate the token from its in-memory Vault projection."""
+    """Read the token from the pod-lifetime Vault projection."""
     token = path.read_text(encoding="utf-8").strip()
     if not token:
         raise ValueError(f"runtime credential is empty: {path}")
     return token
 
 
-def api_url(base_url: str, path: str) -> str:
-    """Return a same-origin Gitea API URL for a validated API path."""
-    base = urllib.parse.urlsplit(base_url.rstrip("/"))
+def configured_base_url() -> str:
+    """Reject attempts to redirect credentials away from the canonical origin."""
+    configured = os.environ.get("GITEA_BASE_URL", CANONICAL_BASE_URL)
+    if configured != CANONICAL_BASE_URL:
+        raise PolicyError("Forgejo origin is fixed to the private Atlas SCM service")
+    return configured
+
+
+def _split_api_path(path: str) -> urllib.parse.SplitResult:
+    if not isinstance(path, str) or len(path) > MAX_API_TARGET_LENGTH:
+        raise PolicyError("API target exceeds the safe size limit")
+    if (
+        not path.isascii()
+        or any(ord(character) < 32 or ord(character) == 127 for character in path)
+        or "\\" in path
+        or not RAW_API_TARGET_RE.fullmatch(path)
+    ):
+        raise PolicyError("API target contains non-canonical raw characters")
     target = urllib.parse.urlsplit(path)
-    if base.scheme not in {"http", "https"} or not base.netloc:
-        raise ValueError("GITEA_BASE_URL must be an absolute HTTP(S) URL")
+    canonical = urllib.parse.urlunsplit(("", "", target.path, target.query, ""))
+    if canonical != path:
+        raise PolicyError("API target is not in exact canonical form")
     if target.scheme or target.netloc or target.fragment:
-        raise ValueError("API path must be relative to the configured Gitea origin")
+        raise PolicyError("API path must be relative to the Atlas SCM origin")
     if not target.path.startswith("/api/v1/"):
-        raise ValueError("API path must start with /api/v1/")
+        raise PolicyError("API path must start with /api/v1/")
+    if len(target.path) > MAX_API_PATH_LENGTH:
+        raise PolicyError("API path exceeds the safe size limit")
+    segments = target.path.split("/")
+    if "//" in target.path or any(segment in {".", ".."} for segment in segments):
+        raise PolicyError("encoded or non-canonical API paths are not allowed")
+    return target
+
+
+def _authorize_read(target: urllib.parse.SplitResult) -> str:
+    """Allow only repository, PR, branch, commit, and status metadata reads."""
+    prefix = "/api/v1/repos/atlas/"
+    remainder = target.path.removeprefix(prefix)
+    if remainder == target.path:
+        raise PolicyError("reads are limited to explicit Atlas repository metadata")
+    repo, separator, suffix = remainder.partition("/")
+    _validate_repo(repo)
+
+    if not separator:
+        _validate_query(target, set())
+        return "repository"
+    if suffix == "pulls":
+        _validate_query(target, {"page", "limit", "state"})
+        return "pull-list"
+    content_match = re.fullmatch(r"pulls/([^/]+)\.(?:patch|diff)", suffix)
+    if content_match:
+        _validate_pr_number_segment(content_match.group(1))
+        raise PolicyError("repository API route is outside the metadata read allowlist")
+    pull_match = re.fullmatch(r"pulls/([^/]+)", suffix)
+    if pull_match:
+        _validate_pr_number_segment(pull_match.group(1))
+        _validate_query(target, set())
+        return "pull"
+    evidence_match = re.fullmatch(r"pulls/([^/]+)/(?:commits|files)", suffix)
+    if evidence_match:
+        _validate_pr_number_segment(evidence_match.group(1))
+        _validate_query(target, {"page", "limit"})
+        return "pull-evidence"
+    if suffix == "branches":
+        _validate_query(target, {"page", "limit"})
+        return "branch-list"
+    branch_match = re.fullmatch(r"branches/([^/]+)", suffix)
+    if branch_match:
+        _validate_ref(branch_match.group(1), "branch")
+        _validate_query(target, set())
+        return "branch"
+    if suffix == "commits":
+        _validate_query(target, {"page", "limit"})
+        return "commit-list"
+    if re.fullmatch(r"git/commits/[0-9a-fA-F]{40}", suffix):
+        _validate_query(target, set())
+        return "commit"
+    if re.fullmatch(
+        r"(?:commits/[0-9a-fA-F]{40}/status|commits/[0-9a-fA-F]{40}/statuses|statuses/[0-9a-fA-F]{40})",
+        suffix,
+    ):
+        _validate_query(target, {"page", "limit"})
+        return "status"
+    raise PolicyError("repository API route is outside the metadata read allowlist")
+
+
+def api_url(base_url: str, path: str) -> str:
+    """Return a canonical same-origin URL without forwarding credentials."""
+    if base_url != CANONICAL_BASE_URL:
+        raise PolicyError("Forgejo origin is fixed to the private Atlas SCM service")
+    target = _split_api_path(path)
     return urllib.parse.urlunsplit(
-        (base.scheme, base.netloc, target.path, target.query, "")
+        ("https", "scm.bstein.dev", target.path, target.query, "")
     )
 
 
+def authorize_request(method: str, path: str, data: object | None) -> str:
+    """Validate one request and return its bounded operation name."""
+    normalized_method = method.upper()
+    target = _split_api_path(path)
+    if normalized_method == "GET":
+        if data is not None:
+            raise PolicyError("read operations cannot include a request body")
+        return _authorize_read(target)
+    if target.query:
+        raise PolicyError("mutating operations cannot include query parameters")
+    if normalized_method != "POST":
+        raise PolicyError("only read and create-draft operations are available")
+    match = CREATE_PATH_RE.fullmatch(target.path)
+    if not match:
+        raise PolicyError("POST is limited to creating an Atlas draft pull request")
+    _validate_repo(match.group(1))
+    if not isinstance(data, dict) or set(data) != {"base", "body", "head", "title"}:
+        raise PolicyError("draft creation accepts only base, body, head, and title")
+    if data["title"] != _draft_title(data["title"]):
+        raise PolicyError("new pull requests must use the Gitea draft-title prefix")
+    _validate_body(data["body"])
+    _validate_ref_bounds(data["base"], "base")
+    _validate_ref_bounds(data["head"], "head")
+    _validate_ref(data["base"], "base")
+    _validate_ref(data["head"], "head")
+    return "create-draft"
+
+
 def build_request(
     method: str,
     path: str,
@@ -49,100 +193,252 @@ def build_request(
     token: str,
     data: object | None = None,
 ) -> urllib.request.Request:
-    """Build one authenticated request without placing the token in its URL."""
+    """Build an authorized request without putting the token in its URL or body."""
+    authorize_request(method, path, data)
+    if method.upper() == "POST":
+        assert isinstance(data, dict)
+        _draft_title(data["title"], forbidden=(token,))
+        _validate_body(data["body"], forbidden=(token,))
     payload = None
     if data is not None:
         payload = json.dumps(data, separators=(",", ":")).encode("utf-8")
     return urllib.request.Request(
         api_url(base_url, path),
         data=payload,
-        method=method,
+        method=method.upper(),
         headers={
             "Accept": "application/json",
             "Authorization": f"token {token}",
             "Content-Type": "application/json",
-            "User-Agent": "hermes-atlas-operator/1",
+            "User-Agent": "hermes-atlas-pr-client/2",
         },
     )
 
 
-def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
-    """Parse a bounded method, API path, and optional JSON request body."""
-    parser = argparse.ArgumentParser(
-        description="Call the Atlas Gitea API using the runtime Vault token."
+def redact_bytes(value: bytes, token: str) -> bytes:
+    """Remove exact runtime-token and authorization-header values from output."""
+    redacted = value.replace(token.encode("utf-8"), b"[REDACTED]")
+    return re.sub(
+        rb"(?i)(authorization\s*[:=]\s*)(?:token|bearer)?\s*[^\s,;\"}]+",
+        rb"\1[REDACTED]",
+        redacted,
     )
-    parser.add_argument("method", choices=ALLOWED_METHODS)
-    parser.add_argument("path", help="Gitea path beginning with /api/v1/")
-    data_group = parser.add_mutually_exclusive_group()
-    data_group.add_argument("--data-json", help="JSON object/array request body")
-    data_group.add_argument(
-        "--data-file", type=Path, help="path to a JSON request body"
-    )
-    data_group.add_argument(
-        "--field",
-        action="append",
-        metavar="KEY=VALUE",
-        help=(
-            "repeatable JSON field; bare text remains a string while true, false, "
-            "null, numbers, objects, and arrays are decoded as JSON"
-        ),
-    )
-    return parser.parse_args(argv)
 
 
-def parse_fields(values: list[str]) -> dict[str, object]:
-    """Build a JSON object from shell-safe, repeatable key/value arguments."""
-    result: dict[str, object] = {}
-    for value in values:
-        key, separator, raw = value.partition("=")
-        if not separator or not key or any(char.isspace() for char in key):
-            raise ValueError("each --field must be KEY=VALUE with a non-space key")
-        try:
-            result[key] = json.loads(raw)
-        except json.JSONDecodeError:
-            result[key] = raw
+def _request(
+    method: str,
+    path: str,
+    data: object | None,
+    *,
+    token: str,
+    opener: Callable[..., object] = _safe_urlopen,
+) -> bytes:
+    request = build_request(
+        method, path, base_url=configured_base_url(), token=token, data=data
+    )
+    with opener(request, timeout=30) as response:  # type: ignore[attr-defined]
+        body = response.read(MAX_RESPONSE_BYTES + 1)  # type: ignore[attr-defined]
+    if len(body) > MAX_RESPONSE_BYTES:
+        raise PolicyError("Forgejo response exceeds the safe size limit")
+    return redact_bytes(body, token)
+
+
+def read(
+    path: str, *, token: str, opener: Callable[..., object] = _safe_urlopen
+) -> bytes:
+    """Read one explicitly allowed Atlas repository metadata resource."""
+    return _request("GET", path, None, token=token, opener=opener)
+
+
+def _nested(document: dict[str, object], *keys: str) -> object:
+    value: object = document
+    for key in keys:
+        if not isinstance(value, dict) or key not in value:
+            raise PolicyError("Forgejo omitted required pull-request evidence")
+        value = value[key]
+    return value
+
+
+def _require_create_response(
+    value: bytes,
+    *,
+    repo: str,
+    base: str,
+    head: str,
+    head_sha: str,
+    title: str,
+    body: str,
+) -> dict[str, object]:
+    """Require exact server evidence for the intended open draft handoff."""
+    try:
+        document = json.loads(value)
+    except json.JSONDecodeError as exc:
+        raise PolicyError("Forgejo returned invalid pull-request metadata") from exc
+    if not isinstance(document, dict):
+        raise PolicyError("Forgejo returned invalid pull-request metadata")
+    number = _validate_pr_number(document.get("number"))
+    full_name = f"{ALLOWED_OWNER}/{repo}"
+    expected_html = f"{CANONICAL_BASE_URL}/{full_name}/pulls/{number}"
+    checks = (
+        document.get("state") == "open",
+        document.get("draft") is True,
+        document.get("merged") is False,
+        document.get("html_url") == expected_html,
+        document.get("url") == expected_html,
+        document.get("title") == title,
+        document.get("body") == body,
+        _nested(document, "base", "ref") == base,
+        _nested(document, "base", "repo", "full_name") == full_name,
+        _nested(document, "head", "ref") == head,
+        _nested(document, "head", "sha") == head_sha,
+        _nested(document, "head", "repo", "full_name") == full_name,
+    )
+    if not all(checks):
+        raise PolicyError("Forgejo did not confirm the exact draft pull-request state")
+    return document
+
+
+def create_draft(
+    repo: str,
+    *,
+    base: str,
+    head: str,
+    head_sha: str,
+    title: str,
+    body: str,
+    token: str,
+    opener: Callable[..., object] = _safe_urlopen,
+) -> bytes:
+    """Create and verify a same-repository draft pull request for human review."""
+    repo = _validate_repo(repo)
+    title = _draft_title(title, forbidden=(token,))
+    body = _validate_body(body, forbidden=(token,))
+    head_sha = _validate_sha(head_sha)
+    _validate_ref_bounds(base, "base")
+    _validate_ref_bounds(head, "head")
+    base = _validate_ref(base, "base")
+    head = _validate_ref(head, "head")
+    data = {"base": base, "body": body, "head": head, "title": title}
+    result = _request(
+        "POST",
+        f"/api/v1/repos/{ALLOWED_OWNER}/{repo}/pulls",
+        data,
+        token=token,
+        opener=opener,
+    )
+    _require_create_response(
+        result,
+        repo=repo,
+        base=base,
+        head=head,
+        head_sha=head_sha,
+        title=title,
+        body=body,
+    )
     return result
 
 
-def load_data(args: argparse.Namespace) -> object | None:
-    """Decode the optional JSON body without involving a shell expansion."""
-    if args.data_json is not None:
-        return json.loads(args.data_json)
-    if args.data_file is not None:
-        return json.loads(args.data_file.read_text(encoding="utf-8"))
-    if args.field is not None:
-        return parse_fields(args.field)
-    return None
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    """Parse the intentionally narrow read and create-draft operations."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="validate and describe the request without reading a token or using the network",
+    )
+    subparsers = parser.add_subparsers(dest="operation", required=True)
+    read_parser = subparsers.add_parser("read", help="read allowed Atlas metadata")
+    read_parser.add_argument("path")
+    create_parser = subparsers.add_parser(
+        "create-draft", help="create a same-repository draft pull request"
+    )
+    create_parser.add_argument("repo")
+    create_parser.add_argument("--base", required=True)
+    create_parser.add_argument("--head", required=True)
+    create_parser.add_argument("--head-sha", required=True)
+    create_parser.add_argument("--title", required=True)
+    create_parser.add_argument("--body", default="")
+    return parser.parse_args(argv)
+
+
+def _operation(args: argparse.Namespace) -> tuple[str, str, object | None]:
+    if args.operation == "read":
+        return "GET", args.path, None
+    repo = _validate_repo(args.repo)
+    _validate_sha(args.head_sha)
+    body = _validate_body(args.body)
+    title = _draft_title(args.title)
+    _validate_ref_bounds(args.base, "base")
+    _validate_ref_bounds(args.head, "head")
+    data = {
+        "base": _validate_ref(args.base, "base"),
+        "body": body,
+        "head": _validate_ref(args.head, "head"),
+        "title": title,
+    }
+    return "POST", f"/api/v1/repos/{ALLOWED_OWNER}/{repo}/pulls", data
+
+
+def _write_body(body: bytes) -> None:
+    if body:
+        sys.stdout.buffer.write(body)
+        if not body.endswith(b"\n"):
+            sys.stdout.buffer.write(b"\n")
 
 
 def main(argv: list[str] | None = None) -> int:
-    """Execute the request, print only its response body, and return HTTP status."""
-    args = parse_args(argv)
+    """Validate, optionally describe, and execute one safe Forgejo operation."""
+    token = ""
     try:
-        request = build_request(
-            args.method,
-            args.path,
-            base_url=os.environ.get("GITEA_BASE_URL", DEFAULT_BASE_URL),
-            token=read_token(),
-            data=load_data(args),
-        )
-        with urllib.request.urlopen(request, timeout=30) as response:
-            body = response.read()
-            if body:
-                sys.stdout.buffer.write(body)
-                if not body.endswith(b"\n"):
-                    sys.stdout.buffer.write(b"\n")
+        args = parse_args(argv)
+        method, path, data = _operation(args)
+        operation = authorize_request(method, path, data)
+        if args.dry_run:
+            print(
+                json.dumps(
+                    {
+                        "dry_run": True,
+                        "host": "scm.bstein.dev",
+                        "operation": operation,
+                        "owner": ALLOWED_OWNER,
+                        "path": path,
+                        "request_fields": sorted(data)
+                        if isinstance(data, dict)
+                        else [],
+                    },
+                    sort_keys=True,
+                )
+            )
+            return 0
+        token = read_token()
+        if args.operation == "read":
+            body = read(path, token=token)
+        else:
+            assert isinstance(data, dict)
+            body = create_draft(
+                args.repo,
+                base=str(data["base"]),
+                head=str(data["head"]),
+                head_sha=args.head_sha,
+                title=str(data["title"]),
+                body=str(data["body"]),
+                token=token,
+            )
+        _write_body(body)
         return 0
     except urllib.error.HTTPError as exc:
-        body = exc.read(65536)
-        print(f"Gitea API returned HTTP {exc.code}", file=sys.stderr)
+        body = redact_bytes(exc.read(MAX_ERROR_BYTES), token) if token else b""
+        print(f"Forgejo request failed with HTTP {exc.code}", file=sys.stderr)
         if body:
             sys.stderr.buffer.write(body)
             if not body.endswith(b"\n"):
                 sys.stderr.buffer.write(b"\n")
         return 1
-    except (OSError, ValueError, json.JSONDecodeError) as exc:
-        print(f"Gitea API request failed: {exc}", file=sys.stderr)
+    except (OSError, PolicyError, ValueError, json.JSONDecodeError):
+        print(
+            "Forgejo request rejected or unavailable; no credential was disclosed",
+            file=sys.stderr,
+        )
         return 1
 
 

services/hermes/scripts/gitea_api_policy.py

@@ -0,0 +1,475 @@
+"""Validation policy for Hermes' least-authority Atlas Forgejo client.
+
+The screening catches structured credential assignments, known token formats,
+and long high-entropy values. It is a fail-closed accident barrier, not proof
+that text is secret-free: a short or multiword secret under an innocuous key is
+not reliably distinguishable from prose and must never be supplied by callers.
+JSON depth/nodes/documents and multiline gaps are deliberately bounded; large
+or unusual configuration blobs are not a supported pull-request field format.
+"""
+
+# ruff: noqa: SIM905
+
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+import urllib.parse
+from collections import Counter
+from math import log2
+
+REPO_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}\Z")
+SHA_RE = re.compile(r"[0-9a-fA-F]{40}\Z")
+DRAFT_TITLE_PREFIX = "WIP: "
+GIT_BIN = "/usr/bin/git"
+MAX_QUERY_LENGTH = 128
+MAX_QUERY_ITEMS = 3
+MAX_QUERY_KEY_LENGTH = 16
+MAX_QUERY_VALUE_LENGTH = 16
+MAX_PAGE = 10_000
+MAX_LIMIT = 50
+MAX_PR_NUMBER = 2_147_483_647
+MAX_PR_NUMBER_DIGITS = 10
+MAX_REF_CHARACTERS = 200
+MAX_REF_UTF8_BYTES = 255
+MAX_TITLE_UTF8_BYTES = 512
+MAX_BODY_UTF8_BYTES = 32_768
+MAX_JSON_DOCUMENTS = 32
+MAX_JSON_NODES = 2_048
+MAX_JSON_DEPTH = 32
+CANONICAL_QUERY_RE = re.compile(r"[A-Za-z0-9_=&-]*\Z")
+ASSIGNMENT_RE = re.compile(
+    r"""(?mx)
+    (?<![A-Za-z0-9_.-])
+    (?P<key_quote>[\"']?)
+    (?P<key>\.?[A-Za-z][A-Za-z0-9_.-]{0,127})
+    (?P=key_quote)
+    [ \t]*(?P<separator>[:=])[ \t]*
+    (?P<value>
+      \"(?:\\.|[^\"\\\r\n])*\" |
+      '(?:\\.|[^'\\\r\n])*' |
+      [^\r\n,;}&]{0,2048}
+    )
+    """
+)
+JSON_KEY_RE = re.compile(r'"(?P<key>[^"]{1,512})"(?P<spacing>[\x00-\x20\x7f]{0,256}):')
+MULTILINE_ASSIGNMENT_RE = re.compile(
+    r"""(?mx)
+    (?<![A-Za-z0-9_.-])
+    (?P<key>\.?[A-Za-z][A-Za-z0-9_.-]{0,127})
+    [ \t]{0,64}(?:\r?\n[ \t]{0,64}){0,4}
+    (?P<separator>[:=])
+    [ \t]{0,64}(?:\r?\n[ \t]{0,64}){1,4}
+    (?P<value>[^\r\n,;}&]{1,2048})
+    """
+)
+CAMEL_BOUNDARY_RE = re.compile(r"(?<=[a-z0-9])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])")
+CREDENTIAL_SCHEME_RE = re.compile(
+    r"(?i)\b(?:bearer|basic|token)\s+"
+    r"(?=[A-Za-z0-9+/_=.:-]{8,}\b)(?=[A-Za-z0-9+/_=.:-]*[0-9+/_=.-])"
+    r"[A-Za-z0-9+/_=.:-]{8,}"
+)
+STANDALONE_SECRET_PATTERNS = tuple(
+    re.compile(pattern, re.IGNORECASE)
+    for pattern in (
+        r"\bgh[pousr]_[A-Za-z0-9]{20,}\b",
+        r"\bgithub_pat_[A-Za-z0-9_]{20,}\b",
+        r"\bglpat-[A-Za-z0-9_-]{20,}\b",
+        r"\bxox[baprs]-[A-Za-z0-9-]{20,}\b",
+        r"\bsk-(?:ant-|proj-)?[A-Za-z0-9_-]{20,}\b",
+        r"\b[rs]k_(?:test|live)_[A-Za-z0-9]{20,}\b",
+        r"\bwhsec_[A-Za-z0-9]{20,}\b",
+        r"\bnpm_[A-Za-z0-9]{20,}\b",
+        r"\bpypi-[A-Za-z0-9_-]{30,}\b",
+        r"\bhf_[A-Za-z0-9]{20,}\b",
+        r"\b(?:gta|gto|gitea|forgejo)_[A-Za-z0-9_-]{20,}\b",
+        r"\bya29\.[A-Za-z0-9_-]{20,}\b",
+        r"\boy2[A-Za-z0-9]{40,}\b",
+        r"\bSG\.[A-Za-z0-9_-]{16,}\.[A-Za-z0-9_-]{20,}\b",
+        r"\bSK[0-9a-f]{32}\b",
+        r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b",
+        r"\bAIza[0-9A-Za-z_-]{35}\b",
+        r"\beyJ[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{8,}\b",
+        r"-----BEGIN\s+[A-Z0-9 ][A-Z0-9 -]{1,62}-----",
+        r"\bssh-(?:rsa|ed25519|dss|ecdsa-[A-Za-z0-9-]+)\s+[A-Za-z0-9+/]{16,}={0,3}",
+        r"https://hooks\.slack\.com/services/[A-Za-z0-9/_-]{20,}",
+        r"https://(?:discord(?:app)?\.com)/api/webhooks/[0-9]+/[A-Za-z0-9._-]{20,}",
+        r"https://[^\s/]*webhook\.office\.com/[^\s]{20,}",
+    )
+)
+HIGH_ENTROPY_TOKEN_RE = re.compile(
+    r"(?<![A-Za-z0-9+/_=-])[A-Za-z0-9+/_=-]{48,16384}(?![A-Za-z0-9+/_=-])"
+)
+PROSE_LEAD_WORDS = frozenset(
+    "add allow change check describe document ensure explain fix handle keep preserve "
+    "prevent reject remove review support test validate".split()
+)
+SENSITIVE_KEY_WORDS = frozenset(
+    "auth authentication authorization auths basic bearer credential credentials "
+    "passwd password pat private sas secret secrets sig signature signing token tokens "
+    "webhook".split()
+)
+KEY_MODIFIER_WORDS = frozenset(
+    "access account api auth client encryption identity private registry secret service "
+    "session signing ssh".split()
+)
+SENSITIVE_COMPACT_KEYS = frozenset(
+    "accessid accesskeyid accountkey clientemail clientid connectionstring "
+    "dockerconfigjson privatekeyid serviceaccountkey sharedaccesssignature".split()
+)
+SENSITIVE_COMPACT_SUFFIXES = tuple(
+    "accesskey accesskeyid accountkey apikey authkey clientemail clientid clientkey "
+    "connectionstring credential credentials encryptionkey identitykey password passwd "
+    "privatekeyid privatekey secret secretkey servicekey sessionkey signature signingkey "
+    "sshkey token webhook".split()
+)
+
+
+class PolicyError(ValueError):
+    """Raised when a requested Forgejo operation exceeds the safe boundary."""
+
+
+def _validate_repo(repo: str) -> str:
+    if not REPO_RE.fullmatch(repo) or repo in {".", ".."}:
+        raise PolicyError("repository name is outside the Atlas allowlist")
+    return repo
+
+
+def _validate_ref_bounds(value: object, name: str) -> str:
+    """Reject oversized or non-encodable refs before invoking Git."""
+    if not isinstance(value, str) or not value or len(value) > MAX_REF_CHARACTERS:
+        raise PolicyError(f"{name} exceeds the safe branch-name limit")
+    try:
+        encoded = value.encode("utf-8", errors="strict")
+    except UnicodeEncodeError as exc:
+        raise PolicyError(f"{name} is not valid UTF-8 text") from exc
+    if len(encoded) > MAX_REF_UTF8_BYTES:
+        raise PolicyError(f"{name} exceeds the safe UTF-8 branch-name limit")
+    return value
+
+
+def _validate_ref(value: object, name: str) -> str:
+    """Validate the complete Git ref grammar using Git itself."""
+    value = _validate_ref_bounds(value, name)
+    if value.startswith("-"):
+        raise PolicyError(f"{name} must be a same-repository branch name")
+    if any(ord(character) < 32 or ord(character) == 127 for character in value):
+        raise PolicyError(f"{name} must be a safe same-repository branch name")
+    result = subprocess.run(
+        [GIT_BIN, "check-ref-format", f"refs/heads/{value}"],
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+        check=False,
+    )
+    if result.returncode != 0:
+        raise PolicyError(f"{name} must be a safe same-repository branch name")
+    return value
+
+
+def _validate_sha(value: object, name: str = "head SHA") -> str:
+    if not isinstance(value, str) or not SHA_RE.fullmatch(value):
+        raise PolicyError(f"{name} must be a full Git SHA-1")
+    return value.lower()
+
+
+def _validate_pr_number_segment(value: object) -> int:
+    """Validate a canonical bounded pull-request number from an API path."""
+    if (
+        not isinstance(value, str)
+        or len(value) > MAX_PR_NUMBER_DIGITS
+        or not re.fullmatch(r"[1-9][0-9]*", value)
+    ):
+        raise PolicyError("pull-request number must use bounded canonical ASCII digits")
+    number = int(value)
+    if number > MAX_PR_NUMBER:
+        raise PolicyError("pull-request number is outside its allowed range")
+    return number
+
+
+def _validate_pr_number(value: object) -> int:
+    """Validate a bounded pull-request number returned by Forgejo."""
+    if (
+        not isinstance(value, int)
+        or isinstance(value, bool)
+        or not 1 <= value <= MAX_PR_NUMBER
+    ):
+        raise PolicyError("Forgejo omitted a valid pull-request number")
+    return value
+
+
+def _validate_text(
+    value: object,
+    name: str,
+    maximum: int,
+    maximum_bytes: int,
+    *,
+    required: bool,
+) -> str:
+    if not isinstance(value, str):
+        raise PolicyError(f"{name} must be text")
+    if required and not value.strip():
+        raise PolicyError(f"{name} must not be empty")
+    try:
+        encoded = value.encode("utf-8", errors="strict")
+    except UnicodeEncodeError as exc:
+        raise PolicyError(f"{name} is not valid UTF-8 text") from exc
+    if len(value) > maximum or len(encoded) > maximum_bytes or "\x00" in value:
+        raise PolicyError(f"{name} exceeds the safe request limit")
+    return value
+
+
+def _normalized_key(value: str) -> tuple[tuple[str, ...], str]:
+    """Split camelCase and separator-based assignment keys into semantics."""
+    separated = CAMEL_BOUNDARY_RE.sub(" ", value.strip(".\"'"))
+    words = tuple(re.findall(r"[A-Za-z0-9]+", separated.lower()))
+    return words, "".join(words)
+
+
+def _is_sensitive_key(value: str) -> bool:
+    words, compact = _normalized_key(value)
+    word_set = set(words)
+    if word_set & SENSITIVE_KEY_WORDS:
+        return True
+    if compact in SENSITIVE_COMPACT_KEYS:
+        return True
+    if compact.endswith(SENSITIVE_COMPACT_SUFFIXES):
+        return True
+    if "key" in word_set and word_set & KEY_MODIFIER_WORDS:
+        return True
+    if {"client", "email"} <= word_set or {"client", "id"} <= word_set:
+        return True
+    if {"access", "id"} <= word_set or {"connection", "string"} <= word_set:
+        return True
+    return compact.endswith("configjson")
+
+
+def _strip_assignment_value(value: str) -> tuple[str, bool]:
+    stripped = value.strip()
+    quoted = (
+        len(stripped) >= 2 and stripped[0] in {'"', "'"} and stripped[-1] == stripped[0]
+    )
+    if quoted:
+        stripped = stripped[1:-1].strip()
+    return stripped, quoted
+
+
+def _looks_like_prose(value: str) -> bool:
+    words = re.findall(r"[A-Za-z]+(?:[-'][A-Za-z]+)?", value)
+    if len(words) < 2:
+        return False
+    if re.search(r"[$`{}\[\]\\@/:=+_]", value):
+        return False
+    return words[0].lower() in PROSE_LEAD_WORDS or len(words) >= 4
+
+
+def _looks_sensitive_assignment_value(value: str) -> bool:
+    stripped, quoted = _strip_assignment_value(value)
+    if not stripped:
+        return False
+    if quoted or stripped.startswith(("{", "[", "|", ">", "!!", "&", "*")):
+        return True
+    if CREDENTIAL_SCHEME_RE.search(stripped):
+        return True
+    if any(pattern.search(stripped) for pattern in STANDALONE_SECRET_PATTERNS):
+        return True
+    if re.search(r"(?:https?://|[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+)", stripped):
+        return True
+    if re.search(r"(?:\$\{|\$\(|\\[nrt]|[A-Za-z0-9+/]{16,}={0,3}\Z)", stripped):
+        return True
+    return not _looks_like_prose(stripped)
+
+
+def _is_structural_credential_assignment(key: str, value: str) -> bool:
+    words, compact_key = _normalized_key(key)
+    stripped, _quoted = _strip_assignment_value(value)
+    _value_words, compact_value = _normalized_key(stripped)
+    return (
+        compact_key == "type" and compact_value in {"serviceaccount", "credential"}
+    ) or ("service" in words and "account" in words and bool(stripped))
+
+
+def _decode_json_key(raw: str) -> tuple[str, bool]:
+    """Decode one bounded JSON key without silently accepting bad escapes."""
+    try:
+        decoded = json.loads(f'"{raw}"')
+    except (json.JSONDecodeError, UnicodeError):
+        approximate = re.sub(r"[^A-Za-z0-9]+", "_", raw.replace("\\", ""))
+        return approximate, False
+    return decoded if isinstance(decoded, str) else "", True
+
+
+def _json_value_has_sensitive_assignment(
+    value: object, *, depth: int = 0, nodes: list[int] | None = None
+) -> bool:
+    """Walk one decoded JSON value with explicit depth and node ceilings."""
+    if nodes is None:
+        nodes = [0]
+    nodes[0] += 1
+    if nodes[0] > MAX_JSON_NODES or depth > MAX_JSON_DEPTH:
+        raise PolicyError("structured pull-request text exceeds the scan limit")
+    if isinstance(value, dict):
+        for key, assigned in value.items():
+            if not isinstance(key, str):
+                continue
+            if _is_sensitive_key(key):
+                return True
+            assigned_text = assigned if isinstance(assigned, str) else ""
+            if _is_structural_credential_assignment(key, assigned_text):
+                return True
+            if _json_value_has_sensitive_assignment(
+                assigned, depth=depth + 1, nodes=nodes
+            ):
+                return True
+    elif isinstance(value, list):
+        return any(
+            _json_value_has_sensitive_assignment(item, depth=depth + 1, nodes=nodes)
+            for item in value
+        )
+    return False
+
+
+def _decoded_json_documents(value: str):
+    """Yield bounded embedded JSON object or array fragments."""
+    decoder = json.JSONDecoder()
+    index = 0
+    attempts = 0
+    while attempts < MAX_JSON_DOCUMENTS:
+        positions = [
+            position for token in "{[" if (position := value.find(token, index)) >= 0
+        ]
+        if not positions:
+            return
+        start = min(positions)
+        attempts += 1
+        try:
+            document, end = decoder.raw_decode(value, start)
+        except (json.JSONDecodeError, RecursionError, ValueError):
+            index = start + 1
+            continue
+        yield document
+        index = max(end, start + 1)
+
+
+def _has_structured_sensitive_assignment(value: str) -> bool:
+    """Detect bounded JSON and multiline YAML/env credential assignments."""
+    for candidate in JSON_KEY_RE.finditer(value):
+        decoded, valid = _decode_json_key(candidate.group("key"))
+        if _is_sensitive_key(decoded):
+            return True
+        if not valid and _is_sensitive_key(candidate.group("key").replace("\\", "")):
+            return True
+    for candidate in MULTILINE_ASSIGNMENT_RE.finditer(value):
+        key = candidate.group("key")
+        assigned = candidate.group("value")
+        if _is_structural_credential_assignment(key, assigned) or (
+            _is_sensitive_key(key) and _looks_sensitive_assignment_value(assigned)
+        ):
+            return True
+    return any(
+        _json_value_has_sensitive_assignment(item)
+        for item in _decoded_json_documents(value)
+    )
+
+
+def _has_high_entropy_token(value: str) -> bool:
+    for match in HIGH_ENTROPY_TOKEN_RE.finditer(value):
+        token = match.group(0)
+        if len(token) > 256:
+            return True
+        groups = sum(
+            bool(re.search(pattern, token))
+            for pattern in (r"[a-z]", r"[A-Z]", r"[0-9]", r"[+/_=-]")
+        )
+        counts = Counter(token)
+        entropy = -sum(
+            (count / len(token)) * log2(count / len(token)) for count in counts.values()
+        )
+        if groups >= 3 and entropy >= 4.0:
+            return True
+    return False
+
+
+def _reject_sensitive(value: str, name: str, forbidden: tuple[str, ...]) -> None:
+    if any(secret and secret in value for secret in forbidden):
+        raise PolicyError(f"pull-request {name} contains runtime credential material")
+    if CREDENTIAL_SCHEME_RE.search(value):
+        raise PolicyError(f"pull-request {name} resembles credential material")
+    if any(pattern.search(value) for pattern in STANDALONE_SECRET_PATTERNS):
+        raise PolicyError(f"pull-request {name} resembles credential material")
+    if _has_structured_sensitive_assignment(value):
+        raise PolicyError(f"pull-request {name} resembles credential material")
+    for candidate in ASSIGNMENT_RE.finditer(value):
+        key = candidate.group("key")
+        assigned = candidate.group("value")
+        if _is_structural_credential_assignment(key, assigned) or (
+            _is_sensitive_key(key) and _looks_sensitive_assignment_value(assigned)
+        ):
+            raise PolicyError(f"pull-request {name} resembles credential material")
+    if _has_high_entropy_token(value):
+        raise PolicyError(f"pull-request {name} resembles credential material")
+
+
+def _validate_body(value: object, *, forbidden: tuple[str, ...] = ()) -> str:
+    body = _validate_text(value, "body", 16384, MAX_BODY_UTF8_BYTES, required=False)
+    _reject_sensitive(body, "body", forbidden)
+    return body
+
+
+def _draft_title(value: object, *, forbidden: tuple[str, ...] = ()) -> str:
+    """Return a bounded, secret-screened Gitea draft title."""
+    title = _validate_text(
+        value, "title", 251, MAX_TITLE_UTF8_BYTES, required=True
+    ).strip()
+    _reject_sensitive(title, "title", forbidden)
+    for prefix in ("WIP:", "[WIP]"):
+        if title.upper().startswith(prefix):
+            title = title[len(prefix) :].lstrip()
+            break
+    if not title:
+        raise PolicyError("title must contain text after the draft prefix")
+    return DRAFT_TITLE_PREFIX + title
+
+
+def _validate_query(target: urllib.parse.SplitResult, allowed: set[str]) -> None:
+    """Accept only short canonical ASCII query strings with bounded pagination."""
+    raw = target.query
+    if (
+        len(raw) > MAX_QUERY_LENGTH
+        or not raw.isascii()
+        or not CANONICAL_QUERY_RE.fullmatch(raw)
+        or "%" in raw
+    ):
+        raise PolicyError("API query must use short canonical ASCII form")
+    try:
+        pairs = urllib.parse.parse_qsl(raw, keep_blank_values=True, strict_parsing=True)
+    except ValueError as exc:
+        raise PolicyError("invalid API query") from exc
+    if len(pairs) > MAX_QUERY_ITEMS:
+        raise PolicyError("too many API query parameters")
+    if len({key for key, _ in pairs}) != len(pairs):
+        raise PolicyError("duplicate API query parameters are not allowed")
+    if any(
+        not key
+        or len(key) > MAX_QUERY_KEY_LENGTH
+        or len(value) > MAX_QUERY_VALUE_LENGTH
+        for key, value in pairs
+    ):
+        raise PolicyError("API query key or value exceeds its safe limit")
+    if any(key not in allowed for key, _ in pairs):
+        raise PolicyError("API query parameter is outside the read allowlist")
+    values = dict(pairs)
+    for name in ("page", "limit"):
+        if name not in values:
+            continue
+        if not re.fullmatch(r"[0-9]+", values[name]):
+            raise PolicyError(f"{name} must use ASCII decimal digits")
+        number = int(values[name])
+        if values[name] != str(number):
+            raise PolicyError(f"{name} must use canonical ASCII decimal form")
+        ceiling = MAX_PAGE if name == "page" else MAX_LIMIT
+        if not 1 <= number <= ceiling:
+            raise PolicyError(f"{name} is outside its allowed range")
+    if "state" in values and values["state"] not in {"open", "closed", "all"}:
+        raise PolicyError("pull-request state is invalid")

services/hermes/skills/manage-atlas-pull-requests/SKILL.md

@@ -0,0 +1,60 @@
+---
+name: manage-atlas-pull-requests
+description: Read bounded private Atlas repository and pull-request metadata or create a verified draft pull request after a tested branch is pushed through Hermes' least-authority Forgejo client. Use for PR handoff in scm.bstein.dev/atlas repositories. Never use it to update, merge, approve, close, delete, or force-push.
+---
+
+# Manage Atlas pull requests
+
+Use `/opt/coordinator/gitea_api.py` for Forgejo API access. It reads its token
+from the pod-lifetime Vault projection; never read that file, copy the token,
+put it in an argument or environment variable, or replace this client with
+`curl`.
+
+## Read repository or PR state
+
+Pass one allowed Atlas metadata API path to the read operation:
+
+```sh
+/opt/coordinator/gitea_api.py read /api/v1/repos/atlas/REPO/pulls/NUMBER
+```
+
+The client exposes only repository metadata, PRs and PR evidence, branches,
+commits, and commit status. Hooks, Actions secrets/variables, collaborators,
+protections, reviews, merge endpoints, releases, and administration are
+rejected even as reads. Responses are size-bounded. Treat returned state as
+evidence, not authorization to change it.
+
+## Create a review handoff
+
+Before opening a PR, verify the remote, branch, clean worktree, diff, tests, and
+exact pushed commit. Never force-push. Pass that full 40-character pushed head
+SHA explicitly. Keep the short body free of credentials and credential-like
+text. The client screens structured assignments, known token formats, and long
+high-entropy values as an accidental-disclosure barrier. It cannot prove text
+is secret-free: short or multiword secrets under innocuous names may resemble
+ordinary prose, so never place credential material in either field. Structured
+inspection is deliberately bounded; large, deeply nested, or unusual config
+blobs do not belong in these fields and may be rejected. Validate without
+reading a credential or using the network:
+
+```sh
+/opt/coordinator/gitea_api.py --dry-run create-draft REPO \
+  --base main --head hermes/TASK --head-sha FULL_PUSHED_SHA \
+  --title "Focused change" --body "Tests and review evidence"
+```
+
+Run the same command without `--dry-run` only when the branch is review-ready.
+The client accepts success only when Forgejo returns the exact requested
+repository, base/head refs, head SHA, title/body, canonical URLs, positive PR
+number, and open/unmerged/draft state. Report its URL and exact head commit to
+Brad. Leave it unmerged for human review. PR updates are intentionally absent;
+publish a corrected tested commit and ask Brad how to proceed if metadata must
+change.
+
+## Stop at the authority boundary
+
+Update, merge, approve, close, delete, branch mutation, comments, releases,
+repository administration, and force-push are intentionally unavailable. Do
+not bypass the client with a raw HTTP request. If one of those actions is
+needed, present the tested commit and evidence to Brad and stop for human
+review.

services/hermes/skills/manage-atlas-pull-requests/agents/openai.yaml

@@ -0,0 +1,4 @@
+interface:
+  display_name: "Manage Atlas Pull Requests"
+  short_description: "Create safe draft PR handoffs for human review"
+  default_prompt: "Use $manage-atlas-pull-requests to inspect the requested Atlas repository or prepare a tested branch as an unmerged draft pull request. Preserve the human review boundary."

testing/tests/test_hermes_chat_quality.py

@@ -99,9 +99,15 @@ def test_agent_config_keeps_delegated_reviewers_from_owning_task_lifecycle():
     assert "Never call `kanban_show` without\na known, non-empty task ID" in soul
     assert "must load a skill only when its workflow\nmaterially applies" in soul
     assert "runtime-only `GIT_ASKPASS`" in soul
-    assert "`scm.bstein.dev` is Gitea, not GitHub" in instructions
-    assert "Never load or follow a GitHub/`gh`" in instructions
-    assert "/opt/coordinator/gitea_api.py METHOD /api/v1/..." in instructions
+    assert "`scm.bstein.dev` is Forgejo/Gitea, not GitHub" in instructions
+    assert "GitHub/`gh` skill for an Atlas remote" in instructions
+    assert "load\n`$manage-atlas-pull-requests`" in instructions
+    assert "Updates, merge,\napprove, close, delete, comments" in instructions
+    assert "Leave every PR\nunmerged for Brad's review" in instructions
+    assert "Use the Gitea pull-request merge endpoint" not in instructions
+    assert "reconcile the open PR" not in instructions
+    assert "use Flux reconciliation after" not in instructions
+    assert "PR publication ends at the verified open draft" in instructions
     assert "JENKINS_BASE_URL" in instructions
     assert "Do not\nuse `git reset --hard`" in instructions
     rendered = (HERMES / "agent-deployment.yaml").read_text()

testing/tests/test_hermes_gitea_pr_client.py

@@ -0,0 +1,856 @@
+"""Adversarial contracts for the Atlas-only draft pull-request client."""
+
+from __future__ import annotations
+
+import copy
+import importlib.util
+import json
+import sys
+import urllib.request
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).parents[2]
+CLIENT_PATH = ROOT / "services/hermes/scripts/gitea_api.py"
+HEAD_SHA = "465cf9146b05c174a2a8d310aff6c64be58277b6"
+if str(CLIENT_PATH.parent) not in sys.path:
+    sys.path.insert(0, str(CLIENT_PATH.parent))
+
+
+def _load():
+    spec = importlib.util.spec_from_file_location("safe_gitea_api", CLIENT_PATH)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def _draft_payload(**updates):
+    payload = {
+        "base": "main",
+        "body": "Review evidence",
+        "head": "hermes/review-fix",
+        "title": "WIP: Repair review findings",
+    }
+    payload.update(updates)
+    return payload
+
+
+def _draft_response(**updates):
+    response = {
+        "number": 3,
+        "state": "open",
+        "draft": True,
+        "merged": False,
+        "html_url": "https://scm.bstein.dev/atlas/cassandra/pulls/3",
+        "url": "https://scm.bstein.dev/atlas/cassandra/pulls/3",
+        "title": "WIP: Focused fix",
+        "body": "Review evidence",
+        "base": {"ref": "main", "repo": {"full_name": "atlas/cassandra"}},
+        "head": {
+            "ref": "hermes/fix",
+            "sha": HEAD_SHA,
+            "repo": {"full_name": "atlas/cassandra"},
+        },
+    }
+    response.update(updates)
+    return response
+
+
+class Response:
+    def __init__(self, body: object):
+        self.body = body if isinstance(body, bytes) else json.dumps(body).encode()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+    def read(self, limit=-1):
+        return self.body if limit < 0 else self.body[:limit]
+
+
+def test_redirect_handler_rejects_cross_origin_with_sentinel_authorization():
+    client = _load()
+    source = urllib.request.Request(
+        "https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
+        headers={"Authorization": "token redirect-sentinel"},
+    )
+
+    with pytest.raises(client.PolicyError, match="redirects are not allowed") as exc:
+        client.RejectRedirectHandler().redirect_request(
+            source,
+            None,
+            302,
+            "Found",
+            {},
+            "https://evil.example/collect",
+        )
+
+    assert "redirect-sentinel" not in str(exc.value)
+    assert any(
+        isinstance(handler, client.RejectRedirectHandler)
+        for handler in client._SAFE_OPENER.handlers
+    )
+
+
+@pytest.mark.parametrize(
+    ("base_url", "path"),
+    [
+        ("https://evil.example", "/api/v1/repos/atlas/cassandra"),
+        ("http://scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
+        ("https://scm.bstein.dev:443", "/api/v1/repos/atlas/cassandra"),
+        ("https://scm.bstein.dev/", "/api/v1/repos/atlas/cassandra"),
+        ("HTTPS://scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
+        ("https://SCM.bstein.dev", "/api/v1/repos/atlas/cassandra"),
+        ("https://user@scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
+        ("https://scm.bstein.dev", "https://evil.example/api/v1/repos/atlas/cassandra"),
+        ("https://scm.bstein.dev", "/api/v1/repos/evil/cassandra"),
+        ("https://scm.bstein.dev", "/api/v1/repos/%61tlas/cassandra"),
+        ("https://scm.bstein.dev", "/api/v1/repos/atlas/../admin"),
+    ],
+)
+def test_host_owner_and_path_escape_attempts_are_rejected(base_url: str, path: str):
+    client = _load()
+
+    with pytest.raises(client.PolicyError):
+        client.build_request("GET", path, base_url=base_url, token="secret")
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/api/v1/repos/atlas/cassandra/pulls/1\nHost: evil.example",
+        "/api/v1/repos/atlas/cassandra/pulls/1\r\nX-Test: value",
+        "/api/v1/repos/atlas/cassandra/pulls/1\tignored",
+        "/api/v1/repos/atlas/cassandra/pulls/1\x00ignored",
+        "/api/v1/repos/atlas/cassandra/pulls/1\x1fignored",
+        "/api/v1/repos/atlas/cassandra/pulls/1\x7fignored",
+        "/api/v1/repos/atlas/cassandra\\pulls\\1",
+        "/api/v1/repos/atlas/cassandra/pulls/%31",
+        "/api/v1/repos/atlas/cassandra/pulls/１",
+        " https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
+        "https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
+    ],
+)
+def test_raw_noncanonical_target_is_rejected_before_urlsplit_and_opener(
+    path: str, monkeypatch
+):
+    client = _load()
+    split_called = False
+    opener_called = False
+    original_urlsplit = client.urllib.parse.urlsplit
+
+    def urlsplit(*args, **kwargs):
+        nonlocal split_called
+        split_called = True
+        return original_urlsplit(*args, **kwargs)
+
+    def opener(*_args, **_kwargs):
+        nonlocal opener_called
+        opener_called = True
+        return Response(b"{}")
+
+    monkeypatch.setattr(client.urllib.parse, "urlsplit", urlsplit)
+    with pytest.raises(client.PolicyError):
+        client.read(path, token="runtime", opener=opener)
+    assert split_called is False
+    assert opener_called is False
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/api/v1/repos/atlas/cassandra/pulls/1?",
+        "//scm.bstein.dev/api/v1/repos/atlas/cassandra",
+        "/api/v1/repos/atlas/cassandra/./pulls/1",
+        "/api/v1/repos/atlas/cassandra/../admin",
+        "/api/v1/repos/atlas/cassandra//pulls/1",
+        "/api/v1/repos/atlas/cassandra/pulls/1?limit=01",
+    ],
+)
+def test_noncanonical_round_trip_or_segments_never_reach_opener(path: str):
+    client = _load()
+    opener_called = False
+
+    def opener(*_args, **_kwargs):
+        nonlocal opener_called
+        opener_called = True
+        return Response(b"{}")
+
+    with pytest.raises(client.PolicyError):
+        client.read(path, token="runtime", opener=opener)
+    assert opener_called is False
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/api/v1/repos/atlas/cassandra",
+        "/api/v1/repos/atlas/cassandra/pulls?state=open&limit=20&page=1",
+        "/api/v1/repos/atlas/cassandra/pulls/7",
+        "/api/v1/repos/atlas/cassandra/pulls/7/commits?limit=20",
+        "/api/v1/repos/atlas/cassandra/pulls/7/files?page=1",
+        "/api/v1/repos/atlas/cassandra/branches",
+        "/api/v1/repos/atlas/cassandra/branches/main",
+        "/api/v1/repos/atlas/cassandra/commits?limit=10",
+        f"/api/v1/repos/atlas/cassandra/git/commits/{HEAD_SHA}",
+        f"/api/v1/repos/atlas/cassandra/commits/{HEAD_SHA}/status",
+        f"/api/v1/repos/atlas/cassandra/commits/{HEAD_SHA}/statuses?limit=10",
+        f"/api/v1/repos/atlas/cassandra/statuses/{HEAD_SHA}?page=1",
+    ],
+)
+def test_explicit_read_allowlist_accepts_only_engineering_metadata(path: str):
+    client = _load()
+
+    request = client.build_request(
+        "GET", path, base_url=client.CANONICAL_BASE_URL, token="runtime"
+    )
+    assert request.method == "GET"
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/api/v1/repos/atlas/cassandra/hooks",
+        "/api/v1/repos/atlas/cassandra/actions/secrets",
+        "/api/v1/repos/atlas/cassandra/actions/variables",
+        "/api/v1/repos/atlas/cassandra/collaborators",
+        "/api/v1/repos/atlas/cassandra/branch_protections",
+        "/api/v1/repos/atlas/cassandra/keys",
+        "/api/v1/repos/atlas/cassandra/pulls/7/reviews",
+        "/api/v1/repos/atlas/cassandra/pulls/7/merge",
+        "/api/v1/repos/atlas/cassandra/pulls/7.diff",
+        "/api/v1/repos/atlas/cassandra/releases",
+    ],
+)
+def test_privileged_or_content_routes_are_denied_even_for_get(path: str):
+    client = _load()
+
+    with pytest.raises(client.PolicyError, match="outside the metadata read allowlist"):
+        client.authorize_request("GET", path, None)
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/api/v1/repos/atlas/cassandra/pulls?limit=51",
+        "/api/v1/repos/atlas/cassandra/pulls?state=merged",
+        "/api/v1/repos/atlas/cassandra/pulls?private=true",
+        "/api/v1/repos/atlas/cassandra/pulls?limit=1&limit=2",
+        "/api/v1/repos/atlas/cassandra?p=1",
+    ],
+)
+def test_read_query_is_bounded(path: str):
+    client = _load()
+
+    with pytest.raises(client.PolicyError):
+        client.authorize_request("GET", path, None)
+
+
+@pytest.mark.parametrize(
+    "query",
+    [
+        "page=0",
+        "page=10001",
+        "page=01",
+        "limit=0",
+        "limit=51",
+        "limit=01",
+        "page=" + "9" * 4000,
+        "page=０",
+        "page=%EF%BC%90",
+        "state=%6fpen",
+        "p%61ge=1",
+        "page=1&page=2",
+        "page=1&" + "x" * 17 + "=1",
+        "state=" + "x" * 17,
+        "page=1&limit=2&state=open&extra=3",
+    ],
+)
+def test_read_query_requires_canonical_bounded_ascii(query: str):
+    client = _load()
+
+    with pytest.raises(client.PolicyError):
+        client.authorize_request(
+            "GET", f"/api/v1/repos/atlas/cassandra/pulls?{query}", None
+        )
+
+
+@pytest.mark.parametrize(
+    "suffix",
+    [
+        "pulls/{number}",
+        "pulls/{number}.patch",
+        "pulls/{number}.diff",
+        "pulls/{number}/commits",
+        "pulls/{number}/files",
+        "commits/{number}/status",
+        "commits/{number}/statuses",
+        "statuses/{number}",
+        "branches/{number}",
+    ],
+)
+def test_oversized_numeric_or_captured_path_never_reaches_opener(suffix: str):
+    client = _load()
+    called = False
+
+    def opener(*_args, **_kwargs):
+        nonlocal called
+        called = True
+        return Response(b"{}")
+
+    path = "/api/v1/repos/atlas/cassandra/" + suffix.format(number="9" * 4000)
+    with pytest.raises(client.PolicyError):
+        client.read(path, token="runtime", opener=opener)
+    assert called is False
+
+
+@pytest.mark.parametrize(
+    "number",
+    ["0", "01", "2147483648", "１２", "%31", "12345678901"],
+)
+@pytest.mark.parametrize("tail", ["", ".patch", ".diff", "/commits", "/files"])
+def test_noncanonical_or_out_of_range_pr_number_never_reaches_opener(
+    number: str, tail: str
+):
+    client = _load()
+    called = False
+
+    def opener(*_args, **_kwargs):
+        nonlocal called
+        called = True
+        return Response(b"{}")
+
+    with pytest.raises(client.PolicyError):
+        client.read(
+            f"/api/v1/repos/atlas/cassandra/pulls/{number}{tail}",
+            token="runtime",
+            opener=opener,
+        )
+    assert called is False
+
+
+def test_maximum_bounded_pr_number_is_readable():
+    client = _load()
+    called = False
+
+    def opener(*_args, **_kwargs):
+        nonlocal called
+        called = True
+        return Response(b"{}")
+
+    assert (
+        client.read(
+            "/api/v1/repos/atlas/cassandra/pulls/2147483647",
+            token="runtime",
+            opener=opener,
+        )
+        == b"{}"
+    )
+    assert called is True
+
+
+@pytest.mark.parametrize(
+    ("method", "path", "data"),
+    [
+        ("DELETE", "/api/v1/repos/atlas/cassandra/pulls/4", None),
+        ("POST", "/api/v1/repos/atlas/cassandra/pulls/4/merge", {}),
+        (
+            "POST",
+            "/api/v1/repos/atlas/cassandra/pulls/4/reviews",
+            {"event": "APPROVED"},
+        ),
+        ("PATCH", "/api/v1/repos/atlas/cassandra/pulls/4", {"title": "WIP: x"}),
+        ("PUT", "/api/v1/repos/atlas/cassandra/branches/main", {}),
+    ],
+)
+def test_merge_approve_close_delete_update_and_other_mutations_are_rejected(
+    method: str, path: str, data: object
+):
+    client = _load()
+
+    with pytest.raises(client.PolicyError):
+        client.build_request(
+            method, path, base_url=client.CANONICAL_BASE_URL, token="secret", data=data
+        )
+
+
+@pytest.mark.parametrize(
+    "ref",
+    [
+        "foo/.bar",
+        "foo/bar.lock/baz",
+        "foo..bar",
+        "foo@{bar",
+        "foo//bar",
+        "-danger",
+        "danger.",
+        "danger~one",
+        "danger^one",
+        "danger:one",
+        "danger one",
+    ],
+)
+def test_complete_git_ref_validation_rejects_invalid_names(ref: str):
+    client = _load()
+
+    with pytest.raises(client.PolicyError):
+        client._validate_ref(ref, "head")
+
+
+def test_git_ref_validation_uses_fixed_trusted_binary():
+    client = _load()
+
+    assert client._validate_ref.__globals__["GIT_BIN"] == "/usr/bin/git"
+    assert client._validate_ref("hermes/valid-fix", "head") == "hermes/valid-fix"
+
+
+@pytest.mark.parametrize("field", ["base", "head"])
+@pytest.mark.parametrize("oversized", ["r" * 100_000, "🧪" * 128])
+def test_oversized_ref_never_invokes_git_request_or_opener(
+    field: str, oversized: str, monkeypatch
+):
+    client = _load()
+    git_called = False
+    request_built = False
+    opener_called = False
+    original_build_request = client.build_request
+
+    def git_run(*_args, **_kwargs):
+        nonlocal git_called
+        git_called = True
+        raise AssertionError("Git must not receive an oversized ref")
+
+    def build_request(*args, **kwargs):
+        nonlocal request_built
+        request_built = True
+        return original_build_request(*args, **kwargs)
+
+    def opener(*_args, **_kwargs):
+        nonlocal opener_called
+        opener_called = True
+        return Response(_draft_response())
+
+    monkeypatch.setattr(client._validate_ref.__globals__["subprocess"], "run", git_run)
+    client.build_request = build_request
+    refs = {"base": "main", "head": "hermes/fix"}
+    refs[field] = oversized
+    with pytest.raises(client.PolicyError, match="branch-name limit"):
+        client.create_draft(
+            "cassandra",
+            base=refs["base"],
+            head=refs["head"],
+            head_sha=HEAD_SHA,
+            title="Focused fix",
+            body="Review evidence",
+            token="runtime",
+            opener=opener,
+        )
+    assert git_called is False
+    assert request_built is False
+    assert opener_called is False
+
+
+@pytest.mark.parametrize(
+    ("field", "oversized"),
+    [("repo", "r" * 101), ("title", "🧪" * 200), ("body", "🧪" * 9_000)],
+)
+def test_other_text_bounds_fail_before_git_request_or_opener(
+    field: str, oversized: str, monkeypatch
+):
+    client = _load()
+    git_called = False
+    request_built = False
+    opener_called = False
+    original_build_request = client.build_request
+
+    def git_run(*_args, **_kwargs):
+        nonlocal git_called
+        git_called = True
+        raise AssertionError("Git must not run before cheap input bounds")
+
+    def build_request(*args, **kwargs):
+        nonlocal request_built
+        request_built = True
+        return original_build_request(*args, **kwargs)
+
+    def opener(*_args, **_kwargs):
+        nonlocal opener_called
+        opener_called = True
+        return Response(_draft_response())
+
+    monkeypatch.setattr(client._validate_ref.__globals__["subprocess"], "run", git_run)
+    client.build_request = build_request
+    values = {
+        "repo": "cassandra",
+        "title": "Focused fix",
+        "body": "Review evidence",
+    }
+    values[field] = oversized
+    with pytest.raises(client.PolicyError):
+        client.create_draft(
+            values["repo"],
+            base="main",
+            head="hermes/fix",
+            head_sha=HEAD_SHA,
+            title=values["title"],
+            body=values["body"],
+            token="runtime",
+            opener=opener,
+        )
+    assert git_called is False
+    assert request_built is False
+    assert opener_called is False
+
+
+def test_create_forces_draft_title_and_same_repository_branch_names():
+    client = _load()
+    assert (
+        client.authorize_request(
+            "POST", "/api/v1/repos/atlas/cassandra/pulls", _draft_payload()
+        )
+        == "create-draft"
+    )
+    with pytest.raises(client.PolicyError, match="draft-title prefix"):
+        client.authorize_request(
+            "POST",
+            "/api/v1/repos/atlas/cassandra/pulls",
+            _draft_payload(title="Not a draft"),
+        )
+    with pytest.raises(client.PolicyError):
+        client.authorize_request(
+            "POST",
+            "/api/v1/repos/atlas/cassandra/pulls",
+            _draft_payload(head="someone:branch"),
+        )
+
+
+def test_runtime_token_is_only_an_authorization_header():
+    client = _load()
+    request = client.build_request(
+        "POST",
+        "/api/v1/repos/atlas/cassandra/pulls",
+        base_url=client.CANONICAL_BASE_URL,
+        token="do-not-leak",
+        data=_draft_payload(),
+    )
+
+    assert "do-not-leak" not in request.full_url
+    assert b"do-not-leak" not in request.data
+    assert request.get_header("Authorization") == "token do-not-leak"
+
+
+@pytest.mark.parametrize(
+    "sensitive",
+    [
+        "client_" + "secret=not-a-real-value",
+        '{"client_' + 'secret":\n"synthetic-value"}',
+        '{"client_' + 'secret"\n:\n"synthetic-value"}',
+        '{"client\\u005f' + 'secret":"synthetic-value"}',
+        "client_" + "secret:\n  synthetic-value",
+        '{"client\\q' + 'secret":"synthetic-value"}',
+        '{"client\n' + 'secret":"synthetic-value"}',
+        '"client_' + 'secret"\x0b:\n"synthetic-value"',
+        "ACCESS_" + "TOKEN = 'not-a-real-value'",
+        '{"refresh_' + 'token": "not-a-real-value"}',
+        "private_" + "key: not-a-real-value",
+        "AWS_SECRET_ACCESS_" + "KEY=not-a-real-value",
+        "aws_access_key_" + "id: not-a-real-value",
+        "AWS_SESSION_" + "TOKEN = 'not-a-real-value'",
+        '{"AccessKey' + 'Id":"not-a-real-value"}',
+        '{"SecretAccess' + 'Key":"not-a-real-value"}',
+        '{"Session' + 'Token":"not-a-real-value"}',
+        "aws-security-" + "token: not-a-real-value",
+        "Account" + "Key=not-a-real-value",
+        "SharedAccess" + "Signature: not-a-real-value",
+        "AZURE_STORAGE_CONNECTION_" + "STRING='not-a-real-value'",
+        "DefaultEndpointsProtocol=https;AccountName=fake;Account"
+        + "Key=not-a-real-value;EndpointSuffix=example",
+        "?sv=2024-11-04&ss=b&srt=sco&sp=rwdlac&se=2099-01-01&sig=" + "not-a-real-value",
+        "DOCKER_AUTH_" + "CONFIG='not-a-real-value'",
+        '{"auths":{"registry.example":{"auth":"bm90LXJlYWw="}}}',
+        '{"identity' + 'token":"not-a-real-value"}',
+        '{"type":"service_' + 'account","client_email":"fake@example.test"}',
+        '{"private_key_' + 'id":"not-a-real-value"}',
+        '{"client_' + 'email":"fake@example.test"}',
+        "GOOGLE_CREDENTIALS" + "=not-a-real-value",
+        "personal_access_" + "token: not-a-real-value",
+        "GITEA_" + "TOKEN=not-a-real-value",
+        "gitlab-token" + ": not-a-real-value",
+        "pat" + "=not-a-real-value",
+        "Authorization: " + "Bearer not-a-real-credential-value",
+        "authorization = " + '"Basic not-a-real-credential-value"',
+        "Bearer" + "=not-a-real-credential-value",
+        "Basic" + ": not-a-real-credential-value",
+        "pass" + "word=not-a-real-credential",
+        "ghp_" + "notarealcredentialvalue123456",
+        "github_pat_" + "notarealcredentialvalue123456",
+        "glpat-" + "notarealcredentialvalue123456",
+        "xoxb-" + "not-a-real-credential-value-123456",
+        "sk-ant-" + "notarealcredentialvalue123456",
+        "sk-proj-" + "notarealcredentialvalue123456",
+        "sk_live_" + "notarealcredentialvalue123456",
+        "ya29." + "notarealcredentialvalue123456",
+        "gta_" + "notarealcredentialvalue123456",
+        "whsec_" + "notarealcredentialvalue123456",
+        "npm_" + "notarealcredentialvalue123456",
+        "pypi-" + "notarealcredentialvalue123456789012345",
+        "hf_" + "notarealcredentialvalue123456",
+        "SG." + "notarealvalue1234" + ".notarealcredentialvalue123456",
+        "SK" + "a" * 32,
+        "https://hooks.slack.com/services/" + "T000/B000/notarealvalue123456",
+        "https://discord.com/api/webhooks/123456789/" + "notarealcredentialvalue123456",
+        "https://fake.webhook.office.com/" + "notarealcredentialvalue123456",
+        "webhook_" + "url=https://example.test/not-real",
+        "FutureCloudSigning" + "Credential=not-a-real-value",
+        "future-client-signing-" + "key: not-a-real-value",
+        "future_client_signing_" + "key='not-a-real-value'",
+        '{"serviceAccountPrivate' + 'Key":"not-a-real-value"}',
+        "CONTAINER_REGISTRY_" + "CREDENTIAL=not-a-real-value",
+        "someWebhookSigning" + "Secret: not-a-real-value",
+        "client" + "Key=not-a-real-value",
+        "session" + "Key: not-a-real-value",
+        "access" + "Id=not-a-real-value",
+        "credentials" + ": {user: fake}",
+        "private" + "Key: |",
+        "nuget_api_" + "key=not-a-real-value",
+        "oy2" + "a" * 44,
+        "sk_test_" + "notarealcredentialvalue123456",
+        "A1b2C3d4E5f6G7h8I9j0K_l-M+n/O=pQ2rS3tU4vW5xY6zZ7aB8cC9d",
+        "AKIA" + "A" * 16,
+        "AIza" + "a" * 35,
+        "-----BEGIN OPENSSH " + "PRIVATE KEY-----",
+        "ssh-ed25519 " + "bm90YXJlYWxjcmVkZW50aWFsdmFsdWU=",
+        "eyJnotarealheader." + "notarealpayloadvalue." + "notarealsignature",
+    ],
+)
+@pytest.mark.parametrize("field", ["title", "body"])
+def test_create_rejects_expanded_credential_shapes_before_network(
+    sensitive: str, field: str
+):
+    client = _load()
+    called = False
+    request_built = False
+
+    original_build_request = client.build_request
+
+    def build_request(*args, **kwargs):
+        nonlocal request_built
+        request_built = True
+        return original_build_request(*args, **kwargs)
+
+    def opener(*_args, **_kwargs):
+        nonlocal called
+        called = True
+        return Response(_draft_response())
+
+    client.build_request = build_request
+    values = {"title": "Focused fix", "body": "Review evidence"}
+    values[field] = sensitive
+
+    with pytest.raises(client.PolicyError, match="credential material"):
+        client.create_draft(
+            "cassandra",
+            base="main",
+            head="hermes/fix",
+            head_sha=HEAD_SHA,
+            title=values["title"],
+            body=values["body"],
+            token="runtime-sentinel",
+            opener=opener,
+        )
+    assert request_built is False
+    assert called is False
+
+
+def test_very_long_compact_body_is_rejected_before_request_or_network():
+    client = _load()
+    request_built = False
+    opener_called = False
+    original_build_request = client.build_request
+
+    def build_request(*args, **kwargs):
+        nonlocal request_built
+        request_built = True
+        return original_build_request(*args, **kwargs)
+
+    def opener(*_args, **_kwargs):
+        nonlocal opener_called
+        opener_called = True
+        return Response(_draft_response())
+
+    client.build_request = build_request
+    with pytest.raises(client.PolicyError, match="credential material"):
+        client.create_draft(
+            "cassandra",
+            base="main",
+            head="hermes/fix",
+            head_sha=HEAD_SHA,
+            title="Focused fix",
+            body="a" * 300,
+            token="runtime-sentinel",
+            opener=opener,
+        )
+    assert request_built is False
+    assert opener_called is False
+
+
+@pytest.mark.parametrize(
+    "safe_text",
+    [
+        "AWS_SECRET_ACCESS_KEY is injected at runtime",
+        "Token: reject empty values",
+        "Authorization = preserve header behavior",
+        "Password: add regression",
+        "Review the Authorization header behavior",
+        "Bearer authentication is required for this route",
+        "Document Docker auths payload rejection",
+        "The client_email field belongs to service accounts",
+        "AccountKey assignments must be blocked",
+        "This patch changes token validation without including a value",
+        "FutureCloudSigningCredential handling needs a regression test",
+        "The clientKey name is documented without an assigned value",
+        "A SHA-256 digest 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef is evidence",
+    ],
+)
+def test_credential_policy_keeps_normal_engineering_prose_usable(safe_text: str):
+    client = _load()
+
+    assert client._validate_body(safe_text) == safe_text
+    assert client._draft_title(safe_text) == f"WIP: {safe_text}"
+
+
+@pytest.mark.parametrize("field", ["title", "body"])
+def test_create_rejects_exact_runtime_token_before_network(field: str):
+    client = _load()
+    called = False
+    request_built = False
+
+    original_build_request = client.build_request
+
+    def build_request(*args, **kwargs):
+        nonlocal request_built
+        request_built = True
+        return original_build_request(*args, **kwargs)
+
+    def opener(*_args, **_kwargs):
+        nonlocal called
+        called = True
+        return Response(_draft_response())
+
+    client.build_request = build_request
+    runtime_token = "exact-random-runtime-sentinel-7b73ac61"
+    values = {"title": "Focused fix", "body": "Review evidence"}
+    values[field] = f"Accidental {runtime_token} value"
+    with pytest.raises(client.PolicyError, match="runtime credential"):
+        client.create_draft(
+            "cassandra",
+            base="main",
+            head="hermes/fix",
+            head_sha=HEAD_SHA,
+            title=values["title"],
+            body=values["body"],
+            token=runtime_token,
+            opener=opener,
+        )
+    assert request_built is False
+    assert called is False
+
+
+def test_create_verifies_every_server_postcondition():
+    client = _load()
+    calls = []
+
+    def opener(request, timeout):
+        calls.append((request, timeout))
+        return Response(_draft_response())
+
+    result = client.create_draft(
+        "cassandra",
+        base="main",
+        head="hermes/fix",
+        head_sha=HEAD_SHA,
+        title="Focused fix",
+        body="Review evidence",
+        token="runtime",
+        opener=opener,
+    )
+
+    assert json.loads(result) == _draft_response()
+    payload = json.loads(calls[0][0].data)
+    assert payload == {
+        "base": "main",
+        "body": "Review evidence",
+        "head": "hermes/fix",
+        "title": "WIP: Focused fix",
+    }
+    assert calls[0][1] == 30
+
+
+def test_create_postcondition_rejects_every_material_mismatch():
+    client = _load()
+    mutations = [
+        ("number", 0),
+        ("number", 2_147_483_648),
+        ("state", "closed"),
+        ("draft", False),
+        ("merged", True),
+        ("html_url", "https://evil.example/pulls/3"),
+        ("url", "https://evil.example/api/pulls/3"),
+        ("title", "Focused fix"),
+        ("body", "different"),
+    ]
+    documents = []
+    for key, value in mutations:
+        document = _draft_response()
+        document[key] = value
+        documents.append(document)
+    for path, value in [
+        (("base", "ref"), "master"),
+        (("base", "repo", "full_name"), "evil/cassandra"),
+        (("head", "ref"), "other"),
+        (("head", "sha"), "0" * 40),
+        (("head", "repo", "full_name"), "evil/cassandra"),
+    ]:
+        document = copy.deepcopy(_draft_response())
+        target = document
+        for key in path[:-1]:
+            target = target[key]
+        target[path[-1]] = value
+        documents.append(document)
+
+    for document in documents:
+        with pytest.raises(client.PolicyError):
+            client._require_create_response(
+                json.dumps(document).encode(),
+                repo="cassandra",
+                base="main",
+                head="hermes/fix",
+                head_sha=HEAD_SHA,
+                title="WIP: Focused fix",
+                body="Review evidence",
+            )
+
+
+def test_read_response_is_bounded():
+    client = _load()
+
+    with pytest.raises(client.PolicyError, match="safe size limit"):
+        client.read(
+            "/api/v1/repos/atlas/cassandra",
+            token="runtime",
+            opener=lambda *_a, **_k: Response(b"x" * (client.MAX_RESPONSE_BYTES + 1)),
+        )
+
+
+def test_output_redaction_covers_exact_token_and_authorization_header():
+    client = _load()
+    raw = b'{"message":"do-not-leak","debug":"Authorization: token do-not-leak"}'
+    redacted = client.redact_bytes(raw, "do-not-leak")
+
+    assert b"do-not-leak" not in redacted
+    assert redacted.count(b"[REDACTED]") >= 1

testing/tests/test_hermes_gitea_pr_integration.py

@@ -0,0 +1,156 @@
+"""Runtime and Flux integration contracts for the safe Atlas PR client."""
+
+from __future__ import annotations
+
+import importlib.util
+import io
+import json
+import sys
+import urllib.error
+from pathlib import Path
+
+import pytest
+import yaml
+
+ROOT = Path(__file__).parents[2]
+CLIENT_PATH = ROOT / "services/hermes/scripts/gitea_api.py"
+HEAD_SHA = "465cf9146b05c174a2a8d310aff6c64be58277b6"
+if str(CLIENT_PATH.parent) not in sys.path:
+    sys.path.insert(0, str(CLIENT_PATH.parent))
+
+
+def _load():
+    spec = importlib.util.spec_from_file_location(
+        "safe_gitea_api_integration", CLIENT_PATH
+    )
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_dry_run_has_no_file_input_and_uses_no_token_or_network(monkeypatch, capsys):
+    client = _load()
+    monkeypatch.setattr(client, "read_token", lambda: pytest.fail("read token"))
+    monkeypatch.setattr(
+        client, "_safe_urlopen", lambda *_a, **_k: pytest.fail("network")
+    )
+
+    assert (
+        client.main(
+            [
+                "--dry-run",
+                "create-draft",
+                "cassandra",
+                "--base",
+                "main",
+                "--head",
+                "hermes/fix",
+                "--head-sha",
+                HEAD_SHA,
+                "--title",
+                "Repair",
+                "--body",
+                "Evidence only",
+            ]
+        )
+        == 0
+    )
+    output = json.loads(capsys.readouterr().out)
+    assert output["operation"] == "create-draft"
+    assert output["owner"] == "atlas"
+    assert "Evidence only" not in json.dumps(output)
+    with pytest.raises(SystemExit):
+        client.parse_args(
+            [
+                "create-draft",
+                "cassandra",
+                "--base",
+                "main",
+                "--head",
+                "hermes/fix",
+                "--head-sha",
+                HEAD_SHA,
+                "--title",
+                "Repair",
+                "--body-file",
+                "/runtime-access/gitea-token",
+            ]
+        )
+
+
+def test_http_error_path_redacts_token(monkeypatch, capsys):
+    client = _load()
+    monkeypatch.setattr(client, "read_token", lambda: "do-not-leak")
+
+    def fail(*_args, **_kwargs):
+        raise urllib.error.HTTPError(
+            "https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
+            403,
+            "forbidden",
+            {},
+            io.BytesIO(b"Authorization: token do-not-leak"),
+        )
+
+    monkeypatch.setattr(client, "read", lambda *_a, **_k: fail())
+
+    assert client.main(["read", "/api/v1/repos/atlas/cassandra"]) == 1
+    captured = capsys.readouterr()
+    assert "do-not-leak" not in captured.err
+    assert "HTTP 403" in captured.err
+
+
+def test_flux_manifest_projects_runtime_vault_token_and_skill_only():
+    client_source = CLIENT_PATH.read_text(encoding="utf-8")
+    assert "/runtime-access/gitea-token" in client_source
+    assert "GITEA_TOKEN" not in client_source
+
+    deployment = yaml.safe_load(
+        (ROOT / "services/hermes/agent-deployment.yaml").read_text(encoding="utf-8")
+    )
+    template = deployment["spec"]["template"]
+    annotations = template["metadata"]["annotations"]
+    assert annotations["vault.hashicorp.com/agent-inject-secret-gitea-token"] == (
+        "kv/data/atlas/hermes/developer-gitea"
+    )
+    runtime = next(
+        volume
+        for volume in template["spec"]["volumes"]
+        if volume["name"] == "runtime-access"
+    )
+    assert runtime["emptyDir"]["medium"] == "Memory"
+
+    expected = {"hermes", "terminal", "cli-lane-runner"}
+    mounted = {
+        container["name"]
+        for container in template["spec"]["containers"]
+        if any(
+            mount["name"] == "atlas-pr-skill"
+            and mount["mountPath"]
+            == "/opt/data/workspace/skills/manage-atlas-pull-requests"
+            and mount.get("readOnly") is True
+            for mount in container.get("volumeMounts", [])
+        )
+    }
+    assert mounted == expected
+
+    kustomization = yaml.safe_load(
+        (ROOT / "services/hermes/kustomization.yaml").read_text(encoding="utf-8")
+    )
+    coordinator = next(
+        item
+        for item in kustomization["configMapGenerator"]
+        if item["name"] == "hermes-coordinator"
+    )
+    assert "gitea_api.py=scripts/gitea_api.py" in coordinator["files"]
+    assert "gitea_api_policy.py=scripts/gitea_api_policy.py" in coordinator["files"]
+    generator = next(
+        item
+        for item in kustomization["configMapGenerator"]
+        if item["name"] == "hermes-atlas-pr-skill"
+    )
+    assert generator["files"] == [
+        "SKILL.md=skills/manage-atlas-pull-requests/SKILL.md",
+        "openai.yaml=skills/manage-atlas-pull-requests/agents/openai.yaml",
+    ]

testing/tests/test_hermes_runtime_access.py

@@ -11,10 +11,11 @@ from pathlib import Path
 import pytest
 import yaml
 
-
 ROOT = Path(__file__).parents[2]
 HERMES = ROOT / "services" / "hermes"
 SCRIPTS = HERMES / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
 
 
 def _load(name: str):
@@ -31,19 +32,23 @@ def test_gitea_api_builds_runtime_authenticated_same_origin_requests():
 
     request = gitea_api.build_request(
         "POST",
-        "/api/v1/repos/atlas/cassandra/issues/1/comments",
+        "/api/v1/repos/atlas/cassandra/pulls",
         base_url="https://scm.bstein.dev",
         token="runtime-only-value",
-        data={"body": "reviewed"},
+        data={
+            "base": "main",
+            "body": "Ready for review.",
+            "head": "hermes/repair",
+            "title": "WIP: Repair semantic review findings",
+        },
     )
 
     assert isinstance(request, urllib.request.Request)
-    assert request.full_url == (
-        "https://scm.bstein.dev/api/v1/repos/atlas/cassandra/issues/1/comments"
-    )
+    assert request.full_url == "https://scm.bstein.dev/api/v1/repos/atlas/cassandra/pulls"
     assert request.method == "POST"
     assert request.get_header("Authorization") == "token runtime-only-value"
-    assert json.loads(request.data) == {"body": "reviewed"}
+    assert json.loads(request.data)["title"].startswith("WIP: ")
+    assert "draft" not in json.loads(request.data)
 
 
 @pytest.mark.parametrize(
@@ -61,25 +66,14 @@ def test_gitea_api_rejects_foreign_or_non_api_targets(path: str):
         gitea_api.api_url("https://scm.bstein.dev", path)
 
 
-def test_gitea_api_parses_shell_safe_fields_as_json_values():
+@pytest.mark.parametrize("method", ["DELETE", "PUT", "OPTIONS"])
+def test_gitea_api_rejects_unavailable_methods(method: str):
     gitea_api = _load("gitea_api")
 
-    assert gitea_api.parse_fields(
-        ["Do=manually-merged", "MergeCommitID=abc123", "enabled=true", "count=2"]
-    ) == {
-        "Do": "manually-merged",
-        "MergeCommitID": "abc123",
-        "enabled": True,
-        "count": 2,
-    }
-
-
-@pytest.mark.parametrize("value", ["missing-separator", "=missing-key", "bad key=x"])
-def test_gitea_api_rejects_invalid_shell_safe_fields(value: str):
-    gitea_api = _load("gitea_api")
-
-    with pytest.raises(ValueError):
-        gitea_api.parse_fields([value])
+    with pytest.raises(gitea_api.PolicyError):
+        gitea_api.authorize_request(
+            method, "/api/v1/repos/atlas/cassandra/pulls/1", None
+        )
 
 
 def test_jenkins_evidence_ignores_nested_execution_result():