agentty 0.13.5

//! Session-scoped persistence adapters and query helpers.

use ag_agent as agent;
use async_trait::async_trait;
use sqlx::SqlitePool;

use super::review::SessionReviewRequestRow;
use crate::domain::agent::{AgentKind, AgentModel, ReasoningLevel};
use crate::domain::session::{SessionFollowUpTask, SessionId, SessionStats};
use crate::domain::session_message::{SessionMessageKind, stored_message_content};
use crate::infra::db::DbError;

/// Transactional turn-metadata payload persisted after one completed agent
/// turn.
///
/// Owns its fields so the persistence trait method stays lifetime-free. A
/// borrowed variant (`SessionTurnMetadata<'a>`) forced the persist method to
/// carry a generic lifetime, which `mockall::automock` drops in the generated
/// mock and newer `clippy` then rejects via `extra_unused_lifetimes`. Owning
/// the data is allocation-cheap on this once-per-turn path and keeps the trait
/// signature stable across toolchains.
pub struct SessionTurnMetadata {
    /// Session-scoped instruction bootstrap marker for app-server providers.
    pub(crate) instruction_conversation_id: Option<String>,
    /// Model identifier used for per-model usage aggregation.
    pub(crate) model: String,
    /// Persisted provider-native conversation identifier for future resumes.
    pub(crate) provider_conversation_id: Option<String>,
    /// Serialized clarification-question payload stored on the session row.
    pub(crate) questions_json: String,
    /// Serialized structured summary payload stored on the session row.
    pub(crate) summary: String,
    /// Token-usage delta attributed to the completed turn.
    pub(crate) token_usage_delta: SessionStats,
}

/// Borrowed values used to persist a newly created session with explicit
/// provider identity and reasoning configuration.
pub struct PersistedSessionCreation<'a> {
    /// Persisted agent provider kind for the session.
    pub agent: &'a str,
    /// Base branch or parent branch used for future worktree materialization.
    pub base_branch: &'a str,
    /// Stable session identifier.
    pub id: &'a str,
    /// Whether the row was created through explicit draft staging.
    pub is_draft: bool,
    /// Persisted model identifier for the session.
    pub model: &'a str,
    /// Optional parent session id for one-level stacked drafts.
    pub parent_session_id: Option<&'a str>,
    /// Owning project identifier.
    pub project_id: i64,
    /// Reasoning level captured from the project default at creation.
    pub reasoning_level: ReasoningLevel,
    /// Initial lifecycle status string.
    pub status: &'a str,
}

/// Borrowed identifiers used to persist one forked session snapshot.
pub struct ForkSessionSnapshot<'a> {
    /// Stable id assigned to the newly forked session.
    pub new_session_id: &'a str,
    /// Stable id of the source session whose metadata and transcript are
    /// copied.
    pub source_session_id: &'a str,
    /// Initial lifecycle status for the forked session.
    pub status: &'a str,
}

/// Row returned when loading a session from the `session` table.
///
/// Includes optional normalized forge review-request linkage metadata loaded
/// through the `session_review_request` table when the session has been
/// published for remote review.
pub struct SessionRow {
    /// Persisted added-line count from the latest diff stats refresh.
    pub added_lines: i64,
    /// Persisted agent provider kind selected for this session.
    pub agent: String,
    /// Base branch used to create the session worktree.
    pub base_branch: String,
    /// Session creation timestamp in Unix seconds.
    pub created_at: i64,
    /// Persisted deleted-line count from the latest diff stats refresh.
    pub deleted_lines: i64,
    /// Stable session identifier.
    pub id: String,
    /// Open active-work interval start timestamp, if any.
    pub in_progress_started_at: Option<i64>,
    /// Completed active-work duration in whole seconds.
    pub in_progress_total_seconds: i64,
    /// Total input tokens accumulated for the session.
    pub input_tokens: i64,
    /// Whether the session is still an explicit draft.
    pub is_draft: bool,
    /// Persisted agent model identifier.
    pub model: String,
    /// Total output tokens accumulated for the session.
    pub output_tokens: i64,
    /// Parent session id when this is a one-level stacked draft.
    pub parent_session_id: Option<String>,
    /// Owning project identifier, when present.
    pub project_id: Option<i64>,
    /// Initial or staged prompt text.
    pub prompt: String,
    /// Published upstream branch reference, when present.
    pub published_upstream_ref: Option<String>,
    /// Serialized clarification-question payload, when present.
    pub questions: Option<String>,
    /// Persisted session-specific reasoning override, when present.
    pub reasoning_level_override: Option<String>,
    /// Joined forge review-request metadata, when present and complete.
    pub review_request: Option<SessionReviewRequestRow>,
    /// Persisted size bucket string.
    pub size: String,
    /// Persisted lifecycle status string.
    pub status: String,
    /// Persisted structured summary text, when present.
    pub summary: Option<String>,
    /// Optional display title.
    pub title: Option<String>,
    /// Last update timestamp in Unix seconds.
    pub updated_at: i64,
}

/// Lightweight row returned when loading session-list metadata.
///
/// Omits transcript-scale fields (`prompt`, `questions`, and `summary`) so
/// list refreshes scale with visible metadata instead of the cumulative size
/// of every saved conversation.
pub struct SessionListRow {
    /// Persisted added-line count from the latest diff stats refresh.
    pub added_lines: i64,
    /// Persisted agent provider kind for this session.
    pub agent: String,
    /// Base branch used to create the session worktree.
    pub base_branch: String,
    /// Session creation timestamp in Unix seconds.
    pub created_at: i64,
    /// Persisted deleted-line count from the latest diff stats refresh.
    pub deleted_lines: i64,
    /// Stable session identifier.
    pub id: String,
    /// Open active-work interval start timestamp, if any.
    pub in_progress_started_at: Option<i64>,
    /// Completed active-work duration in whole seconds.
    pub in_progress_total_seconds: i64,
    /// Total input tokens accumulated for the session.
    pub input_tokens: i64,
    /// Whether the session is still an explicit draft.
    pub is_draft: bool,
    /// Persisted agent model identifier.
    pub model: String,
    /// Total output tokens accumulated for the session.
    pub output_tokens: i64,
    /// Parent session id when this row is a one-level stacked draft.
    pub parent_session_id: Option<String>,
    /// Owning project identifier, when present.
    pub project_id: Option<i64>,
    /// Published upstream branch reference, when present.
    pub published_upstream_ref: Option<String>,
    /// Persisted session-specific reasoning override, when present.
    pub reasoning_level_override: Option<String>,
    /// Joined forge review-request metadata, when present and complete.
    pub review_request: Option<SessionReviewRequestRow>,
    /// Persisted size bucket string.
    pub size: String,
    /// Persisted lifecycle status string.
    pub status: String,
    /// Optional display title.
    pub title: Option<String>,
    /// Last update timestamp in Unix seconds.
    pub updated_at: i64,
}

/// Transcript-detail row loaded lazily for the session being viewed.
#[derive(sqlx::FromRow)]
pub struct SessionDetailRow {
    /// Initial or staged prompt text.
    pub prompt: String,
    /// Serialized clarification-question payload, when present.
    pub questions: Option<String>,
    /// Persisted structured summary text, when present.
    pub summary: Option<String>,
}

/// Row returned when loading one persisted `session_message`.
#[derive(Clone, Debug, Eq, PartialEq, sqlx::FromRow)]
pub struct SessionMessageRow {
    /// Canonical transcript text for this message.
    pub content: String,
    /// Stable message-kind string.
    pub kind: String,
    /// Monotonic position within the owning session transcript.
    pub position: i64,
}

/// Row returned when loading one persisted `session_follow_up_task`.
#[derive(Clone, Debug, Eq, PartialEq, sqlx::FromRow)]
pub struct SessionFollowUpTaskRow {
    /// Stable row identifier generated by `SQLite`.
    pub id: i64,
    /// Session id launched from this follow-up task, when one exists.
    pub launched_session_id: Option<String>,
    /// Display order for the task within its owning session.
    pub position: i64,
    /// Owning session identifier.
    pub session_id: String,
    /// Persisted follow-up task text shown to the user.
    pub text: String,
}

/// Row returned when hydrating persisted focused-review cache entries.
#[derive(Clone, Debug, Eq, PartialEq, sqlx::FromRow)]
pub struct SessionFocusedReviewRow {
    /// Diff-content hash captured when the focused review was generated.
    pub(crate) diff_hash: String,
    /// Stable session identifier.
    pub(crate) session_id: String,
    /// Generated focused-review markdown text.
    pub(crate) text: String,
}

impl SessionFollowUpTaskRow {
    /// Converts one follow-up-task row into the domain snapshot used by the
    /// UI.
    pub(crate) fn into_session_follow_up_task(self) -> SessionFollowUpTask {
        SessionFollowUpTask {
            id: self.id,
            launched_session_id: self.launched_session_id.map(SessionId::from),
            position: usize::try_from(self.position).unwrap_or(usize::MAX),
            text: self.text,
        }
    }
}
/// Session-focused persistence boundary used by app orchestration and tests.
#[async_trait]
pub trait SessionRepository: Send + Sync {
    /// Appends one typed transcript message and refreshes session ordering
    /// metadata.
    async fn append_session_message(
        &self,
        id: &str,
        kind: SessionMessageKind,
        content: &str,
    ) -> Result<(), DbError>;

    /// Sets `project_id` for sessions that do not yet reference a project.
    async fn backfill_session_project(&self, project_id: i64) -> Result<(), DbError>;

    /// Deletes a session row by identifier.
    async fn delete_session(&self, id: &str) -> Result<(), DbError>;

    /// Returns the persisted base branch for a session, when present.
    async fn get_session_base_branch(&self, id: &str) -> Result<Option<String>, DbError>;

