const std = @import("std");const builtin = @import("builtin");const command_effect = @import("../shell_command/command_effect.zig");const command_contract = @import("../execution/command_contract.zig");const sandbox = @import("sandbox.zig");const debug_trace = @import("../shared/debug_trace.zig");const io_mod = @import("../shared/io.zig");const types = @import("../shared/types.zig"); pub const direct_output_limit_bytes: usize = 65_536;const direct_output_read_chunk_bytes: usize = 4096; const DirectOutputProjector = struct { utf8_pending: [3]u8 = undefined, utf8_pending_len: u2 = 0, escaped_controls: usize = 0, escaped_invalid: usize = 0, fn push( self: *DirectOutputProjector, alloc: std.mem.Allocator, raw: []const u8, projected: *std.ArrayList(u8), ) !void { var combined: [direct_output_read_chunk_bytes + 3]u8 = undefined; const pending_len: usize = self.utf8_pending_len; @memcpy(combined[0..pending_len], self.utf8_pending[0..pending_len]); @memcpy(combined[pending_len .. pending_len + raw.len], raw); const bytes = combined[0 .. pending_len + raw.len]; self.utf8_pending_len = 0; var index: usize = 0; while (index < bytes.len) { const byte = bytes[index]; if (byte < 0x80) { if (byte == '\n') { try projected.append(alloc, byte); } else if (byte < 0x20 or byte == 0x7f) { self.escaped_controls += 1; try appendByteEscape(alloc, projected, byte); } else { try projected.append(alloc, byte); } index += 1; continue; } const sequence_len: usize = std.unicode.utf8ByteSequenceLength(byte) catch { self.escaped_invalid += 1; try appendByteEscape(alloc, projected, byte); index += 1; continue; }; if (bytes.len - index < sequence_len) { const remainder = bytes[index..]; @memcpy(self.utf8_pending[0..remainder.len], remainder); self.utf8_pending_len = @intCast(remainder.len); break; } const sequence = bytes[index .. index + sequence_len]; const scalar = std.unicode.utf8Decode(sequence) catch { self.escaped_invalid += 1; try appendByteEscape(alloc, projected, byte); index += 1; continue; }; if (scalar >= 0x80 and scalar <= 0x9f) { self.escaped_controls += 1; try appendC1Escape(alloc, projected, @intCast(scalar)); } else { try projected.appendSlice(alloc, sequence); } index += sequence_len; } } fn finish( self: *DirectOutputProjector, alloc: std.mem.Allocator, projected: *std.ArrayList(u8), ) !void { for (self.utf8_pending[0..self.utf8_pending_len]) |byte| { self.escaped_invalid += 1; try appendByteEscape(alloc, projected, byte); } self.utf8_pending_len = 0; }}; fn appendByteEscape( alloc: std.mem.Allocator, projected: *std.ArrayList(u8), byte: u8,) !void { const digits = "0123456789abcdef"; try projected.appendSlice(alloc, &.{ '\\', 'x', digits[byte >> 4], digits[byte & 0x0f], });} fn appendC1Escape( alloc: std.mem.Allocator, projected: *std.ArrayList(u8), byte: u8,) !void { const digits = "0123456789abcdef"; try projected.appendSlice(alloc, &.{ '\\', 'u', '{', '0', '0', digits[byte >> 4], digits[byte & 0x0f], '}', });} const DirectOutputBudget = struct { limit: usize, admitted: usize = 0, tripped: bool = false, lock: std.Io.Mutex = .init, const Charge = struct { admit_chunk: bool, trip_owner: bool, }; fn charge(self: *DirectOutputBudget, observed_len: usize) Charge { const io = io_mod.getIo(); self.lock.lockUncancelable(io); defer self.lock.unlock(io); if (self.tripped) { return .{ .admit_chunk = false, .trip_owner = false }; } if (observed_len <= self.limit - self.admitted) { self.admitted += observed_len; return .{ .admit_chunk = true, .trip_owner = false }; } self.tripped = true; return .{ .admit_chunk = false, .trip_owner = true }; } fn remaining(self: *DirectOutputBudget) usize { const io = io_mod.getIo(); self.lock.lockUncancelable(io); defer self.lock.unlock(io); return self.limit - self.admitted; }}; pub fn executeDirectReadOnly( cfg: sandbox.Config, alloc: std.mem.Allocator, plan: command_effect.DirectReadOnlyPlan,) !command_contract.RunCommandResult { return executeDirectReadOnlyWithLimit(cfg, alloc, plan, direct_output_limit_bytes);} fn executeDirectReadOnlyWithLimit( cfg: sandbox.Config, alloc: std.mem.Allocator, plan: command_effect.DirectReadOnlyPlan, output_limit: usize,) !command_contract.RunCommandResult { return executeDirectReadOnlyWithLimitAndTestControls(cfg, alloc, plan, output_limit, .{});} const DirectExecutionTestControls = struct { context: ?*anyopaque = null, after_spawn: ?*const fn (*anyopaque, usize, *const std.process.Child) void = null,}; fn executeDirectReadOnlyWithLimitAndTestControls( cfg: sandbox.Config, alloc: std.mem.Allocator, plan: command_effect.DirectReadOnlyPlan, output_limit: usize, test_controls: DirectExecutionTestControls,) !command_contract.RunCommandResult { if (plan.stages.len == 0) return error.InvalidDirectPlan; if (plan.stages.len > command_effect.max_direct_pipeline_stages) { return error.InvalidDirectPlan; } var execution_cfg = cfg; if (execution_cfg.timeout_ms != null and execution_cfg.timeout_started_ms == null) { execution_cfg.timeout_started_ms = io_mod.milliTimestamp(); } try checkControl(execution_cfg); const backend = sandbox.resolveBackend(execution_cfg.backend); if (backend != .none and backend != .macos) { return error.UnsupportedDirectBackend; } if (builtin.os.tag != .macos and builtin.os.tag != .linux) { return error.UnsupportedDirectPlatform; } if (backend == .macos and builtin.os.tag != .macos) { return error.UnsupportedDirectPlatform; } var scratch_state = std.heap.ArenaAllocator.init(alloc); defer scratch_state.deinit(); const scratch = scratch_state.allocator(); var children = try scratch.alloc(std.process.Child, plan.stages.len); var child_count: usize = 0; var group_id: ?std.posix.pid_t = null; var pre_worker_cleanup_pending = true; errdefer if (pre_worker_cleanup_pending) { cleanupChildren(children[0..child_count], group_id); }; const started_ms = io_mod.milliTimestamp(); while (child_count < plan.stages.len) : (child_count += 1) { try checkControl(execution_cfg); const stage = plan.stages[child_count]; var environment = try environmentForProfile(scratch, stage.environment_profile); defer environment.deinit(); const argv = if (backend == .macos) try sandbox.buildDirectMacOSArgv(scratch, execution_cfg, stage.argv) else stage.argv; const child = std.process.spawn(io_mod.getIo(), .{ .argv = argv, .cwd = .{ .path = plan.cwd }, .environ_map = &environment, .stdin = if (child_count == 0) .ignore else .pipe, .stdout = .pipe, .stderr = .pipe, .pgid = if (builtin.os.tag != .windows and builtin.os.tag != .wasi) (if (child_count == 0) 0 else group_id) else null, }) catch |err| { return switch (err) { error.FileNotFound => error.DirectExecutableUnavailable, else => err, }; }; children[child_count] = child; if (child_count == 0 and builtin.os.tag != .windows and builtin.os.tag != .wasi) { group_id = child.id; } if (test_controls.after_spawn) |after_spawn| { after_spawn(test_controls.context.?, child_count, &children[child_count]); } } var output = DirectOutput.init(alloc, execution_cfg); defer output.deinit(); var shared: SharedExecution = .{ .cfg = execution_cfg, .budget = .{ .limit = output_limit }, .output = &output, }; const worker_count = plan.stages.len * 2; const workers = try scratch.alloc(OutputWorker, worker_count); var initialized_workers: usize = 0; for (0..plan.stages.len - 1) |index| { workers[initialized_workers] = .{ .shared = &shared, .source = children[index].stdout.?, .destination = children[index + 1].stdin.?, .kind = .relay, }; children[index].stdout = null; children[index + 1].stdin = null; initialized_workers += 1; } workers[initialized_workers] = .{ .shared = &shared, .source = children[plan.stages.len - 1].stdout.?, .kind = .{ .projected = .stdout }, }; children[plan.stages.len - 1].stdout = null; initialized_workers += 1; for (children[0..child_count]) |*child| { workers[initialized_workers] = .{ .shared = &shared, .source = child.stderr.?, .kind = .{ .projected = .stderr }, }; child.stderr = null; initialized_workers += 1; } std.debug.assert(initialized_workers == worker_count); pre_worker_cleanup_pending = false; var started_workers: usize = 0; while (started_workers < workers.len) : (started_workers += 1) { workers[started_workers].thread = std.Thread.spawn( .{}, OutputWorker.run, .{&workers[started_workers]}, ) catch |err| { shared.commit(.output_failure, err); signalGroup(group_id, true); for (workers[started_workers..]) |*worker| worker.closeUnstarted(); for (workers[0..started_workers]) |*worker| worker.thread.?.join(); waitChildren(children[0..child_count]); const failure = shared.failure() orelse err; debug_trace.logf( "core", "direct command worker start failed route=direct_read_only cause={s} limit={d} admitted_raw_bytes={d} remaining_raw_bytes={d} children_reaped={d} workers_joined={d} artifact=false", .{ @tagName(shared.cause.?), shared.budget.limit, shared.budget.admitted, shared.budget.remaining(), child_count, started_workers, }, ); return failure; }; } var termination_started_ms: ?i64 = null; var force_kill_sent = false; while (!allWorkersDone(workers)) { if (!shared.isStopping()) { if (execution_cfg.cancel_flag) |flag| { if (flag.load(.seq_cst)) shared.commit(.cancelled, error.Cancelled); } if (!shared.isStopping() and deadlineExpired(execution_cfg)) { shared.commit(.timed_out, error.TimeoutExpired); } } if (shared.isStopping()) { const now = io_mod.milliTimestamp(); if (termination_started_ms == null) { signalGroup(group_id, false); termination_started_ms = now; } else if (!force_kill_sent and now - termination_started_ms.? >= 800) { signalGroup(group_id, true); force_kill_sent = true; } } io_mod.sleep(5 * std.time.ns_per_ms); } for (workers) |*worker| worker.thread.?.join(); var final_term: std.process.Child.Term = .{ .unknown = 0 }; for (children[0..child_count], 0..) |*child, index| { const term = child.wait(io_mod.getIo()) catch |err| { shared.commit(.output_failure, err); continue; }; if (index + 1 == child_count) final_term = term; } if (shared.failure()) |failure| { if (shared.cause == .timed_out) { output.flushCallbacks() catch |err| debug_trace.logf( "core", "direct command timeout callback flush failed err={s}", .{@errorName(err)}, ); } debug_trace.logf( "core", "direct command terminated route=direct_read_only cause={s} limit={d} admitted_raw_bytes={d} remaining_raw_bytes={d} projected_bytes={d} escaped_controls={d} escaped_invalid={d} children_reaped={d} workers_joined={d} artifact=false", .{ @tagName(shared.cause.?), shared.budget.limit, shared.budget.admitted, shared.budget.remaining(), output.stdout.items.len + output.stderr.items.len, output.escaped_controls, output.escaped_invalid, child_count, workers.len, }, ); return failure; } try output.flushCallbacks(); const duration_ms = elapsedMs(started_ms, io_mod.milliTimestamp()); debug_trace.logf( "core", "direct command completed route=direct_read_only limit={d} admitted_raw_bytes={d} remaining_raw_bytes={d} projected_bytes={d} escaped_controls={d} escaped_invalid={d} children_reaped={d} workers_joined={d} artifact=false", .{ shared.budget.limit, shared.budget.admitted, shared.budget.remaining(), output.stdout.items.len + output.stderr.items.len, output.escaped_controls, output.escaped_invalid, child_count, workers.len, }, ); return formatDirectResult( alloc, plan, final_term, output.stdout.items, output.stderr.items, output.stdout_bytes, output.stderr_bytes, duration_ms, );} fn environmentForProfile( alloc: std.mem.Allocator, profile: command_effect.EnvironmentProfile,) !std.process.Environ.Map { var environment = std.process.Environ.Map.init(alloc); errdefer environment.deinit(); switch (profile) { .basic_read_only, .git_read_only => { try environment.put("PATH", "/usr/bin:/bin"); try environment.put("LC_ALL", "C"); try environment.put("LANG", "C"); }, } if (profile == .git_read_only) { try environment.put("GIT_CONFIG_NOSYSTEM", "1"); try environment.put("GIT_CONFIG_GLOBAL", "/dev/null"); try environment.put("GIT_OPTIONAL_LOCKS", "0"); try environment.put("GIT_TERMINAL_PROMPT", "0"); try environment.put("GIT_PAGER", "cat"); try environment.put("PAGER", "cat"); } return environment;} const DirectTerminationCause = enum { cancelled, timed_out, output_limit, output_failure,}; const SharedExecution = struct { cfg: sandbox.Config, budget: DirectOutputBudget, output: *DirectOutput, lock: std.Io.Mutex = .init, stopping: std.atomic.Value(bool) = .init(false), cause: ?DirectTerminationCause = null, failure_value: ?anyerror = null, fn commit( self: *SharedExecution, reported: DirectTerminationCause, failure_value: anyerror, ) void { const io = io_mod.getIo(); self.lock.lockUncancelable(io); defer self.lock.unlock(io); if (self.cause != null) return; if (self.cfg.cancel_flag) |flag| { if (flag.load(.seq_cst)) { self.cause = .cancelled; self.failure_value = error.Cancelled; self.stopping.store(true, .release); return; } } if (deadlineExpired(self.cfg)) { self.cause = .timed_out; self.failure_value = error.TimeoutExpired; self.stopping.store(true, .release); return; } self.cause = reported; self.failure_value = failure_value; self.stopping.store(true, .release); } fn isStopping(self: *SharedExecution) bool { return self.stopping.load(.acquire); } fn failure(self: *SharedExecution) ?anyerror { const io = io_mod.getIo(); self.lock.lockUncancelable(io); defer self.lock.unlock(io); return self.failure_value; }}; Lines 1–500 of 1876. Load more
const std = @import("std"); const builtin = @import("builtin"); const command_effect = @import("../shell_command/command_effect.zig"); const command_contract = @import("../execution/command_contract.zig"); const sandbox = @import("sandbox.zig"); const debug_trace = @import("../shared/debug_trace.zig"); const io_mod = @import("../shared/io.zig"); const types = @import("../shared/types.zig"); pub const direct_output_limit_bytes: usize = 65_536; const direct_output_read_chunk_bytes: usize = 4096; const DirectOutputProjector = struct { utf8_pending: [3]u8 = undefined, utf8_pending_len: u2 = 0, escaped_controls: usize = 0, escaped_invalid: usize = 0, fn push( self: *DirectOutputProjector, alloc: std.mem.Allocator, raw: []const u8, projected: *std.ArrayList(u8), ) !void { var combined: [direct_output_read_chunk_bytes + 3]u8 = undefined; const pending_len: usize = self.utf8_pending_len; @memcpy(combined[0..pending_len], self.utf8_pending[0..pending_len]); @memcpy(combined[pending_len .. pending_len + raw.len], raw); const bytes = combined[0 .. pending_len + raw.len]; self.utf8_pending_len = 0; var index: usize = 0; while (index < bytes.len) { const byte = bytes[index]; if (byte < 0x80) { if (byte == '\n') { try projected.append(alloc, byte); } else if (byte < 0x20 or byte == 0x7f) { self.escaped_controls += 1; try appendByteEscape(alloc, projected, byte); } else { try projected.append(alloc, byte); } index += 1; continue; } const sequence_len: usize = std.unicode.utf8ByteSequenceLength(byte) catch { self.escaped_invalid += 1; try appendByteEscape(alloc, projected, byte); index += 1; continue; }; if (bytes.len - index < sequence_len) { const remainder = bytes[index..]; @memcpy(self.utf8_pending[0..remainder.len], remainder); self.utf8_pending_len = @intCast(remainder.len); break; } const sequence = bytes[index .. index + sequence_len]; const scalar = std.unicode.utf8Decode(sequence) catch { self.escaped_invalid += 1; try appendByteEscape(alloc, projected, byte); index += 1; continue; }; if (scalar >= 0x80 and scalar <= 0x9f) { self.escaped_controls += 1; try appendC1Escape(alloc, projected, @intCast(scalar)); } else { try projected.appendSlice(alloc, sequence); } index += sequence_len; } } fn finish( self: *DirectOutputProjector, alloc: std.mem.Allocator, projected: *std.ArrayList(u8), ) !void { for (self.utf8_pending[0..self.utf8_pending_len]) |byte| { self.escaped_invalid += 1; try appendByteEscape(alloc, projected, byte); } self.utf8_pending_len = 0; } }; fn appendByteEscape( alloc: std.mem.Allocator, projected: *std.ArrayList(u8), byte: u8, ) !void { const digits = "0123456789abcdef"; try projected.appendSlice(alloc, &.{ '\\', 'x', digits[byte >> 4], digits[byte & 0x0f], }); } fn appendC1Escape( alloc: std.mem.Allocator, projected: *std.ArrayList(u8), byte: u8, ) !void { const digits = "0123456789abcdef"; try projected.appendSlice(alloc, &.{ '\\', 'u', '{', '0', '0', digits[byte >> 4], digits[byte & 0x0f], '}', }); } const DirectOutputBudget = struct { limit: usize, admitted: usize = 0, tripped: bool = false, lock: std.Io.Mutex = .init, const Charge = struct { admit_chunk: bool, trip_owner: bool, }; fn charge(self: *DirectOutputBudget, observed_len: usize) Charge { const io = io_mod.getIo(); self.lock.lockUncancelable(io); defer self.lock.unlock(io); if (self.tripped) { return .{ .admit_chunk = false, .trip_owner = false }; } if (observed_len <= self.limit - self.admitted) { self.admitted += observed_len; return .{ .admit_chunk = true, .trip_owner = false }; } self.tripped = true; return .{ .admit_chunk = false, .trip_owner = true }; } fn remaining(self: *DirectOutputBudget) usize { const io = io_mod.getIo(); self.lock.lockUncancelable(io); defer self.lock.unlock(io); return self.limit - self.admitted; } }; pub fn executeDirectReadOnly( cfg: sandbox.Config, alloc: std.mem.Allocator, plan: command_effect.DirectReadOnlyPlan, ) !command_contract.RunCommandResult { return executeDirectReadOnlyWithLimit(cfg, alloc, plan, direct_output_limit_bytes); } fn executeDirectReadOnlyWithLimit( cfg: sandbox.Config, alloc: std.mem.Allocator, plan: command_effect.DirectReadOnlyPlan, output_limit: usize, ) !command_contract.RunCommandResult { return executeDirectReadOnlyWithLimitAndTestControls(cfg, alloc, plan, output_limit, .{}); } const DirectExecutionTestControls = struct { context: ?*anyopaque = null, after_spawn: ?*const fn (*anyopaque, usize, *const std.process.Child) void = null, }; fn executeDirectReadOnlyWithLimitAndTestControls( cfg: sandbox.Config, alloc: std.mem.Allocator, plan: command_effect.DirectReadOnlyPlan, output_limit: usize, test_controls: DirectExecutionTestControls, ) !command_contract.RunCommandResult { if (plan.stages.len == 0) return error.InvalidDirectPlan; if (plan.stages.len > command_effect.max_direct_pipeline_stages) { return error.InvalidDirectPlan; } var execution_cfg = cfg; if (execution_cfg.timeout_ms != null and execution_cfg.timeout_started_ms == null) { execution_cfg.timeout_started_ms = io_mod.milliTimestamp(); } try checkControl(execution_cfg); const backend = sandbox.resolveBackend(execution_cfg.backend); if (backend != .none and backend != .macos) { return error.UnsupportedDirectBackend; } if (builtin.os.tag != .macos and builtin.os.tag != .linux) { return error.UnsupportedDirectPlatform; } if (backend == .macos and builtin.os.tag != .macos) { return error.UnsupportedDirectPlatform; } var scratch_state = std.heap.ArenaAllocator.init(alloc); defer scratch_state.deinit(); const scratch = scratch_state.allocator(); var children = try scratch.alloc(std.process.Child, plan.stages.len); var child_count: usize = 0; var group_id: ?std.posix.pid_t = null; var pre_worker_cleanup_pending = true; errdefer if (pre_worker_cleanup_pending) { cleanupChildren(children[0..child_count], group_id); }; const started_ms = io_mod.milliTimestamp(); while (child_count < plan.stages.len) : (child_count += 1) { try checkControl(execution_cfg); const stage = plan.stages[child_count]; var environment = try environmentForProfile(scratch, stage.environment_profile); defer environment.deinit(); const argv = if (backend == .macos) try sandbox.buildDirectMacOSArgv(scratch, execution_cfg, stage.argv) else stage.argv; const child = std.process.spawn(io_mod.getIo(), .{ .argv = argv, .cwd = .{ .path = plan.cwd }, .environ_map = &environment, .stdin = if (child_count == 0) .ignore else .pipe, .stdout = .pipe, .stderr = .pipe, .pgid = if (builtin.os.tag != .windows and builtin.os.tag != .wasi) (if (child_count == 0) 0 else group_id) else null, }) catch |err| { return switch (err) { error.FileNotFound => error.DirectExecutableUnavailable, else => err, }; }; children[child_count] = child; if (child_count == 0 and builtin.os.tag != .windows and builtin.os.tag != .wasi) { group_id = child.id; } if (test_controls.after_spawn) |after_spawn| { after_spawn(test_controls.context.?, child_count, &children[child_count]); } } var output = DirectOutput.init(alloc, execution_cfg); defer output.deinit(); var shared: SharedExecution = .{ .cfg = execution_cfg, .budget = .{ .limit = output_limit }, .output = &output, }; const worker_count = plan.stages.len * 2; const workers = try scratch.alloc(OutputWorker, worker_count); var initialized_workers: usize = 0; for (0..plan.stages.len - 1) |index| { workers[initialized_workers] = .{ .shared = &shared, .source = children[index].stdout.?, .destination = children[index + 1].stdin.?, .kind = .relay, }; children[index].stdout = null; children[index + 1].stdin = null; initialized_workers += 1; } workers[initialized_workers] = .{ .shared = &shared, .source = children[plan.stages.len - 1].stdout.?, .kind = .{ .projected = .stdout }, }; children[plan.stages.len - 1].stdout = null; initialized_workers += 1; for (children[0..child_count]) |*child| { workers[initialized_workers] = .{ .shared = &shared, .source = child.stderr.?, .kind = .{ .projected = .stderr }, }; child.stderr = null; initialized_workers += 1; } std.debug.assert(initialized_workers == worker_count); pre_worker_cleanup_pending = false; var started_workers: usize = 0; while (started_workers < workers.len) : (started_workers += 1) { workers[started_workers].thread = std.Thread.spawn( .{}, OutputWorker.run, .{&workers[started_workers]}, ) catch |err| { shared.commit(.output_failure, err); signalGroup(group_id, true); for (workers[started_workers..]) |*worker| worker.closeUnstarted(); for (workers[0..started_workers]) |*worker| worker.thread.?.join(); waitChildren(children[0..child_count]); const failure = shared.failure() orelse err; debug_trace.logf( "core", "direct command worker start failed route=direct_read_only cause={s} limit={d} admitted_raw_bytes={d} remaining_raw_bytes={d} children_reaped={d} workers_joined={d} artifact=false", .{ @tagName(shared.cause.?), shared.budget.limit, shared.budget.admitted, shared.budget.remaining(), child_count, started_workers, }, ); return failure; }; } var termination_started_ms: ?i64 = null; var force_kill_sent = false; while (!allWorkersDone(workers)) { if (!shared.isStopping()) { if (execution_cfg.cancel_flag) |flag| { if (flag.load(.seq_cst)) shared.commit(.cancelled, error.Cancelled); } if (!shared.isStopping() and deadlineExpired(execution_cfg)) { shared.commit(.timed_out, error.TimeoutExpired); } } if (shared.isStopping()) { const now = io_mod.milliTimestamp(); if (termination_started_ms == null) { signalGroup(group_id, false); termination_started_ms = now; } else if (!force_kill_sent and now - termination_started_ms.? >= 800) { signalGroup(group_id, true); force_kill_sent = true; } } io_mod.sleep(5 * std.time.ns_per_ms); } for (workers) |*worker| worker.thread.?.join(); var final_term: std.process.Child.Term = .{ .unknown = 0 }; for (children[0..child_count], 0..) |*child, index| { const term = child.wait(io_mod.getIo()) catch |err| { shared.commit(.output_failure, err); continue; }; if (index + 1 == child_count) final_term = term; } if (shared.failure()) |failure| { if (shared.cause == .timed_out) { output.flushCallbacks() catch |err| debug_trace.logf( "core", "direct command timeout callback flush failed err={s}", .{@errorName(err)}, ); } debug_trace.logf( "core", "direct command terminated route=direct_read_only cause={s} limit={d} admitted_raw_bytes={d} remaining_raw_bytes={d} projected_bytes={d} escaped_controls={d} escaped_invalid={d} children_reaped={d} workers_joined={d} artifact=false", .{ @tagName(shared.cause.?), shared.budget.limit, shared.budget.admitted, shared.budget.remaining(), output.stdout.items.len + output.stderr.items.len, output.escaped_controls, output.escaped_invalid, child_count, workers.len, }, ); return failure; } try output.flushCallbacks(); const duration_ms = elapsedMs(started_ms, io_mod.milliTimestamp()); debug_trace.logf( "core", "direct command completed route=direct_read_only limit={d} admitted_raw_bytes={d} remaining_raw_bytes={d} projected_bytes={d} escaped_controls={d} escaped_invalid={d} children_reaped={d} workers_joined={d} artifact=false", .{ shared.budget.limit, shared.budget.admitted, shared.budget.remaining(), output.stdout.items.len + output.stderr.items.len, output.escaped_controls, output.escaped_invalid, child_count, workers.len, }, ); return formatDirectResult( alloc, plan, final_term, output.stdout.items, output.stderr.items, output.stdout_bytes, output.stderr_bytes, duration_ms, ); } fn environmentForProfile( alloc: std.mem.Allocator, profile: command_effect.EnvironmentProfile, ) !std.process.Environ.Map { var environment = std.process.Environ.Map.init(alloc); errdefer environment.deinit(); switch (profile) { .basic_read_only, .git_read_only => { try environment.put("PATH", "/usr/bin:/bin"); try environment.put("LC_ALL", "C"); try environment.put("LANG", "C"); }, } if (profile == .git_read_only) { try environment.put("GIT_CONFIG_NOSYSTEM", "1"); try environment.put("GIT_CONFIG_GLOBAL", "/dev/null"); try environment.put("GIT_OPTIONAL_LOCKS", "0"); try environment.put("GIT_TERMINAL_PROMPT", "0"); try environment.put("GIT_PAGER", "cat"); try environment.put("PAGER", "cat"); } return environment; } const DirectTerminationCause = enum { cancelled, timed_out, output_limit, output_failure, }; const SharedExecution = struct { cfg: sandbox.Config, budget: DirectOutputBudget, output: *DirectOutput, lock: std.Io.Mutex = .init, stopping: std.atomic.Value(bool) = .init(false), cause: ?DirectTerminationCause = null, failure_value: ?anyerror = null, fn commit( self: *SharedExecution, reported: DirectTerminationCause, failure_value: anyerror, ) void { const io = io_mod.getIo(); self.lock.lockUncancelable(io); defer self.lock.unlock(io); if (self.cause != null) return; if (self.cfg.cancel_flag) |flag| { if (flag.load(.seq_cst)) { self.cause = .cancelled; self.failure_value = error.Cancelled; self.stopping.store(true, .release); return; } } if (deadlineExpired(self.cfg)) { self.cause = .timed_out; self.failure_value = error.TimeoutExpired; self.stopping.store(true, .release); return; } self.cause = reported; self.failure_value = failure_value; self.stopping.store(true, .release); } fn isStopping(self: *SharedExecution) bool { return self.stopping.load(.acquire); } fn failure(self: *SharedExecution) ?anyerror { const io = io_mod.getIo(); self.lock.lockUncancelable(io); defer self.lock.unlock(io); return self.failure_value; } };