fix: preserve Stop-hook continuation after Channel B completion

src/autoskillit/core/types/execution/_type_subprocess.py

@@ -34,18 +34,18 @@
 #:   stdout=whatever was flushed to the temp file before exit
 #:   Kill-anomaly possible when returncode==0, UNMONITORED, and stdout is success+empty,
 #:   empty_output, or unparseable → _is_kill_anomaly returns True.
-#:   When CHANNEL_A or CHANNEL_B: no kill anomaly; session completed.
+#:   Channel confirmation excludes this kill anomaly; content success is adjudicated separately.
 #:
 #: COMPLETED (Channel A):
 #:   channel_confirmation=CHANNEL_A (heartbeat confirmed type=result in stdout)
 #:   returncode=nonzero (SIGTERM/SIGKILL from async_kill_process_tree)
 #:   stdout=contains a complete type=result NDJSON record
 #:
-#: COMPLETED (Channel B, drain expired OR no heartbeat configured):
-#:   channel_confirmation=CHANNEL_B (session JSONL is sole authority)
-#:   returncode=nonzero (SIGTERM/SIGKILL)
-#:   stdout=may be empty (CLI not yet flushed type=result before kill)
-#:   _compute_success provenance bypass applies: return True immediately.
+#: Channel B corroboration (independent of termination):
+#:   a session JSONL marker never releases a live process.
+#:   It can corroborate an observed exit or remain recorded after an independent
+#:   timeout or infrastructure-abort cause ends the run.
+#:   It cannot replace native terminal or content evidence.
 #:
 #: STALE:
 #:   channel_confirmation=UNMONITORED (typical: stale monitor fired alone)
@@ -138,8 +138,8 @@ class SubprocessResult:
     """How completion was confirmed by the two-channel detection system.
 
     CHANNEL_A: heartbeat confirmed type=result in stdout; data availability guaranteed.
-    CHANNEL_B: session JSONL marker fired; drain expired or no heartbeat configured.
-               stdout may be empty — callers must trust JSONL signal, not stdout content.
+    CHANNEL_B: session JSONL marker corroboration; never releases a live process.
+               May corroborate an exit or remain recorded after timeout or infrastructure abort.
     UNMONITORED: no channel monitoring active (NATURAL_EXIT, STALE, TIMED_OUT, sync path).
     """
     proc_snapshots: list[dict[str, object]] | None = None

src/autoskillit/core/types/foundation/_type_enums.py

@@ -343,7 +343,7 @@ class TerminationAction(StrEnum):
     Produced by decide_termination_action (pure function) and consumed by
     execute_termination_action (the single authorized kill caller in process.py).
 
-    - NO_KILL: process already exited naturally; no kill needed.
+    - NO_KILL: leader exit observed; no kill needed.
     - DRAIN_THEN_KILL_IF_ALIVE: channel confirmed completion but process is still
       alive; wait up to grace_seconds for natural exit, then kill if still running.
     - IMMEDIATE_KILL: timeout, stall, or stale — kill without draining.
@@ -378,8 +378,8 @@ class ChannelConfirmation(StrEnum):
 
     Invariant (from process.py):
     - CHANNEL_A: heartbeat fired; stdout contains non-empty type=result record.
-    - CHANNEL_B: session JSONL marker fired; drain expired OR no heartbeat configured.
-      stdout may be empty. Downstream must not require stdout content.
+    - CHANNEL_B: session JSONL marker observed; never releases a live process;
+      typically corroborates a process exit.
     - UNMONITORED: no channel monitoring active (NATURAL_EXIT, STALE, TIMED_OUT,
       sync path, or heartbeat disabled with no Channel B win).
     - DIR_MISSING: session log directory did not exist when monitoring started.

src/autoskillit/execution/process/AGENTS.md

@@ -4,6 +4,9 @@ Subprocess lifecycle management — spawn, monitor, race, kill.
 
 ## Architecture Notes
 
