const 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);
}