Use the approved plan title as a conventional PR title

The apply pipeline's draft PRs were titled `agent-apply: <task_id>
(diff <hash>)` with a flat body — neither within Sea Haven PR
conventions. That format was deliberate injection-hardening (only
sanitized tokens, never model free-text; §4.6).

Thread the approved plan's title through dispatch as an optional
`pr_title` input, sanitized box-side (single line, no control chars,
70-char cap, capitalized) and RE-VALIDATED in the workflow as defense-
in-depth, with the hardened `agent-apply: <task_id>` title as the
fallback when empty/unsafe. Provenance (task id, diff hash, head) moves
to the PR body. gh consumes both as argv data, never shell-interpolated.
This commit is contained in:
Adam Moussa 2026-06-25 13:43:30 -04:00
parent 2e2d560d2d
commit a705859ea3
4 changed files with 131 additions and 3 deletions

View file

@ -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,28 @@ 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
# Reject any C0/C1 control char (incl. newline/tab) or > 70 chars.
if printf '%s' "$PR_TITLE" | LC_ALL=C grep -qP '[\x00-\x1f\x7f-\x9f]'; then
echo "::warning::pr_title rejected (control chars); using fallback"
elif [ "$(printf '%s' "$PR_TITLE" | wc -m)" -gt 70 ]; then
echo "::warning::pr_title rejected (too long); using fallback"
else
title="$PR_TITLE"
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 \

View file

@ -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()

View file

@ -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,

View file

@ -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):