    /// Returns the parent session id for a stacked session, when present.
    async fn get_session_parent_session_id(&self, id: &str) -> Result<Option<String>, DbError>;

    /// Returns the parent/base commit hash that a stacked child branch was
    /// last known to contain.
    async fn get_session_stack_base_commit_hash(&self, id: &str)
    -> Result<Option<String>, DbError>;

    /// Returns the persisted app-server instruction bootstrap marker for a
    /// session, when present.
    async fn get_session_instruction_conversation_id(
        &self,
        id: &str,
    ) -> Result<Option<String>, DbError>;

    /// Returns the provider conversation identifier for a session, when
    /// present.
    async fn get_session_provider_conversation_id(
        &self,
        id: &str,
    ) -> Result<Option<String>, DbError>;

    /// Inserts a newly created draft-session row.
    async fn insert_draft_session(
        &self,
        id: &str,
        model: &str,
        base_branch: &str,
        status: &str,
        project_id: i64,
    ) -> Result<(), DbError>;

    /// Inserts a newly created stacked draft-session row.
    async fn insert_stacked_draft_session(
        &self,
        id: &str,
        model: &str,
        base_branch: &str,
        status: &str,
        parent_session_id: &str,
        project_id: i64,
    ) -> Result<(), DbError>;

    /// Inserts a newly created session row.
    async fn insert_session(
        &self,
        id: &str,
        model: &str,
        base_branch: &str,
        status: &str,
        project_id: i64,
    ) -> Result<(), DbError>;

    /// Inserts a newly created session row with explicit provider identity.
    async fn insert_session_with_agent(
        &self,
        session: PersistedSessionCreation<'_>,
    ) -> Result<(), DbError>;

    /// Inserts a new session by snapshotting source metadata and ordered
    /// transcript messages while clearing source-specific runtime linkage.
    async fn fork_session_snapshot(&self, snapshot: ForkSessionSnapshot<'_>)
    -> Result<(), DbError>;

    #[cfg(test)]
    /// Loads all sessions ordered by most recent update.
    async fn load_sessions(&self) -> Result<Vec<SessionRow>, DbError>;

    /// Loads lightweight session-list metadata ordered by most recent update
    /// for one project.
    async fn load_sessions_for_project(
        &self,
        project_id: i64,
    ) -> Result<Vec<SessionListRow>, DbError>;

    /// Loads transcript-scale detail for one session when it becomes active.
    async fn load_session_detail(
        &self,
        session_id: &str,
    ) -> Result<Option<SessionDetailRow>, DbError>;

    /// Loads ordered transcript messages for one session.
    async fn load_session_messages(
        &self,
        session_id: &str,
    ) -> Result<Vec<SessionMessageRow>, DbError>;

    /// Loads all persisted session follow-up-task rows in stable display
    /// order.
    async fn load_session_follow_up_tasks(&self) -> Result<Vec<SessionFollowUpTaskRow>, DbError>;

    /// Loads persisted focused-review cache rows for one project.
    async fn load_session_focused_reviews_for_project(
        &self,
        project_id: i64,
    ) -> Result<Vec<SessionFocusedReviewRow>, DbError>;

    /// Loads lightweight session metadata used for cheap change detection.
    async fn load_sessions_metadata(&self) -> Result<(i64, i64), DbError>;

    /// Loads the project identifier associated with one session.
    async fn load_session_project_id(&self, session_id: &str) -> Result<Option<i64>, DbError>;

    /// Loads parentless review-ready sessions that still need their recorded
    /// stack-base commit replayed onto their current base branch.
    async fn load_pending_stack_restack_session_ids(
        &self,
        project_id: i64,
    ) -> Result<Vec<String>, DbError>;

    /// Returns the persisted upstream reference for a published session
    /// branch, when present.
    async fn load_session_published_upstream_ref(
        &self,
        id: &str,
    ) -> Result<Option<String>, DbError>;

    /// Loads the persisted merged commit hash for one session, when present.
    async fn load_session_merged_commit_hash(
        &self,
        session_id: &str,
    ) -> Result<Option<String>, DbError>;

    /// Clears parent links for children after their parent session merges
    /// into its base branch, returning materialized children that may need a
    /// follow-up branch restack.
    async fn restack_child_sessions_after_parent_merge(
        &self,
        parent_session_id: &str,
        base_branch: &str,
        parent_commit_hash: Option<String>,
    ) -> Result<Vec<String>, DbError>;

    /// Loads the persisted session reasoning level.
    async fn load_session_reasoning_level(
        &self,
        session_id: &str,
    ) -> Result<ReasoningLevel, DbError>;

    /// Loads the persisted summary text associated with one session.
    async fn load_session_summary(&self, session_id: &str) -> Result<Option<String>, DbError>;

    /// Returns `(created_at, updated_at)` timestamps for a session.
    async fn load_session_timestamps(
        &self,
        session_id: &str,
    ) -> Result<Option<(i64, i64)>, DbError>;

    /// Persists all canonical turn metadata for one completed agent turn in a
    /// single transaction.
    async fn persist_session_turn_metadata(
        &self,
        session_id: &str,
        turn_metadata: &SessionTurnMetadata,
    ) -> Result<(), DbError>;

    /// Replaces the persisted follow-up task list for one session inside one
    /// transaction so task deletion and reinsertion commit atomically.
    async fn replace_session_follow_up_tasks(
        &self,
        session_id: &str,
        follow_up_tasks: &[String],
    ) -> Result<(), DbError>;

    /// Updates persisted diff-derived size and line-count fields for a
    /// session row.
    async fn update_session_diff_stats(
        &self,
        added_lines: u64,
        deleted_lines: u64,
        id: &str,
        size: &str,
    ) -> Result<(), DbError>;

    /// Updates the launched sibling-session link for one persisted follow-up
    /// task.
    async fn update_session_follow_up_task_launched_session_id(
        &self,
        session_id: &str,
        position: usize,
        launched_session_id: Option<String>,
    ) -> Result<(), DbError>;

    /// Updates the persisted app-server instruction bootstrap marker for a
    /// session.
    async fn update_session_instruction_conversation_id(
        &self,
        id: &str,
        provider_conversation_id: Option<String>,
    ) -> Result<(), DbError>;

    /// Updates the persisted model for a session.
    async fn update_session_model(&self, id: &str, model: &str) -> Result<(), DbError>;

    /// Updates the persisted agent provider and model for a session.
    async fn update_session_agent_model(
        &self,
        id: &str,
        agent: &str,
        model: &str,
    ) -> Result<(), DbError>;

    /// Clears the draft flag for a session row once its staged draft bundle
    /// starts the first live turn.
    async fn clear_session_draft_flag(&self, id: &str) -> Result<(), DbError>;

    /// Updates the persisted merged commit hash for a session row.
    async fn update_session_merged_commit_hash(
        &self,
        id: &str,
        merged_commit_hash: Option<String>,
    ) -> Result<(), DbError>;

    /// Persists or clears the parent/base commit hash used for deterministic
    /// stacked-child rebases.
    async fn update_session_stack_base_commit_hash(
        &self,
        id: &str,
        stack_base_commit_hash: Option<String>,
    ) -> Result<(), DbError>;

    /// Updates the saved prompt for a session row.
    async fn update_session_prompt(&self, id: &str, prompt: &str) -> Result<(), DbError>;

    /// Updates the persisted provider conversation identifier for a session.
    async fn update_session_provider_conversation_id(
        &self,
        id: &str,
        provider_conversation_id: Option<String>,
    ) -> Result<(), DbError>;

    /// Updates the model clarification questions for a session row.
    async fn update_session_questions(&self, id: &str, questions: &str) -> Result<(), DbError>;

    /// Updates the persisted session reasoning level.
    async fn update_session_reasoning_level(
        &self,
        id: &str,
        reasoning_level: ReasoningLevel,
    ) -> Result<(), DbError>;

    /// Updates the persisted upstream reference for a published session
    /// branch.
    async fn update_session_published_upstream_ref(
        &self,
        id: &str,
        published_upstream_ref: Option<String>,
    ) -> Result<(), DbError>;

    /// Accumulates token statistics for a session.
    async fn update_session_stats(&self, id: &str, stats: &SessionStats) -> Result<(), DbError>;

    /// Updates the status for a session row and opens or closes the persisted
    /// cumulative active-work interval when crossing the `InProgress`
    /// boundary.
    async fn update_session_status_with_timing_at(
        &self,
        id: &str,
        status: &str,
        timestamp_seconds: i64,
    ) -> Result<(), DbError>;

    /// Updates the persisted session summary text for a session row.
    async fn update_session_summary(&self, id: &str, summary: &str) -> Result<(), DbError>;

    /// Updates or clears the persisted focused-review cache for a session.
    async fn update_session_focused_review(
        &self,
        id: &str,
        diff_hash: Option<String>,
        text: Option<String>,
    ) -> Result<(), DbError>;

    /// Updates the display title for a session row.
    async fn update_session_title(&self, id: &str, title: &str) -> Result<(), DbError>;

    /// Updates the display title for a session row only when the persisted
    /// prompt still matches the prompt snapshot used to generate that title.
    async fn update_session_title_for_prompt(
        &self,
        id: &str,
        expected_prompt: &str,
        title: &str,
    ) -> Result<bool, DbError>;

    /// Overrides the `created_at` timestamp for one session row.
    #[cfg(test)]
    async fn update_session_created_at(&self, id: &str, created_at: i64) -> Result<(), DbError>;

    #[cfg(test)]
    /// Overrides the `updated_at` timestamp for one session row.
    async fn update_session_updated_at(&self, id: &str, updated_at: i64) -> Result<(), DbError>;
}

/// `SQLite` implementation of [`SessionRepository`].
#[derive(Clone)]
pub(crate) struct SqliteSessionRepository(SqlitePool);

impl SqliteSessionRepository {
    /// Creates a session repository backed by the provided pool.
    pub(crate) fn new(pool: SqlitePool) -> Self {
        Self(pool)
    }
}

/// Row returned when loading a required string scalar value.
struct RequiredStringValueRow {
    value: String,
}

/// Row returned when loading session count and latest-update metadata.
struct SessionStatsMetadataRow {
    /// Latest `session.updated_at` timestamp across rows.
    max_updated_at: i64,
    /// Total number of persisted sessions.
    session_count: i64,
}

/// Row returned when loading an optional `i64` scalar value.
struct OptionalI64ValueRow {
    value: Option<i64>,
}

/// Row returned when loading the persisted instruction bootstrap marker for
/// one session.
#[derive(sqlx::FromRow)]
struct SessionInstructionStateRow {
    app_server_instruction_provider_conversation_id: Option<String>,
}

impl SessionInstructionStateRow {
    /// Converts the optional stored provider conversation id into one
    /// normalized bootstrap conversation id when present and non-empty.
    fn into_instruction_conversation_id(self) -> Option<String> {
        agent::normalize_instruction_conversation_id(
            self.app_server_instruction_provider_conversation_id
                .as_deref(),
        )
    }
}

/// Row returned when loading both persisted timestamps for one session.
struct SessionTimestampsRow {
    created_at: i64,
    updated_at: i64,
}

/// Shared columns for session metadata rows used by both session and
/// session-list mappings.
#[derive(sqlx::FromRow)]
struct SessionRowMetadata {
    added_lines: i64,
    agent: String,
    base_branch: String,
    created_at: i64,
    deleted_lines: i64,
    id: String,
    in_progress_started_at: Option<i64>,
    in_progress_total_seconds: i64,
    input_tokens: i64,
    is_draft: bool,
    model: String,
    output_tokens: i64,
    parent_session_id: Option<String>,
    project_id: Option<i64>,
    published_upstream_ref: Option<String>,
    reasoning_level_override: Option<String>,
    size: String,
    status: String,
    title: Option<String>,
    updated_at: i64,
}

impl SessionRowMetadata {
    /// Converts shared metadata fields into a session-list row.
    fn into_session_list_row(
        self,
        review_request: Option<SessionReviewRequestRow>,
    ) -> SessionListRow {
        SessionListRow {
            added_lines: self.added_lines,
            agent: self.agent,
            base_branch: self.base_branch,
            created_at: self.created_at,
            deleted_lines: self.deleted_lines,
            id: self.id,
            in_progress_started_at: self.in_progress_started_at,
            in_progress_total_seconds: self.in_progress_total_seconds,
            input_tokens: self.input_tokens,
            is_draft: self.is_draft,
            model: self.model,
            output_tokens: self.output_tokens,
            parent_session_id: self.parent_session_id,
            project_id: self.project_id,
            published_upstream_ref: self.published_upstream_ref,
            reasoning_level_override: self.reasoning_level_override,
            review_request,
            size: self.size,
            status: self.status,
            title: self.title,
            updated_at: self.updated_at,
        }
    }
}

/// Row returned when loading one lightweight `session` list entry plus aliased
/// `session_review_request` join columns.
#[derive(sqlx::FromRow)]
struct SessionListJoinRow {
    #[sqlx(flatten)]
    metadata: SessionRowMetadata,
    review_request_display_id: Option<String>,
    review_request_forge_kind: Option<String>,
    review_request_last_refreshed_at: Option<i64>,
    review_request_source_branch: Option<String>,
    review_request_state: Option<String>,
    review_request_status_summary: Option<String>,
    review_request_target_branch: Option<String>,
    review_request_title: Option<String>,
    review_request_web_url: Option<String>,
}

impl SessionListJoinRow {
    /// Converts the query-mapped list join row into a lightweight
    /// [`SessionListRow`].
    fn into_session_list_row(self) -> SessionListRow {
        let Self {
            metadata,
            review_request_display_id,
            review_request_forge_kind,
            review_request_last_refreshed_at,
            review_request_source_branch,
            review_request_state,
            review_request_status_summary,
            review_request_target_branch,
            review_request_title,
            review_request_web_url,
        } = self;

        let review_request = SessionReviewRequestJoinRow {
            display_id: review_request_display_id,
            forge_kind: review_request_forge_kind,
            last_refreshed_at: review_request_last_refreshed_at,
            source_branch: review_request_source_branch,
            state: review_request_state,
            status_summary: review_request_status_summary,
            target_branch: review_request_target_branch,
            title: review_request_title,
            web_url: review_request_web_url,
        }
        .into_review_request_row();

        metadata.into_session_list_row(review_request)
    }
}

/// Aliased nullable `session_review_request` columns loaded through a joined
/// session query.
struct SessionReviewRequestJoinRow {
    display_id: Option<String>,
    forge_kind: Option<String>,
    last_refreshed_at: Option<i64>,
    source_branch: Option<String>,
    state: Option<String>,
    status_summary: Option<String>,
    target_branch: Option<String>,
    title: Option<String>,
    web_url: Option<String>,
}

impl SessionReviewRequestJoinRow {
    /// Converts the joined nullable columns into a review-request row only
    /// when every required field is present.
    fn into_review_request_row(self) -> Option<SessionReviewRequestRow> {
        let Self {
            display_id,
            forge_kind,
            last_refreshed_at,
            source_branch,
            state,
            status_summary,
            target_branch,
            title,
            web_url,
        } = self;

        Some(SessionReviewRequestRow {
            display_id: display_id?,
            forge_kind: forge_kind?,
            last_refreshed_at: last_refreshed_at?,
            source_branch: source_branch?,
            state: state?,
            status_summary,
            target_branch: target_branch?,
            title: title?,
            web_url: web_url?,
        })
    }
}

/// SQL statement that copies session metadata for one fork while clearing
/// runtime- and source-specific linkage.
const FORK_SESSION_ROW_SQL: &str = r"
INSERT INTO session (
    id,
    agent,
    model,
    base_branch,
    status,
    project_id,
    prompt,
    summary,
    title,
    reasoning_level,
    added_lines,
    deleted_lines,
    size,
    input_tokens,
    output_tokens,
    is_draft,
    parent_session_id,
    provider_conversation_id,
    app_server_instruction_provider_conversation_id,
    questions,
    published_upstream_ref,
    merged_commit_hash,
    focused_review_text,
    focused_review_diff_hash,
    stack_base_commit_hash,
    in_progress_total_seconds,
    in_progress_started_at,
    created_at,
    updated_at
)
SELECT ?,
       agent,
       model,
       base_branch,
       ?,
       project_id,
       prompt,
       summary,
       title,
       reasoning_level,
       added_lines,
       deleted_lines,
       size,
       0,
       0,
       0,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       0,
       NULL,
       CAST(strftime('%s', 'now') AS INTEGER),
       CAST(strftime('%s', 'now') AS INTEGER)
FROM session
WHERE id = ?
";

/// SQL statement that copies durable transcript messages for one fork.
const FORK_SESSION_MESSAGES_SQL: &str = r"
INSERT INTO session_message (session_id, position, kind, content, created_at)
SELECT ?,
       position,
       kind,
       content,
       created_at
FROM session_message
WHERE session_id = ?
ORDER BY position, id
";

#[async_trait]
impl SessionRepository for SqliteSessionRepository {
    async fn append_session_message(
        &self,
        id: &str,
        kind: SessionMessageKind,
        content: &str,
    ) -> Result<(), DbError> {
        let content = stored_message_content(kind, content);
        if content.trim().is_empty() {
            return Ok(());
        }

        let mut transaction = self.0.begin().await?;

        let update_result = sqlx::query(
            r"
UPDATE session
SET updated_at = CAST(strftime('%s', 'now') AS INTEGER)
WHERE id = ?
",
        )
        .bind(id)
        .execute(&mut *transaction)
        .await?;

        if update_result.rows_affected() == 0 {
            transaction.commit().await?;

            return Ok(());
        }

        sqlx::query(
            r"
INSERT INTO session_message (session_id, position, kind, content, created_at)
SELECT ?, COALESCE(MAX(position), -1) + 1, ?, ?, CAST(strftime('%s', 'now') AS INTEGER)
FROM session_message
WHERE session_id = ?
",
        )
        .bind(id)
        .bind(kind.as_str())
        .bind(content)
        .bind(id)
        .execute(&mut *transaction)
        .await?;

        transaction.commit().await?;

        Ok(())
    }

    async fn backfill_session_project(&self, project_id: i64) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET project_id = ?
WHERE project_id IS NULL
",
        )
        .bind(project_id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn delete_session(&self, id: &str) -> Result<(), DbError> {
        let mut transaction = self.0.begin().await?;

        // Retarget any stacked children onto this session's base branch before
        // the row is removed. The `ON DELETE SET NULL` foreign key clears the
        // child parent link automatically, but it leaves children pointing at
        // the deleted parent's worktree branch, which no longer exists. Mirror
        // the post-merge restack so a surviving child rebases against the
        // parent's base branch instead of an orphaned `wt/<parent>` ref.
        sqlx::query(
            r"
UPDATE session
SET parent_session_id = NULL,
    base_branch = COALESCE((SELECT base_branch FROM session WHERE id = ?), base_branch)
WHERE parent_session_id = ?
  AND status <> 'Canceled'
",
        )
        .bind(id)
        .bind(id)
        .execute(&mut *transaction)
        .await?;

        sqlx::query(
            r"
DELETE FROM session
WHERE id = ?
",
        )
        .bind(id)
        .execute(&mut *transaction)
        .await?;

        transaction.commit().await?;

        Ok(())
    }

    async fn get_session_base_branch(&self, id: &str) -> Result<Option<String>, DbError> {
        let row = sqlx::query_as!(
            RequiredStringValueRow,
            r#"
SELECT base_branch AS "value!: _"
FROM session
WHERE id = ?
"#,
            id
        )
        .fetch_optional(&self.0)
        .await?;

        Ok(row.map(|row| row.value))
    }

    async fn get_session_parent_session_id(&self, id: &str) -> Result<Option<String>, DbError> {
        let value = sqlx::query_scalar::<_, Option<String>>(
            r"
SELECT parent_session_id
FROM session
WHERE id = ?
",
        )
        .bind(id)
        .fetch_optional(&self.0)
        .await?
        .flatten();

        Ok(value)
    }

    async fn get_session_stack_base_commit_hash(
        &self,
        id: &str,
    ) -> Result<Option<String>, DbError> {
        let value = sqlx::query_scalar::<_, Option<String>>(
            r"
SELECT stack_base_commit_hash
FROM session
WHERE id = ?
",
        )
        .bind(id)
        .fetch_optional(&self.0)
        .await?
        .flatten();

        Ok(value)
    }

    async fn get_session_instruction_conversation_id(
        &self,
        id: &str,
    ) -> Result<Option<String>, DbError> {
        let row = sqlx::query_as::<_, SessionInstructionStateRow>(
            r"
SELECT app_server_instruction_provider_conversation_id
FROM session
WHERE id = ?
",
        )
        .bind(id)
        .fetch_optional(&self.0)
        .await?;

        Ok(row.and_then(SessionInstructionStateRow::into_instruction_conversation_id))
    }

    async fn get_session_provider_conversation_id(
        &self,
        id: &str,
    ) -> Result<Option<String>, DbError> {
        let value = sqlx::query_scalar!(
            r"SELECT provider_conversation_id FROM session WHERE id = ?",
            id
        )
        .fetch_optional(&self.0)
        .await?
        .flatten();

        Ok(value)
    }

    async fn insert_draft_session(
        &self,
        id: &str,
        model: &str,
        base_branch: &str,
        status: &str,
        project_id: i64,
    ) -> Result<(), DbError> {
        let agent = persisted_agent_for_model(model);

        insert_session_with_draft_mode(
            &self.0,
            InsertSessionRow {
                agent: &agent,
                base_branch,
                id,
                is_draft: true,
                model,
                parent_session_id: None,
                project_id,
                reasoning_level: ReasoningLevel::default(),
                status,
            },
        )
        .await
    }

    async fn insert_stacked_draft_session(
        &self,
        id: &str,
        model: &str,
        base_branch: &str,
        status: &str,
        parent_session_id: &str,
        project_id: i64,
    ) -> Result<(), DbError> {
        let agent = persisted_agent_for_model(model);

        insert_session_with_draft_mode(
            &self.0,
            InsertSessionRow {
                agent: &agent,
                base_branch,
                id,
                is_draft: true,
                model,
                parent_session_id: Some(parent_session_id),
                project_id,
                reasoning_level: ReasoningLevel::default(),
                status,
            },
        )
        .await
    }

    async fn insert_session(
        &self,
        id: &str,
        model: &str,
        base_branch: &str,
        status: &str,
        project_id: i64,
    ) -> Result<(), DbError> {
        let agent = persisted_agent_for_model(model);

        insert_session_with_draft_mode(
            &self.0,
            InsertSessionRow {
                agent: &agent,
                base_branch,
                id,
                is_draft: false,
                model,
                parent_session_id: None,
                project_id,
                reasoning_level: ReasoningLevel::default(),
                status,
            },
        )
        .await
    }

    async fn insert_session_with_agent(
        &self,
        session: PersistedSessionCreation<'_>,
    ) -> Result<(), DbError> {
        let PersistedSessionCreation {
            agent,
            base_branch,
            id,
            is_draft,
            model,
            parent_session_id,
            project_id,
            reasoning_level,
            status,
        } = session;

        insert_session_with_draft_mode(
            &self.0,
            InsertSessionRow {
                agent,
                base_branch,
                id,
                is_draft,
                model,
                parent_session_id,
                project_id,
                reasoning_level,
                status,
            },
        )
        .await
    }

    async fn fork_session_snapshot(
        &self,
        snapshot: ForkSessionSnapshot<'_>,
    ) -> Result<(), DbError> {
        let ForkSessionSnapshot {
            new_session_id,
            source_session_id,
            status,
        } = snapshot;
        let mut transaction = self.0.begin().await?;

        let insert_result = sqlx::query(FORK_SESSION_ROW_SQL)
            .bind(new_session_id)
            .bind(status)
            .bind(source_session_id)
            .execute(&mut *transaction)
            .await?;
        if insert_result.rows_affected() != 1 {
            return Err(sqlx::Error::RowNotFound.into());
        }

        sqlx::query(FORK_SESSION_MESSAGES_SQL)
            .bind(new_session_id)
            .bind(source_session_id)
            .execute(&mut *transaction)
            .await?;

        transaction.commit().await?;

        Ok(())
    }

    #[cfg(test)]
    async fn load_sessions(&self) -> Result<Vec<SessionRow>, DbError> {
        let rows = sqlx::query_as::<_, tests::SessionJoinRow>(
            r"
SELECT session.base_branch AS base_branch,
       session.added_lines AS added_lines,
       session.agent AS agent,
       session.created_at AS created_at,
       session.deleted_lines AS deleted_lines,
       session.id AS id,
       session.in_progress_started_at,
       session.in_progress_total_seconds AS in_progress_total_seconds,
       session.input_tokens AS input_tokens,
       session.is_draft AS is_draft,
       session.model AS model,
       session.output_tokens AS output_tokens,
       session.parent_session_id,
       session.project_id,
       session.prompt AS prompt,
       session.reasoning_level AS reasoning_level_override,
       session.published_upstream_ref,
       session.questions,
       session_review_request.display_id AS review_request_display_id,
       session_review_request.forge_kind AS review_request_forge_kind,
       session_review_request.last_refreshed_at AS review_request_last_refreshed_at,
       session_review_request.source_branch AS review_request_source_branch,
       session_review_request.state AS review_request_state,
       session_review_request.status_summary AS review_request_status_summary,
       session_review_request.target_branch AS review_request_target_branch,
       session_review_request.title AS review_request_title,
       session_review_request.web_url AS review_request_web_url,
       session.size AS size,
       session.status AS status,
       session.summary,
       session.title,
       session.updated_at AS updated_at
FROM session
LEFT JOIN session_review_request
ON session_review_request.session_id = session.id
ORDER BY session.updated_at DESC, session.created_at DESC, session.id
",
        )
        .fetch_all(&self.0)
        .await?;

        Ok(rows
            .into_iter()
            .map(tests::SessionJoinRow::into_session_row)
            .collect())
    }

    async fn load_sessions_for_project(
        &self,
        project_id: i64,
    ) -> Result<Vec<SessionListRow>, DbError> {
        let rows = sqlx::query_as::<_, SessionListJoinRow>(
            r"
SELECT session.base_branch AS base_branch,
       session.added_lines AS added_lines,
       session.agent AS agent,
       session.created_at AS created_at,
       session.deleted_lines AS deleted_lines,
       session.id AS id,
       session.in_progress_started_at,
       session.in_progress_total_seconds AS in_progress_total_seconds,
       session.input_tokens AS input_tokens,
       session.is_draft AS is_draft,
       session.model AS model,
       session.output_tokens AS output_tokens,
       session.parent_session_id,
       session.project_id,
       session.reasoning_level AS reasoning_level_override,
       session.published_upstream_ref,
       session_review_request.display_id AS review_request_display_id,
       session_review_request.forge_kind AS review_request_forge_kind,
       session_review_request.last_refreshed_at AS review_request_last_refreshed_at,
       session_review_request.source_branch AS review_request_source_branch,
       session_review_request.state AS review_request_state,
       session_review_request.status_summary AS review_request_status_summary,
       session_review_request.target_branch AS review_request_target_branch,
       session_review_request.title AS review_request_title,
       session_review_request.web_url AS review_request_web_url,
       session.size AS size,
       session.status AS status,
       session.title,
       session.updated_at AS updated_at
FROM session
LEFT JOIN session_review_request
ON session_review_request.session_id = session.id
WHERE session.project_id = ?
ORDER BY session.updated_at DESC, session.created_at DESC, session.id
",
        )
        .bind(project_id)
        .fetch_all(&self.0)
        .await?;

        Ok(rows
            .into_iter()
            .map(SessionListJoinRow::into_session_list_row)
            .collect())
    }

    async fn load_session_detail(
        &self,
        session_id: &str,
    ) -> Result<Option<SessionDetailRow>, DbError> {
        let row = sqlx::query_as::<_, SessionDetailRow>(
            r"
SELECT prompt,
       questions,
       summary
FROM session
WHERE id = ?
",
        )
        .bind(session_id)
        .fetch_optional(&self.0)
        .await?;

        Ok(row)
    }

    async fn load_session_messages(
        &self,
        session_id: &str,
    ) -> Result<Vec<SessionMessageRow>, DbError> {
        let rows = sqlx::query_as::<_, SessionMessageRow>(
            r"
SELECT content,
       kind,
       position
FROM session_message
WHERE session_id = ?
ORDER BY position, id
",
        )
        .bind(session_id)
        .fetch_all(&self.0)
        .await?;

        Ok(rows)
    }

    async fn load_session_follow_up_tasks(&self) -> Result<Vec<SessionFollowUpTaskRow>, DbError> {
        let rows = match sqlx::query_as::<_, SessionFollowUpTaskRow>(
            r"
SELECT id,
       launched_session_id,
       position,
       session_id,
       text
FROM session_follow_up_task
ORDER BY session_id, position, id
",
        )
        .fetch_all(&self.0)
        .await
        {
            Ok(rows) => rows,
            Err(error) if is_missing_follow_up_task_table(&error) => return Ok(Vec::new()),
            Err(error) => return Err(error.into()),
        };

        Ok(rows)
    }

    async fn load_session_focused_reviews_for_project(
        &self,
        project_id: i64,
    ) -> Result<Vec<SessionFocusedReviewRow>, DbError> {
        let rows = sqlx::query_as::<_, SessionFocusedReviewRow>(
            r"
SELECT id AS session_id,
       focused_review_diff_hash AS diff_hash,
       focused_review_text AS text
FROM session
WHERE project_id = ?
  AND focused_review_diff_hash IS NOT NULL
  AND focused_review_text IS NOT NULL
  AND focused_review_text <> ''
ORDER BY updated_at DESC, id
",
        )
        .bind(project_id)
        .fetch_all(&self.0)
        .await?;

        Ok(rows)
    }

    async fn load_sessions_metadata(&self) -> Result<(i64, i64), DbError> {
        let row = sqlx::query_as!(
            SessionStatsMetadataRow,
            r#"
SELECT (SELECT COUNT(*) FROM session) AS "session_count!: _",
       COALESCE(
           (
               SELECT updated_at
               FROM session
               ORDER BY updated_at DESC, id
               LIMIT 1
           ),
           0
       ) AS "max_updated_at!: _"
"#
        )
        .fetch_one(&self.0)
        .await?;

        Ok((row.session_count, row.max_updated_at))
    }

    async fn load_session_project_id(&self, session_id: &str) -> Result<Option<i64>, DbError> {
        let row = sqlx::query_as!(
            OptionalI64ValueRow,
            r#"
SELECT project_id AS "value: _"
FROM session
WHERE id = ?
"#,
            session_id
        )
        .fetch_optional(&self.0)
        .await?;

        Ok(row.and_then(|row| row.value))
    }

    async fn load_pending_stack_restack_session_ids(
        &self,
        project_id: i64,
    ) -> Result<Vec<String>, DbError> {
        let session_ids = sqlx::query_scalar::<_, String>(
            r"
SELECT id
FROM session
WHERE project_id = ?
  AND parent_session_id IS NULL
  AND stack_base_commit_hash IS NOT NULL
  AND status IN ('Review', 'AgentReview')
ORDER BY updated_at ASC, id ASC
",
        )
        .bind(project_id)
        .fetch_all(&self.0)
        .await?;

        Ok(session_ids)
    }

    async fn load_session_published_upstream_ref(
        &self,
        id: &str,
    ) -> Result<Option<String>, DbError> {
        let value = sqlx::query_scalar!(
            r"SELECT published_upstream_ref FROM session WHERE id = ?",
            id
        )
        .fetch_optional(&self.0)
        .await?
        .flatten();

        Ok(value)
    }

    async fn load_session_merged_commit_hash(
        &self,
        session_id: &str,
    ) -> Result<Option<String>, DbError> {
        let row = sqlx::query_scalar::<_, Option<String>>(
            r"
SELECT merged_commit_hash
FROM session
WHERE id = ?
",
        )
        .bind(session_id)
        .fetch_optional(&self.0)
        .await?;

        Ok(row.flatten())
    }

    async fn load_session_reasoning_level(
        &self,
        session_id: &str,
    ) -> Result<ReasoningLevel, DbError> {
        let value = sqlx::query_scalar!(
            r"SELECT reasoning_level FROM session WHERE id = ?",
            session_id
        )
        .fetch_optional(&self.0)
        .await?
        .flatten();

        Ok(value
            .and_then(|value| value.parse::<ReasoningLevel>().ok())
            .unwrap_or_default())
    }

    async fn restack_child_sessions_after_parent_merge(
        &self,
        parent_session_id: &str,
        base_branch: &str,
        parent_commit_hash: Option<String>,
    ) -> Result<Vec<String>, DbError> {
        let mut transaction = self.0.begin().await?;
        let materialized_child_ids = sqlx::query_scalar::<_, String>(
            r"
SELECT id
FROM session
WHERE parent_session_id = ?
  AND status NOT IN ('Canceled', 'Draft')
ORDER BY created_at ASC, id ASC
",
        )
        .bind(parent_session_id)
        .fetch_all(&mut *transaction)
        .await?;

        sqlx::query(
            r"
UPDATE session
SET parent_session_id = NULL,
    base_branch = ?,
    stack_base_commit_hash = CASE
        WHEN status = 'Draft' THEN NULL
        ELSE COALESCE(stack_base_commit_hash, ?)
    END
WHERE parent_session_id = ?
  AND status <> 'Canceled'
",
        )
        .bind(base_branch)
        .bind(parent_commit_hash)
        .bind(parent_session_id)
        .execute(&mut *transaction)
        .await?;

        transaction.commit().await?;

        Ok(materialized_child_ids)
    }

    async fn load_session_summary(&self, session_id: &str) -> Result<Option<String>, DbError> {
        let row = sqlx::query_scalar::<_, Option<String>>(
            r"
SELECT summary
FROM session
WHERE id = ?
",
        )
        .bind(session_id)
        .fetch_optional(&self.0)
        .await?;

        Ok(row.flatten())
    }

    async fn load_session_timestamps(
        &self,
        session_id: &str,
    ) -> Result<Option<(i64, i64)>, DbError> {
        let row = sqlx::query_as!(
            SessionTimestampsRow,
            r#"
SELECT created_at, updated_at
FROM session
WHERE id = ?
            "#,
            session_id
        )
        .fetch_optional(&self.0)
        .await?;

        Ok(row.map(|row| (row.created_at, row.updated_at)))
    }

    async fn persist_session_turn_metadata(
        &self,
        session_id: &str,
        turn_metadata: &SessionTurnMetadata,
    ) -> Result<(), DbError> {
        let mut transaction = self.0.begin().await?;

        let session_update = sqlx::query(
            r"
UPDATE session
SET questions = ?,
    summary = ?,
    provider_conversation_id = ?,
    app_server_instruction_provider_conversation_id = ?
WHERE id = ?
",
        )
        .bind(turn_metadata.questions_json.as_str())
        .bind(turn_metadata.summary.as_str())
        .bind(turn_metadata.provider_conversation_id.as_deref())
        .bind(turn_metadata.instruction_conversation_id.as_deref())
        .bind(session_id)
        .execute(&mut *transaction)
        .await?;
        if session_update.rows_affected() != 1 {
            return Err(sqlx::Error::RowNotFound.into());
        }

        if turn_metadata.token_usage_delta.input_tokens != 0
            || turn_metadata.token_usage_delta.output_tokens != 0
        {
            sqlx::query(
                r"
UPDATE session
SET input_tokens = input_tokens + ?,
    output_tokens = output_tokens + ?
WHERE id = ?
",
            )
            .bind(turn_metadata.token_usage_delta.input_tokens.cast_signed())
            .bind(turn_metadata.token_usage_delta.output_tokens.cast_signed())
            .bind(session_id)
            .execute(&mut *transaction)
            .await?;

            sqlx::query(
                r"
INSERT INTO session_usage (session_id, model, input_tokens, output_tokens, invocation_count)
VALUES (?, ?, ?, ?, 1)
ON CONFLICT(session_id, model) DO UPDATE SET
    input_tokens = input_tokens + excluded.input_tokens,
    output_tokens = output_tokens + excluded.output_tokens,
    invocation_count = invocation_count + 1
",
            )
            .bind(session_id)
            .bind(turn_metadata.model.as_str())
            .bind(turn_metadata.token_usage_delta.input_tokens.cast_signed())
            .bind(turn_metadata.token_usage_delta.output_tokens.cast_signed())
            .execute(&mut *transaction)
            .await?;
        }

        transaction.commit().await?;

        Ok(())
    }

    async fn replace_session_follow_up_tasks(
        &self,
        session_id: &str,
        follow_up_tasks: &[String],
    ) -> Result<(), DbError> {
        let mut transaction = self.0.begin().await?;

        let delete_result = sqlx::query(
            r"
DELETE FROM session_follow_up_task
WHERE session_id = ?
",
        )
        .bind(session_id)
        .execute(&mut *transaction)
        .await;
        match delete_result {
            Ok(_) => {}
            Err(error) if is_missing_follow_up_task_table(&error) => {
                transaction.rollback().await?;
                return Ok(());
            }
            Err(error) => {
                transaction.rollback().await?;
                return Err(error.into());
            }
        }

        for (position, follow_up_task) in follow_up_tasks.iter().enumerate() {
            sqlx::query(
                r"
INSERT INTO session_follow_up_task (session_id, position, text)
VALUES (?, ?, ?)
",
            )
            .bind(session_id)
            .bind(i64::try_from(position).unwrap_or(i64::MAX))
            .bind(follow_up_task)
            .execute(&mut *transaction)
            .await?;
        }

        transaction.commit().await?;

        Ok(())
    }

    async fn update_session_diff_stats(
        &self,
        added_lines: u64,
        deleted_lines: u64,
        id: &str,
        size: &str,
    ) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET added_lines = ?,
    deleted_lines = ?,
    size = ?
WHERE id = ?
  AND (
      added_lines <> ?
      OR deleted_lines <> ?
      OR size <> ?
  )
",
        )
        .bind(added_lines.cast_signed())
        .bind(deleted_lines.cast_signed())
        .bind(size)
        .bind(id)
        .bind(added_lines.cast_signed())
        .bind(deleted_lines.cast_signed())
        .bind(size)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_follow_up_task_launched_session_id(
        &self,
        session_id: &str,
        position: usize,
        launched_session_id: Option<String>,
    ) -> Result<(), DbError> {
        let update_result = sqlx::query(
            r"
UPDATE session_follow_up_task
SET launched_session_id = ?
WHERE session_id = ?
  AND position = ?
",
        )
        .bind(launched_session_id.as_deref())
        .bind(session_id)
        .bind(i64::try_from(position).unwrap_or(i64::MAX))
        .execute(&self.0)
        .await;
        match update_result {
            Ok(_) => {}
            Err(error) if is_missing_follow_up_task_table(&error) => return Ok(()),
            Err(error) => return Err(error.into()),
        }

        Ok(())
    }

    async fn update_session_instruction_conversation_id(
        &self,
        id: &str,
        provider_conversation_id: Option<String>,
    ) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET app_server_instruction_provider_conversation_id = ?
WHERE id = ?
",
        )
        .bind(provider_conversation_id.as_deref())
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_model(&self, id: &str, model: &str) -> Result<(), DbError> {
        let agent = persisted_agent_for_model(model);

        sqlx::query(
            r"
UPDATE session
SET agent = ?,
    model = ?
WHERE id = ?
",
        )
        .bind(agent)
        .bind(model)
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_agent_model(
        &self,
        id: &str,
        agent: &str,
        model: &str,
    ) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET agent = ?,
    model = ?
WHERE id = ?
",
        )
        .bind(agent)
        .bind(model)
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn clear_session_draft_flag(&self, id: &str) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET is_draft = 0
WHERE id = ?
",
        )
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_merged_commit_hash(
        &self,
        id: &str,
        merged_commit_hash: Option<String>,
    ) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET merged_commit_hash = ?
WHERE id = ?
",
        )
        .bind(merged_commit_hash.as_deref())
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_stack_base_commit_hash(
        &self,
        id: &str,
        stack_base_commit_hash: Option<String>,
    ) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET stack_base_commit_hash = ?,
    updated_at = CAST(strftime('%s', 'now') AS INTEGER)
