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

feat(agent-team): Plane-1 Tier-3 fixer — dependency-cve finding -> patch + CI dispatch (opt-in/inert)

Browse source 
The fixer (design §4 fixer row, §7 Phase 5, §3.3.2) takes a CONFIRMED,
low-risk dependency-cve finding (the narrowest fix class) and produces:

  * a fix SPEC (Claude, via the §3.1 billing seam), and
  * a minimal bump PATCH (DeepSeek fast_coder, via the orchestrator run.py
    path that builders_llm uses),

records the candidate diff + its content-hash, and emits the org-CI
workflow_dispatch inputs (task_id / diff_artifact_name / expected_diff_hash /
declared_scope) for the gate-passed P3-live apply/verify surface.

INERT / opt-in / fail-safe, mirroring build_verify_wiring:
  * plan_fix dispatches NOTHING; dispatch_fix has NO default dispatcher
    (the box holds no write token, D2) so an un-wired call can never fire a
    workflow.
  * no git/patch/subprocess/fs-write in executable code — the patch is emitted
    as diff TEXT only; CI applies it and opens a DRAFT PR, the box never
    applies/pushes/merges.
  * untrusted-patch hygiene: the generated diff is confined box-side to the
    single dependency manifest (declared_scope) and rejected via
    ci_gate.denylist_violations if it escapes scope or touches the
    trust-control surface — defense-in-depth with the CI guard.
  * bad/ambiguous findings (wrong check/status/category, missing
    package/fixed_version, ambiguous fixed_version, unparseable/empty diff)
    yield a FAILED no-op plan, never a fabricated fix.

29 new pytest tests under agent-team/tests/test_fixer.py.
This commit is contained in:
Adam Moussa 2026-06-18 16:03:58 -04:00
parent 71b160e61b
commit bf1eed1a31
2 changed files with 914 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

492
agent-team/agent_team/nodes/fixer.py Normal file
View file

