diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml
index 87984530..86d334b6 100644
--- a/services/hermes/kustomization.yaml
+++ b/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
diff --git a/services/hermes/scripts/gitea_api.py b/services/hermes/scripts/gitea_api.py
index b4a98eca..884b0fd4 100755
--- a/services/hermes/scripts/gitea_api.py
+++ b/services/hermes/scripts/gitea_api.py
@@ -7,7 +7,6 @@ import argparse
 import json
 import os
 import re
-import subprocess
 import sys
 import urllib.error
 import urllib.parse
@@ -15,28 +14,22 @@ import urllib.request
 from collections.abc import Callable
 from pathlib import Path
 
+from gitea_api_policy import (
+    PolicyError,
+    _draft_title,
+    _validate_body,
+    _validate_query,
+    _validate_ref,
+    _validate_repo,
+    _validate_sha,
+)
+
 CANONICAL_BASE_URL = "https://scm.bstein.dev"
 ALLOWED_OWNER = "atlas"
 DEFAULT_TOKEN_FILE = Path("/runtime-access/gitea-token")
-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")
 CREATE_PATH_RE = re.compile(r"/api/v1/repos/atlas/([^/]+)/pulls\Z")
 MAX_ERROR_BYTES = 8192
 MAX_RESPONSE_BYTES = 2 * 1024 * 1024
-DRAFT_TITLE_PREFIX = "WIP: "
-GIT_BIN = "/usr/bin/git"
-SENSITIVE_BODY_PATTERNS = (
-    re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
-    re.compile(r"(?i)\b(?:password|passwd|secret|token|api[_-]?key)\s*[:=]"),
-    re.compile(r"(?i)\bauthorization\s*:\s*(?:bearer|token|basic)\s+\S+"),
-    re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9]{20,}|sk-[A-Za-z0-9]{20,})\b"),
-    re.compile(r"\b(?:AKIA[0-9A-Z]{16}|AIza[0-9A-Za-z_-]{35})\b"),
-    re.compile(r"\beyJ[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{8,}\b"),
-)
-
-
-class PolicyError(ValueError):
-    """Raised when a requested Forgejo operation exceeds the safe boundary."""
 
 
 class RejectRedirectHandler(urllib.request.HTTPRedirectHandler):
@@ -70,67 +63,6 @@ def configured_base_url() -> str:
     return configured
 
 