WHERE id = ?
",
        )
        .bind(stack_base_commit_hash)
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_prompt(&self, id: &str, prompt: &str) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET prompt = ?
WHERE id = ?
",
        )
        .bind(prompt)
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_provider_conversation_id(
        &self,
        id: &str,
        provider_conversation_id: Option<String>,
    ) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET provider_conversation_id = ?
WHERE id = ?
",
        )
        .bind(provider_conversation_id.as_deref())
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_questions(&self, id: &str, questions: &str) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET questions = ?
WHERE id = ?
",
        )
        .bind(questions)
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_reasoning_level(
        &self,
        id: &str,
        reasoning_level: ReasoningLevel,
    ) -> Result<(), DbError> {
        sqlx::query!(
            r#"
UPDATE session
SET reasoning_level = ?
WHERE id = ?
            "#,
            reasoning_level.as_str(),
            id
        )
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_published_upstream_ref(
        &self,
        id: &str,
        published_upstream_ref: Option<String>,
    ) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET published_upstream_ref = ?
WHERE id = ?
",
        )
        .bind(published_upstream_ref.as_deref())
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_stats(&self, id: &str, stats: &SessionStats) -> Result<(), DbError> {
        if stats.input_tokens == 0 && stats.output_tokens == 0 {
            return Ok(());
        }

        sqlx::query(
            r"
UPDATE session
SET input_tokens = input_tokens + ?,
    output_tokens = output_tokens + ?
WHERE id = ?
",
        )
        .bind(stats.input_tokens.cast_signed())
        .bind(stats.output_tokens.cast_signed())
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_status_with_timing_at(
        &self,
        id: &str,
        status: &str,
        timestamp_seconds: i64,
    ) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET status = ?,
    in_progress_total_seconds = CASE
        WHEN ? = 'InProgress' OR in_progress_started_at IS NULL THEN in_progress_total_seconds
        ELSE in_progress_total_seconds + MAX(0, ? - in_progress_started_at)
    END,
    in_progress_started_at = CASE
        WHEN ? = 'InProgress' THEN COALESCE(in_progress_started_at, ?)
        ELSE NULL
    END
WHERE id = ?
",
        )
        .bind(status)
        .bind(status)
        .bind(timestamp_seconds)
        .bind(status)
        .bind(timestamp_seconds)
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_summary(&self, id: &str, summary: &str) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET summary = ?
WHERE id = ?
",
        )
        .bind(summary)
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_focused_review(
        &self,
        id: &str,
        diff_hash: Option<String>,
        text: Option<String>,
    ) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET focused_review_diff_hash = ?,
    focused_review_text = ?
WHERE id = ?
",
        )
        .bind(diff_hash.as_deref())
        .bind(text.as_deref())
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_title(&self, id: &str, title: &str) -> Result<(), DbError> {
        sqlx::query!(
            r#"
UPDATE session
SET title = ?
WHERE id = ?
"#,
            title,
            id,
        )
        .execute(&self.0)
        .await?;

        Ok(())
    }

    async fn update_session_title_for_prompt(
        &self,
        id: &str,
        expected_prompt: &str,
        title: &str,
    ) -> Result<bool, DbError> {
        let result = sqlx::query!(
            r#"
UPDATE session
SET title = ?
WHERE id = ?
  AND prompt = ?
"#,
            title,
            id,
            expected_prompt,
        )
        .execute(&self.0)
        .await?;

        Ok(result.rows_affected() > 0)
    }

    #[cfg(test)]
    async fn update_session_created_at(&self, id: &str, created_at: i64) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET created_at = ?
WHERE id = ?
",
        )
        .bind(created_at)
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }

    #[cfg(test)]
    async fn update_session_updated_at(&self, id: &str, updated_at: i64) -> Result<(), DbError> {
        sqlx::query(
            r"
UPDATE session
SET updated_at = ?
WHERE id = ?
",
        )
        .bind(updated_at)
        .bind(id)
        .execute(&self.0)
        .await?;

        Ok(())
    }
}

/// Borrowed values used to insert one newly created session row.
struct InsertSessionRow<'a> {
    /// Agent provider kind persisted alongside the model for this session.
    agent: &'a str,
    /// Base branch or parent branch used for future worktree materialization.
    base_branch: &'a str,
    /// Stable session identifier.
    id: &'a str,
    /// Whether the row was created through explicit draft staging.
    is_draft: bool,
    /// Agent model identifier persisted for the session.
    model: &'a str,
    /// Optional parent session id for one-level stacked drafts.
    parent_session_id: Option<&'a str>,
    /// Owning project identifier.
    project_id: i64,
    /// Reasoning level captured from the project default at creation.
    reasoning_level: ReasoningLevel,
    /// Initial lifecycle status string.
    status: &'a str,
}

