SKY-11222: atomic satisfying-run commit barrier for copilot terminalization (#6755)

skyvern/forge/sdk/copilot/agent.py

@@ -127,7 +127,13 @@ from skyvern.forge.sdk.copilot.streaming_adapter import (
 )
 from skyvern.forge.sdk.copilot.tracing_setup import _copilot_model_name, ensure_tracing_initialized, is_tracing_enabled
 from skyvern.forge.sdk.copilot.turn_context import TurnContextAssembler, TurnContextInputs, TurnContextPacket
-from skyvern.forge.sdk.copilot.turn_halt import CopilotTurnHalt, TurnHalt, turn_halt_to_trace_data
+from skyvern.forge.sdk.copilot.turn_halt import (
+    _INVOLUNTARY_BLOCKER_REASON_CODES,
+    CopilotTurnHalt,
+    TurnHalt,
+    raise_if_turn_halt,
+    turn_halt_to_trace_data,
+)
 from skyvern.forge.sdk.copilot.turn_intent import (
     NO_MUTATION_TURN_INTENT_MODES,
     RequiredContextKey,
@@ -1569,6 +1575,72 @@ except ValueError as _fallback_validation_error:
     )
 
 
+def _verified_terminal_preserve_result(
+    ctx: CopilotContext, result: AgentResult, *, exit_site: str
+) -> AgentResult | None:
+    """When the judge confirmed the outcome, hold the tested proposal and the
+    success reply instead of letting an involuntary blocker render over it."""
+    verified_workflow, verified_yaml = _verified_workflow_or_none(ctx)
+    if verified_workflow is None:
+        return None
+    final_text, outcome = apply_repeated_reply_guard(
+        final_text=_VERIFIED_WORKFLOW_SUCCESS_REPLY,
+        attempted_kind=derive_response_kind(ctx.turn_intent),
+        blocked_signatures=ctx.blocked_reply_signatures,
+        terminal_reason="verified_goal_satisfied",
+        turn_intent=ctx.turn_intent,
+    )
+    output_kind = derive_output_kind(
+        response_type="REPLY",
+        request_policy=ctx.request_policy,
+        updated_workflow=verified_workflow,
+        workflow_was_persisted=ctx.workflow_persisted,
+        workflow_attempted=True,
+        unvalidated=False,
+    )
+    verdict = evaluate_output_policy(
+        request_policy=ctx.request_policy,
+        response_type="REPLY",
+        user_response=final_text,
+        global_llm_context=None,
+        workflow_yaml=verified_yaml,
+        has_workflow_proposal=True,
+        workflow_was_persisted=ctx.workflow_persisted,
+        workflow_attempted=True,
+        unvalidated=False,
+        output_kind=output_kind,
+    )
+    if not verdict.allowed:
+        return None
+    LOG.info(
+        "copilot verified outcome preserved tested proposal over blocker",
+        exit_site=exit_site,
+        workflow_permanent_id=ctx.workflow_permanent_id,
+    )
+    return _make_agent_result(
+        ctx,
+        user_response=final_text,
+        updated_workflow=verified_workflow,
+        global_llm_context=result.global_llm_context,
+        response_type="REPLY",
+        workflow_yaml=verified_yaml,
+        workflow_was_persisted=ctx.workflow_persisted,
+        clear_proposed_workflow=False,
+        total_tokens=result.total_tokens,
+        cancelled=result.cancelled,
+        proposal_disposition="review_tested",
+        turn_outcome=outcome,
+        turn_id=ctx.turn_id,
+        narrative_summary=ctx.narrative_summary,
+        narrative_payload=_build_narrative_payload(
+            ctx,
+            terminal="response",
+            terminal_message=final_text,
+            narrative_summary=ctx.narrative_summary,
+        ),
+    )
+
+
 def _finalize_result_with_blocker_override(
     ctx: CopilotContext, result: AgentResult, *, exit_site: str = "unspecified"
 ) -> AgentResult:
@@ -1581,6 +1653,10 @@ def _finalize_result_with_blocker_override(
         return result
     if not local_signal.renders_final_reply:
         return result
+    if local_signal.internal_reason_code in _INVOLUNTARY_BLOCKER_REASON_CODES and outcome_fully_verified(ctx):
+        preserved = _verified_terminal_preserve_result(ctx, result, exit_site=exit_site)
+        if preserved is not None:
+            return preserved
 
     rendered_reply, rendered_resp_type = _render_blocker_reply(local_signal, exit_site=exit_site)
 
@@ -1973,7 +2049,7 @@ def _build_wip_exit_result(
         )
 
     verified_workflow, verified_yaml = _verified_workflow_or_none(ctx)
-    if ctx.verified_terminal_proposal_ready is True and verified_workflow is not None:
+    if outcome_fully_verified(ctx) and verified_workflow is not None:
         final_text, outcome = _guard(tested_reply)
         return _finalize_result_with_blocker_override(
             ctx,
@@ -2189,6 +2265,48 @@ def _build_cancel_exit_result(ctx: CopilotContext, global_llm_context: str | Non
     )
 
 
+async def _resolve_wrapped_exception_exit_result(
+    ctx: CopilotContext,
+    global_llm_context: str | None,
+    *,
+    goal_satisfied: bool,
+    error: BaseException,
+    workflow_permanent_id: str | None,
+) -> AgentResult:
+    error_type = type(error).__name__
+    try:
+        raise_if_turn_halt(ctx, verified=outcome_fully_verified(ctx))
+    except CopilotTurnHalt as halt_exc:
+        LOG.info(
+            "Copilot run stopped after typed turn halt from wrapped exception",
+            workflow_permanent_id=workflow_permanent_id,
+            error_type=error_type,
+            **turn_halt_to_trace_data(halt_exc.halt),
+        )
+        return _build_turn_halt_exit_result(ctx, global_llm_context, halt_exc.halt)
+    turn_halt = ctx.turn_halt
+    if isinstance(turn_halt, TurnHalt):
+        LOG.info(
+            "Copilot run stopped after typed turn halt from wrapped exception",
+            workflow_permanent_id=workflow_permanent_id,
+            error_type=error_type,
+            **turn_halt_to_trace_data(turn_halt),
+        )
+        return _build_turn_halt_exit_result(ctx, global_llm_context, turn_halt)
+    if goal_satisfied:
+        # The Agents SDK can wrap exceptions raised from hooks; keep this
+        # fallback so a verified-goal stop is not rendered as a generic error.
+        LOG.info(
+            "Copilot run stopped after verified goal satisfaction from wrapped exception",
+            workflow_permanent_id=workflow_permanent_id,
+            workflow_run_id=ctx.last_successful_run_blocks_workflow_run_id,
+            error_type=error_type,
+        )
+        return await _build_goal_satisfied_exit_result(ctx, global_llm_context)
+    LOG.error("Copilot agent error", error=str(error), exc_info=True)
+    return _build_unexpected_error_exit_result(ctx, global_llm_context, error=error)
+
+
 async def _translate_to_agent_result(
     result: RunResultStreaming,
     ctx: CopilotContext,
@@ -2388,10 +2506,7 @@ async def _translate_to_agent_result(
             user_response = _rewrite_failed_test_response(str(user_response), ctx)
     verified_workflow, verified_yaml = _verified_workflow_or_none(ctx)
     verified_terminal_ready = (
-        ctx.verified_terminal_proposal_ready is True
-        and verified_workflow is not None
-        and verified_goal_claim_authorized(ctx)
-        and not blocker_active
+        verified_workflow is not None and verified_goal_claim_authorized(ctx) and not blocker_active
     )
     if verified_terminal_ready:
         resp_type = "REPLY"
@@ -2657,7 +2772,7 @@ def _build_infeasibility_clarification_result(
             response_type="ASK_QUESTION",
             workflow_yaml=prior_workflow_yaml or None,
             workflow_was_persisted=False,
-            clear_proposed_workflow=True,
+            clear_proposed_workflow=not outcome_fully_verified(ctx),
             turn_outcome=outcome,
             turn_id=ctx.turn_id,
             narrative_summary=ctx.narrative_summary,
@@ -2708,7 +2823,7 @@ def _build_request_policy_clarification_result(
             response_type="ASK_QUESTION",
             workflow_yaml=prior_workflow_yaml or None,
             workflow_was_persisted=False,
-            clear_proposed_workflow=True,
+            clear_proposed_workflow=not outcome_fully_verified(ctx),
             turn_outcome=outcome,
             turn_id=ctx.turn_id,
             narrative_summary=ctx.narrative_summary,
@@ -3727,24 +3842,10 @@ async def _run_copilot_turn_impl(
         except Exception:
             LOG.error("Copilot agent error", error=str(e), exc_info=True)
             return _build_unexpected_error_exit_result(ctx, global_llm_context, error=e)
-        turn_halt = getattr(ctx, "turn_halt", None)
-        if isinstance(turn_halt, TurnHalt):
-            LOG.info(
-                "Copilot run stopped after typed turn halt from wrapped exception",
-                workflow_permanent_id=chat_request.workflow_permanent_id,
-                error_type=type(e).__name__,
-                **turn_halt_to_trace_data(turn_halt),
-            )
-            return _build_turn_halt_exit_result(ctx, global_llm_context, turn_halt)
-        if goal_satisfied:
-            # The Agents SDK can wrap exceptions raised from hooks; keep this
-            # fallback so a verified-goal stop is not rendered as a generic error.
-            LOG.info(
-                "Copilot run stopped after verified goal satisfaction from wrapped exception",
-                workflow_permanent_id=chat_request.workflow_permanent_id,
-                workflow_run_id=ctx.last_successful_run_blocks_workflow_run_id,
-                error_type=type(e).__name__,
-            )
-            return await _build_goal_satisfied_exit_result(ctx, global_llm_context)
-        LOG.error("Copilot agent error", error=str(e), exc_info=True)
-        return _build_unexpected_error_exit_result(ctx, global_llm_context, error=e)
+        return await _resolve_wrapped_exception_exit_result(
+            ctx,
+            global_llm_context,
+            goal_satisfied=goal_satisfied,
+            error=e,
+            workflow_permanent_id=chat_request.workflow_permanent_id,
+        )

skyvern/forge/sdk/copilot/blocker_signal.py

@@ -319,6 +319,22 @@ def clear_blocker_signal_for_reason_codes(ctx: _BlockerSignalCtx, internal_reaso
         ctx.blocker_signal = None
 
 
+def clear_tool_blocker_signals_for_reason_codes(ctx: _BlockerSignalCtx, internal_reason_codes: frozenset[str]) -> None:
+    clear_blocker_signal_for_reason_codes(ctx, internal_reason_codes)
+    # getattr matches stash_blocker_signal's defensive read: real contexts type
+    # both fields, but partial test shims may omit them.
+    latest = getattr(ctx, "latest_tool_blocker_signal", None)
+    if isinstance(latest, CopilotToolBlockerSignal) and latest.internal_reason_code in internal_reason_codes:
+        ctx.latest_tool_blocker_signal = None
+    history = getattr(ctx, "tool_blocker_signals", None)
+    if isinstance(history, list):
+        history[:] = [
+            entry
+            for entry in history
+            if not (isinstance(entry, CopilotToolBlockerSignal) and entry.internal_reason_code in internal_reason_codes)
+        ]
+
+
 def stash_blocker_signal(ctx: _BlockerSignalCtx, signal: CopilotToolBlockerSignal) -> str:
     """Mostly first-wins stash + observability log; returns the LLM-visible payload."""
     ctx.latest_tool_blocker_signal = signal

skyvern/forge/sdk/copilot/diagnosis_repair_contract.py

@@ -14,6 +14,7 @@ from skyvern.forge.sdk.copilot.failure_tracking import (
 from skyvern.forge.sdk.copilot.output_policy import url_origin
 from skyvern.forge.sdk.copilot.request_policy import redact_raw_secrets_for_prompt
 from skyvern.forge.sdk.copilot.run_outcome import trusted_terminal_challenge_category_name
+from skyvern.forge.sdk.copilot.terminal_predicates import outcome_fully_verified
 from skyvern.forge.sdk.copilot.workflow_credential_utils import URL_CANDIDATE_RE
 
 if TYPE_CHECKING:
@@ -420,6 +421,10 @@ def _failure_type(
 def _next_action(failure_type: DiagnosisFailureType, ctx: CopilotContext, data: dict[str, Any]) -> RepairNextAction:
     if failure_type == DiagnosisFailureType.NO_FAILURE:
         return RepairNextAction.NO_CHANGE
+    # The judge confirmed every criterion; no failure_type may drive a repair
+    # that overwrites an already-verified terminal proposal.
+    if outcome_fully_verified(ctx):
+        return RepairNextAction.NO_CHANGE
     if (
         data.get("skip_reason") == "workflow_credential_inputs_unbound"
         or failure_type == DiagnosisFailureType.MISSING_CREDENTIAL_OR_INIT

skyvern/forge/sdk/copilot/enforcement.py

@@ -76,6 +76,11 @@ from skyvern.forge.sdk.copilot.run_outcome import (
     run_outcome_display_reason,
 )
 from skyvern.forge.sdk.copilot.screenshot_utils import ScreenshotEntry
+from skyvern.forge.sdk.copilot.terminal_predicates import (
+    artifact_health_blocked,
+    outcome_criteria_evaluated,
+    outcome_fully_verified,
+)
 from skyvern.forge.sdk.copilot.tracing_setup import copilot_span
 from skyvern.forge.sdk.copilot.turn_halt import (
     raise_if_turn_halt,
@@ -558,28 +563,7 @@ def latest_diagnosis_contract_satisfies_goal(ctx: CopilotContext) -> bool:
 
 
 def _outcome_criteria_evaluated(ctx: CopilotContext) -> bool:
-    result = ctx.completion_verification_result
-    return result is not None and result.status == "evaluated"
-
-
-def artifact_health_blocked(ctx: CopilotContext) -> bool:
-    reason = ctx.last_artifact_health_blocker_reason
-    return isinstance(reason, str) and bool(reason.strip())
-
-
-def outcome_fully_verified(ctx: CopilotContext) -> bool:
-    """The judge confirmed every outcome criterion from the evidence this run produced.
-
-    This evidence is authoritative over run status: a run that reached the goal is
-    recognized even when it was canceled or only partially completed. Run status must
-    never suppress recognition of an outcome the user can observe was achieved.
-    """
-    if artifact_health_blocked(ctx):
-        return False
-    if not _outcome_criteria_evaluated(ctx):
-        return False
-    result = ctx.completion_verification_result
-    return result is not None and result.is_fully_satisfied()
+    return outcome_criteria_evaluated(ctx)
 
 
 def verified_goal_satisfied_context(ctx: CopilotContext) -> bool:
@@ -1263,14 +1247,15 @@ def _check_enforcement(
     result: RunResultStreaming | None = None,
     config: CopilotConfig | None = None,
 ) -> str | None:
+    verified = outcome_fully_verified(ctx)
     # Terminal failure-mode signals must pre-empt tool-call hygiene nudges.
     terminal_signal = getattr(ctx, "latest_tool_blocker_signal", None) or getattr(ctx, "blocker_signal", None)
     if terminal_signal is not None:
         stash_turn_halt_from_blocker_signal(ctx, terminal_signal, source="enforcement_backstop")
-    raise_if_turn_halt(ctx)
+    raise_if_turn_halt(ctx, verified=verified)
     _raise_if_unrecoverable_contract_stop(ctx)
 
-    if getattr(ctx, "verified_terminal_proposal_ready", False) is True and verified_goal_satisfied_context(ctx):
+    if verified:
         raise CopilotGoalSatisfied()
 
     if _needs_repair_ceiling_halt(ctx):
@@ -1284,7 +1269,7 @@ def _check_enforcement(
             signal,
             consecutive_identical_repair_count=(state.consecutive_identical_repair_count if state is not None else 0),
         )
-        raise_if_turn_halt(ctx)
+        raise_if_turn_halt(ctx, verified=verified)
 
     # A permanent navigation error (DNS / cert / SSL / invalid URL) cannot be
     # resolved by observing a prior navigate or by testing an updated
@@ -1314,7 +1299,7 @@ def _check_enforcement(
         return None
 
     _maybe_stash_terminal_challenge_halt(ctx)
-    raise_if_turn_halt(ctx)
+    raise_if_turn_halt(ctx, verified=verified)
 
     # If the last run had confirmed challenge evidence, do not misdiagnose a
     # challenge-solving loop as a long-chain budgeting problem.
@@ -1364,7 +1349,7 @@ def _check_enforcement(
         signal = _probable_site_block_stop_signal(ctx, config)
         stash_blocker_signal(ctx, signal)
         stash_turn_halt_from_blocker_signal(ctx, signal, source="enforcement")
-        raise_if_turn_halt(ctx)
+        raise_if_turn_halt(ctx, verified=verified)
 
     if _needs_failed_test_nudge(ctx):
         ctx.failed_test_nudge_count += 1

skyvern/forge/sdk/copilot/hooks.py

@@ -143,7 +143,7 @@ class CopilotRunHooks(RunHooksBase):
             getattr(self._ctx, "latest_tool_blocker_signal", None) or getattr(self._ctx, "blocker_signal", None),
             source="hook",
         )
-        raise_if_turn_halt(self._ctx)
+        raise_if_turn_halt(self._ctx, verified=outcome_fully_verified(self._ctx))
 
         if _tool_completion_satisfies_turn(self._ctx, tool_name, parsed):
             LOG.info(

skyvern/forge/sdk/copilot/streaming_adapter.py

@@ -101,6 +101,7 @@ async def stream_to_sse(
     from skyvern.forge.sdk.copilot.enforcement import (
         CopilotUnrecoverableToolError,
         _maybe_raise_unrecoverable_tool_error,
+        outcome_fully_verified,
     )
     from skyvern.forge.sdk.copilot.turn_halt import (
         CopilotTurnHalt,
@@ -269,7 +270,7 @@ async def stream_to_sse(
                     source="streaming_adapter",
                 )
                 try:
-                    raise_if_turn_halt(ctx)
+                    raise_if_turn_halt(ctx, verified=outcome_fully_verified(ctx))
                 except CopilotTurnHalt:
                     result.cancel()
                     raise

skyvern/forge/sdk/copilot/terminal_predicates.py

@@ -0,0 +1,37 @@
+"""Leaf predicates for terminal outcome verification.
+
+These read only ``CopilotContext`` fields and have no copilot-module imports, so
+the diagnosis, enforcement, and agent layers can all key barrier decisions on the
+same judge verdict without an import cycle.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from skyvern.forge.sdk.copilot.context import CopilotContext
+
+
+def artifact_health_blocked(ctx: CopilotContext) -> bool:
+    reason = ctx.last_artifact_health_blocker_reason
+    return isinstance(reason, str) and bool(reason.strip())
+
+
+def outcome_criteria_evaluated(ctx: CopilotContext) -> bool:
+    result = ctx.completion_verification_result
+    return result is not None and result.status == "evaluated"
+
+
+def outcome_fully_verified(ctx: CopilotContext) -> bool:
+    """Whether the judge confirmed every outcome criterion from this run's evidence.
+
+    The verdict is authoritative over run status: a run that reached the goal is
+    recognized even when it was canceled or only partially completed.
+    """
+    if artifact_health_blocked(ctx):
+        return False
+    if not outcome_criteria_evaluated(ctx):
+        return False
+    result = ctx.completion_verification_result
+    return result is not None and result.is_fully_satisfied()

skyvern/forge/sdk/copilot/tools/completion.py

@@ -6,6 +6,7 @@ import structlog
 from skyvern.config import settings
 from skyvern.forge.sdk.copilot.completion_criteria_store import note_adjudication_on_turn_state
 from skyvern.forge.sdk.copilot.completion_verification import (
+    _STRUCTURED_RECORD_CRITERION_IDS,
     CompletionVerificationResult,
     CriterionVerdict,
     RunEvidenceSnapshot,
@@ -542,6 +543,30 @@ async def _maybe_run_completion_verification(
         if definition_criteria
         else []
     )
+    if run_criteria and all(criterion.id in _STRUCTURED_RECORD_CRITERION_IDS for criterion in run_criteria):
+        # Classifier-fallback criteria are value-agnostic (graded on record shape, not the
+        # requested entity) and the judge cannot disambiguate them either, so a well-shaped record
+        # for the wrong entity must not read as verified. Treat the run plane as criteria-less and
+        # surface only a structural contradiction as a suspicious-success signal.
+        snapshot = _build_run_evidence_snapshot(copilot_ctx, result)
+        contradictions = [
+            verdict for verdict in grade_structured_record_criteria(run_criteria, snapshot) if not verdict.satisfied
+        ]
+        if contradictions:
+            return combine_verification_results(
+                criterion_ids,
+                CompletionVerificationResult(
+                    status="evaluated",
+                    criterion_ids=[criterion.id for criterion in run_criteria],
+                    verdicts=contradictions,
+                ),
+                definition_verdicts,
+            )
+        if not definition_verdicts:
+            return None
+        return combine_verification_results(
+            [criterion.id for criterion in definition_criteria], None, definition_verdicts
+        )
     if not run_criteria:
         return combine_verification_results(criterion_ids, None, definition_verdicts)
     snapshot = _build_run_evidence_snapshot(copilot_ctx, result)

skyvern/forge/sdk/copilot/tools/run_execution.py

@@ -2108,10 +2108,9 @@ def _record_run_blocks_result(
                 ),
             )
         if completion_fully_satisfied:
-            # Terminal proposal promotion is authoritative only after the same
-            # run has satisfied completion verification; clear stale
-            # suspicious-success state before agent/enforcement exits consume
-            # ``verified_terminal_proposal_ready``.
+            # ``verified_terminal_proposal_ready`` is telemetry only (the barrier keys
+            # on ``outcome_fully_verified(ctx)``); clearing the stale suspicious-success
+            # state below is the load-bearing step.
             copilot_ctx.verified_terminal_proposal_ready = True
             copilot_ctx.last_test_suspicious_success = False
             copilot_ctx.last_test_failure_reason = None

skyvern/forge/sdk/copilot/turn_halt.py

@@ -8,13 +8,18 @@ from typing import Any
 
 import structlog
 
-from skyvern.forge.sdk.copilot.blocker_signal import CopilotToolBlockerSignal
+from skyvern.forge.sdk.copilot.blocker_signal import (
+    CopilotToolBlockerSignal,
+    clear_tool_blocker_signals_for_reason_codes,
+)
 from skyvern.forge.sdk.copilot.blocker_signal import to_trace_data as blocker_signal_to_trace_data
 from skyvern.forge.sdk.copilot.failure_tracking import ACTIVE_RUN_TERMINAL_EVIDENCE_REASON_CODE
 from skyvern.forge.sdk.copilot.run_outcome import TERMINAL_CHALLENGE_BLOCKER_REASON_CODE
 
 LOG = structlog.get_logger()
 
+REPAIR_CEILING_REASON_CODE = "repair_ceiling_reached"
+
 
 class TurnHaltKind(StrEnum):
     LOOP_DETECTED = "loop_detected"
@@ -61,6 +66,21 @@ def blocker_signal_is_genuinely_terminal(signal: CopilotToolBlockerSignal | None
     return signal is not None and signal.internal_reason_code in GENUINELY_TERMINAL_BLOCKER_REASON_CODES
 
 
+# Halts the agent did not choose: a verified outcome may suppress these.
+# ACTIVE_TERMINAL_CHALLENGE is voluntary and is deliberately excluded so a
+# future terminal kind defaults to raising rather than being suppressed.
+_INVOLUNTARY_TURN_HALT_KINDS = frozenset(
+    {
+        TurnHaltKind.LOOP_DETECTED,
+        TurnHaltKind.PROBABLE_SITE_BLOCK,
+        TurnHaltKind.REPAIR_CEILING_REACHED,
+    }
+)
+_INVOLUNTARY_BLOCKER_REASON_CODES = (
+    _LOOP_TERMINAL_REASON_CODES | _PROBABLE_SITE_BLOCK_REASON_CODES | frozenset({REPAIR_CEILING_REASON_CODE})
+)
+
+
 @dataclass(frozen=True)
 class TurnHalt:
     kind: TurnHaltKind
@@ -139,10 +159,26 @@ def stash_repair_ceiling_turn_halt(
     return halt
 
 
-def raise_if_turn_halt(ctx: Any) -> None:
+def raise_if_turn_halt(ctx: Any, *, verified: bool = False) -> None:
+    """Raise the stashed turn halt unless a verified outcome suppresses it.
+
+    A judge-confirmed outcome suppresses an involuntary halt and consumes both
+    ``ctx.turn_halt`` and the matching involuntary ``ctx.blocker_signal``;
+    ``verified`` defaults False so an un-updated caller raises rather than
+    falsely suppressing.
+    """
     halt = getattr(ctx, "turn_halt", None)
-    if isinstance(halt, TurnHalt):
-        raise CopilotTurnHalt(halt)
+    if not isinstance(halt, TurnHalt):
+        return
+    if verified and halt.kind in _INVOLUNTARY_TURN_HALT_KINDS:
+        ctx.turn_halt = None
+        clear_tool_blocker_signals_for_reason_codes(ctx, _INVOLUNTARY_BLOCKER_REASON_CODES)
+        LOG.info(
+            "copilot turn halt suppressed by verified outcome",
+            **turn_halt_to_trace_data(halt),
+        )
+        return
+    raise CopilotTurnHalt(halt)
 
 
 def turn_halt_to_trace_data(halt: TurnHalt) -> dict[str, Any]:

tests/unit/test_copilot_agent_blocker_renderer.py

@@ -9,6 +9,7 @@ from skyvern.forge.sdk.copilot import agent as agent_module
 from skyvern.forge.sdk.copilot.agent import (
     _FALLBACK_BLOCKER_REPLY,
     _RAW_SECRET_LEAK_REFUSAL,
+    _VERIFIED_WORKFLOW_SUCCESS_REPLY,
     _build_output_policy_blocked_result,
     _build_turn_halt_exit_result,
     _finalize_result_with_blocker_override,
@@ -21,10 +22,11 @@ from skyvern.forge.sdk.copilot.blocker_signal import (
     CopilotToolBlockerSignal,
     RecoveryHint,
 )
+from skyvern.forge.sdk.copilot.completion_verification import CompletionVerificationResult, CriterionVerdict
 from skyvern.forge.sdk.copilot.context import AgentResult, CopilotContext
 from skyvern.forge.sdk.copilot.output_policy import CopilotOutputKind, OutputPolicyReason, OutputPolicyVerdict
 from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
-from skyvern.forge.sdk.copilot.run_outcome import RecordedRunOutcome
+from skyvern.forge.sdk.copilot.run_outcome import TERMINAL_CHALLENGE_BLOCKER_REASON_CODE, RecordedRunOutcome
 from skyvern.forge.sdk.copilot.turn_halt import TurnHalt, TurnHaltKind
 
 # Source-of-truth deny list lives in blocker_signal.py. Re-importing here
@@ -564,3 +566,95 @@ def test_turn_halt_exit_renders_terminal_reason_when_terminal_blocker_held() ->
     result = _build_turn_halt_exit_result(ctx, global_llm_context=None, halt=halt)
 
     assert result.user_response == terminal.user_facing_reason
+
+
+def _fully_satisfied_result() -> CompletionVerificationResult:
+    return CompletionVerificationResult(
+        status="evaluated",
+        criterion_ids=["c0"],
+        verdicts=[CriterionVerdict(criterion_id="c0", state="satisfied", reason_code="evidence_confirms")],
+    )
+
+
+def _seed_verified_outcome(ctx: CopilotContext) -> None:
+    ctx.completion_verification_result = _fully_satisfied_result()
+    ctx.last_artifact_health_blocker_reason = None
+    ctx.last_workflow = SimpleNamespace(workflow_definition=SimpleNamespace(blocks=[SimpleNamespace()]))
+    ctx.last_workflow_yaml = "title: built\nblocks: []\n"
+
+
+def test_verified_outcome_preserves_tested_proposal_over_blocker() -> None:
+    ctx = _ctx()
+    _seed_verified_outcome(ctx)
+    ctx.blocker_signal = _signal(
+        kind="loop_detected",
+        user_facing="I'm stuck retrying the same step.",
+        internal_reason_code="loop_detected_repeated_failed_step",
+        blocked_tool="update_and_run_blocks",
+    )
+    result = _agent_result()
+
+    overridden = _finalize_result_with_blocker_override(ctx, result)
+
+    assert overridden.user_response == _VERIFIED_WORKFLOW_SUCCESS_REPLY
+    assert overridden.updated_workflow is ctx.last_workflow
+    assert overridden.workflow_yaml == ctx.last_workflow_yaml
+    assert overridden.clear_proposed_workflow is False
+    assert overridden.proposal_disposition == "review_tested"
+    assert overridden.response_type == "REPLY"
+    assert "stuck retrying" not in overridden.user_response
+
+
+def test_verified_outcome_preserve_sets_verified_success_payload() -> None:
+    ctx = _ctx()
+    ctx.turn_id = "turn-verified"
+    _seed_verified_outcome(ctx)
+    ctx.blocker_signal = _signal(
+        kind="loop_detected",
+        user_facing="I'm stuck retrying the same step.",
+        internal_reason_code="loop_detected_repeated_failed_step",
+        blocked_tool="update_and_run_blocks",
+    )
+
+    overridden = _finalize_result_with_blocker_override(ctx, _agent_result())
+
+    assert overridden.narrative_payload is not None
+    assert overridden.narrative_payload["verifiedSuccess"] is True
+    assert overridden.narrative_payload["proposalDisposition"] == "review_tested"
+
+
+def test_unverified_blocker_still_renders_blocker_text() -> None:
+    ctx = _ctx()
+    ctx.completion_verification_result = None
+    ctx.last_workflow = SimpleNamespace(workflow_definition=SimpleNamespace(blocks=[SimpleNamespace()]))
+    ctx.last_workflow_yaml = "title: built\nblocks: []\n"
+    ctx.blocker_signal = _signal(
+        kind="loop_detected",
+        user_facing="I'm stuck retrying the same step.",
+        internal_reason_code="loop_detected_repeated_failed_step",
+        blocked_tool="update_and_run_blocks",
+    )
+
+    overridden = _finalize_result_with_blocker_override(ctx, _agent_result())
+
+    assert overridden.user_response != _VERIFIED_WORKFLOW_SUCCESS_REPLY
+    assert overridden.proposal_disposition != "review_tested"
+
+
+def test_verified_outcome_does_not_suppress_voluntary_terminal_challenge() -> None:
+    ctx = _ctx()
+    _seed_verified_outcome(ctx)
+    challenge_text = "The site requires a verification challenge I can't complete on my own."
+    ctx.blocker_signal = _signal(
+        kind="tool_error",
+        user_facing=challenge_text,
+        recovery_hint="stop",
+        internal_reason_code=TERMINAL_CHALLENGE_BLOCKER_REASON_CODE,
+        blocked_tool="update_and_run_blocks",
+    )
+
+    overridden = _finalize_result_with_blocker_override(ctx, _agent_result())
+
+    assert overridden.user_response == challenge_text
+    assert overridden.user_response != _VERIFIED_WORKFLOW_SUCCESS_REPLY
+    assert overridden.proposal_disposition != "review_tested"

tests/unit/test_copilot_agent_helpers.py

@@ -13,6 +13,12 @@ import yaml
 
 from skyvern.forge.sdk.api.llm.exceptions import LLMProviderError
 from skyvern.forge.sdk.copilot import agent as agent_module
+from skyvern.forge.sdk.copilot.agent import (
+    _VERIFIED_WORKFLOW_SUCCESS_REPLY,
+    _build_goal_satisfied_exit_result,
+    _resolve_wrapped_exception_exit_result,
+)
+from skyvern.forge.sdk.copilot.blocker_signal import CopilotToolBlockerSignal
 from skyvern.forge.sdk.copilot.build_phase import BuildPhase
 from skyvern.forge.sdk.copilot.completion_verification import CompletionVerificationResult, CriterionVerdict
 from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy, CopilotConfig
@@ -29,6 +35,7 @@ from skyvern.forge.sdk.copilot.enforcement import (
     CopilotNonRetriableNavError,
     CopilotTotalTimeoutError,
     CopilotUnrecoverableToolError,
+    verified_goal_satisfied_context,
 )
 from skyvern.forge.sdk.copilot.recoverable_failure import build_recoverable_failure
 from skyvern.forge.sdk.copilot.request_policy import (
@@ -40,8 +47,14 @@ from skyvern.forge.sdk.copilot.request_policy import (
     build_transcript_context,
     redact_raw_secrets_for_prompt,
 )
-from skyvern.forge.sdk.copilot.run_outcome import RecordedRunOutcome
+from skyvern.forge.sdk.copilot.run_outcome import TERMINAL_CHALLENGE_BLOCKER_REASON_CODE, RecordedRunOutcome
 from skyvern.forge.sdk.copilot.turn_context import TranscriptContext, TurnContextOmission, TurnContextPacket
+from skyvern.forge.sdk.copilot.turn_halt import (
+    TurnHalt,
+    TurnHaltKind,
+    raise_if_turn_halt,
+    stash_repair_ceiling_turn_halt,
+)
 from skyvern.forge.sdk.copilot.turn_intent import (
     TurnIntent,
     TurnIntentAuthority,
@@ -785,6 +798,159 @@ class TestVerifiedGoalSatisfiedStop:
 
         assert verified_goal_satisfied_context(ctx)
 
+    @pytest.mark.asyncio
+    async def test_wrapped_exception_fallback_reaches_goal_satisfied_after_verified_consume(self) -> None:
+        ctx = _ctx(
+            last_workflow=SimpleNamespace(workflow_definition=SimpleNamespace(blocks=[SimpleNamespace()])),
+            last_workflow_yaml="workflow_definition:\n  blocks: []\n",
+            last_test_ok=True,
+            last_full_workflow_test_ok=True,
+            latest_diagnosis_repair_contract=_verified_goal_contract(),
+            tool_activity=[{"tool": "update_and_run_blocks", "summary": "OK"}],
+        )
+        ctx.completion_verification_result = CompletionVerificationResult(
+            status="evaluated",
+            criterion_ids=["c0"],
+            verdicts=[CriterionVerdict(criterion_id="c0", state="satisfied", reason_code="evidence_confirms")],
+        )
+        signal = CopilotToolBlockerSignal(
+            blocker_kind="tool_error",
+            agent_steering_text="repair ceiling",
+            user_facing_reason="I could not get the run to pass after several repair attempts.",
+            recovery_hint="report_blocker_to_user",
+            preserves_workflow_draft=True,
+            internal_reason_code="repair_ceiling_reached",
+            blocked_tool="update_and_run_blocks",
+        )
+        ctx.blocker_signal = signal
+        stash_repair_ceiling_turn_halt(ctx, signal, consecutive_identical_repair_count=3)
+
+        raise_if_turn_halt(ctx, verified=True)
+
+        assert not isinstance(ctx.turn_halt, TurnHalt)
+        assert verified_goal_satisfied_context(ctx)
+
+        result = await _build_goal_satisfied_exit_result(ctx, global_llm_context=None)
+
+        assert result.user_response == _VERIFIED_WORKFLOW_SUCCESS_REPLY
+        assert result.proposal_disposition != "no_proposal"
+
+    @pytest.mark.asyncio
+    async def test_wrapped_exception_resolver_renders_success_over_involuntary_halt(self) -> None:
+        ctx = _ctx(
+            last_workflow=SimpleNamespace(workflow_definition=SimpleNamespace(blocks=[SimpleNamespace()])),
+            last_workflow_yaml="workflow_definition:\n  blocks: []\n",
+            last_test_ok=True,
+            last_full_workflow_test_ok=True,
+            latest_diagnosis_repair_contract=_verified_goal_contract(),
+            tool_activity=[{"tool": "update_and_run_blocks", "summary": "OK"}],
+        )
+        ctx.completion_verification_result = CompletionVerificationResult(
+            status="evaluated",
+            criterion_ids=["c0"],
+            verdicts=[CriterionVerdict(criterion_id="c0", state="satisfied", reason_code="evidence_confirms")],
+        )
+        signal = CopilotToolBlockerSignal(
+            blocker_kind="tool_error",
+            agent_steering_text="repair ceiling",
+            user_facing_reason="I could not get the run to pass after several repair attempts.",
+            recovery_hint="report_blocker_to_user",
+            preserves_workflow_draft=True,
+            internal_reason_code="repair_ceiling_reached",
+            blocked_tool="update_and_run_blocks",
+        )
+        ctx.blocker_signal = signal
+        stash_repair_ceiling_turn_halt(ctx, signal, consecutive_identical_repair_count=3)
+
+        result = await _resolve_wrapped_exception_exit_result(
+            ctx,
+            global_llm_context=None,
+            goal_satisfied=True,
+            error=RuntimeError("sdk-wrapped hook exception"),
+            workflow_permanent_id="wfp-1",
+        )
+
+        assert result.user_response == _VERIFIED_WORKFLOW_SUCCESS_REPLY
+        assert result.proposal_disposition != "no_proposal"
+        assert result.updated_workflow is ctx.last_workflow
+        assert not isinstance(ctx.turn_halt, TurnHalt)
+        assert ctx.blocker_signal is None
+        assert signal.user_facing_reason not in result.user_response
+
+    @pytest.mark.asyncio
+    async def test_wrapped_exception_resolver_surfaces_voluntary_challenge(self) -> None:
+        ctx = _ctx(
+            last_workflow=SimpleNamespace(workflow_definition=SimpleNamespace(blocks=[SimpleNamespace()])),
+            last_workflow_yaml="workflow_definition:\n  blocks: []\n",
+            last_test_ok=True,
+            last_full_workflow_test_ok=True,
+            latest_diagnosis_repair_contract=_verified_goal_contract(),
+            tool_activity=[{"tool": "update_and_run_blocks", "summary": "OK"}],
+        )
+        ctx.completion_verification_result = CompletionVerificationResult(
+            status="evaluated",
+            criterion_ids=["c0"],
+            verdicts=[CriterionVerdict(criterion_id="c0", state="satisfied", reason_code="evidence_confirms")],
+        )
+        challenge_text = "The site requires a verification challenge I can't complete on my own."
+        signal = CopilotToolBlockerSignal(
+            blocker_kind="tool_error",
+            agent_steering_text="stop on terminal challenge",
+            user_facing_reason=challenge_text,
+            recovery_hint="stop",
+            internal_reason_code=TERMINAL_CHALLENGE_BLOCKER_REASON_CODE,
+            blocked_tool="update_and_run_blocks",
+        )
+        ctx.blocker_signal = signal
+        ctx.turn_halt = TurnHalt(kind=TurnHaltKind.ACTIVE_TERMINAL_CHALLENGE, blocker_signal=signal)
+
+        result = await _resolve_wrapped_exception_exit_result(
+            ctx,
+            global_llm_context=None,
+            goal_satisfied=True,
+            error=RuntimeError("sdk-wrapped hook exception"),
+            workflow_permanent_id="wfp-1",
+        )
+
+        assert result.user_response == challenge_text
+        assert result.user_response != _VERIFIED_WORKFLOW_SUCCESS_REPLY
+
+    @pytest.mark.asyncio
+    async def test_wrapped_exception_resolver_renders_blocker_when_not_judge_verified(self) -> None:
+        ctx = _ctx(
+            last_workflow=SimpleNamespace(workflow_definition=SimpleNamespace(blocks=[SimpleNamespace()])),
+            last_workflow_yaml="workflow_definition:\n  blocks: []\n",
+            last_test_ok=True,
+            last_full_workflow_test_ok=True,
+            latest_diagnosis_repair_contract=_verified_goal_contract(),
+            tool_activity=[{"tool": "update_and_run_blocks", "summary": "OK"}],
+        )
+        # The broad legacy gate passes (goal_satisfied) but the judge confirmed nothing, so
+        # outcome_fully_verified is False and the involuntary halt must not be suppressed.
+        ctx.completion_verification_result = None
+        signal = CopilotToolBlockerSignal(
+            blocker_kind="tool_error",
+            agent_steering_text="repair ceiling",
+            user_facing_reason="I could not get the run to pass after several repair attempts.",
+            recovery_hint="report_blocker_to_user",
+            preserves_workflow_draft=True,
+            internal_reason_code="repair_ceiling_reached",
+            blocked_tool="update_and_run_blocks",
+        )
+        ctx.blocker_signal = signal
+        stash_repair_ceiling_turn_halt(ctx, signal, consecutive_identical_repair_count=3)
+
+        result = await _resolve_wrapped_exception_exit_result(
+            ctx,
+            global_llm_context=None,
+            goal_satisfied=True,
+            error=RuntimeError("sdk-wrapped hook exception"),
+            workflow_permanent_id="wfp-1",
+        )
+
+        assert result.user_response != _VERIFIED_WORKFLOW_SUCCESS_REPLY
+        assert signal.user_facing_reason in result.user_response
+
     def test_wrapped_goal_satisfied_error_context_requires_no_change(self) -> None:
         from skyvern.forge.sdk.copilot.enforcement import verified_goal_satisfied_context
 

tests/unit/test_copilot_completion_verification.py

@@ -982,6 +982,84 @@ async def test_structured_blocker_run_skips_completion_verification(monkeypatch:
     assert verification is None
 
 
+@pytest.mark.asyncio
+async def test_classifier_fallback_record_is_not_verified_without_judge(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def handler(**_: object) -> dict:
+        raise AssertionError("value-agnostic fallback criteria must not reach the completion judge")
+
+    _patch_completion_handler(monkeypatch, handler)
+    ctx = _ctx_with_blocks("code")
+    ctx.request_policy = RequestPolicy(completion_criteria=_structured_record_criteria())
+
+    result = _structured_record_top_level_output_result()
+    verification = await _maybe_run_completion_verification(ctx, result, time.monotonic())
+    assert verification is None
+
+    _record_run_blocks_result(ctx, result, completion_verification=verification)
+    # The strict barrier predicate and its telemetry flag stay false, so the proposal is
+    # not preserved as a verified success; legacy clean-run flags may still promote, as
+    # they do for any genuine zero-criteria run.
+    assert getattr(ctx, "verified_terminal_proposal_ready", False) is not True
+    assert outcome_fully_verified(ctx) is False
+
+
+@pytest.mark.asyncio
+async def test_non_fallback_judge_confirmed_run_still_fires_barrier(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def handler(**_: object) -> dict:
+        return {"verdicts": [{"criterion_id": "c0", "satisfied": True, "reason_code": "evidence_confirms"}]}
+
+    _patch_completion_handler(monkeypatch, handler)
+    ctx = _ctx_with_blocks("extraction")
+    ctx.request_policy = RequestPolicy(completion_criteria=[_criterion("c0", "item in cart")])
+
+    result = _clean_success_result()
+    verification = await _maybe_run_completion_verification(ctx, result, time.monotonic())
+    assert verification is not None
+    assert verification.is_fully_satisfied() is True
+
+    _record_run_blocks_result(ctx, result, completion_verification=verification)
+    assert outcome_fully_verified(ctx) is True
+    assert verified_goal_satisfied_context(ctx) is True
+
+
+@pytest.mark.asyncio
+async def test_classifier_fallback_record_contradiction_still_surfaces(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def handler(**_: object) -> dict:
+        raise AssertionError("structural contradictions are deterministic, not judged")
+
+    _patch_completion_handler(monkeypatch, handler)
+    ctx = _ctx_with_blocks("code")
+    ctx.request_policy = RequestPolicy(completion_criteria=_structured_record_criteria())
+
+    contradictory_record = _record_payload(
+        items=[{"item_name": "Sample Practice Active", "address": "100 Main St", "status": "Expired"}],
+        overall_status="Expired",
+    )
+    result = {
+        "ok": True,
+        "data": {
+            "workflow_run_id": "wr_contradiction",
+            "overall_status": "completed",
+            "executed_block_labels": ["extract_record_status_record"],
+            "current_url": "https://structured_record.test/entity-details",
+            "blocks": [
+                {
+                    "label": "extract_record_status_record",
+                    "block_type": "CODE",
+                    "status": "completed",
+                    "extracted_data": {"extracted_information": []},
+                }
+            ],
+            "output": {"extract_record_status_record_output": contradictory_record},
+        },
+    }
+
+    verification = await _maybe_run_completion_verification(ctx, result, time.monotonic())
+    assert verification is not None
+    assert verification.is_fully_satisfied() is False
+    assert any(not verdict.satisfied for verdict in verification.verdicts)
+
+
 def _failed_code_block_result() -> dict:
     raw = (
         "code block failed. failure reason: Failed to execute code block. Reason: TimeoutError: "
@@ -1874,13 +1952,13 @@ def test_snapshot_uses_structured_record_top_level_output_parameters() -> None:
 
 
 @pytest.mark.asyncio
-async def test_maybe_run_completion_verification_deterministically_satisfies_structured_record_without_judge_call(
+async def test_maybe_run_completion_verification_treats_fallback_record_as_criteria_less(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     handler_lookup_calls = 0
 
     async def fail_handler(**_: object) -> object:
-        raise AssertionError("deterministically covered structured-record criteria must not call the judge")
+        raise AssertionError("value-agnostic fallback criteria must not call the judge")
 
     async def handler_lookup(_ctx: object) -> object:
         nonlocal handler_lookup_calls
@@ -1900,20 +1978,19 @@ async def test_maybe_run_completion_verification_deterministically_satisfies_str
         time.monotonic(),
     )
 
-    assert verification is not None
-    assert verification.status == "evaluated"
-    assert verification.is_fully_satisfied() is True
-    assert handler_lookup_calls == 1
-    assert _satisfied_criterion_ids(verification.verdicts) == _STRUCTURED_RECORD_CRITERION_IDS
+    # Value-agnostic fallback criteria are criteria-less; a well-shaped record is not a
+    # verified result, and the path short-circuits before any judge lookup.
+    assert verification is None
+    assert handler_lookup_calls == 0
 
 
 @pytest.mark.asyncio
-async def test_maybe_run_completion_verification_fails_closed_on_no_judge_for_deterministic_success(
+async def test_maybe_run_completion_verification_fails_closed_on_no_judge_for_judge_needed_criteria(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     _patch_completion_handler(monkeypatch, None)
     ctx = _run_ctx()
-    ctx.request_policy = RequestPolicy(completion_criteria=_structured_record_criteria())
+    ctx.request_policy = RequestPolicy(completion_criteria=[_criterion("c0", "item in cart")])
 
     verification = await _maybe_run_completion_verification(
         ctx,
@@ -1927,15 +2004,15 @@ async def test_maybe_run_completion_verification_fails_closed_on_no_judge_for_de
 
 
 @pytest.mark.asyncio
-async def test_maybe_run_completion_verification_fails_closed_on_low_budget_for_deterministic_success(
+async def test_maybe_run_completion_verification_fails_closed_on_low_budget_for_judge_needed_criteria(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     async def fail_handler(**_: object) -> object:
-        raise AssertionError("low-budget deterministic verification must not call the judge")
+        raise AssertionError("low-budget verification must not call the judge")
 
     _patch_completion_handler(monkeypatch, fail_handler)
     ctx = _run_ctx()
-    ctx.request_policy = RequestPolicy(completion_criteria=_structured_record_criteria())
+    ctx.request_policy = RequestPolicy(completion_criteria=[_criterion("c0", "item in cart")])
     starved = time.monotonic() - 100_000
 
     verification = await _maybe_run_completion_verification(

tests/unit/test_copilot_diagnosis_repair_contract.py

@@ -93,6 +93,33 @@ def test_contract_shapes_for_failed_suspicious_and_missing_credential_cases() ->
     ) == (DiagnosisFailureType.MISSING_CREDENTIAL_OR_INIT, RepairNextAction.ASK, ["may_answer_without_mutation"])
 
 
+def test_judge_confirmed_suspicious_success_forces_no_change() -> None:
+    ctx = _ctx()
+    ctx.last_test_suspicious_success = True
+    ctx.completion_verification_result = CompletionVerificationResult(
+        status="evaluated",
+        criterion_ids=["c0"],
+        verdicts=[CriterionVerdict(criterion_id="c0", state="satisfied", reason_code="evidence_confirms")],
+    )
+    contract = build_diagnosis_repair_contract(
+        source_tool="update_and_run_blocks",
+        result={
+            "ok": True,
+            "data": {
+                "workflow_run_id": "wr_verified",
+                "overall_status": "completed",
+                "frontier_start_label": "extract",
+                "blocks": [{"label": "extract", "block_type": "EXTRACTION", "status": "completed"}],
+            },
+        },
+        ctx=ctx,
+        workflow_updated=True,
+    )
+
+    assert contract.diagnosis_result.suspected_failure_type == DiagnosisFailureType.SUSPICIOUS_SUCCESS
+    assert contract.repair_decision.next_action == RepairNextAction.NO_CHANGE
+
+
 def test_repairable_block_failure_contract_is_queryable_and_safe() -> None:
     contract = build_diagnosis_repair_contract(
         source_tool="run_blocks_and_collect_debug",

tests/unit/test_copilot_enforcement_pruning.py

@@ -15,6 +15,7 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
+from skyvern.forge.sdk.copilot.blocker_signal import CopilotToolBlockerSignal
 from skyvern.forge.sdk.copilot.completion_verification import CompletionVerificationResult, CriterionVerdict
 from skyvern.forge.sdk.copilot.enforcement import (
     KEEP_RECENT_TOOL_OUTPUTS,
@@ -126,7 +127,7 @@ def test_unrecoverable_browser_session_error_stops_after_second_failure() -> Non
         _maybe_raise_unrecoverable_tool_error,
     )
 
-    ctx = SimpleNamespace()
+    ctx = SimpleNamespace(last_artifact_health_blocker_reason=None, completion_verification_result=None)
     output = {"ok": False, "error": "Browser session not found while taking screenshot (404)."}
 
     _maybe_raise_unrecoverable_tool_error(ctx, "get_browser_screenshot", output)
@@ -546,6 +547,36 @@ def test_enforcement_stops_after_verified_terminal_proposal() -> None:
         _check_enforcement(ctx)
 
 
+def test_verified_outcome_out_orders_same_turn_involuntary_blocker() -> None:
+    ctx = _Ctx()
+    ctx.completion_verification_result = CompletionVerificationResult(
+        status="evaluated",
+        criterion_ids=["c0"],
+        verdicts=[CriterionVerdict(criterion_id="c0", state="satisfied", reason_code="evidence_confirms")],
+    )
+    involuntary = CopilotToolBlockerSignal(
+        blocker_kind="loop_detected",
+        agent_steering_text="repeated failed step",
+        user_facing_reason="I'm stuck retrying the same step.",
+        recovery_hint="report_blocker_to_user",
+        cleared_by_tools=frozenset(),
+        preserves_workflow_draft=True,
+        renders_final_reply=True,
+        internal_reason_code="loop_detected_repeated_failed_step",
+        blocked_tool="update_and_run_blocks",
+        extra={},
+    )
+    ctx.turn_halt = None
+    ctx.blocker_signal = involuntary
+    ctx.latest_tool_blocker_signal = involuntary
+
+    with pytest.raises(CopilotGoalSatisfied):
+        _check_enforcement(ctx)
+
+    assert ctx.turn_halt is None
+    assert ctx.blocker_signal is None
+
+
 def test_record_run_blocks_result_keeps_failure_when_watchdog_cancel_without_timeout() -> None:
     """Stagnation/ceiling cancels mid-session must still set last_test_ok=False
     so the failed-test nudge can fire — only a coincident total timeout softens

tests/unit/test_copilot_hooks.py

@@ -29,6 +29,8 @@ class _FakeContext:
     turn_id: str = "turn_example"
     workflow_copilot_chat_id: str = "chat_example"
     total_tokens_used: int | None = None
+    last_artifact_health_blocker_reason: str | None = None
+    completion_verification_result: Any = None
 
 
 # `on_tool_end(context, agent, tool, result)` only reads `tool` and `result`;

tests/unit/test_copilot_recoverable_failures.py

@@ -119,7 +119,11 @@ async def _capture_tool_result(tool_name: str, parsed_output: dict[str, Any]) ->
     stream.is_disconnected = AsyncMock(return_value=False)
     stream.send = _send
 
-    await stream_to_sse(result, stream, SimpleNamespace())
+    await stream_to_sse(
+        result,
+        stream,
+        SimpleNamespace(last_artifact_health_blocker_reason=None, completion_verification_result=None),
+    )
 
     tool_results = [p for p in sent if getattr(p, "type", None) == WorkflowCopilotStreamMessageType.TOOL_RESULT]
     assert len(tool_results) == 1

tests/unit/test_copilot_response_coverage_gate.py

@@ -61,6 +61,8 @@ class _Ctx:
         self.null_data_streak_count = 0
         self.repeated_failure_streak_count = 0
         self.repeated_failure_nudge_emitted_at_streak = 0
+        self.last_artifact_health_blocker_reason = None
+        self.completion_verification_result = None
 
 
 @dataclass

tests/unit/test_copilot_streaming_adapter.py

@@ -119,7 +119,7 @@ async def test_stream_to_sse_keeps_running_after_client_disconnect() -> None:
     stream.is_disconnected = AsyncMock(return_value=True)
     stream.send = AsyncMock(return_value=True)
 
-    ctx = SimpleNamespace()
+    ctx = SimpleNamespace(last_artifact_health_blocker_reason=None, completion_verification_result=None)
 
     await stream_to_sse(result, stream, ctx)
 
@@ -151,7 +151,7 @@ async def test_tool_call_sse_uses_product_safe_activity_label() -> None:
     stream = MagicMock()
     stream.is_disconnected = AsyncMock(return_value=False)
     stream.send = _send
-    ctx = SimpleNamespace()
+    ctx = SimpleNamespace(last_artifact_health_blocker_reason=None, completion_verification_result=None)
 
     await stream_to_sse(result, stream, ctx)
 
@@ -189,7 +189,7 @@ async def test_stream_to_sse_propagates_cancelled_error() -> None:
     stream.is_disconnected = AsyncMock(return_value=False)
     stream.send = AsyncMock(return_value=True)
 
-    ctx = SimpleNamespace()
+    ctx = SimpleNamespace(last_artifact_health_blocker_reason=None, completion_verification_result=None)
 
     with pytest.raises(asyncio.CancelledError):
         await stream_to_sse(result, stream, ctx)
@@ -327,7 +327,7 @@ async def test_tool_result_sse_summary_translates_loop_detected_failure() -> Non
     stream.is_disconnected = AsyncMock(return_value=False)
     stream.send = _send
 
-    ctx = SimpleNamespace()
+    ctx = SimpleNamespace(last_artifact_health_blocker_reason=None, completion_verification_result=None)
 
     await stream_to_sse(result, stream, ctx)
 
@@ -385,7 +385,12 @@ async def test_tool_result_sse_uses_latest_blocker_signal_for_activity_surface()
     stream = MagicMock()
     stream.is_disconnected = AsyncMock(return_value=False)
     stream.send = _send
-    ctx = SimpleNamespace(latest_tool_blocker_signal=signal, tool_blocker_signals=[signal])
+    ctx = SimpleNamespace(
+        latest_tool_blocker_signal=signal,
+        tool_blocker_signals=[signal],
+        last_artifact_health_blocker_reason=None,
+        completion_verification_result=None,
+    )
 
     await stream_to_sse(result, stream, ctx)
 
@@ -447,7 +452,12 @@ async def test_stream_to_sse_cancels_and_stops_on_terminal_turn_halt() -> None:
     stream = MagicMock()
     stream.is_disconnected = AsyncMock(return_value=False)
     stream.send = AsyncMock(return_value=True)
-    ctx = SimpleNamespace(latest_tool_blocker_signal=signal, tool_blocker_signals=[signal])
+    ctx = SimpleNamespace(
+        latest_tool_blocker_signal=signal,
+        tool_blocker_signals=[signal],
+        last_artifact_health_blocker_reason=None,
+        completion_verification_result=None,
+    )
 
     with pytest.raises(CopilotTurnHalt):
         await stream_to_sse(result, stream, ctx)
@@ -482,7 +492,7 @@ async def test_stream_to_sse_raises_and_cancels_on_repeated_unrecoverable_tool_e
     stream = MagicMock()
     stream.is_disconnected = AsyncMock(return_value=False)
     stream.send = AsyncMock(return_value=True)
-    ctx = SimpleNamespace()
+    ctx = SimpleNamespace(last_artifact_health_blocker_reason=None, completion_verification_result=None)
 
     with pytest.raises(CopilotUnrecoverableToolError):
         await stream_to_sse(result, stream, ctx)
@@ -526,7 +536,7 @@ async def test_tool_result_sse_summary_drops_click_selector_on_success() -> None
     stream.is_disconnected = AsyncMock(return_value=False)
     stream.send = _send
 
-    ctx = SimpleNamespace()
+    ctx = SimpleNamespace(last_artifact_health_blocker_reason=None, completion_verification_result=None)
 
     await stream_to_sse(result, stream, ctx)
 

tests/unit/test_copilot_timeout_wip.py

@@ -176,6 +176,24 @@ class TestBuildTimeoutExitResult:
         assert result.user_response == _TIMEOUT_REPLY_TESTED
         assert result.clear_proposed_workflow is False
 
+    def test_stale_latch_without_judge_verdict_does_not_preserve_proposal(self) -> None:
+        wf = MagicMock(name="wf")
+        ctx = _ctx(
+            last_workflow=wf,
+            last_workflow_yaml="version: '1.0'",
+            last_test_ok=None,
+            last_test_suspicious_success=True,
+        )
+        ctx.verified_terminal_proposal_ready = True
+        ctx.completion_verification_result = None
+        ctx.last_artifact_health_blocker_reason = None
+
+        result = _build_timeout_exit_result(ctx, global_llm_context=None)
+
+        assert result.updated_workflow is None
+        assert result.workflow_yaml is None
+        assert result.user_response == _TIMEOUT_REPLY_DEFAULT
+
     def test_suspicious_current_run_drops_last_good_workflow_without_verified_terminal_state(self) -> None:
         wf = MagicMock(name="wf")
         last_good = MagicMock(name="last_good")

tests/unit/test_copilot_turn_halt.py

@@ -21,6 +21,7 @@ from skyvern.forge.sdk.copilot.turn_halt import (
     CopilotTurnHalt,
     TurnHaltKind,
     raise_if_turn_halt,
+    stash_repair_ceiling_turn_halt,
     stash_turn_halt_from_blocker_signal,
     turn_halt_from_blocker_signal,
 )
@@ -95,7 +96,13 @@ def test_stash_and_raise_turn_halt_sets_context_once() -> None:
 
 
 def test_enforcement_backstop_converts_existing_terminal_blocker_signal() -> None:
-    ctx = SimpleNamespace(turn_halt=None, latest_tool_blocker_signal=None, blocker_signal=None)
+    ctx = SimpleNamespace(
+        turn_halt=None,
+        latest_tool_blocker_signal=None,
+        blocker_signal=None,
+        last_artifact_health_blocker_reason=None,
+        completion_verification_result=None,
+    )
     signal = _signal()
     stash_blocker_signal(ctx, signal)
 
@@ -306,3 +313,149 @@ def test_current_page_challenge_signal_defers_to_populated_result_container_evid
     )
 
     assert signal is None
+
+
+def _involuntary_repair_ceiling_signal() -> CopilotToolBlockerSignal:
+    return CopilotToolBlockerSignal(
+        blocker_kind="tool_error",
+        agent_steering_text="repair ceiling",
+        user_facing_reason="I could not get the run to pass after several repair attempts.",
+        recovery_hint="report_blocker_to_user",
+        cleared_by_tools=frozenset(),
+        preserves_workflow_draft=True,
+        renders_final_reply=True,
+        internal_reason_code="repair_ceiling_reached",
+        blocked_tool="update_and_run_blocks",
+        extra={},
+    )
+
+
+def _consume_ctx(
+    *,
+    turn_halt: object = None,
+    blocker_signal: CopilotToolBlockerSignal | None = None,
+    latest_tool_blocker_signal: CopilotToolBlockerSignal | None = None,
+    tool_blocker_signals: list[CopilotToolBlockerSignal] | None = None,
+) -> SimpleNamespace:
+    return SimpleNamespace(
+        turn_halt=turn_halt,
+        blocker_signal=blocker_signal,
+        latest_tool_blocker_signal=latest_tool_blocker_signal,
+        tool_blocker_signals=tool_blocker_signals if tool_blocker_signals is not None else [],
+    )
+
+
+def test_verified_outcome_suppresses_and_consumes_involuntary_halt() -> None:
+    signal = _involuntary_repair_ceiling_signal()
+    halt = stash_repair_ceiling_turn_halt(SimpleNamespace(turn_halt=None), signal, consecutive_identical_repair_count=3)
+    ctx = _consume_ctx(
+        turn_halt=halt,
+        blocker_signal=signal,
+        latest_tool_blocker_signal=signal,
+        tool_blocker_signals=[signal],
+    )
+
+    raise_if_turn_halt(ctx, verified=True)
+
+    assert ctx.turn_halt is None
+    assert ctx.blocker_signal is None
+    assert ctx.latest_tool_blocker_signal is None
+    assert ctx.tool_blocker_signals == []
+
+
+def test_verified_outcome_consumes_loop_blocker_signal() -> None:
+    signal = _signal(blocker_kind="loop_detected", internal_reason_code="loop_detected_repeated_failed_step")
+    ctx = _consume_ctx(blocker_signal=signal)
+    stash_turn_halt_from_blocker_signal(ctx, signal, source="hook")
+
+    raise_if_turn_halt(ctx, verified=True)
+
+    assert ctx.turn_halt is None
+    assert ctx.blocker_signal is None
+
+
+def test_verified_outcome_does_not_suppress_voluntary_terminal_challenge() -> None:
+    signal = _signal(internal_reason_code=ACTIVE_RUN_TERMINAL_EVIDENCE_REASON_CODE)
+    ctx = _consume_ctx(blocker_signal=signal)
+    stash_turn_halt_from_blocker_signal(ctx, signal, source="hook")
+
+    with pytest.raises(CopilotTurnHalt):
+        raise_if_turn_halt(ctx, verified=True)
+
+    assert ctx.turn_halt is not None
+    assert ctx.blocker_signal is signal
+
+
+def test_verified_outcome_does_not_clear_voluntary_blocker_when_involuntary_absent() -> None:
+    challenge_signal = _signal(internal_reason_code=ACTIVE_RUN_TERMINAL_EVIDENCE_REASON_CODE)
+    loop_halt = stash_turn_halt_from_blocker_signal(
+        SimpleNamespace(turn_halt=None),
+        _signal(blocker_kind="loop_detected", internal_reason_code="loop_detected_repeated_failed_step"),
+        source="hook",
+    )
+    ctx = _consume_ctx(
+        turn_halt=loop_halt,
+        blocker_signal=challenge_signal,
+        latest_tool_blocker_signal=challenge_signal,
+        tool_blocker_signals=[challenge_signal],
+    )
+
+    raise_if_turn_halt(ctx, verified=True)
+
+    assert ctx.turn_halt is None
+    assert ctx.blocker_signal is challenge_signal
+    assert ctx.latest_tool_blocker_signal is challenge_signal
+    assert ctx.tool_blocker_signals == [challenge_signal]
+
+
+def test_verified_outcome_consumes_involuntary_tool_blocker_history() -> None:
+    involuntary = _signal(blocker_kind="loop_detected", internal_reason_code="loop_detected_repeated_failed_step")
+    voluntary = _signal(internal_reason_code=ACTIVE_RUN_TERMINAL_EVIDENCE_REASON_CODE)
+    ctx = _consume_ctx(blocker_signal=involuntary)
+    ctx.latest_tool_blocker_signal = involuntary
+    ctx.tool_blocker_signals = [voluntary, involuntary]
+    stash_turn_halt_from_blocker_signal(ctx, involuntary, source="hook")
+
+    raise_if_turn_halt(ctx, verified=True)
+
+    assert ctx.latest_tool_blocker_signal is None
+    assert ctx.tool_blocker_signals == [voluntary]
+
+
+def test_involuntary_suppression_lets_later_voluntary_challenge_raise() -> None:
+    signal = _involuntary_repair_ceiling_signal()
+    ctx = _consume_ctx(blocker_signal=signal)
+    stash_repair_ceiling_turn_halt(ctx, signal, consecutive_identical_repair_count=3)
+
+    raise_if_turn_halt(ctx, verified=True)
+    assert ctx.turn_halt is None
+    assert ctx.blocker_signal is None
+
+    challenge_signal = _signal(internal_reason_code=ACTIVE_RUN_TERMINAL_EVIDENCE_REASON_CODE)
+    ctx.blocker_signal = challenge_signal
+    stash_turn_halt_from_blocker_signal(ctx, challenge_signal, source="hook")
+
+    with pytest.raises(CopilotTurnHalt):
+        raise_if_turn_halt(ctx, verified=True)
+    assert ctx.blocker_signal is challenge_signal
+
+
+def test_unverified_involuntary_halt_still_raises() -> None:
+    signal = _involuntary_repair_ceiling_signal()
+    ctx = _consume_ctx(blocker_signal=signal)
+    stash_repair_ceiling_turn_halt(ctx, signal, consecutive_identical_repair_count=3)
+
+    with pytest.raises(CopilotTurnHalt):
+        raise_if_turn_halt(ctx, verified=False)
+
+    assert ctx.turn_halt is not None
+    assert ctx.blocker_signal is signal
+
+
+def test_default_verified_argument_is_fail_safe_and_raises() -> None:
+    signal = _involuntary_repair_ceiling_signal()
+    ctx = _consume_ctx(blocker_signal=signal)
+    stash_repair_ceiling_turn_halt(ctx, signal, consecutive_identical_repair_count=3)
+
+    with pytest.raises(CopilotTurnHalt):
+        raise_if_turn_halt(ctx)

tests/unit/test_workflow_copilot_route.py

@@ -1341,3 +1341,36 @@ async def test_legacy_path_persists_copilot_yaml_on_proposal(monkeypatch: pytest
         "legacy V1 path must stash the LLM-emitted YAML on the proposal so "
         "/apply-proposed-workflow can re-create the workflow version"
     )
+
+
+@pytest.mark.asyncio
+async def test_persist_state_keeps_verified_review_tested_proposal(monkeypatch: pytest.MonkeyPatch) -> None:
+    chat = SimpleNamespace(
+        organization_id="org-1",
+        workflow_copilot_chat_id="chat-1",
+        auto_accept=False,
+        proposed_workflow={"existing": True},
+    )
+    monkeypatch.setattr(
+        app.DATABASE,
+        "workflow_params",
+        SimpleNamespace(update_workflow_copilot_chat=AsyncMock()),
+    )
+    agent_result = SimpleNamespace(
+        updated_workflow=SimpleNamespace(model_dump=lambda mode: {"title": "built"}),
+        workflow_yaml="title: built\n",
+        clear_proposed_workflow=False,
+        proposal_disposition="review_tested",
+        cancelled=False,
+        apply_without_review=False,
+        output_policy_diagnostics=None,
+        canonical_was_persisted_due_to_param_change=False,
+    )
+
+    await workflow_copilot_route._persist_proposed_workflow_state(chat, agent_result, restored=False)
+
+    calls = app.DATABASE.workflow_params.update_workflow_copilot_chat.await_args_list
+    assert len(calls) == 1
+    persisted = calls[0].kwargs["proposed_workflow"]
+    assert persisted is not None
+    assert persisted.get("_copilot_unvalidated") is not True