fix(ws): a 426 latches SSE immediately — the server answered authoritatively (#427)

src/agent_ws.zig

@@ -7,7 +7,7 @@
 //!
 //! postLive() is the transport selector called by request(): codex root turns
 //! try ws (postResponsesWs), retry one failed WS with a clean full-history
-//! re-anchor, then latch onto the launch-scoped prewarmed HTTP client for SSE.
+//! re-anchor (a 426 skips it), then latch the prewarmed HTTP client for SSE.
 //!
 //! Observability (issue #134's ask): every ws lifecycle step is routed to the
 //! tracer ("ws" notes: connecting/connected/reuse (delta)/sent Nb/first frame/
@@ -89,7 +89,7 @@ pub fn wsShouldFallback(consecutive_failures: u8) bool {
 
 /// Transport selector: try ws for eligible codex turns, else persistent SSE.
 /// The first WS transport/handshake failure rebuilds full input and retries a
-/// fresh socket. The second latches SSE for the session. Esc/stall propagates.
+/// fresh socket; the second — or any 426 — latches SSE. Esc/stall propagates.
 pub fn postLive(self: *Agent, body: []const u8) ![]u8 {
     // #134/#132 test seam: force a one-shot stall/drop on a live turn so the
     // end-to-end "[response ended early: …]" path (never "[response interrupted
@@ -123,7 +123,11 @@ pub fn postLive(self: *Agent, body: []const u8) ![]u8 {
         if (e == error.CodexWsReanchor) return e;
         self.closeCodexWs();
         self.ws_transport_failures +|= 1;
-        const fallback = wsShouldFallback(self.ws_transport_failures);
+        // (#427) A 426 is the server answering authoritatively — it will not
+        // upgrade this endpoint — so the ladder's free retry would only redial
+        // the same refusal. Latch now (openai/codex: FallbackToHttp).
+        const declined = e == error.UpgradeRequired;
+        const fallback = declined or wsShouldFallback(self.ws_transport_failures);
         // (#codex-ws) A delta body carries previous_response_id + only the new
         // messages, anchored to the WS session that just died — the codex HTTP
         // endpoint rejects previous_response_id outright ("Unsupported
@@ -147,7 +151,7 @@ pub fn postLive(self: *Agent, body: []const u8) ![]u8 {
             return error.CodexWsReanchor;
         }
         self.ws_off = true;
-        if (self.tracer) |tr| tr.note("ws", "transport failed twice — using persistent prewarmed SSE for this session");
+        if (self.tracer) |tr| tr.note("ws", if (declined) "426 upgrade required — using persistent prewarmed SSE for this session" else "transport failed twice — using persistent prewarmed SSE for this session");
         return self.postStream(body);
     };
     self.ws_transport_failures = 0;
@@ -510,10 +514,7 @@ pub fn postResponsesWs(self: *Agent, body: []const u8) ![]u8 {
         //     here is equivalent and lets both regimes share one watchdog arm.
         read: {
             const head_wait = reused and frames_seen == 0;
-            const budget = if (head_wait)
-                http.head_stall_ms
-            else
-                http_stall.budgetMs(http.stream_stall_ms, text_seen);
+            const budget = if (head_wait) http.head_stall_ms else http_stall.budgetMs(http.stream_stall_ms, text_seen);
             const ReadDone = union(enum) { msg: ws.Error!ws.Opcode, stall: WatchdogFired };
             var rd_buf: [2]ReadDone = undefined;
             var rsel: Io.Select(ReadDone) = .init(self.io, &rd_buf);

src/agent_ws_fallback_test.zig

@@ -0,0 +1,198 @@
+//! (#427) The WS→SSE ladder's ROUTING decisions, end to end through
+//! agent_ws.postLive on the agent_ws_mock loopback peer — which serves the
+//! refused handshake AND the SSE turn graff falls back to on one port, so a
+//! test can watch the whole ladder rather than one leg of it.
+//!
+//! ws.zig has always distinguished a 426 handshake ("this endpoint will not
+//! upgrade, ever") from a generic handshake failure, but until #427 nothing
+//! consumed error.UpgradeRequired: a 426 spent the ladder's free retry on a
+//! redial the server had already answered, and only the SECOND 426 latched.
+//! Pinned here: the fast path, the unchanged two-failure ladder beside it, and
+//! that a transport CHOICE is not a stream cut on the event stream.
+
+const std = @import("std");
+const builtin = @import("builtin");
+const Io = std.Io;
+
+const main_mod = @import("main.zig");
+const trace = @import("trace.zig");
+const agent_ws = @import("agent_ws.zig");
+const engine_sink = @import("engine_sink.zig");
+
+const mock = @import("agent_ws_mock.zig");
+const mockAgent = mock.mockAgent;
+const traced = mock.traced;
+
+fn record(ctx: *anyopaque, ev: engine_sink.Stamped) void {
+    const rec: *std.ArrayList(engine_sink.Stamped) = @ptrCast(@alignCast(ctx));
+    rec.append(std.testing.allocator, ev) catch @panic("OOM");
+}
+
+/// A ⚠ notice for the user (TuiSink) / nothing on the wire (JsonSink). Choosing
+/// a transport must emit none: no stream was cut, so there is nothing to report.
+fn sawTransportAbort(rec: []const engine_sink.Stamped) bool {
+    for (rec) |ev| if (std.meta.activeTag(ev.event) == .transport_aborted) return true;
+    return false;
+}
+
+// THE #427 regression test.
+//
+// A 426 is the server answering authoritatively, so the ladder's free retry can
+// only redial the same refusal: one full rebuild plus a fresh dial burned before
+// ws_failures_before_fallback finally latches. Reverting postLive's `declined`
+// term fails this on `dials == 1` — the second handshake is the wasted work.
+test "#427: a 426 latches SSE at once — no retry burned, and the turn is served over SSE" {
+    if (builtin.os.tag == .windows) return error.SkipZigTest;
+    const gpa = std.testing.allocator;
+    const io = std.testing.io;
+
+    const saved_ws = main_mod.g_codex_ws;
+    main_mod.g_codex_ws = true; // wsEligible: pin it, a sibling test may have cleared it
+    defer main_mod.g_codex_ws = saved_ws;
+
+    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); // registered first → torn down LAST, after the task is joined
+    var dials: std.atomic.Value(u8) = .init(0);
+    var sse: std.atomic.Value(u8) = .init(0);
+    var done: std.atomic.Value(bool) = .init(false);
+    var fut = io.async(mock.refuseUpgrade, .{
+        io, &server, @as([]const u8, "426 Upgrade Required"), @as(u8, 1), &dials, &sse, &done,
+    });
+    defer fut.await(io);
+    // LIFO: done, then this, then the join — an assertion that fails before the
+    // ladder reaches its last leg must FAIL, not hang in the mock's accept.
+    defer mock.releaseAccept(io, &server);
+    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) };
+
+    // postLive's SSE leg is a REAL request, so the fallback needs a real client.
+    var client: std.http.Client = .{ .allocator = gpa, .io = io };
+    defer client.deinit();
+    var aw: Io.Writer.Allocating = .init(gpa);
+    defer aw.deinit();
+    var rec: std.ArrayList(engine_sink.Stamped) = .empty;
+    defer rec.deinit(gpa);
+    const vt: engine_sink.VTable = .{ .emit = record, .durable = false };
+
+    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;
+    agent.client = &client;
+    agent.out = &aw.writer; // wsEligible wants a live root stream…
+    agent.sink = .{ .ctx = &rec, .vt = &vt }; // …and the recording sink keeps it off any terminal
+    defer if (agent.codex_ws) |c| {
+        c.dead = true;
+        c.deinit(gpa);
+        agent.codex_ws = null;
+    };
+
+    const body = "{\"model\":\"gpt-5\",\"input\":[]}";
+    const out = try agent_ws.postLive(&agent, body);
+    defer gpa.free(out);
+
+    // ONE dial. The retry the ladder would have spent never happened, and no
+    // error.CodexWsReanchor asked request() to rebuild the body first.
+    try std.testing.expectEqual(@as(u8, 1), dials.load(.acquire));
+    try std.testing.expect(agent.ws_off); // latched for the rest of the session
+    try std.testing.expect(agent.codex_ws == null);
+    // …and THIS attempt already came back over the other transport.
+    try std.testing.expectEqual(@as(u8, 1), sse.load(.acquire));
+    try std.testing.expect(std.mem.indexOf(u8, out, mock.delta_event) != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, mock.completed_event) != null);
+    try std.testing.expect(traced(&tw, "\"detail\":\"426 upgrade required"));
+    try std.testing.expect(!traced(&tw, "\"detail\":\"transport error — retrying one fresh WS\""));
+    try std.testing.expect(!traced(&tw, "\"detail\":\"transport failed twice"));
+    // Picking a transport is not a cut stream: no ⚠ notice, nothing on the wire.
+    try std.testing.expect(!sawTransportAbort(rec.items));
+}
+
+// The other side of the boundary, unchanged by #427: a handshake that failed
+// for any reason the server did NOT declare permanent still gets its one free
+// retry on a fresh socket, and only the second failure latches. Widening the
+// fast path to every handshake error fails this at `dials == 1` on the first
+// call, and at the missing "retrying one fresh WS" note.
+test "#427: a non-426 handshake failure keeps the ladder's free retry, then latches" {
+    if (builtin.os.tag == .windows) return error.SkipZigTest;
+    const gpa = std.testing.allocator;
+    const io = std.testing.io;
+
+    const saved_ws = main_mod.g_codex_ws;
+    main_mod.g_codex_ws = true;
+    defer main_mod.g_codex_ws = saved_ws;
+
+    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 dials: std.atomic.Value(u8) = .init(0);
+    var sse: std.atomic.Value(u8) = .init(0);
+    var done: std.atomic.Value(bool) = .init(false);
+    var fut = io.async(mock.refuseUpgrade, .{
+        io, &server, @as([]const u8, "500 Internal Server Error"), @as(u8, 2), &dials, &sse, &done,
+    });
+    defer fut.await(io);
+    // LIFO: done, then this, then the join — an assertion that fails before the
+    // ladder reaches its last leg must FAIL, not hang in the mock's accept.
+    defer mock.releaseAccept(io, &server);
+    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 client: std.http.Client = .{ .allocator = gpa, .io = io };
+    defer client.deinit();
+    var aw: Io.Writer.Allocating = .init(gpa);
+    defer aw.deinit();
+    var rec: std.ArrayList(engine_sink.Stamped) = .empty;
+    defer rec.deinit(gpa);
+    const vt: engine_sink.VTable = .{ .emit = record, .durable = false };
+
+    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;
+    agent.client = &client;
+    agent.out = &aw.writer;
+    agent.sink = .{ .ctx = &rec, .vt = &vt };
+    defer if (agent.codex_ws) |c| {
+        c.dead = true;
+        c.deinit(gpa);
+        agent.codex_ws = null;
+    };
+
+    const body = "{\"model\":\"gpt-5\",\"input\":[]}";
+
+    // Attempt 1: rebuild full input and retry a fresh WS — no latch, no SSE.
+    try std.testing.expectError(error.CodexWsReanchor, agent_ws.postLive(&agent, body));
+    try std.testing.expect(!agent.ws_off);
+    try std.testing.expectEqual(@as(u8, 1), agent.ws_transport_failures);
+    try std.testing.expectEqual(@as(u8, 1), dials.load(.acquire));
+    try std.testing.expectEqual(@as(u8, 0), sse.load(.acquire));
+    try std.testing.expect(traced(&tw, "\"detail\":\"transport error — retrying one fresh WS\""));
+
+    // Attempt 2: the fresh socket is refused too, so now it latches and serves
+    // this attempt over SSE — the two-failure ladder, exactly as before.
+    const out = try agent_ws.postLive(&agent, body);
+    defer gpa.free(out);
+    try std.testing.expect(agent.ws_off);
+    try std.testing.expectEqual(@as(u8, 2), agent.ws_transport_failures);
+    try std.testing.expectEqual(@as(u8, 2), dials.load(.acquire));
+    try std.testing.expectEqual(@as(u8, 1), sse.load(.acquire));
+    try std.testing.expect(std.mem.indexOf(u8, out, mock.completed_event) != null);
+    try std.testing.expect(traced(&tw, "\"detail\":\"transport failed twice"));
+    try std.testing.expect(!traced(&tw, "\"detail\":\"426 upgrade required"));
+    try std.testing.expect(!sawTransportAbort(rec.items));
+}

src/agent_ws_mock.zig

@@ -1,7 +1,8 @@
 //! (#401) The loopback WebSocket peer the codex-WS transport tests drive, and
 //! the root-shaped Agent pointed at it. Harness only — no tests live here; they
-//! live in agent_ws_stall_test.zig (the budgets and the guards) and
-//! agent_ws_reuse_test.zig (reuse vs fresh connect). Split out because both
+//! live in agent_ws_stall_test.zig (the budgets and the guards),
+//! agent_ws_reuse_test.zig (reuse vs fresh connect) and
+//! agent_ws_fallback_test.zig (#427, the WS→SSE ladder). Split out because both
 //! files would otherwise carry a copy, and a second copy of a mock server is
 //! exactly how two suites start testing different things.
 //!
@@ -206,3 +207,80 @@ pub fn closeObserver(io: Io, server: *std.Io.net.Server, seen: *std.atomic.Value
     seen.store(if ((h[0] & 0x0f) == 0x8) saw_close_frame else saw_nothing, .release);
     Mock.idle(io, done);
 }
+
+/// (#427) The SSE turn the fallback POST is answered with: the same two events
+/// the WS modes frame, as `data:` lines, so a test can assert the turn finished
+/// over the other transport with the same needles.
+pub const sse_body = "data: " ++ delta_event ++ "\n\n" ++ "data: " ++ completed_event ++ "\n\n";
+
+/// (#427) The whole WS→SSE ladder over ONE loopback port. The first `refusals`
+/// connections are WebSocket handshakes answered with `status` — any non-101
+/// status line, i.e. "426 Upgrade Required" (authoritative: never retry) or
+/// something generic like "500 Internal Server Error" — and the connection
+/// after them is the SSE POST graff falls back to. `dials` counts refused
+/// handshakes and `sse` served fallback turns: together they prove whether the
+/// ladder burned a retry the server had already ruled out.
+pub fn refuseUpgrade(
+    io: Io,
+    server: *std.Io.net.Server,
+    status: []const u8,
+    refusals: u8,
+    dials: *std.atomic.Value(u8),
+    sse: *std.atomic.Value(u8),
+    done: *std.atomic.Value(bool),
+) void {
+    var refused: u8 = 0;
+    while (refused < refusals) : (refused += 1) {
+        if (done.load(.acquire)) return; // torn down early — see releaseAccept
+        const c = server.accept(io) catch return Mock.idle(io, done);
+        var rbuf: [8192]u8 = undefined;
+        var wbuf: [4096]u8 = undefined;
+        var sr = std.Io.net.Stream.Reader.init(c, io, &rbuf);
+        var sw = std.Io.net.Stream.Writer.init(c, io, &wbuf);
+        _ = requestHead(&sr.interface) catch {};
+        sw.interface.print("HTTP/1.1 {s}\r\nContent-Length: 0\r\n\r\n", .{status}) catch {};
+        sw.interface.flush() catch {};
+        dials.store(refused + 1, .release);
+        c.close(io); // an upgrade request carries no body, so nothing unread to RST over
+    }
+    if (done.load(.acquire)) return;
+    const c = server.accept(io) catch return Mock.idle(io, done);
+    defer c.close(io);
+    var rbuf: [8192]u8 = undefined;
+    var wbuf: [4096]u8 = undefined;
+    var sr = std.Io.net.Stream.Reader.init(c, io, &rbuf);
+    var sw = std.Io.net.Stream.Writer.init(c, io, &wbuf);
+    // The POST body must be drained: closing a socket with unread bytes in the
+    // receive queue sends an RST, which discards the reply we just wrote.
+    const body_len = requestHead(&sr.interface) catch return Mock.idle(io, done);
+    sr.interface.discardAll(body_len) catch {};
+    sw.interface.print(
+        "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {d}\r\n\r\n{s}",
+        .{ sse_body.len, sse_body },
+    ) catch return Mock.idle(io, done);
+    sw.interface.flush() catch {};
+    sse.store(1, .release);
+    Mock.idle(io, done);
+}
+
+/// (#427) Unblock a task parked in `accept` so it can observe `done` and exit.
+/// A blocked accept is not cancellable, so a test whose assertion fails BEFORE
+/// the ladder reached its last leg would otherwise hang at `fut.await` instead
+/// of reporting the failure. Register it after the await defer (LIFO: `done`,
+/// this, then the join). Same trick as http.zig's #177 poisoned-conn test.
+pub fn releaseAccept(io: Io, server: *std.Io.net.Server) void {
+    var bound = server.socket.address;
+    if (std.Io.net.IpAddress.connect(&bound, io, .{ .mode = .stream })) |s| s.close(io) else |_| {}
+}
+
+/// Consume an HTTP request head, returning its Content-Length (0 when absent).
+fn requestHead(r: *Io.Reader) !usize {
+    var body_len: usize = 0;
+    while (true) {
+        const line = try r.takeDelimiterInclusive('\n');
+        if (line.len <= 2) return body_len; // the blank line ends the head
+        const tag = "content-length:";
+        if (line.len > tag.len and std.ascii.eqlIgnoreCase(line[0..tag.len], tag))
+            body_len = std.fmt.parseInt(usize, std.mem.trim(u8, line[tag.len..], " \r\n"), 10) catch 0;
+    }
+}

src/agent_ws_test.zig

@@ -573,4 +573,5 @@ test "sendDeadlineMs (#401): head budget for a delta, transmit room for a full r
 comptime {
     _ = @import("agent_ws_stall_test.zig");
     _ = @import("agent_ws_reuse_test.zig");
+    _ = @import("agent_ws_fallback_test.zig"); // (#427) the WS→SSE ladder's routing decisions
 }