turnstone/tests/_sse_recovery_server.py

"""Boot the REAL interactive Turnstone server for the SSE recovery e2e
harness: real ``SessionManager`` + real ``ChatSession`` engine driven
through a scripted chat-completions client at the SDK boundary, executing
REAL bash tools, exposed over a real uvicorn socket.

The recipe (verified end-to-end) has four load-bearing pieces:

1. **Provider injection seam.** ``create_app`` takes a PRE-BUILT
   ``SessionManager``, so the harness owns the ``session_factory``: it
   passes ``client=fake_client`` and OMITS the registry, so
   ``ChatSession`` falls back to ``create_provider("openai-compatible")``
   == ``OpenAIChatCompletionsProvider`` — exactly what
   ``tests._session_helpers.scripted_chat_client`` targets. No production
   monkeypatch of the engine.

2. **Auto-title suppression.** The first user message spawns a background
   ``_generate_title`` LLM call that would consume the first scripted
   response (the tool call) and desync a positional script. Setting
   ``session._title_generated = True`` before the first send disables it.

3. **Completion barrier.** ``/send`` returns immediately after spawning
   ``ws.worker_thread``; joining that thread is the true "turn complete,
   every SSE event enqueued" barrier (``stream_end`` is per-LLM-call, not
   per-turn, so it is NOT a completion marker).

4. **Thread hygiene.** ``create_app``'s lifespan starts daemon fan-out
   threads (``_global_fanout_thread`` blocking on ``global_queue.get()``,
   ``_aggregate_emitter_thread`` on a 10s loop).  The harness used to
   swap them for no-ops (they had no shutdown and tripped conftest's
   leaked-thread guard), which kept the global lane dead here; #885 gave
   the lifespan a real shutdown (stop Event + a queue sentinel for the
   fanout, joined in the lifespan exit that ``stop()``'s
   ``should_exit``/join drives), so the harness now runs them REAL — the
   ``roster-restart`` scenario depends on a live global lane — and
   teardown stays clean with no ``allow_thread_leak``.
"""

from __future__ import annotations

import asyncio
import contextlib
import json
import queue as _q
import socket
import threading
import time
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any

import httpx
import uvicorn

from tests._session_helpers import scripted_chat_client
from turnstone.core.adapters.interactive_adapter import InteractiveAdapter
from turnstone.core.auth import JWT_AUD_SERVER, create_jwt
from turnstone.core.session import ChatSession
from turnstone.core.session_manager import SessionManager
from turnstone.core.session_ui_base import SessionUIBase
from turnstone.core.storage import get_storage
from turnstone.core.workstream import WorkstreamKind
from turnstone.prompts import ClientType
from turnstone.server import WebUI, create_app

if TYPE_CHECKING:
    from collections.abc import MutableMapping

    from turnstone.core.workstream import Workstream

_JWT_SECRET = "sse-recovery-e2e-jwt-secret-minimum-32-chars!"
# Small server send buffer so a stalled consumer's in-flight backlog before
# the listener-queue poison stays bounded (paired with the client's small
# SO_RCVBUF in _sse_recovery_helpers). Harmless for prompt readers.
_DEFAULT_SNDBUF = 8192


def _fake_client(scripts: tuple[Any, ...]) -> Any:
    """An SDK-shaped fake whose ``chat.completions.create`` follows a
    positional script (each a :func:`fake_chat_stream` kwargs dict)."""
    create_fn = scripted_chat_client(*scripts)
    client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=create_fn)))
    client.calls = create_fn.calls
    return client


