vikarti.anatra/Skyvern
1
0
Fork 0
mirror of https://github.com/Skyvern-AI/skyvern.git synced 2026-07-09 16:09:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(SKY-11591): actuate the separated-spine output-contract violation instead of returning unchanged (#7099)

Browse source
This commit is contained in:
Andrew Neilson 2026-07-05 20:53:09 -07:00 • committed by GitHub
parent 2d94acf6a2
commit 10d2b44145
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 966 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

28
skyvern/forge/sdk/copilot/agent.py
View file

@ -72,6 +72,7 @@ from skyvern.forge.sdk.copilot.config import (
)
from skyvern.forge.sdk.copilot.context import (
COPILOT_RESPONSE_TYPES,
OUTPUT_OWNER_AMBIGUITY_REASON_CODE,
AgentResult,
CodeAuthoringRepairContext,
CopilotContext,
@ -739,6 +740,33 @@ def _code_authoring_repair_context_prompt(ctx: CopilotContext | None) -> str:
"required_code_return_paths: "
f"{_render_authoring_repair_prompt_list(repair_context.required_code_return_paths)}"
)
if repair_context.required_block_structure:
lines.append(
f"required_block_structure: {_clean_authoring_repair_prompt_atom(repair_context.required_block_structure)}"
)
if repair_context.spine_stage_count is not None:
lines.append(f"spine_stage_count: {repair_context.spine_stage_count}")
if repair_context.spine_split_blockers:
lines.append(
f"spine_split_blockers: {_render_authoring_repair_prompt_list(repair_context.spine_split_blockers)}"
)
lines.append(
"Author one browser-stage code block per scouted mutation stage and a final extraction-only code block "
"that returns the required output paths; do not collapse the browser spine into the extraction block."
)
if repair_context.reason_code == OUTPUT_OWNER_AMBIGUITY_REASON_CODE:
lines.append(
"output_owner_candidate_labels: "
f"{_render_authoring_repair_prompt_list(repair_context.output_owner_candidate_labels)}"
)
lines.append(
"required_output_owner_paths: "
f"{_render_authoring_repair_prompt_list(repair_context.required_code_return_paths)}"
)
lines.append(
"Designate exactly one code block as the sole output owner for the required paths and declare its "
"code_artifact_metadata; do not leave the requested output split across or absent from the code blocks."
)
selector_alternative_lines = _render_selector_repair_alternatives(repair_context.selector_alternatives)
if selector_alternative_lines:
lines.append("same_page_selector_alternatives:")

7
skyvern/forge/sdk/copilot/context.py
View file

@ -153,6 +153,9 @@ class LoadedResultTargetContext(BaseModel):
structure_signature: str = ""
OUTPUT_OWNER_AMBIGUITY_REASON_CODE = "output_owner_ambiguous"
class CodeAuthoringRepairContext(BaseModel):
block_label: str
reason_code: str
@ -188,6 +191,10 @@ class CodeAuthoringRepairContext(BaseModel):
page_result_summaries: list[str] = Field(default_factory=list)
page_action_summaries: list[str] = Field(default_factory=list)
page_challenge_summaries: list[str] = Field(default_factory=list)
required_block_structure: str = ""
spine_stage_count: int | None = None
spine_split_blockers: list[str] = Field(default_factory=list)
output_owner_candidate_labels: list[str] = Field(default_factory=list)
repair_instruction: str = "add workflow-input-like names to parameter_keys, or stop referencing them."

6
skyvern/forge/sdk/copilot/runtime.py
View file

@ -306,7 +306,13 @@ class AgentContext:
# Author-time output-contract cross-turn state, keyed by the contract signature; set lazily by workflow_update.
output_contract_pinned_block_label_by_signature: dict[str, str] = field(default_factory=dict)
output_contract_reject_count_by_signature: dict[str, int] = field(default_factory=dict)
output_contract_deferral_count_by_signature: dict[str, int] = field(default_factory=dict)
runtime_output_repair_attempt_by_signature: dict[str, bool] = field(default_factory=dict)
# Armed when a collapsed-spine violation cannot be split; carries split blockers and stage count to the
# next authoring prompt, keyed by a composite {signature, label, authored-YAML hash} so a new draft re-arms.
output_contract_spine_directive_blockers_by_attempt_key: dict[str, list[str]] = field(default_factory=dict)
output_contract_spine_directive_stage_count_by_attempt_key: dict[str, int] = field(default_factory=dict)
output_contract_output_owner_directive_candidates_by_signature: dict[str, list[str]] = field(default_factory=dict)
synthesized_block_offered: bool = False
synthesized_block_offered_trajectory_len: int = 0
synthesized_block_offered_goal_complete: bool = False

311
skyvern/forge/sdk/copilot/tools/workflow_update.py
View file

