fix: eager session-id persistence via on_session_id_resolved callback (#4072)

The dispatch system silently dropped dispatched_session_id (and related
identity fields) on CancelledError: only status/reason/ended_at were
written via mark_dispatch_interrupted(). The session ID was known in
acc.stdout_session_id/acc.channel_b_session_id inside the task group
but never persisted to state.

Add an on_session_id_resolved callback fired from _extract_stdout_session_id
inside the live-execution task group, mirroring the existing on_pid_resolved
pattern. The closure calls mark_dispatch_session_identity to eagerly persist
the session ID while the dispatch is still RUNNING.

Three layers of defense:
1. Eager callback writes to state during execution
2. CancelledError handler passes captured_session_id + marker_dir to
   mark_dispatch_interrupted (additive — only writes empty fields)
3. Normal completion path prefers eagerly-captured value over skill_result

mark_dispatch_interrupted now accepts optional dispatched_session_id,
session_chain, and dispatched_session_log_dir. New mark_dispatch_session_identity
function is idempotent and status-guarded (RUNNING only).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Trecek

src/autoskillit/core/types/_type_protocols_execution.py

@@ -109,6 +109,7 @@ async def dispatch_food_truck(
         session_id: str | None = None,
         resume_message: str | None = None,
         backend_override: str | None = None,
+        on_session_id_resolved: Callable[[str], None] | None = None,
     ) -> SkillResult: ...
 
 

src/autoskillit/core/types/_type_subprocess.py

@@ -175,4 +175,5 @@ def __call__(
         session_record_types: frozenset[str] = frozenset({"assistant"}),
         inspector_callback: InspectorCallback | None = None,
         workload_basenames: frozenset[str] | None = None,
+        on_session_id_resolved: Callable[[str], None] | None = None,
     ) -> Awaitable[SubprocessResult]: ...

src/autoskillit/execution/headless/__init__.py

@@ -380,6 +380,7 @@ async def dispatch_food_truck(
         session_id: str | None = None,
         resume_message: str | None = None,
         backend_override: str | None = None,
+        on_session_id_resolved: Callable[[str], None] | None = None,
     ) -> SkillResult:
         if self._ctx.backend is not None and not self._ctx.backend.capabilities.food_truck_capable:
             raise RuntimeError(
@@ -483,4 +484,5 @@ async def dispatch_food_truck(
             marker_dir=effective_marker_dir,
             session_id=session_id,
             model_identity=model_identity,
+            on_session_id_resolved=on_session_id_resolved,
         )

src/autoskillit/execution/headless/_headless_execute.py

@@ -111,6 +111,7 @@ async def _execute_claude_headless(
     model_identity: ModelIdentity = ModelIdentity.unknown(),
     inspector_eligible: bool = False,
     inspector_model: str = "",
+    on_session_id_resolved: Callable[[str], None] | None = None,
 ) -> SkillResult:
     """Shared subprocess execution for headless Claude sessions.
 
@@ -262,6 +263,7 @@ async def _execute_claude_headless(
                 max_extension_seconds=max_extension_seconds,
                 marker_dir=marker_dir,
                 session_id=session_id,
+                on_session_id_resolved=on_session_id_resolved,
                 stream_parser=_stream_parser,
                 completion_record_types=_step_backend.capabilities.completion_record_types,
                 session_record_types=_step_backend.capabilities.session_record_types,

src/autoskillit/execution/process/__init__.py

@@ -220,6 +220,7 @@ async def run_managed_async(
     stream_parser: StreamParser | None = None,
     inspector_callback: InspectorCallback | None = None,
     workload_basenames: frozenset[str] | None = None,
+    on_session_id_resolved: Callable[[str], None] | None = None,
 ) -> SubprocessResult:
     """Async subprocess execution with temp file I/O and process tree cleanup.
 
@@ -352,6 +353,7 @@ async def run_managed_async(
                         functools.partial(
                             _extract_stdout_session_id,
                             stream_parser=stream_parser,
+                            on_session_id_resolved=on_session_id_resolved,
                         ),
                         stdout_path,
                         acc,
@@ -632,6 +634,7 @@ async def __call__(
         session_record_types: frozenset[str] = frozenset({"assistant"}),
         inspector_callback: InspectorCallback | None = None,
         workload_basenames: frozenset[str] | None = None,
+        on_session_id_resolved: Callable[[str], None] | None = None,
     ) -> SubprocessResult:
         return await run_managed_async(
             cmd,
@@ -657,4 +660,5 @@ async def __call__(
             session_record_types=session_record_types,
             inspector_callback=inspector_callback,
             workload_basenames=workload_basenames,
+            on_session_id_resolved=on_session_id_resolved,
         )

src/autoskillit/execution/process/_process_race.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
 from dataclasses import asdict, dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, assert_never
@@ -366,6 +367,8 @@ async def _extract_stdout_session_id(
     stream_parser: StreamParser | None = None,
     _poll_interval: float = 0.3,
     _timeout: float = 10.0,
+    *,
+    on_session_id_resolved: Callable[[str], None] | None = None,
 ) -> None:
     """Extract session ID from stdout type=system record and deposit on accumulator.
 
@@ -413,6 +416,8 @@ async def _extract_stdout_session_id(
             if sid:
                 acc.stdout_session_id = sid
                 logger.debug("stdout_session_id_extracted", session_id=sid)
+                if on_session_id_resolved is not None:
+                    on_session_id_resolved(sid)
                 ready.set()
                 return
     logger.debug("stdout_session_id_extraction_timeout", timeout=_timeout)

src/autoskillit/execution/recording.py

@@ -140,6 +140,7 @@ async def __call__(
         session_record_types: frozenset[str] = frozenset({"assistant"}),
         inspector_callback: InspectorCallback | None = None,
         workload_basenames: frozenset[str] | None = None,
+        on_session_id_resolved: Callable[[str], None] | None = None,
     ) -> SubprocessResult:
         step_name = (env or {}).get(SCENARIO_STEP_NAME_ENV, "")
 
@@ -430,6 +431,7 @@ async def __call__(
         session_record_types: frozenset[str] = frozenset({"assistant"}),
         inspector_callback: InspectorCallback | None = None,
         workload_basenames: frozenset[str] | None = None,
+        on_session_id_resolved: Callable[[str], None] | None = None,
     ) -> SubprocessResult:
         step_name = (env or {}).get(SCENARIO_STEP_NAME_ENV, "")
 

src/autoskillit/fleet/__init__.py

@@ -61,6 +61,7 @@
     mark_dispatch_interrupted,
     mark_dispatch_resumable,
     mark_dispatch_running,
+    mark_dispatch_session_identity,
     normalize_dispatch_token_usage,
     read_all_campaign_captures,
     read_state,
@@ -146,6 +147,7 @@
     "record_gate_outcome",
     "mark_dispatch_interrupted",
     "mark_dispatch_resumable",
+    "mark_dispatch_session_identity",
     "mark_dispatch_running",
     "read_all_campaign_captures",
     "read_state",

src/autoskillit/fleet/_api.py

@@ -629,6 +629,7 @@ def _reject_with_state(error_code: FleetErrorCode, message: str) -> DispatchResu
     _dispatched_ticks: list[int] = []
     _dispatched_create_time: list[float] = []
     _dispatched_boot_id: list[str] = []
+    _dispatched_session_id: list[str] = []
 
     # Collect prior dispatch_ids from attempt_history for defense-in-depth parsing
     prior_ids: list[str] = []
@@ -678,6 +679,14 @@ def _on_spawn(pid: int, ticks: int) -> None:
             issue_url=_issue_urls_raw,
         )
 
+    def _on_session_id(session_id: str) -> None:
+        from autoskillit.fleet.state import mark_dispatch_session_identity
+
+        _dispatched_session_id.append(session_id)
+        mark_dispatch_session_identity(
+            state_path, effective_name, dispatched_session_id=session_id
+        )
+
     marker_dir: Path | None = None
     if _locator is not None:
         try:
@@ -710,6 +719,7 @@ def _on_spawn(pid: int, ticks: int) -> None:
                     project_dir=str(tool_ctx.project_dir),
                     marker_dir=marker_dir,
                     session_id=caller_session_id,
+                    on_session_id_resolved=_on_session_id,
                     timeout=resolved_timeout,
                     idle_output_timeout=float(idle_output_timeout)
                     if idle_output_timeout is not None
@@ -733,11 +743,21 @@ def _on_spawn(pid: int, ticks: int) -> None:
             try:
                 from autoskillit.fleet.state import mark_dispatch_interrupted  # noqa: PLC0415
 
+                captured_session_id = _dispatched_session_id[0] if _dispatched_session_id else ""
+                if not captured_session_id:
+                    sr = locals().get("skill_result")
+                    if sr is not None:
+                        captured_session_id = getattr(sr, "session_id", "") or ""
+
                 with anyio.CancelScope(shield=True):
                     mark_dispatch_interrupted(
                         state_path,
                         effective_name,
                         reason="signal_induced_cancellation",
+                        dispatched_session_id=captured_session_id,
+                        dispatched_session_log_dir=str(marker_dir)
+                        if marker_dir is not None
+                        else "",
                     )
             except Exception:
                 logger.warning(
@@ -894,7 +914,9 @@ def _on_spawn(pid: int, ticks: int) -> None:
         dispatch_id=dispatch_id,
         campaign_id=campaign_id,
         caller_session_id=caller_session_id,
-        dispatched_session_id=skill_result.session_id,
+        dispatched_session_id=_dispatched_session_id[0]
+        if _dispatched_session_id
+        else skill_result.session_id,
         session_chain=extended_chain,
         dispatched_session_log_dir=project_log_dir,
         dispatched_pid=_dispatched_pid[0] if _dispatched_pid else 0,

src/autoskillit/fleet/state.py

@@ -82,6 +82,7 @@
     "read_state",
     "mark_dispatch_running",
     "mark_dispatch_interrupted",
+    "mark_dispatch_session_identity",
     "mark_dispatch_resumable",
     "reset_blocking_dispatch",
     "append_dispatch_record",
@@ -395,8 +396,17 @@ def mark_dispatch_interrupted(
     dispatch_name: str,
     *,
     reason: str,
+    dispatched_session_id: str = "",
+    session_chain: list[str] | None = None,
+    dispatched_session_log_dir: str = "",
 ) -> None:
-    """Atomically mark a dispatch as interrupted with a reason."""
+    """Atomically mark a dispatch as interrupted with a reason.
+
+    Identity fields (dispatched_session_id, session_chain, dispatched_session_log_dir)
+    are additive — they are only written if the corresponding field on the record is
+    still empty, preventing the cancellation handler from overwriting values already
+    eagerly persisted by the on_session_id_resolved callback.
+    """
     with CampaignStateMutator(state_path) as m:
         if m.state is None:
             raise FileNotFoundError(f"State file not found or corrupted: {state_path}")
@@ -406,12 +416,43 @@ def mark_dispatch_interrupted(
                 d.status = DispatchStatus.INTERRUPTED
                 d.reason = reason
                 d.ended_at = time.time()
+                if dispatched_session_id and not d.dispatched_session_id:
+                    d.dispatched_session_id = dispatched_session_id
+                if session_chain is not None and not d.session_chain:
+                    d.session_chain = session_chain
+                if dispatched_session_log_dir and not d.dispatched_session_log_dir:
+                    d.dispatched_session_log_dir = dispatched_session_log_dir
                 m.mark_dirty()
                 return
         else:
             raise ValueError(f"Dispatch '{dispatch_name}' not found in state")
 
 
+def mark_dispatch_session_identity(
+    state_path: Path,
+    dispatch_name: str,
+    *,
+    dispatched_session_id: str,
+) -> None:
+    """Eagerly persist the dispatched_session_id once it is discovered during execution.
+
+    Idempotent and status-guarded: only writes when the dispatch is RUNNING and
+    dispatched_session_id is still empty. If the dispatch has already transitioned
+    to INTERRUPTED (race condition), this is a no-op.
+    """
+    with CampaignStateMutator(state_path) as m:
+        if m.state is None:
+            raise FileNotFoundError(f"State file not found or corrupted: {state_path}")
+        for d in m.state.dispatches:
+            if d.name == dispatch_name:
+                if d.status == DispatchStatus.RUNNING and not d.dispatched_session_id:
+                    d.dispatched_session_id = dispatched_session_id
+                    m.mark_dirty()
+                return
+        else:
+            raise ValueError(f"Dispatch '{dispatch_name}' not found in state")
+
+
 def mark_dispatch_resumable(
     state_path: Path,
     dispatch_name: str,