mahbot 0.7.3

//! `Session` — single source of truth for conversation persistence.
//!
//! Owns the in-memory history Vec for the current turn. The orchestrator
//! calls [`init`](Session::init) to load/build history, then passes `&mut Session` to the agent loop.
//!
//! ## Usage
//! 1. `Session::default()` at turn start
//! 2. `session.init(agent_id)` — loads history and the token length, stores
//!    history internally; the turn message is appended separately via
//!    `append_turn_message` after the universal resume-completion step
//! 3. Agent loop calls `session.push_assistant()`, `session.persist_messages()`,
//!    `session.push_messages_unpersisted()`, etc. during tool rounds
//! 4. `session.finalize(agent_id)` on success — persists the final assistant response
//!    and reports a [`FinalizeOutcome`]; an empty unpersisted tail is a no-op
//!    (everything already durable) and is surfaced to the caller to log

use std::fmt::Write;
use std::path::Path;
use std::sync::Arc;

use anyhow::Result;
use arc_swap::ArcSwap;
use futures_util::future::join_all;
use ignore::WalkBuilder;

use crate::pipeline::board::BOARD;
use crate::prompt::{build_workspace_context, format_ticket_block, load_prompt, substitute};
use crate::runtime_events::RuntimeEvent;
use crate::workspace::{WORKSPACES, truncate_workspace_notes};

use crate::agent::skills;
use crate::alarms::Alarm;
use crate::tools::active_models::{ModelKind, ModelSnapshot};
use crate::{ChatMessage, ChatRole, Role, Workspace};

use super::TranscriptSnapshot;

/// Coordinates session persistence and prompt building across a single
/// agent turn.
#[derive(Default)]
pub(crate) struct Session {
    /// In-memory history for the current turn (includes system prompt,
    /// conversation, and the latest user message).
    history: Vec<ChatMessage>,
    /// Number of leading `history` entries already persisted to the store;
    /// `history[..persisted_len]` mirrors the DB contents contiguously and
    /// `history[persisted_len..]` is exactly the unpersisted tail (failed
    /// appends, tool traffic awaiting commit, the final answer). Every
    /// successful persist advances this prefix over the span it wrote.
    persisted_len: usize,
    /// Real provider-reported session length (input + output tokens of the
    /// most recent successful agent LLM call), loaded at [`init`](Self::init)
    /// and updated after every successful agent-purpose LLM call. `None` for
    /// sessions that never recorded a value (new sessions, pre-migration
    /// sessions — approved no-backfill semantics).
    token_length: Option<u64>,
    /// Shared lock-free holder this session publishes live transcript
    /// snapshots into. `None` when the session is not attached to a running
    /// agent (tests, detached sessions). Set by
    /// [`attach_transcript`](Self::attach_transcript) from `Agent::new`.
    transcript: Option<Arc<ArcSwap<TranscriptSnapshot>>>,
}

/// Outcome of a [`Session::finalize`] call.
///
/// The finalize guard is deliberately logging-free: whether an empty
/// unpersisted tail is expected (graceful drain cut right after the last
/// tool-group commit) or anomalous (LLM failure with no output, iteration
/// limit) depends on shutdown state this module does not track. The caller
/// inspects the outcome and logs with full agent/role/workspace/ticket
/// attribution.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum FinalizeOutcome {
    /// The unpersisted tail (final assistant answer / delivered comments)
    /// was flushed to the store and the persisted prefix advanced.
    Flushed,
    /// Nothing was unpersisted — the turn ended with an empty tail. All
    /// committed state is already durable; the caller decides whether the
    /// no-op is expected (drain) or a genuine anomaly.
    NoUnpersistedTail,
}

/// Outcome of [`Session::rewrite_last_user_message`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RewriteOutcome {
    /// The most recent User-role message was durably rewritten and the
    /// in-memory history entry swapped.
    Rewritten,
    /// The most recent User-role message is in the unpersisted tail
    /// (index >= `persisted_len`): the positional store UPDATE would target a
    /// different row, so the rewrite is skipped (conservative no-op — the
    /// caller keeps the original error path).
    UnpersistedTailNoop,
}

/// The unresolved tool calls of the session's LAST tool-call frame plus the
/// frame-level context the sleep-delivery guard needs on the resume path:
/// whether the frame carried non-empty raw text (which is never delivered
/// in a tool round) and how many calls the round bundled.
pub(crate) struct PendingToolFrame {
    pub calls: Vec<crate::ToolCall>,
    pub has_text: bool,
    pub call_count: usize,
}

impl Session {
    // ── Session lifecycle ──────────────────────────────────────────────

    /// Begin a turn. Loads the persisted history and the real
    /// provider-reported session length (last successful agent LLM call's
    /// input + output tokens), records the persisted prefix, and publishes the
    /// initial transcript snapshot. The turn's user message — and the
    /// first-turn system-prompt caching that goes with it — is handled
    /// separately via
    /// [`append_turn_message`](Self::append_turn_message); the orchestrator
    /// runs the universal resume-completion step between the two so any
    /// dangling tool calls are settled BEFORE the new message lands.
    pub(crate) async fn init(&mut self, agent_id: &str) -> Result<()> {
        self.history = crate::session::store().load(agent_id).await;

        // Load the real provider-reported session length (last successful
        // agent LLM call's input + output tokens) so `maybe_summarize` and
        // the Running Agents card see it from the start of the turn. `None`
        // for sessions that never recorded a value — treated as below the
        // summarization threshold.
        self.token_length = crate::session::store().get_token_length(agent_id).await;

        // All loaded history above is persisted — record the prefix so
        // finalize only appends genuinely new assistant output.
        self.persisted_len = self.history.len();

        // Publish the initial live snapshot so the Running Agents card reflects
        // the loaded/unpersisted transcript and the persisted token length from
        // the very start of the turn.
        self.publish_transcript();

        Ok(())
    }

    /// Append this turn's user message and session context into the loaded
    /// session. Moves the message-append half of [`init`](Self::init) here so
    /// the orchestrator can run the universal resume-completion step in between
    /// (see `Agent::work`): the session loads WITHOUT the new message, the
    /// dangling tool calls are settled directly after their frame, and only then
    /// does the user message land. An empty `msg` is a no-op (RecoveryRetry
    /// re-triggers against the existing history without adding a turn).
    ///
    /// `round_ts` pins the timestamp of the appended user message (one value
    /// per round for byte-identical first messages across parallel members);
    /// `None` stamps now.
    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn append_turn_message(
        &mut self,
        agent_id: &str,
        msg: &str,
        ws: &Workspace,
        role: &Role,
        ticket: Option<&crate::pipeline::board::Ticket>,
        channel: &str,
        user_name: &str,
        round_ts: Option<&str>,
        is_admin: bool,
    ) -> Result<()> {
        if !msg.is_empty() {
            let is_new = self.history.is_empty();

            if is_new {
                let (msgs, snapshot) =
                    Self::build_turn_messages(msg, ws, role, ticket, round_ts, is_admin, user_name)
                        .await;

                // Batch-write all messages + session context atomically.
                crate::session::store()
                    .batch_append_with_context(
                        agent_id,
                        &msgs,
                        channel,
                        user_name,
                        &ws.name,
                        role.as_str(),
                    )
                    .await?;
                self.history.extend(msgs);

                if matches!(role, Role::Assistant) {
                    Self::persist_active_models_snapshot(agent_id, &snapshot).await;
                }
            } else {
                // Caching: system prompt is NOT rebuilt on subsequent turns
                // (see doc comment above). The session DB caches the full
                // first-turn message set. The only rebuild path is
                // `apply_summary` below.
                let (content, new_snapshot) = if matches!(role, Role::Assistant) {
                    Self::prepend_model_change(agent_id, msg, user_name).await
                } else {
                    (msg.to_string(), None)
                };
                let user_msg = crate::session::user_msg_with_ts(&content, round_ts);
                // Append the user message and update session context atomically.
                crate::session::store()
                    .append_with_context(
                        agent_id,
                        &user_msg,
                        channel,
                        user_name,
                        &ws.name,
                        role.as_str(),
                    )
                    .await?;
                // Refresh the change-detection baseline only after the message
                // (carrying the change-info) is persisted — otherwise a failed
                // append would lose the change-info while advancing the
                // baseline, so a retry could never communicate the switch.
                // Residual window: an append that succeeds while the baseline
                // write fails leaves a stale baseline, so the next message
                // re-fires the same change-info — benign and self-healing.
                if let Some(snapshot) = new_snapshot {
                    Self::persist_active_models_snapshot(agent_id, &snapshot).await;
                }
                self.history.push(user_msg);
            }
            // The appended message is durably written above — advance the
            // persisted prefix and refresh the live transcript.
            self.persisted_len = self.history.len();
            self.publish_transcript();
        }
        Ok(())
    }

    // ── Live transcript snapshot ─────────────────────────────────────

    /// Attach the shared lock-free holder this session publishes live
    /// transcript snapshots into. Called by `Agent::new` after registering the
    /// holder in the global accessor; must run before the first mutation.
    pub(crate) fn attach_transcript(&mut self, holder: Arc<ArcSwap<TranscriptSnapshot>>) {
        self.transcript = Some(holder);
    }

    /// Publish a fresh read-only snapshot of the current history + token count
    /// into the holder. Copy-on-write: clones the history Vec (the one
    /// per-publish cost) and stores the new snapshot. No-op when no holder is
    /// attached (detached sessions / tests).
    fn publish_transcript(&self) {
        if let Some(holder) = &self.transcript {
            let snapshot = TranscriptSnapshot {
                history: self.history.clone(),
                token_count: self.token_length,
            };
            holder.store(Arc::new(snapshot));
            // Live transcript content changed → the Running Agents cards /
            // footer need a re-render.
            crate::runtime_events::publish(RuntimeEvent::Registries);
        }
    }

    // ── History access (for agent loop) ────────────────────────────────

    /// Read-only access to the in-memory history (for LLM calls).
    #[must_use]
    pub(crate) fn history(&self) -> &[ChatMessage] {
        &self.history
    }

    /// Real provider-reported session length (see the field doc), loaded at
    /// [`init`](Self::init). `None` when no successful usage-bearing agent
    /// call ever recorded a value.
    #[must_use]
    pub(crate) fn token_length(&self) -> Option<u64> {
        self.token_length
    }

    /// Update the in-memory session length after a successful agent LLM call.
    /// The durable store write is the caller's responsibility
    /// ([`crate::Agent::record_session_usage`]).
    pub(crate) fn set_token_length(&mut self, token_length: Option<u64>) {
        self.token_length = token_length;
        self.publish_transcript();
    }

    /// Append an assistant message (final response or intermediate tool-call).
    /// Operates on in-memory history only, not the session store.
    pub(crate) fn push_assistant(&mut self, content: String) {
        self.history.push(ChatMessage::assistant(content));
        self.publish_transcript();
    }

    /// Persist any unpersisted history tail plus `messages` in one
    /// transaction, then deliver `messages` to in-memory history and advance
    /// `persisted_len` over the whole persisted span. On failure nothing is
    /// delivered — the caller aborts or delivers unpersisted via
    /// [`Session::push_messages_unpersisted`].
    ///
    /// `messages` must not already be part of the unpersisted tail — they
    /// would be written twice.
    pub(crate) async fn persist_messages(
        &mut self,
        agent_id: &str,
        messages: &[ChatMessage],
    ) -> Result<()> {
        debug_assert!(self.persisted_len <= self.history.len());
        let mut batch =
            Vec::with_capacity(self.history.len() - self.persisted_len + messages.len());
        batch.extend_from_slice(&self.history[self.persisted_len..]);
        batch.extend_from_slice(messages);
        crate::session::store()
            .batch_append(agent_id, &batch)
            .await?;
        self.persisted_len += batch.len();
        self.history.extend_from_slice(messages);
        self.publish_transcript();
        Ok(())
    }

    /// The tool calls of the session's LAST tool-call frame whose results are
    /// not yet present — the universal resume-completion input.
    ///
    /// Walks back to the last assistant tool-call frame, then forward over the
    /// immediately-following Tool-role entries, collecting the `tool_call_id`s
    /// they resolved. Returns the frame's calls whose ids are missing: `None`
    /// when nothing is missing, or when there is no (undecodable) frame.
    pub(crate) fn pending_tool_frame(&self) -> Option<PendingToolFrame> {
        let frame_idx = self.history.iter().rposition(super::is_tool_call_frame)?;
        let Some(super::DecodedNativeHistoryMessage::Assistant {
            content,
            tool_calls: Some(calls),
            ..
        }) = super::decode_native_history_message(&self.history[frame_idx])
        else {
            return None;
        };
        let has_text = content.as_deref().is_some_and(|t| !t.trim().is_empty());
        let call_count = calls.len();
        // Walk forward over Tool-role entries (stop at the first non-tool).
        let mut completed: std::collections::HashSet<String> = std::collections::HashSet::default();
        for msg in &self.history[frame_idx + 1..] {
            if msg.role == ChatRole::Tool {
                if let Some(super::DecodedNativeHistoryMessage::ToolResult {
                    tool_call_id, ..
                }) = super::decode_native_history_message(msg)
                {
                    completed.insert(tool_call_id);
                }
            } else {
                break;
            }
        }
        let pending: Vec<crate::ToolCall> = calls
            .into_iter()
            .filter(|c| !completed.contains(&c.id))
            .collect();
        if pending.is_empty() {
            None
        } else {
            Some(PendingToolFrame {
                calls: pending,
                has_text,
                call_count,
            })
        }
    }

    /// Test-only thin wrapper: run the store's self-transactional settle and
    /// mirror the rewrite into the in-memory history. Production invokes
    /// [`SessionStore::settle_tool_results_tx`](super::SessionStore::settle_tool_results_tx)
    /// directly inside the agent's atomic settle+terminalize transaction and
    /// mirrors via [`Self::mirror_settled`] only after the commit.
    #[cfg(test)]
    pub(crate) async fn settle_tool_results(
        &mut self,
        agent_id: &str,
        results: &[(String, String)],
        follow_up: &[ChatMessage],
    ) -> Result<()> {
        let settled = crate::session::store()
            .settle_tool_results(agent_id, results, follow_up)
            .await?;
        self.mirror_settled(settled)?;
        Ok(())
    }

    /// Mirror a store settle's returned sequence into the in-memory history:
    /// truncate after the tool-call frame and replay the final sequence
    /// verbatim (role via the strum parse path, content as-is).
    pub(crate) fn mirror_settled(&mut self, settled: Vec<super::SettledRow>) -> Result<()> {
        let Some(frame_idx) = self.history.iter().rposition(super::is_tool_call_frame) else {
            debug_assert!(
                false,
                "settle_tool_results called without a tool-call frame in history"
            );
            return Ok(());
        };
        self.history.truncate(frame_idx + 1);
        for row in settled {
            let role = row.role.parse::<ChatRole>().map_err(|e| {
                anyhow::anyhow!("settled session role {} is invalid: {e}", row.role)
            })?;
            self.history.push(ChatMessage {
                role,
                content: row.content,
            });
        }
        self.persisted_len = self.history.len();
        self.publish_transcript();
        Ok(())
    }

    /// Deliver `messages` to in-memory history without persisting them
    /// (store append failed). They remain in the unpersisted tail; the next
    /// successful persist flushes them in order.
    pub(crate) fn push_messages_unpersisted(&mut self, messages: &[ChatMessage]) {
        self.history.extend_from_slice(messages);
        self.publish_transcript();
    }

    /// Durable rewrite of the most recent User-role message: persist the new
    /// content to the store FIRST, then swap the in-memory history entry.
    ///
    /// Ordering invariant: the store write happens before the in-memory swap,
    /// so a persist failure leaves the in-memory history untouched and the
    /// caller can fall back to the original error path without a half-stripped
    /// continuation or a swallowed error.
    ///
    /// The store UPDATE targets the last `user`-role row, which corresponds to
    /// the most recent User-role message **only while that message is within
    /// the persisted prefix** (`index < persisted_len`) — `history[..persisted_len]`
    /// mirrors the DB contiguously. A message in the unpersisted tail returns
    /// [`RewriteOutcome::UnpersistedTailNoop`] instead of corrupting an earlier
    /// row. The target message is re-derived here (authoritative), not taken
    /// from the caller.
    pub(crate) async fn rewrite_last_user_message(
        &mut self,
        agent_id: &str,
        content: String,
    ) -> Result<RewriteOutcome> {
        debug_assert!(self.persisted_len <= self.history.len());
        let Some(idx) = self.history.iter().rposition(|m| m.role == ChatRole::User) else {
            anyhow::bail!("no user message in history to rewrite");
        };
        if idx >= self.persisted_len {
            return Ok(RewriteOutcome::UnpersistedTailNoop);
        }
        crate::session::store()
            .rewrite_last_user_message(agent_id, &content)
            .await?;
        self.history[idx].content = content;
        self.publish_transcript();
        Ok(RewriteOutcome::Rewritten)
    }

    // ── Summarization — apply summary produced by Agent ─────────────────

    /// Replace the in-memory and persisted history with a compacted version
    /// containing a fresh system prompt (via `build_context_messages`), the
    /// given `summary_text`, and the latest [`super::RETENTION_PER_SIDE`]
    /// user messages + assistant answers from the pre-compaction history
    /// (tool traffic excluded — see `select_retention_window`).
    ///
    /// Before persisting, the messages this compaction deletes are dumped to
    /// a file via [`super::compaction_dump`] and the dump path is appended to
    /// the summary message, so the agent can re-read dropped content with the
    /// read tool. A dump failure is fail-open (no path line).
    ///
    /// The LLM call to produce the summary text is the responsibility of
    /// [`crate::Agent::summarize`] — this method
    /// handles only the history rebuild, dump, and persistence.
    ///
    /// KV-cache preservation: called by Agent after producing a summary
    /// with byte-identical parameters (model, reasoning_effort, tools,
    /// provider routing) so the provider can reuse the cached prefix.
    ///
    /// On persist failure, the full in-memory history is preserved — the
    /// next turn reloads from DB and retries (self-healing).
    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn apply_summary(
        &mut self,
        agent_id: &str,
        summary_text: &str,
        ws: &Workspace,
        role: &Role,
        ticket: Option<&crate::pipeline::board::Ticket>,
        is_admin: bool,
        user_name: &str,
    ) {
        // Build fresh system prompt (may have changed since session start).
        let (mut compacted, snapshot) =
            Self::build_context_messages(ws, role, ticket, is_admin, user_name).await;

        // Append conversation summary, then the retained latest turns in
        // chronological order. The in-flight user message is already among
        // them (newest user message) — no separate re-append here.
        let prefix = load_prompt("context/summary_prefix.md");
        let retained = crate::session::select_retention_window(&self.history);
        // Dump what this compaction deletes BEFORE persisting, so the path
        // rides inside the persisted summary message; a persist failure then
        // orphans the dump (accepted: an orphan is dead weight — its lifecycle
        // is in `compaction_dump`). A dump failure fails open.
        let dump_path = super::compaction_dump::dump_deleted_messages(
            agent_id,
            &self.history,
            &retained,
            role,
            user_name,
        )
        .await;
        let mut summary = format!("{prefix}{summary_text}");
        if let Some(path) = &dump_path {
            summary.push_str("\n\n");
            let rendered_path = path.display().to_string();
            summary.push_str(&substitute(
                &load_prompt("context/summary_spill.md"),
                &[("{{path}}", rendered_path.as_str())],
            ));
        }
        compacted.push(ChatMessage::system(summary));
        compacted.extend(retained);

        // Persist compacted history (system prompt + summary + retained window).
        // On success, update in-memory history to match. On failure, keep the
        // full in-memory history — next turn reloads from DB and retries.
        if let Err(e) = crate::session::store()
            .replace_messages(agent_id, &compacted)
            .await
        {
            tracing::error!(
                agent_id = %agent_id,
                error = %e,
                "Failed to persist compacted session after summarization"
            );
        } else {
            self.history = compacted;
            self.persisted_len = self.history.len();
            self.publish_transcript();
            // Refresh the change-detection baseline when the block was
            // (re)rendered: it is the newest authoritative state — a stale
            // snapshot would fire a spurious change-info on the first turn
            // after every compaction. Sections that did not re-render
            // (catalog down) keep their old baseline so a mid-session switch
            // of that model can still surface as a change line. When no block
            // rendered at all the baseline is left untouched for the same
            // reason — the block itself is dropped from the system prompt
            // until the next compaction or /new (accepted fail-open gap: the
            // block is only re-rendered through this path).
            let block_rendered = snapshot.image.is_some() || snapshot.video.is_some();
            if matches!(role, Role::Assistant) && block_rendered {
                let merged = Self::merge_active_models_snapshot(agent_id, &snapshot).await;
                Self::persist_active_models_snapshot(agent_id, &merged).await;
            }
        }
    }

    // ── Lifecycle ────────────────────────────────────────────────────

    /// Flush any unpersisted history tail to the session store and advance
    /// the persisted prefix — the final assistant answer, or just delivered
    /// comments on aborted turns. No-op on an empty tail (e.g. a drain cut
    /// right after the last tool-group commit, or a turn with no output), so
    /// a retained assistant answer is never duplicated. The no-op is surfaced
    /// via [`FinalizeOutcome::NoUnpersistedTail`] for the caller to log with
    /// attribution. Skipped by the agent-level cancel path (see
    /// `finalize_session`).
    pub(crate) async fn finalize(&mut self, agent_id: &str) -> Result<FinalizeOutcome> {
        debug_assert!(self.persisted_len <= self.history.len());
        if self.persisted_len == self.history.len() {
            return Ok(FinalizeOutcome::NoUnpersistedTail);
        }
        self.persist_messages(agent_id, &[]).await?;
        Ok(FinalizeOutcome::Flushed)
    }

    // ── Internals ────────────────────────────────────────────────────

    /// Build fresh system prompt + ticket context for the current turn.
    /// Returns the system-level messages plus the model snapshot rendered in
    /// the `<active-models-opts>` block (Assistant only; `Default` when no
    /// block was injected). Messages appear in this order:
    ///
    /// ```text
    /// role_description       — from src/prompt/role/{role}.md (always)
    /// onboarding_guide       — the admin's Assistant only, while onboarding
    ///                          state is Init (or via the greeting kickoff's
    ///                          pending-guide bridge)
    /// active_models_opts     — Assistant only, when the catalogs are available
    /// workspace boilerplate  — src/prompt/context/workspace.md, with the
    ///                          per-role frame from context/workspace_frame.md
    ///                          (or context/workspace_frame_admin.md for the
    ///                          admin's Assistant); substituted (always)
    /// skills                 — if any skills exist in the workspace
    /// alarms                 — Assistant only, when the user has active alarms
    /// personal_files         — Assistant only, when the personal workspace has files
    /// workspaces             — the admin's Assistant only, when workspaces are registered
    /// custom_tools           — Assistant only, always (the catalogue, or the
    ///                          brief "nothing available" form)
    /// board_context          — Manager role only, when active tickets exist
    /// ticket_block           — when a ticket is assigned to this session
    /// ```
    ///
    /// # Caching contract
    /// This function is called directly by
    /// [`Session::apply_summary`] when rebuilding context after compaction.
    /// [`Self::build_turn_messages`] wraps it to add the per-turn user message.
    /// The Assistant block is re-emitted on compaction, so long sessions never
    /// lose the capability info.
    async fn build_context_messages(
        ws: &Workspace,
        role: &Role,
        ticket: Option<&crate::pipeline::board::Ticket>,
        is_admin: bool,
        user_name: &str,
    ) -> (Vec<ChatMessage>, ModelSnapshot) {
        let (stored_context, board_context) = tokio::join!(
            lookup_workspace_context(ws, role),
            build_board_context(ws, role),
        );

        let workspace_context = match stored_context.as_deref() {
            Some(ctx) => ctx.to_owned(),
            None => build_workspace_context(ws.as_path()).await,
        };

        let workspace_context = {
            let mut ctx = String::new();

            // Auto-discovered workspace context block.
            ctx.push_str(&crate::prompt::wrap_workspace_context(&workspace_context));

            // User-curated manual notes block (survives rediscover).
            // NOTE: changes to ws.notes do not affect in-flight agent sessions
            // until summarization compacts the history — identical to all other
            // system prompt changes (documented caching contract).
            if !ws.notes.trim().is_empty() {
                let notes = truncate_workspace_notes(&ws.notes);
                let _ = write!(ctx, "\n<user-notes>\n{notes}\n</user-notes>\n");
            }

            ctx
        };

        let workspace_boilerplate = substitute(
            &load_prompt("context/workspace.md"),
            &[
                (
                    "{{path_frame}}",
                    &load_prompt(workspace_frame_key(*role, is_admin)),
                ),
                ("{{operating_system}}", std::env::consts::OS),
                ("{{workspace}}", &ws.as_path().display().to_string()),
                ("{{workspace_context}}", &workspace_context),
                (
                    "{{system_locale}}",
                    &sys_locale::get_locale().unwrap_or_else(|| "unknown".to_string()),
                ),
            ],
        );

        let role_description = role.role_description_for(is_admin);
        let skills = skills::load_skills(ws).await;

        let mut msgs = Vec::with_capacity(7);
        msgs.push(ChatMessage::system(&role_description));

        // Onboarding guide (the admin's Assistant only, while onboarding is
        // live). One-shot / no-re-onboarding contract: the guide disappears
        // once state is `Finished`. The greeting kickoff persists `Finished`
        // BEFORE the pipeline builds the greeting's session (many async hops
        // vs a single fast config_kv upsert), so the state check alone would
        // deterministically miss the guide — the kickoff also arms a
        // process-local pending flag that the admin Assistant
        // build consumes here (see `GREETING_GUIDE_PENDING`). After that
        // single consumption nothing re-arms it and the state can only move
        // `Init` → `Finished`. Audience is global state and admin-only:
        // guests receive no onboarding and none is planned.
        if matches!(role, Role::Assistant) && is_admin {
            let live = crate::config::CONFIG.onboarding_stage()
                != crate::config::OnboardingState::Finished;
            if live || crate::onboarding::take_greeting_guide_pending() {
                msgs.push(ChatMessage::system(crate::prompt::load_prompt(
                    "role/onboarding.md",
                )));
            }
        }

        // Assistant sessions carry the <active-models-opts> block: the active
        // image/video models' parameter envelope, rendered from the live
        // catalogs plus static per-model video-edit nuances. Fail-open — no
        // block when nothing renders (catalogs unavailable, no nuances).
        let mut snapshot = ModelSnapshot::default();
        if matches!(role, Role::Assistant)
            && let Some((block, rendered)) =
                crate::tools::active_models::render_block(user_name).await
        {
            msgs.push(ChatMessage::system(block));
            snapshot = rendered;
        }

        msgs.push(ChatMessage::system(&workspace_boilerplate));
        if !skills.is_empty() {
            msgs.push(ChatMessage::system(skills::skills_to_prompt(&skills, ws)));
        }
        // Assistant sessions carry the user's alarm snapshot, the personal
        // workspace file listing, the registered workspace list (the admin's
        // Assistant only) and the custom-tool catalogue — the same snapshot-at-
        // session-start contract as the board block below.
        if matches!(role, Role::Assistant) {
            let ((alarms, workspaces, personal_files), custom_tools) = tokio::join!(
                fetch_assistant_context(user_name, is_admin),
                crate::tools::custom::context_block(user_name, is_admin),
            );
            for block in assistant_context_blocks(
                is_admin,
                &alarms,
                &workspaces,
                personal_files.as_deref(),
                &custom_tools,
            ) {
                msgs.push(ChatMessage::system(&block));
            }
        }
        if let Some(board_context) = board_context {
            msgs.push(ChatMessage::system(&board_context));
        }
        if let Some(t) = ticket {
            msgs.push(ChatMessage::system(format_ticket_block(t)));
        }
        (msgs, snapshot)
    }

    /// Build fresh system prompt + ticket context + user message for the
    /// current turn.
    /// Returns messages: [role_description, onboarding_guide?, active_models_opts?,
    /// workspace_boilerplate, skills?, alarms?, personal_files?, workspaces?,
    /// custom_tools, board_context?, ticket_block?, user_msg]
    /// plus the rendered active-models snapshot (Assistant only; `Default` when
    /// no block injected).
    ///
    /// # Caching contract
    /// The output of this function is intended to be persisted to the session
    /// DB on the **first** turn (via `batch_append_with_context` in
    /// [`Self::append_turn_message`]) and loaded from there on subsequent turns.
    /// This function is NOT called every turn — only on new sessions.
    ///
    /// System prompts are cached; user messages carry the timestamp block —
    /// round-pinned for parallel-round first messages, fresh for everything
    /// else (see [`crate::session::render_timestamp`]).
    async fn build_turn_messages(
        msg: &str,
        ws: &Workspace,
        role: &Role,
        ticket: Option<&crate::pipeline::board::Ticket>,
        round_ts: Option<&str>,
        is_admin: bool,
        user_name: &str,
    ) -> (Vec<ChatMessage>, ModelSnapshot) {
        let (mut msgs, snapshot) =
            Self::build_context_messages(ws, role, ticket, is_admin, user_name).await;
        msgs.push(crate::session::user_msg_with_ts(msg, round_ts));
        (msgs, snapshot)
    }

    // ── Active-models snapshot (Assistant mid-session change detection) ──

    /// Persist the `<active-models-opts>` baseline after the system prompt
    /// (re)build. Only the model ids actually rendered in the block are
    /// recorded (fail-open: a section absent from the block means no baseline
    /// for that model, so a later switch of it never fires a change-info).
    /// An empty snapshot clears any baseline — used at first turn when no
    /// block rendered (nothing to compare against). At compaction the caller
    /// merges in the preserved baseline for sections that did not re-render
    /// and only calls this when the block (re)rendered.
    async fn persist_active_models_snapshot(agent_id: &str, snapshot: &ModelSnapshot) {
        let json =
            (snapshot.image.is_some() || snapshot.video.is_some()).then(|| snapshot.to_json());
        if let Err(e) = crate::session::store()
            .set_active_models(agent_id, json.as_deref())
            .await
        {
            tracing::warn!(agent_id = %agent_id, error = %e, "Failed to persist active-models snapshot");
        }
    }

    /// Extend a freshly rendered snapshot with the preserved baseline for
    /// sections that did not render (catalog down), so a mid-session switch
    /// of that model can still surface as a change line.
    async fn merge_active_models_snapshot(
        agent_id: &str,
        rendered: &ModelSnapshot,
    ) -> ModelSnapshot {
        let previous = crate::session::store()
            .get_active_models(agent_id)
            .await
            .and_then(|json| ModelSnapshot::from_json(&json));
        ModelSnapshot {
            image: rendered
                .image
                .clone()
                .or_else(|| previous.as_ref().and_then(|p| p.image.clone())),
            video: rendered
                .video
                .clone()
                .or_else(|| previous.as_ref().and_then(|p| p.video.clone())),
        }
    }

    /// Detect a mid-session image/video model change against the persisted
    /// baseline, driven by `user_name`'s per-user model picks. When one
    /// occurred, returns the message with a change-info block prepended plus
    /// the new snapshot to persist (the caller persists it only after the
    /// message is stored); otherwise the message is returned unchanged with no
    /// snapshot.
    async fn prepend_model_change(
        agent_id: &str,
        msg: &str,
        user_name: &str,
    ) -> (String, Option<ModelSnapshot>) {
        let Some(previous) = crate::session::store()
            .get_active_models(agent_id)
            .await
            .and_then(|json| ModelSnapshot::from_json(&json))
        else {
            return (msg.to_string(), None);
        };
        let current = ModelSnapshot::from_user(user_name).await;

        // Advance only the changed model(s) from the persisted baseline — a
        // model whose section never rendered (partial-block fail-open session)
        // must not enter the baseline through an unrelated change, or a later
        // switch of it would fire a change-info for an id never shown to the
        // LLM.
        let mut advanced = previous.clone();
        let mut changed = Vec::new();
        if let (Some(old), Some(new)) = (previous.image.as_deref(), current.image.as_deref())
            && old != new
        {
            advanced.image = current.image.clone();
            changed.push((ModelKind::Image, old, new));
        }
        if let (Some(old), Some(new)) = (previous.video.as_deref(), current.video.as_deref())
            && old != new
        {
            advanced.video = current.video.clone();
            changed.push((ModelKind::Video, old, new));
        }
        if changed.is_empty() {
            return (msg.to_string(), None);
        }

        // Both catalog fetches run concurrently under join_all (each
        // timeout-bounded internally), so a dual image+video switch adds two
        // concurrent fetches — one bounded wall-clock duration, not two
        // sequential ones.
        let blocks = join_all(
            changed
                .into_iter()
                .map(|(kind, old, new)| Self::model_change_block(kind, old, new)),
        )
        .await;

        // The change-info is the newest authoritative state for the changed
        // model(s) — the caller refreshes the baseline once the message is
        // persisted, so the next message does not re-fire the same change.
        (format!("{}\n\n{msg}", blocks.join("\n")), Some(advanced))
    }

    /// Build a change-info block for one model: "old → new" plus the new
    /// model's capabilities from the catalog (or its static edit nuances when
    /// the catalog is unavailable); a change-only line when neither renders.
    async fn model_change_block(kind: ModelKind, old: &str, new: &str) -> String {
        match crate::tools::active_models::render_section(kind, new).await {
            Some(section) => format!(
                "<model-change>Active {} model changed from {} to {}.\nNew model capabilities:\n{section}</model-change>",
                kind.label(),
                old,
                new
            ),
            None => format!(
                "<model-change>Active {} model changed from {} to {}.</model-change>",
                kind.label(),
                old,
                new
            ),
        }
    }
}

/// Build a board-state context block for the Manager role.
///
/// Fetches all tickets for the workspace, filters out unblocking phases,
/// and formats them in the same style as the `list_tickets` tool.
/// Returns `None` for non-Manager roles, or when there are no active tickets.
async fn build_board_context(ws: &Workspace, role: &Role) -> Option<String> {
    if !matches!(role, Role::Manager) {
        return None;
    }
    let board = BOARD.get()?;
    let tickets = board.list_all_tickets(Some(&ws.name), None).await.ok()?;
    let active: Vec<_> = tickets
        .into_iter()
        .filter(|t| !t.phase.is_unblocking())
        .collect();
    if active.is_empty() {
        return None;
    }
    let count = active.len();
    let mut output = format!(
        "<workspace-board>\nTickets in {} ({count} active):\n",
        ws.name
    );
    for t in &active {
        let _ = writeln!(output, "{}", t.short_display());
    }
    output.push_str("</workspace-board>");
    Some(output)
}

// ── Assistant context blocks (alarms + files + workspaces + custom tools) ──

/// Upper char bound for a workspace's collapsed discovery-summary line.
const MAX_WORKSPACE_SUMMARY_CHARS: usize = 1000;

/// Fetch the data for the Assistant's context blocks. Fail-open per store:
/// an uninitialized store (`ALARMS.get()` / `WORKSPACES.get()` returning
/// `None`) or a read error yields empty data, so the corresponding block is
/// omitted — same contract as `build_board_context`. The alarm fetch, the
/// workspace fetch and the personal-files walk run concurrently, and the
/// per-workspace general-context reads are batched via `join_all`.
async fn fetch_assistant_context(
    user_name: &str,
    is_admin: bool,
) -> (Vec<Alarm>, Vec<(Workspace, Option<String>)>, Option<String>) {
    let alarms = async {
        if crate::alarms::ALARMS.get().is_none() {
            return Vec::new();
        }
        crate::alarms::list_user_alarms(user_name)
            .await
            .unwrap_or_default()
    };
    let workspaces = async {
        if !is_admin {
            return Vec::new();
        }
        let Some(workspaces) = WORKSPACES.get() else {
            return Vec::new();
        };
        let Ok(all) = workspaces.list().await else {
            return Vec::new();
        };
        // Defensive: personal workspaces are never registered rows, but never
        // let one leak into the block (or pay a context read for it).
        let registered: Vec<Workspace> = all
            .into_iter()
            .filter(|w| !w.name.starts_with("personal:"))
            .collect();
        let summaries = join_all(
            registered
                .iter()
                .map(|w| workspaces.get_general_context(&w.name)),
        )
        .await
        .into_iter()
        .map(|res| res.ok().flatten())
        .collect::<Vec<_>>();
        registered.into_iter().zip(summaries).collect()
    };
    tokio::join!(alarms, workspaces, fetch_personal_files(user_name))
}

