diff --git a/src/agent_ws.zig b/src/agent_ws.zig index 62e5dcd0c..6ea41af3e 100644 --- a/src/agent_ws.zig +++ b/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); diff --git a/src/agent_ws_fallback_test.zig b/src/agent_ws_fallback_test.zig new file mode 100644 index 000000000..225e19d48 --- /dev/null +++ b/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)); +} diff --git a/src/agent_ws_mock.zig b/src/agent_ws_mock.zig index 04f503668..1f81e65c1 100644 --- a/src/agent_ws_mock.zig +++ b/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; + } +} diff --git a/src/agent_ws_test.zig b/src/agent_ws_test.zig index 40acff928..0bc214a22 100644 --- a/src/agent_ws_test.zig +++ b/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 }