class RecoveryServer:
    """A booted interactive node the recovery scenarios drive."""

    def __init__(
        self,
        *,
        sndbuf: int = _DEFAULT_SNDBUF,
        listener_cap: int | None = None,
        extra_routes: list[Any] | None = None,
        port: int = 0,
        sock: socket.socket | None = None,
    ) -> None:
        self._global_queue: _q.Queue[dict[str, Any]] = _q.Queue(maxsize=100000)
        self._global_listeners: list[_q.Queue[dict[str, Any]]] = []
        self._global_listeners_lock = threading.Lock()
        # Per-ws scripted client, resolved at factory-call time.
        self._pending_client: Any = _fake_client((dict(content="ok", finish_reason="stop"),))
        self._clients: dict[str, Any] = {}

        WebUI._global_queue = self._global_queue

        def session_factory(
            ui: Any,
            model_alias: str | None = None,
            ws_id: str | None = None,
            *,
            skill: Any = None,
            client_type: str = "",
            kind: WorkstreamKind = WorkstreamKind.INTERACTIVE,
            parent_ws_id: str | None = None,
            project_id: str = "",
            **_extra: Any,
        ) -> ChatSession:
            client = self._pending_client
            if ws_id is not None:
                self._clients[ws_id] = client
            return ChatSession(
                client=client,
                model="test-model",
                ui=ui,
                instructions=None,
                temperature=None,
                max_tokens=1024,
                tool_timeout=30,
                ws_id=ws_id,
                user_id="recovery-user",
                client_type=ClientType.WEB,
                kind=kind,
                # Don't truncate large tool outputs: the harness tests
                # recovery, not the tool-result truncation budget, and a
                # truncated /history would diverge from the full live event
                # and defeat the convergence assertions.
                tool_truncation=10_000_000,
            )

        self._adapter = InteractiveAdapter(
            global_queue=self._global_queue,
            ui_factory=lambda ws: WebUI(
                ws_id=ws.id, user_id=ws.user_id, kind=ws.kind, parent_ws_id=ws.parent_ws_id
            ),
            session_factory=session_factory,
        )
        self._manager = SessionManager(
            self._adapter, storage=get_storage(), max_active=32, node_id="recovery-node"
        )
        self._adapter.attach(self._manager)
        WebUI._workstream_mgr = self._manager

        # delay_load knob state (see the method): wraps the storage
        # singleton's load_messages; restored in stop().
        self._load_delay_ms = 0
        self._load_calls = 0
        # load_messages runs on asyncio.to_thread WORKERS, and delay_load
        # exists precisely to overlap two of them — unlike the
        # single-writer HTTP counters, this one has genuine concurrent
        # writers, so the increment takes a lock (a lost update would
        # false-FAIL G7's load_delta === 2, or mask a third load).
        self._load_calls_lock = threading.Lock()
        _storage_obj = get_storage()
        self._orig_load_messages = _storage_obj.load_messages

        def _delayed_load(*a: Any, **k: Any) -> Any:
            with self._load_calls_lock:
                self._load_calls += 1
            result = self._orig_load_messages(*a, **k)
            # Sleep AFTER the load: the held flight must hold the data it
            # actually read (its transaction point), so a flight parked
            # across a rewind genuinely carries PRE-rewind rows — a
            # pre-load sleep would read post-rewind storage and mask a
            # wrongly-joined flight as fresh truth.
            d = self._load_delay_ms
            if d > 0:
                time.sleep(d / 1000.0)
            return result

        _storage_obj.load_messages = _delayed_load  # type: ignore[method-assign]
        self._patched_storage = _storage_obj

        # Optional small listener-queue cap. The cap is a default arg on the
        # registration methods with no config/env override, so lower it by
        # patching their ``__defaults__`` (restored on stop). fix-3's
        # de-amplification makes a real 500-cap overflow need a pathological
        # storm; a small cap exercises the identical _ListenerOverflow ->
        # stream_overflow -> reconnect-replay path within a bounded storm.
        self._orig_defaults: list[tuple[Any, tuple[Any, ...] | None]] = []
        if listener_cap is not None:
            for meth in (
                SessionUIBase._register_listener,
                SessionUIBase.register_listener_with_in_progress_snapshot,
                SessionUIBase.register_listener_with_replay,
            ):
                self._orig_defaults.append((meth, meth.__defaults__))
                meth.__defaults__ = (listener_cap,)

        self._app = create_app(
            workstreams=self._manager,
            global_queue=self._global_queue,
            global_listeners=self._global_listeners,
            global_listeners_lock=self._global_listeners_lock,
            skip_permissions=True,
            jwt_secret=_JWT_SECRET,
            node_id="recovery-node",
            # /history + tenant checks read app.state.auth_storage.
            auth_storage=get_storage(),
        )
        # Same-origin extras (Tier 2 serves its recovery page here so the real
        # Pane's cookie auth + EventSource work without cross-origin plumbing).
        if extra_routes:
            self._app.router.routes.extend(extra_routes)

        # Pre-bind a listening socket with a small SO_SNDBUF (accepted conns
        # inherit it), then hand it to uvicorn.  ``sock`` injection: the
        # gap-free restart scenarios (roster-restart-native) bind a
        # placeholder BEFORE stopping the prior node and hand it in here —
        # a failed EventSource reconnect attempt is TERMINAL per WHATWG
        # (fail-the-connection → CLOSED, no further retries), so the
        # native-retry leg must never observe a refused-window; the
        # placeholder's listen backlog completes the TCP handshake during
        # the boot and uvicorn drains it once serving.
        self._sock = sock if sock is not None else make_listen_socket(port, sndbuf=sndbuf)
        self._port = int(self._sock.getsockname()[1])

        # -- fault injection (public knobs below) ----------------------------
        # In-process arming: the Tier-2 runner holds this RecoveryServer and
        # arms a knob, THEN drives the browser request that consumes it.
        # Single-writer by construction — the runner never arms a knob while
        # the loop thread is mid-consume — and CPython makes each int read /
        # write atomic, so these need no lock even though the uvicorn loop
        # thread increments/decrements them while the runner thread reads.
        self.history_requests = 0
        # /history responses the PRODUCTION route answered 200 (not the
        # fault layer's injected 500s).  See _fault_app for why arrival
        # counting cannot substitute.
        self.history_ok = 0
        self.rewind_requests = 0
        # Per-ws SSE connection opens (``GET …/events`` — the EventSource the
        # pane's connectSSE builds).  A TRANSPORT-FREE heal (the #890 idle-edge
        # staleness backstop, a quiesced REST refetch) must leave this FLAT; a
        # reload-based backstop would bump it once per reconnect (the round-5
        # storm).  Same lock-free single-writer int discipline as above.
        self.events_requests = 0
        # Global-lane SSE connection opens (``GET …/events/global`` — the
        # roster stream app.js's connectGlobalSSE builds).  The
        # roster-restart scenario (#881) asserts the post-restart manual
        # reconnect actually reached the reborn node's real endpoint.
        self.global_events_requests = 0
        self._history_fail_remaining = 0
        self._history_tokenless_remaining = 0
        self._history_delay_ms = 0
        # A thin pure-ASGI fault layer wrapping the REAL app (the production
        # app itself is untouched): count + optionally delay/fail
        # ``GET …/history``, count ``POST …/rewind``, count each per-ws SSE
        # connection open (``GET …/events``), forward everything else (SSE
        # bodies, /send, lifespan, static) verbatim.
        production_app = self._app

        async def _fault_app(scope: dict[str, Any], receive: Any, send: Any) -> None:
            if scope.get("type") == "http":
                path = scope.get("path", "")
                method = scope.get("method", "")
                if path.endswith("/history") and method == "GET":
                    # ARRIVAL, never move.  Scenarios that hold a request open
                    # use this bump as the IN-FLIGHT edge (E6/E7/G1/G7 say so
                    # at their poll sites); counting on forward instead would
                    # delay it past the hold and silently stop those scenarios
                    # testing anything.
                    self.history_requests += 1
                    if self._history_delay_ms > 0:
                        await asyncio.sleep(self._history_delay_ms / 1000.0)
                    if self._history_fail_remaining > 0:
                        self._history_fail_remaining -= 1
                        await send(
                            {
                                "type": "http.response.start",
                                "status": 500,
                                "headers": [(b"content-type", b"application/json")],
                            }
                        )
                        await send({"type": "http.response.body", "body": b'{"error": "injected"}'})
                        return
                    if self._history_tokenless_remaining > 0:
                        # Old/malformed server simulation for the strong
                        # handoff-repair latch. A 200 without handoff_token is
                        # not proof and must never authorize EventSource.
                        self._history_tokenless_remaining -= 1
                        self.history_ok += 1
                        await send(
                            {
                                "type": "http.response.start",
                                "status": 200,
                                "headers": [(b"content-type", b"application/json")],
                            }
                        )
                        await send(
                            {
                                "type": "http.response.body",
                                "body": (b'{"ws_id":"compat","messages":[],"cursor":null}'),
                            }
                        )
                        return

                    # Successful-RESPONSE counter, distinct from the arrival
                    # bump above.  A scenario asserting that a render was
                    # DECLINED needs to know a good payload actually existed —
                    # otherwise "the client refused to render" and "there was
                    # nothing to render" produce identical observables (no
                    # wipe, latch held).  Arrival cannot prove that, and
                    # neither can an injected-fail budget: a PRODUCTION-side
                    # 500/404 would slip through both.  Reading the real
                    # status off the response start is the only honest signal.
                    async def _counting_send(message: MutableMapping[str, Any]) -> None:
                        if (
                            message.get("type") == "http.response.start"
                            and message.get("status") == 200
                        ):
                            self.history_ok += 1
                        await send(message)

                    await production_app(scope, receive, _counting_send)
                    return
                elif path.endswith("/rewind") and method == "POST":
                    self.rewind_requests += 1
                elif path.endswith("/events") and method == "GET":
                    # Per-ws SSE connection open — count it (readable on
                    # RecoveryServer) and forward the long-lived stream
                    # verbatim below.  Uniquely the per-ws stream: the global
                    # lane is ``…/events/global`` (ends ``/global``), and the
                    # route the pane's EventSource hits is
                    # ``…/workstreams/{ws_id}/events`` (session_routes).
                    self.events_requests += 1
                elif path.endswith("/events/global") and method == "GET":
                    self.global_events_requests += 1
            await production_app(scope, receive, send)

        # ``timeout_graceful_shutdown``: an SSE stream that is still open
        # at ``stop()`` would otherwise park uvicorn's graceful drain
        # indefinitely (the 20s thread-join just expires and the browser
        # stays attached to the zombie server — the roster-restart-native
        # scenario is the one caller that stops a node mid-stream).  A
        # bounded drain force-closes the stream after 2s and the lifespan
        # shutdown (#885's daemon-thread teardown) still runs after it.
        self._server = uvicorn.Server(
            uvicorn.Config(
                _fault_app,
                log_level="warning",
                lifespan="on",
                timeout_graceful_shutdown=2,
            )
        )
        self._thread = threading.Thread(
            target=self._serve, name=f"uvicorn-recovery-{self._port}", daemon=True
        )
        self._thread.start()
        if not _tcp_ready(self._port, 10.0):
            self.stop()
            raise AssertionError("recovery server did not accept TCP")

        self._token = create_jwt(
            user_id="recovery-user",
            scopes=frozenset({"read", "write", "approve", "service"}),
            source="recovery",
            secret=_JWT_SECRET,
            audience=JWT_AUD_SERVER,
        )
        self._http = httpx.Client(base_url=self.base_url, timeout=httpx.Timeout(30.0))

    def _serve(self) -> None:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            loop.run_until_complete(self._server.serve(sockets=[self._sock]))
        finally:
            pending = asyncio.all_tasks(loop)
            for task in pending:
                task.cancel()
            if pending:
                with contextlib.suppress(Exception):
                    loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
            loop.close()

    # -- properties ----------------------------------------------------------

    @property
    def base_url(self) -> str:
        return f"http://127.0.0.1:{self._port}"

    @property
    def token(self) -> str:
        return self._token

    @property
    def manager(self) -> SessionManager:
        return self._manager

    # -- workstream lifecycle ------------------------------------------------

    def create_workstream(self, *scripts: Any, name: str = "recovery-ws") -> str:
        """Create a ws whose scripted LLM follows ``scripts`` (positional
        :func:`fake_chat_stream` kwargs). Auto-approves tools and suppresses
        the auto-title call so the positional script stays in sync."""
        self._pending_client = _fake_client(scripts)
        ws = self._manager.create(user_id="recovery-user", name=name)
        self._prime_ws(ws)
        return ws.id

    def open_workstream(self, ws_id: str, *scripts: Any) -> None:
        """Rehydrate a persisted ws on THIS node (the restart path). Fresh
        UI → empty ring + storage-seeded ``_event_id``."""
        if scripts:
            self._pending_client = _fake_client(scripts)
        ws = self._manager.open(ws_id)
        if ws is None:
            raise AssertionError(f"open_workstream: ws {ws_id} not resurrectable")
        self._prime_ws(ws)

    def _prime_ws(self, ws: Workstream) -> None:
        if isinstance(ws.ui, SessionUIBase):
            ws.ui.auto_approve = True  # blanket tool auto-approval
        if ws.session is not None:
            ws.session._title_generated = True  # suppress the auto-title LLM call

    def send(self, ws_id: str, message: str = "go") -> None:
        """POST /send — spawns the worker thread and returns immediately."""
        r = self._http.post(
            f"/v1/api/workstreams/{ws_id}/send",
            headers={"Authorization": f"Bearer {self._token}"},
            json={"message": message},
        )
        r.raise_for_status()

    def wait_turn(self, ws_id: str, *, timeout: float = 45.0) -> None:
        """Block until the turn's worker thread finishes (the true
        turn-complete barrier) and the ws is idle."""
        deadline = time.monotonic() + timeout
        worker: threading.Thread | None = None
        while time.monotonic() < deadline:
            ws = self._manager.get(ws_id)
            worker = ws.worker_thread if ws is not None else None
            if worker is not None:
                break
            time.sleep(0.02)
        if worker is not None:
            worker.join(timeout=max(0.5, deadline - time.monotonic()))
            if worker.is_alive():
                raise AssertionError(f"turn worker for {ws_id} did not finish in {timeout}s")

    def get_ws(self, ws_id: str) -> Workstream | None:
        return self._manager.get(ws_id)

    def ws_state(self, ws_id: str) -> str:
        ws = self._manager.get(ws_id)
        return ws.state.value if ws is not None else ""

    def ring_span(self, ws_id: str) -> tuple[int | None, int]:
        """(earliest retained ring event_id or None, latest counter) — lets a
        scenario wait for the ring to evict a specific cursor."""
        ws = self._manager.get(ws_id)
        ui = ws.ui if ws is not None else None
        if not isinstance(ui, SessionUIBase):
            return None, 0
        buf = ui._event_buffer
        earliest = buf[0][0] if buf else None
        return earliest, ui._event_id

    def listener_poisoned(self, ws_id: str) -> bool:
        """True once any live SSE listener on the ws has poisoned (overflow)."""
        ws = self._manager.get(ws_id)
        ui = ws.ui if ws is not None else None
        if not isinstance(ui, SessionUIBase):
            return False
        return any(getattr(q, "poisoned", False) for q in list(ui._listeners))

    def max_event_id(self, ws_id: str) -> int | None:
        """The storage high-water ``MAX(conversations.event_id)`` — what a
        restarted node's fresh UI seeds ``_event_id`` from."""
        result: int | None = get_storage().get_max_event_id(ws_id)
        return result

    def emit_idle_edge(self, ws_id: str) -> int:
        """Publish one real per-workstream ``state_change: idle`` event.

        Recovery scenarios use this test-server pulse when they need an
        organic-settle-equivalent edge without admitting another user row.
        A normal ``/send`` is not a neutral trigger: it publishes a live
        ``user_turn``, starts model work, and changes the transcript/counts
        these scenarios use to isolate the stale-history backstop.
        """
        ws = self._manager.get(ws_id)
        ui = ws.ui if ws is not None else None
        if not isinstance(ui, SessionUIBase):
            raise AssertionError(f"emit_idle_edge: ws {ws_id} has no session UI")
        before = ui._event_id
        ui.on_state_change("idle")
        if ui._event_id != before + 1:
            raise AssertionError(
                f"emit_idle_edge: expected one event after {before}, got {ui._event_id}"
            )
        return ui._event_id

    def emit_history_resync(self, ws_id: str, reason: str = "recovery_probe") -> int:
        """Publish the real strong repair frame through the ordered UI lane."""

        ws = self._manager.get(ws_id)
        ui = ws.ui if ws is not None else None
        if not isinstance(ui, SessionUIBase):
            raise AssertionError(f"emit_history_resync: ws {ws_id} has no session UI")
        before = ui._event_id
        ui.on_history_resync(reason)
        if ui._event_id != before + 1:
            raise AssertionError(
                f"emit_history_resync: expected one event after {before}, got {ui._event_id}"
            )
        return ui._event_id

    def emit_tool_pending(self, ws_id: str, call_id: str) -> int:
        """Publish a live ``tool_pending`` phase without persisting a turn.

        This drives the coordinator's event-owned ``liveToolCalls`` gate in
        isolation.  ``on_agent_step`` is the production hook that emits this
        exact envelope; the synthetic item is intentionally not added to
        history, so a later authoritative repaint must remove its DOM shell.
        """
        ws = self._manager.get(ws_id)
        ui = ws.ui if ws is not None else None
        if not isinstance(ui, SessionUIBase):
            raise AssertionError(f"emit_tool_pending: ws {ws_id} has no session UI")
        before = ui._event_id
        ui.on_agent_step(
            "",
            {
                "call_id": call_id,
                "func_name": "recovery_probe",
                "approval_label": "recovery probe",
                "header": "recovery render-gate probe",
                "needs_approval": False,
            },
        )
        if ui._event_id != before + 1:
            raise AssertionError(
                f"emit_tool_pending: expected one event after {before}, got {ui._event_id}"
            )
        return ui._event_id

    def emit_tool_result(self, ws_id: str, call_id: str) -> int:
        """Resolve a tool pulse emitted by :meth:`emit_tool_pending`."""
        ws = self._manager.get(ws_id)
        ui = ws.ui if ws is not None else None
        if not isinstance(ui, SessionUIBase):
            raise AssertionError(f"emit_tool_result: ws {ws_id} has no session UI")
        before = ui._event_id
        ui.on_tool_result(call_id, "recovery_probe", "probe complete")
        if ui._event_id != before + 1:
            raise AssertionError(
                f"emit_tool_result: expected one event after {before}, got {ui._event_id}"
            )
        return ui._event_id

    def fetch_history(self, ws_id: str) -> dict[str, Any]:
        r = self._http.get(
            f"/v1/api/workstreams/{ws_id}/history",
            headers={"Authorization": f"Bearer {self._token}"},
        )
        r.raise_for_status()
        result: dict[str, Any] = r.json()
        return result

    # -- fault-injection knobs -----------------------------------------------
    # Armed in-process by the Tier-2 runner (single writer at a time — see
    # __init__).  A plain int is deliberate: CPython makes the loop thread's
    # increment/decrement and the runner thread's read each atomic, and the
    # arm-then-consume ordering means they never race.

    def delay_load(self, ms: int) -> None:
        """Hold ``storage.load_messages`` itself open for ``ms`` (0 = off).

        ``delay_history`` sleeps in the FAULT LAYER — before the route —
        so two delayed requests never overlap inside the #884 flight
        machinery (the first flight completes and pops before the second
        arrives at the route).  This knob sleeps INSIDE the shared
        reconstruction's ``load_messages`` (sync, called via
        ``asyncio.to_thread`` — the sleep parks only that worker), which
        is the same layer the unit tests gate, so held flights genuinely
        overlap and join/miss behavior is observable end to end via
        ``load_calls``.
        """
        self._load_delay_ms = ms

    @property
    def load_calls(self) -> int:
        """``load_messages`` entries (pre-sleep) — the flight-layer twin
        of ``history_requests`` (which counts HTTP arrivals): a JOINED
        request never enters ``load_messages``, so join=1 / miss=2."""
        return self._load_calls

    def fail_history(self, count: int) -> None:
        """Make the next ``count`` ``GET …/history`` responses a 500 — the
        failed refetch the #890 guard-before-wipe must survive."""
        self._history_fail_remaining = count

    def tokenless_history(self, count: int) -> None:
        """Make the next history responses 200 without a handoff proof."""

        self._history_tokenless_remaining = count

    def delay_history(self, ms: int) -> None:
        """Hold each ``GET …/history`` ``ms`` ms before forwarding (0
        clears).  Opens the clear_ui-refetch quiesce window that the row
        affordance gate (``busy || _historyStale``) must close."""
        self._history_delay_ms = ms

    @property
    def history_fail_remaining(self) -> int:
        """Unconsumed forced-failure budget — 0 proves the armed failure
        actually fired (assert backend state, never scripted absence)."""
        return self._history_fail_remaining

    # -- teardown ------------------------------------------------------------

    def stop(self, *, hard: bool = False) -> None:
        """Stop the node.

        ``hard=True`` skips the per-workstream ``manager.close`` sweep — a
        graceful close routes through ``cleanup_session_ui`` →
        ``session.cancel()``, whose bash cancel path PERSISTS a
        synthesized "Cancelled by user" result while the old node is
        still alive, which masks crash states.  A hard stop leaves any
        in-flight tool call genuinely unresulted in storage, modelling a
        SIGKILL/OOM death (the coord-orphan-rewind scenario's premise).
        The 2s graceful-shutdown timeout (uvicorn config) force-closes
        open SSE streams, and the lifespan teardown still runs, so the
        #885 daemon threads are joined on both paths.
        """
        if not hard:
            with contextlib.suppress(Exception):
                for ws in list(self._manager.list_all()):
                    with contextlib.suppress(Exception):
                        self._manager.close(ws.id)
        # hard=True relies on ``timeout_graceful_shutdown=2`` (set in the
        # uvicorn config above) to force-close the pane's EventSource:
        # should_exit alone still runs the ASGI lifespan teardown, so the
        # #885 daemon threads and the sse_executor are joined either way
        # (``force_exit`` would SKIP the lifespan and leak them — the
        # fanout thread blocks on queue.get() forever).  NOTE: the killed
        # workstream's in-flight tool keeps executing on this process's
        # session thread and persists its result at natural completion —
        # hard-kill scenarios must use a paced tool that outlives their
        # observation window.
        self._server.should_exit = True
        self._thread.join(timeout=20)
        with contextlib.suppress(Exception):
            self._http.close()
        with contextlib.suppress(OSError):
            self._sock.close()
        # Restore any patched cap defaults.
        for meth, defaults in self._orig_defaults:
            meth.__defaults__ = defaults
        # Restore the storage singleton's load_messages (delay_load knob).
        with contextlib.suppress(Exception):
            self._patched_storage.load_messages = self._orig_load_messages  # type: ignore[method-assign]


