agentmux_srv/server/agent_handlers/

session.rs

// Copyright 2025-2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0


use std::sync::Arc;


use crate::backend::rpc::engine::WshRpcEngine;
use crate::backend::rpc_types::{
    COMMAND_LIST_RECENT_SESSIONS, CommandListRecentSessionsData,
    ListRecentSessionsResult, RecentSessionRow,
    // Option E (PR 1 of 2) — agent-anchored session zones.
    COMMAND_AGENT_SESSION_READ, COMMAND_AGENT_SESSION_WRITE_STATE,
    COMMAND_AGENT_SESSION_APPEND_OUTPUT, COMMAND_AGENT_SESSION_ARCHIVE,
    COMMAND_AGENT_SESSION_LIST_ARCHIVES,
    CommandAgentSessionReadData, AgentSessionReadResult,
    CommandAgentSessionWriteStateData, AgentSessionWriteStateResult,
    CommandAgentSessionAppendOutputData, AgentSessionAppendOutputResult,
    CommandAgentSessionArchiveData, AgentSessionArchiveResult,
    CommandAgentSessionListArchivesData, AgentArchiveRow,
};
use crate::backend::storage::store::{AgentIdentityLink, AgentInstance, IdentityAccount};

use super::super::AppState;
use super::read_session_preview;

/// At-most-once-per-process-lifetime claim set for definition-summary
/// generation attempts triggered from `listrecentsessions` — see the call
/// site below and `db_agent_activity_summaries` (OBJECT_SCHEMA_VERSION
/// v28). `HashSet::insert` returns `true` only the first time a given
/// `definition_id` is claimed; every later call for the same id (a repeat
/// poll of "My Agents") sees `false` and does not re-spawn.
fn definition_summary_attempted() -> &'static std::sync::Mutex<std::collections::HashSet<String>> {
    static SET: std::sync::OnceLock<std::sync::Mutex<std::collections::HashSet<String>>> =
        std::sync::OnceLock::new();
    SET.get_or_init(|| std::sync::Mutex::new(std::collections::HashSet::new()))
}

