From 85575223088cbe3889f769ae53ddc5d34fc17e34 Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Thu, 11 Jun 2026 10:25:39 -0700 Subject: [PATCH 1/3] fix(orchestrator): evidence-reachability gate at slice close (#3125) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The slice integration branch only advances when a producer pushes (consensus_push at propose time). A commit recorded by the prescribed post-confirmation unblock flow (egg-contract complete-task --commit, #3124) lives only on the agent's local worktree branch, so a slice could reach consensus, open its PR, and be marked complete while a contract task record cites a commit the PR does not contain — and worktree cleanup can then prune the only copy. Close the gap with proposal 2 from #3125: before any close side effect (BRC transcript commit, slice PR, record_complete), verify every commit SHA cited by the closing slice's task records is an ancestor of the integration branch tip. A definitive miss fails the slice with a message listing the lost rows, routing through the existing cascade + HITL escalation machinery instead of closing silently. - contract_completeness.py: evidence_commits/format_evidence_rows pure helpers + independent EGG_EVIDENCE_REACHABILITY_GATE kill switch - gateway_client.py: find_unreachable_evidence_commits — one synthetic session, ls-remote tip + fetch, then merge-base --is-ancestor per SHA; tri-state mapping (exit 1/128 unreachable, anything else skips the gate so a transient gateway failure cannot fail a slice) - routes/pipelines.py: _check_slice_evidence_reachability wired into the slice run loop after consensus, before close side effects; degrades gracefully on contract-read/probe failures (#3081/#3114 posture) --- orchestrator/contract_completeness.py | 54 ++ orchestrator/gateway_client.py | 146 +++++- orchestrator/routes/pipelines.py | 128 +++++ .../tests/test_evidence_reachability_gate.py | 465 ++++++++++++++++++ 4 files changed, 792 insertions(+), 1 deletion(-) create mode 100644 orchestrator/tests/test_evidence_reachability_gate.py diff --git a/orchestrator/contract_completeness.py b/orchestrator/contract_completeness.py index cac78a6662..ccfad6e69e 100644 --- a/orchestrator/contract_completeness.py +++ b/orchestrator/contract_completeness.py @@ -28,6 +28,11 @@ to be pending, and the apply phase (#1557) tracks per-task lifecycle in ``jira_action_status`` instead. +#3125 extends the module with the slice-close evidence-reachability +helpers (``evidence_commits`` / ``evidence_gate_enabled``): commit SHAs +cited by task records are only an integrity contract if the close path +verifies they actually reached the integration branch. + Failure posture: the gate degrades gracefully (returns ``None`` / skips) when the contract cannot be loaded or the slice id does not resolve — an orchestrator-side infrastructure failure must not deadlock @@ -56,6 +61,11 @@ # "off" (or 0/false/no) to disable enforcement without a redeploy. GATE_ENV_VAR = "EGG_CONTRACT_ACK_GATE" +# Operator kill switch for the slice-close evidence-reachability gate +# (#3125). Separate from the ACK/CONFIRM gate so each can be toggled +# independently during an incident. +EVIDENCE_GATE_ENV_VAR = "EGG_EVIDENCE_REACHABILITY_GATE" + _DISABLED_VALUES = frozenset({"off", "0", "false", "no"}) @@ -64,6 +74,11 @@ def gate_enabled() -> bool: return os.environ.get(GATE_ENV_VAR, "on").strip().lower() not in _DISABLED_VALUES +def evidence_gate_enabled() -> bool: + """Return True unless the evidence-reachability kill switch is set.""" + return os.environ.get(EVIDENCE_GATE_ENV_VAR, "on").strip().lower() not in _DISABLED_VALUES + + def load_live_contract( worktree: Path, identifiers: Sequence[int | str], @@ -181,3 +196,42 @@ def format_incomplete_rows(rows: list[dict[str, Any]]) -> str: return "; ".join( f"{r['id']} (role={r['role'] or 'unassigned'}, status={r['status']})" for r in rows ) + + +def evidence_commits( + contract: Contract, + slice_id: str | None = None, +) -> list[dict[str, Any]] | None: + """List the task rows in scope that cite a commit SHA as evidence. + + The slice close-merge gate (#3125) checks every cited SHA for + reachability from the integration branch tip before the slice may + close — a task record pointing at a commit the slice PR does not + contain means the prescribed ``complete-task --commit`` unblock + flow silently dropped a deliverable. + + Rows are included regardless of ``status``: a row can carry a + commit while still pending (the completion CLI links the commit + before flipping status), and a cited-but-unreachable commit is a + gap worth failing on either way. + + Returns one dict per row (``id`` / ``role`` / ``commit``), empty + list when no row cites a commit, or ``None`` when ``slice_id`` was + given but no such slice exists (caller skips the gate). + """ + slices = _slices_in_scope(contract, slice_id) + if slices is None: + return None + return [ + {"id": task.id, "role": task.role, "commit": task.commit} + for sl in slices + for task in sl.tasks or [] + if task.commit + ] + + +def format_evidence_rows(rows: list[dict[str, Any]]) -> str: + """One-line-per-row summary for evidence-reachability messages.""" + return "; ".join( + f"{r['id']} (role={r['role'] or 'unassigned'}, commit={r['commit']})" for r in rows + ) diff --git a/orchestrator/gateway_client.py b/orchestrator/gateway_client.py index 46ced11f8a..e3a7990d95 100644 --- a/orchestrator/gateway_client.py +++ b/orchestrator/gateway_client.py @@ -16,7 +16,7 @@ import subprocess import sys import time -from collections.abc import Callable +from collections.abc import Callable, Sequence from dataclasses import dataclass from datetime import datetime, timedelta from http.client import HTTPException @@ -2428,6 +2428,150 @@ def is_slice_branch_merged_into_parent( except Exception: pass + def find_unreachable_evidence_commits( + self, + pipeline_id: str, + repo_path: str, + *, + commit_shas: Sequence[str], + integration_branch: str, + mode: Literal["public", "private"] = "public", + ) -> list[str] | None: + """Return the subset of ``commit_shas`` NOT reachable from the + integration branch's tip on origin (#3125). + + The slice close path uses this to verify that every commit SHA + cited by a contract task record actually landed on the + integration branch before the slice PR is opened. A producer's + post-confirmation commit (the prescribed ``complete-task + --commit`` unblock flow) lives only on that agent's local + worktree branch unless something pushed it — this check is what + turns that silent loss into a hard stop. + + Tri-state per SHA, derived from ``git merge-base --is-ancestor + `` through ``/api/v1/git/execute``: + + * exit 0 — reachable; + * exit 1 — the commit object exists locally but is not an + ancestor of the tip → unreachable; + * exit 128 — the SHA does not resolve to an object at all + (never pushed and the worktree odb was pruned, or an + abbreviated SHA that no longer resolves) → unreachable. This + is the fully-lost variant of the same gap, so it must fail + the gate, not skip it. + + Returns ``None`` (caller skips the gate with a warning) when + the check cannot be evaluated at all — branch tip unresolvable + on origin, fetch failure, or a gateway/network error on the + merge-base call itself. A transient infrastructure failure must + not fail the slice; the conservative posture matches the other + completeness checks (#3081 / #3114). + + Transport mirrors :meth:`is_slice_branch_merged_into_parent`: + one synthetic launcher-authenticated session shared across the + ls-remote, the fetch, and every merge-base call. + """ + if not integration_branch or not commit_shas: + return [] + + temp_container_id = ( + f"{pipeline_id}-evidence-reachability-{integration_branch.replace('/', '-')}" + ) + session_token: str | None = None + try: + session = self.register_session( + container_id=temp_container_id, + container_ip=self.self_ip, + mode=mode, + pipeline_id=pipeline_id, + synthetic=True, + ) + session_token = session.session_token + + tip_sha = self.get_remote_branch_sha( + pipeline_id, + repo_path, + f"refs/heads/{integration_branch}", + mode=mode, + bearer_token=session_token, + ) + if not tip_sha: + logger.warning( + "Evidence-reachability check skipped: integration branch " + "tip unresolvable on origin (#3125)", + pipeline_id=pipeline_id, + integration_branch=integration_branch, + ) + return None + + # The tip's objects must be locally reachable for merge-base + # to evaluate. Unlike the ancestor probes elsewhere, a failed + # fetch here must SKIP the gate rather than degrade — with no + # tip objects every merge-base would exit 128 and every cited + # commit would be falsely flagged unreachable, failing the + # slice on a network blip. + fetched = self.fetch_branch( + pipeline_id, + repo_path, + args=[f"+refs/heads/{integration_branch}:refs/remotes/origin/{integration_branch}"], + mode=mode, + bearer_token=session_token, + ) + if not fetched: + logger.warning( + "Evidence-reachability check skipped: integration branch fetch failed (#3125)", + pipeline_id=pipeline_id, + integration_branch=integration_branch, + ) + return None + + unreachable: list[str] = [] + for sha in commit_shas: + try: + self._make_request( + "/api/v1/git/execute", + method="POST", + data={ + "repo_path": repo_path, + "operation": "merge-base", + "args": ["--is-ancestor", sha, tip_sha], + }, + bearer_token=session_token, + ) + except GatewayError as exc: + details = exc.details or {} + returncode = details.get("returncode") + if returncode in (1, 128): + # 1 — object present, not an ancestor. + # 128 — SHA unresolvable in the odb (fully lost). + unreachable.append(sha) + continue + logger.warning( + "Evidence-reachability check skipped: merge-base " + "failed unexpectedly (#3125)", + pipeline_id=pipeline_id, + integration_branch=integration_branch, + commit_sha=sha, + returncode=returncode, + error=str(exc), + ) + return None + return unreachable + except Exception as exc: # noqa: BLE001 + logger.warning( + "Evidence-reachability check skipped: gateway request failed (#3125)", + pipeline_id=pipeline_id, + integration_branch=integration_branch, + error=str(exc), + ) + return None + finally: + if session_token: + try: + self.delete_session(session_token) + except Exception: + pass + def create_slice_integration_branch( self, pipeline_id: str, diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 9250afb127..d7b0b70150 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -10666,6 +10666,113 @@ def _escalate_blocked_slice_to_hitl( ) +def _check_slice_evidence_reachability( + pipeline_id: str, + spawner: "ContainerSpawner", # noqa: UP037 + worktree_repo_path: Path, + slice_id: str, + integration_branch: str, + *, + gateway_mode: Literal["public", "private"] = "public", +) -> str | None: + """Verify the slice's cited evidence commits reached the integration branch (#3125). + + The integration branch only advances when a producer pushes + (``consensus_push`` at propose time). A commit recorded by + ``egg-contract complete-task --commit `` *after* that producer + confirmed — the prescribed HITL unblock flow for a post-confirmation + task reassignment (#3124) — lives only on the agent's local worktree + branch, so the slice would otherwise close and open its PR without + the deliverable while the contract task record points at a commit + nothing retains. + + Runs after slice consensus and before any close side effects (BRC + transcript commit, slice PR). Returns ``None`` when the slice may + close, or a human-readable failure string listing every task row + whose cited commit is not an ancestor of the integration branch tip + — the caller records the slice failure with it, which routes + through the existing cascade + HITL escalation machinery instead of + closing silently. + + Failure posture mirrors the other completeness checks (#3081 / + #3114): the gate degrades to ``None`` (close proceeds, warning + logged) when the contract cannot be read, the slice id does not + resolve, or the gateway reachability probe cannot be evaluated. + Only a definitive "this cited commit is not on the branch" verdict + fails the close. ``EGG_EVIDENCE_REACHABILITY_GATE`` is the operator + kill switch. + """ + try: + import contract_completeness as cc + except ImportError: + from .. import contract_completeness as cc # type: ignore[no-redef] + + if not cc.evidence_gate_enabled(): + logger.info( + "Evidence-reachability gate disabled by kill switch (#3125)", + pipeline_id=pipeline_id, + slice_id=slice_id, + ) + return None + + from egg_contracts.loader import load_contract as _load_contract + + try: + with get_pipeline_state_lock(pipeline_id): + contract = _load_contract(pipeline_id, worktree_repo_path) + except Exception as load_err: # noqa: BLE001 + logger.warning( + "Evidence-reachability gate skipped: contract load failed (#3125)", + pipeline_id=pipeline_id, + slice_id=slice_id, + error=str(load_err), + ) + return None + + rows = cc.evidence_commits(contract, slice_id) + if rows is None: + logger.warning( + "Evidence-reachability gate skipped: slice not found in contract (#3125)", + pipeline_id=pipeline_id, + slice_id=slice_id, + ) + return None + if not rows: + return None + + unreachable_shas = spawner.gateway.find_unreachable_evidence_commits( + pipeline_id, + str(worktree_repo_path), + commit_shas=[r["commit"] for r in rows], + integration_branch=integration_branch, + mode=gateway_mode, + ) + if unreachable_shas is None: + # The probe itself could not be evaluated (gateway/network). + # find_unreachable_evidence_commits already logged the cause. + return None + if not unreachable_shas: + return None + + lost = [r for r in rows if r["commit"] in set(unreachable_shas)] + summary = cc.format_evidence_rows(lost) + logger.error( + "Slice close blocked: task records cite commits unreachable from " + "the integration branch (#3125)", + pipeline_id=pipeline_id, + slice_id=slice_id, + integration_branch=integration_branch, + unreachable=summary, + ) + return ( + f"slice {slice_id}: evidence-reachability gate failed — contract task " + f"records cite commits that are not on integration branch " + f"{integration_branch}: {summary}. Cherry-pick (or push) the cited " + f"commits onto {integration_branch}, then re-run the slice close; " + f"set {cc.EVIDENCE_GATE_ENV_VAR}=off to bypass." + ) + + def _commit_slice_brc_history_to_integration_branch( pipeline, spawner: "ContainerSpawner", # noqa: UP037 @@ -16591,6 +16698,27 @@ def _probe_parent_branch_exists(parent_branch: str) -> bool: ) return exit_code_inner, logs_inner + # #3125 — evidence-reachability gate: every commit SHA + # cited by this slice's contract task records must be + # an ancestor of the integration branch tip, or the + # slice PR would ship without a deliverable the task + # record claims is done (the post-confirmation + # ``complete-task --commit`` unblock flow, #3124). + # Fails the slice BEFORE any close side effect so the + # cascade + HITL machinery surfaces the gap loudly. + if pipeline.repo: + evidence_failure = _check_slice_evidence_reachability( + pipeline_id, + spawner, + worktree_repo_path, + slice_id, + integration_branch, + gateway_mode=gateway_mode, # type: ignore[arg-type] + ) + if evidence_failure is not None: + scheduler.record_failure(slice_id) + return 1, evidence_failure + # Slice consensus reached — snapshot the slice's PR # data under the per-pipeline state lock, then RELEASE # the lock before the gateway HTTP round-trip so we diff --git a/orchestrator/tests/test_evidence_reachability_gate.py b/orchestrator/tests/test_evidence_reachability_gate.py new file mode 100644 index 0000000000..c6cdc52fe8 --- /dev/null +++ b/orchestrator/tests/test_evidence_reachability_gate.py @@ -0,0 +1,465 @@ +"""Tests for the slice-close evidence-reachability gate (#3125). + +The slice integration branch only advances when a producer pushes +(``consensus_push`` at propose time). A commit recorded by the +prescribed post-confirmation unblock flow (``egg-contract complete-task +--commit ``, #3124) lives only on the agent's local worktree +branch, so a slice could close and open its PR without a deliverable +its own task record cites as completion evidence. + +Covers: + +* ``contract_completeness.evidence_commits`` / ``format_evidence_rows`` + — row selection (any row citing a commit, regardless of status), the + unknown-slice ``None`` sentinel, and the independent kill switch. +* ``GatewayClient.find_unreachable_evidence_commits`` — the tri-state + merge-base mapping (exit 0 reachable, exit 1 / 128 unreachable, + anything else → ``None`` skip), and the skip-don't-fail posture on + tip-resolution and fetch failures. +* ``routes.pipelines._check_slice_evidence_reachability`` — gate + wiring: kill switch, graceful degradation on contract/probe + failures, and the failure string on a definitive unreachable + verdict. +""" + +from __future__ import annotations + +import json +import sys +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +# Mock heavy dependencies before importing routes.pipelines. +_docker_mock = MagicMock() +sys.modules.setdefault("docker", _docker_mock) +sys.modules.setdefault("docker.errors", _docker_mock.errors) +sys.modules.setdefault("docker.types", _docker_mock.types) + +_orchestrator_path = Path(__file__).parent.parent +if str(_orchestrator_path) not in sys.path: + sys.path.insert(0, str(_orchestrator_path)) +_shared_path = _orchestrator_path.parent / "shared" +if _shared_path.exists() and str(_shared_path) not in sys.path: + sys.path.insert(0, str(_shared_path)) + +import contract_completeness as cc # noqa: E402 +from gateway_client import GatewayClient, GatewayError, SessionInfo # noqa: E402 +from routes.pipelines import _check_slice_evidence_reachability # noqa: E402 + +PIPELINE_ID = "pipeline-evidence-test" +INTEGRATION_BRANCH = "egg/issue-3125/slice-2" +TIP_SHA = "f" * 40 +PUSHED_SHA = "a" * 40 +LATE_SHA = "b" * 40 + + +# ---------------------------------------------------------------------- +# Contract fixtures +# ---------------------------------------------------------------------- + + +def _contract_dict() -> dict[str, Any]: + """Two-slice contract. + + slice-2 rows: + * task-2-1 coder complete, commit PUSHED_SHA + * task-2-2 documenter complete, commit LATE_SHA (the unblock-flow row) + * task-2-3 coder pending, commit LATE_SHA (commit linked, not yet flipped) + * task-2-4 (no role) pending, no commit + """ + return { + "schemaVersion": "1.0", + "issue": {"number": 3125, "title": "evidence test", "url": "http://example"}, + "phases": [ + { + "id": "slice-1", + "name": "first", + "tasks": [ + { + "id": "task-1-1", + "description": "other slice", + "role": "coder", + "status": "complete", + "commit": "c" * 8, + }, + ], + }, + { + "id": "slice-2", + "name": "second", + "tasks": [ + { + "id": "task-2-1", + "description": "pushed work", + "role": "coder", + "status": "complete", + "commit": PUSHED_SHA, + }, + { + "id": "task-2-2", + "description": "late operator commit", + "role": "documenter", + "status": "complete", + "commit": LATE_SHA, + }, + { + "id": "task-2-3", + "description": "commit linked before status flip", + "role": "coder", + "status": "pending", + "commit": LATE_SHA, + }, + { + "id": "task-2-4", + "description": "no evidence cited", + "status": "pending", + }, + ], + }, + ], + } + + +def _write_contract(worktree: Path, identifier: str = PIPELINE_ID) -> Path: + contracts_dir = worktree / ".egg-state" / "contracts" + contracts_dir.mkdir(parents=True, exist_ok=True) + path = contracts_dir / f"{identifier}.json" + path.write_text(json.dumps(_contract_dict())) + return path + + +def _load(worktree: Path): + from egg_contracts.loader import load_contract + + return load_contract(PIPELINE_ID, worktree) + + +# ---------------------------------------------------------------------- +# contract_completeness helpers +# ---------------------------------------------------------------------- + + +class TestEvidenceCommits: + def test_rows_with_commits_only(self, tmp_path: Path) -> None: + _write_contract(tmp_path) + rows = cc.evidence_commits(_load(tmp_path), "slice-2") + assert rows is not None + assert [r["id"] for r in rows] == ["task-2-1", "task-2-2", "task-2-3"] + + def test_pending_row_with_commit_included(self, tmp_path: Path) -> None: + _write_contract(tmp_path) + rows = cc.evidence_commits(_load(tmp_path), "slice-2") + assert rows is not None + assert any(r["id"] == "task-2-3" for r in rows) + + def test_unknown_slice_returns_none_sentinel(self, tmp_path: Path) -> None: + _write_contract(tmp_path) + assert cc.evidence_commits(_load(tmp_path), "slice-9") is None + + def test_no_slice_id_scans_all_slices(self, tmp_path: Path) -> None: + _write_contract(tmp_path) + rows = cc.evidence_commits(_load(tmp_path), None) + assert rows is not None + assert {r["id"] for r in rows} == {"task-1-1", "task-2-1", "task-2-2", "task-2-3"} + + def test_rows_carry_id_role_commit(self, tmp_path: Path) -> None: + _write_contract(tmp_path) + rows = cc.evidence_commits(_load(tmp_path), "slice-2") + assert rows is not None + assert rows[1] == {"id": "task-2-2", "role": "documenter", "commit": LATE_SHA} + + def test_format_evidence_rows(self) -> None: + text = cc.format_evidence_rows( + [ + {"id": "task-2-2", "role": "documenter", "commit": "abc1234"}, + {"id": "task-2-4", "role": None, "commit": "def5678"}, + ] + ) + assert "task-2-2 (role=documenter, commit=abc1234)" in text + assert "task-2-4 (role=unassigned, commit=def5678)" in text + + +class TestEvidenceGateEnabled: + def test_default_on(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(cc.EVIDENCE_GATE_ENV_VAR, raising=False) + assert cc.evidence_gate_enabled() is True + + @pytest.mark.parametrize("value", ["off", "0", "false", "no", " OFF "]) + def test_kill_switch(self, monkeypatch: pytest.MonkeyPatch, value: str) -> None: + monkeypatch.setenv(cc.EVIDENCE_GATE_ENV_VAR, value) + assert cc.evidence_gate_enabled() is False + + def test_independent_of_ack_gate_switch(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(cc.GATE_ENV_VAR, "off") + monkeypatch.delenv(cc.EVIDENCE_GATE_ENV_VAR, raising=False) + assert cc.evidence_gate_enabled() is True + + +# ---------------------------------------------------------------------- +# GatewayClient.find_unreachable_evidence_commits +# ---------------------------------------------------------------------- + + +@pytest.fixture +def gateway_client() -> GatewayClient: + return GatewayClient( + gateway_host="localhost", + gateway_port=19848, + launcher_secret="test-secret", + timeout=5, + ) + + +def _session_info(token: str = "synthetic-tok") -> SessionInfo: + now = datetime.now() + return SessionInfo( + session_token=token, + container_id="temp", + container_ip=None, + mode="public", + created_at=now, + expires_at=now + timedelta(hours=1), + ) + + +def _stub_helpers( + client: GatewayClient, + *, + tip_sha: str | None = TIP_SHA, + fetch_returns: bool = True, +): + return ( + patch.object(client, "register_session", return_value=_session_info()), + patch.object(client, "delete_session", return_value=True), + patch.object(client, "get_remote_branch_sha", return_value=tip_sha), + patch.object(client, "fetch_branch", return_value=fetch_returns), + ) + + +def _gateway_error(returncode: int | None) -> GatewayError: + details = {"returncode": returncode} if returncode is not None else {} + return GatewayError("git execute failed", details=details) + + +class TestFindUnreachableEvidenceCommits: + def test_empty_input_short_circuits(self, gateway_client: GatewayClient) -> None: + with patch.object(gateway_client, "register_session") as mock_register: + result = gateway_client.find_unreachable_evidence_commits( + PIPELINE_ID, + "/repo", + commit_shas=[], + integration_branch=INTEGRATION_BRANCH, + ) + assert result == [] + mock_register.assert_not_called() + + def test_all_reachable(self, gateway_client: GatewayClient) -> None: + stubs = _stub_helpers(gateway_client) + with ( + stubs[0], + stubs[1], + stubs[2], + stubs[3], + patch.object(gateway_client, "_make_request", return_value={"success": True}), + ): + result = gateway_client.find_unreachable_evidence_commits( + PIPELINE_ID, + "/repo", + commit_shas=[PUSHED_SHA, LATE_SHA], + integration_branch=INTEGRATION_BRANCH, + ) + assert result == [] + + @pytest.mark.parametrize("returncode", [1, 128]) + def test_not_ancestor_and_missing_object_flagged( + self, gateway_client: GatewayClient, returncode: int + ) -> None: + def fake_make_request(endpoint, method=None, data=None, **kwargs): + assert endpoint == "/api/v1/git/execute" + assert data["operation"] == "merge-base" + assert data["args"][0] == "--is-ancestor" + assert data["args"][2] == TIP_SHA + if data["args"][1] == LATE_SHA: + raise _gateway_error(returncode) + return {"success": True} + + stubs = _stub_helpers(gateway_client) + with ( + stubs[0], + stubs[1], + stubs[2], + stubs[3], + patch.object(gateway_client, "_make_request", side_effect=fake_make_request), + ): + result = gateway_client.find_unreachable_evidence_commits( + PIPELINE_ID, + "/repo", + commit_shas=[PUSHED_SHA, LATE_SHA], + integration_branch=INTEGRATION_BRANCH, + ) + assert result == [LATE_SHA] + + def test_unexpected_merge_base_failure_skips(self, gateway_client: GatewayClient) -> None: + stubs = _stub_helpers(gateway_client) + with ( + stubs[0], + stubs[1], + stubs[2], + stubs[3], + patch.object(gateway_client, "_make_request", side_effect=_gateway_error(None)), + ): + result = gateway_client.find_unreachable_evidence_commits( + PIPELINE_ID, + "/repo", + commit_shas=[LATE_SHA], + integration_branch=INTEGRATION_BRANCH, + ) + assert result is None + + def test_unresolvable_tip_skips(self, gateway_client: GatewayClient) -> None: + stubs = _stub_helpers(gateway_client, tip_sha=None) + with stubs[0], stubs[1], stubs[2], stubs[3]: + result = gateway_client.find_unreachable_evidence_commits( + PIPELINE_ID, + "/repo", + commit_shas=[LATE_SHA], + integration_branch=INTEGRATION_BRANCH, + ) + assert result is None + + def test_failed_fetch_skips(self, gateway_client: GatewayClient) -> None: + stubs = _stub_helpers(gateway_client, fetch_returns=False) + with stubs[0], stubs[1], stubs[2], stubs[3]: + result = gateway_client.find_unreachable_evidence_commits( + PIPELINE_ID, + "/repo", + commit_shas=[LATE_SHA], + integration_branch=INTEGRATION_BRANCH, + ) + assert result is None + + def test_session_registration_failure_skips(self, gateway_client: GatewayClient) -> None: + with patch.object(gateway_client, "register_session", side_effect=GatewayError("down")): + result = gateway_client.find_unreachable_evidence_commits( + PIPELINE_ID, + "/repo", + commit_shas=[LATE_SHA], + integration_branch=INTEGRATION_BRANCH, + ) + assert result is None + + def test_session_cleaned_up(self, gateway_client: GatewayClient) -> None: + stubs = _stub_helpers(gateway_client) + with ( + stubs[0], + stubs[1] as mock_delete, + stubs[2], + stubs[3], + patch.object(gateway_client, "_make_request", return_value={"success": True}), + ): + gateway_client.find_unreachable_evidence_commits( + PIPELINE_ID, + "/repo", + commit_shas=[PUSHED_SHA], + integration_branch=INTEGRATION_BRANCH, + ) + mock_delete.assert_called_once_with("synthetic-tok") + + +# ---------------------------------------------------------------------- +# routes.pipelines._check_slice_evidence_reachability +# ---------------------------------------------------------------------- + + +def _spawner(unreachable: list[str] | None) -> MagicMock: + spawner = MagicMock() + spawner.gateway.find_unreachable_evidence_commits.return_value = unreachable + return spawner + + +@pytest.fixture +def gate_env(monkeypatch: pytest.MonkeyPatch): + monkeypatch.delenv(cc.EVIDENCE_GATE_ENV_VAR, raising=False) + return monkeypatch + + +class TestCheckSliceEvidenceReachability: + def test_kill_switch_skips_without_gateway_call( + self, tmp_path: Path, gate_env: pytest.MonkeyPatch + ) -> None: + gate_env.setenv(cc.EVIDENCE_GATE_ENV_VAR, "off") + _write_contract(tmp_path) + spawner = _spawner([LATE_SHA]) + result = _check_slice_evidence_reachability( + PIPELINE_ID, spawner, tmp_path, "slice-2", INTEGRATION_BRANCH + ) + assert result is None + spawner.gateway.find_unreachable_evidence_commits.assert_not_called() + + def test_missing_contract_skips(self, tmp_path: Path, gate_env) -> None: + spawner = _spawner([LATE_SHA]) + result = _check_slice_evidence_reachability( + PIPELINE_ID, spawner, tmp_path, "slice-2", INTEGRATION_BRANCH + ) + assert result is None + spawner.gateway.find_unreachable_evidence_commits.assert_not_called() + + def test_unknown_slice_skips(self, tmp_path: Path, gate_env) -> None: + _write_contract(tmp_path) + spawner = _spawner([LATE_SHA]) + result = _check_slice_evidence_reachability( + PIPELINE_ID, spawner, tmp_path, "slice-9", INTEGRATION_BRANCH + ) + assert result is None + spawner.gateway.find_unreachable_evidence_commits.assert_not_called() + + def test_no_cited_commits_skips_probe(self, tmp_path: Path, gate_env) -> None: + contract = _contract_dict() + for task in contract["phases"][1]["tasks"]: + task.pop("commit", None) + contracts_dir = tmp_path / ".egg-state" / "contracts" + contracts_dir.mkdir(parents=True, exist_ok=True) + (contracts_dir / f"{PIPELINE_ID}.json").write_text(json.dumps(contract)) + + spawner = _spawner([LATE_SHA]) + result = _check_slice_evidence_reachability( + PIPELINE_ID, spawner, tmp_path, "slice-2", INTEGRATION_BRANCH + ) + assert result is None + spawner.gateway.find_unreachable_evidence_commits.assert_not_called() + + def test_all_reachable_passes(self, tmp_path: Path, gate_env) -> None: + _write_contract(tmp_path) + spawner = _spawner([]) + result = _check_slice_evidence_reachability( + PIPELINE_ID, spawner, tmp_path, "slice-2", INTEGRATION_BRANCH + ) + assert result is None + call_kwargs = spawner.gateway.find_unreachable_evidence_commits.call_args.kwargs + assert call_kwargs["commit_shas"] == [PUSHED_SHA, LATE_SHA, LATE_SHA] + assert call_kwargs["integration_branch"] == INTEGRATION_BRANCH + + def test_probe_failure_skips(self, tmp_path: Path, gate_env) -> None: + _write_contract(tmp_path) + result = _check_slice_evidence_reachability( + PIPELINE_ID, _spawner(None), tmp_path, "slice-2", INTEGRATION_BRANCH + ) + assert result is None + + def test_unreachable_evidence_fails_with_rows(self, tmp_path: Path, gate_env) -> None: + _write_contract(tmp_path) + result = _check_slice_evidence_reachability( + PIPELINE_ID, _spawner([LATE_SHA]), tmp_path, "slice-2", INTEGRATION_BRANCH + ) + assert result is not None + # Both rows citing the lost SHA are named; the pushed row is not. + assert "task-2-2" in result + assert "task-2-3" in result + assert "task-2-1" not in result + assert LATE_SHA in result + assert INTEGRATION_BRANCH in result + assert cc.EVIDENCE_GATE_ENV_VAR in result From 12e6ba0579d25291786d5026e286f8a899cc63a8 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Thu, 11 Jun 2026 18:00:28 +0000 Subject: [PATCH 2/3] Address review feedback on #3125 evidence-reachability gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * gateway_client.py: an empty integration_branch now returns None (skip the gate) rather than [] (silently approve). The caller's pipeline.repo guard makes this unreachable in production, but defensive matters if a future caller passes through. * routes/pipelines.py: de-duplicate cited commit SHAs before the merge-base probe — multiple task rows can cite the same commit (the #3124 unblock flow often does), and each duplicate previously burned a round-trip. The membership join re-attaches the verdict to every row. * contract_completeness.py: document that evidence_commits row ordering is intentional (slice-task iteration order). * routes/pipelines.py: collapse the two contract loads at slice close into one — both the evidence-reachability gate and the slice PR data snapshot previously took the per-pipeline state lock independently. Gate now accepts an optional pre-loaded contract; the close path loads once and threads through to both readers. * tests: integration test in test_slice_run_loop_integration.py covers the wiring from _run_one_slice_inner to the gate (post-consensus, pre-close, scheduler.record_failure routing, no PR for failing slice, sibling independence). New test in test_evidence_reachability_gate.py covers the pre-loaded contract path (gate skips internal load when caller supplies a contract). Updated test_all_reachable_passes to reflect the de-duplicated probe input. --- orchestrator/contract_completeness.py | 5 ++ orchestrator/gateway_client.py | 7 +- orchestrator/routes/pipelines.py | 82 ++++++++++++----- .../tests/test_evidence_reachability_gate.py | 48 +++++++++- .../tests/test_slice_run_loop_integration.py | 89 +++++++++++++++++++ 5 files changed, 208 insertions(+), 23 deletions(-) diff --git a/orchestrator/contract_completeness.py b/orchestrator/contract_completeness.py index ccfad6e69e..550ae48639 100644 --- a/orchestrator/contract_completeness.py +++ b/orchestrator/contract_completeness.py @@ -218,6 +218,11 @@ def evidence_commits( Returns one dict per row (``id`` / ``role`` / ``commit``), empty list when no row cites a commit, or ``None`` when ``slice_id`` was given but no such slice exists (caller skips the gate). + + Row ordering is intentional: slice-declaration order outermost, + task-declaration order within each slice. Callers (the close-merge + gate, the failure-string formatter) rely on this for deterministic + operator-facing output. """ slices = _slices_in_scope(contract, slice_id) if slices is None: diff --git a/orchestrator/gateway_client.py b/orchestrator/gateway_client.py index e3a7990d95..c024bcf761 100644 --- a/orchestrator/gateway_client.py +++ b/orchestrator/gateway_client.py @@ -2471,8 +2471,13 @@ def find_unreachable_evidence_commits( one synthetic launcher-authenticated session shared across the ls-remote, the fetch, and every merge-base call. """ - if not integration_branch or not commit_shas: + if not commit_shas: return [] + if not integration_branch: + # No branch to probe means we cannot evaluate reachability. + # Skip the gate rather than silently approve — matches the + # other "cannot evaluate" paths in this method (#3125 review). + return None temp_container_id = ( f"{pipeline_id}-evidence-reachability-{integration_branch.replace('/', '-')}" diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index d7b0b70150..709e0ebbd9 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -10674,6 +10674,7 @@ def _check_slice_evidence_reachability( integration_branch: str, *, gateway_mode: Literal["public", "private"] = "public", + contract: Any | None = None, ) -> str | None: """Verify the slice's cited evidence commits reached the integration branch (#3125). @@ -10701,6 +10702,13 @@ def _check_slice_evidence_reachability( Only a definitive "this cited commit is not on the branch" verdict fails the close. ``EGG_EVIDENCE_REACHABILITY_GATE`` is the operator kill switch. + + ``contract`` is an optional pre-loaded contract: the close path + already needs the contract one stretch later for the slice PR data + snapshot, so threading the same load through saves one file read + and one ``get_pipeline_state_lock`` acquisition. When ``None`` + (the default — keeps the gate self-contained for tests), the gate + loads the contract itself under the lock. """ try: import contract_completeness as cc @@ -10715,19 +10723,20 @@ def _check_slice_evidence_reachability( ) return None - from egg_contracts.loader import load_contract as _load_contract + if contract is None: + from egg_contracts.loader import load_contract as _load_contract - try: - with get_pipeline_state_lock(pipeline_id): - contract = _load_contract(pipeline_id, worktree_repo_path) - except Exception as load_err: # noqa: BLE001 - logger.warning( - "Evidence-reachability gate skipped: contract load failed (#3125)", - pipeline_id=pipeline_id, - slice_id=slice_id, - error=str(load_err), - ) - return None + try: + with get_pipeline_state_lock(pipeline_id): + contract = _load_contract(pipeline_id, worktree_repo_path) + except Exception as load_err: # noqa: BLE001 + logger.warning( + "Evidence-reachability gate skipped: contract load failed (#3125)", + pipeline_id=pipeline_id, + slice_id=slice_id, + error=str(load_err), + ) + return None rows = cc.evidence_commits(contract, slice_id) if rows is None: @@ -10740,10 +10749,16 @@ def _check_slice_evidence_reachability( if not rows: return None + # De-duplicate while preserving first-seen order: multiple task rows + # can cite the same commit (the prescribed unblock flow #3124 often + # links one commit across two adjacent rows). Each duplicate would + # otherwise burn one merge-base round-trip per dupe. The membership + # join below re-attaches the verdict to every row that cites it. + probe_shas = list(dict.fromkeys(r["commit"] for r in rows)) unreachable_shas = spawner.gateway.find_unreachable_evidence_commits( pipeline_id, str(worktree_repo_path), - commit_shas=[r["commit"] for r in rows], + commit_shas=probe_shas, integration_branch=integration_branch, mode=gateway_mode, ) @@ -16698,6 +16713,31 @@ def _probe_parent_branch_exists(parent_branch: str) -> bool: ) return exit_code_inner, logs_inner + # Slice consensus reached — load the contract ONCE + # under the per-pipeline state lock and reuse the same + # snapshot for the #3125 evidence-reachability gate + # AND the slice's PR data snapshot below. Both readers + # previously took the lock independently; collapsing + # them eliminates one file read + lock acquire per + # slice close (#3125 review). + # + # The slice_pr_data block below originally documented + # the lock as covering only the contract read so the + # gateway HTTP round-trip wouldn't serialise other + # writers — the same posture applies here: we release + # the lock before the gateway call inside the gate. + contract_post: Any | None = None + try: + with get_pipeline_state_lock(pipeline_id): + contract_post = load_contract(pipeline_id, worktree_repo_path) + except Exception as load_err: # noqa: BLE001 + logger.warning( + "Slice close: contract load failed (continuing) (#3125)", + pipeline_id=pipeline_id, + slice_id=slice_id, + error=str(load_err), + ) + # #3125 — evidence-reachability gate: every commit SHA # cited by this slice's contract task records must be # an ancestor of the integration branch tip, or the @@ -16706,6 +16746,9 @@ def _probe_parent_branch_exists(parent_branch: str) -> bool: # ``complete-task --commit`` unblock flow, #3124). # Fails the slice BEFORE any close side effect so the # cascade + HITL machinery surfaces the gap loudly. + # ``contract_post`` may be None if the load above + # raised — the gate falls back to its own load in that + # case (and skips gracefully if that fails too). if pipeline.repo: evidence_failure = _check_slice_evidence_reachability( pipeline_id, @@ -16714,21 +16757,18 @@ def _probe_parent_branch_exists(parent_branch: str) -> bool: slice_id, integration_branch, gateway_mode=gateway_mode, # type: ignore[arg-type] + contract=contract_post, ) if evidence_failure is not None: scheduler.record_failure(slice_id) return 1, evidence_failure - # Slice consensus reached — snapshot the slice's PR - # data under the per-pipeline state lock, then RELEASE - # the lock before the gateway HTTP round-trip so we - # don't serialise other contract writers for the - # gateway timeout (~30 s). The lock only needs to - # cover the contract read. + # Snapshot the slice's PR data from the same loaded + # contract — no second lock acquire, no second file + # read. slice_pr_data: dict[str, Any] | None = None try: - with get_pipeline_state_lock(pipeline_id): - contract_post = load_contract(pipeline_id, worktree_repo_path) + if contract_post is not None: slice_obj = next( (s for s in contract_post.slices if s.id == slice_id), None, diff --git a/orchestrator/tests/test_evidence_reachability_gate.py b/orchestrator/tests/test_evidence_reachability_gate.py index c6cdc52fe8..df9ce42ad8 100644 --- a/orchestrator/tests/test_evidence_reachability_gate.py +++ b/orchestrator/tests/test_evidence_reachability_gate.py @@ -257,6 +257,22 @@ def test_empty_input_short_circuits(self, gateway_client: GatewayClient) -> None assert result == [] mock_register.assert_not_called() + def test_empty_integration_branch_skips(self, gateway_client: GatewayClient) -> None: + # An empty branch means we cannot probe reachability at all; skip + # the gate rather than silently approve. The caller's + # ``pipeline.repo`` guard makes this unreachable in production, + # but the defensive default matters if a future caller passes + # through with an empty branch. + with patch.object(gateway_client, "register_session") as mock_register: + result = gateway_client.find_unreachable_evidence_commits( + PIPELINE_ID, + "/repo", + commit_shas=[PUSHED_SHA], + integration_branch="", + ) + assert result is None + mock_register.assert_not_called() + def test_all_reachable(self, gateway_client: GatewayClient) -> None: stubs = _stub_helpers(gateway_client) with ( @@ -440,7 +456,10 @@ def test_all_reachable_passes(self, tmp_path: Path, gate_env) -> None: ) assert result is None call_kwargs = spawner.gateway.find_unreachable_evidence_commits.call_args.kwargs - assert call_kwargs["commit_shas"] == [PUSHED_SHA, LATE_SHA, LATE_SHA] + # task-2-2 and task-2-3 cite the same LATE_SHA — the probe input + # is de-duplicated so each unique SHA is round-tripped once. + # Order is first-seen by slice-task iteration. + assert call_kwargs["commit_shas"] == [PUSHED_SHA, LATE_SHA] assert call_kwargs["integration_branch"] == INTEGRATION_BRANCH def test_probe_failure_skips(self, tmp_path: Path, gate_env) -> None: @@ -463,3 +482,30 @@ def test_unreachable_evidence_fails_with_rows(self, tmp_path: Path, gate_env) -> assert LATE_SHA in result assert INTEGRATION_BRANCH in result assert cc.EVIDENCE_GATE_ENV_VAR in result + + def test_pre_loaded_contract_skips_internal_load(self, tmp_path: Path, gate_env) -> None: + """When the caller pre-loads the contract (the close-path + does this to reuse one load for both the gate and the slice + PR data snapshot — #3125 review), the gate uses the supplied + contract and does NOT re-read from disk. + """ + # Write contract to disk so the on-disk fallback would also + # work; we then load it ourselves to simulate the caller + # pre-loading. The gate's internal load is monkey-patched to + # raise so we can be sure the pre-loaded path is taken. + _write_contract(tmp_path) + preloaded = _load(tmp_path) + + with patch( + "egg_contracts.loader.load_contract", + side_effect=RuntimeError("internal load should not run"), + ): + result = _check_slice_evidence_reachability( + PIPELINE_ID, + _spawner([]), + tmp_path, + "slice-2", + INTEGRATION_BRANCH, + contract=preloaded, + ) + assert result is None diff --git a/orchestrator/tests/test_slice_run_loop_integration.py b/orchestrator/tests/test_slice_run_loop_integration.py index b116ed315a..469f9a8d03 100644 --- a/orchestrator/tests/test_slice_run_loop_integration.py +++ b/orchestrator/tests/test_slice_run_loop_integration.py @@ -674,6 +674,95 @@ def test_pr_creation_failure_marks_slice_failed(self) -> None: "Sibling slice must still run regardless of slice-1's PR failure" ) + def test_evidence_reachability_failure_fails_slice_without_pr(self) -> None: + """#3125 review: pin the wiring between + ``_run_one_slice_inner`` (post-consensus) and + ``_check_slice_evidence_reachability``. + + The unit-level gate is covered exhaustively by + ``test_evidence_reachability_gate.py``. This test locks the + 13-line block that calls into it from the slice run loop so a + future refactor — e.g. drift on the ``# type: ignore[arg-type]`` + for ``gateway_mode`` — does not silently lose the wire-up. The + gate verdict (a non-None failure string) must: + + * route through ``scheduler.record_failure(slice_id)`` and the + existing cascade machinery, + * surface as a non-zero overall exit code, + * carry the failure string into the slice logs, and + * skip ``create_slice_pr`` for the failed slice — the close + side effects are exactly what the gate runs *before* to + prevent a silent-loss PR ship. + + Sibling slices remain independent (decision-2): a sibling whose + evidence is intact still runs to completion and opens its PR. + """ + pipeline = _make_pipeline() + failing = _make_slice("slice-1", tasks=[_make_task("task-1-1")]) + sibling = _make_slice("slice-2", tasks=[_make_task("task-2-1")]) + contract = _make_contract(slices=[failing, sibling]) + + evidence_failure_text = ( + "slice slice-1: evidence-reachability gate failed — contract task " + "records cite commits that are not on integration branch " + "egg/issue-9999/slice-1: task-1-1 (role=coder, commit=deadbeef)" + ) + + def _gate_side_effect( + _pipeline_id: str, + _spawner: Any, + _worktree_repo_path: Path, + slice_id: str, + _integration_branch: str, + **_kwargs: Any, + ) -> str | None: + return evidence_failure_text if slice_id == "slice-1" else None + + with ( + patch("egg_contracts.loader.load_contract", return_value=contract), + patch("egg_contracts.loader.save_contract"), + patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon, + patch("routes.pipelines._run_concurrent_phase", return_value=(0, "ok")), + patch( + "routes.pipelines._check_slice_evidence_reachability", + side_effect=_gate_side_effect, + ) as mock_gate, + patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), + ): + mock_start_recon.return_value = (MagicMock(), threading.Event()) + spawner = self._make_spawner() + exit_code, logs = _run_implement_phase_slices( + pipeline_id=pipeline.id, + pipeline=pipeline, + spawner=spawner, + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + store=MagicMock(), + certs_volume=None, + worktree_repo_path=Path("/tmp/x"), + ) + + # Gate was called for each slice post-consensus. + gate_slice_ids = {c.args[3] for c in mock_gate.call_args_list} + assert gate_slice_ids == {"slice-1", "slice-2"} + # Failing slice surfaces a non-zero overall exit code. + assert exit_code != 0, ( + "Evidence-reachability failure must propagate to a non-zero exit " + "via scheduler.record_failure → cascade machinery" + ) + # The gate's failure string lands in the slice logs operator-side. + assert "evidence-reachability gate failed" in logs + # The failing slice does NOT get a PR — close side effects are + # skipped exactly to prevent a silent-loss ship. + pr_calls_slice_ids = [ + c.kwargs["slice_id"] for c in spawner.gateway.create_slice_pr.call_args_list + ] + assert "slice-1" not in pr_calls_slice_ids + # Sibling with reachable evidence still runs and opens its PR. + assert "slice-2" in pr_calls_slice_ids + def test_reconciler_started_and_stopped(self) -> None: pipeline = _make_pipeline() contract = _make_contract(slices=[_make_slice("slice-1")]) From 649c1d9fd4a58f5fe615fff7a697c4fa178955a7 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Thu, 11 Jun 2026 18:26:11 +0000 Subject: [PATCH 3/3] Address #3126 re-review nits: stale comment + close-path contract= wiring assertion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - routes/pipelines.py:16867: rewrite the slice_pr_data try/except comment to match the post-lift scope (nested attribute traversal only; contract load was lifted out into its own block earlier). Rename the exception alias from load_err to attr_err. - tests/test_slice_run_loop_integration.py: assert each _check_slice_evidence_reachability call receives contract=contract through kwargs, so a future refactor cannot silently drop the close-path wiring that lets the gate skip its own (lock-held) internal load. Skipped nit-3 (double warning on contract-load failure) — reviewer labelled it bikeshedding and the defensive double-load is intentional defence-in-depth so the gate is self-contained for direct unit-test invocation. --- orchestrator/routes/pipelines.py | 18 +++++++++--------- .../tests/test_slice_run_loop_integration.py | 9 +++++++++ 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 709e0ebbd9..9a4001871d 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -16864,19 +16864,19 @@ def _probe_parent_branch_exists(parent_branch: str) -> bool: program_pr.manual_steps if program_pr else None ), } - except Exception as load_err: # noqa: BLE001 - # Contract load + nested attribute traversal on - # slice/program PR objects. Surface includes - # loader validation errors, OSError, plus - # AttributeError / KeyError on partially-populated - # PR rollup fields. Continue without slice_pr_data - # (the gateway PR creation just below is gated - # on it being non-None). + except Exception as attr_err: # noqa: BLE001 + # Nested attribute traversal on slice/program PR + # objects (the contract load was lifted out to the + # block above). Surface is AttributeError / + # KeyError on partially-populated PR rollup + # fields. Continue without slice_pr_data (the + # gateway PR creation just below is gated on it + # being non-None). logger.warning( "Slice PR pre-load failed (continuing)", pipeline_id=pipeline_id, slice_id=slice_id, - error=str(load_err), + error=str(attr_err), ) # Persist this slice's per-slice BRC consensus history diff --git a/orchestrator/tests/test_slice_run_loop_integration.py b/orchestrator/tests/test_slice_run_loop_integration.py index 469f9a8d03..d09809f611 100644 --- a/orchestrator/tests/test_slice_run_loop_integration.py +++ b/orchestrator/tests/test_slice_run_loop_integration.py @@ -747,6 +747,15 @@ def _gate_side_effect( # Gate was called for each slice post-consensus. gate_slice_ids = {c.args[3] for c in mock_gate.call_args_list} assert gate_slice_ids == {"slice-1", "slice-2"} + # Each call routes the pre-loaded contract through ``contract=`` + # so the gate skips its own (lock-held) load (#3125 review + # nit-4): locks the wiring that the unit-level + # ``test_pre_loaded_contract_skips_internal_load`` cannot see. + for call in mock_gate.call_args_list: + assert call.kwargs.get("contract") is contract, ( + "Close path must thread the pre-loaded contract through " + "to the gate so the lock-held internal load is skipped" + ) # Failing slice surfaces a non-zero overall exit code. assert exit_code != 0, ( "Evidence-reachability failure must propagate to a non-zero exit "