eager session-id persistence via on_session_id_resolved callback

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,

tests/contracts/test_protocol_satisfaction.py

@@ -410,6 +410,7 @@ def test_req_api_001_public_params_present(self):
             "stream_parser",
             "inspector_callback",
             "workload_basenames",
+            "on_session_id_resolved",
         }
         assert expected == public_params, (
             f"run_managed_async public params changed.\n"
@@ -481,6 +482,7 @@ def test_req_api_002_call_params_match_protocol(self):
             "session_record_types",
             "inspector_callback",
             "workload_basenames",
+            "on_session_id_resolved",
         }
         assert expected == actual, (
             f"DefaultSubprocessRunner.__call__ params changed.\n"

tests/execution/AGENTS.md

@@ -87,6 +87,7 @@ Subprocess integration, headless session, process lifecycle, and session result
 | `test_process_session_log_monitor.py` | Unit tests for _session_log_monitor — core detection + session ID + wiring |
 | `test_process_session_log_monitor_dispatch_marker.py` | Dispatch marker suppression gate tests for _session_log_monitor |
 | `test_process_session_log_monitor_stale_suppression.py` | TCP/CPU stale suppression gate and bounded suppression tests |
+| `test_process_session_id_callback.py` | Tests for the on_session_id_resolved callback fired from inside the task group |
 | `test_process_submodules.py` | Tests verifying process.py decomposition into focused sub-modules (P8-2) |
 | `test_provider_outcome_container.py` | Tests for ProviderOutcome typed container construction — required fields and TypeError on omission |
 | `test_push_trigger_applies.py` | Unit tests for _push_trigger_applies_to_branch and _has_merge_group_trigger |

tests/execution/test_process_session_id_callback.py

@@ -0,0 +1,82 @@
+"""Tests for the on_session_id_resolved callback fired from inside the task group."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+pytestmark = [pytest.mark.layer("execution"), pytest.mark.medium, pytest.mark.feature("execution")]
+
+
+class TestOnSessionIdResolvedCallback:
+    """on_session_id_resolved callback fires during live subprocess execution."""
+
+    @pytest.mark.anyio
+    async def test_callback_fires_with_resolved_session_id(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """When stdout emits a system init record, on_session_id_resolved fires."""
+        import anyio
+
+        from autoskillit.execution.process._process_race import (
+            RaceAccumulator,
+            _extract_stdout_session_id,
+        )
+
+        stdout_path = tmp_path / "stdout.jsonl"
+        session_id = "sess-callback-123"
+        init_record: dict[str, Any] = {
+            "type": "system",
+            "subtype": "init",
+            "session_id": session_id,
+        }
+        stdout_path.write_text(json.dumps(init_record) + "\n")
+
+        acc = RaceAccumulator()
+        ready = anyio.Event()
+        captured: list[str] = []
+
+        def on_sid(sid: str) -> None:
+            captured.append(sid)
+
+        await _extract_stdout_session_id(
+            stdout_path,
+            acc,
+            ready,
+            on_session_id_resolved=on_sid,
+        )
+
+        assert ready.is_set()
+        assert acc.stdout_session_id == session_id
+        assert captured == [session_id]
+
+    @pytest.mark.anyio
+    async def test_callback_not_called_when_no_session_id_found(self, tmp_path: Path) -> None:
+        """When no init record is present, callback is never invoked."""
+        import anyio
+
+        from autoskillit.execution.process._process_race import (
+            RaceAccumulator,
+            _extract_stdout_session_id,
+        )
+
+        stdout_path = tmp_path / "stdout.jsonl"
+        stdout_path.write_text("")
+
+        acc = RaceAccumulator()
+        ready = anyio.Event()
+        captured: list[str] = []
+
+        await _extract_stdout_session_id(
+            stdout_path,
+            acc,
+            ready,
+            on_session_id_resolved=lambda sid: captured.append(sid),
+            _timeout=0.5,
+        )
+
+        assert not acc.stdout_session_id
+        assert captured == []

tests/fakes.py

@@ -132,6 +132,7 @@ class DispatchFoodTruckCall:
     marker_dir: Path | None = None
     session_id: str | None = None
     resume_message: str | None = None
+    on_session_id_resolved: Callable[[str], None] | None = None
 
 
 _DEFAULT_SKILL_RESULT = SkillResult(
@@ -282,6 +283,7 @@ async def dispatch_food_truck(
         marker_dir: Path | None = None,
         session_id: str | None = None,
         resume_message: str | None = None,
+        on_session_id_resolved: Callable[[str], None] | None = None,
     ) -> SkillResult:
         self.dispatch_calls.append(
             DispatchFoodTruckCall(
@@ -312,6 +314,7 @@ async def dispatch_food_truck(
                 marker_dir=marker_dir,
                 session_id=session_id,
                 resume_message=resume_message,
+                on_session_id_resolved=on_session_id_resolved,
             )
         )
         if self._queue: