const std = @import("std");const builtin = @import("builtin");const agent_runtime = @import("../agent/agent_runtime.zig");const permission_request = @import("../permissions/permission_request.zig");const permission_prompter = @import("../permissions/permission_prompter.zig");const session_permission_state = @import("../permissions/session_permission_state.zig");const command_admission = @import("../permissions/command_admission.zig");const permission_auto_classifier = @import("../permissions/auto_classifier.zig");const runtime_assistant_stream = @import("../agent/runtime/assistant_stream.zig");const runtime_config = @import("../agent/runtime/config.zig");const runtime_deps = @import("../agent/runtime/deps.zig");const runtime_lifecycle = @import("../agent/runtime/lifecycle.zig");const worker_runtime = @import("../agent/worker_runtime.zig");const debug_trace = @import("../shared/debug_trace.zig");const io_mod = @import("../shared/io.zig");const text_utils = @import("../shared/text_utils.zig");const session = @import("../session/session.zig");const session_child_store = @import("../session/session_child_store.zig");const session_codec = @import("../session/session_codec.zig");const session_event = @import("../session/session_event.zig");const session_store = @import("../session/session_store.zig");const permissions = @import("../permissions/permissions.zig");const tooling_tool_admission = @import("../tooling/tool_admission.zig");const shell_resolver = @import("../terminal/shell_resolver.zig");const background_runtime = @import("../background/background_runtime.zig");const tool_dispatch = @import("../tooling/tool_dispatch.zig");const types = @import("../shared/types.zig");const control_store = @import("control_store.zig");const authority_mod = @import("authority.zig");const approval_persistence = @import("approval_persistence.zig");const approval_registry_mod = @import("approval_registry.zig");const communication = @import("communication.zig");const communication_manager = @import("communication_manager.zig");const communication_store = @import("communication_store.zig");const domain = @import("domain.zig");const manager_mod = @import("manager.zig");const relationship_index = @import("relationship_index.zig");const work_events = @import("work_events.zig");const agent_test_support = if (builtin.is_test)    @import("../agent/runtime/tests/support.zig")else    struct {};const test_builtin_tools = if (builtin.is_test)    @import("../../builtins/tools.zig")else    struct {}; const Allocator = std.mem.Allocator; pub const TurnPreferences = struct {    model: []const u8,    effort: types.ReasoningEffort,}; /// Resolves child overrides without allocating. The returned model is borrowed/// from either the control record or ordinary session metadata.pub fn resolveTurnPreferences(    configuration: domain.Configuration,    persisted: session_codec.DurableSessionPreferences,) TurnPreferences {    return .{        .model = configuration.model orelse persisted.model,        .effort = configuration.effort orelse persisted.effort,    };} pub const WorkOutcome = enum {    completed,    failed,    awaiting_approval,    paused,}; pub const CompletionDecision = enum {    committed,    cancellation_won,    stale_work,}; pub const TransitionError = error{    OutOfMemory,    InvalidWorkState,    InvalidCancellationReason,}; /// Finds the next FIFO item. Interrupted or approval-blocked work is eligible/// only after an explicit retry/resume request.pub fn nextRunnableIndex(    queue: []const domain.QueuedMessage,    retry_interrupted: bool,) ?usize {    for (queue, 0..) |message, index| {        switch (message.status) {            .completed, .failed, .cancelled => continue,            .pending => return index,            .interrupted, .awaiting_approval => return if (retry_interrupted) index else null,            .running => return null,        }    }    return null;} /// Pure admission reduction over an owned record value.pub fn admitWork(    alloc: Allocator,    record: *control_store.Record,    index: usize,    timestamp_ms: i64,) TransitionError!void {    if (index >= record.queue.len) return error.InvalidWorkState;    const message = &record.queue[index];    const previous = message.status;    switch (message.status) {        .pending, .interrupted, .awaiting_approval => {},        .running, .completed, .failed, .cancelled => return error.InvalidWorkState,    }    if (message.cancellation_reason) |reason| {        if (message.status != .interrupted) return error.InvalidWorkState;        alloc.free(reason);        message.cancellation_reason = null;    }    message.status = .running;    record.state = .running;    record.updated_at_ms = timestamp_ms;    manager_mod.appendWorkRevision(alloc, record, &.{.{        .work_item_id = message.id,        .previous = previous,        .current = .running,    }}, timestamp_ms) catch |err| return switch (err) {        error.OutOfMemory => error.OutOfMemory,        error.GenerationExhausted => error.InvalidWorkState,    };} /// Pure completion reduction. A durable cancellation always wins over a late/// worker result and is never rewritten as success.pub fn finishWork(    alloc: Allocator,    record: *control_store.Record,    work_id: []const u8,    outcome: WorkOutcome,    timestamp_ms: i64,) TransitionError!CompletionDecision {    return finishWorkWithFailureReason(        alloc,        record,        work_id,        outcome,        null,        timestamp_ms,    );} fn finishWorkWithFailureReason(    alloc: Allocator,    record: *control_store.Record,    work_id: []const u8,    outcome: WorkOutcome,    failure_reason: ?[]const u8,    timestamp_ms: i64,) TransitionError!CompletionDecision {    if (outcome != .failed and failure_reason != null) {        return error.InvalidCancellationReason;    }    if (failure_reason) |reason| {        if (reason.len == 0 or reason.len > domain.max_cancellation_reason_bytes or            !text_utils.isModelSafeText(reason))        {            return error.InvalidCancellationReason;        }    }    const message = findWork(record.queue, work_id) orelse return .stale_work;    if (message.status == .cancelled) return .cancellation_won;    if (message.status != .running) return .stale_work;     if (outcome == .paused) {        const owned_reason = try alloc.dupe(u8, recovery_paused_reason);        if (message.cancellation_reason) |old| alloc.free(old);        message.cancellation_reason = owned_reason;    }     message.status = switch (outcome) {        .completed => .completed,        .failed => .failed,        .awaiting_approval => .awaiting_approval,        .paused => .interrupted,    };    const current = message.status;    record.updated_at_ms = timestamp_ms;    if (outcome == .awaiting_approval or outcome == .paused) {        record.state = if (outcome == .awaiting_approval) .awaiting_approval else .interrupted;        try appendSingleTransition(            alloc,            record,            message.id,            .running,            current,            message.cancellation_reason,            timestamp_ms,        );        return .committed;    }    if (remainingWorkState(record.queue)) |state| {        record.state = state;    } else if (record.mode == .one_off) {        record.state = if (outcome == .completed) .completed else .failed;    } else {        record.state = .idle;    }    try appendSingleTransition(        alloc,        record,        message.id,        .running,        current,        failure_reason,        timestamp_ms,    );    return .committed;} /// Pure cancellation reduction over an owned record. Allocation failures leave/// the caller-owned candidate disposable; the shell never publishes it.pub fn cancelWork(    alloc: Allocator,    record: *control_store.Record,    reason: []const u8,    timestamp_ms: i64,) TransitionError!usize {    if (reason.len == 0 or reason.len > domain.max_cancellation_reason_bytes or        !std.unicode.utf8ValidateSlice(reason) or std.mem.indexOfScalar(u8, reason, 0) != null)    {        return error.InvalidCancellationReason;    }    var transitions: std.ArrayList(manager_mod.WorkTransitionInput) = .empty;    defer transitions.deinit(alloc);    for (record.queue) |*message| {        switch (message.status) {            .pending, .running, .awaiting_approval => {},            .completed, .failed, .cancelled, .interrupted => continue,        }        const owned_reason = try alloc.dupe(u8, reason);        if (message.cancellation_reason) |old| alloc.free(old);        message.cancellation_reason = owned_reason;        const previous = message.status;        message.status = .cancelled;        try transitions.append(alloc, .{            .work_item_id = message.id,            .previous = previous,            .current = .cancelled,            .reason = message.cancellation_reason,        });    }    if (transitions.items.len == 0) return 0;    record.updated_at_ms = timestamp_ms;    record.state = if (record.mode == .one_off)        .cancelled    else        remainingWorkState(record.queue) orelse .idle;    manager_mod.appendWorkRevision(alloc, record, transitions.items, timestamp_ms) catch |err|        return switch (err) {            error.OutOfMemory => error.OutOfMemory,            error.GenerationExhausted => error.InvalidWorkState,        };    return transitions.items.len;} pub const RestartRecovery = struct {    interrupted: usize = 0,    completed: usize = 0,}; /// Pure restart reduction. No work becomes runnable and no model/tool callback/// is involved. The caller commits the candidate under the control lock.pub fn recoverAfterRestart(    alloc: Allocator,    record: *control_store.Record,    committed_work_id: ?[]const u8,    timestamp_ms: i64,) TransitionError!RestartRecovery {    const reason = "interrupted by process restart";    var result: RestartRecovery = .{};    var transitions: std.ArrayList(manager_mod.WorkTransitionInput) = .empty;    defer transitions.deinit(alloc);    for (record.queue) |*message| {        switch (message.status) {            .pending, .running, .awaiting_approval => {},            .completed, .failed, .cancelled, .interrupted => continue,        }        const previous = message.status;        if (committed_work_id) |work_id| {            if (previous != .pending and std.mem.eql(u8, message.id, work_id)) {                if (message.cancellation_reason) |old| alloc.free(old);                message.cancellation_reason = null;                message.status = .completed;                result.completed += 1;                try transitions.append(alloc, .{                    .work_item_id = message.id,                    .previous = previous,                    .current = .completed,                });                continue;            }        }        const owned_reason = try alloc.dupe(u8, reason);        if (message.cancellation_reason) |old| alloc.free(old);        message.cancellation_reason = owned_reason;        message.status = .interrupted;        result.interrupted += 1;        try transitions.append(alloc, .{            .work_item_id = message.id,            .previous = previous,            .current = .interrupted,            .reason = message.cancellation_reason,        });    }    if (transitions.items.len != 0) {        record.state = if (result.interrupted != 0)            .interrupted        else if (record.mode == .one_off)            .completed        else            .idle;        record.updated_at_ms = timestamp_ms;        manager_mod.appendWorkRevision(alloc, record, transitions.items, timestamp_ms) catch |err|            return switch (err) {                error.OutOfMemory => error.OutOfMemory,                error.GenerationExhausted => error.InvalidWorkState,            };    }    return result;} fn appendSingleTransition(    alloc: Allocator,    record: *control_store.Record,    work_id: []const u8,    previous: ?domain.QueueStatus,    current: domain.QueueStatus,    reason: ?[]const u8,    timestamp_ms: i64,) TransitionError!void {    manager_mod.appendWorkRevision(alloc, record, &.{.{        .work_item_id = work_id,        .previous = previous,        .current = current,        .reason = reason,    }}, timestamp_ms) catch |err| return switch (err) {        error.OutOfMemory => error.OutOfMemory,        error.GenerationExhausted => error.InvalidWorkState,    };} fn findWork(queue: []domain.QueuedMessage, id: []const u8) ?*domain.QueuedMessage {    for (queue) |*message| if (std.mem.eql(u8, message.id, id)) return message;    return null;} fn hasPending(queue: []const domain.QueuedMessage) bool {    for (queue) |message| if (message.status == .pending) return true;    return false;} fn remainingWorkState(queue: []const domain.QueuedMessage) ?domain.State {    for (queue) |message| switch (message.status) {        .running => return .running,        .awaiting_approval => return .awaiting_approval,        .interrupted => return .interrupted,        .pending => return .queued,        .completed, .failed, .cancelled => {},    };    return null;} pub const CaptureRequest = struct {    child_id: []const u8,    parent_id: []const u8,    source_id: []const u8,    configuration: domain.Configuration,    preferences: TurnPreferences,}; pub const RunOutcome = enum {    completed,    awaiting_approval,    paused,}; pub const ServiceError = error{    OutOfMemory,    AdmissionFailed,    ProviderFailed,    Cancelled,}; pub const Services = struct {    context: ?*anyopaque = null,    capture_fn: *const fn (?*anyopaque, Allocator, CaptureRequest) ServiceError!domain.AdmissionSnapshot,    run_fn: *const fn (?*anyopaque, *TurnContext, domain.QueuedMessage, domain.AdmissionSnapshot, *std.atomic.Value(bool)) ServiceError!RunOutcome,     fn capture(self: Services, alloc: Allocator, request: CaptureRequest) ServiceError!domain.AdmissionSnapshot {        return self.capture_fn(self.context, alloc, request);    }     fn run(        self: Services,        turn: *TurnContext,        message: domain.QueuedMessage,        admission: domain.AdmissionSnapshot,        cancel: *std.atomic.Value(bool),    ) ServiceError!RunOutcome {        return self.run_fn(self.context, turn, message, admission, cancel);    }}; pub const CommitError = error{    OutOfMemory,    InvalidWorkId,    TurnAlreadyCommitted,    SessionCommitFailed,}; pub const max_live_presentation_bytes: usize = 64 * 1024;pub const max_live_tool_activity: usize = 32;pub const max_live_tool_name_bytes: usize = 128;pub const max_live_presentation_events: usize = 256;pub const max_live_presentation_event_bytes: usize = 128 * 1024;const recovery_paused_reason = "model response recovery paused; resume this subagent to continue"; pub const LiveToolActivity = struct {    tool_name: []u8,    phase: runtime_deps.ToolActivityPhase,     pub fn deinit(self: *LiveToolActivity, alloc: Allocator) void {        alloc.free(self.tool_name);        self.* = undefined;    }     fn clone(self: LiveToolActivity, alloc: Allocator) !LiveToolActivity {        return .{            .tool_name = try alloc.dupe(u8, self.tool_name),            .phase = self.phase,        };    }}; /// Allocator-owned copy of presentation that has not entered canonical/// history. It exists only while the corresponding execution slot is live.pub const LivePresentation = struct {    work_id: []u8,    revision: u64,    text: []u8,    text_truncated: bool,    tools: []LiveToolActivity,    tools_truncated: bool,    events: []worker_runtime.WorkerEvent,    events_truncated: bool,     pub fn deinit(self: *LivePresentation, alloc: Allocator) void {        alloc.free(self.work_id);        alloc.free(self.text);        for (self.tools) |*tool| tool.deinit(alloc);        alloc.free(self.tools);        for (self.events) |event| worker_runtime.freeWorkerEvent(alloc, event);        alloc.free(self.events);        self.* = undefined;    }}; const LivePresentationSink = struct {    context: *anyopaque,    append_text_fn: *const fn (*anyopaque, []const u8) void,    append_tool_fn: *const fn (        *anyopaque,        []const u8,        runtime_deps.ToolActivityPhase,    ) void,    append_event_fn: *const fn (        *anyopaque,        worker_runtime.WorkerEvent,    ) void,     fn appendText(self: LivePresentationSink, text: []const u8) void {        self.append_text_fn(self.context, text);    }     fn appendTool(        self: LivePresentationSink,        tool_name: []const u8,        phase: runtime_deps.ToolActivityPhase,    ) void {        self.append_tool_fn(self.context, tool_name, phase);    }     fn appendEvent(        self: LivePresentationSink,        event: worker_runtime.WorkerEvent,    ) void {        self.append_event_fn(self.context, event);    }

Lines 1–500 of 8630. Load more