fix(server): init-message worker exits to error, not idle, on first-turn failure

The initial-message worker (_run_initial) collapsed both cancel and
backend-error exits into one `except (Exception, GenerationCancelled)`
arm that always stamped state=idle, clobbering the state=error that
session.send's _record_fatal_error had persisted+emitted. A spawned
child's first-turn backend failure (unreachable model server, exhausted
quota, auth error) therefore read as an empty, successful turn — the
coordinator's wait/inspect surface reads last_error only for
state=='error' — and the real error surfaced only after a manual nudge
re-ran the turn synchronously.

Split the arm: cancel -> idle, exception -> error. The failed child now
settles at state=error and the first wait_for_workstream returns the
enriched backend error inline. Also fixes the same latent bug for
scheduled tasks, which dispatch through the same endpoint and closure.

A failed first turn is deliberately terminal for automated wakes: it
settles to a non-ready error terminal, not the idle ready-set that
timer/watch wakes recur to, so explicit user/coordinator action
reactivates it rather than a silent auto-retry (a self-healing
wake-from-error would be a separate wake-gate change).

The exception arm routes through a new ChatSession.ensure_error_recorded:
a no-op when send already recorded the error in-line (the common
backend-boundary path — no duplicate state emit), and the recorder when a
pre-try exception (model-registry refresh, user-turn append,
system-message recompose) bypassed send's own handler, so state=error
always carries a meaningful last_error. Its idempotency guard
(_has_persisted_error) is session-lifetime, so ensure_error_recorded is
scoped to _run_initial's FRESH first-turn session only; the docstring
spells out why a session-reuse caller (retry, /send, coord send, wake)
must not route through it until the per-turn error-recorded signal of
#865 lands.

The other half of making an errored workstream cheap for a model to
handle is a stable identifier: the enriched backend error now leads with
the model ALIAS the coordinator references everywhere (list_nodes, spawn)
and annotates the backend id for the operator —
"model=DeepSeek-V4-Flash (id=deepseek-v4-flash)" — so a model routing
around a failed model correlates it against those surfaces without a
lookup, instead of burning reasoning tokens reconciling the alias against
a backend id it never sees anywhere else. Collapses to one token when the
alias and id coincide.

Tests (TestInitialWorkerFailureState) assert the coordinator-visible
manager state and the persisted last_error across the matrix — common-
backend and pre-try errors both settle error with a readable last_error;
cancel-to-idle settles idle with no error recorded. De-forks the
create-app fixture and uses the shared monotonic wait_until helper.

The completion-notification honesty surface and the error-recording
hygiene of the other send-worker closures (retry / main send / coord send
/ wake) are deferred to #865.

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)

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"))

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 ""
 

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)