diff options
Diffstat (limited to 'libpanto/src')
| -rw-r--r-- | libpanto/src/agent.zig | 199 | ||||
| -rw-r--r-- | libpanto/src/provider.zig | 14 | ||||
| -rw-r--r-- | libpanto/src/provider_anthropic_messages.zig | 67 | ||||
| -rw-r--r-- | libpanto/src/provider_openai_chat.zig | 31 |
4 files changed, 305 insertions, 6 deletions
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index 24bc2cf..1da0261 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -1122,6 +1122,19 @@ pub const Stream = struct { /// `provider_retry` notices pushed before the failing attempt) have been /// drained. `next()` yields the queue first, then this error. _pending_error: ?anyerror = null, + /// Count of mid-stream retries attempted for the CURRENT response. Unlike + /// `openWithRetries` (which loops internally with a local counter), a + /// mid-stream failure unwinds back to `.turn_start` and re-opens, so the + /// attempt count must live on the stream to survive across re-opens. + /// Reset to 0 each time a response completes. A persistently-broken + /// stream would otherwise re-open forever without ever exhausting + /// `retry.max_attempts`. + _stream_retries: usize = 0, + /// Owned copy of the provider's diagnostic message for the in-flight + /// mid-stream retry notice. Borrowed by the queued `provider_retry` + /// event, so it must outlive that event: freed on the next mid-stream + /// failure and on `deinit`. + _retry_message: ?[]u8 = null, pub const State = enum { /// Open the next provider response (with retries). @@ -1150,6 +1163,7 @@ pub const Stream = struct { // dropping the stream mid-turn after some messages were committed. self.persistTail(); if (self._response) |ps| ps.deinit(); + if (self._retry_message) |m| self._agent._allocator.free(m); self._queue.deinit(); self._agent._allocator.destroy(self); } @@ -1225,21 +1239,44 @@ pub const Stream = struct { // completion), so retry by re-opening from scratch, // exactly as the prior push loop replayed a failed // streamStep. Non-retryable errors propagate. + // + // Capture the provider's diagnostic message (e.g. an + // Anthropic `overloaded_error`) BEFORE deinit frees + // it, dup'ing it so it outlives the response. Owned + // here and freed after the retry notice is queued. + if (self._retry_message) |old| { + self._agent._allocator.free(old); + self._retry_message = null; + } + if (ps.lastError()) |m| { + self._retry_message = self._agent._allocator.dupe(u8, m) catch null; + } ps.deinit(); self._response = null; if (!provider_mod.isRetryableProviderError(err)) { self.state = .failed; return err; } + // Out of attempts: hard-fail with the last error + // instead of re-opening forever. `_stream_retries` + // counts mid-stream retries already made for this + // response; the initial attempt is the open itself, + // so the Nth retry is attempt N+1. + const cfg = self._agent._config; + self._stream_retries += 1; + if (self._stream_retries >= cfg.retry.max_attempts) { + self.state = .failed; + return err; + } self.state = .turn_start; // Emit a retry notice so consumers see the stall. - const cfg = self._agent._config; - const delay_ms = self._agent.backoffDelayMs(cfg.retry, 1, null); + const delay_ms = self._agent.backoffDelayMs(cfg.retry, self._stream_retries, null); self._queue.push(.{ .provider_retry = .{ - .attempt = 1, + .attempt = self._stream_retries, .max_attempts = cfg.retry.max_attempts, .delay_ms = delay_ms, .err = err, + .message = self._retry_message, } }) catch |e| { self.state = .failed; return e; @@ -1258,6 +1295,9 @@ pub const Stream = struct { if (status == .response_complete) { ps.deinit(); self._response = null; + // A response completed: reset the mid-stream retry + // budget so a later turn starts fresh. + self._stream_retries = 0; self.state = .after_response; } // else `.more`: loop and pump again. @@ -1521,6 +1561,12 @@ const StubProvider = struct { /// transient-retry path. scripted_errors: []const ScriptedError = &.{}, error_idx: usize = 0, + /// A queue of MID-STREAM errors. Unlike `scripted_errors` (which fail at + /// open time), each entry here lets the open succeed but makes the + /// returned response fail on its first `produce`, exercising the + /// `.streaming` retry path. Consumed in order, one per open. + stream_errors: []const StreamFailure = &.{}, + stream_error_idx: usize = 0, /// Count of opens observed (failed + succeeded). Lets tests assert the /// exact number of attempts. calls_made: usize = 0, @@ -1531,6 +1577,12 @@ const StubProvider = struct { retry_after_ms: ?u64 = null, }; + const StreamFailure = struct { + err: anyerror, + /// Optional provider diagnostic surfaced via `ProviderStream.lastError`. + message: ?[]const u8 = null, + }; + const ScriptedTurn = struct { blocks: []const TestBlock, /// Optional provider usage to stamp on the committed assistant @@ -1566,6 +1618,12 @@ const StubResponse = struct { conv: *conversation.Conversation, turn: StubProvider.ScriptedTurn, done: bool = false, + /// If set, the first `produce` returns this error instead of committing + /// (simulates a mid-stream provider failure after a successful open). + produce_error: ?anyerror = null, + /// Optional provider diagnostic surfaced via `lastError` after the + /// mid-stream failure (borrowed; static test strings). + error_message: ?[]const u8 = null, fn create( allocator: Allocator, @@ -1577,13 +1635,37 @@ const StubResponse = struct { return .{ .ptr = self, .vtable = &vtable }; } + fn createFailing( + allocator: Allocator, + conv: *conversation.Conversation, + err: anyerror, + message: ?[]const u8, + ) !provider_mod.ProviderStream { + const self = try allocator.create(StubResponse); + self.* = .{ + .allocator = allocator, + .conv = conv, + .turn = .{ .blocks = &.{} }, + .produce_error = err, + .error_message = message, + }; + return .{ .ptr = self, .vtable = &vtable }; + } + const vtable: provider_mod.ProviderStream.VTable = .{ .produce = produceVT, .deinit = deinitVT, + .last_error = lastErrorVT, }; + fn lastErrorVT(ptr: *anyopaque) ?[]const u8 { + const self: *StubResponse = @ptrCast(@alignCast(ptr)); + return self.error_message; + } + fn produceVT(ptr: *anyopaque, out: *stream_mod.EventQueue) anyerror!provider_mod.ProviderStream.ProduceStatus { const self: *StubResponse = @ptrCast(@alignCast(ptr)); + if (self.produce_error) |err| return err; if (self.done) return .response_complete; self.done = true; @@ -1647,6 +1729,11 @@ fn stubOpenStream( } return e.err; } + if (self.stream_error_idx < self.stream_errors.len) { + const f = self.stream_errors[self.stream_error_idx]; + self.stream_error_idx += 1; + return StubResponse.createFailing(allocator, conv, f.err, f.message); + } if (self.overflow_calls > 0) { self.overflow_calls -= 1; return error.ContextOverflow; @@ -2984,7 +3071,18 @@ const RetryRecorder = struct { infos: std.ArrayList(provider_mod.ProviderRetryInfo) = .empty, allocator: Allocator, + /// Append a copy of `info`, dup'ing its borrowed `message` so assertions + /// stay valid after the source `Stream` is deinit'd. + fn append(self: *RetryRecorder, info: provider_mod.ProviderRetryInfo) !void { + var owned = info; + if (info.message) |m| owned.message = try self.allocator.dupe(u8, m); + try self.infos.append(self.allocator, owned); + } + fn deinit(self: *RetryRecorder) void { + for (self.infos.items) |info| { + if (info.message) |m| self.allocator.free(m); + } self.infos.deinit(self.allocator); } }; @@ -2996,7 +3094,7 @@ fn drainTurnRecording(agent: *Agent, text: []const u8, rr: *RetryRecorder) !void var s = try agent.run(.{ .text = text }); defer s.deinit(); while (try s.next()) |ev| { - if (ev == .provider_retry) try rr.infos.append(rr.allocator, ev.provider_retry); + if (ev == .provider_retry) try rr.append(ev.provider_retry); } } @@ -3172,6 +3270,99 @@ test "runStep: retries exhaust and hard-fail after max_attempts" { try testing.expectEqual(@as(usize, 3), rr.infos.items.len); } +test "runStep: mid-stream failure retries then succeeds" { + const allocator = testing.allocator; + + // The open succeeds every time; the FIRST two responses fail during + // `produce` (mid-stream), the third is a normal scripted turn. + const stream_errs = [_]StubProvider.StreamFailure{ + .{ .err = error.ProviderOverloaded, .message = "overloaded_error: Overloaded" }, + .{ .err = error.ProviderOverloaded, .message = "overloaded_error: Overloaded" }, + }; + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "finally" }} }, + }; + var stub = StubProvider{ + .allocator = allocator, + .scripted = &scripted, + .stream_errors = &stream_errs, + }; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var h = TestHarness.init(allocator); + defer h.deinit(); + fastRetryHarness(&h); + var ns = null_store_mod.NullStore.init(allocator); + const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); + defer agent.deinit(); + h.seedInto(agent); + agent._open_stream_fn = stub.install(); + + const conv = &agent.conversation; + + var rr = RetryRecorder{ .allocator = allocator }; + defer rr.deinit(); + try drainTurnRecording(agent, "hi", &rr); + + // Three opens: two that fail mid-stream + one that succeeds. + try testing.expectEqual(@as(usize, 3), stub.calls_made); + // No duplicate assistant messages: the mid-stream failures committed + // nothing, so we end with user + a single assistant. + try testing.expectEqual(@as(usize, 2), conv.messages.items.len); + try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[1].role); + // Two retry notifications, with a monotonically increasing attempt count + // (the bug was that this stayed pinned at 1 forever). + try testing.expectEqual(@as(usize, 2), rr.infos.items.len); + try testing.expectEqual(@as(usize, 1), rr.infos.items[0].attempt); + try testing.expectEqual(@as(usize, 2), rr.infos.items[1].attempt); + // The provider's diagnostic is surfaced on the retry notice (was the + // whole point: the user should see "overloaded", not a bare error name). + try testing.expectEqualStrings("overloaded_error: Overloaded", rr.infos.items[0].message.?); +} + +test "runStep: mid-stream failures exhaust and hard-fail after max_attempts" { + const allocator = testing.allocator; + + // Every open succeeds but every response fails mid-stream. Without the + // attempt counter this would re-open forever; it must hard-fail after + // `max_attempts` opens. + const stream_errs = [_]StubProvider.StreamFailure{ + .{ .err = error.ProviderStreamMalformed }, + .{ .err = error.ProviderStreamMalformed }, + .{ .err = error.ProviderStreamMalformed }, + .{ .err = error.ProviderStreamMalformed }, + }; + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "unreachable" }} }, + }; + var stub = StubProvider{ + .allocator = allocator, + .scripted = &scripted, + .stream_errors = &stream_errs, + }; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var h = TestHarness.init(allocator); + defer h.deinit(); + fastRetryHarness(&h); + var ns = null_store_mod.NullStore.init(allocator); + const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); + defer agent.deinit(); + h.seedInto(agent); + agent._open_stream_fn = stub.install(); + + var rr = RetryRecorder{ .allocator = allocator }; + defer rr.deinit(); + try testing.expectError(error.ProviderStreamMalformed, drainTurnRecording(agent, "hi", &rr)); + + // 4 opens total (max_attempts), 3 retry notifications — then hard-fail + // instead of looping endlessly. + try testing.expectEqual(@as(usize, 4), stub.calls_made); + try testing.expectEqual(@as(usize, 3), rr.infos.items.len); +} + test "runStep: Retry-After is honored and reported" { const allocator = testing.allocator; diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig index 8ea74f3..e83012d 100644 --- a/libpanto/src/provider.zig +++ b/libpanto/src/provider.zig @@ -102,6 +102,7 @@ pub fn isRetryableProviderError(err: anyerror) bool { error.ProviderServerError, error.ProviderTransport, error.ProviderStreamMalformed, + error.ProviderOverloaded, => true, else => false, }; @@ -217,6 +218,12 @@ pub const ProviderStream = struct { produce: *const fn (*anyopaque, *EventQueue) anyerror!ProduceStatus, /// Free the response and any owned state. deinit: *const fn (*anyopaque) void, + /// Optional: after a failed `produce`, return the provider's + /// diagnostic message for the failure (e.g. an Anthropic + /// `overloaded_error` message), borrowed for the lifetime of the + /// response. Null when the provider has nothing to add beyond the + /// classified error name. + last_error: ?*const fn (*anyopaque) ?[]const u8 = null, }; /// Pump the response, appending decoded events to `out`. Errors are @@ -228,6 +235,13 @@ pub const ProviderStream = struct { pub fn deinit(self: ProviderStream) void { self.vtable.deinit(self.ptr); } + + /// The provider's diagnostic message for the most recent `produce` + /// failure, if any. Borrowed for the lifetime of the response. + pub fn lastError(self: ProviderStream) ?[]const u8 { + const f = self.vtable.last_error orelse return null; + return f(self.ptr); + } }; /// Open one streaming provider turn against the active config snapshot, diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig index 9740c0e..be4268e 100644 --- a/libpanto/src/provider_anthropic_messages.zig +++ b/libpanto/src/provider_anthropic_messages.zig @@ -187,6 +187,7 @@ pub const ResumableResponse = struct { const vtable: provider_mod.ProviderStream.VTable = .{ .produce = produceVT, .deinit = deinitVT, + .last_error = lastErrorVT, }; fn produceVT(ptr: *anyopaque, out: *EventQueue) anyerror!ProduceStatus { @@ -194,6 +195,11 @@ pub const ResumableResponse = struct { return self.produce(out); } + fn lastErrorVT(ptr: *anyopaque) ?[]const u8 { + const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); + return self.state.stream_error_message; + } + fn deinitVT(ptr: *anyopaque) void { const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); self.deinit(); @@ -274,6 +280,10 @@ const StreamState = struct { usage: provider_mod.Usage = .{}, usage_seen: bool = false, stop_reason: ?[]u8 = null, + /// Owned, human-readable description of a mid-stream `error` event + /// (e.g. `"overloaded_error: Overloaded"`), surfaced to the agent via + /// `ProviderStream.lastError` so the retry notice can show *why*. + stream_error_message: ?[]u8 = null, const ActiveBlock = struct { /// Index reported on the wire (Anthropic's content-array index). @@ -304,6 +314,7 @@ const StreamState = struct { for (self.blocks.items) |*b| b.deinit(self.allocator); self.blocks.deinit(self.allocator); if (self.stop_reason) |s| self.allocator.free(s); + if (self.stream_error_message) |s| self.allocator.free(s); } fn ensureStarted(self: *StreamState, out: *EventQueue) !void { @@ -435,6 +446,22 @@ const StreamState = struct { self.stop_reason = if (reason) |r| try self.allocator.dupe(u8, r) else null; } + /// Record a readable description of a mid-stream `error` event, combining + /// the error `kind` and `message` into one owned string (either may be + /// absent). Replaces any previous value. + fn setStreamErrorMessage(self: *StreamState, kind: ?[]const u8, message: ?[]const u8) !void { + if (self.stream_error_message) |old| self.allocator.free(old); + self.stream_error_message = null; + self.stream_error_message = if (kind != null and message != null) + try std.fmt.allocPrint(self.allocator, "{s}: {s}", .{ kind.?, message.? }) + else if (kind) |k| + try self.allocator.dupe(u8, k) + else if (message) |m| + try self.allocator.dupe(u8, m) + else + null; + } + /// Close the active block: append it to `blocks` and emit block_complete. fn closeBlock( self: *StreamState, @@ -576,8 +603,17 @@ fn handleEvent( if (!@import("builtin").is_test) { std.log.err("anthropic stream error: {?s}: {?s}", .{ e.kind, e.message }); } - // Mid-stream error event (e.g. `overloaded_error`) before the - // message was committed: retryable malformed-stream failure. + // Stash a readable description so the agent's retry notice can + // explain *why* the stream failed instead of only showing the + // bare error name. Owned by `state`; freed in `deinit`. + state.setStreamErrorMessage(e.kind, e.message) catch {}; + // Mid-stream error event before the message was committed. Map + // the common overload case to a dedicated retryable error so the + // UI can say "overloaded" rather than "malformed stream"; other + // kinds stay as the generic retryable malformed-stream error. + if (e.kind) |k| { + if (std.mem.eql(u8, k, "overloaded_error")) return error.ProviderOverloaded; + } return error.ProviderStreamMalformed; }, .unknown => { @@ -1073,7 +1109,34 @@ test "error event propagates as Zig error" { &state, &queue, ); + // `overloaded_error` maps to the dedicated retryable error, and the + // provider's diagnostic is stashed for the agent's retry notice. + try testing.expectError(error.ProviderOverloaded, result); + try testing.expectEqualStrings("overloaded_error: too busy", state.stream_error_message.?); +} + +test "non-overloaded error event stays malformed and stashes message" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try addUserText(&conv, "hi"); + + var queue = EventQueue.init(allocator); + defer queue.deinit(); + + var state: StreamState = .init(allocator); + defer state.deinit(); + + const result = handleEvent( + allocator, + \\{"type":"error","error":{"type":"api_error","message":"boom"}} + , + &state, + &queue, + ); try testing.expectError(error.ProviderStreamMalformed, result); + try testing.expectEqualStrings("api_error: boom", state.stream_error_message.?); } test "two streamed turns persist assistant replies in the conversation" { diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index 5152d3e..5713b5d 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -214,8 +214,14 @@ pub const ResumableResponse = struct { const vtable: provider_mod.ProviderStream.VTable = .{ .produce = produceVT, .deinit = deinitVT, + .last_error = lastErrorVT, }; + fn lastErrorVT(ptr: *anyopaque) ?[]const u8 { + const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); + return self.state.stream_error_message; + } + fn produceVT(ptr: *anyopaque, out: *EventQueue) anyerror!ProduceStatus { const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); return self.produce(out); @@ -339,6 +345,11 @@ const StreamState = struct { /// `stream_options.include_usage: true` AND the server honored it). usage: ?provider_mod.Usage = null, + /// Owned, human-readable description of a mid-stream error embedded in an + /// HTTP-200 SSE body, surfaced to the agent via `ProviderStream.lastError` + /// so the retry notice can explain *why*. + stream_error_message: ?[]u8 = null, + const ToolUseInProgress = struct { /// Block index emitted to the receiver for this tool call's /// onBlockStart / onContentDelta / onBlockComplete callbacks. @@ -385,6 +396,23 @@ const StreamState = struct { self.blocks.deinit(self.allocator); if (self.active_tool) |*tu| tu.deinit(self.allocator); self.closed_tool_indices.deinit(); + if (self.stream_error_message) |s| self.allocator.free(s); + } + + /// Record a readable description of an embedded stream error, combining + /// the error `type` and `message` into one owned string (either may be + /// absent). Replaces any previous value. + fn setStreamErrorMessage(self: *StreamState, kind: ?[]const u8, message: ?[]const u8) !void { + if (self.stream_error_message) |old| self.allocator.free(old); + self.stream_error_message = null; + self.stream_error_message = if (kind != null and message != null) + try std.fmt.allocPrint(self.allocator, "{s}: {s}", .{ kind.?, message.? }) + else if (kind) |k| + try self.allocator.dupe(u8, k) + else if (message) |m| + try self.allocator.dupe(u8, m) + else + null; } /// Close the active text/thinking block (if any) and emit @@ -683,6 +711,9 @@ fn handleEvent( d.error_type, d.error_message, }); } + // Stash a readable description so the agent's retry notice can + // explain *why* the stream failed. Owned by `state`. + state.setStreamErrorMessage(d.error_type, d.error_message) catch {}; return error.ProviderStreamMalformed; } |