diff options
| -rw-r--r-- | README.md | 114 | ||||
| -rw-r--r-- | build.zig | 4 | ||||
| -rw-r--r-- | build.zig.zon | 2 | ||||
| -rw-r--r-- | docs/phase-1.md | 2 | ||||
| -rw-r--r-- | docs/phase-4.md | 26 | ||||
| -rw-r--r-- | libpanto/src/config.zig | 46 | ||||
| -rw-r--r-- | libpanto/src/json.zig | 4 | ||||
| -rw-r--r-- | libpanto/src/openai_chat_json.zig | 324 | ||||
| -rw-r--r-- | libpanto/src/provider.zig | 47 | ||||
| -rw-r--r-- | libpanto/src/provider_openai.zig | 6 | ||||
| -rw-r--r-- | libpanto/src/provider_openai_chat.zig | 443 | ||||
| -rw-r--r-- | libpanto/src/root.zig | 11 | ||||
| -rw-r--r-- | libpanto/src/sse.zig | 137 | ||||
| -rw-r--r-- | pantograph-for-studio-rail-system-200-cm.jpg | bin | 0 -> 123641 bytes | |||
| -rw-r--r-- | src/main.zig | 179 |
15 files changed, 1190 insertions, 155 deletions
diff --git a/README.md b/README.md new file mode 100644 index 0000000..de4c962 --- /dev/null +++ b/README.md @@ -0,0 +1,114 @@ +# pantograph: a minimal coding agent + +``` + ?+ + `?] + ]?~ + [?? + [?? + ,?]" + '. ??- + 1111}}}}}}!'' ??? + |(([[[[\[u]]f]]'` {?? + 1((?]][[)tz)[]?' + ]Q?/|)?????v??-??_. + ]Ur )))?]?????n??[)Y'_ + ?]|: `\t-??: + ]]] --- + ]]] ]?-^ + ]]]. -?- + ']]+ ~]-` + [\] --? + [x] "]?` + {]]I ??? + []? '??l + )Y})}}[+ ?-? + \l//[]???-j]u??- .?-? + ^ l\\--??]??]]??_. ??? + i|\---?]]}t>.?~ + )??]}]]]]f?C?], + I??? `f11]?]]}]n][?[ + ?fc ^)1]??]?]Y]}} + ^??? '{{] + ??Y + ^]?? + ]]]. + ^]]] + ]~x' + `]J\ + [}[` + `[X[ + }}[` + .}}[ + /p]` + '1t} +``` + +A pantograph is a draftsman's instrument: a jointed parallelogram of four +rods, pinned at one end to a fixed pivot. The draftsman moves a stylus over +the original drawing; a pen at the far end of the linkage traces the same +path, scaled. Small movements at the stylus become larger movements at the +pen, in exact proportion, without the draftsman ever having to think about +the ratio. The geometry does the work. + +What the tool offers is *controlled leverage*. The operator stays close to +the source — eyes on the original, hand on the stylus — and the mechanism +reproduces that intent at a different scale, faithfully, without slop. The +linkage doesn't make decisions; it amplifies the ones you make. + +That is the goal for `pantograph`: a coding agent built as a precise +linkage between operator and model. Your attention stays on the work; the +tool turns small, deliberate movements into larger ones, in proportion, and +adds nothing of its own. + +## The shape of it + +- **Core in Zig.** One binary, small footprint, fast startup. The agent + loop and nothing else. +- **`libpanto`.** The same loop behind a C ABI, so other programs can + embed it instead of shelling out. +- **Two provider shapes.** Anthropic-shaped and OpenAI-shaped APIs at + arbitrary base URLs. Both work all the way through, or they aren't + shipped. +- **Server mode.** Run as a daemon that speaks those same two API shapes + itself, routing to its configured backends. A small, durable provider + router; the useful slice of something like `omniroute`, without the + memory cost or the fragile bits. + +## What is not in the core + +A coding agent accumulates features the way a train roof accumulates +ironwork. `pantograph` resists that. The following are deliberately *not* +built in: + +- subagents +- MCP +- permission systems +- AGENTS.md automation +- skills +- customizable `/prompts` +- the basic tools — `read`, `write`, `edit`, `bash` — even these ship as + extensions and can be disabled individually + +Each of these is a reasonable thing to want. None of them needs to live in +the core. + +## Extensions + +Extensions are how everything above gets back into the picture, on terms +that don't compromise the rest of the machine. + +- **Lua first.** Cheap to embed, quick to iterate in, easy to sandbox. +- **Native next.** A C ABI for shared-object extensions, so anything that + can produce a `.so` — Zig, Rust, C, C++ — can plug in. +- **Crash isolation as a first-class concern.** A bad extension should + not take the host down with it. The standard tools will themselves be + ported to native extensions once that path is solid. + +Extensions are how `pantograph` stays small without becoming useless. + +## Status + +Early. The thinking lives in `ideas.md`; the staged plan lives in +`docs/phase-1.md` through `docs/phase-4.md`. Source is under `src/` for +the CLI and `libpanto/` for the embeddable library. @@ -49,4 +49,8 @@ pub fn build(b: *std.Build) void { const run_unit_tests = b.addRunArtifact(unit_tests); const test_step = b.step("test", "Run unit tests"); test_step.dependOn(&run_unit_tests.step); + + // Also run libpanto's own tests as part of `zig build test` + const lib_test_step = panto_lib_dep.builder.top_level_steps.get("test").?; + test_step.dependOn(&lib_test_step.step); } diff --git a/build.zig.zon b/build.zig.zon index 57dc3b5..5531d78 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -4,7 +4,7 @@ .version = "0.0.0", .dependencies = .{ .panto = .{ - .path = "panto", + .path = "libpanto", }, }, .paths = .{ diff --git a/docs/phase-1.md b/docs/phase-1.md index 8acaf42..6b149aa 100644 --- a/docs/phase-1.md +++ b/docs/phase-1.md @@ -1,5 +1,7 @@ # Phase 1: libpanto — Minimal Chat Library +**Status: complete.** Streaming chat works end-to-end against OpenAI-compatible APIs; conversation history persists across turns; thinking blocks (`reasoning_content` / `reasoning`) are streamed; `reasoning_effort` is configurable. Open questions resolved: thinking support implemented; mid-stream errors propagate via Zig errors; connections are one-per-turn intentionally (uniform reentry into each turn); long-conversation memory deferred to a later phase. Session persistence is phase 4. + ## Goal A Zig library that can hold a streaming conversation with an LLM via an OpenAI-compatible API. No tools, no extensions — just chat. Includes a minimal CLI for live testing. diff --git a/docs/phase-4.md b/docs/phase-4.md index af68cca..6c4f9b9 100644 --- a/docs/phase-4.md +++ b/docs/phase-4.md @@ -32,7 +32,31 @@ A session persistence system where pantograph saves and resumes conversations. A - Custom entry types (for extensions — future phase) - Labels, bookmarks, session naming - Session export or format migration beyond version 1 -- In-memory-only sessions (every session is persisted) +- In-memory-only sessions from the CLI (every CLI session is persisted; libpanto embedders can opt out by passing a no-op store) + +--- + +## Library / CLI Boundary + +The storage interface lives in libpanto. The default `fs_jsonl` implementation also lives in libpanto. The CLI decides _where_ a session is stored and constructs the store with that path. + +### libpanto owns + +- The `SessionStore` interface (vtable: append entry, load entries, etc.). +- An `fs_jsonl` implementation that takes a base directory at construction time and writes the JSONL format defined below. +- The agent loop's interaction with the store: which events get appended, at what points in a turn. + +### CLI owns + +- Selecting the base directory: XDG resolution (`$XDG_DATA_HOME` or `~/.local/share`), the `panto/sessions/` subpath, and the per-project `<encoded-cwd>/` grouping. +- Selecting the session file: new-file-per-invocation by default; `--resume` / `--resume <id>` to pick an existing file. +- Constructing `fs_jsonl` with the resolved directory and passing it to libpanto. + +Other embedders (a TUI, a daemon, a Lua host, a test harness) bring their own directory policy or their own `SessionStore` implementation. A no-op store is a valid choice for embedders that don't want persistence. + +### Mid-turn writes + +The in-memory `Conversation` is the source of truth during a turn. The store is written between discrete events — never as an inline dependency of streaming. A crash mid-turn loses the in-flight assistant message but never corrupts prior state. (Phase 4 writes per-message; finer-grained streaming-event entries are out of scope.) --- diff --git a/libpanto/src/config.zig b/libpanto/src/config.zig index 295f2c6..74bdcc7 100644 --- a/libpanto/src/config.zig +++ b/libpanto/src/config.zig @@ -1,5 +1,20 @@ pub const APIStyle = enum { - anthropic, + openai_chat, +}; + +/// Reasoning intensity hint sent to providers that support it. +/// +/// `.default` omits the field entirely so the provider's own default applies. +/// `.off` sends `"none"` (supported by OpenRouter/NanoGPT; ignored or rejected +/// elsewhere). The remaining values mirror the `reasoning_effort` parameter +/// accepted by OpenAI reasoning models and OpenAI-compatible proxies. +pub const ReasoningEffort = enum { + default, + off, + minimal, + low, + medium, + high, }; pub const Config = struct { @@ -7,19 +22,32 @@ pub const Config = struct { api_key: []const u8, base_url: []const u8, model: []const u8, + reasoning: ReasoningEffort = .default, }; const t = @import("std").testing; test "Config - construct with all fields" { const cfg = Config{ - .api_style = .anthropic, - .api_key = "sk-ant-test", - .base_url = "https://api.anthropic.com", - .model = "claude-sonnet-4-20250514", + .api_style = .openai_chat, + .api_key = "sk-test", + .base_url = "https://api.openai.com/v1", + .model = "gpt-4o", + .reasoning = .high, + }; + try t.expectEqual(APIStyle.openai_chat, cfg.api_style); + try t.expectEqualStrings("sk-test", cfg.api_key); + try t.expectEqualStrings("https://api.openai.com/v1", cfg.base_url); + try t.expectEqualStrings("gpt-4o", cfg.model); + try t.expectEqual(ReasoningEffort.high, cfg.reasoning); +} + +test "Config - reasoning defaults to .default" { + const cfg = Config{ + .api_style = .openai_chat, + .api_key = "k", + .base_url = "u", + .model = "m", }; - try t.expectEqual(APIStyle.anthropic, cfg.api_style); - try t.expectEqualStrings("sk-ant-test", cfg.api_key); - try t.expectEqualStrings("https://api.anthropic.com", cfg.base_url); - try t.expectEqualStrings("claude-sonnet-4-20250514", cfg.model); + try t.expectEqual(ReasoningEffort.default, cfg.reasoning); } diff --git a/libpanto/src/json.zig b/libpanto/src/json.zig deleted file mode 100644 index 542391e..0000000 --- a/libpanto/src/json.zig +++ /dev/null @@ -1,4 +0,0 @@ -// Serialization helpers — model → wire JSON, deltas → ContentBlocks. -// Implementation pending; this module will handle: -// 1. Serialize Conversation → OpenAI request body JSON -// 2. Parse SSE chunk deltas → ContentBlock updates diff --git a/libpanto/src/openai_chat_json.zig b/libpanto/src/openai_chat_json.zig new file mode 100644 index 0000000..79cccac --- /dev/null +++ b/libpanto/src/openai_chat_json.zig @@ -0,0 +1,324 @@ +//! OpenAI Chat Completions JSON serialization and parsing. +//! +//! Two responsibilities: +//! 1. Serialize a `Conversation` into the OpenAI Chat Completions request body. +//! 2. Parse one streaming SSE event's JSON payload into a strongly-typed +//! `StreamDelta` that the provider can consume. + +const std = @import("std"); +const Allocator = std.mem.Allocator; +const Writer = std.Io.Writer; +const conversation = @import("conversation.zig"); +const config_mod = @import("config.zig"); + +/// A single parsed streaming chunk. Fields are populated only when present +/// in the wire payload; null fields signal "not in this chunk". +/// +/// `content` and `reasoning_content` slices are borrowed from the parsed +/// JSON value, which is owned by the caller's `std.json.Parsed`. +pub const StreamDelta = struct { + role: ?[]const u8 = null, + content: ?[]const u8 = null, + reasoning_content: ?[]const u8 = null, + finish_reason: ?[]const u8 = null, +}; + +/// Serialize a Conversation into a `chat/completions` request body. +/// +/// The caller owns the returned slice (allocated with `allocator`). +pub fn serializeRequest( + allocator: Allocator, + cfg: *const config_mod.Config, + conv: *const conversation.Conversation, +) ![]u8 { + var aw: Writer.Allocating = .init(allocator); + errdefer aw.deinit(); + + var s: std.json.Stringify = .{ .writer = &aw.writer }; + + try s.beginObject(); + + try s.objectField("model"); + try s.write(cfg.model); + + try s.objectField("stream"); + try s.write(true); + + switch (cfg.reasoning) { + .default => {}, + .off => { + try s.objectField("reasoning_effort"); + try s.write("none"); + }, + .minimal, .low, .medium, .high => |eff| { + try s.objectField("reasoning_effort"); + try s.write(@tagName(eff)); + }, + } + + try s.objectField("messages"); + try s.beginArray(); + for (conv.messages.items) |msg| { + try writeMessage(&s, msg, allocator); + } + try s.endArray(); + + try s.endObject(); + + return try aw.toOwnedSlice(); +} + +fn writeMessage(s: *std.json.Stringify, msg: conversation.Message, allocator: Allocator) !void { + try s.beginObject(); + + try s.objectField("role"); + try s.write(@tagName(msg.role)); + + // All roles flatten to a plain `content` string in outbound requests. + // Thinking blocks are intentionally dropped from history: the openai_chat + // dialect has no portable way to round-trip them (OpenAI/DeepSeek strip; + // OpenRouter/NanoGPT want a `reasoning` field instead of an inline block; + // none accept the `{"type":"thinking",...}` shape). Preserving reasoning + // across turns will land with tool-use in a later phase. + try s.objectField("content"); + var buf: std.ArrayList(u8) = .empty; + defer buf.deinit(allocator); + try concatTextBlocks(msg.content.items, &buf, allocator); + try s.write(buf.items); + + try s.endObject(); +} + +fn concatTextBlocks( + blocks: []const conversation.ContentBlock, + out: *std.ArrayList(u8), + allocator: Allocator, +) !void { + for (blocks) |block| { + switch (block) { + .Text => |tb| try out.appendSlice(allocator, tb.items), + // Thinking: dropped. ToolUse/ToolResult: phase 3+. + else => {}, + } + } +} + +/// Parse a single SSE event payload (the JSON object that follows "data: "). +/// +/// Returns a `StreamDelta` borrowed from `parsed`. The caller must keep +/// `parsed` alive for as long as the delta's slices are in use, then call +/// `parsed.deinit()`. +pub const ParsedDelta = struct { + parsed: std.json.Parsed(std.json.Value), + delta: StreamDelta, + + pub fn deinit(self: *ParsedDelta) void { + self.parsed.deinit(); + } +}; + +pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedDelta { + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, payload, .{}); + errdefer parsed.deinit(); + + var delta: StreamDelta = .{}; + + const root = parsed.value; + if (root != .object) return .{ .parsed = parsed, .delta = delta }; + + const choices_v = root.object.get("choices") orelse return .{ .parsed = parsed, .delta = delta }; + if (choices_v != .array or choices_v.array.items.len == 0) { + return .{ .parsed = parsed, .delta = delta }; + } + const choice = choices_v.array.items[0]; + if (choice != .object) return .{ .parsed = parsed, .delta = delta }; + + if (choice.object.get("finish_reason")) |fr| { + if (fr == .string) delta.finish_reason = fr.string; + } + + if (choice.object.get("delta")) |d| { + if (d == .object) { + if (d.object.get("role")) |r| { + if (r == .string) delta.role = r.string; + } + if (d.object.get("content")) |c| { + if (c == .string) delta.content = c.string; + } + // Reasoning content lives under one of these names depending on + // the provider. We accept either. + if (d.object.get("reasoning_content")) |rc| { + if (rc == .string) delta.reasoning_content = rc.string; + } else if (d.object.get("reasoning")) |rc| { + if (rc == .string) delta.reasoning_content = rc.string; + } + } + } + + return .{ .parsed = parsed, .delta = delta }; +} + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; + +fn testConfig(model: []const u8) config_mod.Config { + return .{ + .api_style = .openai_chat, + .api_key = "k", + .base_url = "u", + .model = model, + }; +} + +test "serializeRequest - system + user" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + try conv.addSystemMessage("You are helpful."); + try conv.addUserMessage("Hello!"); + + const cfg = testConfig("gpt-4o"); + const body = try serializeRequest(allocator, &cfg, &conv); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const root = parsed.value.object; + try testing.expectEqualStrings("gpt-4o", root.get("model").?.string); + try testing.expect(root.get("stream").?.bool); + // reasoning_effort is omitted when set to .default. + try testing.expect(root.get("reasoning_effort") == null); + + const msgs = root.get("messages").?.array.items; + try testing.expectEqual(@as(usize, 2), msgs.len); + try testing.expectEqualStrings("system", msgs[0].object.get("role").?.string); + try testing.expectEqualStrings("You are helpful.", msgs[0].object.get("content").?.string); + try testing.expectEqualStrings("user", msgs[1].object.get("role").?.string); + try testing.expectEqualStrings("Hello!", msgs[1].object.get("content").?.string); +} + +test "serializeRequest - assistant Thinking blocks are stripped from outbound history" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + try conv.addAssistantMessage(&.{ + .{ .Thinking = try conversation.textualBlockFromSlice(allocator, "thinking step") }, + .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer here") }, + }); + + const cfg = testConfig("gpt-4o"); + const body = try serializeRequest(allocator, &cfg, &conv); + 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]; + try testing.expectEqualStrings("assistant", msg.object.get("role").?.string); + // Content is a flat string, only the Text block survives. + const content = msg.object.get("content").?.string; + try testing.expectEqualStrings("answer here", content); +} + +test "serializeRequest - reasoning effort level included when set" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("Hi"); + + var cfg = testConfig("gpt-4o"); + cfg.reasoning = .high; + + const body = try serializeRequest(allocator, &cfg, &conv); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + try testing.expectEqualStrings( + "high", + parsed.value.object.get("reasoning_effort").?.string, + ); +} + +test "serializeRequest - reasoning .off sends \"none\"" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("Hi"); + + var cfg = testConfig("gpt-4o"); + cfg.reasoning = .off; + + const body = try serializeRequest(allocator, &cfg, &conv); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + try testing.expectEqualStrings( + "none", + parsed.value.object.get("reasoning_effort").?.string, + ); +} + +test "parseStreamEvent - role only" { + const allocator = testing.allocator; + const payload = + \\{"id":"chatcmpl-1","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]} + ; + + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("assistant", pd.delta.role.?); + try testing.expect(pd.delta.content == null); + try testing.expect(pd.delta.finish_reason == null); +} + +test "parseStreamEvent - content delta" { + const allocator = testing.allocator; + const payload = + \\{"choices":[{"delta":{"content":"Hello"},"finish_reason":null}]} + ; + + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("Hello", pd.delta.content.?); +} + +test "parseStreamEvent - finish_reason stop" { + const allocator = testing.allocator; + const payload = + \\{"choices":[{"delta":{},"finish_reason":"stop"}]} + ; + + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("stop", pd.delta.finish_reason.?); + try testing.expect(pd.delta.content == null); +} + +test "parseStreamEvent - reasoning_content" { + const allocator = testing.allocator; + const payload = + \\{"choices":[{"delta":{"reasoning_content":"hmm"}}]} + ; + + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("hmm", pd.delta.reasoning_content.?); +} diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig index 8d6eea5..16c3dd6 100644 --- a/libpanto/src/provider.zig +++ b/libpanto/src/provider.zig @@ -14,36 +14,53 @@ pub const BlockMeta = struct { tool_name: ?[]const u8 = null, }; +/// Vtable for receiving streaming events from a Provider. +/// +/// The lifecycle callbacks (`onMessageStart` ... `onMessageComplete`) return +/// `anyerror!void`. Returning an error aborts the in-flight turn: the Provider +/// stops streaming, calls `onError(err)`, and propagates `err` out of +/// `streamStep`. No partial assistant message is appended to the conversation. +/// +/// `onError` is the receiver's cleanup hook. It fires exactly once per failed +/// turn, whether the error originated in the receiver itself (a write failure) +/// or in the Provider (HTTP/parse/stream failure). It is the last callback the +/// receiver will see for that turn. `onError` itself cannot fail; receivers +/// must swallow secondary failures during cleanup. pub const ReceiverVTable = struct { - onMessageStart: *const fn (*anyopaque, conversation.MessageRole) void, - onBlockStart: *const fn (*anyopaque, ContentBlockType, usize, ?BlockMeta) void, - onContentDelta: *const fn (*anyopaque, usize, []const u8) void, - onBlockComplete: *const fn (*anyopaque, usize, conversation.ContentBlock) void, - onMessageComplete: *const fn (*anyopaque, conversation.Message) void, + onMessageStart: *const fn (*anyopaque, conversation.MessageRole) anyerror!void, + onBlockStart: *const fn (*anyopaque, ContentBlockType, usize, ?BlockMeta) anyerror!void, + onContentDelta: *const fn (*anyopaque, usize, []const u8) anyerror!void, + onBlockComplete: *const fn (*anyopaque, usize, conversation.ContentBlock) anyerror!void, + onMessageComplete: *const fn (*anyopaque, conversation.Message) anyerror!void, + onError: *const fn (*anyopaque, anyerror) void, }; pub const Receiver = struct { ptr: *anyopaque, vtable: *const ReceiverVTable, - pub fn onMessageStart(self: Receiver, role: conversation.MessageRole) void { - self.vtable.onMessageStart(self.ptr, role); + pub fn onMessageStart(self: Receiver, role: conversation.MessageRole) !void { + try self.vtable.onMessageStart(self.ptr, role); } - pub fn onBlockStart(self: Receiver, block_type: ContentBlockType, index: usize, meta: ?BlockMeta) void { - self.vtable.onBlockStart(self.ptr, block_type, index, meta); + pub fn onBlockStart(self: Receiver, block_type: ContentBlockType, index: usize, meta: ?BlockMeta) !void { + try self.vtable.onBlockStart(self.ptr, block_type, index, meta); } - pub fn onContentDelta(self: Receiver, block_index: usize, delta: []const u8) void { - self.vtable.onContentDelta(self.ptr, block_index, delta); + pub fn onContentDelta(self: Receiver, block_index: usize, delta: []const u8) !void { + try self.vtable.onContentDelta(self.ptr, block_index, delta); } - pub fn onBlockComplete(self: Receiver, block_index: usize, block: conversation.ContentBlock) void { - self.vtable.onBlockComplete(self.ptr, block_index, block); + pub fn onBlockComplete(self: Receiver, block_index: usize, block: conversation.ContentBlock) !void { + try self.vtable.onBlockComplete(self.ptr, block_index, block); } - pub fn onMessageComplete(self: Receiver, message: conversation.Message) void { - self.vtable.onMessageComplete(self.ptr, message); + pub fn onMessageComplete(self: Receiver, message: conversation.Message) !void { + try self.vtable.onMessageComplete(self.ptr, message); + } + + pub fn onError(self: Receiver, err: anyerror) void { + self.vtable.onError(self.ptr, err); } }; diff --git a/libpanto/src/provider_openai.zig b/libpanto/src/provider_openai.zig deleted file mode 100644 index c459a9e..0000000 --- a/libpanto/src/provider_openai.zig +++ /dev/null @@ -1,6 +0,0 @@ -// OpenAI-compatible provider implementation. -// Implementation pending; this module will: -// - Convert Conversation → OpenAI wire JSON -// - Make HTTP POST with stream: true -// - Parse SSE events, drive block boundary state machine -// - Call the full Receiver callback sequence diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig new file mode 100644 index 0000000..e1b9f81 --- /dev/null +++ b/libpanto/src/provider_openai_chat.zig @@ -0,0 +1,443 @@ +//! OpenAI Chat Completions streaming provider. +//! +//! Wire format reference: https://platform.openai.com/docs/api-reference/chat/streaming +//! +//! Responsibilities: +//! - Convert `Conversation` → request JSON (delegated to openai_chat_json.zig) +//! - POST to `{base_url}/chat/completions` with `stream: true` +//! - Read the chunked body, feed bytes through SSEParser +//! - Parse each event payload, drive the block boundary state machine, +//! and emit Receiver callbacks +//! - Assemble the final Message and emit onMessageComplete + +const std = @import("std"); +const Allocator = std.mem.Allocator; +const Io = std.Io; +const http = std.http; +const Uri = std.Uri; + +const conversation = @import("conversation.zig"); +const provider_mod = @import("provider.zig"); +const sse_mod = @import("sse.zig"); +const json_mod = @import("openai_chat_json.zig"); +const config_mod = @import("config.zig"); + +/// Active streaming block type tracked by the state machine. Mirrors the +/// `ContentBlock` union variants but adds `.none` for "no block open yet". +const ActiveBlock = enum { none, text, thinking }; + +pub const OpenAIChatProvider = struct { + allocator: Allocator, + io: Io, + config: config_mod.Config, + http_client: http.Client, + + pub fn init(allocator: Allocator, io: Io, cfg: config_mod.Config) OpenAIChatProvider { + return .{ + .allocator = allocator, + .io = io, + .config = cfg, + .http_client = .{ .allocator = allocator, .io = io }, + }; + } + + pub fn deinit(self: *OpenAIChatProvider) void { + self.http_client.deinit(); + } + + /// Return a `Provider` interface bound to this concrete provider. + pub fn provider(self: *OpenAIChatProvider) provider_mod.Provider { + return .{ .ptr = self, .vtable = &vtable }; + } + + const vtable: provider_mod.ProviderVTable = .{ + .streamStep = vtableStreamStep, + .deinit = vtableDeinit, + }; + + fn vtableStreamStep( + ptr: *anyopaque, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + ) anyerror!void { + const self: *OpenAIChatProvider = @ptrCast(@alignCast(ptr)); + return self.streamStep(conv, receiver); + } + + fn vtableDeinit(ptr: *anyopaque) void { + const self: *OpenAIChatProvider = @ptrCast(@alignCast(ptr)); + self.deinit(); + } + + pub fn streamStep( + self: *OpenAIChatProvider, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + ) !void { + // Outer wrapper guarantees `onError` is called exactly once if + // anything fails — whether the receiver, the HTTP transport, or the + // SSE/JSON parsers. The inner `streamStepInner` does the real work. + self.streamStepInner(conv, receiver) catch |err| { + receiver.onError(err); + return err; + }; + } + + fn streamStepInner( + self: *OpenAIChatProvider, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + ) !void { + // Build URL: "{base_url}/chat/completions" + const url = try std.fmt.allocPrint( + self.allocator, + "{s}/chat/completions", + .{self.config.base_url}, + ); + defer self.allocator.free(url); + + const uri = try Uri.parse(url); + + // Build the request body. + const body = try json_mod.serializeRequest(self.allocator, &self.config, conv); + defer self.allocator.free(body); + + // Auth header + const auth_value = try std.fmt.allocPrint( + self.allocator, + "Bearer {s}", + .{self.config.api_key}, + ); + defer self.allocator.free(auth_value); + + const extra_headers = [_]http.Header{ + .{ .name = "content-type", .value = "application/json" }, + .{ .name = "accept", .value = "text/event-stream" }, + .{ .name = "authorization", .value = auth_value }, + }; + + // Open the request. We can't use `fetch()` because it buffers the + // response; we want to stream the body as it arrives. + var req = try self.http_client.request(.POST, uri, .{ + .extra_headers = &extra_headers, + .keep_alive = false, + .redirect_behavior = .not_allowed, + }); + defer req.deinit(); + + req.transfer_encoding = .{ .content_length = body.len }; + + var send_buf: [4096]u8 = undefined; + var bw = try req.sendBodyUnflushed(&send_buf); + try bw.writer.writeAll(body); + try bw.end(); + try req.connection.?.flush(); + + // Receive response headers. + var redirect_buf: [1024]u8 = undefined; + var response = try req.receiveHead(&redirect_buf); + + if (@intFromEnum(response.head.status) >= 400) { + // Drain body for diagnostics. + var transfer_buf: [4096]u8 = undefined; + const body_reader = response.reader(&transfer_buf); + var err_buf: std.ArrayList(u8) = .empty; + defer err_buf.deinit(self.allocator); + var tmp: [1024]u8 = undefined; + while (true) { + const n = body_reader.readSliceShort(&tmp) catch break; + if (n == 0) break; + 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, + }); + return error.HttpError; + } + + // Stream the body through the SSE parser and event handler. + var transfer_buf: [4096]u8 = undefined; + const body_reader = response.reader(&transfer_buf); + + var parser = sse_mod.SSEParser.init(self.allocator); + defer parser.deinit(); + + var state: StreamState = .init(self.allocator); + defer state.deinit(); + + var chunk: [4096]u8 = undefined; + while (true) { + const n = try body_reader.readSliceShort(&chunk); + if (n == 0) break; + + const events = try parser.feed(chunk[0..n]); + defer parser.freeEvents(events); + + for (events) |ev_payload| { + if (std.mem.eql(u8, ev_payload, "[DONE]")) { + try state.finalize(receiver, conv); + return; + } + try handleEvent(self.allocator, ev_payload, &state, receiver); + if (state.end_of_stream) { + try state.finalize(receiver, conv); + return; + } + } + } + + // Stream ended without [DONE] or finish_reason. Finalize anyway. + try state.finalize(receiver, conv); + } +}; + +/// State maintained across the streaming response: which block is currently +/// being assembled, accumulated content, and the assistant message being +/// built up for the final `onMessageComplete` callback. +const StreamState = struct { + allocator: Allocator, + started: bool = false, + /// Set when the wire stream signals end-of-turn (finish_reason or [DONE]). + /// Tells the outer read loop to stop pulling more events. + end_of_stream: bool = false, + /// Set once `finalize` has run, to make it idempotent. + finalized: bool = false, + active: ActiveBlock = .none, + block_index: usize = 0, + + /// Buffer for the currently-streaming block's content. + /// Owned by this state until the block is completed, at which point + /// ownership transfers to the assembled Message. + current_buf: conversation.TextualBlock = .empty, + + /// Assembled blocks for the final message. + blocks: std.ArrayList(conversation.ContentBlock) = .empty, + + fn init(allocator: Allocator) StreamState { + return .{ .allocator = allocator }; + } + + fn deinit(self: *StreamState) void { + self.current_buf.deinit(self.allocator); + for (self.blocks.items) |*b| b.deinit(self.allocator); + self.blocks.deinit(self.allocator); + } + + /// Close the active block (if any) and emit onBlockComplete. + /// Ownership of `current_buf` transfers into the appended block. + fn closeActive(self: *StreamState, receiver: *provider_mod.Receiver) !void { + if (self.active == .none) return; + + const block: conversation.ContentBlock = switch (self.active) { + .text => .{ .Text = self.current_buf }, + .thinking => .{ .Thinking = self.current_buf }, + .none => unreachable, + }; + // The buffer ownership has moved into `block`; replace with empty. + self.current_buf = .empty; + + try self.blocks.append(self.allocator, block); + try receiver.onBlockComplete(self.block_index, self.blocks.items[self.blocks.items.len - 1]); + + self.active = .none; + } + + /// Open a new block of the given type, possibly closing a prior block. + fn openBlock( + self: *StreamState, + new_active: ActiveBlock, + receiver: *provider_mod.Receiver, + ) !void { + if (self.active == new_active) return; + if (self.active != .none) { + try self.closeActive(receiver); + self.block_index += 1; + } + self.active = new_active; + const block_type: provider_mod.ContentBlockType = switch (new_active) { + .text => .Text, + .thinking => .Thinking, + .none => unreachable, + }; + try receiver.onBlockStart(block_type, self.block_index, null); + } + + fn appendDelta( + self: *StreamState, + receiver: *provider_mod.Receiver, + delta: []const u8, + ) !void { + try self.current_buf.appendSlice(self.allocator, delta); + try receiver.onContentDelta(self.block_index, delta); + } + + /// End the stream: close any open block and emit onMessageComplete. + /// Ownership of the assembled blocks transfers into the conversation. + fn finalize( + self: *StreamState, + receiver: *provider_mod.Receiver, + conv: *conversation.Conversation, + ) !void { + if (self.finalized) return; + self.finalized = true; + + try self.closeActive(receiver); + + // Move blocks into a fresh conversation message. + const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); + defer self.allocator.free(moved_blocks); + + try conv.addAssistantMessage(moved_blocks); + + // The conversation now owns the block buffers. Build a Message view + // for the callback (borrowed from the conversation). + const msg = conv.messages.items[conv.messages.items.len - 1]; + try receiver.onMessageComplete(msg); + } +}; + +fn handleEvent( + allocator: Allocator, + payload: []const u8, + state: *StreamState, + receiver: *provider_mod.Receiver, +) !void { + var parsed = try json_mod.parseStreamEvent(allocator, payload); + defer parsed.deinit(); + const d = parsed.delta; + + if (!state.started and d.role != null) { + state.started = true; + try receiver.onMessageStart(.assistant); + } + + if (d.reasoning_content) |rc| { + if (!state.started) { + state.started = true; + try receiver.onMessageStart(.assistant); + } + try state.openBlock(.thinking, receiver); + try state.appendDelta(receiver, rc); + } + + if (d.content) |c| { + if (!state.started) { + state.started = true; + try receiver.onMessageStart(.assistant); + } + try state.openBlock(.text, receiver); + try state.appendDelta(receiver, c); + } + + if (d.finish_reason) |_| { + state.end_of_stream = true; + } +} + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; + +/// A no-op Receiver that drops every callback. Useful when the test cares +/// about post-stream conversation state rather than callback observability. +const NoopReceiver = struct { + fn make() provider_mod.Receiver { + return .{ .ptr = @constCast(@ptrCast(&dummy)), .vtable = &vt }; + } + var dummy: u8 = 0; + const vt: provider_mod.ReceiverVTable = .{ + .onMessageStart = noopMsgStart, + .onBlockStart = noopBlockStart, + .onContentDelta = noopDelta, + .onBlockComplete = noopBlockComplete, + .onMessageComplete = noopMsgComplete, + .onError = noopErr, + }; + fn noopMsgStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} + fn noopBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize, _: ?provider_mod.BlockMeta) anyerror!void {} + fn noopDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} + fn noopBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} + fn noopMsgComplete(_: *anyopaque, _: conversation.Message) anyerror!void {} + fn noopErr(_: *anyopaque, _: anyerror) void {} +}; + +/// Feed a sequence of SSE event payloads through the state machine as if +/// they had been delivered by the wire, finalizing into `conv`. +fn runStreamedTurn( + allocator: Allocator, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + events: []const []const u8, +) !void { + var state: StreamState = .init(allocator); + defer state.deinit(); + + for (events) |payload| { + if (std.mem.eql(u8, payload, "[DONE]")) break; + try handleEvent(allocator, payload, &state, receiver); + if (state.end_of_stream) break; + } + try state.finalize(receiver, conv); +} + +test "two streamed turns persist assistant replies in the conversation" { + // Regression test for the bug where `finish_reason` arrived before + // `[DONE]` and `finalize` early-returned without appending the assistant + // message, so follow-up turns were sent without prior responses. + + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + try conv.addSystemMessage("You are a helpful assistant."); + try conv.addUserMessage("hello!"); + + var recv = NoopReceiver.make(); + + const turn1 = [_][]const u8{ + \\{"choices":[{"delta":{"role":"assistant"}}]} + , + \\{"choices":[{"delta":{"content":"Hello! "}}]} + , + \\{"choices":[{"delta":{"content":"How can I help you today?"}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"stop"}]} + , + "[DONE]", + }; + try runStreamedTurn(allocator, &conv, &recv, &turn1); + + try testing.expectEqual(@as(usize, 3), conv.messages.items.len); + try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[2].role); + try testing.expectEqualStrings( + "Hello! How can I help you today?", + conv.messages.items[2].content.items[0].Text.items, + ); + + // Second user turn: the assistant must still see its prior response. + try conv.addUserMessage("how did you respond to my greeting just now?"); + + const turn2 = [_][]const u8{ + \\{"choices":[{"delta":{"role":"assistant"}}]} + , + \\{"choices":[{"delta":{"content":"I replied: \"Hello! How can I help you today?\""}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"stop"}]} + , + "[DONE]", + }; + try runStreamedTurn(allocator, &conv, &recv, &turn2); + + // System + user + assistant + user + assistant = 5 messages. + try testing.expectEqual(@as(usize, 5), conv.messages.items.len); + try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[4].role); + try testing.expectEqualStrings( + "I replied: \"Hello! How can I help you today?\"", + conv.messages.items[4].content.items[0].Text.items, + ); +} diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig index 72ceb97..9eb6d0e 100644 --- a/libpanto/src/root.zig +++ b/libpanto/src/root.zig @@ -4,6 +4,11 @@ pub const agent = @import("agent.zig"); pub const config = @import("config.zig"); pub const sse = @import("sse.zig"); -// Internal modules — not re-exported as public API -// pub const json = @import("json.zig"); -// pub const provider_openai = @import("provider_openai.zig"); +// Internal modules — exposed so their tests run with the library test suite, +// but not part of the stable public API. +pub const openai_chat_json = @import("openai_chat_json.zig"); +pub const provider_openai_chat = @import("provider_openai_chat.zig"); + +test { + @import("std").testing.refAllDecls(@This()); +} diff --git a/libpanto/src/sse.zig b/libpanto/src/sse.zig index 150c3ea..c54d532 100644 --- a/libpanto/src/sse.zig +++ b/libpanto/src/sse.zig @@ -1,99 +1,81 @@ const std = @import("std"); -/// Parsed SSE event data — the content after "data: " (prefix removed). -/// For multi-line data events, lines are joined with newlines per the SSE spec. -/// -/// Borrowed from the parser's internal buffer. Call `SSEParser.freeEvents()` -/// to free the events list when done (event data itself is borrowed and -/// does not need individual freeing). -const Event = []const u8; +/// A parsed SSE event payload: the concatenated `data:` content. +/// Owned by the caller; free with `freeEvents`. +const Event = []u8; /// Incremental SSE line parser. /// /// The HTTP client delivers arbitrary-sized read buffers; this module -/// reassembles them into complete `data: ...\n\n` events. -/// -/// Event data is borrowed from the internal buffer. Call `freeEvents()` to -/// free the events list returned by `feed`; the event data itself is borrowed -/// from the parser's buffer and does not need individual freeing. +/// reassembles them into complete `data: ...\n\n` events and returns +/// owned event payload slices. pub const SSEParser = struct { buf: std.ArrayList(u8) = .empty, allocator: std.mem.Allocator, pub fn init(allocator: std.mem.Allocator) SSEParser { - return .{ - .allocator = allocator, - }; + return .{ .allocator = allocator }; } /// Feed a chunk of raw bytes. Returns a list of complete SSE events - /// found in the buffer. Returns an empty list if no complete events - /// are available yet. - /// - /// Event data is borrowed from the internal buffer. Call `freeEvents()` - /// on the returned list when done. + /// found in the buffer (may be empty). The caller owns the returned + /// slice and every event within; free with `freeEvents`. pub fn feed(self: *SSEParser, chunk: []const u8) ![]Event { try self.buf.appendSlice(self.allocator, chunk); - var events = std.ArrayList(Event).init(self.allocator); - errdefer events.deinit(); + var events: std.ArrayList(Event) = .empty; + errdefer { + for (events.items) |ev| self.allocator.free(ev); + events.deinit(self.allocator); + } - var write_pos: usize = 0; var scan_pos: usize = 0; - while (std.mem.indexOf(u8, self.buf.items[scan_pos..], "\n\n")) |rel_delim| { const delim = scan_pos + rel_delim; const event_bytes = self.buf.items[scan_pos..delim]; - // Compact data lines in-place. Compacted data is always shorter - // than the original event bytes (we strip at least "data: " per - // line), so write_pos <= scan_pos and we never overwrite data - // we haven't yet read. - const data_start = write_pos; + // Collect data: lines into an owned payload string. + var payload: std.ArrayList(u8) = .empty; + errdefer payload.deinit(self.allocator); + var line_start: usize = 0; var has_data = false; - while (line_start < event_bytes.len) { const newline = std.mem.indexOfScalarPos(u8, event_bytes, line_start, '\n') orelse event_bytes.len; const line = event_bytes[line_start..newline]; if (std.mem.startsWith(u8, line, "data: ")) { - if (has_data) { - self.buf.items[write_pos] = '\n'; - write_pos += 1; - } - const content = line["data: ".len..]; - std.mem.copyForwards(u8, self.buf.items[write_pos .. write_pos + content.len], content); - write_pos += content.len; + if (has_data) try payload.append(self.allocator, '\n'); + try payload.appendSlice(self.allocator, line["data: ".len..]); has_data = true; } - // Ignore other SSE fields (event:, id:, retry:) and comments (starting with :) + // Ignore other SSE fields (event:, id:, retry:) and comments. line_start = newline + 1; } if (has_data) { - try events.append(self.buf.items[data_start..write_pos]); + try events.append(self.allocator, try payload.toOwnedSlice(self.allocator)); + } else { + payload.deinit(self.allocator); } scan_pos = delim + 2; } - // Shift remaining (incomplete) bytes to the front, retaining - // the backing allocation for reuse. + // Drop processed bytes; retain the buffer allocation. const remaining_len = self.buf.items.len - scan_pos; if (remaining_len > 0 and scan_pos > 0) { std.mem.copyForwards(u8, self.buf.items[0..remaining_len], self.buf.items[scan_pos..]); } self.buf.shrinkRetainingCapacity(remaining_len); - return events.toOwnedSlice(); + return events.toOwnedSlice(self.allocator); } /// Free a list of events returned by `feed`. - /// Event data is borrowed from the parser's buffer and does not need - /// to be freed individually; only the events array itself is freed. pub fn freeEvents(self: *SSEParser, events: []Event) void { + for (events) |ev| self.allocator.free(ev); self.allocator.free(events); } @@ -102,98 +84,87 @@ pub const SSEParser = struct { } }; -test "SSEParser - single complete event" { - const allocator = std.testing.allocator; +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; - var parser = SSEParser.init(allocator); +test "SSEParser - single complete event" { + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); const events = try parser.feed("data: hello\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 1), events.len); - try std.testing.expectEqualStrings("hello", events[0]); + try testing.expectEqual(@as(usize, 1), events.len); + try testing.expectEqualStrings("hello", events[0]); } test "SSEParser - partial then complete" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); - // Feed partial — no complete event yet const events1 = try parser.feed("data: hel"); defer parser.freeEvents(events1); - try std.testing.expectEqual(@as(usize, 0), events1.len); + try testing.expectEqual(@as(usize, 0), events1.len); - // Feed remainder — completes the event const events2 = try parser.feed("lo\n\n"); defer parser.freeEvents(events2); - try std.testing.expectEqual(@as(usize, 1), events2.len); - try std.testing.expectEqualStrings("hello", events2[0]); + try testing.expectEqual(@as(usize, 1), events2.len); + try testing.expectEqualStrings("hello", events2[0]); } test "SSEParser - multiple events in single chunk" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); const events = try parser.feed("data: one\n\ndata: two\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 2), events.len); - try std.testing.expectEqualStrings("one", events[0]); - try std.testing.expectEqualStrings("two", events[1]); + try testing.expectEqual(@as(usize, 2), events.len); + try testing.expectEqualStrings("one", events[0]); + try testing.expectEqualStrings("two", events[1]); } test "SSEParser - data DONE signal" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); const events = try parser.feed("data: [DONE]\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 1), events.len); - try std.testing.expectEqualStrings("[DONE]", events[0]); + try testing.expectEqual(@as(usize, 1), events.len); + try testing.expectEqualStrings("[DONE]", events[0]); } test "SSEParser - empty event (keep-alive)" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); - // A blank line (just \n\n) is a keep-alive comment with no data const events = try parser.feed("\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 0), events.len); + try testing.expectEqual(@as(usize, 0), events.len); } test "SSEParser - ignores non-data fields" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); const events = try parser.feed("event: message\ndata: payload\nid: 42\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 1), events.len); - try std.testing.expectEqualStrings("payload", events[0]); + try testing.expectEqual(@as(usize, 1), events.len); + try testing.expectEqualStrings("payload", events[0]); } test "SSEParser - multi-line data joined with newline" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); const events = try parser.feed("data: line1\ndata: line2\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 1), events.len); - try std.testing.expectEqualStrings("line1\nline2", events[0]); + try testing.expectEqual(@as(usize, 1), events.len); + try testing.expectEqualStrings("line1\nline2", events[0]); } diff --git a/pantograph-for-studio-rail-system-200-cm.jpg b/pantograph-for-studio-rail-system-200-cm.jpg Binary files differnew file mode 100644 index 0000000..e9b2df8 --- /dev/null +++ b/pantograph-for-studio-rail-system-200-cm.jpg diff --git a/src/main.zig b/src/main.zig index 4f9c490..832f037 100644 --- a/src/main.zig +++ b/src/main.zig @@ -1,24 +1,123 @@ const std = @import("std"); -const libpanto = @import("panto"); +const panto = @import("panto"); -pub fn main() !void { - const config = libpanto.config.Config{ - .api_style = .anthropic, - .api_key = "test", - .base_url = "https://api.anthropic.com", - .model = "claude-sonnet-4-20250514", +const Receiver = panto.provider.Receiver; +const ReceiverVTable = panto.provider.ReceiverVTable; +const ContentBlockType = panto.provider.ContentBlockType; +const BlockMeta = panto.provider.BlockMeta; +const MessageRole = panto.conversation.MessageRole; + +/// Receiver that prints streaming deltas to stdout. Thinking blocks are +/// dimmed with ANSI escape codes; text blocks render plain. +const CLIReceiver = struct { + stdout: *std.Io.Writer, + file: *std.Io.File.Writer, + + pub fn receiver(self: *CLIReceiver) Receiver { + return .{ .ptr = self, .vtable = &vtable }; + } + + const vtable: ReceiverVTable = .{ + .onMessageStart = onMessageStart, + .onBlockStart = onBlockStart, + .onContentDelta = onContentDelta, + .onBlockComplete = onBlockComplete, + .onMessageComplete = onMessageComplete, + .onError = onError, }; - var debug_allocator = std.heap.DebugAllocator(.{}).init; - defer _ = debug_allocator.deinit(); - const alloc = debug_allocator.allocator(); + fn onMessageStart(ptr: *anyopaque, role: MessageRole) anyerror!void { + _ = ptr; + _ = role; + } - var conv = libpanto.conversation.Conversation.init(alloc); - defer conv.deinit(); + fn onBlockStart( + ptr: *anyopaque, + block_type: ContentBlockType, + index: usize, + meta: ?BlockMeta, + ) anyerror!void { + _ = index; + _ = meta; + const self: *CLIReceiver = @ptrCast(@alignCast(ptr)); + if (block_type == .Thinking) { + try self.stdout.writeAll("\x1b[2m[thinking] "); + } + try self.file.flush(); + } - try conv.addSystemMessage("You are a helpful assistant."); + fn onContentDelta(ptr: *anyopaque, index: usize, delta: []const u8) anyerror!void { + _ = index; + const self: *CLIReceiver = @ptrCast(@alignCast(ptr)); + try self.stdout.writeAll(delta); + try self.file.flush(); + } - const io = std.Io.Threaded.global_single_threaded.io(); + fn onBlockComplete( + ptr: *anyopaque, + index: usize, + block: panto.conversation.ContentBlock, + ) anyerror!void { + _ = index; + const self: *CLIReceiver = @ptrCast(@alignCast(ptr)); + if (block == .Thinking) { + try self.stdout.writeAll("\x1b[0m\n"); + } + try self.file.flush(); + } + + fn onMessageComplete(ptr: *anyopaque, message: panto.conversation.Message) anyerror!void { + _ = message; + const self: *CLIReceiver = @ptrCast(@alignCast(ptr)); + try self.stdout.writeAll("\n"); + try self.file.flush(); + } + + /// Reset any in-progress display state after a failed turn. Errors here + /// must be swallowed — we're already in the error path. + fn onError(ptr: *anyopaque, err: anyerror) void { + _ = &err; + const self: *CLIReceiver = @ptrCast(@alignCast(ptr)); + // If we were rendering a Thinking block, clear the dim style so + // subsequent output (including the "[error: ...]" line) is readable. + self.stdout.writeAll("\x1b[0m") catch {}; + self.file.flush() catch {}; + } +}; + +pub fn main(init: std.process.Init) !void { + const alloc = init.gpa; + const io = init.io; + + const api_key = init.environ_map.get("PANTO_API_KEY") orelse { + std.debug.print("error: PANTO_API_KEY is required\n", .{}); + return error.MissingApiKey; + }; + const base_url = init.environ_map.get("PANTO_BASE_URL") orelse "https://api.openai.com/v1"; + const model = init.environ_map.get("PANTO_MODEL") orelse { + std.debug.print("error: PANTO_MODEL is required\n", .{}); + return error.MissingModel; + }; + + const reasoning: panto.config.ReasoningEffort = + if (init.environ_map.get("PANTO_REASONING")) |val| + std.meta.stringToEnum(panto.config.ReasoningEffort, val) orelse { + std.debug.print( + "error: PANTO_REASONING must be one of: default, off, minimal, low, medium, high (got: {s})\n", + .{val}, + ); + return error.InvalidReasoning; + } + else + .default; + + const config = panto.config.Config{ + .api_style = .openai_chat, + .api_key = api_key, + .base_url = base_url, + .model = model, + .reasoning = reasoning, + }; var stdout_buffer: [4096]u8 = undefined; var stdout_file = std.Io.File.stdout().writer(io, &stdout_buffer); @@ -28,32 +127,46 @@ pub fn main() !void { var stdin_file = std.Io.File.stdin().reader(io, &stdin_buffer); const stdin = &stdin_file.interface; - try stdout.print("panto — model: {s}\n> ", .{config.model}); + var conv = panto.conversation.Conversation.init(alloc); + defer conv.deinit(); + try conv.addSystemMessage("You are a helpful assistant."); + + var provider_impl = panto.provider_openai_chat.OpenAIChatProvider.init(alloc, io, config); + defer provider_impl.deinit(); + + const agent = panto.agent.Agent.init(alloc, provider_impl.provider()); + + try stdout.print("panto — model: {s} @ {s}\n", .{ config.model, config.base_url }); + try stdout.print("> ", .{}); try stdout_file.flush(); + var cli_recv = CLIReceiver{ .stdout = stdout, .file = &stdout_file }; + var recv = cli_recv.receiver(); + while (true) { - const line = stdin.takeDelimiter('\n') catch |err| { - std.debug.print("Read error: {}\n", .{err}); + const maybe_line = stdin.takeDelimiter('\n') catch |err| { + std.debug.print("read error: {}\n", .{err}); return; }; - if (line) |text| { - if (text.len == 0) { - try stdout.print("> ", .{}); - try stdout_file.flush(); - continue; - } - // Copy the line since the reader buffer may be reused - const owned = try alloc.dupe(u8, text); - defer alloc.free(owned); - try conv.addUserMessage(owned); - // TODO: call agent.runStep() once provider is implemented - try stdout.print("(streaming not yet implemented)\n> ", .{}); - try stdout_file.flush(); - } else { - // EOF (Ctrl+D) - try stdout.print("\n", .{}); + const line = maybe_line orelse { + try stdout.writeAll("\n"); try stdout_file.flush(); break; + }; + + if (line.len == 0) { + try stdout.writeAll("\n> "); + try stdout_file.flush(); + continue; } + + try conv.addUserMessage(line); + + agent.runStep(&conv, &recv) catch |err| { + try stdout.print("\n[error: {s}]\n", .{@errorName(err)}); + }; + + try stdout.writeAll("\n> "); + try stdout_file.flush(); } } |
