Rectify: Optional-Context Structured Skill Input Contract

src/autoskillit/cli/_prompts.py

@@ -169,7 +169,13 @@ def render(self) -> str:
                 "the unchanged recipe_pull. The overview is only a table of contents. "
                 "If subtype=recipe_segment_post_effect_delivery_failure, the operation "
                 "already ran; do not repeat it. Do not assume startup contains "
-                "full-horizon invocation_template_digests."
+                "full-horizon invocation_template_digests. For structured child inputs, "
+                "select "
+                "recipe_execution.skill_input_shapes[step_name], initialize skill_inputs "
+                "with exactly its ordered keys, and replace available values in place. "
+                "For unavailable context, copy only that key's advertised "
+                'unresolved_defaults entry by key presence, so "", 0, and False remain '
+                "verbatim; never delete or invent a key."
             ),
         ),
         _McpStartupRecoveryClause(

src/autoskillit/core/types/_type_constants.py

@@ -797,7 +797,10 @@ class SkillFamilyDef(NamedTuple):
     "an active recipe requires recipe_execution_id and invocation_template_digest; "
     "take both from the recipe_execution block of the complete_recipe_initialization "
     "receipt (bounded delivery) or of the open_kitchen response (inline delivery), "
-    "using invocation_template_digests[step_name] for this step"
+    "using invocation_template_digests[step_name] for this step; structured calls must "
+    "initialize skill_inputs from skill_input_shapes[step_name] ordered keys, replace "
+    "available values in place, copy only advertised unresolved_defaults by key presence "
+    'so "", 0, and False remain verbatim, and never delete or invent a key'
 )
 
 RECIPE_EXECUTION_INACTIVE_MESSAGE: str = (

src/autoskillit/core/types/_type_recipe_binding.py

@@ -197,15 +197,24 @@ class BoundValue:
     context_dependencies: tuple[str, ...] = ()
     input_dependencies: tuple[str, ...] = ()
     template_dependencies: tuple[str, ...] = ()
+    unresolved_default: BoundScalar | None = None
 
     def __post_init__(self) -> None:
         if not isinstance(self.state, BoundValueState):
             raise ValueError("BoundValue.state must be a BoundValueState")
         if not isinstance(self.origin, BoundValueOrigin):
             raise ValueError("BoundValue.origin must be a BoundValueOrigin")
+        if self.unresolved_default is not None and type(self.unresolved_default) not in (
+            str,
+            int,
+            bool,
+        ):
+            raise ValueError("BoundValue.unresolved_default must be a strict scalar or None")
         declared_absent = isinstance(self.declared_value, AbsentBoundValue)
         effective_absent = isinstance(self.effective_value, AbsentBoundValue)
         if self.state is BoundValueState.ABSENT:
+            if self.unresolved_default is not None:
+                raise ValueError("absent bound values cannot declare unresolved_default")
             if (
                 not declared_absent
                 or not effective_absent

src/autoskillit/core/types/_type_recipe_execution.py

@@ -14,7 +14,7 @@
 from enum import StrEnum
 from pathlib import Path
 from types import MappingProxyType
-from typing import Any, Protocol, runtime_checkable
+from typing import Any, Protocol, TypedDict, runtime_checkable
 
 from ..closure_hashing import HASH_RE, compute_canonical_hash
 from ._type_audit_admission import InstallationVersion
@@ -80,6 +80,7 @@ def _bound_value_payload(value: BoundValue) -> dict[str, object]:
         "origin": value.origin.value,
         "state": value.state.value,
         "template_dependencies": list(value.template_dependencies),
+        "unresolved_default": value.unresolved_default,
     }
 
 
@@ -209,18 +210,31 @@ def template_digests(self) -> Mapping[str, str]:
 )
 
 
+class _SkillInputShape(TypedDict):
+    keys: list[str]
+    unresolved_defaults: dict[str, BoundScalar]
+
+
 @dataclass(frozen=True, slots=True)
 class RecipeExecutionCredential:
     """The caller-visible identity of one installed recipe execution."""
 
     execution_id: str
     snapshot_digest: str
     invocation_template_digests: Mapping[str, str]
+    skill_input_shapes: Mapping[str, _SkillInputShape]
 
     def as_wire_block(self) -> dict[str, Any]:
         return {
             "execution_id": self.execution_id,
             "invocation_template_digests": dict(self.invocation_template_digests),
+            "skill_input_shapes": {
+                step_name: {
+                    "keys": list(shape["keys"]),
+                    "unresolved_defaults": dict(shape["unresolved_defaults"]),
+                }
+                for step_name, shape in self.skill_input_shapes.items()
+            },
             "snapshot_digest": self.snapshot_digest,
         }
 
@@ -234,10 +248,22 @@ def build_recipe_execution_credential(
     snapshot: RecipeExecutionSnapshot,
 ) -> RecipeExecutionCredential:
     """Project the sole caller-visible credential for an execution snapshot."""
