a3s 0.10.0

//! Agent stream lifecycle, queue draining, and autonomous-run settlement.

use super::*;
use crate::tui::app_rewind::capture_rewind_checkpoint_seed;

fn take_matching_queue_claim<T>(
    active: &mut Option<PriorityItem<T>>,
    active_token: &mut Option<u64>,
    token: u64,
) -> Option<PriorityItem<T>> {
    if *active_token != Some(token) {
        return None;
    }
    *active_token = None;
    active.take()
}

impl App {
    /// Change the composer mode without mutating the semantics of an admitted
    /// turn. When no turn is active, keep Core's live policy in sync
    /// immediately so the footer and authorization boundary cannot diverge.
    pub(super) fn set_composer_mode(&mut self, mode: Mode) {
        self.mode = mode;
        if self.active_turn_mode.is_none() {
            self.execution_policy.set_mode(mode);
        }
    }

    pub(super) fn start_stream_inner(
        &mut self,
        prompt: String,
        display_task: String,
        clear_turn_artifacts: bool,
        include_attachments: bool,
        synthesis: bool,
    ) -> Option<Cmd<Msg>> {
        self.start_stream_inner_with_runtime(
            prompt,
            display_task,
            clear_turn_artifacts,
            include_attachments,
            synthesis,
            None,
        )
    }

    pub(super) fn start_stream_inner_with_runtime(
        &mut self,
        prompt: String,
        display_task: String,
        clear_turn_artifacts: bool,
        include_attachments: bool,
        synthesis: bool,
        runtime_expectation: Option<RuntimeExpectation>,
    ) -> Option<Cmd<Msg>> {
        let execution_mode = self.mode;
        let submitted_images = if include_attachments {
            std::mem::take(&mut self.pending_images)
        } else {
            Vec::new()
        };
        self.start_stream_inner_with_runtime_and_images(
            prompt,
            display_task,
            clear_turn_artifacts,
            synthesis,
            runtime_expectation,
            submitted_images,
            execution_mode,
            None,
            false,
        )
    }

    #[allow(clippy::too_many_arguments)]
    fn start_stream_inner_with_runtime_and_images(
        &mut self,
        prompt: String,
        display_task: String,
        clear_turn_artifacts: bool,
        synthesis: bool,
        runtime_expectation: Option<RuntimeExpectation>,
        submitted_images: Vec<PendingImage>,
        execution_mode: Mode,
        plan_draft: Option<PlanDraftRequest>,
        capture_rewind: bool,
    ) -> Option<Cmd<Msg>> {
        let attachments = match submitted_images
            .iter()
            .map(PendingImage::attachment)
            .collect::<std::io::Result<Vec<_>>>()
        {
            Ok(attachments) => attachments,
            Err(error) => {
                restore_submitted_images(&mut self.pending_images, submitted_images);
                self.push_notice(
                    NoticeKind::Error,
                    format!("Image attachment could not be read: {error}"),
                );
                self.relayout();
                return None;
            }
        };
        self.active_turn_mode = Some(execution_mode);
        self.active_plan_draft = plan_draft;
        self.execution_policy.set_mode(execution_mode);
        if clear_turn_artifacts && !synthesis {
            self.auto_review.on_user_turn();
            self.last_activity = Instant::now();
        }
        self.streaming.clear();
        self.llm_turn_checkpoint = None;
        self.got_delta = false; // track if this turn streamed any text deltas
        self.turn_text.clear();
        self.turn_had_agent_activity = false;
        self.turn_text_after_activity = false;
        if let Some(expectation) = runtime_expectation {
            self.runtime_expectation = Some(expectation);
        }
        self.stream_join_settling = false;
        self.stream_settle_abort = None;
        self.ultracode_synthesis_inflight = synthesis;
        if !synthesis {
            self.ultracode_synthesis_used = false;
        }
        self.last_paint = None; // first delta of the turn paints immediately
        self.viewport.set_auto_scroll(true); // sending a message jumps to latest
        if clear_turn_artifacts {
            self.plan.clear(); // fresh plan per user turn; planning events refill it

            // Keep completed agents visible until the next user turn; a fresh
            // user turn starts a fresh runtime-entity projection.
            self.runtime.clear_turn_entities();
            self.runtime.set_subagent_task(display_task.clone());
        } else {
            self.runtime.clear_live_tools();
        }
        let rewind_task = display_task.clone();
        self.running_task = Some(display_task);
        self.state = State::Streaming;
        self.relayout();
        self.stream_started = Some(Instant::now());
        self.spinner.start();
        self.rebuild_viewport();
        let session = self.session.clone();
        let rewind_request = capture_rewind.then(|| {
            (
                self.store.clone(),
                self.session_id.clone(),
                PathBuf::from(&self.cwd),
                rewind_task,
                session.history(),
            )
        });
        // Keep the agent aligned with the standing goal (display stays clean).
        // Internal synthesis keeps its marker first so Core can reliably
        // suppress runtime auto-delegation for that final-answer-only turn.
        let prompt = match (&self.goal, synthesis) {
            (_, true) => prompt,
            (Some(g), false) => format!("[Ongoing goal: {g}]\n\n{prompt}"),
            (None, false) => prompt,
        };
        self.stream_start_token = self.stream_start_token.wrapping_add(1);
        let stream_start_token = self.stream_start_token;
        // (A `/ctx <n>` staged transcript window is attached upstream, only to a
        // genuine typed user message — see on_submit — never to a `/loop`,
        // asset review, `?`, or synthesis continuation.)
        // ultracode no longer rewrites the user turn. Whether a turn plans and
        // fans out is decided by the core's message-gated planning
        // (PlanningMode::Auto) plus the `parallel_task` tool description — not an
        // unconditional per-turn imperative, which made even "hi" trigger a plan
        // and workspace exploration.
        let commands = vec![
            cmd::cmd(move || async move {
                let rewind_checkpoint = match rewind_request {
                    Some((store, session_id, workspace, task, history_before)) => Some(
                        capture_rewind_checkpoint_seed(
                            store,
                            session_id,
                            workspace,
                            task,
                            history_before,
                        )
                        .await,
                    ),
                    None => None,
                };
                let res =
                    tokio::time::timeout(Duration::from_millis(STREAM_START_TIMEOUT_MS), async {
                        if attachments.is_empty() {
                            session.stream(prompt.as_str(), None).await
                        } else {
                            session
                                .stream_with_attachments(prompt.as_str(), &attachments, None)
                                .await
                        }
                    })
                    .await;
                match res {
                    Ok(Ok((rx, join))) => Msg::StreamStarted {
                        token: stream_start_token,
                        session: Arc::clone(&session),
                        rx: Arc::new(Mutex::new(rx)),
                        join,
                        submitted_images,
                        rewind_checkpoint,
                    },
                    Ok(Err(error)) => {
                        let retryable_admission = matches!(&error, CodeError::SessionBusy { .. });
                        Msg::StreamError {
                            token: stream_start_token,
                            error: error.to_string(),
                            retryable_admission,
                            submitted_images,
                        }
                    }
                    Err(_) => Msg::StreamError {
                        token: stream_start_token,
                        error: format!(
                            "model stream admission timed out after {STREAM_START_TIMEOUT_MS} ms"
                        ),
                        retryable_admission: true,
                        submitted_images,
                    },
                }
            }),
            spinner_tick(),
            stream_commit_tick(),
        ];
        Some(cmd::batch(commands))
    }

    /// Enqueue a turn together with the execution mode visible at submission.
    /// a3s-lane owns ordering; this sidecar owns immutable execution semantics.
    pub(super) fn enqueue_turn(
        &mut self,
        priority: a3s_lane::Priority,
        queued: Queued,
        mode: Mode,
    ) -> u64 {
        let sequence = self.queue.push(priority, queued);
        self.queued_turn_modes.insert(sequence, mode);
        sequence
    }

    pub(super) fn enqueue_plan_turn(
        &mut self,
        priority: a3s_lane::Priority,
        queued: Queued,
        request: PlanDraftRequest,
    ) -> u64 {
        let sequence = self.enqueue_turn(priority, queued, Mode::Plan);
        self.queued_plan_drafts.insert(sequence, request);
        sequence
    }

    /// Pop exactly one queued user turn.
    pub(super) fn drain_queue(&mut self) -> Option<Cmd<Msg>> {
        // A queued successor must not race a session rebuild or an interrupted
        // stream admission that still has a chance to acquire the core's
        // single-flight lease. The corresponding completion handler calls this
        // method again as soon as the barrier clears.
        if self.state != State::Idle
            || self.session_rebuild_pending.is_some()
            || self.interrupted_stream_start_token.is_some()
            || self.active_queued_turn.is_some()
            || self.pending_plan_review.is_some()
            || self.plan_review.is_some()
        {
            return None;
        }
        let next = panels::queue::take_next_priority_item(
            &mut self.queue,
            &mut self.send_now_queued_sequence,
        );
        if let Some(next) = next {
            let sequence = next.sequence();
            let execution_mode = self
                .queued_turn_modes
                .get(&sequence)
                .copied()
                .unwrap_or(self.mode);
            let plan_draft = self.queued_plan_drafts.get(&sequence).cloned();
            let queued = next.value().clone();
            if let Some((query, evidence_scope)) = queued.deep_research {
                let command = self.start_deep_research_workflow(
                    query,
                    evidence_scope,
                    queued.runtime_expectation,
                );
                if command.is_none() {
                    self.queue.restore(next);
                    self.relayout();
                } else {
                    if self.send_now_queued_sequence == Some(sequence) {
                        self.send_now_queued_sequence = None;
                    }
                    self.queued_turn_modes.remove(&sequence);
                    self.queued_plan_drafts.remove(&sequence);
                }
                return command;
            }
            let command = self.start_stream_inner_with_runtime_and_images(
                queued.text,
                queued.display,
                true,
                false,
                queued.runtime_expectation,
                queued.images,
                execution_mode,
                plan_draft,
                true,
            );
            if command.is_some() {
                self.active_queued_turn_token = Some(self.stream_start_token);
                self.active_queued_turn = Some(next);
            } else {
                self.queue.restore(next);
                self.relayout();
            }
            return command;
        }

        None
    }

    /// Commit the queue claim once Core has admitted its stream. Before this
    /// point the exact a3s-lane entry remains restorable on `SessionBusy` or an
    /// admission timeout.
    pub(super) fn commit_active_queued_turn(&mut self, token: u64) -> bool {
        let Some(item) = take_matching_queue_claim(
            &mut self.active_queued_turn,
            &mut self.active_queued_turn_token,
            token,
        ) else {
            return false;
        };
        self.queued_turn_modes.remove(&item.sequence());
        self.queued_plan_drafts.remove(&item.sequence());
        if self.send_now_queued_sequence == Some(item.sequence()) {
            self.send_now_queued_sequence = None;
        }
        self.queue_retry_generation = self.queue_retry_generation.wrapping_add(1);
        self.queue_retry_attempt = 0;
        true
    }

    /// Return a turn whose stream was never admitted to its original priority
    /// and FIFO position. Its images are Arc-backed, so dropping the attempted
    /// admission copy cannot invalidate the retained queue payload.
    pub(super) fn restore_active_queued_turn(&mut self, token: u64) -> bool {
        if let Some(item) = take_matching_queue_claim(
            &mut self.active_queued_turn,
            &mut self.active_queued_turn_token,
            token,
        ) {
            self.queue.restore(item);
            self.relayout();
            return true;
        }
        false
    }

    pub(super) fn discard_active_queued_turn(&mut self, token: u64) {
        if let Some(item) = take_matching_queue_claim(
            &mut self.active_queued_turn,
            &mut self.active_queued_turn_token,
            token,
        ) {
            self.queued_turn_modes.remove(&item.sequence());
            self.queued_plan_drafts.remove(&item.sequence());
            if self.send_now_queued_sequence == Some(item.sequence()) {
                self.send_now_queued_sequence = None;
            }
        }
    }

    pub(super) fn retry_queued_turn_after_admission_failure(&mut self) -> Cmd<Msg> {
        self.queue_retry_generation = self.queue_retry_generation.wrapping_add(1);
        self.queue_retry_attempt = self.queue_retry_attempt.saturating_add(1);
        let generation = self.queue_retry_generation;
        let shift = u32::from(self.queue_retry_attempt.saturating_sub(1).min(4));
        let delay_ms = QUEUE_ADMISSION_RETRY_BASE_MS
            .saturating_mul(1_u64 << shift)
            .min(QUEUE_ADMISSION_RETRY_MAX_MS);
        cmd::cmd(move || async move {
            tokio::time::sleep(Duration::from_millis(delay_ms)).await;
            Msg::QueueRetry { generation }
        })
    }

    pub(super) fn interrupted_continuation(
        goal_cancelled: bool,
        deep_research_interrupted: bool,
    ) -> InterruptedContinuation {
        if deep_research_interrupted {
            InterruptedContinuation::SettleDeepResearch
        } else if goal_cancelled {
            InterruptedContinuation::RestoreGoalMode
        } else {
            InterruptedContinuation::DrainQueue
        }
    }

    pub(super) fn defer_or_continue_after_interrupt(
        &mut self,
        continuation: InterruptedContinuation,
    ) -> Option<Cmd<Msg>> {
        if self.interrupted_stream_start_token.is_some() {
            self.pending_interrupted_continuation = Some(continuation);
            // Keep submissions queue-only until the stale admission has been
            // cancelled. This closes the race where its cleanup could cancel a
            // newly started queue head on the same session.
            self.state = State::Streaming;
            self.relayout();
            return None;
        }
        self.continue_after_interrupt(continuation)
    }

    pub(super) fn on_interrupted_stream_start_settled(&mut self, token: u64) -> Option<Cmd<Msg>> {
        if self.interrupted_stream_start_token != Some(token) {
            return None;
        }
        self.interrupted_stream_start_token = None;
        let Some(continuation) = self.pending_interrupted_continuation.take() else {
            // The cancellation command has not produced `Msg::Interrupted`
            // yet. That handler will observe the cleared barrier and continue.
            return None;
        };
        self.state = State::Idle;
        self.relayout();
        self.continue_after_interrupt(continuation)
    }

    fn continue_after_interrupt(
        &mut self,
        continuation: InterruptedContinuation,
    ) -> Option<Cmd<Msg>> {
        match continuation {
            InterruptedContinuation::SettleDeepResearch => {
                self.settle_or_finalize_deep_research(DeepResearchSettlementExit::Interrupted)
            }
            InterruptedContinuation::RestoreGoalMode => {
                self.restore_autonomy();
                self.restore_goal_planning_mode()
            }
            InterruptedContinuation::DrainQueue => {
                self.restore_autonomy();
                self.drain_queue()
            }
        }
    }

    pub(super) fn has_queued_turn(&self) -> bool {
        !self.queue.is_empty()
    }

    pub(super) fn wait_for_completed_stream_join(
        &mut self,
        stream_join: StreamJoin,
        synthesis: Option<(String, String)>,
    ) -> Cmd<Msg> {
        self.stream_settle_abort = Some(stream_join.abort_handle());
        self.stream_join_settling = true;
        self.state = State::Streaming;
        self.relayout();
        wait_for_stream_join(
            Arc::clone(&self.session),
            stream_join,
            self.stream_start_token,
            synthesis,
        )
    }

    /// Shared turn-completion: count the turn, wait for the stream lifecycle to
    /// settle, consume queued user input, then continue autonomous work.
    /// Called from BOTH the normal `AgentEvent::End` arm (the happy path, which
    /// returns without re-pumping so `StreamEnded` never fires) and the
    /// `StreamEnded` channel-closed arm — previously this lived only in
    /// `StreamEnded`, so on success the queue never drained and `/loop` ran once.
    pub(super) fn complete_turn(&mut self) -> Option<Cmd<Msg>> {
        if self.deep_research_loop.is_some() {
            self.deep_research_stream_timeout_token =
                self.deep_research_stream_timeout_token.wrapping_add(1);
        }
        if self.deep_research_loop.as_ref().is_some_and(|state| {
            state.started_at.elapsed() >= Duration::from_millis(DEEP_RESEARCH_RUN_HARD_TIMEOUT_MS)
        }) {
            self.loop_remaining = 0;
            self.pending_deep_research_report_resume = false;
        }
        let degraded_deep_research = self.deep_research_loop.is_some()
            && matches!(self.deep_research_outcome, DeepResearchRunOutcome::Degraded);
        if self.state == State::Streaming && !degraded_deep_research {
            self.completed += 1;
        }
        self.warn_missing_runtime_evidence();
        let plan_review = self.take_plan_review_candidate();
        // Goal iterations are evaluated by Core before End. A generic hidden
        // synthesis turn must not replace that event gate or consume the turn
        // as if it were complete.
        let synthesis = if plan_review.is_some()
            || self.deep_research_loop.is_some()
            || degraded_deep_research
            || self.goal_run.is_some()
        {
            None
        } else {
            self.prepare_ultracode_synthesis()
        };
        let completed_stream_join = self.stream_join.take();
        self.finish();
        self.pending_plan_review = plan_review;
        if let Some(completed_stream_join) = completed_stream_join {
            // Keep input queue-only until the worker has completed persistence,
            // cleanup, and release of core's single-flight admission lease.
            return Some(self.wait_for_completed_stream_join(completed_stream_join, synthesis));
        }
        self.discard_active_rewind_checkpoint();
        self.continue_after_stream_settled(synthesis)
    }

    pub(super) fn continue_after_stream_settled(
        &mut self,
        synthesis: Option<(String, String)>,
    ) -> Option<Cmd<Msg>> {
        if self.activate_pending_plan_review() {
            return None;
        }
        if self.goal_run.as_ref().is_some_and(|run| run.achieved) {
            self.pending_goal_failure = None;
            return self.finish_achieved_goal();
        }
        if self.goal_run.is_some() {
            // User messages queued while the iteration ran still execute under
            // the forced-planning goal session. Otherwise schedule the next
            // host-owned iteration without a fixed cap.
            if !self.queue.is_empty() {
                return self.drain_queue();
            }
            let failure = self.pending_goal_failure.take();
            return self.continue_goal_run(failure);
        }
        self.pending_goal_failure = None;
        // A DeepResearch run completes its closed report transaction before
        // queued turns can change session context. Esc remains the explicit
        // way to interrupt the active run.
        if self.deep_research_loop.is_none() && self.has_queued_turn() {
            return self.drain_queue();
        }
        if let Some((prompt, display_task)) = synthesis {
            return self.start_ultracode_synthesis(prompt, display_task);
        }
        self.continue_completed_turn()
    }

    /// Select the next turn only after the prior stream lifecycle has fully
    /// settled. Keeping this selection deferred is important: queued
    /// DeepResearch starts `tool_with_events` synchronously, while normal model
    /// streams start when their returned command is polled.
    pub(super) fn continue_completed_turn(&mut self) -> Option<Cmd<Msg>> {
        // Do not interleave queued turns between DeepResearch retrieval,
        // report generation, its one durable resume, and terminal journaling.
        if self.deep_research_loop.is_none() && self.has_queued_turn() {
            return self.drain_queue();
        }
        if self.loop_remaining > 0 && self.pending_deep_research_report_resume {
            self.pending_deep_research_report_resume = false;
            self.loop_remaining -= 1;
            let n = self.loop_remaining;
            self.push_line(&Style::new().fg(TN_GRAY).render(&format!(
                "  ↻ deep research report resume ({n} left · Esc to stop)"
            )));
            self.loop_continuation = true;
            let query = self
                .deep_research_loop
                .as_ref()
                .map(|state| state.query.clone())
                .unwrap_or_else(|| "report".to_string());
            return self.start_deep_research_report_generation(
                format!("✦\u{200A}resume report {query}"),
                DeepResearchReportGenerationPhase::Resume,
            );
        }
        // Required runtime evidence is a deliverable, not just a warning. In
        // autonomous runs, spend the next loop turn on a targeted correction
        // before falling back to the generic "Continue" prompt.
        if self.loop_remaining > 0 {
            if let Some(prompt) = self
                .runtime_expectation
                .as_ref()
                .and_then(RuntimeExpectation::corrective_prompt)
            {
                self.loop_remaining -= 1;
                let n = self.loop_remaining;
                self.push_line(&Style::new().fg(TN_GRAY).render(&format!(
                    "  ↻ runtime evidence retry ({n} left · Esc to stop)"
                )));
                self.loop_continuation = true;
                return Some(cmd::msg(Msg::Submit(prompt)));
            }
        }
        // /loop: auto-continue until the agent says DONE, the cap is hit, or Esc.
        // DeepResearch may reach this point only for its explicit one-shot
        // report resume above; it never receives a hidden continuation.
        if self.loop_remaining > 0 {
            if self.deep_research_loop.is_some() {
                self.loop_remaining = 0;
            } else {
                self.loop_remaining -= 1;
                let n = self.loop_remaining;
                let prompt =
                    "Continue. If the task is fully complete, reply DONE and stop.".to_string();
                self.push_line(
                    &Style::new()
                        .fg(TN_GRAY)
                        .render(&format!("  ↻ loop ({n} left · Esc to stop)")),
                );
                // Mark the continuation as machine-driven so on_submit doesn't
                // attach a staged `/ctx` window to it.
                self.loop_continuation = true;
                return Some(cmd::msg(Msg::Submit(prompt)));
            }
        }
        // The loop is drained (or was never armed): an autonomous run that
        // auto-switched to auto mode is over — restore the user's mode.
        if self.loop_remaining == 0 {
            if self.deep_research_loop.is_some() {
                self.invalidate_subagent_snapshots();
                return self
                    .settle_or_finalize_deep_research(DeepResearchSettlementExit::ReportReady);
            }
            self.open_pending_deep_research_report_view();
            self.restore_autonomy();
        }
        // Run the next queued message (submitted while busy), if any.
        self.drain_queue()
    }

    pub(super) fn settle_or_finalize_deep_research(
        &mut self,
        exit: DeepResearchSettlementExit,
    ) -> Option<Cmd<Msg>> {
        match self.begin_deep_research_subagent_settlement(exit) {
            Some(settlement) => Some(settlement),
            None => self.finalize_deep_research_settlement(exit),
        }
    }

    pub(super) fn begin_deep_research_subagent_settlement(
        &mut self,
        exit: DeepResearchSettlementExit,
    ) -> Option<Cmd<Msg>> {
        if self.deep_research_loop.is_none() || self.deep_research_subagent_settlement_inflight {
            return None;
        }
        let mut task_ids = self.runtime.subagent_ids();
        if task_ids.is_empty() {
            return None;
        }
        task_ids.sort();
        self.deep_research_subagent_settlement_inflight = true;
        self.state = State::Streaming;
        self.spinner.start();
        self.relayout();
        Some(settle_deep_research_subagents(
            Arc::clone(&self.session),
            self.session_id.clone(),
            self.session_rebuild_seq,
            task_ids,
            exit,
        ))
    }

    pub(super) fn finalize_deep_research_settlement(
        &mut self,
        exit: DeepResearchSettlementExit,
    ) -> Option<Cmd<Msg>> {
        if self.deep_research_journal_finalization_inflight {
            return None;
        }
        let run_id = self
            .deep_research_workflow
            .args
            .as_ref()
            .and_then(|args| args.get("run_id"))
            .and_then(serde_json::Value::as_str)
            .map(str::to_string);
        let host_managed_inquiry = self
            .deep_research_workflow
            .args
            .as_ref()
            .is_some_and(deep_research_host_managed_inquiry);
        let outcome = match self.deep_research_outcome {
            DeepResearchRunOutcome::Active if exit == DeepResearchSettlementExit::Interrupted => {
                Some(ResearchOutcome::Failed)
            }
            DeepResearchRunOutcome::Active => None,
            DeepResearchRunOutcome::Completed => Some(ResearchOutcome::Completed),
            DeepResearchRunOutcome::Qualified => Some(ResearchOutcome::Qualified),
            DeepResearchRunOutcome::Degraded => Some(ResearchOutcome::Degraded),
        };
        if let (Some(run_id), Some(outcome)) = (run_id, outcome) {
            self.deep_research_journal_finalization_inflight = true;
            let workspace = PathBuf::from(&self.cwd);
            let artifacts = self.deep_research_terminal_artifacts.clone();
            let workflow_output = self
                .deep_research_workflow
                .output
                .clone()
                .unwrap_or_default();
            let workflow_metadata = self.deep_research_workflow.metadata.clone();
            return Some(cmd::cmd(move || async move {
                let successful_report = matches!(
                    outcome,
                    ResearchOutcome::Completed | ResearchOutcome::Qualified
                );
                let publication_result = if successful_report {
                    match deep_research_inquiry_publication_outcome(
                        &workflow_output,
                        workflow_metadata.as_ref(),
                    ) {
                        Ok(Some(DeepResearchRunOutcome::Completed))
                            if outcome == ResearchOutcome::Completed =>
                        {
                            Ok(())
                        }
                        Ok(Some(DeepResearchRunOutcome::Qualified))
                            if outcome == ResearchOutcome::Qualified =>
                        {
                            Ok(())
                        }
                        Ok(Some(actual)) => Err(format!(
                            "DeepResearch report outcome disagrees with terminal Inquiry: {actual:?}"
                        )),
                        Ok(None) if host_managed_inquiry => Err(
                            "host-managed DeepResearch cannot journal success without an Inquiry projection"
                                .to_string(),
                        ),
                        Ok(None) => Ok(()),
                        Err(error) => Err(error),
                    }
                } else {
                    Ok(())
                };
                let inquiry_result = match publication_result {
                    Ok(()) => match inquiry_projection_from_workflow(
                        &workflow_output,
                        workflow_metadata.as_ref(),
                    ) {
                        Ok(Some((events, state))) => {
                            record_deep_research_inquiry_state(
                                &workspace,
                                &run_id,
                                &events,
                                &state,
                            )
                            .await
                            .map_err(|error| error.to_string())
                        }
                        Ok(None) if host_managed_inquiry && successful_report => Err(
                            "host-managed DeepResearch terminal journal omitted its Inquiry projection"
                                .to_string(),
                        ),
                        Ok(None) => Ok(()),
                        Err(error) if successful_report => Err(error),
                        Err(_) => Ok(()),
                    },
                    Err(error) => Err(error),
                };
                let result = match inquiry_result {
                    Ok(()) => record_deep_research_run_terminal(
                        &workspace,
                        &run_id,
                        outcome,
                        artifacts.as_ref(),
                    )
                    .await
                    .map_err(|error| error.to_string()),
                    Err(error) => Err(error),
                };
                Msg::DeepResearchJournalFinalized {
                    run_id,
                    exit,
                    result,
                }
            }));
        }
        self.complete_deep_research_settlement(exit)
    }

    pub(super) fn complete_deep_research_settlement(
        &mut self,
        exit: DeepResearchSettlementExit,
    ) -> Option<Cmd<Msg>> {
        if exit.opens_report() && self.deep_research_outcome.report_ready() {
            self.open_pending_deep_research_report_view();
        } else {
            self.pending_deep_research_report_view = None;
        }
        self.restore_autonomy();
        self.drain_queue()
    }

    pub(super) fn resume_after_pending_confirmation(&self) -> Cmd<Msg> {
        resume_after_pending_confirmation_cmd(self.rx.clone())
    }

    /// An autonomous directive run is starting (/sleep, asset reviews,
    /// asset run/deploy, /flow drafts, /loop): switch to non-interactive Auto
    /// so tool prompts cannot stall it, and arm the loop budget that re-prompts until
    /// the deliverable lands. The prior mode is restored when the run ends
    /// (loop drained, interrupt, error, or /clear). A user already in auto
    /// mode keeps it — nothing is remembered or restored.
    pub(super) fn engage_autonomy(&mut self, budget: usize) {
        self.loop_remaining = self.loop_remaining.max(budget);
        if self.mode != Mode::Auto {
            self.autonomy_restore = Some(self.mode);
            self.set_composer_mode(Mode::Auto);
            self.push_line(&Style::new().fg(TN_GRAY).render(
                "  ⏵⏵ auto mode engaged for this task — restores when it completes (Esc stops)",
            ));
        }
    }

    pub(super) fn engage_single_turn_autonomy(&mut self) {
        if self.mode != Mode::Auto {
            self.autonomy_restore = Some(self.mode);
            self.set_composer_mode(Mode::Auto);
            self.push_line(&Style::new().fg(TN_GRAY).render(
                "  ⏵⏵ auto mode engaged for this task — restores when it completes (Esc stops)",
            ));
        }
    }

    pub(super) fn record_runtime_tool_evidence(&mut self, name: &str) {
        if let Some(expectation) = &mut self.runtime_expectation {
            expectation.record_tool(name);
        }
    }

    pub(super) fn record_runtime_parallel_evidence(&mut self) {
        if let Some(expectation) = &mut self.runtime_expectation {
            expectation.record_parallel_work();
        }
    }

    pub(super) fn record_runtime_view_evidence(&mut self) {
        if let Some(expectation) = &mut self.runtime_expectation {
            expectation.record_remote_view();
        }
    }

    pub(super) fn warn_missing_runtime_evidence(&mut self) {
        let warning = self
            .runtime_expectation
            .as_mut()
            .and_then(RuntimeExpectation::missing_warning);
        if let Some(warning) = warning {
            self.push_line(&Style::new().fg(TN_YELLOW).render(&warning));
        }
    }

    /// Restore the pre-autonomy mode (no-op when nothing was auto-switched).
    pub(super) fn restore_autonomy(&mut self) {
        self.runtime_expectation = None;
        self.deep_research_loop = None;
        if let Some((goal, goal_since)) = self.deep_research_goal_restore.take() {
            self.goal = goal;
            self.goal_since = goal_since;
        }
        self.deep_research_report_resume_used = false;
        self.deep_research_workflow.clear();
        self.deep_research_outcome = DeepResearchRunOutcome::Active;
        self.deep_research_journal_finalization_inflight = false;
        self.deep_research_terminal_artifacts = None;
        self.deep_research_agent_event_sequence = 0;
        self.deep_research_projection = None;
        self.pending_deep_research_report_resume = false;
        self.pending_deep_research_report_view = None;
        self.deep_research_report_tool_gate.reset();
        self.deep_research_subagent_settlement_inflight = false;
        if let Some(prev) = self.autonomy_restore.take() {
            self.set_composer_mode(prev);
            self.push_line(
                &Style::new()
                    .fg(TN_GRAY)
                    .render("  ⏵ autonomous task ended — auto mode restored to your previous mode"),
            );
        }
    }

    pub(super) fn record_deep_research_child_event_cmd(
        &mut self,
        task_id: String,
        started: bool,
        payload: serde_json::Value,
    ) -> Option<Cmd<Msg>> {
        let run_id = self
            .deep_research_workflow
            .args
            .as_ref()?
            .get("run_id")?
            .as_str()?
            .to_string();
        self.deep_research_agent_event_sequence =
            self.deep_research_agent_event_sequence.saturating_add(1);
        let sequence = self.deep_research_agent_event_sequence;
        let workspace = PathBuf::from(&self.cwd);
        Some(cmd::cmd(move || async move {
            let result = record_deep_research_child_event(
                &workspace, &run_id, sequence, &task_id, started, payload,
            )
            .await
            .map_err(|error| error.to_string());
            Msg::DeepResearchJournalEventRecorded { run_id, result }
        }))
    }
}

#[cfg(test)]
mod queue_claim_tests {
    use super::*;

    #[test]
    fn admission_failure_restores_the_original_lane_position() {
        let mut queue = PriorityQueue::new();
        queue.push(USER_TURN_PRIORITY, "first");
        queue.push(USER_TURN_PRIORITY, "second");
        queue.push(SYNTHETIC_TURN_PRIORITY, "continuation");

        let mut active = queue.pop();
        let mut active_token = Some(9);
        let claimed = take_matching_queue_claim(&mut active, &mut active_token, 9)
            .expect("matching admission claim");
        queue.restore(claimed);

        let ordered = queue
            .ordered()
            .into_iter()
            .map(|item| *item.value())
            .collect::<Vec<_>>();
        assert_eq!(ordered, ["first", "second", "continuation"]);
    }

    #[test]
    fn stale_stream_token_cannot_consume_the_current_lane_claim() {
        let mut queue = PriorityQueue::new();
        queue.push(USER_TURN_PRIORITY, "current");
        let mut active = queue.pop();
        let original_sequence = active.as_ref().expect("active claim").sequence();
        let mut active_token = Some(12);

        assert!(take_matching_queue_claim(&mut active, &mut active_token, 11).is_none());
        assert_eq!(active_token, Some(12));
        assert_eq!(
            active.as_ref().expect("claim must remain").sequence(),
            original_sequence
        );
    }

    #[test]
    fn matching_lane_claim_is_consumed_at_most_once() {
        let mut queue = PriorityQueue::new();
        queue.push(USER_TURN_PRIORITY, "once");
        let mut active = queue.pop();
        let mut active_token = Some(21);

        assert!(take_matching_queue_claim(&mut active, &mut active_token, 21).is_some());
        assert!(take_matching_queue_claim(&mut active, &mut active_token, 21).is_none());
        assert!(active.is_none());
        assert_eq!(active_token, None);
    }
}