feat(SKY-11879): family-uniform output-contract actuation — no bail returns unchanged (#7141)

tests/unit/test_copilot_blocker_signal.py

@@ -290,6 +290,67 @@ def test_stash_blocker_signal_active_terminal_replaces_per_tool_budget() -> None
     assert ctx.tool_blocker_signals == [budget, active_terminal]
 
 
+def test_stash_output_contract_terminal_replaces_held_churn_stop() -> None:
+    for reason_code in ("output_source_unobservable", "actuation_exhausted"):
+        ctx = _Ctx()
+        churn = _make(
+            kind="loop_detected",
+            internal_reason_code="code_authoring_guardrail_churn",
+            renders_final_reply=True,
+        )
+        oc_terminal = _make(
+            kind="tool_error",
+            internal_reason_code=reason_code,
+            renders_final_reply=True,
+        )
+        stash_blocker_signal(ctx, churn)
+        stash_blocker_signal(ctx, oc_terminal)
+        assert ctx.blocker_signal is oc_terminal
+
+
+def test_stash_output_contract_terminal_replaces_any_held_loop_detected() -> None:
+    for loop_reason in (
+        "loop_detected_generic",
+        "loop_detected_repeated_failed_step",
+        "loop_detected_consecutive_same_tool",
+        "credential_priority_authoring_churn",
+    ):
+        ctx = _Ctx()
+        loop_signal = _make(kind="loop_detected", internal_reason_code=loop_reason, renders_final_reply=True)
+        oc_terminal = _make(
+            kind="tool_error", internal_reason_code="output_source_unobservable", renders_final_reply=True
+        )
+        stash_blocker_signal(ctx, loop_signal)
+        stash_blocker_signal(ctx, oc_terminal)
+        assert ctx.blocker_signal is oc_terminal
+
+
+def test_stash_loop_detected_does_not_replace_held_output_contract_terminal() -> None:
+    ctx = _Ctx()
+    oc_terminal = _make(kind="tool_error", internal_reason_code="actuation_exhausted", renders_final_reply=True)
+    loop_signal = _make(kind="loop_detected", internal_reason_code="loop_detected_generic", renders_final_reply=True)
+    stash_blocker_signal(ctx, oc_terminal)
+    stash_blocker_signal(ctx, loop_signal)
+    assert ctx.blocker_signal is oc_terminal
+
+
+def test_stash_churn_stop_does_not_replace_held_output_contract_terminal() -> None:
+    ctx = _Ctx()
+    oc_terminal = _make(
+        kind="tool_error",
+        internal_reason_code="output_source_unobservable",
+        renders_final_reply=True,
+    )
+    churn = _make(
+        kind="loop_detected",
+        internal_reason_code="code_authoring_guardrail_churn",
+        renders_final_reply=True,
+    )
+    stash_blocker_signal(ctx, oc_terminal)
+    stash_blocker_signal(ctx, churn)
+    assert ctx.blocker_signal is oc_terminal
+
+
 def test_stash_grounding_replaces_synthesized_persistence_tool_error() -> None:
     ctx = _Ctx()
     existing = _make(

tests/unit/test_copilot_cancel_helpers.py

@@ -249,6 +249,10 @@ def _guard_ctx(pending_run_id: str | None = None) -> _NS:
         last_test_non_retriable_nav_error=None,
         pending_reconciliation_run_id=pending_run_id,
         synthesized_block_reopened_for_output_coverage=False,
+        output_contract_actuation_by_signature={},
+        output_contract_reject_count_by_signature={},
+        output_contract_actuation_count_by_signature={},
+        output_contract_armed_directive_fingerprint_by_signature={},
     )
 
 

tests/unit/test_copilot_code_block_persist_seam.py

@@ -47,7 +47,10 @@ from skyvern.forge.sdk.copilot.enforcement import (
     MAX_CREDENTIAL_PRIORITY_AUTHORING_REJECTS,
     _check_enforcement,
 )
-from skyvern.forge.sdk.copilot.output_contracts import code_block_available_binding_keys_by_label
+from skyvern.forge.sdk.copilot.output_contracts import (
+    OutputContractAdvisoryState,
+    code_block_available_binding_keys_by_label,
+)
 from skyvern.forge.sdk.copilot.output_utils import sanitize_tool_result_for_llm
 from skyvern.forge.sdk.copilot.reached_download_target import ReachedDownloadTarget
 from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
@@ -5905,20 +5908,78 @@ class TestSeparatedSpineViolationActuation:
             ctx.last_code_authoring_repair_context.required_block_structure == "separated_browser_spine_plus_extraction"
         )
 
