mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
88facd260e
Inline child approve/deny in the coord tree UI was rendering downstream
of the bulk-live cache (``GET /v1/api/cluster/ws/live``), not the SSE
stream. ``child_ws_state`` events were tiny notifications that fired
an urgent live-bulk fetch on every activity_state transition into/out
of "approval", just to pick up the rich ``pending_approval_detail``
payload. With multiple coord tabs and multi-child workstreams, that
urgent-fetch pattern compounded the SSE-executor pressure Shape A
is unwinding.
Thread the field through every layer so the SSE event itself carries
the rich payload — browser mutates ``liveBadgeCache`` directly,
no urgent fetch:
1. Node ``WebUI._broadcast_state`` emits ``pending_approval_detail``
on ``ws_state`` events. Gated on ``_pending_approval is not None``
so the per-broadcast verdict-cache deepcopy only runs when there
is actually an approval pending. ``_build_node_snapshot`` also
projects the field so the console's reconnect-via-snapshot
resync path delivers it (without this the new collector
forwarding would never see the field on a snapshot row).
2. Console ``ClusterCollector._apply_delta`` (live ``ws_state``
forwarding) and ``_reconcile_node`` (snapshot resync diff) both
forward the field on the emitted ``cluster_state`` event, AND
``_apply_delta`` persists it on the cached ``ws`` dict so the
``get_node_detail`` / ``get_snapshot`` endpoints between
reconciliations don't render stale approve/deny buttons.
3. ``CoordinatorAdapter._dispatch_child_event`` re-emits the field
on the ``child_ws_state`` event sent to coord listener queues.
4. Frontend ``handleChildState`` reads ``ev.pending_approval_detail``
and writes it directly into ``liveBadgeCache``, tagging the
entry with ``sseUpdatedAt``. ``flushLiveFetches`` honors that
tag for ``SSE_AUTHORITATIVE_MS`` (3s) — the upstream
``/dashboard`` cache has its own ~2s TTL, so a bulk-poll
landing right after a transition can otherwise clobber the
fresh SSE-set state with pre-transition data.
The pre-fix ``enteredApproval`` / ``leftApproval`` urgent-fetch
branch is removed. The 409 stale-call_id retry path keeps its own
urgent fetch — that's a different scenario.
Tests cover the forwarding contract at every layer, the broadcast
gate (event includes the field when an approval is pending,
omits it otherwise, and clears after resolution), and the
``flushLiveFetches`` merge-guard structural shape so a refactor
that keeps the symbols but inverts the comparison or drops the
``prev.live`` check can't pass silently.
612 lines
22 KiB
Python
612 lines
22 KiB
Python
"""Tests for CoordinatorAdapter.
|
|
|
|
Mirrors test_interactive_adapter.py: focuses on the transport contract
|
|
(what gets sent to the ClusterCollector) and cleanup_ui behavior
|
|
(unblock listener queues, cancel session). The SessionManager-level
|
|
tests in test_session_manager.py cover the lifecycle path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import queue
|
|
import threading
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
from turnstone.console.coordinator_adapter import CoordinatorAdapter
|
|
from turnstone.core.workstream import Workstream, WorkstreamKind, WorkstreamState
|
|
|
|
|
|
class _StubCoordUI:
|
|
"""Stub matching the subset of ConsoleCoordinatorUI the adapter touches."""
|
|
|
|
def __init__(self) -> None:
|
|
self._approval_event = threading.Event()
|
|
self._approval_result: tuple[bool, str | None] = (True, "initial")
|
|
self._plan_event = threading.Event()
|
|
self._plan_result: str = "accept"
|
|
self._fg_event = threading.Event()
|
|
self._listeners_lock = threading.Lock()
|
|
self._listeners: list[queue.Queue[dict[str, Any]]] = []
|
|
|
|
|
|
class _StubSession:
|
|
def __init__(self) -> None:
|
|
self.cancelled = False
|
|
self.closed = False
|
|
|
|
def cancel(self) -> None:
|
|
self.cancelled = True
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def _make_adapter(
|
|
collector: Any = None,
|
|
*,
|
|
ui_factory: Any = None,
|
|
session_factory: Any = None,
|
|
) -> tuple[CoordinatorAdapter, MagicMock]:
|
|
collector = collector or MagicMock()
|
|
adapter = CoordinatorAdapter(
|
|
collector=collector,
|
|
ui_factory=ui_factory or (lambda ws: _StubCoordUI()),
|
|
session_factory=session_factory or (lambda *a, **kw: _StubSession()),
|
|
)
|
|
return adapter, collector
|
|
|
|
|
|
def _make_ws(**overrides: Any) -> Workstream:
|
|
ws = Workstream(id="coord-1", name="my-coord")
|
|
ws.kind = WorkstreamKind.COORDINATOR
|
|
ws.user_id = "u1"
|
|
ws.ui = _StubCoordUI()
|
|
ws.session = _StubSession()
|
|
for k, v in overrides.items():
|
|
setattr(ws, k, v)
|
|
return ws
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Transport — emit_created / emit_state / emit_closed
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_emit_created_calls_collector_with_coord_fields() -> None:
|
|
adapter, collector = _make_adapter()
|
|
ws = _make_ws()
|
|
adapter.emit_created(ws)
|
|
collector.emit_console_ws_created.assert_called_once_with(
|
|
"coord-1",
|
|
name="my-coord",
|
|
user_id="u1",
|
|
kind=WorkstreamKind.COORDINATOR.value,
|
|
state=WorkstreamState.IDLE.value,
|
|
parent_ws_id=None,
|
|
)
|
|
|
|
|
|
def test_emit_state_calls_collector_state() -> None:
|
|
"""Post-rich-payload, emit_state passes tokens / context_ratio /
|
|
activity / activity_state / content kwargs read from ws.ui's
|
|
snapshot. Default values (zeros / empty strings) when the UI
|
|
hasn't recorded any per-ws metrics yet."""
|
|
adapter, collector = _make_adapter()
|
|
ws = _make_ws()
|
|
adapter.emit_state(ws, WorkstreamState.RUNNING)
|
|
collector.emit_console_ws_state.assert_called_once_with(
|
|
"coord-1",
|
|
WorkstreamState.RUNNING.value,
|
|
tokens=0,
|
|
context_ratio=0.0,
|
|
activity="",
|
|
activity_state="",
|
|
content="",
|
|
)
|
|
|
|
|
|
def test_emit_closed_calls_collector_closed() -> None:
|
|
adapter, collector = _make_adapter()
|
|
adapter.emit_closed("coord-1")
|
|
collector.emit_console_ws_closed.assert_called_once_with("coord-1")
|
|
|
|
|
|
def test_emit_closed_swallows_reason_kwarg() -> None:
|
|
"""The console collector doesn't propagate a 'reason' — the console
|
|
frontend's evicted special-case only fires for real-node
|
|
workstreams. Protocol compatibility only."""
|
|
adapter, collector = _make_adapter()
|
|
adapter.emit_closed("coord-1", reason="evicted")
|
|
collector.emit_console_ws_closed.assert_called_once_with("coord-1")
|
|
|
|
|
|
def test_emit_tolerates_collector_exception() -> None:
|
|
collector = MagicMock()
|
|
collector.emit_console_ws_created.side_effect = RuntimeError("collector dead")
|
|
collector.emit_console_ws_state.side_effect = RuntimeError("collector dead")
|
|
collector.emit_console_ws_closed.side_effect = RuntimeError("collector dead")
|
|
adapter, _ = _make_adapter(collector=collector)
|
|
ws = _make_ws()
|
|
# All three must swallow — the session lifecycle must not break
|
|
# because the collector had a transient failure.
|
|
adapter.emit_created(ws)
|
|
adapter.emit_state(ws, WorkstreamState.RUNNING)
|
|
adapter.emit_closed("coord-1")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cleanup_ui
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cleanup_ui_unblocks_events_and_broadcasts_to_listeners() -> None:
|
|
adapter, _ = _make_adapter()
|
|
ws = _make_ws()
|
|
ws.ui._approval_event.clear() # type: ignore[attr-defined]
|
|
ws.ui._plan_event.clear() # type: ignore[attr-defined]
|
|
ws.ui._fg_event.clear() # type: ignore[attr-defined]
|
|
lq: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=5)
|
|
ws.ui._listeners.append(lq) # type: ignore[attr-defined]
|
|
|
|
adapter.cleanup_ui(ws)
|
|
|
|
assert ws.ui._approval_event.is_set() # type: ignore[attr-defined]
|
|
assert ws.ui._plan_event.is_set() # type: ignore[attr-defined]
|
|
assert ws.ui._fg_event.is_set() # type: ignore[attr-defined]
|
|
assert ws.ui._approval_result == (False, None) # type: ignore[attr-defined]
|
|
assert ws.ui._plan_result == "reject" # type: ignore[attr-defined]
|
|
assert lq.get_nowait() == {"type": "ws_closed"}
|
|
assert ws.ui._listeners == [] # type: ignore[attr-defined]
|
|
assert ws.session.cancelled is True # type: ignore[attr-defined]
|
|
assert ws.session.closed is True # type: ignore[attr-defined]
|
|
|
|
|
|
def test_cleanup_ui_listener_full_queue_evicts_head() -> None:
|
|
adapter, _ = _make_adapter()
|
|
ws = _make_ws()
|
|
lq: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=1)
|
|
lq.put_nowait({"type": "stale"})
|
|
ws.ui._listeners.append(lq) # type: ignore[attr-defined]
|
|
adapter.cleanup_ui(ws)
|
|
assert lq.get_nowait() == {"type": "ws_closed"}
|
|
|
|
|
|
def test_cleanup_ui_tolerates_missing_session_and_ui() -> None:
|
|
adapter, _ = _make_adapter()
|
|
ws = _make_ws()
|
|
ws.session = None
|
|
ws.ui = None
|
|
adapter.cleanup_ui(ws) # no crash
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Construction passthrough
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_session_forwards_skill_model_kind_parent() -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
def _sf(ui: Any, model: str | None, ws_id: str, **kwargs: Any) -> Any:
|
|
captured["ui"] = ui
|
|
captured["model"] = model
|
|
captured["ws_id"] = ws_id
|
|
captured.update(kwargs)
|
|
return _StubSession()
|
|
|
|
adapter, _ = _make_adapter(session_factory=_sf)
|
|
ws = _make_ws()
|
|
ws.parent_ws_id = None
|
|
adapter.build_session(ws, skill="coordinator", model="gpt-5")
|
|
assert captured["ui"] is ws.ui
|
|
assert captured["model"] == "gpt-5"
|
|
assert captured["skill"] == "coordinator"
|
|
assert captured["kind"] == WorkstreamKind.COORDINATOR
|
|
assert captured["parent_ws_id"] is None
|
|
# client_type intentionally NOT forwarded — coord session_factory
|
|
# doesn't accept it (fixed as 'console').
|
|
assert "client_type" not in captured
|
|
|
|
|
|
def test_build_ui_delegates_to_ui_factory() -> None:
|
|
captured_ws: list[Workstream] = []
|
|
|
|
def _ui_factory(ws: Workstream) -> Any:
|
|
captured_ws.append(ws)
|
|
return _StubCoordUI()
|
|
|
|
adapter, _ = _make_adapter(ui_factory=_ui_factory)
|
|
ws = _make_ws()
|
|
result = adapter.build_ui(ws)
|
|
assert captured_ws == [ws]
|
|
assert isinstance(result, _StubCoordUI)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Worker dispatch — _spawn_worker / send
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _SendSession:
|
|
"""ChatSession stub with send / queue_message accounting."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
queue_full: bool = False,
|
|
send_gate: threading.Event | None = None,
|
|
) -> None:
|
|
self.send_calls: list[str] = []
|
|
self.queue_calls: list[str] = []
|
|
self._queue_full = queue_full
|
|
# When set, ``send`` blocks on this event — lets the test pin a
|
|
# worker inside session.send while a second thread races through
|
|
# _spawn_worker, proving the lock gate (not Thread.is_alive) is
|
|
# what serialises them.
|
|
self._send_gate = send_gate
|
|
self._send_lock = threading.Lock()
|
|
self.cancelled = False
|
|
self.closed = False
|
|
|
|
def send(
|
|
self,
|
|
message: str,
|
|
attachments: Any = None,
|
|
send_id: str | None = None,
|
|
) -> None:
|
|
if self._send_gate is not None:
|
|
self._send_gate.wait(timeout=2.0)
|
|
with self._send_lock:
|
|
self.send_calls.append(message)
|
|
|
|
def queue_message(
|
|
self,
|
|
message: str,
|
|
attachment_ids: Any = None,
|
|
queue_msg_id: str | None = None,
|
|
) -> None:
|
|
if self._queue_full:
|
|
raise queue.Full
|
|
self.queue_calls.append(message)
|
|
|
|
def cancel(self) -> None:
|
|
self.cancelled = True
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
class _StubManager:
|
|
"""Minimal SessionManager stub exposing ``get`` for adapter.send."""
|
|
|
|
def __init__(self, ws: Workstream | None = None) -> None:
|
|
self._ws = ws
|
|
|
|
def get(self, ws_id: str) -> Workstream | None:
|
|
if self._ws is not None and self._ws.id == ws_id:
|
|
return self._ws
|
|
return None
|
|
|
|
|
|
class TestCoordinatorAdapterWorkerDispatch:
|
|
def test_spawn_worker_reuses_when_worker_running(self) -> None:
|
|
adapter, _ = _make_adapter()
|
|
ws = _make_ws()
|
|
session = _SendSession()
|
|
ws.session = session # type: ignore[assignment]
|
|
ws._worker_running = True # pre-existing worker
|
|
adapter.attach(_StubManager(ws)) # type: ignore[arg-type]
|
|
|
|
assert adapter.send(ws.id, "hello") is True
|
|
assert session.queue_calls == ["hello"]
|
|
assert session.send_calls == []
|
|
# worker_thread not replaced
|
|
assert ws.worker_thread is None
|
|
|
|
def test_spawn_worker_returns_false_on_queue_full(self) -> None:
|
|
adapter, _ = _make_adapter()
|
|
ws = _make_ws()
|
|
session = _SendSession(queue_full=True)
|
|
ws.session = session # type: ignore[assignment]
|
|
ws._worker_running = True
|
|
adapter.attach(_StubManager(ws)) # type: ignore[arg-type]
|
|
|
|
assert adapter.send(ws.id, "hello") is False
|
|
assert session.send_calls == []
|
|
|
|
def test_spawn_worker_concurrent_calls_produce_one_worker(self) -> None:
|
|
"""Bug-1 reproducer: two simultaneous send() calls under ws._lock
|
|
must land as exactly one ChatSession.send and one queued message,
|
|
not two parallel workers on the same ChatSession."""
|
|
adapter, _ = _make_adapter()
|
|
ws = _make_ws()
|
|
send_gate = threading.Event()
|
|
session = _SendSession(send_gate=send_gate)
|
|
ws.session = session # type: ignore[assignment]
|
|
adapter.attach(_StubManager(ws)) # type: ignore[arg-type]
|
|
|
|
results: list[bool] = []
|
|
start_barrier = threading.Barrier(2)
|
|
results_lock = threading.Lock()
|
|
|
|
def _caller(msg: str) -> None:
|
|
start_barrier.wait(timeout=1.0)
|
|
r = adapter.send(ws.id, msg)
|
|
with results_lock:
|
|
results.append(r)
|
|
|
|
t1 = threading.Thread(target=_caller, args=("first",))
|
|
t2 = threading.Thread(target=_caller, args=("second",))
|
|
t1.start()
|
|
t2.start()
|
|
# Both callers return quickly: the winner spawns the worker
|
|
# (returns True immediately) and the loser queues (returns True).
|
|
t1.join(timeout=3.0)
|
|
t2.join(timeout=3.0)
|
|
assert not t1.is_alive() and not t2.is_alive()
|
|
# At this point session.send is still blocked on send_gate —
|
|
# the second caller MUST have taken the queue path.
|
|
assert len(session.queue_calls) == 1
|
|
# Release the worker and let it finish.
|
|
send_gate.set()
|
|
if ws.worker_thread is not None:
|
|
ws.worker_thread.join(timeout=3.0)
|
|
|
|
assert results == [True, True]
|
|
assert len(session.send_calls) == 1
|
|
assert set(session.send_calls + session.queue_calls) == {"first", "second"}
|
|
assert ws._worker_running is False
|
|
|
|
def test_worker_finally_clears_running_flag(self) -> None:
|
|
adapter, _ = _make_adapter()
|
|
ws = _make_ws()
|
|
session = _SendSession()
|
|
ws.session = session # type: ignore[assignment]
|
|
adapter.attach(_StubManager(ws)) # type: ignore[arg-type]
|
|
|
|
assert adapter.send(ws.id, "hello") is True
|
|
assert ws.worker_thread is not None
|
|
ws.worker_thread.join(timeout=2.0)
|
|
assert ws._worker_running is False
|
|
assert session.send_calls == ["hello"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Children registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCoordinatorAdapterChildrenRegistry:
|
|
def test_emit_created_seeds_empty_children_set(self) -> None:
|
|
adapter, _ = _make_adapter()
|
|
ws = _make_ws()
|
|
adapter.emit_created(ws)
|
|
assert ws.id in adapter._children
|
|
assert adapter._children[ws.id] == set()
|
|
assert adapter._active_coords[ws.id] is ws.ui
|
|
|
|
def test_emit_rehydrated_calls_rebuild(self) -> None:
|
|
adapter, _ = _make_adapter()
|
|
calls: list[str] = []
|
|
# Monkeypatch the rebuild hook to count invocations without
|
|
# requiring a real storage backend.
|
|
adapter._rebuild_children_registry = calls.append # type: ignore[method-assign, assignment]
|
|
ws = _make_ws()
|
|
adapter.emit_created(ws)
|
|
assert calls == []
|
|
adapter.emit_rehydrated(ws)
|
|
assert calls == [ws.id]
|
|
|
|
def test_emit_closed_clears_forward_and_reverse_indexes(self) -> None:
|
|
adapter, _ = _make_adapter()
|
|
with adapter._children_lock:
|
|
adapter._merge_child_ids_locked("coord-a", ["child-a1", "child-a2"])
|
|
adapter._merge_child_ids_locked("coord-b", ["child-b1"])
|
|
adapter._active_coords["coord-a"] = object()
|
|
adapter._active_coords["coord-b"] = object()
|
|
|
|
adapter.emit_closed("coord-a")
|
|
|
|
assert "coord-a" not in adapter._children
|
|
assert "coord-a" not in adapter._active_coords
|
|
assert "child-a1" not in adapter._child_to_coord
|
|
assert "child-a2" not in adapter._child_to_coord
|
|
# coord-b untouched
|
|
assert adapter._child_to_coord["child-b1"] == "coord-b"
|
|
assert "coord-b" in adapter._children
|
|
|
|
def test_merge_child_ids_locked_is_idempotent(self) -> None:
|
|
adapter, _ = _make_adapter()
|
|
with adapter._children_lock:
|
|
adapter._merge_child_ids_locked("coord-a", ["child-1"])
|
|
adapter._merge_child_ids_locked("coord-a", ["child-1"])
|
|
assert adapter._children["coord-a"] == {"child-1"}
|
|
assert adapter._child_to_coord == {"child-1": "coord-a"}
|
|
|
|
def test_prime_children_from_snapshot_merges_without_overwriting(self) -> None:
|
|
adapter, _ = _make_adapter()
|
|
# Seed one in-memory coord + one existing child
|
|
coord_ws = _make_ws()
|
|
coord_ws.id = "coord-a"
|
|
mgr = MagicMock()
|
|
mgr.list_all.return_value = [coord_ws]
|
|
adapter.attach(mgr)
|
|
with adapter._children_lock:
|
|
adapter._merge_child_ids_locked("coord-a", ["child-a1"])
|
|
|
|
snapshot = {
|
|
"nodes": [
|
|
{
|
|
"workstreams": [
|
|
{"id": "child-a2", "parent_ws_id": "coord-a"},
|
|
# Unknown parent — skipped
|
|
{"id": "child-x", "parent_ws_id": "coord-unknown"},
|
|
# Missing fields — skipped
|
|
{"id": "", "parent_ws_id": "coord-a"},
|
|
],
|
|
},
|
|
],
|
|
}
|
|
adapter._prime_children_from_snapshot(snapshot)
|
|
assert adapter._children["coord-a"] == {"child-a1", "child-a2"}
|
|
assert adapter._child_to_coord["child-a2"] == "coord-a"
|
|
assert "child-x" not in adapter._child_to_coord
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatch — _dispatch_child_event
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _UIRecorder:
|
|
"""UI stub capturing _enqueue payloads for dispatch assertions."""
|
|
|
|
def __init__(self) -> None:
|
|
self.enqueued: list[dict[str, Any]] = []
|
|
|
|
def _enqueue(self, payload: dict[str, Any]) -> None:
|
|
self.enqueued.append(payload)
|
|
|
|
|
|
class TestCoordinatorAdapterDispatchChildEvent:
|
|
def _setup(
|
|
self, coord_id: str = "coord-a"
|
|
) -> tuple[CoordinatorAdapter, _UIRecorder, Workstream]:
|
|
adapter, _ = _make_adapter()
|
|
coord_ws = _make_ws()
|
|
coord_ws.id = coord_id
|
|
recorder = _UIRecorder()
|
|
coord_ws.ui = recorder # type: ignore[assignment]
|
|
with adapter._children_lock:
|
|
adapter._children.setdefault(coord_id, set())
|
|
adapter._active_coords[coord_id] = recorder
|
|
adapter.attach(_StubManager(coord_ws)) # type: ignore[arg-type]
|
|
return adapter, recorder, coord_ws
|
|
|
|
def test_dispatch_unknown_parent_drops_event(self) -> None:
|
|
adapter, recorder, _ = self._setup()
|
|
adapter._dispatch_child_event(
|
|
{"type": "ws_created", "ws_id": "orphan", "parent_ws_id": "coord-unknown"}
|
|
)
|
|
adapter._dispatch_child_event({"type": "cluster_state", "ws_id": "orphan"})
|
|
adapter._dispatch_child_event({"type": "ws_closed", "ws_id": "orphan"})
|
|
assert recorder.enqueued == []
|
|
|
|
def test_dispatch_ws_created_routes_to_parent_coord_ui(self) -> None:
|
|
adapter, recorder, _ = self._setup()
|
|
adapter._dispatch_child_event(
|
|
{
|
|
"type": "ws_created",
|
|
"ws_id": "child-a1",
|
|
"parent_ws_id": "coord-a",
|
|
"name": "kid",
|
|
"node_id": "node-1",
|
|
}
|
|
)
|
|
assert len(recorder.enqueued) == 1
|
|
payload = recorder.enqueued[0]
|
|
assert payload["type"] == "child_ws_created"
|
|
assert payload["child_ws_id"] == "child-a1"
|
|
assert payload["parent_ws_id"] == "coord-a"
|
|
# Reverse index updated for subsequent cluster_state events.
|
|
assert adapter._child_to_coord["child-a1"] == "coord-a"
|
|
|
|
def test_dispatch_cluster_state_routes_via_reverse_index(self) -> None:
|
|
adapter, recorder, _ = self._setup()
|
|
with adapter._children_lock:
|
|
adapter._merge_child_ids_locked("coord-a", ["child-a1"])
|
|
adapter._dispatch_child_event(
|
|
{
|
|
"type": "cluster_state",
|
|
"ws_id": "child-a1",
|
|
"state": "running",
|
|
"tokens": 42,
|
|
"node_id": "node-1",
|
|
}
|
|
)
|
|
assert len(recorder.enqueued) == 1
|
|
payload = recorder.enqueued[0]
|
|
assert payload["type"] == "child_ws_state"
|
|
assert payload["state"] == "running"
|
|
assert payload["tokens"] == 42
|
|
|
|
def test_dispatch_ws_closed_routes_to_parent_coord(self) -> None:
|
|
adapter, recorder, _ = self._setup()
|
|
with adapter._children_lock:
|
|
adapter._merge_child_ids_locked("coord-a", ["child-a1"])
|
|
adapter._dispatch_child_event(
|
|
{"type": "ws_closed", "ws_id": "child-a1", "reason": "evicted"}
|
|
)
|
|
assert len(recorder.enqueued) == 1
|
|
payload = recorder.enqueued[0]
|
|
assert payload["type"] == "child_ws_closed"
|
|
assert payload["reason"] == "evicted"
|
|
assert payload["parent_ws_id"] == "coord-a"
|
|
|
|
def test_dispatch_adds_ws_id_in_place(self) -> None:
|
|
"""perf-6: _enqueue_on_ui mutates the payload dict in place with
|
|
the coord's ws_id so the browser can discriminate child events."""
|
|
adapter, recorder, _ = self._setup()
|
|
with adapter._children_lock:
|
|
adapter._merge_child_ids_locked("coord-a", ["child-a1"])
|
|
adapter._dispatch_child_event(
|
|
{
|
|
"type": "cluster_state",
|
|
"ws_id": "child-a1",
|
|
"state": "running",
|
|
}
|
|
)
|
|
assert recorder.enqueued[0]["ws_id"] == "coord-a"
|
|
|
|
def test_dispatch_cluster_state_forwards_pending_approval_detail(self) -> None:
|
|
"""The rich approval payload now rides on child_ws_state directly so
|
|
the browser can mutate liveBadgeCache without a separate live-bulk
|
|
fetch. Drift here means the inline approve/deny buttons would
|
|
regress to chasing the dashboard cache (the load-storm pattern
|
|
Shape A is unwinding)."""
|
|
adapter, recorder, _ = self._setup()
|
|
with adapter._children_lock:
|
|
adapter._merge_child_ids_locked("coord-a", ["child-a1"])
|
|
detail = {
|
|
"items": [{"call_id": "c1", "header": "tool x"}],
|
|
"judge_pending": False,
|
|
}
|
|
adapter._dispatch_child_event(
|
|
{
|
|
"type": "cluster_state",
|
|
"ws_id": "child-a1",
|
|
"state": "running",
|
|
"activity_state": "approval",
|
|
"pending_approval_detail": detail,
|
|
}
|
|
)
|
|
assert len(recorder.enqueued) == 1
|
|
payload = recorder.enqueued[0]
|
|
assert payload["type"] == "child_ws_state"
|
|
assert payload["activity_state"] == "approval"
|
|
assert payload["pending_approval_detail"] == detail
|
|
|
|
def test_dispatch_cluster_state_pending_approval_detail_none_passes_through(
|
|
self,
|
|
) -> None:
|
|
"""Missing pending_approval_detail (no approval pending, or pre-fix
|
|
node mid-rolling-upgrade) must forward as None — not raise, not
|
|
omit — so the browser's handleChildState treats it as "no SSE-
|
|
supplied detail, fall back to cached value"."""
|
|
adapter, recorder, _ = self._setup()
|
|
with adapter._children_lock:
|
|
adapter._merge_child_ids_locked("coord-a", ["child-a1"])
|
|
adapter._dispatch_child_event(
|
|
{
|
|
"type": "cluster_state",
|
|
"ws_id": "child-a1",
|
|
"state": "running",
|
|
"activity_state": "tool",
|
|
}
|
|
)
|
|
assert len(recorder.enqueued) == 1
|
|
payload = recorder.enqueued[0]
|
|
assert "pending_approval_detail" in payload
|
|
assert payload["pending_approval_detail"] is None
|