def make_listen_socket(port: int, *, sndbuf: int = _DEFAULT_SNDBUF) -> socket.socket:
    """Bound + listening socket the way :class:`RecoveryServer` binds its own.

    ``SO_REUSEPORT`` on every listener (same process, same uid) is what
    lets a restart scenario bind the successor's socket while the prior
    node still holds the port — the seam behind the gap-free handoff
    documented at the ``sock`` parameter.  ``SO_SNDBUF`` matches the
    server's small send buffer so accepted connections inherit identical
    backpressure behavior regardless of which side bound the socket.
    """
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, sndbuf)
    s.bind(("127.0.0.1", port))  # port=0 -> ephemeral; fixed -> restart reuse
    s.listen(128)
    return s


def _tcp_ready(port: int, timeout: float) -> bool:
    end = time.monotonic() + timeout
    while time.monotonic() < end:
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=0.3):
                return True
        except OSError:
            time.sleep(0.05)
    return False


def bash_toolcall_script(
    call_id: str, command: str, *, finish_reason: str = "tool_calls"
) -> dict[str, Any]:
    """A scripted assistant turn issuing ONE bash tool call."""
    return dict(
        tool_calls=[{"id": call_id, "name": "bash", "arguments": json.dumps({"command": command})}],
        finish_reason=finish_reason,
    )


def parallel_bash_script(commands: dict[str, str]) -> dict[str, Any]:
    """A scripted assistant turn issuing SEVERAL bash tool calls at once
    (the parallel-pool storm), ``{call_id: command}``.

    Each command is prefixed with a no-op ``: <call_id>;`` so the tool
    ARGUMENTS are distinct per call while the OUTPUT is unchanged (``:``
    ignores its args and prints nothing). Identical-argument parallel
    calls otherwise trip the session's repeat-tool-call guard, which
    appends a warning to the PERSISTED result only (not the live event) —
    an orthogonal divergence that would mask the recovery behavior the
    convergence assertions test.
    """
    return dict(
        tool_calls=[
            {
                "id": cid,
                "name": "bash",
                "arguments": json.dumps({"command": f": {cid}; {cmd}"}),
            }
            for cid, cmd in commands.items()
        ],
        finish_reason="tool_calls",
    )


def final_text_script(content: str = "done") -> dict[str, Any]:
    """The scripted assistant turn that ends the agent loop (no tools)."""
    return dict(content=content, finish_reason="stop")