Something went wrong. Try again.
Experimental Zig-rewrite of the letta code listener.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963const std = @import("std");
pub const PermissionMode = enum { strict, standard, acceptEdits, unrestricted,
pub fn allowsBash(self: PermissionMode) bool { return self == .acceptEdits or self == .unrestricted; }};
pub const Config = struct { api_base_url: []const u8, api_key: []const u8, permission_mode: PermissionMode = .standard, enable_client_tools: bool = false, agent_id: ?[]const u8 = null, acting_user_id: ?[]const u8 = null, /// External abort signal (e.g. `abort_message`). When set, in-flight HTTP /// and SSE work returns `error.Cancelled` within one poll interval. cancel_token: ?*const std.atomic.Value(bool) = null,};
pub const cancel_poll_interval_ms: i64 = 20;
pub fn isCancelled(token: ?*const std.atomic.Value(bool)) bool { return if (token) |t| t.load(.acquire) else false;}
pub const InboundTurn = struct { request_id: ?[]const u8, conversation_id: []const u8, kind: Kind, messages: std.json.Value, agent_id: ?[]const u8 = null, acting_user_id: ?[]const u8 = null, teleport_id: ?[]const u8 = null, client_message_id: ?[]const u8 = null,
pub const Kind = enum { create_message, teleport_continue };};
pub const PendingToolCall = struct { tool_call_id: []u8, name: []u8, arguments: []u8, message_id: ?[]u8 = null,
pub fn deinit(self: PendingToolCall, allocator: std.mem.Allocator) void { allocator.free(self.tool_call_id); allocator.free(self.name); allocator.free(self.arguments); if (self.message_id) |message_id| allocator.free(message_id); }};
pub const ApprovalStatus = enum { success, @"error" };
pub const ApprovalResult = struct { tool_call_id: []const u8, status: ApprovalStatus, tool_return: []const u8,};
pub const Result = union(enum) { completed: void, requires_approval: []PendingToolCall, api_error: struct { status: std.http.Status, body: []u8 },};
pub const ChunkCallback = *const fn (ctx: ?*anyopaque, raw_chunk: []const u8) anyerror!void;
pub const max_sse_line_bytes = 1024 * 1024;pub const max_sse_event_bytes = 4 * 1024 * 1024;const max_api_error_body_bytes = 1024 * 1024;const response_read_buffer_bytes = 16 * 1024;pub const sse_inactivity_timeout_ms: i64 = 90_000;pub const http_operation_timeout_ms: i64 = 90_000;
const StreamState = struct { allocator: std.mem.Allocator, stop_reason: ?[]u8 = null, pending_tool_calls: std.ArrayList(PendingToolCall) = .empty,
fn deinit(self: *StreamState) void { if (self.stop_reason) |reason| self.allocator.free(reason); for (self.pending_tool_calls.items) |call| call.deinit(self.allocator); self.pending_tool_calls.deinit(self.allocator); }};
pub fn parseInboundInput(allocator: std.mem.Allocator, frame: []const u8) !std.json.Parsed(InboundTurn) { const parsed = try std.json.parseFromSlice(std.json.Value, allocator, frame, .{}); errdefer parsed.deinit();
const obj = switch (parsed.value) { .object => |object| object, else => return error.InvalidInboundFrame, }; const typ = jsonString(obj.get("type")) orelse return error.InvalidInboundFrame; if (!std.mem.eql(u8, typ, "input")) return error.NotInputFrame;
const runtime = jsonObject(obj.get("runtime")) orelse return error.InvalidInboundFrame; const conversation_id = jsonString(runtime.get("conversation_id")) orelse return error.MissingConversationId; const payload = jsonObject(obj.get("payload")) orelse return error.InvalidInboundFrame; const kind_string = jsonString(payload.get("kind")) orelse return error.InvalidInboundFrame;
if (std.mem.eql(u8, kind_string, "create_message")) { return .{ .arena = parsed.arena, .value = .{ .request_id = jsonString(obj.get("request_id")), .conversation_id = conversation_id, .kind = .create_message, .messages = payload.get("messages") orelse return error.MissingMessages, .agent_id = jsonString(runtime.get("agent_id")), .acting_user_id = jsonString(runtime.get("acting_user_id")), .teleport_id = null, .client_message_id = firstClientMessageId(payload.get("messages")), }, }; }
if (std.mem.eql(u8, kind_string, "teleport_continue")) { const continuation = jsonObject(payload.get("continuation")); const messages = if (continuation) |c| c.get("approvals") orelse std.json.Value{ .array = std.json.Array.init(parsed.arena.allocator()) } else std.json.Value{ .array = std.json.Array.init(parsed.arena.allocator()) }; return .{ .arena = parsed.arena, .value = .{ .request_id = jsonString(obj.get("request_id")), .conversation_id = conversation_id, .kind = .teleport_continue, .messages = messages, .agent_id = jsonString(runtime.get("agent_id")), .acting_user_id = jsonString(runtime.get("acting_user_id")), .teleport_id = jsonString(payload.get("teleport_id")), .client_message_id = null, }, }; }
return error.UnsupportedInputKind;}
pub fn buildRequestBody(allocator: std.mem.Allocator, inbound: InboundTurn, config: Config) ![]u8 { var out: std.Io.Writer.Allocating = .init(allocator); defer out.deinit();
try out.writer.writeAll("{\"messages\":"); if (inbound.kind == .teleport_continue) { try out.writer.writeByte('['); if (inbound.messages == .array and inbound.messages.array.items.len > 0) { try out.writer.writeAll("{\"type\":\"approval\",\"approvals\":"); try std.json.Stringify.value(inbound.messages, .{}, &out.writer); try out.writer.writeAll(",\"otid\":"); try writeJsonString(&out.writer, inbound.teleport_id orelse "portal-teleport-continuation"); try out.writer.writeAll("},"); } try out.writer.writeAll("{\"role\":\"system\",\"content\":\"<system-reminder>Teleportation to this environment is complete. Continue the existing task from this environment now.</system-reminder>\",\"otid\":"); const teleport_id = inbound.teleport_id orelse "portal-teleport-continuation"; var continue_otid: [512]u8 = undefined; try writeJsonString(&out.writer, try std.fmt.bufPrint(&continue_otid, "{s}:continue", .{teleport_id})); try out.writer.writeAll("}]"); } else { try std.json.Stringify.value(inbound.messages, .{}, &out.writer); } try out.writer.writeAll(",\"streaming\":true,\"stream_tokens\":true,\"include_pings\":true,\"background\":true,\"include_compaction_messages\":true"); if (config.agent_id orelse inbound.agent_id) |agent_id| { try out.writer.writeAll(",\"agent_id\":"); try writeJsonString(&out.writer, agent_id); } try out.writer.writeAll(",\"client_tools\":"); if (config.enable_client_tools) try writeClientTools(&out.writer, config.permission_mode) else try out.writer.writeAll("[]"); try out.writer.writeAll("}"); return out.toOwnedSlice();}
pub fn buildMessagesUrl(allocator: std.mem.Allocator, api_base_url: []const u8, conversation_id: []const u8) ![]u8 { const base = std.mem.trimEnd(u8, api_base_url, "/"); return std.fmt.allocPrint(allocator, "{s}/v1/conversations/{s}/messages", .{ base, conversation_id });}
pub fn runTurnStreaming( allocator: std.mem.Allocator, io: std.Io, config: Config, inbound: InboundTurn, callback: ChunkCallback, callback_ctx: ?*anyopaque,) !Result { const body = try buildRequestBody(allocator, inbound, config); defer allocator.free(body); return runBodyStreaming(allocator, io, config, inbound.conversation_id, body, callback, callback_ctx);}
pub fn runApprovalContinuationStreaming( allocator: std.mem.Allocator, io: std.Io, config: Config, conversation_id: []const u8, approvals: []const ApprovalResult, otid: []const u8, callback: ChunkCallback, callback_ctx: ?*anyopaque,) !Result { var body_writer: std.Io.Writer.Allocating = .init(allocator); defer body_writer.deinit(); try body_writer.writer.writeAll("{\"messages\":[{\"type\":\"approval\",\"approvals\":["); for (approvals, 0..) |approval, i| { if (i != 0) try body_writer.writer.writeByte(','); try body_writer.writer.writeAll("{\"type\":\"tool\",\"tool_call_id\":"); try writeJsonString(&body_writer.writer, approval.tool_call_id); try body_writer.writer.writeAll(",\"status\":"); try writeJsonString(&body_writer.writer, @tagName(approval.status)); try body_writer.writer.writeAll(",\"tool_return\":"); try writeJsonString(&body_writer.writer, approval.tool_return); try body_writer.writer.writeByte('}'); } try body_writer.writer.writeAll("],\"otid\":"); try writeJsonString(&body_writer.writer, otid); try body_writer.writer.writeAll("}],\"streaming\":true,\"stream_tokens\":true,\"include_pings\":true,\"background\":true,\"include_compaction_messages\":true"); if (config.agent_id) |agent_id| { try body_writer.writer.writeAll(",\"agent_id\":"); try writeJsonString(&body_writer.writer, agent_id); } try body_writer.writer.writeAll(",\"client_tools\":"); if (config.enable_client_tools) try writeClientTools(&body_writer.writer, config.permission_mode) else try body_writer.writer.writeAll("[]"); try body_writer.writer.writeByte('}'); const body = try body_writer.toOwnedSlice(); defer allocator.free(body); return runBodyStreaming(allocator, io, config, conversation_id, body, callback, callback_ctx);}
fn runBodyStreaming( allocator: std.mem.Allocator, io: std.Io, config: Config, conversation_id: []const u8, body: []u8, callback: ChunkCallback, callback_ctx: ?*anyopaque,) !Result { const url = try buildMessagesUrl(allocator, config.api_base_url, conversation_id); defer allocator.free(url); const auth = try std.fmt.allocPrint(allocator, "Bearer {s}", .{config.api_key}); defer allocator.free(auth);
var client: std.http.Client = .{ .allocator = allocator, .io = io }; defer client.deinit();
var headers: [5]std.http.Header = undefined; headers[0] = .{ .name = "Authorization", .value = auth }; headers[1] = .{ .name = "Content-Type", .value = "application/json" }; headers[2] = .{ .name = "Accept", .value = "text/event-stream" }; headers[3] = .{ .name = "X-Letta-Source", .value = "letta-code" }; var header_count: usize = 4; if (config.acting_user_id) |acting_user_id| { headers[header_count] = .{ .name = "X-Letta-Acting-User-Id", .value = acting_user_id }; header_count += 1; } if (isCancelled(config.cancel_token)) return error.Cancelled; var request = try requestWithInactivityTimeout(&client, io, .POST, try std.Uri.parse(url), .{ .redirect_behavior = .unhandled, .extra_headers = headers[0..header_count], .keep_alive = false, .headers = .{ .accept_encoding = .{ .override = "identity" } }, }, http_operation_timeout_ms, config.cancel_token); defer request.deinit(); try sendBodyWithInactivityTimeout(io, &request, body, http_operation_timeout_ms, config.cancel_token);
var response = try receiveHeadWithInactivityTimeout(io, &request, http_operation_timeout_ms, config.cancel_token); const status = response.head.status; if (response.head.content_encoding != .identity) return error.UnsupportedCompressionMethod; var transfer_buffer: [response_read_buffer_bytes]u8 = undefined; const reader = response.reader(&transfer_buffer);
if (@intFromEnum(status) < 200 or @intFromEnum(status) >= 300) { var response_body: std.ArrayList(u8) = .empty; defer response_body.deinit(allocator); var read_buffer: [response_read_buffer_bytes]u8 = undefined; while (true) { var vectors = [_][]u8{&read_buffer}; const bytes_read = readApiErrorBodyWithInactivityTimeout(io, reader, &vectors, http_operation_timeout_ms, config.cancel_token) catch |err| switch (err) { error.EndOfStream => break, error.ReadFailed => return response.bodyErr() orelse err, else => return err, }; if (bytes_read == 0) continue; const remaining_capacity = max_api_error_body_bytes - response_body.items.len; try response_body.appendSlice(allocator, read_buffer[0..@min(bytes_read, remaining_capacity)]); } const owned_body = try response_body.toOwnedSlice(allocator); return .{ .api_error = .{ .status = status, .body = owned_body } }; }
var parser = SseParser.init(allocator, callback, callback_ctx); defer parser.deinit(); var read_buffer: [response_read_buffer_bytes]u8 = undefined; while (true) { var vectors = [_][]u8{&read_buffer}; if (isCancelled(config.cancel_token)) return error.Cancelled; const bytes_read = readWithInactivityTimeout(io, reader, &vectors, sse_inactivity_timeout_ms, config.cancel_token) catch |err| switch (err) { error.EndOfStream => break, error.ReadFailed => return response.bodyErr() orelse err, else => return err, }; if (bytes_read == 0) continue; try parser.feed(read_buffer[0..bytes_read]); } if (isCancelled(config.cancel_token)) return error.Cancelled; return parser.finish();}
fn runWithInactivityTimeout( comptime ReturnType: type, io: std.Io, timeout_ms: i64, comptime operation: anytype, operation_args: anytype, comptime timeout_error: anyerror, cancel_token: ?*const std.atomic.Value(bool),) !ReturnType { const OperationResult = @typeInfo(@TypeOf(operation)).@"fn".return_type.?; const Selection = union(enum) { operation: OperationResult, timeout: std.Io.Cancelable!void, cancelled: anyerror!void, }; const Tasks = struct { fn timeout(target_io: std.Io, milliseconds: i64) std.Io.Cancelable!void { return std.Io.sleep(target_io, .fromMilliseconds(milliseconds), .awake); }
fn waitForCancel(target_io: std.Io, token: *const std.atomic.Value(bool)) anyerror!void { while (!token.load(.acquire)) { try std.Io.sleep(target_io, .fromMilliseconds(cancel_poll_interval_ms), .awake); } return error.Cancelled; } }; if (isCancelled(cancel_token)) return error.Cancelled; var results: [3]Selection = undefined; var select = std.Io.Select(Selection).init(io, &results); select.async(.operation, operation, operation_args); select.async(.timeout, Tasks.timeout, .{ io, timeout_ms }); if (cancel_token) |token| select.async(.cancelled, Tasks.waitForCancel, .{ io, token }); defer select.cancelDiscard(); const first = try select.await(); return switch (first) { .operation => |operation_result| try operation_result, .timeout => |timeout_result| { try timeout_result; return timeout_error; }, .cancelled => |cancel_result| { try cancel_result; return error.Cancelled; }, };}
fn requestWithInactivityTimeout( client: *std.http.Client, io: std.Io, method: std.http.Method, uri: std.Uri, options: std.http.Client.RequestOptions, timeout_ms: i64, cancel_token: ?*const std.atomic.Value(bool),) !std.http.Client.Request { const Slot = struct { request: ?std.http.Client.Request = null }; const Tasks = struct { fn create(slot: *Slot, target: *std.http.Client, request_method: std.http.Method, request_uri: std.Uri, request_options: std.http.Client.RequestOptions) anyerror!void { slot.request = try target.request(request_method, request_uri, request_options); } }; var slot = Slot{}; runWithInactivityTimeout(void, io, timeout_ms, Tasks.create, .{ &slot, client, method, uri, options }, error.HttpOperationTimeout, cancel_token) catch |err| { if (slot.request) |*request| request.deinit(); return err; }; const request = slot.request orelse return error.HttpRequestMissing; slot.request = null; return request;}
fn sendBodyWithInactivityTimeout(io: std.Io, request: *std.http.Client.Request, body: []u8, timeout_ms: i64, cancel_token: ?*const std.atomic.Value(bool)) !void { const Tasks = struct { fn send(target: *std.http.Client.Request, request_body: []u8) anyerror!void { return target.sendBodyComplete(request_body); } }; return runWithInactivityTimeout(void, io, timeout_ms, Tasks.send, .{ request, body }, error.HttpOperationTimeout, cancel_token);}
fn receiveHeadWithInactivityTimeout(io: std.Io, request: *std.http.Client.Request, timeout_ms: i64, cancel_token: ?*const std.atomic.Value(bool)) !std.http.Client.Response { const Tasks = struct { fn receiveHead(target: *std.http.Client.Request) std.http.Client.Request.ReceiveHeadError!std.http.Client.Response { return target.receiveHead(&.{}); } }; return runWithInactivityTimeout(std.http.Client.Response, io, timeout_ms, Tasks.receiveHead, .{request}, error.HttpOperationTimeout, cancel_token);}
fn readWithTimeout(io: std.Io, reader: *std.Io.Reader, vectors: [][]u8, timeout_ms: i64, comptime timeout_error: anyerror, cancel_token: ?*const std.atomic.Value(bool)) !usize { const Tasks = struct { fn read(target: *std.Io.Reader, buffers: [][]u8) anyerror!usize { return target.readVec(buffers); } }; return runWithInactivityTimeout(usize, io, timeout_ms, Tasks.read, .{ reader, vectors }, timeout_error, cancel_token);}
fn readWithInactivityTimeout(io: std.Io, reader: *std.Io.Reader, vectors: [][]u8, timeout_ms: i64, cancel_token: ?*const std.atomic.Value(bool)) !usize { return readWithTimeout(io, reader, vectors, timeout_ms, error.SseInactivityTimeout, cancel_token);}
fn readApiErrorBodyWithInactivityTimeout(io: std.Io, reader: *std.Io.Reader, vectors: [][]u8, timeout_ms: i64, cancel_token: ?*const std.atomic.Value(bool)) !usize { return readWithTimeout(io, reader, vectors, timeout_ms, error.HttpOperationTimeout, cancel_token);}
pub fn collectTurnFromSse(allocator: std.mem.Allocator, sse_body: []const u8) !struct { result: Result, chunks: [][]u8 } { var chunks = std.ArrayList([]u8).empty; errdefer { for (chunks.items) |chunk| allocator.free(chunk); chunks.deinit(allocator); } const Context = struct { allocator: std.mem.Allocator, chunks: *std.ArrayList([]u8), }; var context = Context{ .allocator = allocator, .chunks = &chunks }; const cb = struct { fn append(ctx: ?*anyopaque, raw_chunk: []const u8) !void { const context_ptr: *Context = @ptrCast(@alignCast(ctx.?)); try context_ptr.chunks.append(context_ptr.allocator, try context_ptr.allocator.dupe(u8, raw_chunk)); } }.append; const result = try parseSseAndEmit(allocator, sse_body, cb, &context); return .{ .result = result, .chunks = try chunks.toOwnedSlice(allocator) };}
pub fn parseSseAndEmit(allocator: std.mem.Allocator, sse_body: []const u8, callback: ChunkCallback, callback_ctx: ?*anyopaque) !Result { var parser = SseParser.init(allocator, callback, callback_ctx); defer parser.deinit(); try parser.feed(sse_body); return parser.finish();}
const SseParser = struct { allocator: std.mem.Allocator, callback: ChunkCallback, callback_ctx: ?*anyopaque, line: std.ArrayList(u8) = .empty, event_data: std.ArrayList(u8) = .empty, has_data: bool = false, stream_state: StreamState, finished: bool = false,
fn init(allocator: std.mem.Allocator, callback: ChunkCallback, callback_ctx: ?*anyopaque) SseParser { return .{ .allocator = allocator, .callback = callback, .callback_ctx = callback_ctx, .stream_state = .{ .allocator = allocator }, }; }
fn deinit(self: *SseParser) void { self.line.deinit(self.allocator); self.event_data.deinit(self.allocator); self.stream_state.deinit(); }
fn feed(self: *SseParser, bytes: []const u8) !void { if (self.finished) return error.SseParserFinished; var remaining = bytes; while (std.mem.findScalar(u8, remaining, '\n')) |newline| { try self.appendLineBytes(remaining[0..newline]); try self.consumeLine(); remaining = remaining[newline + 1 ..]; } try self.appendLineBytes(remaining); }
fn finish(self: *SseParser) !Result { if (self.finished) return error.SseParserFinished; self.finished = true; if (self.line.items.len > 0) try self.consumeLine(); try self.dispatchEvent();
if (self.stream_state.stop_reason) |reason| { if (std.mem.eql(u8, reason, "requires_approval") and self.stream_state.pending_tool_calls.items.len > 0) { self.allocator.free(reason); self.stream_state.stop_reason = null; return .{ .requires_approval = try self.stream_state.pending_tool_calls.toOwnedSlice(self.allocator) }; } } return .{ .completed = {} }; }
fn appendLineBytes(self: *SseParser, bytes: []const u8) !void { if (self.line.items.len + bytes.len > max_sse_line_bytes) return error.SseLineTooLarge; try self.line.appendSlice(self.allocator, bytes); }
fn consumeLine(self: *SseParser) !void { const line = std.mem.trimEnd(u8, self.line.items, "\r"); if (line.len == 0) { try self.dispatchEvent(); } else if (std.mem.startsWith(u8, line, "data:")) { var value = line[5..]; if (value.len > 0 and value[0] == ' ') value = value[1..]; const separator_len: usize = if (self.has_data) 1 else 0; if (self.event_data.items.len + separator_len + value.len > max_sse_event_bytes) return error.SseEventTooLarge; if (self.has_data) try self.event_data.append(self.allocator, '\n'); try self.event_data.appendSlice(self.allocator, value); self.has_data = true; } self.line.clearRetainingCapacity(); }
fn dispatchEvent(self: *SseParser) !void { if (!self.has_data) return; self.has_data = false; defer self.event_data.clearRetainingCapacity(); const raw_chunk = self.event_data.items; if (std.mem.eql(u8, raw_chunk, "[DONE]")) return;
var parsed = std.json.parseFromSlice(std.json.Value, self.allocator, raw_chunk, .{}) catch return error.InvalidSseJson; defer parsed.deinit(); try absorbChunk(self.allocator, parsed.value, &self.stream_state); try self.callback(self.callback_ctx, raw_chunk); }};
fn absorbChunk(allocator: std.mem.Allocator, value: std.json.Value, stream_state: *StreamState) !void { const obj = jsonObject(value) orelse return; const message_type = jsonString(obj.get("message_type")); if (message_type) |mt| { if (std.mem.eql(u8, mt, "stop_reason")) { if (stream_state.stop_reason) |old| allocator.free(old); stream_state.stop_reason = try allocator.dupe(u8, jsonString(obj.get("stop_reason")) orelse ""); return; } if (std.mem.eql(u8, mt, "approval_request_message")) { const message_id = jsonString(obj.get("id")); if (obj.get("tool_calls")) |tool_calls_value| { if (tool_calls_value == .array) { for (tool_calls_value.array.items) |tool_call_value| { try appendPendingToolCall(allocator, stream_state, tool_call_value, message_id); } } } if (obj.get("tool_call")) |tool_call_value| { try appendPendingToolCall(allocator, stream_state, tool_call_value, message_id); } } }}
fn appendPendingToolCall(allocator: std.mem.Allocator, stream_state: *StreamState, tool_call_value: std.json.Value, message_id: ?[]const u8) !void { const tool_call = jsonObject(tool_call_value) orelse return; const id = jsonString(tool_call.get("tool_call_id")) orelse return; const name = jsonString(tool_call.get("name")) orelse ""; const arguments = jsonString(tool_call.get("arguments")) orelse ""; for (stream_state.pending_tool_calls.items) |*existing| { if (std.mem.eql(u8, existing.tool_call_id, id)) { if (name.len > 0) { allocator.free(existing.name); existing.name = try allocator.dupe(u8, name); } if (arguments.len > 0) { const old = existing.arguments; existing.arguments = try std.fmt.allocPrint(allocator, "{s}{s}", .{ old, arguments }); allocator.free(old); } if (existing.message_id == null and message_id != null) existing.message_id = try allocator.dupe(u8, message_id.?); return; } } try stream_state.pending_tool_calls.append(allocator, .{ .tool_call_id = try allocator.dupe(u8, id), .name = try allocator.dupe(u8, name), .arguments = try allocator.dupe(u8, arguments), .message_id = if (message_id) |mid| try allocator.dupe(u8, mid) else null, });}
fn writeClientTools(writer: *std.Io.Writer, permission_mode: PermissionMode) !void { try writer.writeAll("["); try writeTool(writer, "Read", "Read a file from the local filesystem", "{\"type\":\"object\",\"properties\":{\"file_path\":{\"type\":\"string\"},\"offset\":{\"type\":\"integer\"},\"limit\":{\"type\":\"integer\"}},\"required\":[\"file_path\"]}"); try writer.writeAll(","); try writeTool(writer, "LS", "List directory entries", "{\"type\":\"object\",\"properties\":{\"path\":{\"type\":\"string\"}},\"required\":[\"path\"]}"); try writer.writeAll(","); try writeTool(writer, "Glob", "Find files by glob pattern", "{\"type\":\"object\",\"properties\":{\"pattern\":{\"type\":\"string\"},\"path\":{\"type\":\"string\"}},\"required\":[\"pattern\"]}"); try writer.writeAll(","); try writeTool(writer, "Grep", "Search file contents by pattern", "{\"type\":\"object\",\"properties\":{\"pattern\":{\"type\":\"string\"},\"path\":{\"type\":\"string\"},\"glob\":{\"type\":\"string\"}},\"required\":[\"pattern\"]}"); if (permission_mode.allowsBash()) { try writer.writeAll(","); try writeTool(writer, "Bash", "Execute a shell command when permission mode allows it", "{\"type\":\"object\",\"properties\":{\"command\":{\"type\":\"string\"},\"description\":{\"type\":\"string\"},\"timeout\":{\"type\":\"integer\"},\"run_in_background\":{\"type\":\"boolean\"}},\"required\":[\"command\",\"description\"]}"); } try writer.writeAll("]");}
fn writeTool(writer: *std.Io.Writer, name: []const u8, description: []const u8, parameters_json: []const u8) !void { try writer.writeAll("{\"name\":"); try writeJsonString(writer, name); try writer.writeAll(",\"description\":"); try writeJsonString(writer, description); try writer.writeAll(",\"parameters\":"); try writer.writeAll(parameters_json); try writer.writeAll("}");}
fn writeJsonString(writer: *std.Io.Writer, value: []const u8) !void { try std.json.Stringify.value(value, .{}, writer);}
fn firstClientMessageId(value: ?std.json.Value) ?[]const u8 { const messages = value orelse return null; if (messages != .array) return null; for (messages.array.items) |message| { if (message != .object or message.object.get("content") == null) continue; if (jsonString(message.object.get("client_message_id"))) |client_message_id| { if (client_message_id.len > 0) return client_message_id; } } return null;}
fn jsonString(value: ?std.json.Value) ?[]const u8 { const v = value orelse return null; return switch (v) { .string => |s| s, else => null, };}
fn jsonObject(value: ?std.json.Value) ?std.json.ObjectMap { const v = value orelse return null; return switch (v) { .object => |object| object, else => null, };}
pub fn freePendingToolCalls(allocator: std.mem.Allocator, calls: []PendingToolCall) void { for (calls) |call| call.deinit(allocator); allocator.free(calls);}
pub fn freeChunks(allocator: std.mem.Allocator, chunks: [][]u8) void { for (chunks) |chunk| allocator.free(chunk); allocator.free(chunks);}
test "parse inbound create_message extracts runtime and messages" { const frame = "{\"type\":\"input\",\"request_id\":\"req-1\",\"runtime\":{\"conversation_id\":\"conv-1\",\"agent_id\":\"agent-1\",\"acting_user_id\":\"user-1\"},\"payload\":{\"kind\":\"create_message\",\"messages\":[{\"type\":\"message\",\"role\":\"user\",\"content\":\"hi\"}]}}"; var parsed = try parseInboundInput(std.testing.allocator, frame); defer parsed.deinit(); try std.testing.expectEqualStrings("req-1", parsed.value.request_id.?); try std.testing.expectEqualStrings("conv-1", parsed.value.conversation_id); try std.testing.expect(parsed.value.kind == .create_message); try std.testing.expect(parsed.value.messages == .array); try std.testing.expectEqual(@as(usize, 1), parsed.value.messages.array.items.len); try std.testing.expectEqualStrings("user-1", parsed.value.acting_user_id.?);}
test "request body includes required streaming flags and gated Bash" { const frame = "{\"type\":\"input\",\"runtime\":{\"conversation_id\":\"conv-1\"},\"payload\":{\"kind\":\"create_message\",\"messages\":[{\"type\":\"message\",\"role\":\"user\",\"content\":\"hi\"}]}}"; var parsed = try parseInboundInput(std.testing.allocator, frame); defer parsed.deinit(); const body_standard = try buildRequestBody(std.testing.allocator, parsed.value, .{ .api_base_url = "https://api.letta.com", .api_key = "test", .permission_mode = .standard, .enable_client_tools = true }); defer std.testing.allocator.free(body_standard); try std.testing.expect(std.mem.indexOf(u8, body_standard, "\"streaming\":true") != null); try std.testing.expect(std.mem.indexOf(u8, body_standard, "\"include_compaction_messages\":true") != null); try std.testing.expect(std.mem.indexOf(u8, body_standard, "\"name\":\"Bash\"") == null);
const body_unrestricted = try buildRequestBody(std.testing.allocator, parsed.value, .{ .api_base_url = "https://api.letta.com", .api_key = "test", .permission_mode = .unrestricted, .enable_client_tools = true }); defer std.testing.allocator.free(body_unrestricted); try std.testing.expect(std.mem.indexOf(u8, body_unrestricted, "\"name\":\"Bash\"") != null);}
test "teleport continuation wraps tool results in approval message" { const frame = "{\"type\":\"input\",\"runtime\":{\"conversation_id\":\"conv-1\",\"agent_id\":\"agent-1\"},\"payload\":{\"kind\":\"teleport_continue\",\"teleport_id\":\"teleport-1\",\"continuation\":{\"approvals\":[{\"type\":\"tool\",\"tool_call_id\":\"call-1\",\"status\":\"success\",\"tool_return\":\"done\"}]}}}"; var parsed = try parseInboundInput(std.testing.allocator, frame); defer parsed.deinit(); const body = try buildRequestBody(std.testing.allocator, parsed.value, .{ .api_base_url = "https://api.letta.com", .api_key = "test" }); defer std.testing.allocator.free(body); try std.testing.expect(std.mem.indexOf(u8, body, "\"type\":\"approval\"") != null); try std.testing.expect(std.mem.indexOf(u8, body, "\"approvals\":[{\"type\":\"tool\"") != null); try std.testing.expect(std.mem.indexOf(u8, body, "\"otid\":\"teleport-1\"") != null);}
test "empty teleport continuation includes exact completion reminder" { const frame = "{\"type\":\"input\",\"runtime\":{\"conversation_id\":\"conv-1\",\"agent_id\":\"agent-1\"},\"payload\":{\"kind\":\"teleport_continue\",\"teleport_id\":\"teleport-empty\"}}"; var parsed = try parseInboundInput(std.testing.allocator, frame); defer parsed.deinit(); const body = try buildRequestBody(std.testing.allocator, parsed.value, .{ .api_base_url = "https://api.letta.com", .api_key = "test" }); defer std.testing.allocator.free(body); try std.testing.expect(std.mem.indexOf(u8, body, "\"type\":\"approval\"") == null); try std.testing.expect(std.mem.indexOf(u8, body, "<system-reminder>Teleportation to this environment is complete. Continue the existing task from this environment now.</system-reminder>") != null); try std.testing.expect(std.mem.indexOf(u8, body, "\"otid\":\"teleport-empty:continue\"") != null);}
const TimeoutTestReader = struct { reader: std.Io.Reader, io: std.Io, exited: ?*std.atomic.Value(bool) = null,
const vtable = std.Io.Reader.VTable{ .stream = stream, .readVec = readVec };
fn init(io: std.Io, buffer: []u8, exited: ?*std.atomic.Value(bool)) @This() { return .{ .reader = .{ .vtable = &vtable, .buffer = buffer, .seek = 0, .end = 0 }, .io = io, .exited = exited, }; }
fn wait(self: *@This()) std.Io.Reader.Error!usize { std.Io.sleep(self.io, .fromSeconds(30), .awake) catch { if (self.exited) |exited| exited.store(true, .release); return error.ReadFailed; }; if (self.exited) |exited| exited.store(true, .release); return error.EndOfStream; }
fn stream(reader: *std.Io.Reader, _: *std.Io.Writer, _: std.Io.Limit) std.Io.Reader.StreamError!usize { const self: *@This() = @fieldParentPtr("reader", reader); return self.wait() catch |err| return err; }
fn readVec(reader: *std.Io.Reader, _: [][]u8) std.Io.Reader.Error!usize { const self: *@This() = @fieldParentPtr("reader", reader); return self.wait(); }};
test "SSE inactivity timeout cancels a stalled read" { var internal_buffer: [1]u8 = undefined; var stalled = TimeoutTestReader.init(std.testing.io, &internal_buffer, null); var output: [1]u8 = undefined; var vectors = [_][]u8{&output}; try std.testing.expectError(error.SseInactivityTimeout, readWithInactivityTimeout(std.testing.io, &stalled.reader, &vectors, 10, null));}
test "HTTP request creation timeout cancels stalled operation" { const Tasks = struct { fn stalled(io: std.Io, exited: *std.atomic.Value(bool)) anyerror!usize { std.Io.sleep(io, .fromSeconds(30), .awake) catch { exited.store(true, .release); return error.Canceled; }; return 0; } }; var exited: std.atomic.Value(bool) = .init(false); try std.testing.expectError(error.HttpOperationTimeout, runWithInactivityTimeout(usize, std.testing.io, 10, Tasks.stalled, .{ std.testing.io, &exited }, error.HttpOperationTimeout, null)); try std.testing.expect(exited.load(.acquire));}
test "HTTP send body timeout cancels stalled operation" { const Tasks = struct { fn stalled(io: std.Io, exited: *std.atomic.Value(bool)) anyerror!void { std.Io.sleep(io, .fromSeconds(30), .awake) catch { exited.store(true, .release); return error.Canceled; }; } }; var exited: std.atomic.Value(bool) = .init(false); try std.testing.expectError(error.HttpOperationTimeout, runWithInactivityTimeout(void, std.testing.io, 10, Tasks.stalled, .{ std.testing.io, &exited }, error.HttpOperationTimeout, null)); try std.testing.expect(exited.load(.acquire));}
test "HTTP error body inactivity timeout cancels stalled read" { var internal_buffer: [1]u8 = undefined; var stalled = TimeoutTestReader.init(std.testing.io, &internal_buffer, null); var output: [1]u8 = undefined; var vectors = [_][]u8{&output}; try std.testing.expectError(error.HttpOperationTimeout, readApiErrorBodyWithInactivityTimeout(std.testing.io, &stalled.reader, &vectors, 10, null));}
test "cancel token aborts a stalled SSE read" { var token: std.atomic.Value(bool) = .init(false); const Tasks = struct { fn cancelLater(io: std.Io, t: *std.atomic.Value(bool)) !void { try std.Io.sleep(io, .fromMilliseconds(20), .awake); t.store(true, .release); } }; var group: std.Io.Group = .init; group.async(std.testing.io, Tasks.cancelLater, .{ std.testing.io, &token }); defer group.cancel(std.testing.io); var internal_buffer: [1]u8 = undefined; var stalled = TimeoutTestReader.init(std.testing.io, &internal_buffer, null); var output: [1]u8 = undefined; var vectors = [_][]u8{&output}; try std.testing.expectError(error.Cancelled, readWithInactivityTimeout(std.testing.io, &stalled.reader, &vectors, 30_000, &token));}
test "cancel token aborts a stalled HTTP operation" { var token: std.atomic.Value(bool) = .init(false); const Tasks = struct { fn stalled(io: std.Io, exited: *std.atomic.Value(bool)) anyerror!void { std.Io.sleep(io, .fromSeconds(30), .awake) catch { exited.store(true, .release); return error.Canceled; }; }
fn cancelLater(io: std.Io, t: *std.atomic.Value(bool)) !void { try std.Io.sleep(io, .fromMilliseconds(20), .awake); t.store(true, .release); } }; var exited: std.atomic.Value(bool) = .init(false); var group: std.Io.Group = .init; group.async(std.testing.io, Tasks.cancelLater, .{ std.testing.io, &token }); defer group.cancel(std.testing.io); try std.testing.expectError(error.Cancelled, runWithInactivityTimeout(void, std.testing.io, 30_000, Tasks.stalled, .{ std.testing.io, &exited }, error.HttpOperationTimeout, &token));}
test "absent cancel token preserves timeout behavior" { const Tasks = struct { fn stalled(io: std.Io, exited: *std.atomic.Value(bool)) anyerror!void { std.Io.sleep(io, .fromSeconds(30), .awake) catch { exited.store(true, .release); return error.Canceled; }; } }; var exited: std.atomic.Value(bool) = .init(false); try std.testing.expectError(error.HttpOperationTimeout, runWithInactivityTimeout(void, std.testing.io, 10, Tasks.stalled, .{ std.testing.io, &exited }, error.HttpOperationTimeout, null)); try std.testing.expect(exited.load(.acquire));}
test "parent cancellation waits for nested Select cleanup" { var exited: std.atomic.Value(bool) = .init(false); var internal_buffer: [1]u8 = undefined; var stalled = TimeoutTestReader.init(std.testing.io, &internal_buffer, &exited); var output: [1]u8 = undefined; var vectors = [_][]u8{&output}; const Selection = union(enum) { operation: anyerror!usize }; const Tasks = struct { fn read(io: std.Io, reader: *std.Io.Reader, buffers: [][]u8) anyerror!usize { return readWithInactivityTimeout(io, reader, buffers, 30_000, null); } }; var results: [1]Selection = undefined; var parent = std.Io.Select(Selection).init(std.testing.io, &results); parent.async(.operation, Tasks.read, .{ std.testing.io, &stalled.reader, &vectors }); parent.cancelDiscard(); try std.testing.expect(exited.load(.acquire));}
test "SSE parser rejects malformed JSON before callback" { var callback_count: usize = 0; const cb = struct { fn count(ctx: ?*anyopaque, _: []const u8) !void { const count_ptr: *usize = @ptrCast(@alignCast(ctx.?)); count_ptr.* += 1; } }.count; try std.testing.expectError(error.InvalidSseJson, parseSseAndEmit(std.testing.allocator, "data: {not-json}\n\n", cb, &callback_count)); try std.testing.expectEqual(@as(usize, 0), callback_count);}
test "SSE parser emits chunks and surfaces approval stops" { const sse = "event: message\n" ++ "data: {\"message_type\":\"assistant_message\",\"content\":\"thinking\"}\n\n" ++ "data: {\"message_type\":\"approval_request_message\",\"id\":\"msg-1\",\"tool_calls\":[{\"tool_call_id\":\"call-1\",\"name\":\"Bash\",\"arguments\":\"pw\"}]}\n\n" ++ "data: {\"message_type\":\"approval_request_message\",\"tool_calls\":[{\"tool_call_id\":\"call-1\",\"arguments\":\"d\"}]}\n\n" ++ "data: {\"message_type\":\"stop_reason\",\"stop_reason\":\"requires_approval\"}\n\n"; var emitted = std.ArrayList([]u8).empty; defer { for (emitted.items) |chunk| std.testing.allocator.free(chunk); emitted.deinit(std.testing.allocator); } const cb = struct { fn append(ctx: ?*anyopaque, raw_chunk: []const u8) !void { const list: *std.ArrayList([]u8) = @ptrCast(@alignCast(ctx.?)); try list.append(std.testing.allocator, try std.testing.allocator.dupe(u8, raw_chunk)); } }.append; const result = try parseSseAndEmit(std.testing.allocator, sse, cb, &emitted); try std.testing.expectEqual(@as(usize, 4), emitted.items.len); switch (result) { .requires_approval => |calls| { defer freePendingToolCalls(std.testing.allocator, calls); try std.testing.expectEqual(@as(usize, 1), calls.len); try std.testing.expectEqualStrings("call-1", calls[0].tool_call_id); try std.testing.expectEqualStrings("Bash", calls[0].name); try std.testing.expectEqualStrings("pwd", calls[0].arguments); try std.testing.expectEqualStrings("msg-1", calls[0].message_id.?); }, else => return error.ExpectedApproval, }}
test "incremental SSE parser emits complete CRLF event before stream end" { const Context = struct { count: usize = 0, first: ?[]u8 = null, }; var context = Context{}; defer if (context.first) |first| std.testing.allocator.free(first); const cb = struct { fn capture(ctx: ?*anyopaque, raw_chunk: []const u8) !void { const capture_context: *Context = @ptrCast(@alignCast(ctx.?)); capture_context.count += 1; if (capture_context.first == null) capture_context.first = try std.testing.allocator.dupe(u8, raw_chunk); } }.capture;
var parser = SseParser.init(std.testing.allocator, cb, &context); defer parser.deinit(); try parser.feed("data: {\"message_type\":\"assistant_message\","); try std.testing.expectEqual(@as(usize, 0), context.count); try parser.feed("\r\ndata: \"content\":\"first\"}\r"); try std.testing.expectEqual(@as(usize, 0), context.count); try parser.feed("\n\r\n"); try std.testing.expectEqual(@as(usize, 1), context.count); try std.testing.expectEqualStrings("{\"message_type\":\"assistant_message\",\n\"content\":\"first\"}", context.first.?);
try parser.feed("data:{\"message_type\":\"assistant_message\",\"content\":\"later\"}\n\n"); try std.testing.expectEqual(@as(usize, 2), context.count); const result = try parser.finish(); try std.testing.expect(result == .completed);}
test "incremental SSE parser bounds an unfinished event" { const cb = struct { fn reject(_: ?*anyopaque, _: []const u8) !void { return error.UnexpectedEmission; } }.reject; var parser = SseParser.init(std.testing.allocator, cb, null); defer parser.deinit();
const payload_bytes = 64 * 1024; const line = try std.testing.allocator.alloc(u8, payload_bytes + "data: \n".len); defer std.testing.allocator.free(line); @memcpy(line[0.."data: ".len], "data: "); @memset(line["data: ".len .. line.len - 1], 'x'); line[line.len - 1] = '\n';
const accepted_lines = (max_sse_event_bytes + 1) / (payload_bytes + 1); for (0..accepted_lines) |_| try parser.feed(line); try std.testing.expect(parser.event_data.items.len <= max_sse_event_bytes); try std.testing.expectError(error.SseEventTooLarge, parser.feed(line)); try std.testing.expect(parser.event_data.items.len <= max_sse_event_bytes);}