/// The workspace-frame asset for a role: the admin's Assistant is not confined
/// to the workspace, so it gets the frame that says so (and names what stays
/// closed); a guest's Assistant and every pipeline role keep the workspace
/// frame. `context/workspace.md` carries the rest of the boilerplate either way.
fn workspace_frame_key(role: Role, is_admin: bool) -> &'static str {
    if matches!(role, Role::Assistant) && is_admin {
        "context/workspace_frame_admin.md"
    } else {
        "context/workspace_frame.md"
    }
}

/// Compose the Assistant-only system-prompt blocks from pre-fetched data.
/// Pure and fail-open: each block is omitted entirely when its content is
/// empty, the personal-files listing is supplied pre-rendered (`None` when
/// the workspace is empty or unlistable), and the workspace list appears only
/// for the admin's Assistant. Order: alarms → personal files → workspaces →
/// custom tools — the last being the one an Assistant always gets, because
/// [`crate::tools::custom::context_block`] renders the brief "nothing
/// available" form rather than nothing. The caller gates on `role == Assistant`
/// (before fetching).
fn assistant_context_blocks(
    is_admin: bool,
    alarms: &[Alarm],
    workspaces: &[(Workspace, Option<String>)],
    personal_files: Option<&str>,
    custom_tools: &str,
) -> Vec<String> {
    let mut blocks = Vec::new();
    if let Some(lines) = render_alarm_lines(alarms) {
        blocks.push(substitute(
            &load_prompt("context/alarms.md"),
            &[("{{alarms}}", &lines)],
        ));
    }
    if let Some(lines) = personal_files {
        blocks.push(substitute(
            &load_prompt("context/personal_files.md"),
            &[("{{files}}", lines)],
        ));
    }
    if is_admin && let Some(lines) = render_workspace_lines(workspaces) {
        blocks.push(substitute(
            &load_prompt("context/workspaces.md"),
            &[("{{workspaces}}", &lines)],
        ));
    }
    if !custom_tools.is_empty() {
        blocks.push(custom_tools.to_string());
    }
    blocks
}

/// Render one line per alarm — same shape as the `list_alarms` tool output,
/// plus the interval for periodic alarms. `None` when there are no alarms.
fn render_alarm_lines(alarms: &[Alarm]) -> Option<String> {
    if alarms.is_empty() {
        return None;
    }
    let mut out = String::new();
    for alarm in alarms {
        let fire = crate::alarms::format_fire_time(&alarm.next_fire_at)
            .unwrap_or_else(|_| alarm.next_fire_at.clone());
        let _ = write!(out, "- {}, {}", alarm.id, alarm.kind);
        if let Some(interval) = alarm.interval_seconds {
            let _ = write!(out, " (every {interval} seconds)");
        }
        let _ = write!(out, ", {}, next fire: {}", alarm.text, fire);
        if let Some(trigger) = &alarm.trigger {
            let _ = write!(out, ", trigger: {}", trigger.render());
        }
        out.push('\n');
    }
    Some(out.trim_end().to_string())
}

/// Render one entry per registered workspace — name, status (strum Display),
/// path, plus a one-line collapsed discovery summary when one exists. A
/// workspace without a summary (Pending/Failed, discovery not run) is still
/// listed, just without the summary line. Personal rows are filtered upstream
/// in [`fetch_assistant_context`]. `None` when nothing remains.
fn render_workspace_lines(workspaces: &[(Workspace, Option<String>)]) -> Option<String> {
    let mut out = String::new();
    for (ws, summary) in workspaces {
        let _ = writeln!(out, "- {} ({}): {}", ws.name, ws.status, ws.path);
        if let Some(summary) = summary.as_deref().map(str::trim).filter(|s| !s.is_empty()) {
            let _ = writeln!(
                out,
                "  {}",
                crate::util::truncate(
                    &summary.split_whitespace().collect::<Vec<_>>().join(" "),
                    MAX_WORKSPACE_SUMMARY_CHARS,
                )
            );
        }
    }
    (!out.is_empty()).then(|| out.trim_end().to_string())
}

/// Hard entry cap for the `<personal-files>` listing — the walk stops once
/// exceeded (an exact "N more" count would require an unbounded walk).
const MAX_PERSONAL_FILE_ENTRIES: usize = 200;

/// Secondary stop condition on the rendered `<personal-files>` size, in bytes.
const MAX_PERSONAL_FILES_BYTES: usize = 6 * 1024;

/// List the user's personal-workspace files for the `<personal-files>` block.
/// The path is the canonical `userspaces/<user>` resolution — the same
/// directory the Assistant session is pinned to — so a blank or synthetic
/// user name omits the block entirely instead of listing the userspaces root
/// or other users' files. `None` also when the workspace has no listable
/// files or the walk fails (fail-open, like the other blocks). The walk runs
/// on the blocking pool so a large workspace never stalls the executor.
async fn fetch_personal_files(user_name: &str) -> Option<String> {
    if !crate::users::is_valid_personal_user_name(user_name) {
        return None;
    }
    let root = crate::users::personal_workspace_path(user_name);
    let listed = tokio::task::spawn_blocking(move || walk_personal_files(&root))
        .await
        .ok()?;
    render_personal_file_lines(listed)
}

/// Walk the personal workspace and collect up to `MAX_PERSONAL_FILE_ENTRIES + 1`
/// relative file paths (directories are implied by the paths and skipped).
/// Ripgrep defaults apply: hidden entries (`.git/`, `.DS_Store`, all dotfiles —
/// including the hidden `.compaction` dump dir) are skipped, and `generated/`
/// / `uploads/` directories are pruned at the entry level so large media
/// trees are never traversed. Symlinks are not followed.
fn walk_personal_files(root: &Path) -> Vec<String> {
    let cap = MAX_PERSONAL_FILE_ENTRIES + 1;
    let mut out = Vec::new();
    for entry in WalkBuilder::new(root)
        .filter_entry(|e| {
            e.depth() == 0
                || !e.file_type().is_some_and(|t| t.is_dir())
                || !matches!(e.file_name().to_str(), Some("generated" | "uploads"))
        })
        .build()
        .flatten()
    {
        if entry.depth() == 0 || entry.file_type().is_some_and(|t| t.is_dir()) {
            continue; // the workspace root itself, and directories (implied)
        }
        let Ok(rel) = entry.path().strip_prefix(root) else {
            continue;
        };
        out.push(rel.display().to_string());
        if out.len() >= cap {
            break;
        }
    }
    out
}

/// Render the byte-sorted relative paths for the `<personal-files>` block —
/// directories are implied by the paths. Stops at the entry cap or the byte
/// budget and appends a truncation tail; `None` when nothing remains.
fn render_personal_file_lines(mut paths: Vec<String>) -> Option<String> {
    if paths.is_empty() {
        return None;
    }
    paths.sort();
    let mut out = String::new();
    let mut shown = 0;
    for path in &paths {
        if shown == MAX_PERSONAL_FILE_ENTRIES
            || out.len() + path.len() + 1 > MAX_PERSONAL_FILES_BYTES
        {
            break;
        }
        out.push_str(path);
        out.push('\n');
        shown += 1;
    }
    if shown < paths.len() {
        out.push_str("…listing truncated; use the read tool for the full picture");
    }
    Some(out.trim_end().to_string())
}