+    skill_input_shapes: dict[str, _SkillInputShape] = {}
+    for step_name, template in snapshot.templates.items():
+        present = tuple(value for value in template.invocation.skill_inputs if value.is_present)
+        skill_input_shapes[step_name] = {
+            "keys": [value.name for value in present],
+            "unresolved_defaults": {
+                value.name: value.unresolved_default
+                for value in present
+                if value.unresolved_default is not None
+            },
+        }
     return RecipeExecutionCredential(
         execution_id=snapshot.execution_id,
         snapshot_digest=snapshot.snapshot_digest,
         invocation_template_digests=dict(snapshot.template_digests),
+        skill_input_shapes=skill_input_shapes,
     )
 
 

src/autoskillit/recipe/_binding.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+from dataclasses import replace
 from typing import Any, Final, TypeGuard
 
 import regex as re
@@ -453,6 +454,7 @@ def _structured_skill_inputs(
     contract: SkillContract,
     hidden_inputs: frozenset[str],
     ingredient_values: Mapping[str, BoundScalar],
+    optional_context_refs: frozenset[str],
 ) -> tuple[tuple[BoundValue, ...], tuple[BindingFailure, ...]]:
     input_by_name = {input_def.name: input_def for input_def in contract.inputs}
     failures: list[BindingFailure] = []
@@ -501,6 +503,11 @@ def _structured_skill_inputs(
             ingredient_values=ingredient_values,
         )
         value = _bound_value(input_def.name, declared, resolved)
+        if (
+            not input_def.required
+            and frozenset(value.context_dependencies) & optional_context_refs
+        ):
+            value = replace(value, unresolved_default=input_def.unresolved_default)
         bound.append(value)
         unresolved = bool(value.context_dependencies or value.input_dependencies)
         if not unresolved and not _skill_value_is_valid(resolved, input_def):
@@ -783,6 +790,7 @@ def bind_step_invocation(
                 contract=contract,
                 hidden_inputs=hidden_inputs,
                 ingredient_values=ingredients,
+                optional_context_refs=frozenset(step.optional_context_refs),
             )
             failures.extend(skill_failures)
     else:

src/autoskillit/recipe/_contracts_card.py

