Merge issue-143-remediation: Fix scheduled aggregate runner drift validation (#143)

Author: alexeygrigorev

.github/workflows/scheduled-full-regression.yml

@@ -517,6 +517,7 @@ jobs:
             --report .tmp/verification-report.json \
             --state .tmp/repository-state.json \
             --run-id "$GITHUB_RUN_ID" --run-attempt "$GITHUB_RUN_ATTEMPT" \
+            --allow-hosted-runner-drift \
             --output .tmp/scheduled-state.json
       - name: Preserve the complete aggregate scheduled evidence
         uses: actions/upload-artifact@v4

ci/tests/test_workflows.py

@@ -410,6 +410,13 @@ def test_scheduled_evidence_capture_and_record_symmetrically_allow_runner_drift(
         assert f"--component {name}" in verification_invocations(jobs[name], "environment")[0]
 
 
+def test_scheduled_aggregate_state_explicitly_allows_hosted_runner_drift() -> None:
+    jobs = workflow("scheduled-full-regression.yml")["jobs"]
+    invocations = verification_invocations(jobs["full-regression"], "scheduled-state")
+    assert len(invocations) == 1
+    assert invocations[0].split().count("--allow-hosted-runner-drift") == 1
+
+
 def test_scheduled_playwright_executes_and_records_the_planner_core_command() -> None:
     playwright = workflow("scheduled-full-regression.yml")["jobs"]["playwright"]
     planner_command = json.loads((ROOT / "ci" / "ownership.json").read_text(encoding="utf-8"))[

ci/verification.py

@@ -1027,19 +1027,98 @@ def repository_state(
 ) -> dict[str, Any]:
     graph = dict(graph) if graph is not None else load_graph()
     _manifest, source_tree = git_manifest(repository, revision)
-    environment = environment_fingerprint()
+    environment = validate_environment_fingerprint(environment_fingerprint())
+    component_environment = {
+        component: validate_environment_fingerprint(component_environment_fingerprint(component))
+        for component in PLAN_COMPONENTS
+    }
     payload = {
+        "component_environment": component_environment,
         "component_environment_sha256": {
-            component: component_environment_fingerprint(component)["sha256"]
-            for component in PLAN_COMPONENTS
+            component: fingerprint["sha256"]
+            for component, fingerprint in component_environment.items()
         },
+        "environment": environment,
         "environment_sha256": environment["sha256"],
         "graph_sha256": graph_digest(graph),
         "policy_version": graph["policy_version"],
         "source_manifest_sha256": source_tree["manifest_sha256"],
         "tree_oid": source_tree["tree_oid"],
     }
-    return payload | {"verification_state_sha256": sha256_json(payload)}
+    return _validate_repository_state(payload | {"verification_state_sha256": sha256_json(payload)})
+
+
+def _validate_repository_state(value: object) -> dict[str, Any]:
+    """Validate the aggregate state retained for scheduled coverage decisions."""
+    expected = {
+        "component_environment",
+        "component_environment_sha256",
+        "environment",
+        "environment_sha256",
+        "graph_sha256",
+        "policy_version",
+        "source_manifest_sha256",
+        "tree_oid",
+        "verification_state_sha256",
+    }
+    if not isinstance(value, dict) or set(value) != expected:
+        raise VerificationError("repository state has an invalid shape")
+    try:
+        environment = validate_environment_fingerprint(value["environment"])
+        component_environment = value["component_environment"]
+        if not isinstance(component_environment, dict) or set(component_environment) != set(
+            PLAN_COMPONENTS
+        ):
+            raise VerificationError("repository state component environments are incomplete")
+        validated_components = {
+            component: validate_environment_fingerprint(component_environment[component])
+            for component in PLAN_COMPONENTS
+        }
+    except (TypeError, ValueError) as exc:
+        raise VerificationError("repository state environment fingerprint is invalid") from exc
+
+    if value["environment_sha256"] != environment["sha256"]:
+        raise VerificationError("repository state global environment digest does not match")
+    component_digests = value["component_environment_sha256"]
+    if (
+        not isinstance(component_digests, dict)
+        or set(component_digests) != set(PLAN_COMPONENTS)
+        or component_digests
+        != {
+            component: fingerprint["sha256"]
+            for component, fingerprint in validated_components.items()
+        }
+    ):
+        raise VerificationError("repository state component environment digests do not match")
+    for field in ("graph_sha256", "source_manifest_sha256", "verification_state_sha256"):
+        if not isinstance(value[field], str) or not SHA256_RE.fullmatch(value[field]):
+            raise VerificationError("repository state digest is invalid")
+    if not isinstance(value["tree_oid"], str) or not SHA_RE.fullmatch(value["tree_oid"]):
+        raise VerificationError("repository state tree identity is invalid")
+    if not isinstance(value["policy_version"], int) or isinstance(value["policy_version"], bool):
+        raise VerificationError("repository state policy version is invalid")
+
+    identity = dict(value)
+    digest = identity.pop("verification_state_sha256")
+    if digest != sha256_json(identity):
+        raise VerificationError("repository state digest does not match its contents")
+    return value
+
+
+def _scheduled_environment_matches(
+    actual: Mapping[str, Any],
+    planned: Mapping[str, Any],
+    *,
+    allow_hosted_runner_drift: bool,
+) -> bool:
+    """Apply exact state identity by default; opt into hosted-runner drift explicitly."""
+    if not allow_hosted_runner_drift:
+        return actual == planned
+    return environment_matches_plan(
+        actual,
+        planned,
+        allow_hosted_runner_drift=True,
+    )
 
 
 def build_scheduled_state_envelope(
@@ -1050,22 +1129,35 @@ def build_scheduled_state_envelope(
     run_id: int,
     run_attempt: int,
     workflow: str = ".github/workflows/scheduled-full-regression.yml",
+    allow_hosted_runner_drift: bool = False,
 ) -> dict[str, Any]:
     plan = validate_plan(dict(plan))
     report = validate_report(dict(report))
+    state = _validate_repository_state(dict(state))
     if report["verdict"] != "success" or report["plan_sha256"] != sha256_json(plan):
         raise VerificationError("scheduled state requires a successful plan-bound report")
+    planned_component_environment = {
+        component: item["environment"] for component, item in plan["components"].items()
+    }
     if (
-        state.get("verification_state_sha256") is None
-        or state.get("source_manifest_sha256") != plan["source_tree"]["manifest_sha256"]
-        or state.get("graph_sha256") != plan["graph_sha256"]
-        or state.get("policy_version") != plan["policy_version"]
-        or state.get("environment_sha256") != plan["environment"]["sha256"]
-        or state.get("component_environment_sha256")
-        != {
-            component: item["environment"]["sha256"]
-            for component, item in plan["components"].items()
-        }
+        state["source_manifest_sha256"] != plan["source_tree"]["manifest_sha256"]
+        or state["tree_oid"] != plan["source_tree"]["tree_oid"]
+        or state["graph_sha256"] != plan["graph_sha256"]
+        or state["policy_version"] != plan["policy_version"]
+        or not _scheduled_environment_matches(
+            state["environment"],
+            plan["environment"],
+            allow_hosted_runner_drift=allow_hosted_runner_drift,
+        )
+        or set(state["component_environment"]) != set(planned_component_environment)
+        or any(
+            not _scheduled_environment_matches(
+                state["component_environment"][component],
+                planned_environment,
+                allow_hosted_runner_drift=allow_hosted_runner_drift,
+            )
+            for component, planned_environment in planned_component_environment.items()
+        )
     ):
         raise VerificationError("scheduled repository state does not match the plan")
     evidence_ids = sorted(
@@ -1690,6 +1782,7 @@ def main() -> None:
     scheduled_state_parser.add_argument("--state", required=True)
     scheduled_state_parser.add_argument("--run-id", type=int, required=True)
     scheduled_state_parser.add_argument("--run-attempt", type=int, required=True)
+    scheduled_state_parser.add_argument("--allow-hosted-runner-drift", action="store_true")
     scheduled_state_parser.add_argument("--output", required=True)
 
     args = parser.parse_args()
@@ -1841,6 +1934,7 @@ def main() -> None:
             state=state,
             run_id=args.run_id,
             run_attempt=args.run_attempt,
+            allow_hosted_runner_drift=args.allow_hosted_runner_drift,
         ),
         args.output,
     )

tests_ci/test_verification.py

@@ -8,10 +8,12 @@
 
 from ci.content_invariants import build_invariant_artifact
 from ci.evidence import build_envelope
+from ci.ownership import sha256_json
 from ci.selection import ChangeRecord, classify_records
 from ci.verification import (
     VerificationError,
     build_plan,
+    build_scheduled_state_envelope,
     create_report,
     materialize_reused_evidence,
     read_worktree_change_records,
@@ -40,6 +42,45 @@ def make_plan(tmp_path: Path, changed: dict[str, str]):
     )
 
 
+def scheduled_state_fixture(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    *,
+    planned_image: tuple[str, str] = ("ubuntu24", "20260801.1"),
+    actual_image: tuple[str, str] | None = None,
+):
+    monkeypatch.setenv("ImageOS", planned_image[0])
+    monkeypatch.setenv("ImageVersion", planned_image[1])
+    repository, plan = make_plan(tmp_path, {"api/service.py": "changed\n"})
+    evidence = tmp_path / "evidence"
+    evidence.mkdir()
+    origin = {"issue": 143, "kind": "local", "producer_role": "engineer", "worktree": "143"}
+    for component, item in plan["components"].items():
+        if item["disposition"] != "rerun":
+            continue
+        records, output = component_output(evidence, plan, component)
+        envelope = build_envelope(
+            plan=plan,
+            component=component,
+            result="success",
+            origin=origin,
+            command=item["command"],
+            execution_environment=item["environment"],
+            artifacts=records,
+            machine_output=output,
+            completed_at=NOW,
+        )
+        (evidence / f"{component}-evidence.json").write_text(
+            __import__("json").dumps(envelope, sort_keys=True), encoding="utf-8"
+        )
+    report = create_report(plan=plan, result_directory=evidence, phase="engineer")
+    if actual_image is not None:
+        monkeypatch.setenv("ImageOS", actual_image[0])
+        monkeypatch.setenv("ImageVersion", actual_image[1])
+    state = repository_state(repository, plan["head"])
+    return plan, report, state
+
+
 def test_single_app_plan_reruns_affected_closure_and_preserves_baseline_without_evidence(
     tmp_path: Path,
 ) -> None:
@@ -390,6 +431,120 @@ def test_repository_state_ignores_commit_metadata_but_changes_with_tree(tmp_path
     assert base_state["verification_state_sha256"] != head_state["verification_state_sha256"]
 
 
+def test_scheduled_state_accepts_exact_aggregate_environment_without_opt_in(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    plan, report, state = scheduled_state_fixture(tmp_path, monkeypatch)
+
+    assert set(state["component_environment"]) == set(plan["components"])
+    assert state["environment_sha256"] == state["environment"]["sha256"]
+    assert state["component_environment_sha256"] == {
+        component: fingerprint["sha256"]
+        for component, fingerprint in state["component_environment"].items()
+    }
+
+    envelope = build_scheduled_state_envelope(
+        plan=plan,
+        report=report,
+        state=state,
+        run_id=143,
+        run_attempt=1,
+    )
+
+    assert envelope["verification_state_sha256"] == state["verification_state_sha256"]
+
+
+def test_scheduled_state_requires_explicit_opt_in_for_same_family_drift(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    plan, report, state = scheduled_state_fixture(
+        tmp_path,
+        monkeypatch,
+        actual_image=("ubuntu24", "20260808.1"),
+    )
+
+    with pytest.raises(VerificationError, match="scheduled repository state"):
+        build_scheduled_state_envelope(
+            plan=plan,
+            report=report,
+            state=state,
+            run_id=143,
+            run_attempt=1,
+        )
+
+    envelope = build_scheduled_state_envelope(
+        plan=plan,
+        report=report,
+        state=state,
+        run_id=143,
+        run_attempt=1,
+        allow_hosted_runner_drift=True,
+    )
+    assert envelope["verification_state_sha256"] == state["verification_state_sha256"]
+
+
+def test_scheduled_state_rejects_different_runner_family_even_with_opt_in(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    plan, report, state = scheduled_state_fixture(
+        tmp_path,
+        monkeypatch,
+        actual_image=("windows2025", "20260808.1"),
+    )
+
+    with pytest.raises(VerificationError, match="scheduled repository state"):
+        build_scheduled_state_envelope(
+            plan=plan,
+            report=report,
+            state=state,
+            run_id=143,
+            run_attempt=1,
+            allow_hosted_runner_drift=True,
+        )
+
+
+@pytest.mark.parametrize("field", ["environment", "component_environment"])
+def test_scheduled_state_rejects_tampered_raw_environment_fingerprints(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, field: str
+) -> None:
+    plan, report, state = scheduled_state_fixture(tmp_path, monkeypatch)
+    tampered = deepcopy(state)
+    if field == "environment":
+        tampered["environment"]["browser"] = "firefox"
+    else:
+        tampered["component_environment"]["django"]["browser"] = "firefox"
+
+    with pytest.raises(VerificationError, match="repository state"):
+        build_scheduled_state_envelope(
+            plan=plan,
+            report=report,
+            state=tampered,
+            run_id=143,
+            run_attempt=1,
+            allow_hosted_runner_drift=True,
+        )
+
+
+def test_scheduled_state_rejects_recomputed_tree_identity_tamper(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    plan, report, state = scheduled_state_fixture(tmp_path, monkeypatch)
+    tampered = deepcopy(state)
+    tampered["tree_oid"] = "f" * len(state["tree_oid"])
+    identity = dict(tampered)
+    identity.pop("verification_state_sha256")
+    tampered["verification_state_sha256"] = sha256_json(identity)
+
+    with pytest.raises(VerificationError, match="scheduled repository state"):
+        build_scheduled_state_envelope(
+            plan=plan,
+            report=report,
+            state=tampered,
+            run_id=143,
+            run_attempt=1,
+        )
+
+
 def test_worktree_plan_includes_dirty_tracked_and_untracked_candidate_files(
     tmp_path: Path,
 ) -> None: