const std = @import("std");
const credentials = @import("../auth/credentials.zig");
const secret = @import("../auth/secret.zig");
const io_mod = @import("../shared/io.zig");
const debug_trace = @import("../shared/debug_trace.zig");
const diagnostics = @import("../workspace/diagnostics.zig");
const diff_mod = @import("../output/diff.zig");
const file_mutation_contract = @import("../tooling/file_mutation_contract.zig");
const image_attachments = @import("../images/image_attachments.zig");
const context_contract = @import("../workspace/context_contract.zig");
const auto_classifier_context = @import("../permissions/auto_classifier_context.zig");
const permission_request = @import("../permissions/permission_request.zig");
const notification_contract = @import("../notifications/notification_contract.zig");
const session_runtime = @import("../session/session.zig");
const session_codec = @import("../session/session_codec.zig");
const skill_contract = @import("../skills/skill_contract.zig");
const tool_result_limits = @import("../tooling/tool_result_limits.zig");
const command_output_content = @import("../tooling/command_output_content.zig");
const types = @import("../shared/types.zig");
const model_provider = @import("../config/model_provider.zig");
const assistant_presentation = @import("assistant_presentation.zig");
const compaction_activity = @import("../output/compaction_activity.zig");

pub const AgentTurnSettings = struct {
    max_tool_result_bytes: usize = tool_result_limits.default_max_tool_result_bytes,
    first_call_tool_choice: types.ToolChoice = .auto,
    fast_mode: bool = false,
    effort: types.ReasoningEffort = .auto,
};

pub const SkillBinding = struct {
    name: []u8,
    path: []u8,
};

pub const SkillDisplaySpan = struct {
    raw_start: usize,
    raw_end: usize,
    name: []u8,
    path: []u8,
    display_source: ?skill_contract.SkillSource = null,
    owns_trailing_separator: bool = false,
};

pub const BeginPromptWithSkillBindings = struct {
    prompt: types.UserTurn,
    skill_bindings: []SkillBinding = &.{},
    skill_display_spans: []SkillDisplaySpan = &.{},
};

pub const PromptDelivery = union(enum) {
    ordinary,
    active_turn: u64,
    continuation,

    pub fn activeTurnId(self: PromptDelivery) ?u64 {
        return switch (self) {
            .active_turn => |turn_id| turn_id,
            .ordinary, .continuation => null,
        };
    }

    pub fn isSteering(self: PromptDelivery) bool {
        return switch (self) {
            .ordinary => false,
            .active_turn, .continuation => true,
        };
    }

    pub fn isContinuation(self: PromptDelivery) bool {
        return self == .continuation;
    }
};

pub const SteeringBoundaryKind = enum {
    model,
    cancelled,
    finalizing,
};

const SteeringDelivery = types.SteeringDelivery;

/// Borrowed by the queue. Its owner outlives worker teardown.
pub const SteeringReceipt = struct {
    operation_id: []const u8 = "",
    state: std.atomic.Value(SteeringDelivery) = .init(.queued),
};

pub const SteeringBoundaryResult = union(enum) {
    none,
    continue_turn: [][]u8,
    handoff,
    interrupt,
};

pub const QueuedPrompt = struct {
    turn_id: u64 = 0,
    steering_receipt: ?*SteeringReceipt = null,
    /// Runtime-derived delivery state. The worker retains ownership in every state.
    delivery: PromptDelivery = .ordinary,
    prompt: []u8,
    images: []types.ImageAttachment,
    authorized_image_catalog: []types.ImageAttachment = &.{},
    model: []u8,
    provider: model_provider.ProviderId = .gateway,
    api_key: []u8,
    gateway_team: ?[]u8 = null,
    credential_source: ?types.CredentialSource = null,
    account_id: ?[]u8 = null,
    permission_mode: types.PermissionMode,
    history: []types.HistoryTurn,
    /// Absolute end of the leading canonical-history range loaded without
    /// corrected result provenance. The conservative default protects callers
    /// that do not own a SessionRuntime snapshot.
    unversioned_history_count: usize = std.math.maxInt(usize),
    root_user_intent_context: []u8 = &.{},
    grants: []types.PermissionGrant,
    skill_bindings: []SkillBinding = &.{},
    skill_display_spans: []SkillDisplaySpan = &.{},
    context_snapshot: context_contract.GatheredContextSnapshot = .{},
    agent_settings: AgentTurnSettings = .{},
    snapshot_file_ownerships: []types.SnapshotFileOwnership = &.{},
    /// Present only when an explicit host action resumes a durable paused
    /// model turn. The same ownership rules as the other queued fields apply.
    recovery_checkpoint: ?session_codec.RecoveryCheckpoint = null,
    /// True when this output surface already contains the checkpoint's
    /// assistant source. Recovery still restores the source for overlap and
    /// persistence, but publishes only novel continuation text.
    recovery_source_already_presented: bool = false,
    /// The interactive transcript already contains this exact user turn.
    /// Worker begin publishes only its identity so the UI can consume the
    /// pending owner without painting a duplicate card.
    user_prompt_already_presented: bool = false,
};

pub const ContextCompactionTask = struct {
    turn_id: u64 = 0,
    operation_id: ?compaction_activity.OperationId = null,
    model: []u8,
    provider: model_provider.ProviderId = .gateway,
    api_key: []u8,
    gateway_team: ?[]u8 = null,
    credential_source: ?types.CredentialSource = null,
    account_id: ?[]u8 = null,
    history: []types.HistoryTurn,
    unversioned_history_count: usize = std.math.maxInt(usize),
};

pub const ContextCompactionStatus = enum {
    idle,
    queued,
    running,
};

pub const WorkItem = union(enum) {
    prompt: QueuedPrompt,
    compact_context: ContextCompactionTask,
};

pub const ActivePromptSnapshotOwnership = struct {
    /// Owns snapshot deletion until the files are transferred to accepted
    /// history or handed to a reference-counted finished-turn owner.
    images: []const types.ImageAttachment,
    state: enum {
        active,
        handed_off,
        preserved,
        discarded,
    } = .active,
    shared_ownership: ?types.SnapshotFileOwnership = null,

    pub fn init(images: []const types.ImageAttachment) ActivePromptSnapshotOwnership {
        return .{ .images = images };
    }

    fn ensureSharedOwnership(
        self: *ActivePromptSnapshotOwnership,
        alloc: std.mem.Allocator,
    ) !?types.SnapshotFileOwnership {
        if (self.images.len == 0) return null;
        if (self.state == .preserved or self.state == .discarded) return null;
        if (self.shared_ownership) |ownership| return ownership;
        const ownership = (try SnapshotFileOwnershipState.create(alloc, self.images)).handle();
        self.shared_ownership = ownership;
        return ownership;
    }

    fn handoff(
        self: *ActivePromptSnapshotOwnership,
        alloc: std.mem.Allocator,
    ) !?types.SnapshotFileOwnership {
        if (self.state != .active or self.images.len == 0) return null;
        const ownership = (try self.ensureSharedOwnership(alloc)).?;
        self.state = .handed_off;
        return ownership;
    }

    fn preserve(self: *ActivePromptSnapshotOwnership) bool {
        if (self.images.len == 0) return false;
        switch (self.state) {
            .active => {
                if (self.shared_ownership) |ownership| {
                    ownership.transfer();
                    ownership.release();
                    self.shared_ownership = null;
                }
                self.state = .preserved;
                return true;
            },
            .handed_off => {
                if (self.shared_ownership) |ownership| ownership.transfer();
                return true;
            },
            .preserved => return true,
            .discarded => return false,
        }
    }

    fn deinit(self: *ActivePromptSnapshotOwnership) void {
        if (self.state == .active) {
            if (self.shared_ownership) |ownership| {
                ownership.release();
            } else {
                image_attachments.deleteUnreferencedImageSnapshots(self.images, &.{});
            }
            self.state = .discarded;
        }
    }
};

const SnapshotFileOwnershipState = struct {
    alloc: std.mem.Allocator,
    images: []types.ImageAttachment,
    references: std.atomic.Value(usize) = std.atomic.Value(usize).init(1),
    transferred: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),

    fn create(
        alloc: std.mem.Allocator,
        images: []const types.ImageAttachment,
    ) !*SnapshotFileOwnershipState {
        const owned_images = try types.dupeImageAttachmentSlice(alloc, images);
        errdefer types.freeImageAttachmentSlice(alloc, owned_images);
        const state = try alloc.create(SnapshotFileOwnershipState);
        state.* = .{
            .alloc = alloc,
            .images = owned_images,
        };
        return state;
    }

    fn handle(self: *SnapshotFileOwnershipState) types.SnapshotFileOwnership {
        return .{
            .ctx = self,
            .retain_fn = retain,
            .release_fn = release,
            .transfer_fn = transfer,
        };
    }

    fn retain(raw: *anyopaque) void {
        const self: *SnapshotFileOwnershipState = @ptrCast(@alignCast(raw));
        _ = self.references.fetchAdd(1, .seq_cst);
    }

    fn release(raw: *anyopaque) void {
        const self: *SnapshotFileOwnershipState = @ptrCast(@alignCast(raw));
        if (self.references.fetchSub(1, .seq_cst) != 1) return;
        if (!self.transferred.load(.seq_cst)) {
            image_attachments.deleteUnreferencedImageSnapshots(self.images, &.{});
        }
        const alloc = self.alloc;
        types.freeImageAttachmentSlice(alloc, self.images);
        alloc.destroy(self);
    }

    fn transfer(raw: *anyopaque) void {
        const self: *SnapshotFileOwnershipState = @ptrCast(@alignCast(raw));
        self.transferred.store(true, .seq_cst);
    }
};

fn historyTurnImages(turn: types.HistoryTurn) []const types.ImageAttachment {
    return switch (turn) {
        .assistant => |value| value.user.images,
        .interrupted => |value| value.user.images,
        .compacted_summary => &.{},
    };
}

fn sameImageSnapshots(
    left: []const types.ImageAttachment,
    right: []const types.ImageAttachment,
) bool {
    if (left.len == 0 or left.len != right.len) return false;
    for (left) |image| {
        const other = image_attachments.findById(right, image.id) orelse return false;
        if (!optionalBytesEqual(image.snapshot_path, other.snapshot_path) or
            !optionalBytesEqual(image.snapshot_sha256, other.snapshot_sha256))
        {
            return false;
        }
    }
    return true;
}

fn optionalBytesEqual(left: ?[]const u8, right: ?[]const u8) bool {
    if (left == null or right == null) return left == null and right == null;
    return std.mem.eql(u8, left.?, right.?);
}

pub const PermissionSnapshot = struct {
    mode: types.PermissionMode,
};

pub const CommandOutputChunk = struct {
    lifecycle_id: ?types.ToolLifecycleId = null,
    stream: command_output_content.Stream,
    text: []u8,
};

pub const SteeringPresentationSnapshot = struct {
    messages: [][]u8 = &.{},
    pending_feedback: [][]u8 = &.{},
    waits_for_boundary: bool = false,

    pub fn deinit(self: *SteeringPresentationSnapshot, alloc: std.mem.Allocator) void {
        for (self.messages) |message| alloc.free(message);
        if (self.messages.len > 0) alloc.free(self.messages);
        for (self.pending_feedback) |message| alloc.free(message);
        if (self.pending_feedback.len > 0) alloc.free(self.pending_feedback);
        self.* = .{};
    }
};

const PreparedHistoryPropagation = struct {
    root_user_intent_context: []u8,
    authorized_image_catalog: []types.ImageAttachment,
    snapshot_file_ownerships: ?[]types.SnapshotFileOwnership,
    added_snapshot_ownership: ?types.SnapshotFileOwnership,

    fn deinit(self: PreparedHistoryPropagation, alloc: std.mem.Allocator) void {
        alloc.free(self.root_user_intent_context);
        types.freeImageAttachmentSlice(alloc, self.authorized_image_catalog);
        if (self.added_snapshot_ownership) |ownership| ownership.release();
        if (self.snapshot_file_ownerships) |ownerships| alloc.free(ownerships);
    }
};

pub const StateSnapshot = struct {
    processing: bool = false,
    active_turn_id: u64 = 0,
    queued_count: usize = 0,
    pending_event_count: usize = 0,
    cancel_requested: bool = false,
    cancel_continues_turn: bool = false,
    pending_permission_request: ?permission_request.OwnedPermissionRequest = null,
    pending_permission_review: ?PendingPermissionReview = null,

    pub fn deinit(self: StateSnapshot, alloc: std.mem.Allocator) void {
        if (self.pending_permission_request) |request| {
            var owned = request;
            owned.deinit(alloc);
        }
    }
};

pub const PendingPermissionReview = struct {
    request_id: u64,
    review: *const diff_mod.FileReview,
};

pub const FinalizationFailure = struct {
    turn_id: u64,
    outcome: types.TurnPresentationOutcome,
};

pub const InteractiveAdmissionSnapshot = union(enum) {
    open,
    stopped,
    finalization_failed: FinalizationFailure,
};

pub const PermissionSubmissionResult = enum {
    accepted,
    stale,
    no_pending,
};

const OwnedQuestionOption = struct {
    label: []u8,
    description: ?[]u8 = null,
};

const OwnedQuestionEntry = struct {
    question: []u8,
    options: []OwnedQuestionOption,
    submission: @FieldType(types.QuestionBatchEntry, "submission") = .none,
};

pub const QuestionPromptSource = enum {
    agent_question,
    route_recovery,
    mcp_elicitation,
};

const OwnedQuestionBatch = struct {
    entries: []OwnedQuestionEntry,
    source: QuestionPromptSource = .agent_question,
};

const QuestionResponse = union(enum) {
    pending,
    cancelled,
    answered: [][]u8,
};

pub const PendingQuestionBatchSnapshot = struct {
    entries: []types.QuestionBatchEntry,
    source: QuestionPromptSource = .agent_question,

    pub fn deinit(self: PendingQuestionBatchSnapshot, alloc: std.mem.Allocator) void {
        for (self.entries) |entry| {
            alloc.free(@constCast(entry.question));
            for (entry.options) |opt| {
                alloc.free(@constCast(opt.label));
                if (opt.description) |desc| alloc.free(@constCast(desc));
            }
            alloc.free(@constCast(entry.options));
        }
        alloc.free(self.entries);
    }
};

pub const ContextCompaction = struct {
    turn_id: u64 = 0,
    summary: types.CompactedSummaryHistoryTurn,
    active_prefix: ?types.AssistantHistoryTurn = null,
    retained_from: ?types.ContextHistoryCut = null,
    max_history_turns: usize = 0,
};

pub const ContextCompactionHandler = *const fn (*anyopaque, ContextCompaction) anyerror!void;

pub const FreshPromptPreparation = struct {
    turn_id: u64 = 0,
    user: types.UserTurn,
    prior_turn: ?types.HistoryTurn = null,
};

pub const FreshPromptHistory = struct {
    history: []types.HistoryTurn,
    authorized_image_catalog: []types.ImageAttachment,
    root_user_intent_context: []u8,
    unversioned_history_count: usize,

    pub fn deinit(self: *FreshPromptHistory, alloc: std.mem.Allocator) void {
        types.freeHistoryTurnSlice(alloc, self.history);
        types.freeImageAttachmentSlice(alloc, self.authorized_image_catalog);
        alloc.free(self.root_user_intent_context);
        self.* = undefined;
    }
};

pub const FreshPromptHandler = *const fn (*anyopaque, FreshPromptPreparation) anyerror!FreshPromptHistory;

const HistoryPublication = struct {
    ctx: *anyopaque,
    commit_fn: *const fn (*anyopaque) anyerror!void,
    retained_from: ?types.ContextHistoryCut = null,

    fn commit(self: HistoryPublication) !void {
        try self.commit_fn(self.ctx);
    }
};

const ContextCompactionPublication = struct {
    ctx: *anyopaque,
    handler: ContextCompactionHandler,
    event: ContextCompaction,

    fn commit(raw: *anyopaque) !void {
        const self: *ContextCompactionPublication = @ptrCast(@alignCast(raw));
        try self.handler(self.ctx, self.event);
    }
};

const HistoryPublicationResponse = union(enum) {
    idle,
    waiting: u64,
    committed: ?FreshPromptHistory,
    failed: anyerror,
};

pub const WorkerEvent = union(enum) {
    begin_prompt: types.UserTurn,
    begin_prompt_with_skill_bindings: BeginPromptWithSkillBindings,
    begin_presented_prompt: u64,
    append_user_feedback: []u8,
    assistant_presentation: assistant_presentation.Event,
    notification: notification_contract.Notification,
    question_requested,
    open_model_picker,
    semantic_notice: types.SemanticNotice,
    /// A full-only observability record (network call outcome) for the ctrl+o
    /// full transcript's detail section. Never enters the transcript store.
    full_detail_record: types.SemanticNotice,
    route_recovery_status: types.RouteRecoveryStatus,
    clear_route_recovery_status,
    api_status_text: []u8,
    credential_refreshed: credentials.Credential,
    command_output: CommandOutputChunk,
    command_output_complete: ?types.ToolLifecycleId,
    tool_lifecycle: types.ToolLifecycleEvent,
    turn_token_update: types.TurnTokenProgress,
    turn_phase_update: types.TurnPhaseUpdate,
    diff_block: diff_mod.DiffEntryPayload,
    context_compaction: ContextCompaction,
    prepare_fresh_prompt: FreshPromptPreparation,
    finish_prompt: types.FinishedPrompt,
    session_grant: types.PermissionGrant,
    error_text: types.SemanticNotice,
    /// A turn that can never recover hands the prompt back to the composer.
    /// Owned by the event; the consumer frees it.
    restore_failed_prompt: []u8,
};

pub const WorkerEventBatch = struct {
    events: std.ArrayList(WorkerEvent),
    cancelled_turn_id: ?u64,
};

pub const WorkerRuntime = struct {
    worker_mutex: std.Io.Mutex = .init,
    worker_cond: std.Io.Condition = .init,
    /// One admission-ordered queue for ordinary prompts and steering. Steering
    /// remains in place until its target turn consumes or demotes it.
    queued_prompts: std.ArrayList(QueuedPrompt) = .empty,
    /// One immutable conversation snapshot shared by every queued prompt.
    /// It is transferred or copied only when a prompt actually begins.
    queued_history: []types.HistoryTurn = &.{},
    queued_context_compaction: ?ContextCompactionTask = null,
    active_context_compaction: bool = false,
    history_publication_response: HistoryPublicationResponse = .idle,
    worker_events: std.ArrayList(WorkerEvent) = .empty,
    worker_processing: bool = false,
    active_turn_id: u64 = 0,
    active_tool_calls: std.StringHashMapUnmanaged(void) = .empty,
    /// Allocator used for active call IDs. Allocator handles are capabilities,
    /// not comparable identities; callers guarantee compatible ownership.
    active_tool_allocator: ?std.mem.Allocator = null,
    /// Retains the tool-producing step so later text from that same step cannot
    /// reopen immediate steering before authoritative lifecycle IDs exist.
    tool_phase_step_id: ?u64 = null,
    worker_stop_requested: bool = false,
    finalization_failure: ?FinalizationFailure = null,
    worker_cancel_requested: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
    /// The active turn whose text steering admission owns the cancel flag.
    /// Guarded by `worker_mutex`; explicit cancellation clears this ownership.
    steering_cancel_turn_id: ?u64 = null,
    direct_steering_closed: bool = false,
    worker_recovery_pause_requested: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
    worker_connectivity_wait_active: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
    /// Set under `worker_mutex` once the active turn publishes its terminal
    /// recovery pause. The turn may still be finalizing, but cannot perform
    /// more model work, so one continuation may queue before processing clears.
    recovery_continuation_ready: bool = false,
    /// When true, queued work will not start.
    turn_start_held: bool = false,
    next_permission_request_id: u64 = 1,
    pending_permission_response: ?permission_request.OwnedPermissionResponse = null,
    pending_permission_request_shared: ?permission_request.OwnedPermissionRequest = null,
    pending_permission_review: ?*const diff_mod.FileReview = null,
    pending_permission_waiting: bool = false,
    pending_question_shared: ?OwnedQuestionBatch = null,
    pending_question_response: QuestionResponse = .pending,
    agent_turn_settings: AgentTurnSettings = .{},
    active_agent_turn_settings: ?AgentTurnSettings = null,
    active_context_snapshot: ?*const context_contract.GatheredContextSnapshot = null,
    active_prompt_is_root_authority: bool = false,
    active_prompt_snapshot_ownership: ?*ActivePromptSnapshotOwnership = null,
    preserve_prompt_snapshot_turn_id: ?u64 = null,

    /// Presentation metadata only; all access is protected by worker_mutex.
    compaction_presentation: compaction_activity.State = .{},

    pub fn beginCompactionActivity(self: *WorkerRuntime, origin: compaction_activity.Origin, turn_id: ?u64) compaction_activity.OperationId {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        return self.compaction_presentation.begin(origin, turn_id, io_mod.milliTimestamp());
    }

    /// A rejected intent must not hide an already-running compaction.
    pub fn rejectCompactionActivity(self: *WorkerRuntime, outcome: compaction_activity.Outcome) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.compaction_presentation.snapshot.operation) |op| if (op.active()) return;
        const now = io_mod.milliTimestamp();
        const id = self.compaction_presentation.begin(.manual, null, now);
        self.compaction_presentation.settle(id, .{ .outcome = outcome }, now);
    }

    pub fn runCompactionActivity(self: *WorkerRuntime, id: compaction_activity.OperationId, stage: compaction_activity.Stage) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        self.compaction_presentation.running(id, stage);
    }

    pub fn settleCompactionActivity(self: *WorkerRuntime, id: compaction_activity.OperationId, feedback: compaction_activity.Feedback) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        self.compaction_presentation.settle(id, feedback, io_mod.milliTimestamp());
    }

    /// Cheap by-value read; does not clone permissions or allocate.
    pub fn compactionActivitySnapshot(self: *WorkerRuntime) compaction_activity.Snapshot {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        return self.compaction_presentation.snapshot;
    }

    pub fn dismissCompactionActivity(self: *WorkerRuntime, id: compaction_activity.OperationId, revision: u64) bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        return self.compaction_presentation.dismiss(id, revision);
    }

    pub fn expireCompactionActivity(self: *WorkerRuntime, id: compaction_activity.OperationId, revision: u64, now_ms: i64) bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        return self.compaction_presentation.expire(id, revision, now_ms);
    }

    fn stopCompactionActivityLocked(self: *WorkerRuntime) void {
        const op = self.compaction_presentation.snapshot.operation orelse return;
        if (op.turn_id != self.active_turn_id or !self.worker_processing) return;
        self.compaction_presentation.stopping(op.id);
    }

    fn finishCompactionActivityLocked(self: *WorkerRuntime, turn_id: u64) void {
        const op = self.compaction_presentation.snapshot.operation orelse return;
        if (op.turn_id != turn_id or !op.active()) return;
        diagnostics.traceCompactionLog(true, "unsettled operation at worker finish turn_id={d}", .{turn_id});
        var feedback = compaction_activity.failure(
            if (self.isCancelRequested()) error.Cancelled else error.CompactionInterrupted,
            op.stage(),
            self.isCancelRequested(),
        );
        // No transaction acknowledgement survived this fallback; do not promise rollback.
        if (op.stage() == .publication) feedback.publication = .uncertain;
        self.compaction_presentation.settle(op.id, feedback, io_mod.milliTimestamp());
    }

    fn queuedWorkCountLocked(self: *const WorkerRuntime) usize {
        return self.queued_prompts.items.len +
            @intFromBool(self.queued_context_compaction != null);
    }

    pub fn deinit(self: *WorkerRuntime, alloc: std.mem.Allocator) void {
        if (self.pending_permission_response) |response| {
            self.discardPermissionResponse(response, "deinit");
        }
        self.pending_permission_response = null;
        if (self.pending_permission_request_shared) |*request| request.deinit(alloc);
        self.pending_permission_request_shared = null;
        self.pending_permission_review = null;
        freeOwnedQuestionBatchOpt(alloc, self.pending_question_shared);
        self.pending_question_shared = null;
        freeQuestionResponse(alloc, self.pending_question_response);
        self.pending_question_response = .pending;

        for (self.queued_prompts.items) |prompt| discardQueuedPrompt(alloc, prompt, &.{});
        self.queued_prompts.deinit(alloc);
        types.freeHistoryTurnSlice(alloc, self.queued_history);
        if (self.queued_context_compaction) |task| freeContextCompactionTask(alloc, task);

        self.clearActiveToolCallsLocked();

        for (self.worker_events.items) |event| freeWorkerEvent(alloc, event);
        self.worker_events.deinit(alloc);
    }

    pub fn requestStop(self: *WorkerRuntime) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        self.worker_stop_requested = true;
        if (self.queued_context_compaction) |task| {
            if (task.operation_id) |id| self.compaction_presentation.settle(id, .{ .outcome = .cancelled }, io_mod.milliTimestamp());
        }
        self.stopCompactionActivityLocked();
        self.worker_cond.broadcast(io_mod.getIo());
        self.worker_mutex.unlock(io_mod.getIo());
    }

    pub fn requestShutdown(self: *WorkerRuntime) void {
        self.markExplicitCancellation();
        self.requestStop();
    }

    pub fn requestCancel(self: *WorkerRuntime) void {
        debug_trace.logf("worker", "cancel requested processing={s} queued={d}", .{ if (self.worker_processing) "true" else "false", self.queuedPromptCount() });
        debug_trace.eventf("interrupt", "cancel_requested", .{}, "processing={s} queued={d} active_tool_known=false", .{ if (self.worker_processing) "true" else "false", self.queuedPromptCount() });
        self.markExplicitCancellation();
    }

    /// Interrupt the current gateway wait while preserving its recovery
    /// checkpoint. The ordinary cancel flag wakes the existing socket watcher;
    /// agent policy distinguishes the typed pause flag at the boundary.
    pub fn requestRecoveryPause(self: *WorkerRuntime) void {
        debug_trace.eventf("recovery", "pause_requested", .{}, "source=interactive_try_later", .{});
        self.worker_recovery_pause_requested.store(true, .seq_cst);
        self.markExplicitCancellation();
    }

    fn markExplicitCancellation(self: *WorkerRuntime) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        self.steering_cancel_turn_id = null;
        self.worker_cancel_requested.store(true, .seq_cst);
        self.stopCompactionActivityLocked();
        self.worker_cond.broadcast(io_mod.getIo());
    }

    pub fn requestInteractiveCancel(self: *WorkerRuntime) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        const steering_pending = for (self.queued_prompts.items) |prompt| {
            if (prompt.delivery.activeTurnId() == self.active_turn_id or
                prompt.delivery.isContinuation()) break true;
        } else false;
        debug_trace.logf("worker", "cancel requested processing={s} queued={d} steering_pending={s}", .{
            if (self.worker_processing) "true" else "false",
            self.queuedWorkCountLocked(),
            if (steering_pending) "true" else "false",
        });
        debug_trace.eventf("interrupt", "cancel_requested", .{}, "processing={s} queued={d} steering_pending={s} active_tool_known=false", .{
            if (self.worker_processing) "true" else "false",
            self.queuedWorkCountLocked(),
            if (steering_pending) "true" else "false",
        });
        self.steering_cancel_turn_id = null;
        self.worker_cancel_requested.store(true, .seq_cst);
        self.stopCompactionActivityLocked();
        self.worker_cond.broadcast(io_mod.getIo());
    }

    pub fn cancelApprovalTurn(self: *WorkerRuntime) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        self.steering_cancel_turn_id = null;
        self.worker_cancel_requested.store(true, .seq_cst);
        self.stopCompactionActivityLocked();
        _ = self.resolvePendingPermissionLocked(
            permission_request.OwnedPermissionResponse.init(
                std.heap.c_allocator,
                .deny,
                null,
            ),
        );
    }

    pub fn isCancelRequested(self: *const WorkerRuntime) bool {
        return self.worker_cancel_requested.load(.seq_cst);
    }

    pub fn cancellationStopsTurn(self: *WorkerRuntime) bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        return self.worker_cancel_requested.load(.seq_cst) and
            self.steering_cancel_turn_id != self.active_turn_id;
    }

    pub fn isConnectivityWaitActive(self: *const WorkerRuntime) bool {
        return self.worker_connectivity_wait_active.load(.seq_cst);
    }

    pub fn pushEvent(self: *WorkerRuntime, alloc: std.mem.Allocator, event: WorkerEvent) !void {
        const owned = try dupeWorkerEvent(alloc, event);
        errdefer freeWorkerEvent(alloc, owned);
        try self.pushOwnedEvent(alloc, owned);
    }

    /// Transfers an already-owned event into the queue on success.
    /// Every heap payload in `event` must have been allocated by `alloc`.
    pub fn pushOwnedEvent(self: *WorkerRuntime, alloc: std.mem.Allocator, event: WorkerEvent) !void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        try self.worker_events.ensureUnusedCapacity(alloc, 1);
        var tool_transition = try self.prepareActiveToolTransitionLocked(alloc, event);
        errdefer tool_transition.deinit(alloc);
        self.worker_events.appendAssumeCapacity(event);
        self.applyActiveToolTransitionLocked(tool_transition);
        self.applyToolBoundaryPhaseLocked(event);
        self.applyRecoveryStateEvent(event);
        self.worker_cond.broadcast(io_mod.getIo());
    }

    /// Transfers already-owned events to the front of the queue on success.
    /// Every heap payload must have been allocated by `alloc`.
    pub fn prependOwnedEvents(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        events: []const WorkerEvent,
    ) !void {
        if (events.len == 0) return;
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        try self.worker_events.insertSlice(alloc, 0, events);
        self.worker_cond.broadcast(io_mod.getIo());
    }

    pub fn pushTurnFinished(self: *WorkerRuntime, alloc: std.mem.Allocator, event: types.TurnFinished) !void {
        std.debug.assert(event.turn_id != 0);

        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        self.worker_events.append(alloc, .{ .tool_lifecycle = .{
            .turn_finished = event,
        } }) catch {
            self.latchFinalizationFailureLocked(.{
                .turn_id = event.turn_id,
                .outcome = event.outcome,
            });
            return error.TurnFinalizationDeliveryFailed;
        };
        self.applyRecoveryStateEvent(.{ .tool_lifecycle = .{
            .turn_finished = event,
        } });
        self.worker_cond.broadcast(io_mod.getIo());
    }

    fn applyRecoveryStateEvent(self: *WorkerRuntime, event: WorkerEvent) void {
        switch (event) {
            .route_recovery_status => |status| {
                self.worker_connectivity_wait_active.store(
                    status.action == .waiting_for_connectivity,
                    .seq_cst,
                );
                self.recovery_continuation_ready = status.action == .paused;
            },
            .clear_route_recovery_status,
            .finish_prompt,
            .error_text,
            => {
                self.worker_connectivity_wait_active.store(false, .seq_cst);
                self.recovery_continuation_ready = false;
            },
            .tool_lifecycle => |lifecycle| switch (lifecycle) {
                .turn_finished => self.worker_connectivity_wait_active.store(false, .seq_cst),
                .provisional, .authoritative_started, .progress, .terminal => return,
            },
            else => return,
        }
    }

    pub fn latchFinalizationFailure(self: *WorkerRuntime, failure: FinalizationFailure) void {
        std.debug.assert(failure.turn_id != 0);

        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        self.latchFinalizationFailureLocked(failure);
    }

    fn latchFinalizationFailureLocked(self: *WorkerRuntime, failure: FinalizationFailure) void {
        if (self.finalization_failure == null) {
            self.finalization_failure = failure;
        } else {
            debug_trace.logf(
                "worker",
                "duplicate finalization failure ignored turn_id={d} first_turn_id={d}",
                .{ failure.turn_id, self.finalization_failure.?.turn_id },
            );
        }
        self.worker_stop_requested = true;
        self.steering_cancel_turn_id = null;
        self.worker_cancel_requested.store(true, .seq_cst);
        self.stopCompactionActivityLocked();
        self.worker_cond.broadcast(io_mod.getIo());
    }

    pub fn interactiveAdmissionSnapshot(self: *WorkerRuntime) InteractiveAdmissionSnapshot {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        if (self.finalization_failure) |failure| {
            return .{ .finalization_failed = failure };
        }
        if (self.worker_stop_requested) return .stopped;
        return .open;
    }

    pub fn takeEventBatch(self: *WorkerRuntime) WorkerEventBatch {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        const events = self.worker_events;
        self.worker_events = .empty;
        return .{
            .events = events,
            .cancelled_turn_id = if (self.worker_cancel_requested.load(.seq_cst) and
                self.active_turn_id != 0 and
                self.steering_cancel_turn_id != self.active_turn_id)
                self.active_turn_id
            else
                null,
        };
    }

    pub fn takeEvents(self: *WorkerRuntime) std.ArrayList(WorkerEvent) {
        return self.takeEventBatch().events;
    }

    pub fn enqueuePrompt(self: *WorkerRuntime, alloc: std.mem.Allocator, prompt: QueuedPrompt) !void {
        try self.admitPrompt(alloc, prompt, false);
    }

    pub fn admitInteractivePrompt(self: *WorkerRuntime, alloc: std.mem.Allocator, prompt: QueuedPrompt) !void {
        try self.admitPrompt(alloc, prompt, true);
    }

    /// Takes prompt ownership only on true. Does not cancel an in-flight operation
    /// or fall back to the ordinary FIFO of an ephemeral direct worker.
    pub fn admitActiveSteering(self: *WorkerRuntime, alloc: std.mem.Allocator, prompt: QueuedPrompt) !bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (!self.worker_processing or self.active_turn_id == 0 or
            self.direct_steering_closed or self.worker_stop_requested or
            self.worker_cancel_requested.load(.seq_cst)) return false;
        var queued = prompt;
        queued.delivery = .{ .active_turn = self.active_turn_id };
        try self.enqueuePromptLocked(alloc, queued);
        return true;
    }

    pub fn enqueueContextCompaction(
        self: *WorkerRuntime,
        task: ContextCompactionTask,
    ) !void {
        var queued = task;
        if (queued.turn_id == 0) queued.turn_id = debug_trace.nextTurnId();

        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.finalization_failure != null) return error.TurnFinalizationDeliveryFailed;
        if (self.worker_stop_requested) return error.WorkerStopped;
        if (self.worker_processing or self.queued_prompts.items.len > 0 or
            self.queued_context_compaction != null)
        {
            return error.WorkerBusy;
        }
        if (queued.operation_id == null) queued.operation_id = self.compaction_presentation.begin(.manual, queued.turn_id, io_mod.milliTimestamp());
        self.compaction_presentation.queued(queued.operation_id.?, queued.turn_id, io_mod.milliTimestamp());
        self.queued_context_compaction = queued;
        debug_trace.eventf(
            "worker",
            "context_compaction_enqueue",
            .{ .turn_id = queued.turn_id },
            "queue_depth=1",
            .{},
        );
        self.worker_cond.broadcast(io_mod.getIo());
    }

    pub fn contextCompactionStatus(self: *WorkerRuntime) ContextCompactionStatus {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.active_context_compaction) return .running;
        if (self.queued_context_compaction != null) return .queued;
        return .idle;
    }

    pub fn cancelContextCompaction(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
    ) bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        if (self.queued_context_compaction) |task| {
            self.queued_context_compaction = null;
            if (task.operation_id) |id| self.compaction_presentation.settle(id, .{ .outcome = .cancelled }, io_mod.milliTimestamp());
            self.worker_cond.broadcast(io_mod.getIo());
            self.worker_mutex.unlock(io_mod.getIo());
            freeContextCompactionTask(alloc, task);
            debug_trace.eventf(
                "context_compaction",
                "cancelled",
                .{},
                "phase=queued",
                .{},
            );
            return true;
        }
        if (self.active_context_compaction) {
            self.worker_cancel_requested.store(true, .seq_cst);
            self.stopCompactionActivityLocked();
            self.worker_cond.broadcast(io_mod.getIo());
            self.worker_mutex.unlock(io_mod.getIo());
            debug_trace.eventf(
                "context_compaction",
                "cancel_requested",
                .{},
                "phase=running",
                .{},
            );
            return true;
        }
        self.worker_mutex.unlock(io_mod.getIo());
        return false;
    }

    /// Transfers `prompt` to the active turn when steering is requested and the
    /// turn still accepts guidance. Otherwise it enters the ordinary FIFO.
    /// A steer interrupts the model stream immediately only when no wait
    /// boundary is active; a running tool or an in-flight compaction defers
    /// delivery to the next steering boundary instead of being cancelled.
    fn admitPrompt(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        prompt: QueuedPrompt,
        steer_if_active: bool,
    ) !void {
        var queued = prompt;
        if (queued.turn_id == 0) queued.turn_id = debug_trace.nextTurnId();

        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.finalization_failure != null) return error.TurnFinalizationDeliveryFailed;
        if (self.worker_stop_requested) return error.WorkerStopped;
        if (queued.recovery_checkpoint != null and
            !recoveryContinuationAdmission(
                self.worker_processing,
                self.queuedWorkCountLocked(),
                self.recovery_continuation_ready,
            ))
        {
            return error.RecoveryBusy;
        }
        queued.agent_settings = self.agent_turn_settings;
        var interrupt_after_admission = false;
        const explicitly_cancelled = self.worker_cancel_requested.load(.seq_cst) and
            self.steering_cancel_turn_id != self.active_turn_id;
        if (steer_if_active and self.worker_processing and self.active_turn_id != 0 and
            !explicitly_cancelled)
        {
            queued.delivery = .{ .active_turn = self.active_turn_id };
            interrupt_after_admission = !self.hasSteeringWaitBoundaryLocked();
        }
        try self.enqueuePromptLocked(alloc, queued);
        if (steer_if_active and self.worker_processing) {
            debug_trace.eventf("worker", "steering_admission", .{ .turn_id = self.active_turn_id }, "queued_turn_id={d} targeted={} plain={} tool_boundary={} compaction_active={} interrupt_model={}", .{ queued.turn_id, queued.delivery.activeTurnId() == self.active_turn_id, sameTurnSteeringEligible(queued), self.hasActiveToolBoundaryLocked(), self.compactionInFlightLocked(), interrupt_after_admission });
        }
        if (interrupt_after_admission) {
            self.steering_cancel_turn_id = if (sameTurnSteeringEligible(queued))
                self.active_turn_id
            else
                null;
            self.worker_cancel_requested.store(true, .seq_cst);
        }
    }

    fn sameTurnSteeringEligible(prompt: QueuedPrompt) bool {
        return prompt.images.len == 0 and
            prompt.skill_bindings.len == 0 and
            prompt.skill_display_spans.len == 0;
    }

    fn enqueuePromptLocked(self: *WorkerRuntime, alloc: std.mem.Allocator, queued_value: QueuedPrompt) !void {
        var queued = queued_value;
        try self.queued_prompts.ensureUnusedCapacity(alloc, 1);
        const supplied_history = queued.history;
        queued.history = &.{};
        if (self.queued_prompts.items.len == 0) {
            std.debug.assert(self.queued_history.len == 0);
            self.queued_history = supplied_history;
        } else {
            types.freeHistoryTurnSlice(alloc, supplied_history);
        }
        self.queued_prompts.appendAssumeCapacity(queued);
        debug_trace.logf(
            "worker",
            "queued prompt bytes={d} queue_depth={d} fast_mode={s} effort={s}",
            .{ queued.prompt.len, self.queuedWorkCountLocked(), if (queued.agent_settings.fast_mode) "true" else "false", queued.agent_settings.effort.label() },
        );
        debug_trace.eventf(
            "worker",
            "prompt_enqueue",
            .{ .turn_id = queued.turn_id },
            "prompt_bytes={d} queue_depth={d} fast_mode={s} effort={s}",
            .{ queued.prompt.len, self.queuedWorkCountLocked(), if (queued.agent_settings.fast_mode) "true" else "false", queued.agent_settings.effort.label() },
        );
        self.worker_cond.broadcast(io_mod.getIo());
    }

    pub fn hasPendingPlainSteering(self: *WorkerRuntime) bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (!self.worker_processing or self.active_turn_id == 0 or self.worker_stop_requested) return false;
        var pending = false;
        for (self.queued_prompts.items) |prompt| {
            if (prompt.delivery.activeTurnId() != self.active_turn_id) continue;
            if (!sameTurnSteeringEligible(prompt)) return false;
            pending = true;
        }
        return pending;
    }

    /// Classifies and drains steering at one observed agent boundary. Returned
    /// text is allocator-owned; handoff leaves the queued prompt worker-owned.
    pub fn takeSteeringBoundary(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        turn_id: u64,
        kind: SteeringBoundaryKind,
    ) !SteeringBoundaryResult {
        return self.takeSteeringBoundaryInto(alloc, alloc, turn_id, kind);
    }

    /// Queue/event storage uses alloc; returned guidance uses result_alloc.
    pub fn takeSteeringBoundaryInto(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        result_alloc: std.mem.Allocator,
        turn_id: u64,
        kind: SteeringBoundaryKind,
    ) !SteeringBoundaryResult {
        const result = try self.takeSteeringBoundaryIntoInner(alloc, result_alloc, turn_id, kind);
        debug_trace.eventf("worker", "steering_boundary_check", .{ .turn_id = turn_id }, "kind={s} outcome={s}", .{ @tagName(kind), @tagName(result) });
        return result;
    }

    fn takeSteeringBoundaryIntoInner(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        result_alloc: std.mem.Allocator,
        turn_id: u64,
        kind: SteeringBoundaryKind,
    ) !SteeringBoundaryResult {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        const cancel_requested = self.worker_cancel_requested.load(.seq_cst);
        if (!self.worker_processing or self.active_turn_id != turn_id) {
            return if (kind == .cancelled and cancel_requested) .interrupt else .none;
        }

        switch (kind) {
            .cancelled => {
                if (!cancel_requested) return .none;
                if (self.steering_cancel_turn_id != turn_id) return .interrupt;
                const messages = try self.takeSteeringLocked(alloc, result_alloc, turn_id);
                if (messages.len == 0) return .interrupt;
                self.steering_cancel_turn_id = null;
                self.worker_cancel_requested.store(false, .seq_cst);
                return .{ .continue_turn = messages };
            },
            .model, .finalizing => {
                if (cancel_requested) return .none;
                for (self.queued_prompts.items) |prompt| {
                    if (prompt.delivery.activeTurnId() == turn_id and
                        !sameTurnSteeringEligible(prompt)) return .handoff;
                }
                const messages = try self.takeSteeringLocked(alloc, result_alloc, turn_id);
                if (kind == .finalizing and messages.len == 0) self.direct_steering_closed = true;
                return if (messages.len == 0)
                    .none
                else
                    .{ .continue_turn = messages };
            },
        }
    }

    fn takeSteeringLocked(self: *WorkerRuntime, alloc: std.mem.Allocator, result_alloc: std.mem.Allocator, turn_id: u64) ![][]u8 {
        var steering_count: usize = 0;
        for (self.queued_prompts.items) |prompt| {
            if (prompt.delivery.activeTurnId() != turn_id) continue;
            if (!sameTurnSteeringEligible(prompt)) return &.{};
            steering_count += 1;
        }
        if (steering_count == 0) return &.{};

        const messages = try result_alloc.alloc([]u8, steering_count);
        var copied: usize = 0;
        errdefer {
            for (messages[0..copied]) |text| result_alloc.free(text);
            result_alloc.free(messages);
        }
        const events = try alloc.alloc(WorkerEvent, steering_count);
        var event_count: usize = 0;
        errdefer {
            for (events[0..event_count]) |event| freeWorkerEvent(alloc, event);
            alloc.free(events);
        }
        for (self.queued_prompts.items) |prompt| {
            if (prompt.delivery.activeTurnId() != turn_id) continue;
            messages[copied] = try result_alloc.dupe(u8, prompt.prompt);
            copied += 1;
            if (prompt.steering_receipt == null) {
                events[event_count] = .{
                    .append_user_feedback = try alloc.dupe(u8, prompt.prompt),
                };
                event_count += 1;
            }
        }
        try self.worker_events.ensureUnusedCapacity(alloc, event_count);
        for (events[0..event_count]) |event| self.worker_events.appendAssumeCapacity(event);
        alloc.free(events);

        var index: usize = 0;
        while (index < self.queued_prompts.items.len) {
            if (self.queued_prompts.items[index].delivery.activeTurnId() != turn_id) {
                index += 1;
                continue;
            }
            const prompt = self.queued_prompts.orderedRemove(index);
            if (prompt.steering_receipt) |receipt| {
                receipt.state.store(.applied, .seq_cst);
                debug_trace.eventf("subagent", "feedback_applied", .{}, "operation={s}", .{receipt.operation_id});
            }
            freeQueuedPrompt(alloc, prompt);
        }
        if (self.queued_prompts.items.len == 0) {
            types.freeHistoryTurnSlice(alloc, self.queued_history);
            self.queued_history = &.{};
        }
        debug_trace.eventf("worker", "prompt_steering_consumed", .{ .turn_id = turn_id }, "count={d}", .{messages.len});
        self.worker_cond.broadcast(io_mod.getIo());
        return messages;
    }

    /// Hold turn admission only when no turn is active or queued.
    pub fn tryHoldTurnStart(self: *WorkerRuntime) bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.worker_processing or self.queuedWorkCountLocked() > 0 or self.turn_start_held) {
            return false;
        }
        self.turn_start_held = true;
        debug_trace.logf("worker", "turn start hold acquired", .{});
        return true;
    }

    pub fn releaseTurnStartHold(self: *WorkerRuntime) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (!self.turn_start_held) return;
        self.turn_start_held = false;
        debug_trace.logf("worker", "turn start hold released", .{});
        self.worker_cond.broadcast(io_mod.getIo());
    }

    pub fn removeQueuedPrompt(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        turn_id: u64,
        retained_images: []const types.ImageAttachment,
    ) bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        var removed: ?QueuedPrompt = null;
        for (self.queued_prompts.items, 0..) |queued, index| {
            if (queued.turn_id != turn_id) continue;
            removed = self.queued_prompts.orderedRemove(index);
            break;
        }
        if (removed != null and self.queued_prompts.items.len == 0) {
            types.freeHistoryTurnSlice(alloc, self.queued_history);
            self.queued_history = &.{};
        }
        const remaining = self.queuedWorkCountLocked();
        self.worker_mutex.unlock(io_mod.getIo());

        const prompt = removed orelse return false;
        discardQueuedPrompt(alloc, prompt, retained_images);
        debug_trace.eventf(
            "worker",
            "queued_prompt_removed",
            .{ .turn_id = turn_id },
            "remaining={d}",
            .{remaining},
        );
        return true;
    }

    /// Removes the newest queued text-only steering prompt that still waits for
    /// a tool or compaction boundary and returns an allocator-owned copy of its
    /// text so the composer can restore it for editing. Returns null when
    /// nothing is safely retractable: no active turn, no wait boundary (an
    /// immediate steer is already committed to its interrupt), a cancel
    /// already in flight, or only handoff-class steers with images or skill
    /// bindings.
    pub fn popQueuedSteerForEdit(self: *WorkerRuntime, alloc: std.mem.Allocator) !?[]u8 {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (!self.worker_processing or self.active_turn_id == 0) return null;
        if (!self.hasSteeringWaitBoundaryLocked()) return null;
        if (self.worker_cancel_requested.load(.seq_cst)) return null;

        var retractable: ?usize = null;
        var index: usize = self.queued_prompts.items.len;
        while (index > 0) {
            index -= 1;
            const prompt = self.queued_prompts.items[index];
            if (prompt.delivery.activeTurnId() != self.active_turn_id) continue;
            if (!sameTurnSteeringEligible(prompt)) continue;
            retractable = index;
            break;
        }
        const remove_index = retractable orelse return null;

        const text = try alloc.dupe(u8, self.queued_prompts.items[remove_index].prompt);
        const prompt = self.queued_prompts.orderedRemove(remove_index);
        if (self.queued_prompts.items.len == 0) {
            types.freeHistoryTurnSlice(alloc, self.queued_history);
            self.queued_history = &.{};
        }
        const turn_id = prompt.turn_id;
        const remaining = self.queuedWorkCountLocked();
        freeQueuedPrompt(alloc, prompt);
        debug_trace.eventf(
            "worker",
            "prompt_steering_retracted",
            .{ .turn_id = turn_id },
            "remaining={d}",
            .{remaining},
        );
        self.worker_cond.broadcast(io_mod.getIo());
        return text;
    }

    pub fn waitAndTakeNextPrompt(self: *WorkerRuntime, alloc: std.mem.Allocator) !?QueuedPrompt {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        while ((self.queued_prompts.items.len == 0 or self.turn_start_held) and
            !self.worker_stop_requested)
        {
            self.worker_processing = false;
            self.active_turn_id = 0;
            self.tool_phase_step_id = null;
            self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex) catch break;
        }
        return self.takeNextPromptLocked(alloc);
    }

    pub fn waitAndTakeNextWork(self: *WorkerRuntime, alloc: std.mem.Allocator) !?WorkItem {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        while (((self.queued_prompts.items.len == 0 and
            self.queued_context_compaction == null) or self.turn_start_held) and
            !self.worker_stop_requested)
        {
            self.worker_processing = false;
            self.active_turn_id = 0;
            self.tool_phase_step_id = null;
            self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex) catch break;
        }
        return self.takeNextWorkLocked(alloc);
    }

    /// Nonblocking queue take for single-threaded hosts. Returns null while the
    /// queue is empty, held, or stopped.
    pub fn tryTakeNextPrompt(self: *WorkerRuntime, alloc: std.mem.Allocator) !?QueuedPrompt {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.queued_prompts.items.len == 0 or
            self.turn_start_held or
            self.worker_stop_requested)
        {
            return null;
        }
        return self.takeNextPromptLocked(alloc);
    }

    pub fn tryTakeNextWork(self: *WorkerRuntime, alloc: std.mem.Allocator) !?WorkItem {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if ((self.queued_prompts.items.len == 0 and
            self.queued_context_compaction == null) or self.turn_start_held or
            self.worker_stop_requested)
        {
            return null;
        }
        return self.takeNextWorkLocked(alloc);
    }

    fn takeNextWorkLocked(self: *WorkerRuntime, alloc: std.mem.Allocator) !?WorkItem {
        if (self.worker_stop_requested) return null;
        if (self.queued_context_compaction) |task| {
            self.queued_context_compaction = null;
            self.worker_cancel_requested.store(false, .seq_cst);
            self.worker_recovery_pause_requested.store(false, .seq_cst);
            self.worker_connectivity_wait_active.store(false, .seq_cst);
            self.recovery_continuation_ready = false;
            self.worker_processing = true;
            self.active_turn_id = task.turn_id;
            self.tool_phase_step_id = null;
            self.active_context_compaction = true;
            if (task.operation_id) |id| self.compaction_presentation.running(id, .preparation);
            debug_trace.eventf(
                "worker",
                "context_compaction_begin",
                .{ .turn_id = task.turn_id },
                "remaining_queue={d} cancel_reset=true",
                .{self.queuedWorkCountLocked()},
            );
            return .{ .compact_context = task };
        }
        const prompt = (try self.takeNextPromptLocked(alloc)) orelse return null;
        return .{ .prompt = prompt };
    }

    fn takeNextPromptLocked(self: *WorkerRuntime, alloc: std.mem.Allocator) !?QueuedPrompt {
        if (self.worker_stop_requested or self.queued_prompts.items.len == 0) return null;

        const queued = self.queued_prompts.items[0];
        const transfer_history = self.queued_prompts.items.len == 1;
        const job_history = if (transfer_history)
            self.queued_history
        else
            try session_runtime.snapshotOwnedContextHistory(
                alloc,
                self.queued_history,
                0,
                0,
            );
        errdefer if (!transfer_history) {
            types.freeHistoryTurnSlice(alloc, job_history);
        };
        if (queued.recovery_checkpoint == null and queued.user_prompt_already_presented) {
            try self.worker_events.append(alloc, .{
                .begin_presented_prompt = queued.turn_id,
            });
        } else if (queued.recovery_checkpoint == null) {
            const begin_prompt = try types.dupeUserTurn(alloc, .{ .text = queued.prompt, .images = queued.images });
            errdefer types.freeUserTurn(alloc, begin_prompt);
            if (queued.skill_bindings.len > 0 or queued.skill_display_spans.len > 0) {
                const skill_bindings = try dupeSkillBindings(alloc, queued.skill_bindings);
                errdefer freeSkillBindings(alloc, skill_bindings);
                const skill_display_spans = try dupeSkillDisplaySpans(alloc, queued.skill_display_spans);
                errdefer freeSkillDisplaySpans(alloc, skill_display_spans);
                try self.worker_events.append(alloc, .{
                    .begin_prompt_with_skill_bindings = .{
                        .prompt = begin_prompt,
                        .skill_bindings = skill_bindings,
                        .skill_display_spans = skill_display_spans,
                    },
                });
            } else {
                try self.worker_events.append(alloc, .{ .begin_prompt = begin_prompt });
            }
        }

        var job = self.queued_prompts.orderedRemove(0);
        if (transfer_history) self.queued_history = &.{};
        job.history = job_history;
        self.steering_cancel_turn_id = null;
        self.worker_cancel_requested.store(false, .seq_cst);
        self.worker_recovery_pause_requested.store(false, .seq_cst);
        self.worker_connectivity_wait_active.store(false, .seq_cst);
        self.recovery_continuation_ready = false;
        self.worker_processing = true;
        self.active_turn_id = job.turn_id;
        self.tool_phase_step_id = null;
        if (job.delivery.isContinuation()) {
            for (self.queued_prompts.items) |*prompt| {
                if (!prompt.delivery.isContinuation() or !sameTurnSteeringEligible(prompt.*)) continue;
                prompt.delivery = .{ .active_turn = job.turn_id };
            }
        }
        self.active_context_compaction = false;
        debug_trace.logf(
            "worker",
            "begin prompt bytes={d} remaining_queue={d} fast_mode={s} effort={s}",
            .{ job.prompt.len, self.queuedWorkCountLocked(), if (job.agent_settings.fast_mode) "true" else "false", job.agent_settings.effort.label() },
        );
        debug_trace.eventf(
            "worker",
            "worker_begin",
            .{ .turn_id = job.turn_id },
            "prompt_bytes={d} remaining_queue={d} cancel_reset=true fast_mode={s} effort={s}",
            .{ job.prompt.len, self.queuedWorkCountLocked(), if (job.agent_settings.fast_mode) "true" else "false", job.agent_settings.effort.label() },
        );
        return job;
    }

    pub fn finishProcessing(self: *WorkerRuntime) void {
        debug_trace.logf("worker", "finish processing queued={d}", .{self.queuedPromptCount()});
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        // Close steering admission without moving entries, preserving the exact
        // order in which steering and ordinary prompts were submitted.
        const finished_turn_id = self.active_turn_id;
        self.finishCompactionActivityLocked(finished_turn_id);
        for (self.queued_prompts.items) |*prompt| {
            if (prompt.delivery.activeTurnId() == finished_turn_id) {
                prompt.delivery = .continuation;
                debug_trace.eventf("worker", "steering_deferred_to_next_turn", .{ .turn_id = finished_turn_id }, "queued_turn_id={d} prompt_bytes={d}", .{ prompt.turn_id, prompt.prompt.len });
            }
        }
        self.worker_processing = false;
        self.active_turn_id = 0;
        self.steering_cancel_turn_id = null;
        self.clearActiveToolCallsLocked();
        self.tool_phase_step_id = null;
        self.active_context_compaction = false;
        self.worker_connectivity_wait_active.store(false, .seq_cst);
        self.worker_cond.broadcast(io_mod.getIo());
        self.worker_mutex.unlock(io_mod.getIo());
    }

    /// Marks an agent turn that is executed directly rather than dequeued by
    /// the interactive worker loop. The caller must pair a successful begin
    /// with `finishProcessing`.
    pub fn beginDirectProcessing(self: *WorkerRuntime, turn_id: u64) bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.worker_processing or self.worker_stop_requested) return false;
        self.direct_steering_closed = false;
        self.worker_cancel_requested.store(false, .seq_cst);
        self.worker_recovery_pause_requested.store(false, .seq_cst);
        self.worker_connectivity_wait_active.store(false, .seq_cst);
        self.worker_processing = true;
        self.active_turn_id = turn_id;
        self.tool_phase_step_id = null;
        return true;
    }

    const ActiveToolTransition = union(enum) {
        none,
        started: []u8,
        terminal: []const u8,
        clear,

        fn deinit(self: *ActiveToolTransition, alloc: std.mem.Allocator) void {
            switch (self.*) {
                .started => |call_id| alloc.free(call_id),
                .none, .terminal, .clear => {},
            }
            self.* = .none;
        }
    };

    fn prepareActiveToolTransitionLocked(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        event: WorkerEvent,
    ) !ActiveToolTransition {
        return switch (event) {
            .tool_lifecycle => |lifecycle| switch (lifecycle) {
                .authoritative_started => |started| blk: {
                    if (started.id.turn_id != self.active_turn_id or
                        self.active_tool_calls.contains(started.id.call_id))
                    {
                        break :blk .none;
                    }
                    if (self.active_tool_allocator == null) {
                        self.active_tool_allocator = alloc;
                    }
                    try self.active_tool_calls.ensureUnusedCapacity(alloc, 1);
                    break :blk .{ .started = try alloc.dupe(u8, started.id.call_id) };
                },
                .terminal => |terminal| if (terminal.id.turn_id == self.active_turn_id and
                    self.active_tool_calls.contains(terminal.id.call_id))
                    .{ .terminal = terminal.id.call_id }
                else
                    .none,
                .turn_finished => |finished| if (finished.turn_id == self.active_turn_id) .clear else .none,
                .provisional, .progress => .none,
            },
            else => .none,
        };
    }

    fn applyActiveToolTransitionLocked(
        self: *WorkerRuntime,
        transition: ActiveToolTransition,
    ) void {
        switch (transition) {
            .none => {},
            .started => |call_id| {
                self.active_tool_calls.putAssumeCapacity(call_id, {});
            },
            .terminal => |call_id| {
                const removed = self.active_tool_calls.fetchRemove(call_id) orelse return;
                std.debug.assert(self.active_tool_allocator != null);
                const alloc = self.active_tool_allocator.?;
                alloc.free(removed.key);
            },
            .clear => self.clearActiveToolCallsLocked(),
        }
    }

    fn applyToolBoundaryPhaseLocked(self: *WorkerRuntime, event: WorkerEvent) void {
        switch (event) {
            .turn_phase_update => |update| if (update.turn_id == self.active_turn_id) {
                switch (update.phase) {
                    .running => self.tool_phase_step_id = update.step_id,
                    .waiting_for_subagent => {},
                    .thinking, .generating => if (self.tool_phase_step_id != update.step_id) {
                        self.tool_phase_step_id = null;
                    },
                }
            },
            else => {},
        }
    }

    fn hasActiveToolBoundaryLocked(self: *const WorkerRuntime) bool {
        return self.tool_phase_step_id != null or self.active_tool_calls.count() > 0;
    }

    /// True while a context compaction owns the worker: either the compaction
    /// work item is running, or the active turn is inside a compaction
    /// transaction. Compaction reports through the activity channel rather
    /// than the tool/phase event feed, so it is consulted separately.
    fn compactionInFlightLocked(self: *const WorkerRuntime) bool {
        if (self.active_context_compaction) return true;
        if (self.compaction_presentation.snapshot.operation) |operation| return operation.active();
        return false;
    }

    /// Steering must wait whenever the active work cannot absorb guidance right
    /// now: a running tool or tool-producing step, or a compaction in flight.
    /// Cancelling either would discard in-progress work the user expects to
    /// finish; the message is delivered at the next boundary instead.
    fn hasSteeringWaitBoundaryLocked(self: *const WorkerRuntime) bool {
        return self.hasActiveToolBoundaryLocked() or self.compactionInFlightLocked();
    }

    fn clearActiveToolCallsLocked(self: *WorkerRuntime) void {
        const alloc = self.active_tool_allocator orelse {
            std.debug.assert(self.active_tool_calls.count() == 0);
            return;
        };
        var keys = self.active_tool_calls.keyIterator();
        while (keys.next()) |key| alloc.free(@constCast(key.*));
        self.active_tool_calls.deinit(alloc);
        self.active_tool_calls = .empty;
        self.active_tool_allocator = null;
    }

    pub fn waitUntilIdle(self: *WorkerRuntime) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        while (self.worker_processing) {
            self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex) catch break;
        }
    }

    pub fn isProcessing(self: *WorkerRuntime) bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        return self.worker_processing;
    }

    pub fn snapshotState(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
    ) permission_request.RequestCloneError!StateSnapshot {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        const cancel_requested = self.worker_cancel_requested.load(.seq_cst);
        return .{
            .processing = self.worker_processing,
            .active_turn_id = self.active_turn_id,
            .queued_count = self.queuedWorkCountLocked(),
            .pending_event_count = self.worker_events.items.len,
            .cancel_requested = cancel_requested,
            .cancel_continues_turn = cancel_requested and
                self.steering_cancel_turn_id == self.active_turn_id,
            .pending_permission_request = if (self.permissionRequestAwaitingDecisionLocked()) blk: {
                const request = &self.pending_permission_request_shared.?;
                break :blk try permission_request.OwnedPermissionRequest.dupe(alloc, request.view());
            } else null,
            .pending_permission_review = if (self.permissionRequestAwaitingDecisionLocked())
                if (self.pending_permission_review) |review|
                    .{
                        .request_id = self.pending_permission_request_shared.?.id,
                        .review = review,
                    }
                else
                    null
            else
                null,
        };
    }

    pub fn activeTurnId(self: *WorkerRuntime) u64 {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        return self.active_turn_id;
    }

    /// Returns one coherent, allocator-owned steering presentation snapshot.
    pub fn snapshotSteeringPresentation(self: *WorkerRuntime, alloc: std.mem.Allocator) !SteeringPresentationSnapshot {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        var steering_count: usize = 0;
        for (self.queued_prompts.items) |prompt| {
            if (prompt.delivery.isSteering()) steering_count += 1;
        }
        var snapshot: SteeringPresentationSnapshot = .{
            .waits_for_boundary = steering_count > 0 and self.hasSteeringWaitBoundaryLocked(),
        };
        errdefer snapshot.deinit(alloc);
        snapshot.pending_feedback = pending: {
            var count: usize = 0;
            for (self.worker_events.items) |event| {
                if (event == .append_user_feedback) count += 1;
            }
            if (count == 0) break :pending &.{};
            const feedback = try alloc.alloc([]u8, count);
            errdefer alloc.free(feedback);
            var written: usize = 0;
            errdefer for (feedback[0..written]) |message| alloc.free(message);
            for (self.worker_events.items) |event| {
                if (event == .append_user_feedback) {
                    feedback[written] = try alloc.dupe(u8, event.append_user_feedback);
                    written += 1;
                }
            }
            break :pending feedback;
        };
        if (steering_count == 0) return snapshot;

        const messages = try alloc.alloc([]u8, steering_count);
        var copied: usize = 0;
        errdefer {
            for (messages[0..copied]) |message| alloc.free(message);
            alloc.free(messages);
        }
        for (self.queued_prompts.items) |prompt| {
            if (!prompt.delivery.isSteering()) continue;
            messages[copied] = try alloc.dupe(u8, prompt.prompt);
            copied += 1;
        }
        snapshot.messages = messages;
        return snapshot;
    }

    pub fn queuedPromptCount(self: *WorkerRuntime) usize {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        return self.queuedWorkCountLocked();
    }

    pub fn clearQueuedPrompts(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        retained_images: []const types.ImageAttachment,
    ) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        self.clearQueuedPromptsLocked(alloc, retained_images);
    }

    /// Clears the queue while atomically transferring deletion responsibility
    /// for a matching active or finished turn to the supplied retained images.
    pub fn clearQueuedPromptsForSessionTransition(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        turn_id: u64,
        retained_images: []const types.ImageAttachment,
    ) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        if (retained_images.len > 0) {
            _ = self.preservePromptSnapshotsLocked(turn_id, retained_images);
        }
        self.clearQueuedPromptsLocked(alloc, retained_images);
    }

    fn clearQueuedPromptsLocked(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        retained_images: []const types.ImageAttachment,
    ) void {
        const dropped = self.queuedWorkCountLocked();
        if (dropped > 0) {
            debug_trace.logf("worker", "clear queued prompts dropped={d}", .{dropped});
            debug_trace.eventf("worker", "queued_prompts_cleared", .{}, "dropped={d}", .{dropped});
        }
        for (self.queued_prompts.items) |prompt| {
            discardQueuedPrompt(alloc, prompt, retained_images);
        }
        self.queued_prompts.clearRetainingCapacity();
        types.freeHistoryTurnSlice(alloc, self.queued_history);
        self.queued_history = &.{};
        if (self.queued_context_compaction) |task| {
            if (task.operation_id) |id| self.compaction_presentation.settle(id, .{ .outcome = .cancelled }, io_mod.milliTimestamp());
            freeContextCompactionTask(alloc, task);
            self.queued_context_compaction = null;
        }
        self.worker_cond.broadcast(io_mod.getIo());
    }

    pub fn discardEvents(self: *WorkerRuntime, alloc: std.mem.Allocator) void {
        var events = self.takeEvents();
        defer events.deinit(alloc);
        for (events.items) |event| freeWorkerEvent(alloc, event);
    }

    pub fn syncQueuedPromptModel(self: *WorkerRuntime, alloc: std.mem.Allocator, model: []const u8) !void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        for (self.queued_prompts.items) |*prompt| {
            const next_model = try alloc.dupe(u8, model);
            alloc.free(prompt.model);
            prompt.model = next_model;
        }
    }

    pub fn syncQueuedPromptPermissionSnapshot(self: *WorkerRuntime, snapshot: PermissionSnapshot) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        for (self.queued_prompts.items) |*prompt| {
            prompt.permission_mode = snapshot.mode;
        }
    }

    pub fn syncQueuedPromptFastMode(self: *WorkerRuntime, enabled: bool) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        self.agent_turn_settings.fast_mode = enabled;
        for (self.queued_prompts.items) |*prompt| prompt.agent_settings.fast_mode = enabled;
    }

    pub fn syncQueuedPromptEffort(self: *WorkerRuntime, effort: types.ReasoningEffort) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        self.agent_turn_settings.effort = effort;
        for (self.queued_prompts.items) |*prompt| prompt.agent_settings.effort = effort;
    }

    pub fn setActiveAgentTurnSettings(self: *WorkerRuntime, settings: AgentTurnSettings) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        self.active_agent_turn_settings = settings;
    }

    pub fn clearActiveAgentTurnSettings(self: *WorkerRuntime) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        self.active_agent_turn_settings = null;
    }

    pub fn effectiveAgentTurnSettings(self: *WorkerRuntime) AgentTurnSettings {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        return self.active_agent_turn_settings orelse self.agent_turn_settings;
    }

    pub fn syncQueuedPromptPermissionState(self: *WorkerRuntime, alloc: std.mem.Allocator, grants: []const types.PermissionGrant, snapshot: PermissionSnapshot) !void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        for (self.queued_prompts.items) |*prompt| {
            const next_grants = try types.dupePermissionGrantSlice(alloc, grants);
            types.freePermissionGrantSlice(alloc, prompt.grants);
            prompt.grants = next_grants;
            prompt.permission_mode = snapshot.mode;
        }
    }

    pub fn propagateHistoryTurn(self: *WorkerRuntime, alloc: std.mem.Allocator, turn: types.HistoryTurn, max_history_turns: usize) !void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        try self.propagateHistoryTurnLocked(alloc, turn, max_history_turns, null);
    }

    pub fn commitContextCompaction(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        summary: types.CompactedSummaryHistoryTurn,
        active_prefix: ?types.AssistantHistoryTurn,
        retained_from: ?types.ContextHistoryCut,
        max_history_turns: usize,
    ) !void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        const result = try self.awaitHistoryPublicationLocked(alloc, .{
            .context_compaction = .{
                .turn_id = self.active_turn_id,
                .summary = summary,
                .active_prefix = active_prefix,
                .retained_from = retained_from,
                .max_history_turns = max_history_turns,
            },
        });
        std.debug.assert(result == null);
    }

    pub fn prepareFreshPrompt(self: *WorkerRuntime, alloc: std.mem.Allocator, preparation: FreshPromptPreparation) !FreshPromptHistory {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        var request = preparation;
        request.turn_id = self.active_turn_id;
        return (try self.awaitHistoryPublicationLocked(alloc, .{ .prepare_fresh_prompt = request })) orelse error.InvalidHistoryPublication;
    }

    fn awaitHistoryPublicationLocked(self: *WorkerRuntime, alloc: std.mem.Allocator, event: WorkerEvent) !?FreshPromptHistory {
        if (self.worker_stop_requested or self.isCancelRequested()) return error.Aborted;
        std.debug.assert(self.history_publication_response == .idle);
        const owned_event = try dupeWorkerEvent(alloc, event);
        var owns_event = true;
        defer if (owns_event) freeWorkerEvent(alloc, owned_event);
        try self.worker_events.ensureUnusedCapacity(alloc, 1);
        self.history_publication_response = .{ .waiting = self.active_turn_id };
        defer {
            if (self.history_publication_response == .committed) {
                if (self.history_publication_response.committed) |*snapshot| snapshot.deinit(alloc);
            }
            self.history_publication_response = .idle;
        }
        self.worker_events.appendAssumeCapacity(owned_event);
        owns_event = false;
        self.worker_cond.broadcast(io_mod.getIo());

        while (self.history_publication_response == .waiting and
            !self.worker_stop_requested and !self.isCancelRequested())
        {
            try self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex);
        }
        switch (self.history_publication_response) {
            .committed => |result| {
                self.history_publication_response = .idle;
                return result;
            },
            .failed => |err| return err,
            .waiting, .idle => return error.Aborted,
        }
    }

    pub fn failPendingHistoryPublication(self: *WorkerRuntime, err: anyerror) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.history_publication_response != .waiting) return;
        self.history_publication_response = .{ .failed = err };
        self.worker_cond.broadcast(io_mod.getIo());
    }

    pub fn resolveContextCompaction(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        event: ContextCompaction,
        ctx: *anyopaque,
        handler: ContextCompactionHandler,
    ) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.history_publication_response != .waiting or
            self.history_publication_response.waiting != event.turn_id)
        {
            debug_trace.logf("worker", "compaction delivery dropped reason=stale turn_id={d}", .{event.turn_id});
            return;
        }
        defer self.worker_cond.broadcast(io_mod.getIo());
        if (self.worker_stop_requested or self.isCancelRequested()) {
            self.history_publication_response = .{ .failed = error.Aborted };
            return;
        }
        self.publishContextCompactionLocked(alloc, event, ctx, handler) catch |err| {
            self.history_publication_response = .{ .failed = err };
            return;
        };
        self.history_publication_response = .{ .committed = null };
    }

    pub fn resolveFreshPrompt(self: *WorkerRuntime, alloc: std.mem.Allocator, event: FreshPromptPreparation, ctx: *anyopaque, handler: FreshPromptHandler) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.history_publication_response != .waiting or self.history_publication_response.waiting != event.turn_id) {
            debug_trace.logf("worker", "fresh prompt preparation dropped reason=stale turn_id={d}", .{event.turn_id});
            return;
        }
        defer self.worker_cond.broadcast(io_mod.getIo());
        if (self.worker_stop_requested or self.isCancelRequested()) {
            self.history_publication_response = .{ .failed = error.Aborted };
            return;
        }
        const snapshot = self.publishFreshPromptLocked(alloc, event, ctx, handler) catch |err| {
            self.history_publication_response = .{ .failed = err };
            return;
        };
        self.history_publication_response = .{ .committed = snapshot };
    }

    pub fn publishFreshPrompt(self: *WorkerRuntime, alloc: std.mem.Allocator, event: FreshPromptPreparation, ctx: *anyopaque, handler: FreshPromptHandler) !FreshPromptHistory {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        return self.publishFreshPromptLocked(alloc, event, ctx, handler);
    }

    fn publishFreshPromptLocked(self: *WorkerRuntime, alloc: std.mem.Allocator, event: FreshPromptPreparation, ctx: *anyopaque, handler: FreshPromptHandler) !FreshPromptHistory {
        const prior_turn = event.prior_turn orelse return handler(ctx, event);
        const Publication = struct {
            ctx: *anyopaque,
            handler: FreshPromptHandler,
            event: FreshPromptPreparation,
            result: ?FreshPromptHistory = null,

            fn commit(raw: *anyopaque) !void {
                const self_: *@This() = @ptrCast(@alignCast(raw));
                self_.result = try self_.handler(self_.ctx, self_.event);
            }
        };
        var publication: Publication = .{ .ctx = ctx, .handler = handler, .event = event };
        try self.propagateHistoryTurnLocked(alloc, prior_turn, 0, .{ .ctx = &publication, .commit_fn = Publication.commit });
        return publication.result.?;
    }

    pub fn publishContextCompaction(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        event: ContextCompaction,
        ctx: *anyopaque,
        handler: ContextCompactionHandler,
    ) !void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        try self.publishContextCompactionLocked(alloc, event, ctx, handler);
    }

    fn publishContextCompactionLocked(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        event: ContextCompaction,
        ctx: *anyopaque,
        handler: ContextCompactionHandler,
    ) !void {
        var publication: ContextCompactionPublication = .{ .ctx = ctx, .handler = handler, .event = event };
        try self.propagateHistoryTurnLocked(
            alloc,
            .{ .compacted_summary = event.summary },
            event.max_history_turns,
            .{ .ctx = &publication, .commit_fn = ContextCompactionPublication.commit, .retained_from = event.retained_from },
        );
        if (event.active_prefix != null) {
            if (self.active_prompt_snapshot_ownership) |ownership| {
                _ = ownership.preserve();
            }
        }
    }

    fn propagateHistoryTurnLocked(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        turn: types.HistoryTurn,
        max_history_turns: usize,
        publication: ?HistoryPublication,
    ) !void {
        errdefer |err| if (err == error.SessionPersistenceUncertain) {
            if (self.active_prompt_snapshot_ownership) |ownership| {
                _ = ownership.preserve();
            }
        };
        if (self.queued_prompts.items.len == 0) {
            if (publication) |value| try value.commit();
            return;
        }
        const active_ownership = if (self.active_prompt_snapshot_ownership) |ownership|
            try ownership.ensureSharedOwnership(alloc)
        else
            null;
        const cut = if (publication) |value| value.retained_from else null;
        const next_history = if (cut) |retained_from| try session_runtime.prepareCompactedHistory(alloc, self.queued_history, turn.compacted_summary, retained_from) else try appendHistoryTurnProjection(
            alloc,
            self.queued_history,
            turn,
            max_history_turns,
        );
        var next_history_owned = true;
        defer if (next_history_owned) types.freeHistoryTurnSlice(alloc, next_history);
        const prepared = try alloc.alloc(
            PreparedHistoryPropagation,
            self.queued_prompts.items.len,
        );
        var prepared_count: usize = 0;
        var committed = false;
        defer {
            if (!committed) {
                for (prepared[0..prepared_count]) |entry| entry.deinit(alloc);
            }
            alloc.free(prepared);
        }

        while (prepared_count < self.queued_prompts.items.len) : (prepared_count += 1) {
            const prompt = self.queued_prompts.items[prepared_count];
            const next_image_catalog = try session_runtime.merge_image_catalog_history_turn(
                alloc,
                prompt.authorized_image_catalog,
                turn,
            );
            errdefer types.freeImageAttachmentSlice(alloc, next_image_catalog);
            const next_root_user_intent_context = try auto_classifier_context.refreshQueuedRootUserContext(
                alloc,
                prompt.prompt,
                prompt.root_user_intent_context,
                turn,
            );
            errdefer alloc.free(next_root_user_intent_context);
            const next_snapshot_file_ownerships = if (active_ownership) |ownership| blk: {
                const ownership_count = std.math.add(
                    usize,
                    prompt.snapshot_file_ownerships.len,
                    1,
                ) catch return error.OutOfMemory;
                const ownerships = try alloc.alloc(
                    types.SnapshotFileOwnership,
                    ownership_count,
                );
                std.mem.copyForwards(
                    types.SnapshotFileOwnership,
                    ownerships[0..prompt.snapshot_file_ownerships.len],
                    prompt.snapshot_file_ownerships,
                );
                ownership.retain();
                ownerships[prompt.snapshot_file_ownerships.len] = ownership;
                break :blk ownerships;
            } else null;
            errdefer {
                if (next_snapshot_file_ownerships) |ownerships| {
                    ownerships[ownerships.len - 1].release();
                    alloc.free(ownerships);
                }
            }

            prepared[prepared_count] = .{
                .root_user_intent_context = next_root_user_intent_context,
                .authorized_image_catalog = next_image_catalog,
                .snapshot_file_ownerships = next_snapshot_file_ownerships,
                .added_snapshot_ownership = if (next_snapshot_file_ownerships) |ownerships|
                    ownerships[ownerships.len - 1]
                else
                    null,
            };
        }

        if (publication) |value| try value.commit();
        if (next_history.len <= self.queued_history.len) {
            debug_trace.logf("worker", "queued history replaced by context checkpoint turns={d}", .{self.queued_history.len});
        }
        types.freeHistoryTurnSlice(alloc, self.queued_history);
        self.queued_history = next_history;
        next_history_owned = false;
        for (self.queued_prompts.items, prepared) |*prompt, next| {
            if (turn == .compacted_summary) prompt.unversioned_history_count = 0;
            if (prompt.root_user_intent_context.len > 0) alloc.free(prompt.root_user_intent_context);
            prompt.root_user_intent_context = next.root_user_intent_context;
            types.freeImageAttachmentSlice(alloc, prompt.authorized_image_catalog);
            prompt.authorized_image_catalog = next.authorized_image_catalog;
            if (next.snapshot_file_ownerships) |ownerships| {
                if (prompt.snapshot_file_ownerships.len > 0) {
                    alloc.free(prompt.snapshot_file_ownerships);
                }
                prompt.snapshot_file_ownerships = ownerships;
            }
        }
        committed = true;
    }

    pub fn beginActivePromptSnapshots(
        self: *WorkerRuntime,
        ownership: *ActivePromptSnapshotOwnership,
    ) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        std.debug.assert(self.active_prompt_snapshot_ownership == null);
        if (self.preserve_prompt_snapshot_turn_id == self.active_turn_id) {
            _ = ownership.preserve();
            self.preserve_prompt_snapshot_turn_id = null;
        }
        self.active_prompt_snapshot_ownership = ownership;
    }

    pub fn handoffActivePromptSnapshots(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
    ) !?types.SnapshotFileOwnership {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        const ownership = self.active_prompt_snapshot_ownership orelse return null;
        return ownership.handoff(alloc);
    }

    pub fn endActivePromptSnapshots(
        self: *WorkerRuntime,
        ownership: *ActivePromptSnapshotOwnership,
    ) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        std.debug.assert(self.active_prompt_snapshot_ownership == ownership);
        self.active_prompt_snapshot_ownership = null;
        ownership.deinit();
    }

    pub fn preservePromptSnapshots(self: *WorkerRuntime, turn_id: u64, images: []const types.ImageAttachment) void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        _ = self.preservePromptSnapshotsLocked(turn_id, images);
    }

    fn preservePromptSnapshotsLocked(
        self: *WorkerRuntime,
        turn_id: u64,
        images: []const types.ImageAttachment,
    ) bool {
        if (turn_id == 0 or images.len == 0) return false;

        var preserved = false;
        if (self.active_turn_id == turn_id) {
            if (self.active_prompt_snapshot_ownership) |ownership| {
                if (sameImageSnapshots(ownership.images, images)) {
                    preserved = ownership.preserve();
                }
            } else {
                // The unique active turn bridges queue take to ownership
                // registration while both operations remain mutex-serialized.
                self.preserve_prompt_snapshot_turn_id = turn_id;
                preserved = true;
            }
        }
        for (self.worker_events.items) |*event| {
            if (event.* != .finish_prompt) continue;
            const finished = &event.finish_prompt;
            const finished_images = historyTurnImages(finished.turn);
            if (!sameImageSnapshots(finished_images, images)) continue;
            if (finished.snapshot_file_ownership) |ownership| {
                ownership.transfer();
                preserved = true;
            }
        }
        return preserved;
    }

    pub fn propagateGrant(self: *WorkerRuntime, alloc: std.mem.Allocator, tool_name: []const u8, target_path: []const u8) !void {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        for (self.queued_prompts.items) |*prompt| try appendGrantToQueuedPrompt(alloc, prompt, tool_name, target_path);
    }

    pub fn requestPermissionBlocking(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        request: permission_request.PermissionRequest,
    ) permission_request.RequestCloneError!permission_request.OwnedPermissionResponse {
        return self.requestPermissionBlockingWithReview(alloc, request, null);
    }

    pub fn requestPermissionBlockingWithReview(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        request: permission_request.PermissionRequest,
        review: ?*const diff_mod.FileReview,
    ) permission_request.RequestCloneError!permission_request.OwnedPermissionResponse {
        return self.requestPermissionBlockingObserved(alloc, request, review, null) catch |err| switch (err) {
            error.PermissionRegistrationFailed,
            error.PermissionCapacityExceeded,
            => unreachable,
            else => |request_err| return request_err,
        };
    }

    pub const PermissionRequestObserver = struct {
        context: *anyopaque,
        observe_fn: *const fn (
            *anyopaque,
            *WorkerRuntime,
            permission_request.PermissionRequest,
        ) error{
            OutOfMemory,
            PermissionRegistrationFailed,
            PermissionCapacityExceeded,
        }!void,
    };

    pub fn requestPermissionBlockingObserved(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        request: permission_request.PermissionRequest,
        review: ?*const diff_mod.FileReview,
        observer: ?PermissionRequestObserver,
    ) (permission_request.RequestCloneError || error{
        PermissionRegistrationFailed,
        PermissionCapacityExceeded,
    })!permission_request.OwnedPermissionResponse {
        var shared_request: ?permission_request.OwnedPermissionRequest =
            try permission_request.OwnedPermissionRequest.dupe(
                alloc,
                request,
            );
        errdefer if (shared_request) |*owned| owned.deinit(alloc);

        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        shared_request.?.id = try self.allocatePermissionRequestIdLocked();
        if (self.pending_permission_request_shared) |*old| old.deinit(alloc);
        self.pending_permission_request_shared = shared_request.?;
        shared_request = null;
        self.pending_permission_review = review;
        if (self.pending_permission_response) |response| {
            self.discardPermissionResponse(response, "request_replaced");
        }
        self.pending_permission_response = null;
        self.pending_permission_waiting = true;

        if (observer) |value| {
            value.observe_fn(
                value.context,
                self,
                self.pending_permission_request_shared.?.view(),
            ) catch |err| {
                self.pending_permission_waiting = false;
                self.pending_permission_request_shared.?.deinit(alloc);
                self.pending_permission_request_shared = null;
                self.pending_permission_review = null;
                return err;
            };
        }

        while (self.pending_permission_response == null and !self.worker_stop_requested) {
            self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex) catch break;
        }

        const response = self.pending_permission_response orelse
            permission_request.OwnedPermissionResponse.init(alloc, .deny, null);
        self.pending_permission_waiting = false;
        self.pending_permission_response = null;
        if (self.pending_permission_request_shared) |*old| {
            old.deinit(alloc);
            self.pending_permission_request_shared = null;
        }
        self.pending_permission_review = null;
        return response;
    }

    pub fn submitPermissionResponse(
        self: *WorkerRuntime,
        expected_request_id: u64,
        response: permission_request.OwnedPermissionResponse,
    ) PermissionSubmissionResult {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        if (self.permissionSubmissionBlockedLocked(expected_request_id)) |blocked| {
            self.discardPermissionResponse(response, @tagName(blocked));
            return blocked;
        }
        std.debug.assert(self.resolvePendingPermissionLocked(response));
        return .accepted;
    }

    pub const PermissionCommitError = error{
        OutOfMemory,
        PermissionCommitFailed,
        PermissionCapacityExceeded,
    };

    pub const PermissionCommit = struct {
        context: *anyopaque,
        commit_fn: *const fn (*anyopaque) PermissionCommitError!void,
    };

    /// Keeps canonical pending-request identity stable across a durable host
    /// commit. The response becomes visible and the waiter wakes only after the
    /// callback succeeds. On callback failure the owned response is discarded.
    pub fn submitPermissionResponseAfterCommit(
        self: *WorkerRuntime,
        expected_request_id: u64,
        response: permission_request.OwnedPermissionResponse,
        commit: ?PermissionCommit,
    ) PermissionCommitError!PermissionSubmissionResult {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        if (self.permissionSubmissionBlockedLocked(expected_request_id)) |blocked| {
            self.discardPermissionResponse(response, @tagName(blocked));
            return blocked;
        }
        if (commit) |effect| effect.commit_fn(effect.context) catch |err| {
            self.discardPermissionResponse(response, "commit_failed");
            return err;
        };
        std.debug.assert(self.resolvePendingPermissionLocked(response));
        return .accepted;
    }

    fn permissionSubmissionBlockedLocked(
        self: *const WorkerRuntime,
        expected_request_id: u64,
    ) ?PermissionSubmissionResult {
        if (!self.permissionRequestAwaitedLocked() or
            self.pending_permission_response != null)
        {
            return .no_pending;
        }
        if (self.pending_permission_request_shared.?.id != expected_request_id) {
            return .stale;
        }
        return null;
    }

    fn allocatePermissionRequestIdLocked(
        self: *WorkerRuntime,
    ) error{RequestIdExhausted}!u64 {
        const request_id = self.next_permission_request_id;
        if (request_id == 0) return error.RequestIdExhausted;
        self.next_permission_request_id = std.math.add(
            u64,
            request_id,
            1,
        ) catch 0;
        return request_id;
    }

    fn permissionRequestAwaitedLocked(self: *const WorkerRuntime) bool {
        return self.worker_processing and
            !self.worker_stop_requested and
            self.pending_permission_waiting and
            self.pending_permission_request_shared != null;
    }

    fn permissionRequestAwaitingDecisionLocked(
        self: *const WorkerRuntime,
    ) bool {
        return self.permissionRequestAwaitedLocked() and
            self.pending_permission_response == null;
    }

    fn resolvePendingPermissionLocked(
        self: *WorkerRuntime,
        response: permission_request.OwnedPermissionResponse,
    ) bool {
        if (!self.permissionRequestAwaitingDecisionLocked()) {
            self.discardPermissionResponse(response, "unawaited");
            return false;
        }
        self.pending_permission_response = response;
        self.worker_cond.broadcast(io_mod.getIo());
        return true;
    }

    fn discardPermissionResponse(
        _: *WorkerRuntime,
        response: permission_request.OwnedPermissionResponse,
        reason: []const u8,
    ) void {
        var owned = response;
        if (owned.feedback) |feedback| {
            debug_trace.logf(
                "permission",
                "approval feedback discarded reason={s} bytes={d}",
                .{ reason, feedback.len },
            );
        }
        owned.deinit();
    }

    pub fn requestQuestionBatchAnswerBlocking(self: *WorkerRuntime, alloc: std.mem.Allocator, entries: []const types.QuestionBatchEntry) !?[][]u8 {
        return self.requestQuestionBatchAnswerBlockingWithSource(alloc, entries, .agent_question);
    }

    pub fn requestRouteRecoveryAnswerBlocking(self: *WorkerRuntime, alloc: std.mem.Allocator, entries: []const types.QuestionBatchEntry) !?[][]u8 {
        return self.requestQuestionBatchAnswerBlockingWithSource(alloc, entries, .route_recovery);
    }

    pub fn requestMcpElicitationAnswerBlocking(self: *WorkerRuntime, alloc: std.mem.Allocator, entries: []const types.QuestionBatchEntry) !?[][]u8 {
        return self.requestQuestionBatchAnswerBlockingWithSource(alloc, entries, .mcp_elicitation);
    }

    fn requestQuestionBatchAnswerBlockingWithSource(
        self: *WorkerRuntime,
        alloc: std.mem.Allocator,
        entries: []const types.QuestionBatchEntry,
        source: QuestionPromptSource,
    ) !?[][]u8 {
        const owned = try dupeOwnedQuestionBatch(alloc, entries, source);
        var owns_pending_question = true;
        errdefer if (owns_pending_question) freeOwnedQuestionBatch(alloc, owned);

        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());

        std.debug.assert(self.pending_question_shared == null);
        self.pending_question_shared = owned;
        freeQuestionResponse(alloc, self.pending_question_response);
        self.pending_question_response = .pending;
        self.worker_events.append(alloc, .question_requested) catch |err| {
            self.pending_question_shared = null;
            self.pending_question_response = .pending;
            return err;
        };
        owns_pending_question = false;
        self.worker_cond.broadcast(io_mod.getIo());

        while (self.pending_question_response == .pending and !self.worker_stop_requested) {
            self.worker_cond.wait(io_mod.getIo(), &self.worker_mutex) catch break;
        }

        if (self.pending_question_shared) |pending| {
            freeOwnedQuestionBatch(alloc, pending);
            self.pending_question_shared = null;
        }

        const response = self.pending_question_response;
        self.pending_question_response = .pending;
        return switch (response) {
            .answered => |labels| labels,
            .cancelled, .pending => null,
        };
    }

    pub fn submitQuestionBatchAnswer(self: *WorkerRuntime, alloc: std.mem.Allocator, answers: ?[]const []const u8) !void {
        var new_response: QuestionResponse = .cancelled;
        if (answers) |labels| {
            const dup = try alloc.alloc([]u8, labels.len);
            errdefer alloc.free(dup);
            var filled: usize = 0;
            errdefer {
                var i: usize = 0;
                while (i < filled) : (i += 1) alloc.free(dup[i]);
            }
            while (filled < labels.len) : (filled += 1) dup[filled] = try alloc.dupe(u8, labels[filled]);
            new_response = .{ .answered = dup };
        }

        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.pending_question_shared == null or self.pending_question_response != .pending) {
            freeQuestionResponse(alloc, new_response);
            debug_trace.logf("worker", "ignored late question response pending=false", .{});
            return;
        }
        freeQuestionResponse(alloc, self.pending_question_response);
        self.pending_question_response = new_response;
        self.worker_cond.broadcast(io_mod.getIo());
    }

    pub fn cancelPendingQuestionBatch(self: *WorkerRuntime) bool {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        if (self.pending_question_shared == null or self.pending_question_response != .pending) return false;
        self.pending_question_response = .cancelled;
        self.worker_cond.broadcast(io_mod.getIo());
        return true;
    }

    pub fn snapshotPendingQuestionBatch(self: *WorkerRuntime, alloc: std.mem.Allocator) !?PendingQuestionBatchSnapshot {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        const pending = self.pending_question_shared orelse return null;
        return try dupePendingBatchSnapshot(alloc, pending);
    }

    pub fn pendingQuestionBatchSource(self: *WorkerRuntime) QuestionPromptSource {
        self.worker_mutex.lockUncancelable(io_mod.getIo());
        defer self.worker_mutex.unlock(io_mod.getIo());
        const pending = self.pending_question_shared orelse return .agent_question;
        return pending.source;
    }
};

fn recoveryContinuationAdmission(
    processing: bool,
    queued_count: usize,
    terminal_pause_ready: bool,
) bool {
    return queued_count == 0 and (!processing or terminal_pause_ready);
}

test "recovery continuation admission requires an empty idle or terminally paused worker" {
    try std.testing.expect(recoveryContinuationAdmission(false, 0, false));
    try std.testing.expect(!recoveryContinuationAdmission(true, 0, false));
    try std.testing.expect(recoveryContinuationAdmission(true, 0, true));
    try std.testing.expect(!recoveryContinuationAdmission(false, 1, true));
}

test "terminal recovery pause admits continuation before worker cleanup" {
    const alloc = std.testing.allocator;
    var runtime: WorkerRuntime = .{};
    defer runtime.deinit(alloc);

    const Fixture = struct {
        fn make(allocator: std.mem.Allocator) !QueuedPrompt {
            var prompt = try makePrompt(allocator, "continue preserved turn", "model");
            errdefer freeQueuedPrompt(allocator, prompt);
            prompt.recovery_checkpoint = try (session_codec.RecoveryCheckpoint{
                .turn_id = 41,
                .user = .{ .text = @constCast("unfinished prompt") },
                .assistant_source = @constCast("partial response"),
                .cause = .network_interrupted,
                .action = .paused,
                .authority = .{ .provider = .gateway, .model = @constCast("model") },
                .requested_fast_mode = false,
                .fast_mode = false,
                .max_provider_attempts = 10,
                .consumed_provider_attempts = 10,
            }).dupe(allocator);
            return prompt;
        }
    };

    runtime.worker_processing = true;
    try runtime.pushEvent(alloc, .{ .route_recovery_status = .{
        .kind = .terminal_provider_error,
        .failed_attempt = 10,
        .attempt_limit = 10,
        .cause = .network_interrupted,
        .action = .paused,
        .required_action = .continue_later,
    } });
    try runtime.pushTurnFinished(alloc, .{
        .turn_id = 41,
        .outcome = .paused,
    });

    const continuation = try Fixture.make(alloc);
    runtime.enqueuePrompt(alloc, continuation) catch |err| {
        freeQueuedPrompt(alloc, continuation);
        return err;
    };
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
}

test "connectivity wait state follows worker events before UI presentation" {
    var runtime: WorkerRuntime = .{};
    defer runtime.deinit(std.testing.allocator);

    try runtime.pushEvent(std.testing.allocator, .{ .route_recovery_status = .{
        .kind = .auto_retry,
        .cause = .system_resumed,
        .action = .waiting_for_connectivity,
    } });
    try std.testing.expect(runtime.isConnectivityWaitActive());

    try runtime.pushEvent(std.testing.allocator, .{ .route_recovery_status = .{
        .kind = .auto_retry,
        .cause = .network_interrupted,
        .action = .retrying_request,
    } });
    try std.testing.expect(!runtime.isConnectivityWaitActive());

    try runtime.pushEvent(std.testing.allocator, .{ .route_recovery_status = .{
        .kind = .auto_retry,
        .cause = .system_resumed,
        .action = .waiting_for_connectivity,
    } });
    try runtime.pushEvent(std.testing.allocator, .clear_route_recovery_status);
    try std.testing.expect(!runtime.isConnectivityWaitActive());
}

fn dupeOwnedQuestionEntry(alloc: std.mem.Allocator, entry: types.QuestionBatchEntry) !OwnedQuestionEntry {
    const question_dup = try alloc.dupe(u8, entry.question);
    errdefer alloc.free(question_dup);
    const options_dup = try alloc.alloc(OwnedQuestionOption, entry.options.len);
    errdefer alloc.free(options_dup);

    var filled: usize = 0;
    errdefer {
        var i: usize = 0;
        while (i < filled) : (i += 1) {
            alloc.free(options_dup[i].label);
            if (options_dup[i].description) |d| alloc.free(d);
        }
    }
    while (filled < entry.options.len) : (filled += 1) {
        const src = entry.options[filled];
        const label_dup = try alloc.dupe(u8, src.label);
        errdefer alloc.free(label_dup);
        const desc_dup: ?[]u8 = if (src.description) |d| try alloc.dupe(u8, d) else null;
        options_dup[filled] = .{ .label = label_dup, .description = desc_dup };
    }
    return .{ .question = question_dup, .options = options_dup, .submission = entry.submission };
}

fn freeOwnedQuestionEntry(alloc: std.mem.Allocator, entry: OwnedQuestionEntry) void {
    alloc.free(entry.question);
    for (entry.options) |opt| {
        alloc.free(opt.label);
        if (opt.description) |d| alloc.free(d);
    }
    alloc.free(entry.options);
}

fn dupeOwnedQuestionBatch(
    alloc: std.mem.Allocator,
    entries: []const types.QuestionBatchEntry,
    source: QuestionPromptSource,
) !OwnedQuestionBatch {
    const owned_entries = try alloc.alloc(OwnedQuestionEntry, entries.len);
    errdefer alloc.free(owned_entries);
    var filled: usize = 0;
    errdefer {
        var i: usize = 0;
        while (i < filled) : (i += 1) freeOwnedQuestionEntry(alloc, owned_entries[i]);
    }
    while (filled < entries.len) : (filled += 1) owned_entries[filled] = try dupeOwnedQuestionEntry(alloc, entries[filled]);
    return .{ .entries = owned_entries, .source = source };
}

fn freeOwnedQuestionBatch(alloc: std.mem.Allocator, owned: OwnedQuestionBatch) void {
    for (owned.entries) |entry| freeOwnedQuestionEntry(alloc, entry);
    alloc.free(owned.entries);
}

fn freeOwnedQuestionBatchOpt(alloc: std.mem.Allocator, owned: ?OwnedQuestionBatch) void {
    if (owned) |o| freeOwnedQuestionBatch(alloc, o);
}

fn freeQuestionResponse(alloc: std.mem.Allocator, response: QuestionResponse) void {
    switch (response) {
        .answered => |labels| {
            for (labels) |label| alloc.free(label);
            alloc.free(labels);
        },
        .cancelled, .pending => {},
    }
}

fn dupePendingEntrySnapshot(alloc: std.mem.Allocator, pending: OwnedQuestionEntry) !types.QuestionBatchEntry {
    const question_dup = try alloc.dupe(u8, pending.question);
    errdefer alloc.free(question_dup);
    const options_dup = try alloc.alloc(types.QuestionOption, pending.options.len);
    errdefer alloc.free(options_dup);

    var filled: usize = 0;
    errdefer {
        var i: usize = 0;
        while (i < filled) : (i += 1) {
            alloc.free(@constCast(options_dup[i].label));
            if (options_dup[i].description) |d| alloc.free(@constCast(d));
        }
    }
    while (filled < pending.options.len) : (filled += 1) {
        const src = pending.options[filled];
        const label_dup = try alloc.dupe(u8, src.label);
        errdefer alloc.free(label_dup);
        const desc_dup: ?[]const u8 = if (src.description) |d| try alloc.dupe(u8, d) else null;
        options_dup[filled] = .{ .label = label_dup, .description = desc_dup };
    }
    return .{ .question = question_dup, .options = options_dup, .submission = pending.submission };
}

fn dupePendingBatchSnapshot(alloc: std.mem.Allocator, pending: OwnedQuestionBatch) !PendingQuestionBatchSnapshot {
    const entries_dup = try alloc.alloc(types.QuestionBatchEntry, pending.entries.len);
    errdefer alloc.free(entries_dup);
    var filled: usize = 0;
    errdefer {
        var i: usize = 0;
        while (i < filled) : (i += 1) {
            alloc.free(@constCast(entries_dup[i].question));
            for (entries_dup[i].options) |opt| {
                alloc.free(@constCast(opt.label));
                if (opt.description) |d| alloc.free(@constCast(d));
            }
            alloc.free(@constCast(entries_dup[i].options));
        }
    }
    while (filled < pending.entries.len) : (filled += 1) entries_dup[filled] = try dupePendingEntrySnapshot(alloc, pending.entries[filled]);
    return .{ .entries = entries_dup, .source = pending.source };
}

fn appendGrantToQueuedPrompt(alloc: std.mem.Allocator, prompt: *QueuedPrompt, tool_name: []const u8, target_path: []const u8) !void {
    for (prompt.grants) |grant| {
        if (std.mem.eql(u8, grant.tool_name, tool_name) and std.mem.eql(u8, grant.target_path, target_path)) return;
    }
    const tool_name_dup = try alloc.dupe(u8, tool_name);
    errdefer alloc.free(tool_name_dup);
    const target_path_dup = try alloc.dupe(u8, target_path);
    errdefer alloc.free(target_path_dup);
    const current = prompt.grants;
    const next = try alloc.alloc(types.PermissionGrant, current.len + 1);
    errdefer alloc.free(next);
    if (current.len > 0) std.mem.copyForwards(types.PermissionGrant, next[0..current.len], current);
    next[current.len] = .{ .tool_name = tool_name_dup, .target_path = target_path_dup };
    alloc.free(current);
    prompt.grants = next;
}

fn appendHistoryTurnProjection(
    alloc: std.mem.Allocator,
    current: []const types.HistoryTurn,
    turn: types.HistoryTurn,
    max_history_turns: usize,
) ![]types.HistoryTurn {
    _ = max_history_turns;
    const retained = if (turn == .compacted_summary and
        std.mem.startsWith(u8, turn.compacted_summary.summary, types.context_handoff_open))
        &.{}
    else
        current;
    const next = try alloc.alloc(types.HistoryTurn, retained.len + 1);
    errdefer alloc.free(next);
    var copied: usize = 0;
    errdefer for (next[0..copied]) |owned| types.freeHistoryTurn(alloc, owned);
    for (retained, 0..) |entry, index| {
        next[index] = try types.dupeHistoryTurn(alloc, entry);
        copied += 1;
    }
    next[retained.len] = try types.dupeHistoryTurn(alloc, turn);
    return next;
}

pub fn freeQueuedPrompt(alloc: std.mem.Allocator, prompt: QueuedPrompt) void {
    if (prompt.steering_receipt) |receipt| {
        if (receipt.state.cmpxchgStrong(.queued, .not_applied, .seq_cst, .seq_cst) == null) {
            debug_trace.eventf("subagent", "feedback_not_applied", .{}, "operation={s} reason=queue_discarded", .{receipt.operation_id});
        }
    }
    alloc.free(prompt.prompt);
    types.freeImageAttachmentSlice(alloc, prompt.images);
    types.freeImageAttachmentSlice(alloc, prompt.authorized_image_catalog);
    alloc.free(prompt.model);
    secret.zeroAndFree(alloc, prompt.api_key);
    if (prompt.gateway_team) |team| alloc.free(team);
    if (prompt.account_id) |account_id| alloc.free(account_id);
    types.freeHistoryTurnSlice(alloc, prompt.history);
    if (prompt.root_user_intent_context.len > 0) alloc.free(prompt.root_user_intent_context);
    types.freePermissionGrantSlice(alloc, prompt.grants);
    freeSkillBindings(alloc, prompt.skill_bindings);
    freeSkillDisplaySpans(alloc, prompt.skill_display_spans);
    var context_snapshot = prompt.context_snapshot;
    context_snapshot.deinit(alloc);
    for (prompt.snapshot_file_ownerships) |ownership| ownership.release();
    if (prompt.snapshot_file_ownerships.len > 0) {
        alloc.free(prompt.snapshot_file_ownerships);
    }
    if (prompt.recovery_checkpoint) |checkpoint| {
        var owned = checkpoint;
        owned.deinit(alloc);
    }
}

pub fn freeContextCompactionTask(
    alloc: std.mem.Allocator,
    task: ContextCompactionTask,
) void {
    alloc.free(task.model);
    secret.zeroAndFree(alloc, task.api_key);
    if (task.gateway_team) |team| alloc.free(team);
    if (task.account_id) |account_id| alloc.free(account_id);
    types.freeHistoryTurnSlice(alloc, task.history);
}

pub fn freeWorkItem(alloc: std.mem.Allocator, work: WorkItem) void {
    switch (work) {
        .prompt => |prompt| freeQueuedPrompt(alloc, prompt),
        .compact_context => |task| freeContextCompactionTask(alloc, task),
    }
}

fn discardQueuedPrompt(
    alloc: std.mem.Allocator,
    prompt: QueuedPrompt,
    retained_images: []const types.ImageAttachment,
) void {
    if (prompt.recovery_checkpoint == null) {
        image_attachments.deleteUnreferencedImageSnapshots(prompt.images, retained_images);
    }
    freeQueuedPrompt(alloc, prompt);
}

test "active prompt snapshot ownership discards every pre-transfer boundary" {
    const boundaries = [_][]const u8{
        "preflight_notice",
        "skill_prompt",
        "tool_projection",
        "permission_preparation",
        "process_queued_prompt",
        "postflight_finish",
    };
    for (boundaries, 0..) |_, index| {
        var tmp = std.testing.tmpDir(.{});
        defer tmp.cleanup();
        var name_buffer: [32]u8 = undefined;
        const name = try std.fmt.bufPrint(&name_buffer, "snapshot-{d}.bin", .{index});
        {
            var file = try tmp.dir.createFile(std.testing.io, name, .{});
            defer file.close(std.testing.io);
            try file.writeStreamingAll(std.testing.io, "snapshot");
        }
        const path = try io_mod.dirRealpathAlloc(std.testing.allocator, tmp.dir, name);
        defer std.testing.allocator.free(path);
        const images = [_]types.ImageAttachment{.{
            .id = index + 1,
            .path = @constCast("/tmp/source.png"),
            .media_type = @constCast("image/png"),
            .snapshot_path = path,
            .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
        }};
        var ownership = ActivePromptSnapshotOwnership.init(&images);
        ownership.deinit();
        ownership.deinit();
        try std.testing.expectError(
            error.FileNotFound,
            std.Io.Dir.accessAbsolute(std.testing.io, path, .{}),
        );
    }
}

test "session and checkpoint transfer preserve active and finished prompt snapshots" {
    const alloc = std.testing.allocator;
    const phases = [_]enum { take_before_begin, active, finished }{
        .take_before_begin,
        .active,
        .finished,
    };
    for ([_]bool{ false, true }) |checkpoint_accepted| {
        for (phases) |phase| {
            var tmp = std.testing.tmpDir(.{});
            defer tmp.cleanup();
            const name = switch (phase) {
                .take_before_begin => "take-before-begin.bin",
                .active => "active.bin",
                .finished => "finished.bin",
            };
            {
                var file = try tmp.dir.createFile(std.testing.io, name, .{});
                defer file.close(std.testing.io);
                try file.writeStreamingAll(std.testing.io, "snapshot");
            }
            const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, name);
            defer alloc.free(path);
            const images = [_]types.ImageAttachment{.{
                .id = 1,
                .path = @constCast("/tmp/source.png"),
                .media_type = @constCast("image/png"),
                .snapshot_path = path,
                .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
            }};

            var runtime = WorkerRuntime{};
            defer runtime.deinit(alloc);
            runtime.active_turn_id = 41;
            var active = ActivePromptSnapshotOwnership.init(&images);
            if (phase != .take_before_begin) runtime.beginActivePromptSnapshots(&active);
            if (phase == .finished) {
                const ownership = (try runtime.handoffActivePromptSnapshots(alloc)) orelse
                    return error.TestExpectedSnapshotOwnership;
                try runtime.pushEvent(alloc, .{ .finish_prompt = .{
                    .turn = .{ .assistant = .{
                        .user = .{ .text = @constCast("prompt"), .images = @constCast(&images) },
                        .assistant = @constCast("done"),
                    } },
                    .snapshot_file_ownership = ownership,
                } });
                ownership.release();
                runtime.endActivePromptSnapshots(&active);
                runtime.active_turn_id = 0;
            }

            if (checkpoint_accepted) {
                runtime.preservePromptSnapshots(41, &images);
            } else {
                runtime.clearQueuedPromptsForSessionTransition(alloc, 41, &images);
            }
            if (phase == .take_before_begin) runtime.beginActivePromptSnapshots(&active);
            if (phase != .finished) runtime.endActivePromptSnapshots(&active);
            runtime.discardEvents(alloc);
            try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});
            try std.Io.Dir.deleteFileAbsolute(std.testing.io, path);
        }
    }
}

test "compaction activity survives worker finish without allocating a permission snapshot" {
    var runtime: WorkerRuntime = .{};
    defer runtime.deinit(std.testing.allocator);
    try std.testing.expect(runtime.beginDirectProcessing(17));
    const first = runtime.beginCompactionActivity(.automatic, 17);
    runtime.runCompactionActivity(first, .publication);
    runtime.requestInteractiveCancel();
    try std.testing.expect(runtime.compactionActivitySnapshot().operation.?.phase == .stopping);
    runtime.settleCompactionActivity(first, .{ .outcome = .succeeded, .stage = .publication, .publication = .committed });
    runtime.finishProcessing();
    try std.testing.expectEqual(@as(u64, 0), runtime.activeTurnId());
    const terminal = runtime.compactionActivitySnapshot();
    try std.testing.expectEqual(first, terminal.operation.?.id);
    try std.testing.expectEqual(compaction_activity.Outcome.succeeded, terminal.operation.?.phase.terminal.outcome);
    runtime.settleCompactionActivity(first, .{ .outcome = .cancelled });
    try std.testing.expectEqualDeep(terminal, runtime.compactionActivitySnapshot());

    const next = runtime.beginCompactionActivity(.manual, null);
    const started = runtime.compactionActivitySnapshot();
    runtime.rejectCompactionActivity(.busy);
    try std.testing.expectEqualDeep(started, runtime.compactionActivitySnapshot());
    try std.testing.expect(!runtime.dismissCompactionActivity(first, terminal.revision));
    runtime.settleCompactionActivity(first, compaction_activity.failure(error.Aborted, .publication, true));
    try std.testing.expectEqual(next, runtime.compactionActivitySnapshot().operation.?.id);
    runtime.settleCompactionActivity(next, compaction_activity.failure(error.OutOfMemory, .preparation, false));
    try std.testing.expectEqual(error.OutOfMemory, runtime.compactionActivitySnapshot().operation.?.phase.terminal.err.?);
}

test "manual compaction is a typed worker item and not a prompt" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    const task = ContextCompactionTask{
        .model = try alloc.dupe(u8, "provider/model"),
        .api_key = try alloc.dupe(u8, "key"),
        .history = try alloc.alloc(types.HistoryTurn, 0),
    };
    try runtime.enqueueContextCompaction(task);

    const taken = (try runtime.tryTakeNextWork(alloc)) orelse
        return error.TestExpectedQueuedWork;
    defer freeWorkItem(alloc, taken);
    try std.testing.expect(taken == .compact_context);
    const snapshot = runtime.compactionActivitySnapshot();
    try std.testing.expectEqual(taken.compact_context.operation_id.?, snapshot.operation.?.id);
    try std.testing.expectEqual(taken.compact_context.turn_id, snapshot.operation.?.turn_id.?);
    try std.testing.expect(snapshot.operation.?.phase == .running);
    runtime.finishProcessing();
    try std.testing.expectEqual(error.CompactionInterrupted, runtime.compactionActivitySnapshot().operation.?.phase.terminal.err.?);
}

test "session transition queue clearing settles only the dropped compaction identity" {
    const alloc = std.testing.allocator;
    for ([_]bool{ false, true }) |retain_snapshots| {
        for ([_]bool{ false, true }) |newer_activity| {
            var runtime: WorkerRuntime = .{};
            defer runtime.deinit(alloc);
            try runtime.enqueueContextCompaction(.{
                .turn_id = 41,
                .model = try alloc.dupe(u8, "provider/model"),
                .api_key = try alloc.dupe(u8, "key"),
                .history = try alloc.alloc(types.HistoryTurn, 0),
            });
            const queued = runtime.compactionActivitySnapshot();
            if (newer_activity) _ = runtime.beginCompactionActivity(.manual, null);
            const observed = runtime.compactionActivitySnapshot();

            if (retain_snapshots) {
                runtime.clearQueuedPromptsForSessionTransition(alloc, 41, &.{});
            } else {
                runtime.clearQueuedPrompts(alloc, &.{});
            }
            try std.testing.expectEqual(ContextCompactionStatus.idle, runtime.contextCompactionStatus());
            try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
            try std.testing.expect((try runtime.tryTakeNextWork(alloc)) == null);
            try std.testing.expect(!runtime.isCancelRequested());
            const settled = runtime.compactionActivitySnapshot();
            if (newer_activity) {
                try std.testing.expectEqualDeep(observed, settled);
            } else {
                try std.testing.expectEqual(queued.operation.?.id, settled.operation.?.id);
                try std.testing.expectEqual(compaction_activity.Outcome.cancelled, settled.operation.?.phase.terminal.outcome);
                try std.testing.expect(!settled.operation.?.active());
                try std.testing.expect(settled.revision > queued.revision);
            }
            runtime.clearQueuedPrompts(alloc, &.{});
            try std.testing.expectEqualDeep(settled, runtime.compactionActivitySnapshot());
        }
    }
}

test "queued manual compaction can be cancelled before worker execution" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    const task = ContextCompactionTask{
        .model = try alloc.dupe(u8, "provider/model"),
        .api_key = try alloc.dupe(u8, "key"),
        .history = try alloc.alloc(types.HistoryTurn, 0),
    };
    try runtime.enqueueContextCompaction(task);

    try std.testing.expectEqual(ContextCompactionStatus.queued, runtime.contextCompactionStatus());
    try std.testing.expect(runtime.cancelContextCompaction(alloc));
    try std.testing.expectEqual(ContextCompactionStatus.idle, runtime.contextCompactionStatus());
    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
    try std.testing.expect((try runtime.tryTakeNextWork(alloc)) == null);
    try std.testing.expectEqual(compaction_activity.Outcome.cancelled, runtime.compactionActivitySnapshot().operation.?.phase.terminal.outcome);
}

test "running manual compaction cancellation uses the worker cancel flag" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    const task = ContextCompactionTask{
        .model = try alloc.dupe(u8, "provider/model"),
        .api_key = try alloc.dupe(u8, "key"),
        .history = try alloc.alloc(types.HistoryTurn, 0),
    };
    try runtime.enqueueContextCompaction(task);
    const taken = (try runtime.tryTakeNextWork(alloc)) orelse
        return error.TestExpectedQueuedWork;
    defer freeWorkItem(alloc, taken);

    try std.testing.expectEqual(ContextCompactionStatus.running, runtime.contextCompactionStatus());
    try std.testing.expect(runtime.cancelContextCompaction(alloc));
    try std.testing.expect(runtime.isCancelRequested());
    try std.testing.expect(runtime.compactionActivitySnapshot().operation.?.phase == .stopping);
    runtime.finishProcessing();
    try std.testing.expectEqual(@as(u64, 0), runtime.activeTurnId());
    try std.testing.expectEqual(compaction_activity.Outcome.cancelled, runtime.compactionActivitySnapshot().operation.?.phase.terminal.outcome);
    try std.testing.expectEqual(ContextCompactionStatus.idle, runtime.contextCompactionStatus());
}

test "session transfer discards mismatched active prompt snapshots" {
    const alloc = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    {
        var file = try tmp.dir.createFile(std.testing.io, "active.bin", .{});
        defer file.close(std.testing.io);
        try file.writeStreamingAll(std.testing.io, "snapshot");
    }
    const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "active.bin");
    defer alloc.free(path);
    const active_images = [_]types.ImageAttachment{.{
        .id = 1,
        .path = @constCast("/tmp/source.png"),
        .media_type = @constCast("image/png"),
        .snapshot_path = path,
        .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
    }};
    const unrelated_images = [_]types.ImageAttachment{.{
        .id = 1,
        .path = @constCast("/tmp/other.png"),
        .media_type = @constCast("image/png"),
        .snapshot_path = @constCast("/tmp/other.bin"),
        .snapshot_sha256 = @constCast("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
    }};

    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.active_turn_id = 42;
    var active = ActivePromptSnapshotOwnership.init(&active_images);
    runtime.beginActivePromptSnapshots(&active);

    runtime.clearQueuedPromptsForSessionTransition(alloc, 42, &unrelated_images);
    runtime.endActivePromptSnapshots(&active);

    try std.testing.expectError(
        error.FileNotFound,
        std.Io.Dir.accessAbsolute(std.testing.io, path, .{}),
    );
}

test "accepted active prompt ownership does not delete history files" {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    {
        var first = try tmp.dir.createFile(std.testing.io, "first.bin", .{});
        defer first.close(std.testing.io);
        try first.writeStreamingAll(std.testing.io, "first");
    }
    {
        var second = try tmp.dir.createFile(std.testing.io, "second.bin", .{});
        defer second.close(std.testing.io);
        try second.writeStreamingAll(std.testing.io, "second");
    }
    const first_path = try io_mod.dirRealpathAlloc(std.testing.allocator, tmp.dir, "first.bin");
    defer std.testing.allocator.free(first_path);
    const second_path = try io_mod.dirRealpathAlloc(std.testing.allocator, tmp.dir, "second.bin");
    defer std.testing.allocator.free(second_path);
    const images = [_]types.ImageAttachment{
        .{
            .id = 1,
            .path = @constCast("/tmp/first.png"),
            .media_type = @constCast("image/png"),
            .snapshot_path = first_path,
            .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
        },
        .{
            .id = 2,
            .path = @constCast("/tmp/second.png"),
            .media_type = @constCast("image/png"),
            .snapshot_path = second_path,
            .snapshot_sha256 = @constCast("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
        },
    };
    var ownership = ActivePromptSnapshotOwnership.init(&images);
    const accepted = (try ownership.handoff(std.testing.allocator)) orelse
        return error.TestExpectedEqual;
    accepted.transfer();
    accepted.release();
    ownership.deinit();

    try std.Io.Dir.accessAbsolute(std.testing.io, first_path, .{});
    try std.Io.Dir.accessAbsolute(std.testing.io, second_path, .{});
}

test "propagated queued snapshots survive finish failure until the last borrower releases" {
    const alloc = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    {
        var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{});
        defer file.close(std.testing.io);
        try file.writeStreamingAll(std.testing.io, "snapshot");
    }
    const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin");
    defer alloc.free(path);
    var images = [_]types.ImageAttachment{.{
        .id = 1,
        .path = @constCast("/tmp/source.png"),
        .media_type = @constCast("image/png"),
        .snapshot_path = path,
        .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
    }};
    const turn: types.HistoryTurn = .{ .assistant = .{
        .user = .{ .text = @constCast("active"), .images = &images },
        .assistant = @constCast("done"),
    } };

    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "first queued", "model"));
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "second queued", "model"));

    var active = ActivePromptSnapshotOwnership.init(&images);
    runtime.beginActivePromptSnapshots(&active);
    try runtime.propagateHistoryTurn(alloc, turn, 8);
    const failed_finish_ownership =
        (try runtime.handoffActivePromptSnapshots(alloc)) orelse
        return error.TestExpectedEqual;
    failed_finish_ownership.release();
    runtime.endActivePromptSnapshots(&active);

    for (runtime.queued_prompts.items) |prompt| {
        try std.testing.expectEqual(@as(usize, 1), prompt.authorized_image_catalog.len);
        try std.testing.expectEqual(@as(usize, 1), prompt.snapshot_file_ownerships.len);
        try std.Io.Dir.accessAbsolute(
            std.testing.io,
            prompt.authorized_image_catalog[0].snapshot_path.?,
            .{},
        );
    }

    const first = (try runtime.tryTakeNextPrompt(alloc)).?;
    freeQueuedPrompt(alloc, first);
    try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});

    runtime.clearQueuedPrompts(alloc, &.{});
    try std.testing.expectError(
        error.FileNotFound,
        std.Io.Dir.accessAbsolute(std.testing.io, path, .{}),
    );
}

test "accepted history prevents propagated queue cleanup from deleting snapshots" {
    const alloc = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    {
        var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{});
        defer file.close(std.testing.io);
        try file.writeStreamingAll(std.testing.io, "snapshot");
    }
    const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin");
    defer alloc.free(path);
    var images = [_]types.ImageAttachment{.{
        .id = 1,
        .path = @constCast("/tmp/source.png"),
        .media_type = @constCast("image/png"),
        .snapshot_path = path,
        .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
    }};
    const turn: types.HistoryTurn = .{ .assistant = .{
        .user = .{ .text = @constCast("active"), .images = &images },
        .assistant = @constCast("done"),
    } };

    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model"));

    var active = ActivePromptSnapshotOwnership.init(&images);
    runtime.beginActivePromptSnapshots(&active);
    try runtime.propagateHistoryTurn(alloc, turn, 8);
    const accepted =
        (try runtime.handoffActivePromptSnapshots(alloc)) orelse
        return error.TestExpectedEqual;
    accepted.transfer();
    accepted.release();
    runtime.endActivePromptSnapshots(&active);
    runtime.clearQueuedPrompts(alloc, &.{});

    try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});
}

test "compaction preserves active snapshots only after persistence succeeds" {
    const Persist = struct {
        fail: bool,

        fn commit(raw: *anyopaque, _: ContextCompaction) !void {
            const self: *@This() = @ptrCast(@alignCast(raw));
            if (self.fail) return error.TestPersistenceFailure;
        }
    };
    const alloc = std.testing.allocator;
    for ([_]bool{ false, true }) |queued| {
        for ([_]bool{ false, true }) |fail| {
            var tmp = std.testing.tmpDir(.{});
            defer tmp.cleanup();
            {
                var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{});
                defer file.close(std.testing.io);
                try file.writeStreamingAll(std.testing.io, "snapshot");
            }
            const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin");
            defer alloc.free(path);
            var images = [_]types.ImageAttachment{.{
                .id = 1,
                .path = @constCast("/tmp/source.png"),
                .media_type = @constCast("image/png"),
                .snapshot_path = path,
                .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
            }};
            var runtime: WorkerRuntime = .{};
            defer runtime.deinit(alloc);
            if (queued) try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model"));
            var ownership = ActivePromptSnapshotOwnership.init(&images);
            runtime.beginActivePromptSnapshots(&ownership);
            var active = true;
            defer if (active) runtime.endActivePromptSnapshots(&ownership);
            var persist: Persist = .{ .fail = fail };
            const event: ContextCompaction = .{
                .summary = .{
                    .summary = @constCast("<context_handoff>image checkpoint</context_handoff>"),
                    .removed_turn_count = 1,
                    .compaction_count = 1,
                },
                .active_prefix = .{
                    .user = .{ .text = @constCast("inspect image"), .images = &images },
                    .assistant = @constCast(""),
                },
            };
            if (fail) {
                try std.testing.expectError(error.TestPersistenceFailure, runtime.publishContextCompaction(alloc, event, &persist, Persist.commit));
            } else {
                try runtime.publishContextCompaction(alloc, event, &persist, Persist.commit);
                try runtime.publishContextCompaction(alloc, event, &persist, Persist.commit);
            }
            runtime.endActivePromptSnapshots(&ownership);
            active = false;
            runtime.clearQueuedPrompts(alloc, &.{});
            if (fail) {
                try std.testing.expectError(error.FileNotFound, std.Io.Dir.accessAbsolute(std.testing.io, path, .{}));
            } else {
                try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});
                try std.testing.expect(ownership.shared_ownership == null);
            }
        }
    }
}

test "failed queued prompt dequeue retains borrowed snapshot ownership" {
    const alloc = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    {
        var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{});
        defer file.close(std.testing.io);
        try file.writeStreamingAll(std.testing.io, "snapshot");
    }
    const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin");
    defer alloc.free(path);
    var images = [_]types.ImageAttachment{.{
        .id = 1,
        .path = @constCast("/tmp/source.png"),
        .media_type = @constCast("image/png"),
        .snapshot_path = path,
        .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
    }};
    const turn: types.HistoryTurn = .{ .assistant = .{
        .user = .{ .text = @constCast("active"), .images = &images },
        .assistant = @constCast("done"),
    } };

    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model"));

    var active = ActivePromptSnapshotOwnership.init(&images);
    runtime.beginActivePromptSnapshots(&active);
    try runtime.propagateHistoryTurn(alloc, turn, 8);
    const failed_finish_ownership =
        (try runtime.handoffActivePromptSnapshots(alloc)) orelse
        return error.TestExpectedEqual;
    failed_finish_ownership.release();
    runtime.endActivePromptSnapshots(&active);

    var failing = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 0 });
    try std.testing.expectError(
        error.OutOfMemory,
        runtime.waitAndTakeNextPrompt(failing.allocator()),
    );
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
    try std.testing.expectEqual(
        @as(usize, 1),
        runtime.queued_prompts.items[0].snapshot_file_ownerships.len,
    );
    try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});

    runtime.clearQueuedPrompts(alloc, &.{});
    try std.testing.expectError(
        error.FileNotFound,
        std.Io.Dir.accessAbsolute(std.testing.io, path, .{}),
    );
}

fn checkHistoryPropagationAllocation(
    alloc: std.mem.Allocator,
    snapshot_path: []const u8,
    fail_index: usize,
) !bool {
    {
        var file = try std.Io.Dir.createFileAbsolute(
            std.testing.io,
            snapshot_path,
            .{ .truncate = true },
        );
        defer file.close(std.testing.io);
        try file.writeStreamingAll(std.testing.io, "snapshot");
    }
    var images = [_]types.ImageAttachment{.{
        .id = 1,
        .path = @constCast("/tmp/source.png"),
        .media_type = @constCast("image/png"),
        .snapshot_path = @constCast(snapshot_path),
        .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
    }};
    const turn: types.HistoryTurn = .{ .assistant = .{
        .user = .{ .text = @constCast("active"), .images = &images },
        .assistant = @constCast("done"),
    } };

    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    const first = try makePrompt(alloc, "first queued", "model");
    var owns_first = true;
    errdefer if (owns_first) freeQueuedPrompt(alloc, first);
    try runtime.enqueuePrompt(alloc, first);
    owns_first = false;
    const second = try makePrompt(alloc, "second queued", "model");
    var owns_second = true;
    errdefer if (owns_second) freeQueuedPrompt(alloc, second);
    try runtime.enqueuePrompt(alloc, second);
    owns_second = false;

    var active = ActivePromptSnapshotOwnership.init(&images);
    runtime.beginActivePromptSnapshots(&active);
    defer runtime.endActivePromptSnapshots(&active);

    var failing = std.testing.FailingAllocator.init(
        alloc,
        .{ .fail_index = fail_index },
    );
    runtime.propagateHistoryTurn(failing.allocator(), turn, 8) catch |err| {
        if (!failing.has_induced_failure) return err;
        for (runtime.queued_prompts.items) |prompt| {
            try std.testing.expectEqual(@as(usize, 0), prompt.history.len);
            try std.testing.expectEqual(@as(usize, 0), prompt.authorized_image_catalog.len);
            try std.testing.expectEqual(@as(usize, 0), prompt.snapshot_file_ownerships.len);
        }
        try std.testing.expectEqual(@as(usize, 0), runtime.queued_history.len);
        return false;
    };

    for (runtime.queued_prompts.items) |prompt| {
        try std.testing.expectEqual(@as(usize, 0), prompt.history.len);
        try std.testing.expectEqual(@as(usize, 1), prompt.authorized_image_catalog.len);
        try std.testing.expectEqual(@as(usize, 1), prompt.snapshot_file_ownerships.len);
    }
    try std.testing.expectEqual(@as(usize, 1), runtime.queued_history.len);
    return true;
}

test "multi-queue history propagation is allocation-failure atomic" {
    const alloc = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const root = try io_mod.dirRealpathAlloc(alloc, tmp.dir, ".");
    defer alloc.free(root);
    const snapshot_path = try std.fs.path.join(alloc, &.{ root, "snapshot.bin" });
    defer alloc.free(snapshot_path);

    var fail_index: usize = 0;
    while (fail_index < 256) : (fail_index += 1) {
        if (try checkHistoryPropagationAllocation(alloc, snapshot_path, fail_index)) {
            return;
        }
    }
    return error.TestUnexpectedResult;
}

fn checkContextCompactionCommitAllocation(
    alloc: std.mem.Allocator,
    fail_index: usize,
) !bool {
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    var prompt = try makePrompt(alloc, "queued", "model");
    var owns_prompt = true;
    errdefer if (owns_prompt) freeQueuedPrompt(alloc, prompt);
    prompt.history = try session_runtime.snapshotOwnedContextHistory(alloc, &.{.{ .assistant = .{
        .user = .{ .text = @constCast("prior user") },
        .assistant = @constCast("prior answer"),
    } }}, 0, 0);
    try runtime.enqueuePrompt(alloc, prompt);
    owns_prompt = false;

    const turn: types.HistoryTurn = .{ .compacted_summary = .{
        .summary = @constCast("<context_handoff>checkpoint</context_handoff>"),
        .removed_turn_count = 1,
        .compaction_count = 1,
    } };
    var failing = std.testing.FailingAllocator.init(
        alloc,
        .{ .fail_index = fail_index },
    );
    runtime.publishContextCompaction(failing.allocator(), .{ .summary = turn.compacted_summary, .max_history_turns = 8 }, &runtime, discardCompactionForTest) catch |err| {
        if (!failing.has_induced_failure) return err;
        try std.testing.expectEqual(
            @as(usize, 1),
            runtime.queued_history.len,
        );
        try std.testing.expectEqualStrings("prior answer", runtime.queued_history[0].assistant.assistant);
        try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);
        return false;
    };

    try std.testing.expectEqual(
        @as(usize, 1),
        runtime.queued_history.len,
    );
    try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);
    try std.testing.expectEqualStrings(turn.compacted_summary.summary, runtime.queued_history[0].compacted_summary.summary);
    return true;
}

fn discardCompactionForTest(_: *anyopaque, _: ContextCompaction) !void {}

test "context compaction callback waits for UI persistence" {
    const Probe = struct {
        returned: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
        failure: ?anyerror = null,

        fn run(self: *@This(), runtime: *WorkerRuntime) void {
            runtime.commitContextCompaction(std.testing.allocator, .{
                .summary = @constCast("<context_handoff>checkpoint</context_handoff>"),
                .removed_turn_count = 1,
                .compaction_count = 1,
            }, null, null, 8) catch |err| {
                self.failure = err;
            };
            self.returned.store(true, .release);
        }
    };
    var runtime: WorkerRuntime = .{};
    defer runtime.deinit(std.testing.allocator);
    var probe: Probe = .{};
    const thread = try std.Thread.spawn(.{}, Probe.run, .{ &probe, &runtime });
    defer {
        runtime.requestShutdown();
        thread.join();
    }
    const deadline = io_mod.milliTimestamp() + 5_000;
    while (true) {
        var state = try runtime.snapshotState(std.testing.allocator);
        defer state.deinit(std.testing.allocator);
        if (state.pending_event_count > 0) break;
        if (io_mod.milliTimestamp() >= deadline) return error.TestExpectedCompactionEvent;
        io_mod.sleep(std.time.ns_per_ms);
    }
    io_mod.sleep(10 * std.time.ns_per_ms);
    try std.testing.expect(!probe.returned.load(.acquire));
}

test "context compaction acknowledgment commits or fails before worker continuation" {
    const Probe = struct {
        const Outcome = enum { commit, fail, cancel_after_commit };
        runtime: *WorkerRuntime,
        outcome: Outcome,
        returned: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
        continued: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
        failure: ?anyerror = null,
        persisted: bool = false,

        fn run(self: *@This()) void {
            self.runtime.commitContextCompaction(std.testing.allocator, .{
                .summary = @constCast("<context_handoff>checkpoint</context_handoff>"),
                .removed_turn_count = 1,
                .compaction_count = 1,
            }, .{
                .user = .{ .text = @constCast("active request") },
                .assistant = @constCast(""),
            }, null, 8) catch |err| {
                self.failure = err;
                self.returned.store(true, .release);
                return;
            };
            self.continued.store(true, .release);
            self.returned.store(true, .release);
        }

        fn commit(raw: *anyopaque, event: ContextCompaction) !void {
            const self: *@This() = @ptrCast(@alignCast(raw));
            try std.testing.expectEqualStrings("active request", event.active_prefix.?.user.text);
            try std.testing.expectEqualStrings("old answer", self.runtime.queued_history[0].assistant.assistant);
            if (self.outcome == .fail) return error.TestPersistenceFailure;
            self.persisted = true;
            if (self.outcome == .cancel_after_commit) {
                self.runtime.worker_cancel_requested.store(true, .seq_cst);
            }
        }
    };
    const alloc = std.testing.allocator;
    for ([_]Probe.Outcome{ .commit, .fail, .cancel_after_commit }) |outcome| {
        var runtime: WorkerRuntime = .{};
        defer runtime.deinit(alloc);
        var prompt = try makePrompt(alloc, "queued", "model");
        prompt.history = try session_runtime.snapshotOwnedContextHistory(alloc, &.{.{ .assistant = .{
            .user = .{ .text = @constCast("old request") },
            .assistant = @constCast("old answer"),
        } }}, 0, 0);
        try runtime.enqueuePrompt(alloc, prompt);
        var probe: Probe = .{ .runtime = &runtime, .outcome = outcome };
        const thread = try std.Thread.spawn(.{}, Probe.run, .{&probe});
        var joined = false;
        defer if (!joined) {
            runtime.requestShutdown();
            thread.join();
        };
        const deadline = io_mod.milliTimestamp() + 5_000;
        while (true) {
            var state = try runtime.snapshotState(alloc);
            defer state.deinit(alloc);
            if (state.pending_event_count > 0) break;
            if (io_mod.milliTimestamp() >= deadline) return error.TestExpectedCompactionEvent;
            io_mod.sleep(std.time.ns_per_ms);
        }
        var events = runtime.takeEvents();
        defer events.deinit(alloc);
        defer for (events.items) |event| freeWorkerEvent(alloc, event);
        try std.testing.expectEqual(@as(usize, 1), events.items.len);
        try std.testing.expect(!probe.returned.load(.acquire));
        try std.testing.expect(!probe.continued.load(.acquire));
        runtime.resolveContextCompaction(alloc, events.items[0].context_compaction, &probe, Probe.commit);
        thread.join();
        joined = true;
        if (outcome == .fail) {
            try std.testing.expectEqual(error.TestPersistenceFailure, probe.failure.?);
            try std.testing.expect(!probe.persisted);
            try std.testing.expect(!probe.continued.load(.acquire));
            try std.testing.expectEqualStrings("old answer", runtime.queued_history[0].assistant.assistant);
        } else {
            try std.testing.expect(probe.failure == null);
            try std.testing.expect(probe.persisted);
            try std.testing.expect(probe.continued.load(.acquire));
            try std.testing.expect(runtime.queued_history[0] == .compacted_summary);
        }
    }
}

test "context compaction waiter exits on cancellation and rejects stale delivery" {
    const Probe = struct {
        returned: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
        failure: ?anyerror = null,
        commits: usize = 0,

        fn run(self: *@This(), runtime: *WorkerRuntime) void {
            runtime.commitContextCompaction(std.testing.allocator, .{
                .summary = @constCast("<context_handoff>checkpoint</context_handoff>"),
                .removed_turn_count = 1,
                .compaction_count = 1,
            }, null, null, 8) catch |err| {
                self.failure = err;
            };
            self.returned.store(true, .release);
        }

        fn commit(raw: *anyopaque, _: ContextCompaction) !void {
            const self: *@This() = @ptrCast(@alignCast(raw));
            self.commits += 1;
        }
    };
    const alloc = std.testing.allocator;
    for ([_]bool{ false, true }) |shutdown| {
        var runtime: WorkerRuntime = .{};
        defer runtime.deinit(alloc);
        var probe: Probe = .{};
        const thread = try std.Thread.spawn(.{}, Probe.run, .{ &probe, &runtime });
        var joined = false;
        defer if (!joined) {
            runtime.requestShutdown();
            thread.join();
        };
        const deadline = io_mod.milliTimestamp() + 5_000;
        while (true) {
            var state = try runtime.snapshotState(alloc);
            defer state.deinit(alloc);
            if (state.pending_event_count > 0) break;
            if (io_mod.milliTimestamp() >= deadline) return error.TestExpectedCompactionEvent;
            io_mod.sleep(std.time.ns_per_ms);
        }
        var events = runtime.takeEvents();
        defer events.deinit(alloc);
        defer for (events.items) |event| freeWorkerEvent(alloc, event);
        if (shutdown) runtime.requestShutdown() else runtime.requestInteractiveCancel();
        while (!probe.returned.load(.acquire)) {
            if (io_mod.milliTimestamp() >= deadline) return error.TestExpectedCompactionCancellation;
            io_mod.sleep(std.time.ns_per_ms);
        }
        thread.join();
        joined = true;
        try std.testing.expectEqual(error.Aborted, probe.failure.?);
        runtime.resolveContextCompaction(alloc, events.items[0].context_compaction, &probe, Probe.commit);
        try std.testing.expectEqual(@as(usize, 0), probe.commits);
    }
}

test "context compaction commit is allocation-failure atomic" {
    var fail_index: usize = 0;
    while (fail_index < 128) : (fail_index += 1) {
        if (try checkContextCompactionCommitAllocation(
            std.testing.allocator,
            fail_index,
        )) return;
    }
    return error.TestUnexpectedResult;
}

test "queued prompts begin with the latest checkpoint and its suffix" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    var first = try makePrompt(alloc, "first queued", "model");
    first.history = try session_runtime.snapshotOwnedContextHistory(alloc, &.{.{
        .assistant = .{ .user = .{ .text = @constCast("old request") }, .assistant = @constCast("old answer") },
    }}, 0, 0);
    try runtime.enqueuePrompt(alloc, first);
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "second queued", "model"));

    const checkpoint: types.HistoryTurn = .{ .compacted_summary = .{
        .summary = @constCast("<context_handoff>current summary</context_handoff>"),
        .removed_turn_count = 1,
        .compaction_count = 1,
    } };
    try runtime.publishContextCompaction(alloc, .{ .summary = checkpoint.compacted_summary }, &runtime, discardCompactionForTest);
    try std.testing.expectEqual(@as(usize, 1), runtime.queued_history.len);
    try runtime.propagateHistoryTurn(alloc, .{ .assistant = .{
        .user = .{ .text = @constCast("later request") },
        .assistant = @constCast("later answer"),
    } }, 0);

    for (0..2) |_| {
        const job = (try runtime.tryTakeNextPrompt(alloc)).?;
        defer freeQueuedPrompt(alloc, job);
        try std.testing.expectEqual(@as(usize, 2), job.history.len);
        try std.testing.expectEqualStrings(checkpoint.compacted_summary.summary, job.history[0].compacted_summary.summary);
        try std.testing.expectEqualStrings("later answer", job.history[1].assistant.assistant);
    }
    try std.testing.expectEqual(@as(usize, 0), runtime.queued_history.len);
}

test "finish ownership handoff deletes snapshots after its last unaccepted release" {
    const alloc = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    {
        var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{});
        defer file.close(std.testing.io);
        try file.writeStreamingAll(std.testing.io, "snapshot");
    }
    const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin");
    defer alloc.free(path);
    const images = [_]types.ImageAttachment{.{
        .id = 1,
        .path = @constCast("/tmp/source.png"),
        .media_type = @constCast("image/png"),
        .snapshot_path = path,
        .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
    }};
    var active = ActivePromptSnapshotOwnership.init(&images);
    const ownership = (try active.handoff(alloc)) orelse return error.TestExpectedEqual;
    active.deinit();

    const queued = try types.dupeFinishedPrompt(alloc, .{
        .turn = .{ .compacted_summary = .{
            .summary = @constCast("queued"),
            .removed_turn_count = 1,
            .compaction_count = 1,
        } },
        .snapshot_file_ownership = ownership,
    });
    ownership.release();
    const paced = try types.dupeFinishedPrompt(alloc, queued);
    types.freeFinishedPrompt(alloc, queued);
    try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});

    types.freeFinishedPrompt(alloc, paced);
    try std.testing.expectError(
        error.FileNotFound,
        std.Io.Dir.accessAbsolute(std.testing.io, path, .{}),
    );
}

test "accepted finish ownership survives queue and pacing release" {
    const alloc = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    {
        var file = try tmp.dir.createFile(std.testing.io, "snapshot.bin", .{});
        defer file.close(std.testing.io);
        try file.writeStreamingAll(std.testing.io, "snapshot");
    }
    const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "snapshot.bin");
    defer alloc.free(path);
    const images = [_]types.ImageAttachment{.{
        .id = 1,
        .path = @constCast("/tmp/source.png"),
        .media_type = @constCast("image/png"),
        .snapshot_path = path,
        .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
    }};
    var active = ActivePromptSnapshotOwnership.init(&images);
    const ownership = (try active.handoff(alloc)) orelse return error.TestExpectedEqual;
    active.deinit();

    const finished = try types.dupeFinishedPrompt(alloc, .{
        .turn = .{ .compacted_summary = .{
            .summary = @constCast("accepted"),
            .removed_turn_count = 1,
            .compaction_count = 1,
        } },
        .snapshot_file_ownership = ownership,
    });
    ownership.release();
    finished.snapshot_file_ownership.?.transfer();
    types.freeFinishedPrompt(alloc, finished);

    try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});
}

fn checkFinishOwnershipHandoffAllocation(
    alloc: std.mem.Allocator,
    snapshot_path: []const u8,
) !void {
    {
        var file = try std.Io.Dir.createFileAbsolute(
            std.testing.io,
            snapshot_path,
            .{ .truncate = true },
        );
        defer file.close(std.testing.io);
        try file.writeStreamingAll(std.testing.io, "snapshot");
    }
    const images = [_]types.ImageAttachment{.{
        .id = 1,
        .path = @constCast("/tmp/source.png"),
        .media_type = @constCast("image/png"),
        .snapshot_path = @constCast(snapshot_path),
        .snapshot_sha256 = @constCast("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
    }};
    var active = ActivePromptSnapshotOwnership.init(&images);
    const ownership = active.handoff(alloc) catch |err| {
        active.deinit();
        try std.testing.expectError(
            error.FileNotFound,
            std.Io.Dir.accessAbsolute(std.testing.io, snapshot_path, .{}),
        );
        return err;
    };
    active.deinit();
    ownership.?.release();
    try std.testing.expectError(
        error.FileNotFound,
        std.Io.Dir.accessAbsolute(std.testing.io, snapshot_path, .{}),
    );
}

test "finish ownership handoff preserves allocator and filesystem ownership" {
    const alloc = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    const root = try io_mod.dirRealpathAlloc(alloc, tmp.dir, ".");
    defer alloc.free(root);
    const snapshot_path = try std.fs.path.join(alloc, &.{ root, "snapshot.bin" });
    defer alloc.free(snapshot_path);

    try std.testing.checkAllAllocationFailures(
        alloc,
        checkFinishOwnershipHandoffAllocation,
        .{snapshot_path},
    );
}

pub fn dupeSkillBindings(alloc: std.mem.Allocator, bindings: []const SkillBinding) ![]SkillBinding {
    if (bindings.len == 0) return &.{};
    const copy = try alloc.alloc(SkillBinding, bindings.len);
    var filled: usize = 0;
    errdefer {
        var i: usize = 0;
        while (i < filled) : (i += 1) {
            alloc.free(copy[i].name);
            alloc.free(copy[i].path);
        }
        alloc.free(copy);
    }
    while (filled < bindings.len) : (filled += 1) {
        copy[filled] = .{
            .name = try alloc.dupe(u8, bindings[filled].name),
            .path = try alloc.dupe(u8, bindings[filled].path),
        };
    }
    return copy;
}

pub fn freeSkillBindings(alloc: std.mem.Allocator, bindings: []SkillBinding) void {
    for (bindings) |binding| {
        alloc.free(binding.name);
        alloc.free(binding.path);
    }
    if (bindings.len > 0) alloc.free(bindings);
}

pub fn dupeSkillDisplaySpans(alloc: std.mem.Allocator, spans: []const SkillDisplaySpan) ![]SkillDisplaySpan {
    if (spans.len == 0) return &.{};
    const copy = try alloc.alloc(SkillDisplaySpan, spans.len);
    var filled: usize = 0;
    errdefer {
        var i: usize = 0;
        while (i < filled) : (i += 1) {
            alloc.free(copy[i].name);
            alloc.free(copy[i].path);
        }
        alloc.free(copy);
    }
    while (filled < spans.len) : (filled += 1) {
        copy[filled] = .{
            .raw_start = spans[filled].raw_start,
            .raw_end = spans[filled].raw_end,
            .name = try alloc.dupe(u8, spans[filled].name),
            .path = try alloc.dupe(u8, spans[filled].path),
            .display_source = spans[filled].display_source,
            .owns_trailing_separator = spans[filled].owns_trailing_separator,
        };
    }
    return copy;
}

pub fn freeSkillDisplaySpans(alloc: std.mem.Allocator, spans: []SkillDisplaySpan) void {
    for (spans) |span| {
        alloc.free(span.name);
        alloc.free(span.path);
    }
    if (spans.len > 0) alloc.free(spans);
}

pub fn dupeWorkerEvent(alloc: std.mem.Allocator, event: WorkerEvent) !WorkerEvent {
    return switch (event) {
        .begin_prompt => |prompt| .{ .begin_prompt = try types.dupeUserTurn(alloc, prompt) },
        .begin_prompt_with_skill_bindings => |begin| blk: {
            const prompt = try types.dupeUserTurn(alloc, begin.prompt);
            errdefer types.freeUserTurn(alloc, prompt);
            const skill_bindings = try dupeSkillBindings(alloc, begin.skill_bindings);
            errdefer freeSkillBindings(alloc, skill_bindings);
            const skill_display_spans = try dupeSkillDisplaySpans(alloc, begin.skill_display_spans);
            break :blk .{ .begin_prompt_with_skill_bindings = .{
                .prompt = prompt,
                .skill_bindings = skill_bindings,
                .skill_display_spans = skill_display_spans,
            } };
        },
        .begin_presented_prompt => |turn_id| .{ .begin_presented_prompt = turn_id },
        .append_user_feedback => |text| .{ .append_user_feedback = try alloc.dupe(u8, text) },
        .assistant_presentation => |presentation| .{
            .assistant_presentation = try presentation.clone(alloc),
        },
        .notification => |notification| .{ .notification = notification },
        .question_requested => .question_requested,
        .open_model_picker => .open_model_picker,
        .semantic_notice => |notice| .{ .semantic_notice = try types.dupeSemanticNotice(alloc, notice) },
        .full_detail_record => |notice| .{ .full_detail_record = try types.dupeSemanticNotice(alloc, notice) },
        .route_recovery_status => |status| .{ .route_recovery_status = status },
        .clear_route_recovery_status => .clear_route_recovery_status,
        .api_status_text => |text| .{ .api_status_text = try alloc.dupe(u8, text) },
        .credential_refreshed => |credential| .{
            .credential_refreshed = try credential.clone(alloc),
        },
        .command_output => |chunk| .{ .command_output = .{
            .lifecycle_id = if (chunk.lifecycle_id) |id| .{
                .turn_id = id.turn_id,
                .call_id = try alloc.dupe(u8, id.call_id),
            } else null,
            .stream = chunk.stream,
            .text = try alloc.dupe(u8, chunk.text),
        } },
        .command_output_complete => |lifecycle_id| .{ .command_output_complete = if (lifecycle_id) |id| .{
            .turn_id = id.turn_id,
            .call_id = try alloc.dupe(u8, id.call_id),
        } else null },
        .tool_lifecycle => |lifecycle| .{
            .tool_lifecycle = try dupeToolLifecycleEvent(alloc, lifecycle),
        },
        .turn_token_update => |update| .{ .turn_token_update = update },
        .turn_phase_update => |update| .{ .turn_phase_update = update },
        .diff_block => |payload| blk: {
            const preview = try alloc.dupe(u8, payload.preview);
            errdefer alloc.free(preview);
            const full = if (payload.full) |source| full_value: {
                const content = try alloc.dupe(u8, source.content);
                errdefer alloc.free(content);
                const call_id = try alloc.dupe(u8, source.lifecycle_id.call_id);
                break :full_value diff_mod.FullDiff{
                    .content = content,
                    .lifecycle_id = .{
                        .turn_id = source.lifecycle_id.turn_id,
                        .call_id = call_id,
                    },
                };
            } else null;
            errdefer if (full) |owned| owned.deinit(alloc);
            break :blk .{ .diff_block = .{
                .preview = preview,
                .additions = payload.additions,
                .deletions = payload.deletions,
                .full = full,
            } };
        },
        .prepare_fresh_prompt => |value| blk: {
            const user = try types.dupeUserTurn(alloc, value.user);
            errdefer types.freeUserTurn(alloc, user);
            break :blk .{ .prepare_fresh_prompt = .{
                .turn_id = value.turn_id,
                .user = user,
                .prior_turn = if (value.prior_turn) |prior| try types.dupeHistoryTurn(alloc, prior) else null,
            } };
        },
        .context_compaction => |value| blk: {
            const summary = try types.dupeHistoryTurn(alloc, .{ .compacted_summary = value.summary });
            errdefer types.freeHistoryTurn(alloc, summary);
            const prefix = if (value.active_prefix) |prefix|
                (try types.dupeHistoryTurn(alloc, .{ .assistant = prefix })).assistant
            else
                null;
            break :blk .{ .context_compaction = .{
                .turn_id = value.turn_id,
                .summary = summary.compacted_summary,
                .active_prefix = prefix,
                .retained_from = value.retained_from,
                .max_history_turns = value.max_history_turns,
            } };
        },
        .finish_prompt => |finished| .{ .finish_prompt = try types.dupeFinishedPrompt(alloc, finished) },
        .session_grant => |grant| blk: {
            const tool_name = try alloc.dupe(u8, grant.tool_name);
            errdefer alloc.free(tool_name);
            const target_path = try alloc.dupe(u8, grant.target_path);
            break :blk .{ .session_grant = .{
                .tool_name = tool_name,
                .target_path = target_path,
            } };
        },
        .error_text => |notice| .{ .error_text = try types.dupeSemanticNotice(alloc, notice) },
        .restore_failed_prompt => |text| .{ .restore_failed_prompt = try alloc.dupe(u8, text) },
    };
}

pub fn freeWorkerEvent(alloc: std.mem.Allocator, event: WorkerEvent) void {
    switch (event) {
        .begin_prompt => |prompt| types.freeUserTurn(alloc, prompt),
        .begin_prompt_with_skill_bindings => |begin| {
            types.freeUserTurn(alloc, begin.prompt);
            freeSkillBindings(alloc, begin.skill_bindings);
            freeSkillDisplaySpans(alloc, begin.skill_display_spans);
        },
        .begin_presented_prompt => {},
        .append_user_feedback => |text| alloc.free(text),
        .assistant_presentation => |presentation| {
            var owned = presentation;
            owned.deinit(alloc);
        },
        .notification => {},
        .open_model_picker => {},
        .semantic_notice => |notice| types.freeSemanticNotice(alloc, notice),
        .full_detail_record => |notice| types.freeSemanticNotice(alloc, notice),
        .route_recovery_status => {},
        .clear_route_recovery_status => {},
        .api_status_text => |text| alloc.free(text),
        .credential_refreshed => |credential| {
            var owned = credential;
            owned.deinit(alloc);
        },
        .command_output => |chunk| {
            if (chunk.lifecycle_id) |id| alloc.free(@constCast(id.call_id));
            alloc.free(chunk.text);
        },
        .command_output_complete => |lifecycle_id| {
            if (lifecycle_id) |id| alloc.free(@constCast(id.call_id));
        },
        .tool_lifecycle => |lifecycle| freeToolLifecycleEvent(alloc, lifecycle),
        .diff_block => |payload| diff_mod.freeDiffEntryPayload(alloc, payload),
        .prepare_fresh_prompt => |value| {
            types.freeUserTurn(alloc, value.user);
            if (value.prior_turn) |prior| types.freeHistoryTurn(alloc, prior);
        },
        .context_compaction => |value| {
            types.freeHistoryTurn(alloc, .{ .compacted_summary = value.summary });
            if (value.active_prefix) |prefix| types.freeHistoryTurn(alloc, .{ .assistant = prefix });
        },
        .finish_prompt => |finished| types.freeFinishedPrompt(alloc, finished),
        .session_grant => |grant| {
            alloc.free(grant.tool_name);
            alloc.free(grant.target_path);
        },
        .error_text => |notice| types.freeSemanticNotice(alloc, notice),
        .restore_failed_prompt => |text| alloc.free(text),
        else => {},
    }
}

pub fn dupeToolLifecycleEvent(
    alloc: std.mem.Allocator,
    event: types.ToolLifecycleEvent,
) !types.ToolLifecycleEvent {
    return switch (event) {
        .provisional => |started| blk: {
            const call_id = try alloc.dupe(u8, started.id.call_id);
            errdefer alloc.free(call_id);
            const tool_name = if (started.tool_name) |name| try alloc.dupe(u8, name) else null;
            break :blk .{ .provisional = .{
                .id = .{ .turn_id = started.id.turn_id, .call_id = call_id },
                .presentation_group_id = started.presentation_group_id,
                .tool_name = tool_name,
                .activity_kind = started.activity_kind,
            } };
        },
        .authoritative_started => |started| blk: {
            const call_id = try alloc.dupe(u8, started.id.call_id);
            errdefer alloc.free(call_id);
            const alias = if (started.reconciles_provisional_call_id) |value|
                try alloc.dupe(u8, value)
            else
                null;
            errdefer if (alias) |value| alloc.free(value);
            const tool_name = try alloc.dupe(u8, started.tool_name);
            errdefer alloc.free(tool_name);
            const arguments_json = if (started.arguments_json) |value|
                try alloc.dupe(u8, value)
            else
                null;
            errdefer if (arguments_json) |value| alloc.free(value);
            break :blk .{ .authoritative_started = .{
                .id = .{ .turn_id = started.id.turn_id, .call_id = call_id },
                .presentation_group_id = started.presentation_group_id,
                .reconciles_provisional_call_id = alias,
                .tool_name = tool_name,
                .activity_kind = started.activity_kind,
                .arguments_json = arguments_json,
                .place_after_current_transcript = started.place_after_current_transcript,
            } };
        },
        .progress => |progress| blk: {
            const call_id = try alloc.dupe(u8, progress.id.call_id);
            errdefer alloc.free(call_id);
            const text = try alloc.dupe(u8, progress.text);
            break :blk .{ .progress = .{
                .id = .{ .turn_id = progress.id.turn_id, .call_id = call_id },
                .text = text,
            } };
        },
        .terminal => |terminal| blk: {
            const call_id = try alloc.dupe(u8, terminal.id.call_id);
            errdefer alloc.free(call_id);
            const summary = try alloc.dupe(u8, terminal.outcome.summary);
            errdefer alloc.free(summary);
            const result = if (terminal.result) |value|
                try alloc.dupe(u8, value)
            else
                null;
            errdefer if (result) |value| alloc.free(value);
            const result_memory = try dupeToolResultMemory(alloc, terminal.result_memory);
            errdefer freeToolResultMemory(alloc, result_memory);
            const command_artifact_handle = if (terminal.command_artifact_handle) |handle|
                try alloc.dupe(u8, handle)
            else
                null;
            errdefer if (command_artifact_handle) |handle| alloc.free(handle);
            break :blk .{ .terminal = .{
                .id = .{ .turn_id = terminal.id.turn_id, .call_id = call_id },
                .outcome = .{
                    .kind = terminal.outcome.kind,
                    .summary = summary,
                },
                .result = result,
                .result_memory = result_memory,
                .command_artifact_handle = command_artifact_handle,
            } };
        },
        .turn_finished => |finished| .{ .turn_finished = finished },
    };
}

pub fn freeToolLifecycleEvent(
    alloc: std.mem.Allocator,
    event: types.ToolLifecycleEvent,
) void {
    switch (event) {
        .provisional => |started| {
            alloc.free(@constCast(started.id.call_id));
            if (started.tool_name) |name| alloc.free(@constCast(name));
        },
        .authoritative_started => |started| {
            alloc.free(@constCast(started.id.call_id));
            if (started.reconciles_provisional_call_id) |alias| {
                alloc.free(@constCast(alias));
            }
            alloc.free(@constCast(started.tool_name));
            if (started.arguments_json) |arguments_json| {
                alloc.free(@constCast(arguments_json));
            }
        },
        .progress => |progress| {
            alloc.free(@constCast(progress.id.call_id));
            alloc.free(@constCast(progress.text));
        },
        .terminal => |terminal| {
            alloc.free(@constCast(terminal.id.call_id));
            alloc.free(@constCast(terminal.outcome.summary));
            if (terminal.result) |result| alloc.free(@constCast(result));
            freeToolResultMemory(alloc, terminal.result_memory);
            if (terminal.command_artifact_handle) |handle| alloc.free(@constCast(handle));
        },
        .turn_finished => {},
    }
}

fn dupeToolResultMemory(
    alloc: std.mem.Allocator,
    source: ?types.ToolResultMemory,
) !?types.ToolResultMemory {
    const memory = source orelse return null;
    const output_handle = if (memory.output_handle) |handle|
        try alloc.dupe(u8, handle)
    else
        null;
    errdefer if (output_handle) |handle| alloc.free(handle);
    const preview = if (memory.preview) |value|
        try alloc.dupe(u8, value)
    else
        null;
    errdefer if (preview) |value| alloc.free(value);
    const command_output_replay = if (memory.command_output_replay) |replay|
        try types.dupeCommandOutputReplay(alloc, replay)
    else
        null;
    errdefer if (command_output_replay) |replay| types.freeCommandOutputReplay(alloc, replay);
    return .{
        .output_handle = output_handle,
        .preview = preview,
        .output_bytes = memory.output_bytes,
        .stored_output_bytes = memory.stored_output_bytes,
        .truncated = memory.truncated,
        .model_view_covers_full_file = memory.model_view_covers_full_file,
        .command_output_replay = command_output_replay,
        .command_process_presentation = memory.command_process_presentation,
        .terminal_action_presentation = memory.terminal_action_presentation,
    };
}

fn freeToolResultMemory(
    alloc: std.mem.Allocator,
    memory: ?types.ToolResultMemory,
) void {
    const value = memory orelse return;
    if (value.output_handle) |handle| alloc.free(@constCast(handle));
    if (value.preview) |preview| alloc.free(@constCast(preview));
    if (value.command_output_replay) |replay| {
        types.freeCommandOutputReplay(alloc, replay);
    }
}

fn makePrompt(alloc: std.mem.Allocator, text: []const u8, model: []const u8) !QueuedPrompt {
    return .{
        .prompt = try alloc.dupe(u8, text),
        .images = &.{},
        .model = try alloc.dupe(u8, model),
        .api_key = try alloc.dupe(u8, "key"),
        .permission_mode = .auto,
        .history = try alloc.alloc(types.HistoryTurn, 0),
        .grants = try alloc.alloc(types.PermissionGrant, 0),
    };
}

fn makeImagePrompt(alloc: std.mem.Allocator, text: []const u8, model: []const u8) !QueuedPrompt {
    var prompt = try makePrompt(alloc, text, model);
    errdefer freeQueuedPrompt(alloc, prompt);
    const images = try alloc.alloc(types.ImageAttachment, 1);
    errdefer alloc.free(images);
    const path = try alloc.dupe(u8, "/tmp/steering.png");
    errdefer alloc.free(path);
    const media_type = try alloc.dupe(u8, "image/png");
    errdefer alloc.free(media_type);
    images[0] = .{ .path = path, .media_type = media_type };
    prompt.images = images;
    return prompt;
}

fn makeGrant(alloc: std.mem.Allocator, tool_name: []const u8, target_path: []const u8) !types.PermissionGrant {
    return .{ .tool_name = try alloc.dupe(u8, tool_name), .target_path = try alloc.dupe(u8, target_path) };
}

test "direct steering applies at the boundary without cancellation and seals before teardown" {
    const alloc = std.testing.allocator;
    var receipt = SteeringReceipt{};
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    try std.testing.expect(runtime.beginDirectProcessing(41));
    var prompt = try makePrompt(alloc, "review feedback", "model");
    prompt.steering_receipt = &receipt;
    try std.testing.expect(try runtime.admitActiveSteering(alloc, prompt));
    try std.testing.expect(!runtime.isCancelRequested());
    try std.testing.expectEqual(SteeringDelivery.queued, receipt.state.load(.seq_cst));
    var output = std.heap.ArenaAllocator.init(alloc);
    defer output.deinit();
    const boundary = try runtime.takeSteeringBoundaryInto(alloc, output.allocator(), 41, .finalizing);
    try std.testing.expectEqualStrings("review feedback", boundary.continue_turn[0]);
    try std.testing.expectEqual(SteeringDelivery.applied, receipt.state.load(.seq_cst));
    try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);
    try std.testing.expect((try runtime.takeSteeringBoundary(alloc, 41, .finalizing)) == .none);
    const late = try makePrompt(alloc, "too late", "model");
    defer freeQueuedPrompt(alloc, late);
    try std.testing.expect(!try runtime.admitActiveSteering(alloc, late));
    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
}

test "direct steering output allocation failure preserves feedback and cancellation discards it" {
    const alloc = std.testing.allocator;
    var receipt = SteeringReceipt{};
    var runtime = WorkerRuntime{};
    try std.testing.expect(runtime.beginDirectProcessing(41));
    var prompt = try makePrompt(alloc, "must not disappear", "model");
    prompt.steering_receipt = &receipt;
    try std.testing.expect(try runtime.admitActiveSteering(alloc, prompt));
    var failing = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 0 });
    try std.testing.expectError(error.OutOfMemory, runtime.takeSteeringBoundaryInto(alloc, failing.allocator(), 41, .model));
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
    try std.testing.expectEqual(SteeringDelivery.queued, receipt.state.load(.seq_cst));
    runtime.requestCancel();
    try std.testing.expect((try runtime.takeSteeringBoundary(alloc, 41, .cancelled)) == .interrupt);
    runtime.deinit(alloc);
    try std.testing.expectEqual(SteeringDelivery.not_applied, receipt.state.load(.seq_cst));
}

test "direct steering rejects inactive cancelled and full-allocation admission without taking ownership" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    const prompt = try makePrompt(alloc, "feedback", "model");
    defer freeQueuedPrompt(alloc, prompt);
    try std.testing.expect(!try runtime.admitActiveSteering(alloc, prompt));
    try std.testing.expect(runtime.beginDirectProcessing(41));
    var failing = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 0 });
    try std.testing.expectError(error.OutOfMemory, runtime.admitActiveSteering(failing.allocator(), prompt));
    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
    runtime.requestCancel();
    try std.testing.expect(!try runtime.admitActiveSteering(alloc, prompt));
}

fn makePromptWithGrant(alloc: std.mem.Allocator, text: []const u8, model: []const u8, tool_name: []const u8, target_path: []const u8) !QueuedPrompt {
    var prompt = try makePrompt(alloc, text, model);
    errdefer freeQueuedPrompt(alloc, prompt);
    const grants = try alloc.alloc(types.PermissionGrant, 1);
    grants[0] = try makeGrant(alloc, tool_name, target_path);
    alloc.free(prompt.grants);
    prompt.grants = grants;
    return prompt;
}

fn freeEventList(alloc: std.mem.Allocator, events: *std.ArrayList(WorkerEvent)) void {
    for (events.items) |event| freeWorkerEvent(alloc, event);
    events.deinit(alloc);
}

fn expectContinuedSteering(result: SteeringBoundaryResult) ![]const []const u8 {
    return switch (result) {
        .continue_turn => |messages| messages,
        .none, .handoff, .interrupt => error.TestExpectedContinuedSteering,
    };
}

test "subagent steering peek preserves the existing queue consumer" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    try std.testing.expect(!runtime.hasPendingPlainSteering());
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model"));
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model"));
    try std.testing.expect(runtime.hasPendingPlainSteering());
    try std.testing.expect(runtime.hasPendingPlainSteering());
    {
        var images: [1]types.ImageAttachment = undefined;
        runtime.queued_prompts.items[1].images = &images;
        defer runtime.queued_prompts.items[1].images = &.{};
        try std.testing.expect(!runtime.hasPendingPlainSteering());
    }
    try std.testing.expectEqual(@as(usize, 2), runtime.queued_prompts.items.len);
    const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(alloc, 41, .cancelled));
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqualStrings("first", guidance[0]);
    try std.testing.expectEqualStrings("second", guidance[1]);
    try std.testing.expect(!runtime.hasPendingPlainSteering());
}

test "interactive prompt admission derives steering from active worker state" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));

    const guidance = try expectContinuedSteering(
        try runtime.takeSteeringBoundary(alloc, 41, .cancelled),
    );
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqual(@as(usize, 1), guidance.len);
    try std.testing.expectEqualStrings("steer", guidance[0]);
}

test "interactive prompt during running manual compaction waits without cancelling" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try runtime.enqueueContextCompaction(.{
        .turn_id = 41,
        .model = try alloc.dupe(u8, "provider/model"),
        .api_key = try alloc.dupe(u8, "key"),
        .history = try alloc.alloc(types.HistoryTurn, 0),
    });
    const taken = (try runtime.tryTakeNextWork(alloc)) orelse
        return error.TestExpectedQueuedWork;
    defer freeWorkItem(alloc, taken);
    try std.testing.expectEqual(ContextCompactionStatus.running, runtime.contextCompactionStatus());

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));

    // The compaction is a wait boundary: no cancel, no immediate interrupt.
    try std.testing.expect(!runtime.isCancelRequested());
    try std.testing.expectEqual(@as(?u64, null), runtime.steering_cancel_turn_id);
    var snapshot = try runtime.snapshotSteeringPresentation(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expect(snapshot.waits_for_boundary);

    // A compaction work item never takes a steering boundary; finishing it
    // promotes the waiting prompt to a continuation that runs next.
    runtime.settleCompactionActivity(taken.compact_context.operation_id.?, .{
        .outcome = .succeeded,
        .stage = .publication,
        .publication = .committed,
    });
    runtime.finishProcessing();
    const next = (try runtime.tryTakeNextWork(alloc)) orelse
        return error.TestExpectedNextPrompt;
    defer freeWorkItem(alloc, next);
    try std.testing.expect(next == .prompt);
    try std.testing.expect(next.prompt.delivery.isContinuation());
    try std.testing.expectEqualStrings("steer", next.prompt.prompt);
}

test "interactive prompt during in-turn compaction activity waits without cancelling" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    // In-turn compaction reports through the activity channel, not the tool feed.
    const operation = runtime.beginCompactionActivity(.automatic, 41);
    runtime.runCompactionActivity(operation, .summary);

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));

    try std.testing.expect(!runtime.isCancelRequested());
    try std.testing.expectEqual(@as(?u64, null), runtime.steering_cancel_turn_id);

    // The next model boundary drains the waiting steer as same-turn guidance.
    const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(alloc, 41, .model));
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqual(@as(usize, 1), guidance.len);
    try std.testing.expectEqualStrings("steer", guidance[0]);

    runtime.settleCompactionActivity(operation, .{
        .outcome = .succeeded,
        .stage = .summary,
        .publication = .committed,
    });
    runtime.finishProcessing();
}

test "interactive prompt after in-turn compaction settled steers immediately" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    const operation = runtime.beginCompactionActivity(.automatic, 41);
    runtime.runCompactionActivity(operation, .summary);
    runtime.settleCompactionActivity(operation, .{
        .outcome = .succeeded,
        .stage = .summary,
        .publication = .committed,
    });

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));

    // No active boundary remains, so immediate-steer semantics are unchanged.
    try std.testing.expect(runtime.isCancelRequested());
    try std.testing.expectEqual(@as(?u64, 41), runtime.steering_cancel_turn_id);
    const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(alloc, 41, .cancelled));
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqual(@as(usize, 1), guidance.len);
    try std.testing.expectEqualStrings("steer", guidance[0]);
}

test "active prompt admission drains steering in FIFO order" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model"));
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model"));
    const guidance = try expectContinuedSteering(
        try runtime.takeSteeringBoundary(alloc, 41, .cancelled),
    );
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqual(@as(usize, 2), guidance.len);
    try std.testing.expectEqualStrings("first", guidance[0]);
    try std.testing.expectEqualStrings("second", guidance[1]);
    try std.testing.expectEqual(@as(usize, 0), runtime.queued_prompts.items.len);
    try std.testing.expectEqual(@as(usize, 2), runtime.worker_events.items.len);
    try std.testing.expectEqualStrings("first", runtime.worker_events.items[0].append_user_feedback);
    try std.testing.expectEqualStrings("second", runtime.worker_events.items[1].append_user_feedback);
}

test "immediate steering owns and clears only its cancellation" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model"));
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model"));
    try std.testing.expect(runtime.isCancelRequested());
    try std.testing.expect(!runtime.cancellationStopsTurn());
    var steering_snapshot = try runtime.snapshotState(alloc);
    defer steering_snapshot.deinit(alloc);
    try std.testing.expect(steering_snapshot.cancel_requested);
    try std.testing.expect(steering_snapshot.cancel_continues_turn);

    const guidance = try expectContinuedSteering(
        try runtime.takeSteeringBoundary(alloc, 41, .cancelled),
    );
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqual(@as(usize, 2), guidance.len);
    try std.testing.expectEqualStrings("first", guidance[0]);
    try std.testing.expectEqualStrings("second", guidance[1]);
    try std.testing.expect(!runtime.isCancelRequested());
    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
}

test "explicit interrupt overrides immediate steering cancellation" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));
    runtime.requestInteractiveCancel();
    try std.testing.expect(runtime.cancellationStopsTurn());
    var batch = runtime.takeEventBatch();
    defer freeEventList(alloc, &batch.events);
    try std.testing.expectEqual(@as(?u64, 41), batch.cancelled_turn_id);
    var interrupt_snapshot = try runtime.snapshotState(alloc);
    defer interrupt_snapshot.deinit(alloc);
    try std.testing.expect(interrupt_snapshot.cancel_requested);
    try std.testing.expect(!interrupt_snapshot.cancel_continues_turn);

    const boundary = try runtime.takeSteeringBoundary(alloc, 41, .cancelled);
    try std.testing.expect(boundary == .interrupt);
    try std.testing.expect(runtime.isCancelRequested());
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
}

test "interactive prompt after explicit cancellation starts a new visible turn" {
    const alloc = std.testing.allocator;
    for ([_]bool{ false, true }) |already_presented| {
        var runtime = WorkerRuntime{};
        defer runtime.deinit(alloc);
        try std.testing.expect(runtime.beginDirectProcessing(41));
        runtime.requestInteractiveCancel();

        var prompt = try makePrompt(alloc, "ok", "model");
        prompt.turn_id = 42;
        prompt.user_prompt_already_presented = already_presented;
        try runtime.admitInteractivePrompt(alloc, prompt);

        try std.testing.expect(runtime.cancellationStopsTurn());
        try std.testing.expect(runtime.queued_prompts.items[0].delivery == .ordinary);
        try std.testing.expect(try runtime.takeSteeringBoundary(alloc, 41, .cancelled) == .interrupt);
        try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);

        runtime.finishProcessing();
        const next = (try runtime.tryTakeNextPrompt(alloc)).?;
        defer freeQueuedPrompt(alloc, next);
        try std.testing.expectEqual(@as(u64, 42), next.turn_id);
        try std.testing.expectEqualStrings("ok", next.prompt);
        try std.testing.expect(!runtime.isCancelRequested());
        try std.testing.expectEqual(@as(usize, 1), runtime.worker_events.items.len);
        if (already_presented) {
            try std.testing.expectEqual(@as(u64, 42), runtime.worker_events.items[0].begin_presented_prompt);
        } else {
            try std.testing.expectEqualStrings("ok", runtime.worker_events.items[0].begin_prompt.text);
        }
        runtime.finishProcessing();
    }
}

test "sequential tool lifecycle does not inspect compatible allocator context identities" {
    var first_context: u8 = 0;
    var second_context: u8 = 0;
    var first_view = std.heap.c_allocator;
    var second_view = std.heap.c_allocator;
    first_view.ptr = @ptrCast(&first_context);
    second_view.ptr = @ptrCast(&second_context);
    var runtime = WorkerRuntime{};
    defer runtime.deinit(std.heap.c_allocator);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.pushEvent(first_view, .{ .tool_lifecycle = .{ .authoritative_started = .{
        .id = .{ .turn_id = 41, .call_id = "call_first" },
        .reconciles_provisional_call_id = null,
        .tool_name = "shell",
        .activity_kind = .command,
    } } });
    try runtime.pushEvent(first_view, .{ .tool_lifecycle = .{ .terminal = .{
        .id = .{ .turn_id = 41, .call_id = "call_first" },
        .outcome = .{ .kind = .completed, .summary = "completed" },
    } } });
    try runtime.pushEvent(second_view, .{ .tool_lifecycle = .{ .authoritative_started = .{
        .id = .{ .turn_id = 41, .call_id = "call_second" },
        .reconciles_provisional_call_id = null,
        .tool_name = "shell",
        .activity_kind = .command,
    } } });

    try std.testing.expect(runtime.active_tool_calls.contains("call_second"));
}

test "tool lifecycle decides whether interactive steering interrupts immediately" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{
        .id = .{ .turn_id = 41, .call_id = "call_running" },
        .reconciles_provisional_call_id = null,
        .tool_name = "terminal",
        .activity_kind = .command,
    } } });
    try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{
        .id = .{ .turn_id = 41, .call_id = "call_parallel" },
        .reconciles_provisional_call_id = null,
        .tool_name = "read_file",
        .activity_kind = .read,
    } } });
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "wait for tool", "model"));
    try std.testing.expect(!runtime.isCancelRequested());

    try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .terminal = .{
        .id = .{ .turn_id = 41, .call_id = "call_running" },
        .outcome = .{ .kind = .completed, .summary = "completed" },
    } } });
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "still wait", "model"));
    try std.testing.expect(!runtime.isCancelRequested());

    try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .terminal = .{
        .id = .{ .turn_id = 41, .call_id = "call_parallel" },
        .outcome = .{ .kind = .completed, .summary = "completed" },
    } } });
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "interrupt generation", "model"));
    try std.testing.expect(runtime.isCancelRequested());
}

test "subagent wait phase preserves steering control and existing tool boundaries" {
    const alloc = std.testing.allocator;
    for ([_]?u64{ null, 7 }) |tool_step| {
        var runtime = WorkerRuntime{};
        defer runtime.deinit(alloc);
        runtime.worker_processing = true;
        runtime.active_turn_id = 41;
        runtime.tool_phase_step_id = tool_step;
        for ([_][]const u8{ "first steering", "second steering" }) |text| {
            try runtime.pushEvent(alloc, .{ .turn_phase_update = .{
                .turn_id = 41,
                .step_id = 8,
                .phase = .waiting_for_subagent,
            } });
            try std.testing.expectEqual(tool_step, runtime.tool_phase_step_id);
            try std.testing.expectEqual(@as(usize, 0), runtime.active_tool_calls.count());
            try std.testing.expect(!runtime.isCancelRequested());
            try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, text, "model"));
            try std.testing.expectEqual(tool_step == null, runtime.isCancelRequested());
            const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(
                alloc,
                41,
                if (tool_step == null) .cancelled else .model,
            ));
            defer {
                for (guidance) |item| alloc.free(item);
                alloc.free(guidance);
            }
            try std.testing.expectEqual(@as(usize, 1), guidance.len);
            try std.testing.expectEqualStrings(text, guidance[0]);
            try std.testing.expect(!runtime.isCancelRequested());
        }
        runtime.requestCancel();
        try std.testing.expect((try runtime.takeSteeringBoundary(alloc, 41, .cancelled)) == .interrupt);
    }
}

test "tool handoff phase holds steering only through its model step" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.pushEvent(alloc, .{ .turn_phase_update = .{
        .turn_id = 41,
        .step_id = 7,
        .phase = .running,
    } });
    try runtime.pushEvent(alloc, .{ .turn_phase_update = .{
        .turn_id = 41,
        .step_id = 7,
        .phase = .generating,
    } });
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "wait for pending tools", "model"));

    try std.testing.expect(!runtime.isCancelRequested());
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
    var snapshot = try runtime.snapshotSteeringPresentation(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expect(snapshot.waits_for_boundary);

    const guidance = try expectContinuedSteering(
        try runtime.takeSteeringBoundary(alloc, 41, .model),
    );
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqualStrings("wait for pending tools", guidance[0]);

    try runtime.pushEvent(alloc, .{ .turn_phase_update = .{
        .turn_id = 41,
        .step_id = 8,
        .phase = .thinking,
    } });
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "interrupt next model step", "model"));
    try std.testing.expect(runtime.isCancelRequested());
}

test "failed steering admission does not cancel the active turn" {
    var runtime = WorkerRuntime{};
    defer runtime.deinit(std.testing.allocator);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    var failing = std.testing.FailingAllocator.init(
        std.testing.allocator,
        .{ .fail_index = 0 },
    );
    const prompt = QueuedPrompt{
        .prompt = @constCast("retain me"),
        .images = &.{},
        .model = @constCast("model"),
        .api_key = @constCast("key"),
        .permission_mode = .auto,
        .history = &.{},
        .grants = &.{},
    };

    try std.testing.expectError(
        error.OutOfMemory,
        runtime.admitInteractivePrompt(failing.allocator(), prompt),
    );
    try std.testing.expect(!runtime.isCancelRequested());
    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
}

test "rich interactive input remains steering and hands off at the tool boundary" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{
        .id = .{ .turn_id = 41, .call_id = "call_running" },
        .reconciles_provisional_call_id = null,
        .tool_name = "terminal",
        .activity_kind = .command,
    } } });

    var prompt = try makePrompt(alloc, "inspect this image", "model");
    errdefer freeQueuedPrompt(alloc, prompt);
    prompt.images = try alloc.alloc(types.ImageAttachment, 1);
    prompt.images[0] = .{
        .path = try alloc.dupe(u8, "/tmp/steering.png"),
        .media_type = try alloc.dupe(u8, "image/png"),
    };
    try runtime.admitInteractivePrompt(alloc, prompt);

    var snapshot = try runtime.snapshotSteeringPresentation(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expectEqual(@as(usize, 1), snapshot.messages.len);
    try std.testing.expect(!runtime.isCancelRequested());
    const boundary = try runtime.takeSteeringBoundary(alloc, 41, .model);
    try std.testing.expect(boundary == .handoff);
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
}

test "queued steer waiting at a tool boundary pops back for editing" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;
    try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{
        .id = .{ .turn_id = 41, .call_id = "call_running" },
        .reconciles_provisional_call_id = null,
        .tool_name = "terminal",
        .activity_kind = .command,
    } } });

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first steer", "model"));
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second steer", "model"));
    try std.testing.expect(!runtime.isCancelRequested());

    const second = (try runtime.popQueuedSteerForEdit(alloc)).?;
    defer alloc.free(second);
    try std.testing.expectEqualStrings("second steer", second);
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());

    var snapshot = try runtime.snapshotSteeringPresentation(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expectEqual(@as(usize, 1), snapshot.messages.len);
    try std.testing.expectEqualStrings("first steer", snapshot.messages[0]);
    try std.testing.expect(snapshot.waits_for_boundary);

    const guidance = try expectContinuedSteering(
        try runtime.takeSteeringBoundary(alloc, 41, .model),
    );
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqual(@as(usize, 1), guidance.len);
    try std.testing.expectEqualStrings("first steer", guidance[0]);
    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
    try std.testing.expect(try runtime.popQueuedSteerForEdit(alloc) == null);
}

test "immediate steer already committed to an interrupt is not retractable" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));
    try std.testing.expect(runtime.isCancelRequested());

    try std.testing.expect(try runtime.popQueuedSteerForEdit(alloc) == null);
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());

    const guidance = try expectContinuedSteering(
        try runtime.takeSteeringBoundary(alloc, 41, .cancelled),
    );
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqual(@as(usize, 1), guidance.len);
    try std.testing.expectEqualStrings("steer", guidance[0]);
}

test "steer retraction skips handoff prompts and takes the newest text steer" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;
    try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{
        .id = .{ .turn_id = 41, .call_id = "call_running" },
        .reconciles_provisional_call_id = null,
        .tool_name = "terminal",
        .activity_kind = .command,
    } } });

    var rich = try makePrompt(alloc, "inspect this image", "model");
    rich.images = try alloc.alloc(types.ImageAttachment, 1);
    rich.images[0] = .{
        .path = try alloc.dupe(u8, "/tmp/steering.png"),
        .media_type = try alloc.dupe(u8, "image/png"),
    };
    try runtime.admitInteractivePrompt(alloc, rich);
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "plain text", "model"));

    const text = (try runtime.popQueuedSteerForEdit(alloc)).?;
    defer alloc.free(text);
    try std.testing.expectEqualStrings("plain text", text);

    try std.testing.expect(try runtime.popQueuedSteerForEdit(alloc) == null);
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
    const boundary = try runtime.takeSteeringBoundary(alloc, 41, .model);
    try std.testing.expect(boundary == .handoff);
}

test "steer retraction requires an active processing turn" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model"));
    try std.testing.expect(try runtime.popQueuedSteerForEdit(alloc) == null);
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
}

test "failed steer retraction preserves the queue" {
    const backing = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(backing);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;
    try runtime.pushEvent(backing, .{ .tool_lifecycle = .{ .authoritative_started = .{
        .id = .{ .turn_id = 41, .call_id = "call_running" },
        .reconciles_provisional_call_id = null,
        .tool_name = "terminal",
        .activity_kind = .command,
    } } });
    try runtime.admitInteractivePrompt(backing, try makePrompt(backing, "steer", "model"));

    var failing = std.testing.FailingAllocator.init(
        backing,
        .{ .fail_index = 0 },
    );
    try std.testing.expectError(
        error.OutOfMemory,
        runtime.popQueuedSteerForEdit(failing.allocator()),
    );
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
}

test "immediate steering is visible while the provider cutoff settles" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model"));
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model"));
    var snapshot = try runtime.snapshotSteeringPresentation(alloc);
    defer snapshot.deinit(alloc);

    try std.testing.expectEqual(@as(usize, 2), snapshot.messages.len);
    try std.testing.expectEqualStrings("first", snapshot.messages[0]);
    try std.testing.expectEqualStrings("second", snapshot.messages[1]);
    try std.testing.expect(!snapshot.waits_for_boundary);
}

test "steering presentation survives queue to feedback transfer without duplicate input" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    try std.testing.expect(runtime.beginDirectProcessing(41));
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "same text", "model"));
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "same text", "model"));

    var queued = try runtime.snapshotSteeringPresentation(alloc);
    defer queued.deinit(alloc);
    try std.testing.expectEqual(@as(usize, 2), queued.messages.len);
    try std.testing.expectEqual(@as(usize, 0), queued.pending_feedback.len);

    const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(alloc, 41, .cancelled));
    defer {
        for (guidance) |message| alloc.free(message);
        alloc.free(guidance);
    }
    var consumed = try runtime.snapshotSteeringPresentation(alloc);
    defer consumed.deinit(alloc);
    try std.testing.expectEqual(@as(usize, 0), consumed.messages.len);
    try std.testing.expectEqual(@as(usize, 2), consumed.pending_feedback.len);
    for (consumed.pending_feedback) |message| try std.testing.expectEqualStrings("same text", message);

    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);
    var drained = try runtime.snapshotSteeringPresentation(alloc);
    defer drained.deinit(alloc);
    try std.testing.expectEqual(@as(usize, 0), drained.pending_feedback.len);
    try std.testing.expectEqual(@as(usize, 2), events.items.len);
    runtime.finishProcessing();
}

test "steering snapshot allocation failure preserves queue and feedback ownership" {
    try std.testing.checkAllAllocationFailures(std.testing.allocator, struct {
        fn run(alloc: std.mem.Allocator) !void {
            const backing = std.testing.allocator;
            var runtime = WorkerRuntime{};
            defer runtime.deinit(backing);
            try std.testing.expect(runtime.beginDirectProcessing(41));
            try runtime.pushEvent(backing, .{ .append_user_feedback = @constCast("earlier") });
            try runtime.admitInteractivePrompt(backing, try makePrompt(backing, "later", "model"));
            var snapshot = runtime.snapshotSteeringPresentation(alloc) catch |err| {
                try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
                try std.testing.expectEqual(@as(usize, 1), runtime.worker_events.items.len);
                return err;
            };
            defer snapshot.deinit(alloc);
            try std.testing.expectEqualStrings("earlier", snapshot.pending_feedback[0]);
            try std.testing.expectEqualStrings("later", snapshot.messages[0]);
            runtime.finishProcessing();
        }
    }.run, .{});
}

test "tool-blocked steering snapshot owns visible messages in admission order" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.pushEvent(alloc, .{ .tool_lifecycle = .{ .authoritative_started = .{
        .id = .{ .turn_id = 41, .call_id = "call_running" },
        .reconciles_provisional_call_id = null,
        .tool_name = "terminal",
        .activity_kind = .command,
    } } });
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model"));
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model"));

    var snapshot = try runtime.snapshotSteeringPresentation(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expectEqual(@as(usize, 2), snapshot.messages.len);
    try std.testing.expectEqualStrings("first", snapshot.messages[0]);
    try std.testing.expectEqualStrings("second", snapshot.messages[1]);
    try std.testing.expect(snapshot.waits_for_boundary);
}

test "late steering keeps admission order when demoted on finish" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 9;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer first", "model"));
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queue second", "model"));
    runtime.finishProcessing();

    try std.testing.expect(!runtime.worker_processing);
    try std.testing.expectEqual(@as(u64, 0), runtime.active_turn_id);
    try std.testing.expectEqual(@as(usize, 2), runtime.queuedPromptCount());
    try std.testing.expectEqualStrings("steer first", runtime.queued_prompts.items[0].prompt);
    try std.testing.expect(runtime.queued_prompts.items[0].delivery.isContinuation());
    try std.testing.expectEqualStrings("queue second", runtime.queued_prompts.items[1].prompt);
    try std.testing.expect(!runtime.queued_prompts.items[1].delivery.isContinuation());
}

test "promoted steering retargets remaining guidance without exposing a queue" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 9;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model"));
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "second", "model"));
    runtime.finishProcessing();

    const promoted = (try runtime.tryTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, promoted);
    try std.testing.expect(promoted.delivery.isContinuation());
    var snapshot = try runtime.snapshotSteeringPresentation(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expectEqual(@as(usize, 1), snapshot.messages.len);

    const guidance = try expectContinuedSteering(
        try runtime.takeSteeringBoundary(alloc, promoted.turn_id, .model),
    );
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqual(@as(usize, 1), guidance.len);
    try std.testing.expectEqualStrings("second", guidance[0]);
    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
}

test "promoted steering keeps rich trailing guidance for its own turn" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 9;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "first", "model"));
    var rich = try makePrompt(alloc, "second with image", "model");
    var owns_rich = true;
    errdefer if (owns_rich) freeQueuedPrompt(alloc, rich);
    rich.images = try alloc.alloc(types.ImageAttachment, 1);
    rich.images[0] = .{
        .path = try alloc.dupe(u8, "/tmp/trailing.png"),
        .media_type = try alloc.dupe(u8, "image/png"),
    };
    try runtime.admitInteractivePrompt(alloc, rich);
    owns_rich = false;
    runtime.finishProcessing();

    const promoted = (try runtime.tryTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, promoted);
    try std.testing.expect(promoted.delivery.isContinuation());
    var snapshot = try runtime.snapshotSteeringPresentation(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expectEqual(@as(usize, 1), snapshot.messages.len);
    const boundary = try runtime.takeSteeringBoundary(alloc, promoted.turn_id, .model);
    try std.testing.expect(boundary == .none);
    runtime.requestInteractiveCancel();

    runtime.finishProcessing();
    const trailing = (try runtime.tryTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, trailing);
    try std.testing.expect(trailing.delivery.isContinuation());
    try std.testing.expectEqualStrings("second with image", trailing.prompt);
}

test "clear queued prompts also clears steering" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 9;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model"));
    runtime.clearQueuedPrompts(alloc, &.{});

    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
    const boundary = try runtime.takeSteeringBoundary(alloc, 9, .model);
    try std.testing.expect(boundary == .none);
}

test "idle interactive admission uses ordinary queue" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "next", "model"));
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
    try std.testing.expect(runtime.queued_prompts.items[0].delivery == .ordinary);
}

test "request shutdown sets stop and cancel flags" {
    var runtime = WorkerRuntime{};
    defer runtime.deinit(std.testing.allocator);
    runtime.requestShutdown();
    try std.testing.expect(runtime.worker_stop_requested);
    try std.testing.expect(runtime.worker_cancel_requested.load(.seq_cst));
}

test "takeEventBatch snapshots cancellation with detached events" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    runtime.active_turn_id = 7;
    runtime.worker_cancel_requested.store(true, .seq_cst);
    runtime.worker_recovery_pause_requested.store(true, .seq_cst);
    try runtime.pushEvent(alloc, .{ .assistant_presentation = .{
        .text = @constCast("late"),
    } });

    var batch = runtime.takeEventBatch();
    defer freeEventList(alloc, &batch.events);
    runtime.active_turn_id = 8;
    runtime.worker_cancel_requested.store(false, .seq_cst);
    try std.testing.expectEqual(@as(?u64, 7), batch.cancelled_turn_id);
    try std.testing.expectEqual(@as(usize, 1), batch.events.items.len);
    try std.testing.expect(batch.events.items[0] == .assistant_presentation);
    try std.testing.expect(batch.events.items[0].assistant_presentation == .text);
    try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);
}

test "rich immediate steering revokes an earlier continuable cancellation" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "plain steer", "model"));
    try runtime.admitInteractivePrompt(alloc, try makeImagePrompt(alloc, "rich steer", "model"));
    try std.testing.expect(runtime.isCancelRequested());
    try std.testing.expect(runtime.cancellationStopsTurn());
    try runtime.pushEvent(alloc, .{ .assistant_presentation = .{ .text = @constCast("must stay hidden") } });

    var batch = runtime.takeEventBatch();
    defer freeEventList(alloc, &batch.events);
    try std.testing.expectEqual(@as(?u64, 41), batch.cancelled_turn_id);
    try std.testing.expectEqual(@as(usize, 1), batch.events.items.len);
    try std.testing.expect(
        try runtime.takeSteeringBoundary(alloc, 41, .cancelled) == .interrupt,
    );
}

test "plain input cannot make an earlier rich cancellation continuable" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.admitInteractivePrompt(alloc, try makeImagePrompt(alloc, "rich steer", "model"));
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "plain next turn", "model"));
    try std.testing.expect(runtime.isCancelRequested());
    try std.testing.expect(runtime.cancellationStopsTurn());
    try std.testing.expectEqual(@as(?u64, null), runtime.steering_cancel_turn_id);
    try std.testing.expectEqual(PromptDelivery.ordinary, runtime.queued_prompts.items[1].delivery);

    var batch = runtime.takeEventBatch();
    defer freeEventList(alloc, &batch.events);
    try std.testing.expectEqual(@as(?u64, 41), batch.cancelled_turn_id);
    try std.testing.expect(
        try runtime.takeSteeringBoundary(alloc, 41, .cancelled) == .interrupt,
    );
}

test "takeEventBatch keeps the steering tail live before boundary adoption" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;

    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer", "model"));
    try std.testing.expect(runtime.isCancelRequested());
    try std.testing.expect(!runtime.cancellationStopsTurn());
    try runtime.pushEvent(alloc, .{ .assistant_presentation = .{ .text = @constCast("buffered tail") } });
    var batch = runtime.takeEventBatch();
    defer freeEventList(alloc, &batch.events);
    try std.testing.expectEqual(@as(?u64, null), batch.cancelled_turn_id);
    try std.testing.expectEqual(@as(usize, 1), batch.events.items.len);
    try std.testing.expectEqualStrings("buffered tail", batch.events.items[0].assistant_presentation.text);

    const guidance = try expectContinuedSteering(try runtime.takeSteeringBoundary(alloc, 41, .cancelled));
    defer {
        for (guidance) |text| alloc.free(text);
        alloc.free(guidance);
    }
    try std.testing.expectEqual(@as(usize, 1), guidance.len);
    try std.testing.expectEqualStrings("steer", guidance[0]);
    try std.testing.expect(!runtime.isCancelRequested());
    try std.testing.expectEqual(@as(?u64, null), batch.cancelled_turn_id);
    var feedback = runtime.takeEventBatch();
    defer freeEventList(alloc, &feedback.events);
    try std.testing.expectEqual(@as(usize, 1), feedback.events.items.len);
    try std.testing.expectEqualStrings("steer", feedback.events.items[0].append_user_feedback);
}

test "takeEventBatch requires a matching active steering owner to continue" {
    const cases = [_]struct { active: u64, owner: ?u64, requested: bool, stopped: ?u64 }{
        .{ .active = 7, .owner = null, .requested = true, .stopped = 7 },
        .{ .active = 7, .owner = 6, .requested = true, .stopped = 7 },
        .{ .active = 7, .owner = 7, .requested = true, .stopped = null },
        .{ .active = 7, .owner = null, .requested = false, .stopped = null },
        .{ .active = 0, .owner = null, .requested = true, .stopped = null },
    };
    for (cases) |case| {
        var runtime = WorkerRuntime{};
        defer runtime.deinit(std.testing.allocator);
        runtime.active_turn_id = case.active;
        runtime.steering_cancel_turn_id = case.owner;
        runtime.worker_cancel_requested.store(case.requested, .seq_cst);
        var batch = runtime.takeEventBatch();
        defer freeEventList(std.testing.allocator, &batch.events);
        try std.testing.expectEqual(case.stopped, batch.cancelled_turn_id);
    }
}

test "state snapshot reports completion queued after event batch detach" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try runtime.pushEvent(alloc, .{ .command_output = .{
        .stream = .stdout,
        .text = @constCast("record\n"),
    } });
    var batch = runtime.takeEventBatch();
    defer freeEventList(alloc, &batch.events);
    try runtime.pushEvent(alloc, .{ .command_output_complete = null });

    var snapshot = try runtime.snapshotState(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expectEqual(@as(usize, 1), snapshot.pending_event_count);
    try std.testing.expectEqual(@as(usize, 1), batch.events.items.len);
    try std.testing.expect(batch.events.items[0] == .command_output);
    try std.testing.expect(runtime.worker_events.items[0] == .command_output_complete);
}

fn checkStateSnapshotFailurePreservesPendingEvents(alloc: std.mem.Allocator) !void {
    const owner_alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(owner_alloc);

    try runtime.pushEvent(owner_alloc, .{ .command_output_complete = null });
    runtime.worker_processing = true;
    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared =
        try permission_request.OwnedPermissionRequest.dupe(
            owner_alloc,
            .{ .id = 17, .label = "pending command completion" },
        );

    var snapshot = runtime.snapshotState(alloc) catch |err| {
        try std.testing.expectEqual(@as(usize, 1), runtime.worker_events.items.len);
        try std.testing.expect(runtime.worker_events.items[0] == .command_output_complete);
        try std.testing.expectEqualStrings(
            "pending command completion",
            runtime.pending_permission_request_shared.?.label,
        );
        return err;
    };
    defer snapshot.deinit(alloc);
    try std.testing.expectEqual(@as(usize, 1), snapshot.pending_event_count);
    try std.testing.expectEqual(@as(usize, 1), runtime.worker_events.items.len);
    try std.testing.expect(runtime.worker_events.items[0] == .command_output_complete);
}

test "state snapshot allocation failures preserve pending event ownership" {
    try std.testing.checkAllAllocationFailures(
        std.testing.allocator,
        checkStateSnapshotFailurePreservesPendingEvents,
        .{},
    );
}

test "finalization failure latches fixed metadata and closes interactive admission" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try std.testing.expect(runtime.interactiveAdmissionSnapshot() == .open);
    runtime.latchFinalizationFailure(.{
        .turn_id = 41,
        .outcome = .failed,
    });
    runtime.latchFinalizationFailure(.{
        .turn_id = 42,
        .outcome = .interrupted,
    });

    switch (runtime.interactiveAdmissionSnapshot()) {
        .finalization_failed => |failure| {
            try std.testing.expectEqual(@as(u64, 41), failure.turn_id);
            try std.testing.expectEqual(types.TurnPresentationOutcome.failed, failure.outcome);
        },
        else => return error.TestExpectedEqual,
    }
    try std.testing.expect(runtime.worker_stop_requested);
    try std.testing.expect(runtime.worker_cancel_requested.load(.seq_cst));

    const prompt = try makePrompt(alloc, "rejected", "model");
    try std.testing.expectError(error.TurnFinalizationDeliveryFailed, runtime.enqueuePrompt(alloc, prompt));
    freeQueuedPrompt(alloc, prompt);
    try std.testing.expectEqual(@as(usize, 0), runtime.queued_prompts.items.len);
}

test "dedicated turn finalizer queues before optional finish payload" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try runtime.pushTurnFinished(alloc, .{
        .turn_id = 99,
        .outcome = .completed,
    });
    try runtime.pushEvent(alloc, .{ .finish_prompt = .{
        .turn = .{ .assistant = .{
            .user = .{ .text = @constCast("prompt") },
            .assistant = @constCast("answer"),
        } },
    } });

    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);
    try std.testing.expectEqual(@as(usize, 2), events.items.len);
    try std.testing.expect(events.items[0] == .tool_lifecycle);
    try std.testing.expectEqual(
        @as(u64, 99),
        events.items[0].tool_lifecycle.turn_finished.turn_id,
    );
    try std.testing.expect(events.items[1] == .finish_prompt);
}

test "queue, event, snapshot, sync, history, and grant behavior" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    const first_prompt = try makePrompt(alloc, "first", "old");
    try runtime.enqueuePrompt(alloc, first_prompt);
    const second_prompt = try makePrompt(alloc, "second", "old");
    try runtime.enqueuePrompt(alloc, second_prompt);
    try std.testing.expectEqual(@as(usize, 2), runtime.queuedPromptCount());

    runtime.worker_cancel_requested.store(true, .seq_cst);
    const job = (try runtime.waitAndTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, job);
    try std.testing.expect(job.turn_id != 0);
    try std.testing.expectEqualStrings("first", job.prompt);
    try std.testing.expect(!runtime.isCancelRequested());
    try std.testing.expect(runtime.worker_processing);
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());

    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);
    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .begin_prompt);
    try std.testing.expectEqualStrings("first", events.items[0].begin_prompt.text);
    try std.testing.expect(events.items[0].begin_prompt.text.ptr != job.prompt.ptr);

    try runtime.syncQueuedPromptModel(alloc, "next");
    try std.testing.expectEqualStrings("next", runtime.queued_prompts.items[0].model);
    runtime.syncQueuedPromptPermissionSnapshot(.{
        .mode = .ask,
    });
    try std.testing.expectEqual(types.PermissionMode.ask, runtime.queued_prompts.items[0].permission_mode);
    try std.testing.expectEqual(types.PermissionMode.auto, job.permission_mode);

    const grant_source = try alloc.alloc(types.PermissionGrant, 1);
    defer types.freePermissionGrantSlice(alloc, grant_source);
    grant_source[0] = try makeGrant(alloc, "read_file", "/tmp/a");
    try runtime.syncQueuedPromptPermissionState(alloc, grant_source, .{
        .mode = .auto,
    });
    grant_source[0].tool_name[0] = 'R';
    grant_source[0].target_path[1] = 'T';
    try std.testing.expectEqualStrings("read_file", runtime.queued_prompts.items[0].grants[0].tool_name);
    try std.testing.expectEqualStrings("/tmp/a", runtime.queued_prompts.items[0].grants[0].target_path);

    const old_turn: types.HistoryTurn = .{ .compacted_summary = .{
        .summary = try alloc.dupe(u8, "old"),
        .removed_turn_count = 0,
        .compaction_count = 0,
    } };
    defer types.freeHistoryTurn(alloc, old_turn);
    const seeded_history = try appendHistoryTurnProjection(
        alloc,
        runtime.queued_history,
        old_turn,
        2,
    );
    types.freeHistoryTurnSlice(alloc, runtime.queued_history);
    runtime.queued_history = seeded_history;
    const turn: types.HistoryTurn = .{ .compacted_summary = .{
        .summary = try alloc.dupe(u8, "summary"),
        .removed_turn_count = 0,
        .compaction_count = 0,
    } };
    defer types.freeHistoryTurn(alloc, turn);
    try runtime.propagateHistoryTurn(alloc, turn, 1);
    try std.testing.expectEqual(@as(usize, 2), runtime.queued_history.len);
    try std.testing.expect(runtime.queued_history[1] == .compacted_summary);
    try std.testing.expectEqualStrings("summary", runtime.queued_history[1].compacted_summary.summary);
    try std.testing.expectEqual(@as(usize, 0), runtime.queued_prompts.items[0].unversioned_history_count);

    try runtime.propagateGrant(alloc, "read_file", "/tmp/a");
    try std.testing.expectEqual(@as(usize, 1), runtime.queued_prompts.items[0].grants.len);
    try runtime.propagateGrant(alloc, "read_file", "/tmp/b");
    try std.testing.expectEqual(@as(usize, 2), runtime.queued_prompts.items[0].grants.len);

    runtime.worker_processing = true;
    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared =
        try permission_request.OwnedPermissionRequest.dupe(
            alloc,
            .{ .id = 1, .label = "pending" },
        );
    var snapshot = try runtime.snapshotState(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expectEqualStrings("pending", snapshot.pending_permission_request.?.label);
    try std.testing.expect(
        snapshot.pending_permission_request.?.label.ptr !=
            runtime.pending_permission_request_shared.?.label.ptr,
    );
}

test "permission response durable barrier commits before waiter wake" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared =
        try permission_request.OwnedPermissionRequest.dupe(
            alloc,
            .{ .id = 9, .label = "prepared action" },
        );

    const Capture = struct {
        runtime: *WorkerRuntime,
        committed: bool = false,

        fn commit(raw: *anyopaque) WorkerRuntime.PermissionCommitError!void {
            const self: *@This() = @ptrCast(@alignCast(raw));
            if (self.runtime.pending_permission_response != null) {
                return error.PermissionCommitFailed;
            }
            self.committed = true;
        }
    };
    var capture = Capture{ .runtime = &runtime };
    const result = try runtime.submitPermissionResponseAfterCommit(
        9,
        permission_request.OwnedPermissionResponse.init(alloc, .always, null),
        .{ .context = &capture, .commit_fn = Capture.commit },
    );
    try std.testing.expectEqual(PermissionSubmissionResult.accepted, result);
    try std.testing.expect(capture.committed);
    try std.testing.expect(runtime.pending_permission_response != null);
}

test "permission response durable barrier failure does not wake waiter" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared =
        try permission_request.OwnedPermissionRequest.dupe(
            alloc,
            .{ .id = 10, .label = "prepared action" },
        );

    const Fail = struct {
        fn commit(_: *anyopaque) WorkerRuntime.PermissionCommitError!void {
            return error.PermissionCommitFailed;
        }
    };
    try std.testing.expectError(
        error.PermissionCommitFailed,
        runtime.submitPermissionResponseAfterCommit(
            10,
            permission_request.OwnedPermissionResponse.init(alloc, .once, null),
            .{ .context = &runtime, .commit_fn = Fail.commit },
        ),
    );
    try std.testing.expect(runtime.pending_permission_response == null);
    try std.testing.expect(runtime.pending_permission_waiting);
}

test "permission capacity rejection removes the unpublished waiter" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    const Reject = struct {
        fn observe(
            _: *anyopaque,
            _: *WorkerRuntime,
            _: permission_request.PermissionRequest,
        ) error{
            OutOfMemory,
            PermissionRegistrationFailed,
            PermissionCapacityExceeded,
        }!void {
            return error.PermissionCapacityExceeded;
        }
    };
    try std.testing.expectError(
        error.PermissionCapacityExceeded,
        runtime.requestPermissionBlockingObserved(
            alloc,
            .{ .id = 11, .label = "capacity-bound action" },
            null,
            .{ .context = &runtime, .observe_fn = Reject.observe },
        ),
    );
    try std.testing.expect(!runtime.pending_permission_waiting);
    try std.testing.expect(runtime.pending_permission_request_shared == null);
    try std.testing.expect(runtime.pending_permission_review == null);
    try std.testing.expect(runtime.pending_permission_response == null);
}

test "waitAndTakeNextPrompt emits bound skill prompt event when queued prompt has bindings" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    var prompt = try makePrompt(alloc, "raw $review then $review and $review", "model");
    errdefer freeQueuedPrompt(alloc, prompt);
    prompt.skill_bindings = try dupeSkillBindings(alloc, &[_]SkillBinding{.{
        .name = @constCast("review"),
        .path = @constCast("/tmp/.codex/skills/review"),
    }});
    prompt.skill_display_spans = try dupeSkillDisplaySpans(alloc, &[_]SkillDisplaySpan{
        .{
            .raw_start = "raw $review then ".len,
            .raw_end = "raw $review then $review".len,
            .name = @constCast("review"),
            .path = @constCast("/tmp/.codex/skills/review"),
            .display_source = .workspace_codex,
            .owns_trailing_separator = true,
        },
        .{
            .raw_start = "raw $review then $review and ".len,
            .raw_end = "raw $review then $review and $review".len,
            .name = @constCast("review"),
            .path = @constCast("/tmp/.codex/skills/review"),
        },
    });
    try runtime.enqueuePrompt(alloc, prompt);

    const job = (try runtime.waitAndTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, job);

    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);
    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .begin_prompt_with_skill_bindings);
    try std.testing.expectEqualStrings("raw $review then $review and $review", events.items[0].begin_prompt_with_skill_bindings.prompt.text);
    try std.testing.expectEqual(@as(usize, 1), events.items[0].begin_prompt_with_skill_bindings.skill_bindings.len);
    try std.testing.expectEqualStrings("review", events.items[0].begin_prompt_with_skill_bindings.skill_bindings[0].name);
    try std.testing.expectEqualStrings("/tmp/.codex/skills/review", events.items[0].begin_prompt_with_skill_bindings.skill_bindings[0].path);
    try std.testing.expectEqual(@as(usize, 2), events.items[0].begin_prompt_with_skill_bindings.skill_display_spans.len);
    try std.testing.expectEqual(@as(usize, "raw $review then ".len), events.items[0].begin_prompt_with_skill_bindings.skill_display_spans[0].raw_start);
    try std.testing.expectEqual(@as(usize, "raw $review then $review".len), events.items[0].begin_prompt_with_skill_bindings.skill_display_spans[0].raw_end);
    try std.testing.expectEqualStrings("review", events.items[0].begin_prompt_with_skill_bindings.skill_display_spans[0].name);
    try std.testing.expectEqual(
        skill_contract.SkillSource.workspace_codex,
        events.items[0].begin_prompt_with_skill_bindings.skill_display_spans[0].display_source.?,
    );
    try std.testing.expect(events.items[0].begin_prompt_with_skill_bindings.skill_display_spans[0].owns_trailing_separator);
}

test "permission snapshot keeps the full file review borrowed and request-bound" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    var review = try diff_mod.FileReview.init(
        alloc,
        "",
        "line 1\nline 2\nline 3\nline 4\nline 5\nline 6\nline 7\n",
    );
    defer review.deinit(alloc);

    runtime.worker_processing = true;
    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared =
        try permission_request.OwnedPermissionRequest.dupe(
            alloc,
            .{
                .id = 44,
                .label = "file_mutation",
                .file = .{
                    .kind = .write,
                    .intent = .mutation,
                    .preview = .{
                        .path = "note.txt",
                        .lines = &.{
                            .{ .op = .addition, .new_line = 1, .text = "line 1" },
                            .{ .op = .addition, .new_line = 7, .text = "line 7" },
                        },
                        .additions = 7,
                        .deletions = 0,
                        .truncated = true,
                    },
                    .scope = .workspace_files,
                },
            },
        );
    runtime.pending_permission_review = &review;

    var snapshot = try runtime.snapshotState(alloc);
    defer snapshot.deinit(alloc);

    try std.testing.expectEqual(
        @as(usize, 2),
        snapshot.pending_permission_request.?.file.?.preview.lines.len,
    );
    try std.testing.expectEqual(@as(u64, 44), snapshot.pending_permission_review.?.request_id);
    try std.testing.expect(snapshot.pending_permission_review.?.review == &review);
    try std.testing.expectEqual(@as(usize, 7), snapshot.pending_permission_review.?.review.rowCount());
}

test "held turn propagation refreshes queued prompt trusted reviewer context" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    var queued = try makePrompt(alloc, "queued current prompt", "model");
    var canonical_history = [_]types.HistoryTurn{.{ .assistant = .{
        .user = .{ .text = @constCast("original first request") },
        .assistant = @constCast("original answer"),
    } }};
    queued.root_user_intent_context = try auto_classifier_context.buildCanonicalRootUserContext(
        alloc,
        queued.prompt,
        &canonical_history,
    );
    try runtime.enqueuePrompt(alloc, queued);
    const original_prompt_ptr = runtime.queued_prompts.items[0].prompt.ptr;
    const original_prompt_len = runtime.queued_prompts.items[0].prompt.len;

    const turn: types.HistoryTurn = .{ .assistant = .{
        .user = .{ .text = try alloc.dupe(u8, "Revoke permission: do not modify any more files.") },
        .assistant = try alloc.dupe(u8, "completed prior answer"),
    } };
    defer types.freeHistoryTurn(alloc, turn);

    try runtime.propagateHistoryTurn(alloc, turn, 8);

    try std.testing.expectEqual(@as(usize, 1), runtime.queued_history.len);
    try std.testing.expect(runtime.queued_history[0] == .assistant);
    try std.testing.expectEqualStrings("Revoke permission: do not modify any more files.", runtime.queued_history[0].assistant.user.text);
    try std.testing.expectEqualStrings("completed prior answer", runtime.queued_history[0].assistant.assistant);
    try std.testing.expectEqualStrings("queued current prompt", runtime.queued_prompts.items[0].prompt);
    try std.testing.expectEqual(original_prompt_len, runtime.queued_prompts.items[0].prompt.len);
    try std.testing.expect(original_prompt_ptr == runtime.queued_prompts.items[0].prompt.ptr);

    const review_context = runtime.queued_prompts.items[0].root_user_intent_context;
    try std.testing.expect(std.mem.find(
        u8,
        review_context,
        "first_root_user_request: original first request",
    ) != null);
    try std.testing.expect(std.mem.find(
        u8,
        review_context,
        "recent_root_user_request: Revoke permission: do not modify any more files.",
    ) != null);
    try std.testing.expect(std.mem.find(
        u8,
        review_context,
        "current_request: queued current prompt",
    ) != null);

    const newer_turn: types.HistoryTurn = .{ .assistant = .{
        .user = .{ .text = try alloc.dupe(u8, "newer completed request") },
        .assistant = try alloc.dupe(u8, "newer completed answer"),
    } };
    defer types.freeHistoryTurn(alloc, newer_turn);
    try runtime.propagateHistoryTurn(alloc, newer_turn, 8);
    const refreshed = runtime.queued_prompts.items[0].root_user_intent_context;
    try std.testing.expect(std.mem.find(
        u8,
        refreshed,
        "recent_root_user_request: Revoke permission: do not modify any more files.",
    ) != null);
    try std.testing.expect(std.mem.find(
        u8,
        refreshed,
        "recent_root_user_request: newer completed request",
    ) != null);
}

test "completed image turns propagate authority to queued follow-ups" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    var prompt = try makePrompt(alloc, "queued follow-up", "model");
    prompt.authorized_image_catalog = try types.dupeImageAttachmentSlice(alloc, &.{.{
        .id = 2,
        .path = @constCast("/tmp/older.png"),
        .media_type = @constCast("image/png"),
    }});
    try runtime.enqueuePrompt(alloc, prompt);

    var completed_images = [_]types.ImageAttachment{.{
        .id = 8,
        .path = @constCast("/tmp/completed.png"),
        .media_type = @constCast("image/png"),
    }};
    const turn: types.HistoryTurn = .{ .assistant = .{
        .user = .{ .text = @constCast("active image turn"), .images = &completed_images },
        .assistant = @constCast("done"),
    } };
    try runtime.propagateHistoryTurn(alloc, turn, 1);

    const queued = runtime.queued_prompts.items[0];
    try std.testing.expectEqual(@as(usize, 2), queued.authorized_image_catalog.len);
    try std.testing.expectEqual(@as(usize, 2), queued.authorized_image_catalog[0].id);
    try std.testing.expectEqual(@as(usize, 8), queued.authorized_image_catalog[1].id);
    try std.testing.expectEqualStrings("/tmp/completed.png", queued.authorized_image_catalog[1].path);
}

test "explicit cancellation keeps targeted steering runnable" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 17;
    try runtime.admitInteractivePrompt(alloc, try makePrompt(alloc, "steer now", "model"));

    runtime.requestInteractiveCancel();
    try std.testing.expect(runtime.isCancelRequested());

    runtime.finishProcessing();
    const next = (try runtime.tryTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, next);
    try std.testing.expectEqualStrings("steer now", next.prompt);
}

const PromptTakeThreadState = struct {
    started: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
    finished: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
    err: ?anyerror = null,
    job: ?QueuedPrompt = null,
};

fn runPromptTake(state: *PromptTakeThreadState, runtime: *WorkerRuntime) void {
    state.started.store(true, .seq_cst);
    state.job = runtime.waitAndTakeNextPrompt(std.testing.allocator) catch |err| {
        state.err = err;
        state.finished.store(true, .seq_cst);
        return;
    };
    state.finished.store(true, .seq_cst);
}

test "turn start hold rejects busy worker and blocks take while held" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    runtime.worker_processing = true;
    try std.testing.expect(!runtime.tryHoldTurnStart());
    runtime.worker_processing = false;

    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued blocks park", "model"));
    try std.testing.expect(!runtime.tryHoldTurnStart());
    runtime.clearQueuedPrompts(alloc, &.{});

    try std.testing.expect(runtime.tryHoldTurnStart());
    try std.testing.expect(!runtime.tryHoldTurnStart());

    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "held blocked prompt", "model"));
    var state = PromptTakeThreadState{};
    const thread = try std.Thread.spawn(.{}, runPromptTake, .{ &state, &runtime });
    var joined = false;
    defer if (!joined) {
        runtime.releaseTurnStartHold();
        runtime.requestStop();
        thread.join();
    };

    var attempts: usize = 0;
    while (!state.started.load(.seq_cst) and attempts < 100) : (attempts += 1) {
        io_mod.sleep(std.time.ns_per_ms);
    }
    try std.testing.expect(state.started.load(.seq_cst));
    io_mod.sleep(10 * std.time.ns_per_ms);
    try std.testing.expect(!state.finished.load(.seq_cst));

    runtime.releaseTurnStartHold();
    thread.join();
    joined = true;
    try std.testing.expect(state.err == null);
    const job = state.job.?;
    defer freeQueuedPrompt(alloc, job);
    try std.testing.expectEqualStrings("held blocked prompt", job.prompt);
}

test "waitAndTakeNextPrompt assigns turn id and resets cancellation for next prompt" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    runtime.worker_cancel_requested.store(true, .seq_cst);
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "next", "model"));
    try std.testing.expect(runtime.isCancelRequested());
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());

    const job = (try runtime.waitAndTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, job);

    try std.testing.expect(job.turn_id != 0);
    try std.testing.expect(!runtime.isCancelRequested());
    try std.testing.expect(!runtime.worker_recovery_pause_requested.load(.seq_cst));

    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);
    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .begin_prompt);
}

test "already-presented queued prompt emits identity without duplicating user payload" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    var queued = try makePrompt(alloc, "already visible", "model");
    queued.turn_id = 9001;
    queued.user_prompt_already_presented = true;
    try runtime.enqueuePrompt(alloc, queued);

    const job = (try runtime.tryTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, job);
    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);

    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .begin_presented_prompt);
    try std.testing.expectEqual(@as(u64, 9001), events.items[0].begin_presented_prompt);
}

test "requestRecoveryPause wakes the active operation without losing pause intent" {
    var worker = WorkerRuntime{};
    worker.requestRecoveryPause();
    try std.testing.expect(worker.worker_recovery_pause_requested.load(.seq_cst));
    try std.testing.expect(worker.worker_cancel_requested.load(.seq_cst));
}

test "enqueuePrompt stamps agent settings from worker runtime" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    runtime.agent_turn_settings = .{
        .max_tool_result_bytes = 4096,
        .first_call_tool_choice = .none,
    };
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "next", "model"));
    const job = (try runtime.waitAndTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, job);

    try std.testing.expectEqual(@as(usize, 4096), job.agent_settings.max_tool_result_bytes);
    try std.testing.expectEqual(types.ToolChoice.none, job.agent_settings.first_call_tool_choice);
}

test "enqueuePrompt stamps fast mode and effort from worker runtime" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    runtime.agent_turn_settings = .{
        .fast_mode = true,
        .effort = types.ReasoningEffort.literal("high"),
    };

    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "next", "model"));
    const job = (try runtime.waitAndTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, job);

    try std.testing.expect(job.agent_settings.fast_mode);
    try std.testing.expectEqual(types.ReasoningEffort.literal("high"), job.agent_settings.effort);
}

test "sync queued prompt fast mode and effort updates queued prompts and future defaults" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    runtime.agent_turn_settings = .{
        .fast_mode = false,
        .effort = types.ReasoningEffort.literal("low"),
    };
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "queued", "model"));

    runtime.syncQueuedPromptFastMode(true);
    runtime.syncQueuedPromptEffort(types.ReasoningEffort.literal("high"));

    try std.testing.expect(runtime.queued_prompts.items[0].agent_settings.fast_mode);
    try std.testing.expectEqual(types.ReasoningEffort.literal("high"), runtime.queued_prompts.items[0].agent_settings.effort);
    try std.testing.expect(runtime.agent_turn_settings.fast_mode);
    try std.testing.expectEqual(types.ReasoningEffort.literal("high"), runtime.agent_turn_settings.effort);

    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "future", "model"));

    try std.testing.expect(runtime.queued_prompts.items[1].agent_settings.fast_mode);
    try std.testing.expectEqual(types.ReasoningEffort.literal("high"), runtime.queued_prompts.items[1].agent_settings.effort);
}

const EnqueueThreadState = struct {
    started: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
    err: ?anyerror = null,
};

fn runEnqueuePrompt(state: *EnqueueThreadState, runtime: *WorkerRuntime, text: []const u8, model: []const u8) void {
    const prompt = makePrompt(std.testing.allocator, text, model) catch |err| {
        state.err = err;
        return;
    };
    state.started.store(true, .seq_cst);
    runtime.enqueuePrompt(std.testing.allocator, prompt) catch |err| {
        freeQueuedPrompt(std.testing.allocator, prompt);
        state.err = err;
        return;
    };
}

fn waitForEnqueueThreadStart(state: *EnqueueThreadState) !void {
    var attempts: usize = 0;
    while (attempts < 100) : (attempts += 1) {
        if (state.started.load(.seq_cst)) return;
        io_mod.sleep(1 * std.time.ns_per_ms);
    }
    return error.TestExpectedEqual;
}

test "enqueuePrompt snapshots settings under the same mutex as sync" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    runtime.agent_turn_settings = .{
        .fast_mode = false,
        .effort = types.ReasoningEffort.literal("low"),
    };

    runtime.worker_mutex.lockUncancelable(io_mod.getIo());
    var state = EnqueueThreadState{};
    const thread = try std.Thread.spawn(.{}, runEnqueuePrompt, .{ &state, &runtime, "blocked", "model" });
    try waitForEnqueueThreadStart(&state);
    io_mod.sleep(10 * std.time.ns_per_ms);
    runtime.agent_turn_settings.fast_mode = true;
    runtime.agent_turn_settings.effort = types.ReasoningEffort.literal("high");
    runtime.worker_mutex.unlock(io_mod.getIo());
    thread.join();

    try std.testing.expect(state.err == null);
    try std.testing.expectEqual(@as(usize, 1), runtime.queued_prompts.items.len);
    try std.testing.expect(runtime.queued_prompts.items[0].agent_settings.fast_mode);
    try std.testing.expectEqual(types.ReasoningEffort.literal("high"), runtime.queued_prompts.items[0].agent_settings.effort);
}

test "taken prompt keeps original fast mode and effort after later sync" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    runtime.agent_turn_settings = .{
        .fast_mode = true,
        .effort = types.ReasoningEffort.literal("high"),
    };
    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "active", "model"));

    const job = (try runtime.waitAndTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, job);

    runtime.syncQueuedPromptFastMode(false);
    runtime.syncQueuedPromptEffort(types.ReasoningEffort.literal("low"));

    try std.testing.expect(job.agent_settings.fast_mode);
    try std.testing.expectEqual(types.ReasoningEffort.literal("high"), job.agent_settings.effort);
    try std.testing.expect(!runtime.agent_turn_settings.fast_mode);
    try std.testing.expectEqual(types.ReasoningEffort.literal("low"), runtime.agent_turn_settings.effort);
}

test "effectiveAgentTurnSettings prefers active turn settings until cleared" {
    var runtime = WorkerRuntime{};
    defer runtime.deinit(std.testing.allocator);

    runtime.agent_turn_settings = .{
        .max_tool_result_bytes = 2048,
        .first_call_tool_choice = .auto,
        .fast_mode = false,
        .effort = types.ReasoningEffort.literal("low"),
    };
    try std.testing.expectEqual(@as(usize, 2048), runtime.effectiveAgentTurnSettings().max_tool_result_bytes);
    try std.testing.expect(!runtime.effectiveAgentTurnSettings().fast_mode);
    try std.testing.expectEqual(types.ReasoningEffort.literal("low"), runtime.effectiveAgentTurnSettings().effort);

    runtime.setActiveAgentTurnSettings(.{
        .max_tool_result_bytes = 4096,
        .first_call_tool_choice = .none,
        .fast_mode = true,
        .effort = types.ReasoningEffort.literal("high"),
    });

    const active = runtime.effectiveAgentTurnSettings();
    try std.testing.expectEqual(@as(usize, 4096), active.max_tool_result_bytes);
    try std.testing.expectEqual(types.ToolChoice.none, active.first_call_tool_choice);
    try std.testing.expect(active.fast_mode);
    try std.testing.expectEqual(types.ReasoningEffort.literal("high"), active.effort);

    runtime.clearActiveAgentTurnSettings();
    const fallback = runtime.effectiveAgentTurnSettings();
    try std.testing.expectEqual(@as(usize, 2048), fallback.max_tool_result_bytes);
    try std.testing.expect(!fallback.fast_mode);
    try std.testing.expectEqual(types.ReasoningEffort.literal("low"), fallback.effort);
}

test "queued permission mode updates without mutating dequeued fallback" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    var held_prompt = try makePrompt(alloc, "held", "model");
    held_prompt.permission_mode = .ask;
    try runtime.enqueuePrompt(alloc, held_prompt);
    var queued_prompt = try makePrompt(alloc, "queued", "model");
    queued_prompt.permission_mode = .ask;
    try runtime.enqueuePrompt(alloc, queued_prompt);

    const active = (try runtime.waitAndTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, active);
    runtime.syncQueuedPromptPermissionSnapshot(.{
        .mode = .auto,
    });

    try std.testing.expectEqual(types.PermissionMode.ask, active.permission_mode);
    try std.testing.expectEqual(types.PermissionMode.auto, runtime.queued_prompts.items[0].permission_mode);
}

test "submitted text only queues while a prompt is active" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "use all your tools", "model"));
    const active = (try runtime.waitAndTakeNextPrompt(alloc)).?;
    defer freeQueuedPrompt(alloc, active);
    try std.testing.expect(runtime.worker_processing);
    try std.testing.expect(!runtime.isCancelRequested());

    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared =
        try permission_request.OwnedPermissionRequest.dupe(
            alloc,
            .{ .id = 1, .label = "shell.run launch chrome" },
        );
    const question_options = [_]types.QuestionOption{.{ .label = "Wait", .description = null }};
    const question_entries = [_]types.QuestionBatchEntry{.{
        .question = "Keep waiting?",
        .options = &question_options,
    }};
    runtime.pending_question_shared = try dupeOwnedQuestionBatch(
        alloc,
        &question_entries,
        .agent_question,
    );

    const submitted = [_][]const u8{
        "Stop",
        "continue",
        "What are you doing right now?",
        "What files are in this repository?",
        "checkpoint?",
    };
    for (submitted) |text| {
        try runtime.enqueuePrompt(alloc, try makePrompt(alloc, text, "model"));
        try std.testing.expect(!runtime.isCancelRequested());
        try std.testing.expect(runtime.pending_permission_response == null);
        try std.testing.expect(runtime.pending_permission_request_shared != null);
        try std.testing.expect(runtime.pending_question_shared != null);
        try std.testing.expect(runtime.pending_question_response == .pending);
    }

    try std.testing.expectEqual(submitted.len, runtime.queuedPromptCount());
    for (submitted, runtime.queued_prompts.items) |expected, queued| {
        try std.testing.expectEqualStrings(expected, queued.prompt);
    }
}

test "prompt take and grant append allocation failures preserve state" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try runtime.enqueuePrompt(alloc, try makePromptWithGrant(alloc, "keep", "model", "read_file", "/tmp/a"));
    runtime.worker_cancel_requested.store(true, .seq_cst);
    var failing_take = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 0 });
    try std.testing.expectError(error.OutOfMemory, runtime.waitAndTakeNextPrompt(failing_take.allocator()));
    try std.testing.expectEqual(@as(usize, 1), runtime.queuedPromptCount());
    try std.testing.expectEqualStrings("keep", runtime.queued_prompts.items[0].prompt);
    try std.testing.expect(runtime.isCancelRequested());
    try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);

    const original = runtime.queued_prompts.items[0].grants.ptr;
    var failing_grant = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 1 });
    try std.testing.expectError(error.OutOfMemory, runtime.propagateGrant(failing_grant.allocator(), "write_file", "/tmp/b"));
    try std.testing.expectEqual(original, runtime.queued_prompts.items[0].grants.ptr);
    try std.testing.expectEqual(@as(usize, 1), runtime.queued_prompts.items[0].grants.len);

    var failing_slice = std.testing.FailingAllocator.init(alloc, .{ .fail_index = 2 });
    try std.testing.expectError(error.OutOfMemory, runtime.propagateGrant(failing_slice.allocator(), "write_file", "/tmp/c"));
    try std.testing.expectEqual(original, runtime.queued_prompts.items[0].grants.ptr);
    try std.testing.expectEqual(@as(usize, 1), runtime.queued_prompts.items[0].grants.len);
}

test "clear queued prompts preserves events and discard frees event payloads" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try runtime.enqueuePrompt(alloc, try makePrompt(alloc, "drop", "model"));
    try runtime.pushEvent(alloc, .{ .assistant_presentation = .{
        .text = @constCast("kept"),
    } });
    runtime.clearQueuedPrompts(alloc, &.{});
    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
    try std.testing.expectEqual(@as(usize, 0), runtime.queued_prompts.items.len);
    try std.testing.expectEqual(@as(usize, 1), runtime.worker_events.items.len);

    runtime.discardEvents(alloc);
    try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);

    freeQueuedPrompt(alloc, try makePrompt(alloc, "free", "model"));
    freeWorkerEvent(alloc, .{ .begin_prompt = try types.dupeUserTurn(alloc, .{ .text = @constCast("user"), .images = &.{} }) });
    freeWorkerEvent(alloc, .{ .assistant_presentation = .{
        .text = try alloc.dupe(u8, "append"),
    } });
    const table = try assistant_presentation.parseTablePayload(alloc, "| H |\n|---|\n| v |\n");
    freeWorkerEvent(alloc, .{ .assistant_presentation = .{ .table = table } });
    freeWorkerEvent(alloc, try dupeWorkerEvent(alloc, .{ .semantic_notice = .{
        .topic = "system",
        .tone = .information,
        .body = "notice",
    } }));
    freeWorkerEvent(alloc, .{ .route_recovery_status = .{ .kind = .auto_retry, .failed_attempt = 1, .attempt_limit = 3 } });
    freeWorkerEvent(alloc, .clear_route_recovery_status);
    freeWorkerEvent(alloc, .{ .command_output = .{ .stream = .stdout, .text = try alloc.dupe(u8, "out") } });
    freeWorkerEvent(alloc, .{ .command_output_complete = null });
    freeWorkerEvent(alloc, .{ .command_output_complete = .{ .turn_id = 1, .call_id = try alloc.dupe(u8, "1") } });
    freeWorkerEvent(alloc, .{ .tool_lifecycle = .{
        .provisional = .{
            .id = .{ .turn_id = 1, .call_id = try alloc.dupe(u8, "1") },
            .tool_name = try alloc.dupe(u8, "read_file"),
            .activity_kind = .read,
        },
    } });
    freeWorkerEvent(alloc, .{ .tool_lifecycle = .{
        .progress = .{
            .id = .{ .turn_id = 1, .call_id = try alloc.dupe(u8, "1") },
            .text = try alloc.dupe(u8, "reading"),
        },
    } });
    freeWorkerEvent(alloc, .{ .tool_lifecycle = .{
        .terminal = .{
            .id = .{ .turn_id = 1, .call_id = try alloc.dupe(u8, "1") },
            .outcome = .{
                .kind = .completed,
                .summary = try alloc.dupe(u8, "read"),
            },
        },
    } });
    freeWorkerEvent(alloc, .{ .turn_token_update = .{ .input_tokens = 10, .output_tokens = 5 } });
    freeWorkerEvent(alloc, .{ .diff_block = .{ .preview = try alloc.dupe(u8, "preview") } });
    freeWorkerEvent(alloc, .{ .tool_lifecycle = .{
        .turn_finished = .{ .turn_id = 1, .outcome = .completed },
    } });
    freeWorkerEvent(alloc, .{ .finish_prompt = .{ .turn = .{ .compacted_summary = .{
        .summary = try alloc.dupe(u8, "summary"),
        .removed_turn_count = 1,
        .compaction_count = 1,
    } } } });
    freeWorkerEvent(alloc, .{ .session_grant = try makeGrant(alloc, "read_file", "/tmp/a") });
    freeWorkerEvent(alloc, try dupeWorkerEvent(alloc, .{ .error_text = .{
        .topic = "system",
        .tone = .@"error",
        .body = "err",
    } }));
}

test "typed lifecycle worker events duplicate and free every payload variant" {
    const alloc = std.testing.allocator;
    const events = [_]types.ToolLifecycleEvent{
        .{ .provisional = .{
            .id = .{ .turn_id = 1, .call_id = "provisional" },
            .presentation_group_id = .{ .turn_id = 1, .anchor_step_id = 3 },
            .tool_name = "read_file",
            .activity_kind = .read,
        } },
        .{ .provisional = .{
            .id = .{ .turn_id = 1, .call_id = "provisional" },
            .tool_name = "glob_files",
            .activity_kind = .list,
        } },
        .{ .authoritative_started = .{
            .id = .{ .turn_id = 1, .call_id = "final" },
            .presentation_group_id = .{ .turn_id = 1, .anchor_step_id = 3 },
            .reconciles_provisional_call_id = "provisional",
            .tool_name = "glob_files",
            .activity_kind = .list,
        } },
        .{ .progress = .{
            .id = .{ .turn_id = 1, .call_id = "final" },
            .text = "Listing files",
        } },
        .{ .terminal = .{
            .id = .{ .turn_id = 1, .call_id = "final" },
            .outcome = .{ .kind = .completed, .summary = "Listed files" },
            .command_artifact_handle = "fx-command-final.log",
        } },
        .{ .turn_finished = .{ .turn_id = 1, .outcome = .completed } },
    };

    for (events) |event| {
        const owned = try dupeToolLifecycleEvent(alloc, event);
        freeToolLifecycleEvent(alloc, owned);
    }
}

fn checkToolLifecycleDupAllocationFailure(alloc: std.mem.Allocator) !void {
    const owned = try dupeToolLifecycleEvent(alloc, .{ .authoritative_started = .{
        .id = .{ .turn_id = 7, .call_id = "final" },
        .reconciles_provisional_call_id = "provisional",
        .tool_name = "read_file",
        .activity_kind = .read,
    } });
    defer freeToolLifecycleEvent(alloc, owned);

    try std.testing.expectEqualStrings(
        "final",
        owned.authoritative_started.id.call_id,
    );
    try std.testing.expectEqualStrings(
        "provisional",
        owned.authoritative_started.reconciles_provisional_call_id.?,
    );
    try std.testing.expectEqualStrings(
        "read_file",
        owned.authoritative_started.tool_name,
    );
}

test "typed lifecycle duplication frees every partial allocation" {
    try std.testing.checkAllAllocationFailures(
        std.testing.allocator,
        checkToolLifecycleDupAllocationFailure,
        .{},
    );
}

test "pushEvent copies session grants into queue allocator" {
    const source_alloc = std.testing.allocator;
    const queue_alloc = std.heap.c_allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(queue_alloc);

    const grant = try makeGrant(source_alloc, "read_file", "/tmp/source.zig");
    const tool_ptr = grant.tool_name.ptr;
    const target_ptr = grant.target_path.ptr;

    try runtime.pushEvent(queue_alloc, .{ .session_grant = grant });
    source_alloc.free(grant.tool_name);
    source_alloc.free(grant.target_path);

    var events = runtime.takeEvents();
    defer freeEventList(queue_alloc, &events);

    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .session_grant);
    try std.testing.expectEqualStrings("read_file", events.items[0].session_grant.tool_name);
    try std.testing.expectEqualStrings("/tmp/source.zig", events.items[0].session_grant.target_path);
    try std.testing.expect(events.items[0].session_grant.tool_name.ptr != tool_ptr);
    try std.testing.expect(events.items[0].session_grant.target_path.ptr != target_ptr);
}

test "pushEvent copies string events into queue allocator" {
    const source_alloc = std.testing.allocator;
    const queue_alloc = std.heap.c_allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(queue_alloc);

    const text = try source_alloc.dupe(u8, "borrowed assistant text");
    const text_ptr = text.ptr;

    try runtime.pushEvent(queue_alloc, .{ .assistant_presentation = .{ .text = text } });
    source_alloc.free(text);

    var events = runtime.takeEvents();
    defer freeEventList(queue_alloc, &events);

    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .assistant_presentation);
    try std.testing.expect(events.items[0].assistant_presentation == .text);
    try std.testing.expectEqualStrings("borrowed assistant text", events.items[0].assistant_presentation.text);
    try std.testing.expect(events.items[0].assistant_presentation.text.ptr != text_ptr);
}

test "pushEvent copies semantic table payloads into the queue allocator" {
    const source_alloc = std.testing.allocator;
    const queue_alloc = std.heap.c_allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(queue_alloc);

    var table = try assistant_presentation.parseTablePayload(
        source_alloc,
        "| Name | Count |\n" ++
            "|------|------:|\n" ++
            "| api | 7 |\n",
    );
    const source_cell_ptr = table.rows[1].cells[0].ptr;
    try runtime.pushEvent(queue_alloc, .{ .assistant_presentation = .{ .table = table } });
    table.deinit(source_alloc);

    var events = runtime.takeEvents();
    defer freeEventList(queue_alloc, &events);

    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .assistant_presentation);
    try std.testing.expect(events.items[0].assistant_presentation == .table);
    try std.testing.expectEqualStrings("api", events.items[0].assistant_presentation.table.rows[1].cells[0]);
    try std.testing.expect(events.items[0].assistant_presentation.table.rows[1].cells[0].ptr != source_cell_ptr);
}

test "pushEvent copies code block payload into queue allocator" {
    const source_alloc = std.testing.allocator;
    const queue_alloc = std.heap.c_allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(queue_alloc);

    var block = assistant_presentation.CodeBlockPayload{
        .language = try source_alloc.dupe(u8, "zig"),
        .code = try source_alloc.dupe(u8, "  const value = 1;\n"),
    };
    const source_code_ptr = block.code.ptr;
    try runtime.pushEvent(queue_alloc, .{ .assistant_presentation = .{ .code_block = block } });
    block.deinit(source_alloc);

    var events = runtime.takeEvents();
    defer freeEventList(queue_alloc, &events);

    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .assistant_presentation);
    try std.testing.expect(events.items[0].assistant_presentation == .code_block);
    try std.testing.expectEqualStrings("zig", events.items[0].assistant_presentation.code_block.language);
    try std.testing.expectEqualStrings("  const value = 1;\n", events.items[0].assistant_presentation.code_block.code);
    try std.testing.expect(events.items[0].assistant_presentation.code_block.code.ptr != source_code_ptr);
}

test "pushEvent copies absolute turn token progress into queue allocator" {
    const queue_alloc = std.heap.c_allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(queue_alloc);

    const progress: types.TurnTokenProgress = .{
        .input_tokens = 10,
        .output_tokens = 5,
        .input_exact = false,
        .output_exact = false,
    };
    try runtime.pushEvent(queue_alloc, .{ .turn_token_update = progress });

    var events = runtime.takeEvents();
    defer freeEventList(queue_alloc, &events);

    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .turn_token_update);
    try std.testing.expectEqual(progress, events.items[0].turn_token_update);
}

test "pushEvent copies finish prompt assistant turns into queue allocator" {
    const source_alloc = std.testing.allocator;
    const queue_alloc = std.heap.c_allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(queue_alloc);

    const turn: types.HistoryTurn = .{ .assistant = .{
        .user = .{ .text = try source_alloc.dupe(u8, "source prompt"), .images = &.{} },
        .assistant = try source_alloc.dupe(u8, "source answer"),
    } };
    const prompt_ptr = turn.assistant.user.text.ptr;
    const answer_ptr = turn.assistant.assistant.ptr;

    try runtime.pushEvent(queue_alloc, .{ .finish_prompt = .{
        .turn = turn,
        .summary = .{
            .thinking_duration_ms = 1,
            .turn_duration_ms = 2,
            .token_progress = .{ .input_tokens = 7 },
        },
    } });
    types.freeHistoryTurn(source_alloc, turn);

    var events = runtime.takeEvents();
    defer freeEventList(queue_alloc, &events);

    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .finish_prompt);
    try std.testing.expect(events.items[0].finish_prompt.turn == .assistant);
    try std.testing.expectEqualStrings("source prompt", events.items[0].finish_prompt.turn.assistant.user.text);
    try std.testing.expectEqualStrings("source answer", events.items[0].finish_prompt.turn.assistant.assistant);
    try std.testing.expect(events.items[0].finish_prompt.turn.assistant.user.text.ptr != prompt_ptr);
    try std.testing.expect(events.items[0].finish_prompt.turn.assistant.assistant.ptr != answer_ptr);
    try std.testing.expectEqual(@as(u64, 7), events.items[0].finish_prompt.summary.?.token_progress.input_tokens);
}

test "pushOwnedEvent transfers diff block payloads without copying" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    const preview = try alloc.dupe(u8, "preview");
    try runtime.pushOwnedEvent(alloc, .{ .diff_block = .{
        .preview = preview,
    } });

    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);

    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .diff_block);
    try std.testing.expectEqual(preview.ptr, events.items[0].diff_block.preview.ptr);
}

test "dupeWorkerEvent clones canonical diff payload and full detail" {
    const alloc = std.testing.allocator;
    const source: WorkerEvent = .{ .diff_block = .{
        .preview = @constCast("preview"),
        .additions = 2,
        .deletions = 1,
        .full = .{
            .content = @constCast("full detail"),
            .lifecycle_id = .{ .turn_id = 9, .call_id = "call-9" },
        },
    } };
    const owned = try dupeWorkerEvent(alloc, source);
    defer freeWorkerEvent(alloc, owned);

    try std.testing.expectEqualStrings("preview", owned.diff_block.preview);
    try std.testing.expect(source.diff_block.preview.ptr != owned.diff_block.preview.ptr);
    try std.testing.expectEqual(@as(u32, 2), owned.diff_block.additions);
    try std.testing.expectEqual(@as(u32, 1), owned.diff_block.deletions);
    const full = owned.diff_block.full orelse return error.TestExpectedFullDiff;
    try std.testing.expectEqualStrings("full detail", full.content);
    try std.testing.expectEqual(@as(u64, 9), full.lifecycle_id.turn_id);
    try std.testing.expectEqualStrings("call-9", full.lifecycle_id.call_id);
    try std.testing.expect(source.diff_block.full.?.content.ptr != full.content.ptr);
    try std.testing.expect(source.diff_block.full.?.lifecycle_id.call_id.ptr != full.lifecycle_id.call_id.ptr);
}

test "dupeWorkerEvent owns a complete refreshed credential publication" {
    const alloc = std.testing.allocator;
    const source: WorkerEvent = .{ .credential_refreshed = .{
        .token = @constCast("fresh-token"),
        .source = .fx_login,
        .team_id = @constCast("team_123"),
        .refresh_after_ms = 100,
    } };
    var owned = try dupeWorkerEvent(alloc, source);
    defer freeWorkerEvent(alloc, owned);

    const credential = &owned.credential_refreshed;
    try std.testing.expectEqualStrings("fresh-token", credential.token);
    try std.testing.expect(credential.token.ptr != source.credential_refreshed.token.ptr);
    try std.testing.expectEqualStrings("team_123", credential.team_id.?);
    try std.testing.expectEqual(@as(?i64, 100), credential.refresh_after_ms);
}

test "pushEvent clones every semantic notice field into the queue allocator" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    var topic = [_]u8{ 'c', 'o', 'n', 't', 'e', 'x', 't' };
    var body = [_]u8{ 'w', 'a', 'r', 'n', 'i', 'n', 'g' };
    const topic_ptr = topic[0..].ptr;
    const body_ptr = body[0..].ptr;
    try runtime.pushEvent(alloc, .{ .semantic_notice = .{
        .topic = &topic,
        .tone = .warning,
        .body = &body,
        .visibility = .full_only,
    } });
    topic[0] = 'X';
    body[0] = 'X';

    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);
    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .semantic_notice);
    const notice = events.items[0].semantic_notice;
    try std.testing.expectEqualStrings("context", notice.topic);
    try std.testing.expectEqual(types.NoticeTone.warning, notice.tone);
    try std.testing.expectEqualStrings("warning", notice.body);
    try std.testing.expectEqual(types.NoticeVisibility.full_only, notice.visibility);
    try std.testing.expect(notice.topic.ptr != topic_ptr);
    try std.testing.expect(notice.body.ptr != body_ptr);
}

fn checkSemanticWorkerEventDuplicationAllocationFailure(alloc: std.mem.Allocator) !void {
    const ordinary = try dupeWorkerEvent(alloc, .{ .semantic_notice = .{
        .topic = "context",
        .tone = .warning,
        .body = "source limit reached",
        .visibility = .full_only,
    } });
    defer freeWorkerEvent(alloc, ordinary);

    const error_event = try dupeWorkerEvent(alloc, .{ .error_text = .{
        .topic = "github",
        .tone = .@"error",
        .body = "publish failed",
    } });
    defer freeWorkerEvent(alloc, error_event);

    try std.testing.expectEqualStrings("context", ordinary.semantic_notice.topic);
    try std.testing.expectEqualStrings("source limit reached", ordinary.semantic_notice.body);
    try std.testing.expectEqualStrings("github", error_event.error_text.topic);
    try std.testing.expectEqualStrings("publish failed", error_event.error_text.body);
}

test "ordinary and error semantic worker events free partial duplication allocations" {
    try std.testing.checkAllAllocationFailures(
        std.testing.allocator,
        checkSemanticWorkerEventDuplicationAllocationFailure,
        .{},
    );
}

test "prependOwnedEvents preserves retained order ahead of newer events" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try runtime.pushEvent(alloc, .{ .semantic_notice = .{
        .topic = "system",
        .tone = .information,
        .body = "newer",
    } });
    const retained = [_]WorkerEvent{
        .{ .assistant_presentation = .{
            .text = try alloc.dupe(u8, "first"),
        } },
        try dupeWorkerEvent(alloc, .{ .semantic_notice = .{
            .topic = "system",
            .tone = .information,
            .body = "second",
        } }),
    };
    errdefer {
        for (retained) |event| freeWorkerEvent(alloc, event);
    }
    try runtime.prependOwnedEvents(alloc, &retained);

    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);
    try std.testing.expectEqual(@as(usize, 3), events.items.len);
    try std.testing.expectEqualStrings("first", events.items[0].assistant_presentation.text);
    try std.testing.expectEqualStrings("second", events.items[1].semantic_notice.body);
    try std.testing.expectEqualStrings("newer", events.items[2].semantic_notice.body);
}

fn checkSemanticNoticeEnqueueAllocationFailure(alloc: std.mem.Allocator) !void {
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);

    try runtime.pushEvent(alloc, .{ .semantic_notice = .{
        .topic = "permissions",
        .tone = .success,
        .body = "approval",
        .visibility = .full_only,
    } });
    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);
    try std.testing.expectEqual(@as(usize, 1), events.items.len);
    try std.testing.expect(events.items[0] == .semantic_notice);
    try std.testing.expectEqualStrings("permissions", events.items[0].semantic_notice.topic);
    try std.testing.expectEqualStrings("approval", events.items[0].semantic_notice.body);
}

test "semantic notice enqueue frees partial topic body and queue allocations" {
    try std.testing.checkAllAllocationFailures(
        std.testing.allocator,
        checkSemanticNoticeEnqueueAllocationFailure,
        .{},
    );
}

const PermissionThreadState = struct {
    decision: ?types.ToolPermissionDecision = null,
    err: ?anyerror = null,
};

fn runPermissionRequest(state: *PermissionThreadState, runtime: *WorkerRuntime, label: []const u8) void {
    var response = runtime.requestPermissionBlocking(
        std.testing.allocator,
        .{ .label = label },
    ) catch |err| {
        state.err = err;
        return;
    };
    defer response.deinit();
    state.decision = response.decision;
}

fn submitPermissionDecisionForTest(
    runtime: *WorkerRuntime,
    request_id: u64,
    decision: types.ToolPermissionDecision,
) PermissionSubmissionResult {
    return runtime.submitPermissionResponse(
        request_id,
        permission_request.OwnedPermissionResponse.init(
            std.testing.allocator,
            decision,
            null,
        ),
    );
}

fn waitForPermissionLabel(worker: *WorkerRuntime, expected: []const u8) !u64 {
    var attempts: usize = 0;
    while (attempts < 100) : (attempts += 1) {
        var snapshot = try worker.snapshotState(std.testing.allocator);
        defer snapshot.deinit(std.testing.allocator);
        if (snapshot.pending_permission_request) |request| {
            try std.testing.expectEqualStrings(expected, request.label);
            try std.testing.expect(request.id != 0);
            return request.id;
        }
        io_mod.sleep(1 * std.time.ns_per_ms);
    }
    return error.TestExpectedEqual;
}

test "permission blocking handles submit and stop" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;

    var state = PermissionThreadState{};
    const thread = try std.Thread.spawn(.{}, runPermissionRequest, .{ &state, &runtime, "Allow?" });
    const request_id = try waitForPermissionLabel(&runtime, "Allow?");
    try std.testing.expectEqual(
        PermissionSubmissionResult.accepted,
        submitPermissionDecisionForTest(&runtime, request_id, .once),
    );
    thread.join();
    try std.testing.expect(state.err == null);
    try std.testing.expectEqual(types.ToolPermissionDecision.once, state.decision.?);
    var snapshot = try runtime.snapshotState(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expect(snapshot.pending_permission_request == null);

    var deny_state = PermissionThreadState{};
    const deny_thread = try std.Thread.spawn(.{}, runPermissionRequest, .{ &deny_state, &runtime, "Deny?" });
    const deny_request_id = try waitForPermissionLabel(&runtime, "Deny?");
    try std.testing.expectEqual(
        PermissionSubmissionResult.accepted,
        submitPermissionDecisionForTest(&runtime, deny_request_id, .deny),
    );
    deny_thread.join();
    try std.testing.expect(deny_state.err == null);
    try std.testing.expectEqual(types.ToolPermissionDecision.deny, deny_state.decision.?);
    var deny_snapshot = try runtime.snapshotState(alloc);
    defer deny_snapshot.deinit(alloc);
    try std.testing.expect(deny_snapshot.pending_permission_request == null);

    var stop_runtime = WorkerRuntime{};
    defer stop_runtime.deinit(alloc);
    stop_runtime.worker_processing = true;
    var stop_state = PermissionThreadState{};
    const stop_thread = try std.Thread.spawn(.{}, runPermissionRequest, .{ &stop_state, &stop_runtime, "Stop?" });
    _ = try waitForPermissionLabel(&stop_runtime, "Stop?");
    stop_runtime.requestStop();
    stop_thread.join();
    try std.testing.expect(stop_state.err == null);
    try std.testing.expectEqual(types.ToolPermissionDecision.deny, stop_state.decision.?);
    var stop_snapshot = try stop_runtime.snapshotState(alloc);
    defer stop_snapshot.deinit(alloc);
    try std.testing.expect(stop_snapshot.pending_permission_request == null);

    var shutdown_runtime = WorkerRuntime{};
    defer shutdown_runtime.deinit(alloc);
    shutdown_runtime.worker_processing = true;
    var shutdown_state = PermissionThreadState{};
    const shutdown_thread = try std.Thread.spawn(.{}, runPermissionRequest, .{ &shutdown_state, &shutdown_runtime, "Shutdown?" });
    _ = try waitForPermissionLabel(&shutdown_runtime, "Shutdown?");
    shutdown_runtime.requestShutdown();
    shutdown_thread.join();
    try std.testing.expect(shutdown_state.err == null);
    try std.testing.expectEqual(types.ToolPermissionDecision.deny, shutdown_state.decision.?);
    var shutdown_snapshot = try shutdown_runtime.snapshotState(alloc);
    defer shutdown_snapshot.deinit(alloc);
    try std.testing.expect(shutdown_snapshot.pending_permission_request == null);
}

test "permission response accepts feedback ownership" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared = try permission_request.OwnedPermissionRequest.dupe(
        alloc,
        .{ .id = 44, .label = "feedback" },
    );

    const feedback = try alloc.dupe(u8, "summarize the result");
    try std.testing.expectEqual(
        PermissionSubmissionResult.accepted,
        runtime.submitPermissionResponse(
            44,
            permission_request.OwnedPermissionResponse.init(alloc, .once, feedback),
        ),
    );
    try std.testing.expectEqualStrings(
        "summarize the result",
        runtime.pending_permission_response.?.feedback.?,
    );
}

test "permission submission rejects stale ids and preserves first resolution" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared =
        try permission_request.OwnedPermissionRequest.dupe(
            alloc,
            .{ .id = 22, .label = "B" },
        );

    try std.testing.expectEqual(
        PermissionSubmissionResult.stale,
        submitPermissionDecisionForTest(&runtime, 21, .always),
    );
    try std.testing.expect(runtime.pending_permission_response == null);
    try std.testing.expectEqual(
        PermissionSubmissionResult.accepted,
        submitPermissionDecisionForTest(&runtime, 22, .deny),
    );
    try std.testing.expectEqual(
        types.ToolPermissionDecision.deny,
        runtime.pending_permission_response.?.decision,
    );
    try std.testing.expectEqual(
        PermissionSubmissionResult.no_pending,
        submitPermissionDecisionForTest(&runtime, 22, .once),
    );
    try std.testing.expectEqual(
        PermissionSubmissionResult.no_pending,
        submitPermissionDecisionForTest(&runtime, 22, .always),
    );
    try std.testing.expectEqual(
        PermissionSubmissionResult.no_pending,
        submitPermissionDecisionForTest(&runtime, 22, .deny),
    );
    try std.testing.expectEqual(
        types.ToolPermissionDecision.deny,
        runtime.pending_permission_response.?.decision,
    );

    var snapshot = try runtime.snapshotState(alloc);
    defer snapshot.deinit(alloc);
    try std.testing.expect(snapshot.pending_permission_request == null);
}

test "stale A decisions cannot resolve a newer B request" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;

    var a_state = PermissionThreadState{};
    const a_thread = try std.Thread.spawn(
        .{},
        runPermissionRequest,
        .{ &a_state, &runtime, "A" },
    );
    const a_id = try waitForPermissionLabel(&runtime, "A");
    try std.testing.expectEqual(
        PermissionSubmissionResult.accepted,
        submitPermissionDecisionForTest(&runtime, a_id, .once),
    );
    a_thread.join();
    try std.testing.expectEqual(types.ToolPermissionDecision.once, a_state.decision.?);

    var b_state = PermissionThreadState{};
    const b_thread = try std.Thread.spawn(
        .{},
        runPermissionRequest,
        .{ &b_state, &runtime, "B" },
    );
    const b_id = try waitForPermissionLabel(&runtime, "B");
    try std.testing.expect(b_id != a_id);

    const stale_decisions = [_]types.ToolPermissionDecision{
        .once,
        .always,
        .deny,
        .deny,
    };
    for (stale_decisions) |decision| {
        try std.testing.expectEqual(
            PermissionSubmissionResult.stale,
            submitPermissionDecisionForTest(&runtime, a_id, decision),
        );
    }
    try std.testing.expect(runtime.pending_permission_response == null);
    try std.testing.expectEqual(
        PermissionSubmissionResult.accepted,
        submitPermissionDecisionForTest(&runtime, b_id, .always),
    );
    b_thread.join();
    try std.testing.expectEqual(
        types.ToolPermissionDecision.always,
        b_state.decision.?,
    );
}

test "permission request ids never wrap or reuse zero" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.next_permission_request_id = std.math.maxInt(u64);

    var last_state = PermissionThreadState{};
    const last_thread = try std.Thread.spawn(
        .{},
        runPermissionRequest,
        .{ &last_state, &runtime, "last id" },
    );
    const last_id = try waitForPermissionLabel(&runtime, "last id");
    try std.testing.expectEqual(std.math.maxInt(u64), last_id);
    try std.testing.expectEqual(
        PermissionSubmissionResult.accepted,
        submitPermissionDecisionForTest(&runtime, last_id, .deny),
    );
    last_thread.join();
    try std.testing.expect(last_state.err == null);

    var exhausted_state = PermissionThreadState{};
    const exhausted_thread = try std.Thread.spawn(
        .{},
        runPermissionRequest,
        .{ &exhausted_state, &runtime, "exhausted" },
    );
    exhausted_thread.join();
    try std.testing.expectEqual(
        error.RequestIdExhausted,
        exhausted_state.err.?,
    );
}

test "approval turn cancellation resolves the current worker request once" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared =
        try permission_request.OwnedPermissionRequest.dupe(
            alloc,
            .{ .id = 33, .label = "current" },
        );

    runtime.cancelApprovalTurn();
    try std.testing.expect(runtime.isCancelRequested());
    try std.testing.expectEqual(
        types.ToolPermissionDecision.deny,
        runtime.pending_permission_response.?.decision,
    );
    try std.testing.expectEqual(
        PermissionSubmissionResult.no_pending,
        submitPermissionDecisionForTest(&runtime, 33, .always),
    );
}

test "approval cancellation leaves admitted steering available for the next turn" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;
    runtime.active_turn_id = 41;
    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared =
        try permission_request.OwnedPermissionRequest.dupe(
            alloc,
            .{ .id = 33, .label = "current" },
        );
    try runtime.admitInteractivePrompt(
        alloc,
        try makePrompt(alloc, "continue with the correction", "model"),
    );

    runtime.cancelApprovalTurn();
    runtime.finishProcessing();

    const next = (try runtime.tryTakeNextPrompt(alloc)) orelse
        return error.TestExpectedSteeringContinuation;
    defer freeQueuedPrompt(alloc, next);
    try std.testing.expect(next.delivery.isContinuation());
    try std.testing.expectEqualStrings("continue with the correction", next.prompt);
}

test "text enqueue allocation failure preserves active interactive state" {
    const alloc = std.testing.allocator;
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    runtime.worker_processing = true;

    runtime.pending_permission_waiting = true;
    runtime.pending_permission_request_shared =
        try permission_request.OwnedPermissionRequest.dupe(
            alloc,
            .{ .id = 3, .label = "active permission" },
        );
    const question_options = [_]types.QuestionOption{.{ .label = "Wait", .description = null }};
    const question_entries = [_]types.QuestionBatchEntry{.{
        .question = "Keep waiting?",
        .options = &question_options,
    }};
    runtime.pending_question_shared = try dupeOwnedQuestionBatch(
        alloc,
        &question_entries,
        .agent_question,
    );

    const prompt = try makePrompt(
        alloc,
        "Stop",
        "model",
    );
    var failing = std.testing.FailingAllocator.init(
        alloc,
        .{ .fail_index = 0 },
    );
    try std.testing.expectError(
        error.OutOfMemory,
        runtime.enqueuePrompt(failing.allocator(), prompt),
    );
    freeQueuedPrompt(alloc, prompt);

    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
    try std.testing.expect(!runtime.isCancelRequested());
    try std.testing.expect(runtime.pending_permission_response == null);
    try std.testing.expect(runtime.pending_permission_request_shared != null);
    try std.testing.expectEqualStrings(
        "active permission",
        runtime.pending_permission_request_shared.?.label,
    );
    try std.testing.expect(runtime.pending_question_shared != null);
    try std.testing.expect(runtime.pending_question_response == .pending);
}

const QuestionThreadState = struct {
    answers: ?[][]u8 = null,
    err: ?anyerror = null,
};

fn runQuestionRequest(state: *QuestionThreadState, runtime: *WorkerRuntime, entries: []const types.QuestionBatchEntry) void {
    state.answers = runtime.requestQuestionBatchAnswerBlocking(std.testing.allocator, entries) catch |err| {
        state.err = err;
        return;
    };
}

fn runRouteRecoveryRequest(state: *QuestionThreadState, runtime: *WorkerRuntime, entries: []const types.QuestionBatchEntry) void {
    state.answers = runtime.requestRouteRecoveryAnswerBlocking(std.testing.allocator, entries) catch |err| {
        state.err = err;
        return;
    };
}

fn waitForQuestionSnapshot(worker: *WorkerRuntime) !PendingQuestionBatchSnapshot {
    var attempts: usize = 0;
    while (attempts < 100) : (attempts += 1) {
        if (try worker.snapshotPendingQuestionBatch(std.testing.allocator)) |snapshot| return snapshot;
        io_mod.sleep(1 * std.time.ns_per_ms);
    }
    return error.TestExpectedEqual;
}

fn freeAnswers(alloc: std.mem.Allocator, answers: ?[][]u8) void {
    if (answers) |labels| {
        for (labels) |label| alloc.free(label);
        alloc.free(labels);
    }
}

test "question batch snapshot answer and cancellation" {
    const alloc = std.testing.allocator;
    const options = [_]types.QuestionOption{
        .{ .label = "Yes", .description = "Do it" },
        .{ .label = "No", .description = null },
    };
    const entries = [_]types.QuestionBatchEntry{.{ .question = "Continue?", .options = &options }};

    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    var state = QuestionThreadState{};
    const thread = try std.Thread.spawn(.{}, runQuestionRequest, .{ &state, &runtime, &entries });
    var snapshot = try waitForQuestionSnapshot(&runtime);
    defer snapshot.deinit(alloc);
    try std.testing.expectEqualStrings("Continue?", snapshot.entries[0].question);
    try std.testing.expectEqualStrings("Yes", snapshot.entries[0].options[0].label);
    try std.testing.expect(snapshot.entries[0].question.ptr != entries[0].question.ptr);
    try runtime.submitQuestionBatchAnswer(alloc, &.{"Yes"});
    thread.join();
    defer freeAnswers(alloc, state.answers);
    try std.testing.expect(state.err == null);
    try std.testing.expectEqualStrings("Yes", state.answers.?[0]);
    try std.testing.expect(try runtime.snapshotPendingQuestionBatch(alloc) == null);

    var cancel_runtime = WorkerRuntime{};
    defer cancel_runtime.deinit(alloc);
    var cancel_state = QuestionThreadState{};
    const cancel_thread = try std.Thread.spawn(.{}, runQuestionRequest, .{ &cancel_state, &cancel_runtime, &entries });
    var cancel_snapshot = try waitForQuestionSnapshot(&cancel_runtime);
    cancel_snapshot.deinit(alloc);
    try cancel_runtime.submitQuestionBatchAnswer(alloc, null);
    cancel_thread.join();
    try std.testing.expect(cancel_state.err == null);
    try std.testing.expect(cancel_state.answers == null);
    try std.testing.expect(try cancel_runtime.snapshotPendingQuestionBatch(alloc) == null);
}

test "question batch source distinguishes route recovery from agent questions" {
    const alloc = std.testing.allocator;
    const options = [_]types.QuestionOption{.{ .label = "Try again later", .description = null }};
    const entries = [_]types.QuestionBatchEntry{.{ .question = "Route failed. What should fx do?", .options = &options }};

    var route_runtime = WorkerRuntime{};
    defer route_runtime.deinit(alloc);
    var route_state = QuestionThreadState{};
    const route_thread = try std.Thread.spawn(.{}, runRouteRecoveryRequest, .{ &route_state, &route_runtime, &entries });
    var route_snapshot = try waitForQuestionSnapshot(&route_runtime);
    defer route_snapshot.deinit(alloc);
    try std.testing.expectEqual(QuestionPromptSource.route_recovery, route_snapshot.source);
    try std.testing.expectEqual(QuestionPromptSource.route_recovery, route_runtime.pendingQuestionBatchSource());
    try route_runtime.submitQuestionBatchAnswer(alloc, &.{"Try again later"});
    route_thread.join();
    defer freeAnswers(alloc, route_state.answers);
    try std.testing.expect(route_state.err == null);
    try std.testing.expectEqualStrings("Try again later", route_state.answers.?[0]);

    var agent_runtime = WorkerRuntime{};
    defer agent_runtime.deinit(alloc);
    var agent_state = QuestionThreadState{};
    const agent_thread = try std.Thread.spawn(.{}, runQuestionRequest, .{ &agent_state, &agent_runtime, &entries });
    var agent_snapshot = try waitForQuestionSnapshot(&agent_runtime);
    defer agent_snapshot.deinit(alloc);
    try std.testing.expectEqual(QuestionPromptSource.agent_question, agent_snapshot.source);
    try std.testing.expectEqual(QuestionPromptSource.agent_question, agent_runtime.pendingQuestionBatchSource());
    try agent_runtime.submitQuestionBatchAnswer(alloc, &.{"Try again later"});
    agent_thread.join();
    defer freeAnswers(alloc, agent_state.answers);
    try std.testing.expect(agent_state.err == null);
}

test "question request queues an ordered boundary after prior assistant text" {
    const alloc = std.testing.allocator;
    const options = [_]types.QuestionOption{.{ .label = "Yes", .description = null }};
    const entries = [_]types.QuestionBatchEntry{.{ .question = "Continue?", .options = &options }};

    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    try runtime.pushEvent(alloc, .{ .assistant_presentation = .{
        .text = @constCast("before question"),
    } });

    var state = QuestionThreadState{};
    const thread = try std.Thread.spawn(.{}, runQuestionRequest, .{ &state, &runtime, &entries });
    var snapshot = try waitForQuestionSnapshot(&runtime);
    defer snapshot.deinit(alloc);

    var events = runtime.takeEvents();
    defer freeEventList(alloc, &events);
    try std.testing.expectEqual(@as(usize, 2), events.items.len);
    try std.testing.expect(events.items[0] == .assistant_presentation);
    try std.testing.expect(events.items[0].assistant_presentation == .text);
    try std.testing.expectEqualStrings("before question", events.items[0].assistant_presentation.text);
    try std.testing.expect(events.items[1] == .question_requested);

    try runtime.submitQuestionBatchAnswer(alloc, &.{"Yes"});
    thread.join();
    defer freeAnswers(alloc, state.answers);
    try std.testing.expect(state.err == null);
    try std.testing.expectEqualStrings("Yes", state.answers.?[0]);
}

test "question request rolls back pending state when its boundary event cannot be queued" {
    const options = [_]types.QuestionOption{.{ .label = "Yes", .description = null }};
    const entries = [_]types.QuestionBatchEntry{.{ .question = "Continue?", .options = &options }};

    var probe = std.testing.FailingAllocator.init(std.testing.allocator, .{});
    var probe_runtime = WorkerRuntime{};
    probe_runtime.worker_stop_requested = true;
    const probe_result = try probe_runtime.requestQuestionBatchAnswerBlocking(probe.allocator(), &entries);
    try std.testing.expect(probe_result == null);
    const marker_append_alloc_index = probe.alloc_index - 1;
    probe_runtime.deinit(probe.allocator());
    try std.testing.expectEqual(probe.allocated_bytes, probe.freed_bytes);

    var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{
        .fail_index = marker_append_alloc_index,
    });
    var runtime = WorkerRuntime{};
    defer runtime.deinit(failing.allocator());
    runtime.worker_stop_requested = true;

    try std.testing.expectError(
        error.OutOfMemory,
        runtime.requestQuestionBatchAnswerBlocking(failing.allocator(), &entries),
    );
    try std.testing.expect(runtime.pending_question_shared == null);
    try std.testing.expect(runtime.pending_question_response == .pending);
    try std.testing.expectEqual(@as(usize, 0), runtime.worker_events.items.len);
}

test "discarding queued recovery releases metadata without deleting saved images" {
    const alloc = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "saved.bin", .data = "snapshot" });
    const path = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "saved.bin");
    defer alloc.free(path);
    const images = [_]types.ImageAttachment{.{
        .id = 1,
        .path = @constCast("/missing/source.png"),
        .media_type = @constCast("image/png"),
        .snapshot_path = path,
        .snapshot_sha256 = @constCast("a" ** 64),
    }};
    var runtime = WorkerRuntime{};
    defer runtime.deinit(alloc);
    var prompt = try makePrompt(alloc, "recover", "test/model");
    var owned = true;
    defer if (owned) freeQueuedPrompt(alloc, prompt);
    prompt.images = try types.dupeImageAttachmentSlice(alloc, &images);
    prompt.recovery_checkpoint = try (session_codec.RecoveryCheckpoint{
        .turn_id = 1,
        .user = .{ .text = @constCast("recover"), .images = @constCast(&images) },
        .assistant_source = @constCast(""),
        .cause = .network_interrupted,
        .action = .retrying_request,
        .authority = .{ .provider = .gateway, .model = @constCast("test/model") },
        .requested_fast_mode = false,
        .fast_mode = false,
        .max_provider_attempts = 10,
        .consumed_provider_attempts = 1,
    }).dupe(alloc);
    try runtime.enqueuePrompt(alloc, prompt);
    owned = false;
    runtime.clearQueuedPrompts(alloc, &.{});
    try std.testing.expectEqual(@as(usize, 0), runtime.queuedPromptCount());
    try std.Io.Dir.accessAbsolute(std.testing.io, path, .{});
}