From da5858f2634b48d6394b03075df30fb577328f85 Mon Sep 17 00:00:00 2001
From: Trecek <talontrecek@gmail.com>
Date: Fri, 16 Oct 2026 07:43:30 -0700
Subject: [PATCH 1/6] test: share process and required-join regression helpers

---
 tests/execution/_process_helpers.py         |  17 ++
 tests/execution/test_process_channel_b.py   |  27 +--
 tests/hooks/_session_binding_helpers.py     | 223 +++++++++++++++++++-
 tests/hooks/test_join_enforcement_wiring.py | 191 ++---------------
 4 files changed, 263 insertions(+), 195 deletions(-)

diff --git a/tests/execution/_process_helpers.py b/tests/execution/_process_helpers.py
index f45db53736..f0bae18054 100644
--- a/tests/execution/_process_helpers.py
+++ b/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),
+    )
diff --git a/tests/execution/test_process_channel_b.py b/tests/execution/test_process_channel_b.py
index 1a2e3d7d69..80cdc8079d 100644
--- a/tests/execution/test_process_channel_b.py
+++ b/tests/execution/test_process_channel_b.py
@@ -4,8 +4,6 @@
 
 import sys
 import textwrap
-from functools import partial
-from pathlib import Path
 
 import psutil
 import pytest
@@ -13,10 +11,9 @@
 from autoskillit.core.types import ChannelConfirmation, SubprocessResult, TerminationReason
 from autoskillit.execution.backends.claude import ClaudeCodeBackend
 from autoskillit.execution.process import (
-    _process_race,
-    _session_log_monitor,
     run_managed_async,
 )
+from tests.execution._process_helpers import coordinate_phase2_start
 from tests.execution.conftest import WRITE_RESULT_THEN_HANG_SCRIPT
 
 pytestmark = [pytest.mark.layer("execution"), pytest.mark.medium]
@@ -61,18 +58,6 @@
 """)
 
 
-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),
-    )
-
-
 # Script that writes %%ORDER_UP%% to session JSONL but never writes type=result to stdout.
 # Simulates CLI hung post-completion — drain timeout should expire and kill anyway.
 # Pass session_dir as sys.argv[1].
@@ -244,7 +229,7 @@ async def test_channel_b_does_not_end_live_process(self, tmp_path, monkeypatch):
         """
         session_dir = tmp_path / "session"
         session_dir.mkdir()
-        _coordinate_phase2_start(monkeypatch, session_dir)
+        coordinate_phase2_start(monkeypatch, session_dir)
         script = tmp_path / "channel_b_no_stdout.py"
         script.write_text(CHANNEL_B_NO_STDOUT_SCRIPT)
 
@@ -310,7 +295,7 @@ async def test_channel_b_timeout_remains_nonterminal(self, tmp_path, monkeypatch
         """
         session_dir = tmp_path / "session"
         session_dir.mkdir()
-        _coordinate_phase2_start(monkeypatch, session_dir)
+        coordinate_phase2_start(monkeypatch, session_dir)
         script = tmp_path / "channel_b_no_stdout.py"
         script.write_text(CHANNEL_B_NO_STDOUT_SCRIPT)
 
@@ -390,7 +375,7 @@ async def test_channel_b_then_empty_result_remains_nonterminal(
         """
         session_dir = tmp_path / "session"
         session_dir.mkdir()
-        _coordinate_phase2_start(monkeypatch, session_dir)
+        coordinate_phase2_start(monkeypatch, session_dir)
         script = tmp_path / "channel_b_empty.py"
         script.write_text(CHANNEL_B_THEN_A_EMPTY_RESULT_SCRIPT)
 
@@ -442,7 +427,7 @@ async def test_channel_b_then_empty_result_produces_terminal_timeout(
 
         session_dir = tmp_path / "session"
         session_dir.mkdir()
-        _coordinate_phase2_start(monkeypatch, session_dir)
+        coordinate_phase2_start(monkeypatch, session_dir)
         script = tmp_path / "channel_b_empty.py"
         script.write_text(CHANNEL_B_THEN_A_EMPTY_RESULT_SCRIPT)
 
@@ -496,7 +481,7 @@ async def test_channel_b_timeout_produces_terminal_timeout_result(self, tmp_path
 
         session_dir = tmp_path / "session"
         session_dir.mkdir()
-        _coordinate_phase2_start(monkeypatch, session_dir)
+        coordinate_phase2_start(monkeypatch, session_dir)
         script = tmp_path / "channel_b_no_stdout.py"
         script.write_text(CHANNEL_B_NO_STDOUT_SCRIPT)
 
diff --git a/tests/hooks/_session_binding_helpers.py b/tests/hooks/_session_binding_helpers.py
index fbc542e6d6..2a220bf859 100644
--- a/tests/hooks/_session_binding_helpers.py
+++ b/tests/hooks/_session_binding_helpers.py
@@ -5,9 +5,31 @@
 import json
 import shutil
 from pathlib import Path
+from typing import TYPE_CHECKING
 
-from autoskillit.hooks._session_binding import PROJECTION_MANIFEST_SCHEMA_VERSION
+from autoskillit.hooks._join_ledger import (
+    OUTCOME_SUCCESS,
+    claim_assignment,
+    declare_batch,
+    resolve_flag_dir,
+    settle_assignment,
+)
+from autoskillit.hooks._runtime._hook_constants import MANAGED_JOIN_PARENT_ID_ENV_VAR
+from autoskillit.hooks._session_binding import (
+    PROJECTION_MANIFEST_SCHEMA_VERSION,
+    read_binding,
+    resolve_binding_path,
+)
 from autoskillit.hooks._write_scope import WriteScope, WriteScopeKind, encode_write_scope
+from tests._hook_protocol_oracle import (
+    STATUS_COMPLETED,
+    claude_verdict,
+    codex_verdict,
+    run_hook,
+)
+
+if TYPE_CHECKING:
+    from autoskillit.hooks._runtime._hook_output import HookEmission
 
 _HOOKS_SOURCE = Path(__file__).resolve().parents[2] / "src" / "autoskillit" / "hooks"
 _RUNTIME_SOURCE = _HOOKS_SOURCE / "_runtime"
@@ -102,3 +124,202 @@ def write_projection_manifest(
         encoding="utf-8",
     )
     return manifest_path
+
+
+RETIRED_JOIN_ENV = frozenset(
+    {
+        "AUTOSKILLIT_JOIN_FLAG_PATH",
+        "AUTOSKILLIT_JOIN_REQUIRED",
+        "AUTOSKILLIT_HOOK_EVENT",
+        "AUTOSKILLIT_SESSION_ID",
+        "AUTOSKILLIT_JOIN_SESSION_ID",
+        "AUTOSKILLIT_JOIN_PARENT",
+    }
+)
+
+
+JOIN_HOOK_UNSET = (
+    *RETIRED_JOIN_ENV,
+    "AUTOSKILLIT_LAUNCH_ID",
+    MANAGED_JOIN_PARENT_ID_ENV_VAR,
+    "AUTOSKILLIT_STATE_ROOT",
+    "AUTOSKILLIT_STATE_DIR",
+    "AUTOSKILLIT_HEADLESS",
+)
+
+
+def join_hook_env(tmp_path: Path) -> dict[str, str]:
+    return {
+        "AUTOSKILLIT_AGENT_BACKEND": "claude-code",
+        "AUTOSKILLIT_LOG_DIR": str(tmp_path / "logs"),
+        "AUTOSKILLIT_SESSION_TYPE": "skill",
+    }
+
+
+def run_join_hook(
+    tmp_path: Path,
+    hook: Path,
+    payload: dict[str, object] | str,
+    *,
+    cwd: Path,
+    env_overrides: dict[str, str] | None = None,
+) -> HookEmission:
+    """Run a real hook program with the shared protocol-test environment."""
+    script = hook.relative_to(_HOOKS_SOURCE) if hook.is_relative_to(_HOOKS_SOURCE) else hook
+    return run_hook(
+        script,
+        payload,
+        env={**join_hook_env(tmp_path), **(env_overrides or {})},
+        unset=JOIN_HOOK_UNSET,
+        cwd=cwd,
+    )
+
+
+def skill_load_payload(
+    worktree: Path,
+    *,
+    session_id: str,
+    activation_source: str,
+) -> dict[str, object]:
+    if activation_source == "slash":
+        return {
+            "hook_event_name": "UserPromptExpansion",
+            "expansion_type": "slash_command",
+            "command_name": "autoskillit:join-bearing",
+            "session_id": session_id,
+            "cwd": str(worktree),
+        }
+    return {
+        "hook_event_name": "PostToolUse",
+        "tool_name": "Skill",
+        "tool_input": {"skill": "join-bearing"},
+        "session_id": session_id,
+        "cwd": str(worktree),
+    }
+
+
+def load_join_bearing_skill(
+    tmp_path: Path,
+    *,
+    session_id: str = "session-1",
+    activation_source: str = "PostToolUse",
+    cook_launch_id: str = "",
+    resumed: bool = False,
+) -> Path:
+    worktree, _ = load_join_bearing_skill_with_context(
+        tmp_path,
+        session_id=session_id,
+        activation_source=activation_source,
+        cook_launch_id=cook_launch_id,
+        resumed=resumed,
+    )
+    return worktree
+
+
+def load_join_bearing_skill_with_context(
+    tmp_path: Path,
+    *,
+    session_id: str = "session-1",
+    activation_source: str = "PostToolUse",
+    cook_launch_id: str = "",
+    resumed: bool = False,
+) -> tuple[Path, str]:
+    """Drive the projected skill-load hook so the binding is production-shaped."""
+    worktree = tmp_path / "worktree"
+    (worktree / ".autoskillit").mkdir(parents=True)
+    projection_root, skill_load_hook = copy_projected_hook(tmp_path)
+    write_projection_manifest(projection_root)
+    registry_before: bytes | None = None
+    if cook_launch_id:
+        write_cook_registry(
+            worktree,
+            launch_id=cook_launch_id,
+            session_id=session_id if resumed else None,
+        )
+        if resumed:
+            registry_before = (
+                worktree / ".autoskillit" / "temp" / "session_registry.json"
+            ).read_bytes()
+
+    skill_load_event = skill_load_payload(
+        worktree,
+        session_id=session_id,
+        activation_source=activation_source,
+    )
+    completed = run_join_hook(
+        tmp_path,
+        skill_load_hook,
+        skill_load_event,
+        cwd=worktree,
+        env_overrides=({"AUTOSKILLIT_LAUNCH_ID": cook_launch_id} if cook_launch_id else None),
+    )
+    assert completed.exit_code == 0, completed.stderr
+    output = json.loads(completed.stdout)
+    hook_output = output.get("hookSpecificOutput")
+    context = hook_output.get("additionalContext", "") if isinstance(hook_output, dict) else ""
+    assert isinstance(context, str)
+    assert context
+    if activation_source == "PostToolUse":
+        for verdict_fn in (codex_verdict, claude_verdict):
+            verdict = verdict_fn(
+                "PostToolUse",
+                exit_code=completed.exit_code,
+                stdout=completed.stdout,
+                stderr=completed.stderr,
+            )
+            assert verdict.status == STATUS_COMPLETED
+            assert verdict.contexts
+    binding = read_binding(resolve_binding_path(str(worktree), session_id))
+    assert binding is not None and binding.join_required
+    if registry_before is not None:
+        assert (
+            worktree / ".autoskillit" / "temp" / "session_registry.json"
+        ).read_bytes() == registry_before
+    return worktree, context
+
+
+def declare_one_assignment(worktree: Path, *, session_id: str) -> Path:
+    """Declare one wave at the same state root the hook payload resolves."""
+    flag_dir = resolve_flag_dir(worktree)
+    declare_batch(
+        flag_dir,
+        session_id=session_id,
+        top_level_parent="top_level",
+        skill_name="join-bearing",
+        artifact_digest="artdigest-1",
+        assignments=("worker",),
+    )
+    return flag_dir
+
+
+def write_cook_registry(worktree: Path, *, launch_id: str, session_id: str | None) -> None:
+    registry_path = worktree / ".autoskillit" / "temp" / "session_registry.json"
+    registry_path.parent.mkdir(parents=True, exist_ok=True)
+    registry_path.write_text(
+        json.dumps(
+            {
+                launch_id: {
+                    "session_type": "cook",
+                    "claude_session_id": session_id,
+                }
+            }
+        ),
+        encoding="utf-8",
+    )
+
+
+def complete_one_assignment_wave(worktree: Path, *, session_id: str) -> None:
+    flag_dir = declare_one_assignment(worktree, session_id=session_id)
+    claim_assignment(
+        flag_dir,
+        session_id=session_id,
+        top_level_parent="top_level",
+        tool_use_id="agent-1",
+    )
+    settle_assignment(
+        flag_dir,
+        session_id=session_id,
+        top_level_parent="top_level",
+        tool_use_id="agent-1",
+        outcome=OUTCOME_SUCCESS,
+    )
diff --git a/tests/hooks/test_join_enforcement_wiring.py b/tests/hooks/test_join_enforcement_wiring.py
index 6543b12f6b..e60acdb82b 100644
--- a/tests/hooks/test_join_enforcement_wiring.py
+++ b/tests/hooks/test_join_enforcement_wiring.py
@@ -35,9 +35,25 @@
     STATUS_COMPLETED,
     claude_verdict,
     codex_verdict,
-    run_hook,
 )
-from tests.hooks._session_binding_helpers import copy_projected_hook, write_projection_manifest
+from tests.hooks._session_binding_helpers import (
+    RETIRED_JOIN_ENV as _RETIRED_JOIN_ENV,
+)
+from tests.hooks._session_binding_helpers import (
+    declare_one_assignment as _declare_one_assignment,
+)
+from tests.hooks._session_binding_helpers import (
+    load_join_bearing_skill as _load_join_bearing_skill,
+)
+from tests.hooks._session_binding_helpers import (
+    load_join_bearing_skill_with_context as _load_join_bearing_skill_with_context,
+)
+from tests.hooks._session_binding_helpers import (
+    run_join_hook as _run_hook,
+)
+from tests.hooks._session_binding_helpers import (
+    write_cook_registry as _write_cook_registry,
+)
 
 if TYPE_CHECKING:
     from autoskillit.hooks._runtime._hook_output import HookEmission
@@ -48,16 +64,6 @@
 _PROJECT_ROOT = Path(__file__).resolve().parents[2]
 _GUARDS_DIR = _PROJECT_ROOT / "src" / "autoskillit" / "hooks" / "guards"
 _HOOKS_DIR = _PROJECT_ROOT / "src" / "autoskillit" / "hooks"
-_RETIRED_JOIN_ENV = frozenset(
-    {
-        "AUTOSKILLIT_JOIN_FLAG_PATH",
-        "AUTOSKILLIT_JOIN_REQUIRED",
-        "AUTOSKILLIT_HOOK_EVENT",
-        "AUTOSKILLIT_SESSION_ID",
-        "AUTOSKILLIT_JOIN_SESSION_ID",
-        "AUTOSKILLIT_JOIN_PARENT",
-    }
-)
 _SETTLEMENT_EXPECTATIONS: dict[str, tuple[dict[str, object], str]] = {
     "PostToolUse": ({"tool_response": "complete"}, OUTCOME_SUCCESS),
     "PostToolUseFailure": ({"error": "forced failure"}, OUTCOME_FAILURE),
@@ -85,151 +91,6 @@ def _settlement_event_cases() -> tuple[tuple[str, dict[str, object], str], ...]:
 _SETTLEMENT_EVENT_CASES = _settlement_event_cases()
 
 
-def _run_hook(
-    tmp_path: Path,
-    hook: Path,
-    payload: dict[str, object] | str,
-    *,
-    cwd: Path,
-    env_overrides: dict[str, str] | None = None,
-) -> HookEmission:
-    """Run a real hook program with the shared protocol-test environment."""
-    script = hook.relative_to(_HOOKS_DIR) if hook.is_relative_to(_HOOKS_DIR) else hook
-    return run_hook(
-        script,
-        payload,
-        env={
-            "AUTOSKILLIT_AGENT_BACKEND": "claude-code",
-            "AUTOSKILLIT_LOG_DIR": str(tmp_path / "logs"),
-            "AUTOSKILLIT_SESSION_TYPE": "skill",
-            **(env_overrides or {}),
-        },
-        unset=(
-            *_RETIRED_JOIN_ENV,
-            "AUTOSKILLIT_LAUNCH_ID",
-            MANAGED_JOIN_PARENT_ID_ENV_VAR,
-        ),
-        cwd=cwd,
-    )
-
-
-def _skill_load_payload(
-    worktree: Path,
-    *,
-    session_id: str,
-    activation_source: str,
-) -> dict[str, object]:
-    if activation_source == "slash":
-        return {
-            "hook_event_name": "UserPromptExpansion",
-            "expansion_type": "slash_command",
-            "command_name": "autoskillit:join-bearing",
-            "session_id": session_id,
-            "cwd": str(worktree),
-        }
-    return {
-        "hook_event_name": "PostToolUse",
-        "tool_name": "Skill",
-        "tool_input": {"skill": "join-bearing"},
-        "session_id": session_id,
-        "cwd": str(worktree),
-    }
-
-
-def _load_join_bearing_skill(
-    tmp_path: Path,
-    *,
-    session_id: str = "session-1",
-    activation_source: str = "PostToolUse",
-    cook_launch_id: str = "",
-    resumed: bool = False,
-) -> Path:
-    worktree, _ = _load_join_bearing_skill_with_context(
-        tmp_path,
-        session_id=session_id,
-        activation_source=activation_source,
-        cook_launch_id=cook_launch_id,
-        resumed=resumed,
-    )
-    return worktree
-
-
-def _load_join_bearing_skill_with_context(
-    tmp_path: Path,
-    *,
-    session_id: str = "session-1",
-    activation_source: str = "PostToolUse",
-    cook_launch_id: str = "",
-    resumed: bool = False,
-) -> tuple[Path, str]:
-    """Drive the projected skill-load hook so the binding is production-shaped."""
-    worktree = tmp_path / "worktree"
-    (worktree / ".autoskillit").mkdir(parents=True)
-    projection_root, skill_load_hook = copy_projected_hook(tmp_path)
-    write_projection_manifest(projection_root)
-    registry_before: bytes | None = None
-    if cook_launch_id:
-        _write_cook_registry(
-            worktree,
-            launch_id=cook_launch_id,
-            session_id=session_id if resumed else None,
-        )
-        if resumed:
-            registry_before = (
-                worktree / ".autoskillit" / "temp" / "session_registry.json"
-            ).read_bytes()
-
-    skill_load_event = _skill_load_payload(
-        worktree,
-        session_id=session_id,
-        activation_source=activation_source,
-    )
-    completed = _run_hook(
-        tmp_path,
-        skill_load_hook,
-        skill_load_event,
-        cwd=worktree,
-        env_overrides=({"AUTOSKILLIT_LAUNCH_ID": cook_launch_id} if cook_launch_id else None),
-    )
-    assert completed.exit_code == 0, completed.stderr
-    output = json.loads(completed.stdout)
-    hook_output = output.get("hookSpecificOutput")
-    context = hook_output.get("additionalContext", "") if isinstance(hook_output, dict) else ""
-    assert isinstance(context, str)
-    assert context
-    if activation_source == "PostToolUse":
-        for verdict_fn in (codex_verdict, claude_verdict):
-            verdict = verdict_fn(
-                "PostToolUse",
-                exit_code=completed.exit_code,
-                stdout=completed.stdout,
-                stderr=completed.stderr,
-            )
-            assert verdict.status == STATUS_COMPLETED
-            assert verdict.contexts
-    binding = read_binding(resolve_binding_path(str(worktree), session_id))
-    assert binding is not None and binding.join_required
-    if registry_before is not None:
-        assert (
-            worktree / ".autoskillit" / "temp" / "session_registry.json"
-        ).read_bytes() == registry_before
-    return worktree, context
-
-
-def _declare_one_assignment(worktree: Path, *, session_id: str) -> Path:
-    """Declare one wave at the same state root the hook payload resolves."""
-    flag_dir = resolve_flag_dir(worktree)
-    declare_batch(
-        flag_dir,
-        session_id=session_id,
-        top_level_parent="top_level",
-        skill_name="join-bearing",
-        artifact_digest="artdigest-1",
-        assignments=("worker",),
-    )
-    return flag_dir
-
-
 def _mark_binding_scope_invalid(worktree: Path, *, session_id: str) -> None:
     binding_path = resolve_binding_path(str(worktree), session_id)
     binding = read_binding(binding_path)
@@ -264,22 +125,6 @@ def _block_reason(emission: HookEmission) -> str:
     return emission.stderr
 
 
-def _write_cook_registry(worktree: Path, *, launch_id: str, session_id: str | None) -> None:
-    registry_path = worktree / ".autoskillit" / "temp" / "session_registry.json"
-    registry_path.parent.mkdir(parents=True, exist_ok=True)
-    registry_path.write_text(
-        json.dumps(
-            {
-                launch_id: {
-                    "session_type": "cook",
-                    "claude_session_id": session_id,
-                }
-            }
-        ),
-        encoding="utf-8",
-    )
-
-
 def _configure_non_cook_shape(
     worktree: Path,
     *,

From 63e223807bc7448bac9d8cbb731701725e7afc30 Mon Sep 17 00:00:00 2001
From: Trecek <talontrecek@gmail.com>
Date: Fri, 16 Oct 2026 07:51:56 -0700
Subject: [PATCH 2/6] test: pin Channel B corroboration and Stop continuation
 behavior

---
 tests/_test_filter.py                         |   1 +
 tests/arch/test_layer_enforcement.py          |   4 +
 tests/arch/test_watcher_signal_consistency.py |  68 ++-
 tests/execution/test_process_race.py          |  41 +-
 tests/execution/test_process_run.py           |  11 +-
 .../execution/test_process_stop_hook_join.py  | 450 ++++++++++++++++++
 tests/execution/test_termination_action.py    |   8 +-
 7 files changed, 544 insertions(+), 39 deletions(-)
 create mode 100644 tests/execution/test_process_stop_hook_join.py

diff --git a/tests/_test_filter.py b/tests/_test_filter.py
index 3487b35e69..11fa05cc8c 100644
--- a/tests/_test_filter.py
+++ b/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",
diff --git a/tests/arch/test_layer_enforcement.py b/tests/arch/test_layer_enforcement.py
index 51c8dea33c..69298f3b18 100644
--- a/tests/arch/test_layer_enforcement.py
+++ b/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"}),
diff --git a/tests/arch/test_watcher_signal_consistency.py b/tests/arch/test_watcher_signal_consistency.py
index 6d06039b2c..ba22096473 100644
--- a/tests/arch/test_watcher_signal_consistency.py
+++ b/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:
diff --git a/tests/execution/test_process_race.py b/tests/execution/test_process_race.py
index 28f2f00675..a92b1c1b97 100644
--- a/tests/execution/test_process_race.py
+++ b/tests/execution/test_process_race.py
@@ -20,6 +20,7 @@
 from autoskillit.execution.backends import ClaudeStreamParser
 from autoskillit.execution.process import TetherSpec, run_managed_async, spawn_owned_process
 from autoskillit.execution.process._process_jsonl import EventCursor
+from autoskillit.execution.process._process_monitor import SessionMonitorResult
 from autoskillit.execution.process._process_race import (
     RaceAccumulator,
     RaceSignals,
@@ -306,8 +307,8 @@ class TestRaceAccumulatorFieldCount:
 
     def test_race_accumulator_field_count(self) -> None:
         n = len(dataclasses.fields(RaceAccumulator))
-        assert n == 24, (
-            f"RaceAccumulator has {n} fields (expected 24). Update tests for new fields."
+        assert n == 23, (
+            f"RaceAccumulator has {n} fields (expected 23). Update tests for new fields."
         )
 
 
@@ -367,7 +368,6 @@ def test_lifecycle_accumulator_fields_have_exact_defaults_and_freeze_propagation
     assert acc.schedule_wakeup_violation is False
     assert acc.completion_ceiling_expired is False
     assert acc.channel_a_candidate_at is None
-    assert acc.channel_b_candidate_at is None
     assert acc.stdout_cursor is None
     assert acc.channel_b_cursor is None
     assert not acc.completion_candidate_event.is_set()
@@ -418,7 +418,7 @@ async def test_completion_candidate_without_task_ignores_large_ceiling() -> None
 async def test_completion_ceiling_preserves_pending_ids_for_adjudication() -> None:
     acc = RaceAccumulator(lifecycle_observation_enabled=True)
     acc.pending_task_ids.add("owned")
-    acc.channel_b_candidate_at = anyio.current_time()
+    acc.channel_a_candidate_at = anyio.current_time()
     acc.completion_candidate_event.set()
     trigger = anyio.Event()
 
@@ -460,13 +460,19 @@ async def test_completion_waits_for_terminal_evidence_across_channel_orderings(
     acc = RaceAccumulator(lifecycle_observation_enabled=True)
     acc.stdout_cursor = EventCursor(stdout_path)
     acc.channel_b_cursor = EventCursor(channel_b_path)
+    trigger = anyio.Event()
     if marker_channel == "a":
         acc.channel_a_candidate_at = anyio.current_time()
+        acc.completion_candidate_event.set()
     else:
-        acc.channel_b_candidate_at = anyio.current_time()
-        acc.channel_b_status = ChannelBStatus.COMPLETION
-    acc.completion_candidate_event.set()
-    trigger = anyio.Event()
+        acc.deposit_session_log_result(
+            SessionMonitorResult(
+                ChannelBStatus.COMPLETION, "session", cursor=acc.channel_b_cursor
+            ),
+            anyio.Event(),
+            trigger,
+            None,
+        )
 
     async with anyio.create_task_group() as task_group:
         task_group.start_soon(
@@ -484,13 +490,20 @@ async def test_completion_waits_for_terminal_evidence_across_channel_orderings(
         assert not trigger.is_set()
         with task_path.open("a") as stream:
             stream.write('{"type":"task_notification","task_id":"owned","status":"completed"}\n')
-        with anyio.fail_after(1):
-            await trigger.wait()
+        if marker_channel == "a":
+            with anyio.fail_after(1):
+                await trigger.wait()
+        else:
+            await anyio.sleep(0.03)
+            assert trigger.is_set() is False
+            assert acc.completion_candidate_event.is_set() is False
+            task_group.cancel_scope.cancel()
 
-    signals = acc.to_race_signals()
-    assert signals.pending_task_ids == ()
-    assert signals.terminal_task_ids == ("owned",)
-    assert signals.channel_a_confirmed is (marker_channel == "a")
+    if marker_channel == "a":
+        signals = acc.to_race_signals()
+        assert signals.pending_task_ids == ()
+        assert signals.terminal_task_ids == ("owned",)
+        assert signals.channel_a_confirmed is True
 
 
 @pytest.mark.anyio
diff --git a/tests/execution/test_process_run.py b/tests/execution/test_process_run.py
index ff80039375..0716f9c2bb 100644
--- a/tests/execution/test_process_run.py
+++ b/tests/execution/test_process_run.py
@@ -21,7 +21,7 @@
 import pytest
 import structlog.testing
 
-from autoskillit.core.types import TerminationReason
+from autoskillit.core.types import KillReason, TerminationReason
 from autoskillit.execution.process import (
     _normalize_pass_fds,
     read_temp_output,
@@ -190,6 +190,15 @@ def test_stdin_input_roundtrip_sync(self, tmp_path):
 class TestTimeoutKillsHangingProcess:
     """Timeout fires and kills when process hangs."""
 
+    def test_sync_timeout_reports_infra_kill(self, tmp_path):
+        script = tmp_path / "hang.py"
+        script.write_text(HANG_FOREVER_SCRIPT)
+
+        result = run_managed_sync([sys.executable, str(script)], cwd=tmp_path, timeout=0.3)
+
+        assert result.termination is TerminationReason.TIMED_OUT
+        assert result.kill_reason is KillReason.INFRA_KILL
+
     @pytest.mark.anyio
     async def test_timeout_fires_and_kills_hanging_process(self, tmp_path):
         """Process sleeps forever, timeout kills it, partial output returned."""
diff --git a/tests/execution/test_process_stop_hook_join.py b/tests/execution/test_process_stop_hook_join.py
new file mode 100644
index 0000000000..c7351ee4cf
--- /dev/null
+++ b/tests/execution/test_process_stop_hook_join.py
@@ -0,0 +1,450 @@
+"""Real Stop-guard decisions must survive Channel B completion markers."""
+
+from __future__ import annotations
+
+import json
+import shutil
+import sys
+import textwrap
+import time
+from pathlib import Path
+from uuid import uuid4
+
+import anyio
+import psutil
+import pytest
+
+from autoskillit.core.types import (
+    ChannelBStatus,
+    ChannelConfirmation,
+    KillReason,
+    TerminationReason,
+)
+from autoskillit.execution.backends import ClaudeStreamParser
+from autoskillit.execution.process import run_managed_async
+from autoskillit.execution.process._process_race import RaceAccumulator
+from autoskillit.hook_registry import HOOKS_DIR
+from autoskillit.hooks._join_ledger import resolve_flag_dir
+from autoskillit.hooks._session_binding import read_binding, resolve_binding_path
+from tests.conftest import production_interpreter_env
+from tests.execution._process_helpers import coordinate_phase2_start
+from tests.hooks._session_binding_helpers import (
+    JOIN_HOOK_UNSET,
+    complete_one_assignment_wave,
+    join_hook_env,
+    load_join_bearing_skill,
+)
+
+pytestmark = [pytest.mark.layer("execution"), pytest.mark.medium]
+
+_PTY_CASES = [
+    False,
+    pytest.param(
+        True, marks=pytest.mark.skipif(shutil.which("script") is None, reason="needs script")
+    ),
+]
+_CHILD_PRELUDE = r"""
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+config = json.loads(sys.argv[1])
+root = Path(config["root"])
+session_dir = Path(config["session_dir"])
+sid = config["sid"]
+marker = config["marker"]
+hook_argv = config["hook_argv"]
+hook_env = dict(os.environ)
+stop_payload = {
+    "session_id": sid, "cwd": os.getcwd(),
+    "hook_event_name": "Stop", "stop_hook_active": False,
+}
+transcript = session_dir / (sid + ".jsonl")
+
+
+def emit_stdout(record):
+    print(json.dumps(record), flush=True)
+
+
+def append_transcript(record):
+    with transcript.open("a") as stream:
+        stream.write(json.dumps(record) + "\n")
+
+
+def write_marker(text="plan_path = x.md\n"):
+    append_transcript({
+        "type": "assistant",
+        "message": {"role": "assistant", "content": [{"type": "text", "text": text + marker}]},
+    })
+
+
+def run_hook(argv, payload, env, name):
+    result = subprocess.run(
+        argv, input=json.dumps(payload), env=env,
+        capture_output=True, text=True, timeout=10,
+    )
+    with (root / "hook_results.jsonl").open("a") as stream:
+        record = {"name": name, "rc": result.returncode, "stderr": result.stderr}
+        stream.write(json.dumps(record) + "\n")
+    return result
+
+
+def stop_feedback(reason):
+    text = "Stop hook feedback:\n[" + " ".join(hook_argv) + "]: " + reason
+    user = {"role": "user", "content": [{"type": "text", "text": text}]}
+    append_transcript({"type": "user", "isMeta": True, "message": user})
+    append_transcript({
+        "type": "system", "subtype": "stop_hook_summary",
+        "hookErrors": [reason], "preventedContinuation": False, "level": "suggestion",
+    })
+    emit_stdout({"type": "user", "isSynthetic": True, "message": user})
+    emit_stdout({"type": "system", "subtype": "stop-hook-error", "content": reason})
+
+
+def touch(name):
+    (root / name).write_text(str(time.time()))
+
+
+def wait_for(name):
+    deadline = time.monotonic() + 10
+    while not (root / name).exists():
+        if time.monotonic() >= deadline:
+            raise TimeoutError(name)
+        time.sleep(0.01)
+
+
+def write_pid():
+    (root / "workload.pid").write_text(str(os.getpid()))
+
+
+emit_stdout({"type": "system", "subtype": "init", "session_id": sid})
+append_transcript({"type": "assistant", "message": {"role": "assistant", "content": "working"}})
+deadline = time.monotonic() + 10
+while not (session_dir / "phase2.ready").exists():
+    if time.monotonic() >= deadline:
+        raise TimeoutError("phase2.ready")
+    time.sleep(0.01)
+"""
+
+
+@pytest.fixture
+def stop_case(tmp_path, monkeypatch):
+    sid = str(uuid4())
+    worktree = load_join_bearing_skill(tmp_path, session_id=sid)
+    binding = read_binding(resolve_binding_path(str(worktree), sid))
+    assert binding is not None and binding.join_required
+    assert resolve_flag_dir(worktree).is_relative_to(worktree)
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    coordinate_phase2_start(monkeypatch, session_dir)
+    marker = "%%ORDER_UP%%"
+
+    original_deposit = RaceAccumulator.deposit_session_log_result
+
+    def observe_deposit(self, monitor_result, *args, **kwargs):
+        original_deposit(self, monitor_result, *args, **kwargs)
+        if monitor_result.status is ChannelBStatus.COMPLETION:
+            (tmp_path / "b_deposited").touch()
+
+    monkeypatch.setattr(RaceAccumulator, "deposit_session_log_result", observe_deposit)
+    env = production_interpreter_env()
+    env.update(join_hook_env(tmp_path))
+    for name in JOIN_HOOK_UNSET:
+        env.pop(name, None)
+    config = {
+        "root": str(tmp_path),
+        "session_dir": str(session_dir),
+        "sid": sid,
+        "marker": marker,
+        "hook_argv": [
+            sys.executable,
+            "-B",
+            str(HOOKS_DIR / "_dispatch.py"),
+            "guards/join_stop_guard",
+        ],
+    }
+
+    async def run(body, **overrides):
+        script = tmp_path / "child.py"
+        script.write_text(_CHILD_PRELUDE + textwrap.dedent(body))
+        options = {
+            "lifecycle_observation_enabled": True,
+            "stream_parser": ClaudeStreamParser(completion_marker=marker),
+            "session_log_dir": session_dir,
+            "completion_marker": marker,
+            "_phase1_poll": 0.05,
+            "_phase2_poll": 0.05,
+            "_heartbeat_poll": 0.05,
+            "_session_id_timeout": 0.01,
+            "_phase1_timeout": 120,
+            "timeout": 30,
+            "completion_drain_timeout": 10.0,
+            "natural_exit_grace_seconds": 3.0,
+            "child_deferral_ceiling": 5.0,
+            **overrides,
+        }
+        result = await run_managed_async(
+            [sys.executable, str(script), json.dumps(config)], cwd=worktree, env=env, **options
+        )
+        assert result.session_id == sid
+        assert result.channel_b_session_id == sid
+        assert (tmp_path / "b_deposited").exists()
+        return result
+
+    return worktree, sid, run
+
+
+def _hook_results(tmp_path: Path) -> list[dict]:
+    return [
+        json.loads(line) for line in (tmp_path / "hook_results.jsonl").read_text().splitlines()
+    ]
+
+
+def _guard_statuses(tmp_path: Path) -> list[str]:
+    return [
+        record["status"]
+        for line in (tmp_path / "logs" / "join_diagnostics.jsonl").read_text().splitlines()
+        if (record := json.loads(line)).get("gate") == "join_stop_guard"
+    ]
+
+
+async def _wait_for_file(path: Path) -> None:
+    while not path.exists():
+        await anyio.sleep(0.02)
+
+
+async def _assert_pid_stopped(pid: int) -> None:
+    with anyio.fail_after(5):
+        while psutil.pid_exists(pid):
+            try:
+                if psutil.Process(pid).status() == psutil.STATUS_ZOMBIE:
+                    return
+            except psutil.NoSuchProcess:
+                return
+            await anyio.sleep(0.05)
+
+
+@pytest.mark.anyio
+@pytest.mark.timeout(120)
+@pytest.mark.parametrize("pty_mode", _PTY_CASES)
+async def test_blocked_stop_continuation_completes_required_work_without_preemption(
+    tmp_path, stop_case, pty_mode
+):
+    worktree, sid, run = stop_case
+
+    async def finish_wave():
+        with anyio.move_on_after(30):
+            await _wait_for_file(tmp_path / "first_stop_done")
+            complete_one_assignment_wave(worktree, session_id=sid)
+            (tmp_path / "wave_done").touch()
+
+    async with anyio.create_task_group() as task_group:
+        task_group.start_soon(finish_wave)
+        try:
+            result = await run(
+                """
+                write_marker()
+                wait_for("b_deposited")
+                stop_1 = run_hook(hook_argv, stop_payload, hook_env, "stop_1")
+                stop_feedback(stop_1.stderr)
+                touch("first_stop_done")
+                wait_for("wave_done")
+                time.sleep(0.5)
+                touch("continuation_ran")
+                write_marker()
+                run_hook(hook_argv, {**stop_payload, "stop_hook_active": True}, hook_env, "stop_2")
+                emit_stdout({"type": "result", "subtype": "success", "is_error": False,
+                             "result": "done\\n" + marker, "session_id": sid})
+                time.sleep(0.5)
+                """,
+                pty_mode=pty_mode,
+            )
+        finally:
+            task_group.cancel_scope.cancel()
+
+    first, second = _hook_results(tmp_path)
+    assert first["rc"] == 2
+    assert "join-bearing skill loaded but no declared wave for this parent" in first["stderr"]
+    assert second["rc"] == 0
+    assert _guard_statuses(tmp_path) == ["block", "allow"]
+    assert (tmp_path / "continuation_ran").exists()
+    assert result.returncode == 0
+    assert result.channel_confirmation is ChannelConfirmation.CHANNEL_A
+    assert "Session terminated" not in result.stdout + result.stderr
+    # F-1: the drain's cancelled exit watcher can still label this KILL_AFTER_COMPLETION.
+
+
+@pytest.mark.anyio
+@pytest.mark.timeout(120)
+async def test_approved_completion_stops_without_preemption(tmp_path, stop_case):
+    worktree, sid, run = stop_case
+    complete_one_assignment_wave(worktree, session_id=sid)
+
+    result = await run(
+        """
+        write_marker()
+        wait_for("b_deposited")
+        run_hook(hook_argv, stop_payload, hook_env, "stop_1")
+        emit_stdout({"type": "result", "subtype": "success", "is_error": False,
+                     "result": "done\\n" + marker, "session_id": sid})
+        time.sleep(0.5)
+        """
+    )
+
+    assert _hook_results(tmp_path)[0]["rc"] == 0
+    assert _guard_statuses(tmp_path) == ["allow"]
+    assert result.returncode == 0
+    assert result.channel_confirmation is ChannelConfirmation.CHANNEL_A
+    assert "Session terminated" not in result.stdout + result.stderr
+
+
+@pytest.mark.anyio
+@pytest.mark.timeout(120)
+async def test_natural_exit_after_blocked_stop_is_reported_natural(tmp_path, stop_case):
+    _, _, run = stop_case
+    result = await run(
+        """
+        write_marker()
+        wait_for("b_deposited")
+        stop_1 = run_hook(hook_argv, stop_payload, hook_env, "stop_1")
+        stop_feedback(stop_1.stderr)
+        time.sleep(1.0)
+        """
+    )
+
+    assert _hook_results(tmp_path)[0]["rc"] == 2
+    assert _guard_statuses(tmp_path) == ["block"]
+    assert result.termination is TerminationReason.NATURAL_EXIT
+    assert result.channel_confirmation is ChannelConfirmation.CHANNEL_B
+    assert result.returncode == 0
+    assert result.kill_reason is KillReason.NATURAL_EXIT
+
+
+@pytest.mark.anyio
+@pytest.mark.timeout(120)
+@pytest.mark.parametrize(
+    ("ends_by", "pty_mode"),
+    [
+        ("timeout", False),
+        pytest.param(
+            "timeout",
+            True,
+            marks=pytest.mark.skipif(shutil.which("script") is None, reason="needs script"),
+        ),
+        ("cancel", False),
+    ],
+)
+async def test_repeated_stop_blocks_are_bounded(tmp_path, stop_case, ends_by, pty_mode):
+    _, _, run = stop_case
+    body = """
+        write_pid()
+        for index in range(100):
+            write_marker()
+            if index == 0:
+                wait_for("b_deposited")
+            stopped = run_hook(
+                hook_argv, {**stop_payload, "stop_hook_active": index > 0}, hook_env,
+                "stop_" + str(index),
+            )
+            stop_feedback(stopped.stderr)
+            if index == 0:
+                touch("first_stop_done")
+            time.sleep(0.2)
+    """
+    if ends_by == "cancel":
+        cancelled = False
+
+        async def cancel_after_stop(scope):
+            with anyio.move_on_after(30):
+                await _wait_for_file(tmp_path / "first_stop_done")
+                scope.cancel()
+
+        async with anyio.create_task_group() as task_group:
+            task_group.start_soon(cancel_after_stop, task_group.cancel_scope)
+            try:
+                await run(body)
+            except anyio.get_cancelled_exc_class():
+                cancelled = True
+                raise
+            finally:
+                task_group.cancel_scope.cancel()
+        assert cancelled
+    else:
+        result = await run(body, timeout=10, pty_mode=pty_mode)
+        assert result.termination is TerminationReason.TIMED_OUT
+        assert result.kill_reason is KillReason.INFRA_KILL
+        assert result.cleanup_evidence.complete is True
+
+    records = _hook_results(tmp_path)
+    assert records
+    assert all(record["rc"] == 2 for record in records)
+    assert "block" in _guard_statuses(tmp_path)
+    assert (tmp_path / "b_deposited").exists()
+    await _assert_pid_stopped(int((tmp_path / "workload.pid").read_text()))
+
+
+@pytest.mark.anyio
+@pytest.mark.timeout(120)
+async def test_exit_before_b_detection_retains_channel_b_fallback(stop_case):
+    _, _, run = stop_case
+    result = await run("write_marker()\n", _phase2_poll=0.5)
+
+    assert result.termination is TerminationReason.NATURAL_EXIT
+    assert result.channel_confirmation is ChannelConfirmation.CHANNEL_B
+    assert result.returncode == 0
+    assert result.kill_reason is KillReason.NATURAL_EXIT
+
+
+@pytest.mark.anyio
+@pytest.mark.timeout(120)
+async def test_schedule_wakeup_defers_release_from_native_terminal(tmp_path, stop_case):
+    _, _, run = stop_case
+    result = await run(
+        """
+        write_marker()
+        wait_for("b_deposited")
+        time.sleep(1.5)
+        emit_stdout({"type": "assistant", "message": {"content": [
+            {"type": "tool_use", "id": "wake", "name": "ScheduleWakeup", "input": {}}
+        ]}})
+        touch("a_at")
+        emit_stdout({"type": "result", "result": "done\\n" + marker, "session_id": sid})
+        time.sleep(30)
+        """,
+        child_deferral_ceiling=2.0,
+        natural_exit_grace_seconds=0.2,
+    )
+    ended = time.time()
+
+    assert result.schedule_wakeup_violation is True
+    assert result.completion_ceiling_expired is True
+    assert result.channel_confirmation is ChannelConfirmation.CHANNEL_A
+    assert result.kill_reason is KillReason.KILL_AFTER_COMPLETION
+    assert ended - float((tmp_path / "a_at").read_text()) >= 2.0
+
+
+@pytest.mark.anyio
+@pytest.mark.timeout(120)
+async def test_pending_child_defers_release_until_task_terminal(tmp_path, stop_case):
+    _, _, run = stop_case
+    result = await run(
+        """
+        emit_stdout({"type": "task_started", "task_id": "t1"})
+        write_marker()
+        wait_for("b_deposited")
+        emit_stdout({"type": "result", "result": "done\\n" + marker, "session_id": sid})
+        time.sleep(1.5)
+        touch("task_done_at")
+        emit_stdout({"type": "task_notification", "task_id": "t1", "status": "completed"})
+        time.sleep(0.5)
+        """
+    )
+    ended = time.time()
+
+    assert result.pending_task_ids == ()
+    assert result.completion_ceiling_expired is False
+    assert result.returncode == 0
+    assert ended >= float((tmp_path / "task_done_at").read_text())
diff --git a/tests/execution/test_termination_action.py b/tests/execution/test_termination_action.py
index 21cf70999e..e3145e337e 100644
--- a/tests/execution/test_termination_action.py
+++ b/tests/execution/test_termination_action.py
@@ -24,8 +24,9 @@
         (TerminationReason.COMPLETED, False, False, TerminationAction.DRAIN_THEN_KILL_IF_ALIVE),
         # (NATURAL_EXIT, timeout=False, exited=True) → NO_KILL
         (TerminationReason.NATURAL_EXIT, False, True, TerminationAction.NO_KILL),
-        # (NATURAL_EXIT, timeout=False, exited=False) → NO_KILL
-        (TerminationReason.NATURAL_EXIT, False, False, TerminationAction.NO_KILL),
+        # A live leader needs an honest infrastructure kill regardless of the exit label.
+        (TerminationReason.NATURAL_EXIT, False, False, TerminationAction.IMMEDIATE_KILL),
+        (TerminationReason.SIGNAL_DEATH, False, False, TerminationAction.IMMEDIATE_KILL),
         # (TIMED_OUT, timeout=True, exited=True) → IMMEDIATE_KILL
         (TerminationReason.TIMED_OUT, True, True, TerminationAction.IMMEDIATE_KILL),
         # (TIMED_OUT, timeout=True, exited=False) → IMMEDIATE_KILL
@@ -45,7 +46,8 @@
         "completed_exited_no_kill",
         "completed_alive_drain_kill",
         "natural_exit_exited_no_kill",
-        "natural_exit_alive_no_kill",
+        "natural_exit_alive_is_infra_kill",
+        "signal_death_alive_is_infra_kill",
         "timed_out_exited_immediate_kill",
         "timed_out_alive_immediate_kill",
         "idle_stall_alive_immediate_kill",

From 7a04c0068eca35d2120c1a9cffdff00d2239eaa6 Mon Sep 17 00:00:00 2001
From: Trecek <talontrecek@gmail.com>
Date: Fri, 16 Oct 2026 07:54:32 -0700
Subject: [PATCH 3/6] fix: keep Channel B from releasing live managed workers

---
 .../core/types/execution/_type_subprocess.py      | 14 +++++++-------
 .../core/types/foundation/_type_enums.py          |  6 +++---
 src/autoskillit/execution/process/AGENTS.md       |  3 +++
 src/autoskillit/execution/process/__init__.py     | 15 ++-------------
 .../execution/process/_lifecycle/runner.py        |  6 ++++++
 .../execution/process/_process_race.py            | 10 ++++------
 .../execution/process/_race_watchers.py           | 12 +++++-------
 src/autoskillit/execution/process/_termination.py |  9 +++++----
 src/autoskillit/execution/session/AGENTS.md       |  2 +-
 9 files changed, 36 insertions(+), 41 deletions(-)

diff --git a/src/autoskillit/core/types/execution/_type_subprocess.py b/src/autoskillit/core/types/execution/_type_subprocess.py
index abcc9ab1b8..169b0e3693 100644
--- a/src/autoskillit/core/types/execution/_type_subprocess.py
+++ b/src/autoskillit/core/types/execution/_type_subprocess.py
@@ -41,11 +41,11 @@
 #:   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
diff --git a/src/autoskillit/core/types/foundation/_type_enums.py b/src/autoskillit/core/types/foundation/_type_enums.py
index 332767a998..f7302b2ef4 100644
--- a/src/autoskillit/core/types/foundation/_type_enums.py
+++ b/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.
diff --git a/src/autoskillit/execution/process/AGENTS.md b/src/autoskillit/execution/process/AGENTS.md
index 46c31f6527..245965cc8e 100644
--- a/src/autoskillit/execution/process/AGENTS.md
+++ b/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.
diff --git a/src/autoskillit/execution/process/__init__.py b/src/autoskillit/execution/process/__init__.py
index 115be7083e..33d15e4925 100644
--- a/src/autoskillit/execution/process/__init__.py
+++ b/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):
diff --git a/src/autoskillit/execution/process/_lifecycle/runner.py b/src/autoskillit/execution/process/_lifecycle/runner.py
index 8e274a3e58..06a5bcabfc 100644
--- a/src/autoskillit/execution/process/_lifecycle/runner.py
+++ b/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,
diff --git a/src/autoskillit/execution/process/_process_race.py b/src/autoskillit/execution/process/_process_race.py
index d4cbab90dd..af72a2b4cd 100644
--- a/src/autoskillit/execution/process/_process_race.py
+++ b/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:
diff --git a/src/autoskillit/execution/process/_race_watchers.py b/src/autoskillit/execution/process/_race_watchers.py
index 63a94b4e18..dab2073395 100644
--- a/src/autoskillit/execution/process/_race_watchers.py
+++ b/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)
diff --git a/src/autoskillit/execution/process/_termination.py b/src/autoskillit/execution/process/_termination.py
index eaeed8613b..9ec5f152f9 100644
--- a/src/autoskillit/execution/process/_termination.py
+++ b/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 (
diff --git a/src/autoskillit/execution/session/AGENTS.md b/src/autoskillit/execution/session/AGENTS.md
index cd50923fcb..708e25dcfa 100644
--- a/src/autoskillit/execution/session/AGENTS.md
+++ b/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.

From d2731d1475aa93f4b3e8ea70e15118d24b3347a1 Mon Sep 17 00:00:00 2001
From: Trecek <talontrecek@gmail.com>
Date: Fri, 16 Oct 2026 08:06:22 -0700
Subject: [PATCH 4/6] test: reuse the completed join wave helper

---
 .../core/types/execution/_type_subprocess.py  |  2 +-
 tests/hooks/test_join_enforcement_wiring.py   | 19 ++++---------------
 2 files changed, 5 insertions(+), 16 deletions(-)

diff --git a/src/autoskillit/core/types/execution/_type_subprocess.py b/src/autoskillit/core/types/execution/_type_subprocess.py
index 169b0e3693..34499a56e7 100644
--- a/src/autoskillit/core/types/execution/_type_subprocess.py
+++ b/src/autoskillit/core/types/execution/_type_subprocess.py
@@ -34,7 +34,7 @@
 #:   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)
diff --git a/tests/hooks/test_join_enforcement_wiring.py b/tests/hooks/test_join_enforcement_wiring.py
index e60acdb82b..7d8e924ccf 100644
--- a/tests/hooks/test_join_enforcement_wiring.py
+++ b/tests/hooks/test_join_enforcement_wiring.py
@@ -39,6 +39,9 @@
 from tests.hooks._session_binding_helpers import (
     RETIRED_JOIN_ENV as _RETIRED_JOIN_ENV,
 )
+from tests.hooks._session_binding_helpers import (
+    complete_one_assignment_wave,
+)
 from tests.hooks._session_binding_helpers import (
     declare_one_assignment as _declare_one_assignment,
 )
@@ -917,22 +920,8 @@ def test_join_block_paths_are_codex_blocking(tmp_path: Path, case: str) -> None:
 
 def test_stop_guard_releases_when_the_wave_is_complete(tmp_path: Path) -> None:
     session_id = "stop-release"
-    tool_use_id = "agent-1"
     worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    flag_dir = _declare_one_assignment(worktree, session_id=session_id)
-    claim_assignment(
-        flag_dir,
-        session_id=session_id,
-        top_level_parent="top_level",
-        tool_use_id=tool_use_id,
-    )
-    settle_assignment(
-        flag_dir,
-        session_id=session_id,
-        top_level_parent="top_level",
-        tool_use_id=tool_use_id,
-        outcome=OUTCOME_SUCCESS,
-    )
+    complete_one_assignment_wave(worktree, session_id=session_id)
 
     completed = _run_hook(
         tmp_path,

From d0fea7211ffb27e9d9e07c1224cd5b5c380ba6e6 Mon Sep 17 00:00:00 2001
From: Trecek <talontrecek@gmail.com>
Date: Fri, 16 Oct 2026 11:38:42 -0700
Subject: [PATCH 5/6] fix(review): observe natural exit after race watchers
 stop

---
 src/autoskillit/execution/process/_termination.py | 12 ++++++++----
 tests/execution/test_process_stop_hook_join.py    |  2 +-
 2 files changed, 9 insertions(+), 5 deletions(-)

diff --git a/src/autoskillit/execution/process/_termination.py b/src/autoskillit/execution/process/_termination.py
index 9ec5f152f9..36366cc3f1 100644
--- a/src/autoskillit/execution/process/_termination.py
+++ b/src/autoskillit/execution/process/_termination.py
@@ -92,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
                 )
diff --git a/tests/execution/test_process_stop_hook_join.py b/tests/execution/test_process_stop_hook_join.py
index c7351ee4cf..c80832dee5 100644
--- a/tests/execution/test_process_stop_hook_join.py
+++ b/tests/execution/test_process_stop_hook_join.py
@@ -273,8 +273,8 @@ async def finish_wave():
     assert (tmp_path / "continuation_ran").exists()
     assert result.returncode == 0
     assert result.channel_confirmation is ChannelConfirmation.CHANNEL_A
+    assert result.kill_reason is KillReason.NATURAL_EXIT
     assert "Session terminated" not in result.stdout + result.stderr
-    # F-1: the drain's cancelled exit watcher can still label this KILL_AFTER_COMPLETION.
 
 
 @pytest.mark.anyio

From 708de3619a95f35724cd6deae2a3395086e94e32 Mon Sep 17 00:00:00 2001
From: Trecek <talontrecek@gmail.com>
Date: Fri, 16 Oct 2026 11:39:45 -0700
Subject: [PATCH 6/6] refactor(review): use shared join helper names
 consistently

---
 tests/hooks/test_join_enforcement_wiring.py | 160 +++++++++-----------
 1 file changed, 74 insertions(+), 86 deletions(-)

diff --git a/tests/hooks/test_join_enforcement_wiring.py b/tests/hooks/test_join_enforcement_wiring.py
index 7d8e924ccf..fe33727c04 100644
--- a/tests/hooks/test_join_enforcement_wiring.py
+++ b/tests/hooks/test_join_enforcement_wiring.py
@@ -37,25 +37,13 @@
     codex_verdict,
 )
 from tests.hooks._session_binding_helpers import (
-    RETIRED_JOIN_ENV as _RETIRED_JOIN_ENV,
-)
-from tests.hooks._session_binding_helpers import (
+    RETIRED_JOIN_ENV,
     complete_one_assignment_wave,
-)
-from tests.hooks._session_binding_helpers import (
-    declare_one_assignment as _declare_one_assignment,
-)
-from tests.hooks._session_binding_helpers import (
-    load_join_bearing_skill as _load_join_bearing_skill,
-)
-from tests.hooks._session_binding_helpers import (
-    load_join_bearing_skill_with_context as _load_join_bearing_skill_with_context,
-)
-from tests.hooks._session_binding_helpers import (
-    run_join_hook as _run_hook,
-)
-from tests.hooks._session_binding_helpers import (
-    write_cook_registry as _write_cook_registry,
+    declare_one_assignment,
+    load_join_bearing_skill,
+    load_join_bearing_skill_with_context,
+    run_join_hook,
+    write_cook_registry,
 )
 
 if TYPE_CHECKING:
@@ -150,7 +138,7 @@ def _configure_non_cook_shape(
         registry_path.parent.mkdir(parents=True, exist_ok=True)
         registry_path.write_text("not-json", encoding="utf-8")
     elif case == "ambiguous_registry_identity":
-        _write_cook_registry(worktree, launch_id=launch_id, session_id=session_id)
+        write_cook_registry(worktree, launch_id=launch_id, session_id=session_id)
         registry = json.loads(registry_path.read_text(encoding="utf-8"))
         registry["duplicate-launch"] = {
             "session_type": "cook",
@@ -158,25 +146,25 @@ def _configure_non_cook_shape(
         }
         registry_path.write_text(json.dumps(registry), encoding="utf-8")
     elif case == "launch_session_mismatch":
-        _write_cook_registry(worktree, launch_id=launch_id, session_id="other-session")
+        write_cook_registry(worktree, launch_id=launch_id, session_id="other-session")
     elif case == "headless":
-        _write_cook_registry(worktree, launch_id=launch_id, session_id=session_id)
+        write_cook_registry(worktree, launch_id=launch_id, session_id=session_id)
         env["AUTOSKILLIT_HEADLESS"] = "1"
     elif case in {"non_cook_interactive", "order"}:
-        _write_cook_registry(worktree, launch_id=launch_id, session_id=session_id)
+        write_cook_registry(worktree, launch_id=launch_id, session_id=session_id)
         registry = json.loads(registry_path.read_text(encoding="utf-8"))
         registry[launch_id]["session_type"] = (
             "skill" if case == "non_cook_interactive" else "order"
         )
         registry_path.write_text(json.dumps(registry), encoding="utf-8")
     elif case == "descendant":
-        _write_cook_registry(worktree, launch_id=launch_id, session_id=session_id)
+        write_cook_registry(worktree, launch_id=launch_id, session_id=session_id)
         payload["agent_id"] = "child-agent"
     elif case == "managed_leaf":
         launch_id = session_id
         payload_session_id = "codex-thread"
         payload["session_id"] = payload_session_id
-        _write_cook_registry(worktree, launch_id=launch_id, session_id=None)
+        write_cook_registry(worktree, launch_id=launch_id, session_id=None)
         binding_path = resolve_binding_path(str(worktree), session_id)
         binding = read_binding(binding_path)
         assert binding is not None
@@ -201,9 +189,9 @@ def _configure_non_cook_shape(
 
 def test_claim_guard_denies_an_undeclared_agent_call_in_a_real_session(tmp_path: Path) -> None:
     session_id = "claim-deny"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_claim_guard.py",
         _agent_payload(worktree, session_id=session_id, tool_use_id="agent-1"),
@@ -221,10 +209,10 @@ def test_claim_guard_permits_an_agent_call_that_belongs_to_a_declared_wave(
 ) -> None:
     session_id = "claim-permit"
     tool_use_id = "agent-1"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    flag_dir = _declare_one_assignment(worktree, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    flag_dir = declare_one_assignment(worktree, session_id=session_id)
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_claim_guard.py",
         _agent_payload(worktree, session_id=session_id, tool_use_id=tool_use_id),
@@ -250,8 +238,8 @@ def test_settle_guard_maps_every_registered_event_type(
 ) -> None:
     session_id = f"settle-{event_name}"
     tool_use_id = "agent-1"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    flag_dir = _declare_one_assignment(worktree, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    flag_dir = declare_one_assignment(worktree, session_id=session_id)
     claim_assignment(
         flag_dir,
         session_id=session_id,
@@ -261,7 +249,7 @@ def test_settle_guard_maps_every_registered_event_type(
     payload = _agent_payload(worktree, session_id=session_id, tool_use_id=tool_use_id)
     payload.update({"hook_event_name": event_name, **event_data})
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_settle_guard.py",
         payload,
@@ -291,7 +279,7 @@ def test_authenticated_top_level_cook_full_join_sequence_never_opens_a_wave(
     monkeypatch.setenv("AUTOSKILLIT_AGENT_BACKEND", "claude-code")
     monkeypatch.setenv("AUTOSKILLIT_SESSION_TYPE", "skill")
     monkeypatch.setenv("AUTOSKILLIT_LOG_DIR", str(tmp_path / "logs"))
-    worktree, context = _load_join_bearing_skill_with_context(
+    worktree, context = load_join_bearing_skill_with_context(
         tmp_path,
         session_id=session_id,
         activation_source=activation_source,
@@ -344,7 +332,7 @@ def test_authenticated_top_level_cook_full_join_sequence_never_opens_a_wave(
         ),
     )
     for script_name, payload in guard_events[:2]:
-        completed = _run_hook(
+        completed = run_join_hook(
             tmp_path,
             _GUARDS_DIR / script_name,
             payload,
@@ -366,7 +354,7 @@ def test_authenticated_top_level_cook_full_join_sequence_never_opens_a_wave(
     assert redeclared["wave"] is None
 
     for script_name, payload in guard_events[2:]:
-        completed = _run_hook(
+        completed = run_join_hook(
             tmp_path,
             _GUARDS_DIR / script_name,
             payload,
@@ -409,12 +397,12 @@ def test_cook_redeclaration_is_not_refused_by_a_pre_fix_stranded_wave(
     monkeypatch.setenv("AUTOSKILLIT_AGENT_BACKEND", "claude-code")
     monkeypatch.setenv("AUTOSKILLIT_SESSION_TYPE", "skill")
     monkeypatch.setenv("AUTOSKILLIT_LOG_DIR", str(tmp_path / "logs"))
-    worktree = _load_join_bearing_skill(
+    worktree = load_join_bearing_skill(
         tmp_path,
         session_id=session_id,
         cook_launch_id=launch_id,
     )
-    flag_dir = _declare_one_assignment(worktree, session_id=session_id)
+    flag_dir = declare_one_assignment(worktree, session_id=session_id)
     legacy_batch = active_batch(
         flag_dir,
         session_id=session_id,
@@ -457,7 +445,7 @@ def test_cook_redeclaration_is_not_refused_by_a_pre_fix_stranded_wave(
         ("join_stop_guard.py", {"session_id": session_id, "cwd": str(worktree)}),
     )
     for script_name, payload in events:
-        completed = _run_hook(
+        completed = run_join_hook(
             tmp_path,
             _GUARDS_DIR / script_name,
             payload,
@@ -483,7 +471,7 @@ def test_authenticated_cook_bypass_survives_an_absent_managed_scope(tmp_path: Pa
     launch_id = "cook-launch"
     worktree = tmp_path / "worktree"
     (worktree / ".autoskillit").mkdir(parents=True)
-    _write_cook_registry(worktree, launch_id=launch_id, session_id=session_id)
+    write_cook_registry(worktree, launch_id=launch_id, session_id=session_id)
     env = {"AUTOSKILLIT_LAUNCH_ID": launch_id}
     events = (
         (
@@ -511,7 +499,7 @@ def test_authenticated_cook_bypass_survives_an_absent_managed_scope(tmp_path: Pa
     )
 
     for script_name, payload in events:
-        completed = _run_hook(
+        completed = run_join_hook(
             tmp_path,
             _GUARDS_DIR / script_name,
             payload,
@@ -553,7 +541,7 @@ def test_cook_authentication_rejects_non_top_level_shapes(
     )
 
     session_id = f"non-cook-{case}"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
     payload, binding_session_id, env = _configure_non_cook_shape(
         worktree,
         session_id=session_id,
@@ -576,7 +564,7 @@ def test_cook_authentication_rejects_non_top_level_shapes(
     if case == "descendant":
         return
 
-    claim = _run_hook(
+    claim = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_claim_guard.py",
         {
@@ -593,7 +581,7 @@ def test_cook_authentication_rejects_non_top_level_shapes(
     assert isinstance(claim_output, dict)
     assert claim_output["permissionDecision"] == "deny"
 
-    stop = _run_hook(
+    stop = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_stop_guard.py",
         payload,
@@ -638,8 +626,8 @@ def test_descendant_is_natively_exempt_but_never_authenticated_as_top_level_cook
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     session_id = "descendant-cook-shape"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    flag_dir = _declare_one_assignment(worktree, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    flag_dir = declare_one_assignment(worktree, session_id=session_id)
     payload, binding_session_id, env = _configure_non_cook_shape(
         worktree,
         session_id=session_id,
@@ -677,7 +665,7 @@ def test_descendant_is_natively_exempt_but_never_authenticated_as_top_level_cook
         ),
     )
     for script_name, event in events:
-        completed = _run_hook(
+        completed = run_join_hook(
             tmp_path,
             _GUARDS_DIR / script_name,
             event,
@@ -691,7 +679,7 @@ def test_descendant_is_natively_exempt_but_never_authenticated_as_top_level_cook
     assert batch is not None
     assert batch["wave_outcome"] == "pending"
     assert batch["assignments"][0]["tool_use_id"] is None
-    stop = _run_hook(
+    stop = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_stop_guard.py",
         payload,
@@ -704,7 +692,7 @@ def test_descendant_is_natively_exempt_but_never_authenticated_as_top_level_cook
 def test_claim_and_settle_guards_use_the_managed_binding_identity(tmp_path: Path) -> None:
     payload_session_id = "codex-thread-id"
     managed_join_id = "managed-join-id"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=payload_session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=payload_session_id)
     payload_binding = read_binding(resolve_binding_path(str(worktree), payload_session_id))
     assert payload_binding is not None
     write_binding(
@@ -735,7 +723,7 @@ def test_claim_and_settle_guards_use_the_managed_binding_identity(tmp_path: Path
         tool_use_id="agent-1",
     )
 
-    claimed = _run_hook(
+    claimed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_claim_guard.py",
         payload,
@@ -745,7 +733,7 @@ def test_claim_and_settle_guards_use_the_managed_binding_identity(tmp_path: Path
     assert claimed.exit_code == 0, claimed.stderr
     assert not claimed.stdout
 
-    settled = _run_hook(
+    settled = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_settle_guard.py",
         {
@@ -768,10 +756,10 @@ def test_claim_and_settle_guards_use_the_managed_binding_identity(tmp_path: Path
 
 def test_stop_guard_blocks_on_an_unresolved_wave_using_payload_identity(tmp_path: Path) -> None:
     session_id = "stop-block"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    _declare_one_assignment(worktree, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    declare_one_assignment(worktree, session_id=session_id)
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_stop_guard.py",
         {"session_id": session_id, "cwd": str(worktree)},
@@ -783,10 +771,10 @@ def test_stop_guard_blocks_on_an_unresolved_wave_using_payload_identity(tmp_path
 
 def test_stop_guard_blocks_an_invalid_managed_scope(tmp_path: Path) -> None:
     session_id = "stop-invalid-scope"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
     _mark_binding_scope_invalid(worktree, session_id=session_id)
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_stop_guard.py",
         {"session_id": session_id, "cwd": str(worktree)},
@@ -823,7 +811,7 @@ def test_join_block_paths_are_codex_blocking(tmp_path: Path, case: str) -> None:
         }.get(case, event)
     else:
         session_id = f"block-{case}"
-        worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
+        worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
         event = {"hook_event_name": "Stop", "session_id": session_id, "cwd": str(worktree)}
         if case in {
             "stop_invalid_managed_scope",
@@ -832,7 +820,7 @@ def test_join_block_paths_are_codex_blocking(tmp_path: Path, case: str) -> None:
         }:
             _mark_binding_scope_invalid(worktree, session_id=session_id)
         if case in {"stop_unresolved_wave", "followup_unresolved_wave"}:
-            _declare_one_assignment(worktree, session_id=session_id)
+            declare_one_assignment(worktree, session_id=session_id)
 
         if case.startswith("followup_"):
             hook = _GUARDS_DIR / "join_followup_guard.py"
@@ -877,7 +865,7 @@ def test_join_block_paths_are_codex_blocking(tmp_path: Path, case: str) -> None:
             hook = _GUARDS_DIR / "join_stop_guard.py"
             payload = event
 
-    emission = _run_hook(
+    emission = run_join_hook(
         tmp_path,
         hook,
         payload,
@@ -920,10 +908,10 @@ def test_join_block_paths_are_codex_blocking(tmp_path: Path, case: str) -> None:
 
 def test_stop_guard_releases_when_the_wave_is_complete(tmp_path: Path) -> None:
     session_id = "stop-release"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
     complete_one_assignment_wave(worktree, session_id=session_id)
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_stop_guard.py",
         {"session_id": session_id, "cwd": str(worktree)},
@@ -938,7 +926,7 @@ def test_stop_guard_blocks_on_a_malformed_payload(tmp_path: Path) -> None:
     worktree = tmp_path / "worktree"
     (worktree / ".autoskillit").mkdir(parents=True)
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_stop_guard.py",
         "not valid json",
@@ -950,10 +938,10 @@ def test_stop_guard_blocks_on_a_malformed_payload(tmp_path: Path) -> None:
 
 def test_followup_guard_blocks_a_followup_while_a_wave_is_unresolved(tmp_path: Path) -> None:
     session_id = "followup-block"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    _declare_one_assignment(worktree, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    declare_one_assignment(worktree, session_id=session_id)
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_followup_guard.py",
         {
@@ -980,8 +968,8 @@ def test_followup_guard_allows_exact_recovery_declaration_after_terminal_failure
     tool_name: str,
 ) -> None:
     session_id = "followup-recovery"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    flag_dir = _declare_one_assignment(worktree, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    flag_dir = declare_one_assignment(worktree, session_id=session_id)
     claim_assignment(
         flag_dir,
         session_id=session_id,
@@ -996,7 +984,7 @@ def test_followup_guard_allows_exact_recovery_declaration_after_terminal_failure
         outcome=OUTCOME_FAILURE,
     )
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_followup_guard.py",
         {
@@ -1038,8 +1026,8 @@ def test_followup_guard_keeps_other_effects_blocked_after_terminal_failure(
     tool_name: str,
 ) -> None:
     session_id = f"followup-terminal-block-{tool_name}"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    flag_dir = _declare_one_assignment(worktree, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    flag_dir = declare_one_assignment(worktree, session_id=session_id)
     claim_assignment(
         flag_dir,
         session_id=session_id,
@@ -1054,7 +1042,7 @@ def test_followup_guard_keeps_other_effects_blocked_after_terminal_failure(
         outcome=OUTCOME_FAILURE,
     )
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_followup_guard.py",
         {
@@ -1073,10 +1061,10 @@ def test_followup_guard_blocks_recovery_declaration_while_wave_is_pending(
     tmp_path: Path,
 ) -> None:
     session_id = "followup-pending-recovery"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    _declare_one_assignment(worktree, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    declare_one_assignment(worktree, session_id=session_id)
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_followup_guard.py",
         {
@@ -1095,11 +1083,11 @@ def test_followup_guard_blocks_recovery_declaration_when_ledger_is_corrupt(
     tmp_path: Path,
 ) -> None:
     session_id = "followup-corrupt-recovery"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    flag_dir = _declare_one_assignment(worktree, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    flag_dir = declare_one_assignment(worktree, session_id=session_id)
     (flag_dir / "join_ledger.json").write_text("not-json", encoding="utf-8")
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_followup_guard.py",
         {
@@ -1116,8 +1104,8 @@ def test_followup_guard_blocks_recovery_declaration_when_ledger_is_corrupt(
 
 def test_stop_guard_stays_blocked_after_terminal_failure(tmp_path: Path) -> None:
     session_id = "stop-terminal-failure"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    flag_dir = _declare_one_assignment(worktree, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    flag_dir = declare_one_assignment(worktree, session_id=session_id)
     claim_assignment(
         flag_dir,
         session_id=session_id,
@@ -1132,7 +1120,7 @@ def test_stop_guard_stays_blocked_after_terminal_failure(tmp_path: Path) -> None
         outcome=OUTCOME_FAILURE,
     )
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_stop_guard.py",
         {"session_id": session_id, "cwd": str(worktree)},
@@ -1149,8 +1137,8 @@ def test_stop_guard_is_safe_under_python_optimization(tmp_path: Path) -> None:
     subsequent ``admission.binding_dict`` dereference.
     """
     session_id = "stop-O-flag"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
-    completed = _run_hook(
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_stop_guard.py",
         {"hook_event_name": "Stop", "session_id": session_id, "cwd": str(worktree)},
@@ -1170,7 +1158,7 @@ def test_stop_guard_is_safe_under_python_optimization(tmp_path: Path) -> None:
 def test_followup_guard_prefers_the_managed_join_identity(tmp_path: Path) -> None:
     payload_session_id = "codex-thread-id"
     managed_join_id = "managed-join-id"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=payload_session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=payload_session_id)
     binding = read_binding(resolve_binding_path(str(worktree), payload_session_id))
     assert binding is not None
     write_binding(
@@ -1192,7 +1180,7 @@ def test_followup_guard_prefers_the_managed_join_identity(tmp_path: Path) -> Non
         assignments=("worker",),
     )
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "join_followup_guard.py",
         {
@@ -1213,9 +1201,9 @@ def test_followup_guard_prefers_the_managed_join_identity(tmp_path: Path) -> Non
 
 def test_background_exec_guard_binds_without_any_join_env_var(tmp_path: Path) -> None:
     session_id = "background-bind"
-    worktree = _load_join_bearing_skill(tmp_path, session_id=session_id)
+    worktree = load_join_bearing_skill(tmp_path, session_id=session_id)
 
-    completed = _run_hook(
+    completed = run_join_hook(
         tmp_path,
         _GUARDS_DIR / "background_exec_guard.py",
         {
@@ -1239,7 +1227,7 @@ def test_no_guard_reads_a_retired_join_env_var() -> None:
     for source in _HOOKS_DIR.rglob("*.py"):
         collector = _EnvVarReadCollector()
         collector.visit(ast.parse(source.read_text(encoding="utf-8"), filename=str(source)))
-        retired_reads = collector.reads & _RETIRED_JOIN_ENV
+        retired_reads = collector.reads & RETIRED_JOIN_ENV
         if retired_reads:
             reads[source.relative_to(_PROJECT_ROOT)] = retired_reads