/// Inserts one newly created session row with explicit draft-mode
/// persistence.
async fn insert_session_with_draft_mode(
    pool: &SqlitePool,
    row: InsertSessionRow<'_>,
) -> Result<(), DbError> {
    let InsertSessionRow {
        agent,
        base_branch,
        id,
        is_draft,
        model,
        parent_session_id,
        project_id,
        reasoning_level,
        status,
    } = row;

    sqlx::query(
        r"
INSERT INTO session (
    id,
    agent,
    model,
    base_branch,
    status,
    is_draft,
    parent_session_id,
    project_id,
    reasoning_level,
    prompt
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
",
    )
    .bind(id)
    .bind(agent)
    .bind(model)
    .bind(base_branch)
    .bind(status)
    .bind(is_draft)
    .bind(parent_session_id)
    .bind(project_id)
    .bind(reasoning_level.as_str())
    .bind("")
    .execute(pool)
    .await?;

    Ok(())
}

/// Returns the persisted agent value paired with a newly saved model string.
fn persisted_agent_for_model(model: &str) -> String {
    AgentModel::parse_persisted(model).map_or_else(
        |_| persisted_agent_for_unknown_model(model).to_string(),
        |agent_model| persisted_agent_for_known_model(model, agent_model).to_string(),
    )
}

/// Returns a compatibility agent value for known model strings passed through
/// model-only legacy persistence helpers.
fn persisted_agent_for_known_model(model: &str, agent_model: AgentModel) -> AgentKind {
    if model.starts_with("claude-") {
        return AgentKind::Claude;
    }

    if model.starts_with("gpt-") {
        return AgentKind::Codex;
    }

    if model.starts_with("gemini-") {
        return AgentKind::Antigravity;
    }

    AgentKind::ALL
        .iter()
        .copied()
        .find(|agent_kind| agent_kind.supports_model(agent_model))
        .unwrap_or(AgentKind::Antigravity)
}

/// Returns a compatibility agent value for tests or older callers that pass
/// model strings outside the current curated model set.
fn persisted_agent_for_unknown_model(model: &str) -> AgentKind {
    if model.starts_with("claude-") {
        return AgentKind::Claude;
    }

    if model.starts_with("gpt-") {
        return AgentKind::Codex;
    }

    if model.starts_with("gemini-") {
        return AgentKind::Antigravity;
    }

    AgentKind::Antigravity
}

/// Returns whether one `SQLx` error indicates the optional follow-up-task table
/// is not available in the current database.
fn is_missing_follow_up_task_table(error: &sqlx::Error) -> bool {
    matches!(
        error,
        sqlx::Error::Database(database_error)
            if database_error.message().contains("no such table: session_follow_up_task")
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::domain::session::{
        ForgeKind, ReviewRequest, ReviewRequestState, ReviewRequestSummary,
    };
    use crate::infra::db::AppRepositories;

    /// Session columns that must be reset when snapshotting a fork.
    #[derive(sqlx::FromRow)]
    struct ForkResetRow {
        app_server_instruction_provider_conversation_id: Option<String>,
        focused_review_diff_hash: Option<String>,
        focused_review_text: Option<String>,
        in_progress_started_at: Option<i64>,
        in_progress_total_seconds: i64,
        is_draft: bool,
        merged_commit_hash: Option<String>,
        parent_session_id: Option<String>,
        provider_conversation_id: Option<String>,
        published_upstream_ref: Option<String>,
        questions: Option<String>,
        stack_base_commit_hash: Option<String>,
    }

    impl SessionRowMetadata {
        /// Converts shared metadata fields into a full session row for the
        /// joined-session test mapping.
        fn into_session_row(
            self,
            prompt: String,
            questions: Option<String>,
            summary: Option<String>,
            review_request: Option<SessionReviewRequestRow>,
        ) -> SessionRow {
            SessionRow {
                added_lines: self.added_lines,
                agent: self.agent,
                base_branch: self.base_branch,
                created_at: self.created_at,
                deleted_lines: self.deleted_lines,
                id: self.id,
                in_progress_started_at: self.in_progress_started_at,
                in_progress_total_seconds: self.in_progress_total_seconds,
                input_tokens: self.input_tokens,
                is_draft: self.is_draft,
                model: self.model,
                output_tokens: self.output_tokens,
                parent_session_id: self.parent_session_id,
                project_id: self.project_id,
                prompt,
                published_upstream_ref: self.published_upstream_ref,
                questions,
                reasoning_level_override: self.reasoning_level_override,
                review_request,
                size: self.size,
                status: self.status,
                summary,
                title: self.title,
                updated_at: self.updated_at,
            }
        }
    }

    /// Row returned when loading one `session` plus aliased
    /// `session_review_request` join columns.
    #[derive(sqlx::FromRow)]
    pub(super) struct SessionJoinRow {
        #[sqlx(flatten)]
        metadata: SessionRowMetadata,
        prompt: String,
        questions: Option<String>,
        review_request_display_id: Option<String>,
        review_request_forge_kind: Option<String>,
        review_request_last_refreshed_at: Option<i64>,
        review_request_source_branch: Option<String>,
        review_request_state: Option<String>,
        review_request_status_summary: Option<String>,
        review_request_target_branch: Option<String>,
        review_request_title: Option<String>,
        review_request_web_url: Option<String>,
        summary: Option<String>,
    }

    impl SessionJoinRow {
        /// Converts the query-mapped join row into the public [`SessionRow`]
        /// model.
        pub(super) fn into_session_row(self) -> SessionRow {
            let Self {
                metadata,
                prompt,
                questions,
                review_request_display_id,
                review_request_forge_kind,
                review_request_last_refreshed_at,
                review_request_source_branch,
                review_request_state,
                review_request_status_summary,
                review_request_target_branch,
                review_request_title,
                review_request_web_url,
                summary,
            } = self;

            let review_request = SessionReviewRequestJoinRow {
                display_id: review_request_display_id,
                forge_kind: review_request_forge_kind,
                last_refreshed_at: review_request_last_refreshed_at,
                source_branch: review_request_source_branch,
                state: review_request_state,
                status_summary: review_request_status_summary,
                target_branch: review_request_target_branch,
                title: review_request_title,
                web_url: review_request_web_url,
            }
            .into_review_request_row();

            metadata.into_session_row(prompt, questions, summary, review_request)
        }

        /// Builds a deterministic joined-session row fixture for conversion
        /// tests.
        fn fixture_for_test() -> Self {
            Self {
                metadata: SessionRowMetadata {
                    added_lines: 14,
                    agent: "codex".to_string(),
                    base_branch: "main".to_string(),
                    created_at: 100,
                    deleted_lines: 6,
                    id: "session-a".to_string(),
                    in_progress_started_at: None,
                    in_progress_total_seconds: 0,
                    input_tokens: 11,
                    is_draft: false,
                    model: "gpt-5.5".to_string(),
                    output_tokens: 29,
                    parent_session_id: Some("parent-session".to_string()),
                    project_id: Some(7),
                    published_upstream_ref: Some("origin/session-a".to_string()),
                    reasoning_level_override: None,
                    size: "M".to_string(),
                    status: "Review".to_string(),
                    title: Some("Review session".to_string()),
                    updated_at: 200,
                },
                prompt: "Implement feature".to_string(),
                questions: Some("Question text".to_string()),
                review_request_display_id: Some("#42".to_string()),
                review_request_forge_kind: Some("GitHub".to_string()),
                review_request_last_refreshed_at: Some(456),
                review_request_source_branch: Some("feature/forge".to_string()),
                review_request_state: Some("Open".to_string()),
                review_request_status_summary: Some("2 approvals, checks passing".to_string()),
                review_request_target_branch: Some("main".to_string()),
                review_request_title: Some("Add forge review support".to_string()),
                review_request_web_url: Some(
                    "https://github.com/agentty-xyz/agentty/pull/42".to_string(),
                ),
                summary: Some("Summary text".to_string()),
            }
        }
    }

    /// Builds the fully populated review-request row expected by join-row
    /// conversion tests.
    fn expected_review_request_row() -> SessionReviewRequestRow {
        SessionReviewRequestRow {
            display_id: "#42".to_string(),
            forge_kind: "GitHub".to_string(),
            last_refreshed_at: 456,
            source_branch: "feature/forge".to_string(),
            state: "Open".to_string(),
            status_summary: Some("2 approvals, checks passing".to_string()),
            target_branch: "main".to_string(),
            title: "Add forge review support".to_string(),
            web_url: "https://github.com/agentty-xyz/agentty/pull/42".to_string(),
        }
    }

    /// Builds the review-request domain fixture used by fork snapshot tests.
    fn review_request_fixture() -> ReviewRequest {
        ReviewRequest {
            last_refreshed_at: 456,
            summary: ReviewRequestSummary {
                display_id: "#42".to_string(),
                forge_kind: ForgeKind::GitHub,
                source_branch: "feature/forge".to_string(),
                state: ReviewRequestState::Open,
                status_summary: Some("2 approvals, checks passing".to_string()),
                target_branch: "main".to_string(),
                title: "Add forge review support".to_string(),
                web_url: "https://github.com/agentty-xyz/agentty/pull/42".to_string(),
            },
        }
    }

    /// Loads reset-sensitive fork columns that are not exposed by public
    /// session row projections.
    async fn load_fork_reset_row(pool: &SqlitePool, session_id: &str) -> ForkResetRow {
        sqlx::query_as::<_, ForkResetRow>(
            r"
SELECT app_server_instruction_provider_conversation_id,
       focused_review_diff_hash,
       focused_review_text,
       in_progress_started_at,
       in_progress_total_seconds,
       is_draft,
       merged_commit_hash,
       parent_session_id,
       provider_conversation_id,
       published_upstream_ref,
       questions,
       stack_base_commit_hash
FROM session
WHERE id = ?
",
        )
        .bind(session_id)
        .fetch_one(pool)
        .await
        .expect("failed to load fork reset row")
    }

    /// Seeds a forkable source session with every source-only field that the
    /// snapshot insert is expected to clear.
    async fn seed_fork_snapshot_source(
        database: &AppRepositories,
        pool: &SqlitePool,
    ) -> (ForkResetRow, Option<SessionReviewRequestRow>) {
        let project_id = database
            .projects()
            .upsert_project("/tmp/project", None)
            .await
            .expect("failed to upsert project");
        database
            .sessions()
            .insert_session("parent-session", "gpt-5.5", "main", "Review", project_id)
            .await
            .expect("failed to insert parent session");
        database
            .sessions()
            .insert_stacked_draft_session(
                "source-session",
                "gpt-5.5",
                "wt/parent",
                "Review",
                "parent-session",
                project_id,
            )
            .await
            .expect("failed to insert source session");

        seed_fork_snapshot_source_linkage(database).await;
        seed_fork_snapshot_source_timing(database, pool).await;

        let source_reset_row = load_fork_reset_row(pool, "source-session").await;
        let source_review_request = database
            .reviews()
            .load_session_review_request("source-session")
            .await
            .expect("failed to load source review request");

        (source_reset_row, source_review_request)
    }

    /// Persists source-only linkage and counters on the fork source row.
    async fn seed_fork_snapshot_source_linkage(database: &AppRepositories) {
        database
            .sessions()
            .update_session_provider_conversation_id(
                "source-session",
                Some("provider-thread".to_string()),
            )
            .await
            .expect("failed to update provider conversation id");
        database
            .sessions()
            .update_session_instruction_conversation_id(
                "source-session",
                Some("instruction-thread".to_string()),
            )
            .await
            .expect("failed to update instruction conversation id");
        database
            .sessions()
            .update_session_questions("source-session", r#"["Need detail?"]"#)
            .await
            .expect("failed to update questions");
        database
            .sessions()
            .update_session_published_upstream_ref(
                "source-session",
                Some("origin/wt/source-session".to_string()),
            )
            .await
            .expect("failed to update published upstream ref");
        database
            .sessions()
            .update_session_merged_commit_hash("source-session", Some("merged123".to_string()))
            .await
            .expect("failed to update merged commit hash");
        database
            .sessions()
            .update_session_focused_review(
                "source-session",
                Some("diff123".to_string()),
                Some("Focused review text".to_string()),
            )
            .await
            .expect("failed to update focused review");
        database
            .sessions()
            .update_session_stack_base_commit_hash(
                "source-session",
                Some("stackbase123".to_string()),
            )
            .await
            .expect("failed to update stack base commit hash");
        database
            .sessions()
            .update_session_stats(
                "source-session",
                &SessionStats {
                    added_lines: 0,
                    deleted_lines: 0,
                    input_tokens: 11,
                    output_tokens: 29,
                },
            )
            .await
            .expect("failed to update token stats");
        database
            .reviews()
            .update_session_review_request("source-session", Some(review_request_fixture()))
            .await
            .expect("failed to update review request");
    }

    /// Persists active-work timing fields on the fork source row.
    async fn seed_fork_snapshot_source_timing(database: &AppRepositories, pool: &SqlitePool) {
        database
            .sessions()
            .update_session_status_with_timing_at("source-session", "InProgress", 100)
            .await
            .expect("failed to open timing interval");
        sqlx::query(
            r"
UPDATE session
SET in_progress_total_seconds = ?
WHERE id = ?
",
        )
        .bind(75_i64)
        .bind("source-session")
        .execute(pool)
        .await
        .expect("failed to seed elapsed timing");
    }

    /// Asserts the fixture source row actually had source-only state before
    /// the snapshot was taken.
    fn assert_source_reset_state(
        source_reset_row: &ForkResetRow,
        source_review_request: Option<&SessionReviewRequestRow>,
    ) {
        assert!(source_reset_row.is_draft);
        assert_eq!(
            source_reset_row.parent_session_id.as_deref(),
            Some("parent-session")
        );
        assert_eq!(
            source_reset_row.provider_conversation_id.as_deref(),
            Some("provider-thread")
        );
        assert_eq!(
            source_reset_row
                .app_server_instruction_provider_conversation_id
                .as_deref(),
            Some("instruction-thread")
        );
        assert_eq!(
            source_reset_row.published_upstream_ref.as_deref(),
            Some("origin/wt/source-session")
        );
        assert_eq!(
            source_reset_row.questions.as_deref(),
            Some(r#"["Need detail?"]"#)
        );
        assert_eq!(
            source_reset_row.merged_commit_hash.as_deref(),
            Some("merged123")
        );
        assert_eq!(
            source_reset_row.focused_review_diff_hash.as_deref(),
            Some("diff123")
        );
        assert_eq!(
            source_reset_row.focused_review_text.as_deref(),
            Some("Focused review text")
        );
        assert_eq!(
            source_reset_row.stack_base_commit_hash.as_deref(),
            Some("stackbase123")
        );
        assert_eq!(source_reset_row.in_progress_started_at, Some(100));
        assert_eq!(source_reset_row.in_progress_total_seconds, 75);
        assert_eq!(
            source_review_request.map(|review_request| review_request.display_id.as_str()),
            Some("#42")
        );
    }

    /// Asserts the forked row kept durable snapshot state while clearing
    /// source-only linkage.
    fn assert_fork_reset_state(
        fork_row: &SessionRow,
        fork_reset_row: &ForkResetRow,
        fork_review_request: Option<&SessionReviewRequestRow>,
    ) {
        assert_eq!(fork_row.status, "Review");
        assert!(!fork_row.is_draft);
        assert_eq!(fork_row.parent_session_id, None);
        assert_eq!(fork_row.input_tokens, 0);
        assert_eq!(fork_row.output_tokens, 0);
        assert_eq!(fork_row.questions, None);
        assert_eq!(fork_row.published_upstream_ref, None);
        assert_eq!(fork_row.review_request, None);
        assert_eq!(fork_reset_row.provider_conversation_id, None);
        assert_eq!(
            fork_reset_row.app_server_instruction_provider_conversation_id,
            None
        );
        assert_eq!(fork_reset_row.merged_commit_hash, None);
        assert_eq!(fork_reset_row.focused_review_diff_hash, None);
        assert_eq!(fork_reset_row.focused_review_text, None);
        assert_eq!(fork_reset_row.questions, None);
        assert_eq!(fork_reset_row.stack_base_commit_hash, None);
        assert_eq!(fork_reset_row.in_progress_started_at, None);
        assert_eq!(fork_reset_row.in_progress_total_seconds, 0);
        assert_eq!(fork_review_request, None);
    }

    #[tokio::test]
    async fn test_load_sessions_uses_created_at_to_break_updated_at_ties() {
        // Arrange
        let (database, pool) = AppRepositories::in_memory_with_pool().await;
        let project_id = database
            .projects()
            .upsert_project("/tmp/project", None)
            .await
            .expect("failed to upsert project");
        for session_id in ["a-older", "z-newer"] {
            database
                .sessions()
                .insert_session(session_id, "gpt-5.5", "main", "Review", project_id)
                .await
                .expect("failed to insert session");
        }
        sqlx::query(
            r"
UPDATE session
SET created_at = CASE id WHEN 'a-older' THEN 100 ELSE 200 END,
    updated_at = 300
WHERE id IN ('a-older', 'z-newer')
",
        )
        .execute(&pool)
        .await
        .expect("failed to set session timestamps");

        // Act
        let all_session_ids = database
            .sessions()
            .load_sessions()
            .await
            .expect("failed to load sessions")
            .into_iter()
            .map(|session| session.id)
            .collect::<Vec<_>>();
        let project_session_ids = database
            .sessions()
            .load_sessions_for_project(project_id)
            .await
            .expect("failed to load project sessions")
            .into_iter()
            .map(|session| session.id)
            .collect::<Vec<_>>();

        // Assert
        assert_eq!(all_session_ids, ["z-newer", "a-older"]);
        assert_eq!(project_session_ids, ["z-newer", "a-older"]);
    }

    #[tokio::test]
    async fn test_fork_session_snapshot_resets_source_specific_state() {
        // Arrange
        let (database, pool) = AppRepositories::in_memory_with_pool().await;
        let (source_reset_row, source_review_request) =
            seed_fork_snapshot_source(&database, &pool).await;

        // Act
        database
            .sessions()
            .fork_session_snapshot(ForkSessionSnapshot {
                new_session_id: "fork-session",
                source_session_id: "source-session",
                status: "Review",
            })
            .await
            .expect("failed to fork session snapshot");

        // Assert
        let fork_row = database
            .sessions()
            .load_sessions()
            .await
            .expect("failed to load sessions")
            .into_iter()
            .find(|session_row| session_row.id == "fork-session")
            .expect("missing forked session row");
        let fork_reset_row = load_fork_reset_row(&pool, "fork-session").await;
        let fork_review_request = database
            .reviews()
            .load_session_review_request("fork-session")
            .await
            .expect("failed to load fork review request");

        assert_source_reset_state(&source_reset_row, source_review_request.as_ref());
        assert_fork_reset_state(&fork_row, &fork_reset_row, fork_review_request.as_ref());
    }

    #[tokio::test]
    async fn test_clear_session_draft_flag_marks_draft_session_live() {
        // Arrange
        let (database, _pool) = AppRepositories::in_memory_with_pool().await;
        let project_id = database
            .projects()
            .upsert_project("/tmp/project", None)
            .await
            .expect("failed to upsert project");
        database
            .sessions()
            .insert_draft_session("draft-session", "gpt-5.5", "main", "Draft", project_id)
            .await
            .expect("failed to insert draft session");

        // Act
        database
            .sessions()
            .clear_session_draft_flag("draft-session")
            .await
            .expect("failed to clear session draft flag");

        // Assert
        let session_row = database
            .sessions()
            .load_sessions()
            .await
            .expect("failed to load sessions")
            .into_iter()
            .find(|session_row| session_row.id == "draft-session")
            .expect("missing draft session row");
        assert!(!session_row.is_draft);
    }

    /// Verifies `SessionJoinRow::into_session_row()` drops partially
    /// populated review-request columns instead of surfacing an invalid row
    /// model.
    #[test]
    fn test_session_join_row_ignores_partial_review_request_columns() {
        // Arrange
        let mut session_join_row = SessionJoinRow::fixture_for_test();
        session_join_row.review_request_last_refreshed_at = None;

        // Act
        let session_row = session_join_row.into_session_row();

        // Assert
        assert_eq!(session_row.id, "session-a");
        assert_eq!(session_row.project_id, Some(7));
        assert_eq!(
            session_row.parent_session_id.as_deref(),
            Some("parent-session")
        );
        assert_eq!(session_row.status, "Review");
        assert_eq!(session_row.added_lines, 14);
        assert_eq!(session_row.deleted_lines, 6);
        assert_eq!(session_row.review_request, None);
    }

    /// Verifies `SessionJoinRow::into_session_row()` maps a fully populated
    /// review-request into the public session row model.
    #[test]
    fn test_session_join_row_maps_review_request_columns() {
        // Arrange
        let session_join_row = SessionJoinRow::fixture_for_test();

        // Act
        let session_row = session_join_row.into_session_row();

        // Assert
        assert_eq!(session_row.id, "session-a");
        assert_eq!(session_row.added_lines, 14);
        assert_eq!(session_row.deleted_lines, 6);
        assert_eq!(session_row.project_id, Some(7));
        assert_eq!(
            session_row.parent_session_id.as_deref(),
            Some("parent-session")
        );
        assert_eq!(
            session_row.published_upstream_ref.as_deref(),
            Some("origin/session-a")
        );
        assert_eq!(session_row.questions.as_deref(), Some("Question text"));
        assert_eq!(session_row.summary.as_deref(), Some("Summary text"));
        assert_eq!(session_row.title.as_deref(), Some("Review session"));
        assert_eq!(
            session_row.review_request,
            Some(expected_review_request_row())
        );
    }

    #[test]
    fn test_session_follow_up_task_row_converts_to_domain_task() {
        // Arrange
        let row = SessionFollowUpTaskRow {
            id: 7,
            launched_session_id: Some("launched-session".to_string()),
            position: 3,
            session_id: "source-session".to_string(),
            text: "Follow up on coverage".to_string(),
        };

        // Act
        let follow_up_task = row.into_session_follow_up_task();

        // Assert
        assert_eq!(follow_up_task.id, 7);
        assert_eq!(
            follow_up_task.launched_session_id,
            Some(SessionId::from("launched-session"))
        );
        assert_eq!(follow_up_task.position, 3);
        assert_eq!(follow_up_task.text, "Follow up on coverage");
    }

    #[test]
    fn test_session_follow_up_task_row_clamps_invalid_position() {
        // Arrange
        let row = SessionFollowUpTaskRow {
            id: 8,
            launched_session_id: None,
            position: -1,
            session_id: "source-session".to_string(),
            text: "Handle invalid position".to_string(),
        };

        // Act
        let follow_up_task = row.into_session_follow_up_task();

        // Assert
        assert_eq!(follow_up_task.launched_session_id, None);
        assert_eq!(follow_up_task.position, usize::MAX);
    }
}