adam/orchestrator
Archived
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(agent-team): P3-flip Phase 1 — CI trust-boundary hardening (WIP, gated) #34

Merged
amoussa1229 merged 5 commits from feature/agent-team-p3-flip-phase1 into main 2026-06-22 22:51:53 +00:00
Conversation 0 Commits 5 Files changed 6 +1106 -81
6 changed files with 1106 additions and 81 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

105
agent-team/agent_team/ci_gate.py
View file

@ -71,6 +71,7 @@ __all__ = [
"denylist_violations",
"diff_touched_paths",
"evaluate_ci_gate",
"gate_weakening_violations",
"verify_diff_hash",
]
@ -129,6 +130,38 @@ DENYLIST_GLOBS: tuple[str, ...] = (
"**/*policy*.json",
"**/*.pem",
"**/*.key",
# --- P3-flip §4.2: direct code-execution / supply-chain vectors ---
# Submodule pointers / config — a changed submodule pulls in arbitrary
# external code at the pinned commit.
".gitmodules",
"**/.gitmodules",
# Git hook directories / hook-path redirection — code that runs on git ops.
# (`.git/hooks` itself is never in a checkout; `.husky` / `.githooks` are the
# in-tree hook dirs a repo points `core.hooksPath` at.)
".husky/**",
"**/.husky/**",
".githooks/**",
"**/.githooks/**",
# .gitattributes — clean/smudge filters execute arbitrary processes on checkout.
".gitattributes",
"**/.gitattributes",
# npm/registry config — can inject install scripts, a hostile registry, or auth.
".npmrc",
"**/.npmrc",
# Generated / build artifacts — codegen output is not reviewable as source, so
# a diff that writes into it is escalated to a human rather than auto-built.
"**/__generated__/**",
"**/*.generated.*",
"**/dist/**",
"**/build/**",
"**/*.min.js",
# NOTE (deliberate, for the security gate): lockfiles are NOT wholesale denied
# here. Lockfile-postinstall RCE is already contained by the credential-less,
# egress-blocked build-test sandbox, and the Tier-3 dep-CVE fixer legitimately
# rewrites lockfiles to produce its draft PRs — a wholesale lockfile deny would
# make the fixer un-shippable. The direct code-execution config above (hooks,
# filters, .npmrc, submodules) is the actual §4.2 RCE surface. Revisit if the
# fixer's lockfile writes ever need a scoped allow vs a general deny.
)
# Authenticated GitHub run conclusions that count as a recognised failure (the
@ -336,6 +369,63 @@ def _within_scope(path: str, scope: Sequence[str]) -> bool:
return False
# §4.5 gate-weakening: substrings whose appearance on an ADDED diff line means the
# change is suppressing a quality/security gate (a self-passing move). Matched
# case-insensitively against added-line content.
_GATE_WEAKENING_MARKERS: tuple[tuple[str, str], ...] = (
("noqa", "ruff/flake8 lint suppression (noqa)"),
("type: ignore", "type-check suppression (type: ignore)"),
("type:ignore", "type-check suppression (type:ignore)"),
("pragma: no cover", "coverage suppression (pragma: no cover)"),
("pragma: no-cover", "coverage suppression (pragma: no-cover)"),
("nosec", "bandit security suppression (nosec)"),
("nosemgrep", "semgrep suppression (nosemgrep)"),
("--no-verify", "git/pre-commit hook bypass (--no-verify)"),
)
# Test skip/xfail decorators and calls (carry args, so regex not substring).
_TEST_SKIP_RE: re.Pattern[str] = re.compile(
r"@(?:pytest\.mark\.(?:skip|xfail)|unittest\.skip\w*)\b"
r"|\bpytest\.(?:skip|xfail)\s*\("
r"|\.skipTest\s*\("
)
def _added_diff_lines(unified_diff: str) -> list[str]:
"""Return the content of lines ADDED by a unified diff (excluding ``+++`` headers)."""
if not isinstance(unified_diff, str):
raise CiGateError(
f"unified_diff must be str, got {type(unified_diff).__name__}"
)
return [
line[1:]
for line in unified_diff.splitlines()
if line.startswith("+") and not line.startswith("+++")
]
def gate_weakening_violations(unified_diff: str) -> list[str]:
"""Return reasons the diff WEAKENS a quality/security gate (empty if none).
A diff that ADDS a lint/type/coverage/security suppression, a test
skip/xfail, or a hook bypass could make CI pass falsely — so the pure-code
gate flags it even when the authenticated CI conclusion is ``success``
(§4.5). Such diffs are escalated to a human, never auto-built. Only ADDED
lines are inspected (removing a suppression is fine); matching is
case-insensitive on the marker text.
"""
violations: list[str] = []
for content in _added_diff_lines(unified_diff):
low = content.lower()
snippet = content.strip()[:120]
for marker, desc in _GATE_WEAKENING_MARKERS:
if marker in low:
violations.append(f"gate-weakening: added {desc}: {snippet!r}")
if _TEST_SKIP_RE.search(content):
violations.append(f"gate-weakening: added a test skip/xfail: {snippet!r}")
return violations
def verify_diff_hash(
candidate_diff: str,
*,
@ -432,6 +522,21 @@ def evaluate_ci_gate(
ci_conclusion=None,
)
# (1b) Gate-weakening (§4.5): a diff that suppresses a lint/type/coverage/
# security check or skips tests could make CI pass falsely -> BLOCK regardless
# of the authenticated conclusion. A build cannot pass itself by disabling the
# checks; these escalate to a human.
weakening = gate_weakening_violations(candidate_diff)
if weakening:
reasons.extend(weakening)
return GateResult(
decision=GateDecision.BLOCK,
reasons=reasons,
run_id=expected_run_id,
diff_hash=ledger_hash,
ci_conclusion=None,
)
# (2) Diff-hash integrity: bind the decision to the exact bytes the ledger
# recorded (and, if present, what CI verified before applying).
ci_verified_hash = (

319
agent-team/agent_team/dispatcher.py Normal file
View file

@ -0,0 +1,319 @@
"""Trusted apply-path dispatcher (§4.3): carry a box-emitted diff into org CI.
The read-only R720 box (D2) EMITS a candidate diff + the
``workflow_dispatch`` inputs but holds **no write token**. This dispatcher is
the ONLY component that writes, and it runs on a TRUSTED host (the operator's
Mac, where the GitHub App key / operator ``gh`` auth lives) — never on the box.
It does two things and nothing else:
1. Pushes the candidate diff as a short-lived **head branch** (the PR head).
2. Triggers the ``agent-team-apply-verify.yml`` ``workflow_dispatch``, passing
the diff as ``diff_b64`` plus the integrity inputs.
The CI workflow then RE-VERIFIES everything (materialize re-hashes; guard
re-hashes + denylist + scope; build-test builds/tests credential-less; the
pure-code gate decides pass/fail; the privileged job opens a DRAFT PR only on a
clean pass, gated by the ``agent-apply`` environment's required reviewer). This
dispatcher TRANSPORTS only — it makes no trust decision.
Design discipline (mirrors the rest of agent_team): the network/SDK/subprocess
side effects are behind INJECTED seams so the pure input-assembly + validation
logic is fully unit-testable with no git, no ``gh``, and no network. The real
default seams shell out to ``git`` / ``gh`` and are exercised only in
production.
"""
from __future__ import annotations
import base64
import re
from dataclasses import dataclass
from typing import Protocol
from agent_team.state_store import compute_content_hash
__all__ = [
"DispatchInputs",
"DispatcherError",
"build_dispatch_inputs",
"dispatch_apply_verify",
"head_branch_for",
]
WORKFLOW_FILE = "agent-team-apply-verify.yml"
# The artifact carries exactly this filename; the workflow's materialize/guard
# steps look for ``candidate.diff`` (mirrors nodes.fixer.CANDIDATE_DIFF_FILENAME).
CANDIDATE_DIFF_FILENAME = "candidate.diff"
# Max candidate-diff size. NOT arbitrary: the diff is carried as the base64
# `diff_b64` workflow_dispatch INPUT, and GitHub caps total dispatch inputs at
# ~65,535 bytes. base64 inflates ~4/3, so a diff over ~45 KB cannot be
# dispatched at all. We cap at 40 KB (leaving headroom for the other inputs) and
# fail closed with a clear error rather than letting GitHub reject the dispatch
# opaquely — and to bound resource use on the operator host. Larger diffs need a
# branch-only transport (future), not an inline input.
MAX_DIFF_BYTES = 40_000
# task_id must match the workflow's gate-and-pr safety guard charset (it is
# 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")
# owner/repo path-segment charset (mirrors ci_fetcher / github_intake).
_OWNER_REPO_RE = re.compile(r"\A[A-Za-z0-9_.-]{1,100}\Z")
class DispatcherError(Exception):
"""Raised on structurally invalid dispatch inputs (fail closed, never proceed)."""
def head_branch_for(task_id: str) -> str:
"""Return the head-branch ref the dispatcher pushes for ``task_id``.
A stable, namespaced ref so a re-dispatch for the same task reuses/replaces
one branch rather than littering refs. Validated to the safe charset so the
ref can never carry shell/ref metacharacters.
"""
if not _TASK_ID_RE.match(task_id or ""):
raise DispatcherError(
f"invalid task_id {task_id!r}: must match {_TASK_ID_RE.pattern}"
)
return f"agent-team/apply/{task_id}"
@dataclass(frozen=True)
class DispatchInputs:
"""The exact ``workflow_dispatch`` inputs for the apply/verify workflow.
Field names match ``on.workflow_dispatch.inputs`` 1:1 so :meth:`as_inputs`
can be handed straight to ``gh workflow run -f k=v``.
"""
task_id: str
diff_artifact_name: str
expected_diff_hash: str
declared_scope: str
diff_b64: str
head_branch: str
def as_inputs(self) -> dict[str, str]:
return {
"task_id": self.task_id,
"diff_artifact_name": self.diff_artifact_name,
"expected_diff_hash": self.expected_diff_hash,
"declared_scope": self.declared_scope,
"diff_b64": self.diff_b64,
"head_branch": self.head_branch,
}
def build_dispatch_inputs(
*,
task_id: str,
diff_text: str,
declared_scope: str,
artifact_prefix: str = "agent-team-diff",
) -> DispatchInputs:
"""Assemble the dispatch inputs from a task id + the candidate diff (PURE).
Computes the sha256 the workflow binds to (``expected_diff_hash``), base64s
the diff (``diff_b64``, carried as an input so the credential-less
materialize job can re-create the artifact), derives the head branch, and
names the artifact. No I/O, no network — fully testable.
Fails closed (:class:`DispatcherError`) on an empty diff / empty scope /
unsafe task id, so a malformed task can never be transported.
"""
if not isinstance(diff_text, str) or not diff_text.strip():
raise DispatcherError("diff_text must be a non-empty diff")
if not isinstance(declared_scope, str) or not declared_scope.strip():
raise DispatcherError(
"declared_scope must be non-empty (an empty scope allows any path)"
)
head_branch = head_branch_for(task_id) # validates task_id
raw = diff_text.encode("utf-8")
if len(raw) > MAX_DIFF_BYTES:
raise DispatcherError(
f"diff too large ({len(raw)} bytes > {MAX_DIFF_BYTES}); the diff is "
"carried as a base64 workflow_dispatch input (GitHub caps inputs at "
"~64 KB). Split the change or use a branch-only transport."
)
return DispatchInputs(
task_id=task_id,
diff_artifact_name=f"{artifact_prefix}-{task_id}",
expected_diff_hash=compute_content_hash(raw),
declared_scope=declared_scope,
diff_b64=base64.b64encode(raw).decode("ascii"),
head_branch=head_branch,
)
class BranchPusher(Protocol):
"""Pushes the candidate diff as the head branch (the only WRITE)."""
def __call__(
self, *, owner: str, repo: str, base: str, head_branch: str, diff_text: str
) -> None: ...
class WorkflowDispatcher(Protocol):
"""Triggers the apply/verify ``workflow_dispatch`` with the assembled inputs."""
def __call__(
self, *, owner: str, repo: str, inputs: dict[str, str], ref: str
) -> None: ...
def dispatch_apply_verify(
*,
owner: str,
repo: str,
task_id: str,
diff_text: str,
declared_scope: str,
base: str = "main",
pusher: BranchPusher | None = None,
dispatcher: WorkflowDispatcher | None = None,
) -> DispatchInputs:
"""Transport one candidate diff into org CI: push the head branch, dispatch.
The trusted apply path. Validates owner/repo, assembles the dispatch inputs,
pushes the diff as the head branch via ``pusher``, then triggers the
workflow via ``dispatcher``. Returns the :class:`DispatchInputs` used (for
the ledger/audit). ``pusher`` / ``dispatcher`` are injected so this is
testable without git/gh/network; the real defaults shell out to git/gh.
NOTE: this never runs on the box (the box has no write token, D2). CI owns
every trust decision; this only moves bytes.
"""
if not _OWNER_REPO_RE.match(owner or "") or not _OWNER_REPO_RE.match(repo or ""):
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
)
push = pusher if pusher is not None else _default_branch_pusher()
fire = dispatcher if dispatcher is not None else _default_workflow_dispatcher()
# Push the head branch FIRST: the draft-PR step opens against an
# already-pushed --head, so the branch must exist before the run reaches it.
push(
owner=owner,
repo=repo,
base=base,
head_branch=inputs.head_branch,
diff_text=diff_text,
)
fire(owner=owner, repo=repo, inputs=inputs.as_inputs(), ref=base)
return inputs
def _default_branch_pusher() -> BranchPusher:
"""Real pusher: clone-free apply of the diff onto a fresh branch via git/gh.
Deferred to call time (no subprocess at import). Uses the operator's git
credentials / the GitHub App token present on the trusted host — NEVER a
box-held token. Implemented as a thin shell-out; the heavy lifting is the
pure :func:`build_dispatch_inputs` above, so this stays small.
"""
def _push(
*, owner: str, repo: str, base: str, head_branch: str, diff_text: str
) -> None:
import subprocess
import tempfile
from pathlib import Path
# Operator host only. Use a throwaway worktree, apply the diff, push the
# branch with the host's credentials. Kept intentionally minimal; the
# security comes from CI re-verifying the pushed content by hash.
with tempfile.TemporaryDirectory() as tmp:
tmpdir = Path(tmp)
diff_path = tmpdir / CANDIDATE_DIFF_FILENAME
diff_path.write_text(diff_text, encoding="utf-8")
clone = tmpdir / "repo"
subprocess.run(
[
"gh",
"repo",
"clone",
f"{owner}/{repo}",
str(clone),
"--",
"--depth",
"1",
"--branch",
base,
],
check=True,
capture_output=True,
)
subprocess.run(
["git", "-C", str(clone), "checkout", "-B", head_branch],
check=True,
capture_output=True,
)
# --index applies AND stages exactly the diff's changes (incl. new
# files) and NOTHING else — so the committed head tree is precisely
# base+diff, never stray untracked worktree content. This keeps the
# PR head bound to the same bytes CI hash-verified (LOGIC-1). No
# separate `git add -A` (which would stage unrelated content).
subprocess.run(
["git", "-C", str(clone), "apply", "--index", str(diff_path)],
check=True,
capture_output=True,
)
subprocess.run(
[
"git",
"-C",
str(clone),
"-c",
"user.name=agent-team",
"-c",
"user.email=agent-team@seahavenind.com",
"commit",
"-m",
f"agent-team apply: {head_branch}",
],
check=True,
capture_output=True,
)
subprocess.run(
[
"git",
"-C",
str(clone),
"push",
"--force-with-lease",
"origin",
head_branch,
],
check=True,
capture_output=True,
)
return _push
def _default_workflow_dispatcher() -> WorkflowDispatcher:
"""Real dispatcher: ``gh workflow run`` (operator auth has ``actions:write``)."""
def _fire(*, owner: str, repo: str, inputs: dict[str, str], ref: str) -> None:
import subprocess
args = [
"gh",
"workflow",
"run",
WORKFLOW_FILE,
"--repo",
f"{owner}/{repo}",
"--ref",
ref,
]
for key, value in inputs.items():
args += ["-f", f"{key}={value}"]
subprocess.run(args, check=True, capture_output=True)
return _fire

264
agent-team/ci/agent-team-apply-verify.yml
View file

@ -81,6 +81,21 @@ on:
(boundary 2). A diff that changes anything outside this set fails.
required: true
type: string
diff_b64:
description: >-
Base64 of candidate.diff — the diff CONTENT, carried in by the trusted
dispatcher (the box has no write token, D2). The materialize job
decodes it to an artifact and re-hashes it against expected_diff_hash;
it is UNTRUSTED DATA, never executed in a privileged context.
required: true
type: string
head_branch:
description: >-
The branch the trusted dispatcher already pushed with the diff applied.
The draft PR opens with this as `--head`. The box never pushes it; the
PR head and the verified diff are bound by expected_diff_hash.
required: true
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
@ -94,6 +109,63 @@ concurrency:
cancel-in-progress: true
jobs:
# ───────────────────────────────────────────────────────────────────────────
# JOB 0 — materialize (§4.3 diff transport). Credential-less, no secrets, no
# write token. Decodes the base64 `diff_b64` dispatch input to candidate.diff,
# fail-closed re-hashes it against `expected_diff_hash`, and uploads it as the
# named artifact so guard + build-test can download it IN THIS RUN. This is how
# the read-only box's diff reaches CI without the box holding a write token
# (D2): the trusted dispatcher passes the bytes as an input; CI re-verifies the
# hash here AND again in guard. The diff is DATA — never applied/executed here.
# ───────────────────────────────────────────────────────────────────────────
materialize:
runs-on: ubuntu-latest
timeout-minutes: 5
permissions: {}
steps:
- name: Harden runner (block egress; no secrets present)
uses: step-security/harden-runner@0080882f6c36860b6ba35c610c98ce87d4e2f26f # v2.10.2
with:
egress-policy: block
allowed-endpoints: >
github.com:443
api.github.com:443
objects.githubusercontent.com:443
*.actions.githubusercontent.com:443
- name: Decode candidate diff from the dispatch input (untrusted DATA)
env:
# Untrusted: read from env, NEVER interpolated into the script body
# (CWE-94). base64 -d makes shell-meta in the diff inert here.
DIFF_B64: ${{ inputs.diff_b64 }}
EXPECTED_DIFF_HASH: ${{ inputs.expected_diff_hash }}
run: |
set -euo pipefail
mkdir -p ./_out
printf '%s' "$DIFF_B64" | base64 -d > ./_out/candidate.diff
# Fail-closed: non-empty + bounded size + sha256 must equal the
# ledger-recorded hash the dispatcher passed (guard re-verifies this
# independently). An empty expected hash never counts as a match.
test -s ./_out/candidate.diff
# Size bound (defense-in-depth; the dispatch input is already ~64 KB
# capped by GitHub, the dispatcher caps the diff at 40 KB): reject an
# oversized decoded diff rather than feeding it downstream.
bytes="$(wc -c < ./_out/candidate.diff)"
if [ "$bytes" -gt 49152 ]; then
echo "::error::candidate diff too large (${bytes} bytes)"; exit 1
fi
actual="$(sha256sum ./_out/candidate.diff | cut -d' ' -f1)"
if [ -z "$EXPECTED_DIFF_HASH" ] || [ "$actual" != "$EXPECTED_DIFF_HASH" ]; then
echo "::error::materialized diff hash ${actual} != expected '${EXPECTED_DIFF_HASH}'"
exit 1
fi
- name: Upload candidate diff as the named artifact (same-run only)
uses: actions/upload-artifact@b4b15b8c7c6ac21ea08fcf65892d2ee8f75cf882 # v4.4.3
with:
name: ${{ inputs.diff_artifact_name }}
path: ./_out/candidate.diff
if-no-files-found: error
retention-days: 1
# ───────────────────────────────────────────────────────────────────────────
# JOB 1 — guard (boundaries 2 + 3). Credential-less. Validates the candidate
# diff WITHOUT applying or executing it: re-hashes it (integrity) and runs the
@ -102,6 +174,7 @@ jobs:
# run code here. A failure is terminal: the diff is rejected and ALARM-worthy.
# ───────────────────────────────────────────────────────────────────────────
guard:
needs: materialize
runs-on: ubuntu-latest
timeout-minutes: 5
permissions:
@ -190,6 +263,21 @@ jobs:
"**/*policy*.json",
"**/*.pem",
"**/*.key",
".gitmodules",
"**/.gitmodules",
".husky/**",
"**/.husky/**",
".githooks/**",
"**/.githooks/**",
".gitattributes",
"**/.gitattributes",
".npmrc",
"**/.npmrc",
"**/__generated__/**",
"**/*.generated.*",
"**/dist/**",
"**/build/**",
"**/*.min.js",
)
def canonical(path: str) -> str:
@ -443,6 +531,52 @@ jobs:
sys.exit(main())
PY
- name: Gate-weakening check (§4.5; a diff cannot disable its own checks)
env:
DIFF_PATH: ./_incoming/candidate.diff
run: |
set -euo pipefail
# SECURITY (§4.5): enforce gate-weakening on the LIVE PR-opening path,
# not only box-side. A diff that ADDS a lint/type/coverage/security
# suppression or a test skip/xfail could make build-test pass falsely;
# such a diff is escalated to a human, never auto-built. Mirrors
# agent_team.ci_gate._GATE_WEAKENING_MARKERS. Reads the diff as DATA.
python3 - <<'PY'
import os
import re
import sys
markers = (
"noqa",
"type: ignore",
"type:ignore",
"pragma: no cover",
"pragma: no-cover",
"nosec",
"nosemgrep",
"--no-verify",
)
skip_re = re.compile(
r"@(?:pytest\.mark\.(?:skip|xfail)|unittest\.skip\w*)\b"
r"|\bpytest\.(?:skip|xfail)\s*\("
r"|\.skipTest\s*\("
)
text = open(os.environ["DIFF_PATH"], encoding="utf-8", errors="replace").read()
viol = []
for line in text.splitlines():
if line.startswith("+") and not line.startswith("+++"):
content = line[1:]
low = content.lower()
if any(m in low for m in markers) or skip_re.search(content):
viol.append(content.strip()[:120])
if viol:
print("::error::gate-weakening: a diff cannot add suppressions/skips that disable its own checks")
for v in viol[:20]:
print(f" + {v}")
sys.exit(1)
print("no gate-weakening markers in the diff")
PY
# ───────────────────────────────────────────────────────────────────────────
# JOB 2 — build-test (boundary 1). UNTRUSTED execution. This is the ONLY job
# that applies + runs the patch. It is credential-less: contents:read only, no
@ -549,20 +683,32 @@ jobs:
with:
python-version: "3.12"
- name: Install + build + test (untrusted; result is non-authoritative)
- name: Install + build + test (untrusted; exit code IS authoritative)
id: run
run: |
set -euo pipefail
# Placeholder build/test for the narrowest task class (dep bump /
# single-file fix). At deploy time this is parameterized per target
# repo. Exit code is what matters; any file the patch writes is
# ignored by the authoritative gate (boundary 4).
# Build/test for the narrowest task class (dep bump / single-file fix).
# SECURITY (boundary 4): this step's EXIT CODE is the authoritative
# build-test conclusion the pure-code gate keys on
# (needs.build-test.result). It must therefore FAIL on a red build/test
# — a previous `|| echo` swallowed ruff/pytest failures, which (now that
# the apply path is flipped live) would open draft PRs on red builds.
# The tool installs are best-effort (|| true), but ruff + pytest run
# AUTHORITATIVELY under `set -e` so a real failure fails the job.
#
# DEPLOY: the build/test invocation is parameterized per target repo at
# provisioning (a repo with no pytest suite needs its own command). Until
# then this generic ruff+pytest is the gate; a target whose tests need a
# different runner MUST be wired before it is dispatched.
if [ -f requirements.txt ]; then
python3 -m pip install --quiet -r requirements.txt || true
fi
python3 -m pip install --quiet ruff pytest || true
ruff check . || echo "ruff non-zero (recorded, non-authoritative)"
pytest -q || echo "pytest non-zero (recorded, non-authoritative)"
ruff check .
# pytest exit 5 == "no tests collected"; treat ONLY that as non-fatal
# (a fix with no test suite), every other non-zero is an authoritative
# failure. Never blanket-swallow.
pytest -q || { rc=$?; [ "$rc" -eq 5 ] || exit "$rc"; echo "no tests collected (exit 5)"; }
- name: Post-build denied-path check (build hook may not write the trust surface)
if: always()
@ -629,6 +775,21 @@ jobs:
"**/*policy*.json",
"**/*.pem",
"**/*.key",
".gitmodules",
"**/.gitmodules",
".husky/**",
"**/.husky/**",
".githooks/**",
"**/.githooks/**",
".gitattributes",
"**/.gitattributes",
".npmrc",
"**/.npmrc",
"**/__generated__/**",
"**/*.generated.*",
"**/dist/**",
"**/build/**",
"**/*.min.js",
)
_GLOB_META = set("*?[]")
@ -899,15 +1060,22 @@ jobs:
# Environment at PROVISIONING — GitHub holds the job here until a human
# approves. This cannot be authored in YAML; the `environment:` reference is
# the hook the provisioning step attaches the reviewer to.
# environment: # uncomment at provisioning AFTER the agent-apply environment is created with a required reviewer
# name: agent-apply # uncomment at provisioning AFTER the agent-apply environment is created with a required reviewer
# FLIPPED LIVE 2026-06-22 (provisioning done: agent-apply env + required
# reviewer amoussa1229 + the GitHub App secrets exist). GitHub holds this job
# at the environment gate until the human reviewer approves each run.
# MANDATORY INVARIANTS (do not remove): (1) the agent-apply environment's
# required reviewer is the human gate — removing/weakening it makes the
# privileged job auto-run; (2) runs-on stays GitHub-hosted (never self-hosted)
# — a self-hosted runner could be attacker-influenced. Both are asserted by
# tests/test_apply_verify_workflow_hardening.py.
environment:
name: agent-apply
permissions:
contents: read
# uncomment at provisioning AFTER the agent-apply environment is created with a required reviewer
# The ONLY privileged grant: open a draft PR. Kept COMMENTED until
# provisioning so the job holds ZERO privilege on a test dispatch today;
# the draft-PR + app-token STEPS also stay `if: ${{ false }}` until then.
# pull-requests: write
# The ONLY privileged grant: open a draft PR via the App installation
# token. The required reviewer on the agent-apply environment gates every
# run; the draft-PR + app-token STEPS additionally run only on a clean gate.
pull-requests: write
# NOTE: there is deliberately NO token-federation permission here — this
# workflow uses a GitHub App installation token only, no cloud provider.
steps:
@ -1006,11 +1174,11 @@ jobs:
- name: "Mint GitHub App installation token (pull-requests write only)"
id: app-token
# Hard-disabled with the draft-PR step until provisioning: minting a
# token against an App that does not exist yet would fail, and there is
# nothing to do with it while the PR step is off. The condition is the
# SAME target as the draft-PR step so the two flip together.
if: ${{ false }} # ← flip with the draft-PR step at provisioning (see below).
# LIVE: mint the App token ONLY on a clean pure-code gate pass (same
# condition as the draft-PR step, so the two flip together). The job
# itself is already held at the agent-apply environment's required-reviewer
# gate, so this never runs unapproved.
if: ${{ steps.gate.outputs.gate == 'pass' }}
uses: actions/create-github-app-token@5d869da34e18e7287c1daad50e0b8ea0f506ce69 # v1.11.0
with:
# PROVISIONING: create the GitHub App (single permission:
@ -1031,42 +1199,54 @@ jobs:
# && needs.build-test.result == 'success'
# && steps.gate.outputs.gate == 'pass'
#
# Flip to that condition ONLY after the `agent-apply` environment + the
# GitHub App are provisioned and branch protection is in place (see the
# provisioning runbook). Flipping live against a non-existent environment
# is an unprotected hole, so it stays `${{ false }}` here.
if: ${{ false }} # ← hard-disabled until provisioning (see comment above).
# LIVE: opens the draft PR ONLY on a clean authenticated pass, and only
# after the agent-apply environment's required reviewer approved the job.
if: >-
always()
&& needs.guard.result == 'success'
&& needs.build-test.result == 'success'
&& steps.gate.outputs.gate == 'pass'
env:
# The App installation token (pull-requests: write). gh reads it from
# GH_TOKEN. Untrusted values used below are env-indirected (CWE-94).
GH_TOKEN: ${{ steps.app-token.outputs.token }}
TASK_ID: ${{ inputs.task_id }}
DIFF_HASH: ${{ needs.guard.outputs.diff_hash }}
# 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 }}
# The repo the PR is opened in (the App installation's repo).
GH_REPO: ${{ github.repository }}
run: |
set -euo pipefail
# Draft PR ONLY; never auto-merge (D2). Branch protection + the
# required reviewer on the agent-apply environment + the security
# review + the Claude Code App review are the final enforcement
# (boundary 5). The agent NEVER merges. task_id / diff_hash are read
# from the environment (not interpolated into this script) so an
# untrusted task_id cannot inject shell.
# Draft PR ONLY; never auto-merge (D2). The required reviewer on the
# agent-apply environment + branch protection + the security review +
# the Claude Code App review are the final enforcement (boundary 5).
# The agent NEVER merges.
#
# NOTE: the dispatcher records the agent branch name in the ledger and
# passes it in at provisioning; the branch is created by the trusted
# apply path (the box has no write token, D2), so this step only OPENS
# the PR for an already-pushed agent branch. Wire `--head` to that
# branch input at provisioning.
# NIT-2: build BOTH title and body with `printf '%s'` over the
# env-indirected vars (no `${TASK_ID}` shell interpolation in one and
# printf in the other) so the two are consistent and the untrusted
# values are only ever consumed as printf data.
# §4.6 PR-metadata sanitization: the title/body embed only TASK_ID (a
# coordinator-minted thread id) and DIFF_HASH (a computed sha) — never
# the raw diff or any LLM free-text. Both are env-indirected and only
# ever consumed as printf DATA (no shell interpolation, CWE-94).
# Defense-in-depth: reject a task_id / head_branch that is not a safe
# token before using them, so a malformed dispatch input cannot smuggle
# control characters into the PR text or the git ref.
case "$TASK_ID" in
*[!A-Za-z0-9._-]* | "" ) echo "::error::unsafe task_id"; exit 1 ;;
esac
# Reject not just bad chars/empty, but also a leading/trailing slash,
# any '..' segment, or '//' — so a tampered head_branch can never be a
# git ref-traversal or resolve to an unintended ref (CWE-88).
case "$HEAD_BRANCH" in
*[!A-Za-z0-9._/-]* | "" | /* | */ | *..* | *//* )
echo "::error::unsafe head_branch"; exit 1 ;;
esac
title="$(printf 'agent-apply: %s (diff %s)' "$TASK_ID" "$DIFF_HASH")"
body="$(printf 'Automated draft PR from the agent-team apply/verify pipeline.\n\nTask: %s\nDiff hash: %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")"
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 \
--title "$title" \
--body "$body" \
--base main
echo "draft PR opened (task=$TASK_ID); never auto-merged."
--base main \
--head "$HEAD_BRANCH"
echo "draft PR opened (task=$TASK_ID, head=$HEAD_BRANCH); never auto-merged."

