Classify provider HTTP failures in one place

Six transports turned an HTTP error status into a failure kind with the
same switch, and four wrapped their stream callbacks around the event
sink with the same adapter struct. A new status had to be added in six
places, and a missed copy would misclassify retries for one provider.

FailureKind.fromHttpStatus and the EventSink callback adapters now live
in stream_provider.zig, next to the types they produce, and every
transport uses them. Each status maps to the same kind as before and
each adapter emits the same event. New tests pin both.

src/builtins/gateway.zig

@@ -594,8 +594,8 @@ fn streamAgentCompletion(
         .content_capture_limit = request.content_capture_limit,
         .delivery = request.delivery,
         .admission = request.admission,
-        .on_reasoning_chunk = EventBridge.reasoning,
-        .on_tool_input_chunk = EventBridge.toolInput,
+        .on_reasoning_chunk = agent_stream_provider_contract.EventSink.onReasoningChunk,
+        .on_tool_input_chunk = agent_stream_provider_contract.EventSink.onToolInputChunk,
         .provider_attempt_owner = switch (request.provider_attempt_owner) {
             .transport => .transport,
             .agent => .agent,
@@ -607,8 +607,8 @@ fn streamAgentCompletion(
             alloc,
             stream_request,
             &events,
-            EventBridge.content,
-            EventBridge.toolStart,
+            agent_stream_provider_contract.EventSink.onContentChunk,
+            agent_stream_provider_contract.EventSink.onToolStart,
             deadline,
             request.cancel_flag,
         )
@@ -617,8 +617,8 @@ fn streamAgentCompletion(
             alloc,
             stream_request,
             &events,
-            EventBridge.content,
-            EventBridge.toolStart,
+            agent_stream_provider_contract.EventSink.onContentChunk,
+            agent_stream_provider_contract.EventSink.onToolStart,
             request.cancel_flag,
         )) catch |err| {
         request.attempt_evidence.network_failure = gateway_client.networkFailureEvidence(
@@ -647,32 +647,10 @@ fn streamAgentCompletion(
     } };
 }
 
-const EventBridge = struct {
-    fn sink(raw: *anyopaque) *agent_stream_provider_contract.EventSink {
-        return @ptrCast(@alignCast(raw));
-    }
-
-    fn content(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .content_delta = chunk });
-    }
-
-    fn reasoning(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .reasoning_delta = chunk });
-    }
-
-    fn toolInput(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .tool_input_delta = chunk });
-    }
-
-    fn toolStart(raw: *anyopaque, id: []const u8, name: []const u8, label: ?[]const u8, arguments_json: ?[]const u8) void {
-        sink(raw).emit(.{ .tool_started = .{ .id = id, .name = name, .label = label, .arguments_json = arguments_json } });
-    }
-};
-
 /// Gateway reports the last provider it tried, so a provider's size rejection
 /// can arrive as a 400 from a later fallback. Classify it as too large.
 fn gatewayFailureKind(alloc: Allocator, status: std.http.Status, err_body: ?[]const u8) agent_stream_provider_contract.FailureKind {
-    const kind = failureKind(status);
+    const kind = agent_stream_provider_contract.FailureKind.fromHttpStatus(status);
     if (kind != .invalid_request) return kind;
     const body = err_body orelse return kind;
     return if (vercel_failure_diagnostics.providerRejectedRequestSize(alloc, body)) .request_too_large else kind;
@@ -688,21 +666,6 @@ test "Gateway 400 with a provider 413 attempt is a size failure" {
     try std.testing.expectEqual(agent_stream_provider_contract.FailureKind.rate_limited, gatewayFailureKind(alloc, .too_many_requests, fallback_body));
 }
 
-fn failureKind(status: std.http.Status) agent_stream_provider_contract.FailureKind {
-    return switch (status) {
-        .bad_request => .invalid_request,
-        .unauthorized => .unauthorized,
-        .forbidden => .forbidden,
-        .payload_too_large => .request_too_large,
-        .too_many_requests => .rate_limited,
-        .internal_server_error => .server_error,
-        .bad_gateway => .bad_gateway,
-        .service_unavailable => .unavailable,
-        .gateway_timeout => .gateway_timeout,
-        else => .provider_error,
-    };
-}
-
 fn fetchCredits(
     _: ?*anyopaque,
     alloc: Allocator,

src/core/agent/runtime/tests/support.zig

@@ -347,7 +347,7 @@ pub const FakeGateway = struct {
             errdefer if (failure_schema) |value| alloc.free(value);
             const failure_request_shape = if (completion.failure_request_shape) |value| try alloc.dupe(u8, value) else null;
             return .{ .failed = .{
-                .kind = failureKind(completion.status),
+                .kind = .fromHttpStatus(completion.status),
                 .detail = err_body,
                 .retry_after_seconds = completion.retry_after_seconds,
                 .diagnostics = .{
@@ -410,21 +410,6 @@ pub const FakeGateway = struct {
             .delivery = request.delivery.load(),
         };
     }
-
-    fn failureKind(status: std.http.Status) agent_stream_provider.FailureKind {
-        return switch (status) {
-            .bad_request => .invalid_request,
-            .unauthorized => .unauthorized,
-            .forbidden => .forbidden,
-            .payload_too_large => .request_too_large,
-            .too_many_requests => .rate_limited,
-            .internal_server_error => .server_error,
-            .bad_gateway => .bad_gateway,
-            .service_unavailable => .unavailable,
-            .gateway_timeout => .gateway_timeout,
-            else => .provider_error,
-        };
-    }
 };
 
 pub const ModelCapabilityOverride = struct {

src/core/agent/stream_provider.zig

@@ -43,6 +43,35 @@ pub const EventSink = struct {
     pub fn emit(self: EventSink, event: Event) void {
         self.emit_fn(self.context, event);
     }
+
+    // `StreamCallback` and `ToolStartCallback` adapters for transport parsers
+    // whose callback context is a `*EventSink`.
+
+    pub fn onContentChunk(context: *anyopaque, chunk: []const u8) void {
+        fromContext(context).emit(.{ .content_delta = chunk });
+    }
+
+    pub fn onReasoningChunk(context: *anyopaque, chunk: []const u8) void {
+        fromContext(context).emit(.{ .reasoning_delta = chunk });
+    }
+
+    pub fn onToolInputChunk(context: *anyopaque, chunk: []const u8) void {
+        fromContext(context).emit(.{ .tool_input_delta = chunk });
+    }
+
+    pub fn onToolStart(
+        context: *anyopaque,
+        id: []const u8,
+        name: []const u8,
+        label: ?[]const u8,
+        arguments_json: ?[]const u8,
+    ) void {
+        fromContext(context).emit(.{ .tool_started = .{ .id = id, .name = name, .label = label, .arguments_json = arguments_json } });
+    }
+
+    fn fromContext(context: *anyopaque) *EventSink {
+        return @ptrCast(@alignCast(context));
+    }
 };
 
 pub const Admission = struct {
@@ -284,6 +313,22 @@ pub const FailureKind = enum {
     unavailable,
     gateway_timeout,
     provider_error,
+
+    /// Classifies a provider's HTTP error status.
+    pub fn fromHttpStatus(status: std.http.Status) FailureKind {
+        return switch (status) {
+            .bad_request => .invalid_request,
+            .unauthorized => .unauthorized,
+            .forbidden => .forbidden,
+            .payload_too_large => .request_too_large,
+            .too_many_requests => .rate_limited,
+            .internal_server_error => .server_error,
+            .bad_gateway => .bad_gateway,
+            .service_unavailable => .unavailable,
+            .gateway_timeout => .gateway_timeout,
+            else => .provider_error,
+        };
+    }
 };
 
 pub const FailureDiagnostics = struct {
@@ -587,6 +632,63 @@ test "stream provider accepts one typed request and emits ordered neutral events
     try std.testing.expectEqualStrings("done", result.completed.completion.content.?);
 }
 
+test "provider HTTP statuses classify into failure kinds" {
+    const cases = [_]struct { std.http.Status, FailureKind }{
+        .{ .bad_request, .invalid_request },
+        .{ .unauthorized, .unauthorized },
+        .{ .forbidden, .forbidden },
+        .{ .payload_too_large, .request_too_large },
+        .{ .too_many_requests, .rate_limited },
+        .{ .internal_server_error, .server_error },
+        .{ .bad_gateway, .bad_gateway },
+        .{ .service_unavailable, .unavailable },
+        .{ .gateway_timeout, .gateway_timeout },
+        .{ .not_found, .provider_error },
+        .{ .conflict, .provider_error },
+        .{ .ok, .provider_error },
+    };
+    for (cases) |case| {
+        try std.testing.expectEqual(case[1], FailureKind.fromHttpStatus(case[0]));
+    }
+}
+
+test "event sink callback adapters emit the matching events" {
+    const Capture = struct {
+        last: ?Event = null,
+
+        fn emit(raw: *anyopaque, event: Event) void {
+            const self: *@This() = @ptrCast(@alignCast(raw));
+            self.last = event;
+        }
+    };
+    var capture: Capture = .{};
+    var sink: EventSink = .{ .context = &capture, .emit_fn = Capture.emit };
+
+    const on_content: StreamCallback = EventSink.onContentChunk;
+    on_content(&sink, "text");
+    try std.testing.expectEqualStrings("text", capture.last.?.content_delta);
+
+    const on_reasoning: StreamCallback = EventSink.onReasoningChunk;
+    on_reasoning(&sink, "thought");
+    try std.testing.expectEqualStrings("thought", capture.last.?.reasoning_delta);
+
+    const on_tool_input: StreamCallback = EventSink.onToolInputChunk;
+    on_tool_input(&sink, "{\"a\"");
+    try std.testing.expectEqualStrings("{\"a\"", capture.last.?.tool_input_delta);
+
+    const on_tool_start: ToolStartCallback = EventSink.onToolStart;
+    on_tool_start(&sink, "call_1", "read_file", "Read", "{}");
+    const started = capture.last.?.tool_started;
+    try std.testing.expectEqualStrings("call_1", started.id);
+    try std.testing.expectEqualStrings("read_file", started.name);
+    try std.testing.expectEqualStrings("Read", started.label.?);
+    try std.testing.expectEqualStrings("{}", started.arguments_json.?);
+
+    on_tool_start(&sink, "call_2", "glob_files", null, null);
+    try std.testing.expect(capture.last.?.tool_started.label == null);
+    try std.testing.expect(capture.last.?.tool_started.arguments_json == null);
+}
+
 test "stream provider rejects system messages in conversation before delegation" {
     const Fake = struct {
         calls: usize = 0,

src/gateway/chat_completions.zig

@@ -202,7 +202,7 @@ fn post(alloc: Allocator, definition: *const definitions.Definition, request: st
             detail = redacted;
         }
 
-        return .{ .failed = .{ .kind = failure_kind(response.head.status), .detail = detail, .retry_after_seconds = retry_after, .ownership = .owned } };
+        return .{ .failed = .{ .kind = .fromHttpStatus(response.head.status), .detail = detail, .retry_after_seconds = retry_after, .ownership = .owned } };
     }
     var result = consume(alloc, reader, request, watch.progress()) catch |err| return watch.failure(err);
     if (watch.finish()) |err| {
@@ -212,21 +212,6 @@ fn post(alloc: Allocator, definition: *const definitions.Definition, request: st
     return result;
 }
 
-pub fn failure_kind(status: std.http.Status) streams.FailureKind {
-    return switch (status) {
-        .bad_request => .invalid_request,
-        .unauthorized => .unauthorized,
-        .forbidden => .forbidden,
-        .payload_too_large => .request_too_large,
-        .too_many_requests => .rate_limited,
-        .internal_server_error => .server_error,
-        .bad_gateway => .bad_gateway,
-        .service_unavailable => .unavailable,
-        .gateway_timeout => .gateway_timeout,
-        else => .provider_error,
-    };
-}
-
 /// Parses a successful streamed reply from `reader`, telling `progress` about
 /// each event when it is set.
 pub fn consume(alloc: Allocator, reader: *std.Io.Reader, request: streams.ModelRequest, progress: ?sse.Progress) !streams.Result {

src/gateway/host_chat_completions.zig

@@ -82,7 +82,7 @@ fn stream(raw: ?*anyopaque, alloc: Allocator, request: stream_provider.ModelRequ
     const opened = try host_stream.openPost(alloc, transport, request, endpoint, &headers, payload);
     defer transport.close(opened.handle);
     if (opened.status != .ok) return .{ .failed = .{
-        .kind = chat_completions.failure_kind(opened.status),
+        .kind = .fromHttpStatus(opened.status),
         .detail = try host_stream.readErrorBody(alloc, transport, opened.handle, request),
         .ownership = .owned,
     } };

src/gateway/host_openai_codex.zig

@@ -110,7 +110,7 @@ fn stream(raw: ?*anyopaque, alloc: Allocator, request: stream_provider.ModelRequ
     const opened = try host_stream.openPost(alloc, transport, request, endpoint, headers_buf[0..count], payload);
     defer transport.close(opened.handle);
     if (opened.status != .ok) return .{ .failed = .{
-        .kind = openai_codex.failureKind(opened.status),
+        .kind = .fromHttpStatus(opened.status),
         .detail = try host_stream.readErrorBody(alloc, transport, opened.handle, request),
         .ownership = .owned,
     } };

src/gateway/host_stream_provider.zig

@@ -168,7 +168,7 @@ fn stream(raw: ?*anyopaque, alloc: Allocator, request: stream_provider.ModelRequ
         const body = try readErrorBody(alloc, transport, opened.handle, request);
         // Gateway reports the last provider it tried, so a provider's size
         // rejection can arrive as a 400 from a later fallback.
-        const kind = failureKind(opened.status);
+        const kind = stream_provider.FailureKind.fromHttpStatus(opened.status);
         const size_rejected = kind == .invalid_request and vercel_failure_diagnostics.providerRejectedRequestSize(alloc, body);
         return .{ .failed = .{
             .kind = if (size_rejected) .request_too_large else kind,
@@ -190,9 +190,9 @@ fn stream(raw: ?*anyopaque, alloc: Allocator, request: stream_provider.ModelRequ
         alloc,
         &reader.interface,
         &events,
-        EventBridge.content,
-        EventBridge.toolStart,
-        EventBridge.reasoning,
+        stream_provider.EventSink.onContentChunk,
+        stream_provider.EventSink.onToolStart,
+        stream_provider.EventSink.onReasoningChunk,
         request.cancel_flag,
         request.content_capture_limit,
     ) catch |err| switch (err) {
@@ -456,39 +456,6 @@ fn buildRequest(
     return context.build_fn(alloc, request);
 }
 
-const EventBridge = struct {
-    fn sink(raw: *anyopaque) *stream_provider.EventSink {
-        return @ptrCast(@alignCast(raw));
-    }
-
-    fn content(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .content_delta = chunk });
-    }
-
-    fn reasoning(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .reasoning_delta = chunk });
-    }
-
-    fn toolStart(raw: *anyopaque, id: []const u8, name: []const u8, label: ?[]const u8, arguments_json: ?[]const u8) void {
-        sink(raw).emit(.{ .tool_started = .{ .id = id, .name = name, .label = label, .arguments_json = arguments_json } });
-    }
-};
-
-fn failureKind(status: std.http.Status) stream_provider.FailureKind {
-    return switch (status) {
-        .bad_request => .invalid_request,
-        .unauthorized => .unauthorized,
-        .forbidden => .forbidden,
-        .payload_too_large => .request_too_large,
-        .too_many_requests => .rate_limited,
-        .internal_server_error => .server_error,
-        .bad_gateway => .bad_gateway,
-        .service_unavailable => .unavailable,
-        .gateway_timeout => .gateway_timeout,
-        else => .provider_error,
-    };
-}
-
 fn pulse(value: ?stream_provider.CooperativePulse) !void {
     if (value) |callback| try callback.pulse();
 }

src/gateway/host_xai_grok.zig

@@ -178,7 +178,7 @@ fn stream(raw: ?*anyopaque, alloc: Allocator, request: stream_provider.ModelRequ
     const opened = try host_stream.openPost(alloc, transport, request, endpoint, headers_buf[0 .. extra.len + 1], payload);
     defer transport.close(opened.handle);
     if (opened.status != .ok) return .{ .failed = .{
-        .kind = xai_grok.failureKind(opened.status),
+        .kind = .fromHttpStatus(opened.status),
         .detail = try host_stream.readErrorBody(alloc, transport, opened.handle, request),
         .ownership = .owned,
     } };

src/gateway/openai_codex.zig

@@ -332,7 +332,7 @@ pub fn streamPrepared(
             else => return err,
         };
         return .{ .failed = .{
-            .kind = failureKind(response.head.status),
+            .kind = .fromHttpStatus(response.head.status),
             .detail = body,
             .ownership = .owned,
         } };
@@ -360,10 +360,10 @@ pub fn completeFromStream(
         alloc,
         reader,
         &events,
-        EventBridge.content,
-        EventBridge.toolStart,
-        EventBridge.reasoning,
-        EventBridge.toolInput,
+        stream_provider.EventSink.onContentChunk,
+        stream_provider.EventSink.onToolStart,
+        stream_provider.EventSink.onReasoningChunk,
+        stream_provider.EventSink.onToolInputChunk,
         request.cancel_flag,
         request.content_capture_limit,
         .{},
@@ -382,43 +382,6 @@ pub fn completeFromStream(
     } };
 }
 
-const EventBridge = struct {
-    fn sink(raw: *anyopaque) *stream_provider.EventSink {
-        return @ptrCast(@alignCast(raw));
-    }
-
-    fn content(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .content_delta = chunk });
-    }
-
-    fn reasoning(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .reasoning_delta = chunk });
-    }
-
-    fn toolInput(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .tool_input_delta = chunk });
-    }
-
-    fn toolStart(raw: *anyopaque, id: []const u8, name: []const u8, label: ?[]const u8, arguments_json: ?[]const u8) void {
-        sink(raw).emit(.{ .tool_started = .{ .id = id, .name = name, .label = label, .arguments_json = arguments_json } });
-    }
-};
-
-pub fn failureKind(status: std.http.Status) stream_provider.FailureKind {
-    return switch (status) {
-        .bad_request => .invalid_request,
-        .unauthorized => .unauthorized,
-        .forbidden => .forbidden,
-        .payload_too_large => .request_too_large,
-        .too_many_requests => .rate_limited,
-        .internal_server_error => .server_error,
-        .bad_gateway => .bad_gateway,
-        .service_unavailable => .unavailable,
-        .gateway_timeout => .gateway_timeout,
-        else => .provider_error,
-    };
-}
-
 fn consumeSse(
     alloc: Allocator,
     reader: *std.Io.Reader,

src/gateway/xai_grok.zig

@@ -341,7 +341,7 @@ pub fn streamPrepared(
             break :body try alloc.dupe(u8, "xAI Grok error response exceeded the local limit");
         } else bounded_body;
         return .{ .failed = .{
-            .kind = failureKind(response.head.status),
+            .kind = .fromHttpStatus(response.head.status),
             .detail = body,
             .ownership = .owned,
         } };
@@ -369,10 +369,10 @@ pub fn completeFromStream(
         alloc,
         reader,
         &events,
-        EventBridge.content,
-        EventBridge.toolStart,
-        EventBridge.reasoning,
-        EventBridge.toolInput,
+        stream_provider.EventSink.onContentChunk,
+        stream_provider.EventSink.onToolStart,
+        stream_provider.EventSink.onReasoningChunk,
+        stream_provider.EventSink.onToolInputChunk,
         request.cancel_flag,
         request.content_capture_limit,
         progress,
@@ -390,43 +390,6 @@ pub fn completeFromStream(
     } };
 }
 
-const EventBridge = struct {
-    fn sink(raw: *anyopaque) *stream_provider.EventSink {
-        return @ptrCast(@alignCast(raw));
-    }
-
-    fn content(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .content_delta = chunk });
-    }
-
-    fn reasoning(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .reasoning_delta = chunk });
-    }
-
-    fn toolInput(raw: *anyopaque, chunk: []const u8) void {
-        sink(raw).emit(.{ .tool_input_delta = chunk });
-    }
-
-    fn toolStart(raw: *anyopaque, id: []const u8, name: []const u8, label: ?[]const u8, arguments_json: ?[]const u8) void {
-        sink(raw).emit(.{ .tool_started = .{ .id = id, .name = name, .label = label, .arguments_json = arguments_json } });
-    }
-};
-
-pub fn failureKind(status: std.http.Status) stream_provider.FailureKind {
-    return switch (status) {
-        .bad_request => .invalid_request,
-        .unauthorized => .unauthorized,
-        .forbidden => .forbidden,
-        .payload_too_large => .request_too_large,
-        .too_many_requests => .rate_limited,
-        .internal_server_error => .server_error,
-        .bad_gateway => .bad_gateway,
-        .service_unavailable => .unavailable,
-        .gateway_timeout => .gateway_timeout,
-        else => .provider_error,
-    };
-}
-
 fn consumeSse(
     alloc: Allocator,
     reader: *std.Io.Reader,