.github/workflows/agent-team-apply-verify.yml

@@ -108,6 +108,16 @@ on:
           PR head and the verified diff are bound by expected_diff_hash.
         required: true
         type: string
+      pr_title:
+        description: >-
+          Optional human-readable PR title (the approved plan's title, sanitized
+          box-side). UNTRUSTED model-derived text — consumed ONLY as printf/argv
+          DATA, never shell-interpolated, and re-validated below (single line, no
+          control chars, length-capped). Empty/unsafe -> the hardened
+          `agent-apply: <task_id>` title is used instead.
+        required: false
+        default: ""
+        type: string
 
 # Workflow-level default: least privilege. Every job re-declares its own
 # `permissions:` so the grant is explicit per job and the untrusted job can be
@@ -1237,6 +1247,9 @@ jobs:
           # The branch the trusted dispatcher already pushed with the diff
           # applied; the PR opens with this as --head (the box pushed nothing).
           HEAD_BRANCH: ${{ inputs.head_branch }}
+          # Optional human-readable title (approved plan title, sanitized box-
+          # side). UNTRUSTED — env-indirected and re-validated below before use.
+          PR_TITLE: ${{ inputs.pr_title }}
           # The repo the PR is opened in (the App installation's repo).
           GH_REPO: ${{ github.repository }}
         run: |
@@ -1263,7 +1276,32 @@ jobs:
             *[!A-Za-z0-9._/-]* | "" | /* | */ | *..* | *//* )
               echo "::error::unsafe head_branch"; exit 1 ;;
           esac
-          title="$(printf 'agent-apply: %s (diff %s)' "$TASK_ID" "$DIFF_HASH")"
+          # Hardened, always-safe fallback title (only sanitized tokens).
+          fallback_title="$(printf 'agent-apply: %s (diff %s)' "$TASK_ID" "$DIFF_HASH")"
+          # Re-validate the box-supplied PR_TITLE as DEFENSE-IN-DEPTH (the
+          # dispatcher already sanitized it). Accept it ONLY if it is a single
+          # line, free of control characters, and within the 70-char convention;
+          # otherwise fall back to the hardened title. This keeps the human-
+          # readable plan title in the PR while never letting model-derived text
+          # smuggle newlines/control chars/over-long content into the PR metadata.
+          title="$fallback_title"
+          if [ -n "$PR_TITLE" ]; then
+            # Re-validate the box-supplied title as defense-in-depth (§4.6),
+            # BYTE-FOR-BYTE identical to agent_team.dispatcher.sanitize_pr_title's
+            # acceptance test: reject any C0/C1 control char or DEL
+            # ([\x00-\x1f\x7f-\x9f] — note this DOES cover C1, which a POSIX
+            # [[:cntrl:]] grep misses) and cap at 70 *characters* (not bytes, so a
+            # multibyte title is not spuriously rejected). python3 is present on
+            # the runner; an undecodable/invalid title exits non-zero -> fallback.
+            if printf '%s' "$PR_TITLE" | python3 -c 'import sys, re; t = sys.stdin.read(); sys.exit(0 if t and not re.search(r"[\x00-\x1f\x7f-\x9f]", t) and len(t) <= 70 else 1)'; then
+              title="$PR_TITLE"
+            else
+              echo "::warning::pr_title rejected (control chars or >70 chars); using fallback"
+            fi
+          fi
+          # The provenance (task id, diff hash, head) lives in the BODY so the
+          # title can be the conventional plan title. Both title and body are
+          # consumed by gh as argv DATA — never shell-interpolated (CWE-94).
           body="$(printf 'Automated draft PR from the agent-team apply/verify pipeline.\n\nTask: %s\nDiff hash: %s\nHead: %s\n\nDRAFT ONLY — never auto-merged. Requires: green required checks, security review, Claude Code App review, and human approval (D2, boundary 5).' "$TASK_ID" "$DIFF_HASH" "$HEAD_BRANCH")"
           gh pr create \
             --draft \

agent-team/agent_team/dispatcher.py

@@ -44,6 +44,7 @@ __all__ = [
     "build_dispatch_inputs",
     "dispatch_apply_verify",
     "head_branch_for",
+    "sanitize_pr_title",
     "select_run_id",
 ]
 
