diff --git a/docs/decisions/0004-recipe-redelivery.md b/docs/decisions/0004-recipe-redelivery.md index 6a29095df0..de686f77a1 100644 --- a/docs/decisions/0004-recipe-redelivery.md +++ b/docs/decisions/0004-recipe-redelivery.md @@ -63,6 +63,16 @@ future bodies or an ordered manual `recipe_pull` closure. No raw YAML, LRU looku skip carrier, or separately persisted projection participates in recovery. Carriers are attached by the shared response wrapper from each tool's registry outcome declaration. +Success reachability comes from the registry's `RecipeDeliveryOutcomeDef`. An arm that +cannot be proved failure-only counts as success-reachable. Carrier selection therefore +conservatively supplies possible continuations; it does not evaluate the actual result +branch. Delivering or pulling step text does not execute that step. The orchestrator +still follows the recipe's routing and completed-operation rules. + +A manual pull closure includes every segment it reaches, the same unit as inline +bodies. Outgoing traversal stops at non-root automatic checkpoints, whose carriers +deliver their later segments; automatic checkpoint roots remain walkable. + Acknowledged `run_skill` receipts are replayable only for the same kitchen, request session, and receipt identity. Their tracker effect remains one-shot and cached, so a response retry can reproduce the segment without crediting progress twice. A post-effect diff --git a/src/autoskillit/core/_tool_registry_builders.py b/src/autoskillit/core/_tool_registry_builders.py index 51cb3cd3f9..aa765a8fde 100644 --- a/src/autoskillit/core/_tool_registry_builders.py +++ b/src/autoskillit/core/_tool_registry_builders.py @@ -144,17 +144,20 @@ def _initialization_operation(name: str) -> ToolInitializationOperation: { "check_repo_merge_state": RecipeDeliveryOutcomeDef(), "enqueue_pr": RecipeDeliveryOutcomeDef(success_field="success"), - "wait_for_merge_queue": RecipeDeliveryOutcomeDef(success_field="success"), + "wait_for_merge_queue": RecipeDeliveryOutcomeDef( + success_field="pr_state", success_value="merged" + ), "wait_for_ci": RecipeDeliveryOutcomeDef( success_field="conclusion", success_value="success", ), "push_to_remote": RecipeDeliveryOutcomeDef(success_field="success"), "bootstrap_clone": RecipeDeliveryOutcomeDef(), - "run_python": RecipeDeliveryOutcomeDef(success_field="success"), + "run_python": RecipeDeliveryOutcomeDef(success_field="success", recipe_observable=False), "complete_run_skill_result": RecipeDeliveryOutcomeDef( success_field="run_skill_success", step_name_field="step_name", + recipe_observable=False, ), "merge_worktree": RecipeDeliveryOutcomeDef(success_field="merge_succeeded"), "check_pr_mergeable": RecipeDeliveryOutcomeDef(), diff --git a/src/autoskillit/core/types/recipe/_type_recipe_binding.py b/src/autoskillit/core/types/recipe/_type_recipe_binding.py index 74f2cd9ebb..ef74f8ccd5 100644 --- a/src/autoskillit/core/types/recipe/_type_recipe_binding.py +++ b/src/autoskillit/core/types/recipe/_type_recipe_binding.py @@ -164,11 +164,16 @@ class ToolParamDef: @dataclass(frozen=True, slots=True) class RecipeDeliveryOutcomeDef: - """How the response wrapper reads a delivery tool's outcome from its response.""" + """How the response wrapper reads a delivery tool's outcome from its response. + + ``recipe_observable`` is False when recipe ``result.*`` does not expose + ``success_field`` (the tool nests its result or the field is wrapper-only). + """ success_field: str | None = None success_value: bool | str = True step_name_field: str | None = None + recipe_observable: bool = True def succeeded(self, payload: Mapping[str, object]) -> bool: if self.success_field is None: diff --git a/src/autoskillit/recipe/__init__.py b/src/autoskillit/recipe/__init__.py index dd1e511521..4fe1f16bea 100644 --- a/src/autoskillit/recipe/__init__.py +++ b/src/autoskillit/recipe/__init__.py @@ -386,6 +386,7 @@ ) from autoskillit.recipe.validator import ( edge_routes_success, + result_arms_by_source, ) __all__ = [ @@ -426,6 +427,7 @@ "run_semantic_rules", "analyze_dataflow", "edge_routes_success", + "result_arms_by_source", "check_contract_staleness", "generate_recipe_card", "get_skill_contract", diff --git a/src/autoskillit/recipe/validator.py b/src/autoskillit/recipe/validator.py index 60e32d08d1..3380df8522 100644 --- a/src/autoskillit/recipe/validator.py +++ b/src/autoskillit/recipe/validator.py @@ -408,6 +408,7 @@ def _validate_delivery_graph( ) checkpoint_sources: list[list[str]] = [[] for _ in normalized] + result_arms = result_arms_by_source(ordered_flow_edges) for edge in ordered_flow_edges: source_index = segment_index.get(edge.source) target_index = segment_index.get(edge.target) @@ -435,6 +436,7 @@ def _validate_delivery_graph( edge, automatic=definition.automatic_recipe_delivery, recovery=definition.recovery_recipe_delivery, + result_arms=result_arms.get(edge.source, ()), ) capability = ( definition.automatic_recipe_delivery @@ -504,34 +506,63 @@ def _finalize_delivery_segments( ) +_RESULT_EQUALITY = re.compile( + r"^\s*(?P\$\{\{\s*)?result\.(?P[A-Za-z_]\w*)" + r"(?(wrapped)\s*\}\})\s*==\s*" + r"(?P['\"]?)(?P[\w.-]+)(?P=quote)\s*$" +) + + +def result_arms_by_source( + ordered_flow_edges: tuple[RecipeFlowEdge, ...], +) -> dict[str, tuple[RecipeFlowEdge, ...]]: + """Group finalized result-condition edges by source, preserving declaration order.""" + arms: dict[str, list[RecipeFlowEdge]] = {} + for edge in ordered_flow_edges: + if edge.edge_type == "result_condition": + arms.setdefault(edge.source, []).append(edge) + return {source: tuple(edges) for source, edges in arms.items()} + + +def _arm_matches_success(tool_name: str, condition: str | None) -> bool | None: + """True/False for equality to the registry success field; None otherwise.""" + definition = get_tool_def(tool_name) + outcome = definition.recipe_delivery_outcome if definition is not None else None + if ( + outcome is None + or outcome.success_field is None + or not outcome.recipe_observable + or condition is None + ): + return None + match = _RESULT_EQUALITY.fullmatch(condition) + if match is None or match["field"] != outcome.success_field: + return None + value = match["value"] + if isinstance(outcome.success_value, bool): + return value.lower() == str(outcome.success_value).lower() + return value == outcome.success_value + + def edge_routes_success( tool_name: str, edge: RecipeFlowEdge, *, automatic: bool, recovery: bool, + result_arms: tuple[RecipeFlowEdge, ...], ) -> bool: - """Return whether one finalized tool edge selects the success carrier.""" + """Return whether a successful tool outcome can take this finalized edge.""" if not automatic and recovery: return False if edge.edge_type in {"success", "skip"}: return True if edge.edge_type != "result_condition": return False - condition_parts = (edge.condition or "").split("==") - if len(condition_parts) != 2: - return False - field, value = (part.strip() for part in condition_parts) - if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: - value = value[1:-1] - success_result = { - "wait_for_ci": ("result.conclusion", "success"), - "wait_for_merge_queue": ("result.pr_state", "merged"), - "claim_and_resolve_issue": ("result.claimed", "true"), - }.get(tool_name) - if success_result is not None: - return (field, value) == success_result - return automatic + if edge.condition is None: + preceding = result_arms[: result_arms.index(edge)] + return not any(_arm_matches_success(tool_name, arm.condition) is True for arm in preceding) + return _arm_matches_success(tool_name, edge.condition) is not False # Re-export test-access symbols from their new locations. diff --git a/src/autoskillit/server/recipe/_recipe_segment_delivery.py b/src/autoskillit/server/recipe/_recipe_segment_delivery.py index c0835d9f63..fcc301402b 100644 --- a/src/autoskillit/server/recipe/_recipe_segment_delivery.py +++ b/src/autoskillit/server/recipe/_recipe_segment_delivery.py @@ -23,7 +23,7 @@ load_yaml, ) from autoskillit.pipeline import ReadyRecipe -from autoskillit.recipe import edge_routes_success +from autoskillit.recipe import edge_routes_success, result_arms_by_source from autoskillit.server.recipe._recipe_artifact import ( _finalized_projection_payload, _normalized_recipe_compile_identity, @@ -193,6 +193,8 @@ def _delivery_tool_name(projection: FinalizedRecipeProjection, source_step: str) def _edge_routes_success( projection: FinalizedRecipeProjection, edge: RecipeFlowEdge, + *, + result_arms: tuple[RecipeFlowEdge, ...], ) -> bool: tool_name = _delivery_tool_name(projection, edge.source) definition = get_tool_def(tool_name) if tool_name is not None else None @@ -201,6 +203,7 @@ def _edge_routes_success( edge, automatic=definition is not None and definition.automatic_recipe_delivery, recovery=definition is not None and definition.recovery_recipe_delivery, + result_arms=result_arms, ) @@ -215,13 +218,19 @@ def _target_segment_indices( if source_index is None: return () selected: list[int] = [] + result_arms = result_arms_by_source(projection.ordered_flow_edges) for edge in projection.ordered_flow_edges: target_index = indices.get(edge.target) if ( edge.source != source_step or target_index is None or target_index <= source_index - or (_edge_routes_success(projection, edge) is not success) + or ( + _edge_routes_success( + projection, edge, result_arms=result_arms.get(edge.source, ()) + ) + is not success + ) ): continue if target_index not in selected: @@ -239,8 +248,8 @@ def _manual_closure( projection: FinalizedRecipeProjection, target_steps: tuple[str, ...], ) -> tuple[str, ...]: - ordered = projection.ordered_step_names - known = frozenset(ordered) + """Pull reached segments, stopping traversal at non-root automatic checkpoints.""" + segment_of = _segment_index(projection) automatic_sources = frozenset( source for segment in projection.delivery_segments @@ -250,20 +259,23 @@ def _manual_closure( and definition.automatic_recipe_delivery ) visited: set[str] = set() + entered: set[int] = set() pending = list(target_steps) while pending: step_name = pending.pop() - if step_name in visited or step_name not in known: + if step_name in visited or step_name not in segment_of: continue visited.add(step_name) + segment_index = segment_of[step_name] + if segment_index not in entered: + entered.add(segment_index) + pending.extend(projection.delivery_segments[segment_index].ordered_step_names) if step_name in automatic_sources and step_name not in target_steps: continue pending.extend( - edge.target - for edge in projection.ordered_flow_edges - if edge.source == step_name and edge.target in known + edge.target for edge in projection.ordered_flow_edges if edge.source == step_name ) - return tuple(step_name for step_name in ordered if step_name in visited) + return tuple(step_name for step_name in projection.ordered_step_names if step_name in visited) def _route_pull_closures( @@ -301,6 +313,7 @@ def validate_segment_delivery_projection(projection: FinalizedRecipeProjection) return indices = _segment_index(projection) invocations = projection.binding_projection.invocations + result_arms = result_arms_by_source(projection.ordered_flow_edges) for edge in projection.ordered_flow_edges: source_index = indices.get(edge.source) target_index = indices.get(edge.target) @@ -317,7 +330,7 @@ def validate_segment_delivery_projection(projection: FinalizedRecipeProjection) ) capability = ( definition.automatic_recipe_delivery - if _edge_routes_success(projection, edge) + if _edge_routes_success(projection, edge, result_arms=result_arms.get(edge.source, ())) else definition.recovery_recipe_delivery ) if not capability: diff --git a/src/autoskillit/server/response/_response_budget/_enforce.py b/src/autoskillit/server/response/_response_budget/_enforce.py index 7636301997..03ff3cfe7f 100644 --- a/src/autoskillit/server/response/_response_budget/_enforce.py +++ b/src/autoskillit/server/response/_response_budget/_enforce.py @@ -4,13 +4,16 @@ import hashlib import json +from dataclasses import replace from pathlib import Path from typing import TYPE_CHECKING, Any from autoskillit.core import ( + ASCII_YAML_POLICY, CLIENT_CHARS_PER_TOKEN_POLICY, RESPONSE_BACKSTOP_EXEMPTION_REGISTRY, TokenLimit, + Utf8ByteLimit, ) from autoskillit.server.recipe._recipe_segment_delivery import ( RECIPE_SEGMENT_MAX_BYTES, @@ -326,46 +329,42 @@ def _enforce_checkpoint_segmented_response( return _segment_failure_like(original, carrier, tool_name=tool_name) rendered_size = len(rendered.encode("utf-8")) final_limit = RECIPE_SEGMENT_MAX_BYTES - 1 - over_delivery_bound = ( - selected_result_token_limit is not None - and selected_result_token_limit > 0 - and _estimated_tokens(rendered_size) > selected_result_token_limit - ) - if not force_spill and not over_delivery_bound and rendered_size <= final_limit: + if selected_result_token_limit is not None and selected_result_token_limit > 0: + final_limit = min( + final_limit, ASCII_YAML_POLICY.to_bytes(TokenLimit(selected_result_token_limit)).value + ) + if not force_spill and rendered_size <= final_limit: return rendered if isinstance(original, str) else combined if artifact_dir is None: return _segment_failure_like(original, carrier, tool_name=tool_name) + # Reserve the unchanged carrier, its key, and the separating comma before + # projecting the base; artifact metadata must fit in the remainder too. + carrier_overhead = len(_canonical_json({"recipe_segment": carrier}).encode("utf-8")) - 1 + base_max_bytes = min(config.response_max_bytes, final_limit - carrier_overhead) base_input: Any = _canonical_json(base) if isinstance(original, str) else base - shaped_base = enforce_response_budget( - base_input, - tool_name=tool_name, - artifact_dir=artifact_dir, - config=config, - force_spill=True, - selected_result_token_limit=selected_result_token_limit, - ) try: + shaped_base = enforce_response_budget( + base_input, + tool_name=tool_name, + artifact_dir=artifact_dir, + config=replace(config, response_max_bytes=Utf8ByteLimit(base_max_bytes)), + force_spill=True, + selected_result_token_limit=selected_result_token_limit, + ) shaped_mapping = json.loads(shaped_base) if isinstance(shaped_base, str) else shaped_base - except (TypeError, ValueError, RecursionError): - return _segment_failure_like(original, carrier, tool_name=tool_name) - if not isinstance(shaped_mapping, dict): - return _segment_failure_like(original, carrier, tool_name=tool_name) - shaped_error = shaped_mapping.get("error") - if isinstance(shaped_error, str) and shaped_error.startswith("response_budget_"): - return _segment_failure_like(original, carrier, tool_name=tool_name) + if not isinstance(shaped_mapping, dict): + return _segment_failure_like(original, carrier, tool_name=tool_name) + shaped_error = shaped_mapping.get("error") + if isinstance(shaped_error, str) and shaped_error.startswith("response_budget_"): + return _segment_failure_like(original, carrier, tool_name=tool_name) - recombined = {**shaped_mapping, "recipe_segment": carrier} - try: + recombined = {**shaped_mapping, "recipe_segment": carrier} final_rendered = _canonical_json(recombined) except (TypeError, ValueError, OverflowError, RecursionError): return _segment_failure_like(original, carrier, tool_name=tool_name) final_size = len(final_rendered.encode("utf-8")) - if final_size > final_limit or ( - selected_result_token_limit is not None - and selected_result_token_limit > 0 - and _estimated_tokens(final_size) > selected_result_token_limit - ): + if final_size > final_limit: return _segment_failure_like(original, carrier, tool_name=tool_name) return final_rendered if isinstance(original, str) else recombined diff --git a/tests/_test_filter.py b/tests/_test_filter.py index 11fa05cc8c..122b137764 100644 --- a/tests/_test_filter.py +++ b/tests/_test_filter.py @@ -1339,6 +1339,7 @@ class SourceMapRejection(enum.StrEnum): "server/test_serve_idempotence.py", "server/test_attestation_delivery_reachability.py", "server/test_recipe_execution_receipt_completeness.py", + "server/test_recipe_checkpoint_delivery_regressions.py", "server/test_tools_status_kitchen.py", "server/test_open_kitchen_deferred_recall.py", "server/test_factory_context_construction.py", diff --git a/tests/arch/_recipe_gateway_snapshot.py b/tests/arch/_recipe_gateway_snapshot.py index 4a51c3f5cf..5edb639bf4 100644 --- a/tests/arch/_recipe_gateway_snapshot.py +++ b/tests/arch/_recipe_gateway_snapshot.py @@ -274,6 +274,7 @@ "resolve_input_specs", "resolve_skill_name", "resolve_venue_appendices", + "result_arms_by_source", "rules", "run_semantic_rules", "schema", diff --git a/tests/arch/test_recipe_checkpoint_delivery_authority.py b/tests/arch/test_recipe_checkpoint_delivery_authority.py index c06ecb882e..96cb4ca6f6 100644 --- a/tests/arch/test_recipe_checkpoint_delivery_authority.py +++ b/tests/arch/test_recipe_checkpoint_delivery_authority.py @@ -49,3 +49,19 @@ def test_only_the_delivery_module_prepares_or_attaches_carriers() -> None: "Recipe checkpoint carrier preparation and attachment must stay in " "server/recipe/_recipe_segment_delivery.py:\n" + "\n".join(violations) ) + + +def test_segment_delivery_does_not_read_result_conditions() -> None: + tree = ast.parse( + _DELIVERY_AUTHORITY_MODULE.read_text(encoding="utf-8"), + filename=str(_DELIVERY_AUTHORITY_MODULE), + ) + condition_reads = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.Attribute) + and node.attr == "condition" + and isinstance(node.ctx, ast.Load) + ] + + assert not condition_reads, "Condition interpretation belongs in edge_routes_success." diff --git a/tests/core/test_tool_def_delivery_outcome.py b/tests/core/test_tool_def_delivery_outcome.py index 9848471ab4..147cbdd9f1 100644 --- a/tests/core/test_tool_def_delivery_outcome.py +++ b/tests/core/test_tool_def_delivery_outcome.py @@ -5,6 +5,7 @@ import pytest from autoskillit.core import ( + PRState, RecipeDeliveryOutcomeDef, ToolDef, ToolInitializationOperation, @@ -109,5 +110,54 @@ def test_outcome_reader_semantics( assert outcome.succeeded(payload) is expected +@pytest.mark.parametrize( + ("payload", "expected"), + [({"success": True}, True), ({"success": False}, False)], +) +def test_recipe_observability_does_not_change_outcome_reading( + payload: dict[str, object], + expected: bool, +) -> None: + outcome = RecipeDeliveryOutcomeDef(success_field="success", recipe_observable=False) + + assert outcome.succeeded(payload) is expected + + def test_every_outcome_key_names_a_registered_tool() -> None: assert set(_RECIPE_DELIVERY_OUTCOMES) <= set(TOOL_REGISTRY) + + +def test_wait_for_merge_queue_delivery_outcome_uses_merged_state() -> None: + outcome = TOOL_REGISTRY["wait_for_merge_queue"].recipe_delivery_outcome + assert outcome is not None + + assert outcome.succeeded({"pr_state": PRState.MERGED.value}) is True + assert outcome.succeeded({"success": False, "error": "x"}) is False + + +@pytest.mark.parametrize( + "state", + [state for state in PRState if state is not PRState.MERGED], + ids=lambda state: state.value, +) +def test_wait_for_merge_queue_delivery_outcome_rejects_other_pr_states( + state: PRState, +) -> None: + outcome = TOOL_REGISTRY["wait_for_merge_queue"].recipe_delivery_outcome + assert outcome is not None + + assert outcome.succeeded({"pr_state": state.value}) is False + + +def test_recipe_observable_defaults_true_for_registered_outcomes() -> None: + assert RecipeDeliveryOutcomeDef().recipe_observable is True + for tool_name in ("run_python", "complete_run_skill_result"): + outcome = TOOL_REGISTRY[tool_name].recipe_delivery_outcome + assert outcome is not None + assert outcome.recipe_observable is False + for tool_name, tool_def in TOOL_REGISTRY.items(): + outcome = tool_def.recipe_delivery_outcome + if outcome is None: + continue + if tool_name not in {"run_python", "complete_run_skill_result"}: + assert outcome.recipe_observable is True, tool_name diff --git a/tests/recipe/test_delivery_segments.py b/tests/recipe/test_delivery_segments.py index 36d944d77a..0a7c98be70 100644 --- a/tests/recipe/test_delivery_segments.py +++ b/tests/recipe/test_delivery_segments.py @@ -2,6 +2,8 @@ from __future__ import annotations +import re + import pytest from autoskillit.core import ( @@ -9,11 +11,20 @@ FinalizedRecipeStep, RecipeBindingProjection, RecipeFlowEdge, + ToolDef, ) +from autoskillit.core.tool_registry import TOOL_REGISTRY +from autoskillit.recipe import result_arms_by_source +from autoskillit.recipe._api import load_and_validate from autoskillit.recipe._recipe_composition import _prune_skipped_steps from autoskillit.recipe.io import _parse_recipe, builtin_recipes_dir, load_recipe from autoskillit.recipe.schema import Recipe, RecipeDeliverySegment, RecipeIngredient, RecipeStep -from autoskillit.recipe.validator import _finalize_delivery_segments, validate_recipe_structure +from autoskillit.recipe.validator import ( + _finalize_delivery_segments, + edge_routes_success, + validate_recipe_structure, +) +from tests.server._recipe_fixture_parameters import IMPLEMENTATION_OVERRIDES pytestmark = [pytest.mark.layer("recipe"), pytest.mark.small] @@ -324,3 +335,121 @@ def test_bundled_segmented_recipe_opt_in_is_exact() -> None: } assert opted_in == {"implementation", "remediation"} + + +# These assertions check broad field/value presence, not production's strict equality +# grammar, so they can independently justify why the registry rejects a result arm. +def _mentions_success_field(condition: str | None, success_field: str) -> bool: + return ( + condition is not None + and re.search(rf"\bresult\.{re.escape(success_field)}\b", condition) is not None + ) + + +def _mentions_success_value(condition: str | None, success_value: bool | str) -> bool: + if condition is None: + return False + flags = re.IGNORECASE if isinstance(success_value, bool) else 0 + return re.search(rf"\b{re.escape(str(success_value))}\b", condition, flags=flags) is not None + + +def _assert_registry_justifies_rejected_arm( + tool_name: str, + definition: ToolDef, + result_arms: tuple[RecipeFlowEdge, ...], + arm_index: int, +) -> None: + edge = result_arms[arm_index] + if edge_routes_success( + tool_name, + edge, + automatic=definition.automatic_recipe_delivery, + recovery=definition.recovery_recipe_delivery, + result_arms=result_arms, + ): + return + + outcome = definition.recipe_delivery_outcome + assert outcome is not None and outcome.recipe_observable + assert outcome.success_field is not None + if edge.condition is None: + assert any( + _mentions_success_field(earlier.condition, outcome.success_field) + and _mentions_success_value(earlier.condition, outcome.success_value) + for earlier in result_arms[:arm_index] + ) + else: + assert _mentions_success_field(edge.condition, outcome.success_field) + assert not _mentions_success_value(edge.condition, outcome.success_value) + + +@pytest.mark.parametrize( + ("recipe_name", "defer_unresolved"), + [ + pytest.param("implementation", False, id="implementation-resolved"), + pytest.param("implementation", True, id="implementation-deferred"), + pytest.param("remediation", False, id="remediation-resolved"), + pytest.param("remediation", True, id="remediation-deferred"), + ], +) +def test_registry_anchors_automatic_delivery_classification( + recipe_name: str, + defer_unresolved: bool, +) -> None: + overrides = dict(IMPLEMENTATION_OVERRIDES) + if recipe_name == "remediation": + overrides["investigate"] = "true" + result = load_and_validate( + recipe_name, + ingredient_overrides=overrides, + defer_unresolved=defer_unresolved, + include_finalized_projection=True, + ) + assert result["valid"], result["errors"] + projection = result["_finalized_projection"] + assert projection.delivery_segments + segment_index = { + step_name: index + for index, segment in enumerate(projection.delivery_segments) + for step_name in segment.ordered_step_names + } + check_changes_arms = tuple( + edge + for edge in projection.ordered_flow_edges + if edge.source == "check_changes" and edge.edge_type == "result_condition" + ) + no_change = next( + edge + for edge in check_changes_arms + if edge.target == "close_issue_no_changes" and edge.condition is not None + ) + fallback = next( + edge for edge in check_changes_arms if edge.target == "test" and edge.condition is None + ) + assert no_change.condition is not None and "result.has_changes" in no_change.condition + assert segment_index["close_issue_no_changes"] == segment_index["check_changes"] + assert segment_index["test"] > segment_index["check_changes"] + assert fallback.condition is None + + for source, result_arms in result_arms_by_source(projection.ordered_flow_edges).items(): + source_step = projection.for_step(source) + if source_step is None or source_step.tool is None: + continue + tool_name = ( + "complete_run_skill_result" if source_step.tool == "run_skill" else source_step.tool + ) + definition = TOOL_REGISTRY[tool_name] + if not definition.automatic_recipe_delivery: + continue + source_index = segment_index[source] + for arm_index, edge in enumerate(result_arms): + if edge.target not in segment_index: + continue + if segment_index[edge.target] <= source_index: + continue + _assert_registry_justifies_rejected_arm( + tool_name, + definition, + result_arms, + arm_index, + ) diff --git a/tests/recipe/test_implementation.py b/tests/recipe/test_implementation.py index 1b2aeb3d6f..2189a9798e 100644 --- a/tests/recipe/test_implementation.py +++ b/tests/recipe/test_implementation.py @@ -4,11 +4,15 @@ import hashlib import json +from dataclasses import replace from pathlib import Path +from types import MappingProxyType import pytest -from autoskillit.core import RecipeFlowEdge, get_tool_def, load_yaml +import autoskillit.core.tool_registry as tool_registry +from autoskillit.core import RecipeFlowEdge, load_yaml +from autoskillit.core.tool_registry import TOOL_REGISTRY from autoskillit.recipe.io import builtin_recipes_dir, load_recipe from autoskillit.recipe.validator import ( _extract_routing_edges, @@ -100,24 +104,6 @@ def _assert_delivery_projection_contract( ) assert finalized_steps == frozenset(recipe.steps) - checkpoint_sources = frozenset( - source for segment in segments for source in segment.checkpoint_sources - ) - for edge in edges: - step = recipe.steps[edge.source] - if step.tool is None or edge.source not in checkpoint_sources: - continue - tool_name = "complete_run_skill_result" if step.tool == "run_skill" else step.tool - definition = get_tool_def(tool_name) - assert definition is not None - if edge_routes_success( - tool_name, - edge, - automatic=definition.automatic_recipe_delivery, - recovery=definition.recovery_recipe_delivery, - ): - assert edge.target in finalized_steps - assert { key: recipe.steps[key].tool for key in _CHECKPOINT_HANDLER_TABLE } == _CHECKPOINT_HANDLER_TABLE @@ -126,28 +112,196 @@ def _assert_delivery_projection_contract( } == {key: key for key in _CHECKPOINT_HANDLER_TABLE} +def _result_arms(*conditions: str | None) -> tuple[RecipeFlowEdge, ...]: + return tuple( + RecipeFlowEdge("wait", "result_condition", f"route-{index}", condition, None) + for index, condition in enumerate(conditions) + ) + + @pytest.mark.parametrize( - ("condition", "expected"), + ("tool_name", "conditions", "edge_index", "expected"), [ - (" result.conclusion == 'success' ", True), - ('result.conclusion=="success"', True), - ("other.result.conclusion == 'success'", False), - ("result.conclusion == 'success' == true", False), + ("wait_for_ci", ("${{ result.conclusion }} == 'success'",), 0, True), + ("wait_for_ci", ('result.conclusion == "success"',), 0, True), + ("wait_for_ci", ("${{ result.conclusion }} == 'no_runs'",), 0, False), + ( + "wait_for_ci", + ( + "${{ result.conclusion }} == 'success'", + "${{ result.conclusion }} == 'no_runs'", + None, + ), + 2, + False, + ), + ("wait_for_ci", ("${{ result.conclusion }} == 'no_runs'", None), 1, True), + ("wait_for_ci", ("${{ result.conclusion }} == ${{ context.x }}",), 0, True), + ("wait_for_merge_queue", ("${{ result.pr_state }} == merged",), 0, True), + ("wait_for_merge_queue", ("${{ result.pr_state }} == ejected",), 0, False), + ("run_python", ("${{ result.has_changes }} == false", None), 0, True), + ("run_python", ("${{ result.has_changes }} == false", None), 1, True), + ("run_python", ("${{ result.success }} == 'true'", None), 1, True), + ("run_python", ("${{ result.cleaned }} == 'failed'", None), 1, True), + ( + "wait_for_ci", + ("${{ result.reason == 'ok' and result.status == 'ready' }}",), + 0, + True, + ), + ("wait_for_ci", ("${{ result.conclusion }}",), 0, True), + ("wait_for_ci", ("${{ result.conclusion }} != 'failure'",), 0, True), + ("claim_and_resolve_issue", ("${{ result.claimed }} == true",), 0, False), ], + ids=( + "ci-wrapped-success ci-bare-success ci-no-runs ci-fallback-after-success " + "ci-fallback-no-success ci-dynamic queue-merged queue-ejected python-no-change " + "python-no-change-fallback python-success-fallback python-cleaned-fallback " + "ci-conjunction ci-truthy ci-inequality recovery-only-claim" + ).split(), ) -def test_edge_routes_success_requires_exact_result_equality( - condition: str, +def test_edge_routes_success_from_ordered_recipe_arms( + tool_name: str, + conditions: tuple[str | None, ...], + edge_index: int, + expected: bool, +) -> None: + arms = _result_arms(*conditions) + edge = arms[edge_index] + definition = TOOL_REGISTRY[tool_name] + + assert ( + edge_routes_success( + tool_name, + edge, + automatic=definition.automatic_recipe_delivery, + recovery=definition.recovery_recipe_delivery, + result_arms=arms, + ) + is expected + ) + + +@pytest.mark.parametrize( + ("edge_type", "expected"), + [ + pytest.param("success", True, id="automatic-success"), + pytest.param("skip", True, id="automatic-skip"), + pytest.param("failure", False, id="failure"), + pytest.param("exhausted", False, id="exhausted"), + ], +) +def test_edge_routes_success_handles_non_result_edges( + edge_type: str, expected: bool, ) -> None: - edge = RecipeFlowEdge( - source="wait", - edge_type="result_condition", - target="done", - condition=condition, - result_field=None, + definition = TOOL_REGISTRY["wait_for_ci"] + edge = RecipeFlowEdge("wait", edge_type, "route", None, None) + + assert ( + edge_routes_success( + "wait_for_ci", + edge, + automatic=definition.automatic_recipe_delivery, + recovery=definition.recovery_recipe_delivery, + result_arms=(), + ) + is expected + ) + + +@pytest.mark.parametrize( + "condition", + ["${{ result.success == true", "result.success }} == true"], + ids=("unclosed-wrapper", "unmatched-wrapper-close"), +) +def test_malformed_automatic_result_arms_and_fallback_remain_reachable( + condition: str, +) -> None: + arms = _result_arms(condition, None) + definition = TOOL_REGISTRY["enqueue_pr"] + + for edge in arms: + assert ( + edge_routes_success( + "enqueue_pr", + edge, + automatic=definition.automatic_recipe_delivery, + recovery=definition.recovery_recipe_delivery, + result_arms=arms, + ) + is True + ) + assert ( + edge_routes_success( + "enqueue_pr", + edge, + automatic=False, + recovery=True, + result_arms=arms, + ) + is False + ) + + +@pytest.mark.parametrize( + ("outcome_changes", "condition", "expected_conditioned", "expected_fallback"), + [ + pytest.param( + {"success_field": "checkpoint_status"}, + "${{ result.checkpoint_status }} == 'success'", + True, + False, + id="registry-success-field", + ), + pytest.param( + {"success_value": "delivered"}, + "${{ result.conclusion }} == 'success'", + False, + True, + id="registry-success-value", + ), + pytest.param( + {"recipe_observable": False}, + "${{ result.conclusion }} == 'success'", + True, + True, + id="registry-unobservable-result", + ), + ], +) +def test_edge_routes_success_uses_replaced_registry_outcome( + monkeypatch: pytest.MonkeyPatch, + outcome_changes: dict[str, object], + condition: str, + expected_conditioned: bool, + expected_fallback: bool, +) -> None: + original_tool = TOOL_REGISTRY["wait_for_ci"] + original_outcome = original_tool.recipe_delivery_outcome + assert original_outcome is not None + replacement = replace( + original_tool, + recipe_delivery_outcome=replace(original_outcome, **outcome_changes), + ) + registry = MappingProxyType({**TOOL_REGISTRY, "wait_for_ci": replacement}) + monkeypatch.setattr(tool_registry, "TOOL_REGISTRY", registry) + + registered = tool_registry.get_tool_def("wait_for_ci") + assert registered is replacement + arms = _result_arms(condition, None) + results = tuple( + edge_routes_success( + "wait_for_ci", + edge, + automatic=registered.automatic_recipe_delivery, + recovery=registered.recovery_recipe_delivery, + result_arms=arms, + ) + for edge in arms ) - assert edge_routes_success("wait_for_ci", edge, automatic=True, recovery=True) is expected + assert results == (expected_conditioned, expected_fallback) @pytest.fixture(scope="module") diff --git a/tests/server/_recipe_fixture_parameters.py b/tests/server/_recipe_fixture_parameters.py index f677d00247..28d72c8a5d 100644 --- a/tests/server/_recipe_fixture_parameters.py +++ b/tests/server/_recipe_fixture_parameters.py @@ -10,6 +10,11 @@ IMPLEMENTATION_READY_PARAMETERS = [ pytest.param(("implementation", "plan", IMPLEMENTATION_OVERRIDES), id="implementation"), ] + + +REMEDIATION_READY_PARAMETERS = [ + pytest.param(("remediation", None, IMPLEMENTATION_OVERRIDES), id="remediation"), +] IMPLEMENTATION_REVIEW_READY_PARAMETERS = [ pytest.param(("implementation", "plan", IMPLEMENTATION_OVERRIDES), id="default"), pytest.param(("implementation", "plan", REVIEW_APPROACH_OVERRIDES), id="review-approach"), diff --git a/tests/server/conftest.py b/tests/server/conftest.py index 682ba0773b..4d550ef562 100644 --- a/tests/server/conftest.py +++ b/tests/server/conftest.py @@ -481,9 +481,11 @@ async def tool_ctx_ready_recipe( Indirectly parametrizable over ``(recipe_name, attested_step, ingredient_overrides)`` via ``@pytest.mark.parametrize("tool_ctx_ready_recipe", [(...)], indirect=True)``. + A null attested_step selects the first attested run_skill step during setup. Without parametrization, the fixture uses ``RESEARCH_READY_DEFAULT`` from the shared recipe inputs module. """ + from autoskillit.pipeline import ReadyRecipe from autoskillit.server.tools.tools_recipe import complete_recipe_initialization from tests.server._helpers import ( _credit_initialization_sections, @@ -540,6 +542,19 @@ def load_from_worker_project(name, loaded_project_dir, *args, **kwargs): # required_sections is absent — so no bounded-only assertion belongs here. await _credit_initialization_sections(envelope) + if attested_step is None: + state = tool_ctx_kitchen_open.recipe_initialization_state + assert isinstance(state, ReadyRecipe) + invocations = state.finalized_projection.binding_projection.invocations + attested_step = next( + ( + step_name + for step_name, invocation in invocations.items() + if invocation.tool_name == "run_skill" and invocation.attested + ), + None, + ) + assert attested_step is not None, f"{recipe_name} has no attested run_skill invocation" step_body = await _pull_step_section(envelope, attested_step) forbid_artifact_reads() receipt = json.loads(await complete_recipe_initialization(envelope["initialization_id"])) diff --git a/tests/server/test_recipe_checkpoint_delivery_regressions.py b/tests/server/test_recipe_checkpoint_delivery_regressions.py new file mode 100644 index 0000000000..5e1ef23536 --- /dev/null +++ b/tests/server/test_recipe_checkpoint_delivery_regressions.py @@ -0,0 +1,319 @@ +"""Regression coverage for checkpoint carriers across finalized recipe routes.""" + +from __future__ import annotations + +import json +from typing import Any + +import pytest + +import autoskillit.server.recipe._recipe_segment_delivery as segment_delivery +import autoskillit.server.tools._recipe_section_handler as section_handler +from autoskillit.core import FinalizedRecipeProjection, RecipeFlowEdge, get_tool_def +from autoskillit.core.io import load_yaml +from autoskillit.pipeline import ReadyRecipe +from autoskillit.recipe import result_arms_by_source +from autoskillit.server.recipe._recipe_artifact import load_recipe_artifact +from autoskillit.server.recipe._recipe_segment_delivery import ( + RECIPE_SEGMENT_MAX_BYTES, + attach_recipe_segment, + prepare_recipe_segment_delivery, +) +from autoskillit.server.tools.tools_recipe import get_recipe_section +from tests.server._recipe_fixture_parameters import ( + IMPLEMENTATION_READY_PARAMETERS, + REMEDIATION_READY_PARAMETERS, +) + +pytestmark = [pytest.mark.layer("server"), pytest.mark.anyio, pytest.mark.medium] + +_READY_PARAMETERS = [*IMPLEMENTATION_READY_PARAMETERS, *REMEDIATION_READY_PARAMETERS] + + +def _serialized_size(value: object) -> int: + return len( + json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode( + "utf-8" + ) + ) + + +def _delivered(carrier: dict[str, Any]) -> set[str]: + if carrier["kind"] == "recovery": + return set(carrier["pull_closure"]) + return {body["step"] for body in carrier["bodies"]} | { + step for closure in carrier["pull_closures"] for step in closure["steps"] + } + + +def _carrier_closures(carrier: dict[str, Any]) -> list[tuple[str, ...]]: + if carrier["kind"] == "recovery": + return [tuple(carrier["pull_closure"])] + return [tuple(closure["steps"]) for closure in carrier["pull_closures"]] + + +def _ready_state(ready: Any) -> ReadyRecipe: + state = ready.tool_ctx.recipe_initialization_state + assert isinstance(state, ReadyRecipe) + assert isinstance(state.finalized_projection, FinalizedRecipeProjection) + return state + + +def _automatic_checkpoint_sources( + projection: FinalizedRecipeProjection, +) -> set[str]: + automatic_sources: set[str] = set() + for segment in projection.delivery_segments: + for source in segment.checkpoint_sources: + invocation = projection.binding_projection.invocations.get(source) + if invocation is None: + continue + tool_name = ( + "complete_run_skill_result" + if invocation.tool_name == "run_skill" + else invocation.tool_name + ) + definition = get_tool_def(tool_name) + if definition is not None and definition.automatic_recipe_delivery: + automatic_sources.add(source) + return automatic_sources + + +def _assert_carrier_closures_complete( + projection: FinalizedRecipeProjection, + carriers: list[dict[str, Any]], + *, + recipe_name: str, + source: str, + step_segments: dict[str, Any], + finalized_steps: frozenset[str], + automatic_sources: set[str], +) -> None: + for carrier in carriers: + assert _serialized_size(carrier) < RECIPE_SEGMENT_MAX_BYTES, ( + f"{recipe_name}:{source}:{carrier['kind']} carrier exceeds bound" + ) + body_steps = {body["step"] for body in carrier.get("bodies", ())} + for closure in _carrier_closures(carrier): + closure_steps = set(closure) + delivered = body_steps | closure_steps + assert len(closure_steps) == len(closure) + assert closure_steps <= finalized_steps + assert closure == tuple( + step for step in projection.ordered_step_names if step in closure_steps + ), f"{recipe_name}:{source} closure is not declaration ordered" + for step in delivered: + assert set(step_segments[step].ordered_step_names) <= delivered, ( + f"{recipe_name}:{source} closure splits segment at {step}" + ) + for step in closure_steps - automatic_sources: + for edge in projection.ordered_flow_edges: + if edge.source == step and edge.target in finalized_steps: + assert edge.target in delivered, ( + f"{recipe_name}:{source} closure omits " + f"walkable edge {step}->{edge.target}" + ) + + +def _assert_success_routes_and_response_bounds( + projection: FinalizedRecipeProjection, + prepared: Any, + *, + recipe_name: str, + source: str, + segment_indices: dict[str, int], + result_arms_by_source: dict[str, tuple[RecipeFlowEdge, ...]], +) -> None: + source_index = segment_indices[source] + for edge in projection.ordered_flow_edges: + target_index = segment_indices.get(edge.target) + if ( + edge.source != source + or edge.edge_type == "skip" + or target_index is None + or target_index <= source_index + ): + continue + if not segment_delivery._edge_routes_success( + projection, + edge, + result_arms=result_arms_by_source.get(source, ()), + ): + continue + for carrier in (prepared.success_carrier, prepared.success_pull_only_carrier): + if carrier is not None: + assert edge.target in _delivered(carrier), ( + f"{recipe_name}:{source} success carrier omits {edge.target}" + ) + + for result, success in ( + ( + {"success": True, "result": {"has_changes": True, "commit_count": 1}}, + True, + ), + ({"success": False, "error": "checkpoint failed"}, False), + ): + attached = attach_recipe_segment(result, prepared, success=success) + assert _serialized_size(attached) < RECIPE_SEGMENT_MAX_BYTES, ( + f"{recipe_name}:{source} attached response exceeds bound" + ) + + +@pytest.mark.parametrize("tool_ctx_ready_recipe", _READY_PARAMETERS, indirect=True) +async def test_checkpoint_carriers_deliver_pinned_recipe_routes(tool_ctx_ready_recipe) -> None: + state = _ready_state(tool_ctx_ready_recipe) + tool_ctx = tool_ctx_ready_recipe.tool_ctx + expectations: dict[str, dict[str, set[str]]] = { + "check_changes": { + "success_carrier": {"test"}, + "success_pull_only_carrier": {"test"}, + "recovery_carrier": {"release_issue_failure"}, + }, + "ci_watch": {"success_carrier": {"check_repo_merge_state"}}, + "wait_for_queue": {"success_carrier": {"release_issue_success"}}, + "main_repo_guard": {"success_carrier": {"merge", "release_issue_failure"}}, + "commit_guard": {"recovery_carrier": {"merge", "close_issue_already_done"}}, + "check_merge_fix_loop": {"success_carrier": {"release_issue_failure"}}, + } + if state.recipe_name == "implementation": + expectations.update( + { + "test": {"success_pull_only_carrier": {"commit_guard"}}, + "bind_plan_set": { + "success_pull_only_carrier": { + "verify", + "create_impl_worktree", + "check_changes", + } + }, + } + ) + + for source, slots in expectations.items(): + prepared = prepare_recipe_segment_delivery(tool_ctx, source) + assert prepared is not None, source + for slot, expected_steps in slots.items(): + carrier = getattr(prepared, slot) + assert carrier is not None, f"{state.recipe_name}:{source}:{slot}" + assert expected_steps <= _delivered(carrier), ( + f"{state.recipe_name}:{source}:{slot} omitted " + f"{expected_steps - _delivered(carrier)}" + ) + + check_changes = prepare_recipe_segment_delivery(tool_ctx, "check_changes") + assert check_changes is not None and check_changes.success_carrier is not None + test_body = next( + body for body in check_changes.success_carrier["bodies"] if body["step"] == "test" + ) + parsed_test = load_yaml(test_body["body"]) + assert parsed_test["test"]["on_success"] == ( + "commit_guard" if state.recipe_name == "implementation" else "audit_impl" + ) + attached = attach_recipe_segment( + { + "success": True, + "result": {"has_changes": True, "commit_count": 1}, + }, + check_changes, + success=True, + ) + assert attached["recipe_segment"] == check_changes.success_carrier + assert _serialized_size(attached) < RECIPE_SEGMENT_MAX_BYTES + + +@pytest.mark.parametrize("tool_ctx_ready_recipe", IMPLEMENTATION_READY_PARAMETERS, indirect=True) +async def test_pull_only_check_changes_section_serves_test_body( + tool_ctx_ready_recipe, + monkeypatch: pytest.MonkeyPatch, +) -> None: + state = _ready_state(tool_ctx_ready_recipe) + prepared = prepare_recipe_segment_delivery(tool_ctx_ready_recipe.tool_ctx, "check_changes") + assert prepared is not None and prepared.success_pull_only_carrier is not None + carrier = prepared.success_pull_only_carrier + closure = next(closure for closure in carrier["pull_closures"] if "test" in closure["steps"]) + request = next(request for request in closure["pull_requests"] if request["section"] == "test") + + # The READY fixture poisons artifact reads after initialization; this pull must read + # the production artifact through the production section handler. + monkeypatch.setattr(section_handler, "load_recipe_artifact", load_recipe_artifact) + identity = {key: value for key, value in carrier["recipe_pull"].items() if key != "pull_tool"} + part = request["part"] + page_plan_sha256 = None + continuation = None + content: list[str] = [] + while True: + page = json.loads( + await get_recipe_section( + section=request["section"], + part=part, + page_plan_sha256=page_plan_sha256, + continuation=continuation, + **identity, + ) + ) + assert page["success"] is True + content.append(page["content"]) + if not page["has_more"]: + break + part = page["next_part"] + page_plan_sha256 = page["page_plan_sha256"] + continuation = page["continuation"] + + body = load_yaml("".join(content)) + assert body["test"]["on_success"] == "commit_guard" + assert state.recipe_name == "implementation" + + +@pytest.mark.parametrize("tool_ctx_ready_recipe", _READY_PARAMETERS, indirect=True) +async def test_checkpoint_delivery_closures_and_response_bounds_are_complete( + tool_ctx_ready_recipe, +) -> None: + state = _ready_state(tool_ctx_ready_recipe) + projection = state.finalized_projection + tool_ctx = tool_ctx_ready_recipe.tool_ctx + step_segments = { + step: segment + for segment in projection.delivery_segments + for step in segment.ordered_step_names + } + segment_indices = { + step: index + for index, segment in enumerate(projection.delivery_segments) + for step in segment.ordered_step_names + } + finalized_steps = frozenset(projection.ordered_step_names) + result_arms = result_arms_by_source(projection.ordered_flow_edges) + automatic_sources = _automatic_checkpoint_sources(projection) + + for segment in projection.delivery_segments: + for source in segment.checkpoint_sources: + if source not in projection.binding_projection.invocations: + continue + prepared = prepare_recipe_segment_delivery(tool_ctx, source) + assert prepared is not None, f"{state.recipe_name}:{source}" + carriers = [ + carrier + for carrier in ( + prepared.success_carrier, + prepared.success_pull_only_carrier, + prepared.recovery_carrier, + ) + if carrier is not None + ] + _assert_carrier_closures_complete( + projection, + carriers, + recipe_name=state.recipe_name, + source=source, + step_segments=step_segments, + finalized_steps=finalized_steps, + automatic_sources=automatic_sources, + ) + _assert_success_routes_and_response_bounds( + projection, + prepared, + recipe_name=state.recipe_name, + source=source, + segment_indices=segment_indices, + result_arms_by_source=result_arms, + ) diff --git a/tests/server/test_response_backstop.py b/tests/server/test_response_backstop.py index 0723721441..09d494839d 100644 --- a/tests/server/test_response_backstop.py +++ b/tests/server/test_response_backstop.py @@ -197,6 +197,39 @@ def test_checkpoint_budget_spills_only_base_and_preserves_whole_carrier(tmp_path assert "recipe_segment" not in persisted_base +@pytest.mark.parametrize( + ("body_chars", "token_limit", "max_bytes"), + [(2700, None, 9999), (600, 1000, 2700)], + ids=("carrier-byte-reservation", "carrier-token-reservation"), +) +def test_checkpoint_budget_reserves_carrier_before_projecting_base( + tmp_path, body_chars, token_limit, max_bytes +): + carrier = _checkpoint_carrier() + carrier["bodies"][0]["body"] = "é" * body_chars + base = {"success": True, "stdout": "x" * 12_000} + config = OutputBudgetConfig() + + shaped = enforce_response_budget( + {**base, "recipe_segment": carrier}, + tool_name="test_check", + artifact_dir=tmp_path, + config=config, + selected_result_token_limit=token_limit, + ) + + assert isinstance(shaped, dict) + assert shaped["success"] is True + assert shaped["recipe_segment"] == carrier + assert ( + len(json.dumps(shaped, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) + <= max_bytes + ) + artifact = Path(shaped[RESPONSE_SPILL_METADATA_KEY]["artifact_path"]) + assert json.loads(artifact.read_text()) == base + assert config.response_max_bytes == OutputBudgetConfig().response_max_bytes + + def test_checkpoint_budget_publication_failure_returns_no_repeat_carrier(tmp_path, monkeypatch): from autoskillit.server.response import _response_budget