226
agent-team/tests/test_apply_verify_workflow_hardening.py
View file

@ -8,11 +8,16 @@ later edit cannot silently regress it:
reach a ``run`` step via an ``env:`` binding instead.
* The workflow carries NO cloud token-federation (``id-token``) and NO AWS /
cloud-OIDC references — the auth model is a GitHub App installation token.
* The privileged draft-PR step is NOT flipped live (``if: ${{ false }}``).
* The privileged job's ``pull-requests: write`` grant and ``agent-apply``
``environment:`` are COMMENTED OUT (provisioning-time, not live), and the job
``if:`` guards on guard+build-test success; download-artifact steps are
* The privileged draft-PR + App-token steps are FLIPPED LIVE but run ONLY on a
clean pure-code gate pass (``steps.gate.outputs.gate == 'pass'`` + guard/
build-test success), never unconditionally.
* The privileged job binds the ``agent-apply`` ``environment:`` (required-reviewer
gate) and grants exactly ``pull-requests: write`` (no id-token/cloud), and the
job ``if:`` guards on guard+build-test success; download-artifact steps are
run-id pinned.
* §4.3 transport: a credential-less ``materialize`` job decodes the untrusted
``diff_b64`` input (via env, CWE-94) and fail-closed re-hashes it before
uploading the named artifact; the draft PR opens with an explicit ``--head``.
* The embedded gate's empty-hash handling is fail-closed (extracted + executed).
* The three trust-control denylists (guard YAML, post-build YAML, ci_gate) are
identical (no drift), and the post-build NUL-delimited path parsing is robust.
@ -122,7 +127,9 @@ def test_uses_github_app_token_action() -> None:
# --------------------------------------------------------------------------- #
def test_draft_pr_step_is_not_flipped_live() -> None:
def test_draft_pr_step_is_flipped_live_only_on_gate_pass() -> None:
# FLIPPED LIVE: the draft-PR step now runs, but ONLY on a clean authenticated
# pass — never unconditionally, and never on a failed/skipped upstream.
doc = _doc()
pr_steps = [
s
@ -131,47 +138,46 @@ def test_draft_pr_step_is_not_flipped_live() -> None:
]
assert pr_steps, "expected a draft-PR step"
for step in pr_steps:
# YAML parses `${{ false }}` to the string '${{ false }}' OR bool False
# depending on quoting; both mean "disabled". Assert it is one of those,
# never an always()/success() live condition.
cond = step.get("if")
assert cond is not None, "draft-PR step must keep an explicit if-guard"
assert str(cond).strip().startswith("${{ false }}") or cond is False, (
f"draft-PR step must stay hard-disabled, got if: {cond!r}"
cond = str(step.get("if", ""))
assert "${{ false }}" not in cond, "draft-PR step should be flipped live"
assert "steps.gate.outputs.gate == 'pass'" in cond, (
"draft-PR step must require the pure-code gate pass"
)
assert "needs.guard.result == 'success'" in cond
assert "needs.build-test.result == 'success'" in cond
def test_gate_job_privileged_declarations_are_commented_until_provisioning() -> None:
# BLOCK-1 / FIX-4 / QUESTION-2: privileged declarations must be
# provisioning-time, not live. `pull-requests: write` and `environment:` are
# both COMMENTED OUT until the agent-apply environment exists, so a test
# dispatch today holds ZERO privilege. Assert the live job declares neither.
def test_app_token_step_gated_on_pure_code_pass() -> None:
# The App-token mint runs only on a clean gate pass (same condition as the
# draft-PR step), and the job is additionally held at the environment gate.
doc = _doc()
token_steps = [
s
for _, s in _iter_steps(doc)
if "create-github-app-token" in str(s.get("uses", ""))
]
assert token_steps, "expected the App-token step"
for step in token_steps:
cond = str(step.get("if", ""))
assert "${{ false }}" not in cond
assert "steps.gate.outputs.gate == 'pass'" in cond
def test_gate_job_privileged_declarations_are_live() -> None:
# Provisioning done: the gate-and-pr job now binds the agent-apply environment
# (its required reviewer gates every run) and grants exactly pull-requests:
# write — nothing more. contents stays read; no token-federation/id-token.
job = _doc()["jobs"]["gate-and-pr"]
perms = job.get("permissions") or {}
# contents: read stays active; pull-requests must NOT be granted live.
assert perms.get("contents") == "read"
assert "pull-requests" not in perms, (
"pull-requests: write must stay commented until provisioning"
assert perms.get("pull-requests") == "write", (
"the only privileged grant must be pull-requests: write"
)
# No live environment binding.
assert job.get("environment") is None, (
"environment: must stay commented until provisioning"
)
def test_gate_job_provisioning_lines_are_present_as_comments() -> None:
# The exact deploy-gated scaffolding must remain in the file (commented) with
# the provisioning marker, so provisioning knows what to uncomment.
raw = _raw()
assert "# pull-requests: write" in raw
assert "# environment:" in raw
assert "# name: agent-apply" in raw
assert (
raw.count(
"uncomment at provisioning AFTER the agent-apply environment is created "
"with a required reviewer"
)
>= 2
assert "id-token" not in perms
env = job.get("environment")
env_name = env.get("name") if isinstance(env, dict) else env
assert env_name == "agent-apply", (
"gate-and-pr must bind the agent-apply environment (required-reviewer gate)"
)
@ -320,6 +326,30 @@ def test_three_trust_control_denylists_are_identical() -> None:
)
# --------------------------------------------------------------------------- #
# §4.1 runner-trust: no job may run on a self-hosted / user-provided runner
# --------------------------------------------------------------------------- #
def test_no_job_uses_a_self_hosted_runner() -> None:
"""§4.1: every job — ESPECIALLY the privileged gate-and-pr — must run on a
GitHub-hosted runner. A self-hosted/user-provided runner can be
attacker-influenced and must never be able to pick up the privileged job."""
github_hosted_prefixes = ("ubuntu-", "windows-", "macos-")
jobs = _doc()["jobs"]
assert jobs, "workflow defines no jobs"
for name, job in jobs.items():
runs_on = job.get("runs-on")
labels = [runs_on] if isinstance(runs_on, str) else list(runs_on or [])
assert labels, f"job {name!r} has no runs-on"
assert "self-hosted" not in labels, (
f"job {name!r} must not run on a self-hosted runner"
)
assert all(
any(label.startswith(p) for p in github_hosted_prefixes) for label in labels
), f"job {name!r} runs-on must be GitHub-hosted, got {labels!r}"
# --------------------------------------------------------------------------- #
# FIX-2 / INJ-02: robust NUL-delimited post-build denied-path parsing
# --------------------------------------------------------------------------- #
@ -409,3 +439,121 @@ def test_post_build_diff_z_denied_path_is_caught(tmp_path: Path) -> None:
# The tracked-diff side (git diff -z --name-only) is parsed too.
diff = b"src/app.py\x00main.tf\x00"
assert _run_post_build(tmp_path, diff_z=diff, scope="src/**") == 1
# --------------------------------------------------------------------------- #
# §4.3 diff transport: materialize job + new inputs + head-branch wiring
# --------------------------------------------------------------------------- #
def test_new_transport_inputs_present() -> None:
raw = _raw()
assert "diff_b64:" in raw, "missing diff_b64 dispatch input"
assert "head_branch:" in raw, "missing head_branch dispatch input"
def test_materialize_job_is_credential_less_and_first() -> None:
jobs = _doc()["jobs"]
assert "materialize" in jobs, "expected the materialize job"
assert (jobs["materialize"].get("permissions") or {}) == {}, (
"materialize must hold zero permissions (credential-less)"
)
guard_needs = jobs["guard"].get("needs")
assert guard_needs in ("materialize", ["materialize"]), (
"guard must depend on materialize so the named artifact exists same-run"
)
def test_materialize_decodes_diff_b64_via_env_not_inline() -> None:
# CWE-94: the untrusted diff_b64 is consumed from env, never interpolated.
steps = [
s for _, s in _iter_steps(_doc()) if s.get("run") and "base64 -d" in s["run"]
]
assert steps, "expected a base64-decoding step in materialize"
for s in steps:
assert "${{ inputs.diff_b64 }}" not in s["run"], (
"diff_b64 must not be interpolated into the run body (CWE-94)"
)
env = s.get("env") or {}
assert env.get("DIFF_B64") == "${{ inputs.diff_b64 }}"
assert "EXPECTED_DIFF_HASH" in env, "materialize must fail-closed re-hash"
assert "sha256sum" in s["run"]
def test_materialize_uploads_named_artifact() -> None:
up = [
s
for _, s in _iter_steps(_doc())
if "upload-artifact" in str(s.get("uses", ""))
and (s.get("with") or {}).get("name") == "${{ inputs.diff_artifact_name }}"
]
assert up, "materialize must upload candidate.diff as the named artifact"
def test_draft_pr_opens_with_head_branch_and_sanitizes_inputs() -> None:
pr_steps = [
s
for _, s in _iter_steps(_doc())
if "draft" in s.get("name", "").lower() and s.get("run")
]
assert pr_steps
for s in pr_steps:
assert "HEAD_BRANCH" in (s.get("env") or {}), "head_branch must be env-bound"
assert "--head" in s["run"], "draft PR must open with an explicit --head"
# §4.6 / defense-in-depth: task_id + head_branch are charset-validated.
assert "unsafe task_id" in s["run"]
assert "unsafe head_branch" in s["run"]
def test_head_branch_guard_rejects_ref_traversal() -> None:
# CWE-88: the draft-PR step rejects a head_branch with a leading/trailing
# slash, a '..' segment, or '//' (not just bad chars).
pr_steps = [
s
for _, s in _iter_steps(_doc())
if "draft" in s.get("name", "").lower() and s.get("run")
]
assert pr_steps
for s in pr_steps:
run = s["run"]
assert "*..*" in run and "*//*" in run and "/*" in run, (
"head_branch guard must reject .. // and leading/trailing slash"
)
def test_materialize_bounds_decoded_diff_size() -> None:
steps = [
s for _, s in _iter_steps(_doc()) if s.get("run") and "base64 -d" in s["run"]
]
assert steps
for s in steps:
assert "too large" in s["run"], "materialize must bound the decoded diff size"
def test_build_test_step_does_not_swallow_failures() -> None:
# LOGIC-2: the build/test exit code is the authoritative build-test
# conclusion the gate keys on, so ruff/pytest must NOT be swallowed by a
# `|| echo ... non-authoritative` (which would open draft PRs on red builds).
steps = [
s for _, s in _iter_steps(_doc()) if s.get("run") and "ruff check" in s["run"]
]
assert steps, "expected the build/test step"
for s in steps:
run = s["run"]
assert 'echo "ruff non-zero' not in run, "ruff failure must not be swallowed"
assert 'echo "pytest non-zero' not in run, (
"pytest failure must not be swallowed"
)
# ruff runs authoritatively under set -e (bare invocation).
assert re.search(r"^\s*ruff check \.\s*$", run, re.MULTILINE), (
"ruff check must run authoritatively (no failure swallow)"
)
def test_guard_runs_gate_weakening_check_on_live_path() -> None:
# LOGIC-3: §4.5 enforced on the LIVE PR-opening path (guard), not only box-side.
guard_steps = _doc()["jobs"]["guard"]["steps"]
names = [s.get("name", "").lower() for s in guard_steps]
assert any("gate-weakening" in n for n in names), (
"guard must run a gate-weakening check so the live path enforces §4.5"
)

94
agent-team/tests/test_ci_gate.py
View file

@ -11,6 +11,7 @@ from agent_team.ci_gate import (
denylist_violations,
diff_touched_paths,
evaluate_ci_gate,
gate_weakening_violations,
verify_diff_hash,
)
from agent_team.state_store import compute_content_hash
@ -93,6 +94,21 @@ def test_clean_diff_has_no_violations() -> None:
"deploy/iam/role.json",
"stacks/policies/admin.json",
"modules/main.tf",
# P3-flip §4.2 — direct code-execution / supply-chain vectors.
".gitmodules",
"vendor/.gitmodules",
".husky/pre-commit",
"frontend/.husky/commit-msg",
".githooks/pre-push",
".gitattributes",
"pkg/.gitattributes",
".npmrc",
"web/.npmrc",
"src/__generated__/schema.ts",
"api/types.generated.ts",
"web/dist/bundle.js",
"service/build/output.o",
"assets/app.min.js",
],
)
def test_denylisted_paths_flagged(path: str) -> None:
@ -100,6 +116,16 @@ def test_denylisted_paths_flagged(path: str) -> None:
assert violations, f"expected {path!r} to be denylisted"
def test_lockfiles_are_NOT_denylisted_so_the_tier3_fixer_can_ship() -> None:
"""Deliberate (§4.2 note): lockfile RCE is contained by the credential-less
egress-blocked build sandbox, and the dep-CVE fixer rewrites lockfiles to
produce draft PRs — so lockfiles are intentionally NOT on the denylist."""
for lock in ("package-lock.json", "web/yarn.lock", "poetry.lock", "Cargo.lock"):
assert denylist_violations(_diff_for(lock)) == [], (
f"{lock!r} must not be denylisted (would break the Tier-3 fixer)"
)
def test_rename_into_workflow_is_flagged() -> None:
diff = (
"diff --git a/safe.txt b/.github/workflows/evil.yml\n"
@ -386,3 +412,71 @@ def test_gate_result_carries_provenance() -> None:
assert result.run_id == "run-7"
assert result.diff_hash == h
assert result.reasons # always records why
# --------------------------------------------------------------------------- #
# §4.5 gate-weakening detector
# --------------------------------------------------------------------------- #
def _diff_adding(added_line: str, path: str = "src/mod.py") -> str:
"""A unified diff that ADDS one line (plus an unchanged context line)."""
return (
f"diff --git a/{path} b/{path}\n--- a/{path}\n+++ b/{path}\n"
f"@@ -1 +1,2 @@\n unchanged\n+{added_line}\n"
)
@pytest.mark.parametrize(
"added_line",
[
"x = 1 # noqa",
"y = 2 # noqa: E501",
"z = bad() # type: ignore",
"z = bad() # type:ignore[arg-type]",
"def f(): # pragma: no cover",
"def f(): # pragma: no-cover",
"subprocess.run(cmd) # nosec",
"eval(x) # nosemgrep",
" git commit --no-verify",
"@pytest.mark.skip(reason='flaky')",
"@pytest.mark.xfail",
" pytest.skip('todo')",
" self.skipTest('later')",
],
)
def test_gate_weakening_added_line_flagged(added_line: str) -> None:
violations = gate_weakening_violations(_diff_adding(added_line))
assert violations, f"expected {added_line!r} to be flagged as gate-weakening"
def test_gate_weakening_clean_diff_has_no_violations() -> None:
assert gate_weakening_violations(_diff_for("src/foo.py")) == []
def test_gate_weakening_only_inspects_added_lines() -> None:
# A diff that REMOVES a suppression (line starts with '-') must not be flagged.
removing = (
"diff --git a/m.py b/m.py\n--- a/m.py\n+++ b/m.py\n"
"@@ -1,2 +1 @@\n-x = 1 # noqa\n unchanged\n"
)
assert gate_weakening_violations(removing) == []
def test_gate_weakening_rejects_non_str() -> None:
with pytest.raises(CiGateError):
gate_weakening_violations(None) # type: ignore[arg-type]
def test_gate_blocks_weakening_diff_even_if_ci_success() -> None:
# A diff that adds a coverage suppression BLOCKs even with a green CI run.
diff = _diff_adding("def f(): # pragma: no cover")
result = evaluate_ci_gate(
candidate_diff=diff,
ledger_hash=_ledger_hash(diff),
ci_result=_good_ci("run-w", diff, "success"),
expected_run_id="run-w",
)
assert result.decision is GateDecision.BLOCK
assert result.blocked is True
assert any("gate-weakening" in r for r in result.reasons)

179
agent-team/tests/test_dispatcher.py Normal file
View file

@ -0,0 +1,179 @@
"""Unit tests for agent_team.dispatcher — the trusted apply-path transport (§4.3).
Fully hermetic: the branch-push and workflow-dispatch side effects are injected
fakes, so no git, no ``gh``, and no network are exercised. The tests pin the
pure input-assembly + validation contract and the push-before-dispatch order.
"""
from __future__ import annotations
import base64
import pytest
from agent_team.dispatcher import (
MAX_DIFF_BYTES,
DispatcherError,
DispatchInputs,
build_dispatch_inputs,
dispatch_apply_verify,
head_branch_for,
)
from agent_team.state_store import compute_content_hash
TASK = "0a1b2c3d4e5f6071"
DIFF = "diff --git a/README.md b/README.md\n--- a/README.md\n+++ b/README.md\n@@ -1 +1,2 @@\n title\n+added line\n"
SCOPE = "README.md\ndocs/**"
# --------------------------------------------------------------------------- #
# build_dispatch_inputs (pure)
# --------------------------------------------------------------------------- #
def test_build_dispatch_inputs_binds_hash_and_b64() -> None:
di = build_dispatch_inputs(task_id=TASK, diff_text=DIFF, declared_scope=SCOPE)
# expected_diff_hash is the plain sha256 the workflow's sha256sum reproduces.
assert di.expected_diff_hash == compute_content_hash(DIFF.encode("utf-8"))
# diff_b64 round-trips back to the exact diff bytes.
assert base64.b64decode(di.diff_b64).decode("utf-8") == DIFF
assert di.head_branch == f"agent-team/apply/{TASK}"
assert di.diff_artifact_name == f"agent-team-diff-{TASK}"
assert di.declared_scope == SCOPE
def test_as_inputs_keys_match_the_six_workflow_inputs() -> None:
di = build_dispatch_inputs(task_id=TASK, diff_text=DIFF, declared_scope=SCOPE)
assert set(di.as_inputs()) == {
"task_id",
"diff_artifact_name",
"expected_diff_hash",
"declared_scope",
"diff_b64",
"head_branch",
}
@pytest.mark.parametrize("bad_diff", ["", " ", "\n\n"])
def test_empty_diff_rejected(bad_diff: str) -> None:
with pytest.raises(DispatcherError):
build_dispatch_inputs(task_id=TASK, diff_text=bad_diff, declared_scope=SCOPE)
def test_oversized_diff_rejected() -> None:
# The diff rides a base64 workflow_dispatch input (GitHub ~64 KB cap); a diff
# over MAX_DIFF_BYTES must fail closed in the dispatcher, not be dispatched.
big = "diff --git a/x b/x\n" + "+" + ("x" * (MAX_DIFF_BYTES + 1)) + "\n"
with pytest.raises(DispatcherError):
build_dispatch_inputs(task_id=TASK, diff_text=big, declared_scope=SCOPE)
@pytest.mark.parametrize("bad_scope", ["", " "])
def test_empty_scope_rejected(bad_scope: str) -> None:
# An empty declared scope would let a diff touch ANY path — fail closed.
with pytest.raises(DispatcherError):
build_dispatch_inputs(task_id=TASK, diff_text=DIFF, declared_scope=bad_scope)
@pytest.mark.parametrize(
"bad_task", ["", "has space", "semi;colon", "../escape", "a/b", "x" * 201]
)
def test_unsafe_task_id_rejected(bad_task: str) -> None:
with pytest.raises(DispatcherError):
head_branch_for(bad_task)
with pytest.raises(DispatcherError):
build_dispatch_inputs(task_id=bad_task, diff_text=DIFF, declared_scope=SCOPE)
# --------------------------------------------------------------------------- #
# dispatch_apply_verify (injected seams)
# --------------------------------------------------------------------------- #
class _Recorder:
def __init__(self) -> None:
self.calls: list[dict] = []
def test_dispatch_pushes_then_fires_with_correct_inputs() -> None:
order: list[str] = []
pushed = _Recorder()
fired = _Recorder()
def pusher(*, owner, repo, base, head_branch, diff_text):
order.append("push")
pushed.calls.append(
{
"owner": owner,
"repo": repo,
"base": base,
"head": head_branch,
"diff": diff_text,
}
)
def dispatcher(*, owner, repo, inputs, ref):
order.append("dispatch")
fired.calls.append({"owner": owner, "repo": repo, "inputs": inputs, "ref": ref})
di = dispatch_apply_verify(
owner="Sea-Haven-Industries",
repo="orchestrator",
task_id=TASK,
diff_text=DIFF,
declared_scope=SCOPE,
pusher=pusher,
dispatcher=dispatcher,
)
assert isinstance(di, DispatchInputs)
# Branch is pushed BEFORE the workflow is dispatched (the draft-PR step opens
# against an already-pushed --head).
assert order == ["push", "dispatch"]
assert pushed.calls[0]["head"] == f"agent-team/apply/{TASK}"
assert pushed.calls[0]["diff"] == DIFF
# The dispatch carries all six inputs, including the b64 diff + head branch.
inputs = fired.calls[0]["inputs"]
assert inputs["head_branch"] == f"agent-team/apply/{TASK}"
assert base64.b64decode(inputs["diff_b64"]).decode("utf-8") == DIFF
assert inputs["expected_diff_hash"] == compute_content_hash(DIFF.encode("utf-8"))
assert fired.calls[0]["ref"] == "main"
def test_dispatch_does_not_fire_if_push_fails() -> None:
fired = _Recorder()
def failing_pusher(**_kw):
raise RuntimeError("push failed")
def dispatcher(**kw):
fired.calls.append(kw)
with pytest.raises(RuntimeError):
dispatch_apply_verify(
owner="o",
repo="r",
task_id=TASK,
diff_text=DIFF,
declared_scope=SCOPE,
pusher=failing_pusher,
dispatcher=dispatcher,
)
# A failed push must NOT dispatch a run (no orphan run against a missing head).
assert fired.calls == []
@pytest.mark.parametrize(
"owner,repo", [("", "r"), ("o", ""), ("bad owner", "r"), ("o", "r/x")]
)
def test_unsafe_owner_repo_rejected(owner: str, repo: str) -> None:
with pytest.raises(DispatcherError):
dispatch_apply_verify(
owner=owner,
repo=repo,
task_id=TASK,
diff_text=DIFF,
declared_scope=SCOPE,
pusher=lambda **_k: None,
dispatcher=lambda **_k: None,
)