fix: give kitchen identity one lifecycle owner

scripts/e2e/README.md

@@ -254,10 +254,7 @@ Recipe `expected_failures` rows identify an individual, verified product defect
 accepted; process timeouts, malformed results, other failures and cleanup failures remain
 fatal. An accepted defect is provisional until mandatory branch/PR cleanup succeeds. A clean
 run that observes a configured row makes it stale and fails with instructions to remove the
-row and close the bug. The smoke catalog currently records
-[#5250](https://github.com/TalonT-Org/AutoSkillit/issues/5250) for `open_kitchen_failed`
-during kitchen initialization, before the recipe executes. This row does not accept
-arbitrary failures from the recipe's `failed` stop step. Fleet translates the L3 result's
+row and close the bug. Fleet translates the L3 result's
 `reason` into the failed command envelope's `error` and `user_visible_message` fields;
 the harness matches that envelope rather than the recipe payload's `failed_step`/`reason`.
 

scripts/e2e/catalog.json

@@ -69,15 +69,7 @@
       "recipe": "sandbox-smoke",
       "recipe_fixture": "sandbox-smoke.yaml",
       "ingredients": {},
-      "expected_pull_request_state": "closed",
-      "expected_failures": [
-        {
-          "severity": "error",
-          "check": "open_kitchen_failed",
-          "message": "open_kitchen_failed",
-          "issue": "https://github.com/TalonT-Org/AutoSkillit/issues/5250"
-        }
-      ]
+      "expected_pull_request_state": "closed"
     }
   ]
 }

scripts/e2e/e2e_harness.py

