fix(codex): preserve terminal WebSocket API errors

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

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

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]

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:

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;
                 },
             }

src/agent_responses.zig

@@ -100,6 +100,48 @@ 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;
+    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 = &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();