@ -0,0 +1,492 @@
"""Plane-1 Tier-3 fixer — confirmed finding -> fix patch -> CI draft-PR dispatch.
Design refs: ``docs/r720-agent-team-design.md`` §4 (Tier-3 fixer row), §6.2,
§7 Phase 5, §3.3.2 (the untrusted-patch trust boundary the fixer feeds).
WHAT IT DOES
============
The fixer takes a **CONFIRMED, low-risk Plane-1 finding** (the NARROWEST class:
a ``dependency-cve`` vulnerable-pin finding), produces a fix **spec** (Claude,
via the §3.1 billing seam) and a minimal fix **patch** (DeepSeek ``fast_coder``,
via the local orchestrator ``run.py`` — exactly the path
:mod:`agent_team.nodes.builders_llm` uses), records the candidate diff + its
content-hash, and emits the **dispatch request** for the org CI apply/verify
workflow (``ci/agent-team-apply-verify.yml``, the gate-passed P3-live surface).
The box stays READ-ONLY (D2/D11): it emits a patch + a dispatch request as DATA.
CI applies the patch and opens a DRAFT PR; the box NEVER applies the patch to the
local tree, NEVER pushes, NEVER holds a standing write token, and NEVER merges.
============================ INERT / OPT-IN / FAIL-SAFE ====================
This module mirrors the opt-in/inert discipline of
:func:`agent_team.coordinator.build_verify_wiring` /
:func:`agent_team.coordinator.gated_build_verify_wiring`: it is **not** wired
into the default ``run-team.py`` ``serve`` path, and produces a fix plan WITHOUT
dispatching unless a live dispatcher is explicitly bound.
* **No live dispatch by default.** :func:`plan_fix` produces a
:class:`FixPlan` (spec + patch + dispatch inputs) as DATA and dispatches
NOTHING. Dispatch happens only through :func:`dispatch_fix` with an
explicitly-supplied :data:`WorkflowDispatcher` callable — there is no default
dispatcher (the default is ``None`` -> raise), so an un-wired call can never
fire a workflow.
* **No local mutation.** The patch is produced as diff TEXT via a subprocess
that only ASKS the model for the diff. There is NO ``git apply``/``patch``/
``git``/``write_text``/``open(..., "w")`` path anywhere in this module — by
construction the fixer cannot mutate the working tree or push.
* **Fail-safe.** A finding that is not a confirmed, unambiguous dependency-cve
fix (wrong check/status/category, missing package/fixed_version, an ambiguous
or multi-manifest scope, an unparseable/empty model diff, or a diff that
strays outside the declared single-manifest scope or touches the
trust-control surface) yields a FAILED :class:`FixPlan` (``ok is False``, no
patch) — never a fabricated fix and never a dispatch.
UNTRUSTED-PATCH HYGIENE (§3.3.2)
================================
A generated patch is UNTRUSTED model output. The fixer does NOT trust it: it
flows through the SAME P3-live boundary the builders feed —
``expected_diff_hash`` binding (boundary #3), the declared-scope confinement +
trust-control denylist (boundary #2), and the credential-less apply/draft-PR CI
(boundaries #1/#4/#5). The fixer's job is to (a) confine ``declared_scope`` to
the single dependency manifest the finding names, and (b) reject — box-side,
before any dispatch — any diff that escapes that scope or hits the denylist, so
the CI guard's identical check is defense-in-depth, not the only line.
This module imports the committed contracts verbatim and redefines none of them.
"""
from __future__ import annotations
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field
from typing import Any
from agent_team.billing import BillingMode, claude_invoke
from agent_team.ci_gate import denylist_violations
from agent_team.nodes.builders_llm import BuildCallable, default_build
from agent_team.state_store import compute_content_hash
__all__ = [
"FixDispatchInputs",
"FixPlan",
"SpecCallable",
"WorkflowDispatcher",
"describe_plan",
"dispatch_fix",
"is_fixable_dependency_finding",
"plan_fix",
]
# The CI workflow_dispatch input contract reused VERBATIM from the gate-passed
# P3-live surface (ci/agent-team-apply-verify.yml `on.workflow_dispatch.inputs`).
# The artifact named by ``diff_artifact_name`` MUST contain a file
# ``candidate.diff`` whose sha256 equals ``expected_diff_hash`` (the guard job
# re-hashes it). The trusted apply path (which owns the GitHub App write token)
# uploads the artifact + dispatches; the box only EMITS these inputs (it has no
# write token, D2).
CANDIDATE_DIFF_FILENAME = "candidate.diff"
# A dependency-cve finding's identifying contract (security-review/checkers/
# dependency-cve.sh add_finding). We bind the fixer to EXACTLY this class — the
# narrowest, lowest-risk fix per the design — and refuse anything else.
_FIX_CHECK = "vulnerable-dependency"
_FIX_STATUS = "confirmed"
_FIX_CATEGORY = "other"
# The single manifest file the fixer is allowed to scope a fix to. We start with
# PyPI/requirements.txt ONLY (the narrowest manifest: one ``name==version`` pin
# per line). Other ecosystems (npm lockfiles, NuGet, poetry lockfiles) carry
# resolved transitive trees / integrity hashes a one-line bump cannot safely
# edit, so the fixer scopes to this file and they stay human-routed.
_PYPI_MANIFEST = "requirements.txt"
@dataclass
class FixDispatchInputs:
"""The exact ``workflow_dispatch`` inputs for ci/agent-team-apply-verify.yml.
Field names match the workflow's ``on.workflow_dispatch.inputs`` keys 1:1 so
a dispatcher can pass :meth:`as_inputs` straight through. ``declared_scope``
is newline-separated globs (the workflow splits on newlines); the fixer
confines it to the single manifest the finding names so the CI guard's
boundary-2 scope check confines the apply to exactly that file.
"""
task_id: str
diff_artifact_name: str
expected_diff_hash: str
declared_scope: str
def as_inputs(self) -> dict[str, str]:
"""Return the inputs as the workflow's string->string dispatch map."""
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,
}
@dataclass
class FixPlan:
"""A fix proposed as DATA — spec + patch + dispatch inputs (never applied here).
This is what :func:`plan_fix` returns and what a dry-run prints. On success it
carries everything CI needs (the diff text to upload as ``candidate.diff``,
its content-hash, and the dispatch inputs); on failure it is a NO-OP marked
``ok is False`` with a ``reason`` and no patch — never a fabricated fix.
Attributes:
ok: ``True`` only for a confirmed dependency-cve finding that produced a
valid, in-scope, denylist-clean one-line bump diff.
reason: Human-readable explanation when not ``ok`` (empty on success).
finding_id: The source finding's id (provenance; echoed even on failure).
spec: The Claude-authored fix spec (the intent / what to change).
diff: The candidate unified diff (empty on failure).
diff_hash: sha256 of ``diff`` via
:func:`agent_team.state_store.compute_content_hash` (always computed,
including over the empty diff, so the record shape is uniform).
manifest_path: The single dependency manifest the fix is scoped to.
dispatch: The CI ``workflow_dispatch`` inputs (``None`` on failure).
declared_scope: The canonical scope prefixes (the single manifest).
"""
ok: bool
reason: str = ""
finding_id: str = ""
spec: str = ""
diff: str = ""
diff_hash: str = ""
manifest_path: str = ""
dispatch: FixDispatchInputs | None = None
declared_scope: list[str] = field(default_factory=list)
@classmethod
def failed(cls, reason: str, *, finding_id: str = "") -> "FixPlan":
"""Build a FAILED no-op plan (no patch, no dispatch) the caller rejects."""
return cls(
ok=False,
reason=reason,
finding_id=finding_id,
diff="",
diff_hash=compute_content_hash(b""),
dispatch=None,
)
# The injectable spec seam: given the finding, return the Claude-authored fix
# spec text. Default (:func:`_default_spec`) routes to the §3.1 billing seam
# (Claude). Tests pass a fake. Kept separate from the patch (DeepSeek) seam so
# the design's "spec (Claude) + patch (DeepSeek)" split is explicit.
SpecCallable = Callable[[Mapping[str, Any]], str]
# The injectable workflow dispatcher: given the dispatch inputs + the candidate
# diff content, fire the org CI ``workflow_dispatch`` and return an opaque
# dispatch ref. The box has NO write token, so there is deliberately NO default —
# :func:`dispatch_fix` requires an explicitly-bound dispatcher (provisioning
# supplies the trusted apply path's dispatcher). An un-wired call can therefore
# never fire a workflow.
WorkflowDispatcher = Callable[["FixDispatchInputs", str], str]
def is_fixable_dependency_finding(finding: Any) -> tuple[bool, str]:
"""Return ``(fixable, reason)`` for a candidate Plane-1 finding (fail-safe).
A finding is fixable by THIS module only when it is an unambiguous, confirmed
``dependency-cve`` vulnerable-pin finding whose proof carries the package, the
vulnerable version, and a single concrete ``fixed_version`` to bump to. Any
deviation (wrong shape/check/status/category, missing fields, an ambiguous
fixed version) returns ``(False, why)`` so the caller refuses rather than
guessing. This is a pure predicate (no I/O).
"""
if not isinstance(finding, Mapping):
return False, f"finding must be a mapping, got {type(finding).__name__}"
check = finding.get("check")
if check != _FIX_CHECK:
return False, f"not a dependency-cve finding (check={check!r})"
status = finding.get("status")
if status != _FIX_STATUS:
return False, f"finding is not confirmed (status={status!r})"
category = finding.get("category")
if category != _FIX_CATEGORY:
return False, f"unexpected finding category {category!r}"
proof = finding.get("proof")
if not isinstance(proof, Mapping):
return False, "finding has no structured proof block"
package = proof.get("package")
version = proof.get("version")
fixed = proof.get("fixed_version")
if not _nonempty_str(package):
return False, "proof.package is missing/empty"
if not _nonempty_str(version):
return False, "proof.version is missing/empty"
if not _nonempty_str(fixed):
return False, "proof.fixed_version is missing/empty (nothing to bump to)"
# An ambiguous fixed version (a range/list, not one concrete pin) is NOT a
# safe one-line bump — refuse it (the OSV checker emits a single fixed event,
# but a comma/space/range here means we cannot construct ``pkg==fixed``).
if any(c in str(fixed) for c in (" ", ",", "<", ">", "*")):
return False, f"ambiguous fixed_version {fixed!r}; not a single concrete pin"
# Same defensive parse on the vulnerable version: a non-concrete current pin
# means there is no single ``pkg==version`` line to rewrite.
if any(c in str(version) for c in (" ", ",", "<", ">", "*")):
return False, f"ambiguous version {version!r}; not a single concrete pin"
if str(version) == str(fixed):
return False, "vulnerable version equals fixed_version; nothing to bump"
return True, ""
def _nonempty_str(value: Any) -> bool:
"""True if ``value`` is a non-empty, non-whitespace string."""
return isinstance(value, str) and bool(value.strip())
def _default_spec(finding: Mapping[str, Any]) -> str:
"""Default :data:`SpecCallable`: author the fix spec via Claude (§3.1 seam).
Routes to :func:`agent_team.billing.claude_invoke` (the design's "spec =
Claude" half). The spec describes the intended minimal change — bump the one
vulnerable pin to the advisory's fixed version, touching only the manifest.
The model output is treated as untrusted text and is NEVER the authority for
what gets dispatched (that is the deterministic patch + hash + scope below);
the spec is provenance / the instruction the patch step renders from.
"""
proof = finding.get("proof", {})
package = proof.get("package")
version = proof.get("version")
fixed = proof.get("fixed_version")
advisory = proof.get("advisory_id", "")
repo = finding.get("repo", "")
prompt = (
"You are authoring a minimal, low-risk dependency-bump fix spec for a "
"confirmed vulnerable pinned dependency. State ONLY the single change: "
f"in repo {repo!r}, bump the pinned dependency {package}=={version} to "
f"{package}=={fixed} (advisory {advisory}). The change MUST touch only "
"the dependency manifest and nothing else. Reply with a one-paragraph "
"spec, no code."
)
result = claude_invoke(prompt, mode=BillingMode.SUBSCRIPTION)
return result.text
def _render_patch_instruction(finding: Mapping[str, Any], *, manifest_path: str) -> str:
"""Render the dep-bump into a mechanical-edit instruction for fast_coder.
Pure string assembly (no I/O) so the instruction shape is unit-testable. The
instruction tells DeepSeek to emit ONLY a single unified diff that bumps the
one pin in ``manifest_path`` — the box-side scope + denylist check in
:func:`plan_fix` is the real enforcement, but reinforcing it in the prompt
keeps the model on-task.
"""
proof = finding.get("proof", {})
package = proof.get("package")
version = proof.get("version")
fixed = proof.get("fixed_version")
advisory = proof.get("advisory_id", "")
return (
"You are performing a single mechanical dependency bump. Emit ONLY a "
"git unified diff (no prose, no explanation, no code fences). Change "
f"EXACTLY one line in `{manifest_path}`: the pinned requirement "
f"`{package}=={version}` must become `{package}=={fixed}` (security "
f"advisory {advisory}). Touch ONLY `{manifest_path}` and ONLY that one "
"pin line — do not reformat, reorder, or edit any other line or file. "
f"Both the `---`/`+++` headers must reference `{manifest_path}`."
)
def _extract_diff(text: str) -> str | None:
"""Extract a unified diff from UNTRUSTED model output, or ``None`` (fail-safe).
Delegates to :func:`agent_team.nodes.builders_llm._extract_diff` so the fixer
and the builders share one defensive parser (framing lines, code fences,
header validation). Pure text inspection — it never executes/applies anything.
"""
from agent_team.nodes.builders_llm import _extract_diff as _b_extract
return _b_extract(text)
def plan_fix(
finding: Any,
*,
task_id: str,
spec: SpecCallable | None = None,
build: BuildCallable | None = None,
artifact_prefix: str = "candidate-diff",
) -> FixPlan:
"""Plan a dep-bump fix for ``finding`` as DATA (spec + patch + dispatch inputs).
Validates the finding (:func:`is_fixable_dependency_finding`), authors the fix
spec (Claude via ``spec``, default :func:`_default_spec`), produces the bump
patch (DeepSeek via ``build``, default
:func:`agent_team.nodes.builders_llm.default_build`), parses the UNTRUSTED
diff, confines ``declared_scope`` to the single manifest, and rejects any diff
that escapes that scope or hits the trust-control denylist. Returns a
:class:`FixPlan`.
DISPATCHES NOTHING. The returned plan carries the CI ``workflow_dispatch``
inputs (:class:`FixDispatchInputs`); firing them is :func:`dispatch_fix`'s job
and requires an explicitly-bound dispatcher.
FAIL-SAFE (never fabricate a fix, never escape scope): a non-fixable finding,
a build error, an unparseable/empty diff, or a diff that strays outside the
single-manifest scope / touches the denylist all yield ``FixPlan.failed(...)``
— an ``ok is False`` no-op with no patch and no dispatch.
"""
finding_id = finding.get("id", "") if isinstance(finding, Mapping) else ""
fixable, why = is_fixable_dependency_finding(finding)
if not fixable:
return FixPlan.failed(
f"finding is not a fixable dependency-cve: {why}", finding_id=finding_id
)
if not _nonempty_str(task_id):
return FixPlan.failed("task_id is missing/empty", finding_id=finding_id)
# The single manifest the fix is confined to. PyPI/requirements.txt only (the
# narrowest manifest); other ecosystems are out of scope for this phase.
manifest_path = _PYPI_MANIFEST
declared_scope = [manifest_path]
# 1) Spec (Claude). Untrusted text used only as the instruction/provenance.
spec_fn: SpecCallable = spec if spec is not None else _default_spec
try:
spec_text = spec_fn(finding)
except Exception as exc: # noqa: BLE001 - any spec failure fails SAFE
return FixPlan.failed(
f"fix-spec generation failed: {type(exc).__name__}", finding_id=finding_id
)
if not _nonempty_str(spec_text):
spec_text = "" # a missing spec is non-fatal; the patch is authoritative
# 2) Patch (DeepSeek fast_coder via the orchestrator run.py).
instruction = _render_patch_instruction(finding, manifest_path=manifest_path)
build_fn: BuildCallable = build if build is not None else default_build
try:
raw = build_fn(instruction)
except Exception as exc: # noqa: BLE001 - any build failure fails SAFE
return FixPlan.failed(
f"patch generation failed: {type(exc).__name__}", finding_id=finding_id
)
if not isinstance(raw, str) or not raw.strip():
return FixPlan.failed(
"patch generator produced empty output", finding_id=finding_id
)
diff = _extract_diff(raw)
if diff is None:
return FixPlan.failed(
"patch output is not a usable unified diff", finding_id=finding_id
)
# 3) UNTRUSTED-PATCH HYGIENE (§3.3.2). Box-side enforcement BEFORE any
# dispatch: the generated diff must stay inside the single declared manifest
# scope and must not touch the trust-control surface.
# ci_gate.denylist_violations runs the SAME denylist + scope check the CI
# guard re-runs (defense-in-depth).
violations = denylist_violations(diff, allowed_scope=declared_scope)
if violations:
return FixPlan.failed(
"generated patch escapes the declared single-manifest scope or hits "
f"the trust-control denylist: {'; '.join(violations)}",
finding_id=finding_id,
)
# 4) Diff integrity hash (boundary #3) — the value CI's guard re-hashes and
# the ledger records, so the apply/verify hash binding holds.
diff_hash = compute_content_hash(diff.encode("utf-8"))
dispatch = FixDispatchInputs(
task_id=task_id,
diff_artifact_name=f"{artifact_prefix}-{task_id}",
expected_diff_hash=diff_hash,
# newline-separated globs (the workflow splits on newlines). One entry:
# the single manifest, so the CI guard confines the apply to exactly it.
declared_scope="\n".join(declared_scope),
)
return FixPlan(
ok=True,
reason="",
finding_id=finding_id,
spec=spec_text,
diff=diff,
diff_hash=diff_hash,
manifest_path=manifest_path,
dispatch=dispatch,
declared_scope=declared_scope,
)
def dispatch_fix(plan: FixPlan, *, dispatcher: WorkflowDispatcher | None) -> str:
"""Fire the CI ``workflow_dispatch`` for an ``ok`` :class:`FixPlan` (OPT-IN).
The ONLY live side-effect path. It is fail-closed and opt-in:
* a non-``ok`` plan (or one missing dispatch inputs) raises ``ValueError`` —
a failed/no-op plan is NEVER dispatched;
* ``dispatcher`` has NO default. The box holds no write token, so dispatching
requires the trusted apply path's dispatcher to be explicitly bound at
provisioning. ``dispatcher=None`` raises :class:`RuntimeError`, so an
un-wired call can never fire a workflow (mirrors the build_verify wiring
staying opt-in / off in the default serve path).
On success it forwards the candidate-diff content (to be uploaded as
``candidate.diff``) and the dispatch inputs to the dispatcher and returns its
opaque dispatch ref. This module itself performs NO network, NO git, and NO
write — the dispatcher (provisioned separately, with the GitHub App token)
owns the artifact upload + the actual ``workflow_dispatch`` call + the draft
PR.
"""
if not plan.ok or plan.dispatch is None:
raise ValueError(
f"refusing to dispatch a non-ok fix plan (reason={plan.reason!r})"
)
if dispatcher is None:
raise RuntimeError(
"no workflow dispatcher bound; the box has no write token, so dispatch "
"requires the trusted apply path's dispatcher to be explicitly supplied "
"(opt-in/inert by default — nothing is dispatched without it)"
)
return dispatcher(plan.dispatch, plan.diff)
def describe_plan(plan: FixPlan) -> str:
"""Render a human-readable dry-run summary of a :class:`FixPlan` (no I/O).
Used by the ``fix --dry-run`` CLI to show what WOULD be dispatched (the spec,
the patch, and the CI ``workflow_dispatch`` inputs) without dispatching. Pure
string assembly so it is unit-testable.
"""
lines: list[str] = []
lines.append(f"finding: {plan.finding_id or '(unknown)'}")
if not plan.ok:
lines.append(f"FIX NOT PLANNED (fail-safe): {plan.reason}")
lines.append("dispatch: NONE")
return "\n".join(lines)
lines.append(f"manifest (declared scope): {plan.manifest_path}")
lines.append(f"diff content-hash (sha256): {plan.diff_hash}")
lines.append("")
lines.append("--- fix spec (Claude) ---")
lines.append(plan.spec or "(no spec)")
lines.append("")
lines.append("--- candidate patch (DeepSeek; NOT applied locally) ---")
lines.append(plan.diff)
lines.append("")
lines.append("--- CI workflow_dispatch inputs (would dispatch) ---")
if plan.dispatch is not None:
for key, value in plan.dispatch.as_inputs().items():
lines.append(f" {key}: {value!r}")
lines.append("")
lines.append(
"DRY-RUN: nothing dispatched. CI (agent-team-apply-verify.yml) would "
"apply the patch and open a DRAFT PR; the box never applies/pushes/merges."
)
return "\n".join(lines)