@@ -1029,13 +1029,15 @@ def run_test(
     """Run *test* and always record ``out/result.json``; return the failures."""
     out.mkdir(parents=True, exist_ok=True)
     matched_findings: list[dict[str, str]] = []
+    result: dict[str, Any] = {"test": test.name}
     try:
         failures, matched_findings = _run_test(
             test, catalog, out=out, home=home, env=env, runner=runner
         )
     except Exception as exc:
-        failures = [f"harness error: {exc}"]
-    result = {"test": test.name, "passed": not failures, "failures": failures}
+        failures = [f"harness error: {type(exc).__name__}: {exc}"]
+        result["exception"] = traceback.format_exc()
+    result.update(passed=not failures, failures=failures)
     if test.kind in ("clean-install", "recipe"):
         outcome = "failed" if failures else "expected_failure" if matched_findings else "passed"
         result.update(

src/autoskillit/cli/fleet/_fleet_run.py

@@ -1,4 +1,10 @@
-"""Headless one-shot fleet dispatch command."""
+"""Headless one-shot fleet dispatch command.
+
+A fresh invocation inherits a nonempty ambient campaign id or mints a fresh id. A paired
+dispatch resume preserves the campaign id recorded in its prior state. The resolved parent
+id is forwarded through dispatch arguments and the child's
+AUTOSKILLIT_CAMPAIGN_ID environment value.
+"""
 
 from __future__ import annotations
 
@@ -15,7 +21,7 @@
 if TYPE_CHECKING:
     from autoskillit.config import AutomationConfig
     from autoskillit.core import NativeShellCaptureMode
-    from autoskillit.fleet import DispatchResult
+    from autoskillit.fleet import CampaignState, DispatchResult
 
 logger = get_logger(__name__)
 
@@ -65,7 +71,7 @@ async def _execute_fleet_run(
         new_managed_launch_id,
     )
     from autoskillit.fleet import _build_food_truck_prompt, execute_dispatch
-    from autoskillit.server import make_context
+    from autoskillit.server import establish_kitchen_identity, make_context
     from autoskillit.workspace import (
         SkillProjectionContext,
         compile_session_skill_catalog,
@@ -77,6 +83,19 @@ async def _execute_fleet_run(
         project_dir=Path.cwd(),
         plugin_retirement_coordinator=default_plugin_retirement_coordinator(),
     )
+    prior_state: CampaignState | None = None
+    if resume_session_id is not None and prior_dispatch_id is not None:
+        from autoskillit.fleet import read_state
+
+        prior_state = read_state(ctx.temp_dir / "dispatches" / f"{prior_dispatch_id}.json")
+    establish_kitchen_identity(
+        ctx,
+        campaign_id=(
+            prior_state.campaign_id
+            if prior_state is not None and prior_state.campaign_id
+            else None
+        ),
+    )
     if disable_quota_guard:
         ctx.config.quota_guard.enabled = False
 
@@ -100,17 +119,13 @@ async def _execute_fleet_run(
         from autoskillit.server.managed_join_prelaunch import acquire_managed_join_evidence
 
         managed_join_parent_id = new_managed_launch_id()
-        if resume_session_id is not None and prior_dispatch_id is not None:
-            from autoskillit.fleet import read_state
-
-            prior_state = read_state(ctx.temp_dir / "dispatches" / f"{prior_dispatch_id}.json")
-            if prior_state is not None:
-                prior_record = next(
-                    (record for record in prior_state.dispatches if record.name == recipe),
-                    None,
-                )
-                if prior_record is not None and prior_record.managed_lineage_ref is not None:
-                    managed_join_parent_id = prior_record.managed_lineage_ref.launch_id
+        if prior_state is not None:
+            prior_record = next(
+                (record for record in prior_state.dispatches if record.name == recipe),
+                None,
+            )
+            if prior_record is not None and prior_record.managed_lineage_ref is not None:
+                managed_join_parent_id = prior_record.managed_lineage_ref.launch_id
         evidence = acquire_managed_join_evidence(
             backend=effective_backend,
             configured_model=cfg.model.model_override or cfg.model.default_model,

src/autoskillit/core/runtime/kitchen_state.py

@@ -162,11 +162,14 @@ def sweep_stale_markers(ttl_hours: int = 24) -> int:
 
 
 def resolve_kitchen_id() -> str:
-    """Resolve kitchen identity: inherit campaign ID from environment, or mint fresh.
+    """Return a stable kitchen identity for this process.
 
-    This is the sole mint source used by bootstrap/request transition construction.
-    Later ``ToolContext.kitchen_id`` assignments consume the stored transition value;
-    request handling must never call this function after bootstrap identity exists.
+    It is called only by ``establish_kitchen_identity`` (``autoskillit.server``),
+    the sole owner of the ``CLOSED -> INFRASTRUCTURE_READY`` edge, when no
+    explicit resume campaign id was supplied. All other code reads identity
+    through ``ToolContext.kitchen_id``.
+
+    Empty values are treated as unset per OpenTelemetry's documented convention.
     """
     return os.environ.get("AUTOSKILLIT_CAMPAIGN_ID") or str(uuid4())
 

src/autoskillit/core/types/protocols/_type_protocols_infra.py

@@ -185,7 +185,7 @@ def timeout(self) -> float | None: ...
 
 @runtime_checkable
 class KitchenTransitionLock(Protocol):
-    """Synchronous lock protecting one ToolContext kitchen transition snapshot."""
+    """Reentrant synchronous lock protecting one ToolContext kitchen transition snapshot."""
 
     def __enter__(self) -> Any: ...
 

src/autoskillit/pipeline/context.py

@@ -114,7 +114,9 @@ class TerminalRecipeResponseCacheEntry:
 
 
 class _KitchenIdentityContext(Protocol):
-    kitchen_id: str
+    @property
+    def kitchen_id(self) -> str: ...
+
     kitchen_process_identity: KitchenProcessIdentity | None
     project_dir: Path
 
@@ -196,8 +198,8 @@ class ToolContext:
                           context accounting and recovery service.
     workspace_outcome_ledger: WorkspaceOutcomeLedger — durable shared authority for
                           terminal workspace outcomes across parent and child sessions.
-    kitchen_id:           UUID string assigned when open_kitchen fires; scopes token telemetry
-                          to the current kitchen session lifetime.
+    kitchen_id:           Read-only identity derived from kitchen_open_state; set only through
+                          establish_kitchen_identity and empty while the state is CLOSED.
     kitchen_open_state:   Immutable process-local open-operation lifecycle and effect journal
                           retained for the ToolContext lifetime. Fleet dispatch provenance
                           instead uses request/task-scoped ContextVars; the different owners
@@ -293,7 +295,6 @@ class ToolContext:
     recipe_composite_hash: str = field(default="")
     recipe_version: str = field(default="")
     gate_infrastructure_ready: bool = field(default=False)
-    kitchen_id: str = field(default="")
     kitchen_process_identity: KitchenProcessIdentity | None = field(default_factory=lambda: None)
     kitchen_tracker_key: TrackerParticipantKey | None = field(default_factory=lambda: None)
     tracker_leases: dict[TrackerParticipantKey, ArtifactLease] = field(
@@ -408,6 +409,11 @@ def __post_init__(self) -> None:
         if self.background is None:
             self.background = DefaultBackgroundSupervisor(audit=self.audit)
 
+    @property
+    def kitchen_id(self) -> str:
+        """Return the identity derived from the current lifecycle state."""
+        return self.kitchen_open_state.kitchen_id
+
     @property
     def host_client_attestation(self) -> HostClientAttestation | None:
         """Return the launcher attestation snapshot owned by this context."""

src/autoskillit/server/__init__.py

@@ -38,6 +38,9 @@
     ToolContext,
     gate_error_result,
 )
+from autoskillit.server.lifecycle._kitchen_identity import (
+    establish_kitchen_identity,
+)
 from autoskillit.server.lifecycle._lifespan import (
     _autoskillit_lifespan,
 )
@@ -59,6 +62,7 @@
     # Public utilities consumed by CLI and tests
     "version_info",
     "make_context",
+    "establish_kitchen_identity",
     "resolve_backend_override",
     # Admission helpers consumed by CLI (re-exported to avoid cross-submodule imports)
     "_compute_effective_backend_map",

src/autoskillit/server/lifecycle/AGENTS.md

@@ -17,6 +17,11 @@ boot sequence — the async context manager wired via `lifespan=`, per-session-t
 auto-gate boots, and one-shot startup checks — lives in the `_lifespan/`
 sub-package.
 
+`_kitchen_identity` is the sole kitchen-identity mint and activation authority.
+This is application lifecycle state: `ToolContext.kitchen_id` is derived from
+`kitchen_open_state`, while the remaining context fields stay mutable. It is
+separate from FastMCP transport/session identity.
+
 The pre-deletion editable-install scan that halts `perform_merge()` before a
 worktree is deleted lives at `server/_editable_guard.py`, one level up, not
 here: it is a merge-time filesystem check with a single call site in

src/autoskillit/server/lifecycle/_kitchen_identity.py

@@ -0,0 +1,33 @@
+"""Sole owner of the kitchen-identity CLOSED -> INFRASTRUCTURE_READY edge."""
+
+from autoskillit.core import resolve_kitchen_id
+from autoskillit.pipeline import (
+    KitchenOpenPhase,
+    ToolContext,
+    new_kitchen_open_state,
+)
+
+
+def establish_kitchen_identity(ctx: ToolContext, *, campaign_id: str | None = None) -> str:
+    """Return the lifecycle kitchen id, minting and activating it at most once.
+
+    Callers may already hold the context's reentrant transition lock.
+    """
+    with ctx.kitchen_transition_lock:
+        state = ctx.kitchen_open_state
+        if state.phase is KitchenOpenPhase.CLOSED:
+            from autoskillit.server.recipe._recipe_generation import (  # circular-break
+                activate_kitchen,
+            )
+
+            kitchen_id = campaign_id or resolve_kitchen_id()
+            if not kitchen_id.strip():
+                raise ValueError("kitchen_id must contain non-whitespace characters")
+            activate_kitchen(kitchen_id)
+            ctx.kitchen_process_identity = None
+            state = new_kitchen_open_state(
+                kitchen_id=kitchen_id,
+                context_id=state.context_id,
+            )
+            ctx.kitchen_open_state = state
+        return state.kitchen_id

src/autoskillit/server/lifecycle/_lifespan/AGENTS.md

@@ -8,7 +8,7 @@ FastMCP lifespan boot sequence — relocated from `server/_lifespan/`
 `_startup_checks` runs one-shot synchronous startup checks
 (`run_startup_drift_check`, `run_startup_hook_health_check`,
 `run_startup_install_state_check`, `run_startup_fix_required_coverage_check`,
-`run_startup_join_guard_coverage_check`, `_activate_recipe_kitchen`,
+`run_startup_join_guard_coverage_check`,
 `_finalize_recorder`). `_session_boots`
 holds the per-session-type async auto-gate boots and the
 `_LIFESPAN_BOOT_REGISTRY` dispatch table that picks exactly one boot path

src/autoskillit/server/lifecycle/_lifespan/__init__.py

@@ -9,7 +9,6 @@
   (``run_startup_drift_check``, ``run_startup_hook_health_check``,
   ``run_startup_install_state_check``, ``run_startup_fix_required_coverage_check``,
   ``run_startup_join_guard_coverage_check``,
-  ``_activate_recipe_kitchen``,
   ``_finalize_recorder``).
 - :mod:`_session_boots` — per-session-type async auto-gate boots and the
   ``_LIFESPAN_BOOT_REGISTRY`` dispatch table.
@@ -27,7 +26,6 @@
     _collect_disabled_feature_tags,
     cleanup_readiness_sentinel,
     register_active_kitchen,
-    resolve_kitchen_id,
     write_readiness_sentinel,
 )
 from autoskillit.execution import (
@@ -65,7 +63,6 @@
     _skill_auto_gate_boot,
 )
 from autoskillit.server.lifecycle._lifespan._startup_checks import (
-    _activate_recipe_kitchen,
     _finalize_recorder,
     run_startup_drift_check,
     run_startup_fix_required_coverage_check,
@@ -80,7 +77,6 @@
 
 __all__ = [
     "_LIFESPAN_BOOT_REGISTRY",
-    "_activate_recipe_kitchen",
     "_autoskillit_lifespan",
     "_cleanup_stale_loop",
     "_evidence_reader_auto_gate_boot",