+- A Channel B transcript marker is corroboration and session discovery only; it never releases a live race in any mode. Completion release requires Channel A native terminal evidence or a witnessed exit; timeout, cancellation and infrastructure-abort paths remain bounded.
+- `NO_KILL` requires an observed exit.
+
 **Two-channel completion detection:**
 
 - **Channel A** (stdout heartbeat): polls the subprocess stdout temp file for a `type=result` NDJSON record containing the completion marker. Guarantees stdout data is available.

src/autoskillit/execution/process/__init__.py

@@ -22,7 +22,6 @@
 import anyio
 
 from autoskillit.core import (
-    ChannelBStatus,
     ChannelConfirmation,
     RolloutMarker,
     SessionEvent,
@@ -573,11 +572,6 @@ def _observe_final_stdout(event: SessionEvent) -> None:
                     if event.has_marker and acc.channel_a_candidate_at is None:
                         acc.channel_a_candidate_at = anyio.current_time()
 
-                def _observe_final_channel_b(event: SessionEvent) -> None:
-                    acc.observe_event(event)
-                    if event.has_marker and acc.channel_b_candidate_at is None:
-                        acc.channel_b_candidate_at = anyio.current_time()
-
                 final_fold_complete = False
                 if acc.stdout_cursor is not None:
                     final_fold_complete = fold_event_cursor(
@@ -586,21 +580,16 @@ def _observe_final_channel_b(event: SessionEvent) -> None:
                 if acc.channel_b_cursor is not None:
                     final_fold_complete = (
                         fold_event_cursor(
-                            acc.channel_b_cursor, lifecycle_parser, _observe_final_channel_b
+                            acc.channel_b_cursor, lifecycle_parser, acc.observe_event
                         )
                         and final_fold_complete
                     )
                 acc.channel_a_confirmed = acc.channel_a_candidate_at is not None
-                if acc.channel_b_candidate_at is not None:
-                    acc.channel_b_status = ChannelBStatus.COMPLETION
                 acc.lifecycle_observation_complete = final_fold_complete
             if (
                 lifecycle_observation_enabled
                 and not acc.has_unresolved_obligations()
-                and (
-                    acc.channel_a_candidate_at is not None
-                    or acc.channel_b_candidate_at is not None
-                )
+                and acc.channel_a_candidate_at is not None
                 and not acc.process_exited
             ):
                 with anyio.move_on_after(0.05):

src/autoskillit/execution/process/_lifecycle/runner.py

@@ -10,6 +10,7 @@
 
 from autoskillit.core import (
     ChannelConfirmation,
+    KillReason,
     RolloutMarker,
     SubprocessResult,
     TerminationReason,
@@ -112,6 +113,11 @@ def run_managed_sync(
                 stdout=stdout,
                 stderr=stderr,
                 termination=termination,
+                kill_reason=(
+                    KillReason.INFRA_KILL
+                    if termination is TerminationReason.TIMED_OUT
+                    else KillReason.NATURAL_EXIT
+                ),
                 pid=process.pid,
                 process_group_id=process.pid,
                 channel_confirmation=ChannelConfirmation.UNMONITORED,

src/autoskillit/execution/process/_process_race.py

@@ -83,6 +83,9 @@ class RaceAccumulator:
     ``process_exited_event`` is set by _watch_process BEFORE setting trigger,
     so downstream code can await it as a race primitive. It is passed (by reference)
     to execute_termination_action to support the DRAIN_THEN_KILL_IF_ALIVE window.
+
+    ``completion_candidate_event`` is set only by Channel A native terminal evidence
+    in ``_watch_heartbeat``.
     """
 
     process_exited: bool = False
@@ -104,7 +107,6 @@ class RaceAccumulator:
     schedule_wakeup_violation: bool = False
     completion_ceiling_expired: bool = False
     channel_a_candidate_at: float | None = None
-    channel_b_candidate_at: float | None = None
     stdout_cursor: EventCursor | None = None
     channel_b_cursor: EventCursor | None = None
     completion_candidate_event: anyio.Event = field(default_factory=anyio.Event)
@@ -148,11 +150,7 @@ def deposit_session_log_result(
         if on_session_id_resolved is not None and monitor_result.session_id:
             on_session_id_resolved(monitor_result.session_id)
         channel_b_ready.set()
-        if monitor_result.status is ChannelBStatus.COMPLETION:
-            if self.lifecycle_observation_enabled:
-                self.channel_b_candidate_at = anyio.current_time()
-                self.completion_candidate_event.set()
-        else:
+        if monitor_result.status is not ChannelBStatus.COMPLETION:
             trigger.set()
 
     def to_race_signals(self) -> RaceSignals:

src/autoskillit/execution/process/_race_watchers.py

@@ -258,23 +258,21 @@ async def _watch_completion_eligibility(
     if session_log_enabled and acc.channel_b_cursor is None:
         with anyio.move_on_after(completion_drain_timeout):
             await channel_b_selected.wait()
-    started = min(
-        timestamp
-        for timestamp in (acc.channel_a_candidate_at, acc.channel_b_candidate_at)
-        if timestamp is not None
-    )
+    started = acc.channel_a_candidate_at
+    # Only _watch_heartbeat sets the event, immediately after the timestamp.
+    assert started is not None
     while not trigger.is_set():
         if acc.stdout_cursor is not None:
             fold_event_cursor(acc.stdout_cursor, stream_parser, acc.observe_event)
         if acc.channel_b_cursor is not None:
             fold_event_cursor(acc.channel_b_cursor, stream_parser, acc.observe_event)
         if not acc.has_unresolved_obligations():
-            acc.channel_a_confirmed = acc.channel_a_candidate_at is not None
+            acc.channel_a_confirmed = True
             trigger.set()
             return
         if anyio.current_time() - started >= child_deferral_ceiling:
             acc.completion_ceiling_expired = True
-            acc.channel_a_confirmed = acc.channel_a_candidate_at is not None
+            acc.channel_a_confirmed = True
             trigger.set()
             return
         await anyio.sleep(_poll_interval)

src/autoskillit/execution/process/_termination.py

@@ -45,10 +45,11 @@ def decide_termination_action(
 
     Priority:
     1. timeout_fired → IMMEDIATE_KILL (always overrides)
-    2. process_exited → NO_KILL (process already gone, no signal needed)
-    3. termination-reason dispatch:
+    2. process_exited with unresolved obligations or an expired ceiling → IMMEDIATE_KILL
+    3. process_exited → NO_KILL (leader exit observed, no signal needed)
+    4. live termination-reason dispatch:
        - COMPLETED: channel won but process alive → DRAIN_THEN_KILL_IF_ALIVE
-       - NATURAL_EXIT: fallback case → NO_KILL
+       - NATURAL_EXIT / SIGNAL_DEATH: unobserved exit → IMMEDIATE_KILL
        - IDLE_STALL / STALE / TIMED_OUT: infra kill → IMMEDIATE_KILL
 
     The function is deliberately free of anyio and I/O so it can be tested
@@ -64,7 +65,7 @@ def decide_termination_action(
         return TerminationAction.NO_KILL
     match termination:
         case TerminationReason.NATURAL_EXIT | TerminationReason.SIGNAL_DEATH:
-            return TerminationAction.NO_KILL
+            return TerminationAction.IMMEDIATE_KILL
         case TerminationReason.COMPLETED:
             return TerminationAction.DRAIN_THEN_KILL_IF_ALIVE
         case (
@@ -91,16 +92,20 @@ async def _drain_before_escalation(
 ) -> tuple[int | None, ProcessCleanupResult] | None:
     with anyio.move_on_after(grace_seconds):
         await process_exited_event.wait()
-    if owner.returncode is not None:
-        proc_log.debug("natural_exit_after_drain", returncode=owner.returncode)
+    returncode = await anyio.to_thread.run_sync(partial(owner.observe_exit, include_stopped=False))
+    if returncode is not None:
+        proc_log.debug("natural_exit_after_drain", returncode=returncode)
         return await anyio.to_thread.run_sync(owner.settle_evidence, abandon_on_cancel=False)
     # Child-liveness deferral: same pattern as _session_log_monitor stale-kill suppression
     if pid is not None and child_deferral_ceiling > 0:
         deferral_start = anyio.current_time()
         _poll_interval = 2.0
         while (anyio.current_time() - deferral_start) < child_deferral_ceiling:
-            if owner.returncode is not None:
-                proc_log.debug("natural_exit_during_deferral", returncode=owner.returncode)
+            returncode = await anyio.to_thread.run_sync(
+                partial(owner.observe_exit, include_stopped=False)
+            )
+            if returncode is not None:
+                proc_log.debug("natural_exit_during_deferral", returncode=returncode)
                 return await anyio.to_thread.run_sync(
                     owner.settle_evidence, abandon_on_cancel=False
                 )

src/autoskillit/execution/session/AGENTS.md

@@ -4,4 +4,4 @@ Session result processing — parse, validate content, compute retry, adjudicate
 
 ## Architecture Notes
 
-The sub-modules form a pipeline: parse -> check content -> compute retry -> compute outcome. `_exit_classification.py` provides a parallel infrastructure classification (context exhaustion, API error, process kill) used by `_headless_result.py` for resume routing. When Channel B is the sole confirmation source, `_compute_success` applies a provenance bypass — the session JSONL marker is treated as authoritative proof of success without requiring stdout content.
+The sub-modules form a pipeline: parse -> check content -> compute retry -> compute outcome. `_exit_classification.py` provides a parallel infrastructure classification (context exhaustion, API error, process kill) used by `_headless_result.py` for resume routing. Channel B is marker corroboration only; it cannot replace terminal or content evidence.

tests/_test_filter.py

@@ -1528,6 +1528,7 @@ class SourceMapRejection(enum.StrEnum):
             # file-level: exercises the hook-authored child-outcome snapshot module
             # directly (issue #4623) via autoskillit.hooks's public re-export.
             "execution/test_child_outcomes.py",
+            "execution/test_process_stop_hook_join.py",
             # hook_registry imports from autoskillit.hooks at module scope
             # (HOOK_REGISTRY, generate_hooks_json) for the registry init wiring.
             "hook_registry",

tests/arch/test_layer_enforcement.py

@@ -1989,6 +1989,10 @@ def test_default_classes_only_instantiated_inside_factory_or_allowlist() -> None
     # exercises the hook-authored child-outcome snapshot module directly
     # (issue #4623), via autoskillit.hooks's public re-export
     "tests/execution/test_child_outcomes.py": frozenset({"autoskillit.hooks"}),
+    # The managed runner regression invokes the real join Stop dispatcher and ledger.
+    "tests/execution/test_process_stop_hook_join.py": frozenset(
+        {"autoskillit.hooks", "autoskillit.hook_registry"}
+    ),
     "tests/execution/test_quota_io.py": frozenset({"autoskillit.config"}),
     "tests/execution/test_quota_sleep.py": frozenset({"autoskillit.hooks", "autoskillit.config"}),
     "tests/execution/test_quota_http.py": frozenset({"autoskillit.config"}),

tests/arch/test_watcher_signal_consistency.py

@@ -230,31 +230,57 @@ def test_completion_marker_watchers_do_not_trigger_lifecycle_completion_directly
         and call.func.attr == "set"
         for call in ast.walk(deposition)
     )
-    completion_branch = next(
+    assert not any(
+        isinstance(node, ast.Attribute)
+        and (node.attr == "completion_candidate_event" or node.attr.endswith("_candidate_at"))
+        for node in ast.walk(deposition)
+    )
+    trigger_calls = [
+        call
+        for call in ast.walk(deposition)
+        if isinstance(call, ast.Call) and _calls_trigger_set(call)
+    ]
+    assert len(trigger_calls) == 1
+    release_branch = next(
         node
         for node in deposition.body
-        if isinstance(node, ast.If)
-        and isinstance(node.test, ast.Compare)
-        and any(
-            isinstance(comparator, ast.Attribute) and comparator.attr == "COMPLETION"
-            for comparator in node.test.comparators
-        )
+        if isinstance(node, ast.If) and _calls_trigger_set(ast.Module(body=node.body))
     )
-    assert isinstance(completion_branch.test, ast.Compare)
-    assert isinstance(completion_branch.test.left, ast.Attribute)
-    assert isinstance(completion_branch.test.left.value, ast.Name)
-    assert completion_branch.test.left.value.id == "monitor_result"
-    assert completion_branch.test.left.attr == "status"
-    assert not _calls_trigger_set(ast.Module(body=completion_branch.body))
-    assert any(
-        isinstance(call, ast.Call)
-        and isinstance(call.func, ast.Attribute)
-        and isinstance(call.func.value, ast.Attribute)
-        and call.func.value.attr == "completion_candidate_event"
-        and call.func.attr == "set"
-        for call in ast.walk(ast.Module(body=completion_branch.body))
+    comparison = release_branch.test
+    assert isinstance(comparison, ast.Compare)
+    assert len(comparison.ops) == 1 and isinstance(comparison.ops[0], ast.IsNot)
+    assert isinstance(comparison.left, ast.Attribute)
+    assert isinstance(comparison.left.value, ast.Name)
+    assert comparison.left.value.id == "monitor_result"
+    assert comparison.left.attr == "status"
+    assert len(comparison.comparators) == 1
+    assert isinstance(comparison.comparators[0], ast.Attribute)
+    assert comparison.comparators[0].attr == "COMPLETION"
+
+
+def test_completion_candidate_event_has_single_native_terminal_setter() -> None:
+    setters: list[tuple[Path, int]] = []
+    for source in _EXECUTION_ROOT.rglob("*.py"):
+        tree = ast.parse(source.read_text(encoding="utf-8"))
+        for call in ast.walk(tree):
+            if (
+                isinstance(call, ast.Call)
+                and isinstance(call.func, ast.Attribute)
+                and call.func.attr == "set"
+                and isinstance(call.func.value, ast.Attribute)
+                and call.func.value.attr == "completion_candidate_event"
+            ):
+                setters.append((source, call.lineno))
+    assert len(setters) == 1, setters
+    source, lineno = setters[0]
+    assert source == _PROCESS_RACE
+    tree = ast.parse(source.read_text(encoding="utf-8"))
+    heartbeat = next(
+        node
+        for node in tree.body
+        if isinstance(node, ast.AsyncFunctionDef) and node.name == "_watch_heartbeat"
     )
-    assert _calls_trigger_set(ast.Module(body=completion_branch.orelse))
+    assert heartbeat.lineno < lineno <= heartbeat.end_lineno
 
 
 def test_completion_eligibility_and_final_fold_consume_both_cursors() -> None:

tests/execution/_process_helpers.py

@@ -1,8 +1,13 @@
 """Process doubles shared by liveness watcher tests."""
 
+from functools import partial
+from pathlib import Path
 from types import SimpleNamespace
 
 import psutil
+import pytest
+
+from autoskillit.execution.process import _process_race, _session_log_monitor
 
 
 class ConnectedProcess:
@@ -21,3 +26,15 @@ def connections(self, kind: str | None = None) -> list[SimpleNamespace]:
         ]
 
     net_connections = connections
+
+
+def coordinate_phase2_start(
+    monkeypatch: pytest.MonkeyPatch,
+    session_dir: Path,
+) -> None:
+    ready_path = session_dir / "phase2.ready"
+    monkeypatch.setattr(
+        _process_race,
+        "_session_log_monitor",
+        partial(_session_log_monitor, _on_poll=ready_path.touch),
+    )