Bind audit seals to exact source jobs

CHANGELOG.md

@@ -7,6 +7,11 @@ later entries are regular releases.
 
 ## Unreleased
 
+- Local audit publication now binds each reviewer seal to the exact Actions
+  job, matrix lane, run, and first attempt that produced it. Independent Codex
+  and Claude lanes in one run no longer make publication ambiguous, while a
+  missing, duplicate, wrong-job, or wrong-attempt seal still fails closed
+  (#1032).
 - Hosted-agent installation guidance now includes a Python-based uv bootstrap
   that does not pipe a remote script into a shell. Remote-only orchestrators
   get an explicit packaged-starter doctor command, doctor JSON is identified as

docs/local-audit-runner.md

@@ -165,7 +165,7 @@ metadata only while its original head and freshness checks still hold.
 
 Only canonical metadata leaves the machine: schema, numeric repository ID, PR
 number, reviewer lane, PASS/BLOCKED, full start/end head SHAs, artifact creation
-time, and the originating audit run ID/attempt. The repository name, comment prose, findings, code, prompts, transcript,
+time, and the originating audit run, attempt, and job IDs. The repository name, comment prose, findings, code, prompts, transcript,
 paths and provider output stay local. The SHA-256 digest covers those exact
 canonical metadata bytes. The publisher accepts only equal full start/end SHAs,
 an open PR at that SHA, and artifacts no more than 24 hours old. UNKNOWN, STALE,
@@ -185,12 +185,15 @@ attempt is refused. Keep receipts and reservations for at least the 24-hour
 artifact lifetime. The global publication concurrency group serializes claims;
 GitHub may cancel an older queued dispatch, which requires inspecting its result.
 
-The source job stages the metadata, independently validates it, and completes a
+The source job stages the metadata with its exact GitHub Actions job ID,
+independently validates it, and completes a
 `Code Mower reviewer seal <digest>` step before dispatch. The publisher verifies
-that immutable Actions step record in the matching `audit (claude|codex)` job,
+that immutable Actions step record in that exact `audit (claude|codex)` job,
 source run ID/attempt, trusted `local-cli-audit.yml` `repository_dispatch` event,
 same repository and default branch. The sealed digest binds the PR, lane and
-full start/end head. The small `local-audit-request.yml` trigger requests the
+full start/end head plus the source job identity. Another audit lane in the
+same matrix run cannot satisfy or make ambiguous that binding; a duplicate
+seal within the requested lane is refused. The small `local-audit-request.yml` trigger requests the
 review; the source workflow validates the request against the live PR before
 starting a provider. Both source and publisher use `repository_dispatch`, which
 always executes default-branch code: a builder cannot counterfeit a source job

src/code_mower/audit_publication.py

@@ -41,6 +41,7 @@
         "created_at",
         "source_run_id",
         "source_run_attempt",
+        "source_job_id",
     }
 )
 MARKER = "<!-- CODE_MOWER_AUDIT_PUBLICATION: "
@@ -104,6 +105,7 @@ class Refused(ValueError):
     "reservation lacks publishing run": "RESERVATION_LACKS_PUBLISHING_RUN",
     "run lookup mismatch": "RUN_LOOKUP_MISMATCH",
     "source reviewer seal missing or ambiguous": "SOURCE_REVIEWER_SEAL_MISSING_OR_AMBIGUOUS",
+    "source reviewer job missing or ambiguous": "SOURCE_REVIEWER_JOB_MISSING_OR_AMBIGUOUS",
     "unexpected publishing identity": "UNEXPECTED_PUBLISHING_IDENTITY",
     "unsupported publication command": "UNSUPPORTED_PUBLICATION_COMMAND",
     "unsupported publication schema": "UNSUPPORTED_PUBLICATION_SCHEMA",
@@ -187,6 +189,7 @@ def validate(text, expected_digest, *, now=None):
     require(positive(value["repository_id"]) and positive(value["pr_number"]), "invalid target")
     require(
         positive(value["source_run_id"])
+        and positive(value["source_job_id"])
         and value["source_run_attempt"] == 1
         and type(value["source_run_attempt"]) is int,
         "invalid source run/attempt",
@@ -235,23 +238,51 @@ def verify_source(io, value, repository):
         and source.get("head_repository", {}).get("id") == repository["id"],
         "untrusted source audit run",
     )
-    jobs = io.pages(f"/actions/runs/{source['id']}/attempts/1/jobs", "jobs")
-    matching = [
+    jobs = io.pages(
+        f"/actions/runs/{source['id']}/attempts/{value['source_run_attempt']}/jobs", "jobs"
+    )
+    lane_jobs = [
         job
         for job in jobs
         if job.get("run_id") == source["id"]
+        and job.get("run_attempt") == value["source_run_attempt"]
         and job.get("name") == f"audit ({value['lane']})"
-        and any(
+    ]
+    bound = [job for job in lane_jobs if job.get("id") == value["source_job_id"]]
+    sealed = [
+        job
+        for job in lane_jobs
+        if any(
             step.get("name") == seal_name(value)
             and step.get("status") == "completed"
             and step.get("conclusion") == "success"
             for step in job.get("steps", [])
         )
     ]
-    require(len(matching) == 1, "source reviewer seal missing or ambiguous")
+    require(
+        len(bound) == 1 and len(sealed) == 1 and sealed[0].get("id") == value["source_job_id"],
+        "source reviewer seal missing or ambiguous",
+    )
     return source
 
 
+def current_source_job(io, *, run_id, run_attempt, lane, runner_name):
+    """Resolve this running matrix job before its artifact is sealed."""
+    jobs = io.pages(f"/actions/runs/{run_id}/attempts/{run_attempt}/jobs", "jobs")
+    matching = [
+        job
+        for job in jobs
+        if positive(job.get("id"))
+        and job.get("run_id") == run_id
+        and job.get("run_attempt") == run_attempt
+        and job.get("name") == f"audit ({lane})"
+        and job.get("runner_name") == runner_name
+        and job.get("status") == "in_progress"
+    ]
+    require(len(matching) == 1, "source reviewer job missing or ambiguous")
+    return matching[0]
+
+
 def receipts(run):
     jobs = run.get("publication_jobs")
     require(isinstance(jobs, list), "missing publication receipt")
@@ -571,6 +602,7 @@ def project_local(artifact, repository, *, lane, now):
         "created_at": created_at,
         "source_run_id": artifact.get("source_run_id"),
         "source_run_attempt": artifact.get("source_run_attempt"),
+        "source_job_id": artifact.get("source_job_id"),
     }
     text = canonical(value)
     validate(text, digest(text), now=now)
@@ -622,7 +654,20 @@ def stage(path, *, token, lane, env=None, io=None):
         and env.get("PR_HEAD_SHA") == artifact.get("head_sha_start"),
         "untrusted staging environment",
     )
-    artifact.update(source_run_id=int(env["GITHUB_RUN_ID"]), source_run_attempt=1)
+    run_id = int(env["GITHUB_RUN_ID"])
+    run_attempt = int(env["GITHUB_RUN_ATTEMPT"])
+    source_job = current_source_job(
+        io,
+        run_id=run_id,
+        run_attempt=run_attempt,
+        lane=lane,
+        runner_name=env.get("RUNNER_NAME"),
+    )
+    artifact.update(
+        source_run_id=run_id,
+        source_run_attempt=run_attempt,
+        source_job_id=source_job["id"],
+    )
     value = project_local(artifact, repository, lane=lane, now=int(time.time()))
     current_pr(io, value)
     Path(path).write_text(json.dumps(artifact, indent=2, sort_keys=True) + "\n")

tests/test_audit_publication.py

@@ -46,6 +46,7 @@ def artifact(lane="claude", verdict="PASS", **changes):
             created_at=NOW,
             source_run_id=700,
             source_run_attempt=1,
+            source_job_id=701,
         )
         | changes
     )
@@ -98,6 +99,7 @@ def environment():
         GITHUB_SHA=SOURCE,
         GITHUB_RUN_ID="800",
         GITHUB_RUN_ATTEMPT="1",
+        RUNNER_NAME="code-mower-audit-mac",
     )
 
 
@@ -126,8 +128,12 @@ def source_for(value):
         pull_requests=[dict(number=42, base=dict(ref="main", repo=REPOSITORY))],
         jobs=[
             dict(
+                id=value["source_job_id"],
                 run_id=value["source_run_id"],
+                run_attempt=value["source_run_attempt"],
                 name=f"audit ({value['lane']})",
+                runner_name="code-mower-audit-mac",
+                status="in_progress",
                 steps=[dict(name=pub.seal_name(value), status="completed", conclusion="success")],
             )
         ],
@@ -493,7 +499,9 @@ def test_fabricated_or_mismatched_source_proof_never_publishes(self):
             lambda s: s.update(head_repository={"id": 999}),
             lambda s: s.update(event="pull_request_target"),
             lambda s: s["jobs"][0].update(name="audit (codex)"),
+            lambda s: s["jobs"][0].update(id=702),
             lambda s: s["jobs"][0].update(run_id=701),
+            lambda s: s["jobs"][0].update(run_attempt=2),
             lambda s: s["jobs"][0].update(steps=[]),
             lambda s: s["jobs"][0]["steps"][0].update(name="Code Mower reviewer seal " + "0" * 64),
             lambda s: s["jobs"][0]["steps"][0].update(status="in_progress"),
@@ -512,6 +520,75 @@ def test_fabricated_or_mismatched_source_proof_never_publishes(self):
             pub.publish(event_for(artifact(verdict="PASS")), environment(), api, now=NOW)
         self.assertEqual(api.writes, [])
 
