mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
802d87a57f
Lift the Children primitive out of CoordinatorAdapter into universal SessionManager core primitives, replace the fragile poll + state-event piggyback paths with first-class cluster bus event types for inline approval delivery, and clean up the resulting frontend reducer. Architecture - New `turnstone/core/children_registry.py` — universal parent → children + reverse-lookup primitive with atomic `add_child` (returns parent UI for race-free dispatch). Lifted from `CoordinatorAdapter`. - New `turnstone/core/child_source.py` — `ChildSource` Protocol with `SameNodeChildSource` (in-process via SessionManager state observer) and `ClusterChildSource` (cross-node via ClusterCollector listener). - `SessionManager._on_state_change` upgraded to multi-subscriber (`subscribe_to_state` / `unsubscribe_from_state`) under a dedicated lock; CLI consumer migrated. - `CoordinatorAdapter` shrunk: 731 → ~640 LOC. Children data lives in the registry; fan-out lives in ClusterChildSource. Backward-compat property facades dropped; tests updated to use the registry surface. Cluster bus event vocabulary - New event types `intent_verdict`, `approval_resolved`, `approve_request` flow through both `ClusterCollector._apply_delta` (translation from node SSE) and `emit_console_ws_*` (synthesis on console pseudo-node). - `CoordinatorAdapter._dispatch_child_event` re-emits as `child_ws_intent_verdict` / `child_ws_approval_resolved` / `child_ws_approve_request` on the parent coord's SSE stream. - New `_broadcast_intent_verdict` / `_broadcast_approval_resolved` / `_broadcast_approve_request` no-op hooks on `SessionUIBase`. WebUI pushes to the global queue; ConsoleCoordinatorUI pushes to the collector. `approve_tools` calls `_broadcast_approve_request` right after setting `_pending_approval` so the items reach the coord tree immediately, eliminating the bulk-fetch race. Cleanups - `pending_approval_detail` piggyback on `ws_state` / `cluster_state` removed end-to-end. Bulk fetch + explicit verdict / approve-request push are the canonical carriers. - Browser `_judgePollTick` 90-second poll loop deleted; push path is authoritative. - `urgent` flag on `scheduleLiveFetch` deleted (only caller was 409 retry; replaced with `invalidateLiveBadge` + standard schedule). - Console `_fetch_live_block` derives `pending_approval` from a disjunction (`activity_state="approval"` OR `state="attention"` OR detail present) so the bulk fetch can't return false during the state-transition race window. - Coord-side merge guard in `flushLiveFetches` no longer clobbered: `handleChildState` only stamps `sseUpdatedAt` when authoritatively clearing detail. - `child_locality` capability flag removed (was inert dead code). Reliability - Selective drop on listener queue overflow: critical event types (verdicts, approvals, ws_closed, child_ws_*) evict one oldest item to make room rather than dropping themselves on a full queue. Best-effort events (state ticks, content tokens, status, activity) drop as before. Applied to `SessionUIBase._enqueue`, `ClusterCollector._fanout`, and the `WebUI._global_queue` puts in the new broadcast hooks. - `_state_subscribers` snapshot under a dedicated lock so concurrent subscribe / unsubscribe during dispatch can't shift the iterator. UX / a11y - Loading placeholder in renderChildRow keeps row height stable while the bulk fetch is in-flight (sr-friendly aria-label). - Focus preservation across `_renderChildrenNow` (capture + restore by row + marker) and across targeted `_updateChildRow` swaps. - Layout-shift transition on the approval block max-height; respects `prefers-reduced-motion`. - Sidebar pending count: `(N children · M pending)`. - Risk pill `aria-label` spells out level + confidence for SR users. - Per-coord SSE listener queue depth surfaced in the status bar (`queue N/500`) with color escalation (warn at >50%, danger at >80%). Tests - 305+ test changes across 8 files. New unit tests for `ChildrenRegistry`, `ChildSource` (both impls + multi-subscriber observer), the new collector emit + apply_delta cases, the dispatch cases for new event types, the broadcast hook overrides on both WebUI and ConsoleCoordinatorUI, and the focus / placeholder / pending-count frontend assertions in `test_coordinator_page.py`. 5024 passed, ruff + mypy clean.
130 lines
4.4 KiB
Python
130 lines
4.4 KiB
Python
"""Shared builders for the coordinator-endpoint test files.
|
|
|
|
The four coordinator test modules each ship a copy of the same
|
|
``_AuthMiddleware`` / ``_FakeConfigStore`` / ``_fake_registry`` /
|
|
``_build_mgr`` helpers — this module is the single home for them so
|
|
future edits land once. Named with a leading underscore so pytest
|
|
does not collect it.
|
|
|
|
``_make_client`` stays local to each test module because the route
|
|
list differs per file.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
from unittest.mock import MagicMock
|
|
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
|
|
from turnstone.console.collector import ClusterCollector
|
|
from turnstone.console.coordinator_adapter import CoordinatorAdapter
|
|
from turnstone.console.coordinator_ui import ConsoleCoordinatorUI
|
|
from turnstone.core.auth import AuthResult
|
|
from turnstone.core.session_manager import SessionManager
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Iterable
|
|
|
|
|
|
def _seed_children(
|
|
adapter: CoordinatorAdapter, coord_ws_id: str, child_ws_ids: Iterable[str]
|
|
) -> None:
|
|
"""Seed the coordinator adapter's children registry directly.
|
|
|
|
The production path populates the registry via the cluster-event
|
|
fan-out thread observing ``ws_created`` events. These tests just
|
|
need a known-children set for the endpoint handlers to iterate —
|
|
inject directly via the registry's bulk-merge surface rather than
|
|
spinning up the collector + fan-out plumbing.
|
|
"""
|
|
adapter._registry.merge_children(coord_ws_id, child_ws_ids)
|
|
|
|
|
|
class _AuthMiddleware(BaseHTTPMiddleware):
|
|
"""Inject a configurable AuthResult from a header-based contract.
|
|
|
|
Tests set ``X-Test-Perms`` to a comma-separated permission list, and
|
|
``X-Test-User`` to the user id. Empty or missing → no auth.
|
|
"""
|
|
|
|
async def dispatch(self, request, call_next): # type: ignore[no-untyped-def]
|
|
perms = request.headers.get("X-Test-Perms", "")
|
|
user_id = request.headers.get("X-Test-User", "")
|
|
if perms or user_id:
|
|
request.state.auth_result = AuthResult(
|
|
user_id=user_id,
|
|
scopes=frozenset({"approve"}),
|
|
token_source="test",
|
|
permissions=frozenset(p for p in perms.split(",") if p),
|
|
)
|
|
return await call_next(request)
|
|
|
|
|
|
class _FakeConfigStore:
|
|
"""Minimal ConfigStore stub — returns values from a dict."""
|
|
|
|
def __init__(self, values: dict[str, Any]) -> None:
|
|
self._values = values
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
return self._values.get(key, default)
|
|
|
|
|
|
def _fake_registry() -> MagicMock:
|
|
"""MagicMock whose ``.resolve()`` succeeds so the 503 gate passes."""
|
|
reg = MagicMock()
|
|
reg.resolve.return_value = (MagicMock(), "gpt-4", MagicMock())
|
|
return reg
|
|
|
|
|
|
def _build_mgr_with_factory(storage: Any, session_factory: Any) -> SessionManager:
|
|
"""Build a SessionManager(CoordinatorAdapter) with a caller-supplied factory.
|
|
|
|
Used by tests that need to capture or assert factory kwargs (e.g.
|
|
per-call ``model`` / ``judge_model`` overrides). Plain :func:`_build_mgr`
|
|
is the right entry point when the test doesn't care about the
|
|
factory.
|
|
"""
|
|
adapter = CoordinatorAdapter(
|
|
collector=MagicMock(),
|
|
ui_factory=lambda ws: ConsoleCoordinatorUI(ws_id=ws.id, user_id=ws.user_id or ""),
|
|
session_factory=session_factory,
|
|
)
|
|
mgr = SessionManager(
|
|
adapter,
|
|
storage=storage,
|
|
max_active=3,
|
|
node_id=ClusterCollector.CONSOLE_PSEUDO_NODE_ID,
|
|
event_emitter=adapter,
|
|
)
|
|
adapter.attach(mgr)
|
|
return mgr
|
|
|
|
|
|
def _build_mgr(storage: Any) -> SessionManager:
|
|
"""Build a SessionManager(CoordinatorAdapter) with stub factories (test default)."""
|
|
|
|
def _sf(ui, model_alias=None, ws_id=None, **kw): # type: ignore[no-untyped-def]
|
|
s = MagicMock()
|
|
s.send.return_value = None
|
|
return s
|
|
|
|
return _build_mgr_with_factory(storage, _sf)
|
|
|
|
|
|
class MockStorage:
|
|
"""Minimal storage mock that implements ``list_services``.
|
|
|
|
Used by the collector tests + the console route-walk tests. The
|
|
collector calls ``list_services("turnstone-server", ...)`` to
|
|
discover nodes; tests that don't care about discovery push an
|
|
empty list (the default).
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.services: list[dict[str, str]] = []
|
|
|
|
def list_services(self, service_type: str, max_age_seconds: int = 120) -> list[dict[str, str]]:
|
|
return list(self.services)
|