@ -39,6 +39,7 @@ from skyvern.forge.sdk.copilot.build_test_outcome import (
record_build_test_outcome,
recorded_outcome_from_author_time_reject,
recorded_outcome_from_authoring_repair_context,
run_backed_repair_evidence_exists,
)
from skyvern.forge.sdk.copilot.code_block_preflight import (
SANDBOX_UNRESOLVED_NAME_REASON_CODE,
@ -68,7 +69,11 @@ from skyvern.forge.sdk.copilot.composition_evidence import (
workflow_target_url,
)
from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy, normalize_block_authoring_policy
from skyvern.forge.sdk.copilot.context import CodeAuthoringRepairContext, CopilotContext
from skyvern.forge.sdk.copilot.context import (
OUTPUT_OWNER_AMBIGUITY_REASON_CODE,
CodeAuthoringRepairContext,
CopilotContext,
)
from skyvern.forge.sdk.copilot.data_write_defaults import default_data_write_continue_on_failure
from skyvern.forge.sdk.copilot.enforcement import (
MAX_CODE_AUTHORING_GUARDRAIL_REJECTS,
@ -1740,10 +1745,12 @@ def _metadata_output_repair_context(
_METADATA_CONTRACT_REQUIRED_BEFORE_RUN_REASON_CODE = "metadata_contract_required_before_run"
_SEPARATED_SPINE_SHAPE_REQUIRED_REASON_CODE = "separated_spine_shape_required"
_SEPARATED_BROWSER_SPINE_PLUS_EXTRACTION_STRUCTURE = "separated_browser_spine_plus_extraction"
_OUTPUT_CONTRACT_REJECT_REASON_CODE = "output_contract_required"
_OUTPUT_CONTRACT_REJECT_BUDGET_REASON_CODE = "output_contract_reject_budget_exhausted"
_MAX_OUTPUT_CONTRACT_STEERING_REJECTS = 2
_MAX_OUTPUT_CONTRACT_REJECTS = 4
_MAX_OUTPUT_CONTRACT_DEFERRALS = 3
@dataclass(frozen=True)
@ -2371,6 +2378,30 @@ def _evaluate_output_contract_for_code_block(
reason_code=reason_code,
),
}
if _SEPARATED_SPINE_SHAPE_REQUIRED_REASON_CODE in shape_violations and block_label:
attempt_key = _output_contract_spine_directive_attempt_key(
signature=signature, block_label=block_label, workflow_yaml=workflow_yaml
)
if attempt_key in ctx.output_contract_spine_directive_blockers_by_attempt_key:
blockers = ctx.output_contract_spine_directive_blockers_by_attempt_key[attempt_key]
stage_count = ctx.output_contract_spine_directive_stage_count_by_attempt_key.get(attempt_key)
payload["spine_structure_directive"] = {
"required_block_structure": _SEPARATED_BROWSER_SPINE_PLUS_EXTRACTION_STRUCTURE,
"spine_stage_count": stage_count,
"spine_split_blockers": blockers,
}
if repair is not None:
repair.required_block_structure = _SEPARATED_BROWSER_SPINE_PLUS_EXTRACTION_STRUCTURE
repair.spine_stage_count = stage_count
repair.spine_split_blockers = list(blockers)
if not block_label and signature in ctx.output_contract_output_owner_directive_candidates_by_signature:
repair = _output_owner_ambiguity_repair_context(
required_paths=required_paths,
owner_labels=owner_labels,
source=source,
reason_code=reason_code,
)
payload["output_owner_directive"] = {"output_owner_candidate_labels": repair.output_owner_candidate_labels}
progress_data = _code_repair_progress_data(
repair,
missing_requested_output_facts=payload["missing_requested_output_facts"],
@ -2409,8 +2440,29 @@ def _record_output_contract_reject(
payload["output_contract_reject_count"] = count
payload["output_contract_reject_budget"] = _MAX_OUTPUT_CONTRACT_REJECTS
if count >= _MAX_OUTPUT_CONTRACT_REJECTS:
payload["reason_code"] = _OUTPUT_CONTRACT_REJECT_BUDGET_REASON_CODE
payload["reject_reason"] = _OUTPUT_CONTRACT_REJECT_BUDGET_REASON_CODE
if run_backed_repair_evidence_exists(ctx):
payload["reason_code"] = _OUTPUT_CONTRACT_REJECT_BUDGET_REASON_CODE
payload["reject_reason"] = _OUTPUT_CONTRACT_REJECT_BUDGET_REASON_CODE
else:
deferral_count = _record_output_contract_deferral(ctx, evaluation.required_paths)
if deferral_count >= _MAX_OUTPUT_CONTRACT_DEFERRALS:
payload["reason_code"] = _OUTPUT_CONTRACT_REJECT_BUDGET_REASON_CODE
payload["reject_reason"] = _OUTPUT_CONTRACT_REJECT_BUDGET_REASON_CODE
LOG.info(
"copilot_output_contract_budget_deferral_cap_reached",
block_label=evaluation.block_label,
canonical_output_contract_signature=evaluation.canonical_signature,
output_contract_reject_count=count,
output_contract_deferral_count=deferral_count,
)
else:
LOG.info(
"copilot_output_contract_budget_rewrite_deferred_no_run",
block_label=evaluation.block_label,
canonical_output_contract_signature=evaluation.canonical_signature,
output_contract_reject_count=count,
output_contract_deferral_count=deferral_count,
)
structural_payload = _output_contract_author_time_structural_payload(
ctx,
evaluation.required_paths,
@ -2465,6 +2517,17 @@ def _record_output_contract_family_reject(
return count
def _record_output_contract_deferral(ctx: AgentContext, required_paths: set[str]) -> int:
if not required_paths:
return 0
signature = _stable_output_contract_key(_output_contract_scope_key(ctx), required_paths)
if not signature:
return 0
count = int(ctx.output_contract_deferral_count_by_signature.get(signature, 0) or 0) + 1
ctx.output_contract_deferral_count_by_signature[signature] = count
return count
def _output_contract_reject_result(
evaluation: _OutputContractEvaluation,
*,
@ -2768,6 +2831,167 @@ def _record_runtime_output_repair_attempt(ctx: AgentContext, attempt_key: str) -
ctx.runtime_output_repair_attempt_by_signature = attempts
def _output_contract_spine_directive_attempt_key(*, signature: str, block_label: str, workflow_yaml: str) -> str:
payload = {
"signature": signature,
"block_label": block_label,
"workflow_yaml_hash": hashlib.sha256(workflow_yaml.encode("utf-8")).hexdigest(),
}
return hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()
def _referenced_parameter_keys_in_code(code: str, parameter_keys: list[str]) -> list[str]:
protected = {key for key in parameter_keys if key}
if not protected:
return []
try:
tree = ast.parse(textwrap.dedent(code).strip() or "pass")
except SyntaxError:
return []
referenced = {node.id for node in ast.walk(tree) if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load)}
return sorted(referenced & protected)
class _SpineSplitOutcome(NamedTuple):
imposed_yaml: str | None
blockers: list[str]
stage_count: int | None
def _attempt_separated_spine_split(
*,
ctx: AgentContext,
parsed: dict[str, Any],
label: str,
target_code: str,
synthesized: SynthesizedCodeBlock,
required_paths: set[str],
) -> _SpineSplitOutcome:
code_block = next(
(block for block in _workflow_code_blocks(parsed) if str(block.get("label") or "").strip() == label),
None,
)
if code_block is None:
return _SpineSplitOutcome(None, ["target_block_not_resolved_in_parsed"], None)
reconciliation = _reconcile_synthesized_parameters(
ctx=ctx,
parsed=parsed,
code_block=code_block,
submitted_code=target_code,
synthesized_parameters=synthesized.parameters,
scout_trajectory=ctx.scout_trajectory,
)
if reconciliation.violations:
return _SpineSplitOutcome(None, ["parameter_reconciliation_failed"], None)
reconciled_submitted = _apply_parameter_reconciliation_to_code(textwrap.dedent(target_code), reconciliation)
reconciled_synthesized = _apply_parameter_reconciliation_to_code(textwrap.dedent(synthesized.code), reconciliation)
extraction_suffix = _submitted_suffix_after_synthesized_code(reconciled_submitted, reconciled_synthesized)
if not extraction_suffix:
return _SpineSplitOutcome(None, ["extraction_boundary_ambiguous"], None)
suffix_mutations, _, suffix_ambiguous = _browser_surface_for_code(extraction_suffix)
if suffix_mutations or suffix_ambiguous:
return _SpineSplitOutcome(None, ["extraction_suffix_contains_browser_actions"], None)
keyed_extraction, static_violations = _extraction_code_with_required_static_return(
extraction_suffix, required_paths=required_paths
)
if static_violations:
return _SpineSplitOutcome(None, ["static_return_envelope_unavailable"], None)
if _browser_surface_contains_full_action_spine(keyed_extraction, reconciled_synthesized):
return _SpineSplitOutcome(None, ["extraction_retains_full_spine"], None)
stage_codes = _synthesized_durable_stage_codes(synthesized, source_code=reconciled_synthesized)
if len(stage_codes) < 2:
return _SpineSplitOutcome(None, ["insufficient_durable_stages"], len(stage_codes))
split_violations = _split_selected_output_owner_into_browser_stages(
parsed=parsed,
code_block=code_block,
synthesized=synthesized,
synthesized_code=reconciled_synthesized,
extraction_code=keyed_extraction,
parameter_keys=reconciliation.parameter_keys,
)
if split_violations:
return _SpineSplitOutcome(None, split_violations, len(stage_codes))
referenced_keys = _referenced_parameter_keys_in_code(keyed_extraction, reconciliation.parameter_keys)
if referenced_keys:
code_block["parameter_keys"] = referenced_keys
return _SpineSplitOutcome(yaml.safe_dump(parsed, sort_keys=False), [], len(stage_codes))
def _arm_output_contract_spine_directive(
ctx: AgentContext,
*,
attempt_key: str,
blockers: list[str],
stage_count: int | None,
block_label: str,
signature: str,
) -> None:
already_armed = attempt_key in ctx.output_contract_spine_directive_blockers_by_attempt_key
ctx.output_contract_spine_directive_blockers_by_attempt_key[attempt_key] = list(blockers)
if stage_count is not None:
ctx.output_contract_spine_directive_stage_count_by_attempt_key[attempt_key] = stage_count
if already_armed:
return
LOG.info(
"copilot_output_contract_spine_structure_directive_emitted",
block_label=block_label,
canonical_output_contract_signature=signature,
spine_split_blockers=blockers,
spine_stage_count=stage_count,
)
def _arm_output_contract_output_owner_directive(
ctx: AgentContext,
*,
signature: str,
owner_labels: list[str],
) -> None:
already_armed = signature in ctx.output_contract_output_owner_directive_candidates_by_signature
ctx.output_contract_output_owner_directive_candidates_by_signature[signature] = list(owner_labels)
if already_armed:
return
LOG.info(
"copilot_output_contract_output_owner_directive_emitted",
canonical_output_contract_signature=signature,
output_owner_candidate_labels=owner_labels,
)
def _output_owner_ambiguity_repair_context(
*,
required_paths: set[str],
owner_labels: list[str],
source: str,
reason_code: str,
) -> CodeAuthoringRepairContext:
paths = sorted(dict.fromkeys(str(path).strip() for path in required_paths if str(path).strip()))
candidates = sorted(dict.fromkeys(str(label).strip() for label in owner_labels if str(label).strip()))
path_text = ", ".join(paths)
return CodeAuthoringRepairContext(
block_label="",
reason_code=OUTPUT_OWNER_AMBIGUITY_REASON_CODE,
runtime_failure_class=reason_code,
required_goal_value_paths=paths,
required_extraction_schema_paths=paths,
required_code_return_paths=paths,
metadata_contract_source=source,
metadata_contract_reason_code=reason_code,
output_owner_candidate_labels=candidates,
repair_instruction=(
"Designate exactly one code block as the sole output owner for required paths "
f"{path_text}; declare code_artifact_metadata on that single block and remove competing output owners."
),
)
def _impose_output_contract_envelope_after_steering(
ctx: AgentContext,
workflow_yaml: str,
@ -2795,11 +3019,7 @@ def _impose_output_contract_envelope_after_steering(
required_paths,
)
if not label or owner_labels != [label]:
LOG.info(
"copilot_output_contract_envelope_scaffold_bailed",
reason="ambiguous_or_invalid_output_owner",
canonical_output_contract_signature=signature,
)
_arm_output_contract_output_owner_directive(ctx, signature=signature, owner_labels=owner_labels)
return workflow_yaml, raw_code_artifact_metadata, False
parsed = parse_workflow_yaml(workflow_yaml)
if not isinstance(parsed, dict):
@ -2816,11 +3036,41 @@ def _impose_output_contract_envelope_after_steering(
reached_download_target=getattr(ctx, "reached_download_target", None),
)
if synthesized is not None and _browser_surface_contains_full_action_spine(target_code, synthesized.code):
LOG.info(
"copilot_output_contract_envelope_scaffold_bailed",
reason="separated_spine_shape_violation",
split = _attempt_separated_spine_split(
ctx=ctx,
parsed=parse_workflow_yaml(workflow_yaml),
label=label,
target_code=target_code,
synthesized=synthesized,
required_paths=required_paths,
)
if split.imposed_yaml is not None:
scaffolded_metadata = _apply_metadata_contract_scaffold(
ctx,
split.imposed_yaml,
raw_code_artifact_metadata,
required_paths=required_paths,
source=source,
reason_code=reason_code,
)
_record_runtime_output_repair_attempt(ctx, runtime_attempt_key)
LOG.info(
"copilot_output_contract_spine_split_imposed",
block_label=label,
stage_count=split.stage_count,
canonical_output_contract_signature=signature,
)
return split.imposed_yaml, scaffolded_metadata, True
attempt_key = _output_contract_spine_directive_attempt_key(
signature=signature, block_label=label, workflow_yaml=workflow_yaml
)
_arm_output_contract_spine_directive(
ctx,
attempt_key=attempt_key,
blockers=split.blockers,
stage_count=split.stage_count,
block_label=label,
canonical_output_contract_signature=signature,
signature=signature,
)
return workflow_yaml, raw_code_artifact_metadata, False
keyed_code, violations = _extraction_code_with_required_static_return(target_code, required_paths=required_paths)
@ -2911,6 +3161,11 @@ def _metadata_contract_run_preflight_reject(
evaluation,
summary="Submitted workflow does not satisfy the requested output contract before run.",
)
if evaluation.repair_context is not None and (
evaluation.repair_context.required_block_structure
or evaluation.repair_context.reason_code == OUTPUT_OWNER_AMBIGUITY_REASON_CODE
):
ctx.last_code_authoring_repair_context = evaluation.repair_context
payload = dict(payload)
payload["output_contract_reason_code"] = payload.get("reason_code")
payload["reason_code"] = _METADATA_CONTRACT_REQUIRED_BEFORE_RUN_REASON_CODE
@ -4503,14 +4758,19 @@ def _whole_trajectory_browser_surface_violations(
synthesized_code: str,
) -> _BrowserSurfaceValidation:
violations: list[str] = []
scouted_mutations, _, _ = _browser_surface_for_code(synthesized_code)
scouted_signatures = set(scouted_mutations)
for block in code_blocks:
if block is selected_code_block:
continue
label = _code_block_label(block)
block_code = str(block.get("code") or "")
block_mutations, _, block_ambiguous = _browser_surface_for_code(block_code)
if block_mutations:
action_text = ", ".join(f"{mutation.receiver}.{mutation.method}" for mutation in sorted(block_mutations))
unscouted_mutations = [mutation for mutation in block_mutations if mutation not in scouted_signatures]
if unscouted_mutations:
action_text = ", ".join(
f"{mutation.receiver}.{mutation.method}" for mutation in sorted(unscouted_mutations)
)
violations.append(
f"Unable to impose synthesized code block: `{label}` contains unscouted browser action(s): {action_text}."
)
@ -4523,6 +4783,26 @@ def _whole_trajectory_browser_surface_violations(
return _BrowserSurfaceValidation(violations)
def _separated_spine_already_imposed(
code_blocks: list[dict[str, Any]],
selected_code_block: dict[str, Any],
synthesized_code: str,
) -> bool:
scouted_mutations, _, _ = _browser_surface_for_code(synthesized_code)
if not scouted_mutations:
return False
selected_mutations, _, _ = _browser_surface_for_code(str(selected_code_block.get("code") or ""))
if selected_mutations:
return False
sibling_signatures: set[_BrowserMutationSignature] = set()
for block in code_blocks:
if block is selected_code_block:
continue
block_mutations, _, _ = _browser_surface_for_code(str(block.get("code") or ""))
sibling_signatures.update(block_mutations)
return sibling_signatures == set(scouted_mutations)
def _browser_surface_contains_full_action_spine(submitted_code: str, synthesized_code: str) -> bool:
synthesized_mutations, _, _ = _browser_surface_for_code(synthesized_code)
submitted_mutations, _, submitted_ambiguous = _browser_surface_for_code(submitted_code)
@ -5785,6 +6065,9 @@ def _maybe_impose_synthesized_code_block(workflow_yaml: str, ctx: AgentContext)
violations=["Unable to impose synthesized code block: scout trajectory produced no runnable code."],
)
if _separated_spine_already_imposed(code_blocks, code_block, synthesized.code):
return _SynthesizedCodeImpositionResult(workflow_yaml=workflow_yaml)
diagnostics = synthesized.diagnostics
violations: list[str] = []
if diagnostics.truncated:

630
tests/unit/test_copilot_code_block_persist_seam.py
View file

@ -34,7 +34,11 @@ from skyvern.forge.sdk.copilot.code_block_preflight import (
strip_redundant_sandbox_imports,
)
from skyvern.forge.sdk.copilot.code_block_security import CodeBlockSecurityError
from skyvern.forge.sdk.copilot.code_block_synthesis import _get_by_role_expr, _get_by_role_expr_strict
from skyvern.forge.sdk.copilot.code_block_synthesis import (
SynthesizedCodeBlock,
_get_by_role_expr,
_get_by_role_expr_strict,
)
from skyvern.forge.sdk.copilot.completion_verification import CompletionVerificationResult, CriterionVerdict
from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy
from skyvern.forge.sdk.copilot.context import CodeAuthoringRepairContext, CopilotContext
@ -58,6 +62,7 @@ from skyvern.forge.sdk.copilot.tools import scouting as scouting_module
from skyvern.forge.sdk.copilot.tools import workflow_update as workflow_update_module
from skyvern.forge.sdk.copilot.tools.workflow_update import (
_code_safety_reject_payload,
_OutputContractEvaluation,
_strip_redundant_sandbox_imports_in_yaml,
)
from skyvern.forge.sdk.copilot.turn_halt import CopilotTurnHalt, TurnHaltKind, _kind_for_blocker_signal
@ -3287,7 +3292,7 @@ class TestCodeRepairProgressClassification:
assert result is not None
assert result["ok"] is False
assert result["data"]["reason_code"] == "output_contract_reject_budget_exhausted"
assert result["data"]["reason_code"] == "output_contract_required"
assert result["data"]["output_contract_reject_count"] == 4
assert result["data"]["canonical_required_child_paths"] == ["output.record_id"]
@ -5669,6 +5674,606 @@ class TestCodeRepairProgressClassification:
assert result["data"].get("surface_kind") is None
_SPINE_SYNTH_CODE = 'await page.locator("#stage-a").click()\nawait page.locator("#stage-b").click()'
def _fake_spine_synthesized(
*,
parameters: list[dict[str, str]] | None = None,
steps: list[dict[str, int]] | None = None,
code: str | None = None,
) -> SynthesizedCodeBlock:
return SynthesizedCodeBlock(
code=code if code is not None else _SPINE_SYNTH_CODE,
parameters=parameters if parameters is not None else [],
steps=steps if steps is not None else [{"line_start": 1, "line_end": 1}, {"line_start": 2, "line_end": 2}],
)
def _spine_actuation_ctx() -> CopilotContext:
ctx = _code_only_ctx()
ctx.turn_id = "t-spine"
ctx.scout_trajectory = [
{"tool_name": "click", "selector": "#stage-a", "source_url": "https://example.com/records"},
{"tool_name": "click", "selector": "#stage-b", "source_url": "https://example.com/records"},
]
ctx.request_policy = RequestPolicy(
completion_criteria=[
SimpleNamespace(
id="requested_value",
output_path="output.record_id",
level="run",
method_mandated=False,
kind="outcome",
)
]
)
signature = workflow_update_module._stable_output_contract_key("turn:t-spine", {"output.record_id"})
ctx.output_contract_reject_count_by_signature = {
signature: workflow_update_module._MAX_OUTPUT_CONTRACT_STEERING_REJECTS
}
return ctx
def _collapsed_spine_yaml(code_body: str) -> str:
indented = textwrap.indent(textwrap.dedent(code_body).strip(), " " * 10)
return _yaml(
"title: Entry lookup\n"
"workflow_definition:\n"
" blocks:\n"
" - block_type: code\n"
" label: extract_record\n"
" code: |\n"
f"{indented}\n"
)
class TestSeparatedSpineViolationActuation:
def test_branch_a_split_replaces_collapsed_owner_with_stages_plus_extraction(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
workflow_yaml = _collapsed_spine_yaml(
_SPINE_SYNTH_CODE
+ '\nvalue = await page.locator("#result").inner_text()\nreturn {"output": {"record_id": value}}'
)
new_yaml, _metadata, applied = workflow_update_module._impose_output_contract_envelope_after_steering(
ctx, workflow_yaml, []
)
assert applied is True
blocks = workflow_blocks(parse_workflow_yaml(new_yaml))
assert [block.get("label") for block in blocks] == [
"extract_record_browser_stage_1",
"extract_record_browser_stage_2",
"extract_record",
]
retained_code = str(blocks[-1].get("code") or "")
assert "inner_text" in retained_code
assert ".click()" not in retained_code
def test_branch_a_result_is_idempotent_no_resplit(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
workflow_yaml = _collapsed_spine_yaml(
_SPINE_SYNTH_CODE
+ '\nvalue = await page.locator("#result").inner_text()\nreturn {"output": {"record_id": value}}'
)
split_yaml, _meta, _applied = workflow_update_module._impose_output_contract_envelope_after_steering(
ctx, workflow_yaml, []
)
again_yaml, _meta2, _applied2 = workflow_update_module._impose_output_contract_envelope_after_steering(
ctx, split_yaml, []
)
assert [block.get("label") for block in workflow_blocks(parse_workflow_yaml(again_yaml))] == [
"extract_record_browser_stage_1",
"extract_record_browser_stage_2",
"extract_record",
]
assert not ctx.output_contract_spine_directive_blockers_by_attempt_key
def test_branch_b_arms_directive_and_returns_unchanged_yaml(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
workflow_yaml = _collapsed_spine_yaml(
"_setup = 1\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
)
new_yaml, _metadata, applied = workflow_update_module._impose_output_contract_envelope_after_steering(
ctx, workflow_yaml, []
)
assert applied is False
assert new_yaml == workflow_yaml
signature = workflow_update_module._stable_output_contract_key("turn:t-spine", {"output.record_id"})
attempt_key = workflow_update_module._output_contract_spine_directive_attempt_key(
signature=signature, block_label="extract_record", workflow_yaml=workflow_yaml
)
assert ctx.output_contract_spine_directive_blockers_by_attempt_key[attempt_key] == [
"extraction_boundary_ambiguous"
]
@pytest.mark.parametrize(
"code_body, synth_kwargs, expected_blocker",
[
(
"_setup = 1\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}',
{},
"extraction_boundary_ambiguous",
),
(
_SPINE_SYNTH_CODE + '\nawait page.locator("#extra").click()\nreturn {"output": {"record_id": "X"}}',
{},
"extraction_suffix_contains_browser_actions",
),
(
_SPINE_SYNTH_CODE + '\nvalue = await page.locator("#result").inner_text()',
{},
"static_return_envelope_unavailable",
),
(
_SPINE_SYNTH_CODE
+ '\nvalue = await page.locator("#result").inner_text()\nreturn {"output": {"record_id": value}}',
{"steps": [{"line_start": 1, "line_end": 2}]},
"insufficient_durable_stages",
),
(
_SPINE_SYNTH_CODE
+ '\nvalue = await page.locator("#result").inner_text()\nreturn {"output": {"record_id": value}}',
{"parameters": [{"key": "alpha"}, {"key": "alpha"}]},
"parameter_reconciliation_failed",
),
],
)
def test_branch_b_precondition_failures_arm_matching_blocker(
self,
monkeypatch: pytest.MonkeyPatch,
code_body: str,
synth_kwargs: dict[str, object],
expected_blocker: str,
) -> None:
monkeypatch.setattr(
workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized(**synth_kwargs)
)
ctx = _spine_actuation_ctx()
workflow_yaml = _collapsed_spine_yaml(code_body)
_new_yaml, _metadata, applied = workflow_update_module._impose_output_contract_envelope_after_steering(
ctx, workflow_yaml, []
)
assert applied is False
armed = list(ctx.output_contract_spine_directive_blockers_by_attempt_key.values())
assert armed == [[expected_blocker]]
def test_evaluation_enriches_repair_context_and_payload_after_directive_armed(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
workflow_yaml = _collapsed_spine_yaml(
"_setup = 1\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
)
workflow_update_module._impose_output_contract_envelope_after_steering(ctx, workflow_yaml, [])
evaluation = workflow_update_module._evaluate_output_contract_for_code_block(ctx, workflow_yaml, [])
assert evaluation is not None
assert evaluation.shape_violations == ["separated_spine_shape_required"]
assert evaluation.repair_context is not None
assert evaluation.repair_context.required_block_structure == "separated_browser_spine_plus_extraction"
assert evaluation.repair_context.spine_split_blockers == ["extraction_boundary_ambiguous"]
assert evaluation.payload["spine_structure_directive"]["spine_split_blockers"] == [
"extraction_boundary_ambiguous"
]
def test_directive_renders_into_next_authoring_prompt(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
workflow_yaml = _collapsed_spine_yaml(
"_setup = 1\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
)
workflow_update_module._impose_output_contract_envelope_after_steering(ctx, workflow_yaml, [])
evaluation = workflow_update_module._evaluate_output_contract_for_code_block(ctx, workflow_yaml, [])
assert evaluation is not None
ctx.last_code_authoring_repair_context = evaluation.repair_context
rendered = agent_module._code_authoring_repair_context_prompt(ctx)
assert "required_block_structure: separated_browser_spine_plus_extraction" in rendered
assert "spine_split_blockers:" in rendered
assert "one browser-stage code block per scouted mutation stage" in rendered
def test_preflight_reject_persists_directive_repair_context(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
workflow_yaml = _collapsed_spine_yaml(
"_setup = 1\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
)
result = workflow_update_module._metadata_contract_run_preflight_reject(ctx, workflow_yaml, [])
assert result is not None
assert result["ok"] is False
assert ctx.last_code_authoring_repair_context is not None
assert (
ctx.last_code_authoring_repair_context.required_block_structure == "separated_browser_spine_plus_extraction"
)
def test_composite_key_rearms_on_changed_yaml(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
first_yaml = _collapsed_spine_yaml(
"_setup = 1\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
)
second_yaml = _collapsed_spine_yaml(
"_setup = 2\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
)
workflow_update_module._impose_output_contract_envelope_after_steering(ctx, first_yaml, [])
workflow_update_module._impose_output_contract_envelope_after_steering(ctx, second_yaml, [])
assert len(ctx.output_contract_spine_directive_blockers_by_attempt_key) == 2
def _dual_output_owner_yaml() -> str:
return _yaml(
"title: Entry lookup\n"
"workflow_definition:\n"
" blocks:\n"
" - block_type: code\n"
" label: extract_a\n"
" code: |\n"
' return {"output": {"record_id": "A"}}\n'
" - block_type: code\n"
" label: extract_b\n"
" code: |\n"
' return {"output": {"record_id": "B"}}\n'
)
class TestAmbiguousOutputOwnerActuation:
def _signature(self) -> str:
return workflow_update_module._stable_output_contract_key("turn:t-spine", {"output.record_id"})
def test_ambiguous_owner_arms_directive_instead_of_bailing(self) -> None:
ctx = _spine_actuation_ctx()
workflow_yaml = _dual_output_owner_yaml()
new_yaml, _metadata, applied = workflow_update_module._impose_output_contract_envelope_after_steering(
ctx, workflow_yaml, []
)
assert applied is False
assert new_yaml == workflow_yaml
assert ctx.output_contract_output_owner_directive_candidates_by_signature[self._signature()] == [
"extract_a",
"extract_b",
]
def test_evaluation_enriches_owner_ambiguity_repair_context_after_directive_armed(self) -> None:
ctx = _spine_actuation_ctx()
workflow_yaml = _dual_output_owner_yaml()
workflow_update_module._impose_output_contract_envelope_after_steering(ctx, workflow_yaml, [])
evaluation = workflow_update_module._evaluate_output_contract_for_code_block(ctx, workflow_yaml, [])
assert evaluation is not None
assert "ambiguous_output_owner" in evaluation.shape_violations
assert evaluation.repair_context is not None
assert evaluation.repair_context.reason_code == "output_owner_ambiguous"
assert evaluation.repair_context.output_owner_candidate_labels == ["extract_a", "extract_b"]
assert evaluation.payload["output_owner_directive"]["output_owner_candidate_labels"] == [
"extract_a",
"extract_b",
]
def test_owner_directive_renders_into_next_authoring_prompt(self) -> None:
ctx = _spine_actuation_ctx()
workflow_yaml = _dual_output_owner_yaml()
workflow_update_module._impose_output_contract_envelope_after_steering(ctx, workflow_yaml, [])
evaluation = workflow_update_module._evaluate_output_contract_for_code_block(ctx, workflow_yaml, [])
assert evaluation is not None
ctx.last_code_authoring_repair_context = evaluation.repair_context
rendered = agent_module._code_authoring_repair_context_prompt(ctx)
assert "output_owner_candidate_labels: extract_a, extract_b" in rendered
assert "sole output owner" in rendered
def test_preflight_reject_persists_owner_ambiguity_repair_context(self) -> None:
ctx = _spine_actuation_ctx()
workflow_yaml = _dual_output_owner_yaml()
result = workflow_update_module._metadata_contract_run_preflight_reject(ctx, workflow_yaml, [])
assert result is not None
assert result["ok"] is False
assert ctx.last_code_authoring_repair_context is not None
assert ctx.last_code_authoring_repair_context.reason_code == "output_owner_ambiguous"
def test_owner_directive_emits_fingerprint_once_per_signature(self) -> None:
ctx = _spine_actuation_ctx()
workflow_yaml = _dual_output_owner_yaml()
workflow_update_module._impose_output_contract_envelope_after_steering(ctx, workflow_yaml, [])
workflow_update_module._impose_output_contract_envelope_after_steering(ctx, workflow_yaml, [])
assert list(ctx.output_contract_output_owner_directive_candidates_by_signature.keys()) == [self._signature()]
def _already_split_spine_yaml(extra_sibling_code: str | None = None, *, base: str | None = None) -> str:
base = base or workflow_update_module._SYNTHESIZED_BLOCK_LABEL
extra_block = ""
if extra_sibling_code is not None:
extra_block = (
f" - block_type: code\n label: {base}_browser_stage_extra\n code: |\n"
+ textwrap.indent(textwrap.dedent(extra_sibling_code).strip(), " " * 6)
+ "\n"
)
return _yaml(
"title: Entry lookup\n"
"workflow_definition:\n"
" blocks:\n"
f" - block_type: code\n label: {base}_browser_stage_1\n code: |\n"
' await page.locator("#stage-a").click()\n'
f" - block_type: code\n label: {base}_browser_stage_2\n code: |\n"
' await page.locator("#stage-b").click()\n'
f"{extra_block}"
f" - block_type: code\n label: {base}\n code: |\n"
' value = await page.locator("#result").inner_text()\n'
' return {"output": {"record_id": value}}\n'
)
def _imposition_split_ctx() -> CopilotContext:
ctx = _spine_actuation_ctx()
ctx.impose_synthesized_code_block = True
ctx.raw_code_artifact_metadata = [
{
"block_label": workflow_update_module._SYNTHESIZED_BLOCK_LABEL,
"claimed_outcomes": [{"goal_value_paths": ["output.record_id"], "extraction_schema": '{"type":"object"}'}],
}
]
return ctx
class TestSeparatedSpineImpositionRunEligibility:
def test_imposition_accepts_scouted_browser_stage_siblings(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _imposition_split_ctx()
result = workflow_update_module._maybe_impose_synthesized_code_block(_already_split_spine_yaml(), ctx)
assert result.violations == []
def test_imposition_still_flags_unscouted_sibling_mutation(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _imposition_split_ctx()
result = workflow_update_module._maybe_impose_synthesized_code_block(
_already_split_spine_yaml(extra_sibling_code='await page.locator("#hallucinated").click()'), ctx
)
assert any("unscouted browser action" in violation for violation in result.violations)
assert any("#hallucinated" in violation for violation in result.violations)
assert not any("#stage-a" in violation for violation in result.violations)
def test_whole_trajectory_validation_exempts_spine_covered_sibling_mutations(self) -> None:
parsed = parse_workflow_yaml(_already_split_spine_yaml())
blocks = workflow_update_module._workflow_code_blocks(parsed)
extraction_block = next(
block for block in blocks if block.get("label") == workflow_update_module._SYNTHESIZED_BLOCK_LABEL
)
validation = workflow_update_module._whole_trajectory_browser_surface_violations(
code_blocks=blocks,
selected_code_block=extraction_block,
submitted_selected_code=str(extraction_block.get("code") or ""),
synthesized_code=_SPINE_SYNTH_CODE,
)
assert validation.violations == []
def test_split_imposed_yaml_clears_output_contract_run_gate(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
workflow_yaml = _collapsed_spine_yaml(
_SPINE_SYNTH_CODE
+ '\nvalue = await page.locator("#result").inner_text()\nreturn {"output": {"record_id": value}}'
)
split_yaml, split_metadata, applied = workflow_update_module._impose_output_contract_envelope_after_steering(
ctx, workflow_yaml, []
)
assert applied is True
evaluation = workflow_update_module._evaluate_output_contract_for_code_block(
ctx, split_yaml, split_metadata, allow_static_return_advisory=True
)
assert evaluation is not None
assert workflow_update_module._SEPARATED_SPINE_SHAPE_REQUIRED_REASON_CODE not in evaluation.shape_violations
assert evaluation.can_attempt_run or not evaluation.has_deficiencies
assert workflow_update_module._metadata_contract_run_preflight_reject(ctx, split_yaml, split_metadata) is None
def test_directive_satisfying_reauthor_passes_preflight(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
collapsed_yaml = _collapsed_spine_yaml(
"_setup = 1\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
)
_yaml_out, _meta, armed = workflow_update_module._impose_output_contract_envelope_after_steering(
ctx, collapsed_yaml, []
)
assert armed is False
assert ctx.output_contract_spine_directive_blockers_by_attempt_key
schema = (
'{"type":"object","properties":{"output":{"type":"object","properties":{"record_id":{"type":"string"}}}}}'
)
reauthored_yaml = _already_split_spine_yaml(base="extract_record")
metadata = [
{
"block_label": "extract_record",
"claimed_outcomes": [{"goal_value_paths": ["output.record_id"], "extraction_schema": schema}],
"terminal_verifier_expectations": [
{"goal_value_paths": ["output.record_id"], "extraction_schema": schema}
],
}
]
assert workflow_update_module._metadata_contract_run_preflight_reject(ctx, reauthored_yaml, metadata) is None
def test_armed_attempt_key_survives_scaffold_metadata_owner_shift(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
workflow_yaml = _collapsed_spine_yaml(
"_setup = 1\n" + _SPINE_SYNTH_CODE + '\nreturn {"output": {"record_id": "X"}}'
)
workflow_update_module._impose_output_contract_envelope_after_steering(ctx, workflow_yaml, [])
armed_keys = set(ctx.output_contract_spine_directive_blockers_by_attempt_key)
assert len(armed_keys) == 1
scaffolded_metadata, _applied = workflow_update_module._scaffold_metadata_contract_for_update(
ctx, workflow_yaml, []
)
required_paths, source, reason_code = workflow_update_module._output_contract_required_paths_source(ctx)
read_label, _owner_labels = workflow_update_module._target_output_contract_block_label(
ctx, workflow_yaml, scaffolded_metadata, required_paths
)
read_signature = workflow_update_module._output_contract_signature(
ctx=ctx,
workflow_yaml=workflow_yaml,
source=source,
reason_code=reason_code,
required_paths=required_paths,
)
read_key = workflow_update_module._output_contract_spine_directive_attempt_key(
signature=read_signature, block_label=read_label, workflow_yaml=workflow_yaml
)
assert read_key in armed_keys
def _budget_evaluation(ctx: CopilotContext) -> _OutputContractEvaluation:
workflow_yaml = _collapsed_spine_yaml(
_SPINE_SYNTH_CODE
+ '\nvalue = await page.locator("#result").inner_text()\nreturn {"output": {"record_id": value}}'
)
evaluation = workflow_update_module._evaluate_output_contract_for_code_block(ctx, workflow_yaml, [])
assert evaluation is not None
return evaluation
class TestOutputContractBudgetRunGuard:
def _prime(self, monkeypatch: pytest.MonkeyPatch) -> tuple[CopilotContext, _OutputContractEvaluation]:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
signature = workflow_update_module._stable_output_contract_key("turn:t-spine", {"output.record_id"})
ctx.output_contract_reject_count_by_signature = {
signature: workflow_update_module._MAX_OUTPUT_CONTRACT_REJECTS - 1
}
evaluation = _budget_evaluation(ctx)
return ctx, evaluation
def test_budget_rewrite_deferred_without_run_evidence(self, monkeypatch: pytest.MonkeyPatch) -> None:
ctx, evaluation = self._prime(monkeypatch)
payload = workflow_update_module._record_output_contract_reject(ctx, evaluation, summary="x")
assert payload["output_contract_reject_count"] >= workflow_update_module._MAX_OUTPUT_CONTRACT_REJECTS
assert payload["reason_code"] == "output_contract_required"
def test_budget_rewrite_deferred_for_author_time_reject_outcome(self, monkeypatch: pytest.MonkeyPatch) -> None:
ctx, evaluation = self._prime(monkeypatch)
ctx.latest_recorded_build_test_outcome = RecordedBuildTestOutcome(
phase="author_time_reject", verdict="authoring_rejected", reason_code="metadata_reject"
)
payload = workflow_update_module._record_output_contract_reject(ctx, evaluation, summary="x")
assert payload["reason_code"] == "output_contract_required"
def test_budget_rewrite_fires_with_persisted_run_outcome(self, monkeypatch: pytest.MonkeyPatch) -> None:
ctx, evaluation = self._prime(monkeypatch)
ctx.latest_recorded_build_test_outcome = RecordedBuildTestOutcome(
phase="persisted_block_run",
verdict="repairable_failure",
reason_code="failed_run",
workflow_run_id="wr_1",
)
payload = workflow_update_module._record_output_contract_reject(ctx, evaluation, summary="x")
assert payload["reason_code"] == "output_contract_reject_budget_exhausted"
def _prime_at_reject_count(
self, monkeypatch: pytest.MonkeyPatch, *, reject_count: int, deferral_count: int
) -> tuple[CopilotContext, _OutputContractEvaluation]:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
signature = workflow_update_module._stable_output_contract_key("turn:t-spine", {"output.record_id"})
ctx.output_contract_reject_count_by_signature = {signature: reject_count}
ctx.output_contract_deferral_count_by_signature = {signature: deferral_count}
evaluation = _budget_evaluation(ctx)
return ctx, evaluation
def test_zero_run_reject_loop_terminalizes_at_deferral_cap(self, monkeypatch: pytest.MonkeyPatch) -> None:
ctx, evaluation = self._prime_at_reject_count(
monkeypatch,
reject_count=workflow_update_module._MAX_OUTPUT_CONTRACT_REJECTS,
deferral_count=workflow_update_module._MAX_OUTPUT_CONTRACT_DEFERRALS - 1,
)
payload = workflow_update_module._record_output_contract_reject(ctx, evaluation, summary="x")
assert payload["reason_code"] == "output_contract_reject_budget_exhausted"
def test_zero_run_reject_below_deferral_cap_still_defers(self, monkeypatch: pytest.MonkeyPatch) -> None:
ctx, evaluation = self._prime_at_reject_count(
monkeypatch,
reject_count=workflow_update_module._MAX_OUTPUT_CONTRACT_REJECTS,
deferral_count=0,
)
payload = workflow_update_module._record_output_contract_reject(ctx, evaluation, summary="x")
assert payload["reason_code"] == "output_contract_required"
def test_deferral_cap_matches_max_rejects_plus_two(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(workflow_update_module, "synthesize_code_block", lambda *a, **k: _fake_spine_synthesized())
ctx = _spine_actuation_ctx()
signature = workflow_update_module._stable_output_contract_key("turn:t-spine", {"output.record_id"})
ctx.output_contract_reject_count_by_signature = {
signature: workflow_update_module._MAX_OUTPUT_CONTRACT_REJECTS - 1
}
ctx.output_contract_deferral_count_by_signature = {}
reasons: list[str] = []
for _ in range(6):
evaluation = _budget_evaluation(ctx)
payload = workflow_update_module._record_output_contract_reject(ctx, evaluation, summary="x")
reasons.append(str(payload["reason_code"]))
if payload["reason_code"] == "output_contract_reject_budget_exhausted":
break
assert reasons[-1] == "output_contract_reject_budget_exhausted"
assert reasons[:-1] == ["output_contract_required"] * (len(reasons) - 1)
assert (
ctx.output_contract_reject_count_by_signature[signature]
== workflow_update_module._MAX_OUTPUT_CONTRACT_REJECTS + 2
)
class TestCodeBlockParameterPersistSeam:
@pytest.mark.asyncio
async def test_undeclared_parameter_key_rejects_before_persist(self, monkeypatch: pytest.MonkeyPatch) -> None:
@ -8382,6 +8987,27 @@ class TestCodeAuthoringGuardrailChurnBackstop:
assert accepted["ok"] is True
assert ctx.code_authoring_guardrail_reject_count == 0
@pytest.mark.asyncio
async def test_accepted_persist_at_churn_ceiling_resets_counter_and_clears_held_signals(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
ctx = _code_only_ctx()
for _ in range(MAX_CODE_AUTHORING_GUARDRAIL_REJECTS):
await _update_workflow({"workflow_yaml": _distinct_guardrail_yaml(0)}, ctx)
assert ctx.code_authoring_guardrail_reject_count == MAX_CODE_AUTHORING_GUARDRAIL_REJECTS
held = ctx.blocker_signal
assert isinstance(held, CopilotToolBlockerSignal)
assert held.internal_reason_code == "code_authoring_guardrail_churn"
assert ctx.latest_tool_blocker_signal is held
_stub_successful_update(monkeypatch)
accepted = await _update_workflow({"workflow_yaml": _SAFE_CODE_YAML}, ctx)
assert accepted["ok"] is True
assert ctx.code_authoring_guardrail_reject_count == 0
assert ctx.blocker_signal is None
assert ctx.latest_tool_blocker_signal is None
@pytest.mark.asyncio
async def test_counter_climbs_through_credential_scout_branch(self) -> None:
ctx = _code_only_ctx()