failure retries scheme

1 files changed, 775 insertions, 54 deletions

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);
+}