From 3252f3fd95ec66469c96ebeae43883ab3873963d Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Tue, 11 Aug 2026 21:26:27 -0700 Subject: [PATCH] fix(memory): preserve replay identity and validation --- tests/_replay_helpers.py | 8 +-- tests/test_coordinator_endpoints.py | 41 ++++-------- tests/test_project_memory_scope.py | 33 ++++++++++ tests/test_server_authz.py | 20 +++--- tests/test_sse_reconnect_replay.py | 99 +++++++++++++++++++++++++++-- tests/test_structured_memory.py | 33 +++++++++- turnstone/console/server.py | 28 ++++---- turnstone/core/memory.py | 27 +++++--- turnstone/core/session.py | 9 +++ turnstone/core/session_replay.py | 17 +++-- turnstone/core/session_routes.py | 79 +++++++++++++++-------- turnstone/server.py | 23 +++---- 12 files changed, 302 insertions(+), 115 deletions(-) diff --git a/tests/_replay_helpers.py b/tests/_replay_helpers.py index 171100cf..2e69dfc9 100644 --- a/tests/_replay_helpers.py +++ b/tests/_replay_helpers.py @@ -1,10 +1,8 @@ """Shared mock factory for ``events_replay`` tests. -Both interactive (:func:`turnstone.server._interactive_events_replay`) -and coord (:func:`turnstone.console.server._coord_events_replay`) drive -the same shared preamble at -:func:`turnstone.core.session_replay.session_replay_preamble`. Their -test suites share the underlying mock surface (session.model, +Interactive and coordinator replay tests exercise the same shared preamble at +:func:`turnstone.core.session_replay.session_replay_preamble` plus their +kind-specific tails. Their test suites share the underlying mock surface (session.model, session.model_alias, session._last_usage, ui._pending_*, ui._ws_lock, counters); this module is the single home for that shape so a future field add lands once. diff --git a/tests/test_coordinator_endpoints.py b/tests/test_coordinator_endpoints.py index e0ba0baf..1a6e9dcb 100644 --- a/tests/test_coordinator_endpoints.py +++ b/tests/test_coordinator_endpoints.py @@ -2102,39 +2102,35 @@ def test_coord_cancel_cascade_failure_does_not_fail_owner_cancel(storage): from tests._replay_helpers import make_replay_mocks as _make_coord_replay_mocks # noqa: E402 -def test_coord_events_replay_yields_connected_first(): - """Pre-status-bar coord replay only re-injected pending_approval + - pending_plan_review. Post-status-bar parity with interactive - yields ``connected`` first so the dashboard's status bar populates - the model cell before any history arrives — mirrors the - interactive replay (turnstone/server.py:_interactive_events_replay).""" - from turnstone.console.server import _coord_events_replay +def test_coord_shared_preamble_yields_connected_first(): + """The shared preamble gives coordinator streams the same bootstrap.""" + from turnstone.core.session_replay import session_replay_preamble - ws, ui, request = _make_coord_replay_mocks() - out = list(_coord_events_replay(ws, ui, request)) + ws, ui, _request = _make_coord_replay_mocks() + out = list(session_replay_preamble(ws.session, ui)) assert out[0]["type"] == "connected" assert out[0]["model"] == "gpt-5" assert out[0]["model_alias"] == "default" assert out[0]["skip_permissions"] is False -def test_coord_events_replay_includes_status_only_when_last_usage_present(): +def test_coord_shared_preamble_includes_status_only_when_last_usage_present(): """The ``status`` event populates the per-tab token-usage bar on resume. Skipped when ``session._last_usage`` is None (a freshly- created coordinator that hasn't completed a turn) — matches interactive behaviour.""" - from turnstone.console.server import _coord_events_replay + from turnstone.core.session_replay import session_replay_preamble - ws, ui, request = _make_coord_replay_mocks() - out = list(_coord_events_replay(ws, ui, request)) + ws, ui, _request = _make_coord_replay_mocks() + out = list(session_replay_preamble(ws.session, ui)) assert "status" not in {ev["type"] for ev in out} -def test_coord_events_replay_status_payload_shape(): +def test_coord_shared_preamble_status_payload_shape(): """When ``last_usage`` exists, the replayed ``status`` event carries every field the dashboard's updateStatusBar() reads — same shape SessionUI.on_status emits live.""" - from turnstone.console.server import _coord_events_replay + from turnstone.core.session_replay import session_replay_preamble ws, ui, request = _make_coord_replay_mocks( last_usage={ @@ -2146,7 +2142,7 @@ def test_coord_events_replay_status_payload_shape(): _ws_turn_tool_calls=3, _ws_messages=7, ) - out = list(_coord_events_replay(ws, ui, request)) + out = list(session_replay_preamble(ws.session, ui)) status = next(ev for ev in out if ev["type"] == "status") assert status["prompt_tokens"] == 40000 assert status["completion_tokens"] == 6310 @@ -2174,12 +2170,7 @@ def test_coord_events_replay_skips_session_block_when_no_session(): def test_coord_events_replay_yields_pending_approval(): - """The lifted coord ``events_replay`` callback yields, after the - connected preamble, the pending approval (if any). Pre-lift coord - pushed it onto the listener queue via ``put_nowait``; the lift - restructures as a generator the lifted body iterates and yields as - ``data:`` lines, but the payload identity is preserved. Pure-read — - never mutates ``ui``.""" + """The coord replay tail yields pending approval without mutation.""" from turnstone.console.server import _coord_events_replay ws, ui, request = _make_coord_replay_mocks( @@ -2188,8 +2179,6 @@ def test_coord_events_replay_yields_pending_approval(): out = list(_coord_events_replay(ws, ui, request)) types = [ev["type"] for ev in out] - # Status preamble is yielded first (no last_usage → no status); the - # pending-approval re-injection then matches the pre-lift body. assert types[0] == "connected" assert "approve_request" in types @@ -2246,9 +2235,7 @@ def test_coord_events_replay_skips_verdict_replay_without_pending_approval(): def test_coord_events_replay_yields_only_connected_when_no_pending(): - """A workstream with a session but no pending approval / plan - review and no last_usage yields just the ``connected`` preamble. - The lifted body falls through to the live loop immediately after.""" + """Without controls or usage, replay contains only the preamble.""" from turnstone.console.server import _coord_events_replay ws, ui, request = _make_coord_replay_mocks() diff --git a/tests/test_project_memory_scope.py b/tests/test_project_memory_scope.py index f2c509be..f532fd0d 100644 --- a/tests/test_project_memory_scope.py +++ b/tests/test_project_memory_scope.py @@ -118,6 +118,39 @@ class TestLiveProjectAccess: assert access.attached_project_id == "p1" assert access.project_id == "" + def test_project_display_name_uses_explicit_principal( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + seen: list[tuple[str, str]] = [] + + def _resolve(principal_id: str, project_id: str) -> object: + seen.append((principal_id, project_id)) + return self._access(True, True) + + monkeypatch.setattr(auth, "resolve_project_access", _resolve) + s = _session(user_id="owner", project_id="p1") + s._acting_user_id = "stale-turn-actor" + seen.clear() # Ignore constructor-time system-context composition. + + assert s.project_name_for_principal("reconnecting-viewer") == "P" + assert seen == [("reconnecting-viewer", "p1")] + + def test_project_display_name_does_not_fallback_for_empty_principal( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + seen: list[tuple[str, str]] = [] + + def _resolve(principal_id: str, project_id: str) -> object: + seen.append((principal_id, project_id)) + return self._access(True, True) + + monkeypatch.setattr(auth, "resolve_project_access", _resolve) + s = _session(user_id="owner", project_id="p1") + seen.clear() + + assert s.project_name_for_principal("") == "" + assert seen == [] + class TestProjectRecall: def test_interactive_visible_scopes_includes_project( diff --git a/tests/test_server_authz.py b/tests/test_server_authz.py index 7ccb1114..128466c8 100644 --- a/tests/test_server_authz.py +++ b/tests/test_server_authz.py @@ -1435,28 +1435,26 @@ class TestInteractiveEventsLifted: section. """ - def test_events_replay_yields_connected_first(self): - """Pre-lift ``events_sse`` yielded a ``connected`` event - first (model + skip_permissions). The lifted callback - preserves the order so client SSE handlers that key on - the connected event for state setup keep working.""" - from turnstone.server import _interactive_events_replay + def test_shared_preamble_yields_connected_first(self): + """The shared handler's preamble preserves connected-event shape.""" + from turnstone.core.session_replay import session_replay_preamble ws, ui, request = _make_interactive_replay_mocks() - out = list(_interactive_events_replay(ws, ui, request)) + out = list(session_replay_preamble(ws.session, ui, project_name="Visible Project")) assert out[0]["type"] == "connected" assert out[0]["model"] == "gpt-5" assert out[0]["model_alias"] == "default" + assert out[0]["project_name"] == "Visible Project" assert out[0]["skip_permissions"] is False - def test_events_replay_includes_status_only_when_last_usage_present(self): + def test_shared_preamble_includes_status_only_when_last_usage_present(self): """The ``status`` event populates the per-tab token-usage bar on resume. Skipped when ``session._last_usage`` is None (a freshly-created workstream that hasn't completed a turn).""" - from turnstone.server import _interactive_events_replay + from turnstone.core.session_replay import session_replay_preamble - ws, ui, request = _make_interactive_replay_mocks() - out = list(_interactive_events_replay(ws, ui, request)) + ws, ui, _request = _make_interactive_replay_mocks() + out = list(session_replay_preamble(ws.session, ui)) assert "status" not in {ev["type"] for ev in out} def test_events_replay_yields_pending_approval_then_verdicts(self): diff --git a/tests/test_sse_reconnect_replay.py b/tests/test_sse_reconnect_replay.py index d428984b..9e931b91 100644 --- a/tests/test_sse_reconnect_replay.py +++ b/tests/test_sse_reconnect_replay.py @@ -53,6 +53,7 @@ def _fake_request( headers: dict[str, str] | None = None, query: dict[str, str] | None = None, path_params: dict[str, str] | None = None, + user_id: str = "viewer-1", ) -> Request: """Construct a Starlette ``Request`` for the events handler. @@ -80,7 +81,9 @@ def _fake_request( async def _recv() -> dict[str, Any]: # noqa: RUF029 — async signature required return {"type": "http.disconnect"} - return Request(scope, receive=_recv) + request = Request(scope, receive=_recv) + request.state.auth_result = SimpleNS(user_id=user_id) + return request # --------------------------------------------------------------------------- @@ -673,6 +676,87 @@ def test_handler_emits_retry_on_first_yield() -> None: assert 2500 <= retry <= 4500, f"retry {retry} outside jitter band [2500, 4500]" +def test_handler_resolves_project_off_loop_and_preserves_legacy_replay_callback() -> None: + """Project metadata uses the viewer off-loop; replay keeps its 3-arg API.""" + from turnstone.core.session_replay import ( + request_replay_project_name, + session_replay_preamble, + ) + + event_loop_thread = threading.get_ident() + resolver_threads: list[int] = [] + principals: list[str] = [] + callback_calls: list[tuple[Any, Any, Any]] = [] + + session = MagicMock() + session.model = "test" + session.model_alias = "" + session._last_usage = None + + def _project_name(principal_id: str) -> str: + resolver_threads.append(threading.get_ident()) + principals.append(principal_id) + return "Visible Project" + + session.project_name_for_principal.side_effect = _project_name + + def _replay(ws: Any, ui: Any, request: Any) -> Any: + callback_calls.append((ws, ui, request)) + yield from session_replay_preamble( + ws.session, + ui, + project_name=request_replay_project_name(request), + ) + yield {"type": "tail"} + + ui = _make_ui() + _, blob = _drain_handler_yields( + ui, + session=session, + events_replay=_replay, + max_yields=4, + ) + + assert principals == ["viewer-1"] + assert resolver_threads and resolver_threads[0] != event_loop_thread + assert len(callback_calls) == 1 + assert '"type": "connected"' in blob + assert '"project_name": "Visible Project"' in blob + assert '"type": "tail"' in blob + + +def test_handler_project_lookup_failure_keeps_connected_event() -> None: + """Optional project metadata fails closed without losing bootstrap.""" + from turnstone.core.session_replay import ( + request_replay_project_name, + session_replay_preamble, + ) + + session = MagicMock() + session.model = "test" + session.model_alias = "" + session._last_usage = None + session.project_name_for_principal.side_effect = RuntimeError("storage unavailable") + + def _replay(ws: Any, ui: Any, request: Any) -> Any: + yield from session_replay_preamble( + ws.session, + ui, + project_name=request_replay_project_name(request), + ) + + ui = _make_ui() + _, blob = _drain_handler_yields( + ui, + session=session, + events_replay=_replay, + max_yields=3, + ) + + assert '"type": "connected"' in blob + assert '"project_name": ""' in blob + + def test_handler_replay_ok_skips_snapshot_emits_id(monkeypatch: Any) -> None: """``Last-Event-ID`` + buffer covers gap → emit buffered events with SSE ``id:`` field, SKIP the in-progress snapshot (it would @@ -982,6 +1066,11 @@ class _HandoffSession: self.context_window = 1000 self.reasoning_effort = "low" self._last_usage = {"prompt_tokens": 12, "completion_tokens": 0} + self.project_name_principals: list[str] = [] + + def project_name_for_principal(self, principal_id: str) -> str: + self.project_name_principals.append(principal_id) + return "" def register_listener_for_history_handoff( self, @@ -1093,11 +1182,8 @@ def test_handoff_cursor_replay_keeps_preamble_without_pending_control_duplicate( session = _HandoffSession(ui) def _full_replay(_ws: Any, _ui: Any, _request: Any) -> Any: - # The full replay's own preamble half is what the replay_ok path - # must NOT re-run; the lifted body calls the shared - # session_replay_preamble directly instead (no per-kind hook). - yield {"type": "connected", "model": "test"} - yield {"type": "status", "total_tokens": 12} + # The kind-specific tail must NOT run on replay_ok; the lifted body + # calls the shared preamble directly and the ring owns controls. yield {"type": "approve_request", "items": [{"call_id": "duplicate"}]} _, blob = _drain_handler_yields( @@ -1110,6 +1196,7 @@ def test_handoff_cursor_replay_keeps_preamble_without_pending_control_duplicate( ) assert session.calls == [(session.token, 0)] + assert session.project_name_principals == ["viewer-1"] assert '"type": "connected"' in blob assert '"type": "status"' in blob assert '"type": "state_change"' in blob diff --git a/tests/test_structured_memory.py b/tests/test_structured_memory.py index 271ad1af..b976d7d8 100644 --- a/tests/test_structured_memory.py +++ b/tests/test_structured_memory.py @@ -20,10 +20,39 @@ def _save(name, content, **kwargs): class TestSaveStructuredMemory: + @pytest.mark.parametrize("save", [save_structured_memory, save_structured_memory_strict]) @pytest.mark.parametrize("description", [None, "", " "]) - def test_description_is_required(self, tmp_db, description): + def test_description_is_required(self, tmp_db, description, save): with pytest.raises(ValueError, match="description is required"): - save_structured_memory_strict("test_key", "hello world", description=description) + save("test_key", "hello world", description=description) + + def test_best_effort_save_still_swallows_storage_failures(self, tmp_db, monkeypatch): + from turnstone.core import memory as memory_mod + + class _BoomStorage: + def upsert_structured_memory(self, *_args, **_kwargs): + raise RuntimeError("simulated storage failure") + + monkeypatch.setattr(memory_mod, "get_storage", lambda: _BoomStorage()) + assert save_structured_memory( + "test_key", + "hello world", + description="Test memory", + ) == (None, False) + + def test_best_effort_save_does_not_misclassify_backend_value_error(self, tmp_db, monkeypatch): + from turnstone.core import memory as memory_mod + + class _BoomStorage: + def upsert_structured_memory(self, *_args, **_kwargs): + raise ValueError("backend decode failure") + + monkeypatch.setattr(memory_mod, "get_storage", lambda: _BoomStorage()) + assert save_structured_memory( + "test_key", + "hello world", + description="Test memory", + ) == (None, False) def test_save_new(self, tmp_db): row, was_update = _save("test_key", "hello world") diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 66d1ef41..60f03fff 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -76,7 +76,10 @@ from turnstone.core.model_registry import ( from turnstone.core.model_registry import MODEL_AUTH_MODES as _MODEL_AUTH_MODES from turnstone.core.rendezvous import NoAvailableNodeError, NodeRef from turnstone.core.rerank_calibrate import canonical_caps_value -from turnstone.core.session_replay import session_replay_preamble +from turnstone.core.session_replay import ( + request_replay_project_name, + session_replay_preamble, +) from turnstone.core.session_routes import ( AttachmentUploadHelpers, CoordOnlyVerbHandlers, @@ -3687,20 +3690,15 @@ def _audit_retry_coordinator( def _coord_events_replay( ws: Workstream, ui: Any, - request: Request, # noqa: ARG001 — coord replay doesn't need request context + request: Request, ) -> Iterable[dict[str, Any]]: """Initial SSE replay payload for coord ``events`` connections. - Yields, in order: - - 1. ``connected`` + optional ``status`` via the shared - :func:`turnstone.core.session_replay.session_replay_preamble` - so the dashboard's status bar populates before any live tick. - Same payload shape interactive uses. - 2. Pending approval prompt (if any) and the cached LLM verdicts - that fired since it surfaced. Without this replay a refresh - loses the judge chip on the pending approval until the - operator re-invokes the action. + Yields ``connected`` plus optional ``status``, then the pending approval + prompt (if any) and cached LLM verdicts that fired since it surfaced. The + shared handler resolves viewer-specific project metadata off-loop before + invoking this callback. Without the control replay a refresh loses the + judge chip until the operator re-invokes the action. Coord still skips conversation history — the dashboard fetches it via a separate ``GET /history`` endpoint and doesn't want a @@ -3708,7 +3706,11 @@ def _coord_events_replay( Pure read — never mutates ``ui`` / ``ws`` / ``session``. """ - yield from session_replay_preamble(ws.session, ui) + yield from session_replay_preamble( + ws.session, + ui, + project_name=request_replay_project_name(request), + ) # EVERY live approval cycle replays (parallel task agents can have # several outstanding), each card followed once by the cached LLM diff --git a/turnstone/core/memory.py b/turnstone/core/memory.py index 7576cb38..4cd65b0b 100644 --- a/turnstone/core/memory.py +++ b/turnstone/core/memory.py @@ -855,6 +855,13 @@ def search_history_recent(limit: int = 20, *, user_id: str | None = None) -> lis # -- Structured memories ------------------------------------------------------- +def _require_memory_description(description: str) -> str: + """Return a normalized description or raise the public validation error.""" + if not isinstance(description, str) or not (normalized := description.strip()): + raise ValueError("memory description is required and must be non-empty") + return normalized + + def save_structured_memory( name: str, content: str, @@ -873,16 +880,20 @@ def save_structured_memory( :meth:`StorageBackend.upsert_structured_memory`) -- no preceding read, no IntegrityError round-trip, no TOCTOU window. ``(row, was_update)`` comes straight from that upsert (this passes a fresh ``memory_id``, so a differing - returned id means an existing row was updated in place). A ``None`` - ``description`` is required and must contain non-whitespace text for both - inserts and updates. A ``None`` ``mem_type`` keeps the stored value on an - update and uses the column default on insert. + returned id means an existing row was updated in place). ``description`` + is required and must contain non-whitespace text for both inserts and + updates. A ``None`` ``mem_type`` keeps the stored value on an update and + uses the column default on insert. """ + # Validate outside the best-effort storage boundary. Backend/driver + # ``ValueError`` instances remain operational failures; only this explicit + # caller-input check propagates. + normalized_description = _require_memory_description(description) try: return save_structured_memory_strict( name, content, - description=description, + description=normalized_description, mem_type=mem_type, scope=scope, scope_id=scope_id, @@ -907,14 +918,12 @@ def save_structured_memory_strict( Unlike :func:`save_structured_memory`, storage failures propagate so an API or tool cannot report a database outage as an ordinary failed/not-found - result. Prompt composition keeps using the best-effort facade. + result. Best-effort internal callers keep using the facade. """ import uuid normalized = normalize_key(name) - normalized_description = (description or "").strip() - if not normalized_description: - raise ValueError("memory description is required and must be non-empty") + normalized_description = _require_memory_description(description) row, was_update = get_storage().upsert_structured_memory( str(uuid.uuid4()), normalized, diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 646b0c51..c700a87a 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -18417,6 +18417,15 @@ class ChatSession: project_writable=project_writable, ) + def project_name_for_principal(self, principal_id: str) -> str: + """Return the active attached-project name visible to ``principal_id``. + + The principal is explicit so connection-level presentation never reads + the session's mutable acting-user binding. Project rename, archival, + and access revocation therefore take effect on the next lookup. + """ + return self._memory_access(principal_id).project_name + def _resolve_scope_id(self, scope: str, access: _MemoryAccess | None = None) -> str: """Map a validated scope to the actor-specific storage key.""" resolved = access or self._memory_access() diff --git a/turnstone/core/session_replay.py b/turnstone/core/session_replay.py index f27f16d2..62e2ed12 100644 --- a/turnstone/core/session_replay.py +++ b/turnstone/core/session_replay.py @@ -22,9 +22,17 @@ if TYPE_CHECKING: from turnstone.core.session import ChatSession +def request_replay_project_name(request: Any) -> str: + """Return project metadata pre-resolved by the shared SSE route.""" + value = getattr(getattr(request, "state", None), "_session_replay_project_name", "") + return value if isinstance(value, str) else "" + + def session_replay_preamble( session: ChatSession | None, ui: Any, + *, + project_name: str = "", ) -> Iterable[dict[str, Any]]: """Yield ``connected`` + optional ``status`` events for an SSE replay. @@ -33,7 +41,8 @@ def session_replay_preamble( through to the kind-specific tail. - ``connected`` carries ``model`` / ``model_alias`` / ``skip_permissions`` so the per-tab status bar populates the model cell before any - history arrives. + history arrives. The caller supplies the project display name already + resolved for the authenticated connection principal. - ``status`` only fires when ``session._last_usage`` exists (a session that has completed at least one turn). The payload shape matches :meth:`SessionUI.on_status` so live ticks and replays use @@ -48,9 +57,9 @@ def session_replay_preamble( "type": "connected", "model": session.model, "model_alias": session.model_alias or "", - # The attached project's display name (""=none) so the composer can - # paint its "has a project" badge on connect, beside the model chip. - "project_name": getattr(session, "_project_name", "") or "", + # The visible attached-project display name (""=none) so the composer + # can paint its project badge on connect, beside the model chip. + "project_name": project_name, "skip_permissions": getattr(ui, "auto_approve", False), } diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 21ad3c24..a7a210dc 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -173,15 +173,11 @@ class EventsReplay(Protocol): live event loop starts. Each yielded dict gets JSON-serialised and sent as a single ``data:`` line to the client. - Interactive yields four things on connect: ``connected`` (model + - skip_permissions), ``status`` (token usage + context %, only when - ``session._last_usage`` exists), ``history`` (replayed conversation), - and ``pending_approval`` + cached intent verdicts. Coord yields - just one: ``pending_approval`` (the rest aren't needed because - coord's dashboard fetches history via a separate ``/history`` - endpoint and doesn't render the per-tab status bar). Kinds that - don't need any pre-replay wire ``None`` and the live loop starts - immediately. + The shared handler pre-resolves viewer-specific project metadata onto the + request before calling this stable three-argument callback. Production + callbacks emit ``connected`` plus optional ``status``, then pending controls + and cached verdicts. Conversation history stays on the separate REST + endpoint. Kinds without replay wire ``None``. """ def __call__(self, ws: Workstream, ui: Any, request: Request) -> Iterable[dict[str, Any]]: @@ -444,13 +440,11 @@ class SessionEndpointConfig: # wires ``None`` and lets the cluster collector handle the # transition via ``CoordinatorAdapter.emit_rehydrated``. open_post_load: OpenPostLoad | None = None - # (ws, ui, request) -> Iterable[dict]. Kind-specific initial - # SSE replay payload the lifted ``events`` body yields after - # registering the per-UI listener queue but before the live - # event loop. Both production kinds yield connected + optional status, - # followed by pending approval controls and cached verdicts. Conversation - # history stays on the separate REST ``/history`` bootstrap. Kinds that - # don't need pre-replay wire ``None``. + # (ws, ui, request) -> Iterable[dict]. Kind-specific initial SSE replay + # payload the lifted ``events`` body yields before the live event loop. + # The shared handler pins viewer-specific project metadata on ``request``; + # production callbacks combine it with connected/status and pending + # controls. Conversation history stays on the separate REST bootstrap. events_replay: EventsReplay | None = None # (request) -> Executor for the SSE live-loop's blocking # ``queue.get`` wait. Interactive returns the dedicated @@ -2298,15 +2292,15 @@ def make_open_handler( def make_events_handler(cfg: SessionEndpointConfig) -> Handler: """Lifted body for ``GET {prefix}/{ws_id}/events`` — per-workstream SSE. - Both kinds share the SSE plumbing: register the per-UI listener - queue, run the kind-specific initial replay (``cfg.events_replay``, - typically ``connected`` + ``status`` + ``history`` + pending - approval / plan on interactive; just pending approval / plan on - coord), then drain the queue forever until either the workstream - closes (``ws_closed`` event) or the client disconnects. + Both kinds share the SSE plumbing: resolve viewer-specific project + metadata off-loop, register the per-UI listener queue, run the configured + initial replay (``cfg.events_replay``), then drain the queue forever until + either the workstream closes (``ws_closed`` event) or the client + disconnects. Cursor-only replay emits the shared connected/status preamble + directly because it deliberately skips kind-specific pending controls. - The kind-specific divergence is captured entirely by - ``cfg.events_replay``. The live-loop body, the listener + The kind-specific replay divergence is captured by + ``cfg.events_replay``. The cursor preamble, live-loop body, listener registration, the ``ws_closed`` exit, the disconnect detection, and the SSE-connect/disconnect metric recording are uniform. @@ -2432,6 +2426,33 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler: # ``ui`` is a ``SessionUIBase`` subclass, so the cast is # tightening the type, not weakening it. ui_base = cast("SessionUIBase", ui) + from turnstone.core.web_helpers import auth_user_id + + # Pin the authenticated viewer once per connection. Resolve optional + # project presentation metadata off the event loop and before listener + # registration, so database latency cannot stall unrelated async work + # or let this listener's queue accumulate while it waits. + replay_principal_id = auth_user_id(request).strip() + replay_project_name = "" + session = getattr(ws, "session", None) + project_name_for_principal = getattr(session, "project_name_for_principal", None) + if callable(project_name_for_principal): + try: + resolved_project_name = await asyncio.to_thread( + project_name_for_principal, + replay_principal_id, + ) + if isinstance(resolved_project_name, str): + replay_project_name = resolved_project_name + except Exception: + # Project metadata is optional presentation context. Fail + # closed to no badge without dropping the SSE bootstrap. + log.debug( + "ws.events.project_name_failed ws=%s", + ws_id[:8], + exc_info=True, + ) + request.state._session_replay_project_name = replay_project_name replay_status: str replay_events: list[dict[str, Any]] = [] lost_count = 0 @@ -2675,7 +2696,11 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler: # contract lives in session_replay_preamble alone (it # no-ops on a detached session internally). try: - for ev in session_replay_preamble(ws.session, ui): + for ev in session_replay_preamble( + ws.session, + ui, + project_name=replay_project_name, + ): yield {"data": json.dumps(ev)} except Exception: log.debug( @@ -2745,8 +2770,8 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler: ) } - # Replay phase — stream the kind-specific initial - # payload one event at a time so the client sees + # Replay phase — stream the kind-specific initial payload + # one event at a time so the client sees # the first byte immediately. Pre-building into # a list would block time-to-first-byte until the # entire replay materialized AND let the listener diff --git a/turnstone/server.py b/turnstone/server.py index 6a9bd163..efb838f2 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -78,7 +78,10 @@ from turnstone.core.session_manager import ( STALE_CREATE_SWEEP_INTERVAL_SECONDS, SessionManager, ) -from turnstone.core.session_replay import session_replay_preamble +from turnstone.core.session_replay import ( + request_replay_project_name, + session_replay_preamble, +) from turnstone.core.session_routes import ( AttachmentUploadHelpers, CreatePreCommitError, @@ -801,12 +804,9 @@ def _interactive_events_replay( ) -> Iterable[dict[str, Any]]: """Initial SSE replay payload for interactive ``events`` connections. - Yields a ``connected`` event (model + skip_permissions), a - ``status`` event with the workstream's last token usage + context % - (when a turn has completed), and the pending approval prompt + cached - intent verdicts (if a prompt is pending). The lifted - ``make_events_handler`` body delegates that yield sequence to this - callback so the kind-specific shape stays in this module. + Yields the shared ``connected`` / ``status`` preamble, followed by pending + approval prompts and cached intent verdicts. The lifted handler resolves + viewer-specific project metadata off-loop before invoking this callback. Conversation history is NOT replayed over SSE: the frontend fetches it via ``GET /history`` on page load and re-fetches on the @@ -821,10 +821,11 @@ def _interactive_events_replay( # session can still be detached on the close-then-reopen path. return - # Connected + status preamble — same shape coord replays and the - # lifted reconnect path use; one shared function, no per-kind - # wrapper, so a future field add cannot land on one surface only. - yield from session_replay_preamble(ws.session, ui) + yield from session_replay_preamble( + ws.session, + ui, + project_name=request_replay_project_name(request), + ) # Pending approval re-injection (so a reconnecting tab sees the # prompt) + cached LLM verdicts received since the prompt fired.