@@ -157,12 +157,17 @@ def generate_recipe_card(
                     skill_entry: dict[str, Any] = {
                         "inputs": [
                             {
-                                "name": i.name,
-                                "type": i.type,
-                                "required": i.required,
-                                "recommended": i.recommended,
+                                "name": item.name,
+                                "type": item.type,
+                                "required": item.required,
+                                "recommended": item.recommended,
+                                **(
+                                    {"unresolved_default": item.unresolved_default}
+                                    if item.unresolved_default is not None
+                                    else {}
+                                ),
                             }
-                            for i in contract.inputs
+                            for item in contract.inputs
                         ],
                         "outputs": [{"name": o.name, "type": o.type} for o in contract.outputs],
                         "expected_output_patterns": contract.expected_output_patterns,

src/autoskillit/recipe/_contracts_manifest.py

@@ -11,6 +11,7 @@
 
 from autoskillit.core import (
     VALID_INPUT_SPEC_TYPES,
+    BoundScalar,
     InputSpec,
     InputSpecType,
     get_logger,
@@ -48,22 +49,45 @@ def load_bundled_manifest() -> dict[str, Any]:
     return _MANIFEST_CACHE.get_or_load(manifest_path, load_yaml)
 
 
+def _parse_skill_input(skill_name: str, raw: Mapping[str, Any]) -> SkillInput:
+    input_def = SkillInput(
+        name=raw["name"],
+        type=raw["type"],
+        # Skill inputs default to optional — skills are permissive by design.
+        required=raw.get("required", False),
+        recommended=raw.get("recommended", False),
+    )
+    if "unresolved_default" not in raw:
+        return input_def
+    unresolved_default = raw["unresolved_default"]
+    if type(unresolved_default) not in (str, int, bool):
+        raise ValueError(
+            f"unresolved_default for skill '{skill_name}' input "
+            f"'{input_def.name}' must be a strict string, integer, or boolean"
+        )
+    if input_def.required:
+        raise ValueError(
+            f"required input '{input_def.name}' for skill '{skill_name}' "
+            "cannot declare unresolved_default"
+        )
+    if not input_def.accepts(unresolved_default):
+        raise ValueError(
+            f"unresolved_default for skill '{skill_name}' input "
+            f"'{input_def.name}' does not satisfy type '{input_def.type}'"
+        )
+    return dataclasses.replace(
+        input_def,
+        unresolved_default=cast(BoundScalar, unresolved_default),
+    )
+
+
 def get_skill_contract(skill_name: str, manifest: dict[str, Any]) -> SkillContract | None:
     """Look up a skill in the manifest and return a SkillContract."""
     skills = manifest.get("skills", {})
     skill_data = skills.get(skill_name)
     if skill_data is None:
         return None
-    inputs = tuple(
-        SkillInput(
-            name=inp["name"],
-            type=inp["type"],
-            # Skill inputs default to optional — skills are permissive by design.
-            required=inp.get("required", False),
-            recommended=inp.get("recommended", False),
-        )
-        for inp in skill_data.get("inputs", [])
-    )
+    inputs = tuple(_parse_skill_input(skill_name, inp) for inp in skill_data.get("inputs", []))
     outputs = [
         SkillOutput(
             name=out["name"],
@@ -294,6 +318,7 @@ def compute_skill_contract_identity(
                     "nullable": item.nullable,
                     "required": item.required,
                     "type": item.type,
+                    "unresolved_default": item.unresolved_default,
                 }
                 for item in contract.inputs
             ],

src/autoskillit/recipe/_contracts_types.py

@@ -7,7 +7,7 @@
 
 import regex as re
 
-from autoskillit.core import PreflightKind
+from autoskillit.core import BoundScalar, PreflightKind
 
 _CONTEXT_REF_RE = re.compile(r"\$\{\{\s*context\.(\w+)\s*\}\}")
 INPUT_REF_RE = re.compile(r"\$\{\{\s*inputs\.(\w+)\s*\}\}")
@@ -22,6 +22,15 @@ class SkillInput:
     required: bool
     recommended: bool = False
     nullable: bool = True
+    unresolved_default: BoundScalar | None = None
+
+    def __post_init__(self) -> None:
+        if self.unresolved_default is not None and type(self.unresolved_default) not in (
+            str,
+            int,
+            bool,
+        ):
+            raise ValueError("SkillInput.unresolved_default must be a strict scalar or None")
 
     def accepts(self, value: object) -> bool:
         normalized = self.type

src/autoskillit/recipe/rules/dataflow/rules_dataflow_callable.py

@@ -4,14 +4,22 @@
 
 import importlib
 import inspect
+from collections.abc import Mapping
 
 import regex as re
 
-from autoskillit.core import RUN_PYTHON_SENTINEL_KEYS, SKILL_TOOLS, Severity, get_logger
+from autoskillit.core import (
+    RUN_PYTHON_SENTINEL_KEYS,
+    SKILL_TOOLS,
+    Severity,
+    get_logger,
+    resolve_skill_name,
+)
 from autoskillit.recipe._analysis import ValidationContext
 from autoskillit.recipe.contracts import (
     _CONTEXT_REF_RE,
     get_callable_contract,
+    get_skill_contract,
     load_bundled_manifest,
 )
 from autoskillit.recipe.registry import RuleFinding, make_finding, semantic_rule
@@ -186,15 +194,60 @@ def _check_downstream_context_completeness(ctx: ValidationContext) -> list[RuleF
 @semantic_rule(
     name="nullable-optional-context-ref",
     description=(
-        "run_python steps must not pass optional_context_refs values to non-nullable "
-        "callable inputs without null coercion in the callable"
+        "run_python nullable inputs and run_skill unresolved defaults must safely cover "
+        "values declared in optional_context_refs"
     ),
     severity=Severity.ERROR,
 )
 def _check_nullable_optional_context_ref(ctx: ValidationContext) -> list[RuleFinding]:
     findings = []
     manifest = load_bundled_manifest()
     for step_name, step in ctx.recipe.steps.items():
+        optional_refs = set(step.optional_context_refs)
+        if not optional_refs:
+            continue
+        if step.tool == "run_skill":
+            skill_inputs = step.with_args.get("skill_inputs")
+            skill_command = step.with_args.get("skill_command", "")
+            if not isinstance(skill_inputs, Mapping) or not isinstance(skill_command, str):
+                continue
+            skill_name = resolve_skill_name(skill_command)
+            if not skill_name:
+                continue
+            contract = get_skill_contract(skill_name, manifest)
+            if contract is None:
+                continue
+            input_by_name = {item.name: item for item in contract.inputs}
+            for input_name, value in skill_inputs.items():
+                input_def = input_by_name.get(input_name)
+                if input_def is None or not isinstance(value, str):
+                    continue
+                dependencies = optional_refs.intersection(_CONTEXT_REF_RE.findall(value))
+                if not dependencies:
+                    continue
+                dependency = sorted(dependencies)[0]
+                if input_def.required:
+                    message = (
+                        f"Step '{step_name}' required input '{input_name}' for skill "
+                        f"'{skill_name}' depends on optional context ref '{dependency}'. "
+                        "Capture or gate the value before dispatch."
+                    )
+                elif input_def.unresolved_default is None:
+                    message = (
+                        f"Step '{step_name}' optional input '{input_name}' for skill "
+                        f"'{skill_name}' depends on optional context ref '{dependency}' "
+                        "without a contract unresolved_default."
+                    )
+                else:
+                    continue
+                findings.append(
+                    make_finding(
+                        rule_name="nullable-optional-context-ref",
+                        step_name=step_name,
+                        message=message,
+                    )
+                )
+            continue
         if step.tool != "run_python":
             continue
         callable_path = step.with_args.get("callable", "")
@@ -206,9 +259,6 @@ def _check_nullable_optional_context_ref(ctx: ValidationContext) -> list[RuleFin
         non_nullable = {inp.name for inp in contract.inputs if not inp.nullable}
         if not non_nullable:
             continue
-        optional_refs = set(step.optional_context_refs)
-        if not optional_refs:
-            continue
         args_values = _get_args_values(step.with_args)
         for inp_name in sorted(non_nullable):
             arg_val = args_values.get(inp_name)