diff --git a/tests/test_server_attachments_on_create.py b/tests/test_server_attachments_on_create.py index d0e8d12d..47155a72 100644 --- a/tests/test_server_attachments_on_create.py +++ b/tests/test_server_attachments_on_create.py @@ -9,6 +9,7 @@ Exercises: from __future__ import annotations +import contextlib import json import queue import threading @@ -17,6 +18,8 @@ import time import pytest from starlette.testclient import TestClient +from tests._helpers import wait_until + # Magic-byte-valid 1x1 PNG (matches the fixture in test_server_attachments_endpoints.py) PNG_1x1 = ( b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" @@ -223,6 +226,29 @@ class _FakeSession: self._cancel_event = threading.Event() self.notify_targets = "" self._notify_on_complete = "[]" + # Wired by the app_client session factory so the faithful + # _record_fatal_error model below can emit through the real UI. + self.ui = None + self._has_persisted_error = False + + def _record_fatal_error(self, exc): + """Faithful model of ChatSession._record_fatal_error: sanitize, emit + on_error, persist last_error, set the flag, emit state=error.""" + from turnstone.core.memory import persist_last_error, sanitize_error_text + + safe = sanitize_error_text(f"{type(exc).__name__}: {exc}") + if self.ui is not None: + self.ui.on_error(safe) + persist_last_error(self.ws_id, safe) + self._has_persisted_error = True + if self.ui is not None: + self.ui.on_state_change("error") + + def ensure_error_recorded(self, exc): + """Mirror ChatSession.ensure_error_recorded: idempotent on the flag.""" + if self._has_persisted_error: + return + self._record_fatal_error(exc) def send(self, text, attachments=None, send_id=None): with self._lock: @@ -270,6 +296,10 @@ class _FakeSession: class _FakeUI: + # Set by the app_client fixture; mirrors WebUI._workstream_mgr so + # on_state_change routes to the real manager the coordinator polls. + _workstream_mgr = None + def __init__(self, ws_id="", user_id=""): self.ws_id = ws_id self._user_id = user_id @@ -286,6 +316,14 @@ class _FakeUI: def on_state_change(self, state): self.events.append({"type": "state_change", "state": state}) + # Mirror WebUI.on_state_change's real routing so tests assert the + # state the coordinator polls (via the manager), not just the label. + mgr = type(self)._workstream_mgr + if mgr is not None: + from turnstone.core.workstream import WorkstreamState + + with contextlib.suppress(ValueError): + mgr.set_state(self.ws_id, WorkstreamState(state)) def on_error(self, msg): self.events.append({"type": "error", "message": msg}) @@ -316,6 +354,9 @@ def app_client(tmp_path, monkeypatch): # ui instance so the FakeSession's consume step uses the right scope. user_id = getattr(ui, "_user_id", "") s = _FakeSession(ws_id=ws_id, user_id=user_id) + # Hand the fake its UI so the faithful _record_fatal_error model can + # emit on_error/on_state_change exactly as the real ChatSession does. + s.ui = ui fake_sessions.append(s) return s @@ -332,6 +373,9 @@ def app_client(tmp_path, monkeypatch): mgr = SessionManager( adapter, storage=get_storage(), max_active=10, node_id="node-test", event_emitter=adapter ) + # Route _FakeUI.on_state_change through this manager's set_state (mirrors + # WebUI._workstream_mgr) so a test can read the coordinator-visible state. + monkeypatch.setattr(_FakeUI, "_workstream_mgr", mgr, raising=False) app = create_app( workstreams=mgr, @@ -637,3 +681,98 @@ class TestCreateJsonStillWorks: # dead-by-construction ``enqueue`` branch is still wired (loudly). assert callable(kwargs["run"]) assert callable(kwargs["enqueue"]) + + +# --------------------------------------------------------------------------- +# Initial-message worker failure state +# --------------------------------------------------------------------------- + + +def _post_init(client, name): + """Create a workstream with an initial message and return the response.""" + body = {"name": name, "initial_message": "go"} + return client.post("/v1/api/workstreams/new", json=body, headers=_auth("userA")) + + +class TestInitialWorkerFailureState: + """The initial-message worker classifies its first-turn exit the way the + send/retry closure does, asserted — once the worker has FULLY finished — + against the coordinator-visible manager state (what wait_for_workstream + polls), the persisted last_error, and the emitted events, NOT a default + label: + + * backend failure → state=error carrying a readable last_error, recorded + EXACTLY once (proving ensure_error_recorded's no-op on the common leg); + * cancel-to-idle (send self-handles) → no error recorded. + + (The completion-notification honesty surface is deferred to #865.) + """ + + @pytest.mark.parametrize("pretry", [False, True], ids=["common-backend", "pre-try"]) + def test_backend_error_settles_error_with_detail(self, app_client, monkeypatch, pretry): + from turnstone.core.memory import load_last_error + from turnstone.core.workstream import WorkstreamState + + client, sessions, _gq = app_client + + def _boom(self, *_a, **_k): + exc = RuntimeError("cannot reach openai-compatible at http://x:8000/v1") + # common-backend: send records the fatal error in-line (its own + # except arm) before re-raising → _has_persisted_error set → the + # closure's ensure_error_recorded is a no-op. pre-try: send raises + # BEFORE it would reach _record_fatal_error, so the closure's + # ensure_error_recorded is the ONLY recorder — the finding-[1] path. + if not pretry: + self._record_fatal_error(exc) + raise exc + + monkeypatch.setattr(_FakeSession, "send", _boom) + resp = _post_init(client, "dead-backend") + assert resp.status_code == 200, resp.text + ws_id = resp.json()["ws_id"] + mgr = client.app.state.workstreams + + # Wait for the init worker to FULLY finish (_worker_running clears in the + # runner's finally) — not for a state, which a fresh ws already holds and + # which would race the double-emit count below. + wait_until(lambda: (w := mgr.get(ws_id)) is not None and not w._worker_running) + + # Coordinator-visible state (what wait_for_workstream polls) is ERROR, + # and the failure DETAIL is readable — on the pre-try leg the proof that + # ensure_error_recorded closed [1]. + assert mgr.get(ws_id).state is WorkstreamState.ERROR + assert load_last_error(ws_id) + # Recorded EXACTLY once. On the common-backend leg this is the proof + # that ensure_error_recorded no-oped on send's in-line record — a lost + # _has_persisted_error guard would emit state=error + on_error twice. + events = sessions[0].ui.events + assert len([e for e in events if e.get("state") == "error"]) == 1, events + assert len([e for e in events if e.get("type") == "error"]) == 1, events + + def test_cancel_settles_idle_no_error(self, app_client, monkeypatch): + client, sessions, _gq = app_client + + def _cancel_to_idle(self, *_a, **_k): + # Model send's OWN in-turn cancel handling: it emits idle and returns + # normally (session.py:6474), never re-raising — the reachable + # cancel→idle contract, not the unreachable except-GenerationCancelled + # arm. + if self.ui is not None: + self.ui.on_state_change("idle") + + monkeypatch.setattr(_FakeSession, "send", _cancel_to_idle) + resp = _post_init(client, "stopped") + assert resp.status_code == 200, resp.text + ws_id = resp.json()["ws_id"] + mgr = client.app.state.workstreams + + # Positive barrier: wait for the worker to finish, not for a state — a + # fresh ws already defaults to IDLE, so a state==IDLE gate would pass + # before the worker runs and the assertions below would be vacuous. + wait_until(lambda: (w := mgr.get(ws_id)) is not None and not w._worker_running) + # A clean return records NO error — neither an error state_change nor an + # on_error event ({"type": "error"}, no "state" key). + events = sessions[0].ui.events + assert events, "init worker never emitted" + assert all(e.get("state") != "error" for e in events) + assert all(e.get("type") != "error" for e in events) diff --git a/tests/test_session_backend_error_format.py b/tests/test_session_backend_error_format.py index c63b412c..d65733d2 100644 --- a/tests/test_session_backend_error_format.py +++ b/tests/test_session_backend_error_format.py @@ -142,6 +142,32 @@ def test_not_found_points_at_model_name_mismatch(): assert "/v1/models" in msg # operator hint +# --------------------------------------------------------------------------- +# Model label — lead with the alias the model references, annotate the id +# --------------------------------------------------------------------------- + + +def test_error_leads_with_alias_and_annotates_backend_id(): + """When the display alias differs from the backend model id, the enriched + error leads with the ALIAS (the identifier the model references everywhere — + list_nodes, spawn) and annotates the backend id for the operator, so a + coordinator correlates the failure with those surfaces without a lookup.""" + msg = _format( + _stub(model="deepseek-v4-flash", model_alias="DeepSeek-V4-Flash"), + APIConnectionError("cannot reach"), + ) + assert msg is not None + assert "model=DeepSeek-V4-Flash (id=deepseek-v4-flash)" in msg + + +def test_error_model_label_collapses_when_alias_equals_id(): + """No redundant (id=...) annotation when the alias and backend id coincide.""" + msg = _format(_stub(model="flatspark", model_alias="flatspark"), APIConnectionError("x")) + assert msg is not None + assert "model=flatspark" in msg + assert "(id=" not in msg + + @pytest.mark.parametrize("exc_cls", [AuthenticationError, PermissionDeniedError]) def test_auth_message_mentions_api_key(exc_cls): msg = _format(_stub(), exc_cls("invalid api key")) diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 80c2fce3..6076d289 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -4786,6 +4786,43 @@ class ChatSession: self._has_persisted_error = True self._emit_state("error") + def ensure_error_recorded(self, exc: BaseException) -> None: + """Idempotently route a fatal exception through :meth:`_record_fatal_error`. + + FRESH-SESSION USE ONLY. The sole caller is the initial-message worker + ``_run_initial`` (server.py), which runs on a factory-fresh session's + FIRST send. The idempotency guard is ``_has_persisted_error``, which is + **session-lifetime** — cleared only by an ``idle``/``running`` state + emission, NOT per turn — so it distinguishes "send already recorded THIS + turn's error" from "not yet recorded" only when no prior turn could have + left it set. On a **reused** session (a ``/retry``, main ``/send``, the + coordinator send, a wake) a pre-try raise after a prior errored turn + finds the flag stale-True and this no-ops — silently swallowing the fresh + error and surfacing the stale ``last_error``. Those closures therefore + must NOT route through here until the per-turn error-recorded signal of + #865 lands; see ``_run`` for the deliberate non-use. + + Two cases on the fresh-session path, one primitive: + + * ``send`` already recorded this turn's fatal error in-line — the common + backend-boundary path, where its ``except`` arm ran + ``_record_fatal_error`` (setting ``_has_persisted_error``) before + re-raising. Then this is a **no-op**: no second ``state=error`` + emission, avoiding the duplicate ``state_change`` SSE / ``set_state``. + * the exception escaped ``send`` BEFORE its ``try`` — a pre-try failure + (model-registry refresh, user-turn append, system-message recompose) + that bypasses ``_record_fatal_error``. Then this **records it now**, + so ``state==error ⇒ meaningful last_error`` holds on that path too. + + Sanitizing lives inside ``_record_fatal_error``, so a worker closure must + route raw exceptions here rather than emitting ``str(exc)`` to a UI sink + itself: a credential-bearing ``base_url`` in the exception text would + otherwise cross the confidentiality floor into a dashboard SSE. + """ + if self._has_persisted_error: + return + self._record_fatal_error(exc) + def _format_backend_error(self, exc: BaseException) -> str | None: """Return an enriched message for known backend boundary errors. @@ -4808,10 +4845,22 @@ class ChatSession: base URL are redacted before display / persist. """ # Backend identity shared by every branch — model label + raw tail. - # Hoisted so the context-overflow branch and the class-name branches use - # one derivation. self.model/_model_alias are plain __init__ attributes + # The model references models by ALIAS everywhere it acts (list_nodes + # advertises aliases, spawn takes an alias as model=), so lead with the + # alias here too: a coordinator reading this error can correlate the + # failed model with those surfaces without a lookup and route around it + # (respawn on another node/alias) instead of burning reasoning tokens + # reconciling the alias against a backend id it never sees elsewhere. + # The backend id (what the server was actually asked for) rides as a + # labeled annotation for the operator, collapsing to one token when the + # two coincide. self.model/_model_alias are plain __init__ attributes # (always set), so reading them here can't raise on the fatal path. - model_label = self.model or self._model_alias or "?" + alias = self._model_alias or "" + backend_id = self.model or "" + if alias and backend_id and alias != backend_id: + model_label = f"{alias} (id={backend_id})" + else: + model_label = alias or backend_id or "?" raw_msg = str(exc).strip() raw_tail = f" raw={raw_msg!r}" if raw_msg else "" diff --git a/turnstone/server.py b/turnstone/server.py index fc31d799..ed8b03fe 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -723,6 +723,15 @@ def _interactive_dispatch_retry(ws: Workstream, user_msg: str) -> None: ui.on_stream_end() ui.on_state_change("idle") except Exception as exc: + # Deliberately NOT routed through session.ensure_error_recorded: on a + # REUSED session a pre-try raise after a prior errored turn finds + # _has_persisted_error stale-True (it is session-lifetime — cleared + # only by _emit_state idle/running, not per-turn), so the recorder + # would no-op and swallow the fresh error. The raw-exc UI leak, the + # double state emit, and the pre-try no-persist (the coordinator can + # then read a STALE last_error on a reused-session retry) are known, + # match the /send and coord-send sibling closures, and are tracked in + # #865's per-turn error-signal redesign. if ws.worker_thread is me: ui.on_error(f"Error: {exc}") ui.on_stream_end() @@ -2504,16 +2513,44 @@ async def _interactive_create_post_install( attachments=resolved_atts or None, send_id=send_id if resolved_atts else None, ) - except (Exception, GenerationCancelled): - # Abandoned-worker guard (sibling of _run_cmd/_run_compact/ - # the send closures): a force-cancelled init that unwedges - # late must not stamp idle over — or emit stream_end into — - # a successor turn. + except GenerationCancelled: + # Defense-in-depth, effectively unreachable on the init path: + # send self-handles an in-turn cancel and self-emits idle + # (session.py:6474), and an orphaned/superseded cancel returns + # WITHOUT re-raising — so this arm carries no direct unit test, + # its non-triggering being send's own contract. Kept for + # symmetry with the _run/_run_cmd/_run_compact siblings and to + # settle a stray GenerationCancelled to idle rather than leak it. if ws.worker_thread is me and isinstance(ws.ui, WebUI): ws.ui.on_stream_end() ws.ui.on_state_change("idle") + except Exception as exc: + # A failed first turn settles to state=error, NOT idle. The old + # combined arm stamped idle unconditionally, clobbering the + # error row send had just written, so the coord's wait/inspect + # (which reads last_error only for state=="error") saw an empty + # "successful" turn ("no recent assistant output") and the real + # failure stayed invisible until a manual nudge re-ran it. + # ensure_error_recorded is a no-op when send already recorded + # the error in-line (no double state emit) and the recorder when + # a pre-try exception bypassed send's handler (so state=error + # always carries a last_error). Owner-guarded like every + # sibling closure: a late-unwedging abandoned init must not stamp + # over a successor. + # + # Settling to error (not idle) is deliberately terminal for + # automated wakes: a watch/schedule nudge enqueued during a + # failed init is not re-delivered at worker exit + # (wake_workstream_if_pending is idle-gated) — a failed first + # turn is a non-ready terminal, not the idle ready-set that + # timer/watch wakes recur to; explicit user/coordinator action + # reactivates it. A self-healing wake-from-error would be a + # separate wake-gate change, out of scope here. + if ws.worker_thread is me and isinstance(ws.ui, WebUI): + ws.ui.on_stream_end() + session.ensure_error_recorded(exc) finally: - # Deliberately NOT owner-gated, unlike the except arm above + # Deliberately NOT owner-gated, unlike the except arms above # and the _run_cmd/_run_compact follow-ups: those mutate # live slot/UI state a successor now owns, while the notify # is workstream-scoped — an outward signal that this @@ -2524,7 +2561,10 @@ async def _interactive_create_post_install( # gate to prevent — gating it turned force-cancel into # permanent notification loss for scheduled/unattended # workstreams (the empty-content fallback covers the - # error/cancel exits, as it always did). + # error/cancel exits, as it always did). Outcome-honesty — + # a failed or force-cancelled init still notifies the + # empty-content "(Task completed)" fallback, not "Failed:" — + # is deferred to #865. try: last_content = _extract_last_assistant_content(session) _fire_notify_targets(ws, last_content)