diff --git a/.claude/skills/issue-contract-review/scripts/contract_readiness_check.py b/.claude/skills/issue-contract-review/scripts/contract_readiness_check.py index 3a6691286..4a70e3ab7 100644 --- a/.claude/skills/issue-contract-review/scripts/contract_readiness_check.py +++ b/.claude/skills/issue-contract-review/scripts/contract_readiness_check.py @@ -1117,6 +1117,114 @@ def run_baseline_vc_preflight( } +# Issue #2897: operator-only VC timeout identity pass-through (readiness +# producer side). Everything below only COPIES identity / applied-budget +# information a `baseline_vc_preflight/v1` result item already carries; it +# never estimates, recomputes, or back-derives (e.g. from `duration_ms`) a +# budget, and never re-reads the history store. The same bounded vocabulary +# is re-validated independently by the review-merge consumer +# (`check_issue_contract.py::build_timeout_diagnostics()`), because the +# readiness result crosses a process / file boundary between the two. +_TIMEOUT_PROVENANCE_SOURCES = frozenset( + {"explicit_override", "static_policy", "static_fallback", "history_estimate"} +) +_SHA256_PREFIXED_RE = re.compile(r"^sha256:[0-9a-f]{64}$") +_ESTIMATOR_VERSION_RE = re.compile(r"^[A-Za-z0-9._-]{1,32}$") +_MAX_PROVENANCE_SECONDS = 86_400 + + +def _is_plain_int(value: Any) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def _bounded_sha256(value: Any) -> Optional[str]: + """Return `value` only if it is a `sha256:<64 lowercase hex>` string.""" + if isinstance(value, str) and _SHA256_PREFIXED_RE.match(value): + return value + return None + + +def _bounded_timeout_provenance(raw: Any) -> Optional[dict]: + """Allowlist-copy an existing result item's `timeout_provenance`. + + Returns `None` (never a partially trusted dict) unless all five known + fields are present with their bounded types / enumerated values. + """ + if not isinstance(raw, dict): + return None + timeout_seconds = raw.get("timeout_seconds") + cleanup_tail_seconds = raw.get("cleanup_tail_seconds") + source = raw.get("source") + estimator_version = raw.get("estimator_version") + estimator_input_digest = _bounded_sha256(raw.get("estimator_input_digest")) + if not (_is_plain_int(timeout_seconds) and 0 < timeout_seconds <= _MAX_PROVENANCE_SECONDS): + return None + if not (_is_plain_int(cleanup_tail_seconds) and 0 <= cleanup_tail_seconds <= _MAX_PROVENANCE_SECONDS): + return None + # Type first: an unhashable (list / dict) value must degrade to `None`, + # not raise `TypeError` from the frozenset membership test. + if not isinstance(source, str) or source not in _TIMEOUT_PROVENANCE_SOURCES: + return None + if not (isinstance(estimator_version, str) and _ESTIMATOR_VERSION_RE.match(estimator_version)): + return None + if estimator_input_digest is None: + return None + return { + "timeout_seconds": timeout_seconds, + "cleanup_tail_seconds": cleanup_tail_seconds, + "source": source, + "estimator_version": estimator_version, + "estimator_input_digest": estimator_input_digest, + } + + +def extract_canonical_plan_binding(preflight_result: Optional[dict]) -> dict: + """Read the canonical VC plan digest and the pre-filter `results` count + from an existing `baseline_vc_preflight/v1` payload (Issue #2897 In Scope + (g)). + + The digest source is exclusively the payload's own + `diagnostic_report.canonical_plan_digest`. An early-return / static path + reports `diagnostic_report.status == "not_computed"`; the digest is then + `None` and is never filled in from the body or the current history. + """ + results = (preflight_result or {}).get("results") + results_count = len(results) if isinstance(results, list) else 0 + report = (preflight_result or {}).get("diagnostic_report") + digest: Optional[str] = None + if isinstance(report, dict) and report.get("status") != "not_computed": + digest = _bounded_sha256(report.get("canonical_plan_digest")) + return {"canonical_plan_digest": digest, "results_count": results_count} + + +def _result_item_occurrence_identity(index: int, r: dict, plan_digest: Optional[str]) -> dict: + """Allowlisted identity / applied-budget fields for one result item.""" + runner = r.get("runner") + if runner == "exec": + execution_source = "executed" + elif runner == "dedup_replay": + execution_source = "dedup_replay" + else: + execution_source = "unknown" + dedup = r.get("dedup") + dedup_source_index = dedup.get("source_result_index") if isinstance(dedup, dict) else None + return { + # Canonical occurrence index: the position of this item in the + # UNFILTERED `results` array (primary identifier). + "occurrence_index": index, + # `line` is relative to its fenced block, not to the whole body; it + # is an auxiliary display value only. + "line_coordinate": "block_relative", + "execution_key_hash": _bounded_sha256(r.get("execution_key_hash")), + "execution_source": execution_source, + "dedup_source_result_index": ( + dedup_source_index if _is_plain_int(dedup_source_index) else None + ), + "timeout_provenance": _bounded_timeout_provenance(r.get("timeout_provenance")), + "canonical_plan_digest": plan_digest, + } + + def map_preflight_result_to_errors( preflight_result: dict, ) -> tuple[list[dict], str]: @@ -1238,7 +1346,13 @@ def map_preflight_result_to_errors( aggregate = _raise_status(aggregate, readiness_status) return errors, aggregate - for r in preflight_result.get("results", []): + # Issue #2897: canonical plan digest, read once from the existing + # preflight payload (never recomputed from the body / current history). + _plan_digest = extract_canonical_plan_binding(preflight_result)["canonical_plan_digest"] + + # `enumerate()` runs over the UNFILTERED results array, so `occurrence_index` + # below is the canonical result order, not a position among emitted errors. + for _occurrence_index, r in enumerate(preflight_result.get("results", [])): classification = r.get("classification", "") category = r.get("category", "") decision = r.get("decision", "go") @@ -1319,6 +1433,11 @@ def map_preflight_result_to_errors( "repair": r.get("repair"), "annotations": r.get("annotations"), "runner_env_delta": r.get("runner_env_delta", {}), + # Issue #2897: allowlisted identity / applied-budget + # pass-through (see `_result_item_occurrence_identity`). + **_result_item_occurrence_identity( + _occurrence_index, r, _plan_digest + ), }, } ) @@ -2211,7 +2330,7 @@ def build_result( fix_hint = first_error.get("fix_hint") minimal_context = first_error.get("minimal_context", []) - return { + result: dict = { "schema": "ISSUE_CONTRACT_READINESS_RESULT_V1", "status": overall_status, "body_sha256": body_sha256, @@ -2220,6 +2339,15 @@ def build_result( "minimal_context": minimal_context, "fix_hint": fix_hint, } + if preflight_result is not None: + # Issue #2897 In Scope (g): carry the canonical VC plan digest and + # the pre-filter `results` count to the top level so the review merge + # can bind each timeout occurrence by comparing values inside this + # one readiness result (no recomputation downstream). `None` digest + # means the preflight payload reported `diagnostic_report: + # not_computed`. + result.update(extract_canonical_plan_binding(preflight_result)) + return result # --------------------------------------------------------------------------- diff --git a/.claude/skills/issue-contract-review/tests/test_readiness_timeout_provenance_passthrough.py b/.claude/skills/issue-contract-review/tests/test_readiness_timeout_provenance_passthrough.py new file mode 100644 index 000000000..63bd84efa --- /dev/null +++ b/.claude/skills/issue-contract-review/tests/test_readiness_timeout_provenance_passthrough.py @@ -0,0 +1,328 @@ +"""Issue #2897 AC1 (readiness producer side): `contract_readiness_check.py` +carries the identity / applied-budget information a REAL +`baseline_vc_preflight/v1` result item already holds into +`errors[].source_payload`, and the canonical VC plan digest plus the +pre-filter `results` count into the readiness result's top level. + +Production path exercised (no hand-built `timeout_provenance`): + + fake `baseline_vc_preflight.run_command()` return value (permitted seam 1: + the only fake is the end-of-line execution boundary, using the existing + timeout sentinel `exit_code == -1` and `stderr == "timeout"`) + -> REAL `baseline_vc_preflight.main()` result builder + -> REAL `contract_readiness_check.run_baseline_vc_preflight()` + -> REAL `contract_readiness_check.main()` conversion + +Permitted seam 2 (process-launch mechanics only): the cooperative supervisor +that would spawn `baseline_vc_preflight.py` as a child process is replaced by +an adapter that calls that SAME script's `main()` in-process and captures its +stdout / return code, so the seam-1 `run_command()` replacement is visible to +the "child". The result builder, the conversion and the readiness `main()` +are real. +""" + +from __future__ import annotations + +import contextlib +import io +import json +import signal +import sys +from pathlib import Path +from unittest import mock + +import pytest + +_SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts" +if str(_SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(_SCRIPTS_DIR)) + +import baseline_vc_preflight as bvp # noqa: E402 +import contract_readiness_check as crc # noqa: E402 + +TIMEOUT_OUTCOME = (-1, "", "timeout", 1234, {}) +NOT_FOUND_OUTCOME = (4, "", "ERROR: file or directory not found: x", 5, {}) + +_NEW_TEST_PATH = ".claude/skills/issue-contract-review/tests/test_fixture_target_not_yet_created.py" +_PYTEST_VC = f"uv run --locked pytest {_NEW_TEST_PATH}::test_target" +_PURE_VC = "test -f README.md" + +# Occurrence layout (canonical `results` order, before error filtering): +# 0 pytest VC, block 1 -> not found (expected baseline fail, no error) +# 1 pytest VC, block 2 -> TIMEOUT (same command_hash and same +# block-relative line as occurrence 0) +# 2 pure `test -f` block -> TIMEOUT (real execution) +# 3 pure `test -f` block -> dedup replay of occurrence 2 +_BODY = f"""## Verification Commands + +```bash +# AC1 +# baseline-expect: fail +$ {_PYTEST_VC} +``` + +```bash +# AC1 +# baseline-expect: fail +$ {_PYTEST_VC} +``` + +```bash +# AC2 +$ {_PURE_VC} +``` + +```bash +# AC2 +$ {_PURE_VC} +``` + +## Allowed Paths + +- {_NEW_TEST_PATH} +""" + + +class _RunCommandSeam: + """Seam 1: replaces `baseline_vc_preflight.run_command()` only.""" + + def __init__(self, outcomes_by_call: dict[int, tuple]): + self._outcomes_by_call = outcomes_by_call + self.calls: list[tuple[str, int]] = [] + + def __call__(self, command: str, timeout_seconds: int, cwd: str): + call_index = len(self.calls) + self.calls.append((command, timeout_seconds)) + return self._outcomes_by_call.get(call_index, NOT_FOUND_OUTCOME) + + +class _InProcessBaselineLauncher: + """Seam 2: launch mechanics only. Runs the real + `baseline_vc_preflight.main()` in-process instead of spawning it.""" + + def __init__(self): + self.raw_payloads: list[dict] = [] + + def __call__(self, argv, *, timeout_seconds, cwd=None, env=None, **_ignored): + assert Path(argv[1]).name == "baseline_vc_preflight.py", argv + out, err = io.StringIO(), io.StringIO() + previous_sigterm = signal.getsignal(signal.SIGTERM) + try: + with mock.patch.object(sys, "argv", [argv[1], *argv[2:]]): + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + returncode = bvp.main() + finally: + signal.signal(signal.SIGTERM, previous_sigterm) + self.raw_payloads.append(json.loads(out.getvalue())) + return bvp.SupervisedSubprocessResult(returncode, out.getvalue(), err.getvalue(), False, 0.0) + + +def _run_readiness_main( + monkeypatch, tmp_path: Path, body: str, outcomes_by_call: dict[int, tuple] +) -> tuple[dict, int, _RunCommandSeam, _InProcessBaselineLauncher]: + seam = _RunCommandSeam(outcomes_by_call) + launcher = _InProcessBaselineLauncher() + monkeypatch.setattr(bvp, "run_command", seam) + monkeypatch.setattr(crc, "_run_subprocess_with_cooperative_supervisor", launcher) + body_file = tmp_path / "body.md" + body_file.write_text(body, encoding="utf-8") + out = io.StringIO() + with mock.patch.object( + sys, "argv", ["contract_readiness_check.py", "--body-file", str(body_file), "--mode", "execute"] + ): + with contextlib.redirect_stdout(out): + returncode = crc.main() + return json.loads(out.getvalue()), returncode, seam, launcher + + +def test_readiness_passes_bounded_provenance_and_occurrence_index(monkeypatch, tmp_path): + # Occurrence 1 times out; the pure command (occurrence 2) times out on its + # single real execution and occurrence 3 is its dedup replay. + readiness, returncode, seam, launcher = _run_readiness_main( + monkeypatch, + tmp_path, + _BODY, + # call 0 = occurrence 0, call 1 = occurrence 1, call 2 = occurrence 2 + {1: TIMEOUT_OUTCOME, 2: TIMEOUT_OUTCOME}, + ) + assert len(launcher.raw_payloads) == 1 + raw = launcher.raw_payloads[0] + raw_results = raw["results"] + + assert readiness["status"] == "human_judgment" + assert returncode == 2 + + # Top level: digest and pre-filter results count come from the existing + # preflight payload, not from a recomputation. + assert raw["diagnostic_report"]["status"] == "complete" + assert readiness["canonical_plan_digest"] == raw["diagnostic_report"]["canonical_plan_digest"] + assert readiness["results_count"] == len(raw_results) == 4 + + timeout_errors = [e for e in readiness["errors"] if e["category"] == "timeout"] + # Error-list position is NOT the canonical occurrence index: occurrence 0 + # produced no error, so the first error carries index 1. + assert [e["source_payload"]["occurrence_index"] for e in timeout_errors] == [1, 2, 3] + assert [e["line_start"] for e in timeout_errors] == [raw_results[i]["line"] for i in (1, 2, 3)] + + for error in timeout_errors: + payload = error["source_payload"] + index = payload["occurrence_index"] + raw_item = raw_results[index] + assert payload["line_coordinate"] == "block_relative" + assert payload["command_hash"] == raw_item["command_hash"] + assert payload["execution_key_hash"] == raw_item["execution_key_hash"] + assert payload["canonical_plan_digest"] == raw["diagnostic_report"]["canonical_plan_digest"] + # The budget is the one the real result item carries -- bounded to its + # five known fields, not a recomputation. + assert payload["timeout_provenance"] == raw_item["timeout_provenance"] + assert set(payload["timeout_provenance"]) == { + "timeout_seconds", + "cleanup_tail_seconds", + "source", + "estimator_version", + "estimator_input_digest", + } + # No raw command text in any of the newly carried fields. + carried = { + key: payload[key] + for key in ( + "occurrence_index", + "line_coordinate", + "execution_key_hash", + "execution_source", + "dedup_source_result_index", + "timeout_provenance", + "canonical_plan_digest", + ) + } + serialized = json.dumps(carried) + assert _PYTEST_VC not in serialized and _PURE_VC not in serialized + + # The applied budget equals the timeout actually handed to run_command() + # for that command (this is not derived from duration_ms). + applied_for_pytest = {t for c, t in seam.calls if c == _PYTEST_VC} + assert applied_for_pytest == {timeout_errors[0]["source_payload"]["timeout_provenance"]["timeout_seconds"]} + + # Execution source vs dedup replay are distinguished, and the replay + # points at the canonical index of the real execution. + assert timeout_errors[0]["source_payload"]["execution_source"] == "executed" + assert timeout_errors[0]["source_payload"]["dedup_source_result_index"] is None + assert timeout_errors[1]["source_payload"]["execution_source"] == "executed" + assert timeout_errors[2]["source_payload"]["execution_source"] == "dedup_replay" + assert timeout_errors[2]["source_payload"]["dedup_source_result_index"] == 2 + assert timeout_errors[2]["source_payload"]["execution_key_hash"] == ( + timeout_errors[1]["source_payload"]["execution_key_hash"] + ) + + # Same AC, same command, same block-relative line in two fenced blocks: + # `line + command_hash` would collide, the canonical index does not. + assert raw_results[0]["command_hash"] == raw_results[1]["command_hash"] + assert raw_results[0]["line"] == raw_results[1]["line"] + assert raw_results[0]["ac"] == raw_results[1]["ac"] + assert raw_results[0]["execution_key_hash"] != raw_results[1]["execution_key_hash"] + + +def test_not_computed_diagnostic_report_yields_null_digest_without_completion(monkeypatch, tmp_path): + # A body with no Verification Commands section makes the real preflight + # take an early-return path whose `diagnostic_report` is `not_computed`. + readiness, _returncode, _seam, launcher = _run_readiness_main( + monkeypatch, tmp_path, "## Outcome\n\nno verification commands here\n", {} + ) + raw = launcher.raw_payloads[0] + assert raw["diagnostic_report"]["status"] == "not_computed" + assert readiness["canonical_plan_digest"] is None + assert readiness["results_count"] == len(raw["results"]) + + +def test_static_mode_readiness_result_has_no_plan_binding_keys(tmp_path): + # Static / preflight-static modes never run the preflight, so the legacy + # readiness shape is unchanged (no new top-level keys). + body = tmp_path / "body.md" + body.write_text(_BODY, encoding="utf-8") + out = io.StringIO() + with mock.patch.object( + sys, "argv", ["contract_readiness_check.py", "--body-file", str(body), "--mode", "static"] + ): + with contextlib.redirect_stdout(out): + crc.main() + result = json.loads(out.getvalue()) + assert "canonical_plan_digest" not in result + assert "results_count" not in result + + +@pytest.mark.parametrize( + "bad_provenance", + [ + None, + {}, + {"timeout_seconds": True, "cleanup_tail_seconds": 15, "source": "static_policy", + "estimator_version": "v2", "estimator_input_digest": "sha256:" + "0" * 64}, + {"timeout_seconds": 150, "cleanup_tail_seconds": 15, "source": "made_up_source", + "estimator_version": "v2", "estimator_input_digest": "sha256:" + "0" * 64}, + {"timeout_seconds": 150, "cleanup_tail_seconds": 15, "source": "static_policy", + "estimator_version": "v2", "estimator_input_digest": "not-a-digest"}, + ], +) +def test_malformed_provenance_is_dropped_not_partially_trusted(bad_provenance): + assert crc._bounded_timeout_provenance(bad_provenance) is None + + +_GOOD_PROVENANCE = { + "timeout_seconds": 150, + "cleanup_tail_seconds": 15, + "source": "static_policy", + "estimator_version": "v2", + "estimator_input_digest": "sha256:" + "0" * 64, +} + + +@pytest.mark.parametrize("bad", [[], {}, ["static_policy"], {"k": "v"}], ids=repr) +@pytest.mark.parametrize("field", sorted(_GOOD_PROVENANCE)) +def test_unhashable_enum_like_provenance_value_is_dropped_without_exception(field, bad): + # PR #2901 review fix_delta (P2): a list / dict value must degrade to the + # same `None` as any other malformed value, never raise `TypeError` from + # the frozenset membership test. + assert crc._bounded_timeout_provenance(dict(_GOOD_PROVENANCE)) == _GOOD_PROVENANCE + assert crc._bounded_timeout_provenance(dict(_GOOD_PROVENANCE, **{field: bad})) is None + + +@pytest.mark.parametrize("bad", [[], {}], ids=repr) +def test_real_conversion_with_unhashable_provenance_source_does_not_raise( + monkeypatch, tmp_path, bad +): + # Through the REAL preflight result builder and the REAL readiness + # conversion: only the (permitted seam 2) child payload is degraded. + launcher_calls = [] + monkeypatch.setattr(bvp, "run_command", lambda command, timeout_seconds, cwd: TIMEOUT_OUTCOME) + + def launcher(argv, *, timeout_seconds, cwd=None, env=None, **_ignored): + out, err = io.StringIO(), io.StringIO() + previous_sigterm = signal.getsignal(signal.SIGTERM) + try: + with mock.patch.object(sys, "argv", [argv[1], *argv[2:]]): + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + returncode = bvp.main() + finally: + signal.signal(signal.SIGTERM, previous_sigterm) + payload = json.loads(out.getvalue()) + for item in payload["results"]: + if isinstance(item.get("timeout_provenance"), dict): + item["timeout_provenance"]["source"] = bad + launcher_calls.append(payload) + return bvp.SupervisedSubprocessResult(returncode, json.dumps(payload), err.getvalue(), False, 0.0) + + monkeypatch.setattr(crc, "_run_subprocess_with_cooperative_supervisor", launcher) + body_file = tmp_path / "body.md" + body_file.write_text(_BODY, encoding="utf-8") + out = io.StringIO() + with mock.patch.object( + sys, "argv", ["contract_readiness_check.py", "--body-file", str(body_file), "--mode", "execute"] + ): + with contextlib.redirect_stdout(out): + crc.main() + result = json.loads(out.getvalue()) + assert launcher_calls + timeout_errors = [e for e in result["errors"] if e.get("category") == "timeout"] + assert timeout_errors + for error in timeout_errors: + assert error["source_payload"]["timeout_provenance"] is None diff --git a/.claude/skills/issue-refinement-loop/tests/test_root_review_timeout_diagnostics.py b/.claude/skills/issue-refinement-loop/tests/test_root_review_timeout_diagnostics.py new file mode 100644 index 000000000..b390660fe --- /dev/null +++ b/.claude/skills/issue-refinement-loop/tests/test_root_review_timeout_diagnostics.py @@ -0,0 +1,685 @@ +"""Issue #2897 AC3/AC4 (root review pipeline): the bounded +`timeout_diagnostics` of a command-level VC timeout survives the temporary +readiness artifact cleanup and is retrievable from BOTH the verified +transport artifact (`REVIEWER_COMPACT_ARTIFACT_V2` `semantic_result`, the +canonical readback input) and the full review artifact. + +Production path exercised by every test here, end to end: + + fake `baseline_vc_preflight.run_command()` return value (permitted seam 1: + timeout sentinel `exit_code == -1` and `stderr == "timeout"`) + -> REAL `baseline_vc_preflight.main()` result builder + -> REAL `contract_readiness_check.main()` readiness conversion + -> REAL `check_issue_contract.main()` review check and + `--mode merge_readiness` merge + -> REAL `run_root_review_pipeline.run_checker_pipeline_once()` + (temporary-directory cleanup in its `finally`) + -> REAL `reviewer_transport.run_reviewer_transport()` artifact writer + -> REAL `run_root_review_pipeline._cmd_produce()` full artifact writer + -> persisted bytes + `readback_persisted_artifact()` verified readback + +Permitted seam 2 (process-launch mechanics only): the places that would spawn +`run-checker-attempt`, `check_issue_contract.py`, `contract_readiness_check.py` +or `baseline_vc_preflight.py` as child processes instead call the SAME +script's `main()` in-process and capture stdout / exit code, so the seam-1 +`run_command()` replacement is visible to the "child". The only other +replaced input is the live Issue body fetch (`fetch_and_pin_live_body`, the +same replacement `test_root_review_canonical_delivery.py` uses); it supplies +the pinned body text and nothing about the result. +""" + +from __future__ import annotations + +import argparse +import contextlib +import io +import json +import signal +import subprocess +import sys +import types +from pathlib import Path +from unittest import mock + + +_SKILLS = Path(__file__).resolve().parents[2] +for _path in ( + _SKILLS / "issue-refinement-loop" / "scripts", + _SKILLS / "issue-contract-review" / "scripts", + _SKILLS / "review-issue" / "scripts", +): + if str(_path) not in sys.path: + sys.path.insert(0, str(_path)) + +import baseline_vc_preflight as bvp # noqa: E402 +import check_issue_contract as cic # noqa: E402 +import contract_readiness_check as crc # noqa: E402 +import reviewer_transport as transport # noqa: E402 +import run_root_review_pipeline as pipeline # noqa: E402 +import vc_runtime_history as history # noqa: E402 + +REPO = "squne121/loop-protocol" + +TIMEOUT_OUTCOME = (-1, "", "timeout", 1234, {}) +NOT_FOUND_OUTCOME = (4, "", "ERROR: file or directory not found: x", 5, {}) +SUCCESS_OUTCOME = (0, "", "", 5, {}) + +_NEW_TEST_PATH = ".claude/skills/issue-refinement-loop/tests/test_fixture_target_not_yet_created.py" +_PYTEST_VC = f"uv run --locked pytest {_NEW_TEST_PATH}::test_target" +_PURE_VC = "test -f README.md" + +_BODY_HEADER = """## Machine-Readable Contract + +```yaml +contract_schema_version: v1 +issue_kind: implementation +parent_issue: none +goal_ref: "root review timeout diagnostics fixture" +change_kind: workflow +``` + +## Outcome + +Fixture for the root review timeout diagnostics. + +## Acceptance Criteria + +- [ ] AC1: the root artifact keeps the timed-out occurrence identity. +- [ ] AC2: the root artifact never attributes a timeout to another occurrence. + +## Verification Commands + +""" + +_ALLOWED = f""" +## Allowed Paths + +- {_NEW_TEST_PATH} +""" + +# Call order (pure command, pytest block 1, pytest block 2). The two pytest +# blocks collide on (AC, block-relative line, command_hash). +_TWO_BLOCK_BODY = ( + _BODY_HEADER + + f"""```bash +# AC1 +$ {_PURE_VC} +``` + +```bash +# AC1 +# baseline-expect: fail +$ {_PYTEST_VC} +``` + +```bash +# AC1 +# baseline-expect: fail +$ {_PYTEST_VC} +``` +""" + + _ALLOWED +) + +# 20 identical pure commands: one real execution (occurrence 0, timeout) and +# 19 dedup replays that carry the same timeout outcome -> 20 timeout +# occurrences, above the 16-occurrence bound. +_TWENTY_TIMEOUTS_BODY = ( + _BODY_HEADER + + "".join(f"```bash\n# AC2\n$ {_PURE_VC}\n```\n\n" for _ in range(20)) + + _ALLOWED +) + +_APPROVE_BODY = """## Machine-Readable Contract + +```yaml +contract_schema_version: v1 +issue_kind: research +parent_issue: none +goal_ref: "root review timeout diagnostics fixture (approve branch)" +change_kind: research +``` + +## Outcome + +Fixture proving a normal success carries no timeout diagnostics. + +## Acceptance Criteria + +- [ ] AC1: fixture body is well-formed enough for an approve verdict. + +## Verification Commands + +```bash +# AC1 +# baseline-expect: pass +$ true +``` + +## Allowed Paths + +- fixture/root_review_timeout_diagnostics_approve.md +""" + +_DIAGNOSTIC_TOP_LEVEL_KEYS = { + "schema_version", + "body_sha256", + "canonical_plan_digest", + "results_count", + "total_timeout_occurrences", + "truncated_count", + "occurrences", +} +_DIAGNOSTIC_OCCURRENCE_KEYS = { + "attribution", + "reason_code", + "occurrence_index", + "line", + "line_coordinate", + "command_hash", + "execution_key_hash", + "execution_source", + "dedup_source_result_index", + "timeout_provenance", +} + + +class _Harness: + """Wires the production pipeline to in-process script launches.""" + + def __init__( + self, + monkeypatch, + tmp_path: Path, + body: str, + *, + outcomes_by_call: dict[int, tuple] | None = None, + default_outcome: tuple = NOT_FOUND_OUTCOME, + on_transport_launch=None, + baseline_supervisor_timed_out: bool = False, + ): + self.tmp_path = tmp_path + self.body = body + self.body_sha256 = pipeline.sha256_of(body) + self.outcomes_by_call = outcomes_by_call or {} + self.default_outcome = default_outcome + self.run_command_calls: list[tuple[str, int]] = [] + self.baseline_payloads: list[dict] = [] + self.readiness_payloads: list[dict] = [] + self.readiness_files_seen: list[str] = [] + self.transport_launches = 0 + self._on_transport_launch = on_transport_launch + self._baseline_supervisor_timed_out = baseline_supervisor_timed_out + + monkeypatch.setattr(pipeline, "_REPO_ROOT", tmp_path) + monkeypatch.setattr(pipeline, "fetch_and_pin_live_body", self._fake_fetch) + monkeypatch.setattr(bvp, "run_command", self._fake_run_command) + monkeypatch.setattr(bvp, "run_subprocess_with_cooperative_supervisor", self._supervisor) + monkeypatch.setattr(crc, "_run_subprocess_with_cooperative_supervisor", self._supervisor) + monkeypatch.setattr( + pipeline, + "subprocess", + types.SimpleNamespace(run=self._fake_subprocess_run, TimeoutExpired=subprocess.TimeoutExpired), + ) + monkeypatch.setattr( + transport, + "subprocess", + types.SimpleNamespace( + Popen=self._fake_popen, + DEVNULL=subprocess.DEVNULL, + PIPE=subprocess.PIPE, + TimeoutExpired=subprocess.TimeoutExpired, + ), + ) + + # -- inputs / seam 1 ---------------------------------------------------- + + def _fake_fetch(self, issue_number, repo, timeout_seconds=15): + return self.body, self.body_sha256, None + + def _fake_run_command(self, command: str, timeout_seconds: int, cwd: str): + index = len(self.run_command_calls) + self.run_command_calls.append((command, timeout_seconds)) + return self.outcomes_by_call.get(index, self.default_outcome) + + # -- seam 2: launch mechanics ------------------------------------------- + + def _run_script(self, argv: list[str]) -> tuple[int, str, str]: + script = Path(argv[1]).name + out, err = io.StringIO(), io.StringIO() + previous_sigterm = signal.getsignal(signal.SIGTERM) + code = 0 + try: + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + if script == "run_root_review_pipeline.py": + code = pipeline.main(argv[2:]) + else: + module_main = { + "check_issue_contract.py": cic.main, + "contract_readiness_check.py": crc.main, + "baseline_vc_preflight.py": bvp.main, + }[script] + with mock.patch.object(sys, "argv", [argv[1], *argv[2:]]): + try: + code = module_main() or 0 + except SystemExit as exc: + code = exc.code if isinstance(exc.code, int) else 1 + finally: + signal.signal(signal.SIGTERM, previous_sigterm) + if script == "baseline_vc_preflight.py": + self.baseline_payloads.append(json.loads(out.getvalue())) + if script == "contract_readiness_check.py" and "execute" in argv: + self.readiness_payloads.append(json.loads(out.getvalue())) + if script == "check_issue_contract.py" and "merge_readiness" in argv: + self.readiness_files_seen.append(argv[argv.index("--readiness-result-file") + 1]) + return code, out.getvalue(), err.getvalue() + + def _supervisor(self, argv, *, timeout_seconds, cwd=None, env=None, **_ignored): + script = Path(argv[1]).name + if script == "baseline_vc_preflight.py" and self._baseline_supervisor_timed_out: + # The aggregate wrapper reports its own timeout. + return bvp.SupervisedSubprocessResult(-1, "", "", True, 0.0) + code, stdout, stderr = self._run_script(argv) + return bvp.SupervisedSubprocessResult(code, stdout, stderr, False, 0.0) + + def _fake_subprocess_run(self, cmd, capture_output=True, text=True, timeout=None, **_ignored): + code, stdout, stderr = self._run_script(list(cmd)) + return subprocess.CompletedProcess(cmd, code, stdout, stderr) + + def _fake_popen(self, command, **_ignored): + self.transport_launches += 1 + if self._on_transport_launch is not None: + self._on_transport_launch() + code, stdout, stderr = self._run_script(list(command)) + + class _Process: + pid = 424242 + returncode = code + + def __init__(self): + self.stdout = io.BytesIO(stdout.encode("utf-8")) + self.stderr = io.BytesIO(stderr.encode("utf-8")) + + def wait(self, timeout=None): + return self.returncode + + return _Process() + + # -- production entrypoint ---------------------------------------------- + + def produce(self, issue_number: int) -> tuple[int, dict]: + out = io.StringIO() + with contextlib.redirect_stdout(out): + code = pipeline._cmd_produce(argparse.Namespace(issue_number=issue_number, repo=REPO)) + return code, json.loads(out.getvalue()) + + def verified_readback(self, out: dict, issue_number: int, expected_verdict: str) -> dict: + vta = out["verified_transport_artifact"] + return pipeline.readback_persisted_artifact( + artifact_root=vta["root"], + artifact_relative=vta["relative_path"], + expected_repo=REPO, + expected_issue=issue_number, + expected_body_sha256=self.body_sha256, + expected_invocation_id=vta["invocation_id"], + expected_attempt=vta["attempt"], + expected_artifact_sha256=vta["sha256"], + expected_verdict=expected_verdict, + ) + + +def _persisted_artifacts(out: dict) -> tuple[dict, dict, bytes]: + """(full review artifact JSON, transport semantic_result, transport bytes) + read back from disk.""" + full = json.loads(Path(out["full_review_artifact"]["path"]).read_text(encoding="utf-8")) + vta = out["verified_transport_artifact"] + transport_bytes = (Path(vta["root"]) / vta["relative_path"]).read_bytes() + transport_payload = transport.strict_json_loads(transport_bytes) + return full, transport_payload["semantic_result"], transport_bytes + + +def _attempt_result(out: dict, issue_number: int) -> dict: + vta = out["verified_transport_artifact"] + path = ( + Path(vta["root"]) + / transport.attempt_relative_dir(issue_number, vta["invocation_id"], vta["attempt"]) + / "attempt_result.json" + ) + return json.loads(path.read_text(encoding="utf-8")) + + +def test_production_path_diagnostic_survives_cleanup_in_transport_and_full_artifact( + monkeypatch, tmp_path +): + issue_number = 2897001 + harness = _Harness( + monkeypatch, tmp_path, _TWO_BLOCK_BODY, outcomes_by_call={2: TIMEOUT_OUTCOME} + ) + code, out = harness.produce(issue_number) + assert code == 0 and out["status"] == "ok", out + + # Existing routing facts are unchanged: an inner readiness timeout stays + # the operator-intervention route, not a transport failure. + assert out["compact_result"]["verdict"] == "needs-fix" + assert out["compact_result"]["next_action"] == "request_changes" + assert out["merged_review_result"]["failure_class"] == "contract_readiness_human_judgment" + assert out["canonical_step2_route"] == pipeline.STEP_5_OPERATOR_INTERVENTION_REQUIRED + attempt = _attempt_result(out, issue_number) + assert attempt["transport_status"] == "ok" + assert attempt["timeout"] is False and attempt["exit_code"] == 0 + assert attempt["reason_code"] is None # not misclassified as an outer timeout + + # Temporary readiness artifact (and its scratch directory) are gone. + assert harness.readiness_files_seen, "merge step did not run" + for readiness_file in harness.readiness_files_seen: + assert not Path(readiness_file).exists() + assert not Path(readiness_file).parent.exists() + assert list((tmp_path / "tmp").iterdir()) == [] + + # Persisted bytes: canonical transport artifact and full review artifact. + full_artifact, semantic_result, transport_bytes = _persisted_artifacts(out) + assert b"timeout_diagnostics" in transport_bytes + diagnostics = semantic_result["timeout_diagnostics"] + assert full_artifact["timeout_diagnostics"] == diagnostics + assert out["merged_review_result"]["timeout_diagnostics"] == diagnostics + + # Verified readback (what gate-final-review consumes) returns the same. + readback = harness.verified_readback(out, issue_number, "needs-fix") + assert readback["verdict_identity"] is True, readback + assert readback["payload"]["semantic_result"]["timeout_diagnostics"] == diagnostics + gate = pipeline.gate_final_review(remote_update_ok=True, readback=readback) + assert gate["final_review_allowed"] is True + + # The diagnostic ties the plan digest and the applied budget of the + # timed-out occurrence (index 2) to the REAL preflight result. + raw = harness.baseline_payloads[0] + raw_item = raw["results"][2] + assert diagnostics["body_sha256"] == harness.body_sha256 + assert diagnostics["canonical_plan_digest"] == raw["diagnostic_report"]["canonical_plan_digest"] + assert diagnostics["total_timeout_occurrences"] == 1 + (occurrence,) = diagnostics["occurrences"] + assert occurrence["attribution"] == "attributed" + assert occurrence["occurrence_index"] == 2 + assert occurrence["command_hash"] == raw_item["command_hash"] + assert occurrence["timeout_provenance"] == raw_item["timeout_provenance"] + # ...and that applied budget is the timeout run_command() actually got. + assert harness.run_command_calls[2] == ( + _PYTEST_VC, + occurrence["timeout_provenance"]["timeout_seconds"], + ) + # The other pytest block (same hash, same line) is not attributed. + assert raw["results"][1]["command_hash"] == raw_item["command_hash"] + assert raw["results"][1]["line"] == raw_item["line"] + + +def _seed_history(store: Path, command: str, *, duration_ms: int, count: int) -> None: + resolved_repo_root = bvp.resolve_repo_root_for_history(".") + group_key = history.compute_command_group_key(command, ".", repo_root=resolved_repo_root) + fingerprint = history.compute_environment_fingerprint(bvp._command_family(command)) + for _ in range(count): + outcome = history.record_sample( + store, + execution_id=history.new_execution_id(), + command_group_key=group_key, + environment_fingerprint=fingerprint, + status="success", + command_hash=bvp.compute_command_hash(command), + duration_ms=duration_ms, + applied_timeout_ms=150000, + ) + assert outcome["recorded"], outcome + + +def _pytest_vc_budget_from_store(body: str) -> dict: + resolved_repo_root = bvp.resolve_repo_root_for_history(".") + snapshot = bvp.produce_immutable_history_snapshot(body, cwd=".", repo_root=resolved_repo_root) + plan = bvp.compute_canonical_vc_plan( + body, + cwd=".", + allowed_paths=bvp.extract_allowed_paths(body), + history_snapshot=snapshot, + repo_root=resolved_repo_root, + ) + command_hash = bvp.compute_command_hash(_PYTEST_VC) + (budget,) = [b for b in plan["command_budgets"] if b["command_hash"] == command_hash] + return budget + + +def test_snapshot_budget_unchanged_after_history_mutation(monkeypatch, tmp_path): + issue_number = 2897002 + store = tmp_path / "history.sqlite3" + monkeypatch.setenv("VC_RUNTIME_HISTORY_STORE_PATH", str(store)) + _seed_history(store, _PYTEST_VC, duration_ms=120000, count=5) + + # What the root-owned immutable snapshot (taken once at the start of + # `produce`) resolves for the timed-out command. + budget_at_snapshot = _pytest_vc_budget_from_store(_TWO_BLOCK_BODY) + assert budget_at_snapshot["source"] == "history_estimate" + assert budget_at_snapshot["timeout_seconds"] == 180 + + def mutate_history_after_snapshot(): + # Runs when the transport launches the checker child, i.e. AFTER the + # root snapshot was built and serialized and BEFORE readiness runs. + _seed_history(store, _PYTEST_VC, duration_ms=200000, count=5) + + harness = _Harness( + monkeypatch, + tmp_path, + _TWO_BLOCK_BODY, + outcomes_by_call={2: TIMEOUT_OUTCOME}, + on_transport_launch=mutate_history_after_snapshot, + ) + code, out = harness.produce(issue_number) + assert code == 0 and out["status"] == "ok", out + assert harness.transport_launches == 1 + + # The store really changed: a fresh read now resolves a different budget. + budget_after_mutation = _pytest_vc_budget_from_store(_TWO_BLOCK_BODY) + assert budget_after_mutation["timeout_seconds"] != budget_at_snapshot["timeout_seconds"] + assert budget_after_mutation["estimator_input_digest"] != budget_at_snapshot["estimator_input_digest"] + + # The persisted diagnostic keeps the snapshot-time budget, in BOTH + # persisted artifacts, and equals what run_command() was really given. + full_artifact, semantic_result, _raw_bytes = _persisted_artifacts(out) + for diagnostics in (full_artifact["timeout_diagnostics"], semantic_result["timeout_diagnostics"]): + (occurrence,) = diagnostics["occurrences"] + assert occurrence["attribution"] == "attributed" + provenance = occurrence["timeout_provenance"] + assert provenance["source"] == "history_estimate" + assert provenance["timeout_seconds"] == budget_at_snapshot["timeout_seconds"] == 180 + assert provenance["estimator_input_digest"] == budget_at_snapshot["estimator_input_digest"] + assert provenance["timeout_seconds"] != budget_after_mutation["timeout_seconds"] + timed_out_calls = [call for call in harness.run_command_calls if call[0] == _PYTEST_VC] + assert {seconds for _command, seconds in timed_out_calls} == {180} + + +def _no_leak_run(monkeypatch, tmp_path) -> tuple[dict, dict]: + stdout_marker = "STDOUT_SECRET_MARKER_5d1c" + env_marker = "ENV_SECRET_MARKER_8e2f" + leaky_timeout = (-1, stdout_marker, "timeout", 1234, {"RUNNER_ENV_SECRET": env_marker}) + harness = _Harness(monkeypatch, tmp_path, _TWO_BLOCK_BODY, outcomes_by_call={2: leaky_timeout}) + code, out = harness.produce(2897003) + assert code == 0 and out["status"] == "ok", out + return out, {"stdout": stdout_marker, "env": env_marker} + + +def test_no_leak_and_non_timeout_routes_unchanged(monkeypatch, tmp_path): + # --- 1. no leak: exact key allowlist, no raw command / output / env ----- + out, markers = _no_leak_run(monkeypatch, tmp_path / "leak") + _full, semantic_result, _bytes = _persisted_artifacts(out) + diagnostics = semantic_result["timeout_diagnostics"] + assert set(diagnostics) == _DIAGNOSTIC_TOP_LEVEL_KEYS + for occurrence in diagnostics["occurrences"]: + assert set(occurrence) == _DIAGNOSTIC_OCCURRENCE_KEYS + serialized = json.dumps(diagnostics) + for forbidden in ( + markers["stdout"], + markers["env"], + _PYTEST_VC, + _PURE_VC, + "runner_env_delta", + "minimal_context", + "raw_command", + "stdout_head", + "stderr_head", + ): + assert forbidden not in serialized, forbidden + assert len(serialized.encode("utf-8")) <= 16 * 1024 + + # --- 2. 17+ timeouts: bounded, no capture_failure ---------------------- + many_dir = tmp_path / "many" + many_dir.mkdir() + harness = _Harness(monkeypatch, many_dir, _TWENTY_TIMEOUTS_BODY, outcomes_by_call={0: TIMEOUT_OUTCOME}) + code, many_out = harness.produce(2897004) + assert code == 0 and many_out["status"] == "ok", many_out + many_attempt = _attempt_result(many_out, 2897004) + assert many_attempt["transport_status"] == "ok" + assert many_attempt["reason_code"] != "capture_failure" + _full, many_semantic, _bytes = _persisted_artifacts(many_out) + many_diagnostics = many_semantic["timeout_diagnostics"] + assert many_diagnostics["total_timeout_occurrences"] == 20 + assert len(many_diagnostics["occurrences"]) == 16 + assert many_diagnostics["truncated_count"] == 4 + assert [o["occurrence_index"] for o in many_diagnostics["occurrences"]] == list(range(16)) + assert many_diagnostics["occurrences"][0]["execution_source"] == "executed" + assert all(o["execution_source"] == "dedup_replay" for o in many_diagnostics["occurrences"][1:]) + assert len(json.dumps(many_diagnostics).encode("utf-8")) <= 16 * 1024 + assert many_out["canonical_step2_route"] == pipeline.STEP_5_OPERATOR_INTERVENTION_REQUIRED + + # --- 3. normal success: no diagnostic, no new gate --------------------- + ok_dir = tmp_path / "ok" + ok_dir.mkdir() + harness = _Harness(monkeypatch, ok_dir, _APPROVE_BODY, default_outcome=SUCCESS_OUTCOME) + code, ok_out = harness.produce(2897005) + assert code == 0 and ok_out["status"] == "ok", ok_out + assert ok_out["compact_result"]["verdict"] == "approve" + assert ok_out["canonical_step2_route"] == pipeline.STEP_2_5 + assert "timeout_diagnostics" not in ok_out["merged_review_result"] + _full, ok_semantic, _bytes = _persisted_artifacts(ok_out) + assert "timeout_diagnostics" not in ok_semantic + + # --- 4. non-timeout human_judgment: no diagnostic, same route ---------- + hj_dir = tmp_path / "hj" + hj_dir.mkdir() + harness = _Harness( + monkeypatch, hj_dir, _TWO_BLOCK_BODY, outcomes_by_call={2: (1, "", "boom", 9, {})} + ) + code, hj_out = harness.produce(2897006) + assert code == 0 and hj_out["status"] == "ok", hj_out + assert hj_out["merged_review_result"]["failure_class"] == "contract_readiness_human_judgment" + assert "timeout_diagnostics" not in hj_out["merged_review_result"] + assert hj_out["canonical_step2_route"] == pipeline.STEP_5_OPERATOR_INTERVENTION_REQUIRED + + # --- 5. outer (aggregate wrapper) timeout: existing route, no diagnostic + outer_dir = tmp_path / "outer" + outer_dir.mkdir() + harness = _Harness( + monkeypatch, outer_dir, _TWO_BLOCK_BODY, baseline_supervisor_timed_out=True + ) + code, outer_out = harness.produce(2897007) + assert code == 2 + assert outer_out["status"] == "input_or_runtime_error" + assert outer_out["error_code"] == "reviewer_transport_environment_failure" + assert "merged_review_result" not in outer_out + assert "timeout_diagnostics" not in json.dumps(outer_out) + assert outer_out["canonical_step2_route"] == pipeline.FAIL_CLOSED_ENVIRONMENT_OR_INTEGRITY_FAILURE + + # --- 6. compact V2 wire unchanged ------------------------------------- + wire = "\n".join(out["compact_result"]["stdout_lines"]).encode("utf-8") + b"\n" + assert [line.split(": ", 1)[0] for line in out["compact_result"]["stdout_lines"]] == list( + transport.V2_FIELDS + ) + vta = out["verified_transport_artifact"] + validated = transport.validate_compact_v2( + wire, issue_number=2897003, invocation_id=vta["invocation_id"], attempt=vta["attempt"] + ) + assert validated["validation_status"] == "valid", validated + assert b"timeout_diagnostics" not in wire + + +def test_legacy_merged_result_without_diagnostics_is_still_a_valid_semantic_result(): + # `timeout_diagnostics` is an optional additive field: a legacy merged + # result without it validates, and with it validates too. + legacy = { + "schema": "REVIEW_ISSUE_RESULT_V1", + "schema_version": "1", + "verdict": "approve", + "status": "ok", + "body_sha256": "sha256:" + "0" * 64, + "issue_kind": "implementation", + "generated_at": "2026-10-04T00:00:00Z", + "deterministic_checks": {}, + "blocking_issues": [], + "structured_blockers": [], + "non_blocking_improvements": [], + "findings": [], + "diff_proposal": {}, + "parsed_vc_commands": [], + } + assert transport.validate_semantic_result_schema(legacy) is None + with_diagnostics = dict(legacy, timeout_diagnostics={"schema_version": "TIMEOUT_DIAGNOSTICS_V1"}) + assert transport.validate_semantic_result_schema(with_diagnostics) is None + + +def _big_review_body() -> str: + # 30 long pure VCs: the merged review result is large (~58 KB) even + # without the diagnostic, so adding all 16 occurrences (~10 KB) overflows + # the transport stdout cap although the diagnostic alone is < 16 KiB. + blocks = "".join( + f"```bash\n# AC{(index % 2) + 1}\n$ test -f d{index}/{'a' * 1450}.md\n```\n\n" + for index in range(30) + ) + return _BODY_HEADER + blocks + _ALLOWED + + +def test_large_review_result_with_timeouts_is_not_turned_into_capture_failure( + monkeypatch, tmp_path +): + # The budget the merge side enforces is the transport's own cap. + assert cic.TIMEOUT_DIAGNOSTICS_STDOUT_CAP_BYTES == transport.STDOUT_CAP == 65_536 + + def writer_bytes(result: dict) -> int: + # `_cmd_run_checker_attempt()`: print(json.dumps(merged)) -> utf-8. + return len((json.dumps(result) + "\n").encode("utf-8")) + + issue_number = 2897008 + harness = _Harness( + monkeypatch, + tmp_path, + _big_review_body(), + outcomes_by_call={call: TIMEOUT_OUTCOME for call in range(20)}, + ) + code, out = harness.produce(issue_number) + assert code == 0 and out["status"] == "ok", out + + # Real transport: not a capture_failure, no retry storm, verified artifact. + attempt = _attempt_result(out, issue_number) + assert attempt["transport_status"] == "ok", attempt + assert attempt["reason_code"] != "capture_failure" + assert harness.transport_launches == 1 + full_artifact, semantic_result, _bytes = _persisted_artifacts(out) + merged = out["merged_review_result"] + assert full_artifact["verdict"] == semantic_result["verdict"] == "needs-fix" + + # Precondition of the scenario (computed with the real writer's options): + # diagnostic-free result fits, the naive full diagnostic would not. + base = {key: value for key, value in merged.items() if key != "timeout_diagnostics"} + assert writer_bytes(base) <= transport.STDOUT_CAP + assert len(harness.baseline_payloads[0]["results"]) == 30 + full_diagnostics = cic.build_timeout_diagnostics( + harness.readiness_payloads[0], body_sha256=harness.body_sha256 + ) + assert len(full_diagnostics["occurrences"]) == 16 + assert writer_bytes(dict(base, timeout_diagnostics=full_diagnostics)) > transport.STDOUT_CAP + + # Routing-critical facts are intact; only the optional diagnostic shrank. + assert merged["failure_class"] == "contract_readiness_human_judgment" + assert out["canonical_step2_route"] == pipeline.STEP_5_OPERATOR_INTERVENTION_REQUIRED + assert writer_bytes(merged) <= transport.STDOUT_CAP + diagnostics = semantic_result["timeout_diagnostics"] + assert diagnostics == merged["timeout_diagnostics"] == full_artifact["timeout_diagnostics"] + assert diagnostics["total_timeout_occurrences"] == 20 + assert 0 < len(diagnostics["occurrences"]) < 16 + assert diagnostics["truncated_count"] == 20 - len(diagnostics["occurrences"]) + readback = harness.verified_readback(out, issue_number, "needs-fix") + assert readback["verdict_identity"] is True, readback diff --git a/.claude/skills/review-issue/scripts/check_issue_contract.py b/.claude/skills/review-issue/scripts/check_issue_contract.py index 61aea3593..7631bf951 100644 --- a/.claude/skills/review-issue/scripts/check_issue_contract.py +++ b/.claude/skills/review-issue/scripts/check_issue_contract.py @@ -588,6 +588,361 @@ def readiness_status_to_failure_class(readiness_status: Optional[str]) -> Option return None +# --------------------------------------------------------------------------- +# Issue #2897: bounded `timeout_diagnostics` projection (review merge side). +# +# A readiness `human_judgment` result caused by a command-level VC timeout +# does not become a deterministic `structured_blockers` entry (and must not: +# `failure_class` / the operator-only route are unchanged). Without this +# projection its identity collapses to a `blocking_issues` string. This only +# COPIES values the readiness result already holds (which in turn only copies +# the original `baseline_vc_preflight/v1` result item); it never estimates, +# recomputes, or back-derives a budget (e.g. from `duration_ms` or from the +# current history store) and never reads `parsed_vc_commands`. +# --------------------------------------------------------------------------- + +TIMEOUT_DIAGNOSTICS_SCHEMA_VERSION = "TIMEOUT_DIAGNOSTICS_V1" +TIMEOUT_DIAGNOSTICS_MAX_OCCURRENCES = 16 +# Per-diagnostic bound. This alone does NOT keep the merged result under the +# transport cap (the diagnostic is added to an already large result); the +# whole-result budget below is what is authoritative. +TIMEOUT_DIAGNOSTICS_MAX_SERIALIZED_BYTES = 16 * 1024 +# Mirrors `reviewer_transport.STDOUT_CAP` (65,536 bytes): the transport turns a +# child stdout larger than this into a `capture_failure`. The child writer is +# `run_root_review_pipeline._cmd_run_checker_attempt()`'s +# `print(json.dumps(merged))` (default `ensure_ascii=True`, trailing newline, +# utf-8). A pinned test asserts this constant equals the transport's. +TIMEOUT_DIAGNOSTICS_STDOUT_CAP_BYTES = 65_536 + +TIMEOUT_ATTRIBUTION_ATTRIBUTED = "attributed" +TIMEOUT_ATTRIBUTION_UNKNOWN = "unknown" +TIMEOUT_REASON_BINDING_VERIFIED = "binding_verified" +TIMEOUT_UNKNOWN_REASON_CODES = ( + "plan_digest_missing", + "plan_digest_mismatch", + "provenance_missing", + "execution_key_missing", + "dedup_binding_invalid", + "occurrence_index_out_of_range", +) + +_TIMEOUT_PROVENANCE_SOURCES = frozenset( + {"explicit_override", "static_policy", "static_fallback", "history_estimate"} +) +_TIMEOUT_SHA256_RE = re.compile(r"^sha256:[0-9a-f]{64}$") +_TIMEOUT_ESTIMATOR_VERSION_RE = re.compile(r"^[A-Za-z0-9._-]{1,32}$") +_TIMEOUT_MAX_SECONDS = 86_400 +_TIMEOUT_MAX_INDEX = 1_000_000 +_TIMEOUT_EXECUTION_SOURCES = frozenset({"executed", "dedup_replay"}) + + +def _timeout_plain_int(value: object) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def _timeout_index(value: object) -> Optional[int]: + if _timeout_plain_int(value) and 0 <= value <= _TIMEOUT_MAX_INDEX: # type: ignore[operator] + return value # type: ignore[return-value] + return None + + +def _timeout_sha256(value: object) -> Optional[str]: + if isinstance(value, str) and _TIMEOUT_SHA256_RE.match(value): + return value + return None + + +def _timeout_bounded_provenance(raw: object) -> Optional[dict]: + if not isinstance(raw, dict): + return None + timeout_seconds = raw.get("timeout_seconds") + cleanup_tail_seconds = raw.get("cleanup_tail_seconds") + source = raw.get("source") + estimator_version = raw.get("estimator_version") + estimator_input_digest = _timeout_sha256(raw.get("estimator_input_digest")) + if not (_timeout_plain_int(timeout_seconds) and 0 < timeout_seconds <= _TIMEOUT_MAX_SECONDS): # type: ignore[operator] + return None + if not (_timeout_plain_int(cleanup_tail_seconds) and 0 <= cleanup_tail_seconds <= _TIMEOUT_MAX_SECONDS): # type: ignore[operator] + return None + # Type first: an unhashable (list / dict) value must degrade to `None` + # (-> `unknown`), not raise `TypeError` from the frozenset membership test. + if not isinstance(source, str) or source not in _TIMEOUT_PROVENANCE_SOURCES: + return None + if not (isinstance(estimator_version, str) and _TIMEOUT_ESTIMATOR_VERSION_RE.match(estimator_version)): + return None + if estimator_input_digest is None: + return None + return { + "timeout_seconds": timeout_seconds, + "cleanup_tail_seconds": cleanup_tail_seconds, + "source": source, + "estimator_version": estimator_version, + "estimator_input_digest": estimator_input_digest, + } + + +def _timeout_error_payload(error: dict) -> dict: + payload = error.get("source_payload") + return payload if isinstance(payload, dict) else {} + + +def _timeout_provenance_identity(provenance: Optional[dict]) -> Optional[tuple]: + """Hashable, ordered identity of a bounded provenance (dedup source / + replay comparison). `None` stays `None`.""" + if provenance is None: + return None + return ( + provenance["timeout_seconds"], + provenance["cleanup_tail_seconds"], + provenance["source"], + provenance["estimator_version"], + provenance["estimator_input_digest"], + ) + + +def _timeout_unknown_reason( + *, + top_level_digest: Optional[str], + results_count: Optional[int], + error_digest: Optional[str], + occurrence_index: Optional[int], + provenance: Optional[dict], + execution_key_hash: Optional[str], + execution_source: Optional[str], + dedup_source_index: Optional[int], + executed_binding_by_index: dict, +) -> Optional[str]: + """Return the bounded `unknown` reason code, or `None` if the occurrence + binding is verified. Pure comparison of values already held by the + readiness result; nothing is recomputed.""" + if top_level_digest is None: + return "plan_digest_missing" + if error_digest != top_level_digest: + return "plan_digest_mismatch" + if ( + occurrence_index is None + or results_count is None + or occurrence_index >= results_count + ): + return "occurrence_index_out_of_range" + if provenance is None: + return "provenance_missing" + if execution_key_hash is None: + return "execution_key_missing" + if execution_source == "executed": + if dedup_source_index is not None: + return "dedup_binding_invalid" + elif execution_source == "dedup_replay": + if ( + dedup_source_index is None + or dedup_source_index >= results_count + or dedup_source_index >= occurrence_index + # The source must itself be a binding-verified `executed` + # occurrence (see `build_timeout_diagnostics()`), with the same + # execution key AND the same applied-budget provenance: the + # execution key covers the timeout, and a replay reuses the + # source's execution under the very same budget. + or executed_binding_by_index.get(dedup_source_index) + != (execution_key_hash, _timeout_provenance_identity(provenance)) + ): + return "dedup_binding_invalid" + else: + return "dedup_binding_invalid" + return None + + +def build_timeout_diagnostics( + readiness_result: dict, *, body_sha256: str +) -> Optional[dict]: + """Project command-level VC timeout readiness errors into a bounded + `TIMEOUT_DIAGNOSTICS_V1` object (Issue #2897), or `None` if there is no + timeout occurrence. + + Occurrences are listed in the readiness result's own (canonical result) + order. `occurrence_index` -- the position in the unfiltered `results` + array -- is the primary identifier; `line` is block-relative and only + auxiliary. An occurrence whose binding cannot be verified is recorded as + `attribution: unknown` with a bounded reason code and carries no + provenance / execution identity (never completed from another VC or from + a recomputed budget). + """ + errors = readiness_result.get("errors") or [] + top_level_digest = _timeout_sha256(readiness_result.get("canonical_plan_digest")) + raw_results_count = readiness_result.get("results_count") + results_count = ( + raw_results_count + if _timeout_plain_int(raw_results_count) and 0 <= raw_results_count <= _TIMEOUT_MAX_INDEX + else None + ) + + # `executed` result items by canonical index (any category), used to + # check that a dedup replay points at an actual execution source with the + # same execution key and the same applied-budget provenance. A source is + # registered ONLY if its own binding integrity is verified from values + # already in this readiness result (plan digest, occurrence index range, + # provenance, execution key): an invalid source must never lend + # `binding_verified` to a replay. An index that appears twice with + # conflicting identity is not registered at all. + executed_binding_by_index: dict = {} + conflicting_source_indexes: set = set() + for error in errors: + if not isinstance(error, dict): + continue + payload = _timeout_error_payload(error) + index = _timeout_index(payload.get("occurrence_index")) + key = _timeout_sha256(payload.get("execution_key_hash")) + if index is None or key is None or payload.get("execution_source") != "executed": + continue + source_provenance = _timeout_bounded_provenance(payload.get("timeout_provenance")) + source_reason = _timeout_unknown_reason( + top_level_digest=top_level_digest, + results_count=results_count, + error_digest=_timeout_sha256(payload.get("canonical_plan_digest")), + occurrence_index=index, + provenance=source_provenance, + execution_key_hash=key, + execution_source="executed", + dedup_source_index=_timeout_index(payload.get("dedup_source_result_index")), + executed_binding_by_index={}, + ) + if source_reason is not None: + continue + identity = (key, _timeout_provenance_identity(source_provenance)) + if index in conflicting_source_indexes: + continue + if executed_binding_by_index.get(index, identity) != identity: + del executed_binding_by_index[index] + conflicting_source_indexes.add(index) + continue + executed_binding_by_index[index] = identity + + timeout_errors = [ + error + for error in errors + if isinstance(error, dict) + and error.get("category") == "timeout" + and error.get("source_check") == "baseline_vc_preflight" + ] + if not timeout_errors: + return None + + occurrences: list[dict] = [] + for error in timeout_errors: + payload = _timeout_error_payload(error) + occurrence_index = _timeout_index(payload.get("occurrence_index")) + line = _timeout_index(error.get("line_start")) + command_hash = _timeout_sha256(payload.get("command_hash")) + execution_key_hash = _timeout_sha256(payload.get("execution_key_hash")) + execution_source = payload.get("execution_source") + # Type first (an unhashable list / dict must not raise `TypeError`). + if not isinstance(execution_source, str) or execution_source not in _TIMEOUT_EXECUTION_SOURCES: + execution_source = None + dedup_source_index = _timeout_index(payload.get("dedup_source_result_index")) + provenance = _timeout_bounded_provenance(payload.get("timeout_provenance")) + + reason = _timeout_unknown_reason( + top_level_digest=top_level_digest, + results_count=results_count, + error_digest=_timeout_sha256(payload.get("canonical_plan_digest")), + occurrence_index=occurrence_index, + provenance=provenance, + execution_key_hash=execution_key_hash, + execution_source=execution_source, + dedup_source_index=dedup_source_index, + executed_binding_by_index=executed_binding_by_index, + ) + if reason is None: + occurrences.append( + { + "attribution": TIMEOUT_ATTRIBUTION_ATTRIBUTED, + "reason_code": TIMEOUT_REASON_BINDING_VERIFIED, + "occurrence_index": occurrence_index, + "line": line, + "line_coordinate": "block_relative", + "command_hash": command_hash, + "execution_key_hash": execution_key_hash, + "execution_source": execution_source, + "dedup_source_result_index": dedup_source_index, + "timeout_provenance": provenance, + } + ) + else: + occurrences.append( + { + "attribution": TIMEOUT_ATTRIBUTION_UNKNOWN, + "reason_code": reason, + "occurrence_index": None, + "line": line, + "line_coordinate": "block_relative", + "command_hash": command_hash, + "execution_key_hash": None, + "execution_source": None, + "dedup_source_result_index": None, + "timeout_provenance": None, + } + ) + + total = len(occurrences) + kept = occurrences[:TIMEOUT_DIAGNOSTICS_MAX_OCCURRENCES] + diagnostics = { + "schema_version": TIMEOUT_DIAGNOSTICS_SCHEMA_VERSION, + "body_sha256": body_sha256, + "canonical_plan_digest": top_level_digest, + "results_count": results_count, + "total_timeout_occurrences": total, + "truncated_count": total - len(kept), + "occurrences": kept, + } + # Structural fields are already bounded; this is a last-resort guard so + # the serialized size limit can never be exceeded. + while kept and len(json.dumps(diagnostics, ensure_ascii=True).encode("utf-8")) > ( + TIMEOUT_DIAGNOSTICS_MAX_SERIALIZED_BYTES + ): + kept.pop() + diagnostics["truncated_count"] = total - len(kept) + return diagnostics + + +def _stdout_bytes_of_review_result(result: dict) -> int: + """Byte size of `result` as the root review child actually writes it: + `print(json.dumps(merged))` (default options, trailing newline, utf-8).""" + return len((json.dumps(result) + "\n").encode("utf-8")) + + +def fit_timeout_diagnostics_to_stdout_budget( + base_result: dict, diagnostics: dict +) -> Optional[dict]: + """Shrink the OPTIONAL `timeout_diagnostics` so the WHOLE merged result + (`base_result` + the diagnostic) stays within the transport stdout cap. + + The per-diagnostic 16 KiB bound is not sufficient: the diagnostic is added + to an already large result, and a stdout above + `TIMEOUT_DIAGNOSTICS_STDOUT_CAP_BYTES` becomes a transport + `capture_failure` (a retry-eligible loss of the whole review result). + Only occurrences are dropped (and `truncated_count` kept consistent with + `total_timeout_occurrences`); `base_result` -- verdict, failure_class, + blockers -- is never touched, and JSON is never cut mid-stream. If even + the header-only diagnostic does not fit, `None` is returned and the + caller omits the optional field (the pre-existing contract shape). + + `base_result` must already be in its final form (every other mutation + applied) so the measured size is the size actually written. + """ + if "timeout_diagnostics" in base_result: + base_result = {k: v for k, v in base_result.items() if k != "timeout_diagnostics"} + total = diagnostics["total_timeout_occurrences"] + all_occurrences = list(diagnostics["occurrences"]) + for keep in range(len(all_occurrences), -1, -1): + candidate = dict(diagnostics) + candidate["occurrences"] = all_occurrences[:keep] + candidate["truncated_count"] = total - keep + trial = dict(base_result) + trial["timeout_diagnostics"] = candidate + if _stdout_bytes_of_review_result(trial) <= TIMEOUT_DIAGNOSTICS_STDOUT_CAP_BYTES: + return candidate + return None + + def merge_readiness_into_review_result( review_result: dict, readiness_result: dict, @@ -626,6 +981,7 @@ def merge_readiness_into_review_result( routed into `non_blocking_improvements`. """ merged = json.loads(json.dumps(review_result)) + timeout_diagnostics: Optional[dict] = None review_body_sha256 = merged.get("body_sha256") readiness_body_sha256 = readiness_result.get("body_sha256") readiness_errors = readiness_result.get("errors") or [] @@ -691,6 +1047,17 @@ def merge_readiness_into_review_result( failure_class = readiness_status_to_failure_class(readiness_status) if failure_class: merged["failure_class"] = failure_class + # Issue #2897: bounded timeout occurrence / applied-budget + # identity. Additive optional field only -- no blocker, no + # failure_class / route change. Reached only after the + # body_sha256 fail-closed check above, so a mismatched + # readiness result never produces a diagnostic. + # Attached only after every other mutation below, and fitted + # to the whole-result stdout budget (see + # `fit_timeout_diagnostics_to_stdout_budget()`). + timeout_diagnostics = build_timeout_diagnostics( + readiness_result, body_sha256=review_body_sha256 + ) # `compact_review_result.py` checks `verdict == "approve"` first # and short-circuits to `NEXT_ACTION: proceed` before it ever # looks at `failure_class` (Issue #1791 review remediation @@ -706,6 +1073,11 @@ def merge_readiness_into_review_result( elif new_blockers: merged["verdict"] = "needs-fix" + if timeout_diagnostics is not None: + fitted = fit_timeout_diagnostics_to_stdout_budget(merged, timeout_diagnostics) + if fitted is not None: + merged["timeout_diagnostics"] = fitted + _validate_review_issue_result_payload(merged) return merged diff --git a/.claude/skills/review-issue/tests/test_timeout_diagnostic_projection.py b/.claude/skills/review-issue/tests/test_timeout_diagnostic_projection.py new file mode 100644 index 000000000..baf44c76c --- /dev/null +++ b/.claude/skills/review-issue/tests/test_timeout_diagnostic_projection.py @@ -0,0 +1,774 @@ +"""Issue #2897 AC1/AC2 (review merge side): `check_issue_contract.py +--mode merge_readiness` projects a bounded `timeout_diagnostics` +(`TIMEOUT_DIAGNOSTICS_V1`) from a readiness `human_judgment` produced by a +command-level VC timeout, without fabricating a deterministic blocker and +without changing `failure_class` / the operator-only route. + +Every test goes through the production conversion chain; no completed +`timeout_provenance` / readiness / review dict is hand-built: + + fake `baseline_vc_preflight.run_command()` return value (permitted seam 1, + timeout sentinel `exit_code == -1` and `stderr == "timeout"`) + -> REAL `baseline_vc_preflight.main()` result builder + -> REAL `contract_readiness_check.main()` conversion + -> REAL `check_issue_contract.main()` review check (`--file --json`) + -> REAL `check_issue_contract.main()` `--mode merge_readiness` + +Permitted seam 2 (process-launch mechanics only): the cooperative supervisor +that would spawn `baseline_vc_preflight.py` as a child process is replaced by +an adapter calling that SAME script's `main()` in-process. + +The "unknown attribution" cases degrade an otherwise real artifact by +REMOVING or TAMPERING a binding field (a legacy / partial producer, or a +corrupted file between readiness and merge). They never inject a completed +`timeout_provenance`. +""" + +from __future__ import annotations + +import contextlib +import io +import json +import signal +import sys +from pathlib import Path +from unittest import mock + +import pytest + +_REVIEW_SCRIPTS = Path(__file__).resolve().parent.parent / "scripts" +_CONTRACT_REVIEW_SCRIPTS = Path(__file__).resolve().parents[2] / "issue-contract-review" / "scripts" +for _path in (_REVIEW_SCRIPTS, _CONTRACT_REVIEW_SCRIPTS): + if str(_path) not in sys.path: + sys.path.insert(0, str(_path)) + +import baseline_vc_preflight as bvp # noqa: E402 +import check_issue_contract as cic # noqa: E402 +import contract_readiness_check as crc # noqa: E402 + +TIMEOUT_OUTCOME = (-1, "", "timeout", 1234, {}) +NOT_FOUND_OUTCOME = (4, "", "ERROR: file or directory not found: x", 5, {}) + +_NEW_TEST_PATH = ".claude/skills/review-issue/tests/test_fixture_target_not_yet_created.py" +_PYTEST_VC = f"uv run --locked pytest {_NEW_TEST_PATH}::test_target" +_PURE_VC = "test -f README.md" +_OTHER_DIGEST = "sha256:" + "ab" * 32 + +_BODY_HEADER = """## Machine-Readable Contract + +```yaml +contract_schema_version: v1 +issue_kind: implementation +parent_issue: none +goal_ref: "timeout diagnostic projection fixture" +change_kind: workflow +``` + +## Outcome + +Fixture for the timeout diagnostic projection. + +## Acceptance Criteria + +- [ ] AC1: the projection keeps the timed-out occurrence identity. +- [ ] AC2: the projection never attributes a timeout to another occurrence. + +## Verification Commands + +""" + +_ALLOWED = f""" +## Allowed Paths + +- {_NEW_TEST_PATH} +""" + +# Two fenced blocks holding the SAME AC and the SAME command at the SAME +# block-relative line (`line + command_hash` would collide), preceded by one +# unrelated pure command so the canonical indexes are not trivially 0/1. +_TWO_BLOCK_BODY = ( + _BODY_HEADER + + f"""```bash +# AC1 +$ {_PURE_VC} +``` + +```bash +# AC1 +# baseline-expect: fail +$ {_PYTEST_VC} +``` + +```bash +# AC1 +# baseline-expect: fail +$ {_PYTEST_VC} +``` +""" + + _ALLOWED +) + +# Two pure identical commands: occurrence 0 is the real execution and +# occurrence 1 is its dedup replay. +_DEDUP_BODY = ( + _BODY_HEADER + + f"""```bash +# AC2 +$ {_PURE_VC} +``` + +```bash +# AC2 +$ {_PURE_VC} +``` +""" + + _ALLOWED +) + + +class _RunCommandSeam: + """Seam 1: replaces `baseline_vc_preflight.run_command()` only.""" + + def __init__(self, outcomes_by_call: dict[int, tuple]): + self._outcomes_by_call = outcomes_by_call + self.calls: list[tuple[str, int]] = [] + + def __call__(self, command: str, timeout_seconds: int, cwd: str): + call_index = len(self.calls) + self.calls.append((command, timeout_seconds)) + return self._outcomes_by_call.get(call_index, NOT_FOUND_OUTCOME) + + +class _InProcessBaselineLauncher: + """Seam 2: launch mechanics only. `degrade` (optional) removes / alters a + binding field of the REAL preflight payload to model a legacy or partial + producer; it is never used to supply a completed provenance.""" + + def __init__(self, degrade=None): + self._degrade = degrade + self.raw_payloads: list[dict] = [] + + def __call__(self, argv, *, timeout_seconds, cwd=None, env=None, **_ignored): + assert Path(argv[1]).name == "baseline_vc_preflight.py", argv + out, err = io.StringIO(), io.StringIO() + previous_sigterm = signal.getsignal(signal.SIGTERM) + try: + with mock.patch.object(sys, "argv", [argv[1], *argv[2:]]): + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + returncode = bvp.main() + finally: + signal.signal(signal.SIGTERM, previous_sigterm) + payload = json.loads(out.getvalue()) + self.raw_payloads.append(json.loads(out.getvalue())) + stdout_text = out.getvalue() + if self._degrade is not None: + self._degrade(payload) + stdout_text = json.dumps(payload) + return bvp.SupervisedSubprocessResult(returncode, stdout_text, err.getvalue(), False, 0.0) + + +def _run_main(module_main, argv: list[str]) -> tuple[int, str]: + out = io.StringIO() + code = 0 + with mock.patch.object(sys, "argv", argv): + with contextlib.redirect_stdout(out): + try: + code = module_main() or 0 + except SystemExit as exc: # check_issue_contract.main() exits + code = exc.code if isinstance(exc.code, int) else 1 + return code, out.getvalue() + + +def _readiness_result( + monkeypatch, tmp_path: Path, body: str, outcomes_by_call: dict[int, tuple], degrade=None +) -> tuple[dict, _InProcessBaselineLauncher]: + launcher = _InProcessBaselineLauncher(degrade) + monkeypatch.setattr(bvp, "run_command", _RunCommandSeam(outcomes_by_call)) + monkeypatch.setattr(crc, "_run_subprocess_with_cooperative_supervisor", launcher) + body_file = tmp_path / "body.md" + body_file.write_text(body, encoding="utf-8") + _code, stdout = _run_main( + crc.main, ["contract_readiness_check.py", "--body-file", str(body_file), "--mode", "execute"] + ) + return json.loads(stdout), launcher + + +def _review_result(tmp_path: Path, body: str) -> dict: + body_file = tmp_path / "body.md" + body_file.write_text(body, encoding="utf-8") + _code, stdout = _run_main( + cic.main, ["check_issue_contract.py", "--file", str(body_file), "--json"] + ) + return json.loads(stdout) + + +def _merge(tmp_path: Path, review: dict, readiness: dict) -> tuple[int, dict | None]: + review_file = tmp_path / "review_result.json" + readiness_file = tmp_path / "readiness_result.json" + output_file = tmp_path / "merged_review_result.json" + review_file.write_text(json.dumps(review), encoding="utf-8") + readiness_file.write_text(json.dumps(readiness), encoding="utf-8") + if output_file.exists(): + output_file.unlink() + code, _stdout = _run_main( + cic.main, + [ + "check_issue_contract.py", + "--mode", "merge_readiness", + "--review-result-file", str(review_file), + "--readiness-result-file", str(readiness_file), + "--readiness-artifact-path", str(readiness_file), + "--iteration-id", "timeout_diagnostic_projection_test", + "--output-file", str(output_file), + ], + ) + merged = json.loads(output_file.read_text(encoding="utf-8")) if output_file.exists() else None + return code, merged + + +def _full_chain(monkeypatch, tmp_path, body, outcomes_by_call, degrade=None, tamper_readiness=None): + readiness, launcher = _readiness_result(monkeypatch, tmp_path, body, outcomes_by_call, degrade) + if tamper_readiness is not None: + tamper_readiness(readiness) + review = _review_result(tmp_path, body) + code, merged = _merge(tmp_path, review, readiness) + return review, readiness, merged, code, launcher + + +def test_inner_timeout_preserves_bounded_identity_without_blocker(monkeypatch, tmp_path): + # call 0 = pure command, call 1 = first pytest block, call 2 = second + # pytest block (times out). + review, readiness, merged, code, launcher = _full_chain( + monkeypatch, tmp_path, _TWO_BLOCK_BODY, {2: TIMEOUT_OUTCOME} + ) + raw = launcher.raw_payloads[0] + + # Existing semantics are unchanged: human_judgment failure_class, a + # needs-fix verdict and the timeout string in blocking_issues; no + # deterministic blocker is fabricated for the timeout. + assert readiness["status"] == "human_judgment" + assert merged is not None and code == 1 + assert merged["failure_class"] == "contract_readiness_human_judgment" + assert merged["verdict"] == "needs-fix" + assert "Command exceeded timeout" in merged["blocking_issues"] + assert merged["structured_blockers"] == review["structured_blockers"] + assert not any( + "timeout" in json.dumps(blocker).lower() for blocker in merged["structured_blockers"] + ) + + diagnostics = merged["timeout_diagnostics"] + assert diagnostics["schema_version"] == "TIMEOUT_DIAGNOSTICS_V1" + assert diagnostics["body_sha256"] == review["body_sha256"] == readiness["body_sha256"] + assert diagnostics["canonical_plan_digest"] == raw["diagnostic_report"]["canonical_plan_digest"] + assert diagnostics["results_count"] == len(raw["results"]) == 3 + assert diagnostics["total_timeout_occurrences"] == 1 + assert diagnostics["truncated_count"] == 0 + + (occurrence,) = diagnostics["occurrences"] + raw_item = raw["results"][2] + assert occurrence["attribution"] == "attributed" + assert occurrence["reason_code"] == "binding_verified" + assert occurrence["occurrence_index"] == 2 + assert occurrence["line"] == raw_item["line"] + assert occurrence["line_coordinate"] == "block_relative" + assert occurrence["command_hash"] == raw_item["command_hash"] + assert occurrence["execution_key_hash"] == raw_item["execution_key_hash"] + assert occurrence["execution_source"] == "executed" + assert occurrence["dedup_source_result_index"] is None + assert occurrence["timeout_provenance"] == raw_item["timeout_provenance"] + + # Bounded: no raw command / output / environment in the diagnostic. + serialized = json.dumps(diagnostics) + assert _PYTEST_VC not in serialized and _PURE_VC not in serialized + assert "runner_env_delta" not in serialized and "minimal_context" not in serialized + assert len(serialized.encode("utf-8")) <= 16 * 1024 + + +@pytest.mark.parametrize("timed_out_call, expected_index", [(2, 2), (1, 1)]) +def test_same_hash_same_line_other_block_only_timed_out_occurrence_attributed( + monkeypatch, tmp_path, timed_out_call, expected_index +): + _review, _readiness, merged, _code, launcher = _full_chain( + monkeypatch, tmp_path, _TWO_BLOCK_BODY, {timed_out_call: TIMEOUT_OUTCOME} + ) + raw_results = launcher.raw_payloads[0]["results"] + # Both pytest blocks collide on (ac, line, command_hash)... + assert raw_results[1]["command_hash"] == raw_results[2]["command_hash"] + assert raw_results[1]["line"] == raw_results[2]["line"] + assert raw_results[1]["ac"] == raw_results[2]["ac"] + + # ...but only the block that really timed out is attributed, by index. + diagnostics = merged["timeout_diagnostics"] + assert [o["occurrence_index"] for o in diagnostics["occurrences"]] == [expected_index] + assert diagnostics["total_timeout_occurrences"] == 1 + (occurrence,) = diagnostics["occurrences"] + assert occurrence["attribution"] == "attributed" + assert occurrence["execution_key_hash"] == raw_results[expected_index]["execution_key_hash"] + other = 1 if expected_index == 2 else 2 + assert occurrence["execution_key_hash"] != raw_results[other]["execution_key_hash"] + + +def test_dedup_replay_is_distinguished_from_real_execution(monkeypatch, tmp_path): + _review, _readiness, merged, _code, launcher = _full_chain( + monkeypatch, tmp_path, _DEDUP_BODY, {0: TIMEOUT_OUTCOME} + ) + raw_results = launcher.raw_payloads[0]["results"] + assert raw_results[1]["runner"] == "dedup_replay" + + occurrences = merged["timeout_diagnostics"]["occurrences"] + assert [o["occurrence_index"] for o in occurrences] == [0, 1] + assert [o["execution_source"] for o in occurrences] == ["executed", "dedup_replay"] + assert [o["dedup_source_result_index"] for o in occurrences] == [None, 0] + assert all(o["attribution"] == "attributed" for o in occurrences) + assert occurrences[0]["execution_key_hash"] == occurrences[1]["execution_key_hash"] + + +def _drop_provenance(payload): + for item in payload["results"]: + item.pop("timeout_provenance", None) + + +def _drop_execution_key(payload): + for item in payload["results"]: + item["execution_key_hash"] = None + + +def _not_computed_digest(payload): + payload["diagnostic_report"] = bvp.not_computed_diagnostic_report() + + +def _bad_dedup_source_out_of_range(payload): + for item in payload["results"]: + if item.get("dedup"): + item["dedup"]["source_result_index"] = 99 + + +def _bad_dedup_source_wrong_target(payload): + # A replay can only point at an earlier real execution; point it at itself. + for index, item in enumerate(payload["results"]): + if item.get("dedup"): + item["dedup"]["source_result_index"] = index + + +def _tamper_digest(readiness): + readiness["canonical_plan_digest"] = _OTHER_DIGEST + + +def _tamper_results_count(readiness): + readiness["results_count"] = 1 + + +_ONE_TIMEOUT = {2: TIMEOUT_OUTCOME} +_REPLAY_TIMEOUT = {0: TIMEOUT_OUTCOME} + +# (id, body, outcomes, degrade-preflight, tamper-readiness, expected reason) +_UNKNOWN_CASES = [ + ("plan_digest_missing", _TWO_BLOCK_BODY, _ONE_TIMEOUT, _not_computed_digest, None, + "plan_digest_missing"), + ("plan_digest_mismatch", _TWO_BLOCK_BODY, _ONE_TIMEOUT, None, _tamper_digest, + "plan_digest_mismatch"), + ("provenance_missing", _TWO_BLOCK_BODY, _ONE_TIMEOUT, _drop_provenance, None, + "provenance_missing"), + ("execution_key_missing", _TWO_BLOCK_BODY, _ONE_TIMEOUT, _drop_execution_key, None, + "execution_key_missing"), + ("dedup_source_out_of_range", _DEDUP_BODY, _REPLAY_TIMEOUT, _bad_dedup_source_out_of_range, + None, "dedup_binding_invalid"), + ("dedup_source_wrong_target", _DEDUP_BODY, _REPLAY_TIMEOUT, _bad_dedup_source_wrong_target, + None, "dedup_binding_invalid"), + ("occurrence_index_out_of_range", _TWO_BLOCK_BODY, _ONE_TIMEOUT, None, _tamper_results_count, + "occurrence_index_out_of_range"), +] + + +@pytest.mark.parametrize( + "body, outcomes, degrade, tamper, expected_reason", + [case[1:] for case in _UNKNOWN_CASES], + ids=[case[0] for case in _UNKNOWN_CASES], +) +def test_mismatch_or_missing_binding_records_unknown_without_new_gate( + monkeypatch, tmp_path, body, outcomes, degrade, tamper, expected_reason +): + review, readiness, merged, code, _launcher = _full_chain( + monkeypatch, tmp_path, body, outcomes, degrade=degrade, tamper_readiness=tamper + ) + + # No new gate: the merge succeeds with the very same routing facts as a + # fully bound timeout (human_judgment failure_class, needs-fix verdict, + # timeout string only in blocking_issues, no deterministic blocker). + assert merged is not None and code == 1 + assert readiness["status"] == "human_judgment" + assert merged["failure_class"] == "contract_readiness_human_judgment" + assert merged["verdict"] == "needs-fix" + assert "Command exceeded timeout" in merged["blocking_issues"] + assert merged["structured_blockers"] == review["structured_blockers"] + + diagnostics = merged["timeout_diagnostics"] + assert diagnostics["total_timeout_occurrences"] >= 1 + occurrences = diagnostics["occurrences"] + if expected_reason == "dedup_binding_invalid": + # Only the replay's binding is broken: the real execution (index 0) + # stays attributed, and the replay is not guessed onto it. + assert [o["attribution"] for o in occurrences] == ["attributed", "unknown"] + assert occurrences[0]["occurrence_index"] == 0 + occurrences = occurrences[1:] + for occurrence in occurrences: + assert occurrence["attribution"] == "unknown" + assert occurrence["reason_code"] == expected_reason + # Unknown never carries completed identity / budget for a guessed VC. + assert occurrence["occurrence_index"] is None + assert occurrence["timeout_provenance"] is None + assert occurrence["execution_key_hash"] is None + assert expected_reason in cic.TIMEOUT_UNKNOWN_REASON_CODES + + +def test_body_sha_mismatch_still_fails_closed_without_diagnostic(monkeypatch, tmp_path): + readiness, _launcher = _readiness_result( + monkeypatch, tmp_path, _TWO_BLOCK_BODY, {2: TIMEOUT_OUTCOME} + ) + review = _review_result(tmp_path, _TWO_BLOCK_BODY) + readiness["body_sha256"] = _OTHER_DIGEST + + with pytest.raises(ValueError, match="body_sha256 mismatch"): + cic.merge_readiness_into_review_result( + review, + readiness, + readiness_artifact_path="readiness_result.json", + iteration_id="timeout_diagnostic_projection_test", + ) + code, merged = _merge(tmp_path, review, readiness) + assert code == 1 + assert merged is None + + +def test_non_timeout_human_judgment_gets_no_timeout_diagnostics(monkeypatch, tmp_path): + # A non-timeout inner failure (exit 1, no recognised category) is a + # human_judgment too; it must not produce `timeout_diagnostics`. + _review, readiness, merged, _code, _launcher = _full_chain( + monkeypatch, tmp_path, _TWO_BLOCK_BODY, {2: (1, "", "boom", 9, {})} + ) + assert merged is not None + assert "timeout_diagnostics" not in merged + assert not any(error["category"] == "timeout" for error in readiness["errors"]) + + +def test_more_than_sixteen_timeouts_are_truncated_within_size_bound(): + # Pure projection bound check on a readiness result with 20 timeout + # errors whose bindings are all valid (no routing involved). + digest = "sha256:" + "11" * 32 + errors = [] + for index in range(20): + errors.append( + { + "rule_id": "VCP_TIMEOUT", + "category": "timeout", + "source_check": "baseline_vc_preflight", + "line_start": 3, + "source_payload": { + "occurrence_index": index, + "command_hash": "sha256:" + "22" * 32, + "execution_key_hash": "sha256:" + f"{index:02x}" * 32, + "execution_source": "executed", + "dedup_source_result_index": None, + "canonical_plan_digest": digest, + "timeout_provenance": { + "timeout_seconds": 150, + "cleanup_tail_seconds": 15, + "source": "static_fallback", + "estimator_version": "v2", + "estimator_input_digest": "sha256:" + "33" * 32, + }, + }, + } + ) + readiness = {"errors": errors, "canonical_plan_digest": digest, "results_count": 20} + diagnostics = cic.build_timeout_diagnostics(readiness, body_sha256="sha256:" + "44" * 32) + assert len(diagnostics["occurrences"]) == 16 + assert diagnostics["truncated_count"] == 4 + assert diagnostics["total_timeout_occurrences"] == 20 + assert len(json.dumps(diagnostics).encode("utf-8")) <= 16 * 1024 + + +# --------------------------------------------------------------------------- +# PR #2901 OWNER review fix_delta (Issue #2897) +# --------------------------------------------------------------------------- + +_STDOUT_CAP = 65_536 # reviewer_transport.STDOUT_CAP (pinned in the root test) + + +def _writer_bytes(result: dict) -> int: + """Size of `result` as the root review child really writes it: + `print(json.dumps(merged))` -> default options, trailing newline, utf-8.""" + return len((json.dumps(result) + "\n").encode("utf-8")) + + +def _large_review_body(last_command_padding: int = 1450) -> str: + # 30 long pure VCs (a large `parsed_vc_commands`, hence a large review + # result). The last one's length is the fine-tuning knob (1 byte / char). + blocks = [] + for index in range(30): + padding = last_command_padding if index == 29 else 1450 + blocks.append( + f"```bash\n# AC{(index % 2) + 1}\n$ test -f d{index}/{'a' * padding}.md\n```\n\n" + ) + return _BODY_HEADER + "".join(blocks) + _ALLOWED + + +_TWENTY_TIMEOUTS = {call: TIMEOUT_OUTCOME for call in range(20)} + + +def _without_diagnostics(merged: dict) -> dict: + return {key: value for key, value in merged.items() if key != "timeout_diagnostics"} + + +def _large_chain(monkeypatch, tmp_path, last_command_padding: int = 1450): + body = _large_review_body(last_command_padding) + review, readiness, merged, code, _launcher = _full_chain( + monkeypatch, tmp_path, body, _TWENTY_TIMEOUTS + ) + assert merged is not None and code == 1 + return body, review, readiness, merged + + +def _assert_routing_unchanged(review: dict, merged: dict) -> None: + assert merged["verdict"] == "needs-fix" + assert merged["failure_class"] == "contract_readiness_human_judgment" + assert "Command exceeded timeout" in merged["blocking_issues"] + assert merged["structured_blockers"] == review["structured_blockers"] + assert merged["parsed_vc_commands"] == review["parsed_vc_commands"] + + +def test_whole_result_stdout_budget_shrinks_only_the_diagnostic(monkeypatch, tmp_path): + body, review, readiness, merged = _large_chain(monkeypatch, tmp_path) + base = _without_diagnostics(merged) + full_diagnostics = cic.build_timeout_diagnostics( + readiness, body_sha256=review["body_sha256"] + ) + + # (1) Without the diagnostic the real stdout fits the transport cap. + assert _writer_bytes(base) <= _STDOUT_CAP + # The diagnostic alone is within its own 16 KiB bound... + assert len(json.dumps(full_diagnostics).encode("utf-8")) <= 16 * 1024 + assert len(full_diagnostics["occurrences"]) == 16 + # (2) ...but naively attaching all of it overflows the WHOLE result. + assert _writer_bytes(dict(base, timeout_diagnostics=full_diagnostics)) > _STDOUT_CAP + + # (3) After the fix the production merge result fits. + assert _writer_bytes(merged) <= _STDOUT_CAP + # (4) Routing-critical content is exactly the diagnostic-free content. + _assert_routing_unchanged(review, merged) + assert cic.TIMEOUT_DIAGNOSTICS_STDOUT_CAP_BYTES == _STDOUT_CAP + + # (5) Only the optional diagnostic was shrunk, consistently. + diagnostics = merged["timeout_diagnostics"] + kept = diagnostics["occurrences"] + assert 0 < len(kept) < 16 + assert kept == full_diagnostics["occurrences"][: len(kept)] + assert diagnostics["total_timeout_occurrences"] == 20 + assert diagnostics["truncated_count"] == 20 - len(kept) + assert diagnostics["body_sha256"] == review["body_sha256"] + # Not shrunk more than necessary: one more occurrence would overflow. + one_more = dict(diagnostics, occurrences=full_diagnostics["occurrences"][: len(kept) + 1]) + assert _writer_bytes(dict(base, timeout_diagnostics=one_more)) > _STDOUT_CAP + + +def test_header_only_diagnostic_boundary_then_optional_field_omitted(monkeypatch, tmp_path): + body, review, readiness, merged = _large_chain(monkeypatch, tmp_path) + base_size = _writer_bytes(_without_diagnostics(merged)) + diagnostics = merged["timeout_diagnostics"] + header_only = dict(diagnostics, occurrences=[], truncated_count=20) + overhead = _writer_bytes(dict(_without_diagnostics(merged), timeout_diagnostics=header_only)) - base_size + + # The base result grows 1 byte per padding char: leave exactly `overhead` + # bytes (header-only just fits) and `overhead - 1` bytes (it does not). + fits_padding = 1450 + (_STDOUT_CAP - overhead - base_size) + _body, review_fit, _readiness, merged_fit = _large_chain(monkeypatch, tmp_path, fits_padding) + assert _writer_bytes(_without_diagnostics(merged_fit)) == _STDOUT_CAP - overhead + assert merged_fit["timeout_diagnostics"]["occurrences"] == [] + assert merged_fit["timeout_diagnostics"]["total_timeout_occurrences"] == 20 + assert merged_fit["timeout_diagnostics"]["truncated_count"] == 20 + assert _writer_bytes(merged_fit) == _STDOUT_CAP + _assert_routing_unchanged(review_fit, merged_fit) + + _body, review_omit, _readiness, merged_omit = _large_chain( + monkeypatch, tmp_path, fits_padding + 1 + ) + assert _writer_bytes(_without_diagnostics(merged_omit)) == _STDOUT_CAP - overhead + 1 + # Optional field omitted (pre-existing contract shape); the review result + # as a whole is NOT lost / not turned into a capture_failure. + assert "timeout_diagnostics" not in merged_omit + assert _writer_bytes(merged_omit) <= _STDOUT_CAP + _assert_routing_unchanged(review_omit, merged_omit) + + +# --- Finding B: malformed (unhashable) enum-like values -> `unknown` -------- + +_MALFORMED_ENUM_VALUES = [[], {}, ["static_policy"], {"k": "v"}] + + +def _timeout_error(readiness: dict, occurrence_index: int) -> dict: + (error,) = [ + e + for e in readiness["errors"] + if e.get("category") == "timeout" + and (e.get("source_payload") or {}).get("occurrence_index") == occurrence_index + ] + return error + + +@pytest.mark.parametrize("bad", _MALFORMED_ENUM_VALUES, ids=repr) +@pytest.mark.parametrize( + "field_path, expected_reason", + [ + (("timeout_provenance", "source"), "provenance_missing"), + (("timeout_provenance", "estimator_version"), "provenance_missing"), + (("timeout_provenance", "estimator_input_digest"), "provenance_missing"), + (("timeout_provenance", "timeout_seconds"), "provenance_missing"), + (("timeout_provenance", "cleanup_tail_seconds"), "provenance_missing"), + (("timeout_provenance",), "provenance_missing"), + (("execution_source",), "dedup_binding_invalid"), + (("execution_key_hash",), "execution_key_missing"), + (("canonical_plan_digest",), "plan_digest_mismatch"), + (("occurrence_index",), "occurrence_index_out_of_range"), + ], + ids=lambda value: "/".join(value) if isinstance(value, tuple) else None, +) +def test_unhashable_enum_like_values_degrade_to_unknown_without_exception( + monkeypatch, tmp_path, field_path, expected_reason, bad +): + def tamper(readiness): + payload = _timeout_error(readiness, 2)["source_payload"] + target = payload + for key in field_path[:-1]: + target = target[key] + target[field_path[-1]] = bad + + review, readiness, merged, code, _launcher = _full_chain( + monkeypatch, tmp_path, _TWO_BLOCK_BODY, {2: TIMEOUT_OUTCOME}, tamper_readiness=tamper + ) + # The merge completes with the very same routing facts. + assert merged is not None and code == 1 + _assert_routing_unchanged(review, merged) + (occurrence,) = merged["timeout_diagnostics"]["occurrences"] + assert occurrence["attribution"] == "unknown" + assert occurrence["reason_code"] == expected_reason + assert expected_reason in cic.TIMEOUT_UNKNOWN_REASON_CODES + assert occurrence["occurrence_index"] is None + assert occurrence["timeout_provenance"] is None + assert occurrence["execution_key_hash"] is None + + +@pytest.mark.parametrize("bad", _MALFORMED_ENUM_VALUES, ids=repr) +def test_unhashable_provenance_source_in_real_preflight_payload_degrades( + monkeypatch, tmp_path, bad +): + # Producer boundary, through the REAL conversion: a (legacy / corrupted) + # preflight item with a list / dict `source` must neither raise in the + # readiness producer nor in the review merge. + def degrade(payload): + for item in payload["results"]: + if isinstance(item.get("timeout_provenance"), dict): + item["timeout_provenance"]["source"] = bad + + review, readiness, merged, code, _launcher = _full_chain( + monkeypatch, tmp_path, _TWO_BLOCK_BODY, {2: TIMEOUT_OUTCOME}, degrade=degrade + ) + assert merged is not None and code == 1 + _assert_routing_unchanged(review, merged) + (occurrence,) = merged["timeout_diagnostics"]["occurrences"] + assert occurrence["attribution"] == "unknown" + assert occurrence["reason_code"] == "provenance_missing" + + +@pytest.mark.parametrize("bad", _MALFORMED_ENUM_VALUES, ids=repr) +def test_consumer_bounded_provenance_helper_never_raises(bad): + good = { + "timeout_seconds": 150, + "cleanup_tail_seconds": 15, + "source": "static_fallback", + "estimator_version": "v2", + "estimator_input_digest": "sha256:" + "33" * 32, + } + assert cic._timeout_bounded_provenance(good) == good + for key in good: + assert cic._timeout_bounded_provenance(dict(good, **{key: bad})) is None + + +# --- Finding C: dedup replay bound to a verified source -------------------- + + +def _two_occurrence_chain(monkeypatch, tmp_path, tamper=None, degrade=None): + # _DEDUP_BODY: occurrence 0 real execution (timeout), 1 its dedup replay. + review, readiness, merged, code, launcher = _full_chain( + monkeypatch, tmp_path, _DEDUP_BODY, {0: TIMEOUT_OUTCOME}, + degrade=degrade, tamper_readiness=tamper, + ) + assert merged is not None and code == 1 + _assert_routing_unchanged(review, merged) + first, second, *_extra = merged["timeout_diagnostics"]["occurrences"] + return readiness, first, second, launcher + + +def test_dedup_pair_positive_control_is_attributed(monkeypatch, tmp_path): + _readiness, first, second, _launcher = _two_occurrence_chain(monkeypatch, tmp_path) + assert (first["attribution"], first["reason_code"]) == ("attributed", "binding_verified") + assert (second["attribution"], second["reason_code"]) == ("attributed", "binding_verified") + assert second["dedup_source_result_index"] == 0 + assert second["timeout_provenance"] == first["timeout_provenance"] + + +def test_invalid_source_occurrence_does_not_lend_binding_to_its_replay(monkeypatch, tmp_path): + def tamper(readiness): + # Only the SOURCE occurrence's plan digest disagrees with the top level. + _timeout_error(readiness, 0)["source_payload"]["canonical_plan_digest"] = _OTHER_DIGEST + + _readiness, first, second, _launcher = _two_occurrence_chain(monkeypatch, tmp_path, tamper) + assert (first["attribution"], first["reason_code"]) == ("unknown", "plan_digest_mismatch") + assert (second["attribution"], second["reason_code"]) == ("unknown", "dedup_binding_invalid") + assert second["timeout_provenance"] is None and second["execution_key_hash"] is None + + +def test_source_without_provenance_in_real_payload_makes_replay_unknown(monkeypatch, tmp_path): + def degrade(payload): + # Real preflight payload, source item only loses its provenance. + payload["results"][0].pop("timeout_provenance", None) + + _readiness, first, second, _launcher = _two_occurrence_chain( + monkeypatch, tmp_path, degrade=degrade + ) + assert (first["attribution"], first["reason_code"]) == ("unknown", "provenance_missing") + assert (second["attribution"], second["reason_code"]) == ("unknown", "dedup_binding_invalid") + + +@pytest.mark.parametrize( + "field, value", + [ + ("timeout_seconds", 300), + ("cleanup_tail_seconds", 77), + ("source", "explicit_override"), + ("estimator_version", "v999"), + ("estimator_input_digest", "sha256:" + "ee" * 32), + ], +) +def test_replay_with_contradicting_budget_provenance_is_unknown(monkeypatch, tmp_path, field, value): + def tamper(readiness): + provenance = _timeout_error(readiness, 1)["source_payload"]["timeout_provenance"] + assert provenance[field] != value + provenance[field] = value + + _readiness, first, second, _launcher = _two_occurrence_chain(monkeypatch, tmp_path, tamper) + # The source stays attributed (its own binding is intact); only the + # contradicting replay degrades -- no budget is recomputed or guessed. + assert (first["attribution"], first["reason_code"]) == ("attributed", "binding_verified") + assert (second["attribution"], second["reason_code"]) == ("unknown", "dedup_binding_invalid") + assert second["timeout_provenance"] is None + + +def test_conflicting_duplicate_source_index_is_not_a_verified_source(monkeypatch, tmp_path): + def tamper(readiness): + # A second error claims index 0 as `executed` with another key. + clone = json.loads(json.dumps(_timeout_error(readiness, 0))) + clone["source_payload"]["execution_key_hash"] = "sha256:" + "dd" * 32 + readiness["errors"].append(clone) + + _readiness, _first, second, _launcher = _two_occurrence_chain(monkeypatch, tmp_path, tamper) + assert (second["attribution"], second["reason_code"]) == ("unknown", "dedup_binding_invalid")