From 1a13511586e54871132544012d401ba179bae853 Mon Sep 17 00:00:00 2001 From: yxlyx <85774423+yxlyx@users.noreply.github.com> Date: Mon, 31 Aug 2026 00:04:30 +0800 Subject: [PATCH 1/2] fix(codex): preserve terminal websocket API errors Codex closes its Responses socket immediately after a type:error frame. Graff previously waited for another frame, discarded the accumulated API body on EOF, and retried an unchanged deterministic request across fresh WS and SSE transports. Classify authoritative error frames as terminal API responses, retire the closing socket without charging the transport ladder, and allow only recognized stale chains to rebuild once with full input. Bound parsed code/message diagnostics before tracing or displaying them, including through the fullscreen TUI. Add loopback, interactive PTY, stale-chain, diagnostic-redaction, and headless TUI regressions while retaining the existing fallback and compaction trajectories. Co-Authored-By: Codegraff --- TUI/sim.zig | 37 ++++++ scripts/codex_ws_error_test.py | 205 +++++++++++++++++++++++++++++++++ scripts/codex_ws_mock.py | 9 +- scripts/test-pty-codex-ws.py | 21 ++++ src/agent_request.zig | 17 +-- src/agent_responses.zig | 41 +++++++ src/agent_ws.zig | 14 +-- src/agent_ws_mock.zig | 9 ++ src/agent_ws_reuse_test.zig | 39 +++++++ src/agent_ws_signal.zig | 47 +++++--- src/agent_ws_test.zig | 2 +- src/repl_turn.zig | 4 + 12 files changed, 411 insertions(+), 34 deletions(-) create mode 100644 scripts/codex_ws_error_test.py diff --git a/TUI/sim.zig b/TUI/sim.zig index 9dd735b7..d26ce9e9 100644 --- a/TUI/sim.zig +++ b/TUI/sim.zig @@ -20,10 +20,12 @@ const std = @import("std"); const app = @import("app.zig"); const dump = @import("dump.zig"); +const engine_mod = @import("engine.zig"); const key_mod = @import("key.zig"); const keys = @import("keys.zig"); const render_mod = @import("render.zig"); const theme_mod = @import("theme.zig"); +const turn = @import("turn.zig"); const Model = app.Model; const Effect = app.Effect; @@ -221,6 +223,41 @@ test "typeText lands in the composer dump" { try std.testing.expect(std.mem.indexOf(u8, vis, "›") != null); } +test "#692: terminal API diagnostics survive job finish and the composer recovers" { + const a = std.testing.allocator; + var term: Term = undefined; + term.init(a, 80, 24); + defer term.deinit(); + + try term.model.push(.user, "trigger a provider rejection"); + try term.model.push(.pending, ""); + const job = try a.create(engine_mod.Job); + job.* = .{ + .gpa = a, + .history = &.{}, + .params = .{}, + .stream = .{}, + .raw = .{}, + .threaded = false, + .result = try a.dupe(u8, "codex api error [invalid_request_error]: mock bad request"), + }; + job.events.attach(a); + job.done.store(true, .release); + term.model.pending = job; + turn.finishJob(&term.model); + + const failed = try term.screen(); + defer a.free(failed); + try std.testing.expect(std.mem.indexOf(u8, failed, "invalid_request_error") != null); + try std.testing.expect(std.mem.indexOf(u8, failed, "check /model and your API key") == null); + + _ = term.typeText("continue"); + const recovered = try term.screen(); + defer a.free(recovered); + try std.testing.expect(std.mem.indexOf(u8, recovered, "continue") != null); + try std.testing.expect(std.mem.indexOf(u8, recovered, "›") != null); +} + test "assistant markdown shows lists, quotes, tasks, and rules" { var term: Term = undefined; term.init(std.testing.allocator, 80, 24); diff --git a/scripts/codex_ws_error_test.py b/scripts/codex_ws_error_test.py new file mode 100644 index 00000000..d35c82cc --- /dev/null +++ b/scripts/codex_ws_error_test.py @@ -0,0 +1,205 @@ +#!/usr/bin/env python3 +"""Issue #692 interactive Codex WS error-frame regression scenarios.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +from codex_ws_mock import CodexMock, RecordedRequest +import codex_ws_test as base + +GENERIC_ERROR_CODE = "invalid_request_error" +GENERIC_ERROR_SECRET = "raw-envelope-secret-must-not-leak" +GENERIC_ERROR_TAIL = "generic-error-tail-must-be-truncated" +GENERIC_ERROR_MESSAGE = "mock bad request\r\n" + "x" * 500 + GENERIC_ERROR_TAIL +CHAIN_ERROR_CODE = "previous_response_not_found" +CHAIN_FINAL = "recovered after automatic websocket re-anchor" + + +def generic_error_events(_request: RecordedRequest) -> list[dict]: + """One generic terminal error with raw envelope data that must not leak.""" + return [ + { + "type": "error", + "error": { + "code": GENERIC_ERROR_CODE, + "message": GENERIC_ERROR_MESSAGE, + }, + "debug": {"authorization": GENERIC_ERROR_SECRET}, + } + ] + + +def chain_error_events(request: RecordedRequest) -> list[dict]: + """Tool call, stale chained delta, then success after a full-input redial.""" + if request.ordinal == 1: + call = { + "type": "function_call", + "id": "fc_chain_1", + "call_id": "call_chain_1", + "name": "todo_read", + "arguments": "{}", + "status": "completed", + } + return base.response_events(call, "resp_chain_1", 1_200) + if request.ordinal == 2: + return [ + { + "type": "error", + "error": { + "code": CHAIN_ERROR_CODE, + "message": "Previous response resp_chain_1 was not found", + }, + } + ] + return base.response_events( + base.message_item(CHAIN_FINAL, "msg_chain_final"), + "resp_chain_final", + 1_300, + ) + + +def _env(tmp: str, codex_home: str, port: int) -> tuple[dict[str, str], tuple[str, ...]]: + env = { + "HOME": tmp, + "CODEX_HOME": codex_home, + "CODEGRAFF_API_KEY": "local-pty-test", + "GRAFF_FLEET": "off", + "GRAFF_NO_TELEMETRY": "1", + "GRAFF_CODEX_URL": f"http://127.0.0.1:{port}/backend-api/codex/responses", + "GRAFF_SERVER_COMPACT": "0", + } + ambient = tuple( + key + for key in os.environ + if (key.startswith("GRAFF_") or key.startswith("CODEX_") or key == "NO_COLOR") + and key not in env + ) + return env, ambient + + +def _new_trace(tmp: str, before: set[str]) -> list[dict]: + trace_dir = Path(tmp, ".graff", "traces") + created = sorted(path for path in trace_dir.iterdir() if path.name not in before) + if len(created) != 1: + raise AssertionError(f"expected one new trace, got {created!r}") + return [json.loads(line) for line in created[0].read_text().splitlines() if line] + + +def _assert_ws_stays_primary(session: base.PtySession) -> None: + cursor = len(session.raw) + session.send_line("/models health") + session.wait_for_literal( + "Codex transport: WebSocket primary with automatic SSE fallback", + start=cursor, + ) + session.wait_for_prompt(start=cursor) + + +def _exit(session: base.PtySession, label: str) -> None: + session.send_key("ctrl-d") + result = session.read_until_exit(5.0) + if result.timed_out or result.exit_code != 0: + raise SystemExit( + f"{label}: REPL did not exit cleanly: " + f"exit={result.exit_code} timed_out={result.timed_out}" + ) + + +def run_generic_error_scenario( + tmp: str, codex_home: str, port: int, mock: CodexMock +) -> None: + """A deterministic API error returns once with bounded safe diagnostics.""" + env, ambient = _env(tmp, codex_home, port) + trace_dir = Path(tmp, ".graff", "traces") + before = {path.name for path in trace_dir.iterdir()} if trace_dir.is_dir() else set() + with base.PtySession( + base.GRAFF, + ["--model", "codex", "--no-telemetry"], + cwd=tmp, + env=env, + unset_env=ambient, + timeout=20.0, + ) as session: + session.wait_for_prompt() + cursor = len(session.raw) + session.send_line("trigger the generic websocket rejection") + expected = f"codex api error [{GENERIC_ERROR_CODE}]: mock bad request" + session.wait_for_literal(expected, start=cursor) + session.wait_for_prompt(start=cursor) + rendered = base.terminal_text(bytes(session.raw[cursor:])) + if GENERIC_ERROR_SECRET in rendered or GENERIC_ERROR_TAIL in rendered: + raise AssertionError(f"generic-error: raw/unbounded diagnostics leaked:\n{rendered}") + if mock.ws_turns != 1 or mock.sse_turns != 0 or mock.ws_connections != 1: + raise AssertionError( + "generic-error: deterministic body was retried/fallen back: " + f"connections={mock.ws_connections} ws={mock.ws_turns} sse={mock.sse_turns}" + ) + _assert_ws_stays_primary(session) + _exit(session, "generic-error") + + events = _new_trace(tmp, before) + diagnostics = [event for event in events if event.get("ev") == "ws_api_error"] + if len(diagnostics) != 1: + raise AssertionError(f"generic-error: expected one ws_api_error trace: {events!r}") + detail = diagnostics[0].get("detail", "") + if ( + GENERIC_ERROR_CODE not in detail + or GENERIC_ERROR_SECRET in detail + or GENERIC_ERROR_TAIL in detail + or len(detail.encode("utf-8")) >= 560 + ): + raise AssertionError(f"generic-error: unsafe trace diagnostic: {detail!r}") + ws_details = [event.get("detail", "") for event in events if event.get("ev") == "ws"] + if any("transport error" in detail or "fallback" in detail for detail in ws_details): + raise AssertionError(f"generic-error: API response burned transport ladder: {ws_details!r}") + + +def run_chain_reanchor_scenario( + tmp: str, codex_home: str, port: int, mock: CodexMock +) -> None: + """A recognized stale chain rebuilds once without requiring `continue`.""" + env, ambient = _env(tmp, codex_home, port) + with base.PtySession( + base.GRAFF, + ["--model", "codex", "--no-telemetry"], + cwd=tmp, + env=env, + unset_env=ambient, + timeout=20.0, + ) as session: + session.wait_for_prompt() + cursor = len(session.raw) + session.send_line("exercise automatic stale-chain recovery") + session.wait_for_literal(CHAIN_FINAL, start=cursor) + session.wait_for_prompt(start=cursor) + requests = mock.recorded_requests() + if mock.ws_turns != 3 or mock.sse_turns != 0 or mock.ws_connections != 2: + raise AssertionError( + "chain-error: expected delta rejection then one fresh WS rebuild: " + f"connections={mock.ws_connections} ws={mock.ws_turns} sse={mock.sse_turns}" + ) + if len(requests) != 3: + raise AssertionError(f"chain-error: expected 3 requests, got {requests!r}") + first, rejected, rebuilt = requests + if ( + first.connection_id != rejected.connection_id + or rejected.connection_id == rebuilt.connection_id + or rejected.body.get("previous_response_id") != "resp_chain_1" + or "previous_response_id" in rebuilt.body + ): + raise AssertionError( + "chain-error: rejected delta was not rebuilt as full input on a fresh WS: " + f"{requests!r}" + ) + rebuilt_types = [ + item.get("type") + for item in rebuilt.body.get("input", []) + if isinstance(item, dict) + ] + if "function_call" not in rebuilt_types or "function_call_output" not in rebuilt_types: + raise AssertionError(f"chain-error: full rebuild lost tool history: {rebuilt.body!r}") + _assert_ws_stays_primary(session) + _exit(session, "chain-error") diff --git a/scripts/codex_ws_mock.py b/scripts/codex_ws_mock.py index 0b97a6d8..d5921e33 100644 --- a/scripts/codex_ws_mock.py +++ b/scripts/codex_ws_mock.py @@ -385,13 +385,20 @@ def _serve_ws( self._log(f"ws <- {etype} ({len(message.payload)}b)") if etype != "response.create": continue - for ev in self._events("ws", connection_id, event, headers): + events = self._events("ws", connection_id, event, headers) + for ev in events: _send_frame( conn, OP_TEXT, json.dumps(ev, separators=(",", ":")).encode("utf-8") ) self._log(f"ws -> {ev['type']}") with self._lock: self.ws_turns += 1 + # Real Codex closes immediately after a terminal type:error frame. + # Keeping the mock open would miss #692's exact classification bug: + # the client waited for another frame, saw EOF, and discarded the + # useful API body as a transport reset. + if any(ev.get("type") == "error" for ev in events): + return def _serve_sse( self, conn: socket.socket, reader: _SockReader, headers: dict[str, str] diff --git a/scripts/test-pty-codex-ws.py b/scripts/test-pty-codex-ws.py index 07617519..631d5791 100644 --- a/scripts/test-pty-codex-ws.py +++ b/scripts/test-pty-codex-ws.py @@ -6,6 +6,7 @@ import tempfile from codex_ws_mock import CodexMock +import codex_ws_error_test as error_scenario import codex_ws_test as scenario @@ -100,6 +101,26 @@ def main() -> None: scenario.MIDTURN_CONTEXT_TOKENS = runtime_context scenario.MIDTURN_TOTAL_TOKENS = runtime_context * 9 // 10 + mock = CodexMock(events_for_request=error_scenario.generic_error_events) + port = mock.start() + try: + error_scenario.run_generic_error_scenario(tmp, codex_home, port, mock) + finally: + mock.stop() + print( + "ok generic WS API error: one bounded diagnostic, no transport retry/fallback" + ) + + mock = CodexMock(events_for_request=error_scenario.chain_error_events) + port = mock.start() + try: + error_scenario.run_chain_reanchor_scenario(tmp, codex_home, port, mock) + finally: + mock.stop() + print( + "ok stale WS chain error: automatic full-input re-anchor on one fresh socket" + ) + mock = CodexMock(events_for_request=scenario.midturn_events) port = mock.start() try: diff --git a/src/agent_request.zig b/src/agent_request.zig index 3cbf5ae0..58fff510 100644 --- a/src/agent_request.zig +++ b/src/agent_request.zig @@ -425,13 +425,14 @@ pub fn request(self: *Agent, tools_in: ?[]const u8) !std.json.ObjectMap { }, .err => |failure| { const msg = failure.message; - // (#codex-ws) Belt-and-braces mirror of openai/codex: server - // rejected previous_response_id (stale WS session it no - // longer recognizes) — close it and retry once with full - // input. Gated on codex_prev_id != null (a delta was - // actually sent); it's null after the retry, so this can't - // loop for this request. - if (self.codex_prev_id != null and codex_chain.shouldDropChain(msg, failure.code)) { + const had_chain = self.codex_prev_id != null; + const ws_error_frame = if (self.codex_ws) |c| c.dead else false; + const diagnostic = try responses.failureDiagnostic(self.arena, self.provider.id, failure); + if (ws_error_frame) { + self.closeCodexWs(); + if (self.tracer) |tr| tr.note("ws_api_error", diagnostic); + } + if (had_chain and codex_chain.shouldDropChain(msg, failure.code)) { self.closeCodexWs(); if (self.tracer) |tr| tr.note("ws", "server dropped previous_response_id — re-anchoring with full input"); continue :rebuild; @@ -468,7 +469,7 @@ pub fn request(self: *Agent, tools_in: ?[]const u8) !std.json.ObjectMap { // overload retry path as SSE and JSON error envelopes. if (try retryTransientServerError(self, "", failure.code, msg, &server_retries)) continue :rebuild; if (self.tracer) |tr| tr.api(self.label, self.sub, self.provider.model, ms, body.len, resp_body.len, 0, 0, true); - try self.sayApiError("{s} api error: {s}", .{ self.provider.id, msg }); + try self.sayApiError("{s}", .{diagnostic}); return error.ApiError; }, } diff --git a/src/agent_responses.zig b/src/agent_responses.zig index 7ee0e0b8..f7a1ec57 100644 --- a/src/agent_responses.zig +++ b/src/agent_responses.zig @@ -100,6 +100,47 @@ pub fn parseResponses(self: *Agent, body: []const u8) !ResponsesResult { return error.Unparseable; } +/// Safe diagnostic for a parsed Responses failure. This is the redaction +/// boundary for WS error frames: only the provider, structured code and message +/// survive — never the raw envelope, echoed request, headers or auth fields. +/// Each field is single-lined and bounded before it reaches last_api_error or +/// the default-on trace. +pub fn failureDiagnostic(allocator: std.mem.Allocator, provider: []const u8, failure: ResponsesFailure) ![]u8 { + var buf: [560]u8 = undefined; + var w: std.Io.Writer = .fixed(&buf); + try writeDiagnosticField(&w, provider, 40); + try w.writeAll(" api error"); + if (failure.code) |code| { + try w.writeAll(" ["); + try writeDiagnosticField(&w, code, 96); + try w.writeByte(']'); + } + try w.writeAll(": "); + try writeDiagnosticField(&w, failure.message, 384); + return allocator.dupe(u8, w.buffered()); +} + +fn writeDiagnosticField(w: *std.Io.Writer, raw: []const u8, max: usize) !void { + const prefix = util.utf8Prefix(raw, max); + for (prefix) |b| try w.writeByte(if (b < 0x20 or b == 0x7f) ' ' else b); + if (prefix.len < raw.len) try w.writeAll("…"); +} + +test "failureDiagnostic retains only bounded single-line code and message" { + const a = std.testing.allocator; + const long = "x" ** 500; + const diagnostic = try failureDiagnostic(a, "codex", .{ + .code = "invalid_request_error\nignored-envelope", + .message = "bad request\r\n" ++ long, + }); + defer a.free(diagnostic); + try std.testing.expect(std.mem.startsWith(u8, diagnostic, "codex api error [invalid_request_error ignored-envelope]: bad request ")); + try std.testing.expect(std.mem.endsWith(u8, diagnostic, "…")); + try std.testing.expect(diagnostic.len < 560); + try std.testing.expect(std.mem.indexOfScalar(u8, diagnostic, '\n') == null); + try std.testing.expect(std.mem.indexOfScalar(u8, diagnostic, '\r') == null); +} + test "parseResponses: terminal failure beats partial items; incomplete stays marked" { var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena_state.deinit(); diff --git a/src/agent_ws.zig b/src/agent_ws.zig index 0dfa87fc..caefefe7 100644 --- a/src/agent_ws.zig +++ b/src/agent_ws.zig @@ -579,14 +579,12 @@ pub fn postResponsesWs(self: *Agent, body: []const u8) ![]u8 { try full.writer.writeAll(fbuf.items); try full.writer.writeByte('\n'); const line = try std.fmt.allocPrint(arena, "data: {s}", .{fbuf.items}); - // xAI error frames are terminal but not in isStreamEnd's set. Both - // arms route through postLive's close + redial (resets chain state). - switch (signal.errorFrameAction(fbuf.items)) { - .none => {}, - .retire, .chain_lost => |act| { - if (self.tracer) |tr| tr.note("ws", if (act == .retire) "server connection limit — retiring socket" else "chain anchor gone — re-anchoring full"); - return error.ConnectionResetByPeer; - }, + // Codex closes after `type:error`; return the accumulated API body + // instead of waiting for EOF and misclassifying it as transport loss. + if (signal.errorFrameAction(gpa, fbuf.items) != .none) { + client.dead = true; // no courtesy close write to a closing peer + if (self.tracer) |tr| tr.note("ws", "terminal API error frame"); + break :stream; } if (isStreamEnd(arena, self.provider.kind, line)) { if (self.tracer) |tr| tr.note("ws", "completed"); diff --git a/src/agent_ws_mock.zig b/src/agent_ws_mock.zig index 1f81e65c..70871df0 100644 --- a/src/agent_ws_mock.zig +++ b/src/agent_ws_mock.zig @@ -20,6 +20,7 @@ pub fn nowMs(io: Io) i64 { pub const delta_event = "{\"type\":\"response.output_text.delta\",\"delta\":\"hi\"}"; pub const completed_event = "{\"type\":\"response.completed\"}"; +pub const generic_error_event = "{\"type\":\"error\",\"error\":{\"code\":\"invalid_request_error\",\"message\":\"mock bad request\"}}"; /// What the backend actually puts on the socket in the first milliseconds after /// a send, before the model has produced anything: the two protocol events, then @@ -76,6 +77,9 @@ pub const Mock = struct { /// Upgrade, take the client's frame, stream a whitelisted meta call's /// arguments (`arg_prose_events`) and then go silent. arg_prose_then_silence, + /// Send a generic terminal API error and close immediately, matching + /// Codex's failure sequence from issue #692. + generic_error_then_close, }; pub fn run(io: Io, server: *std.Io.net.Server, mode: Mode, done: *std.atomic.Value(bool)) void { @@ -119,6 +123,11 @@ pub const Mock = struct { for (arg_prose_events) |ev| writeTextFrame(&sw.interface, ev) catch return idle(io, done); }, + .generic_error_then_close => { + readClientFrame(&sr.interface) catch return; + writeTextFrame(&sw.interface, generic_error_event) catch return; + return; + }, } idle(io, done); } diff --git a/src/agent_ws_reuse_test.zig b/src/agent_ws_reuse_test.zig index cb6dbc64..0f7a84b3 100644 --- a/src/agent_ws_reuse_test.zig +++ b/src/agent_ws_reuse_test.zig @@ -233,3 +233,42 @@ test "#401: a fresh connect keeps the full pre-first-token budget for a slow fir try std.testing.expect(!traced(&tw, "\"detail\":\"stall\"")); try std.testing.expect(agent.codex_ws != null); // held for the next delta } + +test "#692: a generic error frame is returned as the terminal API body before peer close" { + if (builtin.os.tag == .windows) return error.SkipZigTest; + const gpa = std.testing.allocator; + const io = std.testing.io; + + var addr = try std.Io.net.IpAddress.parseLiteral("127.0.0.1:0"); + var server = try std.Io.net.IpAddress.listen(&addr, io, .{}); + defer server.deinit(io); + var done: std.atomic.Value(bool) = .init(false); + var fut = io.async(Mock.run, .{ io, &server, Mock.Mode.generic_error_then_close, &done }); + defer fut.await(io); + defer done.store(true, .release); + + var arena_state = std.heap.ArenaAllocator.init(gpa); + defer arena_state.deinit(); + const arena = arena_state.allocator(); + var tw: Io.Writer.Allocating = .init(gpa); + defer tw.deinit(); + var tracer: trace.Tracer = .{ .io = io, .gpa = gpa, .out = &tw.writer, .start = Io.Timestamp.now(io, .awake) }; + + var url_buf: [64]u8 = undefined; + const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/x", .{server.socket.address.getPort()}); + var agent = mockAgent(gpa, arena, io, url); + agent.tracer = &tracer; + defer if (agent.codex_ws) |c| { + c.dead = true; + c.deinit(gpa); + agent.codex_ws = null; + }; + + const out = try agent_ws.postResponsesWs(&agent, "{\"model\":\"gpt-5\",\"input\":[]}"); + defer gpa.free(out); + try std.testing.expect(std.mem.indexOf(u8, out, mock.generic_error_event) != null); + try std.testing.expect(traced(&tw, "\"detail\":\"terminal API error frame\"")); + try std.testing.expect(!traced(&tw, "transport error")); + try std.testing.expect(agent.codex_ws != null); + try std.testing.expect(agent.codex_ws.?.dead); +} diff --git a/src/agent_ws_signal.zig b/src/agent_ws_signal.zig index 8fc7f529..b503c482 100644 --- a/src/agent_ws_signal.zig +++ b/src/agent_ws_signal.zig @@ -127,18 +127,31 @@ pub const TokenSignal = struct { } }; -/// xAI WS error frames ({"type":"error"}) are terminal for the turn but not in -/// isStreamEnd's completed/failed set, so without classification they burn the -/// whole stall budget on a doomed socket. -pub const ErrorFrameAction = enum { none, retire, chain_lost }; +/// Responses WS `type:error` frames are terminal API responses, not transport +/// failures. The two known chain errors retain distinct labels for trace +/// diagnostics, but all three actions return the accumulated response body to +/// parseResponses instead of waiting for the peer's expected close frame. +pub const ErrorFrameAction = enum { none, api_error, retire, chain_lost }; -pub fn errorFrameAction(frame: []const u8) ErrorFrameAction { - if (std.mem.indexOf(u8, frame, "\"type\":\"error\"") == null) return .none; - // The server sends this right before closing a 25-minute-old socket. - if (std.mem.indexOf(u8, frame, "websocket_connection_limit_reached") != null) return .retire; - // Our chain anchor is gone (evicted / never cached under store:false). - if (std.mem.indexOf(u8, frame, "previous_response_not_found") != null) return .chain_lost; - return .none; +pub fn errorFrameAction(gpa: std.mem.Allocator, frame: []const u8) ErrorFrameAction { + // Error frames are rare. Parse only a cheap candidate so whitespace around + // `type` is accepted and prose that merely quotes `"type":"error"` is not. + if (std.mem.indexOf(u8, frame, "error") == null) return .none; + var scratch = std.heap.ArenaAllocator.init(gpa); + defer scratch.deinit(); + const v = std.json.parseFromSliceLeaky(std.json.Value, scratch.allocator(), frame, .{ .allocate = .alloc_always }) catch return .none; + if (v != .object) return .none; + const ty = v.object.get("type") orelse return .none; + if (ty != .string or !std.mem.eql(u8, ty.string, "error")) return .none; + const err = v.object.get("error"); + const code = if (err) |e| (if (e == .object) e.object.get("code") else null) else null; + if (code) |c| if (c == .string) { + // The server sends this right before closing a 25-minute-old socket. + if (std.mem.eql(u8, c.string, "websocket_connection_limit_reached")) return .retire; + // Our chain anchor is gone (evicted / never cached under store:false). + if (std.mem.eql(u8, c.string, "previous_response_not_found")) return .chain_lost; + }; + return .api_error; } /// The deadline for writing a frame of `frame_len` bytes. @@ -156,9 +169,11 @@ pub fn sendDeadlineMs(frame_len: usize, head_ms: u64, stream_ms: u64) u64 { return @min(head_ms +| grow, @max(head_ms, stream_ms)); } -test "errorFrameAction classifies the two xAI ws error codes, ignores prose" { - try std.testing.expectEqual(ErrorFrameAction.retire, errorFrameAction("{\"type\":\"error\",\"error\":{\"code\":\"websocket_connection_limit_reached\"}}")); - try std.testing.expectEqual(ErrorFrameAction.chain_lost, errorFrameAction("{\"type\":\"error\",\"error\":{\"code\":\"previous_response_not_found\"}}")); - try std.testing.expectEqual(ErrorFrameAction.none, errorFrameAction("{\"type\":\"response.output_text.delta\",\"delta\":\"websocket_connection_limit_reached\"}")); - try std.testing.expectEqual(ErrorFrameAction.none, errorFrameAction("{\"type\":\"error\",\"error\":{\"code\":\"other\"}}")); +test "errorFrameAction classifies every actual error frame and ignores quoted prose" { + const a = std.testing.allocator; + try std.testing.expectEqual(ErrorFrameAction.retire, errorFrameAction(a, "{\"type\":\"error\",\"error\":{\"code\":\"websocket_connection_limit_reached\"}}")); + try std.testing.expectEqual(ErrorFrameAction.chain_lost, errorFrameAction(a, "{\"type\":\"error\",\"error\":{\"code\":\"previous_response_not_found\"}}")); + try std.testing.expectEqual(ErrorFrameAction.api_error, errorFrameAction(a, "{ \"type\" : \"error\", \"error\" : {\"code\":\"invalid_request_error\"} }")); + try std.testing.expectEqual(ErrorFrameAction.none, errorFrameAction(a, "{\"type\":\"response.output_text.delta\",\"delta\":\"quoted \\\"type\\\":\\\"error\\\"\"}")); + try std.testing.expectEqual(ErrorFrameAction.none, errorFrameAction(a, "not json: error")); } diff --git a/src/agent_ws_test.zig b/src/agent_ws_test.zig index 52dfb8c6..6894ac34 100644 --- a/src/agent_ws_test.zig +++ b/src/agent_ws_test.zig @@ -79,7 +79,7 @@ test "codexWsIdleExpired: fires only strictly past the idle limit (codex-ws)" { test "the codex .responses arm refreshes auth and re-anchors before resending (#402)" { const src = @embedFile("agent_request.zig"); const arm_start = std.mem.indexOf(u8, src, "unparseable codex response").?; - const arm_end = std.mem.indexOf(u8, src, "{s} api error: {s}").?; + const arm_end = std.mem.indexOf(u8, src, "try self.sayApiError(\"{s}\", .{diagnostic})").?; const arm = src[arm_start..arm_end]; const call = std.mem.indexOf(u8, arm, "retryAfterAuthRefresh(self, msg, &auth_refreshed)") orelse diff --git a/src/repl_turn.zig b/src/repl_turn.zig index 624e9e98..b8cb137b 100644 --- a/src/repl_turn.zig +++ b/src/repl_turn.zig @@ -210,6 +210,10 @@ pub fn replTurnCb(ctx_ptr: ?*anyopaque, gpa: Allocator, history: []const repl.Tu error.StreamStalled => return earlyEnd(gpa, &agent, "stream stalled"), // A mid-stream provider drop (#133), same handling as a stall. error.StreamDropped => return earlyEnd(gpa, &agent, "connection dropped"), + // The Responses WS path now preserves a bounded provider diagnostic. + // Return it as the turn result so the fullscreen TUI does not discard + // the live error and replace it with the misleading API-key fallback. + error.ApiError => return gpa.dupe(u8, agent.last_api_error orelse "provider API error") catch null, error.FallbackConsentRequired => return gpa.dupe(u8, "Saved model unavailable. Allow this provider with /fallback in the standard REPL, or choose another model.") catch null, else => return null, }; From 484152095c118240316cfc2dd3e4cc4147343bf4 Mon Sep 17 00:00:00 2001 From: yxlyx <85774423+yxlyx@users.noreply.github.com> Date: Mon, 31 Aug 2026 00:16:03 +0800 Subject: [PATCH 2/2] test(ci): avoid version-sensitive string repetition The local Zig 0.16 compiler accepted a repeated string expression in the new diagnostic test, but the repository's pinned Zig 0.17 CI compiler rejects it during parsing on both Linux and Windows. Build the same long message with the repository's established @splat pattern so the test remains portable across both toolchains. Co-Authored-By: Codegraff --- src/agent_responses.zig | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/agent_responses.zig b/src/agent_responses.zig index f7a1ec57..90df207a 100644 --- a/src/agent_responses.zig +++ b/src/agent_responses.zig @@ -128,10 +128,11 @@ fn writeDiagnosticField(w: *std.Io.Writer, raw: []const u8, max: usize) !void { test "failureDiagnostic retains only bounded single-line code and message" { const a = std.testing.allocator; - const long = "x" ** 500; + var long: [513]u8 = @splat('x'); + @memcpy(long[0..13], "bad request\r\n"); const diagnostic = try failureDiagnostic(a, "codex", .{ .code = "invalid_request_error\nignored-envelope", - .message = "bad request\r\n" ++ long, + .message = &long, }); defer a.free(diagnostic); try std.testing.expect(std.mem.startsWith(u8, diagnostic, "codex api error [invalid_request_error ignored-envelope]: bad request "));