From 400fd43d7b0a1f7beeb77ceadef66939a1cbc643 Mon Sep 17 00:00:00 2001 From: Andrew Neilson Date: Wed, 8 Jul 2026 14:32:07 -0700 Subject: [PATCH] SKY-11950 typed degraded completion terminal (#7214) --- skyvern/forge/sdk/copilot/agent.py | 44 +++++- skyvern/forge/sdk/copilot/blocker_signal.py | 6 + .../sdk/copilot/completion_verification.py | 70 ++++++++++ skyvern/forge/sdk/copilot/context.py | 3 + .../sdk/copilot/diagnosis_repair_contract.py | 25 +++- skyvern/forge/sdk/copilot/tools/completion.py | 9 +- .../forge/sdk/copilot/tools/run_execution.py | 76 ++++++++++- skyvern/forge/sdk/copilot/turn_halt.py | 22 ++- .../test_copilot_agent_blocker_renderer.py | 25 ++++ .../test_copilot_completion_verification.py | 118 ++++++++++++++++ .../test_copilot_diagnosis_repair_contract.py | 126 ++++++++++++++++++ 11 files changed, 517 insertions(+), 7 deletions(-) diff --git a/skyvern/forge/sdk/copilot/agent.py b/skyvern/forge/sdk/copilot/agent.py index 9ff53af97..d31449be6 100644 --- a/skyvern/forge/sdk/copilot/agent.py +++ b/skyvern/forge/sdk/copilot/agent.py @@ -153,6 +153,7 @@ from skyvern.forge.sdk.copilot.turn_halt import ( _INVOLUNTARY_BLOCKER_REASON_CODES, CopilotTurnHalt, TurnHalt, + TurnHaltKind, raise_if_turn_halt, turn_halt_to_trace_data, ) @@ -1900,6 +1901,16 @@ def _build_turn_halt_exit_result( global_llm_context: str | None, halt: TurnHalt, ) -> AgentResult: + if halt.kind == TurnHaltKind.DELIVERED_UNVERIFIED: + reply = _delivered_unverified_reply(ctx) or _BUILT_UNVERIFIED_COMPLETED_REPLY + return _build_wip_exit_result( + ctx, + global_llm_context, + default_reply=reply, + unvalidated_reply=reply, + tested_reply=reply, + terminal_reason=f"turn_halt:{halt.kind.value}", + ) signal = halt.blocker_signal if isinstance(signal, CopilotToolBlockerSignal) and signal.blocker_kind == "loop_detected": refresh_held_loop_blocker_evidence(ctx) @@ -2031,6 +2042,33 @@ def _terminal_summary_scalar(value: Any) -> str | None: return cleaned +def _delivered_unverified_reply(ctx: CopilotContext) -> str | None: + if getattr(ctx, "delivered_unverified_terminal", False) is not True: + return None + parts: list[str] = [] + observed_outputs = getattr(ctx, "delivered_unverified_observed_outputs", {}) + if not isinstance(observed_outputs, dict): + observed_outputs = {} + for key, value in observed_outputs.items(): + rendered = _terminal_summary_scalar(value) + if rendered is None and isinstance(value, list | dict): + try: + rendered = redact_raw_secrets_for_prompt(json.dumps(value, sort_keys=True)) + except TypeError: + rendered = None + if isinstance(key, str) and key.strip() and rendered and not contains_internal_machinery_leak(rendered): + parts.append(f"{key}: {rendered[:_VERIFIED_TERMINAL_VALUE_MAX_CHARS]}") + if parts: + return ( + "I built and ran the workflow. The latest run returned " + f"{'; '.join(parts[:4])}. That value was not independently verified, so review the draft before using it." + ) + return ( + "I built and ran the workflow, and the latest run returned the requested output. " + "That output was not independently verified, so review the draft before using it." + ) + + def _verified_output_value(ctx: CopilotContext, output_key: str | None) -> Any: if not output_key: return None @@ -2756,7 +2794,11 @@ def _build_wip_exit_result( ): full_test_ok = ctx.last_test_ok is True and ctx.last_full_workflow_test_ok is True unvalidated = not full_test_ok - if unvalidated and recorded_failure_reply: + delivered_reply = _delivered_unverified_reply(ctx) + if delivered_reply is not None: + reply = delivered_reply + unvalidated = True + elif unvalidated and recorded_failure_reply: reply = recorded_failure_reply if halted_mid_progress: reply = _ensure_unvalidated_proposal_affordance(reply) diff --git a/skyvern/forge/sdk/copilot/blocker_signal.py b/skyvern/forge/sdk/copilot/blocker_signal.py index 441d8b45b..d43bb8094 100644 --- a/skyvern/forge/sdk/copilot/blocker_signal.py +++ b/skyvern/forge/sdk/copilot/blocker_signal.py @@ -254,6 +254,9 @@ class _TerminalEvidenceResetCtx(Protocol): last_outcome_gate_reason: str | None last_outcome_gate_workflow_run_id: str | None last_test_anti_bot: str | None + delivered_unverified_terminal: bool + delivered_unverified_workflow_run_id: str | None + delivered_unverified_observed_outputs: dict[str, Any] completion_verification_result: Any | None outcome_verification_trace_snapshot: dict[str, Any] @@ -286,6 +289,9 @@ def clear_terminal_evidence_on_workflow_edit(ctx: _TerminalEvidenceResetCtx) -> ctx.last_outcome_gate_reason = None ctx.last_outcome_gate_workflow_run_id = None ctx.last_test_anti_bot = None + ctx.delivered_unverified_terminal = False + ctx.delivered_unverified_workflow_run_id = None + ctx.delivered_unverified_observed_outputs = {} ctx.completion_verification_result = None ctx.outcome_verification_trace_snapshot = {} diff --git a/skyvern/forge/sdk/copilot/completion_verification.py b/skyvern/forge/sdk/copilot/completion_verification.py index 46b8e42ba..180e88adb 100644 --- a/skyvern/forge/sdk/copilot/completion_verification.py +++ b/skyvern/forge/sdk/copilot/completion_verification.py @@ -254,6 +254,11 @@ class CompletionVerificationResult: } +@dataclass(frozen=True) +class DeliveredUnverifiedTerminalState: + observed_verdicts: tuple[CriterionVerdict, ...] + + @dataclass(frozen=True) class RunEvidenceSnapshot: workflow_run_id: str | None = None @@ -330,6 +335,7 @@ _UNAVAILABLE = CompletionVerificationResult(status="unavailable") _MISSING_VERDICT_EVIDENCE = "judge did not return a verdict for this criterion" _INCOMPLETE_VALIDATION_CLASSIFICATION_CONTRACT = "incomplete typed classification contract" _MISSING_REGISTERED_DOWNLOAD_EVIDENCE = "run output did not include a non-empty registered browser download" +_DELIVERED_UNVERIFIED_OUTPUT_SOURCES = frozenset({"runtime_output", "registered_output_parameter"}) def registered_download_completion_criterion() -> CompletionCriterion: @@ -2242,6 +2248,70 @@ def only_degraded_blocking(result: CompletionVerificationResult) -> bool: return blocking <= degraded +def _delivered_unverified_observed_verdict(verdict: CriterionVerdict) -> bool: + if not verdict.evidence_ref or not verdict.output_path: + return False + normalized_path = verdict.output_path.removeprefix("output.").strip() + if normalized_path.split(".")[-1] == "evidence_text": + return False + if verdict.evidence_source not in _DELIVERED_UNVERIFIED_OUTPUT_SOURCES: + return False + if verdict.self_emitted_judgment_not_independent: + return False + if verdict.requested_output_evidence_source == "independent_run_evidence": + return False + if verdict.grounding_mode in {"shape", "judgment_boolean"}: + return False + if verdict.satisfied: + return ( + verdict.reason_code == "evidence_confirms" + and verdict.grounding_mode == "exact_value" + and verdict.has_exact_value + ) + return _is_structural_requested_output_abstention(verdict) and not verdict.has_exact_value + + +def degraded_contract_delivered_unverified_terminal_state( + result: CompletionVerificationResult | None, + *, + run_ok: bool, + workflow_run_id: str | None, + latest_workflow_run_id: str | None, + artifact_health_blocked: bool, + terminal_blocked: bool, +) -> DeliveredUnverifiedTerminalState | None: + if ( + result is None + or result.status != "evaluated" + or result.is_fully_satisfied() + or not run_ok + or not workflow_run_id + or workflow_run_id != latest_workflow_run_id + or artifact_health_blocked + or terminal_blocked + ): + return None + degraded = set(result.degraded_criterion_ids) + if not degraded: + return None + verdict_by_id = {verdict.criterion_id: verdict for verdict in result.verdicts} + observed: list[CriterionVerdict] = [] + blocking: set[str] = set() + for criterion_id in result.criterion_ids: + verdict = verdict_by_id.get(criterion_id) + if verdict is not None and _delivered_unverified_observed_verdict(verdict): + observed.append(verdict) + continue + if verdict is not None and verdict.satisfied: + continue + if verdict is not None and result.is_structural_contingent_abstention(verdict): + continue + blocking.add(criterion_id) + if not observed or not blocking or not blocking <= degraded: + return None + return DeliveredUnverifiedTerminalState(observed_verdicts=tuple(observed)) + + def carry_degraded_criterion_ids( result: CompletionVerificationResult, criteria: Iterable[CompletionCriterion], diff --git a/skyvern/forge/sdk/copilot/context.py b/skyvern/forge/sdk/copilot/context.py index ddf120e06..0c08d81f4 100644 --- a/skyvern/forge/sdk/copilot/context.py +++ b/skyvern/forge/sdk/copilot/context.py @@ -592,6 +592,9 @@ class CopilotContext(AgentContext): last_run_blocks_workflow_run_id: str | None = None last_successful_run_blocks_workflow_run_id: str | None = None last_outcome_gate_workflow_run_id: str | None = None + delivered_unverified_terminal: bool = False + delivered_unverified_workflow_run_id: str | None = None + delivered_unverified_observed_outputs: dict[str, Any] = field(default_factory=dict) # Consecutive failed runs where navigation completed but the scraper # could not read the page (generic "failed to load the website" template). # Resets on any non-matching run outcome. Streak crosses workflow-shape diff --git a/skyvern/forge/sdk/copilot/diagnosis_repair_contract.py b/skyvern/forge/sdk/copilot/diagnosis_repair_contract.py index f0fa05a7f..e42536077 100644 --- a/skyvern/forge/sdk/copilot/diagnosis_repair_contract.py +++ b/skyvern/forge/sdk/copilot/diagnosis_repair_contract.py @@ -56,6 +56,7 @@ class DiagnosisFailureType(StrEnum): ACTIVE_RUN_TERMINAL_EVIDENCE = "active_run_terminal_evidence" UNRECOVERABLE_TOOL_ERROR = "unrecoverable_tool_error" SCHEMA_INCOMPATIBILITY = "schema_incompatibility" + DELIVERED_UNVERIFIED = "delivered_unverified" UNKNOWN = "unknown" @@ -177,11 +178,18 @@ def build_diagnosis_repair_contract( outcome_verified = outcome_fully_verified(ctx) completion_verification = getattr(ctx, "completion_verification_result", None) completion_verification_failed = _completion_verification_failed(completion_verification) + delivered_unverified = bool( + run_ok + and getattr(ctx, "delivered_unverified_terminal", False) is True + and workflow_run_id + and workflow_run_id == getattr(ctx, "delivered_unverified_workflow_run_id", None) + ) failure_type = _failure_type( run_ok, suspicious, outcome_verified, completion_verification_failed, + delivered_unverified, failed_blocks, categories, terminal_challenge_categories, @@ -247,6 +255,8 @@ def build_diagnosis_repair_contract( "Stop re-authoring: the edited extraction schema declares fields that map to no output the " "workflow produces, so the mismatch is not repairable without user input." ) + elif failure_type == DiagnosisFailureType.DELIVERED_UNVERIFIED: + decision_summary = "No repair selected; the latest run returned requested output but it was not verified." if ( next_action == RepairNextAction.NO_CHANGE and user_goal_satisfied is True @@ -255,7 +265,9 @@ def build_diagnosis_repair_contract( completion_check = "Current run already satisfies the goal." else: completion_check = { - RepairNextAction.NO_CHANGE: "No repair selected; completion remains unverified.", + RepairNextAction.NO_CHANGE: "No repair selected; completion is delivered but not independently verified." + if failure_type == DiagnosisFailureType.DELIVERED_UNVERIFIED + else "No repair selected; completion remains unverified.", RepairNextAction.ASK: "Resume diagnosis after the user supplies the missing context.", RepairNextAction.STOP: "Do not rerun unchanged; user-visible blocker must be resolved first.", }.get( @@ -480,6 +492,7 @@ def _failure_type( suspicious: bool, outcome_verified: bool, completion_verification_failed: bool, + delivered_unverified: bool, failed_blocks: list[str], categories: list[str], terminal_challenge_categories: list[str], @@ -530,7 +543,11 @@ def _failure_type( return DiagnosisFailureType.REPAIRABLE_BLOCK_FAILURE if outcome_verified: return DiagnosisFailureType.NO_FAILURE - if failed_blocks or category_set & _REPAIRABLE_RUNTIME_CATEGORIES: + if failed_blocks: + return DiagnosisFailureType.REPAIRABLE_BLOCK_FAILURE + if delivered_unverified: + return DiagnosisFailureType.DELIVERED_UNVERIFIED + if category_set & _REPAIRABLE_RUNTIME_CATEGORIES: return DiagnosisFailureType.REPAIRABLE_BLOCK_FAILURE if completion_verification_failed: return DiagnosisFailureType.SUSPICIOUS_SUCCESS @@ -551,7 +568,7 @@ def _next_action( data: dict[str, Any], repair_context: CodeAuthoringRepairContext | None, ) -> RepairNextAction: - if failure_type == DiagnosisFailureType.NO_FAILURE: + if failure_type in {DiagnosisFailureType.NO_FAILURE, DiagnosisFailureType.DELIVERED_UNVERIFIED}: return RepairNextAction.NO_CHANGE if failure_type == DiagnosisFailureType.UNRECOVERABLE_TOOL_ERROR: return RepairNextAction.STOP @@ -645,6 +662,8 @@ def _verification_satisfaction( ) -> tuple[bool | None, bool | None]: if failure_type == DiagnosisFailureType.TERMINAL_CHALLENGE_BLOCKER: return False, False + if failure_type == DiagnosisFailureType.DELIVERED_UNVERIFIED: + return True, False if isinstance(data, dict) and data.get("active_run_terminal_evidence_detected") is True: trace = data.get("active_run_terminal_completion_verification") fully_satisfied = isinstance(trace, dict) and trace.get("fully_satisfied") is True diff --git a/skyvern/forge/sdk/copilot/tools/completion.py b/skyvern/forge/sdk/copilot/tools/completion.py index 443ae6e00..5f37e460e 100644 --- a/skyvern/forge/sdk/copilot/tools/completion.py +++ b/skyvern/forge/sdk/copilot/tools/completion.py @@ -1520,10 +1520,17 @@ def _tool_visible_result_after_completion_verification( outcome_unverified_reason = _outcome_unverified_reason(copilot_ctx, completion_verification) if outcome_unverified_reason is None: return result + data = result.get("data") + run_id = data.get("workflow_run_id") if isinstance(data, dict) else None + if ( + copilot_ctx.delivered_unverified_terminal + and isinstance(run_id, str) + and run_id == copilot_ctx.delivered_unverified_workflow_run_id + ): + return result if not _outcome_failure_warrants_repair(copilot_ctx, completion_verification): return result - data = result.get("data") copied_data = dict(data) if isinstance(data, dict) else {} copied_data["failure_reason"] = outcome_unverified_reason copied_data["completion_verification"] = ( diff --git a/skyvern/forge/sdk/copilot/tools/run_execution.py b/skyvern/forge/sdk/copilot/tools/run_execution.py index 747f2d1be..ca2d9ffef 100644 --- a/skyvern/forge/sdk/copilot/tools/run_execution.py +++ b/skyvern/forge/sdk/copilot/tools/run_execution.py @@ -10,7 +10,7 @@ from collections import defaultdict from collections.abc import Mapping, Sequence from dataclasses import dataclass, replace from datetime import datetime, timezone -from typing import Any, Literal +from typing import Any, Literal, cast from urllib.parse import urlparse import structlog @@ -49,6 +49,8 @@ from skyvern.forge.sdk.copilot.completion_output_grounding import page_evidence_ from skyvern.forge.sdk.copilot.completion_verification import ( CompletionVerificationResult, CriterionVerdict, + DeliveredUnverifiedTerminalState, + degraded_contract_delivered_unverified_terminal_state, only_structural_requested_output_abstentions, ) from skyvern.forge.sdk.copilot.composition_evidence import has_bounded_page_schema @@ -115,6 +117,7 @@ from skyvern.forge.sdk.copilot.runtime_authoring_repair import ( from skyvern.forge.sdk.copilot.terminal_predicates import outcome_fully_verified from skyvern.forge.sdk.copilot.tracing_setup import copilot_span from skyvern.forge.sdk.copilot.turn_halt import ( + stash_delivered_unverified_turn_halt, stash_repair_ceiling_turn_halt, stash_turn_halt_from_blocker_signal, ) @@ -2480,6 +2483,43 @@ def _update_verification_evidence_from_run_result(copilot_ctx: AgentContext, res evidence.failure_reason = " ".join(failure_reason.split())[:240] +def _read_mapping_path(payload: dict[str, Any], path: str) -> Any: + current: Any = payload + for part in path.split("."): + if not part or not isinstance(current, dict): + return None + current = current.get(part) + return current + + +def _delivered_unverified_single_block_outputs( + result: dict[str, Any], + terminal_state: DeliveredUnverifiedTerminalState, +) -> dict[str, Any]: + data = result.get("data") + blocks = data.get("blocks") if isinstance(data, dict) else None + outputs: list[dict[str, Any]] = [] + if isinstance(blocks, list): + for block in blocks: + if isinstance(block, dict) and isinstance(block.get("extracted_data"), dict): + outputs.append(cast(dict[str, Any], block["extracted_data"])) + if len(outputs) != 1: + return {} + block_output = outputs[0] + observed: dict[str, Any] = {} + for verdict in terminal_state.observed_verdicts: + output_path = verdict.output_path or "" + normalized_path = output_path.removeprefix("output.").strip() + if not normalized_path or normalized_path.split(".")[-1] == "evidence_text": + continue + value = _read_mapping_path(block_output, normalized_path) + if value is None: + value = _read_mapping_path(block_output, output_path) + if value is not None: + observed[normalized_path] = value + return observed + + def _record_run_blocks_result( copilot_ctx: Any, result: dict[str, Any], completion_verification: CompletionVerificationResult | None = None ) -> RecordedRunOutcome | None: @@ -2513,6 +2553,9 @@ def _record_run_blocks_result( _emit_completion_verification_trace(copilot_ctx, completion_verification) copilot_ctx.last_run_blocks_workflow_run_id = run_id if isinstance(run_id, str) else None copilot_ctx.last_successful_run_blocks_workflow_run_id = run_id if run_ok and isinstance(run_id, str) else None + copilot_ctx.delivered_unverified_terminal = False + copilot_ctx.delivered_unverified_workflow_run_id = None + copilot_ctx.delivered_unverified_observed_outputs = {} # Watchdog cancels normally count as ok=False; only a coincident total # timeout softens to ``None`` to keep the unvalidated WIP rescue open. cancelled_by_watchdog = result.get(_INTERNAL_RUN_CANCELLED_BY_WATCHDOG_KEY) is True @@ -2693,6 +2736,37 @@ def _record_run_blocks_result( copilot_ctx.last_test_suspicious_success = False copilot_ctx.last_test_failure_reason = None copilot_ctx.suspicious_success_nudge_count = 0 + delivered_unverified = degraded_contract_delivered_unverified_terminal_state( + completion_verification, + run_ok=run_ok, + workflow_run_id=run_id if isinstance(run_id, str) else None, + latest_workflow_run_id=copilot_ctx.last_run_blocks_workflow_run_id, + artifact_health_blocked=artifact_reason is not None, + terminal_blocked=terminal_challenge is not None or bool(copilot_ctx.last_test_anti_bot), + ) + if delivered_unverified is not None: + observed_outputs = _delivered_unverified_single_block_outputs(result, delivered_unverified) + if observed_outputs: + workflow_run_id = run_id if isinstance(run_id, str) else None + copilot_ctx.last_test_suspicious_success = False + copilot_ctx.last_test_failure_reason = None + copilot_ctx.suspicious_success_nudge_count = 0 + copilot_ctx.last_full_workflow_test_ok = False + copilot_ctx.delivered_unverified_terminal = True + copilot_ctx.delivered_unverified_workflow_run_id = workflow_run_id + copilot_ctx.delivered_unverified_observed_outputs = observed_outputs + stash_delivered_unverified_turn_halt(copilot_ctx, workflow_run_id=workflow_run_id) + update_repeated_failure_state(copilot_ctx, result) + _update_verification_evidence_from_run_result(copilot_ctx, result) + recorded_outcome = RecordedRunOutcome( + verdict="not_evaluated", + display_reason=run_outcome_display_reason( + "The latest run returned the requested output, but it was not independently verified." + ), + workflow_run_id=workflow_run_id, + ) + _record_adjudicated_build_test_outcome(copilot_ctx, result, completion_verification, recorded_outcome) + return _stash_recorded_run_outcome(copilot_ctx, recorded_outcome) if empty_data_blocks and not completion_verification_evaluated: copilot_ctx.last_test_ok = None copilot_ctx.last_test_suspicious_success = True diff --git a/skyvern/forge/sdk/copilot/turn_halt.py b/skyvern/forge/sdk/copilot/turn_halt.py index 7879f1561..aca046fd8 100644 --- a/skyvern/forge/sdk/copilot/turn_halt.py +++ b/skyvern/forge/sdk/copilot/turn_halt.py @@ -4,7 +4,7 @@ from __future__ import annotations from dataclasses import dataclass, field from enum import StrEnum -from typing import Any +from typing import TYPE_CHECKING, Any import structlog @@ -27,6 +27,9 @@ from skyvern.forge.sdk.copilot.schema_incompatibility import SCHEMA_INCOMPATIBIL LOG = structlog.get_logger() +if TYPE_CHECKING: + from skyvern.forge.sdk.copilot.runtime import AgentContext + REPAIR_CEILING_REASON_CODE = "repair_ceiling_reached" ADVISORY_DISPATCH_STALLED_REASON_CODE = "advisory_dispatch_stalled" @@ -38,10 +41,12 @@ class TurnHaltKind(StrEnum): REPAIR_CEILING_REACHED = "repair_ceiling_reached" SCHEMA_INCOMPATIBILITY = "schema_incompatibility" OUTPUT_SOURCE_UNOBSERVABLE = "output_source_unobservable" + DELIVERED_UNVERIFIED = "delivered_unverified" class TurnHaltVerdict(StrEnum): BLOCKED = "blocked" + DELIVERED_UNVERIFIED = "delivered_unverified" _LOOP_TERMINAL_REASON_CODES = frozenset( @@ -280,6 +285,21 @@ def stash_repair_ceiling_turn_halt( return halt +def stash_delivered_unverified_turn_halt(ctx: AgentContext, *, workflow_run_id: str | None) -> TurnHalt | None: + if isinstance(ctx.turn_halt, TurnHalt): + return ctx.turn_halt + run_refs = {"workflow_run_id": workflow_run_id} if workflow_run_id else {} + halt = TurnHalt( + kind=TurnHaltKind.DELIVERED_UNVERIFIED, + verdict=TurnHaltVerdict.DELIVERED_UNVERIFIED, + run_refs=run_refs, + extra={"source": "run_execution"}, + ) + ctx.turn_halt = halt + LOG.info("copilot turn halt stashed", **turn_halt_to_trace_data(halt)) + return halt + + def raise_if_turn_halt(ctx: Any, *, verified: bool = False) -> None: """Raise the stashed turn halt unless a verified outcome suppresses it. diff --git a/tests/unit/test_copilot_agent_blocker_renderer.py b/tests/unit/test_copilot_agent_blocker_renderer.py index 7ccb13a43..399ccfc98 100644 --- a/tests/unit/test_copilot_agent_blocker_renderer.py +++ b/tests/unit/test_copilot_agent_blocker_renderer.py @@ -13,6 +13,7 @@ from skyvern.forge.sdk.copilot.agent import ( _build_goal_satisfied_exit_result, _build_output_policy_blocked_result, _build_turn_halt_exit_result, + _build_wip_exit_result, _finalize_result_with_blocker_override, _render_blocker_reply, _verified_workflow_success_reply, @@ -816,6 +817,30 @@ def test_unverified_blocker_still_renders_blocker_text() -> None: assert overridden.proposal_disposition != "review_tested" +def test_delivered_unverified_exit_renders_observed_output_as_unvalidated() -> None: + ctx = _ctx() + ctx.last_workflow = SimpleNamespace(workflow_definition=SimpleNamespace(blocks=[SimpleNamespace()])) + ctx.last_workflow_yaml = "workflow_definition:\n blocks: []\n" + ctx.last_test_ok = True + ctx.last_full_workflow_test_ok = False + ctx.delivered_unverified_terminal = True + ctx.delivered_unverified_observed_outputs = {"document_name": "Resale Demand Package"} + + result = _build_wip_exit_result( + ctx, + None, + default_reply="default", + unvalidated_reply="unvalidated", + tested_reply="tested", + terminal_reason="turn_halt:delivered_unverified", + ) + + assert "document_name: Resale Demand Package" in result.user_response + assert "not independently verified" in result.user_response + assert result.proposal_disposition == "review_untested" + assert result.workflow_yaml == ctx.last_workflow_yaml + + def test_verified_outcome_does_not_suppress_voluntary_terminal_challenge() -> None: ctx = _ctx() _seed_verified_outcome(ctx) diff --git a/tests/unit/test_copilot_completion_verification.py b/tests/unit/test_copilot_completion_verification.py index ec8f30693..4bff2c844 100644 --- a/tests/unit/test_copilot_completion_verification.py +++ b/tests/unit/test_copilot_completion_verification.py @@ -31,10 +31,12 @@ from skyvern.forge.sdk.copilot.completion_verification import ( REGISTERED_DOWNLOAD_COMPLETION_CRITERION_ID, CompletionVerificationResult, CriterionVerdict, + DeliveredUnverifiedTerminalState, RunEvidenceSnapshot, _coerce_result, _structured_record_has_identifier, combine_verification_results, + degraded_contract_delivered_unverified_terminal_state, evaluate_completion_criteria, grade_definition_criteria, grade_fallback_floor_reached_end_state_criteria, @@ -3130,6 +3132,122 @@ def test_record_run_blocks_keeps_clean_structural_abstention_as_built_unverified assert outcome.is_authoritative is False +def _degraded_delivered_result(*verdicts: CriterionVerdict) -> CompletionVerificationResult: + return CompletionVerificationResult( + status="evaluated", + criterion_ids=["__copilot_fallback_floor__run", *[verdict.criterion_id for verdict in verdicts]], + verdicts=[ + CriterionVerdict( + criterion_id="__copilot_fallback_floor__run", + state="unsatisfied", + reason_code="no_evidence", + ), + *verdicts, + ], + degraded_criterion_ids=["__copilot_fallback_floor__run"], + ) + + +def _observed_structural_abstention( + criterion_id: str = "requested_output", + *, + evidence_source: str = "runtime_output", + reason_code: str = "structurally_abstained", + output_path: str = "output.document_name", + grounding_mode: str | None = "missing", +) -> CriterionVerdict: + return CriterionVerdict( + criterion_id=criterion_id, + state="unsatisfied", + reason_code=reason_code, + evidence_ref="block_outputs:extract.document_name", + output_path=output_path, + grounding_mode=grounding_mode, + evidence_source=evidence_source, + ) + + +def _delivered_terminal_state( + result: CompletionVerificationResult, + *, + run_ok: bool = True, + workflow_run_id: str | None = "wr_x", + latest_workflow_run_id: str | None = "wr_x", + artifact_health_blocked: bool = False, + terminal_blocked: bool = False, +) -> DeliveredUnverifiedTerminalState | None: + return degraded_contract_delivered_unverified_terminal_state( + result, + run_ok=run_ok, + workflow_run_id=workflow_run_id, + latest_workflow_run_id=latest_workflow_run_id, + artifact_health_blocked=artifact_health_blocked, + terminal_blocked=terminal_blocked, + ) + + +def test_degraded_delivered_unverified_terminal_state_allows_observed_runtime_output() -> None: + terminal_state = _delivered_terminal_state(_degraded_delivered_result(_observed_structural_abstention())) + + assert terminal_state is not None + assert [verdict.criterion_id for verdict in terminal_state.observed_verdicts] == ["requested_output"] + + +@pytest.mark.parametrize( + "kwargs", + [ + {"run_ok": False}, + {"workflow_run_id": "wr_old"}, + {"latest_workflow_run_id": "wr_old"}, + {"artifact_health_blocked": True}, + {"terminal_blocked": True}, + ], +) +def test_degraded_delivered_unverified_terminal_state_rejects_run_and_blocker_exclusions( + kwargs: dict[str, bool | str | None], +) -> None: + assert _delivered_terminal_state(_degraded_delivered_result(_observed_structural_abstention()), **kwargs) is None + + +@pytest.mark.parametrize( + "verdict", + [ + CriterionVerdict( + criterion_id="requested_output", + state="unsatisfied", + reason_code="missing_exact_field", + evidence_ref="block_outputs:extract.document_name", + output_path="output.document_name", + evidence_source="runtime_output", + ), + _observed_structural_abstention(output_path="output.evidence_text"), + _observed_structural_abstention(grounding_mode="shape"), + _observed_structural_abstention(evidence_source="independent_page_evidence"), + ], +) +def test_degraded_delivered_unverified_terminal_state_rejects_non_value_output(verdict: CriterionVerdict) -> None: + assert _delivered_terminal_state(_degraded_delivered_result(verdict)) is None + + +def test_degraded_delivered_unverified_terminal_state_is_not_verified_success() -> None: + ctx = _ctx_with_blocks("extraction") + result = _clean_success_result() + result["data"]["blocks"][0]["label"] = "extract" + result["data"]["blocks"][0]["extracted_data"] = {"document_name": "Resale Demand Package"} + verification = _degraded_delivered_result(_observed_structural_abstention()) + + recorded = _record_run_blocks_result(ctx, result, completion_verification=verification) + + assert recorded is not None + assert recorded.verdict == "not_evaluated" + assert ctx.delivered_unverified_terminal is True + assert ctx.delivered_unverified_observed_outputs == {"document_name": "Resale Demand Package"} + assert ctx.turn_halt is not None + assert ctx.turn_halt.kind.value == "delivered_unverified" + assert ctx.last_full_workflow_test_ok is False + assert ctx.last_test_suspicious_success is False + + def test_record_run_blocks_verifies_structural_requested_output_with_run_corroborator() -> None: ctx = _ctx_with_blocks("extraction") verification = _mixed( diff --git a/tests/unit/test_copilot_diagnosis_repair_contract.py b/tests/unit/test_copilot_diagnosis_repair_contract.py index 565adf9c9..241d6ebeb 100644 --- a/tests/unit/test_copilot_diagnosis_repair_contract.py +++ b/tests/unit/test_copilot_diagnosis_repair_contract.py @@ -1399,6 +1399,132 @@ def test_unverified_completion_evidence_does_not_suppress_suspicious_success( assert contract.verification_result.user_goal_satisfied is False +def test_degraded_delivered_unverified_completion_does_not_route_repair() -> None: + ctx = _ctx() + ctx.last_test_suspicious_success = True + ctx.last_run_blocks_workflow_run_id = "wr_unverified" + ctx.delivered_unverified_terminal = True + ctx.delivered_unverified_workflow_run_id = "wr_unverified" + ctx.delivered_unverified_observed_outputs = {"document_name": "Resale Demand Package"} + ctx.completion_verification_result = CompletionVerificationResult( + status="evaluated", + criterion_ids=["__copilot_fallback_floor__run", "requested_output"], + verdicts=[ + CriterionVerdict( + criterion_id="__copilot_fallback_floor__run", + state="unsatisfied", + reason_code="no_evidence", + ), + CriterionVerdict( + criterion_id="requested_output", + state="unsatisfied", + reason_code="structurally_abstained", + evidence_ref="block_outputs:extract.document_name", + output_path="output.document_name", + grounding_mode="missing", + evidence_source="runtime_output", + ), + ], + degraded_criterion_ids=["__copilot_fallback_floor__run"], + ) + + contract = build_diagnosis_repair_contract( + source_tool="update_and_run_blocks", + result={ + "ok": True, + "data": { + "workflow_run_id": "wr_unverified", + "overall_status": "completed", + "frontier_start_label": "extract", + "failure_categories": [{"category": "OUTCOME_UNVERIFIED"}], + "blocks": [{"label": "extract", "block_type": "EXTRACTION", "status": "completed"}], + }, + }, + ctx=ctx, + workflow_updated=True, + ) + + assert contract.diagnosis_result.suspected_failure_type == DiagnosisFailureType.DELIVERED_UNVERIFIED + assert contract.repair_decision.next_action == RepairNextAction.NO_CHANGE + assert contract.verification_result.user_goal_satisfied is True + assert contract.verification_result.completion_contract_satisfied is False + + +def test_delivered_unverified_does_not_mask_failed_blocks() -> None: + ctx = _ctx() + ctx.delivered_unverified_terminal = True + ctx.delivered_unverified_workflow_run_id = "wr_unverified" + ctx.delivered_unverified_observed_outputs = {"document_name": "Resale Demand Package"} + + contract = build_diagnosis_repair_contract( + source_tool="update_and_run_blocks", + result={ + "ok": True, + "data": { + "workflow_run_id": "wr_unverified", + "overall_status": "completed", + "blocks": [ + { + "label": "extract", + "block_type": "EXTRACTION", + "status": "failed", + } + ], + }, + }, + ctx=ctx, + workflow_updated=True, + ) + + assert contract.diagnosis_result.suspected_failure_type == DiagnosisFailureType.REPAIRABLE_BLOCK_FAILURE + assert contract.repair_decision.next_action == RepairNextAction.REPAIR + + +def test_degraded_path_without_terminal_state_still_routes_repair() -> None: + ctx = _ctx() + ctx.last_test_suspicious_success = True + ctx.last_run_blocks_workflow_run_id = "wr_unverified" + ctx.completion_verification_result = CompletionVerificationResult( + status="evaluated", + criterion_ids=["__copilot_fallback_floor__run", "requested_output"], + verdicts=[ + CriterionVerdict( + criterion_id="__copilot_fallback_floor__run", + state="unsatisfied", + reason_code="no_evidence", + ), + CriterionVerdict( + criterion_id="requested_output", + state="unsatisfied", + reason_code="structurally_abstained", + evidence_ref="block_outputs:extract.document_name", + output_path="output.document_name", + grounding_mode="missing", + evidence_source="runtime_output", + ), + ], + degraded_criterion_ids=["__copilot_fallback_floor__run"], + ) + + contract = build_diagnosis_repair_contract( + source_tool="update_and_run_blocks", + result={ + "ok": True, + "data": { + "workflow_run_id": "wr_unverified", + "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.REPAIR + + @pytest.mark.parametrize( "completion_verification", [