-    def test_composite_key_rearms_on_changed_yaml(self, monkeypatch: pytest.MonkeyPatch) -> None:
+    def test_cosmetic_churn_escalates_to_advisory_instead_of_rearming(self, monkeypatch: pytest.MonkeyPatch) -> None:
         monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
         ctx = _spine_actuation_ctx()
         first_yaml = _collapsed_spine_yaml(
             "_setup = 1\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
         )
-        second_yaml = _collapsed_spine_yaml(
+        cosmetic_yaml = _collapsed_spine_yaml(
             "_setup = 2\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
         )
 
         workflow_update_module._impose_output_contract_envelope_after_steering(ctx, first_yaml, [])
-        workflow_update_module._impose_output_contract_envelope_after_steering(ctx, second_yaml, [])
+        assert len(ctx.output_contract_spine_directive_blockers_by_attempt_key) == 1
+        assert not ctx.output_contract_actuation_by_signature
+        assert ctx.turn_halt is None
 
-        assert len(ctx.output_contract_spine_directive_blockers_by_attempt_key) == 2
+        workflow_update_module._impose_output_contract_envelope_after_steering(ctx, cosmetic_yaml, [])
+        assert list(ctx.output_contract_actuation_by_signature.values()) == [OutputContractAdvisoryState.GRANTED]
+        assert ctx.turn_halt is None
+        assert len(ctx.output_contract_spine_directive_blockers_by_attempt_key) == 1
+
+    def test_granted_advisory_arms_run_evidence_before_consume_when_run_attemptable(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        ctx = _spine_actuation_ctx()
+        signature = "sig_preflight_dispatch"
+        ctx.output_contract_actuation_by_signature[signature] = OutputContractAdvisoryState.GRANTED
+        evaluation = workflow_update_module._OutputContractEvaluation(
+            block_label="extract_record",
+            artifact_id="art-1",
+            required_paths={"output.confirmation_number"},
+            source="requested_output_contract",
+            reason_code="requested_output_contract_missing_output_coverage",
+            missing_metadata_paths=[],
+            missing_schema_paths=[],
+            missing_return_paths=[],
+            shape_violations=["separated_spine_shape_violation"],
+            canonical_signature=signature,
+            payload={
+                "reason_code": "requested_output_contract_missing_output_coverage",
+                "missing_requested_output_facts": [],
+            },
+            repair_context=None,
+            can_attempt_run=True,
+        )
+        monkeypatch.setattr(
+            workflow_update_module,
+            "_impose_output_contract_envelope_after_steering",
+            lambda ctx, wf, meta: (wf, meta, False),
+        )
+        monkeypatch.setattr(
+            workflow_update_module,
+            "_scaffold_metadata_contract_for_update",
+            lambda ctx, wf, meta: (meta, False),
+        )
+        monkeypatch.setattr(
+            workflow_update_module,
+            "_evaluate_output_contract_for_code_block",
+            lambda *a, **k: evaluation,
+        )
+
+        result = workflow_update_module._metadata_contract_run_preflight_reject(ctx, "title: x", [])
+
+        assert result is None
+        assert ctx.output_contract_pending_run_evidence.get(signature) == ["output.confirmation_number"]
+        assert ctx.output_contract_actuation_by_signature[signature] == OutputContractAdvisoryState.CONSUMED
+
+        run_result = {
+            "data": {"blocks": [{"label": "extract_record", "extracted_data": {"confirmation_number": "R-9"}}]}
+        }
+        workflow_update_module.record_output_contract_run_output_evidence(ctx, run_result)
+        assert ctx.output_contract_run_output_observed_by_signature[signature] is True
+        assert ctx.output_contract_run_bound_required_path_by_signature[signature] is True
 
 
 def _dual_output_owner_yaml() -> str:
@@ -6163,6 +6224,32 @@ class TestSeparatedSpineImpositionRunEligibility:
 
         assert read_key in armed_keys
 
+    def test_granted_structural_advisory_lifts_output_contract_run_gate_on_unsplit_draft(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
+        ctx = _spine_actuation_ctx()
+        collapsed_yaml = _collapsed_spine_yaml(
+            "_setup = 1\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
+        )
+
+        workflow_update_module._impose_output_contract_envelope_after_steering(ctx, collapsed_yaml, [])
+        blocked = workflow_update_module._evaluate_output_contract_for_code_block(
+            ctx, collapsed_yaml, [], allow_static_return_advisory=True
+        )
+        assert blocked is not None and blocked.can_attempt_run is False
+
+        workflow_update_module._impose_output_contract_envelope_after_steering(ctx, collapsed_yaml, [])
+        assert list(ctx.output_contract_actuation_by_signature.values()) == [OutputContractAdvisoryState.GRANTED]
+
+        scaffolded_metadata, _applied = workflow_update_module._scaffold_metadata_contract_for_update(
+            ctx, collapsed_yaml, []
+        )
+        granted = workflow_update_module._evaluate_output_contract_for_code_block(
+            ctx, collapsed_yaml, scaffolded_metadata, allow_static_return_advisory=True
+        )
+        assert granted is not None and granted.can_attempt_run is True
+
 
 def _budget_evaluation(ctx: CopilotContext) -> _OutputContractEvaluation:
     workflow_yaml = _collapsed_spine_yaml(
@@ -9106,6 +9193,26 @@ class TestCodeAuthoringGuardrailChurnBackstop:
         assert ctx.blocker_signal is terminal
         assert ctx.latest_tool_blocker_signal is terminal
 
+    def test_churn_stop_defers_while_output_contract_ladder_active_then_fires(self) -> None:
+        ctx = _code_only_ctx()
+        ctx.output_contract_actuation_count_by_signature["sig_active"] = 1
+        for _ in range(MAX_CODE_AUTHORING_GUARDRAIL_REJECTS):
+            workflow_update_module._record_code_authoring_guardrail_reject(ctx, frontier_unchanged=True)
+        assert ctx.code_authoring_guardrail_reject_count == MAX_CODE_AUTHORING_GUARDRAIL_REJECTS
+        assert ctx.blocker_signal is None
+        ctx.output_contract_actuation_count_by_signature.clear()
+        workflow_update_module._record_code_authoring_guardrail_reject(ctx, frontier_unchanged=True)
+        churn = ctx.blocker_signal
+        assert isinstance(churn, CopilotToolBlockerSignal)
+        assert churn.internal_reason_code == "code_authoring_guardrail_churn"
+
+    def test_churn_stop_defers_while_advisory_grant_held(self) -> None:
+        ctx = _code_only_ctx()
+        ctx.output_contract_actuation_by_signature["sig_granted"] = OutputContractAdvisoryState.GRANTED
+        for _ in range(MAX_CODE_AUTHORING_GUARDRAIL_REJECTS):
+            workflow_update_module._record_code_authoring_guardrail_reject(ctx, frontier_unchanged=True)
+        assert ctx.blocker_signal is None
+
     @pytest.mark.asyncio
     async def test_mixed_credential_and_unresolved_name_reject_returns_code_repair_progress(
         self, monkeypatch: pytest.MonkeyPatch

tests/unit/test_copilot_enforcement_pruning.py

@@ -36,6 +36,7 @@ from skyvern.forge.sdk.copilot.enforcement import (
     _needs_suspicious_success_nudge,
     _prune_input_list,
     _should_block_mutating_tool_after_synthesized_offer,
+    _should_force_advisory_run_dispatch,
     _should_force_synthesized_block_persistence,
     _summarize_tool_output,
     _uncovered_output_reject_admits_evaluate,
@@ -54,6 +55,10 @@ from skyvern.forge.sdk.copilot.mcp_adapter import (
     _restore_post_hook_context,
     _snapshot_post_hook_context,
 )
+from skyvern.forge.sdk.copilot.output_contracts import (
+    OUTPUT_SOURCE_UNOBSERVABLE_REASON_CODE,
+    OutputContractAdvisoryState,
+)
 from skyvern.forge.sdk.copilot.reached_download_target import ReachedDownloadTarget
 from skyvern.forge.sdk.copilot.request_policy import CompletionCriterion
 from skyvern.forge.sdk.copilot.streaming_adapter import _update_enforcement_from_tool
@@ -64,6 +69,7 @@ from skyvern.forge.sdk.copilot.tools import (
     _record_run_blocks_result,
     _record_workflow_update_result,
 )
+from skyvern.forge.sdk.copilot.turn_halt import stash_turn_halt_from_blocker_signal
 from skyvern.forge.sdk.copilot.turn_intent import TurnIntent, TurnIntentAuthority, TurnIntentMode
 from skyvern.forge.sdk.copilot.verification_evidence import WorkflowVerificationEvidence
 from tests.unit.conftest import make_copilot_context
@@ -1225,6 +1231,7 @@ def _fresh_ctx_for_record() -> SimpleNamespace:
         repeated_action_fingerprint_streak_count=0,
         copilot_total_timeout_exceeded=False,
         workflow_verification_evidence=WorkflowVerificationEvidence(),
+        output_contract_pending_run_evidence={},
     )
 
 
@@ -2341,3 +2348,78 @@ class TestScoutOutputCoverageGate:
         _restore_post_hook_context(ctx, snapshot)
         assert ctx.scouted_output_covered_paths == {"output.document_name"}
         assert ctx.synthesized_block_reopened_for_output_coverage is False
+
+
+class TestAdvisoryRunDispatchForceLane:
+    """A granted output-contract advisory run is forced onto update_and_run_blocks through the same
+    tool_choice forcing lane as the synthesized-persistence force, and releases on consume or terminal."""
+
+    def _granted_ctx(self) -> _Ctx:
+        ctx = _Ctx()
+        ctx.turn_intent = TurnIntent(
+            mode=TurnIntentMode.BUILD,
+            authority=TurnIntentAuthority(may_update_workflow=True, may_run_blocks=True),
+        )
+        ctx.block_authoring_policy = BlockAuthoringPolicy.CODE_ONLY_BROWSER
+        ctx.turn_halt = None
+        ctx.blocker_signal = None
+        ctx.output_contract_actuation_by_signature = {"sig_a": OutputContractAdvisoryState.GRANTED}
+        return ctx
+
+    def test_granted_advisory_forces_run_dispatch(self) -> None:
+        assert _should_force_advisory_run_dispatch(self._granted_ctx()) is True
+
+    def test_consumed_advisory_releases_the_force(self) -> None:
+        ctx = self._granted_ctx()
+        ctx.output_contract_actuation_by_signature = {"sig_a": OutputContractAdvisoryState.CONSUMED}
+        assert _should_force_advisory_run_dispatch(ctx) is False
+
+    def test_no_grant_does_not_force(self) -> None:
+        ctx = self._granted_ctx()
+        ctx.output_contract_actuation_by_signature = {}
+        assert _should_force_advisory_run_dispatch(ctx) is False
+
+    def test_authority_forbidding_run_never_forces(self) -> None:
+        ctx = self._granted_ctx()
+        ctx.turn_intent = TurnIntent(
+            mode=TurnIntentMode.BUILD,
+            authority=TurnIntentAuthority(may_update_workflow=True, may_run_blocks=False),
+        )
+        assert _should_force_advisory_run_dispatch(ctx) is False
+
+    def test_held_genuinely_terminal_blocker_releases_the_force(self) -> None:
+        ctx = self._granted_ctx()
+        ctx.blocker_signal = CopilotToolBlockerSignal(
+            blocker_kind="tool_error",
+            agent_steering_text="",
+            user_facing_reason="",
+            recovery_hint="stop",
+            internal_reason_code=OUTPUT_SOURCE_UNOBSERVABLE_REASON_CODE,
+        )
+        assert _should_force_advisory_run_dispatch(ctx) is False
+
+    def test_non_code_only_policy_does_not_force(self) -> None:
+        ctx = self._granted_ctx()
+        ctx.block_authoring_policy = None
+        assert _should_force_advisory_run_dispatch(ctx) is False
+
+    def test_granted_advisory_survives_model_churn_and_stays_force_eligible(self) -> None:
+        ctx = self._granted_ctx()
+        ctx.latest_tool_blocker_signal = None
+        ctx.tool_blocker_signals = []
+        ctx.output_contract_actuation_count_by_signature = {}
+        ctx.output_contract_run_output_observed_by_signature = {}
+        ctx.output_contract_page_extraction_imposed_by_signature = {}
+        ctx.output_contract_pending_run_evidence = {"sig_a": ["output.confirmation_number"]}
+        churn = CopilotToolBlockerSignal(
+            blocker_kind="loop_detected",
+            agent_steering_text="",
+            user_facing_reason="",
+            recovery_hint="stop",
+            internal_reason_code="code_authoring_guardrail_churn",
+        )
+        stash_turn_halt_from_blocker_signal(ctx, churn, source="enforcement_backstop")
+        stash_turn_halt_from_blocker_signal(ctx, churn, source="enforcement_backstop")
+        assert ctx.turn_halt is None
+        assert ctx.output_contract_actuation_by_signature["sig_a"] == OutputContractAdvisoryState.GRANTED
+        assert _should_force_advisory_run_dispatch(ctx) is True

tests/unit/test_copilot_tool_loop_error_block_running.py

@@ -15,6 +15,7 @@ from typing import Any
 
 from skyvern.forge.sdk.copilot.enforcement import TOTAL_TIMEOUT_SECONDS
 from skyvern.forge.sdk.copilot.failure_tracking import compute_action_sequence_fingerprint
+from skyvern.forge.sdk.copilot.output_contracts import OutputContractAdvisoryState
 from skyvern.forge.sdk.copilot.tools import (
     BLOCK_RUNNING_TOOLS,
     COPILOT_FINAL_REPLY_RESERVE_SECONDS,
@@ -43,6 +44,11 @@ def _ctx(**overrides: Any) -> Any:
         "staged_workflow_yaml": None,
         "has_staged_proposal": False,
         "synthesized_block_reopened_for_output_coverage": False,
+        "output_contract_actuation_by_signature": {},
+        "output_contract_reject_count_by_signature": {},
+        "output_contract_actuation_count_by_signature": {},
+        "output_contract_armed_directive_fingerprint_by_signature": {},
+        "output_contract_dispatch_reopened_by_signature": {},
     }
     base.update(overrides)
     return SimpleNamespace(**base)
@@ -127,6 +133,31 @@ def test_planning_tool_still_trips_name_only_streak() -> None:
     assert "LOOP DETECTED" in msg
 
 
+def test_unresolved_output_contract_ladder_skips_name_only_streak_for_authoring_tool() -> None:
+    ctx = _ctx(output_contract_actuation_count_by_signature={"sig_active": 1})
+    for _ in range(5):
+        assert _tool_loop_error(ctx, "update_workflow") is None
+
+
+def test_resolved_output_contract_ladder_re_enables_name_only_streak() -> None:
+    ctx = _ctx(
+        output_contract_actuation_count_by_signature={"sig_done": 3},
+        output_contract_actuation_by_signature={"sig_done": OutputContractAdvisoryState.CONSUMED},
+    )
+    assert _tool_loop_error(ctx, "update_workflow") is None
+    assert _tool_loop_error(ctx, "update_workflow") is None
+    msg = _tool_loop_error(ctx, "update_workflow")
+    assert msg is not None
+    assert "LOOP DETECTED" in msg
+
+
+def test_block_running_tool_is_blocked_by_pending_reconciliation() -> None:
+    ctx = _ctx(pending_reconciliation_run_id="wr_123")
+    msg = _tool_loop_error(ctx, "update_and_run_blocks")
+    assert msg is not None
+    assert "wr_123" in msg
+
+
 def test_block_running_tool_blocks_late_retry_after_failed_workflow() -> None:
     ctx = _ctx(
         last_failed_workflow_yaml="version: '1.0'",

tests/unit/test_copilot_turn_halt.py

@@ -12,6 +12,7 @@ from skyvern.forge.sdk.copilot.enforcement import (
     terminal_challenge_blocker_signal_from_page_evidence,
 )
 from skyvern.forge.sdk.copilot.failure_tracking import ACTIVE_RUN_TERMINAL_EVIDENCE_REASON_CODE
+from skyvern.forge.sdk.copilot.output_contracts import OutputContractAdvisoryState
 from skyvern.forge.sdk.copilot.run_outcome import (
     TERMINAL_CHALLENGE_BLOCKER_REASON_CODE,
     TERMINAL_CHALLENGE_RUN_OUTCOME_REASON_CODE,
@@ -450,3 +451,126 @@ def test_default_verified_argument_is_fail_safe_and_raises() -> None:
 
     with pytest.raises(CopilotTurnHalt):
         raise_if_turn_halt(ctx)
+
+
+def _output_contract_ctx(*, granted: bool) -> SimpleNamespace:
+    states = {"sig_a": OutputContractAdvisoryState.GRANTED} if granted else {}
+    return SimpleNamespace(
+        turn_halt=None,
+        output_contract_actuation_by_signature=states,
+        output_contract_actuation_count_by_signature={},
+    )
+
+
+def test_loop_detected_deferred_while_output_contract_ladder_unresolved() -> None:
+    ctx = _output_contract_ctx(granted=True)
+    signal = _signal(blocker_kind="loop_detected", internal_reason_code="code_authoring_guardrail_churn")
+
+    halt = stash_turn_halt_from_blocker_signal(ctx, signal, source="enforcement_backstop")
+
+    assert halt is None
+    assert ctx.turn_halt is None
+
+
+def test_loop_detected_promotes_once_output_contract_ladder_resolves() -> None:
+    ctx = _output_contract_ctx(granted=False)
+    signal = _signal(blocker_kind="loop_detected", internal_reason_code="code_authoring_guardrail_churn")
+
+    halt = stash_turn_halt_from_blocker_signal(ctx, signal, source="enforcement_backstop")
+
+    assert halt is not None
+    assert halt.kind == TurnHaltKind.LOOP_DETECTED
+
+
+def test_output_source_unobservable_terminal_promotes_even_while_ladder_unresolved() -> None:
+    ctx = _output_contract_ctx(granted=True)
+    signal = _signal(blocker_kind="tool_error", internal_reason_code="output_source_unobservable")
+
+    halt = stash_turn_halt_from_blocker_signal(ctx, signal, source="workflow_update")
+
+    assert halt is not None
+    assert halt.kind == TurnHaltKind.OUTPUT_SOURCE_UNOBSERVABLE
+
+
+def test_active_terminal_challenge_promotes_while_ladder_unresolved() -> None:
+    ctx = _output_contract_ctx(granted=True)
+    signal = _signal(internal_reason_code=ACTIVE_RUN_TERMINAL_EVIDENCE_REASON_CODE)
+
+    halt = stash_turn_halt_from_blocker_signal(ctx, signal, source="run_execution")
+
+    assert halt is not None
+    assert halt.kind == TurnHaltKind.ACTIVE_TERMINAL_CHALLENGE
+
+
+def _defer_ledger_ctx() -> SimpleNamespace:
+    return SimpleNamespace(
+        turn_halt=None,
+        blocker_signal=None,
+        latest_tool_blocker_signal=None,
+        tool_blocker_signals=[],
+        output_contract_actuation_by_signature={"sig_a": OutputContractAdvisoryState.GRANTED},
+        output_contract_actuation_count_by_signature={},
+        output_contract_run_output_observed_by_signature={},
+        output_contract_page_extraction_imposed_by_signature={},
+        output_contract_pending_run_evidence={"sig_a": ["output.confirmation_number"]},
+    )
+
+
+def _defer_count_ledger_ctx() -> SimpleNamespace:
+    return SimpleNamespace(
+        turn_halt=None,
+        blocker_signal=None,
+        latest_tool_blocker_signal=None,
+        tool_blocker_signals=[],
+        output_contract_actuation_by_signature={"sig_a": OutputContractAdvisoryState.UNUSED},
+        output_contract_actuation_count_by_signature={"sig_a": 1},
+        output_contract_run_output_observed_by_signature={},
+        output_contract_page_extraction_imposed_by_signature={},
+        output_contract_pending_run_evidence={"sig_a": ["output.confirmation_number"]},
+    )
+
+
+def _loop_signal() -> CopilotToolBlockerSignal:
+    return _signal(blocker_kind="loop_detected", internal_reason_code="code_authoring_guardrail_churn")
+
+
+def test_defer_swallows_first_loop_signal_and_snapshots_progress() -> None:
+    ctx = _defer_ledger_ctx()
+    assert stash_turn_halt_from_blocker_signal(ctx, _loop_signal(), source="enforcement_backstop") is None
+    assert ctx.turn_halt is None
+    assert ctx.output_contract_defer_progress_token is not None
+
+
+def test_defer_never_expires_granted_grant_awaiting_forced_dispatch() -> None:
+    ctx = _defer_ledger_ctx()
+    stash_turn_halt_from_blocker_signal(ctx, _loop_signal(), source="enforcement_backstop")
+    stash_turn_halt_from_blocker_signal(ctx, _loop_signal(), source="enforcement_backstop")
+    assert ctx.turn_halt is None
+    assert ctx.output_contract_actuation_by_signature["sig_a"] == OutputContractAdvisoryState.GRANTED
+
+
+def test_defer_expires_countonly_ladder_into_typed_terminal() -> None:
+    ctx = _defer_count_ledger_ctx()
+    stash_turn_halt_from_blocker_signal(ctx, _loop_signal(), source="enforcement_backstop")
+    stash_turn_halt_from_blocker_signal(ctx, _loop_signal(), source="enforcement_backstop")
+    assert ctx.turn_halt is not None
+    assert ctx.turn_halt.kind == TurnHaltKind.OUTPUT_SOURCE_UNOBSERVABLE
+    assert ctx.output_contract_actuation_by_signature["sig_a"] == OutputContractAdvisoryState.EXPIRED
+
+
+def test_defer_re_arms_when_lifecycle_advances() -> None:
+    ctx = _defer_ledger_ctx()
+    stash_turn_halt_from_blocker_signal(ctx, _loop_signal(), source="enforcement_backstop")
+    ctx.output_contract_run_output_observed_by_signature["sig_a"] = True
+    stash_turn_halt_from_blocker_signal(ctx, _loop_signal(), source="enforcement_backstop")
+    assert ctx.turn_halt is None
+    assert ctx.output_contract_actuation_by_signature["sig_a"] == OutputContractAdvisoryState.GRANTED
+
+
+def test_defer_countonly_ladder_reaches_a_terminal_within_two_stalled_signals() -> None:
+    ctx = _defer_count_ledger_ctx()
+    for _ in range(6):
+        stash_turn_halt_from_blocker_signal(ctx, _loop_signal(), source="enforcement_backstop")
+        if ctx.turn_halt is not None:
+            break
+    assert ctx.turn_halt is not None

tests/unit/test_copilot_watchdog.py

@@ -465,6 +465,10 @@ def test_reconciliation_guard_message_does_not_say_timed_out() -> None:
         last_test_non_retriable_nav_error=None,
         pending_reconciliation_run_id="wr_guarded",
         synthesized_block_reopened_for_output_coverage=False,
+        output_contract_actuation_by_signature={},
+        output_contract_reject_count_by_signature={},
+        output_contract_actuation_count_by_signature={},
+        output_contract_armed_directive_fingerprint_by_signature={},
     )
     msg = _tool_loop_error(ctx, "update_and_run_blocks")
     assert isinstance(msg, str)