fix: deliver checkpoint continuations from registry success routes

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

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(),

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:

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",

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<wrapped>\$\{\{\s*)?result\.(?P<field>[A-Za-z_]\w*)"
+    r"(?(wrapped)\s*\}\})\s*==\s*"
+    r"(?P<quote>['\"]?)(?P<value>[\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.

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:

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
 

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",

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",

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."