diff --git a/coworker/engine.py b/coworker/engine.py index 3c14b179..bbfd75ce 100644 --- a/coworker/engine.py +++ b/coworker/engine.py @@ -177,6 +177,31 @@ class TurnEngine: async for event in self._loop(): yield event + def _append_notice(self, kind: str, text: Optional[str] = None) -> None: + """Persist a turn-ending marker (error/interrupted) as a display-only `notice` + message: it survives reload like the transcript does, but `_outbound_messages` + drops the role so no provider ever sees it.""" + notice: dict[str, Any] = {"role": "notice", "kind": kind, "ts": time.time()} + if text: + notice["text"] = text + self.messages.append(notice) + + async def retry(self) -> AsyncIterator[Event]: + """Re-run the model loop after a provider error — no new user message; the failed + turn's input is already the tail of history. Guarded on the tail being an error + notice so a stray retry frame can't re-answer a completed turn.""" + last = self.messages[-1] if self.messages else None + if not ( + isinstance(last, dict) + and last.get("role") == "notice" + and last.get("kind") == "error" + ): + return + self._cancel.clear() + yield Event(EventType.TURN_START, {"input": ""}) + async for event in self._loop(): + yield event + async def resume(self) -> AsyncIterator[Event]: """Continue a turn that was suspended at a prompt and persisted — durable resume after a restart (or engine eviction). Re-process the trailing assistant message's UNANSWERED @@ -257,6 +282,7 @@ class TurnEngine: } if friendly: payload["raw"] = str(exc) + self._append_notice("error", friendly or str(exc)) yield Event(EventType.ERROR, payload) return if self._cancel.is_set() and turn is None: @@ -267,6 +293,7 @@ class TurnEngine: self.messages.append( _assistant_message(AssistantTurn(text="".join(streamed))) ) + self._append_notice("interrupted") yield Event(EventType.INTERRUPTED, {"iterations": iterations}) return if turn is None: @@ -294,6 +321,7 @@ class TurnEngine: yield Event(EventType.ITERATION_END, {"iteration": iterations}) if self._cancel.is_set(): + self._append_notice("interrupted") yield Event(EventType.INTERRUPTED, {"iterations": iterations}) return if self._steering: @@ -802,7 +830,8 @@ class TurnEngine: """ # Strip the display-only sidecars — `source` (connector cards), `_display` # (e.g. filter-hidden counts), and `ts` (append-time timestamps) — copying only - # messages that carry one. + # messages that carry one. Whole `notice` messages (error/interrupted markers) + # are display-only too: dropped entirely. _SIDECARS = ("source", "_display", "ts") out = [ ( @@ -811,6 +840,7 @@ class TurnEngine: else msg ) for msg in self.messages + if msg.get("role") != "notice" ] # PDF attachments (stored as `file` parts) are adapted to the ACTIVE model right # here — never in the persisted history — so a mid-session model switch always diff --git a/coworker/server/app.py b/coworker/server/app.py index 5c50bcb3..3908ccc3 100644 --- a/coworker/server/app.py +++ b/coworker/server/app.py @@ -1576,12 +1576,13 @@ def create_app(manager: SessionManager) -> FastAPI: "iteration_end", } - async def run_turn(content) -> None: + async def run_turn(content, *, retry: bool = False) -> None: manager.mark_running( session_id ) # busy → self-wakes steer instead of colliding try: - async for event in engine.run(content): + events = engine.retry() if retry else engine.run(content) + async for event in events: # Broadcast to every socket viewing this session (this socket included — it's a # registered client), so a second view of the same session stays in sync too. await manager.broadcast_session( @@ -1629,6 +1630,11 @@ def create_app(manager: SessionManager) -> FastAPI: _resolve_pending(str(message.get("answer", ""))) elif kind == "interrupt": engine.request_interrupt() + elif kind == "retry": + # Re-run after a provider error (engine guards on the error-notice + # tail, so a stray frame is a no-op that still ends with turn_done). + if not manager.is_running(session_id): + asyncio.create_task(run_turn(None, retry=True)) elif kind == "set_mode": try: engine.permissions.mode = Mode(message.get("mode")) diff --git a/surfaces/gui/e2e/error-retry.spec.ts b/surfaces/gui/e2e/error-retry.spec.ts new file mode 100644 index 00000000..12fd3395 --- /dev/null +++ b/surfaces/gui/e2e/error-retry.spec.ts @@ -0,0 +1,27 @@ +// Model-layer roadmap item 1 (2026-07-22): a turn that dies on a provider error leaves a +// visible, persistent marker with a Retry affordance. Retry re-runs the failed turn with NO +// new user bubble; once the turn recovers, the button disappears (the notice is history). +import { expect } from "@playwright/test"; +import { test } from "./fixtures"; + +test("provider error shows a retriable notice; Retry re-runs without a new user message", async ({ + page, +}) => { + await page.goto("/"); + await page.getByText("Draft the launch note").first().click(); + const box = page.getByPlaceholder(/Ask the coworker/); + await box.fill("please fail the turn"); + await box.press("Enter"); + + await expect(page.getByText("Error: model unreachable").first()).toBeVisible({ timeout: 10_000 }); + const retry = page.getByTestId("notice-retry"); + await expect(retry).toBeVisible(); + + await retry.click(); + await expect(page.getByText("Recovered after retry.").first()).toBeVisible({ timeout: 10_000 }); + + // No fake user bubble from the retry turn, exactly one real one… + await expect(page.locator(".bubble-user")).toHaveCount(1); + // …and the button is gone now that the error notice is no longer the transcript tail. + await expect(page.getByTestId("notice-retry")).toHaveCount(0); +}); diff --git a/surfaces/gui/e2e/fixtures.ts b/surfaces/gui/e2e/fixtures.ts index 3bf66025..5d11d97e 100644 --- a/surfaces/gui/e2e/fixtures.ts +++ b/surfaces/gui/e2e/fixtures.ts @@ -646,6 +646,12 @@ export async function mockApi(page: import("@playwright/test").Page) { }); return; } + // A turn that dies on a provider error; the follow-up {type:"retry"} recovers. + if (/fail the turn/i.test(msg.text)) { + send("error", { error: "model unreachable" }); + send("turn_done"); + return; + } // A deliberately SLOW multi-second stream (~40 ticks × 120ms) so specs can // interact mid-turn — the follow/pin scroll contract (FB-004) is untestable // against the instant echo below. @@ -697,6 +703,11 @@ export async function mockApi(page: import("@playwright/test").Page) { } send("interrupted", {}); send("turn_done"); + } else if (msg.type === "retry") { + // Like the real engine: re-runs with NO new user message (turn_start input is empty). + send("turn_start", { input: "" }); + send("assistant_message", { text: "Recovered after retry." }); + send("turn_done"); } }); }); diff --git a/surfaces/gui/src/App.tsx b/surfaces/gui/src/App.tsx index 64223c89..62ca9fff 100644 --- a/surfaces/gui/src/App.tsx +++ b/surfaces/gui/src/App.tsx @@ -644,7 +644,10 @@ export function App() { break; case "error": flushPartialStream(); - setItems((p) => [...p, { kind: "notice", tone: "warn", text: "Error: " + (d.error || "unknown") }]); + setItems((p) => [ + ...p, + { kind: "notice", tone: "warn", text: "Error: " + (d.error || "unknown"), retriable: true }, + ]); break; case "turn_done": setRunning(false); @@ -799,6 +802,11 @@ export function App() { const prefillComposer = (text: string, attachments?: Attachment[]) => setComposerPrefill((p) => ({ text, attachments, nonce: (p?.nonce ?? 0) + 1 })); const interrupt = () => sessionRef.current?.interrupt(); + const retry = () => { + // Optimistic running: turn_start confirms; a rejected retry still ends in turn_done. + setRunning(true); + sessionRef.current?.retry(); + }; const changeMode = (m: string) => { setMode(m); sessionRef.current?.setMode(m); @@ -1400,6 +1408,7 @@ export function App() { items={items} onApprove={approve} running={running} + onRetry={retry} // §33 ref #3: sub-threshold streamed text renders INSIDE the live turn // group (header when collapsed, quiet line when expanded) — never as a // floating paragraph. diff --git a/surfaces/gui/src/api.ts b/surfaces/gui/src/api.ts index 10592365..b1589719 100644 --- a/surfaces/gui/src/api.ts +++ b/surfaces/gui/src/api.ts @@ -1746,6 +1746,12 @@ export class Session { this.send({ type: "interrupt" }); } + // Re-run a turn that ended in a provider error — no new user message; the server + // guards on the history tail so a stray frame is a no-op. + retry() { + this.send({ type: "retry" }); + } + setMode(mode: string) { this.send({ type: "set_mode", mode }); } diff --git a/surfaces/gui/src/components/Transcript.tsx b/surfaces/gui/src/components/Transcript.tsx index dcb1342b..3551d661 100644 --- a/surfaces/gui/src/components/Transcript.tsx +++ b/surfaces/gui/src/components/Transcript.tsx @@ -280,9 +280,12 @@ interface Props { running?: boolean; // Sub-threshold streamed text (streamGate mode "quiet") — handed to the live turn group. streamingText?: string; + // Re-run the failed turn (no new user message). Offered only on a retriable notice that + // is the transcript tail of an idle session — anywhere else the error is history. + onRetry?: () => void; } -export function Transcript({ items, running, streamingText }: Props) { +export function Transcript({ items, running, streamingText, onRetry }: Props) { // §33 grouping: a turn = the maximal run of assistant/tool/resolved-approval items between // breakers (user, connector, notices, plan/dir requests…). Trailing assistant texts are the // ANSWER and render as bubbles after the group; interior assistant texts are narration and @@ -396,6 +399,11 @@ export function Transcript({ items, running, streamingText }: Props) { return (
{item.text} + {item.retriable && !running && onRetry && block.i === items.length - 1 && ( + + )}
); default: diff --git a/surfaces/gui/src/itemsFromMessages.test.ts b/surfaces/gui/src/itemsFromMessages.test.ts index 3ae23e86..f8b9cf95 100644 --- a/surfaces/gui/src/itemsFromMessages.test.ts +++ b/surfaces/gui/src/itemsFromMessages.test.ts @@ -48,3 +48,23 @@ describe("itemsFromMessages timestamps", () => { ]); }); }); + +describe("itemsFromMessages notices", () => { + it("replays persisted error/interrupted markers; only errors are retriable", () => { + const items = itemsFromMessages([ + { role: "user", content: "hi" }, + { role: "assistant", content: "partial ans" }, + { role: "notice", kind: "interrupted", ts: 1752969720 }, + { role: "user", content: "again" }, + { role: "notice", kind: "error", text: "model down", ts: 1752969724 }, + ] as any); + + expect(items).toEqual([ + { kind: "user", text: "hi" }, + { kind: "assistant", text: "partial ans" }, + { kind: "notice", tone: "warn", text: "Interrupted." }, + { kind: "user", text: "again" }, + { kind: "notice", tone: "warn", text: "Error: model down", retriable: true }, + ]); + }); +}); diff --git a/surfaces/gui/src/itemsFromMessages.ts b/surfaces/gui/src/itemsFromMessages.ts index e7c67f8b..1ac8b53c 100644 --- a/surfaces/gui/src/itemsFromMessages.ts +++ b/surfaces/gui/src/itemsFromMessages.ts @@ -58,6 +58,15 @@ export function itemsFromMessages(messages: ConversationMessage[]): Item[] { ...(hidden ? { hidden } : {}), }); } + } else if (m.role === "notice") { + // Persisted turn-ending marker (engine `_append_notice`): error/interrupted survive + // reload exactly like the live view rendered them. An error notice is retriable — + // the Transcript only offers the button when it's the transcript tail. + items.push( + m.kind === "interrupted" + ? { kind: "notice", tone: "warn", text: "Interrupted." } + : { kind: "notice", tone: "warn", text: "Error: " + (m.text || "unknown"), retriable: true }, + ); } // system messages are omitted; tool-result messages are folded into the tool row above } diff --git a/surfaces/gui/src/types.ts b/surfaces/gui/src/types.ts index b135bed7..b835eeee 100644 --- a/surfaces/gui/src/types.ts +++ b/surfaces/gui/src/types.ts @@ -114,4 +114,4 @@ export type Item = multi?: boolean; resolved?: string; } - | { kind: "notice"; tone: "info" | "warn"; text: string }; + | { kind: "notice"; tone: "info" | "warn"; text: string; retriable?: boolean }; diff --git a/tests/test_engine_stop.py b/tests/test_engine_stop.py index 790fc08b..38da2d68 100644 --- a/tests/test_engine_stop.py +++ b/tests/test_engine_stop.py @@ -86,10 +86,18 @@ def test_stop_mid_stream_keeps_partial_text(tmp_path): assert events[-1].type == EventType.INTERRUPTED # Far fewer than the full 200 chunks were consumed… assert provider.chunks_produced < 100 - # …and the partial text the user watched is persisted, with no tool calls. - last = engine.messages[-1] - assert last["role"] == "assistant" and last["content"].startswith("w0 ") - assert "tool_calls" not in last + # …and the partial text the user watched is persisted, with no tool calls, + # capped by the interrupted marker (display-only notice role). + assert engine.messages[-1] == { + "role": "notice", + "kind": "interrupted", + "ts": engine.messages[-1]["ts"], + } + partial = engine.messages[-2] + assert partial["role"] == "assistant" and partial["content"].startswith("w0 ") + assert "tool_calls" not in partial + # The notice never reaches a provider. + assert all(m.get("role") != "notice" for m in engine._outbound_messages()) class FailingStreamProvider(ProviderClient): @@ -120,9 +128,72 @@ def test_provider_error_mid_stream_keeps_partial_text(tmp_path): events = asyncio.run(run()) assert events[-1].type == EventType.ERROR - last = engine.messages[-1] - assert last["role"] == "assistant" and last["content"] == "partial answer" - assert "tool_calls" not in last + notice = engine.messages[-1] + assert notice["role"] == "notice" and notice["kind"] == "error" + assert "provider went away" in notice["text"] + partial = engine.messages[-2] + assert partial["role"] == "assistant" and partial["content"] == "partial answer" + assert "tool_calls" not in partial + + +class FlakyProvider(ProviderClient): + """Fails the first N stream calls, then answers — a provider outage that recovers.""" + + def __init__(self, failures=1): + self._failures = failures + self.calls = 0 + + def complete(self, **kwargs): # pragma: no cover + raise NotImplementedError + + def capabilities(self, model): + return ModelCapabilities() + + def stream(self, *, model, messages, tools=None, **settings): + self.calls += 1 + if self.calls <= self._failures: + raise RuntimeError("outage") + yield StreamChunk(turn=AssistantTurn(text="recovered", finish_reason="stop")) + + +def test_retry_reruns_failed_turn_without_new_user_message(tmp_path): + provider = FlakyProvider(failures=1) + engine = TurnEngine( + provider=provider, + registry=ToolRegistry(), + permissions=PermissionEngine(workspace_root=tmp_path), + model="gpt-5.5", + ) + + async def scenario(): + first = [ev async for ev in engine.run("hello")] + second = [ev async for ev in engine.retry()] + return first, second + + first, second = asyncio.run(scenario()) + assert first[-1].type == EventType.ERROR + assert second[-1].type == EventType.TURN_END + # Exactly one user message — retry re-runs, it doesn't re-ask. + assert sum(1 for m in engine.messages if m.get("role") == "user") == 1 + assert engine.messages[-1]["content"] == "recovered" + + +def test_retry_is_noop_unless_tail_is_error_notice(tmp_path): + engine = TurnEngine( + provider=OneTurnProvider(AssistantTurn(text="done", finish_reason="stop")), + registry=ToolRegistry(), + permissions=PermissionEngine(workspace_root=tmp_path), + model="gpt-5.5", + ) + + async def scenario(): + async for _ in engine.run("hello"): + pass + return [ev async for ev in engine.retry()] + + # A completed session must not grow a second answer from a stray retry frame. + assert asyncio.run(scenario()) == [] + assert engine.messages[-1]["content"] == "done" def test_stop_while_awaiting_approval(tmp_path): diff --git a/tests/test_server.py b/tests/test_server.py index b6eded07..40614c64 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -323,6 +323,38 @@ def test_ws_simple_turn(tmp_path): assert "turn_end" in types +def test_ws_error_persists_notice_and_retry_reruns(tmp_path): + class FlakyProvider(ProviderClient): + def __init__(self): + self.calls = 0 + + def complete(self, *, model, messages, tools=None, **settings): + self.calls += 1 + if self.calls == 1: + raise RuntimeError("outage") + return _text("recovered") + + def capabilities(self, model): + return ModelCapabilities() + + manager = SessionManager(workspace=tmp_path, provider=FlakyProvider()) + client = TestClient(create_app(manager)) + with client.websocket_connect("/ws/session/flaky") as ws: + assert ws.receive_json()["type"] == "ready" + ws.send_json({"type": "user_message", "text": "hello"}) + assert "error" in _drain(ws) + # The error survives as a persisted notice (reload shows what happened)… + messages = client.get("/v1/sessions/flaky/messages").json()["messages"] + assert messages[-1]["role"] == "notice" and messages[-1]["kind"] == "error" + # …and retry re-runs the turn without a new user message. + ws.send_json({"type": "retry"}) + types = _drain(ws) + assert "turn_start" in types and "assistant_message" in types + messages = client.get("/v1/sessions/flaky/messages").json()["messages"] + assert messages[-1]["role"] == "assistant" and messages[-1]["content"] == "recovered" + assert sum(1 for m in messages if m["role"] == "user") == 1 + + # -- origin gate (local-API hardening): a browser page on a foreign origin must not be able to # read the API cross-origin or open the driving WebSocket. -------------------------------------