From 3f1ace16afc7877b0bfad374cb286d4d84140960 Mon Sep 17 00:00:00 2001 From: t Date: Thu, 4 Jun 2026 09:46:49 -0600 Subject: failure retries scheme --- docs/archive/retries.md | 341 +++++++++++ docs/todos.md | 5 +- libpanto/src/agent.zig | 829 +++++++++++++++++++++++++-- libpanto/src/anthropic_messages_json.zig | 38 ++ libpanto/src/config.zig | 20 + libpanto/src/conversation.zig | 6 + libpanto/src/openai_chat_json.zig | 2 +- libpanto/src/provider.zig | 193 +++++++ libpanto/src/provider_anthropic_messages.zig | 32 +- libpanto/src/provider_openai_chat.zig | 39 +- libpanto/src/session.zig | 72 ++- libpanto/src/tool.zig | 17 +- libpanto/src/tool_source.zig | 15 +- src/main.zig | 20 + 14 files changed, 1533 insertions(+), 96 deletions(-) create mode 100644 docs/archive/retries.md diff --git a/docs/archive/retries.md b/docs/archive/retries.md new file mode 100644 index 0000000..26aced1 --- /dev/null +++ b/docs/archive/retries.md @@ -0,0 +1,341 @@ +# Message-level error retries + +## Summary + +The agent loop should make an explicit decision for every failure that occurs while producing the next conversation message. Depending on the failure, the agent should either: + +1. retry the same provider request with backoff, +2. compact/rewrite the conversation and retry the same provider request, +3. append conversation-visible tool error results and continue the agent loop, or +4. hard-fail and return the error to the embedder. + +The important distinction is between provider/API failures and tool-call failures: + +- Provider/API failures usually mean the assistant message was not produced. The correct recovery is to retry the same request without mutating the conversation. +- Tool-call failures usually mean the model made a recoverable choice. The correct recovery is to append `ToolResult` blocks describing the failures, then let the model continue from there. + +This plan treats retries as **message-level**: a failed provider attempt should not create a partial duplicate message, and a tool failure should be represented as the next conversation message in the normal assistant/tool-result exchange. + +## Existing behavior + +We already have several special cases: + +- Context overflow errors are handled by automatic compaction followed by one retry of the same request. +- Max-tokens overflow during tool argument generation can surface as a partially generated tool input; this is currently handled as invalid/incomplete JSON-ish tool input and sent back as a tool result. +- Some tool-result post-processing failures are already degraded into text, e.g. unrecognized binary attachments are dropped with a note. + +However, error handling is not yet holistic: + +- Provider HTTP failures are often collapsed into broad errors such as `HttpError`, losing retryability information. +- Transient provider/API failures do not have a general backoff retry loop. +- Tool execution errors usually abort the agent loop instead of becoming tool error results. +- Retry attempts are not surfaced to the UI, which makes coding-agent behavior opaque during provider outages or rate limits. + +## Error actions + +### 1. Retry the same provider request + +Use this when the provider request appears transiently unavailable or unreliable, and the conversation has not gained a completed assistant message. + +Examples: + +- HTTP 429 rate limit. +- HTTP 408, 409, 425 where retry is appropriate. +- HTTP 500, 502, 503, 504. +- Connection reset, DNS/connect failure, TLS failure, timeout. +- Stream interruption or malformed provider stream before a complete assistant message. +- Possibly an empty assistant response, with a small retry cap. + +Behavior: + +- Do not append or modify conversation messages. +- Retry the same request against the same conversation snapshot. +- Use exponential backoff with jitter. +- Honor `Retry-After` when the provider sends it. +- Stop after a configurable attempt cap and then hard-fail. +- Notify the receiver/UI before each delayed retry. + +### 2. Compact/rewrite context, then retry the same provider request + +Use this when the provider rejects the request because the input context is too large. + +Examples: + +- OpenAI-compatible `context_length_exceeded` / `maximum context length` responses. +- Anthropic `prompt is too long` responses. +- Other provider-specific context-window diagnostics. + +Behavior: + +- Run automatic compaction if configured. +- Rewrite the conversation into the compacted form. +- Retry the same request once against the compacted conversation. +- If compaction cannot shed enough context, or if the retry also overflows, hard-fail. + +This is already mostly implemented; it should be integrated into the same provider-error decision path as ordinary retries. + +### 3. Append tool error results and continue + +Use this when the model emitted a tool call and the failure is meaningful to the model. + +Examples: + +- Unknown tool name. +- Invalid/incomplete tool input. +- Future schema validation failures. +- Native tool handler returned an error. +- Lua/runtime exception. +- `ToolSource` per-call error. +- `ToolSource` whole-batch failure, mapped to each affected call. +- Tool output shape/encoding problems that can be explained to the model. +- Media processing failures where the rest of the agent is healthy. + +Behavior: + +- Append a user message containing one `ToolResult` for every `ToolUse` block, preserving original call order. +- Successful calls keep their normal results. +- Failed calls get model-readable error text. +- The agent loop continues, allowing the model to correct arguments, try a different tool, or explain the failure to the user. +- Prefer adding an internal `is_error` flag to tool results, serialized where supported. + +Example error text: + +```text +Tool execution failed for `std.read`: file not found: src/foo.zig +You may fix the arguments, try a different tool, or explain the failure to the user. +``` + +Provider serialization: + +- Anthropic supports an `is_error`-style tool result marker; serialize it when present. +- OpenAI Chat Completions has no direct equivalent; include the error as textual tool content. + +### 4. Hard-fail + +Use this when retrying would be unsafe or misleading, or when the failure belongs to the host application rather than the model/provider interaction. + +Examples: + +- User cancellation / abort. +- Out of memory. +- Receiver/UI/session persistence failure. +- Invalid local configuration: missing API key, malformed URL, unknown model alias. +- Provider authentication/authorization failure: HTTP 401/403. +- Non-retryable provider request errors: most HTTP 400s except context overflow. +- Internal invariant violation / conversation corruption. +- Concurrency unavailable, unless we add a sequential fallback. + +Behavior: + +- Return the error to the embedder. +- Do not synthesize a model-visible tool result for host/system failures. + +## Provider retry notifications + +Retry visibility should be front-loaded. Coding-agent UIs need to show that the agent is waiting on the provider rather than silently hanging. + +Add a receiver callback for provider retry scheduling, something like: + +```zig +onProviderRetry(info: ProviderRetryInfo) void +``` + +Possible fields: + +```zig +const ProviderRetryInfo = struct { + attempt: usize, // next attempt number, 1-based or 0-based; define clearly + max_attempts: usize, + delay_ms: u64, + err: anyerror, + status_code: ?u16 = null, + retry_after_ms: ?u64 = null, + message: ?[]const u8 = null, +}; +``` + +Semantics: + +- Fire after a provider attempt fails and before sleeping for the next attempt. +- Fire for ordinary provider retry/backoff. +- Fire for context-overflow compaction retry too, either through this callback or through a sibling callback that says compaction is being attempted. +- Do not fire for tool-call failures that are being sent back as tool results; those are normal conversation continuations. + +Initial CLI behavior can be simple, e.g. print a dim status line: + +```text +provider unavailable: retrying in 2.4s (attempt 2/4) +``` + +## Provider error classification + +The providers should stop collapsing all HTTP/API failures into a single broad error when the agent needs to distinguish retryable from non-retryable failures. + +Minimum useful public errors: + +```zig +error.ProviderRateLimited +error.ProviderUnavailable +error.ProviderServerError +error.ProviderTransport +error.ProviderStreamMalformed +error.ProviderAuthFailed +error.ProviderBadRequest +error.ProviderModelNotFound +error.ContextOverflow +``` + +Suggested HTTP mapping: + +- `400` + context marker: `ContextOverflow` +- other `400`: `ProviderBadRequest` +- `401`, `403`: `ProviderAuthFailed` +- `404`: `ProviderModelNotFound` or `ProviderBadRequest`, depending on provider/API shape +- `408`: retryable transport/request timeout +- `409`, `425`: retryable if provider semantics indicate temporary conflict/not-ready +- `429`: `ProviderRateLimited` +- `500..599`: `ProviderServerError` / `ProviderUnavailable` + +Longer term, `anyerror` alone is too lossy for status code, provider message, and `Retry-After`. A richer provider error/reporting channel would be useful. For the first implementation, distinct error names plus logging may be enough. + +## Agent retry policy + +Add a retry policy to config, with conservative defaults. + +Possible shape: + +```zig +const RetryConfig = struct { + max_attempts: usize = 4, + initial_delay_ms: u64 = 500, + max_delay_ms: u64 = 10_000, + multiplier: f64 = 2.0, + jitter: bool = true, +}; +``` + +Rules: + +- Attempt count includes the first attempt; `max_attempts = 4` means one initial try plus up to three retries. +- Retry only errors classified as retryable provider failures. +- Never retry cancellation, local receiver errors, auth failures, or bad local configuration. +- If the provider supplies `Retry-After`, prefer that delay, capped by policy. +- Keep context-overflow compaction retry as a separate one-shot path, not an unbounded retry loop. + +## Tool error result policy + +Change tool dispatch so normal tool failures do not abort the turn. + +Current behavior to replace: + +- Unknown tool returns `error.UnknownTool`. +- `Tool.invoke` error records `FlatCall.err`; after dispatch, the first error aborts the turn. +- `ToolSource` per-call errors and whole-batch errors abort the turn. + +Desired behavior: + +- Unknown tool becomes an error `ToolResult`. +- Invalid/incomplete input remains an error `ToolResult`. +- Native tool errors become error `ToolResult`s unless classified as hard host failures. +- Source per-call errors become per-call error `ToolResult`s. +- Source whole-batch errors become error `ToolResult`s for every member call unless the batch failed due to hard host failure. +- Successful calls in the same assistant message should still return successful results. + +Hard host failures should still abort, e.g. cancellation and out-of-memory. + +This requires a helper like: + +```zig +fn classifyToolError(err: anyerror) ToolErrorAction { + return switch (err) { + error.Canceled, error.OutOfMemory => .hard_fail, + else => .tool_result, + }; +} +``` + +The exact list will need auditing across native tools, Lua tools, source runtimes, and media processing. + +## Data model changes + +Consider extending `ToolResultBlock` with an error marker: + +```zig +pub const ToolResultBlock = struct { + tool_use_id: []const u8, + parts: std.ArrayList(ResultPartStored), + is_error: bool = false, +}; +``` + +Update: + +- conversation cloning/deinit paths, +- session persistence format, +- Anthropic serializer/parser, +- OpenAI serializer/parser, +- tests. + +Backward compatibility: + +- Missing `is_error` in old sessions should default to `false`. +- OpenAI Chat serialization can ignore the flag except for the textual content already present. + +## Implementation plan + +### Phase 1: Classify provider errors + +- Expand provider HTTP status handling to return specific provider errors. +- Preserve context-overflow detection. +- Add tests for status-to-error mapping where practical. + +### Phase 2: Add provider retry loop + +- Add retry config defaults. +- Implement `streamWithRetries` around `stream_fn` in `Agent.runStep`. +- Retry only classified transient provider errors. +- Add exponential backoff with jitter and `Retry-After` support if available. +- Ensure failed attempts do not mutate the conversation. + +### Phase 3: Add retry notifications + +- Extend `ReceiverVTable` with provider retry notification. +- Update all receivers, test receivers, and CLI receivers. +- Print visible retry status in the CLI. + +### Phase 4: Convert tool failures to tool results + +- Add helper to build textual error `ResultPart`s. +- Change unknown-tool handling to append an error result instead of aborting. +- Change `FlatCall.err` assembly to synthesize error results for model-visible failures. +- Keep hard host failures as returned errors. +- Update tool API docs to say returned errors normally become model-visible tool failures. + +### Phase 5: Add tool-result error marker + +- Add `is_error` to `ToolResultBlock`. +- Serialize to Anthropic where supported. +- Preserve compatibility for OpenAI and old session files. +- Update tests. + +### Phase 6: Holistic tests + +Add tests for: + +- provider 429 retries and then succeeds without duplicate messages, +- provider 500 retries with backoff notification, +- provider auth failure does not retry, +- context overflow still compacts once and retries, +- unknown tool produces an error tool result and the agent continues, +- tool handler error produces an error tool result and the model gets another turn, +- source per-call and batch errors produce error results, +- cancellation/OOM-like errors still hard-fail, +- retry notification callbacks are delivered before delay. + +## Open questions + +- Should malformed provider streams always be retryable, or only before any complete content block has been committed? +- Do we need a richer provider error object sooner rather than later, mainly for `Retry-After` and UI diagnostics? +- Should tool implementations have a first-class `model_error` return path distinct from host errors, instead of overloading `anyerror`? +- Should `ConcurrencyUnavailable` fall back to sequential tool execution instead of hard-failing? +- How should retry notifications be represented in persisted sessions, if at all? Initial answer: probably not persisted; they are UI/runtime events, not conversation history. diff --git a/docs/todos.md b/docs/todos.md index 55a97fc..1388ecf 100644 --- a/docs/todos.md +++ b/docs/todos.md @@ -5,13 +5,14 @@ - [ ] C ABI - [x] Agent compaction with custom compaction prompts - [x] Agent auto-compaction -- [ ] image upload support +- [x] image upload support - [ ] google gemini native provider - [ ] openai responses API native provider - [ ] non-streaming - [ ] one-shot simple API -- [ ] message-level error retries +- [x] message-level error retries - [ ] abort/cancellation +- [ ] user message queueing: steering, follow-up - [ ] step cap, stop conditions ## panto cli 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); +} diff --git a/libpanto/src/anthropic_messages_json.zig b/libpanto/src/anthropic_messages_json.zig index 8e33539..940d55e 100644 --- a/libpanto/src/anthropic_messages_json.zig +++ b/libpanto/src/anthropic_messages_json.zig @@ -232,6 +232,13 @@ fn writeBlock(s: *std.json.Stringify, block: conversation.ContentBlock) !void { try s.write("tool_result"); try s.objectField("tool_use_id"); try s.write(tr.tool_use_id); + // Anthropic supports an `is_error` marker on tool results; + // emit it only when set (omitting it defaults to a success + // result on the wire). + if (tr.is_error) { + try s.objectField("is_error"); + try s.write(true); + } // Anthropic accepts `content` as an array of typed blocks: // text, image (base64 source), and document (PDF). try s.objectField("content"); @@ -1053,12 +1060,43 @@ test "serializeRequest - user ToolResult becomes tool_result content block" { try testing.expectEqual(@as(usize, 1), arr.len); try testing.expectEqualStrings("tool_result", arr[0].object.get("type").?.string); try testing.expectEqualStrings("tu_1", arr[0].object.get("tool_use_id").?.string); + // A successful result omits the is_error marker entirely. + try testing.expect(arr[0].object.get("is_error") == null); const tr_content = arr[0].object.get("content").?.array.items; try testing.expectEqual(@as(usize, 1), tr_content.len); try testing.expectEqualStrings("text", tr_content[0].object.get("type").?.string); try testing.expectEqualStrings("the answer is 42", tr_content[0].object.get("text").?.string); } +test "serializeRequest - error ToolResult emits is_error: true" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + const id = try allocator.dupe(u8, "tu_err"); + var parts: std.ArrayList(conversation.ResultPartStored) = .empty; + try parts.append(allocator, .{ .text = try conversation.textualBlockFromSlice(allocator, "file not found") }); + + var blocks: std.ArrayList(conversation.ContentBlock) = .empty; + try blocks.append(allocator, .{ .ToolResult = .{ .tool_use_id = id, .parts = parts, .is_error = true } }); + try conv.messages.append(allocator, .{ .role = .user, .content = blocks }); + + var empty_tools = tool_registry_mod.ToolRegistry.init(allocator); + defer empty_tools.deinit(); + const cfg = testConfig("claude-x"); + const body = try serializeRequest(allocator, &cfg, &conv, &empty_tools); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const msg = parsed.value.object.get("messages").?.array.items[0].object; + const arr = msg.get("content").?.array.items; + try testing.expectEqualStrings("tool_result", arr[0].object.get("type").?.string); + try testing.expect(arr[0].object.get("is_error").?.bool); +} + test "serializeRequest - tool result with image part emits image block" { const allocator = testing.allocator; diff --git a/libpanto/src/config.zig b/libpanto/src/config.zig index 114280c..eca81ff 100644 --- a/libpanto/src/config.zig +++ b/libpanto/src/config.zig @@ -87,6 +87,25 @@ pub const CompactionConfig = struct { model: ?ProviderConfig = null, }; +/// Policy for retrying transient provider/API failures. Conservative +/// defaults: four attempts (one initial + three retries) with exponential +/// backoff and jitter, capped at 10s per delay. +pub const RetryConfig = struct { + /// Total attempts including the first. `4` => initial try + up to 3 + /// retries. Must be >= 1. + max_attempts: usize = 4, + /// Base delay before the first retry, in milliseconds. + initial_delay_ms: u64 = 500, + /// Upper bound on any single backoff delay, in milliseconds. Also caps + /// a provider-supplied `Retry-After`. + max_delay_ms: u64 = 10_000, + /// Exponential growth factor applied per retry. + multiplier: f64 = 2.0, + /// When true, apply random jitter in `[0, computed_delay)` (full + /// jitter) to avoid thundering-herd retries. + jitter: bool = true, +}; + /// An immutable snapshot of everything the agent consults per turn: which /// provider/model to talk to, and which tools to expose. The agent holds a /// `*const Config` and re-reads it each turn; replacing the pointer swaps @@ -99,6 +118,7 @@ pub const Config = struct { provider: ProviderConfig, registry: *const ToolRegistry, compaction: CompactionConfig = .{}, + retry: RetryConfig = .{}, pub fn style(self: Config) APIStyle { return self.provider.style(); diff --git a/libpanto/src/conversation.zig b/libpanto/src/conversation.zig index 2c29f8e..f0c822d 100644 --- a/libpanto/src/conversation.zig +++ b/libpanto/src/conversation.zig @@ -71,6 +71,12 @@ pub const ResultPartStored = union(enum) { pub const ToolResultBlock = struct { tool_use_id: []const u8, parts: std.ArrayList(ResultPartStored) = .empty, + /// True when this result reports a tool failure rather than success. + /// Serialized to providers that support it (Anthropic's `is_error`); + /// recorded but unserialized for OpenAI Chat (the error text in `parts` + /// carries the signal there). Defaults to false for backward + /// compatibility with sessions written before this field existed. + is_error: bool = false, pub fn deinit(self: *ToolResultBlock, alloc: Allocator) void { alloc.free(self.tool_use_id); diff --git a/libpanto/src/openai_chat_json.zig b/libpanto/src/openai_chat_json.zig index f0d12ed..2307ce8 100644 --- a/libpanto/src/openai_chat_json.zig +++ b/libpanto/src/openai_chat_json.zig @@ -91,7 +91,7 @@ pub fn serializeRequest( try s.objectField("stream"); try s.write(true); - try s.objectField("max_tokens"); + try s.objectField("max_completion_tokens"); try s.write(cfg.max_tokens); // Ask for the final-chunk usage block. Without this the server diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig index 6ccbefe..0258b6b 100644 --- a/libpanto/src/provider.zig +++ b/libpanto/src/provider.zig @@ -36,6 +36,142 @@ pub fn isContextOverflowBody(body: []const u8) bool { return false; } +/// Distinct provider/API failure classes the agent needs in order to decide +/// between retrying, compacting, and hard-failing. The transport/stream layer +/// maps HTTP status codes and connection failures onto these so the agent's +/// retry policy can switch on a stable, provider-agnostic name rather than a +/// broad `error.HttpError`. +/// +/// Zig errors cannot carry payloads, so status code, `Retry-After`, and the +/// provider's diagnostic message ride alongside via `ProviderDiagnostic` (an +/// out-parameter the provider fills before returning the error). +pub const ProviderError = error{ + /// HTTP 429. Retryable; honor `Retry-After` when present. + ProviderRateLimited, + /// HTTP 503 or a server explicitly signalling unavailability. Retryable. + ProviderUnavailable, + /// HTTP 500/502/504. Retryable. + ProviderServerError, + /// Connection reset, DNS/connect failure, TLS failure, request timeout + /// (HTTP 408), or other transport-level failure before a response. Also + /// covers retryable conflict/not-ready statuses (409, 425). Retryable. + ProviderTransport, + /// The provider stream ended or was malformed before a complete + /// assistant message was committed. Retryable (subject to policy). + ProviderStreamMalformed, + /// HTTP 401/403. Not retryable — a credentials/permissions problem. + ProviderAuthFailed, + /// HTTP 400 (other than context overflow). Not retryable — the request + /// itself is malformed. + ProviderBadRequest, + /// HTTP 404 shaped as an unknown-model error. Not retryable. + ProviderModelNotFound, + /// The input context exceeds the model's window (HTTP 400 + a recognized + /// context marker). Handled by one-shot compaction, not ordinary retry. + ContextOverflow, +}; + +/// Side-channel for the payload a `ProviderError` cannot carry. The provider +/// fills the relevant fields immediately before returning a classified error; +/// the agent reads them to drive backoff and retry notifications. Reset to +/// `.{}` before each provider attempt. +pub const ProviderDiagnostic = struct { + /// The HTTP status code, when the failure carried one. + status_code: ?u16 = null, + /// Parsed `Retry-After` delay in milliseconds, when the provider sent it. + retry_after_ms: ?u64 = null, + /// The provider's diagnostic message (borrowed/owned per caller; the + /// agent treats it as borrowed for the lifetime of the failed attempt). + message: ?[]const u8 = null, + + pub fn reset(self: *ProviderDiagnostic) void { + self.* = .{}; + } +}; + +/// True for the provider errors the agent's retry policy may retry. Context +/// overflow is deliberately excluded — it has a separate one-shot compaction +/// path. Auth, bad-request, and model-not-found are terminal. +pub fn isRetryableProviderError(err: anyerror) bool { + return switch (err) { + error.ProviderRateLimited, + error.ProviderUnavailable, + error.ProviderServerError, + error.ProviderTransport, + error.ProviderStreamMalformed, + => true, + else => false, + }; +} + +/// Map an HTTP status code from a provider response onto a `ProviderError`. +/// The `body` is inspected only for the 400 case, to separate context +/// overflow (compact-and-retry) from an ordinary bad request (hard-fail). +/// +/// Caller is responsible for stashing the status code (and any `Retry-After`) +/// into a `ProviderDiagnostic`; this function only chooses the error name. +pub fn classifyHttpStatus(status: u16, body: []const u8) ProviderError { + return switch (status) { + 400 => if (isContextOverflowBody(body)) error.ContextOverflow else error.ProviderBadRequest, + 401, 403 => error.ProviderAuthFailed, + 404 => error.ProviderModelNotFound, + 408 => error.ProviderTransport, + 409, 425 => error.ProviderTransport, + 429 => error.ProviderRateLimited, + 503 => error.ProviderUnavailable, + 500, 502, 504 => error.ProviderServerError, + else => if (status >= 500) error.ProviderServerError else error.ProviderBadRequest, + }; +} + +/// Parse an HTTP `Retry-After` header value into milliseconds. Supports the +/// delta-seconds form (`"120"`); the HTTP-date form is not parsed and returns +/// null (the agent then falls back to its computed backoff). Returns null for +/// empty or unparseable values. +pub fn parseRetryAfterMs(value: []const u8) ?u64 { + const trimmed = std.mem.trim(u8, value, " \t\r\n"); + if (trimmed.len == 0) return null; + const secs = std.fmt.parseInt(u64, trimmed, 10) catch return null; + return secs *| std.time.ms_per_s; +} + +/// Find a `Retry-After` header (case-insensitive) in an HTTP response head +/// and parse it into milliseconds. `head` is a `std.http.Client.Response.Head` +/// (taken as `anytype` to avoid importing the http types here). Returns null +/// when absent or unparseable. +pub fn retryAfterFromHead(head: anytype) ?u64 { + var it = head.iterateHeaders(); + while (it.next()) |h| { + if (std.ascii.eqlIgnoreCase(h.name, "retry-after")) { + return parseRetryAfterMs(h.value); + } + } + return null; +} + +/// Details handed to the receiver before the agent sleeps for a provider +/// retry. Purely a UI/runtime event — never persisted to the session. +pub const ProviderRetryInfo = struct { + /// The attempt that just failed, 1-based (1 = the initial attempt). + attempt: usize, + /// Total attempts the policy will make, including the first. + max_attempts: usize, + /// How long the agent will sleep before the next attempt, in ms. + delay_ms: u64, + /// The classified error that triggered the retry. + err: anyerror, + /// HTTP status code, when known. + status_code: ?u16 = null, + /// Provider-sent `Retry-After`, when present, in ms. + retry_after_ms: ?u64 = null, + /// Provider diagnostic message, when known. + message: ?[]const u8 = null, + /// True when the retry is a context-overflow compaction attempt rather + /// than an ordinary backoff retry. Compaction retries report + /// `delay_ms == 0`. + compaction: bool = false, +}; + /// Vtable for receiving streaming events from a Provider. /// /// The lifecycle callbacks (`onMessageStart` ... `onMessageComplete`) return @@ -75,6 +211,12 @@ pub const ReceiverVTable = struct { onBlockComplete: *const fn (*anyopaque, usize, conversation.ContentBlock) anyerror!void, onMessageComplete: *const fn (*anyopaque, conversation.Message, ?Usage) anyerror!void, onError: *const fn (*anyopaque, anyerror) void, + /// Fired by the agent after a provider attempt fails and before it + /// sleeps for the next attempt. Purely informational: receivers surface + /// it (e.g. a dim CLI status line) but cannot abort or alter the retry. + /// Never fires for tool-call failures (those become tool results) nor + /// for terminal provider errors. May be a no-op. + onProviderRetry: *const fn (*anyopaque, ProviderRetryInfo) void, }; pub const Receiver = struct { @@ -108,6 +250,10 @@ pub const Receiver = struct { pub fn onError(self: Receiver, err: anyerror) void { self.vtable.onError(self.ptr, err); } + + pub fn onProviderRetry(self: Receiver, info: ProviderRetryInfo) void { + self.vtable.onProviderRetry(self.ptr, info); + } }; /// Drive one streaming provider turn against the active config snapshot. @@ -127,6 +273,7 @@ pub fn streamStep( cfg: *const config_mod.Config, conv: *conversation.Conversation, receiver: *Receiver, + diag: ?*ProviderDiagnostic, ) anyerror!void { // Imported lazily to break the circular module graph: // provider.zig <- provider_openai_chat.zig <- provider.zig. @@ -140,6 +287,7 @@ pub fn streamStep( .io = io, .config = c, .http_client = client, + .diag = diag, }; return req.streamStep(conv, cfg.registry, receiver); }, @@ -149,6 +297,7 @@ pub fn streamStep( .io = io, .config = c, .http_client = client, + .diag = diag, }; return req.streamStep(conv, cfg.registry, receiver); }, @@ -164,6 +313,7 @@ pub const StreamFn = *const fn ( cfg: *const config_mod.Config, conv: *conversation.Conversation, receiver: *Receiver, + diag: ?*ProviderDiagnostic, ) anyerror!void; test "isContextOverflowBody - matches known markers, rejects others" { @@ -174,3 +324,46 @@ test "isContextOverflowBody - matches known markers, rejects others" { try t2.expect(!isContextOverflowBody("{\"error\":{\"code\":\"invalid_api_key\"}}")); try t2.expect(!isContextOverflowBody("rate limit exceeded")); } + +test "classifyHttpStatus - maps statuses to provider errors" { + const t2 = std.testing; + try t2.expectEqual(error.ContextOverflow, classifyHttpStatus(400, "prompt is too long")); + try t2.expectEqual(error.ProviderBadRequest, classifyHttpStatus(400, "bad json")); + try t2.expectEqual(error.ProviderAuthFailed, classifyHttpStatus(401, "")); + try t2.expectEqual(error.ProviderAuthFailed, classifyHttpStatus(403, "")); + try t2.expectEqual(error.ProviderModelNotFound, classifyHttpStatus(404, "")); + try t2.expectEqual(error.ProviderTransport, classifyHttpStatus(408, "")); + try t2.expectEqual(error.ProviderTransport, classifyHttpStatus(409, "")); + try t2.expectEqual(error.ProviderTransport, classifyHttpStatus(425, "")); + try t2.expectEqual(error.ProviderRateLimited, classifyHttpStatus(429, "")); + try t2.expectEqual(error.ProviderServerError, classifyHttpStatus(500, "")); + try t2.expectEqual(error.ProviderServerError, classifyHttpStatus(502, "")); + try t2.expectEqual(error.ProviderUnavailable, classifyHttpStatus(503, "")); + try t2.expectEqual(error.ProviderServerError, classifyHttpStatus(504, "")); + try t2.expectEqual(error.ProviderServerError, classifyHttpStatus(599, "")); + try t2.expectEqual(error.ProviderBadRequest, classifyHttpStatus(418, "")); +} + +test "isRetryableProviderError - retryable vs terminal" { + const t2 = std.testing; + try t2.expect(isRetryableProviderError(error.ProviderRateLimited)); + try t2.expect(isRetryableProviderError(error.ProviderUnavailable)); + try t2.expect(isRetryableProviderError(error.ProviderServerError)); + try t2.expect(isRetryableProviderError(error.ProviderTransport)); + try t2.expect(isRetryableProviderError(error.ProviderStreamMalformed)); + try t2.expect(!isRetryableProviderError(error.ProviderAuthFailed)); + try t2.expect(!isRetryableProviderError(error.ProviderBadRequest)); + try t2.expect(!isRetryableProviderError(error.ProviderModelNotFound)); + try t2.expect(!isRetryableProviderError(error.ContextOverflow)); + try t2.expect(!isRetryableProviderError(error.Canceled)); +} + +test "parseRetryAfterMs - delta-seconds and rejects" { + const t2 = std.testing; + try t2.expectEqual(@as(?u64, 120_000), parseRetryAfterMs("120")); + try t2.expectEqual(@as(?u64, 0), parseRetryAfterMs("0")); + try t2.expectEqual(@as(?u64, 2_000), parseRetryAfterMs(" 2 ")); + try t2.expectEqual(@as(?u64, null), parseRetryAfterMs("")); + // HTTP-date form is not parsed. + try t2.expectEqual(@as(?u64, null), parseRetryAfterMs("Wed, 21 Oct 2015 07:28:00 GMT")); +} diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig index e77d6a5..55efb5f 100644 --- a/libpanto/src/provider_anthropic_messages.zig +++ b/libpanto/src/provider_anthropic_messages.zig @@ -33,6 +33,8 @@ pub const AnthropicMessagesRequest = struct { io: Io, config: *const config_mod.AnthropicMessagesConfig, http_client: *http.Client, + /// Optional diagnostic side-channel; see `OpenAIChatRequest.diag`. + diag: ?*provider_mod.ProviderDiagnostic = null, pub fn streamStep( self: *AnthropicMessagesRequest, @@ -106,17 +108,18 @@ pub const AnthropicMessagesRequest = struct { try err_buf.appendSlice(self.allocator, tmp[0..n]); if (err_buf.items.len > 16 * 1024) break; } - std.log.err("anthropic_messages HTTP {d}: {s}", .{ - @intFromEnum(response.head.status), - err_buf.items, - }); + const status: u16 = @intFromEnum(response.head.status); + std.log.err("anthropic_messages HTTP {d}: {s}", .{ status, err_buf.items }); // Anthropic rejects oversized requests with HTTP 400 and a - // "prompt is too long" message; surface as ContextOverflow so - // the caller can compact and retry. - if (@intFromEnum(response.head.status) == 400 and provider_mod.isContextOverflowBody(err_buf.items)) { - return error.ContextOverflow; + // "prompt is too long" message; `classifyHttpStatus` maps that to + // ContextOverflow so the caller can compact and retry. Other + // statuses map to retryable/terminal provider errors. + const classified = provider_mod.classifyHttpStatus(status, err_buf.items); + if (self.diag) |d| { + d.status_code = status; + d.retry_after_ms = provider_mod.retryAfterFromHead(response.head); } - return error.HttpError; + return classified; } var transfer_buf: [4096]u8 = undefined; @@ -140,7 +143,8 @@ pub const AnthropicMessagesRequest = struct { while (true) { const n = body_reader.readVec(&vecs) catch |err| switch (err) { error.EndOfStream => break, - else => return err, + // Transport failure before the message completed: retryable. + else => return error.ProviderStreamMalformed, }; if (n == 0) continue; @@ -473,7 +477,9 @@ fn handleEvent( if (!@import("builtin").is_test) { std.log.err("anthropic stream error: {?s}: {?s}", .{ e.kind, e.message }); } - return error.StreamError; + // Mid-stream error event (e.g. `overloaded_error`) before the + // message was committed: retryable malformed-stream failure. + return error.ProviderStreamMalformed; }, .unknown => { // Forward-compatible: ignore unknown event types per Anthropic's @@ -543,12 +549,14 @@ const RecordingReceiver = struct { .onBlockComplete = onBlockComplete, .onMessageComplete = onMessageComplete, .onError = onError, + .onProviderRetry = onProviderRetry, }; fn onMessageStart(ptr: *anyopaque, role: conversation.MessageRole) anyerror!void { const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); try self.events.append(self.allocator, .{ .message_start = role }); } + fn onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} fn onBlockStart( ptr: *anyopaque, bt: provider_mod.ContentBlockType, @@ -988,7 +996,7 @@ test "error event propagates as Zig error" { &state, &recv, ); - try testing.expectError(error.StreamError, result); + try testing.expectError(error.ProviderStreamMalformed, result); } test "two streamed turns persist assistant replies in the conversation" { diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index c454679..ca01d7c 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -46,6 +46,11 @@ pub const OpenAIChatRequest = struct { io: Io, config: *const config_mod.OpenAIChatConfig, http_client: *http.Client, + /// Optional diagnostic side-channel. When non-null, classified failures + /// stash the HTTP status code and any `Retry-After` here for the agent's + /// retry policy. Strings written here are not owned by the diagnostic; + /// they live only as long as this request object. + diag: ?*provider_mod.ProviderDiagnostic = null, pub fn streamStep( self: *OpenAIChatRequest, @@ -133,19 +138,17 @@ pub const OpenAIChatRequest = struct { try err_buf.appendSlice(self.allocator, tmp[0..n]); if (err_buf.items.len > 16 * 1024) break; } - std.log.err("openai_chat HTTP {d}: {s}", .{ - @intFromEnum(response.head.status), - err_buf.items, - }); - // Detect a context-overflow rejection so the caller can - // compact and retry rather than treating it as a hard error. - // OpenAI-compatible APIs return HTTP 400 with a - // `context_length_exceeded` code / "maximum context length" - // message on the input side. - if (@intFromEnum(response.head.status) == 400 and provider_mod.isContextOverflowBody(err_buf.items)) { - return error.ContextOverflow; + const status: u16 = @intFromEnum(response.head.status); + std.log.err("openai_chat HTTP {d}: {s}", .{ status, err_buf.items }); + // Classify the status into a retryable/terminal provider error. + // HTTP 400 with a context marker becomes `ContextOverflow` so the + // caller can compact and retry rather than hard-fail. + const classified = provider_mod.classifyHttpStatus(status, err_buf.items); + if (self.diag) |d| { + d.status_code = status; + d.retry_after_ms = provider_mod.retryAfterFromHead(response.head); } - return error.HttpError; + return classified; } // Stream the body through the SSE parser and event handler. @@ -172,7 +175,10 @@ pub const OpenAIChatRequest = struct { while (true) { const n = body_reader.readVec(&vecs) catch |err| switch (err) { error.EndOfStream => break, - else => return err, + // A transport read failure mid-stream (reset, TLS, timeout) + // before `[DONE]` means no assistant message was committed. + // Surface it as a retryable malformed-stream error. + else => return error.ProviderStreamMalformed, }; if (n == 0) continue; @@ -576,7 +582,7 @@ fn handleEvent( d.error_type, d.error_message, }); } - return error.StreamError; + return error.ProviderStreamMalformed; } if (!state.started and d.role != null) { @@ -636,6 +642,7 @@ const NoopReceiver = struct { .onBlockComplete = noopBlockComplete, .onMessageComplete = noopMsgComplete, .onError = noopErr, + .onProviderRetry = noopRetry, }; fn noopMsgStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} fn noopBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} @@ -644,6 +651,7 @@ const NoopReceiver = struct { fn noopBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} fn noopMsgComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} fn noopErr(_: *anyopaque, _: anyerror) void {} + fn noopRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} }; /// Feed a sequence of SSE event payloads through the state machine as if @@ -883,8 +891,11 @@ const RecordingReceiver = struct { .onBlockComplete = onBlockComplete, .onMessageComplete = onMessageComplete, .onError = onError, + .onProviderRetry = onProviderRetry, }; + fn onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} + fn record(self: *RecordingReceiver, s: []const u8) !void { const owned = try self.allocator.dupe(u8, s); try self.events.append(self.allocator, owned); diff --git a/libpanto/src/session.zig b/libpanto/src/session.zig index daa6cfa..0491157 100644 --- a/libpanto/src/session.zig +++ b/libpanto/src/session.zig @@ -208,6 +208,7 @@ pub const DiskResultPart = union(enum) { pub const DiskToolResultBlock = struct { tool_use_id: []const u8, // owned parts: []DiskResultPart, // owned + is_error: bool = false, pub fn deinit(self: DiskToolResultBlock, alloc: Allocator) void { alloc.free(self.tool_use_id); for (self.parts) |p| p.deinit(alloc); @@ -402,6 +403,12 @@ fn writeDiskBlock(s: *std.json.Stringify, block: DiskContentBlock) !void { try s.write("toolResult"); try s.objectField("toolUseId"); try s.write(b.tool_use_id); + // Persist the error marker only when set, so existing + // (success) tool-result logs serialize byte-identically. + if (b.is_error) { + try s.objectField("isError"); + try s.write(true); + } // `parts` is an array of {type:"text",text} and // {type:"image",mimeType,data} (data = inline base64). try s.objectField("parts"); @@ -613,7 +620,9 @@ fn parseDiskBlock(allocator: Allocator, obj: std.json.ObjectMap) ParseError!Disk const tuid = try dupeStringField(allocator, obj, "toolUseId"); errdefer allocator.free(tuid); const parts = try parseDiskResultParts(allocator, obj); - return .{ .tool_result = .{ .tool_use_id = tuid, .parts = parts } }; + // Missing `isError` in older logs defaults to false. + const is_err = readBool(obj, "isError"); + return .{ .tool_result = .{ .tool_use_id = tuid, .parts = parts, .is_error = is_err } }; } else if (std.mem.eql(u8, t, "compactionSummary")) { const text = try dupeStringField(allocator, obj, "text"); return .{ .compaction_summary = .{ .text = text } }; @@ -659,6 +668,12 @@ fn parseDiskResultParts(allocator: Allocator, obj: std.json.ObjectMap) ParseErro return list.toOwnedSlice(allocator); } +fn readBool(obj: std.json.ObjectMap, name: []const u8) bool { + const v = obj.get(name) orelse return false; + if (v != .bool) return false; + return v.bool; +} + fn readU64(obj: std.json.ObjectMap, name: []const u8) u64 { const v = obj.get(name) orelse return 0; if (v != .integer) return 0; @@ -729,7 +744,11 @@ pub fn contentBlockToDisk( }, } } - return .{ .tool_result = .{ .tool_use_id = tuid, .parts = try parts.toOwnedSlice(allocator) } }; + return .{ .tool_result = .{ + .tool_use_id = tuid, + .parts = try parts.toOwnedSlice(allocator), + .is_error = tr.is_error, + } }; }, // A `.System` block becomes a disk text block; its mode rides on // the enclosing `DiskMessage.mode` (set by the session manager), @@ -794,7 +813,7 @@ pub fn diskContentBlockToInternal( }, } } - return .{ .ToolResult = .{ .tool_use_id = tuid, .parts = parts } }; + return .{ .ToolResult = .{ .tool_use_id = tuid, .parts = parts, .is_error = b.is_error } }; }, .compaction_summary => |b| { const tb = try conversation.textualBlockFromSlice(allocator, b.text); @@ -959,6 +978,53 @@ test "serialize/parse tool result message entry" { try testing.expectEqual(@as(usize, 1), got.message.content[0].tool_result.parts.len); try testing.expectEqualStrings("file1.txt\nfile2.txt", got.message.content[0].tool_result.parts[0].text); try testing.expectEqualStrings("anthropic", got.provider.?); + // Unset is_error defaults to false and serializes without the field. + try testing.expect(!got.message.content[0].tool_result.is_error); + try testing.expect(std.mem.indexOf(u8, line, "isError") == null); +} + +test "serialize/parse tool result preserves is_error = true" { + const a = testing.allocator; + + var content = try a.alloc(DiskContentBlock, 1); + var trp = try a.alloc(DiskResultPart, 1); + trp[0] = .{ .text = try dupe(a, "file not found") }; + content[0] = .{ .tool_result = .{ + .tool_use_id = try dupe(a, "tool_err"), + .parts = trp, + .is_error = true, + } }; + + const entry: SessionEntry = .{ .message = .{ + .base = .{ + .id = try dupe(a, "e1"), + .parent_id = try dupe(a, "e0"), + .timestamp = try dupe(a, "2026-04-25T17:40:18.000Z"), + }, + .provider = try dupe(a, "anthropic"), + .model = try dupe(a, "claude-sonnet-4-20250514"), + .message = .{ .role = .user, .content = content }, + } }; + defer entry.deinit(a); + + const line = try serializeEntry(a, entry); + defer a.free(line); + try testing.expect(std.mem.indexOf(u8, line, "\"isError\":true") != null); + + var fe = try parseLine(a, line); + defer fe.deinit(a); + try testing.expect(fe.entry.message.message.content[0].tool_result.is_error); +} + +test "parse tool result without isError defaults to false" { + const a = testing.allocator; + // A legacy line predating the is_error field. + const line = + \\{"type":"message","id":"x","parentId":"y","timestamp":"t","provider":"anthropic","model":"m","message":{"role":"user","content":[{"type":"toolResult","toolUseId":"t1","parts":[{"type":"text","text":"ok"}]}]}} + ; + var fe = try parseLine(a, line); + defer fe.deinit(a); + try testing.expect(!fe.entry.message.message.content[0].tool_result.is_error); } test "serialize/parse tool result with text + image part round-trips" { diff --git a/libpanto/src/tool.zig b/libpanto/src/tool.zig index c912099..60a7736 100644 --- a/libpanto/src/tool.zig +++ b/libpanto/src/tool.zig @@ -84,11 +84,18 @@ pub const Tool = struct { /// agent takes ownership and frees the slice and every part (see /// `freeResultParts`). /// - /// Returning an error aborts the current turn. The agent surfaces - /// the error to the user. Native tool implementations are - /// responsible for catching their own panics — a panic in `invoke` - /// will crash the process. Adapters that bridge to safer languages - /// (Lua, Python, Go) should convert panics/exceptions into errors. + /// Returning an error normally becomes a model-visible error + /// `ToolResult`: the agent synthesizes an error result for this + /// call (and keeps the matching `ToolResult` for every other call + /// in the batch), then lets the model continue so it can correct + /// arguments, try another tool, or explain the failure. Only hard + /// host failures (`error.Canceled`, `error.OutOfMemory`) abort the + /// whole turn and propagate to the embedder. + /// + /// Native tool implementations are responsible for catching their + /// own panics — a panic in `invoke` will crash the process. + /// Adapters that bridge to safer languages (Lua, Python, Go) should + /// convert panics/exceptions into errors. invoke: *const fn ( ctx: *anyopaque, input: []const u8, diff --git a/libpanto/src/tool_source.zig b/libpanto/src/tool_source.zig index 4b9e104..bf1f5a1 100644 --- a/libpanto/src/tool_source.zig +++ b/libpanto/src/tool_source.zig @@ -77,11 +77,16 @@ pub const ToolSource = struct { /// slot. The source decides internal scheduling — sequential, /// coroutine fan-out, worker pool, etc. /// - /// On any uncaught error returned from this function (i.e. the - /// fn itself returning an error rather than recording one in a - /// `results[i]` slot), libpanto treats the *entire batch* as - /// failed: it frees any `ok` slots already filled and aborts the - /// turn with that error. + /// Two failure modes, both normally model-visible: + /// - Per-call: record `.{ .err = e }` in a `results[i]` slot. + /// That call gets an error `ToolResult`; siblings are + /// unaffected. + /// - Whole-batch: return an error from this function. libpanto + /// frees any `ok` slots already filled and maps the error onto + /// *every* member call as an error `ToolResult`. + /// In both cases the agent loop continues so the model can react. + /// Only hard host failures (`error.Canceled`, `error.OutOfMemory`) + /// abort the whole turn and propagate to the embedder. invoke_batch: *const fn ( ctx: *anyopaque, calls: []const Call, diff --git a/src/main.zig b/src/main.zig index be0ac86..61e78d5 100644 --- a/src/main.zig +++ b/src/main.zig @@ -93,8 +93,28 @@ const CLIReceiver = struct { .onBlockComplete = onBlockComplete, .onMessageComplete = onMessageComplete, .onError = onError, + .onProviderRetry = onProviderRetry, }; + /// Surface provider retry scheduling as a dim status line so the user + /// can see the agent is waiting on the provider rather than hung. + fn onProviderRetry(ptr: *anyopaque, info: panto.provider.ProviderRetryInfo) void { + const self: *CLIReceiver = @ptrCast(@alignCast(ptr)); + if (info.compaction) { + self.stdout.print( + "\x1b[2mcontext overflow: compacting and retrying\x1b[0m\n", + .{}, + ) catch {}; + } else { + const secs = @as(f64, @floatFromInt(info.delay_ms)) / 1000.0; + self.stdout.print( + "\x1b[2mprovider unavailable ({s}): retrying in {d:.1}s (attempt {d}/{d})\x1b[0m\n", + .{ @errorName(info.err), secs, info.attempt + 1, info.max_attempts }, + ) catch {}; + } + self.file.flush() catch {}; + } + /// The print-based CLI defers tool-name rendering to onBlockComplete /// to avoid cursor gymnastics (we can't go back and edit the prefix). /// Receivers backed by a TUI capable of in-place updates would render -- cgit v1.3