@@ -79,6 +80,42 @@ MAX_DIFF_BYTES = 40_000
 # embedded in the head ref and the PR text). A coordinator thread_id is a uuid,
 # well within this set; we validate to fail closed on anything else.
 _TASK_ID_RE = re.compile(r"\A[A-Za-z0-9._-]{1,200}\Z")
+
+# Max PR-title length (chars). Sea Haven's pull-requests.md caps titles at 70; we
+# leave a little headroom for the model-derived plan title to be readable.
+MAX_PR_TITLE_LEN = 70
+# Any C0/C1 control char (incl. newline/tab) — stripped from the title so a
+# model-derived plan title can never smuggle control chars / extra lines into the
+# PR metadata (defense-in-depth; the workflow ALSO re-validates, §4.6).
+_CONTROL_CHARS_RE = re.compile(r"[\x00-\x1f\x7f-\x9f]")
+_WHITESPACE_RUN_RE = re.compile(r"\s+")
+
+
+def sanitize_pr_title(raw: Any) -> str:
+    """Reduce a model-derived plan title to a SAFE, single-line PR title.
+
+    The agent-team's PR title is the approved plan's title — UNTRUSTED model
+    output. This collapses it to a single clean line the apply workflow can use
+    verbatim: strip control chars, collapse internal whitespace, trim, cap to
+    :data:`MAX_PR_TITLE_LEN`, and capitalize the first letter (Sea Haven commit/PR
+    convention). Returns ``""`` when the input is not a usable string — the
+    workflow then falls back to the hardened ``agent-apply: <task_id>`` title, so
+    a missing/garbage title never blocks the PR. Pure; no I/O.
+    """
+    if not isinstance(raw, str):
+        return ""
+    text = _CONTROL_CHARS_RE.sub(" ", raw)
+    text = _WHITESPACE_RUN_RE.sub(" ", text).strip()
+    if not text:
+        return ""
+    if len(text) > MAX_PR_TITLE_LEN:
+        # Trim to the cap on a word boundary where possible, else hard-cut.
+        clipped = text[:MAX_PR_TITLE_LEN].rstrip()
+        cut = clipped.rsplit(" ", 1)[0] if " " in clipped else clipped
+        text = (cut or clipped).rstrip()
+    return text[0].upper() + text[1:] if text else ""
+
+
 # owner/repo path-segment charset (mirrors ci_fetcher / github_intake).
 _OWNER_REPO_RE = re.compile(r"\A[A-Za-z0-9_.-]{1,100}\Z")
 
@@ -115,6 +152,7 @@ class DispatchInputs:
     declared_scope: str
     diff_b64: str
     head_branch: str
+    pr_title: str = ""
 
     def as_inputs(self) -> dict[str, str]:
         return {
@@ -124,6 +162,7 @@ class DispatchInputs:
             "declared_scope": self.declared_scope,
             "diff_b64": self.diff_b64,
             "head_branch": self.head_branch,
+            "pr_title": self.pr_title,
         }
 
 
