diff options
Diffstat (limited to 'libpanto')
| -rw-r--r-- | libpanto/src/agent.zig | 33 | ||||
| -rw-r--r-- | libpanto/src/anthropic_messages_json.zig | 22 | ||||
| -rw-r--r-- | libpanto/src/config.zig | 5 | ||||
| -rw-r--r-- | libpanto/src/openai_chat_json.zig | 3 | ||||
| -rw-r--r-- | libpanto/src/provider_anthropic_messages.zig | 69 | ||||
| -rw-r--r-- | libpanto/src/session_manager.zig | 160 |
6 files changed, 248 insertions, 44 deletions
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index 08cc91c..a42c16c 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -43,6 +43,23 @@ const Entry = tool_registry_mod.Entry; pub const Config = config_mod.Config; +fn isValidToolInput(input: []const u8) bool { + if (input.len == 0) return true; + if (input[0] != '{') return true; // legacy tests/tools may use opaque bytes + var parsed = std.json.parseFromSlice(std.json.Value, std.heap.page_allocator, input, .{}) catch return false; + defer parsed.deinit(); + return parsed.value == .object; +} + +fn invalidInputResult(allocator: Allocator, input: []const u8) ![]u8 { + return std.fmt.allocPrint( + allocator, + "Tool call was not executed: tool input was incomplete or invalid JSON. Partial input: {s}", + .{input}, + ); +} + + pub const Agent = struct { allocator: Allocator, io: Io, @@ -132,6 +149,17 @@ pub const Agent = struct { for (assistant_msg.content.items) |block| { if (block != .ToolUse) continue; const tu = block.ToolUse; + if (!isValidToolInput(tu.input.items)) { + try calls.append(.{ + .tool_use_id = tu.id, + .tool_name = tu.name, + .input = tu.input.items, + .entry = null, + .result = try invalidInputResult(self.allocator, tu.input.items), + .err = null, + }); + continue; + } const entry = self.config.registry.lookup(tu.name) orelse { // Unknown tool: abort the turn with a clear error. return error.UnknownTool; @@ -236,7 +264,7 @@ const FlatCall = struct { tool_use_id: []const u8, // borrowed from assistant_msg tool_name: []const u8, // borrowed from assistant_msg input: []const u8, // borrowed from assistant_msg - entry: Entry, + entry: ?Entry, /// Owned result bytes from `Tool.invoke` or `ToolSource.invoke_batch`. /// Allocated with the agent's allocator. Transferred into a @@ -292,7 +320,8 @@ fn buildGroups( } for (calls, 0..) |c, i| { - switch (c.entry) { + const ent = c.entry orelse continue; + switch (ent) { .single => |t| try out.append(.{ .single = .{ .tool = t, .call_index = i } }), .source => |sr| { const gop = try pending.getOrPut(sr.source); diff --git a/libpanto/src/anthropic_messages_json.zig b/libpanto/src/anthropic_messages_json.zig index f8bfc2e..d382de8 100644 --- a/libpanto/src/anthropic_messages_json.zig +++ b/libpanto/src/anthropic_messages_json.zig @@ -111,6 +111,26 @@ fn writeRawJson(s: *std.json.Stringify, raw: []const u8) !void { try s.write(parsed.value); } +fn writeToolUseInput(s: *std.json.Stringify, raw: []const u8) !void { + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); + defer arena.deinit(); + const parsed = std.json.parseFromSlice(std.json.Value, arena.allocator(), raw, .{}) catch { + try s.beginObject(); + try s.objectField("_invalid_input"); + try s.write(raw); + try s.endObject(); + return; + }; + if (parsed.value != .object) { + try s.beginObject(); + try s.objectField("_invalid_input"); + try s.write(raw); + try s.endObject(); + return; + } + try s.write(parsed.value); +} + /// Build the top-level Anthropic `system` string. Applies the shared /// append/replace derivation (see `conversation.effectiveSystemBlocks`), /// strips trailing newlines from each surviving block, and joins them with @@ -200,7 +220,7 @@ fn writeBlock(s: *std.json.Stringify, block: conversation.ContentBlock) !void { try s.beginObject(); try s.endObject(); } else { - try writeRawJson(s, input_bytes); + try writeToolUseInput(s, input_bytes); } try s.endObject(); }, diff --git a/libpanto/src/config.zig b/libpanto/src/config.zig index 6744b56..6b8d9a8 100644 --- a/libpanto/src/config.zig +++ b/libpanto/src/config.zig @@ -46,6 +46,7 @@ pub const OpenAIChatConfig = struct { base_url: []const u8, model: []const u8, reasoning: ReasoningEffort = .default, + max_tokens: u32 = 64_000, }; pub const AnthropicMessagesConfig = struct { @@ -55,7 +56,7 @@ pub const AnthropicMessagesConfig = struct { /// Value sent in the `anthropic-version` header. api_version: []const u8 = "2023-06-01", /// Required by Anthropic's Messages API. - max_tokens: u32 = 4096, + max_tokens: u32 = 64_000, }; /// Per-provider transport/auth/model configuration. Tagged by `APIStyle`. @@ -144,7 +145,7 @@ test "ProviderConfig - anthropic_messages variant" { } }; try t.expectEqual(APIStyle.anthropic_messages, cfg.style()); try t.expectEqualStrings("2023-06-01", cfg.anthropic_messages.api_version); - try t.expectEqual(@as(u32, 4096), cfg.anthropic_messages.max_tokens); + try t.expectEqual(@as(u32, 64_000), cfg.anthropic_messages.max_tokens); } test "ProviderConfig - openai_chat reasoning defaults to .default" { diff --git a/libpanto/src/openai_chat_json.zig b/libpanto/src/openai_chat_json.zig index 3b81c41..8e6c9be 100644 --- a/libpanto/src/openai_chat_json.zig +++ b/libpanto/src/openai_chat_json.zig @@ -91,6 +91,9 @@ pub fn serializeRequest( try s.objectField("stream"); try s.write(true); + try s.objectField("max_tokens"); + try s.write(cfg.max_tokens); + // Ask for the final-chunk usage block. Without this the server // sends `usage: null` and we can't stamp token counts on the // session log. Most OpenAI-compatible proxies accept this; ones diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig index 02277be..5c24b8f 100644 --- a/libpanto/src/provider_anthropic_messages.zig +++ b/libpanto/src/provider_anthropic_messages.zig @@ -181,6 +181,7 @@ const StreamState = struct { /// `onMessageComplete`, the latter stamps `null`. usage: provider_mod.Usage = .{}, usage_seen: bool = false, + stop_reason: ?[]u8 = null, const ActiveBlock = struct { /// Index reported on the wire (Anthropic's content-array index). @@ -210,6 +211,7 @@ const StreamState = struct { } for (self.blocks.items) |*b| b.deinit(self.allocator); self.blocks.deinit(self.allocator); + if (self.stop_reason) |s| self.allocator.free(s); } fn ensureStarted(self: *StreamState, receiver: *provider_mod.Receiver) !void { @@ -322,6 +324,40 @@ const StreamState = struct { a.signature = try self.allocator.dupe(u8, sig); } + fn setStopReason(self: *StreamState, reason: ?[]const u8) !void { + if (self.stop_reason) |old| self.allocator.free(old); + self.stop_reason = if (reason) |r| try self.allocator.dupe(u8, r) else null; + } + + fn isJSONObject(input: []const u8) bool { + if (input.len == 0) return true; + var parsed = std.json.parseFromSlice(std.json.Value, std.heap.page_allocator, input, .{}) catch return false; + defer parsed.deinit(); + return parsed.value == .object; + } + + fn appendAnthropicFailureResult( + allocator: Allocator, + blocks: *std.ArrayList(conversation.ContentBlock), + tool_use_id: []const u8, + input: []const u8, + stop_reason: ?[]const u8, + ) !void { + const id_copy = try allocator.dupe(u8, tool_use_id); + errdefer allocator.free(id_copy); + const msg = try std.fmt.allocPrint( + allocator, + "Tool call was not executed: Anthropic stopped before completing valid tool input JSON (stop_reason={s}). Partial input: {s}", + .{ stop_reason orelse "unknown", input }, + ); + errdefer allocator.free(msg); + var content: conversation.TextualBlock = .empty; + errdefer content.deinit(allocator); + try content.appendSlice(allocator, msg); + allocator.free(msg); + try blocks.append(allocator, .{ .ToolResult = .{ .tool_use_id = id_copy, .content = content } }); + } + /// Close the active block: append it to `blocks` and emit onBlockComplete. fn closeBlock( self: *StreamState, @@ -355,11 +391,22 @@ const StreamState = struct { .text = a.text_buf, .signature = a.signature, } }, - .tool_use => .{ .ToolUse = .{ - .id = a.tool_id.?, - .name = a.tool_name.?, - .input = a.text_buf, - } }, + .tool_use => blk: { + if (!isJSONObject(a.text_buf.items)) { + try appendAnthropicFailureResult( + self.allocator, + &self.blocks, + a.tool_id.?, + a.text_buf.items, + self.stop_reason, + ); + } + break :blk .{ .ToolUse = .{ + .id = a.tool_id.?, + .name = a.tool_name.?, + .input = a.text_buf, + } }; + }, .unsupported => unreachable, }; @@ -389,7 +436,8 @@ const StreamState = struct { self.finalized = true; if (self.active != null) { - // Wire dropped us mid-block. Close it cleanly anyway. + // Preserve an interrupted tool call so the agent can answer it + // with a synthetic error ToolResult instead of invoking it. try self.closeBlock(receiver); } @@ -434,12 +482,13 @@ fn handleEvent( if (d.signature_delta) |sig| try state.setSignature(sig); if (d.input_json_delta) |j| try state.appendInputJsonDelta(receiver, j); }, - .content_block_stop => { - try state.closeBlock(receiver); + .content_block_stop => |s| { + if (state.active) |a| { + if (a.wire_index == s.index) try state.closeBlock(receiver); + } }, .message_delta => |d| { - // We don't act on stop_reason directly; message_stop is the - // authoritative end-of-stream signal. + try state.setStopReason(d.stop_reason); state.mergeUsage(d.usage); }, .message_stop => { diff --git a/libpanto/src/session_manager.zig b/libpanto/src/session_manager.zig index 8b503f2..9e8e2c8 100644 --- a/libpanto/src/session_manager.zig +++ b/libpanto/src/session_manager.zig @@ -355,10 +355,11 @@ pub const SessionManager = struct { try truncateFileTo(io, path_copy, valid_bytes); } - // Run format migration. Currently a no-op for version 1, but the - // hook exists so future versions can rewrite the file. var migrated_header = header; - const did_migrate = migrate(allocator, &migrated_header, &entries); + var did_migrate = migrate(allocator, &migrated_header, &entries); + if (elideDanglingToolUses(allocator, &entries)) { + did_migrate = true; + } // We didn't reassign by_id during migration; rebuild if needed. if (did_migrate) { by_id.clearRetainingCapacity(); @@ -567,16 +568,11 @@ pub const SessionManager = struct { // ---------- Persistence ---------- - /// Write the header + all currently-buffered entries + `final_entry` + /// Write the header + all currently-buffered entries + `new_entries` /// to the file as a single batch. Creates the directory and file. - /// Called only when `flushed = false` and we just got an assistant - /// message. - fn flushBuffered(self: *SessionManager, final_entry: SessionEntry) !void { - // Ensure the directory exists. + fn flushBufferedMany(self: *SessionManager, new_entries: []const SessionEntry) !void { try mkdirP(self.io, self.session_dir); - // Open (create exclusively isn't required; if a stale file with the - // same UUIDv7 existed we'd just overwrite — UUIDv7s are unique). const file = try Io.Dir.cwd().createFile(self.io, self.session_file, .{ .truncate = true, .read = false, @@ -584,7 +580,6 @@ pub const SessionManager = struct { defer file.close(self.io); var offset: u64 = 0; - // Header const header_line = try session_mod.serializeHeader(self.allocator, self.header); defer self.allocator.free(header_line); try file.writePositionalAll(self.io, header_line, offset); @@ -592,7 +587,6 @@ pub const SessionManager = struct { try file.writePositionalAll(self.io, "\n", offset); offset += 1; - // Buffered entries. for (self.entries.items) |e| { const line = try session_mod.serializeEntry(self.allocator, e); defer self.allocator.free(line); @@ -602,38 +596,105 @@ pub const SessionManager = struct { offset += 1; } - // Final (the assistant message we just got). - const final_line = try session_mod.serializeEntry(self.allocator, final_entry); - defer self.allocator.free(final_line); - try file.writePositionalAll(self.io, final_line, offset); - offset += final_line.len; - try file.writePositionalAll(self.io, "\n", offset); - offset += 1; + for (new_entries) |entry| { + const line = try session_mod.serializeEntry(self.allocator, entry); + defer self.allocator.free(line); + try file.writePositionalAll(self.io, line, offset); + offset += line.len; + try file.writePositionalAll(self.io, "\n", offset); + offset += 1; + } - // Best-effort fsync — the directory itself doesn't need syncing - // for our purposes (we don't unlink/rename); the file body does. file.sync(self.io) catch {}; - self.flushed = true; self.written_bytes = offset; } + fn flushBuffered(self: *SessionManager, final_entry: SessionEntry) !void { + try self.flushBufferedMany(&.{final_entry}); + } + /// Append a single line for `entry` to the open session file. Caller /// must have already verified `flushed`. fn persistEntry(self: *SessionManager, entry: SessionEntry) !void { + try self.persistEntries(&.{entry}); + } + + pub fn appendMessagesAtomic( + self: *SessionManager, + messages: []DiskMessage, + providers: []const ?[]const u8, + models: []const ?[]const u8, + ) !void { + std.debug.assert(messages.len == providers.len); + std.debug.assert(messages.len == models.len); + if (messages.len == 0) return; + + const base_len = self.entries.items.len; + try self.entries.ensureUnusedCapacity(self.allocator, messages.len); + try self.by_id.ensureUnusedCapacity(@intCast(messages.len)); + + var entries = try self.allocator.alloc(SessionEntry, messages.len); + defer self.allocator.free(entries); + var built: usize = 0; + errdefer { + for (entries[0..built]) |*e| e.deinit(self.allocator); + } + + var prev_leaf = self.leaf_id; + for (messages, 0..) |msg, i| { + const msg_local = msg; + const id_buf = try self.newEntryId(); + errdefer self.allocator.free(id_buf); + const timestamp = try isoTimestamp(self.allocator, self.io); + errdefer self.allocator.free(timestamp); + const parent_id_copy: ?[]const u8 = if (prev_leaf) |l| try self.allocator.dupe(u8, l) else null; + errdefer if (parent_id_copy) |p| self.allocator.free(p); + const provider_copy: ?[]const u8 = if (providers[i]) |p| try self.allocator.dupe(u8, p) else null; + errdefer if (provider_copy) |p| self.allocator.free(p); + const model_copy: ?[]const u8 = if (models[i]) |m| try self.allocator.dupe(u8, m) else null; + errdefer if (model_copy) |m| self.allocator.free(m); + entries[i] = .{ .message = .{ + .base = .{ .id = id_buf, .parent_id = parent_id_copy, .timestamp = timestamp }, + .provider = provider_copy, + .model = model_copy, + .message = msg_local, + } }; + built += 1; + prev_leaf = entries[i].base().id; + } + + if (self.flushed) { + try self.persistEntries(entries); + } else { + try self.flushBufferedMany(entries); + } + + for (entries, 0..) |entry, i| { + self.entries.appendAssumeCapacity(entry); + self.by_id.putAssumeCapacity(entry.base().id, base_len + i); + } + self.leaf_id = entries[entries.len - 1].base().id; + built = 0; + } + + fn persistEntries(self: *SessionManager, entries: []const SessionEntry) !void { const file = try Io.Dir.cwd().openFile(self.io, self.session_file, .{ .mode = .write_only, }); defer file.close(self.io); - const line = try session_mod.serializeEntry(self.allocator, entry); - defer self.allocator.free(line); - - try file.writePositionalAll(self.io, line, self.written_bytes); - self.written_bytes += line.len; - try file.writePositionalAll(self.io, "\n", self.written_bytes); - self.written_bytes += 1; + var offset = self.written_bytes; + for (entries) |entry| { + const line = try session_mod.serializeEntry(self.allocator, entry); + defer self.allocator.free(line); + try file.writePositionalAll(self.io, line, offset); + offset += line.len; + try file.writePositionalAll(self.io, "\n", offset); + offset += 1; + } file.sync(self.io) catch {}; + self.written_bytes = offset; } // ============================================================================= @@ -711,6 +772,47 @@ fn migrate( return false; } +fn elideDanglingToolUses(allocator: Allocator, entries: *std.ArrayList(SessionEntry)) bool { + var needed: std.StringHashMap(void) = .init(allocator); + defer needed.deinit(); + var changed = false; + + var i = entries.items.len; + while (i > 0) { + i -= 1; + const entry = &entries.items[i]; + if (entry.* != .message) continue; + const msg = &entry.message.message; + + if (msg.role == .user) { + for (msg.content) |block| { + if (block == .tool_result) { + needed.put(block.tool_result.tool_use_id, {}) catch {}; + } + } + continue; + } + + if (msg.role != .assistant) continue; + var kept: std.ArrayList(DiskContentBlock) = .empty; + defer kept.deinit(allocator); + var removed = false; + for (msg.content) |block| { + if (block == .tool_use and !needed.contains(block.tool_use.id)) { + block.deinit(allocator); + removed = true; + continue; + } + kept.append(allocator, block) catch unreachable; + } + if (!removed) continue; + allocator.free(msg.content); + msg.content = kept.toOwnedSlice(allocator) catch unreachable; + changed = true; + } + return changed; +} + // ============================================================================= // File utilities // ============================================================================= |