-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(value: object, name: str) -> str:
-    """Validate the complete Git ref grammar using Git itself."""
-    if not isinstance(value, str) or not value or 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_text(value: object, name: str, maximum: 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")
-    if len(value) > maximum or "\x00" in value:
-        raise PolicyError(f"{name} exceeds the safe request limit")
-    return value
-
-
-def _validate_body(value: object, *, forbidden: tuple[str, ...] = ()) -> str:
-    body = _validate_text(value, "body", 16384, required=False)
-    if any(secret and secret in body for secret in forbidden):
-        raise PolicyError("pull-request body contains runtime credential material")
-    if any(pattern.search(body) for pattern in SENSITIVE_BODY_PATTERNS):
-        raise PolicyError("pull-request body resembles credential material")
-    return body
-
-
-def _draft_title(value: object) -> str:
-    """Return a bounded title using Gitea's configured default draft prefix."""
-    title = _validate_text(value, "title", 251, required=True).strip()
-    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 _split_api_path(path: str) -> urllib.parse.SplitResult:
     target = urllib.parse.urlsplit(path)
     if target.scheme or target.netloc or target.fragment:
@@ -142,29 +74,6 @@ def _split_api_path(path: str) -> urllib.parse.SplitResult:
     return target
 
 
-def _validate_query(target: urllib.parse.SplitResult, allowed: set[str]) -> None:
-    try:
-        pairs = urllib.parse.parse_qsl(
-            target.query, keep_blank_values=True, strict_parsing=True
-        )
-    except ValueError as exc:
-        raise PolicyError("invalid API query") from exc
-    if len({key for key, _ in pairs}) != len(pairs):
-        raise PolicyError("duplicate API query parameters are not allowed")
-    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 values[name].isdigit() or int(values[name]) < 1:
-            raise PolicyError(f"{name} must be a positive integer")
-    if "limit" in values and int(values["limit"]) > 50:
-        raise PolicyError("read limit cannot exceed 50")
-    if "state" in values and values["state"] not in {"open", "closed", "all"}:
-        raise PolicyError("pull-request state is invalid")
-
-
 def _authorize_read(target: urllib.parse.SplitResult) -> str:
     """Allow only repository, PR, branch, commit, and status metadata reads."""
     prefix = "/api/v1/repos/atlas/"
@@ -255,6 +164,10 @@ def build_request(
 ) -> urllib.request.Request:
     """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")
@@ -372,7 +285,7 @@ def create_draft(
     base = _validate_ref(base, "base")
     head = _validate_ref(head, "head")
     head_sha = _validate_sha(head_sha)
-    title = _draft_title(title)
+    title = _draft_title(title, forbidden=(token,))
     body = _validate_body(body, forbidden=(token,))
     data = {"base": base, "body": body, "head": head, "title": title}
     result = _request(
diff --git a/services/hermes/scripts/gitea_api_policy.py b/services/hermes/scripts/gitea_api_policy.py
new file mode 100644
index 00000000..0ad461dc
--- /dev/null
+++ b/services/hermes/scripts/gitea_api_policy.py
@@ -0,0 +1,160 @@
+"""Pure validation policy for Hermes' least-authority Atlas Forgejo client."""
+
+from __future__ import annotations
+
+import re
+import subprocess
+import urllib.parse
+
+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
+CANONICAL_QUERY_RE = re.compile(r"[A-Za-z0-9_=&-]*\Z")
+SENSITIVE_TEXT_RE = re.compile(
+    r"""(?ix)
+    (?:
+      (?<![A-Za-z0-9])
+      (?:client[_-]?secret|access[_-]?token|refresh[_-]?token|private[_-]?key|
+         api[_-]?key|password|passwd|secret|token|authorization|bearer|basic)
+      [\"']?\s*[:=]\s*[\"']?
+      |
+      \b(?:bearer|basic)\s+[A-Za-z0-9+/_=.:-]{8,}
+      |
+      -----BEGIN\s+[A-Z0-9 ][A-Z0-9 -]{1,62}-----
+      |
+      \bssh-(?:rsa|ed25519|dss|ecdsa-[A-Za-z0-9-]+)\s+[A-Za-z0-9+/]{16,}={0,3}
+      |
+      \b(?:gh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,}|
+          glpat-[A-Za-z0-9_-]{20,}|xox[baprs]-[A-Za-z0-9-]{20,}|
+          sk-ant-[A-Za-z0-9_-]{20,}|sk-(?:proj-|live_)?[A-Za-z0-9_-]{20,}|
+          [sr]k_live_[A-Za-z0-9]{20,}|
+          ya29\.[A-Za-z0-9_-]{20,})\b
+      |
+      \b(?:AKIA|ASIA)[0-9A-Z]{16}\b
+      |
+      \bAIza[0-9A-Za-z_-]{35}\b
+      |
+      \beyJ[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{8,}\b
+    )
+    """
+)
+
+
+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(value: object, name: str) -> str:
+    """Validate the complete Git ref grammar using Git itself."""
+    if not isinstance(value, str) or not value or 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_text(value: object, name: str, maximum: 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")
+    if len(value) > maximum or "\x00" in value:
+        raise PolicyError(f"{name} exceeds the safe request limit")
+    return value
+
+
+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 SENSITIVE_TEXT_RE.search(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, 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, 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")
diff --git a/testing/tests/test_hermes_gitea_pr_client.py b/testing/tests/test_hermes_gitea_pr_client.py
index 447f3067..a2f4dbf8 100644
--- a/testing/tests/test_hermes_gitea_pr_client.py
+++ b/testing/tests/test_hermes_gitea_pr_client.py
@@ -14,6 +14,8 @@ 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():
@@ -179,6 +181,35 @@ def test_read_query_is_bounded(path: str):
         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(
     ("method", "path", "data"),
     [
@@ -230,7 +261,7 @@ def test_complete_git_ref_validation_rejects_invalid_names(ref: str):
 def test_git_ref_validation_uses_fixed_trusted_binary():
     client = _load()
 
-    assert client.GIT_BIN == "/usr/bin/git"
+    assert client._validate_ref.__globals__["GIT_BIN"] == "/usr/bin/git"
     assert client._validate_ref("hermes/valid-fix", "head") == "hermes/valid-fix"
 
 
@@ -272,23 +303,36 @@ def test_runtime_token_is_only_an_authorization_header():
 
 
 @pytest.mark.parametrize(
-    "body",
+    "sensitive",
     [
-        "pass" + "word=not-a-real-credential",
+        "client_" + "secret=not-a-real-value",
+        "ACCESS_" + "TOKEN = 'not-a-real-value'",
+        '{"refresh_' + 'token": "not-a-real-value"}',
+        "private_" + "key: not-a-real-value",
         "Authorization: " + "Bearer not-a-real-credential-value",
-        "token: " + "ghp_" + "notarealcredentialvalue123456",
+        "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",
+        "AKIA" + "A" * 16,
+        "AIza" + "a" * 35,
         "-----BEGIN OPENSSH " + "PRIVATE KEY-----",
+        "ssh-ed25519 " + "bm90YXJlYWxjcmVkZW50aWFsdmFsdWU=",
         "eyJnotarealheader." + "notarealpayloadvalue." + "notarealsignature",
     ],
 )
-def test_body_rejects_common_credential_shapes(body: str):
-    client = _load()
-
-    with pytest.raises(client.PolicyError, match="credential material"):
-        client._validate_body(body)
-
-
-def test_create_rejects_exact_runtime_token_before_network():
+@pytest.mark.parametrize("field", ["title", "body"])
+def test_create_rejects_expanded_credential_shapes_before_network(
+    sensitive: str, field: str
+):
     client = _load()
     called = False
 
@@ -297,15 +341,45 @@ def test_create_rejects_exact_runtime_token_before_network():
         called = True
         return Response(_draft_response())
 
+    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 called is False
+
+
+@pytest.mark.parametrize("field", ["title", "body"])
+def test_create_rejects_exact_runtime_token_before_network(field: str):
+    client = _load()
+    called = False
+
+    def opener(*_args, **_kwargs):
+        nonlocal called
+        called = True
+        return Response(_draft_response())
+
+    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="Focused fix",
-            body="accidental runtime-sentinel value",
-            token="runtime-sentinel",
+            title=values["title"],
+            body=values["body"],
+            token=runtime_token,
             opener=opener,
         )
     assert called is False
diff --git a/testing/tests/test_hermes_gitea_pr_integration.py b/testing/tests/test_hermes_gitea_pr_integration.py
index d3b47eef..4a589f4c 100644
--- a/testing/tests/test_hermes_gitea_pr_integration.py
+++ b/testing/tests/test_hermes_gitea_pr_integration.py
@@ -15,6 +15,8 @@ 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():
@@ -136,6 +138,13 @@ def test_flux_manifest_projects_runtime_vault_token_and_skill_only():
     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"]
diff --git a/testing/tests/test_hermes_runtime_access.py b/testing/tests/test_hermes_runtime_access.py
index 7ad5087a..b27806bf 100644
--- a/testing/tests/test_hermes_runtime_access.py
+++ b/testing/tests/test_hermes_runtime_access.py
@@ -14,6 +14,8 @@ 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):