@@ -132,6 +171,7 @@ def build_dispatch_inputs(
     task_id: str,
     diff_text: str,
     declared_scope: str,
+    pr_title: str = "",
     artifact_prefix: str = "agent-team-diff",
 ) -> DispatchInputs:
     """Assemble the dispatch inputs from a task id + the candidate diff (PURE).
@@ -165,6 +205,7 @@ def build_dispatch_inputs(
         declared_scope=declared_scope,
         diff_b64=base64.b64encode(raw).decode("ascii"),
         head_branch=head_branch,
+        pr_title=sanitize_pr_title(pr_title),
     )
 
 
@@ -234,6 +275,7 @@ def dispatch_apply_verify(
     task_id: str,
     diff_text: str,
     declared_scope: str,
+    pr_title: str = "",
     base: str = "main",
     pusher: BranchPusher | None = None,
     dispatcher: WorkflowDispatcher | None = None,
@@ -260,7 +302,10 @@ def dispatch_apply_verify(
         raise DispatcherError(f"invalid owner/repo {owner!r}/{repo!r}")
 
     inputs = build_dispatch_inputs(
-        task_id=task_id, diff_text=diff_text, declared_scope=declared_scope
+        task_id=task_id,
+        diff_text=diff_text,
+        declared_scope=declared_scope,
+        pr_title=pr_title,
     )
     push = pusher if pusher is not None else _default_branch_pusher()
     fire = dispatcher if dispatcher is not None else _default_workflow_dispatcher()

agent-team/agent_team/nodes/dispatch_invoker.py

@@ -74,6 +74,10 @@ def make_dispatch_node(
             plan.get("scope") or [] if isinstance(plan, dict) else []
         )
         declared_scope: str = "\n".join(str(s) for s in scope_list if s)
+        # The approved plan's title becomes a conventional, human-readable PR
+        # title (sanitized in the dispatcher; the workflow falls back to the
+        # hardened "agent-apply: <task_id>" form if it is empty/unsafe).
+        pr_title: str = plan.get("title", "") if isinstance(plan, dict) else ""
 
         _parked: dict[str, Any] = {
             "status": TaskStatus.PARKED.value,
@@ -106,6 +110,7 @@ def make_dispatch_node(
                 task_id=thread_id,
                 diff_text=diff_text,
                 declared_scope=declared_scope,
+                pr_title=pr_title,
                 base=base,
                 pusher=pusher,
                 dispatcher=dispatcher,

agent-team/tests/test_dispatcher.py

@@ -13,6 +13,7 @@ import pytest
 
 from agent_team.dispatcher import (
     MAX_DIFF_BYTES,
+    MAX_PR_TITLE_LEN,
     DispatcherError,
     DispatchInputs,
     DispatchResult,
@@ -23,6 +24,7 @@ from agent_team.dispatcher import (
     dispatch_apply_verify,
     head_branch_for,
     run_name_for,
+    sanitize_pr_title,
     select_run_id,
 )
 from agent_team.state_store import compute_content_hash
@@ -48,7 +50,7 @@ def test_build_dispatch_inputs_binds_hash_and_b64() -> None:
     assert di.declared_scope == SCOPE
 
 
-def test_as_inputs_keys_match_the_six_workflow_inputs() -> None:
+def test_as_inputs_keys_match_the_workflow_inputs() -> None:
     di = build_dispatch_inputs(task_id=TASK, diff_text=DIFF, declared_scope=SCOPE)
     assert set(di.as_inputs()) == {
         "task_id",
@@ -57,9 +59,51 @@ def test_as_inputs_keys_match_the_six_workflow_inputs() -> None:
         "declared_scope",
         "diff_b64",
         "head_branch",
+        "pr_title",
     }
 
 
+def test_sanitize_pr_title_collapses_whitespace_and_capitalizes() -> None:
+    assert (
+        sanitize_pr_title("  add\tconfluence  writer\nsection  ")
+        == "Add confluence writer section"
+    )
+
+
+def test_sanitize_pr_title_strips_control_chars() -> None:
+    # A newline / control char must never survive into the single-line PR title.
+    out = sanitize_pr_title("Add section\r\n\x00malicious")
+    assert "\n" not in out and "\r" not in out and "\x00" not in out
+    assert out == "Add section malicious"
+
+
+def test_sanitize_pr_title_caps_length_on_word_boundary() -> None:
+    long = "Add a very long descriptive title " + "word " * 40
+    out = sanitize_pr_title(long)
+    assert len(out) <= MAX_PR_TITLE_LEN
+    assert not out.endswith(" ")  # trimmed cleanly
+
+
+@pytest.mark.parametrize("bad", ["", "   ", "\n\t ", None, 123])
+def test_sanitize_pr_title_returns_empty_for_unusable(bad) -> None:
+    assert sanitize_pr_title(bad) == ""
+
+
+def test_build_dispatch_inputs_carries_sanitized_pr_title() -> None:
+    di = build_dispatch_inputs(
+        task_id=TASK,
+        diff_text=DIFF,
+        declared_scope=SCOPE,
+        pr_title="  document the writer\nlane  ",
+    )
+    assert di.pr_title == "Document the writer lane"
+
+
+def test_build_dispatch_inputs_pr_title_defaults_empty() -> None:
+    di = build_dispatch_inputs(task_id=TASK, diff_text=DIFF, declared_scope=SCOPE)
+    assert di.pr_title == ""
+
+
 @pytest.mark.parametrize("bad_diff", ["", "   ", "\n\n"])
 def test_empty_diff_rejected(bad_diff: str) -> None:
     with pytest.raises(DispatcherError):