From 3f1ace16afc7877b0bfad374cb286d4d84140960 Mon Sep 17 00:00:00 2001 From: t Date: Thu, 4 Jun 2026 09:46:49 -0600 Subject: failure retries scheme --- libpanto/src/agent.zig | 829 +++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 775 insertions(+), 54 deletions(-) (limited to 'libpanto/src/agent.zig') diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index c078c24..c7d0536 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -107,7 +107,7 @@ fn cloneBlock(alloc: Allocator, block: conversation.ContentBlock) !conversation. }, } } - break :blk .{ .ToolResult = .{ .tool_use_id = tuid, .parts = parts } }; + break :blk .{ .ToolResult = .{ .tool_use_id = tuid, .parts = parts, .is_error = b.is_error } }; }, .System => |b| .{ .System = .{ .text = try conversation.textualBlockFromSlice(alloc, b.text.items), @@ -139,6 +139,7 @@ const CompactionCapture = struct { .onBlockComplete = onBlockComplete, .onMessageComplete = onMessageComplete, .onError = onError, + .onProviderRetry = onProviderRetry, }; fn onMessageStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} fn onBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} @@ -147,6 +148,7 @@ const CompactionCapture = struct { fn onBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} fn onMessageComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} fn onError(_: *anyopaque, _: anyerror) void {} + fn onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} }; fn isValidToolInput(input: []const u8) bool { @@ -166,6 +168,43 @@ fn invalidInputResult(allocator: Allocator, input: []const u8) ![]tool_mod.Resul return tool_mod.ownedTextResult(allocator, msg); } +/// What to do with an error returned by tool dispatch. +const ToolErrorAction = enum { + /// Surface the failure to the model as an error `ToolResult`, then let + /// the agent loop continue. + tool_result, + /// Abort the whole turn and propagate to the embedder. Reserved for + /// failures that belong to the host, not the model/provider exchange. + hard_fail, +}; + +/// Decide how to handle a tool dispatch error. Only genuine host failures +/// abort the turn; everything else becomes a model-visible tool result so +/// the model can correct course (and so every `ToolUse` keeps its matching +/// `ToolResult`, which providers require). +fn classifyToolError(err: anyerror) ToolErrorAction { + return switch (err) { + error.Canceled, error.OutOfMemory => .hard_fail, + else => .tool_result, + }; +} + +/// Build an error `ResultPart` describing a failed tool call, in the +/// model-readable form the plan specifies. +fn toolErrorResult( + allocator: Allocator, + tool_name: []const u8, + err: anyerror, +) ![]tool_mod.ResultPart { + const msg = try std.fmt.allocPrint( + allocator, + "Tool execution failed for `{s}`: {s}\n" ++ + "You may fix the arguments, try a different tool, or explain the failure to the user.", + .{ tool_name, @errorName(err) }, + ); + return tool_mod.ownedTextResult(allocator, msg); +} + pub const Agent = struct { allocator: Allocator, @@ -188,6 +227,10 @@ pub const Agent = struct { /// automatic compaction occurred this turn (so it can persist the /// rewritten conversation). Reset at the top of each `runStep`. auto_compacted: bool = false, + /// PRNG state for backoff jitter. Seeded lazily on first retry. Only + /// touched from the single agent-loop thread (retries are serial), so + /// no synchronization is needed. + retry_prng: ?std.Random.DefaultPrng = null, pub fn init(allocator: Allocator, io: Io, config: *const Config) Agent { return .{ @@ -227,19 +270,7 @@ pub const Agent = struct { // Re-read the config snapshot at the top of each turn so a // mid-conversation swap takes effect here, never mid-stream. const cfg = self.config; - self.stream_fn(self.allocator, self.io, cfg, conv, receiver) catch |err| { - // Automatic compaction on context overflow: compact the - // conversation once and retry the failed request a single - // time. If the retry also overflows, surface the error. - if (err != error.ContextOverflow) return err; - if (self.auto_compacted) return err; // already retried once - const sys = self.compaction_system_prompt orelse return err; - const res = try self.compact(conv, sys, null); - if (!res.compacted) return err; // nothing to shed; give up - self.auto_compacted = true; - // Retry the same request against the compacted context. - try self.stream_fn(self.allocator, self.io, cfg, conv, receiver); - }; + try self.streamWithRetries(cfg, conv, receiver); const last = conv.messages.items[conv.messages.items.len - 1]; std.debug.assert(last.role == .assistant); @@ -262,6 +293,121 @@ pub const Agent = struct { return false; } + /// Drive one provider turn with the configured retry policy. + /// + /// Decision path for a failed attempt: + /// - `ContextOverflow`: compact once, then retry the same request a + /// single time against the compacted conversation (a one-shot path, + /// independent of the transient-retry budget). + /// - retryable provider error (rate limit, server, transport, + /// malformed stream): sleep with exponential backoff + jitter + /// (honoring `Retry-After` when present) and retry, up to + /// `retry.max_attempts` total attempts. + /// - anything else (auth, bad request, cancellation, local errors): + /// propagate immediately. + /// + /// A failed attempt never mutates the conversation (providers commit the + /// assistant message only on success), so each retry runs against the + /// same snapshot. + fn streamWithRetries( + self: *Agent, + cfg: *const Config, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + ) !void { + const policy = cfg.retry; + var attempt: usize = 1; + while (true) { + var diag: provider_mod.ProviderDiagnostic = .{}; + self.stream_fn(self.allocator, self.io, cfg, conv, receiver, &diag) catch |err| { + if (err == error.ContextOverflow) { + try self.handleContextOverflow(cfg, conv, receiver, err); + return; + } + if (!provider_mod.isRetryableProviderError(err)) return err; + // Out of attempts: hard-fail with the last error. + if (attempt >= policy.max_attempts) return err; + + const delay_ms = self.backoffDelayMs(policy, attempt, diag.retry_after_ms); + receiver.onProviderRetry(.{ + .attempt = attempt, + .max_attempts = policy.max_attempts, + .delay_ms = delay_ms, + .err = err, + .status_code = diag.status_code, + .retry_after_ms = diag.retry_after_ms, + .message = diag.message, + }); + if (delay_ms > 0) { + const ms: i64 = @intCast(@min(delay_ms, std.math.maxInt(i64))); + self.io.sleep(.fromMilliseconds(ms), .real) catch |e| return e; + } + attempt += 1; + continue; + }; + return; + } + } + + /// One-shot context-overflow recovery: compact once, retry once. Mirrors + /// the prior inline behavior, now fired from `streamWithRetries`. The + /// retry is announced through `onProviderRetry` with `compaction = true` + /// and `delay_ms = 0`. + fn handleContextOverflow( + self: *Agent, + cfg: *const Config, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + err: anyerror, + ) !void { + if (self.auto_compacted) return err; // already retried once this turn + const sys = self.compaction_system_prompt orelse return err; + const res = try self.compact(conv, sys, null); + if (!res.compacted) return err; // nothing to shed; give up + self.auto_compacted = true; + receiver.onProviderRetry(.{ + .attempt = 1, + .max_attempts = 2, + .delay_ms = 0, + .err = err, + .compaction = true, + }); + // Retry the same request against the compacted context. A second + // overflow (or any other error) propagates. + var diag: provider_mod.ProviderDiagnostic = .{}; + try self.stream_fn(self.allocator, self.io, cfg, conv, receiver, &diag); + } + + /// Compute the backoff delay (ms) for the just-failed `attempt` + /// (1-based). Prefers a provider `Retry-After` (capped by policy); + /// otherwise exponential `initial * multiplier^(attempt-1)`, capped, + /// with optional full jitter in `[0, delay)`. + fn backoffDelayMs( + self: *Agent, + policy: config_mod.RetryConfig, + attempt: usize, + retry_after_ms: ?u64, + ) u64 { + if (retry_after_ms) |ra| { + return @min(ra, policy.max_delay_ms); + } + const exp: f64 = @floatFromInt(attempt - 1); + const base: f64 = @as(f64, @floatFromInt(policy.initial_delay_ms)) * + std.math.pow(f64, policy.multiplier, exp); + const capped: f64 = @min(base, @as(f64, @floatFromInt(policy.max_delay_ms))); + var delay: u64 = @intFromFloat(capped); + if (policy.jitter and delay > 0) { + if (self.retry_prng == null) { + const ns = std.Io.Clock.now(.real, self.io).nanoseconds; + const seed: u64 = @truncate(@as(u128, @bitCast(@as(i128, ns)))); + self.retry_prng = std.Random.DefaultPrng.init(seed); + } + const r = self.retry_prng.?.random(); + delay = r.intRangeLessThan(u64, 0, delay + 1); + } + return delay; + } + /// Outcome of a compaction attempt. pub const CompactionResult = struct { /// Whether the conversation was actually compacted. False means the @@ -477,7 +623,7 @@ pub const Agent = struct { defer capture.deinit(); var recv = capture.receiver(); - try self.stream_fn(alloc, self.io, cfg, &conv, &recv); + try self.stream_fn(alloc, self.io, cfg, &conv, &recv, null); // The provider appended an assistant message; gather its text. const last = conv.messages.items[conv.messages.items.len - 1]; @@ -515,12 +661,25 @@ pub const Agent = struct { .entry = null, .result = try invalidInputResult(self.allocator, tu.input.items), .err = null, + .is_error = true, }); continue; } const entry = self.config.registry.lookup(tu.name) orelse { - // Unknown tool: abort the turn with a clear error. - return error.UnknownTool; + // Unknown tool: don't abort. Synthesize an error result so + // the model can correct, and so this ToolUse still gets its + // matching ToolResult (providers reject a follow-up request + // otherwise). + try calls.append(.{ + .tool_use_id = tu.id, + .tool_name = tu.name, + .input = tu.input.items, + .entry = null, + .result = try toolErrorResult(self.allocator, tu.name, error.UnknownTool), + .err = null, + .is_error = true, + }); + continue; }; try calls.append(.{ .tool_use_id = tu.id, @@ -561,16 +720,59 @@ pub const Agent = struct { } } + // Try real concurrency first. If the `Io` implementation can't + // provide it (`error.ConcurrencyUnavailable`), fall back to running + // every group sequentially on this thread — tool batches are small + // (rarely more than a handful of calls) so the serial path is a fine + // safety net rather than a hard failure. + var ran_concurrently = true; for (groups.items) |*g| { - try task_group.concurrent(self.io, runGroup, .{ self, g, calls.items }); + task_group.concurrent(self.io, runGroup, .{ self, g, calls.items }) catch |e| { + if (e == error.ConcurrencyUnavailable) { + ran_concurrently = false; + break; + } + return e; + }; + } + if (ran_concurrently) { + // `error.Canceled` here means cancellation propagated into this + // dispatch from above; surface it like any other error. + try task_group.await(self.io); + } else { + // Cancel any tasks that were spawned before the failure, then + // run all groups serially. Only entry-bearing calls are touched + // by `runGroup`; the pre-seeded error results (unknown tool, + // invalid input) have `entry == null` and must be left intact. + task_group.cancel(self.io); + for (calls.items) |*c| { + if (c.entry == null) continue; + if (c.result) |r| tool_mod.freeResultParts(self.allocator, r); + c.result = null; + c.err = null; + } + for (groups.items) |*g| runGroup(self, g, calls.items); } - // `error.Canceled` here means cancellation propagated into this - // dispatch from above; surface it like any other error. - try task_group.await(self.io); - // Assemble ToolResult blocks in original call order. If any - // call errored, prefer to abort the turn — but only after the - // standard errdefer above has freed remaining results. + // Pre-pass: resolve worker-reported errors. A hard host failure + // (cancellation, OOM) aborts the whole turn. Every other failure is + // converted into a model-visible error `ToolResult` so the model can + // recover and so each `ToolUse` keeps its matching `ToolResult` + // (providers reject the next request otherwise). + for (calls.items) |*c| { + const e = c.err orelse continue; + if (classifyToolError(e) == .hard_fail) return e; + // Replace any partial result with a synthesized error result. + if (c.result) |r| { + tool_mod.freeResultParts(self.allocator, r); + c.result = null; + } + c.result = try toolErrorResult(self.allocator, c.tool_name, e); + c.err = null; + c.is_error = true; + } + + // Assemble ToolResult blocks in original call order. var content: std.ArrayList(conversation.ContentBlock) = .empty; errdefer { for (content.items) |*b| b.deinit(self.allocator); @@ -578,17 +780,11 @@ pub const Agent = struct { } try content.ensureTotalCapacity(self.allocator, calls.items.len); - var first_err: ?anyerror = null; for (calls.items) |*c| { - if (c.err) |e| { - first_err = e; - continue; - } const result_parts = c.result orelse { - // Internal error: every successful call should have left - // parts behind. Treat as MissingToolResult. - first_err = error.MissingToolResult; - continue; + // Internal invariant: every call should now have a result + // (success, synthesized error, or pre-seeded error). + return error.MissingToolResult; }; c.result = null; // ownership transferred below defer tool_mod.freeResultParts(self.allocator, result_parts); @@ -616,16 +812,25 @@ pub const Agent = struct { // rasters, then base64-encode for storage. Tools // hand over raw bytes only. const processed = image_mod.process(self.allocator, m.data, m.media_type) catch |e| { - // Unrecognized bytes: keep the turn alive by - // dropping the attachment and noting it as text. + // Media processing failure: keep the turn alive by + // dropping the attachment and noting it as text, + // rather than aborting. `UnknownMediaType` gets a + // friendly note; other failures name the error. + var note: conversation.TextualBlock = .empty; + errdefer note.deinit(self.allocator); if (e == error.UnknownMediaType) { - var note: conversation.TextualBlock = .empty; - errdefer note.deinit(self.allocator); try note.appendSlice(self.allocator, "[unrecognized binary attachment dropped]"); - stored.appendAssumeCapacity(.{ .text = note }); - continue; + } else { + const txt = try std.fmt.allocPrint( + self.allocator, + "[media attachment dropped: {s}]", + .{@errorName(e)}, + ); + defer self.allocator.free(txt); + try note.appendSlice(self.allocator, txt); } - return e; + stored.appendAssumeCapacity(.{ .text = note }); + continue; }; defer self.allocator.free(processed.data); @@ -646,11 +851,10 @@ pub const Agent = struct { content.appendAssumeCapacity(.{ .ToolResult = .{ .tool_use_id = id_copy, .parts = stored, + .is_error = c.is_error, } }); } - if (first_err) |e| return e; - try conv.messages.append(self.allocator, .{ .role = .user, .content = content, @@ -672,8 +876,15 @@ const FlatCall = struct { /// ToolResultBlock on success. result: ?[]tool_mod.ResultPart, - /// If non-null, the call failed and the turn must abort. + /// If non-null, the worker reported a failure for this call. After + /// dispatch it is classified: host failures abort the turn, everything + /// else is converted into an error `ToolResult`. err: ?anyerror, + + /// True when `result` already holds a synthesized error result (unknown + /// tool, invalid input). Worker-reported `err`s are folded into this + /// during assembly. + is_error: bool = false, }; /// One dispatch group. Either a single Tool invocation, or a batch of @@ -844,6 +1055,21 @@ const StubProvider = struct { /// `error.ContextOverflow` before any scripted turn is served. Used to /// drive the auto-compaction path. Decremented on each overflow. overflow_calls: usize = 0, + /// A queue of provider errors to return, in order, before any scripted + /// turn is served. Each entry is consumed on one stream call. Used to + /// drive the transient-retry path. `diag_retry_after_ms`, when set on an + /// entry, is stashed into the caller's `ProviderDiagnostic`. + scripted_errors: []const ScriptedError = &.{}, + error_idx: usize = 0, + /// Count of stream calls observed (failed + succeeded). Lets tests + /// assert the exact number of attempts. + calls_made: usize = 0, + + const ScriptedError = struct { + err: anyerror, + status_code: ?u16 = null, + retry_after_ms: ?u64 = null, + }; const ScriptedTurn = struct { blocks: []const TestBlock, @@ -873,9 +1099,20 @@ fn stubStreamStep( _: *const config_mod.Config, conv: *conversation.Conversation, _: *provider_mod.Receiver, + diag: ?*provider_mod.ProviderDiagnostic, ) anyerror!void { const self = stub_active orelse return error.NoStubInstalled; _ = allocator; + self.calls_made += 1; + if (self.error_idx < self.scripted_errors.len) { + const e = self.scripted_errors[self.error_idx]; + self.error_idx += 1; + if (diag) |d| { + d.status_code = e.status_code; + d.retry_after_ms = e.retry_after_ms; + } + return e.err; + } if (self.overflow_calls > 0) { self.overflow_calls -= 1; return error.ContextOverflow; @@ -1063,6 +1300,35 @@ const FailingTool = struct { } }; +/// A tool that returns a hard host failure (`error.Canceled`), which must +/// abort the whole turn rather than degrade into a tool result. +const HardFailTool = struct { + name_owned: []u8, + + fn create(allocator: Allocator, name: []const u8) !Tool { + const self = try allocator.create(HardFailTool); + errdefer allocator.destroy(self); + self.name_owned = try allocator.dupe(u8, name); + return .{ + .decl = .{ .name = self.name_owned, .description = "hard fail", .schema_json = "{}" }, + .ctx = self, + .vtable = &vt, + }; + } + + const vt: Tool.VTable = .{ .invoke = invoke, .deinit = deinit }; + + fn invoke(_: *anyopaque, _: []const u8, _: Allocator) anyerror![]tool_mod.ResultPart { + return error.Canceled; + } + + fn deinit(ctx: *anyopaque, allocator: Allocator) void { + const self: *HardFailTool = @ptrCast(@alignCast(ctx)); + allocator.free(self.name_owned); + allocator.destroy(self); + } +}; + const NoopReceiver = struct { fn make() provider_mod.Receiver { return .{ .ptr = @ptrCast(@constCast(&dummy)), .vtable = &vt }; @@ -1076,6 +1342,7 @@ const NoopReceiver = struct { .onBlockComplete = noop4, .onMessageComplete = noop5, .onError = noop6, + .onProviderRetry = noop7, }; fn noop1(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} fn noop2(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} @@ -1084,6 +1351,7 @@ const NoopReceiver = struct { fn noop4(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} fn noop5(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} fn noop6(_: *anyopaque, _: anyerror) void {} + fn noop7(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} }; /// A configurable ToolSource for testing the grouped-dispatch path. @@ -1267,6 +1535,66 @@ const FailingSource = struct { } }; +/// A source that succeeds the first member call and fails the rest with a +/// per-call error (returning void from `invoke_batch`). Exercises the +/// per-call error path distinct from a whole-batch failure. +const PartialSource = struct { + name_owned: []u8, + decls: []tool_source_mod.ToolDecl, + decl_strings: std.array_list.Managed([]u8), + allocator: Allocator, + + fn create(allocator: Allocator, source_name: []const u8, tool_names: []const []const u8) !ToolSource { + const self = try allocator.create(PartialSource); + errdefer allocator.destroy(self); + var strings = std.array_list.Managed([]u8).init(allocator); + errdefer { + for (strings.items) |s| allocator.free(s); + strings.deinit(); + } + const name_owned = try allocator.dupe(u8, source_name); + try strings.append(name_owned); + const decls = try allocator.alloc(tool_source_mod.ToolDecl, tool_names.len); + errdefer allocator.free(decls); + for (tool_names, 0..) |tn, i| { + const n = try allocator.dupe(u8, tn); + try strings.append(n); + const d = try allocator.dupe(u8, "partial"); + try strings.append(d); + const s = try allocator.dupe(u8, "{}"); + try strings.append(s); + decls[i] = .{ .name = n, .description = d, .schema_json = s }; + } + self.* = .{ .name_owned = name_owned, .decls = decls, .decl_strings = strings, .allocator = allocator }; + return ToolSource{ .name = self.name_owned, .tools = self.decls, .ctx = self, .vtable = &vt }; + } + + const vt: ToolSource.VTable = .{ .invoke_batch = invokeBatch, .deinit = deinitSrc }; + + fn invokeBatch( + _: *anyopaque, + calls: []const tool_source_mod.Call, + results: []tool_source_mod.CallResult, + allocator: Allocator, + ) anyerror!void { + for (calls, 0..) |_, j| { + if (j == 0) { + results[j] = .{ .ok = try tool_mod.textResult(allocator, "ok") }; + } else { + results[j] = .{ .err = error.PerCallBoom }; + } + } + } + + fn deinitSrc(ctx: *anyopaque, _: Allocator) void { + const self: *PartialSource = @ptrCast(@alignCast(ctx)); + for (self.decl_strings.items) |s| self.allocator.free(s); + self.decl_strings.deinit(); + self.allocator.free(self.decls); + self.allocator.destroy(self); + } +}; + test "registry register and lookup" { var h = TestHarness.init(testing.allocator); defer h.deinit(); @@ -1378,14 +1706,14 @@ test "runStep dispatches multiple tool calls in parallel" { try testing.expect(t0 != t1 and t1 != t2 and t0 != t2); } -test "runStep propagates tool errors and aborts the turn" { +test "runStep: native tool handler error becomes an error result and the model gets another turn" { const allocator = testing.allocator; const scripted = [_]StubProvider.ScriptedTurn{ .{ .blocks = &.{ .{ .ToolUse = .{ .id = "x", .name = "boom", .input = "" } }, } }, - .{ .blocks = &.{.{ .Text = "should-not-see" }} }, + .{ .blocks = &.{.{ .Text = "i will recover" }} }, }; var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; var threaded: std.Io.Threaded = .init(allocator, .{}); @@ -1403,18 +1731,24 @@ test "runStep propagates tool errors and aborts the turn" { try conv.addUserMessage("break it"); var recv = NoopReceiver.make(); - try testing.expectError(error.ToolExploded, agent.runStep(&conv, &recv)); + try agent.runStep(&conv, &recv); - try testing.expectEqual(@as(usize, 2), conv.messages.items.len); + // user, assistant(tool_use), user(tool_result), assistant(text) + try testing.expectEqual(@as(usize, 4), conv.messages.items.len); + const tr = conv.messages.items[2].content.items[0].ToolResult; + try testing.expectEqualStrings("x", tr.tool_use_id); + try testing.expect(tr.is_error); + try testing.expect(std.mem.indexOf(u8, trText(tr), "ToolExploded") != null); } -test "runStep errors UnknownTool when the model calls something unregistered" { +test "runStep: unknown tool becomes an error tool result and the loop continues" { const allocator = testing.allocator; const scripted = [_]StubProvider.ScriptedTurn{ .{ .blocks = &.{ .{ .ToolUse = .{ .id = "z", .name = "ghost", .input = "" } }, } }, + .{ .blocks = &.{.{ .Text = "ok, that tool does not exist" }} }, }; var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; var threaded: std.Io.Threaded = .init(allocator, .{}); @@ -1431,7 +1765,16 @@ test "runStep errors UnknownTool when the model calls something unregistered" { try conv.addUserMessage("call a ghost"); var recv = NoopReceiver.make(); - try testing.expectError(error.UnknownTool, agent.runStep(&conv, &recv)); + try agent.runStep(&conv, &recv); + + // messages: user, assistant(tool_use), user(tool_result), assistant(text) + try testing.expectEqual(@as(usize, 4), conv.messages.items.len); + const tr_msg = conv.messages.items[2]; + try testing.expectEqual(conversation.MessageRole.user, tr_msg.role); + const tr = tr_msg.content.items[0].ToolResult; + try testing.expectEqualStrings("z", tr.tool_use_id); + try testing.expect(tr.is_error); + try testing.expect(std.mem.indexOf(u8, trText(tr), "UnknownTool") != null); } test "runStep with no tool calls returns after one provider step" { @@ -1579,7 +1922,7 @@ test "runStep: distinct sources run on distinct threads in parallel" { try testing.expect(sa.batches.items[0].thread_id != sb.batches.items[0].thread_id); } -test "runStep: source whole-batch error aborts the turn" { +test "runStep: source whole-batch error becomes per-call error results and continues" { const allocator = testing.allocator; const scripted = [_]StubProvider.ScriptedTurn{ @@ -1587,7 +1930,7 @@ test "runStep: source whole-batch error aborts the turn" { .{ .ToolUse = .{ .id = "a", .name = "fa", .input = "" } }, .{ .ToolUse = .{ .id = "b", .name = "fb", .input = "" } }, } }, - .{ .blocks = &.{.{ .Text = "never" }} }, + .{ .blocks = &.{.{ .Text = "recovered" }} }, }; var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; var threaded: std.Io.Threaded = .init(allocator, .{}); @@ -1605,10 +1948,20 @@ test "runStep: source whole-batch error aborts the turn" { try conv.addUserMessage("kaboom"); var recv = NoopReceiver.make(); - try testing.expectError(error.SourceExploded, agent.runStep(&conv, &recv)); + try agent.runStep(&conv, &recv); - // Conversation stops at user + assistant(tool_use). No ToolResult appended. - try testing.expectEqual(@as(usize, 2), conv.messages.items.len); + // user, assistant(tool_use x2), user(tool_result x2), assistant(text) + try testing.expectEqual(@as(usize, 4), conv.messages.items.len); + const tr_msg = conv.messages.items[2]; + try testing.expectEqual(@as(usize, 2), tr_msg.content.items.len); + // Every member of the failed batch produced an error result, in order. + const tr_a = tr_msg.content.items[0].ToolResult; + const tr_b = tr_msg.content.items[1].ToolResult; + try testing.expectEqualStrings("a", tr_a.tool_use_id); + try testing.expectEqualStrings("b", tr_b.tool_use_id); + try testing.expect(tr_a.is_error); + try testing.expect(tr_b.is_error); + try testing.expect(std.mem.indexOf(u8, trText(tr_a), "SourceExploded") != null); } test "runStep: mixed single Tools and source-backed tools coexist in one turn" { @@ -1911,3 +2264,371 @@ test "runStep: context overflow without compaction prompt propagates" { var recv = NoopReceiver.make(); try testing.expectError(error.ContextOverflow, agent.runStep(&conv, &recv)); } + +// ----------------------------------------------------------------------------- +// Phase 6: provider retry + tool-error holistic tests +// ----------------------------------------------------------------------------- + +/// Receiver that records `onProviderRetry` notifications (and nothing else), +/// so tests can assert retry scheduling without a live provider. +const RetryRecordingReceiver = struct { + infos: std.ArrayList(provider_mod.ProviderRetryInfo) = .empty, + allocator: Allocator, + + fn make(self: *RetryRecordingReceiver) provider_mod.Receiver { + return .{ .ptr = self, .vtable = &vt }; + } + fn deinit(self: *RetryRecordingReceiver) void { + self.infos.deinit(self.allocator); + } + const vt: provider_mod.ReceiverVTable = .{ + .onMessageStart = onMessageStart, + .onBlockStart = onBlockStart, + .onToolDetails = onToolDetails, + .onContentDelta = onContentDelta, + .onBlockComplete = onBlockComplete, + .onMessageComplete = onMessageComplete, + .onError = onError, + .onProviderRetry = onProviderRetry, + }; + fn onMessageStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} + fn onBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} + fn onToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {} + fn onContentDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} + fn onBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} + fn onMessageComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} + fn onError(_: *anyopaque, _: anyerror) void {} + fn onProviderRetry(ptr: *anyopaque, info: provider_mod.ProviderRetryInfo) void { + const self: *RetryRecordingReceiver = @ptrCast(@alignCast(ptr)); + self.infos.append(self.allocator, info) catch {}; + } +}; + +/// Build an agent + harness with near-zero backoff so retry tests don't +/// actually sleep. Caller owns the harness and must keep it alive. +fn fastRetryHarness(h: *TestHarness) void { + h.activate(); + // Make sleeps negligible and deterministic (no jitter). + h.config.retry = .{ + .max_attempts = 4, + .initial_delay_ms = 0, + .max_delay_ms = 0, + .multiplier = 2.0, + .jitter = false, + }; +} + +test "runStep: provider 429 retries then succeeds without duplicate messages" { + const allocator = testing.allocator; + + const errs = [_]StubProvider.ScriptedError{ + .{ .err = error.ProviderRateLimited, .status_code = 429 }, + .{ .err = error.ProviderRateLimited, .status_code = 429 }, + }; + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "finally" }} }, + }; + var stub = StubProvider{ + .allocator = allocator, + .scripted = &scripted, + .scripted_errors = &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 agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rr = RetryRecordingReceiver{ .allocator = allocator }; + defer rr.deinit(); + var recv = rr.make(); + try agent.runStep(&conv, &recv); + + // Two failures + one success. + try testing.expectEqual(@as(usize, 3), stub.calls_made); + // No duplicate assistant messages: user + 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, delivered before each delayed retry. + try testing.expectEqual(@as(usize, 2), rr.infos.items.len); + try testing.expectEqual(@as(?u16, 429), rr.infos.items[0].status_code); + try testing.expectEqual(@as(usize, 1), rr.infos.items[0].attempt); + try testing.expectEqual(@as(usize, 2), rr.infos.items[1].attempt); +} + +test "runStep: provider 500 retries with backoff notification" { + const allocator = testing.allocator; + + const errs = [_]StubProvider.ScriptedError{ + .{ .err = error.ProviderServerError, .status_code = 500 }, + }; + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "ok" }} }, + }; + var stub = StubProvider{ + .allocator = allocator, + .scripted = &scripted, + .scripted_errors = &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 agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rr = RetryRecordingReceiver{ .allocator = allocator }; + defer rr.deinit(); + var recv = rr.make(); + try agent.runStep(&conv, &recv); + + try testing.expectEqual(@as(usize, 2), stub.calls_made); + try testing.expectEqual(@as(usize, 1), rr.infos.items.len); + try testing.expectEqual(error.ProviderServerError, rr.infos.items[0].err); + try testing.expectEqual(@as(usize, 4), rr.infos.items[0].max_attempts); + try testing.expect(!rr.infos.items[0].compaction); +} + +test "runStep: provider auth failure does not retry" { + const allocator = testing.allocator; + + const errs = [_]StubProvider.ScriptedError{ + .{ .err = error.ProviderAuthFailed, .status_code = 401 }, + }; + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "unreachable" }} }, + }; + var stub = StubProvider{ + .allocator = allocator, + .scripted = &scripted, + .scripted_errors = &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 agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rr = RetryRecordingReceiver{ .allocator = allocator }; + defer rr.deinit(); + var recv = rr.make(); + try testing.expectError(error.ProviderAuthFailed, agent.runStep(&conv, &recv)); + + // Exactly one attempt, no retry notification. + try testing.expectEqual(@as(usize, 1), stub.calls_made); + try testing.expectEqual(@as(usize, 0), rr.infos.items.len); +} + +test "runStep: retries exhaust and hard-fail after max_attempts" { + const allocator = testing.allocator; + + const errs = [_]StubProvider.ScriptedError{ + .{ .err = error.ProviderUnavailable, .status_code = 503 }, + .{ .err = error.ProviderUnavailable, .status_code = 503 }, + .{ .err = error.ProviderUnavailable, .status_code = 503 }, + .{ .err = error.ProviderUnavailable, .status_code = 503 }, + }; + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "unreachable" }} }, + }; + var stub = StubProvider{ + .allocator = allocator, + .scripted = &scripted, + .scripted_errors = &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 agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rr = RetryRecordingReceiver{ .allocator = allocator }; + defer rr.deinit(); + var recv = rr.make(); + try testing.expectError(error.ProviderUnavailable, agent.runStep(&conv, &recv)); + + // 4 attempts total (max_attempts), 3 retry notifications. + 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; + + const errs = [_]StubProvider.ScriptedError{ + .{ .err = error.ProviderRateLimited, .status_code = 429, .retry_after_ms = 7000 }, + }; + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "ok" }} }, + }; + var stub = StubProvider{ + .allocator = allocator, + .scripted = &scripted, + .scripted_errors = &errs, + }; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var h = TestHarness.init(allocator); + defer h.deinit(); + h.activate(); + // Cap below the Retry-After to verify the policy cap applies. + h.config.retry = .{ .initial_delay_ms = 0, .max_delay_ms = 1, .jitter = false }; + var agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rr = RetryRecordingReceiver{ .allocator = allocator }; + defer rr.deinit(); + var recv = rr.make(); + try agent.runStep(&conv, &recv); + + try testing.expectEqual(@as(usize, 1), rr.infos.items.len); + // Reported Retry-After is the raw provider value... + try testing.expectEqual(@as(?u64, 7000), rr.infos.items[0].retry_after_ms); + // ...but the actual delay is capped by policy.max_delay_ms. + try testing.expectEqual(@as(u64, 1), rr.infos.items[0].delay_ms); +} + +test "runStep: cancellation from a tool still hard-fails" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{ + .{ .ToolUse = .{ .id = "x", .name = "hard", .input = "" } }, + } }, + .{ .blocks = &.{.{ .Text = "unreachable" }} }, + }; + var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var h = TestHarness.init(allocator); + defer h.deinit(); + try h.registry.register(try HardFailTool.create(allocator, "hard")); + h.activate(); + var agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("go"); + + var recv = NoopReceiver.make(); + try testing.expectError(error.Canceled, agent.runStep(&conv, &recv)); + // Turn aborts: no tool result appended (user + assistant only). + try testing.expectEqual(@as(usize, 2), conv.messages.items.len); +} + +test "runStep: source per-call error produces a per-call error result and continues" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{ + .{ .ToolUse = .{ .id = "a", .name = "pa", .input = "" } }, + .{ .ToolUse = .{ .id = "b", .name = "pb", .input = "" } }, + } }, + .{ .blocks = &.{.{ .Text = "moving on" }} }, + }; + var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var h = TestHarness.init(allocator); + defer h.deinit(); + try h.registry.registerSource(try PartialSource.create(allocator, "ps", &.{ "pa", "pb" })); + h.activate(); + var agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("go"); + + var recv = NoopReceiver.make(); + try agent.runStep(&conv, &recv); + + try testing.expectEqual(@as(usize, 4), conv.messages.items.len); + const tr_msg = conv.messages.items[2]; + const tr_a = tr_msg.content.items[0].ToolResult; // first call succeeded + const tr_b = tr_msg.content.items[1].ToolResult; // second failed + try testing.expect(!tr_a.is_error); + try testing.expectEqualStrings("ok", trText(tr_a)); + try testing.expect(tr_b.is_error); + try testing.expect(std.mem.indexOf(u8, trText(tr_b), "PerCallBoom") != null); +} + +test "runStep: context-overflow compaction fires a compaction retry notification" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "COMPACTED SUMMARY" }} }, // compaction call + .{ .blocks = &.{.{ .Text = "final answer" }} }, // retried main call + }; + var stub = StubProvider{ + .allocator = allocator, + .scripted = &scripted, + .overflow_calls = 1, + }; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var h = TestHarness.init(allocator); + defer h.deinit(); + h.activate(); + h.config.compaction = .{ .keep_verbatim = 10 }; + var agent = Agent.init(allocator, io, &h.config); + agent.stream_fn = stub.install(); + agent.compaction_system_prompt = "Summarize the conversation."; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addSystemMessage("you are helpful"); + try conv.addUserMessage("first question with several words here"); + try conv.addAssistantMessage(&.{ + .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") }, + }); + try conv.addUserMessage("second recent question"); + + var rr = RetryRecordingReceiver{ .allocator = allocator }; + defer rr.deinit(); + var recv = rr.make(); + try agent.runStep(&conv, &recv); + + try testing.expect(agent.auto_compacted); + // Exactly one notification, flagged as a compaction retry with no delay. + try testing.expectEqual(@as(usize, 1), rr.infos.items.len); + try testing.expect(rr.infos.items[0].compaction); + try testing.expectEqual(@as(u64, 0), rr.infos.items[0].delay_ms); + try testing.expectEqual(error.ContextOverflow, rr.infos.items[0].err); +} -- cgit v1.3