fix(SKY-11854): recycle a scouting-only avoidable output-field-confirmation ASK instead of refusing (#7098)

skyvern/forge/sdk/copilot/agent.py

@@ -91,6 +91,7 @@ from skyvern.forge.sdk.copilot.enforcement import (
     BUILT_UNVERIFIED_REPAIR_INERT_TERMINAL_REASON,
     artifact_health_blocked,
     outcome_fully_verified,
+    recycle_admits_present_completion_contract_ask,
     synthesized_persistence_reopened,
     synthesized_persistence_reopened_after_failed_run,
     synthesized_trajectory_is_goal_complete,
@@ -1633,6 +1634,15 @@ def _build_narrative_payload(
     }
 
 
+def _log_output_policy_parity(ctx: CopilotContext, *, has_workflow_proposal: bool, workflow_attempted: bool) -> None:
+    LOG.info(
+        "copilot.output_policy_parity",
+        has_workflow_proposal=has_workflow_proposal,
+        workflow_attempted=workflow_attempted,
+        **ctx.genuine_attempt_parity_fields(),
+    )
+
+
 def _build_exit_result(
     ctx: CopilotContext,
     user_response: str,
@@ -1648,7 +1658,10 @@ def _build_exit_result(
         blocked_signatures=ctx.blocked_reply_signatures,
         terminal_reason=terminal_reason or ("cancel" if cancelled else None),
     )
-    workflow_attempted = ctx.last_update_block_count is not None or ctx.last_test_ok is not None
+    workflow_attempted = ctx.has_genuine_workflow_attempt()
+    _log_output_policy_parity(
+        ctx, has_workflow_proposal=verified_workflow is not None, workflow_attempted=workflow_attempted
+    )
     output_kind = derive_output_kind(
         response_type="REPLY",
         request_policy=ctx.request_policy,
@@ -3159,7 +3172,10 @@ async def _translate_to_agent_result(
         structured = StructuredContext.from_json_str(llm_context_raw)
     structured.merge_turn_summary(ctx.tool_activity)
     enriched_context = structured.to_json_str()
-    workflow_attempted = ctx.last_update_block_count is not None or ctx.last_test_ok is not None
+    workflow_attempted = ctx.has_genuine_workflow_attempt()
+    _log_output_policy_parity(
+        ctx, has_workflow_proposal=last_workflow is not None, workflow_attempted=workflow_attempted
+    )
     if _should_use_built_unverified_completed_reply(
         ctx,
         response_type=resp_type,
@@ -3507,7 +3523,12 @@ def _evaluate_copilot_final_output_policy(
     elif isinstance(getattr(ctx, "last_workflow_yaml", None), str):
         workflow_yaml = ctx.last_workflow_yaml
 
-    workflow_attempted = ctx.last_update_block_count is not None or ctx.last_test_ok is not None
+    workflow_attempted = ctx.has_genuine_workflow_attempt()
+    _log_output_policy_parity(
+        ctx,
+        has_workflow_proposal=bool(workflow_yaml or ctx.last_workflow is not None),
+        workflow_attempted=workflow_attempted,
+    )
     surface_untested_draft = _should_surface_untested_draft_despite_question(ctx, response_type)
     policy_response_type = "REPLY" if surface_untested_draft else response_type
     if surface_untested_draft:
@@ -3543,6 +3564,9 @@ def _evaluate_copilot_final_output_policy(
         output_kind=output_kind,
     )
     hard_verdict = hard_block_output_policy_verdict(raw_verdict)
+    deferred_reason_codes = _defer_avoidable_ask_to_recycle(ctx, hard_verdict, response_type)
+    if deferred_reason_codes is not None:
+        hard_verdict = OutputPolicyVerdict(allowed=True, output_kind=hard_verdict.output_kind)
     diagnostics = build_output_policy_diagnostics(
         raw_verdict=raw_verdict,
         final_verdict=hard_verdict,
@@ -3552,9 +3576,31 @@ def _evaluate_copilot_final_output_policy(
         hard_block_reason_codes=list(hard_verdict.reason_codes),
         soft_rewrite_reason_codes=[],
     )
+    if deferred_reason_codes is not None:
+        diagnostics["deferred_to_recycle"] = True
+        diagnostics["deferred_reason_codes"] = [reason.value for reason in deferred_reason_codes]
     return hard_verdict, response_type, diagnostics
 
 
+def _defer_avoidable_ask_to_recycle(
+    ctx: CopilotContext,
+    hard_verdict: OutputPolicyVerdict,
+    response_type: str,
+) -> list[OutputPolicyReason] | None:
+    if hard_verdict.allowed or response_type != "ASK_QUESTION":
+        return None
+    if list(hard_verdict.reason_codes) != [OutputPolicyReason.AVOIDABLE_OUTPUT_FIELD_CONFIRMATION]:
+        return None
+    if not recycle_admits_present_completion_contract_ask(ctx):
+        return None
+    LOG.info(
+        "copilot.output_policy_avoidable_deferred_to_recycle",
+        deferred_reason_codes=[reason.value for reason in hard_verdict.reason_codes],
+        **ctx.genuine_attempt_parity_fields(),
+    )
+    return list(hard_verdict.reason_codes)
+
+
 def _build_copilot_input_guardrails(
     InputGuardrailCls: Any,
     GuardrailFunctionOutputCls: Any,
@@ -3780,7 +3826,7 @@ def _build_output_policy_blocked_result(
             workflow_yaml=preserved_workflow_yaml or prior_workflow_yaml,
             has_workflow_proposal=preserved_workflow is not None,
             workflow_was_persisted=ctx.workflow_persisted,
-            workflow_attempted=bool(ctx.last_run_blocks_workflow_run_id),
+            workflow_attempted=ctx.has_genuine_workflow_attempt(),
             unvalidated=ctx.last_test_ok is not True,
             output_kind=verdict.output_kind,
         )

skyvern/forge/sdk/copilot/context.py

@@ -735,3 +735,35 @@ class CopilotContext(AgentContext):
     block_ended_at_map: dict[str, str] = field(default_factory=dict)
     turn_started_at: str | None = None
     turn_ended_at: str | None = None
+
+    def has_genuine_workflow_attempt(self) -> bool:
+        """This turn persisted a workflow proposal or executed a real build-test run; excludes
+        ``test_after_update_done``, which is stamped for any ``run_blocks_and_collect_debug`` scout
+        probe (including early-return probes that record no run) and so is not a genuine-attempt signal."""
+        if self.update_workflow_called:
+            return True
+        if self.last_update_block_count is not None:
+            return True
+        if self.last_test_ok is not None:
+            return True
+        for run_id in (
+            self.last_run_blocks_workflow_run_id,
+            self.last_successful_run_blocks_workflow_run_id,
+            self.last_outcome_gate_workflow_run_id,
+        ):
+            if run_id is not None and run_id.strip():
+                return True
+        return False
+
+    def genuine_attempt_parity_fields(self) -> dict[str, bool | int | str | None]:
+        return {
+            "has_genuine_workflow_attempt": self.has_genuine_workflow_attempt(),
+            "update_workflow_called": self.update_workflow_called,
+            "test_after_update_done": self.test_after_update_done,
+            "last_update_block_count": self.last_update_block_count,
+            "last_test_ok": self.last_test_ok,
+            "last_run_blocks_workflow_run_id": self.last_run_blocks_workflow_run_id,
+            "last_successful_run_blocks_workflow_run_id": self.last_successful_run_blocks_workflow_run_id,
+            "last_outcome_gate_workflow_run_id": self.last_outcome_gate_workflow_run_id,
+            "ctx_last_workflow_present": self.last_workflow is not None,
+        }

skyvern/forge/sdk/copilot/enforcement.py

@@ -992,46 +992,33 @@ def _turn_intent_can_update_and_run_without_user_input(turn_intent: Any) -> bool
     return bool(turn_intent.authority.may_run_blocks)
 
 
-def _has_current_turn_update_or_test_marker(ctx: Any) -> bool:
-    if bool(getattr(ctx, "update_workflow_called", False)):
-        return True
-    if bool(getattr(ctx, "test_after_update_done", False)):
-        return True
-    if getattr(ctx, "last_update_block_count", None) is not None:
-        return True
-    if getattr(ctx, "last_test_ok", None) is not None:
-        return True
-    for field_name in (
-        "last_run_blocks_workflow_run_id",
-        "last_successful_run_blocks_workflow_run_id",
-        "last_outcome_gate_workflow_run_id",
-    ):
-        marker = getattr(ctx, field_name, None)
-        if isinstance(marker, str) and marker.strip():
-            return True
-    return False
+def recycle_admits_present_completion_contract_ask(ctx: CopilotContext) -> bool:
+    request_policy = ctx.request_policy
+    if not isinstance(request_policy, RequestPolicy):
+        return False
+    if not request_policy_has_present_completion_contract(request_policy):
+        return False
+    if request_policy.user_response_policy == "ask_clarification":
+        return False
+    if request_policy.clarification_reason not in (None, "none"):
+        return False
+    if not _turn_intent_can_author_without_user_input(ctx.turn_intent):
+        return False
+    if ctx.has_genuine_workflow_attempt():
+        return False
+    return True
 
 
-def _present_completion_contract_ask_retry(ctx: Any, parsed: dict[str, Any]) -> str | None:
+def _present_completion_contract_ask_retry(ctx: CopilotContext, parsed: dict[str, Any]) -> str | None:
     if parsed.get("type") != "ASK_QUESTION":
         return None
-    request_policy = getattr(ctx, "request_policy", None)
-    if not isinstance(request_policy, RequestPolicy):
-        return None
-    if not request_policy_has_present_completion_contract(request_policy):
-        return None
-    if request_policy.user_response_policy == "ask_clarification":
-        return None
-    if request_policy.clarification_reason not in (None, "none"):
-        return None
-    if not _turn_intent_can_author_without_user_input(getattr(ctx, "turn_intent", None)):
-        return None
-    if _has_current_turn_update_or_test_marker(ctx):
+    if not recycle_admits_present_completion_contract_ask(ctx):
         return None
     LOG.info(
         "copilot.present_completion_contract_ask_retry",
         reason_code="present_completion_contract_ask_internal_retry",
-        turn_intent_mode=getattr(getattr(ctx, "turn_intent", None), "mode", None),
+        turn_intent_mode=ctx.turn_intent.mode if ctx.turn_intent else None,
+        **ctx.genuine_attempt_parity_fields(),
     )
     return PRESENT_COMPLETION_CONTRACT_ASK_RETRY
 

tests/unit/test_copilot_output_policy.py

@@ -1,18 +1,25 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 import json
 from types import SimpleNamespace
+from typing import Any
 from unittest.mock import MagicMock, patch
 
 import pytest
-from agents import ToolInputGuardrailData
+from agents import GuardrailFunctionOutput, OutputGuardrail, ToolInputGuardrailData
+from agents.run_context import RunContextWrapper
 from agents.tool_context import ToolContext
 
 from skyvern.forge.sdk.copilot import agent as agent_module
 from skyvern.forge.sdk.copilot.blocker_signal import CopilotToolBlockerSignal
 from skyvern.forge.sdk.copilot.build_phase import BuildPhase
 from skyvern.forge.sdk.copilot.context import CopilotContext
+from skyvern.forge.sdk.copilot.enforcement import (
+    PRESENT_COMPLETION_CONTRACT_ASK_RETRY,
+    _response_coverage_nudge,
+)
 from skyvern.forge.sdk.copilot.output_policy import (
     CopilotOutputKind,
     OutputPolicyReason,
@@ -23,12 +30,12 @@ from skyvern.forge.sdk.copilot.output_policy import (
     hard_block_output_policy_verdict,
     normalize_response_scaffolding,
 )
-from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
+from skyvern.forge.sdk.copilot.request_policy import CompletionCriterion, RequestPolicy
 from skyvern.forge.sdk.copilot.tools import (
     _WORKFLOW_YAML_OUTPUT_POLICY_GUARDRAIL,
     NATIVE_TOOLS,
 )
-from skyvern.forge.sdk.copilot.turn_intent import TurnIntent, TurnIntentMode
+from skyvern.forge.sdk.copilot.turn_intent import TurnIntent, TurnIntentAuthority, TurnIntentMode
 
 
 def _credential(credential_id: str = "cred_safe", tested_url: str = "https://login.example.test/login") -> object:
@@ -2530,3 +2537,265 @@ def test_rejects_discovered_credential_bound_to_new_origin() -> None:
 
     assert not verdict.allowed
     assert OutputPolicyReason.CREDENTIAL_SCOPE_BROADENED in verdict.reason_codes
+
+
+def _present_contract_confirmation_policy() -> RequestPolicy:
+    return _policy(
+        user_response_policy="proceed",
+        completion_contract_status="present",
+        completion_criteria=[
+            SimpleNamespace(id="record_identity", outcome="The returned record identifies the target record."),
+        ],
+    )
+
+
+def test_genuine_attempt_true_suppresses_avoidable_backstop() -> None:
+    verdict = evaluate_output_policy(
+        request_policy=_present_contract_confirmation_policy(),
+        response_type="ASK_QUESTION",
+        user_response="Please confirm the output fields before I build this record status workflow.",
+        has_workflow_proposal=False,
+        workflow_attempted=True,
+    )
+
+    assert OutputPolicyReason.AVOIDABLE_OUTPUT_FIELD_CONFIRMATION not in verdict.reason_codes
+
+
+def test_genuine_attempt_true_suppresses_unbacked_delivery_claim() -> None:
+    verdict = evaluate_output_policy(
+        request_policy=_policy(),
+        response_type="REPLY",
+        user_response="Here's the workflow.",
+        has_workflow_proposal=False,
+        workflow_attempted=True,
+    )
+
+    assert OutputPolicyReason.UNBACKED_WORKFLOW_DELIVERY_CLAIM not in verdict.reason_codes
+
+
+class _CaptureSentinel(Exception):
+    pass
+
+
+def _run_id_only_ctx() -> CopilotContext:
+    ctx = _ctx()
+    ctx.last_run_blocks_workflow_run_id = "wr_discriminating"
+    ctx.last_update_block_count = None
+    ctx.last_test_ok = None
+    ctx.last_workflow = None
+    return ctx
+
+
+def test_build_exit_result_passes_genuine_predicate_as_workflow_attempted(monkeypatch) -> None:
+    ctx = _run_id_only_ctx()
+    assert ctx.has_genuine_workflow_attempt() is True
+
+    captured: dict[str, Any] = {}
+
+    def _capture_eval(**kwargs: Any) -> None:
+        captured["evaluate"] = kwargs["workflow_attempted"]
+        raise _CaptureSentinel
+
+    def _capture_kind(**kwargs: Any) -> None:
+        captured["derive"] = kwargs["workflow_attempted"]
+        return None
+
+    monkeypatch.setattr(agent_module, "derive_output_kind", _capture_kind)
+    monkeypatch.setattr(agent_module, "evaluate_output_policy", _capture_eval)
+
+    with pytest.raises(_CaptureSentinel):
+        agent_module._build_exit_result(ctx, user_response="Done.", global_llm_context=None)
+
+    assert captured["derive"] is True
+    assert captured["evaluate"] is True
+    assert captured["evaluate"] == ctx.has_genuine_workflow_attempt()
+
+
+def test_watchdog_run_id_only_counts_as_genuine_and_suppresses_unbacked() -> None:
+    ctx = _run_id_only_ctx()
+    legacy_two_field_attempted = ctx.last_update_block_count is not None or ctx.last_test_ok is not None
+    assert legacy_two_field_attempted is False
+    assert ctx.has_genuine_workflow_attempt() is True
+
+    verdict = evaluate_output_policy(
+        request_policy=_policy(),
+        response_type="REPLY",
+        user_response="Here's the workflow.",
+        has_workflow_proposal=False,
+        workflow_attempted=ctx.has_genuine_workflow_attempt(),
+    )
+
+    assert OutputPolicyReason.UNBACKED_WORKFLOW_DELIVERY_CLAIM not in verdict.reason_codes
+
+
+def test_agent_no_longer_derives_workflow_attempted_from_two_marker_fields() -> None:
+    source = inspect.getsource(agent_module)
+    assert "last_update_block_count is not None or ctx.last_test_ok is not None" not in source
+    assert "workflow_attempted=bool(ctx.last_run_blocks_workflow_run_id)" not in source
+
+
+_AVOIDABLE_ASK = {
+    "type": "ASK_QUESTION",
+    "user_response": "Please confirm the output fields before I build this workflow.",
+}
+
+
+def _recoverable_authoring_intent(*, may_update_workflow: bool = True) -> TurnIntent:
+    return TurnIntent(
+        mode=TurnIntentMode.BUILD,
+        authority=TurnIntentAuthority(
+            may_update_workflow=may_update_workflow,
+            may_run_blocks=may_update_workflow,
+            requires_user_input=False,
+        ),
+    )
+
+
+def _present_contract_ask_policy(**overrides: object) -> RequestPolicy:
+    defaults: dict[str, object] = dict(
+        user_response_policy="proceed",
+        clarification_reason="none",
+        completion_contract_status="present",
+        completion_criteria=[
+            CompletionCriterion(id="record_id", outcome="The returned record includes the requested id."),
+        ],
+    )
+    defaults.update(overrides)
+    return _policy(**defaults)
+
+
+def _recoverable_ask_ctx(
+    *,
+    policy: RequestPolicy | None = None,
+    turn_intent: TurnIntent | None = None,
+    **marker: object,
+) -> CopilotContext:
+    ctx = _ctx(
+        request_policy=policy or _present_contract_ask_policy(),
+        turn_intent=turn_intent or _recoverable_authoring_intent(),
+    )
+    for name, value in marker.items():
+        setattr(ctx, name, value)
+    return ctx
+
+
+def test_seam_ordering_avoidable_ask_defers_then_recycle_fires() -> None:
+    ctx = _recoverable_ask_ctx()
+
+    verdict, response_type, diagnostics = agent_module._evaluate_copilot_final_output_policy(ctx, _AVOIDABLE_ASK)
+
+    assert response_type == "ASK_QUESTION"
+    assert verdict.allowed is True
+    assert diagnostics["deferred_to_recycle"] is True
+    assert diagnostics["deferred_reason_codes"] == [OutputPolicyReason.AVOIDABLE_OUTPUT_FIELD_CONFIRMATION.value]
+    assert diagnostics["final_output_policy_allowed"] is True
+
+    assert _response_coverage_nudge(ctx, _AVOIDABLE_ASK) == PRESENT_COMPLETION_CONTRACT_ASK_RETRY
+
+
+@pytest.mark.asyncio
+async def test_sdk_output_guardrail_defers_avoidable_ask_without_tripwire() -> None:
+    output_guardrails = agent_module._build_copilot_output_guardrails(OutputGuardrail, GuardrailFunctionOutput)
+    result = await output_guardrails[0].run(
+        RunContextWrapper(context=_recoverable_ask_ctx()),
+        SimpleNamespace(),
+        _AVOIDABLE_ASK,
+    )
+
+    assert result.output.tripwire_triggered is False
+    assert result.output.output_info["deferred_to_recycle"] is True
+
+
+@pytest.mark.parametrize(
+    ("policy_overrides", "marker", "may_author", "expected_admit"),
+    [
+        pytest.param({}, {}, True, True, id="recoverable_admits"),
+        pytest.param({}, {"test_after_update_done": True}, True, True, id="scout_only_admits"),
+        pytest.param({}, {"last_test_ok": True}, True, False, id="genuine_test"),
+        pytest.param({}, {"last_run_blocks_workflow_run_id": "wr_1"}, True, False, id="genuine_run"),
+        pytest.param({"user_response_policy": "ask_clarification"}, {}, True, False, id="ask_clarification"),
+        pytest.param(
+            {"clarification_reason": "credential_name_unresolved"}, {}, True, False, id="clarification_reason"
+        ),
+        pytest.param({}, {}, False, False, id="non_authoring"),
+    ],
+)
+def test_avoidable_deferral_iff_recycle_admits(
+    policy_overrides: dict[str, object],
+    marker: dict[str, object],
+    may_author: bool,
+    expected_admit: bool,
+) -> None:
+    ctx = _recoverable_ask_ctx(
+        policy=_present_contract_ask_policy(**policy_overrides),
+        turn_intent=_recoverable_authoring_intent(may_update_workflow=may_author),
+        **marker,
+    )
+
+    recycle_admits = _response_coverage_nudge(ctx, _AVOIDABLE_ASK) == PRESENT_COMPLETION_CONTRACT_ASK_RETRY
+    assert recycle_admits is expected_admit
+
+    _, _, diagnostics = agent_module._evaluate_copilot_final_output_policy(ctx, _AVOIDABLE_ASK)
+    assert diagnostics.get("deferred_to_recycle", False) is recycle_admits
+
+
+@pytest.mark.parametrize(
+    ("policy_overrides", "may_author"),
+    [
+        pytest.param({"clarification_reason": "credential_name_unresolved"}, True, id="clarification_reason"),
+        pytest.param({}, False, id="non_authoring"),
+    ],
+)
+def test_backstop_trips_as_failsafe_when_recycle_does_not_admit(
+    policy_overrides: dict[str, object],
+    may_author: bool,
+) -> None:
+    ctx = _recoverable_ask_ctx(
+        policy=_present_contract_ask_policy(**policy_overrides),
+        turn_intent=_recoverable_authoring_intent(may_update_workflow=may_author),
+    )
+
+    verdict, _, diagnostics = agent_module._evaluate_copilot_final_output_policy(ctx, _AVOIDABLE_ASK)
+
+    assert verdict.allowed is False
+    assert OutputPolicyReason.AVOIDABLE_OUTPUT_FIELD_CONFIRMATION in verdict.reason_codes
+    assert diagnostics.get("deferred_to_recycle", False) is False
+
+
+def test_defer_avoidable_ask_requires_ask_shape_and_sole_avoidable_code() -> None:
+    ctx = _recoverable_ask_ctx()
+    avoidable = OutputPolicyReason.AVOIDABLE_OUTPUT_FIELD_CONFIRMATION
+
+    assert agent_module._defer_avoidable_ask_to_recycle(
+        ctx, OutputPolicyVerdict(reason_codes=[avoidable]), "ASK_QUESTION"
+    ) == [avoidable]
+
+    mixed = OutputPolicyVerdict(reason_codes=[avoidable, OutputPolicyReason.RAW_SECRET_LEAK])
+    assert agent_module._defer_avoidable_ask_to_recycle(ctx, mixed, "ASK_QUESTION") is None
+
+    assert (
+        agent_module._defer_avoidable_ask_to_recycle(ctx, OutputPolicyVerdict(reason_codes=[avoidable]), "REPLY")
+        is None
+    )
+
+    wrong = OutputPolicyVerdict(reason_codes=[OutputPolicyReason.PERSISTENCE_STATE_MISMATCH])
+    assert agent_module._defer_avoidable_ask_to_recycle(ctx, wrong, "ASK_QUESTION") is None
+
+    assert agent_module._defer_avoidable_ask_to_recycle(ctx, OutputPolicyVerdict(), "ASK_QUESTION") is None
+
+
+def test_avoidable_ask_with_co_firing_secret_leak_is_not_deferred() -> None:
+    ctx = _recoverable_ask_ctx()
+    ask = {
+        "type": "ASK_QUESTION",
+        "user_response": (
+            "Please confirm the output fields before I build this workflow. I used password: hunter2 to test the login."
+        ),
+    }
+
+    verdict, _, diagnostics = agent_module._evaluate_copilot_final_output_policy(ctx, ask)
+
+    assert verdict.allowed is False
+    assert OutputPolicyReason.AVOIDABLE_OUTPUT_FIELD_CONFIRMATION in verdict.reason_codes
+    assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes
+    assert diagnostics.get("deferred_to_recycle", False) is False

tests/unit/test_copilot_pre_discovery_url_nudge.py

@@ -17,6 +17,7 @@ from types import SimpleNamespace
 import pytest
 
 from skyvern.forge.sdk.copilot.build_phase import BuildPhase
+from skyvern.forge.sdk.copilot.context import CopilotContext
 from skyvern.forge.sdk.copilot.enforcement import (
     MAX_PRE_DISCOVERY_URL_QUESTION_NUDGES,
     PRE_DISCOVERY_URL_QUESTION_NUDGE,
@@ -45,12 +46,19 @@ class _Ctx:
         self.format_nudge_count = 0
         self.test_after_update_done = False
         self.workflow_persisted = False
+        self.last_workflow = None
         self.last_update_block_count = None
         self.last_test_ok = None
         self.last_run_blocks_workflow_run_id = None
         self.last_successful_run_blocks_workflow_run_id = None
         self.last_outcome_gate_workflow_run_id = None
 
+    def has_genuine_workflow_attempt(self) -> bool:
+        return CopilotContext.has_genuine_workflow_attempt(self)  # type: ignore[arg-type]
+
+    def genuine_attempt_parity_fields(self) -> dict[str, bool | int | str | None]:
+        return CopilotContext.genuine_attempt_parity_fields(self)  # type: ignore[arg-type]
+
 
 _URL_ASK = {
     "type": "ASK_QUESTION",
@@ -252,13 +260,51 @@ def test_present_completion_contract_ask_allows_clarification(overrides: dict[st
 @pytest.mark.parametrize(
     "marker",
     [
-        {"update_workflow_called": True},
-        {"last_update_block_count": 1},
-        {"last_test_ok": False},
-        {"last_run_blocks_workflow_run_id": "wr_test"},
+        pytest.param({"update_workflow_called": True}, id="persisted_update"),
+        pytest.param({"last_update_block_count": 1}, id="persisted_block_count"),
+        pytest.param({"last_test_ok": False}, id="failed_build_test"),
+        pytest.param({"last_run_blocks_workflow_run_id": "wr_test"}, id="run_id"),
+        pytest.param(
+            {"last_run_blocks_workflow_run_id": "wr_test", "last_test_ok": None},
+            id="watchdog_softened_run_id",
+        ),
     ],
 )
-def test_present_completion_contract_ask_allows_after_update_or_test_marker(marker: dict[str, object]) -> None:
+def test_present_completion_contract_ask_suppressed_by_genuine_attempt(marker: dict[str, object]) -> None:
     ctx = _present_contract_ctx(**marker)
 
+    assert ctx.has_genuine_workflow_attempt() is True
     assert _response_coverage_nudge(ctx, _OUTPUT_CONFIRMATION_ASK) is None
+
+
+def test_present_completion_contract_ask_admits_after_scout_only_marker() -> None:
+    ctx = _present_contract_ctx(test_after_update_done=True)
+
+    assert ctx.has_genuine_workflow_attempt() is False
+    assert _response_coverage_nudge(ctx, _OUTPUT_CONFIRMATION_ASK) == PRESENT_COMPLETION_CONTRACT_ASK_RETRY
+
+
+_PARITY_MARKER_STATES = [
+    pytest.param({}, id="no_markers"),
+    pytest.param({"test_after_update_done": True}, id="scout_only"),
+    pytest.param({"update_workflow_called": True}, id="persisted_update"),
+    pytest.param({"last_update_block_count": 0}, id="zero_block_count"),
+    pytest.param({"last_test_ok": True}, id="passed_build_test"),
+    pytest.param({"last_test_ok": False}, id="failed_build_test"),
+    pytest.param({"last_run_blocks_workflow_run_id": "wr_1"}, id="run_id"),
+    pytest.param({"last_outcome_gate_workflow_run_id": "wr_2"}, id="outcome_gate_run_id"),
+    pytest.param({"test_after_update_done": True, "last_test_ok": True}, id="scout_and_genuine"),
+]
+
+
+@pytest.mark.parametrize("marker", _PARITY_MARKER_STATES)
+def test_recycle_admission_is_superset_of_backstop_block(marker: dict[str, object]) -> None:
+    ctx = _present_contract_ctx(**marker)
+    workflow_attempted = ctx.has_genuine_workflow_attempt()
+    recycle_admits = _response_coverage_nudge(ctx, _OUTPUT_CONFIRMATION_ASK) == PRESENT_COMPLETION_CONTRACT_ASK_RETRY
+
+    backstop_would_fire = not workflow_attempted
+    if backstop_would_fire:
+        assert recycle_admits
+    else:
+        assert not recycle_admits

tests/unit/test_copilot_run_outcome_event.py

@@ -23,6 +23,7 @@ from skyvern.forge.sdk.copilot.request_policy import CompletionCriterion, Reques
 from skyvern.forge.sdk.copilot.run_outcome import RecordedRunOutcome, run_outcome_display_reason
 from skyvern.forge.sdk.copilot.tools import run_execution
 from skyvern.forge.sdk.copilot.tools.run_execution import (
+    _INTERNAL_RUN_CANCELLED_BY_WATCHDOG_KEY,
     _adjudicated_run_outcome,
     _record_run_blocks_result,
     _stash_recorded_run_outcome,
@@ -607,3 +608,28 @@ def test_narrative_payload_without_recorded_outcome_has_no_outcome_keys() -> Non
     for block in payload["blocks"]:
         assert "outcome" not in block
         assert "outcomeReason" not in block
+
+
+class TestGenuineAttemptRunStamp:
+    def test_ok_run_counts_as_genuine_attempt(self) -> None:
+        ctx = _ctx([_code_block("b0", {"records": [{"id": 1}]})])
+        _record_run_blocks_result(ctx, _run_result([_code_block("b0", {"records": [{"id": 1}]})]))
+        assert ctx.last_test_ok is True
+        assert ctx.last_run_blocks_workflow_run_id == "wr_test"
+        assert ctx.has_genuine_workflow_attempt() is True
+
+    def test_failed_run_counts_as_genuine_attempt(self) -> None:
+        ctx = _ctx([_code_block("b0", {})])
+        _record_run_blocks_result(ctx, _run_result([_code_block("b0", {})], ok=False))
+        assert ctx.last_test_ok is False
+        assert ctx.has_genuine_workflow_attempt() is True
+
+    def test_watchdog_softened_run_counts_as_genuine_attempt(self) -> None:
+        ctx = _ctx([_code_block("b0", {})])
+        ctx.copilot_total_timeout_exceeded = True
+        result = _run_result([_code_block("b0", {})], ok=False)
+        result[_INTERNAL_RUN_CANCELLED_BY_WATCHDOG_KEY] = True
+        _record_run_blocks_result(ctx, result)
+        assert ctx.last_test_ok is None
+        assert ctx.last_run_blocks_workflow_run_id == "wr_test"
+        assert ctx.has_genuine_workflow_attempt() is True

tests/unit/test_copilot_streaming_adapter.py

@@ -878,6 +878,37 @@ def _copilot_ctx() -> Any:
     )
 
 
+class TestGenuineAttemptScoutStamp:
+    def test_run_blocks_scout_stamp_does_not_count_as_genuine_attempt(self) -> None:
+        ctx = _copilot_ctx()
+        _update_enforcement_from_tool(ctx, "run_blocks_and_collect_debug", {"ok": True})
+        assert ctx.test_after_update_done is True
+        assert ctx.has_genuine_workflow_attempt() is False
+
+    def test_failed_run_blocks_scout_stamp_does_not_count_as_genuine_attempt(self) -> None:
+        ctx = _copilot_ctx()
+        _update_enforcement_from_tool(ctx, "run_blocks_and_collect_debug", {"ok": False})
+        assert ctx.test_after_update_done is True
+        assert ctx.has_genuine_workflow_attempt() is False
+
+    def test_persisted_update_counts_as_genuine_attempt(self) -> None:
+        ctx = _copilot_ctx()
+        _update_enforcement_from_tool(ctx, "update_workflow", {"ok": True, "data": {"block_count": 2}})
+        assert ctx.update_workflow_called is True
+        assert ctx.has_genuine_workflow_attempt() is True
+
+    def test_update_and_run_blocks_counts_as_genuine_attempt(self) -> None:
+        ctx = _copilot_ctx()
+        _update_enforcement_from_tool(ctx, "update_and_run_blocks", {"ok": True, "data": {"block_count": 2}})
+        assert ctx.has_genuine_workflow_attempt() is True
+
+    def test_update_without_blocks_is_not_a_genuine_attempt(self) -> None:
+        ctx = _copilot_ctx()
+        _update_enforcement_from_tool(ctx, "update_workflow", {"ok": True, "data": {"block_count": 0}})
+        assert ctx.update_workflow_called is False
+        assert ctx.has_genuine_workflow_attempt() is False
+
+
 class TestCodeRepairProgressStreaming:
     @pytest.mark.asyncio
     async def test_repeated_classified_rejects_collapse_to_one_progress_frame(self) -> None: