feat(SKY-11897): typed judgment-evidence grounding — confirm/refute/abstain from independent packet (#7224)

This commit is contained in:
Andrew Neilson 2026-07-08 19:53:47 -07:00 committed by GitHub
parent bb7fe7158f
commit 6a5226d197
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 1536 additions and 46 deletions

View file

@ -25,13 +25,15 @@ completion_contract:
- Set to null when the user does not provide an explicit completion condition.
completion_criteria:
- An array of the END-STATE OUTCOMES that must be true for the user's goal to be done. Each item is {outcome, contingent_on, contingent_antecedent_output_path, deliverable_kind, implicit, method_mandated, level, output_path, expected_output_value, expected_output_shape, requested_output_evidence_source, kind, terminal_action_family, classification_output_key, expected_classification}.
- An array of the END-STATE OUTCOMES that must be true for the user's goal to be done. Each item is {outcome, contingent_on, contingent_antecedent_output_path, deliverable_kind, implicit, method_mandated, level, output_path, expected_output_value, expected_output_shape, requested_output_evidence_source, kind, terminal_action_family, classification_output_key, expected_classification, judgment_predicate, judgment_polarity_when_holds}.
- outcome: a short statement of an end state the workflow must achieve (an END, not a step). Extract the goal as outcomes the user wants to be true, not the recipe they describe.
- contingent_on: null for ordinary criteria. Use a short antecedent string only when this criterion should be graded only if that antecedent happens, for example {"contingent_on": "the provider site blocks online submission", "contingent_antecedent_output_path": "output.blocker", "outcome": "the blocker is reported to the user"}. The antecedent goes in contingent_on; the consequent to grade goes in outcome. Do not encode if/then criteria only in prose.
- contingent_antecedent_output_path: null unless the antecedent is represented by a structured run output field. The only valid form is output.<field> (for example output.blocker). Do not emit transcript paths, selectors, regexes, JSONPath, screenshots, nested paths, or prose conditions here.
- deliverable_kind: null unless this criterion's requested output is satisfied by execution-owned browser-download registration rather than a user-authored extraction field. The only supported non-null value is "registered_download".
- expected_output_value: null unless this criterion requires an exact structured output value the user explicitly provided. Put that value here as data; never hide it in outcome prose. For a goal-judgment field whose expected answer is a boolean (e.g. selected_highest_priority), provide the JSON true or false here.
- expected_output_shape: null unless this criterion requires a generated/provider structured output value whose exact value is not known before the run. Use only one of: reference_code, numeric_identifier, date, address, status_label, money_amount, owner_label, goal_judgment_boolean. goal_judgment_boolean declares that the field is an artifact-computed true/false judgment; the typed boolean expected_output_value is the deterministic backstop. Do not use this as a non-empty-value proxy; leave null when the field's shape is not one of those declared types.
- judgment_predicate: null unless expected_output_shape=goal_judgment_boolean and requested_output_evidence_source=independent_run_evidence. Use only the closed-vocabulary value login_gate_blocks_target, and only when the requested judgment is whether login gating blocks the target path.
- judgment_polarity_when_holds: null unless judgment_predicate is non-null. Set it to the boolean output value that means the predicate holds; for login_gate_blocks_target this is true.
- requested_output_evidence_source: runtime_output|independent_run_evidence|registered_output_parameter|registered_artifact_content. Use runtime_output for site-produced values the code block directly reads from the page or browser state, such as dates, amounts, statuses, identifiers, names, or downloaded-file descriptors. Use independent_run_evidence only for artifact-computed judgments, selections, superlatives, rankings, validation booleans, or "best/correct/cheapest/highest" claims that cannot prove themselves by returning their own answer. Use registered_output_parameter or registered_artifact_content only when the requested datum must be proven by execution-owned registered output parameters or registered artifact content. Do not choose this from output_path names, fixture names, domain keywords, or prose labels; choose it from the user's requested evidence contract.
- When a selection, priority, ranking, or superlative criterion governs a returned requested-output field, set output_path for that returned field and requested_output_evidence_source=independent_run_evidence; the selected value cannot prove its own correctness.
- Example: if the user asks to return the selected or highest-priority document name, emit the criterion with output_path=output.document_name and requested_output_evidence_source=independent_run_evidence.
@ -40,6 +42,7 @@ completion_criteria:
- classification_output_key: null unless kind=validation_classification. Set it to the exact top-level block output field/key that will carry the classification scalar. Do not use JSONPath, regex, transcript paths, screenshots, prose conditions, or nested paths.
- expected_classification: null unless kind=validation_classification. Set it to the exact expected classification target as a boolean or canonical enum/string value. Do not hide this target in outcome prose.
- For validation-only utility/path checks whose goal is to classify whether the reachable path is login-only/login-gated, prefer kind=validation_classification with classification_output_key=login_only and expected_classification=true. Do not represent that classification target as a requested-output criterion such as output.login_only.
- For validation-only utility/path checks whose goal is to return whether the target path is blocked by a login gate, emit this as a requested-output judgment with kind=outcome, output_path=output.login_gate_blocks_target, expected_output_shape=goal_judgment_boolean, requested_output_evidence_source=independent_run_evidence, judgment_predicate=login_gate_blocks_target, and judgment_polarity_when_holds=true. Do not use kind=validation_classification for this returned field.
- Capture ALL distinct outcomes the goal requires, including implicit constraints the user did not state as a step (for example "added exactly once / not duplicated", "the item is present in the cart"). Mark such items implicit=true.
- A prescribed HOW step is NOT a criterion. "search for X", "open the result lightbox", "click add to cart", "go to the cart page" are methods/navigation, not outcomes — do not list them. A different or shorter path that reaches the same end state must still satisfy the goal.
- Telling/showing/reporting results to the user is NOT a criterion. "tell me what it extracted", "show me the data", "confirm to me it worked" describe the chat reply, which no workflow definition or run can evidence — capture only the underlying workflow outcome (the data is extracted into the run output, the form is submitted) and list nothing for the reporting itself.

View file

@ -1073,6 +1073,8 @@ def build_artifact_metadata_skeleton(
"text": _FILL_CRITERION_TEXT,
"level": "terminal",
"terminal": True,
"judgment_predicate": None,
"judgment_polarity_when_holds": None,
}
],
"terminal_verifier_expectations": [

View file

@ -28,10 +28,12 @@ from skyvern.forge.sdk.copilot.request_policy import (
_coerce_expected_classification,
_coerce_expected_output_shape,
_coerce_expected_output_value,
_coerce_judgment_truth_condition,
_coerce_requested_output_evidence_source,
_normalize_contingent_antecedent_output_path,
_normalize_deliverable_kind,
is_fallback_floor_criterion,
judgment_truth_condition_key,
normalized_criterion_outcome_key,
requested_output_path_for_field,
typed_expected_output_value_key,
@ -147,6 +149,9 @@ def criteria_to_json(criteria: tuple[CompletionCriterion, ...] | list[Completion
item["requested_output_corroborator"] = True
if criterion.mint_degrade is not None:
item["mint_degrade"] = criterion.mint_degrade
if criterion.judgment_truth_condition is not None:
item["judgment_predicate"] = criterion.judgment_truth_condition.predicate
item["judgment_polarity_when_holds"] = criterion.judgment_truth_condition.polarity_when_holds
items.append(item)
return items
@ -222,6 +227,9 @@ def criteria_from_json(raw: Any) -> tuple[CompletionCriterion, ...]:
mint_degrade="turn_unsatisfiable_fallback"
if item.get("mint_degrade") == "turn_unsatisfiable_fallback"
else None,
judgment_truth_condition=_coerce_judgment_truth_condition(
item.get("judgment_predicate"), item.get("judgment_polarity_when_holds")
),
)
)
return tuple(criteria)
@ -230,7 +238,10 @@ def criteria_from_json(raw: Any) -> tuple[CompletionCriterion, ...]:
def _criterion_reconcile_key(criterion: CompletionCriterion) -> str:
contingent_key = criterion.contingent_on or ""
contingent_path_key = criterion.contingent_antecedent_output_path or ""
deliverable_kind_key = f"{criterion.deliverable_kind or ''}\x1fmint_degrade:{criterion.mint_degrade or ''}"
deliverable_kind_key = (
f"{criterion.deliverable_kind or ''}\x1fmint_degrade:{criterion.mint_degrade or ''}"
f"\x1fjudgment:{judgment_truth_condition_key(criterion.judgment_truth_condition)}"
)
expected_output_value_key = typed_expected_output_value_key(criterion.expected_output_value)
expected_output_shape_key = criterion.expected_output_shape or ""
requested_output_evidence_source_key = criterion.requested_output_evidence_source

View file

@ -7,19 +7,25 @@ import textwrap
from collections.abc import Mapping
from typing import Any, Literal, Protocol
import structlog
import yaml
from skyvern.forge.sdk.copilot.completion_verification import CriterionVerdict, EvidenceSourceKind, RunEvidenceSnapshot
from skyvern.forge.sdk.copilot.request_policy import (
CompletionCriterion,
ExpectedOutputValue,
JudgmentPredicate,
JudgmentTruthCondition,
RequestedOutputEvidenceSource,
_coerce_judgment_truth_condition,
_is_judgment_boolean_criterion,
lookup_requested_output_path_alias,
schema_output_path_aliases_from_criteria,
)
from skyvern.utils.yaml_loader import safe_load_no_dates
LOG = structlog.get_logger()
_GOAL_PATH_INDEX_PATTERN = re.compile(r"\[\d+\]")
_REQUESTED_OUTPUT_PREFIX = "output."
_STRUCTURAL_ABSTENTION_REASON_CODE = "structurally_abstained"
@ -28,6 +34,9 @@ _IGNORED_RUNTIME_FIELD_NAMES = frozenset({"evidence_text"})
_INDEPENDENT_EVIDENCE_SOURCES: frozenset[EvidenceSourceKind] = frozenset(
{"independent_page_evidence", "registered_output_parameter", "registered_artifact_content"}
)
# A judgment boolean may only be decided by page evidence observed after the run. The other
# "independent" sources are registered by the generated code, so they can carry a fabricated packet.
_JUDGMENT_EVIDENCE_SOURCES: frozenset[EvidenceSourceKind] = frozenset({"independent_page_evidence"})
_POST_RUN_PAGE_OBSERVATION_LABEL = "post_run_page_observation"
_REGISTERED_ARTIFACT_OBSERVATION_LABEL = "registered_artifact_observation"
_MIN_CARRIER_VALUE_CHARS = 4
@ -37,10 +46,15 @@ _PAGE_EVIDENCE_STAMP_KEYS = frozenset({"workflow_run_id", "observed_after_workfl
class _GroundingCtx(Protocol):
code_artifact_metadata: object
workflow_verification_evidence: _WorkflowVerificationEvidenceCtx | None
last_workflow_yaml: str | None
workflow_yaml: str | None
class _WorkflowVerificationEvidenceCtx(Protocol):
code_artifact_metadata: object
def split_requested_output_criteria(
criteria: list[CompletionCriterion],
) -> tuple[list[CompletionCriterion], list[CompletionCriterion]]:
@ -49,7 +63,7 @@ def split_requested_output_criteria(
for criterion in criteria:
if (
criterion.output_path
and criterion.level != "definition"
and (criterion.level != "definition" or _is_judgment_boolean_criterion(criterion))
and not criterion.method_mandated
and criterion.kind != "validation_classification"
and _normalize_output_path(criterion.output_path)
@ -66,16 +80,20 @@ def grade_requested_output_criteria(
snapshot: RunEvidenceSnapshot,
) -> list[CriterionVerdict]:
requested_path_aliases = schema_output_path_aliases_from_criteria(criteria)
raw_authored_paths_by_label = _authored_output_contract_paths_by_label(copilot_ctx)
authored_paths_by_label = _authored_output_contract_paths_by_label(copilot_ctx, requested_path_aliases)
declared_judgment_output_paths = _producer_declared_judgment_output_paths(copilot_ctx.code_artifact_metadata)
metadata = _grounding_code_artifact_metadata(copilot_ctx)
raw_authored_paths_by_label = _authored_output_contract_paths_by_label(copilot_ctx, metadata=metadata)
authored_paths_by_label = _authored_output_contract_paths_by_label(
copilot_ctx, requested_path_aliases, metadata=metadata
)
declared_judgment_output_paths = _producer_declared_judgment_output_paths(metadata)
declared_judgment_truth_conditions = _producer_declared_judgment_truth_conditions(metadata)
verdicts: list[CriterionVerdict] = []
for criterion in criteria:
path = _normalize_output_path(criterion.output_path)
# A carrier verdict is already sourced from independent evidence and only returned on a
# positive confirmation, so the early return skips the self-emission veto path that no
# longer applies; abstentions fall through to normal authored-path grading.
carrier_verdict = _carrier_confirmation(criterion, path, snapshot)
# A carrier verdict is sourced from independent evidence. The judgment arm may return an
# unsatisfied evidence_contradicts verdict, which must reach the grade loop unswallowed;
# scalar-carrier abstentions and non-confirmations fall through to normal authored-path grading.
carrier_verdict = _carrier_confirmation(criterion, path, snapshot, declared_judgment_truth_conditions)
if carrier_verdict is not None:
verdicts.append(carrier_verdict)
continue
@ -92,22 +110,6 @@ def grade_requested_output_criteria(
)
)
continue
independent_labels = {
label for label, source in snapshot.block_output_sources.items() if source in _INDEPENDENT_EVIDENCE_SOURCES
}
accepted_runtime_outputs = dict(
_accepted_runtime_outputs(snapshot.block_outputs, accepted_labels, independent_labels)
)
accepted_output_sources = {
label: source
for label, source in snapshot.block_output_sources.items()
if label in accepted_runtime_outputs
}
accepted_authored_paths = set().union(*(authored_paths_by_label[label] for label in accepted_labels))
accepted_raw_authored_paths = set().union(
*(raw_authored_paths_by_label.get(label, set()) for label in accepted_labels)
)
projection_roots = _runtime_projection_roots(accepted_authored_paths | accepted_raw_authored_paths)
expected_value = criterion.expected_output_value
expected_shape = criterion.expected_output_shape
grounding_mode = _criterion_grounding_mode(criterion)
@ -121,6 +123,28 @@ def grade_requested_output_criteria(
_is_judgment_boolean_criterion(criterion) or requires_declared_independent_evidence
)
trace_fields = _criterion_trace_fields(criterion, grounding_mode, effective_evidence_source)
independent_labels = {
label for label, source in snapshot.block_output_sources.items() if source in _INDEPENDENT_EVIDENCE_SOURCES
}
accepted_runtime_output_items = _accepted_runtime_outputs(
snapshot.block_outputs, accepted_labels, independent_labels
)
if judgment_grounding_bars_self_emission:
accepted_runtime_output_items = sorted(
accepted_runtime_output_items,
key=lambda item: 0 if item[0] in independent_labels else 1,
)
accepted_runtime_outputs = dict(accepted_runtime_output_items)
accepted_output_sources = {
label: source
for label, source in snapshot.block_output_sources.items()
if label in accepted_runtime_outputs
}
accepted_authored_paths = set().union(*(authored_paths_by_label[label] for label in accepted_labels))
accepted_raw_authored_paths = set().union(
*(raw_authored_paths_by_label.get(label, set()) for label in accepted_labels)
)
projection_roots = _runtime_projection_roots(accepted_authored_paths | accepted_raw_authored_paths)
if expected_value is not None:
refutation = _independent_evidence_text_refutation(
accepted_runtime_outputs,
@ -266,10 +290,13 @@ def grade_requested_output_criteria(
def _carrier_confirmation(
criterion: CompletionCriterion, path: str, snapshot: RunEvidenceSnapshot
criterion: CompletionCriterion,
path: str,
snapshot: RunEvidenceSnapshot,
declared_judgment_truth_conditions: Mapping[str, JudgmentTruthCondition],
) -> CriterionVerdict | None:
if _is_judgment_boolean_criterion(criterion):
return None
if _is_judgment_boolean_criterion(criterion) or path in declared_judgment_truth_conditions:
return _judgment_carrier_verdict(criterion, path, snapshot, declared_judgment_truth_conditions)
value_text = _carrier_scalar_text(criterion.expected_output_value)
if value_text is None or len(value_text) < _MIN_CARRIER_VALUE_CHARS:
return None
@ -509,6 +536,233 @@ def _producer_declared_judgment_output_paths(metadata: object) -> set[str]:
return paths
def _producer_declared_judgment_truth_conditions(metadata: object) -> dict[str, JudgmentTruthCondition]:
if not isinstance(metadata, Mapping):
return {}
conditions: dict[str, JudgmentTruthCondition] = {}
for artifact in metadata.values():
if not isinstance(artifact, Mapping):
continue
declared_criteria = artifact.get("completion_criteria")
if not isinstance(declared_criteria, list):
continue
for declared in declared_criteria:
if not isinstance(declared, Mapping):
continue
normalized = _normalize_output_path(declared.get("output_path"))
condition = _coerce_judgment_truth_condition(
declared.get("judgment_predicate"), declared.get("judgment_polarity_when_holds")
)
if condition is None and declared.get("requested_output_evidence_source") == "independent_run_evidence":
condition = _producer_declared_path_judgment_truth_condition(normalized)
if normalized and condition is not None:
conditions.setdefault(normalized, condition)
return conditions
def _producer_declared_path_judgment_truth_condition(output_path: str) -> JudgmentTruthCondition | None:
if output_path == "login_gate_blocks_target":
return JudgmentTruthCondition(predicate="login_gate_blocks_target", polarity_when_holds=True)
return None
def _grounding_code_artifact_metadata(copilot_ctx: _GroundingCtx) -> object:
metadata = copilot_ctx.code_artifact_metadata
if _metadata_has_authored_output_contract(metadata):
return metadata
evidence = copilot_ctx.workflow_verification_evidence
if evidence is not None and _metadata_has_authored_output_contract(evidence.code_artifact_metadata):
return evidence.code_artifact_metadata
return metadata
def _metadata_has_authored_output_contract(metadata: object) -> bool:
if not isinstance(metadata, Mapping):
return False
return any(
isinstance(artifact, Mapping) and bool(_artifact_contract_paths(artifact)) for artifact in metadata.values()
)
def _judgment_carrier_verdict(
criterion: CompletionCriterion,
path: str,
snapshot: RunEvidenceSnapshot,
declared_judgment_truth_conditions: Mapping[str, JudgmentTruthCondition],
) -> CriterionVerdict | None:
expected_value = _bool_canonical(criterion.expected_output_value)
if expected_value is None:
expected_value = _emitted_judgment_boolean(snapshot, path)
if expected_value is None:
return None
condition = criterion.judgment_truth_condition or declared_judgment_truth_conditions.get(path)
if condition is None:
return None
origin = "criterion" if criterion.judgment_truth_condition is not None else "producer_metadata"
trace_fields = _criterion_trace_fields(
criterion,
"judgment_boolean",
"independent_run_evidence"
if path in declared_judgment_truth_conditions
else criterion.requested_output_evidence_source,
)
confirmed: CriterionVerdict | None = None
confirmed_label: str | None = None
abstention_label: str | None = None
abstention_source: EvidenceSourceKind | None = None
for label, payload in snapshot.block_outputs.items():
source = snapshot.block_output_sources.get(label)
if source not in _JUDGMENT_EVIDENCE_SOURCES or not isinstance(payload, Mapping):
continue
if abstention_label is None:
abstention_label = label
abstention_source = source
predicate_holds = _judgment_packet_predicate_holds(condition.predicate, payload)
if predicate_holds is None:
continue
observed_value = condition.polarity_when_holds if predicate_holds else not condition.polarity_when_holds
if observed_value != expected_value:
_log_judgment_evidence_verdict(condition, label, "evidence_contradicts", origin)
return CriterionVerdict(
criterion_id=criterion.id,
state="unsatisfied",
reason_code="evidence_contradicts",
evidence_ref=f"block_outputs:{label}",
missing_evidence=f"independent evidence contradicted output.{path}",
evidence_source=source,
**trace_fields,
)
if confirmed is None:
confirmed_label = label
confirmed = CriterionVerdict(
criterion_id=criterion.id,
state="satisfied",
reason_code="evidence_confirms",
evidence_ref=f"block_outputs:{label}",
evidence_source=source,
**trace_fields,
)
if confirmed is not None and confirmed_label is not None:
_log_judgment_evidence_verdict(condition, confirmed_label, "evidence_confirms", origin)
if confirmed is None and abstention_label is not None:
return CriterionVerdict(
criterion_id=criterion.id,
state="unsatisfied",
reason_code=_STRUCTURAL_ABSTENTION_REASON_CODE,
evidence_ref=f"block_outputs:{abstention_label}",
missing_evidence=f"independent evidence did not structurally decide output.{path}",
evidence_source=abstention_source,
**trace_fields,
)
return confirmed
def _emitted_judgment_boolean(snapshot: RunEvidenceSnapshot, path: str) -> bool | None:
parts = _path_parts(path)
observed_values: set[bool] = set()
for label, payload in snapshot.block_outputs.items():
if snapshot.block_output_sources.get(label) in _INDEPENDENT_EVIDENCE_SOURCES:
continue
values, _source_path = _runtime_path_values(payload, parts, path, set())
for value in values:
canonical = _bool_canonical(value)
if canonical is None:
return None
observed_values.add(canonical)
if len(observed_values) == 1:
return next(iter(observed_values))
return None
def _log_judgment_evidence_verdict(
condition: JudgmentTruthCondition, packet_label: str, verdict: str, origin: str
) -> None:
LOG.info(
"copilot_judgment_evidence_verdict",
predicate=condition.predicate,
packet_label=packet_label,
verdict=verdict,
origin=origin,
)
def _judgment_packet_predicate_holds(predicate: JudgmentPredicate, packet: Mapping[str, object]) -> bool | None:
if predicate == "login_gate_blocks_target":
return _login_gate_blocks_target(packet)
return None
def _login_gate_blocks_target(packet: Mapping[str, object]) -> bool | None:
forms = _packet_mapping_list(packet.get("forms"))
result_containers = _packet_mapping_list(packet.get("result_containers"))
navigation_targets = _packet_mapping_list(packet.get("navigation_targets"))
if _has_structural_result_container(result_containers):
return False
if _has_enabled_password_field(forms) and (_has_submit_control(forms) or _has_gated_submit_control(packet, forms)):
return True
if not forms and not result_containers and not navigation_targets:
return None
return None
def _packet_mapping_list(value: object) -> list[Mapping[str, object]]:
if not isinstance(value, list):
return []
return [item for item in value if isinstance(item, Mapping)]
def _has_enabled_password_field(forms: list[Mapping[str, object]]) -> bool:
for form in forms:
for field in _packet_mapping_list(form.get("fields")):
if field.get("disabled") is True:
continue
field_type = field.get("type")
if isinstance(field_type, str) and field_type.casefold() == "password":
return True
return False
def _has_submit_control(forms: list[Mapping[str, object]]) -> bool:
for form in forms:
for control in _packet_mapping_list(form.get("submit_controls")):
control_type = control.get("type")
selector = control.get("selector")
if isinstance(control_type, str) and control_type.casefold() in {"", "button", "submit"}:
return True
if isinstance(selector, str) and selector.strip():
return True
return False
def _has_gated_submit_control(packet: Mapping[str, object], forms: list[Mapping[str, object]]) -> bool:
for form in forms:
for control in _packet_mapping_list(form.get("submit_controls")):
if control.get("disabled") is True:
return True
challenge_state = packet.get("challenge_state")
if isinstance(challenge_state, Mapping):
if challenge_state.get("gates_submit_controls") is True:
return True
if _packet_mapping_list(challenge_state.get("gated_submit_controls")):
return True
return False
def _has_structural_result_container(containers: list[Mapping[str, object]]) -> bool:
for container in containers:
row_count = container.get("row_count")
if isinstance(row_count, int) and row_count > 0:
return True
for key in ("selector", "id", "tag", "row_selector"):
value = container.get(key)
if isinstance(value, str) and value.strip():
return True
rows = container.get("sample_rows")
if isinstance(rows, list) and rows:
return True
return False
def _schema_boolean_output_paths(schema: Mapping[str, Any] | None, prefix: str = "") -> set[str]:
if not isinstance(schema, Mapping):
return set()
@ -534,9 +788,9 @@ def _schema_boolean_output_paths(schema: Mapping[str, Any] | None, prefix: str =
def _authored_output_contract_paths_by_label(
copilot_ctx: _GroundingCtx, aliases: dict[str, str] | None = None
copilot_ctx: _GroundingCtx, aliases: dict[str, str] | None = None, *, metadata: object | None = None
) -> dict[str, set[str]]:
metadata = copilot_ctx.code_artifact_metadata
metadata = metadata if metadata is not None else copilot_ctx.code_artifact_metadata
code_by_label = _code_blocks_by_label(copilot_ctx)
paths_by_label: dict[str, set[str]] = {}
if isinstance(metadata, Mapping):

View file

@ -203,6 +203,8 @@ RequestedOutputEvidenceSource = Literal[
"registered_output_parameter",
"registered_artifact_content",
]
JudgmentPredicate = Literal["login_gate_blocks_target"]
_JUDGMENT_PREDICATES: frozenset[str] = frozenset(get_args(JudgmentPredicate))
_CRITERION_KINDS: frozenset[str] = frozenset({"outcome", "terminal_action", "validation_classification"})
_TERMINAL_ACTION_FAMILIES: frozenset[str] = frozenset({"request", "application", "form", "order"})
_EXPECTED_OUTPUT_SHAPES: frozenset[str] = frozenset(get_args(ExpectedOutputShape))
@ -230,8 +232,11 @@ _OUTPUT_NEGATED_INTENT_PREFIX_RE = re.compile(
)
_OUTPUT_EXPLICIT_FIELD_KEY_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
_OUTPUT_NAMED_FIELD_RE = re.compile(
r"\b(?:(?:requested[-\s]+output|output|return(?:ed)?|final|structured)\s+)?fields?\s+"
r"(?:named|called)\s+([A-Za-z_][A-Za-z0-9_]*)\b",
r"\b(?:"
r"(?:(?:requested[-\s]+output|output|return(?:ed)?|final|structured)\s+)?fields?\s+(?:named|called)"
r"|(?:requested[-\s]+)?output\s+(?:named|called)"
r"|named\s+(?:requested[-\s]+)?output"
r")\s+`?([A-Za-z_][A-Za-z0-9_]*)`?\b",
re.I,
)
_OUTPUT_ACRONYM_RE = re.compile(r"\b[A-Z][A-Z0-9]{1,10}\b")
@ -254,6 +259,15 @@ _OUTPUT_OUTCOME_WORDS = frozenset(
)
@dataclass(frozen=True)
class JudgmentTruthCondition:
"""Per-polarity page-evidence predicate a judgment boolean is decidable by. ``polarity_when_holds``
is the emitted boolean value that corresponds to the predicate holding on the independent packet."""
predicate: JudgmentPredicate
polarity_when_holds: bool
@dataclass(frozen=True)
class CompletionCriterion:
id: str
@ -276,6 +290,7 @@ class CompletionCriterion:
expected_classification: ClassificationTarget | None = None
requested_output_corroborator: bool = False
mint_degrade: Literal["turn_unsatisfiable_fallback"] | None = None
judgment_truth_condition: JudgmentTruthCondition | None = None
@dataclass
@ -447,6 +462,12 @@ def typed_expected_output_value_key(value: ExpectedOutputValue | None) -> str:
return ""
def judgment_truth_condition_key(condition: JudgmentTruthCondition | None) -> str:
if condition is None:
return ""
return f"{condition.predicate}:{'t' if condition.polarity_when_holds else 'f'}"
def _criterion_grounding_mode(
criterion: CompletionCriterion,
) -> Literal["exact_value", "shape", "missing", "judgment_boolean"]:
@ -684,6 +705,14 @@ def _coerce_expected_output_shape(value: Any) -> ExpectedOutputShape | None:
return None
def _coerce_judgment_truth_condition(predicate: Any, polarity: Any) -> JudgmentTruthCondition | None:
if not isinstance(predicate, str) or predicate not in _JUDGMENT_PREDICATES:
return None
if not isinstance(polarity, bool):
return None
return JudgmentTruthCondition(predicate=cast(JudgmentPredicate, predicate), polarity_when_holds=polarity)
def _coerce_requested_output_evidence_source(value: Any) -> RequestedOutputEvidenceSource:
if isinstance(value, str) and value in _REQUESTED_OUTPUT_EVIDENCE_SOURCES:
return cast(RequestedOutputEvidenceSource, value)
@ -805,6 +834,9 @@ def _parse_completion_criteria(raw: Any) -> list[CompletionCriterion]:
coerced_judgment_bool = _canonical_bool_string(expected_output_value)
if coerced_judgment_bool is not None:
expected_output_value = coerced_judgment_bool
judgment_truth_condition = _coerce_judgment_truth_condition(
item.get("judgment_predicate"), item.get("judgment_polarity_when_holds")
)
contingent_on_raw = item.get("contingent_on")
contingent_on = (
" ".join(contingent_on_raw.split())[:_COMPLETION_CRITERION_CONTINGENT_ON_MAX_CHARS].strip()
@ -844,6 +876,7 @@ def _parse_completion_criteria(raw: Any) -> list[CompletionCriterion]:
typed_expected_output_value_key(expected_output_value),
expected_output_shape or "",
requested_output_evidence_source,
judgment_truth_condition_key(judgment_truth_condition),
)
if key in seen:
continue
@ -869,8 +902,17 @@ def _parse_completion_criteria(raw: Any) -> list[CompletionCriterion]:
terminal_action_family=_coerce_terminal_action_family(item.get("terminal_action_family"), kind),
classification_output_key=classification_output_key,
expected_classification=expected_classification,
judgment_truth_condition=judgment_truth_condition,
)
)
if judgment_truth_condition is not None:
LOG.info(
"copilot_judgment_truth_condition_minted",
mint_source="classifier",
predicate=judgment_truth_condition.predicate,
polarity_when_holds=judgment_truth_condition.polarity_when_holds,
criterion_id=criteria[-1].id,
)
if len(criteria) >= _MAX_COMPLETION_CRITERIA:
break
return criteria
@ -1208,6 +1250,23 @@ def _requested_output_evidence_sources_from_criteria(
return sources
def _requested_output_judgment_conditions_from_criteria(
criteria: list[CompletionCriterion],
requested_specs: list[tuple[str, str, str]],
) -> dict[str, JudgmentTruthCondition]:
conditions: dict[str, JudgmentTruthCondition] = {}
for criterion in criteria:
if criterion.level == "definition" or criterion.method_mandated or criterion.judgment_truth_condition is None:
continue
for field_name, output_path, _field_label in requested_specs:
if criterion.output_path != output_path and not _criterion_text_covers_requested_output(
criterion, field_name
):
continue
conditions.setdefault(output_path, criterion.judgment_truth_condition)
return conditions
def _requested_output_criterion_id(output_path: str) -> str:
slug = "_".join(_word_tokens(output_path)) or "field"
return f"{_REQUESTED_OUTPUT_CRITERION_ID_PREFIX}{slug}"
@ -1429,6 +1488,9 @@ def _apply_requested_output_completion_criteria(
source_by_output_path = _requested_output_evidence_sources_from_criteria(
policy.completion_criteria, requested_specs
)
judgment_condition_by_output_path = _requested_output_judgment_conditions_from_criteria(
policy.completion_criteria, requested_specs
)
metadata_by_output_path: dict[str, tuple[str | None, str | None, Literal["registered_download"] | None]] = {}
for criterion in policy.completion_criteria:
@ -1477,6 +1539,7 @@ def _apply_requested_output_completion_criteria(
contingent_on=metadata_by_output_path.get(output_path, (None, None, None))[0],
contingent_antecedent_output_path=metadata_by_output_path.get(output_path, (None, None, None))[1],
deliverable_kind=metadata_by_output_path.get(output_path, (None, None, None))[2],
judgment_truth_condition=judgment_condition_by_output_path.get(output_path),
)
for _field_name, output_path, field_label in requested_specs
]
@ -1571,6 +1634,9 @@ def _render_active_criteria_for_prompt(criteria: list[CompletionCriterion] | Non
item["expected_output_shape"] = criterion.expected_output_shape
if criterion.requested_output_evidence_source != "runtime_output":
item["requested_output_evidence_source"] = criterion.requested_output_evidence_source
if criterion.judgment_truth_condition is not None:
item["judgment_predicate"] = criterion.judgment_truth_condition.predicate
item["judgment_polarity_when_holds"] = criterion.judgment_truth_condition.polarity_when_holds
if criterion.classification_output_key:
item["classification_output_key"] = criterion.classification_output_key
if criterion.expected_classification is not None:

View file

@ -51,7 +51,11 @@ from skyvern.forge.sdk.copilot.reached_download_target import (
ReachedDownloadTarget,
derive_from_block_outputs,
)
from skyvern.forge.sdk.copilot.request_policy import CompletionCriterion, is_fallback_floor_criterion
from skyvern.forge.sdk.copilot.request_policy import (
CompletionCriterion,
_is_judgment_boolean_criterion,
is_fallback_floor_criterion,
)
from skyvern.forge.sdk.copilot.runtime import PreRunPageReference, RegisteredArtifactEvidence
from skyvern.forge.sdk.copilot.terminal_predicates import outcome_fully_verified
from skyvern.forge.sdk.copilot.tracing_setup import copilot_span
@ -294,9 +298,16 @@ def _authored_output_contract_path(value: object) -> str:
return f"output.{path}"
def _split_criteria_by_plane(criteria: list[Any]) -> tuple[list[CompletionCriterion], list[CompletionCriterion]]:
run_criteria = [c for c in criteria if getattr(c, "level", "run") != "definition"]
definition_criteria = [c for c in criteria if getattr(c, "level", "run") == "definition"]
def _split_criteria_by_plane(
criteria: list[CompletionCriterion],
) -> tuple[list[CompletionCriterion], list[CompletionCriterion]]:
run_criteria: list[CompletionCriterion] = []
definition_criteria: list[CompletionCriterion] = []
for criterion in criteria:
if criterion.level != "definition" or (criterion.output_path and _is_judgment_boolean_criterion(criterion)):
run_criteria.append(criterion)
else:
definition_criteria.append(criterion)
return run_criteria, definition_criteria
@ -792,6 +803,15 @@ def _bind_independent_post_run_page_evidence(
key: value for key, value in evidence.items() if key not in _POST_RUN_PAGE_EVIDENCE_STAMP_KEYS
}
block_output_sources[_POST_RUN_PAGE_OBSERVATION_LABEL] = "independent_page_evidence"
LOG.info(
"copilot_post_run_page_evidence_snapshot_binding",
workflow_run_id=run_id,
packet_label=_POST_RUN_PAGE_OBSERVATION_LABEL,
evidence_source="independent_page_evidence",
evidence_workflow_run_id=evidence.get("workflow_run_id"),
evidence_observed_after_workflow_run=evidence.get("observed_after_workflow_run"),
evidence_source_tool=evidence.get("source_tool"),
)
def _bind_registered_artifact_evidence(

View file

@ -110,8 +110,11 @@ from skyvern.forge.sdk.copilot.reached_download_target import (
code_is_download_intent,
)
from skyvern.forge.sdk.copilot.request_policy import (
CompletionCriterion,
JudgmentPredicate,
RequestedOutputEvidenceSource,
_coerce_requested_output_evidence_source,
_is_judgment_boolean_criterion,
)
from skyvern.forge.sdk.copilot.runtime import (
AgentContext,
@ -252,6 +255,17 @@ class CodeArtifactCompletionCriterion(BaseModel):
terminal: bool | None = None
output_path: str | None = None
requested_output_evidence_source: RequestedOutputEvidenceSource | None = None
judgment_predicate: JudgmentPredicate | None = Field(
default=None,
description=(
"For a judgment-boolean criterion, the closed-vocabulary page-evidence predicate the "
"independent post-run packet decides this boolean by (e.g. `login_gate_blocks_target`)."
),
)
judgment_polarity_when_holds: bool | None = Field(
default=None,
description="The emitted boolean value that corresponds to `judgment_predicate` holding on the packet.",
)
class CodeArtifactScopedRef(BaseModel):
@ -1656,13 +1670,10 @@ def _recorded_outcome_missing_output_paths(ctx: AgentContext) -> set[str]:
def _requested_output_child_paths(ctx: AgentContext) -> set[str]:
if _copilot_block_authoring_policy(ctx) != BlockAuthoringPolicy.CODE_ONLY_BROWSER:
return set()
request_policy = getattr(ctx, "request_policy", None)
criteria_source = getattr(request_policy, "graded_completion_criteria", None)
criteria = criteria_source() if callable(criteria_source) else getattr(request_policy, "completion_criteria", [])
if not isinstance(criteria, list):
return set()
paths: set[str] = set()
for criterion in criteria:
for criterion in _active_completion_criteria(ctx):
if isinstance(criterion, CompletionCriterion) and _independent_judgment_output_criterion(criterion):
continue
if getattr(criterion, "level", None) == "definition":
continue
if getattr(criterion, "method_mandated", False):
@ -1675,6 +1686,32 @@ def _requested_output_child_paths(ctx: AgentContext) -> set[str]:
return paths
def _active_completion_criteria(ctx: AgentContext) -> list[CompletionCriterion]:
request_policy = ctx.request_policy
if request_policy is None:
return []
return request_policy.graded_completion_criteria()
def _independent_judgment_output_criterion(criterion: CompletionCriterion) -> bool:
return criterion.requested_output_evidence_source == "independent_run_evidence" and _is_judgment_boolean_criterion(
criterion
)
def _independent_judgment_output_paths(ctx: AgentContext) -> set[str]:
paths: set[str] = set()
for criterion in _active_completion_criteria(ctx):
if not isinstance(criterion, CompletionCriterion):
continue
if not _independent_judgment_output_criterion(criterion):
continue
path = _canonical_requested_output_path(criterion.output_path)
if path and _output_path_has_child(path):
paths.add(path)
return paths
def _canonical_requested_output_path(value: object) -> str:
if not isinstance(value, str):
return ""
@ -2296,7 +2333,11 @@ def _runtime_output_repair_contract_from_recorded_outcome(ctx: AgentContext) ->
def _output_contract_required_paths_source(ctx: AgentContext) -> tuple[set[str], str, str]:
runtime_contract = _runtime_output_repair_contract_from_recorded_outcome(ctx)
if runtime_contract is not None:
return runtime_contract.required_paths, runtime_contract.source, runtime_contract.reason_code
return (
runtime_contract.required_paths - _independent_judgment_output_paths(ctx),
runtime_contract.source,
runtime_contract.reason_code,
)
required_paths, source, reason_code = _required_child_output_paths_for_authoring(ctx)
repair_context = getattr(ctx, "last_code_authoring_repair_context", None)
if required_paths or not isinstance(repair_context, CodeAuthoringRepairContext):
@ -2310,6 +2351,7 @@ def _output_contract_required_paths_source(ctx: AgentContext) -> tuple[set[str],
*repair_context.required_code_return_paths,
]
)
required_paths -= _independent_judgment_output_paths(ctx)
source = str(repair_context.metadata_contract_source or "").strip() or "metadata_reject"
reason_code = (
str(repair_context.metadata_contract_reason_code or "").strip()

View file

@ -55,7 +55,7 @@ from skyvern.forge.sdk.copilot.output_contracts import (
)
from skyvern.forge.sdk.copilot.output_utils import sanitize_tool_result_for_llm
from skyvern.forge.sdk.copilot.reached_download_target import ReachedDownloadTarget
from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
from skyvern.forge.sdk.copilot.request_policy import CompletionCriterion, JudgmentTruthCondition, RequestPolicy
from skyvern.forge.sdk.copilot.run_outcome import TERMINAL_CHALLENGE_BLOCKER_REASON_CODE, RecordedRunOutcome
from skyvern.forge.sdk.copilot.runtime import AgentContext
from skyvern.forge.sdk.copilot.tools import (
@ -4110,6 +4110,223 @@ class TestCodeRepairProgressClassification:
schema = json.loads(scaffolded[0]["claimed_outcomes"][0]["extraction_schema"])
assert schema["properties"]["output"]["properties"]["recorded_value"] == {}
def test_independent_judgment_output_is_not_required_as_code_return_path(self) -> None:
judgment_criterion = CompletionCriterion(
id="login_gate",
outcome="the target path is blocked by a login gate",
output_path="output.login_gate_blocks_target",
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target",
polarity_when_holds=True,
),
)
judgment_only_ctx = _code_only_ctx()
judgment_only_ctx.request_policy = RequestPolicy(completion_criteria=[judgment_criterion])
judgment_only_paths, _, _ = workflow_update_module._output_contract_required_paths_source(judgment_only_ctx)
assert judgment_only_paths == set()
ctx = _code_only_ctx()
ctx.request_policy = RequestPolicy(
completion_criteria=[
judgment_criterion,
CompletionCriterion(
id="record_id",
outcome="the record id is returned",
output_path="output.record_id",
),
]
)
required_paths, source, reason_code = workflow_update_module._output_contract_required_paths_source(ctx)
assert required_paths == {"output.record_id"}
assert source == "requested_output_contract"
assert reason_code == "requested_output_contract_missing_output_coverage"
def test_independent_judgment_shape_output_is_not_required_as_code_return_path(self) -> None:
ctx = _code_only_ctx()
ctx.request_policy = RequestPolicy(
completion_criteria=[
CompletionCriterion(
id="login_gate",
outcome="the target path is blocked by a login gate",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
)
]
)
required_paths, _, _ = workflow_update_module._output_contract_required_paths_source(ctx)
assert required_paths == set()
def test_independent_judgment_repair_context_is_not_rehydrated_as_code_return_path(self) -> None:
ctx = _code_only_ctx()
ctx.request_policy = RequestPolicy(
completion_criteria=[
CompletionCriterion(
id="login_gate",
outcome="the target path is blocked by a login gate",
output_path="output.login_gate_blocks_target",
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target",
polarity_when_holds=True,
),
)
]
)
ctx.last_code_authoring_repair_context = CodeAuthoringRepairContext(
block_label="judge_login_gate_blocks_target",
reason_code="metadata_reject",
required_goal_value_paths=["output.login_gate_blocks_target"],
required_extraction_schema_paths=["output.login_gate_blocks_target"],
required_code_return_paths=["output.login_gate_blocks_target"],
metadata_contract_source="requested_output_contract",
metadata_contract_reason_code="requested_output_contract_missing_output_coverage",
)
required_paths, _, _ = workflow_update_module._output_contract_required_paths_source(ctx)
result = workflow_update_module._metadata_contract_run_preflight_reject(ctx, _SAFE_CODE_YAML, [])
assert required_paths == set()
assert result is None
def test_mixed_repair_context_keeps_non_judgment_code_return_path(self) -> None:
ctx = _code_only_ctx()
ctx.request_policy = RequestPolicy(
completion_criteria=[
CompletionCriterion(
id="login_gate",
outcome="the target path is blocked by a login gate",
output_path="output.login_gate_blocks_target",
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target",
polarity_when_holds=True,
),
)
]
)
ctx.last_code_authoring_repair_context = CodeAuthoringRepairContext(
block_label="extract_entry_output",
reason_code="metadata_reject",
required_goal_value_paths=["output.login_gate_blocks_target", "output.record_id"],
required_extraction_schema_paths=["output.login_gate_blocks_target", "output.record_id"],
required_code_return_paths=["output.login_gate_blocks_target", "output.record_id"],
metadata_contract_source="requested_output_contract",
metadata_contract_reason_code="requested_output_contract_missing_output_coverage",
)
required_paths, source, reason_code = workflow_update_module._output_contract_required_paths_source(ctx)
assert required_paths == {"output.record_id"}
assert source == "requested_output_contract"
assert reason_code == "requested_output_contract_missing_output_coverage"
def test_independent_judgment_runtime_repair_fact_is_not_rehydrated_as_code_return_path(self) -> None:
ctx = _code_only_ctx()
ctx.request_policy = RequestPolicy(
completion_criteria=[
CompletionCriterion(
id="login_gate",
outcome="the target path is blocked by a login gate",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
)
]
)
ctx.latest_recorded_build_test_outcome = RecordedBuildTestOutcome(
phase="persisted_block_run",
attempted_tool="update_and_run_blocks",
verdict="repairable_failure",
reason_code="outcome_not_demonstrated",
workflow_run_id="wr_current",
structural_failure_identity="completion:runtime-output",
runtime_output_repair_facts=[
{
"workflow_run_id": "wr_current",
"block_label": "judge_login_gate_blocks_target",
"output_path": "output.login_gate_blocks_target",
"output_root": "output",
"criterion_id": "__copilot_requested_output__output_login_gate_blocks_target",
"reason_code": "structurally_abstained",
"grounding_mode": "judgment_boolean",
"value_status": "structural_abstained",
}
],
)
required_paths, _, _ = workflow_update_module._output_contract_required_paths_source(ctx)
assert required_paths == set()
def test_mixed_runtime_repair_facts_keep_non_judgment_code_return_path(self) -> None:
ctx = _code_only_ctx()
ctx.request_policy = RequestPolicy(
completion_criteria=[
CompletionCriterion(
id="login_gate",
outcome="the target path is blocked by a login gate",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
),
CompletionCriterion(
id="record_id",
outcome="the record id is returned",
output_path="output.record_id",
),
]
)
ctx.latest_recorded_build_test_outcome = RecordedBuildTestOutcome(
phase="persisted_block_run",
attempted_tool="update_and_run_blocks",
verdict="repairable_failure",
reason_code="outcome_not_demonstrated",
workflow_run_id="wr_current",
structural_failure_identity="completion:runtime-output",
runtime_output_repair_facts=[
{
"workflow_run_id": "wr_current",
"block_label": "extract_entry_output",
"output_path": "output.login_gate_blocks_target",
"output_root": "output",
"criterion_id": "__copilot_requested_output__output_login_gate_blocks_target",
"reason_code": "structurally_abstained",
"grounding_mode": "judgment_boolean",
"value_status": "structural_abstained",
},
{
"workflow_run_id": "wr_current",
"block_label": "extract_entry_output",
"output_path": "output.record_id",
"output_root": "output",
"criterion_id": "record_id",
"reason_code": "structurally_abstained",
"grounding_mode": "missing",
"value_status": "structural_abstained",
},
],
)
required_paths, source, reason_code = workflow_update_module._output_contract_required_paths_source(ctx)
assert required_paths == {"output.record_id"}
assert source == "runtime_output_repair"
assert reason_code == "runtime_output_repair_required"
def test_runtime_output_facts_record_same_run_null_without_evidence_text_backfill(self) -> None:
result = {
"ok": True,

View file

@ -9,6 +9,7 @@ from types import SimpleNamespace
from typing import Any
import pytest
from structlog.testing import capture_logs
from skyvern.config import settings
from skyvern.forge.sdk.copilot.agent import (
@ -75,6 +76,7 @@ from skyvern.forge.sdk.copilot.hooks import _tool_completion_satisfies_turn
from skyvern.forge.sdk.copilot.reached_download_target import ReachedDownloadTarget
from skyvern.forge.sdk.copilot.request_policy import (
CompletionCriterion,
JudgmentTruthCondition,
RequestPolicy,
_parse_completion_criteria,
build_classifier_fallback_floor,
@ -103,6 +105,7 @@ from skyvern.forge.sdk.copilot.tools import (
from skyvern.forge.sdk.copilot.tools.completion import (
_POST_RUN_PAGE_OBSERVATION_LABEL,
_artifact_health_blocker_from_result,
_completion_verification_from_run_result,
_reconcile_download_completion_criterion,
)
from skyvern.forge.sdk.copilot.tools.composition_capture import _active_run_terminal_monitor_enabled
@ -9428,6 +9431,771 @@ def test_judgment_boolean_refuted_when_independent_evidence_resolves_false_leaf(
assert verdicts[0].evidence_source == "independent_page_evidence"
def _login_gate_packet() -> dict[str, object]:
return {
"forms": [
{
"fields": [{"type": "password", "selector": "#password", "disabled": False}],
"submit_controls": [{"type": "submit", "selector": "button[type=submit]", "disabled": False}],
}
],
"navigation_targets": [{"href": "https://example.test/account", "selector": "a.account"}],
"result_containers": [],
}
def _gated_login_packet() -> dict[str, object]:
return {
"forms": [
{
"fields": [{"type": "password", "selector": "#password", "disabled": False}],
"submit_controls": [{"type": "submit", "selector": "button[type=submit]", "disabled": True}],
}
],
"navigation_targets": [{"href": "https://example.test/account", "selector": "a.account"}],
"result_containers": [],
"challenge_state": {
"gates_submit_controls": True,
"gated_submit_controls": [{"selector": "button[type=submit]", "disabled": True}],
},
}
def _target_reached_packet() -> dict[str, object]:
return {
"forms": [],
"navigation_targets": [{"href": "https://example.test/account", "selector": "a.account"}],
"result_containers": [{"selector": "#account-summary", "row_count": 1}],
}
def test_judgment_truth_condition_confirms_login_gate_from_independent_packet() -> None:
ctx = _run_ctx()
ctx.code_artifact_metadata = _metadata_for_requested_paths("login_gate_blocks_target")
criteria = [
replace(
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
requested_output_evidence_source="independent_run_evidence",
),
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
),
)
]
with capture_logs() as logs:
verdicts = grade_requested_output_criteria(
ctx,
criteria,
RunEvidenceSnapshot(
block_outputs={_POST_RUN_PAGE_OBSERVATION_LABEL: _login_gate_packet()},
block_output_sources={_POST_RUN_PAGE_OBSERVATION_LABEL: "independent_page_evidence"},
),
)
assert verdicts[0].state == "satisfied"
assert verdicts[0].reason_code == "evidence_confirms"
assert verdicts[0].evidence_ref == f"block_outputs:{_POST_RUN_PAGE_OBSERVATION_LABEL}"
assert verdicts[0].evidence_source == "independent_page_evidence"
assert logs[-1]["event"] == "copilot_judgment_evidence_verdict"
assert logs[-1]["predicate"] == "login_gate_blocks_target"
assert logs[-1]["packet_label"] == _POST_RUN_PAGE_OBSERVATION_LABEL
assert logs[-1]["verdict"] == "evidence_confirms"
assert logs[-1]["origin"] == "criterion"
def test_bound_gated_submit_login_packet_wins_over_self_emitted_judgment() -> None:
ctx = _run_ctx()
_set_workflow_labels(ctx, "extract_profile")
ctx.code_artifact_metadata = {
"extract_profile": {
"claimed_outcomes": [{"goal_value_paths": ["login_gate_blocks_target"]}],
"completion_criteria": [
{
"output_path": "output.login_gate_blocks_target",
"requested_output_evidence_source": "independent_run_evidence",
"judgment_predicate": "login_gate_blocks_target",
"judgment_polarity_when_holds": True,
}
],
}
}
ctx.composition_page_evidence = _post_run_page_evidence(
run_id="wr_requested_output",
visible_text_excerpt="",
**_gated_login_packet(),
)
criteria = [
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
)
]
snapshot = _build_run_evidence_snapshot(
ctx,
_requested_output_result({"login_gate_blocks_target": True}),
)
with capture_logs() as logs:
verdicts = grade_requested_output_criteria(ctx, criteria, snapshot)
assert snapshot.block_output_sources[_POST_RUN_PAGE_OBSERVATION_LABEL] == "independent_page_evidence"
assert verdicts[0].state == "satisfied"
assert verdicts[0].reason_code == "evidence_confirms"
assert verdicts[0].evidence_source == "independent_page_evidence"
assert verdicts[0].self_emitted_judgment_not_independent is False
assert any(
log["event"] == "copilot_judgment_evidence_verdict" and log["verdict"] == "evidence_confirms" for log in logs
)
def test_bound_independent_page_output_wins_over_self_emitted_judgment_value() -> None:
ctx = _run_ctx()
ctx.code_artifact_metadata = _metadata_for_requested_paths("login_gate_blocks_target")
criteria = [
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
)
]
verdicts = grade_requested_output_criteria(
ctx,
criteria,
RunEvidenceSnapshot(
block_outputs={
"extract_profile": {"login_gate_blocks_target": True},
_POST_RUN_PAGE_OBSERVATION_LABEL: {"login_gate_blocks_target": True},
},
block_output_sources={
"extract_profile": "runtime_output",
_POST_RUN_PAGE_OBSERVATION_LABEL: "independent_page_evidence",
},
),
)
assert verdicts[0].state == "satisfied"
assert verdicts[0].reason_code == "evidence_confirms"
assert verdicts[0].evidence_ref == f"block_outputs:{_POST_RUN_PAGE_OBSERVATION_LABEL}.login_gate_blocks_target"
assert verdicts[0].evidence_source == "independent_page_evidence"
assert verdicts[0].self_emitted_judgment_not_independent is False
def test_value_less_emitted_false_judgment_is_refuted_by_login_gate_packet() -> None:
ctx = _run_ctx()
_set_workflow_labels(ctx, "extract_profile")
ctx.code_artifact_metadata = {
"extract_profile": {
"claimed_outcomes": [{"goal_value_paths": ["login_gate_blocks_target"]}],
"completion_criteria": [
{
"output_path": "output.login_gate_blocks_target",
"requested_output_evidence_source": "independent_run_evidence",
"judgment_predicate": "login_gate_blocks_target",
"judgment_polarity_when_holds": True,
}
],
}
}
ctx.composition_page_evidence = _post_run_page_evidence(
run_id="wr_requested_output",
visible_text_excerpt="",
**_gated_login_packet(),
)
criteria = [
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
)
]
snapshot = _build_run_evidence_snapshot(
ctx,
_requested_output_result({"login_gate_blocks_target": False}),
)
with capture_logs() as logs:
verdicts = grade_requested_output_criteria(ctx, criteria, snapshot)
assert verdicts[0].state == "unsatisfied"
assert verdicts[0].reason_code == "evidence_contradicts"
assert verdicts[0].evidence_ref == f"block_outputs:{_POST_RUN_PAGE_OBSERVATION_LABEL}"
assert verdicts[0].evidence_source == "independent_page_evidence"
assert verdicts[0].has_exact_value is False
assert any(
log["event"] == "copilot_judgment_evidence_verdict"
and log["predicate"] == "login_gate_blocks_target"
and log["packet_label"] == _POST_RUN_PAGE_OBSERVATION_LABEL
and log["verdict"] == "evidence_contradicts"
for log in logs
)
def test_producer_declared_value_less_judgment_reaches_packet_verdict() -> None:
ctx = _run_ctx()
_set_workflow_labels(ctx, "extract_profile")
ctx.code_artifact_metadata = {
"extract_profile": {
"claimed_outcomes": [{"goal_value_paths": ["login_gate_blocks_target"]}],
"completion_criteria": [
{
"output_path": "output.login_gate_blocks_target",
"requested_output_evidence_source": "independent_run_evidence",
"judgment_predicate": "login_gate_blocks_target",
"judgment_polarity_when_holds": True,
}
],
}
}
ctx.composition_page_evidence = _post_run_page_evidence(
run_id="wr_requested_output",
visible_text_excerpt="",
**_gated_login_packet(),
)
criteria = [
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
requested_output_evidence_source="independent_run_evidence",
)
]
snapshot = _build_run_evidence_snapshot(
ctx,
_requested_output_result({"login_gate_blocks_target": True}),
)
with capture_logs() as logs:
verdicts = grade_requested_output_criteria(ctx, criteria, snapshot)
assert verdicts[0].state == "satisfied"
assert verdicts[0].reason_code == "evidence_confirms"
assert verdicts[0].grounding_mode == "judgment_boolean"
assert verdicts[0].has_exact_value is False
assert verdicts[0].evidence_ref == f"block_outputs:{_POST_RUN_PAGE_OBSERVATION_LABEL}"
assert verdicts[0].evidence_source == "independent_page_evidence"
assert any(
log["event"] == "copilot_judgment_evidence_verdict"
and log["predicate"] == "login_gate_blocks_target"
and log["packet_label"] == _POST_RUN_PAGE_OBSERVATION_LABEL
and log["verdict"] == "evidence_confirms"
and log["origin"] == "producer_metadata"
for log in logs
)
def test_producer_declared_truth_condition_without_metadata_source_reaches_packet_verdict() -> None:
ctx = _run_ctx()
_set_workflow_labels(ctx, "extract_profile")
ctx.code_artifact_metadata = {
"extract_profile": {
"claimed_outcomes": [{"goal_value_paths": ["login_gate_blocks_target"]}],
"completion_criteria": [
{
"output_path": "output.login_gate_blocks_target",
"judgment_predicate": "login_gate_blocks_target",
"judgment_polarity_when_holds": True,
}
],
}
}
ctx.composition_page_evidence = _post_run_page_evidence(
run_id="wr_requested_output",
visible_text_excerpt="",
**_gated_login_packet(),
)
criteria = [
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
requested_output_evidence_source="independent_run_evidence",
)
]
snapshot = _build_run_evidence_snapshot(
ctx,
_requested_output_result({"login_gate_blocks_target": True}),
)
with capture_logs() as logs:
verdicts = grade_requested_output_criteria(ctx, criteria, snapshot)
assert verdicts[0].state == "satisfied"
assert verdicts[0].reason_code == "evidence_confirms"
assert verdicts[0].grounding_mode == "judgment_boolean"
assert verdicts[0].has_exact_value is False
assert verdicts[0].evidence_source == "independent_page_evidence"
assert any(
log["event"] == "copilot_judgment_evidence_verdict"
and log["predicate"] == "login_gate_blocks_target"
and log["origin"] == "producer_metadata"
for log in logs
)
@pytest.mark.asyncio
async def test_definition_level_judgment_output_reaches_independent_packet_grader() -> None:
ctx = _run_ctx()
_set_workflow_labels(ctx, "extract_profile")
ctx.code_artifact_metadata = {
"extract_profile": {
"claimed_outcomes": [{"goal_value_paths": ["login_gate_blocks_target"]}],
"completion_criteria": [
{
"output_path": "output.login_gate_blocks_target",
"requested_output_evidence_source": "independent_run_evidence",
"judgment_predicate": "login_gate_blocks_target",
"judgment_polarity_when_holds": True,
}
],
}
}
ctx.composition_page_evidence = _post_run_page_evidence(
run_id="wr_requested_output",
visible_text_excerpt="",
**_gated_login_packet(),
)
criteria = [
replace(
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
level="definition",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
),
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
),
)
]
with capture_logs() as logs:
verification = await _completion_verification_from_run_result(
ctx,
_requested_output_result({"login_gate_blocks_target": True}),
time.monotonic(),
criteria,
)
assert verification is not None
assert verification.verdicts[0].state == "satisfied"
assert verification.verdicts[0].reason_code == "evidence_confirms"
assert verification.verdicts[0].evidence_source == "independent_page_evidence"
assert any(
log["event"] == "copilot_judgment_evidence_verdict" and log["verdict"] == "evidence_confirms" for log in logs
)
@pytest.mark.asyncio
async def test_authored_output_fallback_preserves_staged_judgment_truth_condition() -> None:
ctx = _run_ctx()
_set_workflow_labels(ctx, "extract_profile")
metadata = {
"extract_profile": {
"claimed_outcomes": [{"goal_value_paths": ["login_gate_blocks_target"]}],
"completion_criteria": [
{
"output_path": "output.login_gate_blocks_target",
"requested_output_evidence_source": "independent_run_evidence",
"judgment_predicate": "login_gate_blocks_target",
"judgment_polarity_when_holds": True,
}
],
}
}
ctx.workflow_verification_evidence = SimpleNamespace(code_artifact_metadata=metadata)
ctx.code_artifact_metadata = {}
ctx.request_policy = RequestPolicy(completion_criteria=build_classifier_fallback_floor([]))
ctx.composition_page_evidence = _post_run_page_evidence(
run_id="wr_requested_output",
visible_text_excerpt="",
**_gated_login_packet(),
)
with capture_logs() as logs:
verification = await _maybe_run_completion_verification(
ctx,
_requested_output_result({"login_gate_blocks_target": True}),
time.monotonic(),
)
assert verification is not None
verdict = verification.verdicts[0]
assert verdict.criterion_id == "__copilot_authored_output__output_login_gate_blocks_target"
assert verdict.state == "satisfied"
assert verdict.reason_code == "evidence_confirms"
assert verdict.grounding_mode == "judgment_boolean"
assert verdict.evidence_source == "independent_page_evidence"
assert any(
log["event"] == "copilot_judgment_evidence_verdict"
and log["predicate"] == "login_gate_blocks_target"
and log["origin"] == "producer_metadata"
for log in logs
)
@pytest.mark.asyncio
async def test_authored_output_fallback_recovers_path_declared_judgment_truth_condition() -> None:
ctx = _run_ctx()
_set_workflow_labels(ctx, "extract_profile")
metadata = {
"extract_profile": {
"claimed_outcomes": [{"goal_value_paths": ["login_gate_blocks_target"]}],
"completion_criteria": [
{
"output_path": "output.login_gate_blocks_target",
"requested_output_evidence_source": "independent_run_evidence",
}
],
}
}
ctx.workflow_verification_evidence = SimpleNamespace(code_artifact_metadata=metadata)
ctx.code_artifact_metadata = {}
ctx.request_policy = RequestPolicy(completion_criteria=build_classifier_fallback_floor([]))
ctx.composition_page_evidence = _post_run_page_evidence(
run_id="wr_requested_output",
visible_text_excerpt="",
**_gated_login_packet(),
)
with capture_logs() as logs:
verification = await _maybe_run_completion_verification(
ctx,
_requested_output_result({"login_gate_blocks_target": True}),
time.monotonic(),
)
assert verification is not None
verdict = verification.verdicts[0]
assert verdict.criterion_id == "__copilot_authored_output__output_login_gate_blocks_target"
assert verdict.state == "satisfied"
assert verdict.reason_code == "evidence_confirms"
assert verdict.grounding_mode == "judgment_boolean"
assert verdict.evidence_source == "independent_page_evidence"
assert any(
log["event"] == "copilot_judgment_evidence_verdict"
and log["predicate"] == "login_gate_blocks_target"
and log["origin"] == "producer_metadata"
for log in logs
)
@pytest.mark.asyncio
async def test_authored_output_duplicate_path_judgment_row_reaches_packet_grader() -> None:
ctx = _run_ctx()
_set_workflow_labels(ctx, "validate_login_gate_blocks_target")
metadata = {
"validate_login_gate_blocks_target": {
"claimed_outcomes": [{"goal_value_paths": ["login_gate_blocks_target"]}],
"completion_criteria": [
{
"output_path": "login_gate_blocks_target",
"requested_output_evidence_source": "runtime_output",
},
{
"output_path": "login_gate_blocks_target",
"requested_output_evidence_source": "independent_run_evidence",
"judgment_predicate": "login_gate_blocks_target",
"judgment_polarity_when_holds": True,
},
],
}
}
ctx.workflow_verification_evidence = SimpleNamespace(code_artifact_metadata=metadata)
ctx.code_artifact_metadata = {}
ctx.request_policy = RequestPolicy(completion_criteria=build_classifier_fallback_floor([]))
ctx.composition_page_evidence = _post_run_page_evidence(
run_id="wr_requested_output",
visible_text_excerpt="",
**_gated_login_packet(),
)
with capture_logs() as logs:
verification = await _maybe_run_completion_verification(
ctx,
{
"ok": True,
"data": {
"workflow_run_id": "wr_requested_output",
"overall_status": "completed",
"executed_block_labels": ["validate_login_gate_blocks_target"],
"current_url": "https://example.test/profile",
"blocks": [
{
"label": "validate_login_gate_blocks_target",
"block_type": "CODE",
"status": "completed",
"extracted_data": {"login_gate_blocks_target": False},
}
],
},
},
time.monotonic(),
)
assert verification is not None
verdict = verification.verdicts[0]
assert verdict.criterion_id == "__copilot_authored_output__output_login_gate_blocks_target"
assert verdict.state == "unsatisfied"
assert verdict.reason_code == "evidence_contradicts"
assert verdict.grounding_mode == "judgment_boolean"
assert verdict.evidence_source == "independent_page_evidence"
assert any(
log["event"] == "copilot_judgment_evidence_verdict"
and log["predicate"] == "login_gate_blocks_target"
and log["verdict"] == "evidence_contradicts"
for log in logs
)
def test_judgment_truth_condition_refutes_inverted_login_gate_packet() -> None:
ctx = _run_ctx()
ctx.code_artifact_metadata = _metadata_for_requested_paths("login_gate_blocks_target")
criteria = [
replace(
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
requested_output_evidence_source="independent_run_evidence",
),
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
),
)
]
verdicts = grade_requested_output_criteria(
ctx,
criteria,
RunEvidenceSnapshot(
block_outputs={_POST_RUN_PAGE_OBSERVATION_LABEL: _target_reached_packet()},
block_output_sources={_POST_RUN_PAGE_OBSERVATION_LABEL: "independent_page_evidence"},
),
)
assert verdicts[0].state == "unsatisfied"
assert verdicts[0].reason_code == "evidence_contradicts"
assert verdicts[0].evidence_ref == f"block_outputs:{_POST_RUN_PAGE_OBSERVATION_LABEL}"
assert verdicts[0].evidence_source == "independent_page_evidence"
def test_judgment_truth_condition_hollow_packet_abstains_before_self_emitted_output() -> None:
ctx = _run_ctx()
ctx.code_artifact_metadata = _metadata_for_requested_paths("login_gate_blocks_target")
criteria = [
replace(
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
requested_output_evidence_source="independent_run_evidence",
),
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
),
)
]
verdicts = grade_requested_output_criteria(
ctx,
criteria,
RunEvidenceSnapshot(
block_outputs={
"extract_profile": {"login_gate_blocks_target": True},
_POST_RUN_PAGE_OBSERVATION_LABEL: {
"forms": [],
"navigation_targets": [],
"result_containers": [],
},
},
block_output_sources={
"extract_profile": "runtime_output",
_POST_RUN_PAGE_OBSERVATION_LABEL: "independent_page_evidence",
},
),
)
assert verdicts[0].state == "unsatisfied"
assert verdicts[0].reason_code == "structurally_abstained"
assert verdicts[0].evidence_ref == f"block_outputs:{_POST_RUN_PAGE_OBSERVATION_LABEL}"
assert verdicts[0].evidence_source == "independent_page_evidence"
assert verdicts[0].self_emitted_judgment_not_independent is False
def test_judgment_truth_condition_refuses_fabricated_packet_from_registered_output() -> None:
ctx = _run_ctx()
ctx.code_artifact_metadata = _metadata_for_requested_paths("login_gate_blocks_target")
criteria = [
replace(
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
requested_output_evidence_source="independent_run_evidence",
),
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
),
)
]
verdicts = grade_requested_output_criteria(
ctx,
criteria,
RunEvidenceSnapshot(
block_outputs={
"extract_profile": {"login_gate_blocks_target": True},
"fake_packet_output": _gated_login_packet(),
},
block_output_sources={
"extract_profile": "runtime_output",
"fake_packet_output": "registered_output_parameter",
},
),
)
assert verdicts[0].state == "unsatisfied"
assert verdicts[0].reason_code != "evidence_confirms"
assert verdicts[0].evidence_ref != "block_outputs:fake_packet_output"
def test_judgment_truth_condition_ignores_packet_label_and_visible_text() -> None:
ctx = _run_ctx()
ctx.code_artifact_metadata = _metadata_for_requested_paths("login_gate_blocks_target")
criteria = [
replace(
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
requested_output_evidence_source="independent_run_evidence",
),
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
),
)
]
verdicts = grade_requested_output_criteria(
ctx,
criteria,
RunEvidenceSnapshot(
block_outputs={
"login_gate_blocks_target": {
"visible_text_excerpt": "Login required. Enter password to continue.",
"evidence_text": "Login required.",
}
},
block_output_sources={"login_gate_blocks_target": "independent_page_evidence"},
),
)
assert verdicts[0].reason_code != "evidence_confirms"
def test_producer_declared_judgment_truth_condition_fallback_confirms_packet() -> None:
ctx = _run_ctx()
ctx.code_artifact_metadata = {
"extract_profile": {
"claimed_outcomes": [{"goal_value_paths": ["login_gate_blocks_target"]}],
"completion_criteria": [
{
"output_path": "output.login_gate_blocks_target",
"requested_output_evidence_source": "independent_run_evidence",
"judgment_predicate": "login_gate_blocks_target",
"judgment_polarity_when_holds": True,
}
],
}
}
criteria = [
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
requested_output_evidence_source="independent_run_evidence",
)
]
verdicts = grade_requested_output_criteria(
ctx,
criteria,
RunEvidenceSnapshot(
block_outputs={_POST_RUN_PAGE_OBSERVATION_LABEL: _login_gate_packet()},
block_output_sources={_POST_RUN_PAGE_OBSERVATION_LABEL: "independent_page_evidence"},
),
)
assert verdicts[0].state == "satisfied"
assert verdicts[0].reason_code == "evidence_confirms"
def test_criterion_judgment_truth_condition_takes_precedence_over_metadata() -> None:
ctx = _run_ctx()
ctx.code_artifact_metadata = {
"extract_profile": {
"claimed_outcomes": [{"goal_value_paths": ["login_gate_blocks_target"]}],
"completion_criteria": [
{
"output_path": "output.login_gate_blocks_target",
"requested_output_evidence_source": "independent_run_evidence",
"judgment_predicate": "login_gate_blocks_target",
"judgment_polarity_when_holds": True,
}
],
}
}
criteria = [
replace(
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_value=False,
requested_output_evidence_source="independent_run_evidence",
),
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=False
),
)
]
verdicts = grade_requested_output_criteria(
ctx,
criteria,
RunEvidenceSnapshot(
block_outputs={_POST_RUN_PAGE_OBSERVATION_LABEL: _login_gate_packet()},
block_output_sources={_POST_RUN_PAGE_OBSERVATION_LABEL: "independent_page_evidence"},
),
)
assert verdicts[0].state == "satisfied"
assert verdicts[0].reason_code == "evidence_confirms"
def _judgment_shape_abstention() -> CriterionVerdict:
return CriterionVerdict(
criterion_id="c_judgment",

View file

@ -35,6 +35,7 @@ _FLOOR_ID = "__copilot_fallback_floor__run"
class _GroundingCtx:
def __init__(self) -> None:
self.code_artifact_metadata: dict[str, object] = {}
self.workflow_verification_evidence: SimpleNamespace | None = None
self.last_workflow_yaml: str | None = None
self.workflow_yaml: str | None = None

View file

@ -2,6 +2,7 @@ from __future__ import annotations
import hashlib
import json
from dataclasses import replace
from typing import Any
import pytest
@ -17,6 +18,7 @@ from skyvern.forge.sdk.copilot.completion_criteria_store import (
from skyvern.forge.sdk.copilot.config import CopilotConfig
from skyvern.forge.sdk.copilot.request_policy import (
CompletionCriterion,
JudgmentTruthCondition,
RequestPolicy,
_apply_requested_output_completion_criteria,
_apply_validation_classification_completion_criteria,
@ -1718,6 +1720,56 @@ def test_parser_promotes_goal_judgment_boolean_shape_without_explicit_value() ->
assert _criterion_grounding_mode(criteria[0]) == "judgment_boolean"
def test_parser_mints_login_gate_judgment_truth_condition_from_classifier_fields() -> None:
criteria = _parse_completion_criteria(
[
{
"outcome": "The returned record reports whether the login gate blocks the target.",
"output_path": "output.login_gate_blocks_target",
"expected_output_shape": "goal_judgment_boolean",
"requested_output_evidence_source": "independent_run_evidence",
"judgment_predicate": "login_gate_blocks_target",
"judgment_polarity_when_holds": True,
}
]
)
assert criteria[0].judgment_truth_condition == JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
)
def test_backticked_named_output_mints_login_gate_requested_output_truth_condition() -> None:
policy = RequestPolicy(
completion_criteria=[
replace(
_criterion(
"c_login_gate",
"The returned record reports login gate blocks target.",
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
),
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
),
)
]
)
_apply_requested_output_completion_criteria(
policy,
"Create a validation workflow and return the named output `login_gate_blocks_target`.",
)
criteria = _criteria_for_path(policy, "output.login_gate_blocks_target")
assert len(criteria) == 1
assert criteria[0].expected_output_shape == "goal_judgment_boolean"
assert criteria[0].requested_output_evidence_source == "independent_run_evidence"
assert criteria[0].judgment_truth_condition == JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
)
def test_parser_dedup_keeps_runtime_string_and_judgment_boolean_on_same_path() -> None:
criteria = _parse_completion_criteria(
[
@ -1788,6 +1840,26 @@ def test_store_round_trip_preserves_boolean_expected_value_and_source() -> None:
assert _criterion_grounding_mode(round_tripped[0]) == "judgment_boolean"
def test_store_round_trip_preserves_judgment_truth_condition() -> None:
criterion = replace(
_criterion(
"c_login_gate",
"The returned record reports whether the login gate blocks the target.",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
requested_output_evidence_source="independent_run_evidence",
),
judgment_truth_condition=JudgmentTruthCondition(predicate="login_gate_blocks_target", polarity_when_holds=True),
)
round_tripped = criteria_from_json(criteria_to_json([criterion]))
assert round_tripped[0].judgment_truth_condition == JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
)
assert _criterion_reconcile_key(round_tripped[0]) == _criterion_reconcile_key(criterion)
def test_expected_false_survives_active_criteria_rendering() -> None:
criterion = _criterion(
"c_bool",

View file

@ -11,6 +11,7 @@ from skyvern.forge.sdk.copilot.request_policy import (
PROMPT_NAME,
RAW_SECRET_REFUSAL_SENTINEL,
CompletionCriterion,
JudgmentTruthCondition,
_classifier_fallback_policy,
_classify_request,
_credential_ids,
@ -91,7 +92,7 @@ class TestRequestPolicyPromptStructure:
"{outcome, contingent_on, contingent_antecedent_output_path, "
"deliverable_kind, implicit, method_mandated, level, output_path, expected_output_value, "
"expected_output_shape, requested_output_evidence_source, kind, terminal_action_family, "
"classification_output_key, expected_classification}"
"classification_output_key, expected_classification, judgment_predicate, judgment_polarity_when_holds}"
) in rendered
assert "never hide it in outcome prose" in rendered
assert (
@ -108,6 +109,17 @@ class TestRequestPolicyPromptStructure:
assert "classification_output_key=login_only and expected_classification=true" in rendered
assert 'The only supported non-null value is "registered_download"' in rendered
def test_completion_criteria_schema_exposes_judgment_truth_condition_fields(self) -> None:
rendered = _render()
assert "judgment_predicate: null unless expected_output_shape=goal_judgment_boolean" in rendered
assert "Use only the closed-vocabulary value login_gate_blocks_target" in rendered
assert "judgment_polarity_when_holds: null unless judgment_predicate is non-null" in rendered
assert "kind=outcome" in rendered
assert "output_path=output.login_gate_blocks_target" in rendered
assert "judgment_predicate=login_gate_blocks_target" in rendered
assert "judgment_polarity_when_holds=true" in rendered
assert "Do not use kind=validation_classification for this returned field" in rendered
def test_priority_requested_output_evidence_source_guidance_is_present(self) -> None:
rendered = _render()
assert (
@ -140,6 +152,28 @@ class TestRequestPolicyPromptStructure:
assert '"expected_output_value": "WTR-1842-DEMO"' in rendered
assert '"expected_output_shape": "reference_code"' in rendered
def test_active_completion_criteria_render_judgment_truth_condition(self) -> None:
active = _render_active_criteria_for_prompt(
[
CompletionCriterion(
id="c0",
outcome="the returned record reports whether the login gate blocks the target",
output_path="output.login_gate_blocks_target",
expected_output_value=True,
expected_output_shape="goal_judgment_boolean",
requested_output_evidence_source="independent_run_evidence",
judgment_truth_condition=JudgmentTruthCondition(
predicate="login_gate_blocks_target", polarity_when_holds=True
),
)
]
)
rendered = _render(active_completion_criteria=active)
assert '"judgment_predicate": "login_gate_blocks_target"' in rendered
assert '"judgment_polarity_when_holds": true' in rendered
class TestRawSecretBackstop:
@pytest.mark.parametrize(