const std = @import("std");const build_options = @import("build_options");const acp_server = @import("acp/server.zig");const jsonrpc = @import("acp/jsonrpc.zig");const background_process_provider = @import("core/execution/background_process_provider.zig");const gateway_provider = @import("core/gateway/gateway_provider.zig");const generation_usage_provider = @import("core/session/generation_usage_provider.zig");const host = @import("core/hosts/host.zig");const debug_trace = @import("core/shared/debug_trace.zig");const io_mod = @import("core/shared/io.zig");const fetch_state = @import("napi_fetch_state.zig");const streamable_http = @import("core/mcp/streamable_http.zig");const host_stream_provider = @import("gateway/host_stream_provider.zig");const oauth_transport = @import("core/auth/oauth_transport.zig");const builtin_context = @import("builtins/context.zig");const builtin_gateway = @import("builtins/gateway.zig");const builtin_modes = @import("builtins/modes.zig"); const c = @cImport({    @cInclude("node_api.h");}); const Allocator = std.mem.Allocator;const max_drain_bytes = 1024 * 1024;const max_input_bytes = 8 * 1024 * 1024;const max_output_bytes = 8 * 1024 * 1024;const max_fetch_request_bytes = 8 * 1024 * 1024;const max_fetch_response_bytes = 8 * 1024 * 1024;const max_api_key_bytes = 64 * 1024;const max_model_bytes = 1024;const max_path_bytes = 16 * 1024;const max_url_bytes = 16 * 1024;const max_active_runtimes = 64;const runtime_handle_type_tag = c.napi_type_tag{    .lower = 0x4c4942465852544d,    .upper = 0xa71d7c52e9314b08,}; comptime {    if (build_options.napi_surface != .core) {        @compileError("libfx N-API core requires -Dnapi-surface=core");    }} const InputQueue = struct {    mutex: std.Io.Mutex = .init,    wake: std.Io.Condition = .init,    bytes: std.ArrayList(u8) = .empty,    offset: usize = 0,    closed: bool = false,     fn read(self: *InputQueue, alloc: Allocator, destination: []u8) usize {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        while (self.offset == self.bytes.items.len and !self.closed) {            self.bytes.clearRetainingCapacity();            self.offset = 0;            self.wake.waitUncancelable(io, &self.mutex);        }        const available = self.bytes.items.len - self.offset;        if (available == 0) return 0;        const len = @min(destination.len, available);        @memcpy(destination[0..len], self.bytes.items[self.offset..][0..len]);        self.offset += len;        if (self.offset == self.bytes.items.len) {            self.bytes.clearRetainingCapacity();            self.offset = 0;        }        _ = alloc;        return len;    }     fn write(self: *InputQueue, alloc: Allocator, data: []const u8) !void {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        if (self.closed) return error.InputClosed;        const queued = self.bytes.items.len - self.offset;        if (data.len > max_input_bytes or queued > max_input_bytes - data.len) return error.InputQueueFull;        if (self.offset > 0) {            std.mem.copyForwards(u8, self.bytes.items[0..queued], self.bytes.items[self.offset..]);            self.bytes.items.len = queued;            self.offset = 0;        }        try self.bytes.appendSlice(alloc, data);        self.wake.signal(io);    }     fn close(self: *InputQueue) void {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        self.closed = true;        self.wake.broadcast(io);        self.mutex.unlock(io);    }     fn deinit(self: *InputQueue, alloc: Allocator) void {        self.bytes.deinit(alloc);    }}; const OutputQueue = struct {    mutex: std.Io.Mutex = .init,    bytes: std.ArrayList(u8) = .empty,    offset: usize = 0,     fn write(self: *OutputQueue, alloc: Allocator, data: []const u8) !void {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        const queued = self.bytes.items.len - self.offset;        if (data.len > max_output_bytes or queued > max_output_bytes - data.len) return error.OutputQueueFull;        if (self.offset > 0) {            std.mem.copyForwards(u8, self.bytes.items[0..queued], self.bytes.items[self.offset..]);            self.bytes.items.len = queued;            self.offset = 0;        }        try self.bytes.appendSlice(alloc, data);    }     fn drain(self: *OutputQueue, destination: []u8) usize {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        const available_bytes = self.bytes.items.len - self.offset;        const len = @min(destination.len, available_bytes);        if (len == 0) return 0;        @memcpy(destination[0..len], self.bytes.items[self.offset..][0..len]);        self.offset += len;        if (self.offset == self.bytes.items.len) {            self.bytes.clearRetainingCapacity();            self.offset = 0;        }        return len;    }     fn available(self: *OutputQueue) usize {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        return self.bytes.items.len - self.offset;    }     fn deinit(self: *OutputQueue, alloc: Allocator) void {        self.bytes.deinit(alloc);    }}; const FetchBridge = struct {    mutex: std.Io.Mutex = .init,    wake: std.Io.Condition = .init,    request: std.ArrayList(u8) = .empty,    response: std.ArrayList(u8) = .empty,    response_offset: usize = 0,    phase: fetch_state.Phase = .idle,    next_handle: fetch_state.Handle = 1,    status: u16 = 0,     fn clearPendingRequest(self: *FetchBridge, reason: []const u8) void {        if (self.request.items.len == 0) return;        debug_trace.logf("napi", "dropping pending host fetch reason={s} bytes={d}", .{ reason, self.request.items.len });        self.request.clearRetainingCapacity();    }     fn trace_stale(operation: []const u8, handle: fetch_state.Handle, reason: fetch_state.StaleReason) void {        debug_trace.logf(            "napi",            "dropping stale host fetch operation={s} handle={d} reason={s}",            .{ operation, handle, @tagName(reason) },        );    }     fn advance_handle(self: *FetchBridge) void {        self.next_handle = if (self.next_handle == std.math.maxInt(fetch_state.Handle))            0        else            self.next_handle + 1;    }     fn open(raw: ?*anyopaque, method: []const u8, url: []const u8, headers: []const u8, body: []const u8) !i32 {        const self: *FetchBridge = @ptrCast(@alignCast(raw.?));        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        if (self.next_handle == 0) return error.HostStreamUnavailable;        const handle = self.next_handle;        const decision = fetch_state.decide(self.phase, .{ .open = handle });        switch (decision.action) {            .cancelled => {                self.phase = decision.phase;                return error.Cancelled;            },            .unavailable, .shutting_down => return error.HostStreamUnavailable,            .applied => {},            else => unreachable,        }        self.response_offset = 0;        self.response.clearRetainingCapacity();        var writer: std.Io.Writer.Allocating = .init(std.heap.c_allocator);        defer writer.deinit();        const encoded_body = std.base64.standard.Encoder.calcSize(body.len);        const body_base64 = try std.heap.c_allocator.alloc(u8, encoded_body);        defer std.heap.c_allocator.free(body_base64);        _ = std.base64.standard.Encoder.encode(body_base64, body);        try std.json.Stringify.value(.{            .handle = handle,            .method = method,            .url = url,            .headers = headers,            .body = body_base64,        }, .{}, &writer.writer);        const bytes = writer.writer.buffered();        if (bytes.len > max_fetch_request_bytes) return error.HostStreamBackpressure;        self.request.clearRetainingCapacity();        try self.request.appendSlice(std.heap.c_allocator, bytes);        self.phase = decision.phase;        self.advance_handle();        self.wake.broadcast(io);        return handle;    }     fn statusFn(raw: ?*anyopaque, handle: i32, status_out: *u16) i32 {        const self: *FetchBridge = @ptrCast(@alignCast(raw.?));        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        while (true) switch (self.phase) {            .request_pending, .awaiting_response => |active| {                if (active != handle) return -2;                self.wake.waitUncancelable(io, &self.mutex);            },            .streaming => |active| {                if (active != handle) return -2;                status_out.* = self.status;                return 1;            },            .terminal => |terminal| {                if (terminal.handle != handle) return -2;                return switch (terminal.kind) {                    .success => result: {                        status_out.* = self.status;                        break :result 1;                    },                    .failure => -1,                    .aborted => -2,                };            },            else => return -2,        };    }     fn next(raw: ?*anyopaque, handle: i32, out: []u8) i32 {        const self: *FetchBridge = @ptrCast(@alignCast(raw.?));        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        while (true) {            const terminal_kind: ?fetch_state.TerminalKind = switch (self.phase) {                .streaming => |active| if (active == handle) null else return -2,                .terminal => |terminal| if (terminal.handle == handle) terminal.kind else return -2,                else => return -2,            };            if (terminal_kind) |kind| switch (kind) {                .failure => return -1,                .aborted => return -2,                .success => {},            };            const available = self.response.items.len - self.response_offset;            if (available > 0) {                const len = @min(out.len, available);                @memcpy(out[0..len], self.response.items[self.response_offset..][0..len]);                self.response_offset += len;                self.wake.broadcast(io);                return @intCast(len);            }            self.response.clearRetainingCapacity();            self.response_offset = 0;            if (terminal_kind != null) return 0;            self.wake.waitUncancelable(io, &self.mutex);        }    }     fn close(raw: ?*anyopaque, handle: i32) void {        const self: *FetchBridge = @ptrCast(@alignCast(raw.?));        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        const decision = fetch_state.decide(self.phase, .{ .close = handle });        if (decision.action == .stale) {            trace_stale("close", handle, decision.stale_reason.?);            return;        }        self.clearPendingRequest("stream_close");        self.phase = decision.phase;        self.status = 0;        self.response.clearRetainingCapacity();        self.response_offset = 0;        self.wake.broadcast(io);    }     fn startResponse(self: *FetchBridge, handle: fetch_state.Handle, status: u16) FetchOperationResult {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        const decision = fetch_state.decide(self.phase, .{ .start = handle });        if (decision.action == .stale) {            trace_stale("start", handle, decision.stale_reason.?);            return .stale;        }        self.status = status;        self.phase = decision.phase;        self.wake.broadcast(io);        return .applied;    }     fn pushResponse(self: *FetchBridge, handle: fetch_state.Handle, data: []const u8) !FetchOperationResult {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        const decision = fetch_state.decide(self.phase, .{ .push = handle });        if (decision.action == .stale) {            trace_stale("push", handle, decision.stale_reason.?);            return .stale;        }        const queued = self.response.items.len - self.response_offset;        if (data.len > max_fetch_response_bytes or queued > max_fetch_response_bytes - data.len) return .backpressure;        if (self.response_offset > 0) {            std.mem.copyForwards(u8, self.response.items[0..queued], self.response.items[self.response_offset..]);            self.response.items.len = queued;            self.response_offset = 0;        }        try self.response.appendSlice(std.heap.c_allocator, data);        self.wake.broadcast(io);        return .applied;    }     fn finishResponse(self: *FetchBridge, handle: fetch_state.Handle) FetchOperationResult {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        const decision = fetch_state.decide(self.phase, .{ .finish = handle });        if (decision.action == .stale) {            trace_stale("finish", handle, decision.stale_reason.?);            return .stale;        }        self.phase = decision.phase;        self.wake.broadcast(io);        return .applied;    }     fn failResponse(self: *FetchBridge, handle: fetch_state.Handle) FetchOperationResult {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        const decision = fetch_state.decide(self.phase, .{ .fail = handle });        if (decision.action == .stale) {            trace_stale("fail", handle, decision.stale_reason.?);            return .stale;        }        self.phase = decision.phase;        self.wake.broadcast(io);        return .applied;    }     fn is_active(self: *FetchBridge, handle: fetch_state.Handle) bool {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        return fetch_state.is_active(self.phase, handle);    }     fn abort(self: *FetchBridge) void {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        const decision = fetch_state.decide(self.phase, .cancel);        self.clearPendingRequest("abort");        self.phase = decision.phase;        self.wake.broadcast(io);    }     fn shutdown(self: *FetchBridge) void {        const io = io_mod.getIo();        self.mutex.lockUncancelable(io);        defer self.mutex.unlock(io);        const decision = fetch_state.decide(self.phase, .shutdown);        self.clearPendingRequest("shutdown");        self.phase = decision.phase;        self.wake.broadcast(io);    }     fn deinit(self: *FetchBridge) void {        self.request.deinit(std.heap.c_allocator);        self.response.deinit(std.heap.c_allocator);    }}; const FetchOperationResult = enum(u8) {    stale = 0,    applied = 1,    backpressure = 2,}; const Runtime = struct {    alloc: Allocator,    fetch: FetchBridge = .{},    stream_context: host_stream_provider.ProviderContext = undefined,    input: InputQueue = .{},    output: OutputQueue = .{},    credential: []u8,    model: ?[]u8,    home: []u8,    workspace_root: []u8,    gateway_chat_url: []u8,    thread: std.Thread,    exited: std.atomic.Value(bool) = .init(false),    exit_code: std.atomic.Value(u8) = .init(0),     fn readInput(context: ?*anyopaque, destination: []u8) usize {        const self: *Runtime = @ptrCast(@alignCast(context.?));        return self.input.read(self.alloc, destination);    }     fn writeOutput(context: ?*anyopaque, bytes: []const u8) !void {        const self: *Runtime = @ptrCast(@alignCast(context.?));        self.output.write(self.alloc, bytes) catch |err| {            debug_trace.logf("napi", "native output failed err={s}", .{@errorName(err)});            self.exit_code.store(1, .seq_cst);            return err;        };    }     fn run(self: *Runtime) void {        const provider = gateway_provider.Provider{            .agent_stream = host_stream_provider.provider(&self.stream_context),            .oauth_transport = oauth_transport.unavailable_provider,            .chat_url = builtin_gateway.provider.chat_url,            .cli_model_catalog = builtin_gateway.provider.cli_model_catalog,            .credits = builtin_gateway.provider.credits,            .generation_usage = generation_usage_provider.unavailable_provider,            .web_search = builtin_gateway.provider.web_search,            .model_catalog = builtin_gateway.provider.model_catalog,        };        acp_server.runWithTransport(            self.alloc,            .{                .default_model = builtin_gateway.default_model,                .default_agent_step_limit = 64,                .gateway_retry_count = 0,                .gateway_chat_url = self.gateway_chat_url,                .gateway_models_path = builtin_gateway.models_path,                .gateway_provider = provider,                .background_process_provider = background_process_provider.unavailable_provider,                .secret_store = host.unavailable_secret_store,                .prompt_policy = builtin_context.prompt_policy,                .ignored_list_entries = &.{},                .max_list_entries = 0,                .max_read_file_bytes = 0,                .max_read_file_lines = 0,                .max_read_file_line_len = 0,                .max_command_output_bytes = 0,                .max_tool_result_bytes = 64 * 1024,                .max_history_turns = 100,                .context_registry = .{ .default_provider = builtin_context.provider },                .mode_registry = builtin_modes.registry,                .credential_override = self.credential,                .model_override = self.model,                .home_override = self.home,                .workspace_root_override = self.workspace_root,                .allow_acp_mcp = false,                .allow_native_tools = false,            },            jsonrpc.Reader.initCallback(self, Runtime.readInput),            jsonrpc.Writer.initCallback(self, Runtime.writeOutput),        ) catch |err| {            debug_trace.logf("napi", "native runtime failed err={s}", .{@errorName(err)});            self.exit_code.store(1, .seq_cst);        };        self.exited.store(true, .seq_cst);    }     fn closeInput(self: *Runtime) void {        self.input.close();    }     fn abortHostEffects(self: *Runtime) void {        self.fetch.abort();    }     fn deinit(self: *Runtime) void {        self.closeInput();        self.fetch.shutdown();        self.thread.join();        self.fetch.deinit();        self.input.deinit(self.alloc);        self.output.deinit(self.alloc);        self.alloc.free(self.credential);        if (self.model) |model| self.alloc.free(model);        self.alloc.free(self.home);

Lines 1–500 of 961. Load more