summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--docs/oauth-provider-auth.md6
-rw-r--r--libpanto/src/agent.zig11
-rw-r--r--libpanto/src/anthropic_messages_json.zig54
-rw-r--r--libpanto/src/config.zig10
-rw-r--r--libpanto/src/conversation.zig60
-rw-r--r--libpanto/src/file_system_jsonl_store.zig59
-rw-r--r--libpanto/src/openai_responses_json.zig319
-rw-r--r--libpanto/src/provider.zig12
-rw-r--r--libpanto/src/provider_anthropic_messages.zig37
-rw-r--r--libpanto/src/provider_openai_chat.zig12
-rw-r--r--libpanto/src/provider_openai_responses.zig626
-rw-r--r--libpanto/src/public.zig1
-rw-r--r--scripts/analyze_pi_sessions.py458
-rw-r--r--src/auth_manager.zig5
-rw-r--r--src/config_file.zig69
-rw-r--r--src/debug_log.zig220
-rw-r--r--src/lua_runtime.zig140
-rw-r--r--src/luarocks_runtime.zig103
-rw-r--r--src/main.zig17
-rw-r--r--src/models_toml.zig40
-rw-r--r--src/tui_app.zig7
-rw-r--r--src/tui_components.zig167
-rw-r--r--src/tui_selectors.zig10
-rw-r--r--tmp/pi_session_cost_analysis.json288
24 files changed, 2591 insertions, 140 deletions
diff --git a/docs/oauth-provider-auth.md b/docs/oauth-provider-auth.md
index 96312e1..268c7e9 100644
--- a/docs/oauth-provider-auth.md
+++ b/docs/oauth-provider-auth.md
@@ -573,9 +573,9 @@ it is the OpenAI **Responses API** at
items, `reasoning`, `include: ["reasoning.encrypted_content"]`, `store: false`;
headers `Authorization: Bearer`, `chatgpt-account-id`,
`OpenAI-Beta: responses=experimental`, `originator: codex_cli_rs`). That is a
-new wire dialect, implemented as the **`openai_responses` provider style**
-(`provider_openai_responses.zig` + `openai_responses_json.zig`), distinct from
-`openai_chat`. The default config ships a commented `[providers.codex]` /
+Responses-family dialect: configure it as `style = "openai_responses"` with
+`dialect = "codex"`, which maps to the internal `openai_codex_responses`
+`APIStyle`. The default config ships a commented `[providers.codex]` /
`[auth.openai_codex]` example. The serializer maps system→`instructions`,
user/assistant text→message items, tool calls→`function_call` /
`function_call_output` items, and tools→flat function entries; the stream
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig
index 1da0261..0ff231a 100644
--- a/libpanto/src/agent.zig
+++ b/libpanto/src/agent.zig
@@ -78,7 +78,12 @@ fn cloneMessage(alloc: Allocator, msg: conversation.Message) !conversation.Messa
for (msg.content.items) |block| {
content.appendAssumeCapacity(try cloneBlock(alloc, block));
}
- return .{ .role = msg.role, .content = content, .usage = msg.usage };
+ return .{
+ .role = msg.role,
+ .content = content,
+ .usage = msg.usage,
+ .metadata = if (msg.metadata) |m| try alloc.dupe(u8, m) else null,
+ };
}
fn cloneBlock(alloc: Allocator, block: conversation.ContentBlock) !conversation.ContentBlock {
@@ -91,7 +96,9 @@ fn cloneBlock(alloc: Allocator, block: conversation.ContentBlock) !conversation.
mut.deinit(alloc);
}
const sig: ?[]const u8 = if (b.signature) |s| try alloc.dupe(u8, s) else null;
- break :blk .{ .Thinking = .{ .text = tb, .signature = sig } };
+ errdefer if (sig) |s| alloc.free(s);
+ const origin: ?conversation.SignatureOrigin = if (b.signature_origin) |o| try o.dupe(alloc) else null;
+ break :blk .{ .Thinking = .{ .text = tb, .signature = sig, .signature_origin = origin } };
},
.ToolUse => |b| blk: {
const id = try alloc.dupe(u8, b.id);
diff --git a/libpanto/src/anthropic_messages_json.zig b/libpanto/src/anthropic_messages_json.zig
index f7df80b..1b1cf6e 100644
--- a/libpanto/src/anthropic_messages_json.zig
+++ b/libpanto/src/anthropic_messages_json.zig
@@ -138,7 +138,7 @@ pub fn serializeRequest(
for (window, 0..) |msg, mi| {
if (msg.role == .system) continue;
const mark: ?usize = if (cache_at) |c| (if (c.message == mi) c.block else null) else null;
- try writeMessage(&s, msg, mark);
+ try writeMessage(&s, msg, mark, cfg);
}
try s.endArray();
@@ -251,6 +251,7 @@ fn writeMessage(
s: *std.json.Stringify,
msg: conversation.Message,
cache_block: ?usize,
+ cfg: *const config_mod.AnthropicMessagesConfig,
) !void {
try s.beginObject();
@@ -261,7 +262,7 @@ fn writeMessage(
try s.beginArray();
for (msg.content.items, 0..) |block, bi| {
const mark = if (cache_block) |cb| cb == bi else false;
- try writeBlock(s, block, mark);
+ try writeBlock(s, block, mark, cfg);
}
try s.endArray();
@@ -278,7 +279,12 @@ fn writeCacheControl(s: *std.json.Stringify) !void {
try s.endObject();
}
-fn writeBlock(s: *std.json.Stringify, block: conversation.ContentBlock, mark_cache: bool) !void {
+fn writeBlock(
+ s: *std.json.Stringify,
+ block: conversation.ContentBlock,
+ mark_cache: bool,
+ cfg: *const config_mod.AnthropicMessagesConfig,
+) !void {
switch (block) {
.Text => |tb| {
try s.beginObject();
@@ -292,9 +298,14 @@ fn writeBlock(s: *std.json.Stringify, block: conversation.ContentBlock, mark_cac
.Thinking => |tb| {
// Anthropic requires the signature field to round-trip a thinking
// block. If we don't have one (e.g. block was synthesized from a
- // non-Anthropic provider), skip the block: Anthropic will reject
- // unsigned thinking on incoming messages.
+ // non-Anthropic provider), or if the signature came from a
+ // different provider/model than the current request, skip the
+ // block: Anthropic will reject unsigned thinking on incoming
+ // messages, and opaque signatures are not portable across
+ // provider/model boundaries.
const sig = tb.signature orelse return;
+ const origin = tb.signature_origin orelse return;
+ if (!origin.matches(.anthropic_messages, cfg.base_url, cfg.model)) return;
try s.beginObject();
try s.objectField("type");
try s.write("thinking");
@@ -789,6 +800,7 @@ test "serializeRequest - cache breakpoint skips a trailing thinking block" {
.signature = sig,
} },
}, null);
+ try conversation.setThinkingOrigins(allocator, conv.messages.items[0].content.items, .anthropic_messages, "u", "claude-x");
var cfg = testConfig("claude-x");
var empty_tools = tool_registry_mod.ToolRegistry.init(allocator);
@@ -935,6 +947,7 @@ test "serializeRequest - signed assistant Thinking blocks round-trip" {
} },
.{ .Text = try conversation.textualBlockFromSlice(allocator, "the answer is 42") },
}, null);
+ try conversation.setThinkingOrigins(allocator, conv.messages.items[0].content.items, .anthropic_messages, "u", "claude-x");
const cfg = testConfig("claude-x");
var empty_tools = tool_registry_mod.ToolRegistry.init(allocator);
@@ -962,6 +975,37 @@ test "serializeRequest - signed assistant Thinking blocks round-trip" {
try testing.expectEqualStrings("the answer is 42", content[1].object.get("text").?.string);
}
+test "serializeRequest - mismatched thinking origin drops the block" {
+ const allocator = testing.allocator;
+
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+
+ const sig = try allocator.dupe(u8, "EqQBCgIYAhIM1gbcDa9GJwZA");
+ try conv.addAssistantMessage(&.{
+ .{ .Thinking = .{
+ .text = try conversation.textualBlockFromSlice(allocator, "other provider thinking"),
+ .signature = sig,
+ } },
+ .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer") },
+ }, null);
+ try conversation.setThinkingOrigins(allocator, conv.messages.items[0].content.items, .openai_responses, "https://api.individual.githubcopilot.com", "gpt-5.4-mini");
+
+ const cfg = testConfig("claude-x");
+ var empty_tools = tool_registry_mod.ToolRegistry.init(allocator);
+ defer empty_tools.deinit();
+ 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 content = parsed.value.object.get("messages").?.array.items[0]
+ .object.get("content").?.array.items;
+ try testing.expectEqual(@as(usize, 1), content.len);
+ try testing.expectEqualStrings("text", content[0].object.get("type").?.string);
+}
+
test "serializeRequest - unsigned Thinking blocks are dropped" {
// Anthropic rejects thinking blocks without a valid signature on inbound
// messages. If a block lacks one (e.g. from a different provider or an
diff --git a/libpanto/src/config.zig b/libpanto/src/config.zig
index c48cefc..7c8a495 100644
--- a/libpanto/src/config.zig
+++ b/libpanto/src/config.zig
@@ -27,6 +27,9 @@ pub const APIStyle = enum {
/// OpenAI Responses API (`/responses`). Used by the ChatGPT-subscription
/// Codex backend, whose request/stream shape differs from Chat Completions.
openai_responses,
+ /// ChatGPT-subscription Codex Responses dialect. User config selects this
+ /// with `style = "openai_responses"` plus `dialect = "codex"`.
+ openai_codex_responses,
};
/// A single HTTP request header (name/value). Used for provider
@@ -155,6 +158,7 @@ pub const ProviderConfig = union(APIStyle) {
openai_chat: OpenAIChatConfig,
anthropic_messages: AnthropicMessagesConfig,
openai_responses: OpenAIResponsesConfig,
+ openai_codex_responses: OpenAIResponsesConfig,
pub fn style(self: ProviderConfig) APIStyle {
return @as(APIStyle, self);
@@ -187,6 +191,12 @@ pub const ProviderConfig = union(APIStyle) {
.model = c.model,
.reasoning = c.reasoning,
},
+ .openai_codex_responses => |c| .{
+ .api_style = .openai_codex_responses,
+ .base_url = c.base_url,
+ .model = c.model,
+ .reasoning = c.reasoning,
+ },
};
}
};
diff --git a/libpanto/src/conversation.zig b/libpanto/src/conversation.zig
index 1942a0c..760f3ee 100644
--- a/libpanto/src/conversation.zig
+++ b/libpanto/src/conversation.zig
@@ -1,5 +1,6 @@
const std = @import("std");
const Allocator = std.mem.Allocator;
+const config = @import("config.zig");
/// A streaming text buffer used by content blocks.
/// Thin alias over ArrayList(u8) — amortized O(1) appends,
@@ -13,19 +14,60 @@ pub fn textualBlockFromSlice(alloc: Allocator, slice: []const u8) !TextualBlock
return buf;
}
+/// Provenance for a replayable thinking signature. Scoped conservatively to
+/// the exact provider endpoint + wire model that produced the enclosing
+/// assistant message.
+pub const SignatureOrigin = struct {
+ api_style: config.APIStyle,
+ base_url: []const u8,
+ model: []const u8,
+
+ pub fn init(
+ alloc: Allocator,
+ api_style: config.APIStyle,
+ base_url: []const u8,
+ model: []const u8,
+ ) !SignatureOrigin {
+ const burl = try alloc.dupe(u8, base_url);
+ errdefer alloc.free(burl);
+ const mdl = try alloc.dupe(u8, model);
+ return .{ .api_style = api_style, .base_url = burl, .model = mdl };
+ }
+
+ pub fn deinit(self: *SignatureOrigin, alloc: Allocator) void {
+ alloc.free(self.base_url);
+ alloc.free(self.model);
+ }
+
+ pub fn dupe(self: SignatureOrigin, alloc: Allocator) !SignatureOrigin {
+ return init(alloc, self.api_style, self.base_url, self.model);
+ }
+
+ pub fn matches(self: SignatureOrigin, api_style: config.APIStyle, base_url: []const u8, model: []const u8) bool {
+ return self.api_style == api_style and
+ std.mem.eql(u8, self.base_url, base_url) and
+ std.mem.eql(u8, self.model, model);
+ }
+};
+
/// A reasoning/thinking block from the assistant.
///
/// `text` is the streamed reasoning content. `signature` is the opaque
/// integrity token Anthropic emits with extended-thinking responses and
/// requires echoed back verbatim on follow-up turns. It is optional because
-/// other providers (OpenAI-compatible APIs) do not produce it.
+/// other providers (OpenAI-compatible APIs) do not produce it. When present,
+/// `signature_origin` records the exact provider/style/model that emitted the
+/// enclosing assistant message so serializers can decide whether that opaque
+/// signature is safe to replay on a later turn.
pub const ThinkingBlock = struct {
text: TextualBlock = .empty,
signature: ?[]const u8 = null,
+ signature_origin: ?SignatureOrigin = null,
pub fn deinit(self: *ThinkingBlock, alloc: Allocator) void {
self.text.deinit(alloc);
if (self.signature) |sig| alloc.free(sig);
+ if (self.signature_origin) |*origin| origin.deinit(alloc);
}
};
@@ -222,6 +264,7 @@ pub const Message = struct {
block.deinit(alloc);
}
self.content.deinit(alloc);
+ if (self.metadata) |m| alloc.free(m);
}
};
@@ -335,6 +378,21 @@ pub const Conversation = struct {
}
};
+pub fn setThinkingOrigins(
+ allocator: Allocator,
+ blocks: []ContentBlock,
+ api_style: config.APIStyle,
+ base_url: []const u8,
+ model: []const u8,
+) !void {
+ for (blocks) |*block| {
+ if (block.* != .Thinking) continue;
+ const origin = try SignatureOrigin.init(allocator, api_style, base_url, model);
+ if (block.Thinking.signature_origin) |*old| old.deinit(allocator);
+ block.Thinking.signature_origin = origin;
+ }
+}
+
/// Derive the effective ordered list of system-text blocks from a slice of
/// messages. This is the single shared rule that governs both provider
/// serialization and session rebuild.
diff --git a/libpanto/src/file_system_jsonl_store.zig b/libpanto/src/file_system_jsonl_store.zig
index 1d56ae9..e47698c 100644
--- a/libpanto/src/file_system_jsonl_store.zig
+++ b/libpanto/src/file_system_jsonl_store.zig
@@ -709,7 +709,7 @@ pub const SessionFile = struct {
for (self.entries.items) |entry| {
switch (entry) {
- .message => |me| try appendMessageToConv(&conv, self.allocator, me.message),
+ .message => |me| try appendMessageToConv(&conv, self.allocator, me.message, me.stamp),
}
}
return conv;
@@ -733,6 +733,7 @@ fn appendMessageToConv(
conv: *conversation_mod.Conversation,
allocator: Allocator,
disk_msg: StoredMessage,
+ stamp: ?session_mod.WireStamp,
) !void {
var content: std.ArrayList(conversation_mod.ContentBlock) = .empty;
errdefer {
@@ -756,6 +757,14 @@ fn appendMessageToConv(
const tb = block.Text;
block = .{ .System = .{ .text = tb, .mode = sys_mode } };
}
+ if (block == .Thinking and stamp != null) {
+ block.Thinking.signature_origin = try conversation_mod.SignatureOrigin.init(
+ allocator,
+ stamp.?.api_style,
+ stamp.?.base_url,
+ stamp.?.model,
+ );
+ }
content.appendAssumeCapacity(block);
}
const role: conversation_mod.MessageRole = switch (disk_msg.role) {
@@ -2090,3 +2099,51 @@ test "FileSystemJSONLStore catalog: create → append → load round-trips" {
try testing.expectEqual(@as(usize, 2), loaded.messages.items.len);
try testing.expectEqual(conversation_mod.MessageRole.user, loaded.messages.items[0].role);
}
+
+test "FileSystemJSONLStore load restores thinking signature origin from message stamp" {
+ const io = testing.io;
+
+ var td = try TmpSessionDir.init(testing.allocator);
+ defer td.deinit(testing.allocator);
+ const sessions = try std.fs.path.join(testing.allocator, &.{ td.abs_path, "sessions" });
+ defer testing.allocator.free(sessions);
+
+ var catalog = try FileSystemJSONLStore.init(testing.allocator, io, sessions);
+ defer catalog.deinit();
+ const store = catalog.store();
+
+ var sess = store.create();
+ defer sess.info.deinit(testing.allocator);
+
+ var conv = conversation_mod.Conversation.init(testing.allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "ping");
+ try conv.addAssistantMessage(&.{
+ .{ .Thinking = .{
+ .text = try conversation_mod.textualBlockFromSlice(testing.allocator, "thinking..."),
+ .signature = try testing.allocator.dupe(u8, "sig123"),
+ } },
+ .{ .Text = try conversation_mod.textualBlockFromSlice(testing.allocator, "pong") },
+ }, null);
+
+ const id: session_store_mod.WireIdentity = .{
+ .api_style = .anthropic_messages,
+ .base_url = "https://api.anthropic.com",
+ .model = "claude-sonnet-4-20250514",
+ };
+ var batch = [_]session_store_mod.PersistentMessage{
+ .{ .message = conv.messages.items[0], .identity = id },
+ .{ .message = conv.messages.items[1], .identity = id },
+ };
+ try sess.append(&batch);
+
+ var loaded = (try store.load(sess.info.id)).?;
+ defer loaded.deinit();
+ const thinking = loaded.messages.items[1].content.items[0].Thinking;
+ try testing.expect(thinking.signature_origin != null);
+ try testing.expect(thinking.signature_origin.?.matches(
+ .anthropic_messages,
+ "https://api.anthropic.com",
+ "claude-sonnet-4-20250514",
+ ));
+}
diff --git a/libpanto/src/openai_responses_json.zig b/libpanto/src/openai_responses_json.zig
index 8ff7712..295e79c 100644
--- a/libpanto/src/openai_responses_json.zig
+++ b/libpanto/src/openai_responses_json.zig
@@ -30,6 +30,11 @@ const tool_registry_mod = @import("tool_registry.zig");
// Request serialization
// ===========================================================================
+pub const RequestDialect = enum {
+ public,
+ codex,
+};
+
/// Serialize a Conversation into a `/responses` request body. Caller owns the
/// returned slice.
pub fn serializeRequest(
@@ -37,6 +42,7 @@ pub fn serializeRequest(
cfg: *const config_mod.OpenAIResponsesConfig,
conv: *const conversation.Conversation,
tools: *const tool_registry_mod.ToolRegistry,
+ dialect: RequestDialect,
) ![]u8 {
var aw: Writer.Allocating = .init(allocator);
errdefer aw.deinit();
@@ -54,8 +60,24 @@ pub fn serializeRequest(
try s.objectField("store");
try s.write(false);
- try s.objectField("max_output_tokens");
- try s.write(cfg.max_tokens);
+ if (dialect == .public) {
+ try s.objectField("max_output_tokens");
+ try s.write(cfg.max_tokens);
+ }
+
+ if (dialect == .codex) {
+ try s.objectField("text");
+ try s.beginObject();
+ try s.objectField("verbosity");
+ try s.write("low");
+ try s.endObject();
+
+ try s.objectField("tool_choice");
+ try s.write("auto");
+
+ try s.objectField("parallel_tool_calls");
+ try s.write(true);
+ }
// Carry encrypted reasoning so multi-turn reasoning can continue without
// server-side storage.
@@ -110,7 +132,7 @@ pub fn serializeRequest(
try s.beginArray();
for (conversation.activeMessageWindow(conv.messages.items)) |msg| {
if (msg.role == .system) continue;
- try writeInputForMessage(&s, msg, allocator);
+ try writeInputForMessage(&s, msg, allocator, cfg, dialect);
}
try s.endArray();
@@ -129,7 +151,13 @@ fn writeReasoning(s: *std.json.Stringify, effort: []const u8) !void {
}
/// Emit the `input` item(s) for one conversation message.
-fn writeInputForMessage(s: *std.json.Stringify, msg: conversation.Message, allocator: Allocator) !void {
+fn writeInputForMessage(
+ s: *std.json.Stringify,
+ msg: conversation.Message,
+ allocator: Allocator,
+ cfg: *const config_mod.OpenAIResponsesConfig,
+ dialect: RequestDialect,
+) !void {
switch (msg.role) {
.system => {},
.user => {
@@ -161,17 +189,33 @@ fn writeInputForMessage(s: *std.json.Stringify, msg: conversation.Message, alloc
defer text_buf.deinit(allocator);
try concatTextBlocks(msg.content.items, &text_buf, allocator);
if (text_buf.items.len > 0) {
- try writeRoleMessage(s, "user", "input_text", text_buf.items);
+ try writeRoleMessage(s, "user", "input_text", text_buf.items, null);
}
},
.assistant => {
+ // Replay opaque reasoning items first so stateless follow-up turns
+ // preserve encrypted reasoning continuity.
+ for (msg.content.items) |block| {
+ if (block != .Thinking) continue;
+ const tb = block.Thinking;
+ const sig = tb.signature orelse continue;
+ const origin = tb.signature_origin orelse continue;
+ if (!origin.matches(
+ if (dialect == .codex) .openai_codex_responses else .openai_responses,
+ cfg.base_url,
+ cfg.model,
+ )) continue;
+ if (sig.len == 0 or sig[0] != '{') continue;
+ try writeRawJson(s, sig);
+ }
+
// Assistant text first (as an output_text message), then each
// tool call as a `function_call` item.
var text_buf: std.ArrayList(u8) = .empty;
defer text_buf.deinit(allocator);
try concatTextBlocks(msg.content.items, &text_buf, allocator);
if (text_buf.items.len > 0) {
- try writeRoleMessage(s, "assistant", "output_text", text_buf.items);
+ try writeRoleMessage(s, "assistant", "output_text", text_buf.items, openAIPhaseFromMetadata(msg.metadata));
}
for (msg.content.items) |block| {
if (block != .ToolUse) continue;
@@ -192,10 +236,20 @@ fn writeInputForMessage(s: *std.json.Stringify, msg: conversation.Message, alloc
}
}
-fn writeRoleMessage(s: *std.json.Stringify, role: []const u8, content_type: []const u8, text: []const u8) !void {
+fn writeRoleMessage(
+ s: *std.json.Stringify,
+ role: []const u8,
+ content_type: []const u8,
+ text: []const u8,
+ phase: ?[]const u8,
+) !void {
try s.beginObject();
try s.objectField("role");
try s.write(role);
+ if (phase) |p| {
+ try s.objectField("phase");
+ try s.write(p);
+ }
try s.objectField("content");
try s.beginArray();
try s.beginObject();
@@ -208,6 +262,17 @@ fn writeRoleMessage(s: *std.json.Stringify, role: []const u8, content_type: []co
try s.endObject();
}
+fn openAIPhaseFromMetadata(metadata: ?[]const u8) ?[]const u8 {
+ const md = metadata orelse return null;
+ var parsed = std.json.parseFromSlice(std.json.Value, std.heap.page_allocator, md, .{}) catch return null;
+ defer parsed.deinit();
+ if (parsed.value != .object) return null;
+ const phase = strField(parsed.value.object, "openai_responses_phase") orelse return null;
+ if (std.mem.eql(u8, phase, "commentary")) return "commentary";
+ if (std.mem.eql(u8, phase, "final_answer")) return "final_answer";
+ return null;
+}
+
fn concatTextBlocks(
blocks: []const conversation.ContentBlock,
out: *std.ArrayList(u8),
@@ -244,6 +309,7 @@ pub const EventKind = enum {
output_text_delta,
reasoning_summary_delta,
function_call_arguments_delta,
+ function_call_arguments_done,
output_item_done,
completed,
failed,
@@ -261,8 +327,16 @@ pub const StreamEvent = struct {
/// Item id (`response.output_item.added/done`, function-call argument
/// deltas reference `item_id`).
item_id: ?[]const u8 = null,
+ /// Output array index. Some function-call argument events identify the
+ /// item by this instead of repeating `item_id`.
+ output_index: ?usize = null,
/// Item type on add/done: "message" | "function_call" | "reasoning".
item_type: ?[]const u8 = null,
+ /// Assistant message phase on message output items.
+ item_phase: ?[]const u8 = null,
+ /// Raw reasoning output item JSON, used for stateless encrypted reasoning
+ /// replay on follow-up turns.
+ reasoning_item_json: ?[]const u8 = null,
/// Function-call identity (on `output_item.added`/`done`).
call_id: ?[]const u8 = null,
name: ?[]const u8 = null,
@@ -272,6 +346,8 @@ pub const StreamEvent = struct {
error_message: ?[]const u8 = null,
/// Usage on `response.completed`.
usage: ?Usage = null,
+ /// Function-call output items included in a terminal `response.completed`.
+ completed_items: []const OutputItem = &.{},
pub const Usage = struct {
input_tokens: u64 = 0,
@@ -280,6 +356,15 @@ pub const StreamEvent = struct {
reasoning_tokens: u64 = 0,
};
+ pub const OutputItem = struct {
+ output_index: usize,
+ item_id: ?[]const u8 = null,
+ item_type: ?[]const u8 = null,
+ call_id: ?[]const u8 = null,
+ name: ?[]const u8 = null,
+ arguments: ?[]const u8 = null,
+ };
+
pub fn deinit(self: *StreamEvent) void {
self.parsed.deinit();
}
@@ -305,23 +390,36 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !StreamEvent
ev.kind = .output_text_delta;
ev.delta = strField(obj, "delta");
ev.item_id = strField(obj, "item_id");
+ ev.output_index = usizeField(obj, "output_index");
} else if (std.mem.eql(u8, type_str, "response.reasoning_summary_text.delta")) {
ev.kind = .reasoning_summary_delta;
ev.delta = strField(obj, "delta");
ev.item_id = strField(obj, "item_id");
+ ev.output_index = usizeField(obj, "output_index");
} else if (std.mem.eql(u8, type_str, "response.function_call_arguments.delta")) {
ev.kind = .function_call_arguments_delta;
- ev.delta = strField(obj, "delta");
+ ev.delta = strField(obj, "delta") orelse
+ strField(obj, "arguments_delta") orelse
+ strField(obj, "arguments");
ev.item_id = strField(obj, "item_id");
+ ev.output_index = usizeField(obj, "output_index");
+ } else if (std.mem.eql(u8, type_str, "response.function_call_arguments.done")) {
+ ev.kind = .function_call_arguments_done;
+ ev.arguments = strField(obj, "arguments");
+ ev.item_id = strField(obj, "item_id");
+ ev.output_index = usizeField(obj, "output_index");
} else if (std.mem.eql(u8, type_str, "response.output_item.added")) {
ev.kind = .output_item_added;
- readItem(obj, &ev);
+ ev.output_index = usizeField(obj, "output_index");
+ try readItem(parsed.arena.allocator(), obj, &ev);
} else if (std.mem.eql(u8, type_str, "response.output_item.done")) {
ev.kind = .output_item_done;
- readItem(obj, &ev);
- } else if (std.mem.eql(u8, type_str, "response.completed")) {
+ ev.output_index = usizeField(obj, "output_index");
+ try readItem(parsed.arena.allocator(), obj, &ev);
+ } else if (std.mem.eql(u8, type_str, "response.completed") or std.mem.eql(u8, type_str, "response.done")) {
ev.kind = .completed;
readUsage(obj, &ev);
+ try readCompletedOutput(parsed.arena.allocator(), obj, &ev);
} else if (std.mem.eql(u8, type_str, "response.failed") or std.mem.eql(u8, type_str, "response.incomplete")) {
ev.kind = .failed;
ev.error_message = readResponseError(obj);
@@ -334,15 +432,31 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !StreamEvent
return ev;
}
-fn readItem(obj: std.json.ObjectMap, ev: *StreamEvent) void {
+fn readItem(allocator: Allocator, obj: std.json.ObjectMap, ev: *StreamEvent) !void {
const item = obj.get("item") orelse return;
if (item != .object) return;
const io = item.object;
ev.item_id = strField(io, "id");
ev.item_type = strField(io, "type");
+ ev.item_phase = strField(io, "phase");
ev.call_id = strField(io, "call_id");
ev.name = strField(io, "name");
ev.arguments = strField(io, "arguments");
+ if (ev.item_type) |it| {
+ if (std.mem.eql(u8, it, "reasoning")) {
+ if (io.get("encrypted_content") != null) {
+ ev.reasoning_item_json = try stringifyValue(allocator, item);
+ }
+ }
+ }
+}
+
+fn stringifyValue(allocator: Allocator, value: std.json.Value) ![]const u8 {
+ var aw: Writer.Allocating = .init(allocator);
+ errdefer aw.deinit();
+ var s: std.json.Stringify = .{ .writer = &aw.writer };
+ try s.write(value);
+ return try aw.toOwnedSlice();
}
fn readUsage(obj: std.json.ObjectMap, ev: *StreamEvent) void {
@@ -366,6 +480,32 @@ fn readUsage(obj: std.json.ObjectMap, ev: *StreamEvent) void {
ev.usage = usage;
}
+fn readCompletedOutput(allocator: Allocator, obj: std.json.ObjectMap, ev: *StreamEvent) !void {
+ const resp = obj.get("response") orelse return;
+ if (resp != .object) return;
+ const output = resp.object.get("output") orelse return;
+ if (output != .array) return;
+
+ var items: std.ArrayList(StreamEvent.OutputItem) = .empty;
+ errdefer items.deinit(allocator);
+ for (output.array.items, 0..) |v, i| {
+ if (v != .object) continue;
+ const item_type = strField(v.object, "type") orelse continue;
+ if (!std.mem.eql(u8, item_type, "function_call")) continue;
+ try items.append(allocator, .{
+ .output_index = i,
+ .item_id = strField(v.object, "id"),
+ .item_type = item_type,
+ .call_id = strField(v.object, "call_id"),
+ .name = strField(v.object, "name"),
+ .arguments = strField(v.object, "arguments"),
+ });
+ }
+ if (items.items.len == 0) return;
+ const buf = try items.toOwnedSlice(allocator);
+ ev.completed_items = buf;
+}
+
fn readResponseError(obj: std.json.ObjectMap) ?[]const u8 {
const resp = obj.get("response") orelse return null;
if (resp != .object) return null;
@@ -389,6 +529,12 @@ fn u64Field(obj: std.json.ObjectMap, name: []const u8) u64 {
return @intCast(v.integer);
}
+fn usizeField(obj: std.json.ObjectMap, name: []const u8) ?usize {
+ const v = obj.get(name) orelse return null;
+ if (v != .integer or v.integer < 0) return null;
+ return @intCast(v.integer);
+}
+
// ===========================================================================
// Tests
// ===========================================================================
@@ -421,7 +567,7 @@ test "responses serializeRequest - instructions, input, store/include" {
cfg.reasoning = .high;
var tools = emptyTools();
defer tools.deinit();
- const body = try serializeRequest(allocator, &cfg, &conv, &tools);
+ const body = try serializeRequest(allocator, &cfg, &conv, &tools, .public);
defer allocator.free(body);
var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{});
@@ -458,7 +604,7 @@ test "responses serializeRequest - tools are flat function items" {
});
const cfg = testConfig("gpt-5.1-codex");
- const body = try serializeRequest(allocator, &cfg, &conv, &tools);
+ const body = try serializeRequest(allocator, &cfg, &conv, &tools, .public);
defer allocator.free(body);
var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{});
defer parsed.deinit();
@@ -471,6 +617,87 @@ test "responses serializeRequest - tools are flat function items" {
try testing.expect(tool0.get("parameters").? == .object);
}
+test "responses serializeRequest - codex dialect omits max tokens and sends codex defaults" {
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "Hello!");
+
+ var tools = emptyTools();
+ defer tools.deinit();
+ const cfg = testConfig("gpt-5.5");
+ const body = try serializeRequest(allocator, &cfg, &conv, &tools, .codex);
+ 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.expect(root.get("max_output_tokens") == null);
+ try testing.expectEqualStrings("low", root.get("text").?.object.get("verbosity").?.string);
+ try testing.expectEqualStrings("auto", root.get("tool_choice").?.string);
+ try testing.expect(root.get("parallel_tool_calls").?.bool);
+}
+
+test "responses serializeRequest - assistant phase metadata and reasoning signature replay" {
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+
+ const sig = try allocator.dupe(u8,
+ "{\"type\":\"reasoning\",\"id\":\"rs_1\",\"encrypted_content\":\"sealed\"}"
+ );
+ const thinking = try conversation.textualBlockFromSlice(allocator, "thinking");
+ const text = try conversation.textualBlockFromSlice(allocator, "answer");
+ try conv.addAssistantMessage(&.{
+ .{ .Thinking = .{ .text = thinking, .signature = sig } },
+ .{ .Text = text },
+ }, null);
+ try conversation.setThinkingOrigins(allocator, conv.messages.items[0].content.items, .openai_codex_responses, "u", "gpt-5.5");
+ conv.messages.items[0].metadata = try allocator.dupe(u8, "{\"openai_responses_phase\":\"final_answer\"}");
+
+ var tools = emptyTools();
+ defer tools.deinit();
+ const cfg = testConfig("gpt-5.5");
+ const body = try serializeRequest(allocator, &cfg, &conv, &tools, .codex);
+ defer allocator.free(body);
+ var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{});
+ defer parsed.deinit();
+
+ const input = parsed.value.object.get("input").?.array.items;
+ try testing.expectEqual(@as(usize, 2), input.len);
+ try testing.expectEqualStrings("reasoning", input[0].object.get("type").?.string);
+ try testing.expectEqualStrings("sealed", input[0].object.get("encrypted_content").?.string);
+ try testing.expectEqualStrings("assistant", input[1].object.get("role").?.string);
+ try testing.expectEqualStrings("final_answer", input[1].object.get("phase").?.string);
+}
+
+test "responses serializeRequest - mismatched signature origin skips reasoning replay" {
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+
+ const sig = try allocator.dupe(u8,
+ "{\"type\":\"reasoning\",\"id\":\"rs_1\",\"encrypted_content\":\"sealed\"}"
+ );
+ try conv.addAssistantMessage(&.{
+ .{ .Thinking = .{ .text = try conversation.textualBlockFromSlice(allocator, "thinking"), .signature = sig } },
+ .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer") },
+ }, null);
+ try conversation.setThinkingOrigins(allocator, conv.messages.items[0].content.items, .openai_responses, "https://api.individual.githubcopilot.com", "gpt-5.4-mini");
+
+ var tools = emptyTools();
+ defer tools.deinit();
+ const cfg = testConfig("gpt-5.5");
+ const body = try serializeRequest(allocator, &cfg, &conv, &tools, .codex);
+ defer allocator.free(body);
+ var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{});
+ defer parsed.deinit();
+
+ const input = parsed.value.object.get("input").?.array.items;
+ try testing.expectEqual(@as(usize, 1), input.len);
+ try testing.expectEqualStrings("assistant", input[0].object.get("role").?.string);
+}
+
test "responses serializeRequest - assistant tool_use + tool result round-trip" {
const allocator = testing.allocator;
var conv = conversation.Conversation.init(allocator);
@@ -495,7 +722,7 @@ test "responses serializeRequest - assistant tool_use + tool result round-trip"
var tools = emptyTools();
defer tools.deinit();
const cfg = testConfig("gpt-5.1-codex");
- const body = try serializeRequest(allocator, &cfg, &conv, &tools);
+ const body = try serializeRequest(allocator, &cfg, &conv, &tools, .public);
defer allocator.free(body);
var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{});
defer parsed.deinit();
@@ -540,6 +767,29 @@ test "responses parseStreamEvent - function_call item added/done + args delta" {
try testing.expectEqual(EventKind.function_call_arguments_delta, d.kind);
try testing.expectEqualStrings("fc_1", d.item_id.?);
try testing.expectEqualStrings("{\"x\":1}", d.delta.?);
+
+ var alt = try parseStreamEvent(allocator,
+ \\{"type":"response.function_call_arguments.delta","item_id":"fc_1","arguments_delta":"{\"x\":1}"}
+ );
+ defer alt.deinit();
+ try testing.expectEqual(EventKind.function_call_arguments_delta, alt.kind);
+ try testing.expectEqualStrings("{\"x\":1}", alt.delta.?);
+
+ var by_index = try parseStreamEvent(allocator,
+ \\{"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"x\":1}"}
+ );
+ defer by_index.deinit();
+ try testing.expectEqual(EventKind.function_call_arguments_delta, by_index.kind);
+ try testing.expectEqual(@as(usize, 0), by_index.output_index.?);
+ try testing.expectEqualStrings("{\"x\":1}", by_index.delta.?);
+
+ var done = try parseStreamEvent(allocator,
+ \\{"type":"response.function_call_arguments.done","output_index":0,"arguments":"{\"x\":1}"}
+ );
+ defer done.deinit();
+ try testing.expectEqual(EventKind.function_call_arguments_done, done.kind);
+ try testing.expectEqual(@as(usize, 0), done.output_index.?);
+ try testing.expectEqualStrings("{\"x\":1}", done.arguments.?);
}
test "responses parseStreamEvent - completed usage" {
@@ -555,6 +805,45 @@ test "responses parseStreamEvent - completed usage" {
try testing.expectEqual(@as(u64, 8), ev.usage.?.reasoning_tokens);
}
+test "responses parseStreamEvent - done alias, phase, encrypted reasoning item" {
+ const allocator = testing.allocator;
+ var done = try parseStreamEvent(allocator,
+ \\{"type":"response.done","response":{"usage":{"input_tokens":1,"output_tokens":2}}}
+ );
+ defer done.deinit();
+ try testing.expectEqual(EventKind.completed, done.kind);
+ try testing.expectEqual(@as(u64, 1), done.usage.?.input_tokens);
+
+ var msg = try parseStreamEvent(allocator,
+ \\{"type":"response.output_item.done","item":{"type":"message","id":"msg_1","phase":"commentary"}}
+ );
+ defer msg.deinit();
+ try testing.expectEqual(EventKind.output_item_done, msg.kind);
+ try testing.expectEqualStrings("commentary", msg.item_phase.?);
+
+ var reasoning = try parseStreamEvent(allocator,
+ \\{"type":"response.output_item.done","item":{"type":"reasoning","id":"rs_1","encrypted_content":"sealed"}}
+ );
+ defer reasoning.deinit();
+ try testing.expectEqual(EventKind.output_item_done, reasoning.kind);
+ try testing.expect(reasoning.reasoning_item_json != null);
+}
+
+test "responses parseStreamEvent - completed output function calls" {
+ const allocator = testing.allocator;
+ var ev = try parseStreamEvent(allocator,
+ \\{"type":"response.completed","response":{"output":[{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":"{\"path\":\"a\"}","status":"completed"}],"usage":{"input_tokens":100,"output_tokens":20}}}
+ );
+ defer ev.deinit();
+ try testing.expectEqual(EventKind.completed, ev.kind);
+ try testing.expectEqual(@as(usize, 1), ev.completed_items.len);
+ try testing.expectEqual(@as(usize, 0), ev.completed_items[0].output_index);
+ try testing.expectEqualStrings("fc_1", ev.completed_items[0].item_id.?);
+ try testing.expectEqualStrings("call_9", ev.completed_items[0].call_id.?);
+ try testing.expectEqualStrings("std__read", ev.completed_items[0].name.?);
+ try testing.expectEqualStrings("{\"path\":\"a\"}", ev.completed_items[0].arguments.?);
+}
+
test "responses parseStreamEvent - error + failed" {
const allocator = testing.allocator;
var e = try parseStreamEvent(allocator,
diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig
index 9c715b7..3c3b52e 100644
--- a/libpanto/src/provider.zig
+++ b/libpanto/src/provider.zig
@@ -326,6 +326,18 @@ pub fn openStream(
const rr = try req.open(conv, registry);
return rr.providerStream();
},
+ .openai_codex_responses => |*c| {
+ var req: provider_openai_responses.OpenAIResponsesRequest = .{
+ .allocator = allocator,
+ .io = io,
+ .config = c,
+ .dialect = .codex,
+ .http_client = client,
+ .diag = diag,
+ };
+ const rr = try req.open(conv, registry);
+ return rr.providerStream();
+ },
}
}
diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig
index 8f5d200..95013d1 100644
--- a/libpanto/src/provider_anthropic_messages.zig
+++ b/libpanto/src/provider_anthropic_messages.zig
@@ -5,7 +5,10 @@
//!
//! Responsibilities:
//! - Convert `Conversation` → request JSON (delegated to anthropic_messages_json.zig)
-//! - POST to `{base_url}/v1/messages` with `stream: true`
+//! - POST to `{base_url}/messages` with `stream: true`
+//! (caller is responsible for the `/v1` version segment; panto will
+//! introduce a new `anthropic_messages_v2` API style for v2 rather than
+//! pulling the version into the provider)
//! - Read the chunked body, feed bytes through SSEParser
//! - Parse each event payload, drive a thin assembly loop, and emit Receiver
//! callbacks. Anthropic gives us explicit block boundaries, so no
@@ -58,6 +61,12 @@ pub const AnthropicMessagesRequest = struct {
.parser = sse_mod.SSEParser.init(self.allocator),
.state = .init(self.allocator),
};
+ rr.state.signature_origin = try conversation.SignatureOrigin.init(
+ self.allocator,
+ .anthropic_messages,
+ self.config.base_url,
+ self.config.model,
+ );
errdefer {
rr.parser.deinit();
rr.state.deinit();
@@ -65,7 +74,7 @@ pub const AnthropicMessagesRequest = struct {
const url = try std.fmt.allocPrint(
self.allocator,
- "{s}/v1/messages",
+ "{s}/messages",
.{self.config.base_url},
);
defer self.allocator.free(url);
@@ -74,6 +83,7 @@ pub const AnthropicMessagesRequest = struct {
const body = try json_mod.serializeRequest(self.allocator, self.config, conv, tools);
defer self.allocator.free(body);
+ std.log.debug("anthropic_messages => {s}", .{body});
// Build headers. The four base headers are always present; the
// interleaved-thinking beta header is added only when the config
@@ -96,7 +106,7 @@ pub const AnthropicMessagesRequest = struct {
.value = "interleaved-thinking-2025-05-14",
};
}
- const base_headers = headers_buf[0 .. if (send_interleaved) @as(usize, 5) else @as(usize, 4)];
+ const base_headers = headers_buf[0..if (send_interleaved) @as(usize, 5) else @as(usize, 4)];
// Merge any provider `extra_headers` onto the base set. Freed at the
// end of `open` — after the request body has been flushed.
const extra_headers = try provider_mod.mergeHeaders(
@@ -147,12 +157,20 @@ pub const AnthropicMessagesRequest = struct {
if (err_buf.items.len > 16 * 1024) break;
}
const status: u16 = @intFromEnum(rr.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; `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);
+ // 401/403 is routinely recovered by the turn-runner's forced token
+ // refresh + reopen; demote it to `.debug` (still in the debug log)
+ // so a transparent refresh doesn't surface a scary error line. The
+ // retry layer raises a hard error only if recovery ultimately fails.
+ if (classified == error.ProviderAuthFailed) {
+ std.log.debug("anthropic_messages HTTP {d} (recoverable auth): {s}", .{ status, err_buf.items });
+ } else {
+ std.log.err("anthropic_messages HTTP {d}: {s}", .{ status, err_buf.items });
+ }
if (self.diag) |d| {
d.status_code = status;
d.retry_after_ms = retry_after_ms;
@@ -287,6 +305,7 @@ const StreamState = struct {
/// `onMessageComplete`, the latter stamps `null`.
usage: provider_mod.Usage = .{},
usage_seen: bool = false,
+ signature_origin: ?conversation.SignatureOrigin = null,
stop_reason: ?[]u8 = null,
/// Owned, human-readable description of a mid-stream `error` event
/// (e.g. `"overloaded_error: Overloaded"`), surfaced to the agent via
@@ -321,6 +340,7 @@ const StreamState = struct {
}
for (self.blocks.items) |*b| b.deinit(self.allocator);
self.blocks.deinit(self.allocator);
+ if (self.signature_origin) |*o| o.deinit(self.allocator);
if (self.stop_reason) |s| self.allocator.free(s);
if (self.stream_error_message) |s| self.allocator.free(s);
}
@@ -557,6 +577,15 @@ const StreamState = struct {
defer self.allocator.free(moved_blocks);
const usage: ?provider_mod.Usage = if (self.usage_seen) self.usage else null;
+ if (self.signature_origin) |origin| {
+ try conversation.setThinkingOrigins(
+ self.allocator,
+ moved_blocks,
+ origin.api_style,
+ origin.base_url,
+ origin.model,
+ );
+ }
try conv.addAssistantMessage(moved_blocks, usage);
const msg = conv.messages.items[conv.messages.items.len - 1];
diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig
index b236a60..3f01e7f 100644
--- a/libpanto/src/provider_openai_chat.zig
+++ b/libpanto/src/provider_openai_chat.zig
@@ -94,6 +94,7 @@ pub const OpenAIChatRequest = struct {
// Build the request body.
const body = try json_mod.serializeRequest(self.allocator, self.config, conv, tools);
defer self.allocator.free(body);
+ std.log.debug("openai_chat => {s}", .{body});
// Auth header
const auth_value = try std.fmt.allocPrint(
@@ -165,11 +166,19 @@ pub const OpenAIChatRequest = struct {
if (err_buf.items.len > 16 * 1024) break;
}
const status: u16 = @intFromEnum(rr.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);
+ // 401/403 is routinely recovered by the turn-runner's forced token
+ // refresh + reopen; demote it to `.debug` (still in the debug log)
+ // so a transparent refresh doesn't surface a scary error line. The
+ // retry layer raises a hard error only if recovery ultimately fails.
+ if (classified == error.ProviderAuthFailed) {
+ std.log.debug("openai_chat HTTP {d} (recoverable auth): {s}", .{ status, err_buf.items });
+ } else {
+ std.log.err("openai_chat HTTP {d}: {s}", .{ status, err_buf.items });
+ }
if (self.diag) |d| {
d.status_code = status;
d.retry_after_ms = retry_after_ms;
@@ -1032,7 +1041,6 @@ test "inbound wire tool name is decoded to dotted form (even split across __)" {
try testing.expectEqualStrings("std.read", tu.name);
}
-
test "parallel tool_calls emit one complete start/delta/complete cycle per block" {
// Regression test: previously, the OpenAI provider deferred ALL
// tool_use onBlockComplete callbacks to finalize, so a four-tool
diff --git a/libpanto/src/provider_openai_responses.zig b/libpanto/src/provider_openai_responses.zig
index 92a309b..c172f6d 100644
--- a/libpanto/src/provider_openai_responses.zig
+++ b/libpanto/src/provider_openai_responses.zig
@@ -36,6 +36,11 @@ const tool_registry_mod = @import("tool_registry.zig");
const Event = stream_mod.Event;
const EventQueue = stream_mod.EventQueue;
+pub const OpenAIResponsesDialect = enum {
+ public,
+ codex,
+};
+
fn decodeNameInPlace(name_buf: *conversation.TextualBlock) void {
const decoded = tool_registry_mod.decodeName(name_buf.items, name_buf.items);
name_buf.items.len = decoded.len;
@@ -45,6 +50,7 @@ pub const OpenAIResponsesRequest = struct {
allocator: Allocator,
io: Io,
config: *const config_mod.OpenAIResponsesConfig,
+ dialect: OpenAIResponsesDialect = .public,
http_client: *http.Client,
diag: ?*provider_mod.ProviderDiagnostic = null,
@@ -61,17 +67,27 @@ pub const OpenAIResponsesRequest = struct {
.parser = sse_mod.SSEParser.init(self.allocator),
.state = .init(self.allocator),
};
+ rr.state.signature_origin = try conversation.SignatureOrigin.init(
+ self.allocator,
+ if (self.dialect == .codex) .openai_codex_responses else .openai_responses,
+ self.config.base_url,
+ self.config.model,
+ );
errdefer {
rr.parser.deinit();
rr.state.deinit();
}
- const url = try std.fmt.allocPrint(self.allocator, "{s}/responses", .{self.config.base_url});
+ const url = try responsesURL(self.allocator, self.config.base_url);
defer self.allocator.free(url);
const uri = try Uri.parse(url);
- const body = try json_mod.serializeRequest(self.allocator, self.config, conv, tools);
+ const body = try json_mod.serializeRequest(self.allocator, self.config, conv, tools, switch (self.dialect) {
+ .public => .public,
+ .codex => .codex,
+ });
defer self.allocator.free(body);
+ std.log.debug("openai_responses => {s}", .{body});
const auth_value = try std.fmt.allocPrint(self.allocator, "Bearer {s}", .{self.config.api_key});
defer self.allocator.free(auth_value);
@@ -123,8 +139,18 @@ pub const OpenAIResponsesRequest = struct {
if (err_buf.items.len > 16 * 1024) break;
}
const status: u16 = @intFromEnum(rr.response.head.status);
- std.log.err("openai_responses HTTP {d}: {s}", .{ status, err_buf.items });
const classified = provider_mod.classifyHttpStatus(status, err_buf.items);
+ // 401/403 is routinely recovered by the turn-runner's forced token
+ // refresh + reopen (see `driveTurn` in tui_app.zig). Logging it at
+ // `.err` surfaces a scary "token expired" line for what is a
+ // transparent, recoverable refresh. Demote it to `.debug` (still
+ // captured in the debug log); the retry layer raises a hard error
+ // only if recovery ultimately fails.
+ if (classified == error.ProviderAuthFailed) {
+ std.log.debug("openai_responses HTTP {d} (recoverable auth): {s}", .{ status, err_buf.items });
+ } else {
+ std.log.err("openai_responses HTTP {d}: {s}", .{ status, err_buf.items });
+ }
if (self.diag) |d| {
d.status_code = status;
d.retry_after_ms = retry_after_ms;
@@ -137,6 +163,16 @@ pub const OpenAIResponsesRequest = struct {
}
};
+// Appends `/responses` to `base_url` (with trailing slashes trimmed). The
+// caller is responsible for any path segment preceding `/responses` — for
+// the Codex ChatGPT-subscription endpoint, that means putting `/codex`
+// (or any other prefix) in `base_url` directly. No `endsWith` guessing: the
+// contract is the same regardless of `dialect`.
+fn responsesURL(allocator: Allocator, base_url: []const u8) ![]u8 {
+ const trimmed = std.mem.trim(u8, base_url, "/");
+ return std.fmt.allocPrint(allocator, "{s}/responses", .{trimmed});
+}
+
pub const ResumableResponse = struct {
allocator: Allocator,
conv: *conversation.Conversation,
@@ -226,6 +262,29 @@ pub const ResumableResponse = struct {
const ActiveBlock = enum { none, text, thinking };
+/// Streaming-response assembly state.
+///
+/// Modeled on `provider_openai_chat.zig`'s `StreamState`: one active
+/// text/thinking block at a time, plus a set of in-progress tool calls. The
+/// Responses protocol is friendlier than Chat Completions here — every
+/// function-call event carries an explicit `item_id` (and `output_index`),
+/// and the lifecycle is spelled out by `output_item.added` →
+/// `function_call_arguments.delta`* → `function_call_arguments.done`/
+/// `output_item.done` → `response.completed`. So tool calls are keyed by
+/// `item_id` (the stable identity) and need none of Chat Completions'
+/// contiguity inference.
+///
+/// Argument-accumulation rule (the crux of the tool-input correctness): the
+/// concatenation of `function_call_arguments.delta` fragments is the reliable
+/// source of the tool input. A terminal `function_call_arguments.done` /
+/// `output_item.done` / `response.completed` event also restates the full
+/// `arguments`, but we apply it only as a *non-empty* override: a restated
+/// value can improve the accumulation (e.g. if it is more complete) but never
+/// wipe it. This matters because these events are observed to restate
+/// `arguments` as `""` once the value has already streamed via deltas, and an
+/// unconditional overwrite there destroys the real input — the original
+/// empty-tool-input bug. Empty arguments normalize to `"{}"` so a tool never
+/// receives an unparseable empty string.
const StreamState = struct {
allocator: Allocator,
started: bool = false,
@@ -233,14 +292,32 @@ const StreamState = struct {
active: ActiveBlock = .none,
block_index: usize = 0,
current_buf: conversation.TextualBlock = .empty,
+ current_thinking_signature: ?[]const u8 = null,
+ signature_origin: ?conversation.SignatureOrigin = null,
+ assistant_phase: ?AssistantPhase = null,
blocks: std.ArrayList(conversation.ContentBlock) = .empty,
- /// In-progress tool calls keyed by their streaming `item_id`.
- tools: std.StringArrayHashMapUnmanaged(ToolUseInProgress) = .empty,
+ /// In-progress and completed tool calls, keyed by `output_index`, in
+ /// first-seen order. Completed calls are retained (flagged `closed`) so
+ /// the terminal `response.completed`, which restates every output item,
+ /// does not re-emit a call that `output_item.done` already closed.
+ ///
+ /// We key by `output_index`, NOT the per-item `item_id`. The OpenAI spec
+ /// promises a stable `item.id` per output item, but the GitHub Copilot
+ /// Responses proxy emits a *fresh, opaque `item.id` on every event* for
+ /// the same call (`output_item.added`, each `…arguments.delta`,
+ /// `output_item.done`, and the `response.completed` restatement all
+ /// differ). Keying by `item_id` there made dedup miss, so one call
+ /// fanned out into three identical ToolUse blocks sharing one `call_id`.
+ /// `output_index` is stable across all of a call's events on both
+ /// backends — and is what `provider_openai_chat` keys on, too.
+ tools: std.AutoArrayHashMapUnmanaged(usize, ToolUseInProgress) = .empty,
usage: ?provider_mod.Usage = null,
stream_error_message: ?[]u8 = null,
const ToolUseInProgress = struct {
block_index: usize,
+ /// Set once `block_complete` has been emitted for this call.
+ closed: bool = false,
id_buf: conversation.TextualBlock = .empty, // call_id
name_buf: conversation.TextualBlock = .empty,
arguments: conversation.TextualBlock = .empty,
@@ -258,11 +335,12 @@ const StreamState = struct {
fn deinit(self: *StreamState) void {
self.current_buf.deinit(self.allocator);
+ if (self.current_thinking_signature) |s| self.allocator.free(s);
+ if (self.signature_origin) |*o| o.deinit(self.allocator);
for (self.blocks.items) |*b| b.deinit(self.allocator);
self.blocks.deinit(self.allocator);
var it = self.tools.iterator();
while (it.next()) |e| {
- self.allocator.free(e.key_ptr.*);
e.value_ptr.deinit(self.allocator);
}
self.tools.deinit(self.allocator);
@@ -284,10 +362,11 @@ const StreamState = struct {
if (self.active == .none) return;
const block: conversation.ContentBlock = switch (self.active) {
.text => .{ .Text = self.current_buf },
- .thinking => .{ .Thinking = .{ .text = self.current_buf } },
+ .thinking => .{ .Thinking = .{ .text = self.current_buf, .signature = self.current_thinking_signature } },
.none => unreachable,
};
self.current_buf = .empty;
+ self.current_thinking_signature = null;
try self.blocks.append(self.allocator, block);
try out.push(.{ .block_complete = .{
.index = self.block_index,
@@ -296,6 +375,19 @@ const StreamState = struct {
self.active = .none;
}
+ fn setThinkingSignature(self: *StreamState, signature: []const u8) !void {
+ if (self.current_thinking_signature) |old| self.allocator.free(old);
+ self.current_thinking_signature = try self.allocator.dupe(u8, signature);
+ }
+
+ fn setAssistantPhase(self: *StreamState, phase: []const u8) void {
+ if (std.mem.eql(u8, phase, "commentary")) {
+ self.assistant_phase = .commentary;
+ } else if (std.mem.eql(u8, phase, "final_answer")) {
+ self.assistant_phase = .final_answer;
+ }
+ }
+
fn openBlock(self: *StreamState, new_active: ActiveBlock, out: *EventQueue) !void {
std.debug.assert(new_active == .text or new_active == .thinking);
if (self.active == new_active) return;
@@ -320,24 +412,46 @@ const StreamState = struct {
} });
}
- /// Open a new ToolUse block for a `function_call` output item.
+ /// Normalize a final/seed arguments string: an empty value becomes `"{}"`
+ /// so a tool never sees an unparseable empty input.
+ fn normalizedArguments(arguments: []const u8) []const u8 {
+ return if (arguments.len == 0) "{}" else arguments;
+ }
+
+ /// Resolve an in-progress (not-yet-closed) tool call by `output_index`.
+ /// Returns null for an unknown index or an already-closed call.
+ fn lookupTool(self: *StreamState, output_index: ?usize) ?*ToolUseInProgress {
+ const idx = output_index orelse return null;
+ const tu = self.tools.getPtr(idx) orelse return null;
+ if (tu.closed) return null;
+ return tu;
+ }
+
+ /// Open a ToolUse block for a `function_call` output item. No-op if a
+ /// call at this `output_index` already exists (open or closed): the call
+ /// is announced once by `output_item.added`, then restated by
+ /// `output_item.done` and again by `response.completed` — and on the
+ /// Copilot proxy each restatement carries a different `item_id`, so
+ /// `output_index` is the only reliable dedup key.
fn openToolUse(
self: *StreamState,
out: *EventQueue,
- item_id: []const u8,
+ output_index: ?usize,
call_id: ?[]const u8,
name: ?[]const u8,
) !void {
+ const idx = output_index orelse return;
+ if (self.tools.contains(idx)) return;
// Close any active text/thinking block so the tool gets its own index.
if (self.active != .none) {
try self.closeActive(out);
self.block_index += 1;
}
var tu: ToolUseInProgress = .{ .block_index = self.block_index };
+ errdefer tu.deinit(self.allocator);
if (call_id) |c| try tu.id_buf.appendSlice(self.allocator, c);
if (name) |nm| try tu.name_buf.appendSlice(self.allocator, nm);
- const key = try self.allocator.dupe(u8, item_id);
- try self.tools.put(self.allocator, key, tu);
+ try self.tools.put(self.allocator, idx, tu);
try out.push(.{ .block_start = .{ .block_type = .ToolUse, .index = self.block_index } });
if (call_id != null and name != null) {
@@ -350,8 +464,11 @@ const StreamState = struct {
self.block_index += 1;
}
- fn appendToolArgs(self: *StreamState, out: *EventQueue, item_id: []const u8, delta: []const u8) !void {
- const tu = self.tools.getPtr(item_id) orelse return;
+ /// Append a streaming argument fragment to the matching call. Fragments
+ /// for an unknown or already-closed call are dropped (the protocol always
+ /// opens a call before streaming its arguments).
+ fn appendToolArgs(self: *StreamState, out: *EventQueue, output_index: ?usize, delta: []const u8) !void {
+ const tu = self.lookupTool(output_index) orelse return;
try tu.arguments.appendSlice(self.allocator, delta);
try out.push(.{ .content_delta = .{
.index = tu.block_index,
@@ -359,36 +476,51 @@ const StreamState = struct {
} });
}
- /// Close a ToolUse block on `output_item.done`. `final_args` (when the
- /// done event carries the full arguments) overrides the accumulated ones.
+ /// Override the matching call's arguments with a restated full value, but
+ /// only when non-empty — terminal events frequently restate already-
+ /// streamed arguments as `""`, which must not wipe the accumulated input.
+ fn setToolArgs(self: *StreamState, output_index: ?usize, arguments: []const u8) !void {
+ if (arguments.len == 0) return;
+ const tu = self.lookupTool(output_index) orelse return;
+ tu.arguments.clearRetainingCapacity();
+ try tu.arguments.appendSlice(self.allocator, arguments);
+ }
+
+ /// Close a ToolUse block. `final_args` (when a `done`/`completed` event
+ /// restates the full arguments) overrides the accumulated value only when
+ /// non-empty. A call whose identity never resolved (missing id or name)
+ /// is dropped. Idempotent: closing an already-closed call is a no-op.
fn closeToolUse(
self: *StreamState,
out: *EventQueue,
- item_id: []const u8,
+ output_index: ?usize,
final_args: ?[]const u8,
) !void {
- const entry = self.tools.fetchSwapRemove(item_id) orelse return;
- self.allocator.free(entry.key);
- var tu = entry.value;
+ const tu = self.lookupTool(output_index) orelse return;
+ if (final_args) |fa| {
+ if (fa.len > 0) {
+ tu.arguments.clearRetainingCapacity();
+ try tu.arguments.appendSlice(self.allocator, fa);
+ }
+ }
+ tu.closed = true;
if (tu.id_buf.items.len == 0 or tu.name_buf.items.len == 0) {
- tu.deinit(self.allocator);
+ // Identity never resolved — nothing dispatchable. Free the buffers
+ // now; the (empty) entry stays in the map for dedup + final free.
+ tu.id_buf.clearAndFree(self.allocator);
+ tu.name_buf.clearAndFree(self.allocator);
+ tu.arguments.clearAndFree(self.allocator);
return;
}
- if (final_args) |fa| {
- tu.arguments.clearRetainingCapacity();
- try tu.arguments.appendSlice(self.allocator, fa);
- }
- decodeNameInPlace(&tu.name_buf);
- const id_owned = try tu.id_buf.toOwnedSlice(self.allocator);
- const name_owned = try tu.name_buf.toOwnedSlice(self.allocator);
+ const input = try conversation.textualBlockFromSlice(self.allocator, normalizedArguments(tu.arguments.items));
+ decodeNameInPlace(&tu.name_buf);
const block: conversation.ContentBlock = .{ .ToolUse = .{
- .id = id_owned,
- .name = name_owned,
- .input = tu.arguments,
+ .id = try self.allocator.dupe(u8, tu.id_buf.items),
+ .name = try self.allocator.dupe(u8, tu.name_buf.items),
+ .input = input,
} };
- tu.arguments = .empty;
try self.blocks.append(self.allocator, block);
try out.push(.{ .block_complete = .{
.index = tu.block_index,
@@ -402,19 +534,38 @@ const StreamState = struct {
try self.closeActive(out);
// Close any tool calls that never received an explicit done event.
var it = self.tools.iterator();
- var leftover: std.ArrayList([]const u8) = .empty;
- defer leftover.deinit(self.allocator);
- while (it.next()) |e| leftover.append(self.allocator, e.key_ptr.*) catch {};
- for (leftover.items) |k| try self.closeToolUse(out, k, null);
+ while (it.next()) |e| {
+ if (!e.value_ptr.closed) try self.closeToolUse(out, e.key_ptr.*, null);
+ }
const moved_blocks = try self.blocks.toOwnedSlice(self.allocator);
defer self.allocator.free(moved_blocks);
+ if (self.signature_origin) |origin| {
+ try conversation.setThinkingOrigins(
+ self.allocator,
+ moved_blocks,
+ origin.api_style,
+ origin.base_url,
+ origin.model,
+ );
+ }
try conv.addAssistantMessage(moved_blocks, self.usage);
+ if (self.assistant_phase) |phase| {
+ const md = switch (phase) {
+ .commentary => openai_phase_commentary_metadata,
+ .final_answer => openai_phase_final_answer_metadata,
+ };
+ conv.messages.items[conv.messages.items.len - 1].metadata = try conv.allocator.dupe(u8, md);
+ }
const msg = conv.messages.items[conv.messages.items.len - 1];
try out.push(.{ .message_complete = .{ .message = msg, .usage = self.usage } });
}
};
+const AssistantPhase = enum { commentary, final_answer };
+const openai_phase_commentary_metadata = "{\"openai_responses_phase\":\"commentary\"}";
+const openai_phase_final_answer_metadata = "{\"openai_responses_phase\":\"final_answer\"}";
+
/// Handle one parsed event. Returns true when the stream is terminal
/// (`response.completed`) so the caller can finalize.
fn handleEvent(
@@ -445,23 +596,45 @@ fn handleEvent(
if (ev.item_type) |it| {
if (std.mem.eql(u8, it, "function_call")) {
try state.ensureStarted(out);
- if (ev.item_id) |id| try state.openToolUse(out, id, ev.call_id, ev.name);
+ try state.openToolUse(out, ev.output_index, ev.call_id, ev.name);
+ // `output_item.added` sometimes seeds the full args.
+ if (ev.arguments) |args| try state.setToolArgs(ev.output_index, args);
}
}
},
.function_call_arguments_delta => {
- if (ev.item_id) |id| {
- if (ev.delta) |d| try state.appendToolArgs(out, id, d);
- }
+ if (ev.delta) |d| try state.appendToolArgs(out, ev.output_index, d);
+ },
+ .function_call_arguments_done => {
+ if (ev.arguments) |args| try state.setToolArgs(ev.output_index, args);
},
.output_item_done => {
if (ev.item_type) |it| {
if (std.mem.eql(u8, it, "function_call")) {
- if (ev.item_id) |id| try state.closeToolUse(out, id, ev.arguments);
+ try state.ensureStarted(out);
+ try state.openToolUse(out, ev.output_index, ev.call_id, ev.name);
+ try state.closeToolUse(out, ev.output_index, ev.arguments);
+ } else if (std.mem.eql(u8, it, "reasoning")) {
+ if (ev.reasoning_item_json) |sig| {
+ try state.ensureStarted(out);
+ try state.openBlock(.thinking, out);
+ try state.setThinkingSignature(sig);
+ try state.closeActive(out);
+ state.block_index += 1;
+ }
+ } else if (std.mem.eql(u8, it, "message")) {
+ if (ev.item_phase) |phase| state.setAssistantPhase(phase);
}
}
},
.completed => {
+ // `response.completed` restates every output item; open+close any
+ // function call not already emitted via `output_item.done`.
+ for (ev.completed_items) |item| {
+ try state.ensureStarted(out);
+ try state.openToolUse(out, item.output_index, item.call_id, item.name);
+ try state.closeToolUse(out, item.output_index, item.arguments);
+ }
if (ev.usage) |u| {
const cached = u.cached_tokens;
const total_in = u.input_tokens;
@@ -593,13 +766,13 @@ test "responses stream: function call assembles a ToolUse" {
defer rec.deinit();
const events = [_][]const u8{
- \\{"type":"response.output_item.added","item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read"}}
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read"}}
,
- \\{"type":"response.function_call_arguments.delta","item_id":"fc_1","delta":"{\"path\":"}
+ \\{"type":"response.function_call_arguments.delta","output_index":0,"item_id":"fc_1","delta":"{\"path\":"}
,
- \\{"type":"response.function_call_arguments.delta","item_id":"fc_1","delta":"\"a\"}"}
+ \\{"type":"response.function_call_arguments.delta","output_index":0,"item_id":"fc_1","delta":"\"a\"}"}
,
- \\{"type":"response.output_item.done","item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":"{\"path\":\"a\"}"}}
+ \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":"{\"path\":\"a\"}"}}
,
\\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
,
@@ -628,6 +801,195 @@ test "responses stream: function call assembles a ToolUse" {
for (expect, rec.events.items) |w, g| try testing.expectEqualStrings(w, g);
}
+test "responses stream: function call arguments can be keyed by output_index" {
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "call it");
+
+ const events = [_][]const u8{
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read"}}
+ ,
+ \\{"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"path\":"}
+ ,
+ \\{"type":"response.function_call_arguments.delta","output_index":0,"delta":"\"a\"}"}
+ ,
+ \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":""}}
+ ,
+ \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
+ ,
+ };
+ try runStreamedTurn(allocator, &conv, null, &events);
+
+ const tu = conv.messages.items[1].content.items[0].ToolUse;
+ try testing.expectEqualStrings("call_9", tu.id);
+ try testing.expectEqualStrings("std.read", tu.name);
+ try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);
+}
+
+test "responses stream: function call arguments on item added are retained" {
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "call it");
+
+ const events = [_][]const u8{
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__bash","arguments":"{\"command\":\"ls\"}"}}
+ ,
+ \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
+ ,
+ };
+ try runStreamedTurn(allocator, &conv, null, &events);
+
+ const tu = conv.messages.items[1].content.items[0].ToolUse;
+ try testing.expectEqualStrings("call_9", tu.id);
+ try testing.expectEqualStrings("std.bash", tu.name);
+ try testing.expectEqualStrings("{\"command\":\"ls\"}", tu.input.items);
+}
+
+test "responses stream: function call arguments done supplies final input" {
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "call it");
+
+ const events = [_][]const u8{
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read"}}
+ ,
+ \\{"type":"response.function_call_arguments.done","output_index":0,"arguments":"{\"path\":\"a\"}"}
+ ,
+ \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":""}}
+ ,
+ \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
+ ,
+ };
+ try runStreamedTurn(allocator, &conv, null, &events);
+
+ const tu = conv.messages.items[1].content.items[0].ToolUse;
+ try testing.expectEqualStrings("call_9", tu.id);
+ try testing.expectEqualStrings("std.read", tu.name);
+ try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);
+}
+
+test "responses stream: completed output supplies final function call input" {
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "call it");
+
+ const events = [_][]const u8{
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":""}}
+ ,
+ \\{"type":"response.completed","response":{"output":[{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":"{\"path\":\"a\"}","status":"completed"}],"usage":{"input_tokens":3,"output_tokens":1}}}
+ ,
+ };
+ try runStreamedTurn(allocator, &conv, null, &events);
+
+ const tu = conv.messages.items[1].content.items[0].ToolUse;
+ try testing.expectEqualStrings("call_9", tu.id);
+ try testing.expectEqualStrings("std.read", tu.name);
+ try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);
+}
+
+test "responses stream: completed output does not duplicate closed function call" {
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "call it");
+
+ const events = [_][]const u8{
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":""}}
+ ,
+ \\{"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"path\":\"a\"}"}
+ ,
+ \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":""}}
+ ,
+ \\{"type":"response.completed","response":{"output":[{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":"{\"path\":\"a\"}","status":"completed"}],"usage":{"input_tokens":3,"output_tokens":1}}}
+ ,
+ };
+ try runStreamedTurn(allocator, &conv, null, &events);
+
+ const asst = conv.messages.items[1];
+ try testing.expectEqual(@as(usize, 1), asst.content.items.len);
+ try testing.expectEqualStrings("{\"path\":\"a\"}", asst.content.items[0].ToolUse.input.items);
+}
+
+test "responses stream: empty function call arguments normalize to object" {
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "call it");
+
+ const events = [_][]const u8{
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"ping","arguments":""}}
+ ,
+ \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"ping","arguments":""}}
+ ,
+ \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
+ ,
+ };
+ try runStreamedTurn(allocator, &conv, null, &events);
+
+ const tu = conv.messages.items[1].content.items[0].ToolUse;
+ try testing.expectEqualStrings("call_9", tu.id);
+ try testing.expectEqualStrings("ping", tu.name);
+ try testing.expectEqualStrings("{}", tu.input.items);
+}
+
+test "responses stream: finalization closes an open function call" {
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "call it");
+
+ const events = [_][]const u8{
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__bash"}}
+ ,
+ \\{"type":"response.function_call_arguments.delta","output_index":0,"item_id":"fc_1","delta":"{\"command\":\"ls\"}"}
+ ,
+ \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
+ ,
+ };
+ try runStreamedTurn(allocator, &conv, null, &events);
+
+ const tu = conv.messages.items[1].content.items[0].ToolUse;
+ try testing.expectEqualStrings("call_9", tu.id);
+ try testing.expectEqualStrings("std.bash", tu.name);
+ try testing.expectEqualStrings("{\"command\":\"ls\"}", tu.input.items);
+}
+
+test "responses stream: streamed args survive an empty arguments.done" {
+ // Codex streams the arguments as `function_call_arguments.delta`
+ // fragments and then emits a terminal `function_call_arguments.done`
+ // whose `arguments` field is empty (the value already arrived via the
+ // deltas). The accumulated input must NOT be wiped by that empty done.
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "call it");
+
+ const events = [_][]const u8{
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","status":"in_progress","arguments":"","call_id":"call_9","name":"std__read"}}
+ ,
+ \\{"type":"response.function_call_arguments.delta","item_id":"fc_1","output_index":0,"delta":"{\"path\":"}
+ ,
+ \\{"type":"response.function_call_arguments.delta","item_id":"fc_1","output_index":0,"delta":"\"a\"}"}
+ ,
+ \\{"type":"response.function_call_arguments.done","item_id":"fc_1","output_index":0,"arguments":""}
+ ,
+ \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","status":"completed","arguments":"","call_id":"call_9","name":"std__read"}}
+ ,
+ \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
+ ,
+ };
+ try runStreamedTurn(allocator, &conv, null, &events);
+
+ const tu = conv.messages.items[1].content.items[0].ToolUse;
+ try testing.expectEqualStrings("call_9", tu.id);
+ try testing.expectEqualStrings("std.read", tu.name);
+ try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);
+}
+
test "responses stream: text then tool call keeps block order" {
const allocator = testing.allocator;
var conv = conversation.Conversation.init(allocator);
@@ -635,13 +997,13 @@ test "responses stream: text then tool call keeps block order" {
try addUserText(&conv, "go");
const events = [_][]const u8{
- \\{"type":"response.output_text.delta","item_id":"msg_1","delta":"working"}
+ \\{"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"delta":"working"}
,
- \\{"type":"response.output_item.added","item":{"type":"function_call","id":"fc_1","call_id":"c1","name":"ping"}}
+ \\{"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","id":"fc_1","call_id":"c1","name":"ping"}}
,
- \\{"type":"response.function_call_arguments.delta","item_id":"fc_1","delta":"{}"}
+ \\{"type":"response.function_call_arguments.delta","output_index":1,"item_id":"fc_1","delta":"{}"}
,
- \\{"type":"response.output_item.done","item":{"type":"function_call","id":"fc_1","call_id":"c1","name":"ping","arguments":"{}"}}
+ \\{"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","id":"fc_1","call_id":"c1","name":"ping","arguments":"{}"}}
,
\\{"type":"response.completed","response":{"usage":{"input_tokens":1,"output_tokens":1}}}
,
@@ -654,3 +1016,171 @@ test "responses stream: text then tool call keeps block order" {
try testing.expectEqualStrings("ping", asst.content.items[1].ToolUse.name);
try testing.expectEqualStrings("{}", asst.content.items[1].ToolUse.input.items);
}
+
+/// Drive raw SSE bytes through the same path `produce` uses live: feed
+/// arbitrary byte chunks to the `SSEParser`, decode each event, and drain the
+/// `EventQueue` after every chunk (which resets its arena). This exercises
+/// payload lifetimes the all-at-once `runStreamedTurn` helper does not.
+fn runRawStream(
+ allocator: Allocator,
+ conv: *conversation.Conversation,
+ raw: []const u8,
+ chunk_len: usize,
+) !void {
+ var parser = sse_mod.SSEParser.init(allocator);
+ defer parser.deinit();
+ var state: StreamState = .init(allocator);
+ defer state.deinit();
+ var queue = EventQueue.init(allocator);
+ defer queue.deinit();
+
+ var off: usize = 0;
+ var done = false;
+ while (off < raw.len and !done) {
+ const end = @min(off + chunk_len, raw.len);
+ const events = try parser.feed(raw[off..end]);
+ defer parser.freeEvents(events);
+ off = end;
+ for (events) |payload| {
+ if (try handleEvent(allocator, payload, &state, &queue)) {
+ done = true;
+ break;
+ }
+ }
+ // Drain (and reset the arena) between chunks, as the agent loop does.
+ while (queue.pop()) |_| {}
+ }
+ try state.finalize(&queue, conv);
+ while (queue.pop()) |_| {}
+}
+
+test "responses stream: realistic codex SSE assembles tool input across chunks" {
+ // A representative ChatGPT-Codex function-call stream: explicit `event:`
+ // lines, args streamed as `function_call_arguments.delta` fragments, and
+ // terminal `done`/`completed` events that restate `arguments` as "".
+ // Sliced into small byte chunks so events straddle reads and the queue
+ // arena resets mid-stream.
+ const allocator = testing.allocator;
+
+ const raw =
+ "event: response.created\n" ++
+ "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\"}}\n\n" ++
+ "event: response.output_item.added\n" ++
+ "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"fc_1\",\"type\":\"function_call\",\"status\":\"in_progress\",\"arguments\":\"\",\"call_id\":\"call_9\",\"name\":\"std__read\"}}\n\n" ++
+ "event: response.function_call_arguments.delta\n" ++
+ "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"output_index\":0,\"delta\":\"{\\\"path\\\":\"}\n\n" ++
+ "event: response.function_call_arguments.delta\n" ++
+ "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"output_index\":0,\"delta\":\"\\\"/tmp/x\\\"}\"}\n\n" ++
+ "event: response.function_call_arguments.done\n" ++
+ "data: {\"type\":\"response.function_call_arguments.done\",\"item_id\":\"fc_1\",\"output_index\":0,\"arguments\":\"\"}\n\n" ++
+ "event: response.output_item.done\n" ++
+ "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"id\":\"fc_1\",\"type\":\"function_call\",\"status\":\"completed\",\"arguments\":\"\",\"call_id\":\"call_9\",\"name\":\"std__read\"}}\n\n" ++
+ "event: response.completed\n" ++
+ "data: {\"type\":\"response.completed\",\"response\":{\"output\":[{\"id\":\"fc_1\",\"type\":\"function_call\",\"status\":\"completed\",\"arguments\":\"{\\\"path\\\":\\\"/tmp/x\\\"}\",\"call_id\":\"call_9\",\"name\":\"std__read\"}],\"usage\":{\"input_tokens\":10,\"output_tokens\":5}}}\n\n";
+
+ // Try several chunk sizes so events land on different read boundaries.
+ for ([_]usize{ 1, 7, 64, raw.len }) |chunk_len| {
+ var c = conversation.Conversation.init(allocator);
+ defer c.deinit();
+ try addUserText(&c, "read the file");
+ try runRawStream(allocator, &c, raw, chunk_len);
+
+ const asst = c.messages.items[1];
+ try testing.expectEqual(@as(usize, 1), asst.content.items.len);
+ const tu = asst.content.items[0].ToolUse;
+ try testing.expectEqualStrings("call_9", tu.id);
+ try testing.expectEqualStrings("std.read", tu.name);
+ try testing.expectEqualStrings("{\"path\":\"/tmp/x\"}", tu.input.items);
+ }
+}
+
+test "responses stream: parallel function calls assemble distinct ToolUse blocks" {
+ // The model emits several function calls in one turn, each with its own
+ // item_id / output_index / call_id, with their argument deltas
+ // interleaved. Each must become its own ToolUse block with the right
+ // input — no cross-talk between calls.
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "ls three dirs");
+
+ const events = [_][]const u8{
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_a","call_id":"call_a","name":"std__shell","arguments":""}}
+ ,
+ \\{"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","id":"fc_b","call_id":"call_b","name":"std__shell","arguments":""}}
+ ,
+ \\{"type":"response.function_call_arguments.delta","item_id":"fc_a","output_index":0,"delta":"{\"command\":\"ls a\"}"}
+ ,
+ \\{"type":"response.function_call_arguments.delta","item_id":"fc_b","output_index":1,"delta":"{\"command\":"}
+ ,
+ \\{"type":"response.output_item.added","output_index":2,"item":{"type":"function_call","id":"fc_c","call_id":"call_c","name":"std__shell","arguments":""}}
+ ,
+ \\{"type":"response.function_call_arguments.delta","item_id":"fc_b","output_index":1,"delta":"\"ls b\"}"}
+ ,
+ \\{"type":"response.function_call_arguments.delta","item_id":"fc_c","output_index":2,"delta":"{\"command\":\"ls c\"}"}
+ ,
+ \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_a","call_id":"call_a","name":"std__shell","arguments":""}}
+ ,
+ \\{"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","id":"fc_b","call_id":"call_b","name":"std__shell","arguments":""}}
+ ,
+ \\{"type":"response.output_item.done","output_index":2,"item":{"type":"function_call","id":"fc_c","call_id":"call_c","name":"std__shell","arguments":""}}
+ ,
+ \\{"type":"response.completed","response":{"output":[{"type":"function_call","id":"fc_a","call_id":"call_a","name":"std__shell","arguments":"{\"command\":\"ls a\"}"},{"type":"function_call","id":"fc_b","call_id":"call_b","name":"std__shell","arguments":"{\"command\":\"ls b\"}"},{"type":"function_call","id":"fc_c","call_id":"call_c","name":"std__shell","arguments":"{\"command\":\"ls c\"}"}],"usage":{"input_tokens":5,"output_tokens":9}}}
+ ,
+ };
+ try runStreamedTurn(allocator, &conv, null, &events);
+
+ const asst = conv.messages.items[1];
+ try testing.expectEqual(@as(usize, 3), asst.content.items.len);
+ try testing.expectEqualStrings("call_a", asst.content.items[0].ToolUse.id);
+ try testing.expectEqualStrings("{\"command\":\"ls a\"}", asst.content.items[0].ToolUse.input.items);
+ try testing.expectEqualStrings("call_b", asst.content.items[1].ToolUse.id);
+ try testing.expectEqualStrings("{\"command\":\"ls b\"}", asst.content.items[1].ToolUse.input.items);
+ try testing.expectEqualStrings("call_c", asst.content.items[2].ToolUse.id);
+ try testing.expectEqualStrings("{\"command\":\"ls c\"}", asst.content.items[2].ToolUse.input.items);
+ for (asst.content.items) |b| try testing.expectEqualStrings("std.shell", b.ToolUse.name);
+}
+
+test "responses stream: one call with a mutating item_id stays a single block" {
+ // Regression for the GitHub Copilot Responses proxy: it emits a *fresh*,
+ // opaque `item_id` on every event for the SAME call — the
+ // `output_item.added`, each `…arguments.delta`, the `…arguments.done`, the
+ // `output_item.done`, and the `response.completed` restatement all carry
+ // different `item_id`s. Only `output_index` (and `call_id`) are stable.
+ // Keying tool calls by `item_id` made dedup miss, fanning this single call
+ // out into three identical ToolUse blocks sharing one `call_id` (which in
+ // turn stranded two UI result boxes at the `(…)` placeholder). Keying by
+ // `output_index` must collapse it back to exactly one block. A leading
+ // reasoning item (also with mutating ids) must not spawn a phantom tool.
+ const allocator = testing.allocator;
+ var conv = conversation.Conversation.init(allocator);
+ defer conv.deinit();
+ try addUserText(&conv, "what's in the current directory?");
+
+ const events = [_][]const u8{
+ \\{"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"rsn_AAAA"}}
+ ,
+ \\{"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","id":"rsn_BBBB"}}
+ ,
+ \\{"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","id":"itm_AAAA","call_id":"call_KOEB","name":"std__shell","arguments":""}}
+ ,
+ \\{"type":"response.function_call_arguments.delta","output_index":1,"item_id":"itm_BBBB","delta":"{\"command\":"}
+ ,
+ \\{"type":"response.function_call_arguments.delta","output_index":1,"item_id":"itm_CCCC","delta":"\"pwd && ls -la\"}"}
+ ,
+ \\{"type":"response.function_call_arguments.done","output_index":1,"item_id":"itm_DDDD","arguments":""}
+ ,
+ \\{"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","id":"itm_EEEE","call_id":"call_KOEB","name":"std__shell","arguments":"{\"command\":\"pwd && ls -la\"}"}}
+ ,
+ \\{"type":"response.completed","response":{"output":[{"type":"reasoning","id":"rsn_CCCC"},{"type":"function_call","id":"itm_FFFF","call_id":"call_KOEB","name":"std__shell","arguments":"{\"command\":\"pwd && ls -la\"}"}],"usage":{"input_tokens":5,"output_tokens":9}}}
+ ,
+ };
+ try runStreamedTurn(allocator, &conv, null, &events);
+
+ const asst = conv.messages.items[1];
+ try testing.expectEqual(@as(usize, 1), asst.content.items.len);
+ const tu = asst.content.items[0].ToolUse;
+ try testing.expectEqualStrings("call_KOEB", tu.id);
+ try testing.expectEqualStrings("std.shell", tu.name);
+ try testing.expectEqualStrings("{\"command\":\"pwd && ls -la\"}", tu.input.items);
+}
diff --git a/libpanto/src/public.zig b/libpanto/src/public.zig
index 339c6f1..8a55e47 100644
--- a/libpanto/src/public.zig
+++ b/libpanto/src/public.zig
@@ -131,6 +131,7 @@ pub const TextualBlock = conversation_mod.TextualBlock;
/// for binding code to construct `.Text`/`.Thinking`/tool-result text when
/// assembling `ContentBlock`s by hand (e.g. for conversation resumption).
pub const textualBlockFromSlice = conversation_mod.textualBlockFromSlice;
+pub const SignatureOrigin = conversation_mod.SignatureOrigin;
pub const ThinkingBlock = conversation_mod.ThinkingBlock;
pub const ToolUseBlock = conversation_mod.ToolUseBlock;
pub const ToolResultBlock = conversation_mod.ToolResultBlock;
diff --git a/scripts/analyze_pi_sessions.py b/scripts/analyze_pi_sessions.py
new file mode 100644
index 0000000..ffa9f00
--- /dev/null
+++ b/scripts/analyze_pi_sessions.py
@@ -0,0 +1,458 @@
+#!/usr/bin/env python3
+
+import argparse
+import json
+import math
+from collections import Counter
+from pathlib import Path
+from statistics import mean, median
+from typing import Any
+
+
+BUCKETS = ("input", "output", "cache_read", "cache_write")
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser(
+ description="Analyze Pi agent session logs for token mix and cache behavior."
+ )
+ parser.add_argument(
+ "root",
+ nargs="?",
+ default="~/.pi/agent/sessions",
+ help="Directory containing Pi session .jsonl files.",
+ )
+ parser.add_argument(
+ "--json-out",
+ help="Optional path to write the full report as JSON.",
+ )
+ return parser.parse_args()
+
+
+def usage_value(usage: dict[str, Any], *keys: str) -> int:
+ for key in keys:
+ value = usage.get(key)
+ if value is None:
+ continue
+ if isinstance(value, bool):
+ return int(value)
+ if isinstance(value, (int, float)):
+ return int(value)
+ return 0
+
+
+def safe_div(numerator: float, denominator: float) -> float | None:
+ if denominator == 0:
+ return None
+ return numerator / denominator
+
+
+def percentile(values: list[float], p: float) -> float | None:
+ if not values:
+ return None
+ if len(values) == 1:
+ return values[0]
+ ordered = sorted(values)
+ rank = (len(ordered) - 1) * p
+ low = math.floor(rank)
+ high = math.ceil(rank)
+ if low == high:
+ return ordered[low]
+ weight = rank - low
+ return ordered[low] * (1 - weight) + ordered[high] * weight
+
+
+def summarize(values: list[float]) -> dict[str, float | None]:
+ return {
+ "count": len(values),
+ "mean": mean(values) if values else None,
+ "median": median(values) if values else None,
+ "p10": percentile(values, 0.10),
+ "p90": percentile(values, 0.90),
+ "min": min(values) if values else None,
+ "max": max(values) if values else None,
+ }
+
+
+def session_primary(counter: Counter[str]) -> str | None:
+ if not counter:
+ return None
+ return max(counter.items(), key=lambda item: (item[1], item[0]))[0]
+
+
+def format_number(value: float | None, digits: int = 3) -> str:
+ if value is None:
+ return "n/a"
+ return f"{value:,.{digits}f}"
+
+
+def format_pct(value: float | None, digits: int = 1) -> str:
+ if value is None:
+ return "n/a"
+ return f"{value * 100:.{digits}f}%"
+
+
+def analyze_session(path: Path) -> dict[str, Any]:
+ totals = {bucket: 0 for bucket in BUCKETS}
+ usage_messages = 0
+ assistant_messages = 0
+ models = Counter()
+ providers = Counter()
+ apis = Counter()
+ line_count = 0
+ parse_errors = 0
+
+ with path.open("r", encoding="utf-8") as handle:
+ for raw_line in handle:
+ line_count += 1
+ line = raw_line.strip()
+ if not line:
+ continue
+ try:
+ entry = json.loads(line)
+ except json.JSONDecodeError:
+ parse_errors += 1
+ continue
+
+ message = entry.get("message")
+ if not isinstance(message, dict):
+ continue
+
+ role = message.get("role")
+ if role == "assistant":
+ assistant_messages += 1
+
+ usage = entry.get("usage") or message.get("usage")
+ if role != "assistant" or not isinstance(usage, dict):
+ continue
+
+ usage_messages += 1
+ bucket_values = {
+ "input": usage_value(usage, "input", "input_tokens"),
+ "output": usage_value(usage, "output", "output_tokens"),
+ "cache_read": usage_value(
+ usage,
+ "cacheRead",
+ "cache_read",
+ "cache_read_input_tokens",
+ ),
+ "cache_write": usage_value(
+ usage,
+ "cacheWrite",
+ "cache_write",
+ "cache_creation_input_tokens",
+ ),
+ }
+ for bucket, value in bucket_values.items():
+ totals[bucket] += value
+
+ billable = sum(bucket_values.values())
+ if billable <= 0:
+ continue
+
+ model = (
+ entry.get("responseModel")
+ or entry.get("model")
+ or message.get("responseModel")
+ or message.get("model")
+ )
+ provider = entry.get("provider") or message.get("provider")
+ api = entry.get("api") or message.get("api")
+ if isinstance(model, str):
+ models[model] += billable
+ if isinstance(provider, str):
+ providers[provider] += billable
+ if isinstance(api, str):
+ apis[api] += billable
+
+ prompt_tokens = totals["input"] + totals["cache_read"] + totals["cache_write"]
+ billable_tokens = prompt_tokens + totals["output"]
+ cache_activity = totals["cache_read"] + totals["cache_write"]
+
+ shares = {
+ bucket: safe_div(totals[bucket], billable_tokens) for bucket in BUCKETS
+ }
+
+ return {
+ "path": str(path),
+ "lines": line_count,
+ "parse_errors": parse_errors,
+ "assistant_messages": assistant_messages,
+ "usage_messages": usage_messages,
+ "totals": totals,
+ "prompt_tokens": prompt_tokens,
+ "billable_tokens": billable_tokens,
+ "cache_activity_tokens": cache_activity,
+ "primary_model": session_primary(models),
+ "primary_provider": session_primary(providers),
+ "primary_api": session_primary(apis),
+ "metrics": {
+ "prompt_to_output_ratio": safe_div(prompt_tokens, totals["output"]),
+ "input_to_output_ratio": safe_div(totals["input"], totals["output"]),
+ "cache_hit_rate": safe_div(totals["cache_read"], cache_activity),
+ "cache_coverage": safe_div(cache_activity, prompt_tokens),
+ "prompt_reuse_rate": safe_div(totals["cache_read"], prompt_tokens),
+ "output_share": safe_div(totals["output"], billable_tokens),
+ "prompt_share": safe_div(prompt_tokens, billable_tokens),
+ **{f"{bucket}_share": value for bucket, value in shares.items()},
+ },
+ }
+
+
+def build_report(root: Path) -> dict[str, Any]:
+ session_files = sorted(root.rglob("*.jsonl"))
+ sessions = [analyze_session(path) for path in session_files]
+ analyzed_sessions = [s for s in sessions if s["billable_tokens"] > 0]
+
+ pooled_totals = {bucket: 0 for bucket in BUCKETS}
+ total_prompt = 0
+ total_billable = 0
+ total_usage_messages = 0
+ total_assistant_messages = 0
+ total_parse_errors = 0
+ provider_sessions = Counter()
+ model_sessions = Counter()
+
+ per_session_metric_values: dict[str, list[float]] = {}
+ cache_active_metric_values: dict[str, list[float]] = {}
+ cache_active_sessions = 0
+
+ for session in analyzed_sessions:
+ for bucket in BUCKETS:
+ pooled_totals[bucket] += session["totals"][bucket]
+ total_prompt += session["prompt_tokens"]
+ total_billable += session["billable_tokens"]
+ total_usage_messages += session["usage_messages"]
+ total_assistant_messages += session["assistant_messages"]
+ total_parse_errors += session["parse_errors"]
+ if session["primary_provider"]:
+ provider_sessions[session["primary_provider"]] += 1
+ if session["primary_model"]:
+ model_sessions[session["primary_model"]] += 1
+ for name, value in session["metrics"].items():
+ if value is None:
+ continue
+ per_session_metric_values.setdefault(name, []).append(value)
+ if session["cache_activity_tokens"] > 0 and name in {
+ "cache_hit_rate",
+ "cache_coverage",
+ "prompt_reuse_rate",
+ "cache_read_share",
+ "cache_write_share",
+ }:
+ cache_active_metric_values.setdefault(name, []).append(value)
+ per_session_metric_values.setdefault("billable_tokens", []).append(
+ session["billable_tokens"]
+ )
+ per_session_metric_values.setdefault("prompt_tokens", []).append(
+ session["prompt_tokens"]
+ )
+ per_session_metric_values.setdefault("output_tokens", []).append(
+ session["totals"]["output"]
+ )
+ if session["cache_activity_tokens"] > 0:
+ cache_active_sessions += 1
+
+ pooled_cache_activity = pooled_totals["cache_read"] + pooled_totals["cache_write"]
+ pooled_weights = {
+ bucket: safe_div(pooled_totals[bucket], total_billable) for bucket in BUCKETS
+ }
+
+ mean_session_weights = {
+ bucket: mean(per_session_metric_values.get(f"{bucket}_share", []))
+ if per_session_metric_values.get(f"{bucket}_share")
+ else None
+ for bucket in BUCKETS
+ }
+ median_session_weights = {
+ bucket: median(per_session_metric_values.get(f"{bucket}_share", []))
+ if per_session_metric_values.get(f"{bucket}_share")
+ else None
+ for bucket in BUCKETS
+ }
+
+ return {
+ "root": str(root),
+ "sessions_scanned": len(session_files),
+ "sessions_with_usage": len(analyzed_sessions),
+ "sessions_without_usage": len(session_files) - len(analyzed_sessions),
+ "assistant_messages": total_assistant_messages,
+ "assistant_messages_with_usage": total_usage_messages,
+ "parse_errors": total_parse_errors,
+ "cache_active_sessions": cache_active_sessions,
+ "pooled_totals": pooled_totals,
+ "pooled_metrics": {
+ "prompt_tokens": total_prompt,
+ "billable_tokens": total_billable,
+ "prompt_to_output_ratio": safe_div(total_prompt, pooled_totals["output"]),
+ "input_to_output_ratio": safe_div(
+ pooled_totals["input"], pooled_totals["output"]
+ ),
+ "cache_hit_rate": safe_div(
+ pooled_totals["cache_read"], pooled_cache_activity
+ ),
+ "cache_coverage": safe_div(pooled_cache_activity, total_prompt),
+ "prompt_reuse_rate": safe_div(pooled_totals["cache_read"], total_prompt),
+ "output_share": safe_div(pooled_totals["output"], total_billable),
+ "prompt_share": safe_div(total_prompt, total_billable),
+ **{f"{bucket}_share": value for bucket, value in pooled_weights.items()},
+ },
+ "recommended_weights": {
+ "pooled_token_mix": pooled_weights,
+ "mean_session_mix": mean_session_weights,
+ "median_session_mix": median_session_weights,
+ },
+ "per_session_distributions": {
+ name: summarize(values)
+ for name, values in sorted(per_session_metric_values.items())
+ },
+ "cache_active_session_distributions": {
+ name: summarize(values)
+ for name, values in sorted(cache_active_metric_values.items())
+ },
+ "top_primary_providers": provider_sessions.most_common(10),
+ "top_primary_models": model_sessions.most_common(10),
+ }
+
+
+def print_report(report: dict[str, Any]) -> None:
+ print("Pi Session Cost Analysis")
+ print("========================")
+ print(f"Root: {report['root']}")
+ print(
+ f"Sessions scanned: {report['sessions_scanned']} "
+ f"({report['sessions_with_usage']} with usage)"
+ )
+ print(
+ f"Assistant messages with usage: {report['assistant_messages_with_usage']} "
+ f"of {report['assistant_messages']}"
+ )
+ print(f"Cache-active sessions: {report['cache_active_sessions']}")
+ print()
+
+ pooled = report["pooled_metrics"]
+ totals = report["pooled_totals"]
+ weights = report["recommended_weights"]
+
+ print("Pooled Totals")
+ print("-------------")
+ print(f"Input: {totals['input']:,}")
+ print(f"Output: {totals['output']:,}")
+ print(f"Cache read: {totals['cache_read']:,}")
+ print(f"Cache write: {totals['cache_write']:,}")
+ print(f"Billable: {pooled['billable_tokens']:,}")
+ print()
+
+ print("Recommended Weight Vectors")
+ print("--------------------------")
+ for label, vector in (
+ ("Pooled token mix", weights["pooled_token_mix"]),
+ ("Mean session mix", weights["mean_session_mix"]),
+ ("Median session mix", weights["median_session_mix"]),
+ ):
+ print(
+ f"{label:18} "
+ f"input={format_pct(vector['input'])} "
+ f"output={format_pct(vector['output'])} "
+ f"cache_read={format_pct(vector['cache_read'])} "
+ f"cache_write={format_pct(vector['cache_write'])}"
+ )
+ print()
+
+ print("Key Corpus Metrics")
+ print("------------------")
+ print(
+ f"Prompt/output ratio: {format_number(pooled['prompt_to_output_ratio'])}x "
+ f"(prompt = input + cache_read + cache_write)"
+ )
+ print(
+ f"Uncached input/output ratio: {format_number(pooled['input_to_output_ratio'])}x"
+ )
+ print(f"Cache hit rate: {format_pct(pooled['cache_hit_rate'])}")
+ print(f"Cache coverage of prompt side: {format_pct(pooled['cache_coverage'])}")
+ print(f"Prompt reuse rate: {format_pct(pooled['prompt_reuse_rate'])}")
+ print()
+
+ print("Per-Session Distributions")
+ print("-------------------------")
+ metric_names = [
+ "billable_tokens",
+ "prompt_to_output_ratio",
+ "input_to_output_ratio",
+ "cache_hit_rate",
+ "cache_coverage",
+ "prompt_reuse_rate",
+ "input_share",
+ "output_share",
+ "cache_read_share",
+ "cache_write_share",
+ ]
+ for name in metric_names:
+ stats = report["per_session_distributions"].get(name)
+ if not stats:
+ continue
+ unit = "%" if (
+ name.endswith("_share")
+ or "rate" in name
+ or "coverage" in name
+ ) else ""
+ value_fmt = format_pct if unit == "%" else format_number
+ print(
+ f"{name:22} "
+ f"mean={value_fmt(stats['mean'])} "
+ f"median={value_fmt(stats['median'])} "
+ f"p10={value_fmt(stats['p10'])} "
+ f"p90={value_fmt(stats['p90'])}"
+ )
+ print()
+
+ if report["cache_active_session_distributions"]:
+ print("Cache-Active Session Distributions")
+ print("----------------------------------")
+ for name in [
+ "cache_hit_rate",
+ "cache_coverage",
+ "prompt_reuse_rate",
+ "cache_read_share",
+ "cache_write_share",
+ ]:
+ stats = report["cache_active_session_distributions"].get(name)
+ if not stats:
+ continue
+ print(
+ f"{name:22} "
+ f"mean={format_pct(stats['mean'])} "
+ f"median={format_pct(stats['median'])} "
+ f"p10={format_pct(stats['p10'])} "
+ f"p90={format_pct(stats['p90'])}"
+ )
+ print()
+
+ print("Top Primary Providers")
+ print("---------------------")
+ for provider, count in report["top_primary_providers"]:
+ print(f"{provider:20} {count}")
+ print()
+
+ print("Top Primary Models")
+ print("------------------")
+ for model, count in report["top_primary_models"]:
+ print(f"{model:35} {count}")
+
+
+def main() -> int:
+ args = parse_args()
+ root = Path(args.root).expanduser().resolve()
+ report = build_report(root)
+
+ if args.json_out:
+ out_path = Path(args.json_out).expanduser().resolve()
+ out_path.parent.mkdir(parents=True, exist_ok=True)
+ out_path.write_text(json.dumps(report, indent=2), encoding="utf-8")
+
+ print_report(report)
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/src/auth_manager.zig b/src/auth_manager.zig
index e26e28a..5504807 100644
--- a/src/auth_manager.zig
+++ b/src/auth_manager.zig
@@ -198,6 +198,11 @@ fn patchCredential(
c.base_url = base_url;
c.extra_headers = merged;
},
+ .openai_codex_responses => |*c| {
+ c.api_key = cred.api_key;
+ c.base_url = base_url;
+ c.extra_headers = merged;
+ },
}
}
diff --git a/src/config_file.zig b/src/config_file.zig
index 1823177..8f6725c 100644
--- a/src/config_file.zig
+++ b/src/config_file.zig
@@ -77,6 +77,10 @@ pub const Provider = struct {
/// the name used on the left of a `<provider>:<model>` reference.
name: []const u8,
style: APIStyle,
+ /// User-facing sub-dialect for styles that share a protocol family.
+ /// Currently only `style = "openai_responses"` supports `dialect = "codex"`,
+ /// which maps to internal `APIStyle.openai_codex_responses`.
+ dialect: ?[]const u8 = null,
base_url: []const u8,
/// The `auth.<name>` session this provider draws its credential from.
auth_name: []const u8,
@@ -90,6 +94,7 @@ pub const Provider = struct {
pub fn deinit(self: Provider, alloc: Allocator) void {
alloc.free(self.name);
+ if (self.dialect) |d| alloc.free(d);
alloc.free(self.base_url);
alloc.free(self.auth_name);
// `extra_headers` lives in the Config auth arena; not freed here.
@@ -213,6 +218,14 @@ pub fn buildProviderConfig(
.max_tokens = max_tokens,
.extra_headers = prov.extra_headers,
} },
+ .openai_codex_responses => return .{ .openai_codex_responses = .{
+ .api_key = api_key,
+ .base_url = prov.base_url,
+ .model = wire_model,
+ .reasoning = reasoning,
+ .max_tokens = max_tokens,
+ .extra_headers = prov.extra_headers,
+ } },
}
}
@@ -677,7 +690,16 @@ fn resolveProvider(
const style_str = (val.get("style") orelse return error.InvalidProvider).asString() orelse
return error.InvalidProvider;
- const style = std.meta.stringToEnum(APIStyle, style_str) orelse return error.InvalidProvider;
+ var style = std.meta.stringToEnum(APIStyle, style_str) orelse return error.InvalidProvider;
+ if (style == .openai_codex_responses) return error.InvalidProvider;
+ const dialect = if (tomlStr(val, "dialect")) |d| d else null;
+ if (dialect) |d| {
+ if (std.mem.eql(u8, style_str, "openai_responses") and std.mem.eql(u8, d, "codex")) {
+ style = .openai_codex_responses;
+ } else {
+ return error.InvalidProvider;
+ }
+ }
const base_url = (val.get("base_url") orelse return error.InvalidProvider).asString() orelse
return error.InvalidProvider;
@@ -699,6 +721,7 @@ fn resolveProvider(
return .{
.name = try allocator.dupe(u8, name),
.style = style,
+ .dialect = if (dialect) |d| try allocator.dupe(u8, d) else null,
.base_url = try allocator.dupe(u8, base_url),
.auth_name = try allocator.dupe(u8, auth_name),
.prompt_cache = prompt_cache,
@@ -1231,6 +1254,50 @@ test "resolve: provider extra_headers flow into the built provider config" {
try testing.expectEqualStrings("tok", pc.openai_chat.api_key);
}
+test "resolve: openai_responses codex dialect maps to internal API style" {
+ const a = testing.allocator;
+ var env = emptyEnv(a);
+ defer env.deinit();
+ const src =
+ \\[providers.codex]
+ \\style = "openai_responses"
+ \\dialect = "codex"
+ \\base_url = "https://chatgpt.com/backend-api"
+ \\auth = "k"
+ \\
+ \\[auth.k]
+ \\key = "tok"
+ ;
+ const doc = try parseDoc(a, src);
+ defer doc.deinit();
+ var cfg = try resolve(a, &env, doc.root);
+ defer cfg.deinit();
+
+ try testing.expectEqual(APIStyle.openai_codex_responses, cfg.provider("codex").?.style);
+ var defs = models_toml.ModelRegistry.init(a);
+ defer defs.deinit();
+ const pc = try buildProviderConfig(&cfg, &defs, .{ .provider = "codex", .model = "gpt-5.5" });
+ try testing.expectEqual(APIStyle.openai_codex_responses, pc.style());
+}
+
+test "resolve: internal openai_codex_responses style is not user-facing" {
+ const a = testing.allocator;
+ var env = emptyEnv(a);
+ defer env.deinit();
+ const src =
+ \\[providers.codex]
+ \\style = "openai_codex_responses"
+ \\base_url = "https://chatgpt.com/backend-api"
+ \\auth = "k"
+ \\
+ \\[auth.k]
+ \\key = "tok"
+ ;
+ const doc = try parseDoc(a, src);
+ defer doc.deinit();
+ try testing.expectError(error.InvalidProvider, resolve(a, &env, doc.root));
+}
+
test "resolve: prompt_cache defaults true and is parsed when set" {
const a = testing.allocator;
var env = emptyEnv(a);
diff --git a/src/debug_log.zig b/src/debug_log.zig
new file mode 100644
index 0000000..0adbe63
--- /dev/null
+++ b/src/debug_log.zig
@@ -0,0 +1,220 @@
+//! Debug-build log redirection.
+//!
+//! Debug builds emit a *lot* of `std.log.debug` traffic — most usefully the
+//! raw JSON of every provider API request and response. Writing that to stderr
+//! makes the interactive TUI unusable (every line scribbles over the frame).
+//!
+//! This module installs a custom `std.Options.logFn` (wired up via
+//! `std_options` in `main.zig`). In **Debug** builds it routes the entire log
+//! stream to a per-session file under
+//!
+//! ($PANTO_HOME | $XDG_DATA_HOME/panto | ~/.local/share/panto)/debug/<session-id>.log
+//!
+//! and writes *nothing* to the terminal, leaving the TUI pristine. In any
+//! other build mode it delegates to `std.log.defaultLog` (stderr, level-gated)
+//! so release behavior is unchanged.
+//!
+//! The file is opened with `O_TRUNC`, so each session starts fresh; the file
+//! holds exactly one session's logs and never grows across runs. (A *new*
+//! session id per invocation already gives a fresh file; truncation guards the
+//! resume case where the same id is reused.)
+//!
+//! The log function may be called from any thread (the `Io.Threaded` worker
+//! pool as well as the main loop), so all writes are serialized behind a
+//! mutex. Writes go straight to the fd via `std.c.write` — no `Io` handle is
+//! needed (and none is available at a `logFn` call site), matching the raw-fd
+//! approach already used by `tui_terminal.zig`.
+
+const std = @import("std");
+const builtin = @import("builtin");
+const Io = std.Io;
+const Allocator = std.mem.Allocator;
+
+const panto_home = @import("panto_home.zig");
+
+/// True only in Debug builds; gates the file-redirect path at comptime.
+const enabled = builtin.mode == .Debug;
+
+/// A tiny atomic spinlock serializing writes. `logFn` can be called from any
+/// thread but has no `Io` handle, so `Io.Mutex` (which needs one) is out;
+/// contention is rare and each critical section is a single bounded write.
+var log_lock = std.atomic.Value(bool).init(false);
+
+fn lockLog() void {
+ while (log_lock.cmpxchgWeak(false, true, .acquire, .monotonic) != null) {
+ std.atomic.spinLoopHint();
+ }
+}
+
+fn unlockLog() void {
+ log_lock.store(false, .release);
+}
+/// The open log file descriptor, or -1 before `init` (or if init failed).
+var log_fd: std.c.fd_t = -1;
+/// Monotonic per-line counter. Wall-clock time needs an `Io` handle (not
+/// available at a `logFn` call site), so lines are stamped with a sequence
+/// number instead — enough to order events within the session.
+var seq: u64 = 0;
+/// Buffer backing the per-call `Writer`. Guarded by `log_mutex`.
+var writer_buf: [16 * 1024]u8 = undefined;
+
+/// Open the per-session debug log file and arm `logFn` to write to it. No-op
+/// in non-Debug builds. Best-effort: any failure leaves `log_fd == -1`, and
+/// `logFn` silently drops messages (debug logging must never break startup).
+///
+/// Safe to call once, after the session id is known. `environ_map` is used to
+/// resolve `$PANTO_HOME`; `io` is used only to create the `debug/` directory.
+pub fn init(
+ allocator: Allocator,
+ io: Io,
+ environ_map: *const std.process.Environ.Map,
+ session_id: []const u8,
+) void {
+ if (!enabled) return;
+
+ var layout = panto_home.resolve(allocator, environ_map) catch return;
+ defer layout.deinit();
+
+ const debug_dir = std.fs.path.join(allocator, &.{ layout.home, "debug" }) catch return;
+ defer allocator.free(debug_dir);
+
+ Io.Dir.cwd().createDirPath(io, debug_dir) catch |err| switch (err) {
+ error.PathAlreadyExists => {},
+ else => return,
+ };
+
+ const file_name = std.fmt.allocPrint(allocator, "{s}.log", .{session_id}) catch return;
+ defer allocator.free(file_name);
+ const path = std.fs.path.joinZ(allocator, &.{ debug_dir, file_name }) catch return;
+ defer allocator.free(path);
+
+ const flags = std.c.O{ .ACCMODE = .WRONLY, .CREAT = true, .TRUNC = true };
+ const fd = std.c.open(path.ptr, flags, @as(std.c.mode_t, 0o600));
+ if (fd < 0) return;
+
+ lockLog();
+ defer unlockLog();
+ log_fd = fd;
+ seq = 0;
+}
+
+/// Custom `std.Options.logFn`. Debug builds capture everything to the session
+/// file (terminal stays clean); other builds use the stderr default.
+pub fn logFn(
+ comptime level: std.log.Level,
+ comptime scope: @EnumLiteral(),
+ comptime format: []const u8,
+ args: anytype,
+) void {
+ if (!enabled) {
+ std.log.defaultLog(level, scope, format, args);
+ return;
+ }
+
+ lockLog();
+ defer unlockLog();
+
+ const fd = log_fd;
+ // Before the file is open (early bootstrap), drop the message rather than
+ // letting it reach the terminal — keeping the TUI pristine is the point.
+ if (fd < 0) return;
+
+ var fw: FdWriter = .{
+ .fd = fd,
+ .interface = .{ .vtable = &fd_vtable, .buffer = &writer_buf },
+ };
+ const w = &fw.interface;
+
+ seq += 1;
+ w.print("#{d:>6} {s}", .{ seq, level.asText() }) catch {};
+ if (scope != .default) w.print("({t})", .{scope}) catch {};
+ w.writeAll(": ") catch {};
+ w.print(format, args) catch {};
+ w.writeAll("\n") catch {};
+ w.flush() catch {};
+}
+
+/// A minimal `std.Io.Writer` whose sink is a raw file descriptor. Writes are
+/// best-effort (logging must never fault): short writes are looped, `EINTR` is
+/// retried, and any other error silently drops the remaining bytes.
+const FdWriter = struct {
+ fd: std.c.fd_t,
+ interface: std.Io.Writer,
+};
+
+const fd_vtable: std.Io.Writer.VTable = .{ .drain = drainFd };
+
+fn drainFd(w: *std.Io.Writer, data: []const []const u8, splat: usize) std.Io.Writer.Error!usize {
+ const self: *FdWriter = @alignCast(@fieldParentPtr("interface", w));
+ const fd = self.fd;
+
+ // 1. Flush any buffered bytes first (they were logically written already).
+ if (w.end != 0) {
+ writeAllFd(fd, w.buffer[0..w.end]);
+ w.end = 0;
+ }
+
+ // 2. Write each data slice in order; the last is repeated `splat` times.
+ if (data.len == 0) return 0;
+ var consumed: usize = 0;
+ for (data[0 .. data.len - 1]) |slice| {
+ writeAllFd(fd, slice);
+ consumed += slice.len;
+ }
+ const last = data[data.len - 1];
+ var i: usize = 0;
+ while (i < splat) : (i += 1) {
+ writeAllFd(fd, last);
+ consumed += last.len;
+ }
+ return consumed;
+}
+
+fn writeAllFd(fd: std.c.fd_t, bytes: []const u8) void {
+ var off: usize = 0;
+ while (off < bytes.len) {
+ const n = std.c.write(fd, bytes.ptr + off, bytes.len - off);
+ if (n < 0) {
+ if (std.posix.errno(n) == .INTR) continue;
+ return; // best-effort: drop the rest
+ }
+ if (n == 0) return;
+ off += @intCast(n);
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+const testing = std.testing;
+
+test "FdWriter: buffered + splat writes reach the fd intact" {
+ const path = "/tmp/panto_debug_log_drain_test.txt";
+ const wr_flags = std.c.O{ .ACCMODE = .WRONLY, .CREAT = true, .TRUNC = true };
+ const fd = std.c.open(path, wr_flags, @as(std.c.mode_t, 0o600));
+ try testing.expect(fd >= 0);
+
+ {
+ // A deliberately tiny buffer forces several drains mid-message,
+ // exercising the buffer-flush path and the splat repetition.
+ var buf: [8]u8 = undefined;
+ var fw: FdWriter = .{
+ .fd = fd,
+ .interface = .{ .vtable = &fd_vtable, .buffer = &buf },
+ };
+ const w = &fw.interface;
+ try w.print("hello {d} world {s}", .{ 42, "xyz" });
+ try w.splatBytesAll("ab", 3); // splat: "ababab"
+ try w.flush();
+ }
+ _ = std.c.close(fd);
+
+ const rfd = std.c.open(path, std.c.O{ .ACCMODE = .RDONLY }, @as(std.c.mode_t, 0));
+ try testing.expect(rfd >= 0);
+ defer _ = std.c.close(rfd);
+ var rbuf: [128]u8 = undefined;
+ const n = std.c.read(rfd, &rbuf, rbuf.len);
+ try testing.expect(n > 0);
+ try testing.expectEqualStrings("hello 42 world xyzababab", rbuf[0..@intCast(n)]);
+}
diff --git a/src/lua_runtime.zig b/src/lua_runtime.zig
index 2574fae..35247bb 100644
--- a/src/lua_runtime.zig
+++ b/src/lua_runtime.zig
@@ -87,6 +87,15 @@ pub const LuaRuntime = struct {
/// calls to drive libuv to completion. `0` until
/// `installScheduler` runs.
uv_run_ref: c_int = 0,
+ /// Registry ref to `require("luv").update_time`. libuv caches the
+ /// loop's notion of "now" and only refreshes it while `uv.run`
+ /// executes; the loop is idle between batches (and during a slow
+ /// model's streaming), so its clock goes stale. The scheduler calls
+ /// this at the start of each batch so handlers that arm timers
+ /// (e.g. the shell tool's timeout) compute deadlines against the
+ /// real current time, not a frozen one. `0` until
+ /// `installScheduler` runs.
+ uv_update_time_ref: c_int = 0,
/// Pointer to the in-flight batch, valid only for the duration of
/// one `invoke_batch` call. The `panto._record_result` C function
@@ -220,6 +229,7 @@ pub const LuaRuntime = struct {
try installRecordResult(self);
try installWrapperClosure(self);
try cacheUvRun(self);
+ try cacheUvUpdateTime(self);
}
/// Tear down the runtime: free every owned string, unref every
@@ -244,6 +254,9 @@ pub const LuaRuntime = struct {
if (self.uv_run_ref != 0) {
c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, self.uv_run_ref);
}
+ if (self.uv_update_time_ref != 0) {
+ c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, self.uv_update_time_ref);
+ }
// Free the UI event bridge (Lua handler/component refs + caches)
// BEFORE closing the state, since it unrefs registry entries.
@@ -724,6 +737,12 @@ fn runBatch(
return;
}
+ // Refresh libuv's cached loop clock before any handler arms a timer.
+ // The loop sits idle between batches (and while a slow model streams),
+ // so its notion of "now" is stale; a timeout armed against it would
+ // otherwise fire the instant `uv.run` refreshes time below.
+ refreshLoopClock(self);
+
var slots = try allocator.alloc(Slot, calls.len);
defer allocator.free(slots);
for (slots) |*s| s.* = .{};
@@ -1156,6 +1175,45 @@ fn cacheUvRun(self: *LuaRuntime) !void {
self.uv_run_ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX);
}
+/// Cache `require("luv").update_time` so the scheduler can refresh the
+/// loop clock at the start of each batch. See `uv_update_time_ref`.
+fn cacheUvUpdateTime(self: *LuaRuntime) !void {
+ const L = self.L;
+ const snippet =
+ \\local uv = require("luv")
+ \\return uv.update_time
+ ;
+ if (c.luaL_loadstring(L, snippet) != 0) {
+ logTopAsError(L, "panto-lua: failed to compile luv.update_time lookup");
+ c.lua_settop(L, c.lua_gettop(L) - 1);
+ return RuntimeError.LuaInitFailed;
+ }
+ if (c.lua_pcallk(L, 0, 1, 0, 0, null) != 0) {
+ logTopAsError(L, "panto-lua: require('luv') failed (was the bootstrap successful?)");
+ c.lua_settop(L, c.lua_gettop(L) - 1);
+ return RuntimeError.LuaInitFailed;
+ }
+ if (c.lua_type(L, -1) != lua_bridge.T_FUNCTION) {
+ c.lua_settop(L, c.lua_gettop(L) - 1);
+ return RuntimeError.LuaInitFailed;
+ }
+ self.uv_update_time_ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX);
+}
+
+/// Refresh libuv's cached loop clock (`uv.update_time()`). Best-effort:
+/// a failure here only means timeout deadlines may be computed against a
+/// slightly stale clock, so we log and continue rather than failing the
+/// batch.
+fn refreshLoopClock(self: *LuaRuntime) void {
+ if (self.uv_update_time_ref == 0) return;
+ const L = self.L;
+ _ = c.lua_rawgeti(L, lua_bridge.LUA_REGISTRYINDEX, @intCast(self.uv_update_time_ref));
+ if (c.lua_pcallk(L, 0, 0, 0, 0, null) != 0) {
+ logTopAsError(L, "panto-lua: uv.update_time failed");
+ c.lua_settop(L, c.lua_gettop(L) - 1);
+ }
+}
+
// ---------------------------------------------------------------------------
// Internals
// ---------------------------------------------------------------------------
@@ -1778,6 +1836,88 @@ test "scheduler: two real std.shell calls in one batch do not deadlock" {
try testing.expect(std.mem.indexOf(u8, okText(results[1]), "second-done") != null);
}
+test "scheduler: shell timeout is measured from dispatch, not a stale loop clock" {
+ // Regression: libuv caches the loop's "now" and only refreshes it while
+ // `uv.run` runs. The loop is idle between batches, so a tool that arms a
+ // timeout timer against the stale clock would have it fire the instant
+ // `uv.run` refreshes time. We force staleness with `uv.sleep` (a blocking
+ // sleep that does NOT run the loop), then dispatch a 1s-timeout shell call
+ // and assert it does not spuriously time out.
+ var rt = try LuaRuntime.create(testing.allocator);
+ defer rt.deinit();
+ const luarocks_rt = bootstrapRealRuntime(rt) catch |e| return e;
+ defer luarocks_rt.deinit();
+
+ const tools_dir = try findAgentToolsDir();
+ defer testing.allocator.free(tools_dir);
+ const shell_path = try std.fs.path.join(testing.allocator, &.{ tools_dir, "shell.lua" });
+ defer testing.allocator.free(shell_path);
+ try rt.loadTool(shell_path, tools_dir);
+
+ var src = rt.toolSource();
+
+ // Prime the loop clock with one run.
+ {
+ const warm = [_]panto.ToolCall{.{ .tool_name = "std.shell", .input = "{\"command\":\"true\"}" }};
+ var wres: [1]panto.ToolCallResult = .{.{ .err = error.SourceDroppedCall }};
+ try src.vtable.invoke_batch(src.ctx, &warm, &wres, testing.allocator);
+ freeResults(&wres);
+ }
+
+ // Advance real time ~1.2s WITHOUT running the loop, so its cached clock
+ // goes stale relative to the wall clock by more than the timeout below.
+ {
+ const snippet = "require('luv').sleep(1200)";
+ if (c.luaL_loadstring(rt.L, snippet) != 0) return error.SleepSnippetFailed;
+ if (c.lua_pcallk(rt.L, 0, 0, 0, 0, null) != 0) return error.SleepSnippetFailed;
+ }
+
+ const calls = [_]panto.ToolCall{
+ .{ .tool_name = "std.shell", .input = "{\"command\":\"echo prompt-done\",\"timeout\":1}" },
+ };
+ var results: [1]panto.ToolCallResult = .{.{ .err = error.SourceDroppedCall }};
+ try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
+ defer freeResults(&results);
+
+ try testing.expect(std.mem.indexOf(u8, okText(results[0]), "prompt-done") != null);
+ try testing.expect(std.mem.indexOf(u8, okText(results[0]), "timed out") == null);
+}
+
+test "scheduler: three real std.shell calls with explicit timeout all succeed" {
+ var rt = try LuaRuntime.create(testing.allocator);
+ defer rt.deinit();
+ const luarocks_rt = bootstrapRealRuntime(rt) catch |e| return e;
+ defer luarocks_rt.deinit();
+
+ const tools_dir = try findAgentToolsDir();
+ defer testing.allocator.free(tools_dir);
+ const shell_path = try std.fs.path.join(testing.allocator, &.{ tools_dir, "shell.lua" });
+ defer testing.allocator.free(shell_path);
+ try rt.loadTool(shell_path, tools_dir);
+
+ var src = rt.toolSource();
+ const calls = [_]panto.ToolCall{
+ .{ .tool_name = "std.shell", .input = "{\"command\":\"echo alpha-done\",\"timeout\":30}" },
+ .{ .tool_name = "std.shell", .input = "{\"command\":\"echo bravo-done\",\"timeout\":30}" },
+ .{ .tool_name = "std.shell", .input = "{\"command\":\"echo charlie-done\",\"timeout\":30}" },
+ };
+ var results: [3]panto.ToolCallResult = .{
+ .{ .err = error.SourceDroppedCall },
+ .{ .err = error.SourceDroppedCall },
+ .{ .err = error.SourceDroppedCall },
+ };
+ try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator);
+ defer freeResults(&results);
+
+ try testing.expect(std.mem.indexOf(u8, okText(results[0]), "alpha-done") != null);
+ try testing.expect(std.mem.indexOf(u8, okText(results[1]), "bravo-done") != null);
+ try testing.expect(std.mem.indexOf(u8, okText(results[2]), "charlie-done") != null);
+ // None should have hit the timeout path.
+ for (results) |r| {
+ try testing.expect(std.mem.indexOf(u8, okText(r), "timed out") == null);
+ }
+}
+
// Reproduction: two REAL `std.read` calls in one batch. read.lua uses
// only synchronous `uv.fs_*` calls and never yields, so both should
// complete during dispatch without entering the drive loop.
diff --git a/src/luarocks_runtime.zig b/src/luarocks_runtime.zig
index f76a60e..7f2e161 100644
--- a/src/luarocks_runtime.zig
+++ b/src/luarocks_runtime.zig
@@ -29,6 +29,7 @@
//! `<panto-binary> lua` (with the absolute path of the running `panto`
//! binary) into the luarocks config as `variables.LUA`.
+const builtin = @import("builtin");
const std = @import("std");
const Allocator = std.mem.Allocator;
const Io = std.Io;
@@ -70,6 +71,7 @@ pub const BootstrapError = error{
LuarocksInjectionFailed,
LuarocksInstallFailed,
LuarocksSearcherInstallFailed,
+ PantoModuleSigningFailed,
PathConfigFailed,
PantoExecutablePathUnknown,
} || Allocator.Error;
@@ -229,6 +231,98 @@ fn stagePantoModule(
var dir = try Io.Dir.cwd().openDir(io, layout.lib_lua_dir, .{});
defer dir.close(io);
try writeIfDifferent(allocator, io, dir, "panto.so", embedded_panto_so.bytes);
+ try ensurePantoModuleSignature(allocator, layout);
+}
+
+/// Ad-hoc re-sign the staged `panto.so` so macOS will `dlopen` it.
+///
+/// We deliberately spawn `codesign` via `posix_spawn` rather than
+/// `std.process.run`. `std.process.run` (via `Io.Threaded`) implements
+/// spawning as a raw `fork()` + `execve()`. On macOS a `fork()` from a
+/// process that has *any* live secondary thread runs the registered
+/// `pthread_atfork` child handlers in the forked child; system-framework
+/// handlers can dereference per-thread state that no longer exists in the
+/// child and fault (`EXC_BAD_ACCESS`). In a normal `panto` run bootstrap
+/// is still single-threaded here so the fork is safe, but under the
+/// multithreaded Zig test runner (whose pool already has worker threads,
+/// and which installs a debug `SIGSEGV` handler that turns that fault into
+/// an `abort()`) the spawned child dies with `SIGABRT` before `codesign`
+/// ever execs. `posix_spawn` is a single Darwin primitive that does *not*
+/// run userspace `pthread_atfork` handlers — Apple's recommended way to
+/// spawn from a multithreaded process — so it sidesteps the hazard
+/// entirely while behaving identically to the caller. The child's stdio is
+/// redirected to /dev/null (see below); failures surface via the exit
+/// status, which we translate into `error.PantoModuleSigningFailed`.
+fn ensurePantoModuleSignature(
+ allocator: Allocator,
+ layout: panto_home.Layout,
+) !void {
+ if (builtin.os.tag != .macos) return;
+
+ const module_path = try std.fs.path.joinZ(allocator, &.{ layout.lib_lua_dir, "panto.so" });
+ defer allocator.free(module_path);
+
+ const argv = [_:null]?[*:0]const u8{
+ "/usr/bin/codesign",
+ "--force",
+ "--sign",
+ "-",
+ module_path.ptr,
+ null,
+ };
+
+ // Redirect the child's stdio to /dev/null: `codesign` is silent on
+ // success and writes progress/errors to stderr, which we don't want
+ // leaking into panto's own output (and, under the test runner, racing
+ // with the build-server IPC). Failures are surfaced via the exit
+ // status and our own `std.log.err` below.
+ var actions: std.c.posix_spawn_file_actions_t = undefined;
+ if (std.c.posix_spawn_file_actions_init(&actions) != 0) {
+ return error.PantoModuleSigningFailed;
+ }
+ defer _ = std.c.posix_spawn_file_actions_destroy(&actions);
+ const rdonly: c_int = @bitCast(std.c.O{ .ACCMODE = .RDONLY });
+ const wronly: c_int = @bitCast(std.c.O{ .ACCMODE = .WRONLY });
+ _ = std.c.posix_spawn_file_actions_addopen(&actions, std.c.STDIN_FILENO, "/dev/null", rdonly, 0);
+ _ = std.c.posix_spawn_file_actions_addopen(&actions, std.c.STDOUT_FILENO, "/dev/null", wronly, 0);
+ _ = std.c.posix_spawn_file_actions_addopen(&actions, std.c.STDERR_FILENO, "/dev/null", wronly, 0);
+
+ var pid: std.c.pid_t = undefined;
+ const spawn_rc = std.c.posix_spawn(
+ &pid,
+ "/usr/bin/codesign",
+ &actions,
+ null,
+ &argv,
+ @ptrCast(std.c.environ),
+ );
+ if (spawn_rc != 0) {
+ std.log.err(
+ "failed to spawn codesign for staged panto module at {s}: posix_spawn errno={d}",
+ .{ module_path, spawn_rc },
+ );
+ return error.PantoModuleSigningFailed;
+ }
+
+ const status = blk: while (true) {
+ var status: c_int = undefined;
+ const rc = std.c.waitpid(pid, &status, 0);
+ if (rc == -1) {
+ const err = std.posix.errno(rc);
+ if (err == .INTR) continue;
+ std.log.err("failed to wait for codesign: errno={t}", .{err});
+ return error.PantoModuleSigningFailed;
+ }
+ break :blk @as(u32, @bitCast(status));
+ };
+
+ if (std.c.W.IFEXITED(status) and std.c.W.EXITSTATUS(status) == 0) return;
+
+ std.log.err(
+ "codesign failed for staged panto module at {s}: status=0x{x}",
+ .{ module_path, status },
+ );
+ return error.PantoModuleSigningFailed;
}
// ---------------------------------------------------------------------------
@@ -338,14 +432,13 @@ const default_base_config =
\\# exchange_expires_path = "expires_at"
\\# exchange_base_url_path = "endpoints.api"
\\
- \\# OpenAI Codex via a ChatGPT subscription (device login). Codex speaks the
- \\# OpenAI Responses API, not Chat Completions, so style = "openai_responses".
- \\# NOTE: this path is implemented but not yet verified against a live
- \\# ChatGPT plan — see docs/oauth-provider-auth.md.
+ \\# OpenAI Codex via a ChatGPT subscription (device login). Codex uses the
+ \\# Responses protocol with a stricter ChatGPT-subscription dialect.
\\#
\\# [providers.codex]
\\# style = "openai_responses"
- \\# base_url = "https://chatgpt.com/backend-api/codex"
+ \\# dialect = "codex"
+ \\# base_url = "https://chatgpt.com/backend-api"
\\# auth = "openai_codex"
\\#
\\# [providers.codex.extra_headers]
diff --git a/src/main.zig b/src/main.zig
index 58502ad..8ac5e8b 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -16,6 +16,12 @@ const glob = @import("glob.zig");
const system_prompt = @import("system_prompt.zig");
const command = @import("command.zig");
const command_compaction = @import("compaction.zig");
+const debug_log = @import("debug_log.zig");
+
+/// Route the process-wide log stream. In Debug builds this redirects every
+/// `std.log` call to a per-session file (keeping the TUI clean); other builds
+/// keep the stderr default. See `debug_log.zig`.
+pub const std_options: std.Options = .{ .logFn = debug_log.logFn };
// TUI foundation layer (Phase 1, sub-phase 1). Not yet wired into the REPL;
// referenced here so `zig build test` type-checks and exercises them.
@@ -60,6 +66,7 @@ test {
_ = system_prompt;
_ = command;
_ = command_compaction;
+ _ = debug_log;
_ = tui_terminal;
_ = tui_key;
_ = tui_input;
@@ -255,7 +262,9 @@ pub fn main(init: std.process.Init) !void {
// per-cwd `session_dir` (the CLI owns the cwd→dir grouping; the store
// is cwd-agnostic). Must outlive the agent, which appends through a
// `Session` handle minted/resolved below.
- const session_metadata = try std.fmt.allocPrint(alloc, "{{\"cwd\":{s}}}", .{try std.json.Stringify.valueAlloc(alloc, cwd, .{})});
+ const cwd_json = try std.json.Stringify.valueAlloc(alloc, cwd, .{});
+ defer alloc.free(cwd_json);
+ const session_metadata = try std.fmt.allocPrint(alloc, "{{\"cwd\":{s}}}", .{cwd_json});
defer alloc.free(session_metadata);
var session_store_impl = try panto.FileSystemJSONLStore.initWithMetadata(alloc, io, session_dir, session_metadata);
defer session_store_impl.deinit();
@@ -277,6 +286,12 @@ pub fn main(init: std.process.Init) !void {
// `session.info` is adopted by the agent below (`Agent.init` can't fail)
// and freed in the agent's `deinit`; no separate cleanup here.
+ // Arm the per-session debug log now that we know the session id. In Debug
+ // builds this opens `$PANTO_HOME/debug/<id>.log` and redirects all
+ // `std.log` output there; no-op in release. Best-effort — failures leave
+ // logging disabled rather than aborting startup.
+ debug_log.init(alloc, io, init.environ_map, session.info.id);
+
const is_resume = cli_flags.resume_kind != .none and session.info.message_count > 0;
// On resume, reconstruct the conversation from the store. (The
diff --git a/src/models_toml.zig b/src/models_toml.zig
index 955640d..622bdf6 100644
--- a/src/models_toml.zig
+++ b/src/models_toml.zig
@@ -21,7 +21,8 @@
//!
//! # anthropic_messages-only knobs:
//! api_version = <string> # Anthropic-Version header; default "2023-06-01"
-//! thinking = <string> # disabled | enabled | adaptive (default: disabled)
+//! thinking = <string> # disabled | enabled | adaptive (default: disabled,
+//! # or adaptive when `effort` is present)
//! effort = <string> # low | medium | high | xhigh | max (default: medium)
//! # only used when thinking = "adaptive"
//! thinking_budget_tokens = <int> # max reasoning tokens for thinking = "enabled"
@@ -268,13 +269,15 @@ fn ingestModel(
break :blk null;
};
+ const has_effort = v.get("effort") != null;
+
const thinking: Thinking = blk: {
if (v.get("thinking")) |t| {
if (t.asString()) |s| {
break :blk std.meta.stringToEnum(Thinking, s) orelse .disabled;
}
}
- break :blk .disabled;
+ break :blk if (has_effort) .adaptive else .disabled;
};
const effort: Effort = blk: {
@@ -527,6 +530,39 @@ test "parseInto: anthropic adaptive thinking with effort" {
try testing.expectEqual(false, def.thinking_interleaved); // default
}
+test "parseInto: anthropic effort implies adaptive thinking" {
+ var models = emptyModels(testing.allocator);
+ defer models.deinit();
+
+ const src =
+ \\[anthropic.opus]
+ \\model = "claude-opus-4-8"
+ \\effort = "high"
+ ;
+ try parseInto(&models, src);
+
+ const def = models.defs.get("anthropic", "opus").?;
+ try testing.expectEqual(Thinking.adaptive, def.thinking);
+ try testing.expectEqual(Effort.high, def.effort);
+}
+
+test "parseInto: explicit thinking overrides effort-implied adaptive thinking" {
+ var models = emptyModels(testing.allocator);
+ defer models.deinit();
+
+ const src =
+ \\[anthropic.haiku]
+ \\model = "claude-haiku-4-5-20251001"
+ \\thinking = "disabled"
+ \\effort = "high"
+ ;
+ try parseInto(&models, src);
+
+ const def = models.defs.get("anthropic", "haiku").?;
+ try testing.expectEqual(Thinking.disabled, def.thinking);
+ try testing.expectEqual(Effort.high, def.effort);
+}
+
test "parseInto: anthropic thinking disabled" {
var models = emptyModels(testing.allocator);
defer models.deinit();
diff --git a/src/tui_app.zig b/src/tui_app.zig
index 1182777..69eb8fa 100644
--- a/src/tui_app.zig
+++ b/src/tui_app.zig
@@ -1402,16 +1402,13 @@ pub fn runLoop(app: *App, term: *Terminal, opts: RunOptions) !void {
// that here. Otherwise seed the plain model label.
if (app.selectors == null) try app.footer.setModel(opts.model_label);
- // Session-start welcome banner as the first transcript entry. cwd is read
- // from the process; the model label comes from the run options. (Version
- // is not threaded through the run options yet; the banner omits it.)
+ // Session-start welcome banner as the first transcript entry.
{
const welcome = try app.spawnWelcome(.{
.version = opts.version,
.cwd = opts.cwd,
.model = opts.model_label,
});
- try welcome.setModel(opts.model_label);
if (opts.cwd.len != 0) try welcome.setCwd(opts.cwd);
if (opts.version.len != 0) try welcome.setVersion(opts.version);
}
@@ -2371,7 +2368,7 @@ test "spawnWelcome shows a session-start banner entry" {
defer h.teardown(alloc);
const w = try h.app.spawnWelcome(.{});
- try w.setModel("m");
+ _ = w;
try testing.expectEqual(@as(usize, 1), h.app.transcript.items.len);
try testing.expect(h.app.transcript.items[0].kind == .welcome);
}
diff --git a/src/tui_components.zig b/src/tui_components.zig
index e64b5bd..14c61d9 100644
--- a/src/tui_components.zig
+++ b/src/tui_components.zig
@@ -983,9 +983,10 @@ pub const InputBox = struct {
// Locate the cursor's (row, byte-col-in-row).
const focused = self.focusable.focused;
- // Walk lines, tracking byte offset so we know which row holds cursor.
- // `cursor_row` records which produced row carries the cursor block, so
- // the scroll-window below can keep it visible.
+ // Walk logical lines, but render HARD-WRAPPED visual rows. Wrapping is
+ // purely visual: no '\n' bytes are inserted into the editor buffer.
+ // `cursor_row` records which produced visual row carries the cursor
+ // block, so the scroll-window below can keep it visible.
var line_byte_start: usize = 0;
var produced_any = false;
var cursor_row: usize = 0;
@@ -994,22 +995,19 @@ pub const InputBox = struct {
const line_start = line_byte_start;
const line_end = line_start + line.len;
const cursor_in_line = focused and self.cursor >= line_start and self.cursor <= line_end and
- // The cursor belongs to the FIRST line whose range contains it
- // (at a '\n' boundary it stays on the line before the break,
- // i.e. == line_end). Disambiguate the boundary: if cursor ==
- // line_end and there are more lines, it belongs to the NEXT
- // line's start unless this is the last line.
+ // At an explicit '\n' boundary, the cursor belongs to the next
+ // logical line's start unless this is the final line.
(self.cursor < line_end or it.peek() == null);
- if (cursor_in_line) cursor_row = rows.items.len;
- const row = try self.renderRow(line, if (cursor_in_line) self.cursor - line_start else null, cursor_style, width, focused);
- try rows.append(a, row);
- produced_any = true;
+ const first_row = rows.items.len;
+ try self.appendWrappedRows(line, line_start, cursor_in_line, cursor_style, width, focused, &rows, &cursor_row);
+ produced_any = produced_any or rows.items.len > first_row;
line_byte_start = line_end + 1; // skip the '\n'
}
if (!produced_any) {
// Empty buffer: a single (possibly cursor-bearing) row.
const row = try self.renderRow("", if (focused) @as(?usize, 0) else null, cursor_style, width, focused);
+ if (focused) cursor_row = 0;
try rows.append(a, row);
}
@@ -1036,6 +1034,80 @@ pub const InputBox = struct {
return cacheLines(&self.cache);
}
+ /// Append the hard-wrapped visual rows for one logical line. The slices are
+ /// rendered immediately into owned row bytes; the editor buffer itself is
+ /// unchanged. Cursor placement follows terminal wrapping semantics: a
+ /// cursor exactly at a wrap boundary appears at column 0 of the next visual
+ /// row, not at the end of the previous one.
+ fn appendWrappedRows(
+ self: *InputBox,
+ line: []const u8,
+ line_start: usize,
+ cursor_in_line: bool,
+ cursor_style: Style,
+ width: usize,
+ focused: bool,
+ rows: *std.ArrayList([]const u8),
+ cursor_row: *usize,
+ ) !void {
+ const a = self.alloc;
+
+ if (width == 0) {
+ const row = try self.renderRow("", if (cursor_in_line) @as(?usize, 0) else null, cursor_style, width, focused);
+ if (cursor_in_line) cursor_row.* = rows.items.len;
+ try rows.append(a, row);
+ return;
+ }
+
+ if (line.len == 0) {
+ const row = try self.renderRow("", if (cursor_in_line) @as(?usize, 0) else null, cursor_style, width, focused);
+ if (cursor_in_line) cursor_row.* = rows.items.len;
+ try rows.append(a, row);
+ return;
+ }
+
+ var i: usize = 0;
+ while (i < line.len) {
+ const chunk_start = i;
+ var cols: usize = 0;
+ while (i < line.len) {
+ const seq_len = std.unicode.utf8ByteSequenceLength(line[i]) catch 1;
+ const adv = @min(seq_len, line.len - i);
+ const cp = std.unicode.utf8Decode(line[i .. i + adv]) catch '?';
+ const w = codepointWidth(cp);
+ if (cols > 0 and cols + w > width) break;
+ i += adv;
+ cols += w;
+ if (cols >= width) break;
+ }
+ if (i == chunk_start) i = self.nextBoundary(line_start + i) - line_start; // defensive progress
+
+ const chunk = line[chunk_start..i];
+ const abs_start = line_start + chunk_start;
+ const abs_end = line_start + i;
+ var cursor_col: ?usize = null;
+ if (cursor_in_line) {
+ if (self.cursor >= abs_start and self.cursor < abs_end) {
+ cursor_col = self.cursor - abs_start;
+ } else if (i == line.len and self.cursor == abs_end) {
+ cursor_col = chunk.len;
+ }
+ }
+ if (cursor_col != null) cursor_row.* = rows.items.len;
+ const row = try self.renderRow(chunk, cursor_col, cursor_style, width, focused);
+ try rows.append(a, row);
+ }
+
+ // If the cursor is at the end of a line that exactly fills the last
+ // visual row, show the cursor on the following wrapped row instead of
+ // replacing the last visible character with the block.
+ if (cursor_in_line and self.cursor == line_start + line.len and displayWidth(line) % width == 0) {
+ const row = try self.renderRow("", @as(?usize, 0), cursor_style, width, focused);
+ cursor_row.* = rows.items.len;
+ try rows.append(a, row);
+ }
+ }
+
/// Build a dim, full-width horizontal rule (box-drawing `─`). Caller owns
/// the returned slice.
fn horizontalRule(self: *InputBox, width: usize) ![]u8 {
@@ -1079,6 +1151,7 @@ pub const InputBox = struct {
/// cursor (or a space at end-of-line) and emits CURSOR_MARKER there.
fn renderRow(self: *InputBox, line: []const u8, cursor_col: ?usize, cursor_style: Style, width: usize, focused: bool) ![]u8 {
const a = self.alloc;
+ if (width == 0) return a.dupe(u8, "");
// The cursor block consumes one visible column, so usable text width
// is width-1 when the cursor sits at/after the truncated end and we
// must show the block. To keep it simple and always-safe: truncate the
@@ -1632,19 +1705,18 @@ pub const Selector = struct {
};
// ===========================================================================
-// Welcome — session-start banner (plan §6: "version, cwd, model info")
+// Welcome — session-start banner
// ===========================================================================
/// A static banner shown as the first transcript entry at session start.
-/// Structured data in (version / cwd / model label via setters), lines out.
-/// Re-rendered only when one of the fields changes (markDirty), which in
-/// practice is once during bring-up.
+/// Structured data in (version / cwd via setters), lines out. Re-rendered
+/// only when one of the visible fields changes (markDirty), which in practice
+/// is once during bring-up.
pub const Welcome = struct {
alloc: std.mem.Allocator,
cache: RenderCache,
version: std.ArrayList(u8) = .empty,
cwd: std.ArrayList(u8) = .empty,
- model: std.ArrayList(u8) = .empty,
pub fn init(alloc: std.mem.Allocator) Welcome {
return .{ .alloc = alloc, .cache = RenderCache.init(alloc) };
@@ -1653,7 +1725,6 @@ pub const Welcome = struct {
pub fn deinit(self: *Welcome) void {
self.version.deinit(self.alloc);
self.cwd.deinit(self.alloc);
- self.model.deinit(self.alloc);
self.cache.deinit();
}
@@ -1673,11 +1744,6 @@ pub const Welcome = struct {
try self.setField(&self.cwd, value);
}
- /// Set the model label shown in the banner.
- pub fn setModel(self: *Welcome, value: []const u8) !void {
- try self.setField(&self.model, value);
- }
-
fn renderImpl(ptr: *anyopaque, width: usize, alloc: std.mem.Allocator) anyerror![]const []const u8 {
_ = alloc;
const self: *Welcome = @ptrCast(@alignCast(ptr));
@@ -1707,19 +1773,13 @@ pub const Welcome = struct {
try lines.append(a, try std.fmt.allocPrint(a, "{s}{s}{s}", .{ accent.open(), vis, accent.close() }));
}
- // Detail lines (dim, with leading space): cwd and model, only when set.
+ // Detail lines (dim, with leading space): cwd only when set.
if (self.cwd.items.len != 0) {
const txt = try std.fmt.allocPrint(a, " cwd: {s}", .{self.cwd.items});
defer a.free(txt);
const vis = truncateToCols(txt, width);
try lines.append(a, try std.fmt.allocPrint(a, "{s}{s}{s}", .{ dim.open(), vis, dim.close() }));
}
- if (self.model.items.len != 0) {
- const txt = try std.fmt.allocPrint(a, " model: {s}", .{self.model.items});
- defer a.free(txt);
- const vis = truncateToCols(txt, width);
- try lines.append(a, try std.fmt.allocPrint(a, "{s}{s}{s}", .{ dim.open(), vis, dim.close() }));
- }
_ = plain_bg;
// Blank margin below.
@@ -2366,7 +2426,35 @@ test "InputBox: cursor block fits within width at end of a full line" {
for ("abcde") |c| try ib.applyKey(charKey(c, &[_]u8{c}));
// width 5, cursor at end: the block must not overflow.
const lines = try ib.comp().render(5, testing.allocator);
- try testing.expect(vw(lines[1]) <= 5);
+ for (lines) |ln| try testing.expect(vw(ln) <= 5);
+}
+
+test "InputBox: long logical line hard-wraps visually without inserting newlines" {
+ var ib = InputBox.init(testing.allocator);
+ defer ib.deinit();
+ ib.setFocused(true);
+ try typeStr(&ib, "abcdef");
+ const lines = try ib.comp().render(5, testing.allocator);
+ // Two content rows ("abcde" and "f"+cursor), plus top/bottom rules.
+ try testing.expectEqual(@as(usize, 4), lines.len);
+ try testing.expectEqualStrings("abcdef", ib.text.items);
+ try testing.expect(std.mem.indexOf(u8, lines[1], "abcde") != null);
+ try testing.expect(std.mem.indexOf(u8, lines[2], "f") != null);
+ try testing.expect(std.mem.indexOf(u8, lines[2], CURSOR_MARKER) != null);
+ for (lines) |ln| try testing.expect(vw(ln) <= 5);
+}
+
+test "InputBox: line cap applies to visual wrapped rows" {
+ var ib = InputBox.init(testing.allocator);
+ defer ib.deinit();
+ ib.line_cap = 2;
+ try typeStr(&ib, "abcde");
+ const lines = try ib.comp().render(2, testing.allocator);
+ // Three wrapped content rows are capped to the tail two, plus rules.
+ try testing.expectEqual(@as(usize, 4), lines.len);
+ try testing.expect(std.mem.indexOf(u8, lines[1], "cd") != null);
+ try testing.expect(std.mem.indexOf(u8, lines[2], "e") != null);
+ for (lines) |ln| try testing.expect(vw(ln) <= 2);
}
fn ctrlKey(letter: u8) Key {
@@ -2873,22 +2961,20 @@ test "components drive the real engine without a TTY" {
// -- Welcome / Thinking / CompactionSummary / ToolUse (P2) ------------------
-test "Welcome: renders title + cwd + model, all within width" {
+test "Welcome: renders title + cwd, all within width" {
var w = Welcome.init(testing.allocator);
defer w.deinit();
try w.setVersion("0.1.0");
try w.setCwd("/tmp/project");
- try w.setModel("anthropic:claude");
const lines = try w.comp().render(40, testing.allocator);
- // 3 content lines + 2 margin = 5.
- try testing.expectEqual(@as(usize, 5), lines.len);
+ // 2 content lines + 2 margin = 4.
+ try testing.expectEqual(@as(usize, 4), lines.len);
for (lines) |l| try testing.expect(vw(l) <= 40);
try testing.expect(std.mem.indexOf(u8, lines[1], "panto v0.1.0") != null);
try testing.expect(std.mem.indexOf(u8, lines[2], "/tmp/project") != null);
- try testing.expect(std.mem.indexOf(u8, lines[3], "anthropic:claude") != null);
}
-test "Welcome: title only when cwd/model unset" {
+test "Welcome: title only when cwd unset" {
var w = Welcome.init(testing.allocator);
defer w.deinit();
const lines = try w.comp().render(20, testing.allocator);
@@ -2902,11 +2988,10 @@ test "Welcome: honors the width contract at a tiny width" {
defer w.deinit();
try w.setVersion("0.1.0");
try w.setCwd("/a/very/long/working/directory/path/that/overflows");
- try w.setModel("anthropic:claude-some-very-long-model-id");
- // Width 6: every banner row (title + cwd + model) must truncate to fit.
- // 3 content lines + 2 margin = 5.
+ // Width 6: every banner row (title + cwd) must truncate to fit.
+ // 2 content lines + 2 margin = 4.
const lines = try w.comp().render(6, testing.allocator);
- try testing.expectEqual(@as(usize, 5), lines.len);
+ try testing.expectEqual(@as(usize, 4), lines.len);
for (lines) |l| try testing.expect(vw(l) <= 6);
}
diff --git a/src/tui_selectors.zig b/src/tui_selectors.zig
index 590d321..e76f82d 100644
--- a/src/tui_selectors.zig
+++ b/src/tui_selectors.zig
@@ -89,7 +89,7 @@ pub const ReasoningOption = struct {
pub fn forStyle(style: APIStyle) []const ReasoningOption {
return switch (style) {
// The Responses style takes the same reasoning-effort options.
- .openai_chat, .openai_responses => &openai,
+ .openai_chat, .openai_responses, .openai_codex_responses => &openai,
.anthropic_messages => &anthropic,
};
}
@@ -100,6 +100,7 @@ pub const ReasoningOption = struct {
.openai_chat => |r| switch (cfg.*) {
.openai_chat => |*c| c.reasoning = r,
.openai_responses => |*c| c.reasoning = r,
+ .openai_codex_responses => |*c| c.reasoning = r,
.anthropic_messages => {},
},
.anthropic_messages => |sel| switch (cfg.*) {
@@ -110,7 +111,7 @@ pub const ReasoningOption = struct {
c.effort = e;
},
},
- .openai_chat, .openai_responses => {},
+ .openai_chat, .openai_responses, .openai_codex_responses => {},
},
}
}
@@ -120,7 +121,8 @@ pub const ReasoningOption = struct {
pub fn matches(self: ReasoningOption, cfg: ProviderConfig) bool {
switch (self.value) {
.openai_chat => |r| return (cfg == .openai_chat and cfg.openai_chat.reasoning == r) or
- (cfg == .openai_responses and cfg.openai_responses.reasoning == r),
+ (cfg == .openai_responses and cfg.openai_responses.reasoning == r) or
+ (cfg == .openai_codex_responses and cfg.openai_codex_responses.reasoning == r),
.anthropic_messages => |sel| {
if (cfg != .anthropic_messages) return false;
const c = cfg.anthropic_messages;
@@ -173,7 +175,7 @@ pub fn adaptiveFallback(cfg: *ProviderConfig) bool {
c.thinking_budget_tokens = budget;
return true;
},
- .openai_chat, .openai_responses => return false,
+ .openai_chat, .openai_responses, .openai_codex_responses => return false,
}
}
diff --git a/tmp/pi_session_cost_analysis.json b/tmp/pi_session_cost_analysis.json
new file mode 100644
index 0000000..cc40030
--- /dev/null
+++ b/tmp/pi_session_cost_analysis.json
@@ -0,0 +1,288 @@
+{
+ "root": "/Users/travis/Code/config/dot-pi/agent/sessions",
+ "sessions_scanned": 495,
+ "sessions_with_usage": 454,
+ "sessions_without_usage": 41,
+ "assistant_messages": 28636,
+ "assistant_messages_with_usage": 28636,
+ "parse_errors": 0,
+ "cache_active_sessions": 216,
+ "pooled_totals": {
+ "input": 648085868,
+ "output": 11386363,
+ "cache_read": 984042157,
+ "cache_write": 65075433
+ },
+ "pooled_metrics": {
+ "prompt_tokens": 1697203458,
+ "billable_tokens": 1708589821,
+ "prompt_to_output_ratio": 149.0558010490268,
+ "input_to_output_ratio": 56.91772412314626,
+ "cache_hit_rate": 0.9379712687878963,
+ "cache_coverage": 0.6181448576803453,
+ "prompt_reuse_rate": 0.5798021164531472,
+ "output_share": 0.006664187542294857,
+ "prompt_share": 0.9933358124577052,
+ "input_share": 0.37931038803724676,
+ "cache_read_share": 0.5759382064116839,
+ "cache_write_share": 0.0380872180087745
+ },
+ "recommended_weights": {
+ "pooled_token_mix": {
+ "input": 0.37931038803724676,
+ "output": 0.006664187542294857,
+ "cache_read": 0.5759382064116839,
+ "cache_write": 0.0380872180087745
+ },
+ "mean_session_mix": {
+ "input": 0.549819370457156,
+ "output": 0.05134274693943982,
+ "cache_read": 0.3229895728469125,
+ "cache_write": 0.07584830975649165
+ },
+ "median_session_mix": {
+ "input": 0.7903769253077484,
+ "output": 0.014961511819985934,
+ "cache_read": 0.0,
+ "cache_write": 0.0
+ }
+ },
+ "per_session_distributions": {
+ "billable_tokens": {
+ "count": 454,
+ "mean": 3763413.702643172,
+ "median": 315231.0,
+ "p10": 20182.100000000002,
+ "p90": 11804297.1,
+ "min": 2244,
+ "max": 101170942
+ },
+ "cache_coverage": {
+ "count": 454,
+ "mean": 0.40473111896757075,
+ "median": 0.0,
+ "p10": 0.0,
+ "p90": 0.9999679722217543,
+ "min": 0.0,
+ "max": 0.9999922748401895
+ },
+ "cache_hit_rate": {
+ "count": 216,
+ "mean": 0.8218859540276632,
+ "median": 0.9487389145525554,
+ "p10": 0.374890408374274,
+ "p90": 1.0,
+ "min": 0.0,
+ "max": 1.0
+ },
+ "cache_read_share": {
+ "count": 454,
+ "mean": 0.3229895728469125,
+ "median": 0.0,
+ "p10": 0.0,
+ "p90": 0.9389935967096287,
+ "min": 0.0,
+ "max": 0.9893885889087582
+ },
+ "cache_write_share": {
+ "count": 454,
+ "mean": 0.07584830975649165,
+ "median": 0.0,
+ "p10": 0.0,
+ "p90": 0.267090715504709,
+ "min": 0.0,
+ "max": 0.999839897534422
+ },
+ "input_share": {
+ "count": 454,
+ "mean": 0.549819370457156,
+ "median": 0.7903769253077484,
+ "p10": 3.1707600618505675e-05,
+ "p90": 0.9974440745899954,
+ "min": 7.709723608187813e-06,
+ "max": 0.9999909002957817
+ },
+ "input_to_output_ratio": {
+ "count": 454,
+ "mean": 829.6580689032315,
+ "median": 8.629654523169638,
+ "p10": 0.00421753985473397,
+ "p90": 409.9444444444443,
+ "min": 0.0010703582132153562,
+ "max": 109892.68181818182
+ },
+ "output_share": {
+ "count": 454,
+ "mean": 0.05134274693943982,
+ "median": 0.014961511819985934,
+ "p10": 0.0018531251145351706,
+ "p90": 0.16915366650275257,
+ "min": 9.099704218250615e-06,
+ "max": 0.5140911569511498
+ },
+ "output_tokens": {
+ "count": 454,
+ "mean": 25080.09471365639,
+ "median": 7470.5,
+ "p10": 365.8000000000001,
+ "p90": 78289.59999999998,
+ "min": 1,
+ "max": 279572
+ },
+ "prompt_reuse_rate": {
+ "count": 454,
+ "mean": 0.32683124761341736,
+ "median": 0.0,
+ "p10": 0.0,
+ "p90": 0.9461329490069706,
+ "min": 0.0,
+ "max": 0.9914821057091003
+ },
+ "prompt_share": {
+ "count": 454,
+ "mean": 0.9486572530605601,
+ "median": 0.985038488180014,
+ "p10": 0.8308463334972475,
+ "p90": 0.9981468748854648,
+ "min": 0.48590884304885024,
+ "max": 0.9999909002957817
+ },
+ "prompt_to_output_ratio": {
+ "count": 454,
+ "mean": 938.8894985366835,
+ "median": 65.83823653274615,
+ "p10": 4.91178438425,
+ "p90": 538.8489050453238,
+ "min": 0.9451803176902779,
+ "max": 109892.68181818182
+ },
+ "prompt_tokens": {
+ "count": 454,
+ "mean": 3738333.607929515,
+ "median": 308096.0,
+ "p10": 19041.5,
+ "p90": 11756748.7,
+ "min": 2149,
+ "max": 100968785
+ }
+ },
+ "cache_active_session_distributions": {
+ "cache_coverage": {
+ "count": 216,
+ "mean": 0.8506848519040607,
+ "median": 0.9997448273674421,
+ "p10": 0.43102818092852674,
+ "p90": 0.999982772379876,
+ "min": 0.0021478132287605567,
+ "max": 0.9999922748401895
+ },
+ "cache_hit_rate": {
+ "count": 216,
+ "mean": 0.8218859540276632,
+ "median": 0.9487389145525554,
+ "p10": 0.374890408374274,
+ "p90": 1.0,
+ "min": 0.0,
+ "max": 1.0
+ },
+ "cache_read_share": {
+ "count": 216,
+ "mean": 0.6788762318171216,
+ "median": 0.7882742441214121,
+ "p10": 0.15606025798620213,
+ "p90": 0.9641169584856126,
+ "min": 0.0,
+ "max": 0.9893885889087582
+ },
+ "cache_write_share": {
+ "count": 216,
+ "mean": 0.15942191032151487,
+ "median": 0.043386068762250346,
+ "p10": 0.0,
+ "p90": 0.5204273102542047,
+ "min": 0.0,
+ "max": 0.999839897534422
+ },
+ "prompt_reuse_rate": {
+ "count": 216,
+ "mean": 0.6869508630393124,
+ "median": 0.7947498412123791,
+ "p10": 0.15876630503361927,
+ "p90": 0.971281214123751,
+ "min": 0.0,
+ "max": 0.9914821057091003
+ }
+ },
+ "top_primary_providers": [
+ [
+ "omni",
+ 263
+ ],
+ [
+ "anthropic",
+ 110
+ ],
+ [
+ "pi-claude-cli",
+ 32
+ ],
+ [
+ "opencode",
+ 27
+ ],
+ [
+ "openai",
+ 10
+ ],
+ [
+ "nanogpt",
+ 9
+ ],
+ [
+ "github-copilot",
+ 3
+ ]
+ ],
+ "top_primary_models": [
+ [
+ "claude-opus-4-7",
+ 100
+ ],
+ [
+ "claude-opus-4-8",
+ 77
+ ],
+ [
+ "claude-opus-4-6",
+ 46
+ ],
+ [
+ "opencode-zen/glm-5.1",
+ 41
+ ],
+ [
+ "glm-5.1",
+ 27
+ ],
+ [
+ "gpt-5.5-2026-04-23",
+ 24
+ ],
+ [
+ "claude-sonnet-4-6",
+ 21
+ ],
+ [
+ "gpt-5.4-2026-03-05",
+ 10
+ ],
+ [
+ "anthropic/claude-opus-4-7",
+ 10
+ ],
+ [
+ "gpt-5.5",
+ 9
+ ]
+ ]
+} \ No newline at end of file
generated by cgit v1.3 (git 2.53.0) at 2026-07-09 07:18:24 +0000