422
agent-team/tests/test_fixer.py Normal file
View file

@ -0,0 +1,422 @@
"""Unit tests for agent_team.nodes.fixer (Plane-1 Tier-3 fixer, §4 / §7 Phase 5).
The fixer is exercised with FAKE spec/build callables — no Claude, no DeepSeek,
no network, no subprocess. The load-bearing properties under test:
* **Happy path.** A confirmed dependency-cve finding + a fake build returning a
valid one-line bump diff yields an ``ok`` :class:`FixPlan` with the right diff,
a real content-hash, the single-manifest declared scope, and the exact CI
``workflow_dispatch`` inputs (task_id / artifact / hash / scope).
* **Scope confinement (untrusted-patch hygiene).** A generated diff that touches
anything beyond the single declared manifest — or the trust-control surface —
fails SAFE to a no-op plan (no patch, no dispatch).
* **INERT / opt-in.** ``plan_fix`` dispatches nothing; ``dispatch_fix`` with no
dispatcher raises (the box has no write token) and never fires a workflow.
* **Fail-safe on bad/ambiguous findings.** Wrong check/status/category, missing
package/fixed_version, an ambiguous fixed version, or an unparseable/empty
model diff all yield a FAILED plan, never a fabricated fix.
* **No mutation surface.** The module exposes no git/patch/fs-write path.
"""
from __future__ import annotations
import inspect
from typing import Any
import pytest
from agent_team.nodes import fixer
from agent_team.nodes.fixer import (
FixDispatchInputs,
FixPlan,
describe_plan,
dispatch_fix,
is_fixable_dependency_finding,
plan_fix,
)
from agent_team.state_store import compute_content_hash
# A confirmed dependency-cve finding, exactly the shape
# security-review/checkers/dependency-cve.sh emits.
def _finding(**overrides: Any) -> dict[str, Any]:
base: dict[str, Any] = {
"repo": "example-repo",
"id": "example-repo-vuln-requests-2-19-0-CVE-2018-18074",
"title": "requests 2.19.0 is vulnerable (CVE-2018-18074)",
"severity": "high",
"category": "other",
"check": "vulnerable-dependency",
"status": "confirmed",
"proof": {
"package": "requests",
"version": "2.19.0",
"advisory_id": "CVE-2018-18074",
"summary": "requests before 2.20.0 leaks auth on redirect",
"fixed_version": "2.20.0",
},
}
base.update(overrides)
return base
# A valid one-line bump diff confined to requirements.txt.
_BUMP_DIFF = (
"diff --git a/requirements.txt b/requirements.txt\n"
"--- a/requirements.txt\n"
"+++ b/requirements.txt\n"
"@@ -1,1 +1,1 @@\n"
"-requests==2.19.0\n"
"+requests==2.20.0\n"
)
def _fake_build(diff_text: str) -> Any:
"""Return a fake BuildCallable that yields ``diff_text`` (ignores instruction)."""
def _build(_instruction: str) -> str:
return diff_text
return _build
def _fake_spec(_finding: Any) -> str:
return "Bump requests==2.19.0 to requests==2.20.0 in the manifest only."
# --------------------------------------------------------------------------- #
# Happy path: correct dep-bump patch + hash + dispatch inputs
# --------------------------------------------------------------------------- #
def test_plan_fix_produces_correct_patch_hash_and_dispatch() -> None:
plan = plan_fix(
_finding(),
task_id="t-123",
spec=_fake_spec,
build=_fake_build(_BUMP_DIFF),
)
assert plan.ok is True
assert plan.reason == ""
# _extract_diff normalizes (strips trailing whitespace); the body is intact.
assert plan.diff == _BUMP_DIFF.strip()
assert "requests==2.19.0" in plan.diff and "requests==2.20.0" in plan.diff
# Hash matches the foundation content-hash CI re-computes over the SAME bytes
# CI uploads as candidate.diff (boundary #3).
assert plan.diff_hash == compute_content_hash(plan.diff.encode("utf-8"))
assert plan.manifest_path == "requirements.txt"
assert plan.declared_scope == ["requirements.txt"]
# Exact CI workflow_dispatch contract (ci/agent-team-apply-verify.yml inputs).
assert plan.dispatch is not None
inputs = plan.dispatch.as_inputs()
assert set(inputs) == {
"task_id",
"diff_artifact_name",
"expected_diff_hash",
"declared_scope",
}
assert inputs["task_id"] == "t-123"
assert inputs["diff_artifact_name"] == "candidate-diff-t-123"
assert inputs["expected_diff_hash"] == plan.diff_hash
# Single-manifest scope, newline-joined (the workflow splits on newlines).
assert inputs["declared_scope"] == "requirements.txt"
def test_dispatch_hash_binds_to_diff_for_ci_gate() -> None:
"""The dispatched expected_diff_hash equals sha256(candidate.diff) — the
binding ci_gate / the guard job verify before apply."""
plan = plan_fix(
_finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
)
assert plan.dispatch is not None
# What CI uploads as candidate.diff is plan.diff; its hash is the binding.
assert (
compute_content_hash(plan.diff.encode("utf-8"))
== plan.dispatch.expected_diff_hash
)
def test_spec_callable_raising_fails_safe() -> None:
"""A spec callable that raises fails SAFE to a no-op plan (no patch/dispatch)."""
def _boom(_f: Any) -> str:
raise RuntimeError("claude down")
plan = plan_fix(
_finding(), task_id="t-1", spec=_boom, build=_fake_build(_BUMP_DIFF)
)
assert plan.ok is False
assert "fix-spec generation failed" in plan.reason
assert plan.diff == ""
assert plan.dispatch is None
def test_empty_spec_is_tolerated_but_patch_still_required() -> None:
plan = plan_fix(
_finding(),
task_id="t-1",
spec=lambda _f: " ",
build=_fake_build(_BUMP_DIFF),
)
assert plan.ok is True
assert plan.spec == "" # empty spec normalized; patch carries the fix
# --------------------------------------------------------------------------- #
# Scope confinement / untrusted-patch hygiene
# --------------------------------------------------------------------------- #
def test_patch_touching_extra_file_is_rejected() -> None:
out_of_scope = _BUMP_DIFF + (
"diff --git a/agent_team/secret.py b/agent_team/secret.py\n"
"--- a/agent_team/secret.py\n"
"+++ b/agent_team/secret.py\n"
"@@ -1,1 +1,1 @@\n"
"-x = 1\n"
"+x = 2\n"
)
plan = plan_fix(
_finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(out_of_scope)
)
assert plan.ok is False
assert "escapes the declared single-manifest scope" in plan.reason
assert plan.dispatch is None
def test_patch_touching_trust_control_surface_is_rejected() -> None:
"""A diff that touches a denylisted path (a workflow) fails SAFE even though
it claims to be a dep bump."""
malicious = (
"diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n"
"--- a/.github/workflows/ci.yml\n"
"+++ b/.github/workflows/ci.yml\n"
"@@ -1,1 +1,1 @@\n"
"-on: push\n"
"+on: pull_request_target\n"
)
plan = plan_fix(
_finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(malicious)
)
assert plan.ok is False
assert "trust-control denylist" in plan.reason
assert plan.dispatch is None
def test_patch_renaming_into_denied_path_is_rejected() -> None:
rename_diff = (
"diff --git a/requirements.txt b/.github/workflows/evil.yml\n"
"rename from requirements.txt\n"
"rename to .github/workflows/evil.yml\n"
)
plan = plan_fix(
_finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(rename_diff)
)
assert plan.ok is False
assert plan.dispatch is None
# --------------------------------------------------------------------------- #
# INERT / opt-in: nothing dispatched by default
# --------------------------------------------------------------------------- #
def test_plan_fix_dispatches_nothing() -> None:
"""plan_fix only PLANS — it never calls a dispatcher (none is bindable here)."""
plan = plan_fix(
_finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
)
# The plan carries dispatch inputs but the act of planning fired nothing.
assert isinstance(plan.dispatch, FixDispatchInputs)
def test_dispatch_fix_without_dispatcher_raises_and_fires_nothing() -> None:
plan = plan_fix(
_finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
)
with pytest.raises(RuntimeError, match="no workflow dispatcher bound"):
dispatch_fix(plan, dispatcher=None)
def test_dispatch_fix_refuses_non_ok_plan() -> None:
bad = FixPlan.failed("nope", finding_id="x")
calls: list[Any] = []
def _disp(_inputs: FixDispatchInputs, _diff: str) -> str:
calls.append(_inputs)
return "ref"
with pytest.raises(ValueError, match="non-ok fix plan"):
dispatch_fix(bad, dispatcher=_disp)
assert calls == [] # the dispatcher was never called
def test_dispatch_fix_with_explicit_dispatcher_forwards_diff_and_inputs() -> None:
"""When (and only when) a dispatcher is explicitly bound does anything fire."""
plan = plan_fix(
_finding(), task_id="t-9", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
)
seen: dict[str, Any] = {}
def _disp(inputs: FixDispatchInputs, diff: str) -> str:
seen["inputs"] = inputs
seen["diff"] = diff
return "dispatch-ref-42"
ref = dispatch_fix(plan, dispatcher=_disp)
assert ref == "dispatch-ref-42"
assert seen["diff"] == plan.diff
assert seen["inputs"].expected_diff_hash == plan.diff_hash
# --------------------------------------------------------------------------- #
# Fail-safe on bad / ambiguous findings
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"mutation, why_substr",
[
({"check": "secrets"}, "not a dependency-cve"),
({"status": "unverified"}, "not confirmed"),
({"category": "injection"}, "unexpected finding category"),
],
)
def test_non_dependency_findings_are_not_fixable(
mutation: dict[str, Any], why_substr: str
) -> None:
fixable, why = is_fixable_dependency_finding(_finding(**mutation))
assert fixable is False
assert why_substr in why
@pytest.mark.parametrize(
"proof_mutation, why_substr",
[
({"package": ""}, "package is missing"),
({"version": ""}, "version is missing"),
({"fixed_version": ""}, "fixed_version is missing"),
({"fixed_version": ">=2.20.0"}, "ambiguous fixed_version"),
({"fixed_version": "2.20.0, 2.21.0"}, "ambiguous fixed_version"),
({"fixed_version": "2.19.0"}, "equals fixed_version"),
],
)
def test_bad_proof_findings_are_not_fixable(
proof_mutation: dict[str, Any], why_substr: str
) -> None:
f = _finding()
f["proof"].update(proof_mutation)
fixable, why = is_fixable_dependency_finding(f)
assert fixable is False
assert why_substr in why
def test_non_mapping_finding_is_not_fixable() -> None:
fixable, why = is_fixable_dependency_finding("not a dict")
assert fixable is False
assert "must be a mapping" in why
def test_plan_fix_on_bad_finding_yields_failed_plan_no_patch() -> None:
plan = plan_fix(
_finding(check="secrets"),
task_id="t-1",
spec=_fake_spec,
build=_fake_build(_BUMP_DIFF),
)
assert plan.ok is False
assert "not a fixable dependency-cve" in plan.reason
assert plan.diff == ""
assert plan.dispatch is None
def test_plan_fix_missing_task_id_fails_safe() -> None:
plan = plan_fix(
_finding(), task_id=" ", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
)
assert plan.ok is False
assert "task_id is missing" in plan.reason
def test_plan_fix_unparseable_diff_fails_safe() -> None:
plan = plan_fix(
_finding(),
task_id="t-1",
spec=_fake_spec,
build=_fake_build("I could not generate a diff, sorry."),
)
assert plan.ok is False
assert "not a usable unified diff" in plan.reason
assert plan.dispatch is None
def test_plan_fix_empty_build_output_fails_safe() -> None:
plan = plan_fix(
_finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(" ")
)
assert plan.ok is False
assert "empty output" in plan.reason
def test_plan_fix_build_raises_fails_safe() -> None:
def _boom(_instruction: str) -> str:
raise RuntimeError("orchestrator down")
plan = plan_fix(_finding(), task_id="t-1", spec=_fake_spec, build=_boom)
assert plan.ok is False
assert "patch generation failed" in plan.reason
# --------------------------------------------------------------------------- #
# describe_plan (dry-run rendering)
# --------------------------------------------------------------------------- #
def test_describe_plan_ok_shows_spec_patch_and_dispatch() -> None:
plan = plan_fix(
_finding(), task_id="t-7", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
)
text = describe_plan(plan)
assert "fix spec (Claude)" in text
assert "candidate patch (DeepSeek" in text
assert "workflow_dispatch inputs" in text
assert "t-7" in text
assert plan.diff_hash in text
assert "DRY-RUN: nothing dispatched" in text
def test_describe_plan_failed_shows_failsafe_and_no_dispatch() -> None:
plan = FixPlan.failed("bad finding", finding_id="abc")
text = describe_plan(plan)
assert "FIX NOT PLANNED" in text
assert "dispatch: NONE" in text
# --------------------------------------------------------------------------- #
# No mutation surface (inert by construction)
# --------------------------------------------------------------------------- #
def test_module_has_no_patch_apply_or_git_or_write_surface() -> None:
"""No git/patch/subprocess/fs-write CALLS in executable code.
Docstrings legitimately MENTION these terms (to state what the fixer does
NOT do), so we strip docstrings + comments via the tokenizer and inspect only
executable tokens. By construction the fixer cannot mutate the tree or push.
"""
import io
import tokenize
src = inspect.getsource(fixer)
code_tokens: list[str] = []
for tok in tokenize.generate_tokens(io.StringIO(src).readline):
if tok.type in (tokenize.COMMENT, tokenize.STRING):
continue
code_tokens.append(tok.string)
code = " ".join(code_tokens)
# The fixer must not call out to git/patch/subprocess or write the filesystem.
for forbidden in ("subprocess", "Popen", "write_text", "system", "popen"):
assert forbidden not in code, f"fixer must not reference {forbidden!r} in code"
# No bare write-mode open(...) or git invocation in executable code.
assert "git" not in code, "fixer must not invoke git in code"