diff options
Diffstat (limited to 'libpanto/src/agent.zig')
| -rw-r--r-- | libpanto/src/agent.zig | 199 |
1 files changed, 195 insertions, 4 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; |