eager session-id persistence via on_session_id_resolved callback (#4078)

The dispatch system has a structural weakness where runtime-discovered
identity fields (`dispatched_session_id`, `session_chain`,
`dispatched_session_log_dir`, `token_usage`, `resume_checkpoint`,
`branch_name`) are only written to state on the normal completion path
(`_api.py:927` via `upsert_dispatch_record_by_name`). On signal
interruption (`CancelledError`), only `status`, `reason`, and `ended_at`
are written via `mark_dispatch_interrupted()`. This leaves the manifest
non-resumable because downstream consumers (state recovery, fleet
session, label cleanup) rely on `dispatched_session_id` being non-empty.

Part A addresses the core architectural immunity: eager session-identity
persistence via an `on_session_id_resolved` callback fired from inside
the live-execution task group (mirroring the existing `on_pid_resolved`
pattern), `mark_dispatch_interrupted` signature extension for
defense-in-depth identity capture, and test coverage that makes this bug
class instantly caught. Part B (separate task) will cover R3
(`sessions.jsonl` fallback lookup), R4 (L3 orchestrator dispatch history
visibility), and the remaining deferred-write fields (`token_usage`,
`resume_checkpoint`, `branch_name`).

## Requirements

`dispatched_session_id` is recorded as `""` (empty string) in the
dispatch manifest when a signal interrupts the dispatch — even when the
headless session completed all pipeline steps successfully. This makes
the dispatch non-resumable and, combined with the stale `in-progress`
GitHub label, prevents the orchestrator from resuming the interrupted
session. Recovery requires manual `reset_dispatch(force=True)` followed
by a fresh dispatch, losing the ability to continue from where the
session left off.

### R1: Eager `dispatched_session_id` Write on Interruption

Extend `mark_dispatch_interrupted()` to accept an optional
`dispatched_session_id` parameter. In the `except
asyncio.CancelledError` handler at `_api.py:731`:

1. Guard against `NameError`: `skill_result` is assigned inside the
`try` block at line 698. If `CancelledError` fires BEFORE
`dispatch_food_truck()` returns, `skill_result` is never assigned. The
handler must use `skill_result = locals().get("skill_result")` or
pre-initialize `skill_result = None` before the `try` block.
2. If `skill_result` is available and has a non-empty `session_id`, pass
it to `mark_dispatch_interrupted()`.
3. Also capture `session_chain` — extend `mark_dispatch_interrupted()`
to accept an optional `session_chain` parameter, or add a separate
`set_dispatch_session_identity()` state mutation function.

This follows the same eager-write pattern used for PID capture via
`mark_dispatch_running()` and for `issue_url` capture added in PR #3828.

### R2: Intermediate Session ID Write at Process Layer

When Channel B or stdout extraction succeeds during the race monitoring
phase, write `dispatched_session_id` to the manifest immediately —
before `SubprocessResult` is fully constructed. This covers the case
where a signal arrives during the session (before
`dispatch_food_truck()` returns) but the session ID is already known.

**Architectural constraint:** The execution package is IL-1; fleet state
is IL-2. Direct imports from `execution` → `fleet` would violate import
layering. This must be implemented via a callback pattern:
`_run_dispatch` passes a `on_session_id_resolved: Callable[[str], None]`
callback down to the process runner, which invokes it when Channel B or
stdout extraction succeeds. The callback closes over `state_path` and
`effective_name` and calls a new state mutation function. This mirrors
the existing `on_pid_resolved` callback pattern at `_api.py:657-679`.

### R3: `sessions.jsonl` Fallback Lookup

When `dispatched_session_id` is empty on a resumable/interrupted
dispatch that has a non-empty `dispatch_id`, search `sessions.jsonl` for
entries matching that `dispatch_id` to recover the session_id.
Defense-in-depth, not a substitute for R1/R2.

### R4: L3 Orchestrator Dispatch History Visibility

Give the L3 fleet dispatch orchestrator read access to the history of
dispatches it launched and their current status. Currently, the L3 has
no mechanism to query dispatch state — it relies entirely on the return
value of `dispatch_food_truck`, which is lost on interruption. The
orchestrator should be able to:
- Query dispatches by `caller_session_id` (its own session) or
`campaign_id`
- See the current status, `dispatched_session_id`, `dispatched_pid`, and
`branch_name` for each dispatch
- Use this information to make informed resume/reset/skip decisions when
re-entering after a crash

This would eliminate the class of bugs where the L3 orchestrator has no
context about what happened during an interrupted dispatch and must rely
on indirect signals (labels, sidecar files) that may themselves be
stale.

### R5: Test Coverage

- Add assertion on `dispatched_session_id` in
`TestCancelledErrorRecordsInterruptedState`
- Add test for post-completion race window: `CancelledError` after
`dispatch_food_truck()` returns but before
`upsert_dispatch_record_by_name()`
- Add test verifying `session_chain` is also captured on the
interruption path
- Add test for the resume-blocked scenario: INTERRUPTED + empty
`dispatched_session_id` + stale `in-progress` label → verify
`reset_dispatch` recovery works

Closes #4072

## Implementation Plan

Plan file:
`/home/talon/projects/autoskillit-runs/remediation-20260611-155102-572761/.autoskillit/temp/rectify/rectify_dispatch_session_id_signal_persistence_2026-06-11_155102.md`

🤖 Generated with [Claude Code](https://claude.com/claude-code) via
AutoSkillit
<!-- autoskillit:pipeline-signature
steps=prepare_pr,run_arch_lenses,compose_pr,annotate_pr_diff,review_pr
-->

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,