agentty 0.13.0

//! Session lifecycle workflows and direct user actions.

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

use ag_agent as agent;
use ag_agent::AgentRequestKind;
use ag_forge as forge;
use ag_git as git;
use ag_protocol::{AgentResponse, parse_agent_response_strict};
use askama::Template;
use tokio::sync::mpsc;
use tracing::warn;
use uuid::Uuid;

use super::task::SessionTranscriptMessageAppend;
use super::worker::{SessionCommand, TurnMetadata};
use super::{
    SessionTaskService, StatusTransition, draft, isolation, session_branch, session_folder,
    unix_timestamp_from_system_time,
};
use crate::app::session::SessionError;
use crate::app::{
    AppEvent, AppServices, ProjectManager, SessionManager, agentty_home, review_request, setting,
};
use crate::domain::agent::{AgentKind, AgentSelection, AgentSelectionMetadata, ReasoningLevel};
use crate::domain::session::{
    ReviewRequest, SESSION_DATA_DIR, Session, SessionHandles, SessionId, Status,
    can_merge_session_branch_in_stack as stack_can_merge_session_branch,
    can_mutate_session_branch_in_stack as stack_can_mutate_session_branch,
    can_rebase_session_branch_in_stack as stack_can_rebase_session_branch,
    can_reply_to_session_in_stack as stack_can_reply_to_session,
    can_start_staged_session_in_stack as stack_can_start_staged_session,
};
use crate::domain::session_message::{SessionMessageKind, SessionTranscript};
use crate::domain::session_order;
use crate::domain::setting::SettingName;
use crate::domain::transcript_notice::TranscriptNotice;
use crate::domain::turn_prompt::{TurnPrompt, TurnPromptAttachment, TurnPromptTextSource};
use crate::infra::db;
use crate::infra::fs::{FsClient, FsError};

/// Maximum accepted length for generated session titles.
///
/// Longer candidates are treated as likely non-title prose instead of being
/// truncated into misleading session labels.
const GENERATED_SESSION_TITLE_MAX_CHARACTERS: usize = 72;
/// Progress/status prefixes that indicate the model returned process prose
/// instead of a requested-work title.
const GENERATED_SESSION_TITLE_PROGRESS_PREFIXES: &[&str] = &[
    "checking ",
    "confirming ",
    "gathering ",
    "inspecting ",
    "investigating ",
    "reviewing ",
    "validating ",
    "working ",
];
const USER_PROMPT_PREFIX: &str = " › ";
const USER_PROMPT_CONTINUATION_PREFIX: &str = "   ";

/// Input bag for constructing a queued session command.
struct BuildSessionCommandInput {
    is_first_message: bool,
    published_upstream_ref: Option<String>,
    prompt: TurnPrompt,
    replay_transcript: Option<String>,
    session_agent: AgentSelection,
}

/// Intermediate values captured while preparing a session reply.
type ReplyContext = (Option<String>, bool, SessionId, Option<String>);

/// Cleanup payload for a deleted session's git and filesystem resources.
struct DeletedSessionCleanup {
    branch_name: String,
    folder: PathBuf,
    has_git_branch: bool,
    session_id: SessionId,
    staged_draft_root: PathBuf,
    working_dir: PathBuf,
}

/// Askama view model for rendering one-shot title-generation prompts.
#[derive(Template)]
#[template(path = "session_title_generation_prompt.md", escape = "none")]
struct SessionTitleGenerationPromptTemplate<'a> {
    prompt: &'a str,
}

/// Identifies one tracked draft-title generation task completion event.
struct TitleGenerationTaskCompletion {
    generation: u64,
    session_id: SessionId,
}

impl SessionManager {
    /// Moves selection to the next selectable session in grouped list order.
    ///
    /// Group header rows are non-selectable and are skipped by design.
    pub fn next(&mut self) {
        if let Some(index) = session_order::next_selectable_session_index(
            &self.state.sessions,
            self.state.table_state.selected(),
        ) {
            self.state.table_state.select(Some(index));
        }
    }

    /// Moves selection to the previous selectable session in grouped list
    /// order.
    ///
    /// Group header rows are non-selectable and are skipped by design.
    pub fn previous(&mut self) {
        if let Some(index) = session_order::previous_selectable_session_index(
            &self.state.sessions,
            self.state.table_state.selected(),
        ) {
            self.state.table_state.select(Some(index));
        }
    }

    /// Creates a blank session with an empty prompt and output.
    ///
    /// Returns the identifier of the newly created session.
    /// The session is created with `Draft` status and no agent is started —
    /// call [`SessionManager::start_session`] to submit a prompt and launch
    /// the agent.
    ///
    /// # Errors
    /// Returns an error if the worktree, session files, database record, or
    /// backend setup cannot be created.
    pub async fn create_session(
        &mut self,
        projects: &ProjectManager,
        services: &AppServices,
    ) -> Result<String, SessionError> {
        self.create_live_session(projects, services).await
    }

    /// Creates a blank draft session that stages prompts until explicitly
    /// started.
    ///
    /// Draft sessions defer worktree creation until the staged bundle starts
    /// so the session branch can be based on the latest local base-branch
    /// state.
    ///
    /// Returns the identifier of the newly created session.
    ///
    /// # Errors
    /// Returns an error if the session files or database record cannot be
    /// created, or if regular-session worktree/backend setup fails.
    pub async fn create_draft_session(
        &mut self,
        projects: &ProjectManager,
        services: &AppServices,
    ) -> Result<String, SessionError> {
        let base_branch = projects.git_branch().ok_or_else(|| {
            SessionError::Workflow("Git branch is required to create a session".to_string())
        })?;

        self.create_draft_session_for_project(services, projects.active_project_id(), base_branch)
            .await
    }

    /// Creates a blank draft session stacked on top of a selected parent
    /// session branch.
    ///
    /// The child remains an explicit draft while prompts are staged. It can
    /// start once the parent is review-ready and no other stack member is
    /// doing branch work. Its lazy worktree is based on the stored parent
    /// branch, and the parent link is kept so review publishing can target the
    /// parent branch while the stack is active.
    ///
    /// # Errors
    /// Returns an error when the parent is missing, already stacked, terminal,
    /// an unmaterialized draft, missing project metadata, or when draft
    /// persistence fails.
    pub async fn create_stacked_draft_session(
        &mut self,
        services: &AppServices,
        parent_session_id: &str,
    ) -> Result<String, SessionError> {
        let (base_branch, parent_id) = {
            let parent_session = self.session_or_err(parent_session_id)?;
            if !parent_session.allows_stacked_child_creation() {
                return Err(SessionError::Workflow(
                    "Stacked sessions can only be created from root sessions with active branches"
                        .to_string(),
                ));
            }

            let parent_branch = self
                .session_branch_name(&parent_session.id)
                .map_or_else(|| session_branch(&parent_session.id), str::to_string);

            (parent_branch, parent_session.id.clone())
        };
        let project_id = services
            .db()
            .sessions()
            .load_session_project_id(parent_id.as_str())
            .await?
            .ok_or_else(|| {
                SessionError::Workflow(
                    "Parent session has no project association for stacked draft creation"
                        .to_string(),
                )
            })?;

        self.create_draft_session_for_project_with_parent(
            services,
            project_id,
            &base_branch,
            Some(parent_id.as_str()),
        )
        .await
    }

    /// Creates one blank draft session for an explicit persisted project.
    ///
    /// Continuation flows use the source session project instead of the
    /// currently active project so the later lazy worktree is materialized from
    /// the same repository and base branch as the terminal source session.
    ///
    /// Returns the identifier of the newly created session.
    ///
    /// # Errors
    /// Returns an error if the session files or database record cannot be
    /// created.
    pub async fn create_draft_session_for_project(
        &mut self,
        services: &AppServices,
        project_id: i64,
        base_branch: &str,
    ) -> Result<String, SessionError> {
        self.create_draft_session_for_project_with_parent(services, project_id, base_branch, None)
            .await
    }

    /// Creates one blank draft session with an optional persisted parent
    /// session id.
    async fn create_draft_session_for_project_with_parent(
        &mut self,
        services: &AppServices,
        project_id: i64,
        base_branch: &str,
        parent_session_id: Option<&str>,
    ) -> Result<String, SessionError> {
        let session_agent = self
            .resolve_default_session_agent(services, project_id)
            .await;
        let session_model = session_agent.model();
        let reasoning_level = services
            .db()
            .settings()
            .load_project_reasoning_level(project_id)
            .await?;
        self.default_session_model = session_model;

        let session_id = Uuid::new_v4().to_string();
        let folder = session_folder(services.base_path(), &session_id);
        if services.fs_client().exists(folder.clone()) {
            return Err(SessionError::Workflow(format!(
                "Session folder {session_id} already exists"
            )));
        }

        let session_agent_kind = session_agent.kind().to_string();
        let status = Status::Draft.to_string();
        let insert_result = services
            .db()
            .sessions()
            .insert_session_with_agent(db::PersistedSessionCreation {
                agent: &session_agent_kind,
                base_branch,
                id: &session_id,
                is_draft: true,
                model: session_model.as_str(),
                parent_session_id,
                project_id,
                reasoning_level,
                status: &status,
            })
            .await;

        insert_result.map_err(|error| {
            SessionError::Workflow(format!("Failed to save session metadata: {error}"))
        })?;

        Self::record_session_creation_activity(services, &session_id).await;
        services.emit_session_and_project_refresh_events();

        Ok(session_id)
    }

    /// Forks a root review-ready session into a new independent review
    /// session.
    ///
    /// The fork creates a new worktree branch from the source session branch,
    /// snapshots persisted transcript messages, clears provider-native
    /// conversation and publish/review-request linkage, and marks the new
    /// session for one-time history replay on its first reply.
    ///
    /// # Errors
    /// Returns an error if the source session is missing, not root
    /// review-ready, repository metadata cannot be resolved, the worktree
    /// cannot be created, or the metadata snapshot cannot be persisted.
    pub async fn fork_session(
        &mut self,
        services: &AppServices,
        source_session_id: &str,
    ) -> Result<String, SessionError> {
        let (source_branch, source_agent) = {
            let source_session = self.session_or_err(source_session_id)?;
            if !source_session.allows_fork_action() {
                return Err(SessionError::Workflow(
                    "Only root review-ready sessions can be forked".to_string(),
                ));
            }

            let source_branch = self
                .session_branch_name(&source_session.id)
                .map_or_else(|| session_branch(&source_session.id), str::to_string);

            (source_branch, source_session.agent)
        };
        services
            .db()
            .sessions()
            .load_session_project_id(source_session_id)
            .await?
            .ok_or_else(|| {
                SessionError::Workflow(
                    "Source session has no project association for session forking".to_string(),
                )
            })?;

        let repo_root = self
            .load_session_repo_root(services, source_session_id)
            .await?;
        let session_id = Uuid::new_v4().to_string();
        let folder = session_folder(services.base_path(), &session_id);
        if services.fs_client().exists(folder.clone()) {
            return Err(SessionError::Workflow(format!(
                "Session folder {session_id} already exists"
            )));
        }

        let worktree_branch = session_branch(&session_id);
        self.create_session_worktree(
            services,
            &session_id,
            &folder,
            &repo_root,
            &worktree_branch,
            &source_branch,
        )
        .await?;

        let fork_status = Status::Review.to_string();
        let snapshot = db::ForkSessionSnapshot {
            new_session_id: &session_id,
            source_session_id,
            status: &fork_status,
        };
        if let Err(error) = services
            .db()
            .sessions()
            .fork_session_snapshot(snapshot)
            .await
        {
            self.rollback_failed_session_creation(
                services,
                &folder,
                &repo_root,
                &session_id,
                &worktree_branch,
                false,
            )
            .await;

            return Err(SessionError::Workflow(format!(
                "Failed to save forked session metadata: {error}"
            )));
        }

        Self::record_session_creation_activity(services, &session_id).await;

        if let Err(error) = agent::create_backend(source_agent.kind()).setup(&folder) {
            self.rollback_failed_session_creation(
                services,
                &folder,
                &repo_root,
                &session_id,
                &worktree_branch,
                true,
            )
            .await;

            return Err(SessionError::Workflow(format!(
                "Failed to setup session backend: {error}"
            )));
        }

        self.mark_history_replay_pending(&session_id);
        services.emit_session_and_project_refresh_events();

        Ok(session_id)
    }

    /// Creates one regular session whose worktree is materialized before the
    /// first prompt is submitted.
    ///
    /// # Errors
    /// Returns an error if the worktree, session files, database record, or
    /// backend setup cannot be created.
    async fn create_live_session(
        &mut self,
        projects: &ProjectManager,
        services: &AppServices,
    ) -> Result<String, SessionError> {
        let base_branch = projects.git_branch().ok_or_else(|| {
            SessionError::Workflow("Git branch is required to create a session".to_string())
        })?;
        let session_agent = self
            .resolve_default_session_agent(services, projects.active_project_id())
            .await;
        let session_model = session_agent.model();
        let reasoning_level = services
            .db()
            .settings()
            .load_project_reasoning_level(projects.active_project_id())
            .await?;
        self.default_session_model = session_model;

        let session_id = Uuid::new_v4().to_string();
        let folder = session_folder(services.base_path(), &session_id);
        let fs_client = services.fs_client();
        if fs_client.exists(folder.clone()) {
            return Err(SessionError::Workflow(format!(
                "Session folder {session_id} already exists"
            )));
        }

        let worktree_branch = session_branch(&session_id);
        let working_dir = projects.working_dir().to_path_buf();
        let git_client = services.git_client();
        let repo_root = git_client
            .find_git_repo_root(working_dir)
            .await
            .ok_or_else(|| {
                SessionError::Workflow("Failed to find git repository root".to_string())
            })?;
        self.create_session_worktree(
            services,
            &session_id,
            &folder,
            &repo_root,
            &worktree_branch,
            base_branch,
        )
        .await?;

        let session_agent_kind = session_agent.kind().to_string();
        let status = Status::Draft.to_string();
        if let Err(error) = services
            .db()
            .sessions()
            .insert_session_with_agent(db::PersistedSessionCreation {
                agent: &session_agent_kind,
                base_branch,
                id: &session_id,
                is_draft: false,
                model: session_model.as_str(),
                parent_session_id: None,
                project_id: projects.active_project_id(),
                reasoning_level,
                status: &status,
            })
            .await
        {
            self.rollback_failed_session_creation(
                services,
                &folder,
                &repo_root,
                &session_id,
                &worktree_branch,
                false,
            )
            .await;

            return Err(SessionError::Workflow(format!(
                "Failed to save session metadata: {error}"
            )));
        }

        Self::record_session_creation_activity(services, &session_id).await;

        if let Err(error) = agent::create_backend(session_agent.kind()).setup(&folder) {
            self.rollback_failed_session_creation(
                services,
                &folder,
                &repo_root,
                &session_id,
                &worktree_branch,
                true,
            )
            .await;

            return Err(SessionError::Workflow(format!(
                "Failed to setup session backend: {error}"
            )));
        }
        services.emit_session_and_project_refresh_events();

        Ok(session_id)
    }

    /// Creates the git worktree and session-local metadata directory for one
    /// session branch from an explicit start ref.
    ///
    /// Regular sessions pass the local base branch as the start ref, stacked
    /// drafts pass their parent branch, and forks pass the source session
    /// branch so the new worktree preserves the source branch state at fork
    /// time.
    ///
    /// # Errors
    /// Returns an error if git worktree creation fails or the `.agentty`
    /// metadata directory cannot be created inside the worktree.
    async fn create_session_worktree(
        &self,
        services: &AppServices,
        session_id: &str,
        folder: &Path,
        repo_root: &Path,
        worktree_branch: &str,
        start_ref: &str,
    ) -> Result<(), SessionError> {
        services
            .git_client()
            .create_worktree(
                repo_root.to_path_buf(),
                folder.to_path_buf(),
                worktree_branch.to_string(),
                start_ref.to_string(),
            )
            .await
            .map_err(|error| {
                SessionError::Workflow(format!("Failed to create git worktree: {error}"))
            })?;

        let data_dir = folder.join(SESSION_DATA_DIR);
        if let Err(error) = services.fs_client().create_dir_all(data_dir).await {
            self.rollback_failed_session_creation(
                services,
                folder,
                repo_root,
                session_id,
                worktree_branch,
                false,
            )
            .await;

            return Err(SessionError::Workflow(format!(
                "Failed to create session metadata directory: {error}"
            )));
        }

        Ok(())
    }

    /// Ensures a draft session has a usable worktree and backend setup before
    /// its first live turn starts.
    ///
    /// Non-draft sessions are created eagerly and therefore skip this path.
    /// Draft sessions create their worktree lazily here so staged prompts can
    /// remain detached from the base branch until the user starts the session.
    ///
    /// # Errors
    /// Returns an error if repository discovery, worktree creation, or
    /// backend setup fails.
    async fn ensure_session_worktree_ready(
        &mut self,
        services: &AppServices,
        session_id: &str,
    ) -> Result<(), SessionError> {
        let (base_branch, folder, parent_session_id, persisted_session_id, session_agent) = {
            let session = self.session_or_err(session_id)?;
            if !session.is_draft_session() {
                return Ok(());
            }

            (
                session.base_branch.clone(),
                session.folder.clone(),
                session.parent_session_id.clone(),
                session.id.clone(),
                session.agent,
            )
        };

        let worktree_branch = session_branch(&persisted_session_id);
        if services.fs_client().is_dir(folder.clone()) {
            isolation::validate_session_worktree(
                services.fs_client().as_ref(),
                services.git_client().as_ref(),
                &folder,
                &persisted_session_id,
            )
            .await?;
            agent::create_backend(session_agent.kind())
                .setup(&folder)
                .map_err(|error| {
                    SessionError::Workflow(format!("Failed to setup session backend: {error}"))
                })?;
            self.persist_stack_base_for_stacked_draft_worktree(
                services,
                &folder,
                parent_session_id.as_ref(),
                &persisted_session_id,
            )
            .await?;
            self.set_session_worktree_available(session_id, true);

            return Ok(());
        }

        let repo_root = self.load_session_repo_root(services, session_id).await?;

        self.create_session_worktree(
            services,
            &persisted_session_id,
            &folder,
            &repo_root,
            &worktree_branch,
            &base_branch,
        )
        .await?;

        if let Err(error) = agent::create_backend(session_agent.kind()).setup(&folder) {
            let cleanup_errors = Self::cleanup_session_worktree_resources(
                services.fs_client().clone(),
                services.git_client(),
                folder,
                worktree_branch,
                Some(repo_root),
                true,
            )
            .await;

            if !cleanup_errors.is_empty() {
                return Err(SessionError::Workflow(format!(
                    "Failed to setup session backend: {error}. Cleanup also failed: {}",
                    cleanup_errors.join("; ")
                )));
            }

            return Err(SessionError::Workflow(format!(
                "Failed to setup session backend: {error}"
            )));
        }
        self.persist_stack_base_for_stacked_draft_worktree(
            services,
            &folder,
            parent_session_id.as_ref(),
            &persisted_session_id,
        )
        .await?;
        self.set_session_worktree_available(session_id, true);

        Ok(())
    }

    /// Clears the persisted and in-memory draft flag once a draft session
    /// starts its first live turn.
    ///
    /// The flag only means "still staging draft prompts", so it must not
    /// outlive the session start; a sticky flag would keep draft-only
    /// restrictions, such as the fork gate, active for the session's whole
    /// life. Non-draft sessions skip the write.
    ///
    /// # Errors
    /// Returns an error if the session is missing or persistence fails.
    async fn clear_session_draft_flag(
        &mut self,
        services: &AppServices,
        session_id: &str,
    ) -> Result<(), SessionError> {
        if !self.session_or_err(session_id)?.is_draft_session() {
            return Ok(());
        }

        services
            .db()
            .sessions()
            .clear_session_draft_flag(session_id)
            .await?;

        let session_index = self.session_index_or_err(session_id)?;
        if let Some(session) = self.session_at_mut(session_index) {
            session.is_draft = false;
        }

        Ok(())
    }

    /// Persists the parent tip used by a stacked draft's newly materialized
    /// worktree.
    ///
    /// The stored hash lets later stacked-child rebases use
    /// `git rebase --onto` to replay only the child's commits when the parent
    /// branch moves or squash-merges.
    ///
    /// # Errors
    /// Returns an error when the worktree `HEAD` cannot be resolved or stack
    /// metadata cannot be persisted.
    async fn persist_stack_base_for_stacked_draft_worktree(
        &self,
        services: &AppServices,
        folder: &Path,
        parent_session_id: Option<&SessionId>,
        session_id: &str,
    ) -> Result<(), SessionError> {
        if parent_session_id.is_none() {
            return Ok(());
        }

        let stack_base_commit_hash = services
            .git_client()
            .head_hash(folder.to_path_buf())
            .await
            .map_err(SessionError::Git)?;
        services
            .db()
            .sessions()
            .update_session_stack_base_commit_hash(session_id, Some(stack_base_commit_hash))
            .await
            .map_err(SessionError::Db)?;

        Ok(())
    }

    /// Resolves the repository root for one persisted session.
    ///
    /// # Errors
    /// Returns an error if the session project cannot be resolved or no git
    /// repository root can be found for the project path.
    async fn load_session_repo_root(
        &self,
        services: &AppServices,
        session_id: &str,
    ) -> Result<PathBuf, SessionError> {
        let project_id = services
            .db()
            .sessions()
            .load_session_project_id(session_id)
            .await?
            .ok_or_else(|| {
                SessionError::Workflow(
                    "Session project is required to create a worktree".to_string(),
                )
            })?;
        let project_path = self.load_project_path(services, project_id).await?;

        services
            .git_client()
            .find_git_repo_root(project_path)
            .await
            .ok_or_else(|| SessionError::Workflow("Failed to find git repository root".to_string()))
    }

    /// Loads the persisted project path for one project identifier.
    ///
    /// # Errors
    /// Returns an error if the project row does not exist or cannot be loaded.
    async fn load_project_path(
        &self,
        services: &AppServices,
        project_id: i64,
    ) -> Result<PathBuf, SessionError> {
        let project_row = services
            .db()
            .projects()
            .get_project(project_id)
            .await?
            .ok_or_else(|| {
                SessionError::Workflow(format!("Project with id `{project_id}` was not found"))
            })?;

        Ok(PathBuf::from(project_row.path))
    }

    async fn persist_staged_draft(
        services: &AppServices,
        session_id: &str,
        staged_attachments: &[TurnPromptAttachment],
        staged_prompt: &str,
        title_to_save: Option<&str>,
    ) -> Result<(), SessionError> {
        draft::store_staged_draft_attachments(
            services.fs_client().as_ref(),
            services.base_path(),
            session_id,
            staged_attachments,
        )
        .await?;
        services
            .db()
            .sessions()
            .update_session_prompt(session_id, staged_prompt)
            .await?;
        if let Some(title) = title_to_save {
            services
                .db()
                .sessions()
                .update_session_title(session_id, title)
                .await?;
        }
        Ok(())
    }

    /// Appends one staged draft message to a `Draft` session without launching
    /// the agent yet.
    ///
    /// This emits a [`AppEvent::SessionUpdated`] signal so memoized session
    /// views refresh immediately after local draft updates. The signal is
    /// best-effort after staged state has already been persisted; if the
    /// foreground event channel is closed, staging still succeeds and the next
    /// session refresh observes the committed prompt.
    ///
    /// The first staged prompt seeds a fallback title, while later staged
    /// prompts keep the current visible title in place until the refreshed
    /// generated title arrives.
    ///
    /// # Errors
    /// Returns an error if the session is missing, was not created as a draft
    /// session, is no longer `Draft`, or the staged bundle cannot be persisted.
    pub async fn stage_draft_message(
        &mut self,
        services: &AppServices,
        session_id: &str,
        prompt: impl Into<TurnPrompt>,
    ) -> Result<(), SessionError> {
        let prompt = prompt.into();
        let session_index = self.session_index_or_err(session_id)?;
        let (
            folder,
            persisted_session_id,
            session_agent,
            staged_attachments,
            staged_prompt,
            title_to_save,
        ) = {
            let session = self
                .session_at(session_index)
                .ok_or(SessionError::NotFound)?;
            if !session.is_draft_session() {
                return Err(SessionError::Workflow(
                    "Only draft sessions can stage drafts".to_string(),
                ));
            }
            if session.status != Status::Draft {
                return Err(SessionError::Workflow(
                    "Only `Draft` sessions can stage drafts".to_string(),
                ));
            }

            let next_attachment_number = session.draft_attachments.len().saturating_add(1);
            let staged_prompt =
                Self::append_staged_prompt(&session.prompt, &prompt, next_attachment_number);
            let mut staged_attachments = session.draft_attachments.clone();
            staged_attachments.extend(Self::renumbered_attachments(
                &prompt,
                next_attachment_number,
            ));
            let title_to_save = session.title.is_none().then(|| prompt.transcript_text());

            (
                session.folder.clone(),
                session.id.clone(),
                session.agent,
                staged_attachments,
                staged_prompt,
                title_to_save,
            )
        };
        let project_id = services
            .db()
            .sessions()
            .load_session_project_id(&persisted_session_id)
            .await?
            .ok_or_else(|| {
                SessionError::Workflow(
                    "Session project is required to stage draft prompts".to_string(),
                )
            })?;
        let (title_generation_folder, title_generation_agent) = self
            .draft_title_generation_context(services, project_id, session_agent, folder)
            .await?;

        Self::persist_staged_draft(
            services,
            &persisted_session_id,
            &staged_attachments,
            &staged_prompt,
            title_to_save.as_deref(),
        )
        .await?;

        let title_generation_prompt = staged_prompt.clone();

        if let Some(session) = self.session_at_mut(session_index) {
            session.prompt = staged_prompt;
            session.draft_attachments = staged_attachments;
            if let Some(title_to_save) = title_to_save {
                session.title = Some(title_to_save);
            }
        }

        let title_generation_task_generation =
            self.next_title_generation_task_generation(&persisted_session_id);
        let title_generation_task = Self::spawn_session_title_generation_task(
            services.event_sender(),
            services.db().clone(),
            &persisted_session_id,
            &title_generation_folder,
            &title_generation_prompt,
            title_generation_agent,
            Some(title_generation_task_generation),
        );
        self.replace_title_generation_task(
            &persisted_session_id,
            title_generation_task_generation,
            title_generation_task,
        );

        SessionTaskService::emit_session_updated(
            &services.event_sender(),
            &services.session_update_versions(),
            persisted_session_id.as_str(),
        );

        Ok(())
    }

    /// Loads the folder and agent/model selection used for draft title
    /// generation.
    async fn draft_title_generation_context(
        &self,
        services: &AppServices,
        project_id: i64,
        session_agent: AgentSelection,
        session_folder: PathBuf,
    ) -> Result<(PathBuf, AgentSelection), SessionError> {
        let project_working_dir = self.load_project_path(services, project_id).await?;
        let title_generation_agent =
            setting::load_default_fast_agent_setting(services, Some(project_id), session_agent)
                .await;
        let title_generation_folder = if services.fs_client().is_dir(session_folder.clone()) {
            session_folder
        } else {
            project_working_dir
        };

        Ok((title_generation_folder, title_generation_agent))
    }

    /// Returns whether a staged draft can start under the current one-level
    /// stack constraints.
    pub(crate) fn can_start_staged_session(&self, session_id: &str) -> bool {
        stack_can_start_staged_session(&self.state.sessions, session_id)
    }

    /// Returns whether a session can start branch-mutating work without
    /// competing with another member of its one-level stack.
    pub(crate) fn can_mutate_session_branch_in_stack(&self, session_id: &str) -> bool {
        stack_can_mutate_session_branch(&self.state.sessions, session_id)
    }

    /// Returns whether a session can enter the merge queue without competing
    /// with another member of its one-level stack.
    pub(crate) fn can_merge_session_branch_in_stack(&self, session_id: &str) -> bool {
        stack_can_merge_session_branch(&self.state.sessions, session_id)
    }

    /// Returns whether a session can start sync work without competing with
    /// another member of its one-level stack.
    pub(crate) fn can_rebase_session_branch_in_stack(&self, session_id: &str) -> bool {
        stack_can_rebase_session_branch(&self.state.sessions, session_id)
    }

    /// Returns whether a session can accept a reply without another stack
    /// member already owning active branch work.
    pub(crate) fn can_reply_to_session_in_stack(&self, session_id: &str) -> bool {
        stack_can_reply_to_session(&self.state.sessions, session_id)
    }

    /// Starts a `Draft` session from its persisted staged draft bundle.
    ///
    /// This materializes the deferred draft worktree before launching the
    /// first live turn. Stacked drafts additionally wait for a review-ready
    /// parent and an otherwise idle stack so only one branch-mutating session
    /// runs in that stack.
    ///
    /// # Errors
    /// Returns an error if the session is missing, is not a draft session, no
    /// drafts are staged, or launching the first turn fails.
    pub async fn start_staged_session(
        &mut self,
        services: &AppServices,
        session_id: &str,
    ) -> Result<(), SessionError> {
        let prompt = {
            let session = self.session_or_err(session_id)?;
            if !session.is_draft_session() {
                return Err(SessionError::Workflow(
                    "Only draft sessions can be started from staged drafts".to_string(),
                ));
            }
            if session.status != Status::Draft {
                return Err(SessionError::Workflow(
                    "Only `Draft` sessions can be started from staged drafts".to_string(),
                ));
            }
            if session.prompt.is_empty() {
                return Err(SessionError::Workflow(
                    "Stage at least one draft before starting the session".to_string(),
                ));
            }
            if !self.can_start_staged_session(session_id) {
                return Err(SessionError::Workflow(
                    "Stacked sessions can only start when their parent is in review and the stack \
                     has no other active branch work"
                        .to_string(),
                ));
            }

            TurnPrompt {
                attachments: session.draft_attachments.clone(),
                text: session.prompt.clone(),
                text_source: TurnPromptTextSource::UserPrompt,
            }
        };

        self.start_session(services, session_id, prompt).await?;

        if let Ok(session_index) = self.session_index_or_err(session_id)
            && let Some(session) = self.session_at_mut(session_index)
        {
            session.draft_attachments.clear();
        }

        if let Err(error) = draft::store_staged_draft_attachments(
            services.fs_client().as_ref(),
            services.base_path(),
            session_id,
            &[],
        )
        .await
        {
            warn!(
                session_id = session_id,
                error = %error,
                "failed to clear staged draft attachments after session start"
            );
        }

        Ok(())
    }

    /// Submits the first prompt for a blank session and starts the agent.
    ///
    /// The first prompt is persisted as both session prompt and session title.
    /// A detached one-shot title-generation task from the start-turn worker
    /// may replace that initial title once.
    ///
    /// # Errors
    /// Returns an error if the session is missing, its worktree cannot be
    /// prepared, or prompt persistence fails.
    pub async fn start_session(
        &mut self,
        services: &AppServices,
        session_id: &str,
        prompt: impl Into<TurnPrompt>,
    ) -> Result<(), SessionError> {
        let prompt = prompt.into();
        self.ensure_session_worktree_ready(services, session_id)
            .await?;

        let session_index = self.session_index_or_err(session_id)?;
        let (persisted_session_id, session_agent, title) = {
            let session = self
                .session_at_mut(session_index)
                .ok_or(SessionError::NotFound)?;

            session.prompt.clone_from(&prompt.text);

            let title = prompt.text.clone();
            session.title = Some(title.clone());
            let session_agent = session.agent;

            (session.id.clone(), session_agent, title)
        };

        let handles = self.session_handles_or_err(&persisted_session_id)?;
        let transcript = Arc::clone(&handles.transcript);
        let status_transition =
            StatusTransition::from_services(services, handles, persisted_session_id.clone());
        let app_event_tx = services.event_sender();

        self.persist_first_message_metadata(services, &persisted_session_id, &prompt.text, &title)
            .await;

        let prompt_transcript_text = prompt.transcript_text();
        let initial_output = Self::formatted_prompt_output(&prompt, false);
        SessionTaskService::append_session_transcript_message(
            &transcript,
            services.db(),
            &app_event_tx,
            &services.session_update_versions(),
            &persisted_session_id,
            SessionTranscriptMessageAppend {
                kind: SessionMessageKind::UserPrompt,
                raw_content: &prompt_transcript_text,
            },
        )
        .await;
        self.set_active_prompt_output(&persisted_session_id, initial_output);

        if !status_transition.apply(Status::InProgress).await {
            warn!(
                session_id = %persisted_session_id,
                "skipped session start status update because the in-memory status did not transition to in-progress"
            );
        }

        let operation_id = Uuid::new_v4().to_string();
        let command = SessionCommand::Run {
            operation_id,
            request_kind: AgentRequestKind::SessionStart,
            replay_transcript: None,
            prompt: prompt.clone(),
            turn_metadata: TurnMetadata {
                published_upstream_ref: None,
                session_agent,
            },
        };
        if let Err(error) = self
            .enqueue_session_command(services, &persisted_session_id, command)
            .await
        {
            self.cleanup_prompt_attachment_files(services, &prompt)
                .await;

            return Err(error);
        }

        if let Err(error) = self.clear_session_draft_flag(services, session_id).await {
            warn!(
                session_id,
                %error,
                "failed to clear draft flag after session start"
            );
        }

        Ok(())
    }

    /// Submits a follow-up prompt to an existing session.
    ///
    /// Returns `true` when the reply command was enqueued on the session
    /// worker, letting callers gate optimistic status advances on a real
    /// enqueue.
    pub async fn reply(
        &mut self,
        services: &AppServices,
        session_id: &str,
        prompt: impl Into<TurnPrompt>,
    ) -> bool {
        let prompt = prompt.into();
        let Ok(session) = self.session_or_err(session_id) else {
            return false;
        };
        let session_agent = session.agent;

        self.reply_impl(services, session_id, prompt, session_agent)
            .await
    }

    /// Stages one chat prompt into the in-memory queue for the running turn.
    ///
    /// The queue is owned by [`SessionHandles::queued_messages`] and lives
    /// only for the active app session, so queued prompts are discarded on
    /// `agentty` restart. The session worker drains the queue between turns
    /// without bouncing through `Review` and pauses drainage while the
    /// session sits in `Question`. `Ctrl+C` on the running turn drops the
    /// most recently queued chat message (LIFO) one press at a time without
    /// interrupting the running turn, and once the queue is empty a further
    /// press cancels the active turn.
    ///
    /// The just-pushed entry is mirrored into the render snapshot via
    /// [`SessionState::sync_session_from_handle`] so the inline `queued ›`
    /// row appears on the very next frame, and the targeted
    /// [`AppEvent::SessionUpdated`] event triggers a single-session redraw
    /// without paying for a full DB-backed `RefreshSessions` reload.
    ///
    /// # Errors
    /// Returns [`SessionError::NotFound`] when the session id does not
    /// resolve to a known session, or [`SessionError::Workflow`] when the
    /// payload is empty after trimming.
    pub fn enqueue_message(
        &mut self,
        services: &AppServices,
        session_id: &str,
        prompt: impl Into<TurnPrompt>,
    ) -> Result<(), SessionError> {
        let prompt = prompt.into();
        if prompt.is_empty() {
            return Err(SessionError::Workflow(
                "Cannot queue an empty chat message".to_string(),
            ));
        }

        let handles = self.session_handles_or_err(session_id)?;

        // Sync critical section (single push, no `.await`); `std::sync::Mutex`
        // is the correct choice per CLAUDE.md §"Mutex Selection".
        if let Ok(mut guard) = handles.queued_messages.lock() {
            guard.push_back(prompt);
        }

        self.state.sync_session_from_handle(session_id);

        SessionTaskService::emit_session_updated(
            &services.event_sender(),
            &services.session_update_versions(),
            session_id,
        );

        Ok(())
    }

    /// Updates and persists the agent/model selection for a single session.
    ///
    /// When `LastUsedModelAsDefault` is enabled, this also persists the chosen
    /// session agent/model pair as `DefaultSmartAgent` and
    /// `DefaultSmartModel`.
    ///
    /// When the model changes, this also clears any persisted provider-native
    /// conversation identifier so incompatible runtimes do not attempt resume
    /// with stale ids, and drops the existing session worker so the next turn
    /// creates a fresh worker with the correct [`AgentChannel`] type.
    ///
    /// # Errors
    /// Returns an error if the session is missing or persistence fails.
    pub async fn set_session_model(
        &mut self,
        services: &AppServices,
        session_id: &str,
        session_agent: AgentSelection,
    ) -> Result<(), SessionError> {
        let session_model = session_agent.model();
        let session_index = self.session_index_or_err(session_id)?;
        let agent_changed = self
            .session_at(session_index)
            .is_some_and(|session| session.agent != session_agent);
        let model_changed = self
            .session_at(session_index)
            .is_some_and(|session| session.agent.model() != session_model);
        let session_agent_kind = session_agent.kind().to_string();

        services
            .db()
            .sessions()
            .update_session_agent_model(session_id, &session_agent_kind, session_model.as_str())
            .await?;
        if agent_changed {
            services
                .db()
                .sessions()
                .update_session_provider_conversation_id(session_id, None)
                .await?;
            services
                .db()
                .sessions()
                .update_session_instruction_conversation_id(session_id, None)
                .await?;

            self.clear_session_worker(session_id);
        }

        let session_project_id = services
            .db()
            .sessions()
            .load_session_project_id(session_id)
            .await?;

        if Self::should_persist_last_used_model_as_default(services, session_project_id).await?
            && let Some(project_id) = session_project_id
        {
            services
                .db()
                .settings()
                .upsert_project_setting(
                    project_id,
                    SettingName::DefaultSmartAgent,
                    session_agent.kind().name(),
                )
                .await?;
            services
                .db()
                .settings()
                .upsert_project_setting(
                    project_id,
                    SettingName::DefaultSmartModel,
                    session_model.as_str(),
                )
                .await?;
        }

        services.emit_app_event(AppEvent::SessionModelUpdated {
            session_id: SessionId::from(session_id),
            session_agent,
        });

        if agent_changed || model_changed {
            self.mark_history_replay_pending(session_id);
        }

        Ok(())
    }

    /// Updates and persists the reasoning level for a single session.
    ///
    /// # Errors
    /// Returns an error if the session is missing or persistence fails.
    pub async fn set_session_reasoning_level(
        &mut self,
        services: &AppServices,
        session_id: &str,
        reasoning_level: ReasoningLevel,
    ) -> Result<(), SessionError> {
        self.session_index_or_err(session_id)?;

        services
            .db()
            .sessions()
            .update_session_reasoning_level(session_id, reasoning_level)
            .await?;

        services.emit_app_event(AppEvent::SessionReasoningLevelUpdated {
            reasoning_level,
            session_id: SessionId::from(session_id),
        });

        Ok(())
    }

    /// Returns whether session model switches should also persist the
    /// `DefaultSmartAgent` and `DefaultSmartModel` setting pair.
    async fn should_persist_last_used_model_as_default(
        services: &AppServices,
        project_id: Option<i64>,
    ) -> Result<bool, SessionError> {
        let Some(project_id) = project_id else {
            return Ok(false);
        };

        let should_persist = services
            .db()
            .settings()
            .get_project_setting(project_id, SettingName::LastUsedModelAsDefault)
            .await?
            .and_then(|setting_value| setting_value.parse::<bool>().ok())
            .unwrap_or(false);

        Ok(should_persist)
    }

    /// Returns the currently selected session, if any.
    pub fn selected_session(&self) -> Option<&Session> {
        self.state
            .table_state
            .selected()
            .and_then(|index| self.state.sessions.get(index))
    }

    /// Returns the session snapshot for one list index, if it still exists.
    pub fn session_at(&self, session_index: usize) -> Option<&Session> {
        self.state.sessions.get(session_index)
    }

    /// Returns the session identifier for the given list index.
    pub fn session_id_for_index(&self, session_index: usize) -> Option<SessionId> {
        self.state
            .sessions
            .get(session_index)
            .map(|session| session.id.clone())
    }

    /// Resolves a stable session identifier to the current list index.
    pub fn session_index_for_id(&self, session_id: &str) -> Option<usize> {
        self.state.session_index_for_id(session_id)
    }

    /// Publishes a review-ready session branch and creates or refreshes the
    /// linked forge review request.
    ///
    /// Existing links are refreshed after the branch push so repeated publish
    /// requests update the same remote review request instead of returning a
    /// stale stored summary. When no link is stored yet, the workflow reuses
    /// an existing remote review request for the session branch before
    /// creating a new one.
    ///
    /// # Errors
    /// Returns an error if the session is missing, cannot be published, git
    /// push fails, forge detection fails, the review-request operation fails,
    /// or persistence fails.
    pub async fn publish_review_request(
        &mut self,
        services: &AppServices,
        session_id: &str,
    ) -> Result<ReviewRequest, SessionError> {
        let session_index = self.session_index_or_err(session_id)?;
        let Some(session) = self.state.sessions.get(session_index) else {
            return Err(SessionError::NotFound);
        };

        if !session.status.allows_review_actions() {
            return Err(SessionError::Workflow(
                "Session must be in review to create a review request".to_string(),
            ));
        }

        let folder = session.folder.clone();
        let source_branch = session_branch(session_id);
        let linked_review_request = session.review_request.clone();
        let git_client = services.git_client();
        let published_upstream_ref = git_client
            .push_current_branch(folder.clone())
            .await
            .map_err(|error| {
                SessionError::Workflow(format!("Failed to publish session branch: {error}"))
            })?;
        self.store_published_upstream_ref(services, session_id, published_upstream_ref)
            .await?;

        let session = self
            .state
            .sessions
            .get(session_index)
            .ok_or(SessionError::NotFound)?;
        let review_request_client = services.review_request_client();
        let remote = self
            .review_request_remote(services, session, linked_review_request.as_ref())
            .await?;
        let review_request_summary = if let Some(review_request) = linked_review_request {
            review_request_client
                .refresh_review_request(remote, review_request.summary.display_id)
                .await
                .map_err(|error| SessionError::Workflow(error.detail_message()))?
        } else {
            match review_request_client
                .find_by_source_branch(remote.clone(), source_branch.clone())
                .await
                .map_err(|error| SessionError::Workflow(error.detail_message()))?
            {
                Some(existing_review_request) => review_request_client
                    .refresh_review_request(remote, existing_review_request.display_id)
                    .await
                    .map_err(|error| SessionError::Workflow(error.detail_message()))?,
                None => review_request_client
                    .create_review_request(
                        remote,
                        Self::load_review_request_create_input(
                            git_client.as_ref(),
                            session,
                            source_branch.clone(),
                        )
                        .await?,
                    )
                    .await
                    .map_err(|error| SessionError::Workflow(error.detail_message()))?,
            }
        };
        self.store_review_request_summary(services, session_id, review_request_summary)
            .await
    }

    /// Returns the browser-openable URL for one linked review request.
    ///
    /// # Errors
    /// Returns an error if the session is missing, has no linked review
    /// request, or the stored summary is missing a usable web URL.
    pub fn review_request_web_url(
        &self,
        services: &AppServices,
        session_id: &str,
    ) -> Result<String, SessionError> {
        let session = self.session_or_err(session_id)?;
        let review_request = session.review_request.as_ref().ok_or_else(|| {
            SessionError::Workflow("Session has no linked review request".to_string())
        })?;

        services
            .review_request_client()
            .review_request_web_url(&review_request.summary)
            .map_err(|error| SessionError::Workflow(error.detail_message()))
    }

    /// Deletes the currently selected session and cleans related resources.
    ///
    /// After persistence and filesystem cleanup, this triggers session and
    /// project-list reloads through app refresh events.
    pub async fn delete_selected_session(
        &mut self,
        projects: &ProjectManager,
        services: &AppServices,
    ) {
        let Some(cleanup) = self
            .remove_selected_session_from_state_and_db(projects, services)
            .await
        else {
            return;
        };

        Self::cleanup_deleted_session_resources(
            services.fs_client(),
            services.git_client(),
            cleanup,
        )
        .await;
    }

    /// Deletes the selected session while deferring filesystem cleanup to a
    /// background task.
    pub async fn delete_selected_session_deferred_cleanup(
        &mut self,
        projects: &ProjectManager,
        services: &AppServices,
    ) {
        let Some(cleanup) = self
            .remove_selected_session_from_state_and_db(projects, services)
            .await
        else {
            return;
        };

        let fs_client = services.fs_client();
        let git_client = services.git_client();
        tokio::spawn(async move {
            SessionManager::cleanup_deleted_session_resources(fs_client, git_client, cleanup).await;
        });
    }

    /// Removes the selected session from app state and persistence, returning
    /// deferred cleanup instructions for git and filesystem resources.
    async fn remove_selected_session_from_state_and_db(
        &mut self,
        projects: &ProjectManager,
        services: &AppServices,
    ) -> Option<DeletedSessionCleanup> {
        let selected_index = self.state.table_state.selected()?;
        if selected_index >= self.state.sessions.len() {
            return None;
        }

        let session = self.remove_session_at(selected_index)?;
        self.state.handles.remove(&session.id);
        self.remove_session_worktree_availability(&session.id);
        self.remove_at_mention_index_for_root(&session.folder);
        self.abort_title_generation_task(&session.id);
        self.clear_history_replay_pending(&session.id);
        SessionTaskService::remove_session_update_version(
            &services.session_update_versions(),
            &session.id,
        );

        if let Err(error) = services
            .db()
            .operations()
            .request_cancel_for_session_operations(&session.id)
            .await
        {
            warn!(
                session_id = %session.id,
                error = %error,
                "failed to cancel pending session operations during deletion"
            );
        }
        self.clear_session_worker(&session.id);
        if let Err(error) = services.db().sessions().delete_session(&session.id).await {
            warn!(
                session_id = %session.id,
                error = %error,
                "failed to delete session record during session deletion"
            );
        }
        services.emit_session_and_project_refresh_events();

        let staged_draft_root = services.base_path().join(&session.id);

        Some(DeletedSessionCleanup {
            branch_name: session_branch(&session.id),
            folder: session.folder,
            has_git_branch: projects.has_git_branch(),
            session_id: session.id,
            staged_draft_root,
            working_dir: projects.working_dir().to_path_buf(),
        })
    }

    /// Deletes worktree resources for a previously removed session.
    async fn cleanup_deleted_session_resources(
        fs_client: Arc<dyn FsClient>,
        git_client: Arc<dyn git::GitClient>,
        cleanup: DeletedSessionCleanup,
    ) {
        let repo_root = if cleanup.has_git_branch {
            git_client.find_git_repo_root(cleanup.working_dir).await
        } else {
            None
        };

        let cleanup_errors = Self::cleanup_session_worktree_resources(
            fs_client.clone(),
            git_client,
            cleanup.folder,
            cleanup.branch_name,
            repo_root,
            cleanup.has_git_branch,
        )
        .await;
        Self::warn_cleanup_errors(&cleanup.session_id, &cleanup_errors);
        if fs_client.is_dir(cleanup.staged_draft_root.clone())
            && let Err(error) = fs_client.remove_dir_all(cleanup.staged_draft_root).await
        {
            warn!(
                session_id = %cleanup.session_id,
                error = %error,
                "failed to remove staged draft directory during session deletion"
            );
        }
        Self::cleanup_session_temp_directory(fs_client, &cleanup.session_id).await;
    }

    /// Builds one normalized create-request payload from the session branch
    /// commit message.
    async fn load_review_request_create_input(
        git_client: &dyn git::GitClient,
        session: &Session,
        source_branch: String,
    ) -> Result<forge::CreateReviewRequestInput, SessionError> {
        let commit_message = git_client
            .head_commit_message(session.folder.clone())
            .await
            .map_err(|error| {
                SessionError::Workflow(format!(
                    "Failed to load session branch commit message: {error}"
                ))
            })?
            .ok_or_else(|| {
                SessionError::Workflow(
                    "Session branch has no commit message for review-request publishing."
                        .to_string(),
                )
            })?;
        let review_request_commit_message = review_request::parse_review_request_commit_message(
            &commit_message,
        )
        .ok_or_else(|| {
            SessionError::Workflow(
                "Session branch commit message must have a non-empty title for review-request \
                 publishing."
                    .to_string(),
            )
        })?;

        Ok(forge::CreateReviewRequestInput {
            body: review_request_commit_message.body,
            source_branch,
            target_branch: session.base_branch.clone(),
            title: review_request_commit_message.title,
        })
    }

    /// Converts one refreshed summary into persisted review-request metadata.
    pub(super) fn build_review_request(
        &self,
        summary: forge::ReviewRequestSummary,
    ) -> ReviewRequest {
        ReviewRequest {
            last_refreshed_at: unix_timestamp_from_system_time(self.state.clock.now_system_time()),
            summary,
        }
    }

    /// Persists one normalized review-request summary for a session.
    ///
    /// # Errors
    /// Returns an error if the session disappears or persistence fails.
    pub(crate) async fn store_review_request_summary(
        &mut self,
        services: &AppServices,
        session_id: &str,
        summary: forge::ReviewRequestSummary,
    ) -> Result<ReviewRequest, SessionError> {
        let session_index = self.session_index_or_err(session_id)?;
        let review_request = self.build_review_request(summary);

        self.store_review_request(services, session_index, review_request)
            .await
    }

    /// Persists one linked review request in memory and the database.
    ///
    /// # Errors
    /// Returns an error if the session disappears or persistence fails.
    pub(super) async fn store_review_request(
        &mut self,
        services: &AppServices,
        session_index: usize,
        review_request: ReviewRequest,
    ) -> Result<ReviewRequest, SessionError> {
        let session_id = self
            .state
            .sessions
            .get(session_index)
            .map(|session| session.id.clone())
            .ok_or(SessionError::NotFound)?;
        services
            .db()
            .reviews()
            .update_session_review_request(&session_id, Some(review_request.clone()))
            .await?;

        let Some(session) = self.state.sessions.get_mut(session_index) else {
            return Err(SessionError::NotFound);
        };
        session.review_request = Some(review_request.clone());

        Ok(review_request)
    }

    /// Persists one published upstream reference in memory and the database.
    ///
    /// # Errors
    /// Returns an error if the session disappears or persistence fails.
    pub(super) async fn store_published_upstream_ref(
        &mut self,
        services: &AppServices,
        session_id: &str,
        published_upstream_ref: String,
    ) -> Result<(), SessionError> {
        services
            .db()
            .sessions()
            .update_session_published_upstream_ref(session_id, Some(published_upstream_ref.clone()))
            .await?;

        let session_index = self.session_index_or_err(session_id)?;
        let Some(session) = self.state.sessions.get_mut(session_index) else {
            return Err(SessionError::NotFound);
        };
        session.published_upstream_ref = Some(published_upstream_ref);

        Ok(())
    }

    /// Validates and queues a follow-up prompt for an existing session.
    ///
    /// Gathers reply context, appends the prompt line to session output, builds
    /// a [`SessionCommand::Run`] with the appropriate [`AgentRequestKind`],
    /// and enqueues it on the session worker. Returns `true` only when the
    /// command reached the worker queue, so callers can defer optimistic status
    /// advances until the reply is genuinely in flight.
    async fn reply_impl(
        &mut self,
        services: &AppServices,
        session_id: &str,
        prompt: TurnPrompt,
        session_agent: AgentSelection,
    ) -> bool {
        let Ok(session_index) = self.session_index_or_err(session_id) else {
            return false;
        };
        let should_replay_history = self.should_replay_history(session_id);
        let (replay_transcript, is_first_message, persisted_session_id, title_to_save) =
            match self.prepare_reply_context(session_index, &prompt, should_replay_history) {
                Ok(Some(reply_context)) => reply_context,
                Ok(None) => return false,
                Err(error) => {
                    self.append_reply_status_error(services, session_id, &error)
                        .await;

                    return false;
                }
            };

        if should_replay_history {
            self.clear_history_replay_pending(&persisted_session_id);
        }

        let app_event_tx = services.event_sender();

        let Ok(handles) = self.session_handles_or_err(&persisted_session_id) else {
            return false;
        };

        let transcript = Arc::clone(&handles.transcript);
        let status_transition =
            StatusTransition::from_services(services, handles, persisted_session_id.clone());

        let effective_prompt = prompt;

        if let Some(title) = title_to_save {
            self.persist_first_message_metadata(
                services,
                &persisted_session_id,
                &effective_prompt.text,
                &title,
            )
            .await;

            if !status_transition.apply(Status::InProgress).await {
                warn!(
                    session_id = %persisted_session_id,
                    "skipped reply status update because the in-memory status did not transition to in-progress"
                );
            }
        }

        self.append_reply_prompt_line(
            services,
            &transcript,
            &app_event_tx,
            &persisted_session_id,
            &effective_prompt,
        )
        .await;
        let published_upstream_ref = self
            .session_or_err(&persisted_session_id)
            .ok()
            .and_then(|session| session.published_upstream_ref.clone());

        let command = Self::build_session_command(BuildSessionCommandInput {
            is_first_message,
            published_upstream_ref,
            prompt: effective_prompt.clone(),
            replay_transcript,
            session_agent,
        });
        self.enqueue_reply_command(
            services,
            &transcript,
            &persisted_session_id,
            &effective_prompt,
            command,
        )
        .await
    }

    /// Validates reply eligibility and gathers per-session values needed for
    /// queueing a reply command.
    ///
    /// # Errors
    /// Returns a [`SessionError::Workflow`] when session status does not allow
    /// replying.
    fn prepare_reply_context(
        &mut self,
        session_index: usize,
        prompt: &TurnPrompt,
        should_replay_history: bool,
    ) -> Result<Option<ReplyContext>, SessionError> {
        let Some(session_id) = self
            .state
            .sessions
            .get(session_index)
            .map(|session| session.id.clone())
        else {
            return Ok(None);
        };
        if !self.can_reply_to_session_in_stack(session_id.as_str()) {
            return Err(SessionError::Workflow(
                "Stacked replies can only run when no other stack session is active".to_string(),
            ));
        }

        let Some(session) = self.state.sessions.get_mut(session_index) else {
            return Ok(None);
        };

        let is_first_message = session.prompt.is_empty();
        let allowed = session.status.allows_review_actions()
            || session.status == Status::Question
            || (is_first_message && session.status == Status::Draft);
        if !allowed {
            return Err(SessionError::Workflow(
                "Session must be in review status".to_string(),
            ));
        }

        let mut title_to_save = None;
        if is_first_message {
            session.prompt.clone_from(&prompt.text);
            let title = prompt.text.clone();
            session.title = Some(title.clone());
            title_to_save = Some(title);
        }

        let replay_transcript = if !is_first_message
            && (should_replay_history
                || agent::transport_mode(session.agent.kind()).uses_app_server())
        {
            session
                .transcript
                .as_ref()
                .and_then(SessionTranscript::replay_text)
        } else {
            None
        };

        Ok(Some((
            replay_transcript,
            is_first_message,
            session.id.clone(),
            title_to_save,
        )))
    }

    /// Persists first-message prompt/title metadata before queueing execution.
    ///
    /// This writes the initial prompt/title.
    ///
    /// Title generation itself is triggered once from the start-turn worker
    /// path as soon as the first turn starts running.
    async fn persist_first_message_metadata(
        &self,
        services: &AppServices,
        session_id: &str,
        prompt: &str,
        title: &str,
    ) {
        if let Err(error) = services
            .db()
            .sessions()
            .update_session_title(session_id, title)
            .await
        {
            warn!(
                session_id = session_id,
                error = %error,
                "failed to persist first-message session title"
            );
        }

        if let Err(error) = services
            .db()
            .sessions()
            .update_session_prompt(session_id, prompt)
            .await
        {
            warn!(
                session_id = session_id,
                error = %error,
                "failed to persist first-message session prompt"
            );
        }
    }

    /// Appends the user reply marker line to session output.
    async fn append_reply_prompt_line(
        &mut self,
        services: &AppServices,
        transcript: &Arc<Mutex<SessionTranscript>>,
        app_event_tx: &mpsc::UnboundedSender<AppEvent>,
        session_id: &str,
        prompt: &TurnPrompt,
    ) {
        let prompt_transcript_text = prompt.transcript_text();
        let reply_line = Self::formatted_prompt_output(prompt, true);
        SessionTaskService::append_session_transcript_message(
            transcript,
            services.db(),
            app_event_tx,
            &services.session_update_versions(),
            session_id,
            SessionTranscriptMessageAppend {
                kind: SessionMessageKind::UserPrompt,
                raw_content: &prompt_transcript_text,
            },
        )
        .await;
        self.set_active_prompt_output(session_id, reply_line);
    }

    /// Formats one user prompt block for persisted session output.
    ///
    /// The first line uses `USER_PROMPT_PREFIX`; continuation lines use
    /// `USER_PROMPT_CONTINUATION_PREFIX` so embedded blank lines remain inside
    /// the prompt block instead of being interpreted as prompt terminators.
    fn formatted_prompt_output(prompt: &TurnPrompt, prepend_newline: bool) -> String {
        let prompt_text = prompt.transcript_text();
        let prompt_lines = prompt_text.split('\n').collect::<Vec<_>>();
        let mut formatted_lines = Vec::with_capacity(prompt_lines.len());

        for (index, prompt_line) in prompt_lines.into_iter().enumerate() {
            let prefix = if index == 0 {
                USER_PROMPT_PREFIX
            } else {
                USER_PROMPT_CONTINUATION_PREFIX
            };

            formatted_lines.push(format!("{prefix}{prompt_line}"));
        }

        let prompt_block = formatted_lines.join("\n");
        if prepend_newline {
            return format!("\n{prompt_block}\n\n");
        }

        format!("{prompt_block}\n\n")
    }

    /// Appends one newly staged prompt onto the persisted draft-session
    /// prompt text stored in `session.prompt`.
    ///
    /// Attachment placeholders are renumbered sequentially so draft sessions
    /// can keep one flat prompt string while preserving a stable attachment
    /// order across multiple staging passes.
    fn append_staged_prompt(
        existing_prompt: &str,
        prompt: &TurnPrompt,
        next_attachment_number: usize,
    ) -> String {
        let staged_prompt = Self::renumbered_prompt_text(prompt, next_attachment_number);
        if existing_prompt.is_empty() {
            return staged_prompt;
        }

        format!("{existing_prompt}\n\n{staged_prompt}")
    }

    /// Returns the staged prompt text after renumbering any attachment
    /// placeholders to their global draft-session positions.
    fn renumbered_prompt_text(prompt: &TurnPrompt, next_attachment_number: usize) -> String {
        let mut prompt_text = prompt.text.clone();

        for (offset, attachment) in prompt.attachments.iter().enumerate() {
            let placeholder = format!("[Image #{}]", next_attachment_number.saturating_add(offset));
            prompt_text = replace_first(&prompt_text, &attachment.placeholder, &placeholder);
        }

        prompt_text
    }

    /// Returns the prompt attachments rewritten to the global draft-session
    /// placeholder sequence.
    fn renumbered_attachments(
        prompt: &TurnPrompt,
        next_attachment_number: usize,
    ) -> Vec<TurnPromptAttachment> {
        prompt
            .attachments
            .iter()
            .enumerate()
            .map(|(offset, attachment)| TurnPromptAttachment {
                placeholder: format!("[Image #{}]", next_attachment_number.saturating_add(offset)),
                local_image_path: attachment.local_image_path.clone(),
            })
            .collect()
    }

    /// Builds a queued command for starting or resuming a session interaction.
    ///
    /// Creates a [`SessionCommand::Run`] with
    /// [`AgentRequestKind::SessionStart`] for first messages and
    /// [`AgentRequestKind::SessionResume`] with optional transcript replay
    /// for subsequent replies.
    fn build_session_command(input: BuildSessionCommandInput) -> SessionCommand {
        let BuildSessionCommandInput {
            is_first_message,
            published_upstream_ref,
            prompt,
            replay_transcript,
            session_agent,
        } = input;
        let operation_id = Uuid::new_v4().to_string();
        let request_kind = if is_first_message {
            AgentRequestKind::SessionStart
        } else {
            AgentRequestKind::SessionResume
        };

        SessionCommand::Run {
            operation_id,
            request_kind,
            replay_transcript,
            prompt,
            turn_metadata: TurnMetadata {
                published_upstream_ref,
                session_agent,
            },
        }
    }

    /// Appends a reply-error notice to the session output so the user sees
    /// why the reply was rejected.
    async fn append_reply_status_error(
        &self,
        services: &AppServices,
        session_id: &str,
        error: &SessionError,
    ) {
        let status_error = TranscriptNotice::ReplyError.format(error);
        let Ok(handles) = self.session_handles_or_err(session_id) else {
            return;
        };
        let app_event_tx = services.event_sender();

        SessionTaskService::append_workflow_notice(
            &handles.transcript,
            services.db(),
            &app_event_tx,
            &services.session_update_versions(),
            session_id,
            &status_error,
        )
        .await;
    }

    /// Returns `true` when the command reached the session worker queue and
    /// `false` when enqueueing failed and a reply-error notice was appended.
    async fn enqueue_reply_command(
        &mut self,
        services: &AppServices,
        transcript: &Arc<Mutex<SessionTranscript>>,
        persisted_session_id: &str,
        prompt: &TurnPrompt,
        command: SessionCommand,
    ) -> bool {
        if let Err(error) = self
            .enqueue_session_command(services, persisted_session_id, command)
            .await
        {
            self.cleanup_prompt_attachment_files(services, prompt).await;

            let error_line = TranscriptNotice::ReplyError.format(error);
            let app_event_tx = services.event_sender();
            SessionTaskService::append_workflow_notice(
                transcript,
                services.db(),
                &app_event_tx,
                &services.session_update_versions(),
                persisted_session_id,
                &error_line,
            )
            .await;

            return false;
        }

        true
    }

    /// Spawns one detached model command that generates a session title for
    /// one prompt snapshot.
    ///
    /// The generated title is persisted only when the session prompt still
    /// matches the prompt used to generate it, then a `RefreshSessions` event
    /// is emitted so list-mode snapshots pick up the new title. Callers that
    /// can supersede draft-title generation should retain the returned task
    /// handle and abort any older in-flight task before replacing it.
    pub(crate) fn spawn_session_title_generation_task(
        app_event_tx: mpsc::UnboundedSender<AppEvent>,
        db: db::AppRepositories,
        session_id: &str,
        folder: &Path,
        prompt: &str,
        session_agent: AgentSelection,
        tracked_generation: Option<u64>,
    ) -> tokio::task::JoinHandle<()> {
        let folder = folder.to_path_buf();
        let prompt = prompt.to_string();
        let persisted_session_id = SessionId::from(session_id);
        let tracked_completion =
            tracked_generation.map(|generation| TitleGenerationTaskCompletion {
                generation,
                session_id: persisted_session_id.clone(),
            });

        tokio::spawn(async move {
            let Ok(title_generation_prompt) =
                SessionManager::session_title_generation_prompt(&prompt)
            else {
                SessionManager::emit_title_generation_finished_event(
                    &app_event_tx,
                    tracked_completion.as_ref(),
                );
                return;
            };

            let Some(title_response) = SessionManager::run_title_generation_command(
                folder.as_path(),
                &title_generation_prompt,
                session_agent,
            )
            .await
            else {
                SessionManager::emit_title_generation_finished_event(
                    &app_event_tx,
                    tracked_completion.as_ref(),
                );
                return;
            };

            let Some(generated_title) =
                SessionManager::parse_generated_session_title(&title_response)
            else {
                SessionManager::emit_title_generation_finished_event(
                    &app_event_tx,
                    tracked_completion.as_ref(),
                );
                return;
            };

            if generated_title == prompt {
                SessionManager::emit_title_generation_finished_event(
                    &app_event_tx,
                    tracked_completion.as_ref(),
                );
                return;
            }

            match db
                .sessions()
                .update_session_title_for_prompt(&persisted_session_id, &prompt, &generated_title)
                .await
            {
                Ok(true) => {
                    if app_event_tx.send(AppEvent::RefreshSessions).is_err() {
                        warn!(
                            session_id = %persisted_session_id,
                            "failed to refresh sessions after title generation because the app event receiver is closed"
                        );
                    }
                }
                Ok(false) => {}
                Err(error) => {
                    warn!(
                        session_id = %persisted_session_id,
                        error = %error,
                        "failed to persist generated session title"
                    );
                }
            }

            SessionManager::emit_title_generation_finished_event(
                &app_event_tx,
                tracked_completion.as_ref(),
            );
        })
    }

    /// Emits one tracked title-generation completion event when the task was
    /// registered in the per-session task map.
    fn emit_title_generation_finished_event(
        app_event_tx: &mpsc::UnboundedSender<AppEvent>,
        tracked_completion: Option<&TitleGenerationTaskCompletion>,
    ) {
        let Some(tracked_completion) = tracked_completion else {
            return;
        };

        if app_event_tx
            .send(AppEvent::SessionTitleGenerationFinished {
                generation: tracked_completion.generation,
                session_id: tracked_completion.session_id.clone(),
            })
            .is_err()
        {
            warn!(
                session_id = %tracked_completion.session_id,
                generation = tracked_completion.generation,
                "failed to send session title generation completion event because the app event receiver is closed"
            );
        }
    }

    /// Executes one detached title-generation command and returns parsed
    /// content.
    async fn run_title_generation_command(
        folder: &Path,
        prompt: &str,
        session_agent: AgentSelection,
    ) -> Option<String> {
        let response = agent::submit_one_shot(agent::OneShotRequest {
            agent_kind: session_agent.kind(),
            child_pid: None,
            folder,
            model: session_agent.model(),
            prompt,
            request_kind: AgentRequestKind::UtilityPrompt,
            reasoning_level: ReasoningLevel::default(),
        })
        .await
        .ok()?;

        Some(response.to_answer_display_text())
    }

    /// Builds the title-generation instruction prompt from the user message.
    ///
    /// # Errors
    /// Returns an error if Askama template rendering fails.
    fn session_title_generation_prompt(prompt: &str) -> Result<String, SessionError> {
        let template = SessionTitleGenerationPromptTemplate { prompt };

        template.render().map_err(|error| {
            SessionError::Workflow(format!(
                "Failed to render `session_title_generation_prompt.md`: {error}"
            ))
        })
    }

    /// Parses model output into a normalized one-line session title.
    ///
    /// Accepts either a plain-text title line or a protocol-wrapped response
    /// (`{"answer":"..."}`) whose first answer line contains the title.
    ///
    /// Returns [`None`] when no usable title line is present.
    fn parse_generated_session_title(content: &str) -> Option<String> {
        let content = content.trim();
        if content.is_empty() {
            return None;
        }

        if let Ok(protocol_response) = parse_agent_response_strict(content) {
            return Self::parse_generated_session_title_from_protocol_response(&protocol_response);
        }

        let first_line = Self::first_nonempty_line(content)?;

        Self::normalize_generated_session_title(first_line)
    }

    /// Extracts the first usable title candidate from protocol `answer`
    /// content.
    fn parse_generated_session_title_from_protocol_response(
        protocol_response: &AgentResponse,
    ) -> Option<String> {
        for answer in protocol_response.answers() {
            if let Some(first_line) = Self::first_nonempty_line(&answer)
                && let Some(parsed_title) = Self::normalize_generated_session_title(first_line)
            {
                return Some(parsed_title);
            }
        }

        None
    }

    /// Returns the first non-empty line from model output content.
    fn first_nonempty_line(content: &str) -> Option<&str> {
        content.lines().find_map(|line| {
            let trimmed_line = line.trim();
            if trimmed_line.is_empty() {
                return None;
            }

            Some(trimmed_line)
        })
    }

    /// Normalizes one candidate title and rejects status-like model output.
    ///
    /// Title generation runs through a general utility prompt, so providers can
    /// occasionally return first-person progress prose. Those candidates are
    /// rejected instead of overwriting the user-prompt fallback title.
    fn normalize_generated_session_title(candidate: &str) -> Option<String> {
        let mut title = candidate.trim().to_string();
        if let Some((prefix, remainder)) = title.split_once(':')
            && prefix.trim().eq_ignore_ascii_case("title")
        {
            title = remainder.trim().to_string();
        }

        title = title
            .trim_matches(|ch| matches!(ch, '"' | '\'' | '`'))
            .trim()
            .to_string();

        if title.is_empty() {
            return None;
        }

        if !Self::is_generated_session_title_candidate(&title) {
            return None;
        }

        Some(title)
    }

    /// Returns whether a normalized generated title looks like requested work
    /// rather than model progress, narration, or other non-title prose.
    fn is_generated_session_title_candidate(title: &str) -> bool {
        if title.chars().count() > GENERATED_SESSION_TITLE_MAX_CHARACTERS {
            return false;
        }

        if Self::starts_with_first_person_pronoun(title) {
            return false;
        }

        if Self::starts_with_progress_prefix(title) {
            return false;
        }

        true
    }

    /// Returns whether the title begins with a first-person pronoun shape.
    fn starts_with_first_person_pronoun(title: &str) -> bool {
        let mut characters = title.chars();
        if !matches!(characters.next(), Some('I' | 'i')) {
            return false;
        }

        matches!(characters.next(), Some(' ' | '\'' | '\u{2019}'))
    }

    /// Returns whether the title begins with a progress/status gerund.
    fn starts_with_progress_prefix(title: &str) -> bool {
        let lower_title = title.to_ascii_lowercase();

        GENERATED_SESSION_TITLE_PROGRESS_PREFIXES
            .iter()
            .any(|prefix| lower_title.starts_with(prefix))
    }

    /// Resolves the default agent/model selection for a new session.
    async fn resolve_default_session_agent(
        &self,
        services: &AppServices,
        project_id: i64,
    ) -> AgentSelection {
        let available_agent_kinds = services.available_agent_kinds();
        let fallback_agent_kind = available_agent_kinds
            .first()
            .copied()
            .unwrap_or(AgentKind::Antigravity);
        let fallback_selection = crate::domain::agent::resolve_agent_selection_for_model(
            self.default_session_model,
            fallback_agent_kind,
            &available_agent_kinds,
        );

        setting::load_default_smart_agent_setting(services, Some(project_id), fallback_selection)
            .await
    }

    /// Reverts filesystem and database changes after session creation failure.
    async fn rollback_failed_session_creation(
        &self,
        services: &AppServices,
        folder: &Path,
        repo_root: &Path,
        session_id: &str,
        worktree_branch: &str,
        session_saved: bool,
    ) {
        if session_saved {
            if let Err(error) = services.db().sessions().delete_session(session_id).await {
                warn!(
                    session_id = session_id,
                    error = %error,
                    "failed to roll back persisted session metadata"
                );
            }
            SessionTaskService::remove_session_update_version(
                &services.session_update_versions(),
                session_id,
            );
        }

        {
            let git_client = services.git_client();
            let folder = folder.to_path_buf();
            let repo_root = repo_root.to_path_buf();
            let worktree_branch = worktree_branch.to_string();
            if let Err(error) = git_client.remove_worktree(folder).await {
                warn!(
                    session_id = session_id,
                    error = %error,
                    "failed to remove worktree while rolling back session creation"
                );
            }

            if let Err(error) = git_client.delete_branch(repo_root, worktree_branch).await {
                warn!(
                    session_id = session_id,
                    error = %error,
                    "failed to delete branch while rolling back session creation"
                );
            }
        }

        if let Err(error) = services
            .fs_client()
            .remove_dir_all(folder.to_path_buf())
            .await
        {
            warn!(
                session_id = session_id,
                error = %error,
                "failed to remove session worktree directory while rolling back session creation"
            );
        }

        Self::cleanup_session_temp_directory(services.fs_client(), session_id).await;
    }

    /// Records that one session was created and warns if analytics persistence
    /// fails.
    async fn record_session_creation_activity(services: &AppServices, session_id: &str) {
        if let Err(error) = services
            .db()
            .activity()
            .insert_session_creation_activity_now(session_id)
            .await
        {
            warn!(
                session_id = session_id,
                error = %error,
                "failed to record session creation activity"
            );
        }
    }

    /// Appends text to a specific session output stream.
    pub(crate) async fn append_output_for_session(
        &self,
        services: &AppServices,
        session_id: &str,
        output: &str,
    ) {
        let Ok((session, handles)) = self.session_and_handles_or_err(session_id) else {
            return;
        };
        let app_event_tx = services.event_sender();

        SessionTaskService::append_workflow_notice(
            &handles.transcript,
            services.db(),
            &app_event_tx,
            &services.session_update_versions(),
            &session.id,
            output,
        )
        .await;
    }

    /// Removes prompt attachment files that are no longer owned by the
    /// composer or worker.
    ///
    /// Only Agentty-managed temp files under `AGENTTY_ROOT/tmp/` are removed.
    pub(crate) async fn cleanup_prompt_attachment_files(
        &self,
        services: &AppServices,
        prompt: &TurnPrompt,
    ) {
        Self::cleanup_prompt_attachment_paths(
            services.fs_client(),
            prompt.local_image_paths().cloned().collect(),
        )
        .await;
    }

    /// Cancels a review, running, or unstarted draft session.
    ///
    /// Persisted transcript metadata remains available after the worktree
    /// checkout and session branch are removed. Draft sessions that never
    /// created a worktree only update persisted state and skip worktree
    /// cleanup. Running sessions first request operation cancellation and fire
    /// the active turn's cancellation token so provider work stops before the
    /// terminal `Canceled` status is persisted.
    ///
    /// # Errors
    /// Returns an error if the session is not found or is not cancelable.
    pub async fn cancel_session(
        &self,
        services: &AppServices,
        session_id: &str,
    ) -> Result<(), SessionError> {
        let status_updated = self.cancel_single_session(services, session_id).await?;
        if status_updated {
            self.cancel_stacked_child_sessions(services, session_id)
                .await;
        }

        Ok(())
    }

    /// Cancels one session without cascading into stacked children.
    async fn cancel_single_session(
        &self,
        services: &AppServices,
        session_id: &str,
    ) -> Result<bool, SessionError> {
        let session = self.session_or_err(session_id)?;
        if !session.allows_cancel_action() {
            return Err(SessionError::Workflow(
                "Session must be running, in review, or be an unstarted draft to be canceled"
                    .to_string(),
            ));
        }

        let branch_name = session_branch(&session.id);
        let folder = session.folder.clone();
        let is_running = session.status == Status::InProgress;
        let has_worktree = services.fs_client().is_dir(folder.clone());
        let handles = self.session_handles_or_err(session_id)?;
        let status_transition = StatusTransition::from_services(services, handles, session_id);

        if is_running {
            Self::signal_running_session_cancellation(services, handles, session_id).await;
        }

        let status_updated = status_transition.apply(Status::Canceled).await;

        if status_updated {
            Self::spawn_canceled_session_cleanup(
                services,
                folder,
                branch_name,
                has_worktree,
                session_id.to_string(),
            );
        }

        Ok(status_updated)
    }

    /// Cancels every loaded one-level stacked child of `parent_session_id`.
    ///
    /// Child cancellation is best-effort after the parent has already reached
    /// `Canceled`, but it reuses the same single-session cancellation path so
    /// staged draft temp data and any future child worktree resources are
    /// cleaned consistently.
    pub(crate) async fn cancel_stacked_child_sessions(
        &self,
        services: &AppServices,
        parent_session_id: &str,
    ) {
        for child_session_id in self.stacked_child_session_ids(parent_session_id) {
            if let Err(error) = self
                .cancel_single_session(services, child_session_id.as_str())
                .await
            {
                warn!(
                    parent_session_id = parent_session_id,
                    child_session_id = %child_session_id,
                    error = %error,
                    "failed to cancel stacked child session after parent cancellation"
                );
            }
        }
    }

    /// Returns loaded one-level child ids for one parent session.
    fn stacked_child_session_ids(&self, parent_session_id: &str) -> Vec<SessionId> {
        self.state
            .sessions
            .iter()
            .filter(|session| {
                session
                    .parent_session_id
                    .as_ref()
                    .is_some_and(|parent_id| parent_id.as_str() == parent_session_id)
            })
            .map(|session| session.id.clone())
            .collect()
    }

    /// Defers terminal cancellation cleanup so foreground key handling returns
    /// after persisted status changes instead of waiting on git and filesystem
    /// removal.
    ///
    /// The background task resolves the shared repository root only when a
    /// worktree exists, removes git worktree/branch resources, then clears the
    /// session-scoped prompt temp directory. Cleanup remains best-effort and
    /// reports failures through debug-visible warnings.
    fn spawn_canceled_session_cleanup(
        services: &AppServices,
        folder: PathBuf,
        branch_name: String,
        has_worktree: bool,
        session_id: String,
    ) {
        let fs_client = services.fs_client();
        let git_client = services.git_client();
        let cleanup_task_handle = tokio::spawn(async move {
            if has_worktree {
                let repo_root = git_client.main_repo_root(folder.clone()).await.ok();
                let cleanup_errors = Self::cleanup_session_worktree_resources(
                    Arc::clone(&fs_client),
                    Arc::clone(&git_client),
                    folder,
                    branch_name,
                    repo_root,
                    true,
                )
                .await;
                Self::warn_cleanup_errors(&session_id, &cleanup_errors);
            }

            Self::cleanup_session_temp_directory(fs_client, &session_id).await;
        });
        services.track_cleanup_task(cleanup_task_handle);
    }

    /// Requests cancellation for queued operations and signals the active
    /// running turn for a session that is being terminally canceled.
    async fn signal_running_session_cancellation(
        services: &AppServices,
        handles: &SessionHandles,
        session_id: &str,
    ) {
        if let Err(error) = services
            .db()
            .operations()
            .request_cancel_for_session_operations(session_id)
            .await
        {
            warn!(
                session_id = session_id,
                error = %error,
                "failed to request cancellation for running session operations"
            );
        }

        if let Ok(mut queued_messages) = handles.queued_messages.lock() {
            queued_messages.clear();
        }

        match handles.cancel_token.lock() {
            Ok(cancel_token) => cancel_token.cancel(),
            Err(error) => {
                warn!(
                    session_id = session_id,
                    error = %error,
                    "failed to lock running session cancel token"
                );
            }
        }
    }

    /// Removes git and filesystem resources for one session worktree.
    ///
    /// This best-effort helper is shared by terminal-state cleanup and session
    /// deletion so both paths remove the linked worktree checkout, delete the
    /// session branch when the shared repository root is known, and finally
    /// remove the directory from disk. Any cleanup failures are returned as
    /// human-readable messages so callers can surface them when needed.
    #[must_use]
    async fn cleanup_session_worktree_resources(
        fs_client: Arc<dyn FsClient>,
        git_client: Arc<dyn git::GitClient>,
        folder: PathBuf,
        branch_name: String,
        repo_root: Option<PathBuf>,
        remove_git_resources: bool,
    ) -> Vec<String> {
        let mut cleanup_errors = Vec::new();

        if remove_git_resources {
            if let Err(error) = git_client.remove_worktree(folder.clone()).await {
                cleanup_errors.push(format!("failed to remove worktree: {error}"));
            }

            if let Some(repo_root) = repo_root
                && let Err(error) = git_client.delete_branch(repo_root, branch_name).await
            {
                cleanup_errors.push(format!("failed to delete branch: {error}"));
            }
        }

        if let Err(error) = fs_client.remove_dir_all(folder).await {
            cleanup_errors.push(format!("failed to remove worktree directory: {error}"));
        }

        cleanup_errors
    }

    /// Emits debug-visible warnings for best-effort cleanup failures.
    fn warn_cleanup_errors(session_id: &str, cleanup_errors: &[String]) {
        for cleanup_error in cleanup_errors {
            warn!(session_id = session_id, "{cleanup_error}");
        }
    }

    /// Removes Agentty-managed prompt attachment files and prunes their
    /// now-empty image directory when possible.
    pub(crate) async fn cleanup_prompt_attachment_paths(
        fs_client: Arc<dyn FsClient>,
        attachment_paths: Vec<PathBuf>,
    ) {
        Self::cleanup_prompt_attachment_paths_in_root(
            fs_client,
            &prompt_attachment_tmp_root(),
            attachment_paths,
        )
        .await;
    }

    /// Removes Agentty-managed prompt attachment files inside one explicit tmp
    /// root and prunes their shared image directory only when it is empty.
    ///
    /// The image directory is shared per session, so other queued prompts or
    /// the active composer may still reference sibling files there. Pruning
    /// uses [`FsClient::remove_dir`] (empty-only) and silently tolerates the
    /// `DirectoryNotEmpty` and `NotFound` cases so retracting one prompt
    /// never deletes another prompt's attachments.
    async fn cleanup_prompt_attachment_paths_in_root(
        fs_client: Arc<dyn FsClient>,
        managed_tmp_root: &Path,
        attachment_paths: Vec<PathBuf>,
    ) {
        if attachment_paths.is_empty() {
            return;
        }

        let image_directory =
            managed_prompt_attachment_directory(&attachment_paths, managed_tmp_root);

        for attachment_path in attachment_paths {
            if is_managed_prompt_attachment_path(&attachment_path, managed_tmp_root)
                && let Err(error) = fs_client.remove_file(attachment_path).await
            {
                warn!(
                    error = %error,
                    "failed to remove managed prompt attachment file"
                );
            }
        }

        if let Some(image_directory) = image_directory
            && let Err(error) = fs_client.remove_dir(image_directory).await
        {
            let FsError::Io(io_error) = &error;
            if !matches!(
                io_error.kind(),
                std::io::ErrorKind::DirectoryNotEmpty | std::io::ErrorKind::NotFound
            ) {
                warn!(
                    error = %error,
                    "failed to remove managed prompt attachment directory"
                );
            }
        }
    }

    /// Removes the session-scoped temp directory used for pasted prompt
    /// images.
    async fn cleanup_session_temp_directory(fs_client: Arc<dyn FsClient>, session_id: &str) {
        if let Err(error) = fs_client
            .remove_dir_all(session_prompt_temp_directory(session_id))
            .await
        {
            warn!(
                session_id = session_id,
                error = %error,
                "failed to remove session prompt temp directory"
            );
        }
    }
}

/// Replaces only the first occurrence of `needle` in `haystack`.
///
/// If `needle` is absent, the original string is returned unchanged.
fn replace_first(haystack: &str, needle: &str, replacement: &str) -> String {
    let Some(match_index) = haystack.find(needle) else {
        return haystack.to_string();
    };

    let mut replaced = String::with_capacity(
        haystack
            .len()
            .saturating_sub(needle.len())
            .saturating_add(replacement.len()),
    );
    replaced.push_str(&haystack[..match_index]);
    replaced.push_str(replacement);
    replaced.push_str(&haystack[match_index + needle.len()..]);

    replaced
}

/// Returns the session-scoped temp directory used for pasted prompt images.
fn session_prompt_temp_directory(session_id: &str) -> PathBuf {
    agentty_home().join("tmp").join(session_id)
}

/// Returns the Agentty-owned tmp root used for pasted prompt attachments.
fn prompt_attachment_tmp_root() -> PathBuf {
    agentty_home().join("tmp")
}

/// Returns the shared managed image directory for the given attachment paths
/// when every path stays within the Agentty temp root.
fn managed_prompt_attachment_directory(
    attachment_paths: &[PathBuf],
    managed_tmp_root: &Path,
) -> Option<PathBuf> {
    let image_directory = attachment_paths.first()?.parent()?.to_path_buf();
    if !is_managed_prompt_attachment_directory(&image_directory, managed_tmp_root) {
        return None;
    }

    attachment_paths
        .iter()
        .all(|attachment_path| {
            attachment_path.parent() == Some(image_directory.as_path())
                && is_managed_prompt_attachment_path(attachment_path, managed_tmp_root)
        })
        .then_some(image_directory)
}

/// Returns whether one attachment path is owned by Agentty under the managed
/// prompt-image tmp root.
fn is_managed_prompt_attachment_path(path: &Path, managed_tmp_root: &Path) -> bool {
    path.parent().is_some_and(|parent| {
        is_managed_prompt_attachment_directory(parent, managed_tmp_root)
            && path.starts_with(managed_tmp_root)
    })
}

/// Returns whether one directory is an Agentty-managed prompt-image directory.
fn is_managed_prompt_attachment_directory(path: &Path, managed_tmp_root: &Path) -> bool {
    path.starts_with(managed_tmp_root) && path.ends_with("images")
}

#[cfg(test)]
mod test_support {
    use std::sync::Arc;

    use super::*;
    use crate::domain::agent::AgentModel;

    impl SessionManager {
        /// Submits a follow-up prompt using a pre-built backend for
        /// deterministic test execution.
        ///
        /// Creates a test CLI channel backed by the given
        /// [`agent::AgentBackend`] and registers it in the session-local
        /// channel map so the worker uses it instead of the default factory.
        /// This allows tests to control spawned process commands without
        /// relying on a real provider binary.
        pub(crate) async fn reply_with_backend(
            &mut self,
            services: &AppServices,
            session_id: &str,
            prompt: impl Into<TurnPrompt>,
            backend: Arc<dyn agent::AgentBackend>,
            session_model: AgentModel,
        ) {
            let prompt = prompt.into();
            let session_agent = self.session_or_err(session_id).map_or(
                AgentSelection::new(AgentKind::Antigravity, session_model),
                |session| session.agent,
            );
            let channel =
                ag_agent::create_cli_agent_channel_with_backend(backend, session_agent.kind());
            self.worker_service
                .test_agent_channels
                .insert(session_id.to_string().into(), channel);
            self.reply_impl(services, session_id, prompt, session_agent)
                .await;
        }
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::sync::Arc;

    use ag_forge as forge;
    use tokio::sync::mpsc;

    use super::*;
    use crate::app::session::SessionDefaults;
    use crate::app::{AppEvent, AppServices, SessionState};
    use crate::domain::agent::{AgentKind, AgentModel, ReasoningLevel};
    use crate::domain::selection::SelectionState;
    use crate::domain::session::{
        ForgeKind, ReviewRequestState, ReviewRequestSummary, SessionHandles,
    };
    use crate::domain::turn_prompt::{TurnPromptAttachment, TurnPromptTextSource};
    use crate::infra::clock::RealClock;
    use crate::infra::db::{self, AppRepositories};
    use crate::infra::fs;

    /// Builds a session manager with one session for reply-context tests.
    fn session_manager_with_one_session(session: Session) -> SessionManager {
        let mut handles = HashMap::new();
        handles.insert(
            session.id.clone(),
            SessionHandles::new_with_transcript(
                session.status,
                session.transcript.clone().unwrap_or_default(),
            ),
        );

        let state = SessionState::new(
            handles,
            vec![session],
            SelectionState::default(),
            Arc::new(RealClock),
            1,
            0,
        );

        SessionManager::new(
            SessionDefaults {
                model: AgentModel::Gpt55,
            },
            Arc::new(git::MockGitClient::new()),
            state,
            Vec::new(),
        )
    }

    /// Builds a minimal in-memory session snapshot for lifecycle unit tests.
    fn test_session(prompt: &str, status: Status, title: Option<&str>, output: &str) -> Session {
        crate::test_support::SessionFixtureBuilder::new()
            .agent(crate::domain::agent::AgentSelection::new(
                crate::domain::agent::AgentKind::Claude,
                AgentModel::ClaudeSonnet5,
            ))
            .folder(PathBuf::from("/tmp/session"))
            .transcript(output)
            .prompt(prompt)
            .status(status)
            .title(title.map(ToString::to_string))
            .build()
    }

    /// Builds a filesystem mock that delegates simple checks to local disk.
    fn create_passthrough_mock_fs_client() -> fs::MockFsClient {
        let mut mock_fs_client = fs::MockFsClient::new();
        mock_fs_client
            .expect_create_dir_all()
            .times(0..)
            .returning(|_| Box::pin(async { Ok(()) }));
        mock_fs_client
            .expect_remove_dir_all()
            .times(0..)
            .returning(|_| Box::pin(async { Ok(()) }));
        mock_fs_client
            .expect_read_file()
            .times(0..)
            .returning(|path| {
                Box::pin(async move { tokio::fs::read(path).await.map_err(fs::FsError::from) })
            });
        mock_fs_client
            .expect_remove_file()
            .times(0..)
            .returning(|_| Box::pin(async { Ok(()) }));
        mock_fs_client
            .expect_exists()
            .times(0..)
            .returning(|path| path.exists());
        mock_fs_client
            .expect_is_dir()
            .times(0..)
            .returning(|path| path.is_dir());

        mock_fs_client
    }

    /// Persists one session row that matches the in-memory fixture.
    async fn database_with_session(session: &Session) -> AppRepositories {
        let database = AppRepositories::in_memory().await;
        let project_id = database
            .projects()
            .upsert_project("/tmp/project", Some("main".to_string()))
            .await
            .expect("failed to upsert project");
        if session.is_draft {
            database
                .sessions()
                .insert_draft_session(
                    &session.id,
                    session.agent.model().as_str(),
                    &session.base_branch,
                    &session.status.to_string(),
                    project_id,
                )
                .await
                .expect("failed to insert draft session");
        } else {
            database
                .sessions()
                .insert_session(
                    &session.id,
                    session.agent.model().as_str(),
                    &session.base_branch,
                    &session.status.to_string(),
                    project_id,
                )
                .await
                .expect("failed to insert session");
        }
        database
            .sessions()
            .update_session_prompt(&session.id, &session.prompt)
            .await
            .expect("failed to persist session prompt");
        if let Some(title) = &session.title {
            database
                .sessions()
                .update_session_title(&session.id, title)
                .await
                .expect("failed to persist session title");
        }
        if let Some(review_request) = &session.review_request {
            database
                .reviews()
                .update_session_review_request(&session.id, Some(review_request.clone()))
                .await
                .expect("failed to persist session review request");
        }

        database
    }

    /// Builds app services with caller-provided filesystem, git, and forge
    /// boundaries.
    fn test_services_with_fs_client(
        database: &AppRepositories,
        fs_client: Arc<dyn fs::FsClient>,
        git_client: Arc<dyn git::GitClient>,
        review_request_client: Arc<dyn forge::ReviewRequestClient>,
    ) -> AppServices {
        let (event_tx, _event_rx) = mpsc::unbounded_channel();

        AppServices::new_with_agent_clis(
            PathBuf::from("/tmp/agentty-tests"),
            Arc::new(crate::infra::clock::RealClock),
            event_tx,
            crate::app::service::AppServiceDeps {
                app_server_client_override: Some(crate::test_support::mock_app_server()),
                available_agent_kinds: AgentKind::ALL.to_vec(),
                clipboard_image_client_override: None,
                fs_client,
                git_client,
                repositories: database.clone(),
                review_request_client,
            },
            crate::domain::agent::AgentCliInfo::from_kinds(AgentKind::ALL),
        )
    }

    /// Builds app services with caller-provided git and forge boundaries.
    fn test_services(
        database: &AppRepositories,
        git_client: Arc<dyn git::GitClient>,
        review_request_client: Arc<dyn forge::ReviewRequestClient>,
    ) -> AppServices {
        test_services_with_fs_client(
            database,
            Arc::new(create_passthrough_mock_fs_client()),
            git_client,
            review_request_client,
        )
    }

    /// Builds app services plus an event receiver for reducer-event
    /// assertions.
    fn test_services_with_event_receiver(
        database: &AppRepositories,
        git_client: Arc<dyn git::GitClient>,
        review_request_client: Arc<dyn forge::ReviewRequestClient>,
    ) -> (AppServices, mpsc::UnboundedReceiver<AppEvent>) {
        let (event_tx, event_rx) = mpsc::unbounded_channel();
        let services = AppServices::new_with_agent_clis(
            PathBuf::from("/tmp/agentty-tests"),
            Arc::new(crate::infra::clock::RealClock),
            event_tx,
            crate::app::service::AppServiceDeps {
                app_server_client_override: Some(crate::test_support::mock_app_server()),
                available_agent_kinds: AgentKind::ALL.to_vec(),
                clipboard_image_client_override: None,
                fs_client: Arc::new(create_passthrough_mock_fs_client()),
                git_client,
                repositories: database.clone(),
                review_request_client,
            },
            crate::domain::agent::AgentCliInfo::from_kinds(AgentKind::ALL),
        );

        (services, event_rx)
    }

    /// Builds one normalized review-request summary for workflow tests.
    fn review_request_summary(display_id: &str) -> ReviewRequestSummary {
        ReviewRequestSummary {
            display_id: display_id.to_string(),
            forge_kind: ForgeKind::GitHub,
            source_branch: session_branch("session-id"),
            state: ReviewRequestState::Open,
            status_summary: Some("Checks pending".to_string()),
            target_branch: "main".to_string(),
            title: "Add forge review support".to_string(),
            web_url: format!(
                "https://github.com/agentty-xyz/agentty/pull/{}",
                &display_id[1..]
            ),
        }
    }

    /// Returns one GitHub forge-remote fixture for review-request tests.
    fn github_remote() -> forge::ForgeRemote {
        forge::ForgeRemote {
            command_working_directory: Some(PathBuf::from("/tmp/session")),
            forge_kind: ForgeKind::GitHub,
            host: "github.com".to_string(),
            namespace: "agentty-xyz".to_string(),
            project: "agentty".to_string(),
            repo_url: "https://github.com/agentty-xyz/agentty.git".to_string(),
            web_url: "https://github.com/agentty-xyz/agentty".to_string(),
        }
    }

    /// Returns the expected create payload for one session review request.
    fn expected_create_input() -> forge::CreateReviewRequestInput {
        forge::CreateReviewRequestInput {
            body: Some("- Keep title in sync".to_string()),
            source_branch: session_branch("session-id"),
            target_branch: "main".to_string(),
            title: "Refine session commit message".to_string(),
        }
    }
    /// Configures git expectations for review-request publication.
    fn expect_published_session_branch(mock_git_client: &mut git::MockGitClient) {
        mock_git_client
            .expect_push_current_branch()
            .times(1)
            .returning(|_| Box::pin(async { Ok("origin/wt/session-id".to_string()) }));
        mock_git_client.expect_repo_url().times(1).returning(|_| {
            Box::pin(async { Ok("https://github.com/agentty-xyz/agentty.git".to_string()) })
        });
    }

    /// Loads the persisted session row used by workflow assertions.
    async fn load_persisted_session_row(database: &AppRepositories) -> db::SessionRow {
        database
            .sessions()
            .load_sessions()
            .await
            .expect("failed to load session rows")
            .into_iter()
            .find(|row| row.id == "session-id")
            .expect("session row should exist")
    }

    #[tokio::test]
    /// Ensures `set_session_reasoning_level()` persists the level and
    /// emits the matching reducer event.
    async fn test_set_session_reasoning_level_persists_level_and_emits_event() {
        // Arrange
        let session = test_session("Prompt", Status::Review, Some("Title"), "");
        let database = database_with_session(&session).await;
        let mut session_manager = session_manager_with_one_session(session);
        let (services, mut event_rx) = test_services_with_event_receiver(
            &database,
            Arc::new(git::MockGitClient::new()),
            Arc::new(forge::MockReviewRequestClient::new()),
        );

        // Act
        session_manager
            .set_session_reasoning_level(&services, "session-id", ReasoningLevel::High)
            .await
            .expect("reasoning level update should succeed");
        let persisted_reasoning_level = database
            .sessions()
            .load_session_reasoning_level("session-id")
            .await
            .expect("reasoning level should load");
        let emitted_event = event_rx
            .try_recv()
            .expect("expected reasoning update event");

        // Assert
        assert_eq!(persisted_reasoning_level, ReasoningLevel::High);
        assert_eq!(
            emitted_event,
            AppEvent::SessionReasoningLevelUpdated {
                reasoning_level: ReasoningLevel::High,
                session_id: "session-id".into(),
            }
        );
        assert!(event_rx.try_recv().is_err());
    }

    #[tokio::test]
    /// Verifies `enqueue_message()` emits a single targeted
    /// [`AppEvent::SessionUpdated`] for the touched session and never falls
    /// back to [`AppEvent::RefreshSessions`]. The targeted event lets the
    /// reducer re-sync only the affected snapshot from handles instead of
    /// paying for a full DB-backed reload, which is the contract that makes
    /// queued chat rows appear without a perceptible delay.
    async fn test_enqueue_message_emits_session_updated_event_only() {
        // Arrange
        let session = test_session("Prompt", Status::InProgress, Some("Title"), "");
        let database = database_with_session(&session).await;
        let mut session_manager = session_manager_with_one_session(session);
        let (services, mut event_rx) = test_services_with_event_receiver(
            &database,
            Arc::new(git::MockGitClient::new()),
            Arc::new(forge::MockReviewRequestClient::new()),
        );

        // Act
        session_manager
            .enqueue_message(&services, "session-id", "queued reply")
            .expect("enqueue_message should succeed for InProgress session");

        // Assert
        let emitted_event = event_rx
            .try_recv()
            .expect("expected SessionUpdated event from enqueue_message");
        assert!(
            matches!(
                &emitted_event,
                AppEvent::SessionUpdated { session_id, .. }
                    if AsRef::<str>::as_ref(session_id) == "session-id"
            ),
            "enqueue_message must emit SessionUpdated, got {emitted_event:?}"
        );
        assert!(
            event_rx.try_recv().is_err(),
            "enqueue_message must not emit additional events (especially not RefreshSessions) so \
             the reducer skips the full DB-backed reload"
        );
    }

    #[tokio::test]
    async fn test_publish_review_request_creates_and_persists_link_when_lookup_misses() {
        // Arrange
        let session = test_session(
            "Implement forge review support",
            Status::Review,
            Some("Add forge review support"),
            "",
        );
        let database = database_with_session(&session).await;
        let mut session_manager = session_manager_with_one_session(session);
        let source_branch = session_branch("session-id");
        let expected_create_input = expected_create_input();
        let remote = github_remote();
        let created_summary = review_request_summary("#42");
        let mut mock_git_client = git::MockGitClient::new();
        expect_published_session_branch(&mut mock_git_client);
        mock_git_client
            .expect_head_commit_message()
            .times(1)
            .returning(|_| {
                Box::pin(async {
                    Ok(Some(
                        "Refine session commit message\n\n- Keep title in sync".to_string(),
                    ))
                })
            });
        let mut mock_review_request_client = forge::MockReviewRequestClient::new();
        mock_review_request_client
            .expect_detect_remote()
            .times(1)
            .returning({
                let remote = remote.clone();
                move |_| Ok(remote.clone())
            });
        mock_review_request_client
            .expect_find_by_source_branch()
            .times(1)
            .withf({
                let remote = remote.clone();
                let source_branch = source_branch.clone();
                move |candidate_remote, candidate_source_branch| {
                    candidate_remote == &remote && candidate_source_branch == &source_branch
                }
            })
            .returning(|_, _| Box::pin(async { Ok(None) }));
        mock_review_request_client
            .expect_create_review_request()
            .times(1)
            .withf({
                let remote = remote.clone();
                let expected_create_input = expected_create_input.clone();
                move |candidate_remote, candidate_input| {
                    candidate_remote == &remote && candidate_input == &expected_create_input
                }
            })
            .returning(move |_, _| {
                let created_summary = created_summary.clone();

                Box::pin(async move { Ok(created_summary) })
            });
        let services = test_services(
            &database,
            Arc::new(mock_git_client),
            Arc::new(mock_review_request_client),
        );

        // Act
        let review_request = session_manager
            .publish_review_request(&services, "session-id")
            .await
            .expect("review request should be created");
        let persisted_row = load_persisted_session_row(&database).await;

        // Assert
        assert_eq!(review_request.summary.display_id, "#42");
        assert_eq!(
            session_manager.state.sessions[0].review_request,
            Some(review_request.clone())
        );
        assert_eq!(
            persisted_row
                .review_request
                .as_ref()
                .map(|row| row.display_id.as_str()),
            Some("#42")
        );
        assert_eq!(
            persisted_row
                .review_request
                .as_ref()
                .map(|row| row.last_refreshed_at),
            Some(review_request.last_refreshed_at)
        );
        assert_eq!(
            persisted_row.published_upstream_ref.as_deref(),
            Some("origin/wt/session-id")
        );
        assert_eq!(
            session_manager.state.sessions[0]
                .published_upstream_ref
                .as_deref(),
            Some("origin/wt/session-id")
        );
    }

    #[tokio::test]
    async fn test_stage_draft_message_preserves_persisted_prompt_when_attachment_write_fails() {
        // Arrange
        let mut session = test_session("", Status::Draft, None, "");
        session.is_draft = true;
        let database = database_with_session(&session).await;
        let mut session_manager = session_manager_with_one_session(session);
        let mut mock_fs_client = fs::MockFsClient::new();
        mock_fs_client
            .expect_create_dir_all()
            .times(0..)
            .returning(|_| Box::pin(async { Ok(()) }));
        mock_fs_client
            .expect_remove_dir_all()
            .times(0..)
            .returning(|_| Box::pin(async { Ok(()) }));
        mock_fs_client
            .expect_read_file()
            .times(0..)
            .returning(|_| Box::pin(async { Ok(Vec::new()) }));
        mock_fs_client
            .expect_remove_file()
            .times(0..)
            .returning(|_| Box::pin(async { Ok(()) }));
        mock_fs_client
            .expect_exists()
            .times(0..)
            .returning(|path| path.exists());
        mock_fs_client
            .expect_is_dir()
            .times(0..)
            .returning(|path| path.is_dir());
        mock_fs_client.expect_write_file().once().returning(|_, _| {
            Box::pin(async {
                Err(fs::FsError::Io(std::io::Error::other(
                    "simulated attachment write failure",
                )))
            })
        });
        let services = test_services_with_fs_client(
            &database,
            Arc::new(mock_fs_client),
            Arc::new(git::MockGitClient::new()),
            Arc::new(forge::MockReviewRequestClient::new()),
        );
        let prompt = TurnPrompt {
            attachments: vec![TurnPromptAttachment {
                placeholder: "[Image #1]".to_string(),
                local_image_path: PathBuf::from("/tmp/image-1.png"),
            }],
            text: "Review [Image #1]".to_string(),
            text_source: TurnPromptTextSource::UserPrompt,
        };

        // Act
        let error = session_manager
            .stage_draft_message(&services, "session-id", prompt)
            .await
            .expect_err("attachment metadata failure should abort draft staging");
        let persisted_session = load_persisted_session_row(&database).await;

        // Assert
        assert!(matches!(error, SessionError::Fs(_)));
        assert!(persisted_session.prompt.is_empty());
        assert!(session_manager.sessions()[0].prompt.is_empty());
        assert!(session_manager.sessions()[0].draft_attachments.is_empty());
    }

    #[tokio::test]
    async fn test_ensure_session_worktree_ready_skips_non_draft_sessions() {
        // Arrange
        let session = test_session("", Status::Draft, None, "");
        let database = database_with_session(&session).await;
        let mut session_manager = session_manager_with_one_session(session);
        let mut mock_fs_client = fs::MockFsClient::new();
        mock_fs_client.expect_is_dir().times(0);
        let mut mock_git_client = git::MockGitClient::new();
        mock_git_client.expect_create_worktree().times(0);
        mock_git_client.expect_find_git_repo_root().times(0);
        let services = test_services_with_fs_client(
            &database,
            Arc::new(mock_fs_client),
            Arc::new(mock_git_client),
            Arc::new(forge::MockReviewRequestClient::new()),
        );

        // Act
        let result = session_manager
            .ensure_session_worktree_ready(&services, "session-id")
            .await;

        // Assert
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_ensure_session_worktree_ready_reuses_existing_draft_worktree() {
        // Arrange
        let mut session = test_session("", Status::Draft, None, "");
        session.is_draft = true;
        let database = database_with_session(&session).await;
        let mut session_manager = session_manager_with_one_session(session);
        let mut mock_fs_client = fs::MockFsClient::new();
        mock_fs_client.expect_is_dir().times(3).return_const(true);
        mock_fs_client
            .expect_canonicalize()
            .times(2)
            .returning(|path| {
                Box::pin(async move {
                    if path == Path::new("/tmp/project") {
                        Ok(PathBuf::from("/tmp/project"))
                    } else {
                        Ok(PathBuf::from("/tmp/session"))
                    }
                })
            });
        let mut mock_git_client = git::MockGitClient::new();
        mock_git_client
            .expect_detect_git_info()
            .once()
            .returning(|_| Box::pin(async { Some("wt/session-".to_string()) }));
        mock_git_client
            .expect_main_checkout_working_tree()
            .once()
            .returning(|_| Box::pin(async { Ok(Some(PathBuf::from("/tmp/project"))) }));
        mock_git_client.expect_create_worktree().times(0);
        mock_git_client.expect_find_git_repo_root().times(0);
        let services = test_services_with_fs_client(
            &database,
            Arc::new(mock_fs_client),
            Arc::new(mock_git_client),
            Arc::new(forge::MockReviewRequestClient::new()),
        );

        // Act
        let result = session_manager
            .ensure_session_worktree_ready(&services, "session-id")
            .await;

        // Assert
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_create_session_worktree_uses_local_base_branch_ref() {
        // Arrange
        let session = test_session("", Status::Draft, None, "");
        let database = database_with_session(&session).await;
        let session_manager = session_manager_with_one_session(session);
        let repo_root = PathBuf::from("/tmp/project");
        let folder = PathBuf::from("/tmp/session-worktree");
        let expected_repo_root = repo_root.clone();
        let expected_folder = folder.clone();
        let mut mock_git_client = git::MockGitClient::new();
        mock_git_client
            .expect_create_worktree()
            .once()
            .withf(
                move |candidate_repo_root, candidate_folder, worktree_branch, start_ref| {
                    candidate_repo_root == &expected_repo_root
                        && candidate_folder == &expected_folder
                        && worktree_branch == "wt/session-id"
                        && start_ref == "main"
                },
            )
            .returning(|_, _, _, _| Box::pin(async { Ok(()) }));
        let services = test_services_with_fs_client(
            &database,
            Arc::new(create_passthrough_mock_fs_client()),
            Arc::new(mock_git_client),
            Arc::new(forge::MockReviewRequestClient::new()),
        );

        // Act
        let result = session_manager
            .create_session_worktree(
                &services,
                "session-id",
                folder.as_path(),
                repo_root.as_path(),
                "wt/session-id",
                "main",
            )
            .await;

        // Assert
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_cleanup_session_worktree_resources_collects_cleanup_errors() {
        // Arrange
        let mut mock_fs_client = fs::MockFsClient::new();
        mock_fs_client
            .expect_remove_dir_all()
            .once()
            .returning(|_| {
                Box::pin(async {
                    Err(fs::FsError::Io(std::io::Error::other(
                        "simulated directory cleanup failure",
                    )))
                })
            });
        let mut mock_git_client = git::MockGitClient::new();
        mock_git_client
            .expect_remove_worktree()
            .once()
            .returning(|_| {
                Box::pin(async {
                    Err(git::GitError::CommandFailed {
                        command: "git worktree remove".to_string(),
                        stderr: "simulated worktree removal failure".to_string(),
                    })
                })
            });
        mock_git_client
            .expect_delete_branch()
            .once()
            .returning(|_, _| {
                Box::pin(async {
                    Err(git::GitError::CommandFailed {
                        command: "git branch -D".to_string(),
                        stderr: "simulated branch deletion failure".to_string(),
                    })
                })
            });

        // Act
        let cleanup_errors = SessionManager::cleanup_session_worktree_resources(
            Arc::new(mock_fs_client),
            Arc::new(mock_git_client),
            PathBuf::from("/tmp/session"),
            "wt/session-id".to_string(),
            Some(PathBuf::from("/tmp/repo")),
            true,
        )
        .await;

        // Assert
        assert_eq!(cleanup_errors.len(), 3);
        assert!(
            cleanup_errors
                .iter()
                .any(|message| message.contains("failed to remove worktree"))
        );
        assert!(
            cleanup_errors
                .iter()
                .any(|message| message.contains("failed to delete branch"))
        );
        assert!(
            cleanup_errors
                .iter()
                .any(|message| message.contains("failed to remove worktree directory"))
        );
    }

    #[tokio::test]
    async fn test_publish_review_request_reuses_existing_remote_link_before_create() {
        // Arrange
        let session = test_session(
            "Implement forge review support",
            Status::Review,
            Some("Add forge review support"),
            "",
        );
        let database = database_with_session(&session).await;
        let mut session_manager = session_manager_with_one_session(session);
        let source_branch = session_branch("session-id");
        let remote = github_remote();
        let existing_summary = review_request_summary("#24");
        let mut mock_git_client = git::MockGitClient::new();
        expect_published_session_branch(&mut mock_git_client);
        let mut mock_review_request_client = forge::MockReviewRequestClient::new();
        mock_review_request_client
            .expect_detect_remote()
            .times(1)
            .returning({
                let remote = remote.clone();
                move |_| Ok(remote.clone())
            });
        mock_review_request_client
            .expect_find_by_source_branch()
            .times(1)
            .withf({
                let remote = remote.clone();
                let source_branch = source_branch.clone();
                move |candidate_remote, candidate_source_branch| {
                    candidate_remote == &remote && candidate_source_branch == &source_branch
                }
            })
            .returning(move |_, _| {
                let existing_summary = existing_summary.clone();

                Box::pin(async move { Ok(Some(existing_summary)) })
            });
        mock_review_request_client
            .expect_refresh_review_request()
            .times(1)
            .withf({
                let remote = remote.clone();
                move |candidate_remote, display_id| {
                    candidate_remote == &remote && display_id == "#24"
                }
            })
            .returning(|_, _| Box::pin(async { Ok(review_request_summary("#24")) }));
        mock_review_request_client
            .expect_create_review_request()
            .times(0);
        let services = test_services(
            &database,
            Arc::new(mock_git_client),
            Arc::new(mock_review_request_client),
        );

        // Act
        let review_request = session_manager
            .publish_review_request(&services, "session-id")
            .await
            .expect("existing remote review request should be reused");

        // Assert
        assert_eq!(review_request.summary.display_id, "#24");
        assert_eq!(
            session_manager.state.sessions[0]
                .review_request
                .as_ref()
                .map(|review_request| review_request.summary.display_id.as_str()),
            Some("#24")
        );
    }

    #[tokio::test]
    async fn test_publish_review_request_refreshes_stored_link_after_push() {
        // Arrange
        let mut session = test_session(
            "Implement forge review support",
            Status::Review,
            Some("Add forge review support"),
            "",
        );
        session.review_request = Some(ReviewRequest {
            last_refreshed_at: 42,
            summary: review_request_summary("#11"),
        });
        let database = database_with_session(&session).await;
        let mut session_manager = session_manager_with_one_session(session);
        let remote = github_remote();
        let refreshed_summary = review_request_summary("#11");
        let mut mock_git_client = git::MockGitClient::new();
        expect_published_session_branch(&mut mock_git_client);
        let mut mock_review_request_client = forge::MockReviewRequestClient::new();
        mock_review_request_client
            .expect_detect_remote()
            .times(1)
            .returning({
                let remote = remote.clone();
                move |_| Ok(remote.clone())
            });
        mock_review_request_client
            .expect_refresh_review_request()
            .times(1)
            .withf({
                let remote = remote.clone();
                move |candidate_remote, display_id| {
                    candidate_remote == &remote && display_id == "#11"
                }
            })
            .returning(move |_, _| {
                let refreshed_summary = refreshed_summary.clone();

                Box::pin(async move { Ok(refreshed_summary) })
            });
        let services = test_services(
            &database,
            Arc::new(mock_git_client),
            Arc::new(mock_review_request_client),
        );

        // Act
        let review_request = session_manager
            .publish_review_request(&services, "session-id")
            .await
            .expect("stored review request should be refreshed");
        let persisted_row = load_persisted_session_row(&database).await;

        // Assert
        assert_eq!(review_request.summary.display_id, "#11");
        assert!(review_request.last_refreshed_at >= 42);
        assert_eq!(
            session_manager.state.sessions[0]
                .published_upstream_ref
                .as_deref(),
            Some("origin/wt/session-id")
        );
        assert_eq!(
            persisted_row
                .review_request
                .as_ref()
                .map(|row| row.display_id.as_str()),
            Some("#11")
        );
    }

    #[tokio::test]
    async fn test_review_request_web_url_returns_linked_review_request_url() {
        // Arrange
        let mut session = test_session(
            "Implement forge review support",
            Status::Done,
            Some("Add forge review support"),
            "",
        );
        session.review_request = Some(ReviewRequest {
            last_refreshed_at: 42,
            summary: review_request_summary("#11"),
        });
        let session_manager = session_manager_with_one_session(session);
        let database = database_with_session(
            session_manager
                .state
                .sessions
                .first()
                .expect("fixture session should exist"),
        )
        .await;
        let mut mock_review_request_client = forge::MockReviewRequestClient::new();
        mock_review_request_client
            .expect_review_request_web_url()
            .times(1)
            .returning(|summary| Ok(summary.web_url.clone()));
        let services = test_services(
            &database,
            Arc::new(git::MockGitClient::new()),
            Arc::new(mock_review_request_client),
        );

        // Act
        let review_request_url = session_manager
            .review_request_web_url(&services, "session-id")
            .expect("linked review request URL should be returned");

        // Assert
        assert_eq!(
            review_request_url,
            "https://github.com/agentty-xyz/agentty/pull/11"
        );
    }

    #[test]
    fn test_formatted_prompt_output_formats_multiline_prompt_with_continuation_prefix() {
        // Arrange
        let prompt = TurnPrompt::from_text("first line\n\n\nafter gap".to_string());

        // Act
        let formatted_prompt = SessionManager::formatted_prompt_output(&prompt, false);

        // Assert
        assert_eq!(
            formatted_prompt,
            " › first line\n   \n   \n   after gap\n\n"
        );
    }

    #[test]
    fn test_formatted_prompt_output_prepends_newline_for_replies() {
        // Arrange
        let prompt = TurnPrompt::from_text("reply line".to_string());

        // Act
        let formatted_prompt = SessionManager::formatted_prompt_output(&prompt, true);

        // Assert
        assert_eq!(formatted_prompt, "\n › reply line\n\n");
    }

    #[test]
    /// Ensures transcript formatting keeps prompt image markers visible.
    fn test_formatted_prompt_output_preserves_image_placeholders_in_transcript() {
        // Arrange
        let prompt = TurnPrompt {
            attachments: vec![TurnPromptAttachment {
                placeholder: "[Image #1]".to_string(),
                local_image_path: PathBuf::from("/tmp/image-1.png"),
            }],
            text: "Review [Image #1]".to_string(),
            text_source: TurnPromptTextSource::UserPrompt,
        };

        // Act
        let formatted_prompt = SessionManager::formatted_prompt_output(&prompt, false);

        // Assert
        assert_eq!(formatted_prompt, " › Review [Image #1]\n\n");
    }

    #[test]
    fn test_renumbered_prompt_text_rewrites_only_attachment_occurrences() {
        // Arrange
        let prompt = TurnPrompt {
            attachments: vec![TurnPromptAttachment {
                placeholder: "[Image #1]".to_string(),
                local_image_path: PathBuf::from("/tmp/image-1.png"),
            }],
            text: "Attach [Image #1] but keep literal [Image #1] text".to_string(),
            text_source: TurnPromptTextSource::UserPrompt,
        };

        // Act
        let renumbered_prompt = SessionManager::renumbered_prompt_text(&prompt, 2);

        // Assert
        assert_eq!(
            renumbered_prompt,
            "Attach [Image #2] but keep literal [Image #1] text"
        );
    }

    #[tokio::test]
    /// Ensures prompt attachment cleanup removes temp files and their image
    /// directory after handoff.
    async fn test_cleanup_prompt_attachment_paths_removes_files_and_directory() {
        // Arrange
        let temp_dir = tempfile::tempdir().expect("temp dir should exist");
        let managed_tmp_root = temp_dir.path().join("tmp");
        let image_directory = managed_tmp_root.join("session-id").join("images");
        std::fs::create_dir_all(&image_directory).expect("image directory should exist");
        let first_image = image_directory.join("image-1.png");
        let second_image = image_directory.join("image-2.png");
        std::fs::write(&first_image, b"png").expect("first image should exist");
        std::fs::write(&second_image, b"png").expect("second image should exist");

        // Act
        SessionManager::cleanup_prompt_attachment_paths_in_root(
            Arc::new(fs::RealFsClient),
            &managed_tmp_root,
            vec![first_image.clone(), second_image.clone()],
        )
        .await;

        // Assert
        assert!(!first_image.exists());
        assert!(!second_image.exists());
        assert!(!image_directory.exists());
    }

    #[tokio::test]
    /// Ensures cleanup of one prompt's attachments preserves sibling files
    /// owned by other queued prompts in the same shared image directory and
    /// keeps the directory in place when it is still non-empty.
    async fn test_cleanup_prompt_attachment_paths_preserves_sibling_files_in_shared_directory() {
        // Arrange — two managed image files share one session image directory,
        // mirroring two queued prompts with image attachments under the same
        // `AGENTTY_ROOT/tmp/<session-id>/images/` root.
        let temp_dir = tempfile::tempdir().expect("temp dir should exist");
        let managed_tmp_root = temp_dir.path().join("tmp");
        let image_directory = managed_tmp_root.join("session-id").join("images");
        std::fs::create_dir_all(&image_directory).expect("image directory should exist");
        let popped_image = image_directory.join("image-1.png");
        let sibling_image = image_directory.join("image-2.png");
        std::fs::write(&popped_image, b"png").expect("popped image should exist");
        std::fs::write(&sibling_image, b"png").expect("sibling image should exist");

        // Act — clean up only the popped prompt's attachment.
        SessionManager::cleanup_prompt_attachment_paths_in_root(
            Arc::new(fs::RealFsClient),
            &managed_tmp_root,
            vec![popped_image.clone()],
        )
        .await;

        // Assert — popped image is gone, sibling image survives, and the
        // shared directory is preserved because it is still non-empty.
        assert!(
            !popped_image.exists(),
            "popped attachment file should be removed"
        );
        assert!(
            sibling_image.exists(),
            "sibling attachment file from another queued prompt must survive cleanup"
        );
        assert!(
            image_directory.exists(),
            "shared image directory must be preserved while sibling attachments remain"
        );
    }

    #[tokio::test]
    /// Ensures cleanup ignores attachment paths outside the managed Agentty
    /// temp root.
    async fn test_cleanup_prompt_attachment_paths_leaves_unmanaged_files_untouched() {
        // Arrange
        let temp_dir = tempfile::tempdir().expect("temp dir should exist");
        let managed_tmp_root = temp_dir.path().join("tmp");
        let image_directory = temp_dir.path().join("user-images");
        std::fs::create_dir_all(&image_directory).expect("image directory should exist");
        let image_path = image_directory.join("image-1.png");
        std::fs::write(&image_path, b"png").expect("image file should exist");

        // Act
        SessionManager::cleanup_prompt_attachment_paths_in_root(
            Arc::new(fs::RealFsClient),
            &managed_tmp_root,
            vec![image_path.clone()],
        )
        .await;

        // Assert
        assert!(image_path.exists());
        assert!(image_directory.exists());
    }

    #[test]
    /// Ensures first replies persist the full prompt as the one-time title.
    fn test_prepare_reply_context_first_message_sets_title_from_prompt() {
        // Arrange
        let prompt = "Implement optimistic retry path";
        let turn_prompt = TurnPrompt::from_text(prompt.to_string());
        let session = test_session("", Status::Draft, None, "");
        let mut session_manager = session_manager_with_one_session(session);

        // Act
        let context = session_manager
            .prepare_reply_context(0, &turn_prompt, false)
            .expect("reply context should be available")
            .expect("session should produce reply context");

        // Assert
        assert_eq!(context.0, None);
        assert!(context.1);
        assert_eq!(context.2, "session-id");
        assert_eq!(context.3, Some(prompt.to_string()));
        assert_eq!(session_manager.sessions()[0].prompt, prompt);
        assert_eq!(
            session_manager.sessions()[0].title,
            Some(prompt.to_string())
        );
    }

    #[test]
    /// Ensures follow-up replies keep the existing title unchanged.
    fn test_prepare_reply_context_follow_up_keeps_existing_title() {
        // Arrange
        let session = test_session(
            "Initial prompt",
            Status::Review,
            Some("Initial prompt"),
            "existing output",
        );
        let mut session_manager = session_manager_with_one_session(session);
        let prompt = TurnPrompt::from_text("Follow-up prompt".to_string());

        // Act
        let context = session_manager
            .prepare_reply_context(0, &prompt, false)
            .expect("reply context should be available")
            .expect("session should produce reply context");

        // Assert
        assert_eq!(context.0, None);
        assert!(!context.1);
        assert_eq!(context.2, "session-id");
        assert_eq!(context.3, None);
        assert_eq!(session_manager.sessions()[0].prompt, "Initial prompt");
        assert_eq!(
            session_manager.sessions()[0].title,
            Some("Initial prompt".to_string())
        );
    }

    #[test]
    /// Ensures replying to an in-progress session returns a typed
    /// [`SessionError::Workflow`] instead of a raw string.
    fn test_prepare_reply_context_returns_workflow_error_when_status_blocks_reply() {
        // Arrange
        let session = test_session("Initial prompt", Status::InProgress, Some("Title"), "");
        let mut session_manager = session_manager_with_one_session(session);
        let prompt = TurnPrompt::from_text("Another prompt".to_string());

        // Act
        let result = session_manager.prepare_reply_context(0, &prompt, false);

        // Assert
        let error = result.expect_err("in-progress session should block reply");
        assert!(
            matches!(error, SessionError::Workflow(_)),
            "expected SessionError::Workflow, got: {error:?}"
        );
    }

    #[test]
    /// Ensures title-generation prompt rendering includes session request text.
    fn test_session_title_generation_prompt_includes_request() {
        // Arrange
        let request_prompt = "Refactor session lifecycle updates";

        // Act
        let title_prompt = SessionManager::session_title_generation_prompt(request_prompt)
            .expect("title generation prompt should render");

        // Assert
        assert!(title_prompt.contains("Generate a concise, commit-style title"));
        assert!(title_prompt.contains("Describe what the user wants to do"));
        assert!(title_prompt.contains("Keep it high-level and intent-focused."));
        assert!(title_prompt.contains("Do not include long file names"));
        assert!(title_prompt.contains("Do not describe your own progress"));
        assert!(title_prompt.contains("Do not use first-person phrasing"));
        assert!(title_prompt.contains("Put only the title text in `answer`"));
        assert!(!title_prompt.contains("Return only the title text."));
        assert!(title_prompt.contains(request_prompt));
    }

    #[test]
    /// Ensures single-line title responses are normalized and accepted.
    fn test_parse_generated_session_title_accepts_plain_title() {
        // Arrange
        let response_content = "Refine session startup flow";

        // Act
        let parsed_title = SessionManager::parse_generated_session_title(response_content);

        // Assert
        assert_eq!(
            parsed_title,
            Some("Refine session startup flow".to_string())
        );
    }

    #[test]
    /// Ensures protocol-wrapped plain answer lines are accepted.
    fn test_parse_generated_session_title_accepts_protocol_answer_plain_text() {
        // Arrange
        let response_content = r#"{"answer":"Polish title parsing","questions":[],"summary":null}"#;

        // Act
        let parsed_title = SessionManager::parse_generated_session_title(response_content);

        // Assert
        assert_eq!(parsed_title, Some("Polish title parsing".to_string()));
    }

    #[test]
    /// Ensures plain-text responses with extra lines keep only the first
    /// non-empty title line.
    fn test_parse_generated_session_title_uses_first_nonempty_line_for_multiline_response() {
        // Arrange
        let response_content = "Polish title parsing\nExtra detail that should be ignored";

        // Act
        let parsed_title = SessionManager::parse_generated_session_title(response_content);

        // Assert
        assert_eq!(parsed_title, Some("Polish title parsing".to_string()));
    }

    #[test]
    /// Ensures protocol payloads without `answer` text do not update
    /// titles.
    fn test_parse_generated_session_title_returns_none_for_question_only_protocol_payload() {
        // Arrange
        let response_content = r#"{"answer":"","questions":[{"text":"Need confirmation?","options":[]}],"summary":null}"#;

        // Act
        let parsed_title = SessionManager::parse_generated_session_title(response_content);

        // Assert
        assert_eq!(parsed_title, None);
    }

    #[test]
    /// Ensures `Title:` prefixes are normalized before persistence.
    fn test_parse_generated_session_title_normalizes_title_prefix() {
        // Arrange
        let response_content = "Title: \"Polish merge queue behavior\"";

        // Act
        let parsed_title = SessionManager::parse_generated_session_title(response_content);

        // Assert
        assert_eq!(
            parsed_title,
            Some("Polish merge queue behavior".to_string())
        );
    }

    #[test]
    /// Ensures first-person progress output cannot overwrite fallback
    /// titles.
    fn test_parse_generated_session_title_rejects_first_person_progress_output() {
        // Arrange
        let response_content = r#"{"answer":"I am checking the exact commit-message constraints.","questions":[],"summary":null}"#;

        // Act
        let parsed_title = SessionManager::parse_generated_session_title(response_content);

        // Assert
        assert_eq!(parsed_title, None);
    }

    #[test]
    /// Ensures progress-gerund output is rejected as status prose.
    fn test_parse_generated_session_title_rejects_progress_prefix() {
        // Arrange
        let response_content = "Checking commit-message constraints";

        // Act
        let parsed_title = SessionManager::parse_generated_session_title(response_content);

        // Assert
        assert_eq!(parsed_title, None);
    }

    #[test]
    /// Ensures overlong model prose is rejected instead of being truncated
    /// into a misleading generated title.
    fn test_parse_generated_session_title_rejects_overlong_candidate() {
        // Arrange
        let response_content =
            "Refine session title generation for utility outputs that are unexpectedly verbose";

        // Act
        let parsed_title = SessionManager::parse_generated_session_title(response_content);

        // Assert
        assert_eq!(parsed_title, None);
    }

    /// Builds a session manager containing the supplied sessions with no
    /// pre-selected row.
    fn session_manager_with_sessions(sessions: Vec<Session>) -> SessionManager {
        let mut handles = HashMap::new();
        for session in &sessions {
            handles.insert(
                session.id.clone(),
                SessionHandles::new_with_transcript(
                    session.status,
                    session.transcript.clone().unwrap_or_default(),
                ),
            );
        }
        let row_count = i64::try_from(sessions.len()).unwrap_or(0);
        let state = SessionState::new(
            handles,
            sessions,
            SelectionState::default(),
            Arc::new(RealClock),
            row_count,
            0,
        );

        SessionManager::new(
            SessionDefaults {
                model: AgentModel::Gpt55,
            },
            Arc::new(git::MockGitClient::new()),
            state,
            Vec::new(),
        )
    }

    /// Returns one session with a custom identifier and status for navigation
    /// tests.
    fn session_with_id(id: &str, status: Status) -> Session {
        let mut session = test_session("prompt", status, None, "");
        session.id = id.to_string().into();

        session
    }

    #[test]
    fn next_starts_at_first_selectable_row_when_no_prior_selection() {
        // Arrange
        let mut session_manager = session_manager_with_sessions(vec![
            session_with_id("session-active", Status::InProgress),
            session_with_id("session-archive", Status::Done),
        ]);

        // Act
        session_manager.next();

        // Assert
        assert_eq!(session_manager.state.table_state.selected(), Some(0));
    }

    #[test]
    fn next_advances_selection_to_next_grouped_row() {
        // Arrange
        let mut session_manager = session_manager_with_sessions(vec![
            session_with_id("session-active-1", Status::InProgress),
            session_with_id("session-active-2", Status::Review),
        ]);
        session_manager.state.table_state.select(Some(0));

        // Act
        session_manager.next();

        // Assert
        assert_eq!(session_manager.state.table_state.selected(), Some(1));
    }

    #[test]
    fn next_wraps_to_first_selectable_row_after_last_row() {
        // Arrange
        let mut session_manager = session_manager_with_sessions(vec![
            session_with_id("session-active", Status::InProgress),
            session_with_id("session-archive", Status::Done),
        ]);
        session_manager.state.table_state.select(Some(1));

        // Act
        session_manager.next();

        // Assert
        assert_eq!(session_manager.state.table_state.selected(), Some(0));
    }

    #[test]
    fn next_is_no_op_when_no_sessions_present() {
        // Arrange
        let mut session_manager = session_manager_with_sessions(Vec::new());

        // Act
        session_manager.next();

        // Assert
        assert_eq!(session_manager.state.table_state.selected(), None);
    }

    #[test]
    fn previous_starts_at_first_selectable_row_when_no_prior_selection() {
        // Arrange
        let mut session_manager = session_manager_with_sessions(vec![
            session_with_id("session-active", Status::InProgress),
            session_with_id("session-archive", Status::Done),
        ]);

        // Act
        session_manager.previous();

        // Assert
        assert_eq!(session_manager.state.table_state.selected(), Some(0));
    }

    #[test]
    fn previous_moves_selection_back_one_grouped_row() {
        // Arrange
        let mut session_manager = session_manager_with_sessions(vec![
            session_with_id("session-active-1", Status::InProgress),
            session_with_id("session-active-2", Status::Review),
        ]);
        session_manager.state.table_state.select(Some(1));

        // Act
        session_manager.previous();

        // Assert
        assert_eq!(session_manager.state.table_state.selected(), Some(0));
    }

    #[test]
    fn previous_wraps_to_last_selectable_row_when_at_first_row() {
        // Arrange
        let mut session_manager = session_manager_with_sessions(vec![
            session_with_id("session-active", Status::InProgress),
            session_with_id("session-archive", Status::Done),
        ]);
        session_manager.state.table_state.select(Some(0));

        // Act
        session_manager.previous();

        // Assert
        assert_eq!(session_manager.state.table_state.selected(), Some(1));
    }

    #[test]
    fn previous_is_no_op_when_no_sessions_present() {
        // Arrange
        let mut session_manager = session_manager_with_sessions(Vec::new());

        // Act
        session_manager.previous();

        // Assert
        assert_eq!(session_manager.state.table_state.selected(), None);
    }

    #[test]
    fn selected_session_returns_currently_selected_session_or_none() {
        // Arrange
        let mut session_manager = session_manager_with_sessions(vec![
            session_with_id("session-a", Status::InProgress),
            session_with_id("session-b", Status::Review),
        ]);

        // Act / Assert
        assert!(session_manager.selected_session().is_none());

        session_manager.state.table_state.select(Some(1));
        assert_eq!(
            session_manager
                .selected_session()
                .map(|session| session.id.clone()),
            Some("session-b".into())
        );
    }

    #[test]
    fn session_at_returns_session_by_index_or_none_for_out_of_range() {
        // Arrange
        let session_manager = session_manager_with_sessions(vec![
            session_with_id("session-a", Status::InProgress),
            session_with_id("session-b", Status::Review),
        ]);

        // Act / Assert
        assert_eq!(
            session_manager
                .session_at(0)
                .map(|session| session.id.as_str()),
            Some("session-a")
        );
        assert_eq!(
            session_manager
                .session_at(1)
                .map(|session| session.id.as_str()),
            Some("session-b")
        );
        assert!(session_manager.session_at(99).is_none());
    }

    #[test]
    fn session_id_for_index_returns_owned_id_or_none_for_out_of_range() {
        // Arrange
        let session_manager =
            session_manager_with_sessions(vec![session_with_id("session-a", Status::InProgress)]);

        // Act / Assert
        assert_eq!(
            session_manager.session_id_for_index(0),
            Some("session-a".into())
        );
        assert!(session_manager.session_id_for_index(1).is_none());
    }

    #[tokio::test]
    async fn set_session_model_persists_new_model_and_clears_conversation_state() {
        // Arrange
        let mut session = test_session("Prompt", Status::Review, Some("Title"), "");
        session.agent = AgentSelection::new(AgentKind::Claude, AgentModel::ClaudeSonnet5);
        let database = database_with_session(&session).await;
        database
            .sessions()
            .update_session_provider_conversation_id(
                "session-id",
                Some("provider-conv".to_string()),
            )
            .await
            .expect("seed provider conversation id");
        database
            .sessions()
            .update_session_instruction_conversation_id(
                "session-id",
                Some("instruction-conv".to_string()),
            )
            .await
            .expect("seed instruction conversation id");
        let mut session_manager = session_manager_with_one_session(session);
        let (services, mut event_rx) = test_services_with_event_receiver(
            &database,
            Arc::new(git::MockGitClient::new()),
            Arc::new(forge::MockReviewRequestClient::new()),
        );

        // Act
        session_manager
            .set_session_model(
                &services,
                "session-id",
                AgentSelection::new(AgentKind::Codex, AgentModel::Gpt55),
            )
            .await
            .expect("set session model should succeed");
        let persisted_model = database
            .sessions()
            .load_sessions()
            .await
            .expect("load sessions should succeed")
            .into_iter()
            .find(|row| row.id == "session-id")
            .expect("session row should exist")
            .model;
        let cleared_provider = database
            .sessions()
            .get_session_provider_conversation_id("session-id")
            .await
            .expect("provider id load should succeed");
        let cleared_instruction = database
            .sessions()
            .get_session_instruction_conversation_id("session-id")
            .await
            .expect("instruction id load should succeed");
        let emitted_event = event_rx.try_recv().expect("model event expected");

        // Assert
        assert_eq!(persisted_model, AgentModel::Gpt55.as_str());
        assert!(cleared_provider.is_none());
        assert!(cleared_instruction.is_none());
        assert_eq!(
            emitted_event,
            AppEvent::SessionModelUpdated {
                session_id: "session-id".into(),
                session_agent: AgentSelection::new(AgentKind::Codex, AgentModel::Gpt55),
            }
        );
        assert!(session_manager.should_replay_history("session-id"));
    }

    #[tokio::test]
    async fn set_session_model_keeps_conversation_state_when_model_does_not_change() {
        // Arrange
        let mut session = test_session("Prompt", Status::InProgress, Some("Title"), "");
        session.agent = AgentSelection::new(AgentKind::Claude, AgentModel::ClaudeSonnet5);
        let database = database_with_session(&session).await;
        database
            .sessions()
            .update_session_provider_conversation_id(
                "session-id",
                Some("provider-conv".to_string()),
            )
            .await
            .expect("seed provider conversation id");
        let mut session_manager = session_manager_with_one_session(session);
        let (services, mut event_rx) = test_services_with_event_receiver(
            &database,
            Arc::new(git::MockGitClient::new()),
            Arc::new(forge::MockReviewRequestClient::new()),
        );

        // Act
        session_manager
            .set_session_model(
                &services,
                "session-id",
                AgentSelection::new(AgentKind::Claude, AgentModel::ClaudeSonnet5),
            )
            .await
            .expect("set session model should succeed");
        let preserved_provider = database
            .sessions()
            .get_session_provider_conversation_id("session-id")
            .await
            .expect("provider id load should succeed");
        let emitted_event = event_rx.try_recv().expect("model event expected");

        // Assert
        assert_eq!(preserved_provider.as_deref(), Some("provider-conv"));
        assert_eq!(
            emitted_event,
            AppEvent::SessionModelUpdated {
                session_id: "session-id".into(),
                session_agent: AgentSelection::new(AgentKind::Claude, AgentModel::ClaudeSonnet5),
            }
        );
        assert!(!session_manager.should_replay_history("session-id"));
    }

    #[tokio::test]
    async fn set_session_model_returns_error_for_missing_session() {
        // Arrange
        let session = test_session("Prompt", Status::Review, Some("Title"), "");
        let database = database_with_session(&session).await;
        let mut session_manager = session_manager_with_one_session(session);
        let services = test_services(
            &database,
            Arc::new(git::MockGitClient::new()),
            Arc::new(forge::MockReviewRequestClient::new()),
        );

        // Act
        let result = session_manager
            .set_session_model(
                &services,
                "missing",
                AgentSelection::new(AgentKind::Codex, AgentModel::Gpt55),
            )
            .await;

        // Assert
        assert!(
            result.is_err(),
            "missing session should return SessionError"
        );
    }

    #[test]
    fn session_index_for_id_returns_index_or_none_for_unknown_session() {
        // Arrange
        let session_manager = session_manager_with_sessions(vec![
            session_with_id("session-a", Status::InProgress),
            session_with_id("session-b", Status::Review),
        ]);

        // Act / Assert
        assert_eq!(session_manager.session_index_for_id("session-a"), Some(0));
        assert_eq!(session_manager.session_index_for_id("session-b"), Some(1));
        assert!(session_manager.session_index_for_id("missing").is_none());
    }
}