+    def test_two_lanes_bind_only_their_exact_source_jobs(self):
+        claude = artifact(source_job_id=701)
+        codex = artifact("codex", source_job_id=702)
+        jobs = source_for(claude)["jobs"] + source_for(codex)["jobs"]
+        for value in (claude, codex):
+            api = MemoryGitHub(value)
+            api.source["jobs"] = deepcopy(jobs)
+            pub.publish(event_for(value), environment(), api, now=NOW)
+            self.assertEqual([method for method, _, _ in api.writes], ["POST", "PATCH"])
+
+    def test_source_job_resolution_requires_one_running_lane_on_this_runner(self):
+        value = artifact()
+        api = MemoryGitHub(value)
+        self.assertEqual(
+            pub.current_source_job(
+                api,
+                run_id=value["source_run_id"],
+                run_attempt=value["source_run_attempt"],
+                lane=value["lane"],
+                runner_name="code-mower-audit-mac",
+            )["id"],
+            value["source_job_id"],
+        )
+        other_lane = source_for(artifact("codex", source_job_id=702))["jobs"][0]
+        api.source["jobs"].append(other_lane)
+        self.assertEqual(
+            pub.current_source_job(
+                api,
+                run_id=value["source_run_id"],
+                run_attempt=value["source_run_attempt"],
+                lane=value["lane"],
+                runner_name="code-mower-audit-mac",
+            )["id"],
+            value["source_job_id"],
+        )
+        for change in (
+            {"status": "completed"},
+            {"runner_name": "other-runner"},
+            {"run_attempt": 2},
+            {"duplicate": True},
+        ):
+            failed = MemoryGitHub(value)
+            if change.get("duplicate"):
+                failed.source["jobs"].append(failed.source["jobs"][0] | {"id": 703})
+            else:
+                failed.source["jobs"][0].update(change)
+            with self.assertRaises(pub.Refused):
+                pub.current_source_job(
+                    failed,
+                    run_id=value["source_run_id"],
+                    run_attempt=value["source_run_attempt"],
+                    lane=value["lane"],
+                    runner_name="code-mower-audit-mac",
+                )
+
+    def test_other_lane_cannot_substitute_for_bound_source_job(self):
+        value = artifact()
+        api = MemoryGitHub(value)
+        api.source["jobs"][0]["steps"] = []
+        forged = deepcopy(api.source["jobs"][0])
+        forged.update(id=702, name="audit (codex)")
+        forged["steps"] = [
+            dict(name=pub.seal_name(value), status="completed", conclusion="success")
+        ]
+        api.source["jobs"].append(forged)
+        with self.assertRaises(pub.Refused):
+            pub.publish(event_for(value), environment(), api, now=NOW)
+        self.assertEqual(api.writes, [])
+
     def test_canonical_exact_schema_digest_and_field_types(self):
         value = artifact()
         for lane in ("claude", "codex"):
@@ -537,6 +614,8 @@ def test_canonical_exact_schema_digest_and_field_types(self):
             dict(created_at=True),
             dict(source_run_id=True),
             dict(source_run_id=0),
+            dict(source_job_id=True),
+            dict(source_job_id=0),
             dict(source_run_attempt=2),
             dict(source_run_attempt=True),
             dict(source="private source"),
@@ -927,6 +1006,7 @@ def test_stage_saves_source_binding_without_dispatch_and_can_resume(self):
         local = self.local()
         local.pop("source_run_id")
         local.pop("source_run_attempt")
+        local.pop("source_job_id")
         api = MemoryGitHub(artifact(created_at=int(time.time())))
         with tempfile.TemporaryDirectory() as tmp:
             path, staged = Path(tmp) / "local.json", Path(tmp) / "metadata.json"
@@ -935,6 +1015,7 @@ def test_stage_saves_source_binding_without_dispatch_and_can_resume(self):
                 GITHUB_EVENT_NAME="repository_dispatch",
                 GITHUB_RUN_ID="700",
                 GITHUB_RUN_ATTEMPT="1",
+                RUNNER_NAME="code-mower-audit-mac",
                 PR_HEAD_SHA=HEAD,
                 GITHUB_WORKFLOW_REF=f"{REPO}/{pub.SOURCE_WORKFLOW}@refs/heads/main",
                 CODE_MOWER_LOCAL_AUDIT_LANE="claude",
@@ -943,6 +1024,7 @@ def test_stage_saves_source_binding_without_dispatch_and_can_resume(self):
             pub.stage(path, token="fixture", lane="claude", env=env, io=api)
             value = pub.validate(staged.read_text(), pub.digest(staged.read_text()))
             self.assertEqual(value["source_run_id"], 700)
+            self.assertEqual(value["source_job_id"], 701)
             self.assertNotIn("PRIVATE_SOURCE", staged.read_text())
             self.assertEqual(api.writes, [])
             api.source = source_for(value)
@@ -988,6 +1070,7 @@ def local(self, lane="claude", **changes):
                 posted_comment_url=None,
                 source_run_id=700,
                 source_run_attempt=1,
+                source_job_id=701,
             )
             | changes
         )