mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
16916dc257
Coord initial-message + create-time-attachments coordination: - ``CoordinatorAdapter.send`` gains optional ``attachments`` + ``send_id`` kwargs so the worker dispatched at create time can carry the uploaded files onto the first turn. Mirrors interactive's pre-existing worker-thread pattern. The ``send_id`` reservation token soft-locks the rows; the adapter's failure path unreserves so a worker crash returns them to pending. - ``_coord_create_post_install`` reserves any uploaded ``attachment_ids`` via the lifted ``reserve_and_resolve_attachments`` helper before dispatching through the adapter — closes the parity gap with interactive's create-with-attachments+initial_message flow. - ``_reserve_and_resolve_attachments`` lifted from ``turnstone/server.py`` to ``turnstone/core/attachments.py`` as ``reserve_and_resolve_attachments`` so both processes use one kind-agnostic implementation. Copilot review fixes on PR #416: - Skill lookup now calls ``storage.get_prompt_template_by_name`` directly rather than going through ``turnstone.core.memory.get_skill_by_name``; that helper swallows storage exceptions into ``None`` which would have masked outages as the 400 "Skill not found" branch. Calling storage directly lets exceptions bubble to the lifted body's correlation_id'd 500 path so operators chasing skill-related reports can distinguish real misses from registry outages. - ``_interactive_create_build_kwargs`` / ``_coord_create_build_kwargs`` thread ``skill_data["name"]`` (the canonical row name) into ``mgr.create`` instead of the raw ``body["skill"]`` value. Pre-fix a whitespace-padded request body ``"skill": " my-skill "`` would have persisted the dirty name even though the lookup ran on the stripped key. - ``make_create_handler`` docstring corrected: audit-emit failures return 200 (not 201). - ``_audit_workstream_created`` docstring corrected: factory keeps the successful 200 response on audit-emit failure (was 201). New regression test: ``test_create_with_multipart_attachments_and_initial_message_reserves`` asserts attachments are reserved (not pending) when both ``initial_message`` and uploads land in the same coord create request. Updated ``_SendSession`` stub in ``test_coordinator_adapter.py`` to match the new ``send`` / ``queue_message`` signatures. 4501 tests passing; ruff + mypy clean.
551 lines
20 KiB
Python
551 lines
20 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:
|
|
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
|
|
)
|
|
|
|
|
|
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"
|