/// Try to find a stored workspace context for the given workspace and role.
async fn lookup_workspace_context(ws: &Workspace, role: &Role) -> Option<String> {
    let workspaces = crate::workspace::store();
    workspaces.get_context(&ws.name, role.as_str()).await.ok()?
}

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

    /// Mid-session change detection: switching image AND video models against
    /// a persisted baseline fires one change-info per model (change-only
    /// lines — both shared catalog caches are seeded fail-open, so the change
    /// path performs no network I/O), returns the new snapshot for the caller
    /// to persist after the message append, and does not re-fire once the
    /// baseline is advanced. A missing baseline is a no-op.
    #[tokio::test]
    #[serial_test::serial(active_models)]
    async fn prepend_model_change_detects_switch_and_refreshes_baseline() {
        let user = "test_assistant_change_user";
        crate::util::test::init_test_stores().await;
        // Empty-catalog seeds: both catalog lookups return a fresh (empty)
        // catalog with no network fetch, so render_section finds no model and
        // the change path exercises the change-only fallback hermetically.
        // Unlike a None seed (1-min negative-cache Backoff residue), the
        // caches stay Fresh — no cross-test coupling.
        crate::tools::media_catalog::image::seed_cache(Some(std::sync::Arc::new(
            crate::tools::media_catalog::image::ImageCatalog::default(),
        )));
        crate::tools::media_catalog::video::seed_cache(Some(std::sync::Arc::new(
            crate::tools::media_catalog::video::VideoCatalog::default(),
        )));
        let agent_id = "test_assistant_change";

        crate::users::store()
            .set_image_gen_model(user, "model-a")
            .await
            .expect("set image model");
        crate::users::store()
            .set_video_model(user, "video-a")
            .await
            .expect("set video model");
        let baseline = ModelSnapshot::from_user(user).await;
        crate::session::store()
            .set_active_models(agent_id, Some(&baseline.to_json()))
            .await
            .expect("baseline persisted");

        // Switch both models mid-session → one change-info per model.
        crate::users::store()
            .set_image_gen_model(user, "model-b")
            .await
            .expect("set image model");
        crate::users::store()
            .set_video_model(user, "video-b")
            .await
            .expect("set video model");
        let (out, new_snapshot) = Session::prepend_model_change(agent_id, "hello", user).await;
        assert!(
            out.starts_with(
                "<model-change>Active image model changed from model-a to model-b.</model-change>\n\
                 <model-change>Active video model changed from video-a to video-b.</model-change>"
            ),
            "unexpected change-info: {out:?}"
        );
        assert!(out.ends_with("\n\nhello"));
        let new_snapshot = new_snapshot.expect("snapshot returned for persistence");

        // Baseline refreshed (as the caller does after a successful append) →
        // the same switches do not re-fire.
        crate::session::store()
            .set_active_models(agent_id, Some(&new_snapshot.to_json()))
            .await
            .expect("baseline advanced");
        let (out, snapshot) = Session::prepend_model_change(agent_id, "again", user).await;
        assert_eq!(out, "again");
        assert!(snapshot.is_none());
    }

    #[tokio::test]
    async fn prepend_model_change_without_baseline_is_noop() {
        crate::util::test::init_test_stores().await;
        let (out, snapshot) =
            Session::prepend_model_change("test_assistant_no_baseline", "hello", "no_such_user")
                .await;
        assert_eq!(out, "hello");
        assert!(snapshot.is_none());
    }

    /// Partial-block session (one catalog down at start): switching the model
    /// whose section DID render must not mint a baseline for the never-rendered
    /// model — a later switch of it stays silent.
    #[tokio::test]
    #[serial_test::serial(active_models)]
    async fn prepend_model_change_preserves_absent_sections() {
        let user = "test_assistant_partial_user";
        crate::util::test::init_test_stores().await;
        crate::tools::media_catalog::image::seed_cache(Some(std::sync::Arc::new(
            crate::tools::media_catalog::image::ImageCatalog::default(),
        )));
        crate::tools::media_catalog::video::seed_cache(Some(std::sync::Arc::new(
            crate::tools::media_catalog::video::VideoCatalog::default(),
        )));
        let agent_id = "test_assistant_partial";

        // Session started with only the video section rendered (image catalog
        // was down at session start) — the image id is NOT in the baseline.
        crate::users::store()
            .set_video_model(user, "video-a")
            .await
            .expect("set video model");
        let baseline = ModelSnapshot {
            image: None,
            video: Some("video-a".into()),
        };
        crate::session::store()
            .set_active_models(agent_id, Some(&baseline.to_json()))
            .await
            .expect("baseline persisted");

        // Switch the video model → change-info for video; the returned
        // snapshot advances only video, leaving the absent image baseline
        // untouched.
        crate::users::store()
            .set_video_model(user, "video-b")
            .await
            .expect("set video model");
        let (out, new_snapshot) = Session::prepend_model_change(agent_id, "hello", user).await;
        assert!(
            out.starts_with("<model-change>Active video model changed from video-a to video-b.")
        );
        let new_snapshot = new_snapshot.expect("snapshot returned for persistence");
        assert_eq!(new_snapshot.image, None);
        assert_eq!(new_snapshot.video.as_deref(), Some("video-b"));
        crate::session::store()
            .set_active_models(agent_id, Some(&new_snapshot.to_json()))
            .await
            .expect("baseline advanced");

        // A later image switch stays silent — the image id was never rendered.
        crate::users::store()
            .set_image_gen_model(user, "image-b")
            .await
            .expect("set image model");
        let (out, snapshot) = Session::prepend_model_change(agent_id, "again", user).await;
        assert_eq!(out, "again");
        assert!(snapshot.is_none());
    }

    /// Compaction merge: a rendered snapshot is extended with the preserved
    /// baseline for sections that did not re-render (catalog down), so a
    /// mid-session switch of that model can still surface as a change line.
    #[tokio::test]
    async fn merge_active_models_snapshot_preserves_unrendered_sections() {
        crate::util::test::init_test_stores().await;
        let previous = ModelSnapshot {
            image: Some("model-a".into()),
            video: Some("video-a".into()),
        };
        crate::session::store()
            .set_active_models("test_assistant_merge", Some(&previous.to_json()))
            .await
            .expect("baseline persisted");

        // Compaction re-rendered only the image section (video catalog down).
        let rendered = ModelSnapshot {
            image: Some("model-b".into()),
            video: None,
        };
        let merged = Session::merge_active_models_snapshot("test_assistant_merge", &rendered).await;
        assert_eq!(
            merged,
            ModelSnapshot {
                image: Some("model-b".into()),
                video: Some("video-a".into()),
            }
        );

        // No prior baseline → the merged snapshot is exactly the rendered one.
        let merged =
            Session::merge_active_models_snapshot("test_assistant_no_merge_baseline", &rendered)
                .await;
        assert_eq!(merged, rendered);
    }

    // ── rewrite_last_user_message (input-image-rejection strip durability) ─

    #[tokio::test]
    async fn rewrite_last_user_message_persists_then_swaps() {
        crate::util::test::init_test_stores().await;
        let agent_id = "test_rewrite_ok";
        // The in-memory history must mirror the store (persisted prefix).
        let seed = vec![
            ChatMessage::system("role"),
            ChatMessage::user("[IMAGE:/tmp/a.png] first"),
            ChatMessage::user("[IMAGE:/tmp/b.png] second"),
        ];
        crate::session::store()
            .batch_append(agent_id, &seed)
            .await
            .unwrap();
        let mut session = Session::default();
        session.history = seed;
        session.persisted_len = session.history.len();

        let outcome = session
            .rewrite_last_user_message(agent_id, "rewritten".to_string())
            .await
            .unwrap();
        assert_eq!(outcome, RewriteOutcome::Rewritten);

        // In-memory swap happened AFTER the durable write.
        assert_eq!(session.history[2].content, "rewritten");
        assert_eq!(
            session.history[1].content, "[IMAGE:/tmp/a.png] first",
            "earlier user row untouched"
        );

        // Durable: the store row reflects the rewrite.
        let msgs = crate::session::store().load(agent_id).await;
        assert_eq!(msgs.len(), 3);
        assert_eq!(msgs[2].content, "rewritten");
    }

    #[tokio::test]
    async fn rewrite_last_user_message_unpersisted_tail_is_conservative_noop() {
        crate::util::test::init_test_stores().await;
        let agent_id = "test_rewrite_noop";
        // Seed the DB with one persisted user row.
        crate::session::store()
            .batch_append(
                agent_id,
                &[ChatMessage::user("[IMAGE:/tmp/a.png] persisted")],
            )
            .await
            .unwrap();

        // The most recent user message lives in the UNPERSISTED tail
        // (index 1 >= persisted_len 1): a positional UPDATE would rewrite the
        // persisted row instead — conservative no-op.
        let mut session = Session {
            history: vec![
                ChatMessage::user("[IMAGE:/tmp/a.png] persisted"),
                ChatMessage::user("[IMAGE:/tmp/b.png] unpersisted"),
            ],
            persisted_len: 1,
            ..Default::default()
        };

        let outcome = session
            .rewrite_last_user_message(agent_id, "rewritten".to_string())
            .await
            .unwrap();
        assert_eq!(outcome, RewriteOutcome::UnpersistedTailNoop);
        assert_eq!(
            session.history[1].content, "[IMAGE:/tmp/b.png] unpersisted",
            "in-memory history untouched"
        );

        // Durable row untouched.
        let msgs = crate::session::store().load(agent_id).await;
        assert_eq!(msgs[0].content, "[IMAGE:/tmp/a.png] persisted");
    }

    #[tokio::test]
    async fn rewrite_last_user_message_no_user_message_errors() {
        crate::util::test::init_test_stores().await;
        let mut session = Session {
            history: vec![ChatMessage::system("role")],
            persisted_len: 1,
            ..Default::default()
        };
        let err = session
            .rewrite_last_user_message("test_rewrite_err", "x".to_string())
            .await
            .unwrap_err();
        assert!(err.to_string().contains("no user message"), "{err}");
    }

    #[tokio::test]
    async fn rewrite_last_user_message_persist_failure_leaves_history_untouched() {
        crate::util::test::init_test_stores().await;
        let agent_id = "test_rewrite_persist_fail";
        // Seed a DIFFERENT agent's row so the sessions table is non-empty:
        // the positional UPDATE for `agent_id` targets only that agent's last
        // `user` row — none exists, so the store reports 0 rows affected and
        // errors.
        crate::session::store()
            .batch_append(
                "other_agent",
                &[ChatMessage::user("[IMAGE:/tmp/a.png] other")],
            )
            .await
            .unwrap();

        let mut session = Session {
            history: vec![
                ChatMessage::system("role"),
                ChatMessage::user("[IMAGE:/tmp/a.png] mine"),
            ],
            persisted_len: 2, // idx 1 < persisted_len → the guard passes
            ..Default::default()
        };

        let err = session
            .rewrite_last_user_message(agent_id, "rewritten".to_string())
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("no user message row"),
            "store error propagates: {err}"
        );
        assert_eq!(
            session.history[1].content, "[IMAGE:/tmp/a.png] mine",
            "in-memory history untouched on persist failure (ordering invariant)"
        );
    }

    // ── Tool-call frame persist / settle ─────────────────────────────

    /// Helper: the assistant frame JSON for an analyze tool call.
    fn analyze_frame(call_id: &str) -> String {
        crate::providers::reasoning::assistant_replay_payload(
            Some(""),
            &[crate::ToolCall {
                id: call_id.to_string(),
                name: "analyze".to_string(),
                arguments: serde_json::json!({"analyze": "analyze task"}),
            }],
            None,
        )
        .to_string()
    }

    /// (a) `pending_tool_frame` returns the session's LAST tool-call frame's
    /// result-less calls. No frame → None; a frame with calls and no results →
    /// all of them; one result present → only the missing one remains.
    #[test]
    fn pending_tool_frame_none_and_dangling_and_partial() {
        // No frame → None.
        assert!(Session::default().pending_tool_frame().is_none());

        // Frame with two calls, no results → both dangling.
        let frame = crate::providers::reasoning::assistant_replay_payload(
            Some(""),
            &[
                crate::ToolCall {
                    id: "call_a".to_string(),
                    name: "read".to_string(),
                    arguments: serde_json::json!({"path": "a"}),
                },
                crate::ToolCall {
                    id: "call_b".to_string(),
                    name: "read".to_string(),
                    arguments: serde_json::json!({"path": "b"}),
                },
            ],
            None,
        )
        .to_string();
        let mut session = Session {
            history: vec![ChatMessage::assistant(frame)],
            persisted_len: 1,
            ..Default::default()
        };
        let pending = session
            .pending_tool_frame()
            .expect("dangling calls present")
            .calls;
        assert_eq!(pending.len(), 2);
        assert_eq!(pending[0].id, "call_a");
        assert_eq!(pending[1].id, "call_b");

        // One result present → only the missing one remains.
        session
            .history
            .push(ChatMessage::tool_result("call_a", "result a"));
        let pending = session
            .pending_tool_frame()
            .expect("still one dangling")
            .calls;
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].id, "call_b");

        // All results present → None (nothing dangling).
        session
            .history
            .push(ChatMessage::tool_result("call_b", "result b"));
        assert!(session.pending_tool_frame().is_none());
    }

    /// (b) `settle_tool_results` inserts the settled result directly after the
    /// session's dangling tool-call frame, bumps `message_count` by the result
    /// count, and mirrors the insert in-memory with `persisted_len` ==
    /// `history.len()`.
    #[tokio::test]
    async fn settle_tool_results_inserts_result_after_frame() {
        crate::util::test::init_test_stores().await;
        let agent_id = "settle_tool_results_test";
        let frame = analyze_frame("call_analyze_1");

        let mut session = Session::default();
        session
            .persist_messages(
                agent_id,
                &[ChatMessage::user("first"), ChatMessage::assistant(frame)],
            )
            .await
            .unwrap();
        assert_eq!(session.persisted_len, 2, "both seeded rows persisted");

        let conn = &crate::session::store().conn;
        let count_before: i64 = conn
            .query_optional(
                "SELECT message_count FROM session_metadata WHERE agent_id = ?1",
                crate::db::params![agent_id],
                |r| r.get::<i64>(0),
            )
            .await
            .unwrap()
            .expect("metadata row exists");

        session
            .settle_tool_results(
                agent_id,
                &[("call_analyze_1".to_string(), "the result".to_string())],
                &[],
            )
            .await
            .unwrap();

        // DB rows by id: user, assistant(frame), tool(result).
        let rows = conn
            .query(
                "SELECT id, role, content FROM sessions WHERE agent_id = ?1 ORDER BY id",
                crate::db::params![agent_id],
            )
            .await
            .unwrap();
        assert_eq!(rows.len(), 3, "one result row inserted after the frame");
        let roles: Vec<String> = rows.iter().map(|r| r.get::<String>(1).unwrap()).collect();
        assert_eq!(roles, vec!["user", "assistant", "tool"]);
        assert_eq!(
            rows[2].get::<String>(2).unwrap(),
            serde_json::to_string(&crate::ToolResultPayload {
                tool_call_id: "call_analyze_1".to_string(),
                content: "the result".to_string(),
            })
            .unwrap(),
            "the tool row carries the settled result with the original call id"
        );

        // message_count delta == 1 (the result row).
        let count_after: i64 = conn
            .query_optional(
                "SELECT message_count FROM session_metadata WHERE agent_id = ?1",
                crate::db::params![agent_id],
                |r| r.get::<i64>(0),
            )
            .await
            .unwrap()
            .unwrap();
        assert_eq!(count_after - count_before, 1);

        // In-memory history mirrors the insert; persisted prefix covers it.
        let roles: Vec<crate::ChatRole> = session.history().iter().map(|m| m.role).collect();
        assert_eq!(
            roles,
            vec![ChatRole::User, ChatRole::Assistant, ChatRole::Tool]
        );
        assert_eq!(
            session.history()[2].content,
            serde_json::to_string(&crate::ToolResultPayload {
                tool_call_id: "call_analyze_1".to_string(),
                content: "the result".to_string(),
            })
            .unwrap()
        );
        assert_eq!(session.persisted_len, session.history().len());
    }

    /// (b2) `settle_tool_results` preserves the frame's tool-call ORDER when a
    /// partial result set is settled: a frame with calls `call_a`/`call_b`,
    /// only `call_b`'s row already committed, settles `call_a` — the final DB
    /// order is frame, tool(call_a), tool(call_b) (the recovered result keeps
    /// the frame call order; the captured sibling stays in the frame's tool
    /// block, not displaced by "new results first").
    #[tokio::test]
    async fn settle_tool_results_preserves_frame_call_order() {
        crate::util::test::init_test_stores().await;
        let agent_id = "settle_tool_results_order_test";
        let frame = crate::providers::reasoning::assistant_replay_payload(
            Some(""),
            &[
                crate::ToolCall {
                    id: "call_a".to_string(),
                    name: "read".to_string(),
                    arguments: serde_json::json!({"path": "a"}),
                },
                crate::ToolCall {
                    id: "call_b".to_string(),
                    name: "read".to_string(),
                    arguments: serde_json::json!({"path": "b"}),
                },
            ],
            None,
        )
        .to_string();

        let mut session = Session::default();
        session
            .persist_messages(
                agent_id,
                &[
                    ChatMessage::user("first"),
                    ChatMessage::assistant(frame),
                    ChatMessage::tool_result("call_b", "result b"),
                ],
            )
            .await
            .unwrap();

        // Settle only call_a; call_b's row is the captured sibling.
        session
            .settle_tool_results(
                agent_id,
                &[("call_a".to_string(), "result a".to_string())],
                &[],
            )
            .await
            .unwrap();

        // DB rows by id: user, assistant(frame), tool(call_a), tool(call_b) —
        // the recovered result keeps the frame's call order, and the captured
        // sibling verbatim.
        let conn = &crate::session::store().conn;
        let rows = conn
            .query(
                "SELECT id, role, content FROM sessions WHERE agent_id = ?1 ORDER BY id",
                crate::db::params![agent_id],
            )
            .await
            .unwrap();
        assert_eq!(
            rows.len(),
            4,
            "frame + captured sibling + one recovered result"
        );
        let roles: Vec<String> = rows.iter().map(|r| r.get::<String>(1).unwrap()).collect();
        assert_eq!(roles, vec!["user", "assistant", "tool", "tool"]);
        let mut ids: Vec<String> = Vec::new();
        for row in &rows[2..] {
            let payload: crate::ToolResultPayload =
                serde_json::from_str(&row.get::<String>(2).unwrap()).unwrap();
            ids.push(payload.tool_call_id);
        }
        assert_eq!(ids, vec!["call_a", "call_b"], "frame call order preserved");
        let b_payload: crate::ToolResultPayload =
            serde_json::from_str(&rows[3].get::<String>(2).unwrap()).unwrap();
        assert_eq!(b_payload.content, "result b", "captured sibling verbatim");
    }

    /// The workspace frame the admin's Assistant receives states what is closed
    /// instead of reading as confinement, and no other role stops being told
    /// where the workspace is — the two frame assets the gating rule can pick
    /// between, rendered into the one boilerplate the roles share.
    #[test]
    fn the_admin_workspace_frame_states_what_is_closed() {
        let shared = crate::prompt::load_prompt("context/workspace.md");
        assert!(
            shared.contains("{{path_frame}}"),
            "the shared boilerplate takes the role's frame: {shared}"
        );
        let frame = crate::prompt::load_prompt("context/workspace_frame.md");
        assert!(
            frame.contains("outside the workspace"),
            "the shared frame keeps the workspace rule: {frame}"
        );
        let admin = crate::prompt::load_prompt("context/workspace_frame_admin.md");
        assert!(
            !admin.contains("outside the workspace"),
            "the admin's frame must not read as confinement: {admin}"
        );
        for named in ["registered project workspaces", "live databases"] {
            assert!(
                admin.contains(named),
                "the admin's frame must name what stays closed ({named}): {admin}"
            );
        }

        // The frame is a fragment: the boilerplate's own line break separates it
        // from the line below, so a trailing newline would inject a blank line
        // into every role's system prompt.
        for asset in [&frame, &admin] {
            assert!(
                !asset.ends_with('\n'),
                "the frame assets are fragments and must not end with a newline"
            );
        }

        // Only the admin's Assistant gets that frame.
        assert_eq!(
            workspace_frame_key(Role::Assistant, true),
            "context/workspace_frame_admin.md"
        );
        for pick in [
            workspace_frame_key(Role::Assistant, false),
            workspace_frame_key(Role::Manager, true),
            workspace_frame_key(Role::Engineer, false),
        ] {
            assert_eq!(pick, "context/workspace_frame.md");
        }
    }

    /// Pure gating of the Assistant context blocks: a guest's Assistant gets
    /// the alarms, personal-files and custom-tool blocks (the workspace list is
    /// the admin's only); the admin's Assistant gets the first three in
    /// alarms → personal files → workspaces order plus the custom-tool block
    /// last; empty data omits the conditional blocks, while the custom-tool
    /// block is always present (its "nothing available" form is rendered
    /// upstream). (Role gating lives at the call site.)
    #[test]
    fn assistant_blocks_gating() {
        let alarm = Alarm {
            id: "a1".to_string(),
            session_id: "s1".to_string(),
            user_name: "u".to_string(),
            kind: "one-shot".to_string(),
            text: "hello".to_string(),
            trigger: None,
            interval_seconds: None,
            next_fire_at: "2026-01-01T00:00:00+00:00".to_string(),
        };
        let workspaces = vec![(
            crate::workspace::test_ws_named("/tmp/proj", "proj"),
            Some("summary".to_string()),
        )];
        let files = Some("MEMORY.md\nnotes/projects.md");
        let custom = "<custom-tools>\n- weather\n</custom-tools>";

        // Guest's Assistant: alarms + personal files + custom tools, no
        // workspace list.
        let blocks = assistant_context_blocks(
            false,
            std::slice::from_ref(&alarm),
            &workspaces,
            files,
            custom,
        );
        assert_eq!(blocks.len(), 3);
        assert!(blocks[0].contains("<user-alarms>"));
        assert!(blocks[1].contains("<personal-files>"));
        assert!(!blocks[1].contains("<registered-workspaces>"));
        assert_eq!(blocks[2], custom);

        // The admin's Assistant: the workspace list joins them, and the
        // custom-tool block stays last.
        let blocks = assistant_context_blocks(
            true,
            std::slice::from_ref(&alarm),
            &workspaces,
            files,
            custom,
        );
        assert_eq!(blocks.len(), 4);
        assert!(blocks[0].contains("<user-alarms>"));
        assert!(blocks[1].contains("<personal-files>"));
        assert!(blocks[2].contains("<registered-workspaces>"));
        assert_eq!(blocks[3], custom);

        // Empty data omits the conditional blocks but never the custom tools.
        assert_eq!(
            assistant_context_blocks(true, &[], &[], None, custom),
            vec![custom.to_string()]
        );
    }

    /// Personal-files walk: hidden entries and `generated/` / `uploads/`
    /// directories never appear, while regular nested files do. An absent
    /// workspace yields an empty listing.
    #[test]
    fn personal_files_walk_prunes_excluded_dirs() {
        let root = tempfile::tempdir().expect("tempdir");
        std::fs::write(root.path().join("MEMORY.md"), "m").expect("write");
        std::fs::create_dir_all(root.path().join("notes")).expect("mkdir");
        std::fs::write(root.path().join("notes/topic.md"), "t").expect("write");
        std::fs::create_dir_all(root.path().join("generated")).expect("mkdir");
        std::fs::write(root.path().join("generated/big.bin"), "b").expect("write");
        std::fs::create_dir_all(root.path().join(".git")).expect("mkdir");
        std::fs::write(root.path().join(".git/config"), "c").expect("write");

        let listed = walk_personal_files(root.path());
        assert!(listed.contains(&"MEMORY.md".to_string()));
        assert!(listed.contains(&Path::new("notes").join("topic.md").display().to_string()));
        assert!(!listed.iter().any(|p| p.contains("generated")));
        assert!(!listed.iter().any(|p| p.contains(".git")));
        // Directories are implied by the file paths, never listed themselves.
        assert!(!listed.iter().any(|p| p == "notes"));

        assert!(walk_personal_files(&root.path().join("absent")).is_empty());
    }

    /// Path-safety guard: blank or synthetic user names (path separators,
    /// dot segments) never resolve to a workspace — the block is omitted
    /// instead of listing the userspaces root or another user's files.
    #[tokio::test]
    async fn personal_files_fetch_rejects_synthetic_names() {
        for name in ["", "  ", ".", "..", "a/b", "a\\b"] {
            assert!(fetch_personal_files(name).await.is_none(), "name: {name}");
        }
    }

    /// Personal-files rendering: byte-sorted output stopped by the entry cap
    /// or the byte budget, both appending a truncation tail; an empty listing
    /// renders as `None`.
    #[test]
    fn personal_files_render_caps_budgets_and_sorts() {
        let paths: Vec<String> = (0..MAX_PERSONAL_FILE_ENTRIES + 50)
            .map(|i| format!("f{i:03}.txt"))
            .collect();
        let rendered = render_personal_file_lines(paths).expect("render");
        let lines: Vec<&str> = rendered.lines().collect();
        assert_eq!(lines.len(), MAX_PERSONAL_FILE_ENTRIES + 1); // entries + tail
        assert!(lines[0] < lines[1]);
        assert!(rendered.ends_with("use the read tool for the full picture"));

        // Byte budget stops the listing before the entry cap is reached.
        let long = "x".repeat(MAX_PERSONAL_FILES_BYTES / 2);
        let rendered =
            render_personal_file_lines(vec![long.clone(), long.clone(), long]).expect("render");
        assert_eq!(rendered.lines().count(), 2); // one entry + tail

        assert!(render_personal_file_lines(Vec::new()).is_none());
    }

    /// Alarm line rendering: a periodic interval is annotated, the trigger is
    /// rendered verbatim (no scrubbing — alarm-born text is not redacted on the
    /// alarm side), and an empty alarm set renders as `None`.
    #[test]
    fn alarm_lines_show_trigger_and_interval() {
        let one_shot = Alarm {
            id: "a1".to_string(),
            session_id: "s1".to_string(),
            user_name: "u".to_string(),
            kind: "one-shot".to_string(),
            text: "hello".to_string(),
            trigger: None,
            interval_seconds: None,
            next_fire_at: "2026-01-01T00:00:00+00:00".to_string(),
        };
        let periodic = Alarm {
            id: "a2".to_string(),
            session_id: "s1".to_string(),
            user_name: "u".to_string(),
            kind: "periodic".to_string(),
            text: "check".to_string(),
            trigger: Some(crate::alarms::StoredTrigger::Tool(crate::alarms::Trigger {
                tool: "weather".to_string(),
                args: serde_json::Map::from_iter([
                    ("city".to_string(), serde_json::json!("Minsk")),
                    (
                        "api_key".to_string(),
                        serde_json::json!("supersecretvalue123"),
                    ),
                ]),
            })),
            interval_seconds: Some(3600),
            next_fire_at: "2026-01-01T00:00:00+00:00".to_string(),
        };

        let lines = render_alarm_lines(&[one_shot, periodic]).expect("alarms render");
        assert!(lines.contains("every 3600 seconds"));
        assert!(lines.contains("trigger: weather "));
        assert!(lines.contains("\"city\":\"Minsk\""));
        // Scrubbing was deliberately removed: a credential-shaped argument
        // reaches the persisted prompt verbatim.
        assert!(lines.contains("\"api_key\":\"supersecretvalue123\""));
        assert!(lines.contains("next fire:"));

        assert!(render_alarm_lines(&[]).is_none());
    }

    /// Workspace list rendering: the discovery summary is collapsed to one
    /// line and a summary-less workspace is still listed. (Personal-row
    /// filtering lives in the fetcher.) An empty slice renders as `None`.
    #[test]
    fn workspace_lines_summary_collapse() {
        let workspaces = vec![
            (
                crate::workspace::test_ws_named("/data/proj", "proj"),
                Some("Multi   line\nsummary text".to_string()),
            ),
            (
                crate::workspace::test_ws_named("/data/other", "other"),
                None,
            ),
        ];

        let lines = render_workspace_lines(&workspaces).expect("workspaces render");
        assert!(lines.contains("- proj (pending): /data/proj"));
        assert!(lines.contains("\n  Multi line summary text"));
        assert!(lines.contains("- other (pending): /data/other"));

        assert!(render_workspace_lines(&[]).is_none());
    }
}