pub fn register(engine: &Arc<WshRpcEngine>, state: &AppState) {
    // ---- Recent sessions (cascade follow-up 2026-05-23) ----
    //
    // listrecentsessions — joins `db_agent_instances` with the
    // filestore `output.state.json` snapshot for each instance's
    // block_id_hint, producing a preview + node count so the
    // AgentPicker can show actual conversation context instead of just
    // metadata. Sort key is the snapshot modts (last activity)
    // descending; rows without a snapshot fall back to the instance
    // started_at and are de-prioritized. Cap at 20 rows.
    //
    // The reattach mechanism is the existing continuation flow:
    // continueOfInstanceId + workDirOverride (see PR #977). This RPC
    // is a more discoverable surface for finding sessions to continue
    // — particularly orphaned ones whose pane crashed.
    let mstore = state.mstore.clone();
    let id_store_lrs = state.id_store.clone();
    let identity_store_lrs = state.identity_store.clone();
    let filestore = state.filestore.clone();
    let broker_lrs = state.broker.clone();
    // `Value` request: a malformed one means the defaults (unwrap_or_default).
    engine.register_typed(COMMAND_LIST_RECENT_SESSIONS, move |data: serde_json::Value, _ctx| {
        let mstore = mstore.clone();
        let id_store = id_store_lrs.clone();
        let identity_store = identity_store_lrs.clone();
        let filestore = filestore.clone();
        let broker = broker_lrs.clone();
        async move {
                let t0 = std::time::Instant::now();
                let cmd: CommandListRecentSessionsData =
                    serde_json::from_value(data).unwrap_or_default();
                let limit = if cmd.limit == 0 {
                    20
                } else {
                    cmd.limit.min(100)
                };
                // Tracks which of this handler's data sources degraded to
                // empty this call. Every source below used to `.map_err(..)?`
                // — one bad row anywhere aborted the ENTIRE "My Agents" list,
                // and that exact failure mode has already broken it to an
                // empty list at least twice in production (PR #2296's oauth
                // serde-tag mismatch; see
                // retro-my-agents-fresh-channel-regression-2026_07_27 (never committed to this repo)
                // §4/§9 rec 1). A single malformed row must degrade THAT
                // source only, not the whole response — logged here so the
                // next incident's log can actually show what happened
                // (§9 rec 3), unlike this retro's investigation, which had
                // no way to tell whether this handler ran at all.
                let mut degraded: Vec<&'static str> = Vec::new();
                // Pull up to ~10x the requested cap so we can post-
                // filter by snapshot presence + identity_id without
                // running out of candidates. 10x is a safety margin
                // and stays well inside the 200 default of
                // instance_list_named.
                let raw_limit = (limit * 10).max(50).min(500);

                // Identity filter is pushed INTO `instance_list_named`
                // (codex P2 #3 on PR #1096): when a chain has
                // continuations under different accounts, the
                // ranking must run on identity-matching rows so the
                // newest match wins. Post-query filtering would drop
                // the chain entirely if the newest row used a
                // different identity, even when older rows match.
                let identity_filter = cmd
                    .identity_id
                    .as_deref()
                    .map(|s| s.trim())
                    .filter(|s| !s.is_empty());

                // "My Agents" sources from the cross-version REGISTRY when it's
                // available, so agents created in ANY build / channel / version
                // appear here — not just this instance's local SQLite sessions
                // (the live mirror, registry_mirror.rs, keeps the registry current
                // for global workspaces). Falls back to local SQLite when the
                // registry couldn't be resolved (CI / odd envs). Cross-channel rows
                // arrive as synthetic instances (no live block); the per-instance
                // snapshot enrichment below lights up the ones that ALSO ran here.
                let instances: Vec<AgentInstance> = match mstore.shared_agent_registry() {
                    Some(reg) => {
                        let agents_root = mstore.registry_agents_base();
                        let mut records = match reg.list_active() {
                            Ok(v) => v,
                            Err(e) => {
                                tracing::warn!(
                                    error = %e,
                                    "listrecentsessions: registry list_active failed — \
                                     degrading to local-only instances for this call"
                                );
                                degraded.push("registry");
                                Vec::new()
                            }
                        };
                        if let Some(idf) = identity_filter {
                            records.retain(|r| r.data.identity_id.as_deref() == Some(idf));
                        }
                        // Dedup by (definition_id, instance_name) keeping the newest
                        // launch — the registry read path lacks SQLite's chain-root
                        // collapse, so two fresh heads of one logical agent would
                        // otherwise double up. Sort newest-first within each group,
                        // collapse, then re-sort by recency.
                        records.sort_by(|a, b| {
                            a.data
                                .definition_id
                                .cmp(&b.data.definition_id)
                                .then_with(|| a.data.instance_name.cmp(&b.data.instance_name))
                                .then_with(|| {
                                    b.data.last_launched_at_ms.cmp(&a.data.last_launched_at_ms)
                                })
                        });
                        records.dedup_by(|a, b| {
                            a.data.definition_id == b.data.definition_id
                                && a.data.instance_name == b.data.instance_name
                        });
                        records.sort_by(|a, b| {
                            b.data.last_launched_at_ms.cmp(&a.data.last_launched_at_ms)
                        });
                        records.truncate(raw_limit);
                        // Local agents in PICKER mode (include_continuations=true):
                        // collapses each chain to one row AND surfaces orphan
                        // continuations (head hard-deleted) as their own root, so
                        // they don't vanish from "My Agents" (reagent P2). Indexed by
                        // (definition_id, instance_name) — the SAME key the registry
                        // dedup uses — keeping the newest, so overlay AND the
                        // local-only append agree on identity (reagent P1).
                        let local = mstore
                            .instance_list_named(raw_limit, None, identity_filter, true)
                            .unwrap_or_else(|e| {
                                tracing::warn!(
                                    error = %e,
                                    "listrecentsessions: local instance_list_named failed — \
                                     degrading to registry-only instances for this call"
                                );
                                degraded.push("local_instances");
                                Vec::new()
                            });
                        let mut local_by_key: std::collections::HashMap<
                            (String, String),
                            AgentInstance,
                        > = std::collections::HashMap::new();
                        for li in local {
                            let key = (li.definition_id.clone(), li.instance_name.clone());
                            match local_by_key.get(&key) {
                                Some(e) if e.started_at >= li.started_at => {}
                                _ => {
                                    local_by_key.insert(key, li);
                                }
                            }
                        }
                        let mut out: Vec<AgentInstance> = records
                            .into_iter()
                            .map(|rec| {
                                let d = rec.data;
                                let key = (d.definition_id.clone(), d.instance_name.clone());
                                if let Some(li) = local_by_key.get(&key) {
                                    // Defense-in-depth (SPEC_PANE_CLOSE_REOPEN_
                                    // CONTINUITY_GUARANTEE_2026_07_27.md §4.1):
                                    // the local row is normally the fresher
                                    // source (persist_session_id keeps it live
                                    // as of that spec), but fall back to the
                                    // registry's session_id when the local one
                                    // is empty rather than surfacing an empty
                                    // session id the picker's reattach flow
                                    // would silently treat as "start fresh."
                                    // Correct regardless of whether the
                                    // primary live-write fix is in place for
                                    // this particular row (e.g. it predates
                                    // that change, or the write raced/failed).
                                    if li.session_id.is_empty() {
                                        if let Some(ref registry_sid) = d.session_id {
                                            if !registry_sid.is_empty() {
                                                let mut merged = li.clone();
                                                merged.session_id = registry_sid.clone();
                                                return merged;
                                            }
                                        }
                                    }
                                    return li.clone();
                                }
                                // Reconstruct the absolute workdir from the record's
                                // source base (v3) or the current channel (legacy).
                                let working_directory = match d.source_agents_base.as_deref() {
                                    Some(src) => std::path::Path::new(src)
                                        .join(&d.working_dir)
                                        .to_string_lossy()
                                        .to_string(),
                                    None => match agents_root.as_ref() {
                                        Some(root) => root
                                            .join(&d.working_dir)
                                            .to_string_lossy()
                                            .to_string(),
                                        None => d.working_dir.clone(),
                                    },
                                };
                                AgentInstance {
                                    id: d.instance_id,
                                    definition_id: d.definition_id,
                                    parent_instance_id: String::new(),
                                    block_id: String::new(),
                                    session_id: d.session_id.unwrap_or_default(),
                                    status: "available".to_string(),
                                    github_context: String::new(),
                                    started_at: d.last_launched_at_ms,
                                    ended_at: 0,
                                    created_at: d.created_at_ms,
                                    identity_id: d.identity_id.unwrap_or_default(),
                                    memory_id: d.memory_id.unwrap_or_default(),
                                    instance_name: d.instance_name,
                                    working_directory,
                                    display_hidden: false,
                                }
                            })
                            .collect();
                        // APPEND local-only agents — those whose (definition_id,
                        // instance_name) no registry record represents (created
                        // before the live mirror could register them, or orphan
                        // continuations). Keyed identically to the dedup, so a deduped
                        // agent's local head never re-appears as a duplicate row.
                        let have_keys: std::collections::HashSet<(String, String)> = out
                            .iter()
                            .map(|i| (i.definition_id.clone(), i.instance_name.clone()))
                            .collect();
                        for (key, li) in local_by_key {
                            if !have_keys.contains(&key) {
                                out.push(li);
                            }
                        }
                        out
                    }
                    None => mstore
                        .instance_list_named(raw_limit, None, identity_filter, true)
                        .unwrap_or_else(|e| {
                            tracing::warn!(
                                error = %e,
                                "listrecentsessions: local instance_list_named failed \
                                 (no shared registry attached) — degrading to empty list"
                            );
                            degraded.push("local_instances");
                            Vec::new()
                        }),
                };

                let defs = mstore.agent_def_list().unwrap_or_else(|e| {
                    tracing::warn!(
                        error = %e,
                        "listrecentsessions: agent_def_list failed — degrading to empty \
                         (rows will show \"(missing definition)\")"
                    );
                    degraded.push("agent_defs");
                    Vec::new()
                });
                // Identity display names resolve off the direct
                // agent<->account links now (db_agent_identity_links /
                // db_accounts), not the retired bundle tables — see
                // SPEC_ARMORY_PHASE4_STORAGE_RENAME_COMPLETION_2026_07_12.md
                // §4 item 1. Bulk-fetched once and grouped by definition_id
                // rather than queried per-row.
                let agent_identity_links = identity_store.agent_identity_list_all().unwrap_or_else(|e| {
                    tracing::warn!(
                        error = %e,
                        // Fires alongside degraded.push("identity_links")
                        // below; the row loop then leaves the account line
                        // empty (a failed lookup is not "No auth").
                        "listrecentsessions: agent_identity_list_all failed — degrading to \
                         empty (rows will show no account line)"
                    );
                    degraded.push("identity_links");
                    Vec::new()
                });
                let accounts = id_store.identity_list(None).unwrap_or_else(|e| {
                    // The exact call PR #2296 broke (an oauth secret_ref serde-tag
                    // mismatch aborted this and, before this hardening, the whole
                    // handler with it) — now degrades in isolation instead.
                    tracing::warn!(
                        error = %e,
                        "listrecentsessions: identity_list failed — degrading to empty \
                         (the global mirror fills in what it can; a link it cannot resolve shows no account line)"
                    );
                    degraded.push("accounts");
                    Vec::new()
                });
                // The global mirror, read only when a link points at an account
                // this channel's store does not hold (an isolated channel).
                let own_ids: std::collections::HashSet<&str> = accounts.iter().map(|a| a.id.as_str()).collect();
                let mut mirror_failed = false;
                let mirror_accounts: Vec<IdentityAccount> =
                    if agent_identity_links.iter().any(|l| !own_ids.contains(l.account_id.as_str())) {
                        identity_store.identity_list(None).unwrap_or_else(|e| {
                            tracing::warn!(error = %e, "listrecentsessions: the global account mirror could not be read");
                            mirror_failed = true;
                            Vec::new()
                        })
                    } else {
                        Vec::new()
                    };
                let account_names = super::account_label::names_by_id(&accounts, &mirror_accounts);
                let mut unresolved_links = 0usize;
                let mut links_by_agent: std::collections::HashMap<&str, Vec<&AgentIdentityLink>> =
                    std::collections::HashMap::new();
                for link in &agent_identity_links {
                    links_by_agent
                        .entry(link.agent_id.as_str())
                        .or_default()
                        .push(link);
                }
                let memories = id_store.bundle_list().unwrap_or_else(|e| {
                    tracing::warn!(
                        error = %e,
                        "listrecentsessions: bundle_list failed — degrading to empty \
                         (rows will show \"(missing memory)\")"
                    );
                    degraded.push("memories");
                    Vec::new()
                });

                // docs/reports/REPORT_AGENT_PICKER_FIELD_ORDER_SORT_AND_DATA_GAPS_AUDIT_2026_08_24.md
                // §4: when `agent_identity_list_all()` itself failed above,
                // `links_by_agent` is empty for EVERY definition_id — not
                // just the ones that genuinely have no bound identity —
                // so every row would otherwise read "No auth", making a
                // source failure indistinguishable from a legitimate state.
                // The same goes for a failed accounts lookup that the global
                // mirror could not make up for: a link resolving to nothing
                // then proves nothing. Computed once, outside the loop: these
                // sources only ever degrade before the loop starts, not per-row.
                let degraded_lookups = super::account_label::Degraded {
                    links: degraded.contains(&"identity_links"),
                    accounts: degraded.contains(&"accounts") && (mirror_failed || mirror_accounts.is_empty()),
                };

                // Build rows. Hits filestore once per instance; with
                // raw_limit ≤ 500 and stat() being a single indexed
                // SQLite query, the per-call cost is dominated by
                // the eventual snapshot read for the top-20.
                let mut rows: Vec<RecentSessionRow> = Vec::with_capacity(instances.len());
                // What each agent was last asked comes from its global transcript
                // (keyed by agent id), not from this channel's per-block
                // snapshot, so a closed pane, an agent never launched here and a
                // registry-only row all get a real preview.
                let histories = crate::backend::agent_session::history_summary::read_histories(
                    instances.iter().map(|i| i.definition_id.clone()).collect(),
                )
                .await;
                if histories
                    .values()
                    .any(|h| matches!(h, crate::backend::agent_session::history_summary::HistoryRead::Failed))
                {
                    degraded.push("transcript");
                }
                for inst in instances {
                    // Consolidation Phase 3b: `definition_id` is the agent's
                    // own id, so its own row answers every display question
                    // (name, provider, model, type, created_at) — those are
                    // the agent's current values, which may well have drifted
                    // from the template it was launched off.
                    let def = defs.iter().find(|d| d.id == inst.definition_id);
                    // The template it came from, used ONLY for the identity-
                    // link fallback below (empty `parent_id` for an agent that
                    // was never cloned from one).
                    let template = def
                        .filter(|d| !d.parent_id.is_empty())
                        .and_then(|d| defs.iter().find(|t| t.id == d.parent_id));
                    // The agent's own account links, falling back to the
                    // template's. `db_agent_identity_links`' FK now targets
                    // `db_agents` (Phase 3c, #3088) — a launch-only agent
                    // CAN hold a link of its own — but nothing writes one
                    // there yet: the Agent Setup modal, the only identity
                    // binding UI, only ever targets a template or user
                    // agent's own definition id. Until a launch-scoped
                    // binding path exists, this fallback stays exactly as
                    // necessary as it was before the FK moved; removing it
                    // now would just make a launched agent's identity
                    // silently disappear. Retire alongside that writer path
                    // (Phase 3d).
                    let links_key = match links_by_agent.get(inst.definition_id.as_str()) {
                        Some(_) => inst.definition_id.as_str(),
                        None => template
                            .map(|t| t.id.as_str())
                            .unwrap_or(inst.definition_id.as_str()),
                    };
                    let line = super::account_label::account_line(
                        links_by_agent.get(links_key).map(|v| v.as_slice()),
                        &account_names,
                        degraded_lookups,
                    );
                    unresolved_links += line.unresolved;
                    let identity_name = line.text;
                    let memory_name = if inst.memory_id.is_empty() {
                        "(vanilla CLI)".to_string()
                    } else {
                        memories
                            .iter()
                            .find(|m| m.id == inst.memory_id)
                            .map(|m| m.name.clone())
                            .unwrap_or_else(|| "(missing memory)".to_string())
                    };

                    // Stat first (cheap) — gives us the modts for
                    // sorting. Only fetch the full content if the
                    // snapshot exists.
                    //
                    // docs/reports/REPORT_AGENT_PICKER_FIELD_ORDER_SORT_AND_DATA_GAPS_AUDIT_2026_08_24.md
                    // §5: `Ok(None)` (the block genuinely never wrote a
                    // snapshot) and `Err(_)` (a transient filestore I/O
                    // error, lock contention, DB read failure) used to
                    // collapse into the same `_ =>` arm — a real error on a
                    // row that DOES have history rendered identically to
                    // "never had one," silently. Split so the row can carry
                    // that distinction downstream instead of losing it here.
                    use crate::backend::agent_session::history_summary::HistoryRead;
                    // `node_count` is 0 here: counting messages would mean reading
                    // the whole transcript, so the tile shows none rather than a
                    // guess.
                    let from_transcript = match histories.get(&inst.definition_id) {
                        Some(HistoryRead::Transcript(h)) => Some((
                            true,
                            if h.last_activity_ms > 0 { h.last_activity_ms } else { inst.started_at },
                            h.last_user_message.clone(),
                            0usize,
                            false,
                        )),
                        Some(HistoryRead::Failed) => Some((false, inst.started_at, String::new(), 0usize, true)),
                        // No transcript: fall back to this channel's per-block
                        // snapshot, which old builds wrote.
                        Some(HistoryRead::NoTranscript) | None => None,
                    };
                    let (has_snapshot, last_active_at, mut preview, node_count, snapshot_check_failed) =
                        if let Some(t) = from_transcript {
                            t
                        } else if inst.block_id.is_empty() {
                            (false, inst.started_at, String::new(), 0usize, false)
                        } else {
                            match filestore.stat(&inst.block_id, "output.state.json") {
                                Ok(Some(file)) => {
                                    let modts = if file.modts > 0 {
                                        file.modts
                                    } else {
                                        inst.started_at
                                    };
                                    let (preview, node_count) = read_session_preview(
                                        &filestore,
                                        &inst.block_id,
                                    );
                                    (true, modts, preview, node_count, false)
                                }
                                Ok(None) => (false, inst.started_at, String::new(), 0usize, false),
                                Err(e) => {
                                    tracing::warn!(
                                        error = %e,
                                        definition_id = %inst.definition_id,
                                        block_id = %inst.block_id,
                                        "listrecentsessions: filestore.stat(output.state.json) failed — \
                                         row will report snapshot_check_failed instead of silently \
                                         looking like it never had a snapshot"
                                    );
                                    if !degraded.contains(&"snapshot_stat") {
                                        degraded.push("snapshot_stat");
                                    }
                                    (false, inst.started_at, String::new(), 0usize, true)
                                }
                            }
                        };

                    // No structured snapshot to preview — fall back to a
                    // once-per-definition, on-demand Haiku summary of the
                    // instance's raw terminal output (see
                    // docs/reports/REPORT_AGENT_PICKER_FIELD_ORDER_SORT_AND_DATA_GAPS_AUDIT_2026_08_24.md
                    // §5a). Only meaningful for a LOCAL row with a real
                    // block_id — a cross-channel row's synthetic empty
                    // block_id has no local raw output to summarize either,
                    // so it's left alone (still shows the plain fallback
                    // text, which is accurate for that case: this channel
                    // genuinely has nothing).
                    if !has_snapshot && !inst.block_id.is_empty() {
                        match mstore.agent_activity_summary_get(&inst.definition_id) {
                            Ok(Some(existing)) if !existing.summary.is_empty() => {
                                preview = existing.summary;
                            }
                            // `insert` returns `true` only the FIRST time this
                            // definition_id is claimed for this process's
                            // lifetime — mirrors `SubagentWatcher.naming_triggered`'s
                            // "claim once regardless of success/failure"
                            // pattern (reagent P1, PR #2786). Without this,
                            // `ambient::gateway().admit()` alone only
                            // coalesces CONCURRENT triggers, not sequential
                            // ones — a definition that can never be
                            // summarized (no cmd meta, no raw output, CLI
                            // failure) would otherwise be re-spawned on
                            // EVERY subsequent listrecentsessions call (tab
                            // visibility regain, agents:changed refetch, …)
                            // forever. A successful generation is unaffected
                            // by this gate: it persists to
                            // db_agent_activity_summaries, which the branch
                            // above already checks first.
                            // `Ok(Some(existing))` with an empty `summary`
                            // shouldn't occur in normal operation (a row is
                            // only ever inserted once generation succeeds —
                            // see the table's own doc comment) but is
                            // handled the same as `Ok(None)` regardless,
                            // rather than relying on that invariant to make
                            // the match exhaustive.
                            Ok(None) | Ok(Some(_))
                                if definition_summary_attempted().lock().unwrap().insert(inst.definition_id.clone()) =>
                            {
                                // Fire-and-forget: generation (a real Haiku
                                // CLI round-trip) must not block this
                                // response.
                                let mstore_bg = mstore.clone();
                                let filestore_bg = filestore.clone();
                                let broker_bg = broker.clone();
                                let definition_id_bg = inst.definition_id.clone();
                                let block_id_bg = inst.block_id.clone();
                                let provider_id_bg = def.map(|d| d.provider.clone()).unwrap_or_default();
                                tokio::spawn(async move {
                                    let result = crate::ambient::tasks::generate_definition_activity_summary(
                                        &mstore_bg,
                                        &filestore_bg,
                                        &broker_bg,
                                        &definition_id_bg,
                                        &block_id_bg,
                                        &provider_id_bg,
                                    )
                                    .await;
                                    // reagent P2, PR #2786: at minimum, log the
                                    // spend so it's observable — no UI-facing
                                    // total exists for ANY background-triggered
                                    // ambient call today (dispatch_name/
                                    // subagent_name via trigger_eager_naming,
                                    // generate_recovered_title all
                                    // discard their own returned tokens the
                                    // same way; only the live, RPC-response
                                    // pull paths — session:activity_summary,
                                    // session:next_prompt_suggestion — thread
                                    // usage to the frontend, since only they
                                    // have a response channel to thread it
                                    // through). A real aggregate sink for
                                    // background ambient spend is a separate,
                                    // cross-cutting follow-up, not scoped to
                                    // this one call site.
                                    if let Some(generated) = result {
                                        if let Some(tokens) = generated.tokens {
                                            tracing::debug!(
                                                definition_id = %definition_id_bg,
                                                ?tokens,
                                                "listrecentsessions: definition activity summary generated"
                                            );
                                        }
                                    }
                                });
                            }
                            Ok(None) | Ok(Some(_)) => {
                                // Already attempted this process lifetime —
                                // no-op, same as any other row without a
                                // persisted summary.
                            }
                            Err(e) => {
                                tracing::warn!(
                                    error = %e,
                                    definition_id = %inst.definition_id,
                                    "listrecentsessions: agent_activity_summary_get failed — \
                                     row keeps its plain fallback preview text"
                                );
                                if !degraded.contains(&"activity_summary") {
                                    degraded.push("activity_summary");
                                }
                            }
                        }
                    }

                    rows.push(RecentSessionRow {
                        instance_id: inst.id,
                        instance_name: inst.instance_name,
                        definition_id: inst.definition_id.clone(),
                        definition_name: def
                            .map(|d| d.name.clone())
                            .unwrap_or_else(|| "(missing definition)".to_string()),
                        provider: def.map(|d| d.provider.clone()).unwrap_or_default(),
                        model_vendor_base_url: def.map(|d| d.model_vendor_base_url.clone()).unwrap_or_default(),
                        working_directory: inst.working_directory,
                        identity_id: inst.identity_id,
                        identity_name,
                        memory_id: inst.memory_id,
                        memory_name,
                        block_id_hint: inst.block_id,
                        // Surface the CLI-captured session id so the
                        // picker reattach can `--resume <sid>` on the
                        // FIRST turn of the new block. Without this
                        // the new subprocess starts a fresh session
                        // and the CLI re-injects the startup context.
                        session_id: inst.session_id,
                        preview,
                        node_count,
                        last_active_at,
                        has_snapshot,
                        snapshot_check_failed,
                        agent_created_at: def.map(|d| d.created_at).unwrap_or(0),
                        started_at: inst.started_at,
                        agent_type: def.map(|d| d.agent_type.clone()).unwrap_or_default(),
                    });
                }

                // Sort: rows with a snapshot first (descending by
                // modts), then no-snapshot rows by started_at desc.
                // This keeps live conversations at the top while
                // still surfacing legacy rows.
                rows.sort_by(|a, b| match (a.has_snapshot, b.has_snapshot) {
                    (true, true) | (false, false) => {
                        b.last_active_at.cmp(&a.last_active_at)
                    }
                    (true, false) => std::cmp::Ordering::Less,
                    (false, true) => std::cmp::Ordering::Greater,
                });
                rows.truncate(limit);

                // Command-level completion trace (retro §9 rec 3) — the
                // prior incident's own log had no way to show whether this
                // RPC ran, succeeded, or returned degraded data; this line
                // answers all three going forward. `degraded` is non-empty
                // only when at least one source above fell back to empty —
                // an all-clear call logs an empty list here.
                tracing::info!(
                    rows = rows.len(),
                    degraded = ?degraded,
                    // Links to an account this store cannot find (deleted, or an
                    // unparseable secret_ref skipped by identity_list): those
                    // agents show "No auth", and this says how many.
                    unresolved_account_links = unresolved_links,
                    elapsed_ms = t0.elapsed().as_millis() as u64,
                    "listrecentsessions: completed"
                );

                // `degraded` must also reach the CALLER, not just the log
                // (reagent P1 on PR #2327): once every source degrades to
                // empty instead of erroring the whole RPC, a transport-level
                // success/failure check can no longer tell "genuinely zero
                // agents" apart from "a source failed and we got nothing" —
                // the exact ambiguity this hardening exists to close, just
                // pushed from the RPC layer down into the response body.
                // See ListRecentSessionsResult's own doc comment.
                let result = ListRecentSessionsResult {
                    rows,
                    degraded: degraded.iter().map(|s| s.to_string()).collect(),
                };
                Ok(result)
        }
    });

    // ---- agent:session:read ----
    let filestore = state.filestore.clone();
    engine.register_typed(
        COMMAND_AGENT_SESSION_READ,
        move |cmd: CommandAgentSessionReadData, _ctx| {
            let filestore = filestore.clone();
            async move {
                let (content, modts) =
                    crate::backend::agent_session::read_session_state(&filestore, &cmd.definition_id)
                        .map_err(|e| format!("agent:session:read: {e}"))?;
                Ok(AgentSessionReadResult { content, modts })
            }
        },
    );

    // ---- agent:session:write_state ----
    let filestore = state.filestore.clone();
    engine.register_typed(
        COMMAND_AGENT_SESSION_WRITE_STATE,
        move |cmd: CommandAgentSessionWriteStateData, _ctx| {
            let filestore = filestore.clone();
            async move {
                let bytes = cmd.content.as_bytes();
                let bytes_written = bytes.len() as u64;
                crate::backend::agent_session::write_session_state(
                    &filestore,
                    &cmd.definition_id,
                    bytes,
                )
                .map_err(|e| format!("agent:session:write_state: {e}"))?;
                Ok(AgentSessionWriteStateResult { bytes_written })
            }
        },
    );

    // ---- agent:session:append_output ----
    let filestore = state.filestore.clone();
    engine.register_typed(
        COMMAND_AGENT_SESSION_APPEND_OUTPUT,
        move |cmd: CommandAgentSessionAppendOutputData, _ctx| {
            let filestore = filestore.clone();
            async move {
                let bytes_written = crate::backend::agent_session::append_session_output(
                    &filestore,
                    &cmd.definition_id,
                    &cmd.line,
                )
                .map_err(|e| format!("agent:session:append_output: {e}"))?;
                Ok(AgentSessionAppendOutputResult { bytes_written })
            }
        },
    );

    // ---- agent:session:archive ----
    let filestore = state.filestore.clone();
    engine.register_typed(
        COMMAND_AGENT_SESSION_ARCHIVE,
        move |cmd: CommandAgentSessionArchiveData, _ctx| {
            let filestore = filestore.clone();
            async move {
                let result =
                    crate::backend::agent_session::archive_session(&filestore, &cmd.definition_id)
                        .map_err(|e| format!("agent:session:archive: {e}"))?;
                let (archive_zoneid, archived_at_ms) = match result {
                    Some((z, ts)) => (z, ts),
                    None => (String::new(), 0),
                };
                Ok(AgentSessionArchiveResult {
                    archive_zoneid,
                    archived_at_ms,
                })
            }
        },
    );

    // ---- agent:session:list_archives ----
    let filestore = state.filestore.clone();
    engine.register_typed(
        COMMAND_AGENT_SESSION_LIST_ARCHIVES,
        move |cmd: CommandAgentSessionListArchivesData, _ctx| {
            let filestore = filestore.clone();
            async move {
                let summaries = crate::backend::agent_session::list_archives(
                    &filestore,
                    &cmd.definition_id,
                    cmd.limit,
                )
                .map_err(|e| format!("agent:session:list_archives: {e}"))?;
                let rows: Vec<AgentArchiveRow> = summaries
                    .into_iter()
                    .map(|s| AgentArchiveRow {
                        archive_zoneid: s.archive_zoneid,
                        archived_at_ms: s.archived_at_ms,
                        preview: s.preview,
                        node_count: s.node_count,
                    })
                    .collect();
                Ok(rows)
            }
        },
    );
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::storage::store::AgentDefinition;
    use crate::backend::rpc_types::RpcMessage;
    use crate::registry::{
        DefinitionRecord, DefinitionRecordV1, DefinitionStore, NamedAgentRecord,
        NamedAgentRecordV1, Registry,
    };
    use crate::server::tests::test_state;

    fn named_record(instance_id: &str, definition_id: &str) -> NamedAgentRecord {
        NamedAgentRecord {
            schema_version: 1,
            data: NamedAgentRecordV1 {
                instance_id: instance_id.to_string(),
                instance_name: format!("session-{instance_id}"),
                definition_id: definition_id.to_string(),
                identity_id: None,
                memory_id: None,
                session_id: None,
                working_dir: format!("{definition_id}-workdir"),
                source_agents_base: None,
                created_at_ms: 1,
                last_launched_at_ms: 1,
                created_by_version: "test".to_string(),
                last_launched_by_version: "test".to_string(),
            },
        }
    }

    fn definition_record(id: &str) -> DefinitionRecord {
        DefinitionRecord {
            schema_version: 1,
            data: DefinitionRecordV1 {
                id: id.to_string(),
                name: format!("Agent {id}"),
                provider: "claude".to_string(),
                is_seeded: 0,
                ..Default::default()
            },
        }
    }

    /// Seeds `state.mstore` with N cross-channel registry + definition
    /// records (as they'd exist on a REAL machine before a fresh channel
    /// ever boots) and returns the dispatch machinery ready to call
    /// `listrecentsessions`.
    fn setup_with_n_cross_channel_agents(
        n: usize,
    ) -> (
        AppState,
        std::sync::Arc<WshRpcEngine>,
        tokio::sync::mpsc::UnboundedReceiver<RpcMessage>,
        tempfile::TempDir,
        tempfile::TempDir,
    ) {
        let state = test_state();
        let registry_dir = tempfile::tempdir().unwrap();
        let def_dir = tempfile::tempdir().unwrap();
        let registry = Registry::open(registry_dir.path().to_path_buf()).unwrap();
        let def_store = DefinitionStore::open(def_dir.path().to_path_buf()).unwrap();
        for i in 0..n {
            let id = format!("inst-{i}");
            let def_id = format!("def-{i}");
            registry.upsert(&named_record(&id, &def_id)).unwrap();
            def_store.upsert(&definition_record(&def_id)).unwrap();
        }
        state.mstore.set_registry(std::sync::Arc::new(registry));
        state.mstore.set_def_registry(std::sync::Arc::new(def_store));

        let (engine, output_rx) = WshRpcEngine::new();
        register(&engine, &state);
        (state, engine, output_rx, registry_dir, def_dir)
    }

    async fn dispatch_list_recent_sessions(
        engine: &std::sync::Arc<WshRpcEngine>,
        output_rx: &mut tokio::sync::mpsc::UnboundedReceiver<RpcMessage>,
    ) -> RpcMessage {
        engine.handle_message(RpcMessage {
            command: COMMAND_LIST_RECENT_SESSIONS.to_string(),
            reqid: "req-1".to_string(),
            data: Some(serde_json::json!({ "limit": 20 })),
            ..Default::default()
        });
        tokio::time::timeout(std::time::Duration::from_secs(2), output_rx.recv())
            .await
            .unwrap()
            .unwrap()
    }

    /// Retro
    /// `docs/retro/retro-my-agents-fresh-channel-regression-2026-07-27.md`
    /// §9 rec 5: this exact path (fresh channel + pre-populated global
    /// registry) had broken three separate times with no regression test
    /// covering it. Directly exercises the RPC end-to-end against a real
    /// `Registry`/`DefinitionStore` on disk — the same code path a fresh
    /// `task package` channel hits on its very first "My Agents" fetch.
    #[tokio::test]
    async fn cross_channel_registry_agents_populate_a_fresh_channels_my_agents_list() {
        let (_state, engine, mut output_rx, _reg_dir, _def_dir) =
            setup_with_n_cross_channel_agents(3);

        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;

        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult =
            serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        assert_eq!(
            result.rows.len(),
            3,
            "all 3 cross-channel registry agents must appear on a channel that never created any of them locally"
        );
        assert!(
            result.degraded.is_empty(),
            "a fully healthy call must report no degraded sources, got: {:?}",
            result.degraded
        );
    }

    /// Retro §4/§9 rec 1 — the core hardening this test locks in. Before
    /// this fix, `identity_list()` failing on one unparseable `db_accounts`
    /// row (PR #2296's exact incident: an oauth secret_ref serde-tag
    /// mismatch) `?`-aborted the ENTIRE `listrecentsessions` response,
    /// zeroing "My Agents" even though the registry/definitions themselves
    /// were completely healthy. Reproduces that exact malformed-row shape
    /// directly via raw SQL (bypassing `identity_upsert`, which would
    /// never produce an invalid `secret_ref`) and asserts the response
    /// still contains every agent — only the identity-name enrichment
    /// degrades, not the whole list.
    #[tokio::test]
    async fn a_malformed_identity_account_row_is_tolerated_without_degrading_the_accounts_source() {
        let (state, engine, mut output_rx, _reg_dir, _def_dir) =
            setup_with_n_cross_channel_agents(2);

        // Same malformed shape as the real PR #2296 incident: a
        // `secret_ref.backend` tag no `SecretRef` variant matches.
        //
        // Updated for ANALYSIS_ARMORY_STASH_CREDENTIAL_VISIBILITY_GAP_2026_08_04:
        // `identity_list` itself now skips a malformed row (with a warning)
        // instead of erroring the whole call (backend/storage/identities.rs),
        // so this scenario no longer reaches the `unwrap_or_else` degrade
        // path below at all — the "accounts" source is no longer degraded
        // by a single bad row, it's just quietly correct. The `degraded`
        // push at that call site still exists and is still meaningful for a
        // genuine `identity_list` error (e.g. a real DB failure); it's
        // simply no longer reachable via this specific fixture.
        {
            let conn = state.mstore.conn().lock().unwrap();
            conn.execute(
                "INSERT INTO db_accounts
                    (id, name, provider, kind, display_name, secret_ref, context, status, created_at, updated_at)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
                rusqlite::params![
                    "acct-broken",
                    "broken-account",
                    "claude",
                    "oauth",
                    "",
                    r#"{"backend":"totally_bogus_backend"}"#,
                    "{}",
                    "unknown",
                    0i64,
                    0i64,
                ],
            )
            .unwrap();
        }

        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;

        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult =
            serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        assert_eq!(
            result.rows.len(),
            2,
            "a single malformed db_accounts row must not zero out the whole \
             My Agents list"
        );
        // The improved behavior: identity_list's own per-row tolerance means
        // this is no longer a "source failure" at all — nothing degraded,
        // the malformed row was just silently excluded. Contrast with the
        // OLD assertion this test used to make (`degraded` contains
        // "accounts") — that was evidence of a real, if survivable, failure;
        // this is evidence there wasn't one. The `degraded.push("accounts")`
        // call site (above, in the handler) is still meaningful for a
        // genuine `identity_list` error — just no longer reachable via this
        // fixture.
        assert!(
            !result.degraded.contains(&"accounts".to_string()),
            "a single malformed row must not degrade the accounts source at all \
             (identity_list now tolerates it internally), got: {:?}",
            result.degraded
        );
    }

    /// docs/reports/REPORT_AGENT_PICKER_FIELD_ORDER_SORT_AND_DATA_GAPS_AUDIT_2026_08_24.md
    /// §4: when `agent_identity_list_all()` itself fails, no row can say whether
    /// its agent is bound. A failure must not read like a healthy "No auth", so
    /// the account line is left empty (SPEC_MY_AGENTS_TILES_AUTH_AND_HISTORY_2026_10_03.md
    /// §5.2 R5). Forces the failure by dropping `db_agent_identity_links` outright
    /// and asserts every row has an empty account line, with `degraded`
    /// reporting "identity_links".
    #[tokio::test]
    async fn identity_links_source_failure_leaves_the_account_line_empty_not_no_auth() {
        let (state, engine, mut output_rx, _reg_dir, _def_dir) =
            setup_with_n_cross_channel_agents(2);

        {
            let conn = state.mstore.conn().lock().unwrap();
            conn.execute("DROP TABLE db_agent_identity_links", []).unwrap();
        }

        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;

        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult =
            serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        assert_eq!(result.rows.len(), 2, "the identity_links failure must not zero out the whole list");
        assert!(
            result.degraded.contains(&"identity_links".to_string()),
            "got: {:?}",
            result.degraded
        );
        for row in &result.rows {
            assert_eq!(
                row.identity_name, "",
                "a source FAILURE must claim nothing: an unbound agent reads \"No auth\" and a \
                 backend outage must not look identical to that"
            );
        }
    }

    /// A failed ACCOUNTS lookup (the global mirror unreadable too) must not turn a
    /// bound agent into "No auth": its account may be perfectly good. The linked
    /// agent's line is left empty; an agent with no link is still plainly unbound.
    #[tokio::test]
    async fn a_failed_accounts_lookup_leaves_a_linked_agents_line_empty_not_no_auth() {
        let (state, engine, mut output_rx, _reg_dir, _def_dir) = setup_with_n_cross_channel_agents(2);
        {
            let conn = state.identity_store.conn().lock().unwrap();
            conn.execute_batch(
                "PRAGMA foreign_keys=OFF; INSERT INTO db_agent_identity_links (agent_id, account_id, provider)                  VALUES ('def-0', 'acct-that-is-fine', 'claude');",
            )
            .unwrap();
        }
        {
            let conn = state.id_store.conn().lock().unwrap();
            conn.execute_batch("PRAGMA foreign_keys=OFF; DROP TABLE db_accounts;").unwrap();
        }
        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;
        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult = serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        assert!(result.degraded.contains(&"accounts".to_string()), "got: {:?}", result.degraded);
        let line = |def: &str| result.rows.iter().find(|r| r.definition_id == def).unwrap().identity_name.clone();
        assert_eq!(line("def-0"), "", "a link we cannot resolve because the lookup failed is unknown, not unbound");
        assert_eq!(line("def-1"), "No auth", "an agent with no link is unbound whatever the accounts lookup did");
    }

    /// A healthy list: agents with no account link read "No auth", and none of the
    /// retired labels appears anywhere.
    #[tokio::test]
    async fn an_agent_with_no_account_link_says_no_auth_and_the_old_labels_are_gone() {
        let (_state, engine, mut output_rx, _reg_dir, _def_dir) = setup_with_n_cross_channel_agents(2);
        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;
        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult = serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        assert_eq!(result.rows.len(), 2);
        for row in &result.rows {
            assert_eq!(row.identity_name, "No auth");
            for old in ["(missing account)", "(ambient creds)", "(unknown account)"] {
                assert_ne!(row.identity_name, old);
            }
        }
    }

    /// Minimal local (not cross-channel registry) `AgentDefinition` —
    /// mirrors `storage/agents/mod.rs`'s own private `test_agent_def` test fixture
    /// shape, duplicated here rather than widening that helper's
    /// visibility for one caller.
    fn local_agent_def(id: &str) -> AgentDefinition {
        AgentDefinition {
            id: id.to_string(),
            slug: id.to_string(),
            name: format!("Agent {id}"),
            icon: String::new(),
            provider: "claude".to_string(),
            description: String::new(),
            working_directory: String::new(),
            shell: String::new(),
            provider_flags: String::new(),
            auto_start: 0,
            restart_on_crash: 0,
            idle_timeout_minutes: 0,
            created_at: 1,
            agent_type: "standalone".to_string(),
            environment: String::new(),
            agent_bus_id: String::new(),
            is_seeded: 0,
            accounts: String::new(),
            parent_id: String::new(),
            branch_label: String::new(),
            updated_at: 1,
            user_hidden: 0,
            container_image: String::new(),
            container_volumes: "[]".to_string(),
            container_name: String::new(),
            use_ambient_login: 0,
            model_vendor_base_url: String::new(),
            auto_continue_enabled: 0,
            memory_id: String::new(),
            conversation_visibility: "private".to_string(),
        }
    }

    fn local_agent_instance(id: &str, definition_id: &str, block_id: &str) -> AgentInstance {
        AgentInstance {
            id: id.to_string(),
            definition_id: definition_id.to_string(),
            parent_instance_id: String::new(),
            block_id: block_id.to_string(),
            session_id: String::new(),
            status: "available".to_string(),
            github_context: String::new(),
            started_at: 1,
            ended_at: 0,
            created_at: 1,
            identity_id: String::new(),
            memory_id: String::new(),
            instance_name: id.to_string(),
            working_directory: String::new(),
            display_hidden: false,
        }
    }

    /// docs/reports/REPORT_AGENT_PICKER_FIELD_ORDER_SORT_AND_DATA_GAPS_AUDIT_2026_08_24.md
    /// §5a: a row with no structured `output.state.json` snapshot but a
    /// PERSISTED `db_agent_activity_summaries` row for its definition must
    /// surface that summary as `preview`, still reporting `has_snapshot:
    /// false` (accurate — there genuinely is no structured snapshot; the
    /// summary is a substitute, not a claim that one exists).
    #[tokio::test]
    async fn snapshot_less_row_uses_a_persisted_definition_summary_as_preview() {
        let state = test_state();
        let mut def = local_agent_def("def-local-1");
        state.mstore.agent_def_insert(&mut def).unwrap();
        state
            .mstore
            .instance_create(&local_agent_instance("inst-local-1", "def-local-1", "block-local-1"))
            .unwrap();
        state
            .mstore
            .agent_activity_summary_set("def-local-1", "Fixed the login race condition", 1000)
            .unwrap();

        let (engine, mut output_rx) = WshRpcEngine::new();
        register(&engine, &state);
        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;

        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult =
            serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        assert_eq!(result.rows.len(), 1);
        let row = &result.rows[0];
        assert!(!row.has_snapshot, "no output.state.json was ever written for this instance");
        assert_eq!(
            row.preview, "Fixed the login race condition",
            "persisted definition summary must surface as the row's preview \
             when there's no real snapshot to preview instead"
        );
    }

    /// Counterpart to the above: when NOTHING is persisted yet for the
    /// definition, the row's `preview` stays empty and the frontend shows its
    /// own empty-state text ("No conversations yet").
    /// Generation itself (a real Haiku CLI round-trip) is fire-and-forget
    /// and not asserted here — this only locks in that a missing summary
    /// never surfaces as a fabricated non-empty preview.
    #[tokio::test]
    async fn snapshot_less_row_with_no_persisted_summary_leaves_preview_empty() {
        let state = test_state();
        let mut def = local_agent_def("def-local-2");
        state.mstore.agent_def_insert(&mut def).unwrap();
        state
            .mstore
            .instance_create(&local_agent_instance("inst-local-2", "def-local-2", "block-local-2"))
            .unwrap();

        let (engine, mut output_rx) = WshRpcEngine::new();
        register(&engine, &state);
        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;

        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult =
            serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        assert_eq!(result.rows.len(), 1);
        let row = &result.rows[0];
        assert!(!row.has_snapshot);
        assert_eq!(row.preview, "", "no summary persisted yet — preview must stay empty, not fabricated");
    }

    /// Put an `output` file in the (test) global transcript store for `def_id`.
    fn install_transcript(store: &std::sync::Arc<crate::backend::storage::filestore::FileStore>, def_id: &str, ndjson: &str) {
        let zone = format!("agent:{def_id}:current");
        store
            .make_file(&zone, "output", Default::default(), crate::backend::storage::filestore::FileOpts::default())
            .unwrap();
        store.write_file(&zone, "output", ndjson.as_bytes()).unwrap();
    }

    fn user_line(text: &str) -> String {
        serde_json::json!({"type":"user","message":{"role":"user","content":text}}).to_string()
    }

    /// Clears the thread's test transcript store when a test ends, pass or fail.
    struct TranscriptStoreGuard;
    impl Drop for TranscriptStoreGuard {
        fn drop(&mut self) {
            crate::backend::agent_session::history_summary::set_test_store(None);
        }
    }

    /// The preview, last-active time and "has history" flag come from the agent's
    /// global transcript, so a registry-only row (no local instance, no block id)
    /// with a transcript shows what it was last asked.
    #[tokio::test]
    async fn a_row_with_no_block_takes_its_preview_from_the_global_transcript() {
        let (_state, engine, mut output_rx, _reg_dir, _def_dir) = setup_with_n_cross_channel_agents(2);
        let store = std::sync::Arc::new(crate::backend::storage::filestore::FileStore::open_in_memory().unwrap());
        install_transcript(&store, "def-0", &format!("{}\n{}", user_line("# Session Context\nboot"), user_line("ship the tiles")));
        crate::backend::agent_session::history_summary::set_test_store(Some(store));
        let _guard = TranscriptStoreGuard;

        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;
        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult = serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        let with = result.rows.iter().find(|r| r.definition_id == "def-0").unwrap();
        assert!(with.block_id_hint.is_empty(), "the row has no local block");
        assert!(with.has_snapshot);
        assert!(!with.snapshot_check_failed);
        assert_eq!(with.preview, "ship the tiles");
        assert!(with.last_active_at > 1, "last-active comes from the transcript, got {}", with.last_active_at);
        let without = result.rows.iter().find(|r| r.definition_id == "def-1").unwrap();
        assert!(!without.has_snapshot, "an agent with no transcript has no history");
        assert!(!without.snapshot_check_failed, "no transcript is a state, not a failed read");
        assert_eq!(without.preview, "");
    }

    /// A transcript that holds no user prompt is still history: the row says so
    /// (has_snapshot) rather than claiming the agent never ran.
    #[tokio::test]
    async fn a_transcript_without_a_prompt_still_counts_as_history() {
        let (_state, engine, mut output_rx, _reg_dir, _def_dir) = setup_with_n_cross_channel_agents(1);
        let store = std::sync::Arc::new(crate::backend::storage::filestore::FileStore::open_in_memory().unwrap());
        install_transcript(&store, "def-0", r#"{"type":"stream_event","event":{}}"#);
        crate::backend::agent_session::history_summary::set_test_store(Some(store));
        let _guard = TranscriptStoreGuard;

        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;
        let result: ListRecentSessionsResult = serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        assert_eq!(result.rows.len(), 1);
        assert!(result.rows[0].has_snapshot);
        assert_eq!(result.rows[0].preview, "");
    }

    /// Consolidation Phase 3b (reagent P1 on PR #3080): the row's display
    /// fields must come from the AGENT's own row, never from the template it
    /// was launched off. The template is consulted for exactly one thing —
    /// the legacy-shaped identity-link fallback — and an agent whose
    /// provider/name/type has since drifted from its template must report its
    /// own current values, as it did before consolidation.
    #[tokio::test]
    async fn a_launched_agents_row_reports_its_own_config_not_its_templates() {
        let state = test_state();
        let mut tmpl = local_agent_def("tmpl-drift");
        tmpl.is_seeded = 1;
        tmpl.name = "Template".to_string();
        tmpl.provider = "claude".to_string();
        tmpl.agent_type = "standalone".to_string();
        state.mstore.agent_def_insert(&mut tmpl).unwrap();

        let mut own = local_agent_def("agent-drift");
        own.parent_id = "tmpl-drift".to_string();
        state.mstore.agent_def_insert(&mut own).unwrap();
        state
            .mstore
            .instance_create(&local_agent_instance("launch-1", "agent-drift", "block-drift"))
            .unwrap();

        // The agent drifts from the template it was cloned from.
        own.name = "Renamed by the user".to_string();
        own.provider = "codex".to_string();
        own.agent_type = "orchestrator".to_string();
        state.mstore.agent_def_update(&mut own).unwrap();

        let (engine, mut output_rx) = WshRpcEngine::new();
        register(&engine, &state);
        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;

        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult =
            serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        assert_eq!(result.rows.len(), 1);
        let row = &result.rows[0];
        assert_eq!(row.definition_name, "Renamed by the user");
        assert_eq!(row.provider, "codex", "the template's 'claude' must not shadow the agent's own provider");
        assert_eq!(row.agent_type, "orchestrator");
    }

    /// docs/reports/REPORT_AGENT_PICKER_FIELD_ORDER_SORT_AND_DATA_GAPS_AUDIT_2026_08_24.md
    /// §5, bullet 1: `filestore.stat()`'s `Err` (I/O error, lock
    /// contention, DB read failure) used to collapse into the exact same
    /// branch as `Ok(None)` (genuinely no snapshot) — a transient storage
    /// error on a row that DOES have real history rendered identically to
    /// "never had one," with no log line. Forces the failure
    /// deterministically the same way the identity_links test above does
    /// (`DROP TABLE`), rather than relying on real I/O flakiness.
    #[tokio::test]
    async fn filestore_stat_error_is_distinguished_from_genuinely_no_snapshot() {
        let state = test_state();
        let mut def = local_agent_def("def-local-3");
        state.mstore.agent_def_insert(&mut def).unwrap();
        state
            .mstore
            .instance_create(&local_agent_instance("inst-local-3", "def-local-3", "block-local-3"))
            .unwrap();

        {
            let conn = state.filestore.conn().lock().unwrap();
            conn.execute("DROP TABLE db_wave_file", []).unwrap();
        }

        let (engine, mut output_rx) = WshRpcEngine::new();
        register(&engine, &state);
        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;

        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult =
            serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        assert_eq!(result.rows.len(), 1, "a stat() failure on one row must not zero out the whole list");
        let row = &result.rows[0];
        assert!(!row.has_snapshot, "a failed stat() is not evidence a snapshot exists");
        assert!(
            row.snapshot_check_failed,
            "a stat() I/O error must be distinguishable from a genuine Ok(None) — \
             otherwise a transient failure looks identical to \"never had history\""
        );
        assert!(
            result.degraded.contains(&"snapshot_stat".to_string()),
            "got: {:?}",
            result.degraded
        );
    }

    /// Counterpart: a genuine `Ok(None)` (the block really never wrote a
    /// snapshot) must NOT set `snapshot_check_failed` — only an actual
    /// `Err` does. Without this, the fix above could be made to pass by
    /// setting the flag unconditionally whenever `has_snapshot` is false,
    /// which would just move the ambiguity rather than remove it.
    #[tokio::test]
    async fn genuinely_missing_snapshot_does_not_set_the_check_failed_flag() {
        let state = test_state();
        let mut def = local_agent_def("def-local-4");
        state.mstore.agent_def_insert(&mut def).unwrap();
        state
            .mstore
            .instance_create(&local_agent_instance("inst-local-4", "def-local-4", "block-local-4"))
            .unwrap();

        let (engine, mut output_rx) = WshRpcEngine::new();
        register(&engine, &state);
        let resp = dispatch_list_recent_sessions(&engine, &mut output_rx).await;

        assert!(resp.error.is_empty(), "unexpected error: {}", resp.error);
        let result: ListRecentSessionsResult =
            serde_json::from_value(resp.data.expect("expected result data")).unwrap();
        let row = &result.rows[0];
        assert!(!row.has_snapshot);
        assert!(
            !row.snapshot_check_failed,
            "a genuine Ok(None) — the block simply never wrote a snapshot — is not a check failure"
        );
    }

    /// reagent P1, PR #2786: without a dedup gate, a definition that can
    /// never be summarized (no cmd meta, no raw output, CLI failure) gets
    /// re-spawned on every `listrecentsessions` poll forever —
    /// `ambient::gateway().admit()` alone only coalesces CONCURRENT
    /// triggers, not sequential ones, since its guard drops (freeing the
    /// slot) the moment one call finishes. This tests
    /// `definition_summary_attempted()`'s claim set directly rather than
    /// the RPC dispatch (which can't observe whether a background
    /// `tokio::spawn` happened without mocking the Haiku CLI) — a unique
    /// id avoids cross-test interference with the process-wide static set.
    #[test]
    fn definition_summary_attempted_claims_a_definition_id_at_most_once() {
        let id = "def-dedup-test-unique-9f3a1b".to_string();
        assert!(
            definition_summary_attempted().lock().unwrap().insert(id.clone()),
            "first claim for a fresh id must succeed"
        );
        assert!(
            !definition_summary_attempted().lock().unwrap().insert(id.clone()),
            "second claim for the same id must be rejected — this is the fix for \
             reagent's unbounded-retry finding on PR #2786"
        );
    }
}