Skip to main content

supercode_interchange/
catalog.rs

1//! Discovery and durable addressing for sessions written by external harnesses.
2//!
3//! [`HarnessCatalog`] is intentionally about persisted state. It does not
4//! claim that the process which wrote a session is still alive or attachable.
5
6use std::collections::{BTreeSet, HashMap, HashSet};
7use std::fs::{self, File};
8use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
9use std::path::{Component, Path, PathBuf};
10use std::time::UNIX_EPOCH;
11
12use rusqlite::Connection;
13use serde::{Deserialize, Serialize};
14use serde_json::Value;
15
16use crate::native_store::load_native_store_family;
17use crate::ontology::{Binding, OrchestratorBindingRow};
18use crate::session::{
19    hermes_capture_nouns, openclaw_agent_id_from_path, openclaw_capture_header_nouns,
20    percent_decode_path, OrchestrationNouns, SessionMeta, SessionSource,
21};
22use crate::{Error, Fidelity, Result, Session, SessionFollower};
23
24pub use crate::ontology::HarnessId;
25
26/// Durable storage address for a persisted session.
27#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
28#[serde(tag = "kind", rename_all = "snake_case")]
29pub enum StorageLocator {
30    /// One session stored in one file.
31    File {
32        /// Absolute or caller-resolvable path to the transcript.
33        path: PathBuf,
34    },
35    /// One logical session selected from a SQLite store.
36    Sqlite {
37        /// Path to the SQLite database.
38        path: PathBuf,
39        /// Harness-native stable selector, currently an OpenCode session id.
40        selector: String,
41    },
42}
43
44impl StorageLocator {
45    /// Return the underlying file or database path.
46    pub fn path(&self) -> &Path {
47        match self {
48            Self::File { path } | Self::Sqlite { path, .. } => path,
49        }
50    }
51}
52
53/// Stable identity for a persisted harness session.
54#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
55pub struct SessionLocator {
56    /// Harness which owns the storage format.
57    pub harness: HarnessId,
58    /// Harness-native session identity.
59    pub session_id: String,
60    /// Exact storage address needed to load the session again.
61    pub storage: StorageLocator,
62}
63
64/// Lightweight metadata returned by catalog discovery.
65#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
66pub struct SessionDescriptor {
67    /// Durable address accepted by [`HarnessCatalog::load`] and
68    /// [`HarnessCatalog::follow`].
69    pub locator: SessionLocator,
70    /// Working directory recorded by the harness.
71    pub cwd: Option<PathBuf>,
72    /// Harness-provided title, when cheaply available.
73    pub title: Option<String>,
74    /// Oldest-first bounded conversation messages for a fallback topic when
75    /// the harness does not publish a useful title. These are read only for
76    /// the returned page and interpreted by the same presentation projection
77    /// as an opened conversation.
78    #[serde(default, skip_serializing_if = "Vec::is_empty")]
79    pub preview_candidates: Vec<SessionPreviewCandidate>,
80    /// Newest-first bounded conversation messages for compact list previews.
81    /// These are read only for the returned page, never for the entire
82    /// catalog, and are interpreted by the same presentation projection as
83    /// an opened conversation.
84    #[serde(default, skip_serializing_if = "Vec::is_empty")]
85    pub latest_message_candidates: Vec<SessionPreviewCandidate>,
86    /// Time of the last recorded TURN as Unix epoch milliseconds.
87    ///
88    /// Not the transcript file's mtime: a harness rewrites its transcript for
89    /// reasons that are not conversation (Claude Code appends untimestamped
90    /// `bridge-session` records while a session merely sits open, and moves
91    /// mtime again on resume), so mtime ranks idle-but-open sessions above
92    /// genuinely active ones. Falls back to mtime only when no timestamped
93    /// record is readable.
94    pub updated_at_ms: Option<u64>,
95    /// Harness message-record count, when available without loading the session.
96    /// STORED message-record count in the native store (cheap line scan) —
97    /// for branched record-tree dialects this can exceed the active-path
98    /// message count a full load renders; `inspect` labels both (SUP-58).
99    pub message_count: Option<usize>,
100    /// Model recorded in lightweight session metadata.
101    pub model: Option<String>,
102    /// Direct parent session for a harness-native child rollout. Ordinary
103    /// conversation lists exclude these children, while tree/fidelity callers
104    /// can request them explicitly without losing the native relationship.
105    #[serde(default, skip_serializing_if = "Option::is_none")]
106    pub parent_session_id: Option<String>,
107    /// Number of proven harness-native descendants represented by this root.
108    /// Child identities remain behind the trusted catalog boundary until a
109    /// caller explicitly requests this session family.
110    #[serde(default, skip_serializing_if = "is_zero")]
111    pub child_session_count: usize,
112    /// ORCH-6: the ORCH-3 conversation nouns (`trigger`, `surface`,
113    /// `profile`, `recurrence`, `cross_surface`, `workspace`), flattened onto
114    /// the row so the wire stays additive. Filled by `finalize_nouns` for
115    /// every harness; only Hermes and OpenClaw publish more than the
116    /// `trigger`/`workspace` defaults today.
117    #[serde(flatten)]
118    pub nouns: OrchestrationNouns,
119}
120
121/// One bounded normalized conversation-message candidate for list projection.
122#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
123pub struct SessionPreviewCandidate {
124    /// Opaque identity of this native message boundary. Consumers may retain
125    /// it to reconcile bounded discovery windows without treating a growing
126    /// preview or a native-store heartbeat as a new conversation message.
127    #[serde(default, skip_serializing_if = "Option::is_none")]
128    pub cursor: Option<String>,
129    /// Canonical conversation role. Older clients may assume `user` when this
130    /// field is absent from an older server.
131    pub role: String,
132    /// Canonical text content.
133    pub content: String,
134    /// Canonical provenance used by the normal conversation visibility rules.
135    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
136    pub metadata: HashMap<String, String>,
137}
138
139/// One stable newest-first discovery page.
140#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
141pub struct DiscoveryPage {
142    /// Sessions in this page.
143    pub sessions: Vec<SessionDescriptor>,
144    /// Opaque cursor for the next page, or `None` at the end.
145    pub next_cursor: Option<String>,
146    /// Coverage receipt (UNI-7 / SUP-54): what was asked and what came back,
147    /// so incomplete coverage can never read as complete.
148    #[serde(default)]
149    pub receipt: DiscoveryReceipt,
150}
151
152/// Coverage evidence for one discovery page.
153#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
154#[serde(default)]
155pub struct DiscoveryReceipt {
156    /// True only when bounded topic/latest message search was applied before
157    /// pagination. Omitted for ordinary metadata-only discovery.
158    #[serde(skip_serializing_if = "is_false")]
159    pub searched_previews: bool,
160    /// The requested lower time bound (epoch ms), when one was given.
161    #[serde(skip_serializing_if = "Option::is_none")]
162    pub requested_after_ms: Option<u64>,
163    /// The requested upper time bound (epoch ms), when one was given.
164    #[serde(skip_serializing_if = "Option::is_none")]
165    pub requested_before_ms: Option<u64>,
166    /// The requested page limit, when one was given.
167    #[serde(skip_serializing_if = "Option::is_none")]
168    pub requested_limit: Option<usize>,
169    /// Oldest `updated_at_ms` among the RETURNED sessions.
170    #[serde(skip_serializing_if = "Option::is_none")]
171    pub oldest_returned_ms: Option<u64>,
172    /// Newest `updated_at_ms` among the RETURNED sessions.
173    #[serde(skip_serializing_if = "Option::is_none")]
174    pub newest_returned_ms: Option<u64>,
175    /// Sessions in this page.
176    pub returned: usize,
177    /// Sessions matching the query across ALL pages.
178    pub total_matched: usize,
179    /// True when matches beyond this page exist (`next_cursor` is the resume
180    /// point). Explicit so a consumer cannot mistake a capped page for the
181    /// complete result set.
182    pub truncated: bool,
183}
184
185/// Append-aware index of the stable topic records in Codex `history.jsonl`.
186///
187/// Long-lived session-list clients can retain this index and refresh it after
188/// filesystem invalidations. Ordinary appends read and parse only the new
189/// bytes; truncation, replacement, and in-place rewrites rebuild the index so
190/// the result remains identical to a fresh catalog discovery.
191#[derive(Debug)]
192pub struct CodexHistoryTopicIndex {
193    path: PathBuf,
194    fingerprint: Option<CodexHistoryFingerprint>,
195    offset: u64,
196    trailing: Vec<u8>,
197    topics: HashMap<String, Vec<SessionPreviewCandidate>>,
198}
199
200#[derive(Debug, Clone, Copy, PartialEq, Eq)]
201struct CodexHistoryFingerprint {
202    len: u64,
203    modified_ns: u128,
204    identity: u128,
205}
206
207impl CodexHistoryTopicIndex {
208    /// Create an empty index for the Codex sessions root from
209    /// [`HarnessHomes::codex`]. Call [`Self::refresh`] before first use.
210    pub fn new(sessions_root: &Path) -> Self {
211        let root = sessions_root.parent().unwrap_or(sessions_root);
212        Self {
213            path: root.join("history.jsonl"),
214            fingerprint: None,
215            offset: 0,
216            trailing: Vec::new(),
217            topics: HashMap::new(),
218        }
219    }
220
221    /// Return the native history file watched by this index.
222    pub fn path(&self) -> &Path {
223        &self.path
224    }
225
226    /// Refresh from durable state and return session ids whose topic changed.
227    ///
228    /// A missing file is a valid empty history. Read errors leave the previous
229    /// successful index intact so a transient filesystem error cannot erase
230    /// topics from a live session list.
231    pub fn refresh(&mut self) -> Result<BTreeSet<String>> {
232        let metadata = match fs::metadata(&self.path) {
233            Ok(metadata) => metadata,
234            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
235                let changed = self.topics.keys().cloned().collect();
236                self.fingerprint = None;
237                self.offset = 0;
238                self.trailing.clear();
239                self.topics.clear();
240                return Ok(changed);
241            }
242            Err(error) => return Err(error.into()),
243        };
244        let fingerprint = codex_history_fingerprint(&metadata)?;
245        if self.fingerprint == Some(fingerprint) {
246            return Ok(BTreeSet::new());
247        }
248
249        let is_append = self.fingerprint.is_some_and(|previous| {
250            previous.identity == fingerprint.identity
251                && previous.len < fingerprint.len
252                && self.offset <= previous.len
253        });
254        if is_append {
255            let mut file = File::open(&self.path)?;
256            file.seek(SeekFrom::Start(self.offset))?;
257            let mut bytes = Vec::with_capacity(
258                usize::try_from(fingerprint.len.saturating_sub(self.offset)).unwrap_or(0),
259            );
260            file.read_to_end(&mut bytes)?;
261            self.offset = file.stream_position()?;
262            let changed = self.ingest(bytes);
263            self.fingerprint = Some(CodexHistoryFingerprint {
264                len: self.offset,
265                ..fingerprint
266            });
267            return Ok(changed);
268        }
269
270        let previous = std::mem::take(&mut self.topics);
271        let mut file = File::open(&self.path)?;
272        let mut bytes = Vec::with_capacity(usize::try_from(fingerprint.len).unwrap_or(0));
273        file.read_to_end(&mut bytes)?;
274        self.offset = file.stream_position()?;
275        self.trailing.clear();
276        self.ingest(bytes);
277        self.fingerprint = Some(CodexHistoryFingerprint {
278            len: self.offset,
279            ..fingerprint
280        });
281        Ok(changed_topic_ids(&previous, &self.topics))
282    }
283
284    fn ingest(&mut self, bytes: Vec<u8>) -> BTreeSet<String> {
285        let mut input = std::mem::take(&mut self.trailing);
286        input.extend(bytes);
287        let complete_len = input
288            .iter()
289            .rposition(|byte| *byte == b'\n')
290            .map_or(0, |index| index + 1);
291        let mut changed = BTreeSet::new();
292        for line in input[..complete_len].split(|byte| *byte == b'\n') {
293            self.ingest_line(line, &mut changed);
294        }
295        self.trailing.extend_from_slice(&input[complete_len..]);
296        if !self.trailing.is_empty() {
297            let trailing = self.trailing.clone();
298            if self.ingest_line(&trailing, &mut changed) {
299                self.trailing.clear();
300            }
301        }
302        changed
303    }
304
305    fn ingest_line(&mut self, line: &[u8], changed: &mut BTreeSet<String>) -> bool {
306        let Ok(value) = serde_json::from_slice::<Value>(line) else {
307            return false;
308        };
309        let Some(session_id) = value.get("session_id").and_then(Value::as_str) else {
310            return true;
311        };
312        if self.topics.contains_key(session_id) {
313            return true;
314        }
315        let mut candidates = Vec::new();
316        push_message_candidate(&mut candidates, "user", value.get("text"), HashMap::new());
317        if !candidates.is_empty() {
318            self.topics.insert(session_id.to_string(), candidates);
319            changed.insert(session_id.to_string());
320        }
321        true
322    }
323}
324
325/// Configurable session roots for the built-in harnesses.
326#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
327#[serde(default)]
328pub struct HarnessHomes {
329    /// Directory containing Claude Code project session directories.
330    pub claude_code: PathBuf,
331    /// Directory containing Codex rollout sessions.
332    pub codex: PathBuf,
333    /// Directory containing Pi project session directories.
334    pub pi: PathBuf,
335    /// OpenCode data root, or an explicit `opencode*.db` path.
336    pub opencode: PathBuf,
337    /// Grok session root containing percent-encoded workspace directories.
338    pub grok: PathBuf,
339    /// Gemini CLI configuration root containing `projects.json` and `tmp/`.
340    pub gemini: PathBuf,
341    /// Goose `sessions.db`, or a directory containing it.
342    pub goose: PathBuf,
343    /// Supercode's native saved-session directory.
344    pub supercode: PathBuf,
345    /// OpenClaw home root (contains `agents/<agentId>/sessions/*.jsonl`,
346    /// openclaw >= 2026.7 — plain pi-v3 dialect files; see
347    /// `SessionSource::OpenClaw`). READ-ONLY discovery (UNI-16).
348    pub openclaw: PathBuf,
349    /// Hermes `state.db` SQLite store path (see `SessionSource::Hermes`).
350    /// READ-ONLY discovery (UNI-15).
351    pub hermes: PathBuf,
352    /// The orchestrator's home FOLDER (`SUPERCODE_ORCHESTRATOR_HOME`, default
353    /// `~/.supercode/orchestrator`). Unlike every other entry this addresses a
354    /// directory, because the orchestrator's serialization IS its folder
355    /// (`docs/ORCHESTRATOR-IR.md` §6): the root is the `default` profile and
356    /// `profiles/<name>/` are the named ones. READ-ONLY (ORC-7).
357    pub orchestrator: PathBuf,
358}
359
360/// Every profile folder under an orchestrator home, in listing order: the
361/// root (the implicit `default` profile) then each `profiles/<name>/`.
362///
363/// One helper, used by every reader that answers `--harness orchestrator`, so
364/// the folder layout of `docs/ORCHESTRATOR-IR.md` §6 is stated once.
365pub fn orchestrator_profile_dirs(root: &Path) -> Vec<(String, PathBuf)> {
366    if !root.is_dir() {
367        return Vec::new();
368    }
369    let mut dirs = vec![("default".to_string(), root.to_path_buf())];
370    if let Ok(entries) = fs::read_dir(root.join("profiles")) {
371        let mut named: Vec<(String, PathBuf)> = entries
372            .flatten()
373            .filter(|entry| entry.path().is_dir())
374            .filter_map(|entry| {
375                entry
376                    .file_name()
377                    .into_string()
378                    .ok()
379                    .map(|name| (name, entry.path()))
380            })
381            // A dot-prefixed folder is never a profile: `profiles/.trash/`
382            // holds the homes `profiles.delete` moved aside (ORC-13), and the
383            // package's own loader skips it for the same reason.
384            .filter(|(name, _)| !name.starts_with('.'))
385            .collect();
386        named.sort();
387        dirs.extend(named);
388    }
389    dirs
390}
391
392impl Default for HarnessHomes {
393    fn default() -> Self {
394        let home = std::env::var_os("HOME")
395            .map(PathBuf::from)
396            .unwrap_or_else(|| PathBuf::from("."));
397        let claude_root = std::env::var_os("CLAUDE_CONFIG_DIR")
398            .map(PathBuf::from)
399            .unwrap_or_else(|| home.join(".claude"));
400        let codex_root = std::env::var_os("CODEX_HOME")
401            .map(PathBuf::from)
402            .unwrap_or_else(|| home.join(".codex"));
403        let pi = std::env::var_os("PI_CODING_AGENT_SESSION_DIR")
404            .map(PathBuf::from)
405            .unwrap_or_else(|| {
406                std::env::var_os("PI_CODING_AGENT_DIR")
407                    .map(PathBuf::from)
408                    .unwrap_or_else(|| home.join(".pi/agent"))
409                    .join("sessions")
410            });
411        let opencode = std::env::var_os("OPENCODE_DB")
412            .map(PathBuf::from)
413            .unwrap_or_else(|| {
414                std::env::var_os("XDG_DATA_HOME")
415                    .map(PathBuf::from)
416                    .unwrap_or_else(|| home.join(".local/share"))
417                    .join("opencode")
418            });
419        let grok = std::env::var_os("GROK_HOME")
420            .map(PathBuf::from)
421            .unwrap_or_else(|| home.join(".grok"))
422            .join("sessions");
423        let gemini = std::env::var_os("GEMINI_CLI_HOME")
424            .map(PathBuf::from)
425            .unwrap_or_else(|| home.join(".gemini"));
426        // Blind-walk finding 2026-08-31: openclaw itself treats
427        // OPENCLAW_HOME as a HOME replacement (state lives at
428        // `$OPENCLAW_HOME/.openclaw`), and names the state dir directly with
429        // OPENCLAW_STATE_DIR. Mirror those semantics exactly so an inherited
430        // environment means the same thing to us and to any openclaw process
431        // we spawn.
432        let openclaw = std::env::var_os("OPENCLAW_STATE_DIR")
433            .map(PathBuf::from)
434            .or_else(|| {
435                std::env::var_os("OPENCLAW_HOME").map(|root| PathBuf::from(root).join(".openclaw"))
436            })
437            .unwrap_or_else(|| home.join(".openclaw"));
438        let orchestrator = std::env::var_os("SUPERCODE_ORCHESTRATOR_HOME")
439            .map(PathBuf::from)
440            .unwrap_or_else(|| home.join(".supercode/orchestrator"));
441        let hermes = std::env::var_os("HERMES_HOME")
442            .map(PathBuf::from)
443            .unwrap_or_else(|| home.join(".hermes"))
444            .join("state.db");
445        let goose = std::env::var_os("GOOSE_PATH_ROOT")
446            .map(PathBuf::from)
447            .map(|root| root.join("data/sessions/sessions.db"))
448            .unwrap_or_else(|| {
449                #[cfg(target_os = "macos")]
450                {
451                    home.join("Library/Application Support/Block/goose/sessions/sessions.db")
452                }
453                #[cfg(target_os = "windows")]
454                {
455                    std::env::var_os("APPDATA")
456                        .map(PathBuf::from)
457                        .unwrap_or_else(|| home.join("AppData/Roaming"))
458                        .join("Block/goose/sessions/sessions.db")
459                }
460                #[cfg(not(any(target_os = "macos", target_os = "windows")))]
461                {
462                    std::env::var_os("XDG_DATA_HOME")
463                        .map(PathBuf::from)
464                        .unwrap_or_else(|| home.join(".local/share"))
465                        .join("goose/sessions/sessions.db")
466                }
467            });
468        let supercode = std::env::var_os("SUPERCODE_HOME")
469            .map(PathBuf::from)
470            .unwrap_or_else(|| {
471                std::env::var_os("XDG_CONFIG_HOME")
472                    .map(PathBuf::from)
473                    .unwrap_or_else(|| home.join(".config"))
474                    .join("supercode")
475            })
476            .join("sessions");
477        Self {
478            claude_code: claude_root.join("projects"),
479            codex: codex_root.join("sessions"),
480            gemini,
481            goose,
482            supercode,
483            openclaw,
484            hermes,
485            orchestrator,
486            pi,
487            opencode,
488            grok,
489        }
490    }
491}
492
493fn codex_history_fingerprint(metadata: &fs::Metadata) -> Result<CodexHistoryFingerprint> {
494    let modified_ns = metadata
495        .modified()?
496        .duration_since(UNIX_EPOCH)
497        .map_err(|error| Error::Other(format!("history timestamp predates Unix epoch: {error}")))?
498        .as_nanos();
499    #[cfg(unix)]
500    let identity = {
501        use std::os::unix::fs::MetadataExt;
502        (u128::from(metadata.dev()) << 64) | u128::from(metadata.ino())
503    };
504    #[cfg(not(unix))]
505    let identity = 0;
506    Ok(CodexHistoryFingerprint {
507        len: metadata.len(),
508        modified_ns,
509        identity,
510    })
511}
512
513fn changed_topic_ids(
514    before: &HashMap<String, Vec<SessionPreviewCandidate>>,
515    after: &HashMap<String, Vec<SessionPreviewCandidate>>,
516) -> BTreeSet<String> {
517    before
518        .keys()
519        .chain(after.keys())
520        .filter(|session_id| before.get(*session_id) != after.get(*session_id))
521        .cloned()
522        .collect()
523}
524
525/// Filters and roots used for one catalog scan.
526#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
527#[serde(default)]
528pub struct DiscoveryQuery {
529    /// Only return sessions whose recorded working directory is this path.
530    pub workspace: Option<PathBuf>,
531    /// Only return sessions whose recorded working directory belongs to the
532    /// same REPOSITORY FAMILY as this path (UNI-7 / SUP-54 decision): the
533    /// family identity is the realpath of the git common directory, so a main
534    /// checkout and all its worktrees join as one family; the origin URL is
535    /// the tie-breaker that also joins separate clones of the same remote. A
536    /// non-repository path degrades to exact-realpath matching. Combines with
537    /// `workspace` as AND when both are set (exact-cwd stays available).
538    pub workspace_family: Option<PathBuf>,
539    /// Only return sessions updated at or after this epoch-ms instant.
540    pub updated_after_ms: Option<u64>,
541    /// Only return sessions updated at or before this epoch-ms instant.
542    pub updated_before_ms: Option<u64>,
543    /// Harnesses to scan. Empty means all built-ins.
544    pub harnesses: Vec<HarnessId>,
545    /// Storage roots to scan.
546    pub homes: HarnessHomes,
547    /// Case-insensitive search over harness, id, title, workspace, and model.
548    pub query: Option<String>,
549    /// Also match bounded opening/latest message candidates. Explicitly opt-in:
550    /// this scans previews of eligible sessions before pagination, not just the
551    /// returned page. Requires a nonempty query; not a full-transcript search.
552    pub search_previews: bool,
553    /// Opaque cursor returned by a prior [`HarnessCatalog::discover_page`].
554    pub cursor: Option<String>,
555    /// Maximum number of results after newest-first sorting.
556    pub limit: Option<usize>,
557    /// Include oldest-first bounded topic candidates for harnesses whose
558    /// native store does not publish a useful title. Off by default because
559    /// topics are stable and list clients can retain them across refreshes.
560    pub include_topic_candidates: bool,
561    /// Include harness-native child rollouts such as Codex subagents. Off by
562    /// default because they are parts of a parent conversation, not chats the
563    /// user independently started. Translation/tree callers can opt in.
564    pub include_child_sessions: bool,
565    /// Restrict an explicit child-inclusive discovery to one root and every
566    /// descendant linked to it by native lineage. Applied before pagination.
567    pub root_session_id: Option<String>,
568    /// ORCH-6: only return sessions routed through this config home (Hermes
569    /// `profile_name` / gateway key namespace, OpenClaw agent id). Exact
570    /// match; a session with no profile never matches. `harnesses` is the
571    /// harness filter and needs no second spelling.
572    pub profile: Option<String>,
573}
574
575/// Repository-family identity (UNI-7 / SUP-54 decision): the realpath of the
576/// git common directory joins a main checkout with all its worktrees, and the
577/// origin URL is the tie-breaker that also joins separate clones of the same
578/// remote. Resolved from the filesystem alone (`.git` file/dir + config), no
579/// subprocess, so discovery stays deterministic and sandbox-friendly.
580#[derive(Debug, Clone, PartialEq, Eq)]
581struct RepoFamily {
582    /// Realpath of the git common directory, or the realpath of the queried
583    /// path itself when it is not inside a git repository.
584    identity: PathBuf,
585    /// True when `identity` is a git common directory (not a bare-path
586    /// degradation) — only then may origin tie-breaking apply.
587    is_repository: bool,
588    /// `[remote "origin"] url` from the common directory's config, if any.
589    origin_url: Option<String>,
590}
591
592impl RepoFamily {
593    fn of(path: &Path) -> Self {
594        let start = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
595        let mut current = Some(start.as_path());
596        while let Some(dir) = current {
597            let dot_git = dir.join(".git");
598            if dot_git.is_dir() {
599                let identity = std::fs::canonicalize(&dot_git).unwrap_or_else(|_| dot_git.clone());
600                let origin_url = read_origin_url(&identity);
601                return Self {
602                    identity,
603                    is_repository: true,
604                    origin_url,
605                };
606            }
607            if dot_git.is_file() {
608                // Worktree checkout: `.git` is a one-line pointer file
609                // `gitdir: <main>/.git/worktrees/<name>`; the common directory
610                // is everything before `/worktrees/<name>`.
611                if let Ok(text) = std::fs::read_to_string(&dot_git) {
612                    if let Some(gitdir) = text
613                        .lines()
614                        .find_map(|line| line.trim().strip_prefix("gitdir:"))
615                    {
616                        let gitdir = PathBuf::from(gitdir.trim());
617                        let gitdir = if gitdir.is_absolute() {
618                            gitdir
619                        } else {
620                            dir.join(gitdir)
621                        };
622                        let common = gitdir
623                            .parent()
624                            .filter(|parent| parent.ends_with("worktrees"))
625                            .and_then(Path::parent)
626                            .map(Path::to_path_buf)
627                            .unwrap_or(gitdir);
628                        let identity = std::fs::canonicalize(&common).unwrap_or(common);
629                        let origin_url = read_origin_url(&identity);
630                        return Self {
631                            identity,
632                            is_repository: true,
633                            origin_url,
634                        };
635                    }
636                }
637            }
638            current = dir.parent();
639        }
640        Self {
641            identity: start,
642            is_repository: false,
643            origin_url: None,
644        }
645    }
646
647    /// Two paths join one family when their common directories match, or —
648    /// for genuine repositories only — when both declare the same origin URL
649    /// (the clone tie-breaker). Bare-path degradations never origin-match.
650    fn joins(&self, other: &Self) -> bool {
651        if self.identity == other.identity {
652            return true;
653        }
654        self.is_repository
655            && other.is_repository
656            && matches!((&self.origin_url, &other.origin_url), (Some(a), Some(b)) if a == b)
657    }
658}
659
660/// Minimal git-config scan for `[remote "origin"] url = ...`.
661fn read_origin_url(common_dir: &Path) -> Option<String> {
662    let text = std::fs::read_to_string(common_dir.join("config")).ok()?;
663    let mut in_origin = false;
664    for line in text.lines() {
665        let line = line.trim();
666        if line.starts_with('[') {
667            in_origin = line.starts_with("[remote \"origin\"]");
668            continue;
669        }
670        if in_origin {
671            if let Some(value) = line.strip_prefix("url") {
672                let value = value.trim_start();
673                if let Some(url) = value.strip_prefix('=') {
674                    let url = url.trim();
675                    if !url.is_empty() {
676                        return Some(url.to_string());
677                    }
678                }
679            }
680        }
681    }
682    None
683}
684
685/// Read-only entry point for discovering, loading, and following persisted
686/// harness sessions.
687#[derive(Debug, Default, Clone, Copy)]
688pub struct HarnessCatalog;
689
690impl HarnessCatalog {
691    /// Construct a catalog. It holds no cache or global mutable state.
692    pub fn new() -> Self {
693        Self
694    }
695
696    /// Discover sessions using lightweight headers/indexes rather than full
697    /// transcript normalization. Malformed or concurrently-created entries
698    /// are skipped without aborting the rest of the scan.
699    pub fn discover(&self, query: &DiscoveryQuery) -> Result<Vec<SessionDescriptor>> {
700        Ok(self.discover_page(query)?.sessions)
701    }
702
703    /// Scan file-backed session metadata without pagination or conversation
704    /// previews. Child sessions are retained so a long-lived caller can keep
705    /// an exact in-memory lineage index and project it without rescanning the
706    /// native stores.
707    pub fn discover_raw_index(&self, query: &DiscoveryQuery) -> Vec<SessionDescriptor> {
708        self.scan_descriptors(query, true)
709    }
710
711    /// Project a raw metadata index through the query's lineage, search, and
712    /// pagination rules without reading any transcript content.
713    pub fn project_index(
714        &self,
715        query: &DiscoveryQuery,
716        descriptors: impl IntoIterator<Item = SessionDescriptor>,
717    ) -> Result<Vec<SessionDescriptor>> {
718        Ok(self.project_index_page(query, descriptors)?.sessions)
719    }
720
721    /// Project a retained metadata index without reading transcripts, preserving
722    /// the full-match count and successor cursor of ordinary discovery.
723    pub fn project_index_page(
724        &self,
725        query: &DiscoveryQuery,
726        descriptors: impl IntoIterator<Item = SessionDescriptor>,
727    ) -> Result<DiscoveryPage> {
728        if query.search_previews {
729            return Err(Error::Other(
730                "preview search requires discover_page, not a metadata-only index projection"
731                    .into(),
732            ));
733        }
734        let mut found = descriptors.into_iter().collect::<Vec<_>>();
735        project_descriptors(query, &mut found);
736        let total_matched = found.len();
737        let (sessions, next_cursor) = paginate_descriptors(query, found)?;
738        let receipt = DiscoveryReceipt {
739            searched_previews: false,
740            requested_after_ms: query.updated_after_ms,
741            requested_before_ms: query.updated_before_ms,
742            requested_limit: query.limit,
743            oldest_returned_ms: sessions.iter().filter_map(|s| s.updated_at_ms).min(),
744            newest_returned_ms: sessions.iter().filter_map(|s| s.updated_at_ms).max(),
745            returned: sessions.len(),
746            total_matched,
747            truncated: next_cursor.is_some(),
748        };
749        Ok(DiscoveryPage {
750            sessions,
751            next_cursor,
752            receipt,
753        })
754    }
755
756    /// Add the bounded topic/latest-message previews used by list clients to
757    /// an already projected metadata page.
758    pub fn enrich_index_page(
759        &self,
760        query: &DiscoveryQuery,
761        mut sessions: Vec<SessionDescriptor>,
762    ) -> Result<Vec<SessionDescriptor>> {
763        enrich_descriptors(query, &mut sessions, None)?;
764        Ok(sessions)
765    }
766
767    /// Add list previews using a retained append-aware Codex history index.
768    ///
769    /// Results are identical to [`Self::enrich_index_page`], while a
770    /// long-lived caller avoids reparsing all of `history.jsonl` for every
771    /// active transcript write.
772    pub fn enrich_index_page_with_codex_history(
773        &self,
774        query: &DiscoveryQuery,
775        mut sessions: Vec<SessionDescriptor>,
776        codex_history: &CodexHistoryTopicIndex,
777    ) -> Result<Vec<SessionDescriptor>> {
778        enrich_descriptors(query, &mut sessions, Some(codex_history))?;
779        Ok(sessions)
780    }
781
782    /// Discover one stable page and return the cursor for its successor.
783    pub fn discover_page(&self, query: &DiscoveryQuery) -> Result<DiscoveryPage> {
784        if query.search_previews {
785            return self.discover_preview_page(query);
786        }
787        let mut page = self.project_index_page(
788            query,
789            self.scan_descriptors(query, query.include_child_sessions),
790        )?;
791        enrich_descriptors(query, &mut page.sessions, None)?;
792        Ok(page)
793    }
794
795    fn discover_preview_page(&self, query: &DiscoveryQuery) -> Result<DiscoveryPage> {
796        let search = query
797            .query
798            .as_deref()
799            .map(str::trim)
800            .filter(|text| !text.is_empty())
801            .ok_or_else(|| Error::Other("search_previews requires a nonempty query".into()))?
802            .to_lowercase();
803        if query.limit == Some(0) {
804            return Err(Error::Other("preview search limit must be positive".into()));
805        }
806        let cursor = query.cursor.as_deref().map(decode_cursor).transpose()?;
807        let mut eligible_query = query.clone();
808        eligible_query.query = None;
809        eligible_query.search_previews = false;
810        eligible_query.include_topic_candidates = true;
811        let mut eligible = self.scan_descriptors(query, query.include_child_sessions);
812        project_descriptors(&eligible_query, &mut eligible);
813        // Read Codex's first history topic once per scan, never once per row.
814        let topics = codex_history_topics(&query.homes.codex, &eligible).unwrap_or_default();
815        let mut sessions = Vec::new();
816        let mut total_matched = 0;
817        let mut cursor_seen = cursor.is_none();
818        let mut more = false;
819        for mut descriptor in eligible {
820            let metadata_match = descriptor_matches(&descriptor, &search);
821            if !metadata_match {
822                let topic = topics
823                    .get(&descriptor.locator.session_id)
824                    .filter(|_| descriptor.locator.harness.as_str() == HarnessId::CODEX);
825                enrich_descriptor(&eligible_query, &mut descriptor, topic);
826                if !descriptor
827                    .preview_candidates
828                    .iter()
829                    .chain(&descriptor.latest_message_candidates)
830                    .any(|candidate| candidate.content.to_lowercase().contains(&search))
831                {
832                    continue;
833                }
834            }
835            total_matched += 1;
836            if !cursor_seen {
837                cursor_seen = cursor.as_ref() == Some(&descriptor_cursor_key(&descriptor));
838                continue;
839            }
840            if sessions.len() < query.limit.unwrap_or(usize::MAX) {
841                if metadata_match {
842                    let topic = topics
843                        .get(&descriptor.locator.session_id)
844                        .filter(|_| descriptor.locator.harness.as_str() == HarnessId::CODEX);
845                    enrich_descriptor(&eligible_query, &mut descriptor, topic);
846                }
847                sessions.push(descriptor);
848            } else {
849                more = true;
850            }
851        }
852        if !cursor_seen {
853            return Err(Error::Other("discovery cursor is stale or invalid".into()));
854        }
855        let next_cursor = more.then(|| sessions.last().map(encode_cursor)).flatten();
856        let receipt = DiscoveryReceipt {
857            searched_previews: true,
858            requested_after_ms: query.updated_after_ms,
859            requested_before_ms: query.updated_before_ms,
860            requested_limit: query.limit,
861            oldest_returned_ms: sessions.iter().filter_map(|s| s.updated_at_ms).min(),
862            newest_returned_ms: sessions.iter().filter_map(|s| s.updated_at_ms).max(),
863            returned: sessions.len(),
864            total_matched,
865            truncated: more,
866        };
867        Ok(DiscoveryPage {
868            sessions,
869            next_cursor,
870            receipt,
871        })
872    }
873
874    fn scan_descriptors(
875        &self,
876        query: &DiscoveryQuery,
877        include_child_sessions: bool,
878    ) -> Vec<SessionDescriptor> {
879        let selected: HashSet<&str> = if query.harnesses.is_empty() {
880            [
881                HarnessId::CLAUDE_CODE,
882                HarnessId::CODEX,
883                HarnessId::PI,
884                HarnessId::OPENCODE,
885                HarnessId::GROK,
886                HarnessId::GEMINI,
887                HarnessId::GOOSE,
888                HarnessId::SUPERCODE,
889                HarnessId::OPENCLAW,
890                HarnessId::HERMES,
891                HarnessId::ORCHESTRATOR,
892            ]
893            .into_iter()
894            .collect()
895        } else {
896            query.harnesses.iter().map(HarnessId::as_str).collect()
897        };
898        let mut found = Vec::new();
899        if selected.contains(HarnessId::CLAUDE_CODE) {
900            discover_jsonl(
901                &query.homes.claude_code,
902                HarnessId::CLAUDE_CODE,
903                query.workspace.as_deref(),
904                include_child_sessions,
905                &mut found,
906            );
907        }
908        if selected.contains(HarnessId::CODEX) {
909            discover_jsonl(
910                &query.homes.codex,
911                HarnessId::CODEX,
912                query.workspace.as_deref(),
913                include_child_sessions,
914                &mut found,
915            );
916        }
917        if selected.contains(HarnessId::PI) {
918            discover_jsonl(
919                &query.homes.pi,
920                HarnessId::PI,
921                query.workspace.as_deref(),
922                include_child_sessions,
923                &mut found,
924            );
925        }
926        if selected.contains(HarnessId::OPENCODE) {
927            discover_opencode(
928                &query.homes.opencode,
929                query.workspace.as_deref(),
930                &mut found,
931            );
932        }
933        if selected.contains(HarnessId::GROK) {
934            discover_grok(&query.homes.grok, query.workspace.as_deref(), &mut found);
935        }
936        if selected.contains(HarnessId::GEMINI) {
937            discover_gemini(&query.homes.gemini, query.workspace.as_deref(), &mut found);
938        }
939        if selected.contains(HarnessId::GOOSE) {
940            discover_goose(&query.homes.goose, query.workspace.as_deref(), &mut found);
941        }
942        if selected.contains(HarnessId::OPENCLAW) {
943            discover_openclaw(
944                &query.homes.openclaw,
945                query.workspace.as_deref(),
946                &mut found,
947            );
948        }
949        if selected.contains(HarnessId::HERMES) {
950            discover_hermes(&query.homes.hermes, query.workspace.as_deref(), &mut found);
951        }
952        if selected.contains(HarnessId::ORCHESTRATOR) {
953            discover_orchestrator(
954                &query.homes.orchestrator,
955                query.workspace.as_deref(),
956                &mut found,
957            );
958        }
959        if selected.contains(HarnessId::SUPERCODE) {
960            discover_supercode(
961                &query.homes.supercode,
962                query.workspace.as_deref(),
963                &mut found,
964            );
965        }
966        for descriptor in &mut found {
967            finalize_nouns(descriptor);
968        }
969        found
970    }
971
972    /// Refresh one file-backed descriptor without rescanning its native store.
973    ///
974    /// This is the incremental counterpart to [`Self::discover_page`]: a
975    /// filesystem notification is only an invalidation hint, so callers
976    /// re-read the durable file and derive the complete current descriptor.
977    /// `None` means the path disappeared or no longer contains a recognizable
978    /// session. SQLite-backed harnesses retain their indexed discovery path.
979    pub fn refresh_file_descriptor(
980        &self,
981        locator: &SessionLocator,
982        workspace: Option<&Path>,
983        include_topic_candidates: bool,
984    ) -> Result<Option<SessionDescriptor>> {
985        let Some(mut descriptor) = self.refresh_file_index_descriptor(locator, workspace)? else {
986            return Ok(None);
987        };
988        if include_topic_candidates {
989            descriptor.preview_candidates =
990                topic_message_candidates(&descriptor.locator).unwrap_or_default();
991        }
992        descriptor.latest_message_candidates =
993            latest_message_candidates(&descriptor.locator).unwrap_or_default();
994        Ok(Some(descriptor))
995    }
996
997    /// Refresh only stable list metadata for one file-backed descriptor.
998    /// This avoids reading conversation preview windows for background index
999    /// maintenance.
1000    pub fn refresh_file_index_descriptor(
1001        &self,
1002        locator: &SessionLocator,
1003        workspace: Option<&Path>,
1004    ) -> Result<Option<SessionDescriptor>> {
1005        let StorageLocator::File { path } = &locator.storage else {
1006            return Ok(None);
1007        };
1008        if !matches!(
1009            locator.harness.as_str(),
1010            HarnessId::CLAUDE_CODE | HarnessId::CODEX
1011        ) {
1012            return Ok(None);
1013        }
1014        if !path.is_file() {
1015            return Ok(None);
1016        }
1017        let Ok(meta) = read_header(path, locator.harness.as_str()) else {
1018            // Harnesses append the header and first turn non-atomically. A
1019            // transiently incomplete new file is not a service error; the
1020            // next native event or reconciliation pass will retry it.
1021            return Ok(None);
1022        };
1023        if workspace.is_some_and(|wanted| {
1024            meta.cwd
1025                .as_deref()
1026                .is_none_or(|cwd| !recorded_cwd_matches(cwd, wanted))
1027        }) {
1028            return Ok(None);
1029        }
1030        let parent_session_id = meta.parent_session_id.or_else(|| {
1031            (locator.harness.as_str() == HarnessId::CLAUDE_CODE)
1032                .then(|| claude_subagent_parent_id(path))
1033                .flatten()
1034        });
1035        let tail = tail_facts(path, locator.harness.as_str());
1036        let descriptor = SessionDescriptor {
1037            locator: SessionLocator {
1038                harness: locator.harness.clone(),
1039                session_id: meta
1040                    .session_id
1041                    .unwrap_or_else(|| locator.session_id.clone()),
1042                storage: StorageLocator::File { path: path.clone() },
1043            },
1044            cwd: meta.cwd,
1045            title: meta.title,
1046            preview_candidates: Vec::new(),
1047            latest_message_candidates: Vec::new(),
1048            updated_at_ms: tail.last_turn_ms.or_else(|| modified_ms(path)),
1049            message_count: None,
1050            model: tail.model.or(meta.model),
1051            parent_session_id,
1052            child_session_count: 0,
1053            nouns: OrchestrationNouns::default(),
1054        };
1055        Ok(Some(descriptor))
1056    }
1057
1058    /// Load the complete normalized session named by a durable locator.
1059    pub fn load(&self, locator: &SessionLocator) -> Result<Session> {
1060        self.load_with_fidelity(locator, Fidelity::ByteLossless)
1061    }
1062
1063    /// [`Self::load`] at a declared fidelity.
1064    ///
1065    /// Read-only surfaces (a session mirror, `follow`) pass
1066    /// [`Fidelity::Semantic`] so a compacted transcript renders instead of
1067    /// erroring; every continuation/transfer/export caller keeps the strict
1068    /// default. See [`Session::load_with_fidelity`].
1069    pub fn load_with_fidelity(
1070        &self,
1071        locator: &SessionLocator,
1072        fidelity: Fidelity,
1073    ) -> Result<Session> {
1074        if let Some(session) = load_hermes_locator(locator) {
1075            return session;
1076        }
1077        match &locator.storage {
1078            StorageLocator::File { path } => {
1079                if let Some(session) = load_native_store_family(path)? {
1080                    Ok(session)
1081                } else {
1082                    Ok(Session::load_with_fidelity(path, fidelity)?)
1083                }
1084            }
1085            StorageLocator::Sqlite { path, selector } => {
1086                if locator.harness.as_str() == HarnessId::GOOSE {
1087                    Ok(Session::from_goose_sqlite(path, selector)?)
1088                } else {
1089                    Ok(Session::from_opencode_sqlite(path, Some(selector))?)
1090                }
1091            }
1092        }
1093    }
1094
1095    /// Load the selected parent transcript without recursively attaching
1096    /// Claude Code child sessions. This is the bounded frontend-view seam;
1097    /// lossless operations continue to use [`Self::load_with_fidelity`].
1098    #[doc(hidden)]
1099    pub fn load_parent_with_fidelity(
1100        &self,
1101        locator: &SessionLocator,
1102        fidelity: Fidelity,
1103    ) -> Result<Session> {
1104        if let Some(session) = load_hermes_locator(locator) {
1105            return session;
1106        }
1107        match &locator.storage {
1108            StorageLocator::File { path } => {
1109                if let Some(session) = load_native_store_family(path)? {
1110                    Ok(session)
1111                } else {
1112                    Ok(Session::load_parent_with_fidelity(path, fidelity)?)
1113                }
1114            }
1115            StorageLocator::Sqlite { path, selector } => {
1116                if locator.harness.as_str() == HarnessId::GOOSE {
1117                    Ok(Session::from_goose_sqlite(path, selector)?)
1118                } else {
1119                    Ok(Session::from_opencode_sqlite(path, Some(selector))?)
1120                }
1121            }
1122        }
1123    }
1124
1125    /// Load bounded parent-only human-visible history. Codex compaction
1126    /// changes resumable context but does not erase earlier visible turns.
1127    #[doc(hidden)]
1128    pub fn load_display_view(
1129        &self,
1130        locator: &SessionLocator,
1131        fidelity: Fidelity,
1132        message_limit: usize,
1133    ) -> Result<Session> {
1134        if let Some(session) = load_hermes_locator(locator) {
1135            let mut session = session?;
1136            if session.messages.len() > message_limit.max(1) {
1137                session
1138                    .messages
1139                    .drain(..session.messages.len() - message_limit.max(1));
1140            }
1141            return Ok(session);
1142        }
1143        match &locator.storage {
1144            StorageLocator::File { path } => {
1145                if let Some(mut session) = load_native_store_family(path)? {
1146                    if session.messages.len() > message_limit.max(1) {
1147                        session
1148                            .messages
1149                            .drain(..session.messages.len() - message_limit.max(1));
1150                    }
1151                    Ok(session)
1152                } else {
1153                    Ok(Session::load_display_view(path, fidelity, message_limit)?)
1154                }
1155            }
1156            StorageLocator::Sqlite { path, selector } => {
1157                let mut session = if locator.harness.as_str() == HarnessId::GOOSE {
1158                    Session::from_goose_sqlite_display(path, selector, message_limit)?
1159                } else {
1160                    Session::from_opencode_sqlite(path, Some(selector))?
1161                };
1162                if session.messages.len() > message_limit.max(1) {
1163                    session
1164                        .messages
1165                        .drain(..session.messages.len() - message_limit.max(1));
1166                }
1167                Ok(session)
1168            }
1169        }
1170    }
1171
1172    /// Open a passive change-triggered follower for a durable locator.
1173    pub fn follow(&self, locator: &SessionLocator) -> Result<SessionFollower> {
1174        self.follow_with_fidelity(locator, Fidelity::ByteLossless)
1175    }
1176
1177    /// [`Self::follow`] at a declared fidelity — see [`Self::load_with_fidelity`].
1178    pub fn follow_with_fidelity(
1179        &self,
1180        locator: &SessionLocator,
1181        fidelity: Fidelity,
1182    ) -> Result<SessionFollower> {
1183        SessionFollower::open_locator_with_fidelity(locator, fidelity)
1184    }
1185
1186    /// Follow a read-only view with explicit child-tree and history bounds.
1187    #[doc(hidden)]
1188    pub fn follow_read_view(
1189        &self,
1190        locator: &SessionLocator,
1191        fidelity: Fidelity,
1192        include_subagents: bool,
1193        message_limit: Option<usize>,
1194        max_message_chars: Option<usize>,
1195        display_history: bool,
1196    ) -> Result<SessionFollower> {
1197        SessionFollower::open_locator_with_view(
1198            locator,
1199            fidelity,
1200            include_subagents,
1201            message_limit,
1202            max_message_chars,
1203            display_history,
1204        )
1205    }
1206}
1207
1208/// ORCH-6: a Hermes row's durable address is the pair `{state.db, session
1209/// id}` — one store holds every session — so the generic file loader (which
1210/// opens the store's most recent session) would answer with the WRONG
1211/// conversation for every discovered locator but one, and its nouns with it.
1212/// Route by the locator's own session id instead.
1213fn load_hermes_locator(locator: &SessionLocator) -> Option<Result<Session>> {
1214    if locator.harness.as_str() != HarnessId::HERMES {
1215        return None;
1216    }
1217    let StorageLocator::File { path } = &locator.storage else {
1218        return None;
1219    };
1220    Some(Session::from_hermes_sqlite(path, Some(&locator.session_id)))
1221}
1222
1223/// ORCH-6: resolve every discovered row's `trigger` and `workspace` through
1224/// ORCH-3's own derivation. Harness scanners fill only what their native store
1225/// states (`surface`, `profile`, `recurrence`, `cross_surface`, and an explicit
1226/// `trigger` when the source says); this pass rebuilds a `SessionMeta` from
1227/// those facts plus `cwd` and re-reads the nouns off it, so a discovered row
1228/// and a loaded session can never disagree about the same session.
1229fn finalize_nouns(descriptor: &mut SessionDescriptor) {
1230    let mut meta = SessionMeta::new(SessionSource::Native);
1231    meta.cwd = descriptor.cwd.clone();
1232    meta.trigger = descriptor.nouns.trigger;
1233    meta.surface = descriptor.nouns.surface.clone();
1234    meta.profile = descriptor.nouns.profile.clone();
1235    meta.recurrence = descriptor.nouns.recurrence.clone();
1236    meta.cross_surface = descriptor.nouns.cross_surface.clone();
1237    descriptor.nouns = OrchestrationNouns::from_meta(&meta);
1238}
1239
1240fn project_descriptors(query: &DiscoveryQuery, found: &mut Vec<SessionDescriptor>) {
1241    roll_up_session_children(found, query.include_child_sessions);
1242    if let Some(root_session_id) = query.root_session_id.as_deref() {
1243        retain_session_family(found, root_session_id);
1244    }
1245    if let Some(family_path) = query.workspace_family.as_deref() {
1246        let family = RepoFamily::of(family_path);
1247        let mut cache: HashMap<PathBuf, bool> = HashMap::new();
1248        found.retain(|descriptor| {
1249            let Some(cwd) = descriptor.cwd.as_deref() else {
1250                return false;
1251            };
1252            *cache
1253                .entry(cwd.to_path_buf())
1254                .or_insert_with(|| RepoFamily::of(cwd).joins(&family))
1255        });
1256    }
1257    if let Some(profile) = query
1258        .profile
1259        .as_deref()
1260        .map(str::trim)
1261        .filter(|p| !p.is_empty())
1262    {
1263        found.retain(|descriptor| descriptor.nouns.profile.as_deref() == Some(profile));
1264    }
1265    if let Some(after) = query.updated_after_ms {
1266        found.retain(|descriptor| descriptor.updated_at_ms.is_some_and(|at| at >= after));
1267    }
1268    if let Some(before) = query.updated_before_ms {
1269        found.retain(|descriptor| descriptor.updated_at_ms.is_some_and(|at| at <= before));
1270    }
1271    found.sort_by(|a, b| {
1272        b.updated_at_ms
1273            .cmp(&a.updated_at_ms)
1274            .then_with(|| a.locator.harness.cmp(&b.locator.harness))
1275            .then_with(|| a.locator.session_id.cmp(&b.locator.session_id))
1276    });
1277    if let Some(search) = query
1278        .query
1279        .as_deref()
1280        .map(str::trim)
1281        .filter(|query| !query.is_empty())
1282    {
1283        let search = search.to_lowercase();
1284        found.retain(|descriptor| descriptor_matches(descriptor, &search));
1285    }
1286}
1287
1288fn paginate_descriptors(
1289    query: &DiscoveryQuery,
1290    found: Vec<SessionDescriptor>,
1291) -> Result<(Vec<SessionDescriptor>, Option<String>)> {
1292    let start = match query.cursor.as_deref() {
1293        Some(cursor) => {
1294            let key = decode_cursor(cursor)?;
1295            found
1296                .iter()
1297                .position(|descriptor| descriptor_cursor_key(descriptor) == key)
1298                .map(|index| index + 1)
1299                .ok_or_else(|| Error::Other("discovery cursor is stale or invalid".into()))?
1300        }
1301        None => 0,
1302    };
1303    let end = query
1304        .limit
1305        .map(|limit| start.saturating_add(limit).min(found.len()))
1306        .unwrap_or(found.len());
1307    let sessions = found[start.min(found.len())..end].to_vec();
1308    let next_cursor = (end < found.len())
1309        .then(|| sessions.last().map(encode_cursor))
1310        .flatten();
1311    Ok((sessions, next_cursor))
1312}
1313
1314fn enrich_descriptors(
1315    query: &DiscoveryQuery,
1316    sessions: &mut [SessionDescriptor],
1317    codex_history: Option<&CodexHistoryTopicIndex>,
1318) -> Result<()> {
1319    let codex_topics = if query.include_topic_candidates && codex_history.is_none() {
1320        codex_history_topics(&query.homes.codex, sessions).unwrap_or_default()
1321    } else {
1322        HashMap::new()
1323    };
1324    for descriptor in sessions {
1325        let topic = (descriptor.locator.harness.as_str() == HarnessId::CODEX)
1326            .then(|| {
1327                codex_history
1328                    .and_then(|history| history.topics.get(&descriptor.locator.session_id))
1329                    .or_else(|| codex_topics.get(&descriptor.locator.session_id))
1330            })
1331            .flatten();
1332        enrich_descriptor(query, descriptor, topic);
1333    }
1334    Ok(())
1335}
1336
1337fn enrich_descriptor(
1338    query: &DiscoveryQuery,
1339    descriptor: &mut SessionDescriptor,
1340    codex_topic: Option<&Vec<SessionPreviewCandidate>>,
1341) {
1342    if query.include_topic_candidates {
1343        descriptor.preview_candidates = codex_topic
1344            .cloned()
1345            .unwrap_or_else(|| topic_message_candidates(&descriptor.locator).unwrap_or_default());
1346    }
1347    descriptor.latest_message_candidates =
1348        latest_message_candidates(&descriptor.locator).unwrap_or_default();
1349}
1350
1351fn is_false(value: &bool) -> bool {
1352    !value
1353}
1354
1355fn descriptor_matches(descriptor: &SessionDescriptor, search: &str) -> bool {
1356    [
1357        Some(descriptor.locator.harness.as_str()),
1358        Some(descriptor.locator.session_id.as_str()),
1359        descriptor.title.as_deref(),
1360        descriptor.cwd.as_ref().and_then(|path| path.to_str()),
1361        descriptor.model.as_deref(),
1362    ]
1363    .into_iter()
1364    .flatten()
1365    .any(|value| value.to_lowercase().contains(search))
1366}
1367
1368fn descriptor_cursor_key(descriptor: &SessionDescriptor) -> (Option<u64>, String, String) {
1369    (
1370        descriptor.updated_at_ms,
1371        descriptor.locator.harness.as_str().to_string(),
1372        descriptor.locator.session_id.clone(),
1373    )
1374}
1375
1376fn encode_cursor(descriptor: &SessionDescriptor) -> String {
1377    let json = serde_json::to_vec(&descriptor_cursor_key(descriptor)).unwrap_or_default();
1378    let mut encoded = String::with_capacity(json.len() * 2);
1379    for byte in json {
1380        use std::fmt::Write;
1381        let _ = write!(&mut encoded, "{byte:02x}");
1382    }
1383    encoded
1384}
1385
1386fn decode_cursor(cursor: &str) -> Result<(Option<u64>, String, String)> {
1387    if cursor.len() % 2 != 0 {
1388        return Err(Error::Other("discovery cursor is invalid".into()));
1389    }
1390    let bytes = (0..cursor.len())
1391        .step_by(2)
1392        .map(|index| u8::from_str_radix(&cursor[index..index + 2], 16))
1393        .collect::<std::result::Result<Vec<_>, _>>()
1394        .map_err(|_| Error::Other("discovery cursor is invalid".into()))?;
1395    serde_json::from_slice(&bytes).map_err(|_| Error::Other("discovery cursor is invalid".into()))
1396}
1397
1398#[derive(Default)]
1399struct HeaderMeta {
1400    session_id: Option<String>,
1401    cwd: Option<PathBuf>,
1402    title: Option<String>,
1403    model: Option<String>,
1404    parent_session_id: Option<String>,
1405}
1406
1407fn discover_jsonl(
1408    root: &Path,
1409    harness: &str,
1410    workspace: Option<&Path>,
1411    include_child_sessions: bool,
1412    found: &mut Vec<SessionDescriptor>,
1413) {
1414    let mut files = Vec::new();
1415    collect_jsonl(root, harness, include_child_sessions, &mut files);
1416    for path in files {
1417        let Ok(meta) = read_header(&path, harness) else {
1418            continue;
1419        };
1420        if workspace.is_some_and(|wanted| {
1421            meta.cwd
1422                .as_deref()
1423                .is_none_or(|cwd| !recorded_cwd_matches(cwd, wanted))
1424        }) {
1425            continue;
1426        }
1427        let session_id = meta.session_id.unwrap_or_else(|| {
1428            path.file_stem()
1429                .and_then(|value| value.to_str())
1430                .unwrap_or("unknown")
1431                .to_string()
1432        });
1433        let parent_session_id = meta.parent_session_id.or_else(|| {
1434            (harness == HarnessId::CLAUDE_CODE)
1435                .then(|| claude_subagent_parent_id(&path))
1436                .flatten()
1437        });
1438        let tail = tail_facts(&path, harness);
1439        found.push(SessionDescriptor {
1440            locator: SessionLocator {
1441                harness: HarnessId::new(harness),
1442                session_id,
1443                storage: StorageLocator::File { path: path.clone() },
1444            },
1445            cwd: meta.cwd,
1446            title: meta.title,
1447            preview_candidates: Vec::new(),
1448            latest_message_candidates: Vec::new(),
1449            updated_at_ms: tail.last_turn_ms.or_else(|| modified_ms(&path)),
1450            message_count: None,
1451            model: tail.model.or(meta.model),
1452            parent_session_id,
1453            child_session_count: if harness == HarnessId::CLAUDE_CODE && !include_child_sessions {
1454                count_claude_subagents(&path)
1455            } else {
1456                0
1457            },
1458            nouns: OrchestrationNouns::default(),
1459        });
1460    }
1461}
1462
1463fn collect_jsonl(root: &Path, harness: &str, include_child_sessions: bool, out: &mut Vec<PathBuf>) {
1464    let Ok(entries) = fs::read_dir(root) else {
1465        return;
1466    };
1467    for entry in entries.flatten() {
1468        let Ok(kind) = entry.file_type() else {
1469            continue;
1470        };
1471        let path = entry.path();
1472        if kind.is_dir() {
1473            if harness == HarnessId::CLAUDE_CODE
1474                && path.file_name().and_then(|v| v.to_str()) == Some("subagents")
1475                && !include_child_sessions
1476            {
1477                continue;
1478            }
1479            collect_jsonl(&path, harness, include_child_sessions, out);
1480        } else if kind.is_file() && path.extension().and_then(|v| v.to_str()) == Some("jsonl") {
1481            out.push(path);
1482        }
1483    }
1484}
1485
1486fn read_header(path: &Path, harness: &str) -> Result<HeaderMeta> {
1487    let file = File::open(path)?;
1488    let mut result = HeaderMeta::default();
1489    let mut bytes = 0usize;
1490    for line in BufReader::new(file).lines().take(32) {
1491        let line = line?;
1492        bytes += line.len();
1493        if bytes > 256 * 1024 {
1494            break;
1495        }
1496        let Ok(value) = serde_json::from_str::<Value>(&line) else {
1497            continue;
1498        };
1499        update_header_meta(&mut result, &value, harness);
1500        if result.session_id.is_some() && result.cwd.is_some() && result.model.is_some() {
1501            break;
1502        }
1503    }
1504    if result.session_id.is_none() && result.cwd.is_none() {
1505        return Err(Error::Other(format!(
1506            "{} has no recognizable {harness} session header",
1507            path.display()
1508        )));
1509    }
1510    Ok(result)
1511}
1512
1513fn update_header_meta(result: &mut HeaderMeta, value: &Value, harness: &str) {
1514    match harness {
1515        HarnessId::CLAUDE_CODE => {
1516            fill_string(&mut result.session_id, value.get("sessionId"));
1517            fill_path(&mut result.cwd, value.get("cwd"));
1518            fill_string(
1519                &mut result.model,
1520                value.get("message").and_then(|v| v.get("model")),
1521            );
1522        }
1523        HarnessId::CODEX => {
1524            let payload = value.get("payload").unwrap_or(&Value::Null);
1525            if value.get("type").and_then(Value::as_str) == Some("session_meta") {
1526                fill_string(&mut result.session_id, payload.get("id"));
1527                fill_path(&mut result.cwd, payload.get("cwd"));
1528                fill_string(&mut result.title, payload.get("thread_name"));
1529                fill_string(&mut result.title, payload.get("title"));
1530                fill_string(
1531                    &mut result.parent_session_id,
1532                    payload.get("parent_thread_id"),
1533                );
1534                if let Some(parent) = payload
1535                    .pointer("/source/subagent/thread_spawn/parent_thread_id")
1536                    .and_then(Value::as_str)
1537                {
1538                    result.parent_session_id = Some(parent.to_string());
1539                }
1540                if result.title.is_none() {
1541                    result.title = payload
1542                        .pointer("/source/subagent/thread_spawn/agent_path")
1543                        .and_then(Value::as_str)
1544                        .and_then(|path| path.rsplit('/').find(|part| !part.is_empty()))
1545                        .map(humanize_topic);
1546                }
1547            }
1548            if value.get("type").and_then(Value::as_str) == Some("turn_context") {
1549                fill_path(&mut result.cwd, payload.get("cwd"));
1550                fill_string(&mut result.model, payload.get("model"));
1551            }
1552        }
1553        HarnessId::PI => {
1554            if value.get("type").and_then(Value::as_str) == Some("session") {
1555                fill_string(&mut result.session_id, value.get("id"));
1556                fill_path(&mut result.cwd, value.get("cwd"));
1557            }
1558            fill_string(
1559                &mut result.model,
1560                value.get("message").and_then(|v| v.get("model")),
1561            );
1562        }
1563        _ => {}
1564    }
1565}
1566
1567/// Collapse native child rollouts into their root conversation before sorting
1568/// and pagination. A child's write time contributes to the root so active
1569/// delegated work keeps the conversation visible without creating extra rows.
1570fn roll_up_session_children(found: &mut Vec<SessionDescriptor>, include_children: bool) {
1571    let by_id = found
1572        .iter()
1573        .enumerate()
1574        .map(|(index, descriptor)| {
1575            (
1576                (
1577                    descriptor.locator.harness.as_str().to_string(),
1578                    descriptor.locator.session_id.clone(),
1579                ),
1580                index,
1581            )
1582        })
1583        .collect::<HashMap<_, _>>();
1584    let mut root_updates = HashMap::<usize, u64>::new();
1585    let mut root_child_counts = HashMap::<usize, usize>::new();
1586
1587    for descriptor in found.iter() {
1588        let Some(mut parent_id) = descriptor.parent_session_id.as_deref() else {
1589            continue;
1590        };
1591        let harness = descriptor.locator.harness.as_str();
1592        let mut root = None;
1593        let mut visited = HashSet::new();
1594        while visited.insert(parent_id.to_string()) {
1595            let Some(&parent_index) = by_id.get(&(harness.to_string(), parent_id.to_string()))
1596            else {
1597                break;
1598            };
1599            root = Some(parent_index);
1600            let Some(next_parent) = found[parent_index].parent_session_id.as_deref() else {
1601                break;
1602            };
1603            parent_id = next_parent;
1604        }
1605        if let (Some(root), Some(updated_at_ms)) = (root, descriptor.updated_at_ms) {
1606            root_updates
1607                .entry(root)
1608                .and_modify(|current| *current = (*current).max(updated_at_ms))
1609                .or_insert(updated_at_ms);
1610        }
1611        if let Some(root) = root {
1612            *root_child_counts.entry(root).or_default() += 1;
1613        }
1614    }
1615
1616    for (root, child_updated_at_ms) in root_updates {
1617        found[root].updated_at_ms = Some(
1618            found[root]
1619                .updated_at_ms
1620                .unwrap_or_default()
1621                .max(child_updated_at_ms),
1622        );
1623    }
1624    for (root, child_count) in root_child_counts {
1625        found[root].child_session_count = child_count;
1626    }
1627    if !include_children {
1628        found.retain(|descriptor| descriptor.parent_session_id.is_none());
1629    }
1630}
1631
1632fn retain_session_family(found: &mut Vec<SessionDescriptor>, root_session_id: &str) {
1633    let parent_by_id = found
1634        .iter()
1635        .map(|descriptor| {
1636            (
1637                descriptor.locator.session_id.clone(),
1638                descriptor.parent_session_id.clone(),
1639            )
1640        })
1641        .collect::<HashMap<_, _>>();
1642    found.retain(|descriptor| {
1643        let mut current = descriptor.locator.session_id.clone();
1644        let mut visited = HashSet::new();
1645        while visited.insert(current.clone()) {
1646            if current == root_session_id {
1647                return true;
1648            }
1649            let Some(Some(parent)) = parent_by_id.get(&current) else {
1650                return false;
1651            };
1652            current = parent.clone();
1653        }
1654        false
1655    });
1656}
1657
1658fn claude_subagent_parent_id(path: &Path) -> Option<String> {
1659    let subagents = path.parent()?;
1660    if subagents.file_name()?.to_str()? != "subagents" {
1661        return None;
1662    }
1663    subagents
1664        .parent()?
1665        .file_name()?
1666        .to_str()
1667        .map(str::to_string)
1668}
1669
1670fn count_claude_subagents(parent_path: &Path) -> usize {
1671    let Some(parent) = parent_path.parent() else {
1672        return 0;
1673    };
1674    let Some(stem) = parent_path.file_stem() else {
1675        return 0;
1676    };
1677    let root = parent.join(stem).join("subagents");
1678    let mut files = Vec::new();
1679    collect_jsonl(&root, HarnessId::CLAUDE_CODE, true, &mut files);
1680    files.len()
1681}
1682
1683fn is_zero(value: &usize) -> bool {
1684    *value == 0
1685}
1686
1687fn humanize_topic(value: &str) -> String {
1688    let text = value.replace(['_', '-'], " ");
1689    let mut characters = text.chars();
1690    match characters.next() {
1691        Some(first) => first.to_uppercase().collect::<String>() + characters.as_str(),
1692        None => text,
1693    }
1694}
1695
1696fn discover_gemini(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
1697    let slug_to_cwd = std::fs::read_to_string(root.join("projects.json"))
1698        .ok()
1699        .and_then(|text| serde_json::from_str::<Value>(&text).ok())
1700        .and_then(|value| value.get("projects").and_then(Value::as_object).cloned())
1701        .map(|projects| {
1702            projects
1703                .into_iter()
1704                .filter_map(|(cwd, slug)| Some((slug.as_str()?.to_string(), PathBuf::from(cwd))))
1705                .collect::<HashMap<_, _>>()
1706        })
1707        .unwrap_or_default();
1708    let mut files = Vec::new();
1709    collect_jsonl(&root.join("tmp"), HarnessId::GEMINI, false, &mut files);
1710    let worker_count = std::thread::available_parallelism()
1711        .map(usize::from)
1712        .unwrap_or(4)
1713        .clamp(1, 8)
1714        .min(files.len().max(1));
1715    let chunk_size = files.len().max(1).div_ceil(worker_count);
1716    let discovered = std::thread::scope(|scope| {
1717        files
1718            .chunks(chunk_size)
1719            .map(|paths| {
1720                scope.spawn(|| {
1721                    paths
1722                        .iter()
1723                        .filter_map(|path| gemini_descriptor(path, &slug_to_cwd, workspace))
1724                        .collect::<Vec<_>>()
1725                })
1726            })
1727            .collect::<Vec<_>>()
1728            .into_iter()
1729            .flat_map(|worker| {
1730                worker
1731                    .join()
1732                    .expect("Gemini discovery worker must not panic")
1733            })
1734            .collect::<Vec<_>>()
1735    });
1736    found.extend(discovered);
1737}
1738
1739fn gemini_descriptor(
1740    path: &Path,
1741    slug_to_cwd: &HashMap<String, PathBuf>,
1742    workspace: Option<&Path>,
1743) -> Option<SessionDescriptor> {
1744    if path
1745        .parent()
1746        .and_then(Path::file_name)
1747        .and_then(|name| name.to_str())
1748        != Some("chats")
1749    {
1750        return None;
1751    }
1752    let slug = path
1753        .parent()
1754        .and_then(Path::parent)
1755        .and_then(Path::file_name)
1756        .and_then(|name| name.to_str());
1757    let cwd = slug.and_then(|slug| slug_to_cwd.get(slug)).cloned();
1758    if workspace.is_some_and(|wanted| {
1759        cwd.as_deref()
1760            .is_none_or(|actual| !recorded_cwd_matches(actual, wanted))
1761    }) {
1762        return None;
1763    }
1764
1765    // The native session id lives on line one. A small decoration budget keeps
1766    // the common title/model case without turning 1,800 sessions into a
1767    // sequential 60 MiB read before the list can render.
1768    let file = File::open(path).ok()?;
1769    let mut reader = BufReader::new(file.take(64 * 1024));
1770    let mut header = String::new();
1771    reader.read_line(&mut header).ok()?;
1772    let header = serde_json::from_str::<Value>(&header).ok()?;
1773    let session_id = header.get("sessionId")?.as_str()?.to_string();
1774    let mut model = None;
1775    for line in reader
1776        .take(4 * 1024)
1777        .lines()
1778        .map_while(std::result::Result::ok)
1779    {
1780        let Ok(value) = serde_json::from_str::<Value>(&line) else {
1781            continue;
1782        };
1783        let kind = value.get("type").and_then(Value::as_str);
1784        if kind != Some("user") && kind != Some("gemini") {
1785            continue;
1786        }
1787        if model.is_none() {
1788            model = value
1789                .get("model")
1790                .and_then(Value::as_str)
1791                .map(str::to_string);
1792        }
1793        if model.is_some() {
1794            break;
1795        }
1796    }
1797    Some(SessionDescriptor {
1798        locator: SessionLocator {
1799            harness: HarnessId::from(HarnessId::GEMINI),
1800            session_id,
1801            storage: StorageLocator::File {
1802                path: path.to_path_buf(),
1803            },
1804        },
1805        cwd,
1806        title: None,
1807        preview_candidates: Vec::new(),
1808        latest_message_candidates: Vec::new(),
1809        updated_at_ms: tail_facts(path, HarnessId::GEMINI)
1810            .last_turn_ms
1811            .or_else(|| modified_ms(path)),
1812        message_count: None,
1813        model,
1814        parent_session_id: None,
1815        child_session_count: 0,
1816        nouns: OrchestrationNouns::default(),
1817    })
1818}
1819
1820fn display_text(content: Option<&Value>) -> Option<String> {
1821    match content? {
1822        Value::String(text) => Some(text.clone()),
1823        Value::Array(parts) => Some(
1824            parts
1825                .iter()
1826                .filter_map(|part| part.get("text").and_then(Value::as_str))
1827                .collect::<Vec<_>>()
1828                .join(" ")
1829                .trim()
1830                .to_string(),
1831        ),
1832        _ => None,
1833    }
1834}
1835
1836/// Hermes (UNI-15): enumerate sessions from the single `state.db` SQLite
1837/// store, strictly read-only (the store is a live, shared, WAL,
1838/// single-writer database owned by a running Hermes install). A missing or
1839/// non-Hermes file is skipped silently, like every other absent home.
1840fn discover_hermes(db_path: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
1841    if !db_path.is_file() {
1842        return;
1843    }
1844    let Ok(conn) = Connection::open_with_flags(
1845        db_path,
1846        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
1847    ) else {
1848        return;
1849    };
1850    let fingerprint_ok = ["sessions", "messages", "schema_version"].iter().all(|t| {
1851        conn.query_row(
1852            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1",
1853            [t],
1854            |_| Ok(()),
1855        )
1856        .is_ok()
1857    });
1858    if !fingerprint_ok {
1859        return;
1860    }
1861    let Ok(mut statement) = conn.prepare(
1862        "SELECT id, cwd, title, model, message_count, started_at, ended_at, parent_session_id, \
1863         source, model_config FROM sessions ORDER BY started_at DESC",
1864    ) else {
1865        return;
1866    };
1867    let Ok(rows) = statement.query_map([], |row| {
1868        Ok((
1869            row.get::<_, String>(0)?,
1870            row.get::<_, Option<String>>(1)?,
1871            row.get::<_, Option<String>>(2)?,
1872            row.get::<_, Option<String>>(3)?,
1873            row.get::<_, Option<i64>>(4)?,
1874            row.get::<_, Option<f64>>(5)?,
1875            row.get::<_, Option<f64>>(6)?,
1876            row.get::<_, Option<String>>(7)?,
1877            row.get::<_, Option<String>>(8)?,
1878            row.get::<_, Option<String>>(9)?,
1879        ))
1880    }) else {
1881        return;
1882    };
1883    for row in rows.flatten() {
1884        let (
1885            id,
1886            cwd,
1887            title,
1888            model,
1889            message_count,
1890            started_at,
1891            ended_at,
1892            parent,
1893            source,
1894            model_config,
1895        ) = row;
1896        let cwd = cwd.map(PathBuf::from);
1897        if let Some(filter) = workspace {
1898            if cwd.as_deref() != Some(filter) {
1899                continue;
1900            }
1901        }
1902        let updated_at_ms = ended_at
1903            .or(started_at)
1904            .map(|seconds| (seconds * 1000.0) as u64);
1905        // ORCH-6: the same derivation `Session::from_hermes_sqlite` runs, over
1906        // the same row — discovery reads the gateway columns it already has
1907        // open instead of guessing a lighter-weight variant.
1908        let mut meta = SessionMeta::new(SessionSource::Hermes);
1909        meta.cwd = cwd.clone();
1910        if let Some(hermes_source) = source.filter(|value| !value.is_empty()) {
1911            meta.lineage
1912                .insert("hermes_source".to_string(), hermes_source);
1913        }
1914        if let Some(parent_id) = parent.as_deref() {
1915            meta.lineage.insert(
1916                "hermes_lineage_kind".to_string(),
1917                crate::session::hermes_lineage_kind(
1918                    &conn,
1919                    parent_id,
1920                    model_config.as_deref(),
1921                    started_at,
1922                )
1923                .to_string(),
1924            );
1925        }
1926        hermes_capture_nouns(&conn, &id, &mut meta);
1927        found.push(SessionDescriptor {
1928            locator: SessionLocator {
1929                harness: HarnessId::new(HarnessId::HERMES),
1930                session_id: id,
1931                storage: StorageLocator::File {
1932                    path: db_path.to_path_buf(),
1933                },
1934            },
1935            cwd,
1936            title: title.filter(|t| !t.is_empty()),
1937            preview_candidates: Vec::new(),
1938            latest_message_candidates: Vec::new(),
1939            updated_at_ms,
1940            message_count: message_count.map(|count| count.max(0) as usize),
1941            model,
1942            parent_session_id: parent,
1943            child_session_count: 0,
1944            nouns: OrchestrationNouns::from_meta(&meta),
1945        });
1946    }
1947}
1948
1949/// The orchestrator (ORC-7): every profile folder's `state.db` `bindings`
1950/// table is one row per conversation the orchestrator holds. A binding is not
1951/// a transcript — the transcript belongs to the WORKER harness it points at —
1952/// so the row carries the surface, trigger, profile and worker identity, and
1953/// its locator addresses the worker's own storage when the binding recorded
1954/// one. Strictly read-only, like every other store here.
1955fn discover_orchestrator(
1956    root: &Path,
1957    workspace: Option<&Path>,
1958    found: &mut Vec<SessionDescriptor>,
1959) {
1960    // A binding has no cwd of its own; a workspace filter can only exclude it.
1961    if workspace.is_some() {
1962        return;
1963    }
1964    for (profile, dir) in orchestrator_profile_dirs(root) {
1965        let db_path = dir.join("state.db");
1966        if !db_path.is_file() {
1967            continue;
1968        }
1969        let Ok(conn) = Connection::open_with_flags(
1970            &db_path,
1971            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
1972        ) else {
1973            continue;
1974        };
1975        let Ok(mut statement) = conn.prepare(
1976            "SELECT platform, chat_type, chat_id, thread_id, participant_id, worker_harness, \
1977             worker_session_id, worker_locator, started_at, last_activity_at, ended_at, \
1978             end_reason, handoff_to, handoff_state, handoff_error, recurrence_job_id, \
1979             CAST(strftime('%s', last_activity_at) AS INTEGER) \
1980             FROM bindings ORDER BY last_activity_at DESC, started_at DESC",
1981        ) else {
1982            continue;
1983        };
1984        let Ok(rows) = statement.query_map([], |row| {
1985            let text = |index: usize| -> rusqlite::Result<Option<String>> {
1986                Ok(row
1987                    .get::<_, Option<String>>(index)?
1988                    .filter(|value| !value.is_empty()))
1989            };
1990            Ok((
1991                OrchestratorBindingRow {
1992                    platform: text(0)?.unwrap_or_default(),
1993                    chat_type: text(1)?.unwrap_or_default(),
1994                    chat_id: text(2)?,
1995                    thread_id: text(3)?,
1996                    participant_id: text(4)?,
1997                    worker_harness: text(5)?.unwrap_or_default(),
1998                    worker_session_id: text(6)?.filter(|s| !s.is_empty()),
1999                    worker_locator: text(7)?,
2000                    started_at: text(8)?,
2001                    last_activity_at: text(9)?,
2002                    ended_at: text(10)?,
2003                    end_reason: text(11)?,
2004                    handoff_to: text(12)?,
2005                    handoff_state: text(13)?,
2006                    handoff_error: text(14)?,
2007                    recurrence_job_id: text(15)?,
2008                },
2009                row.get::<_, Option<i64>>(16)?,
2010            ))
2011        }) else {
2012            continue;
2013        };
2014        for (row, last_activity_epoch) in rows.flatten() {
2015            found.push(orchestrator_descriptor(
2016                &db_path,
2017                &profile,
2018                &row,
2019                last_activity_epoch,
2020            ));
2021        }
2022    }
2023}
2024
2025fn orchestrator_descriptor(
2026    db_path: &Path,
2027    profile: &str,
2028    row: &OrchestratorBindingRow,
2029    last_activity_epoch: Option<i64>,
2030) -> SessionDescriptor {
2031    let binding = Binding::from_orchestrator_row(profile, row);
2032    let nouns = binding.nouns();
2033    // The row's title names the worker the binding points at: that pair is
2034    // the only address from which the conversation itself can be read.
2035    let mut title = format!(
2036        "{} {}",
2037        row.worker_harness,
2038        row.worker_session_id
2039            .as_deref()
2040            .unwrap_or("(no worker session yet)")
2041    );
2042    if let Some(reason) = row.end_reason.as_deref().filter(|_| row.ended_at.is_some()) {
2043        title.push_str(&format!(" (ended: {reason})"));
2044    }
2045    SessionDescriptor {
2046        locator: SessionLocator {
2047            harness: HarnessId::new(HarnessId::ORCHESTRATOR),
2048            session_id: row.worker_session_id.clone().unwrap_or_default(),
2049            storage: StorageLocator::File {
2050                path: row
2051                    .worker_locator
2052                    .clone()
2053                    .map_or_else(|| db_path.to_path_buf(), PathBuf::from),
2054            },
2055        },
2056        cwd: None,
2057        title: Some(title),
2058        preview_candidates: Vec::new(),
2059        latest_message_candidates: Vec::new(),
2060        updated_at_ms: last_activity_epoch.map(|seconds| (seconds.max(0) as u64) * 1000),
2061        message_count: None,
2062        model: None,
2063        parent_session_id: None,
2064        child_session_count: 0,
2065        nouns,
2066    }
2067}
2068
2069/// OpenClaw >= 2026.7: `<home>/agents/<agentId>/sessions/<uuid>.jsonl` are
2070/// plain pi-v3 dialect session files. `.trajectory.jsonl` runtime traces and
2071/// `.trajectory-path.json` pointers live in the SAME directory and are
2072/// excluded by suffix plus a header check (their first line carries
2073/// `traceSchema`, never `type:"session"`). Read-only discovery (UNI-16).
2074fn discover_openclaw(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
2075    let agents = root.join("agents");
2076    let Ok(agent_dirs) = std::fs::read_dir(&agents) else {
2077        return;
2078    };
2079    for agent_dir in agent_dirs.flatten() {
2080        let sessions = agent_dir.path().join("sessions");
2081        let Ok(files) = std::fs::read_dir(&sessions) else {
2082            continue;
2083        };
2084        for file in files.flatten() {
2085            let path = file.path();
2086            let name = file.file_name();
2087            let name = name.to_string_lossy();
2088            if !name.ends_with(".jsonl") || name.ends_with(".trajectory.jsonl") {
2089                continue;
2090            }
2091            let Ok(text) = std::fs::read_to_string(&path) else {
2092                continue;
2093            };
2094            let Some(header_line) = text.lines().find(|line| !line.trim().is_empty()) else {
2095                continue;
2096            };
2097            let Ok(header) = serde_json::from_str::<serde_json::Value>(header_line) else {
2098                continue;
2099            };
2100            if header.get("type").and_then(serde_json::Value::as_str) != Some("session") {
2101                continue;
2102            }
2103            let session_id = header
2104                .get("id")
2105                .and_then(serde_json::Value::as_str)
2106                .unwrap_or_else(|| name.trim_end_matches(".jsonl"))
2107                .to_string();
2108            let cwd = header
2109                .get("cwd")
2110                .and_then(serde_json::Value::as_str)
2111                .map(PathBuf::from);
2112            if let Some(filter) = workspace {
2113                if cwd.as_deref() != Some(filter) {
2114                    continue;
2115                }
2116            }
2117            let updated_at_ms = tail_facts(&path, HarnessId::OPENCLAW)
2118                .last_turn_ms
2119                .or_else(|| {
2120                    file.metadata()
2121                        .ok()
2122                        .and_then(|metadata| metadata.modified().ok())
2123                        .and_then(|modified| modified.duration_since(std::time::UNIX_EPOCH).ok())
2124                        .map(|elapsed| elapsed.as_millis() as u64)
2125                });
2126            let message_count = text
2127                .lines()
2128                .filter(|line| line.contains("\"type\":\"message\""))
2129                .count();
2130            // ORCH-6: the same two facts `Session::load` reads for an OpenClaw
2131            // file — the gateway `sessionKey` in the header, and the agent id
2132            // in `agents/<id>/`.
2133            let mut meta = SessionMeta::new(SessionSource::OpenClaw);
2134            meta.cwd = cwd.clone();
2135            openclaw_capture_header_nouns(&header, &mut meta);
2136            if meta.profile.is_none() {
2137                meta.profile = openclaw_agent_id_from_path(&path);
2138            }
2139            found.push(SessionDescriptor {
2140                locator: SessionLocator {
2141                    harness: HarnessId::new(HarnessId::OPENCLAW),
2142                    session_id,
2143                    storage: StorageLocator::File { path },
2144                },
2145                cwd,
2146                title: None,
2147                preview_candidates: Vec::new(),
2148                latest_message_candidates: Vec::new(),
2149                updated_at_ms,
2150                message_count: Some(message_count),
2151                model: None,
2152                parent_session_id: None,
2153                child_session_count: 0,
2154                nouns: OrchestrationNouns::from_meta(&meta),
2155            });
2156        }
2157    }
2158}
2159
2160fn discover_supercode(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
2161    for info in list_native_store(root) {
2162        let path = if info.archived {
2163            root.join("archived").join(format!("{}.jsonl", info.name))
2164        } else {
2165            root.join(format!("{}.jsonl", info.name))
2166        };
2167        // The sidecar is the store's authoritative content whenever it exists
2168        // (see `read_native_store_header`), and a reduced session can outlive its
2169        // working transcript entirely. Address the file that IS there: a locator
2170        // naming a deleted `<name>.jsonl` is one discovery's own loader rejects.
2171        let sidecar = path.with_extension("sidecar.jsonl");
2172        let path = if path.is_file() {
2173            path
2174        } else {
2175            sidecar.clone()
2176        };
2177        let header = read_native_store_header(&path);
2178        if workspace.is_some_and(|wanted| {
2179            header
2180                .as_ref()
2181                .and_then(|meta| meta.cwd.as_deref())
2182                .is_none_or(|cwd| !recorded_cwd_matches(cwd, wanted))
2183        }) {
2184            continue;
2185        }
2186        let title = (!info.title.trim().is_empty()).then_some(info.title);
2187        // The native store's own records carry no clock; the sidecar beside it
2188        // stamps every turn. Prefer that, and keep mtime as the last resort.
2189        let updated_at_ms = tail_facts(&sidecar, HarnessId::SUPERCODE)
2190            .last_turn_ms
2191            .or_else(|| tail_facts(&path, HarnessId::SUPERCODE).last_turn_ms)
2192            .or_else(|| modified_ms(&path))
2193            .or_else(|| modified_ms(&sidecar));
2194        found.push(SessionDescriptor {
2195            locator: SessionLocator {
2196                harness: HarnessId::from(HarnessId::SUPERCODE),
2197                session_id: info.name,
2198                storage: StorageLocator::File { path: path.clone() },
2199            },
2200            cwd: header.as_ref().and_then(|meta| meta.cwd.clone()),
2201            title,
2202            preview_candidates: Vec::new(),
2203            latest_message_candidates: Vec::new(),
2204            updated_at_ms,
2205            message_count: None,
2206            model: header.and_then(|meta| meta.model),
2207            parent_session_id: None,
2208            child_session_count: 0,
2209            nouns: OrchestrationNouns::default(),
2210        });
2211    }
2212}
2213
2214/// Read only the bounded native envelope needed by discovery. Loading a
2215/// sidecar-backed session here used to deserialize the complete byte-lossless
2216/// transcript family, making a workspace list proportional to every saved
2217/// Supercode transcript on the machine.
2218fn read_native_store_header(path: &Path) -> Option<HeaderMeta> {
2219    let name = path.file_stem()?.to_str()?;
2220    let sidecar = path.with_file_name(format!("{name}.sidecar.jsonl"));
2221    let source_path = if sidecar.is_file() {
2222        sidecar
2223    } else {
2224        path.to_path_buf()
2225    };
2226    let file = File::open(source_path).ok()?;
2227    let mut result = HeaderMeta::default();
2228    let mut source = None;
2229    let mut bytes = 0usize;
2230    for line in BufReader::new(file).lines().take(32) {
2231        let line = line.ok()?;
2232        bytes += line.len();
2233        if bytes > 256 * 1024 {
2234            break;
2235        }
2236        let Ok(value) = serde_json::from_str::<Value>(&line) else {
2237            continue;
2238        };
2239        if source.is_none() {
2240            source = value.get("source").and_then(Value::as_str).map(|source| {
2241                if source == "claude_code" {
2242                    HarnessId::CLAUDE_CODE.to_string()
2243                } else {
2244                    source.to_string()
2245                }
2246            });
2247            fill_string(&mut result.session_id, value.get("session_id"));
2248        }
2249        if let Some(harness) = source.as_deref() {
2250            update_header_meta(&mut result, &value, harness);
2251        }
2252        if result.cwd.is_some() && result.model.is_some() {
2253            break;
2254        }
2255    }
2256    Some(result)
2257}
2258
2259#[derive(Deserialize)]
2260struct NativeStoreInfo {
2261    name: String,
2262    #[serde(default)]
2263    title: String,
2264    #[serde(skip)]
2265    archived: bool,
2266}
2267
2268fn list_native_store(root: &Path) -> Vec<NativeStoreInfo> {
2269    let mut sessions = Vec::new();
2270    for archived in [false, true] {
2271        let directory = if archived {
2272            root.join("archived")
2273        } else {
2274            root.to_path_buf()
2275        };
2276        let Ok(entries) = fs::read_dir(directory) else {
2277            continue;
2278        };
2279        for entry in entries.flatten() {
2280            let path = entry.path();
2281            if !path.to_string_lossy().ends_with(".meta.json") {
2282                continue;
2283            }
2284            let Ok(text) = fs::read_to_string(path) else {
2285                continue;
2286            };
2287            let Ok(mut info) = serde_json::from_str::<NativeStoreInfo>(&text) else {
2288                continue;
2289            };
2290            info.archived = archived;
2291            sessions.push(info);
2292        }
2293    }
2294    sessions.sort_by(|left, right| left.name.cmp(&right.name));
2295    sessions
2296}
2297
2298fn discover_grok(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
2299    let Ok(workspaces) = fs::read_dir(root) else {
2300        return;
2301    };
2302    for workspace_entry in workspaces.flatten() {
2303        let encoded = workspace_entry.file_name();
2304        let Some(cwd) = encoded
2305            .to_str()
2306            .and_then(percent_decode_path)
2307            .map(PathBuf::from)
2308        else {
2309            continue;
2310        };
2311        if workspace.is_some_and(|wanted| !recorded_cwd_matches(&cwd, wanted)) {
2312            continue;
2313        }
2314        let Ok(sessions) = fs::read_dir(workspace_entry.path()) else {
2315            continue;
2316        };
2317        for session_entry in sessions.flatten() {
2318            let session_dir = session_entry.path();
2319            if !session_dir.is_dir() {
2320                continue;
2321            }
2322            let transcript = session_dir.join("chat_history.jsonl");
2323            if !transcript.is_file() {
2324                continue;
2325            }
2326            let Some(session_id) = session_dir
2327                .file_name()
2328                .and_then(|name| name.to_str())
2329                .map(str::to_string)
2330            else {
2331                continue;
2332            };
2333            let summary = fs::read_to_string(session_dir.join("summary.json"))
2334                .ok()
2335                .and_then(|text| serde_json::from_str::<Value>(&text).ok());
2336            let title = summary
2337                .as_ref()
2338                .and_then(|value| value.get("generated_title"))
2339                .and_then(Value::as_str)
2340                .filter(|title| !title.is_empty())
2341                .map(str::to_string);
2342            let model = summary
2343                .as_ref()
2344                .and_then(|value| value.get("current_model_id"))
2345                .and_then(Value::as_str)
2346                .map(str::to_string);
2347            let message_count = summary
2348                .as_ref()
2349                .and_then(|value| value.get("num_chat_messages"))
2350                .and_then(Value::as_u64)
2351                .and_then(|count| usize::try_from(count).ok());
2352            let updated_at_ms = summary
2353                .as_ref()
2354                .and_then(|value| value.get("updated_at"))
2355                .and_then(Value::as_str)
2356                .and_then(crate::sidecar::rfc3339_to_ms)
2357                .and_then(|millis| u64::try_from(millis).ok())
2358                .or_else(|| modified_ms(&transcript));
2359            found.push(SessionDescriptor {
2360                locator: SessionLocator {
2361                    harness: HarnessId::from(HarnessId::GROK),
2362                    session_id,
2363                    storage: StorageLocator::File { path: transcript },
2364                },
2365                cwd: Some(cwd.clone()),
2366                title,
2367                preview_candidates: Vec::new(),
2368                latest_message_candidates: Vec::new(),
2369                updated_at_ms,
2370                message_count,
2371                model,
2372                parent_session_id: None,
2373                child_session_count: 0,
2374                nouns: OrchestrationNouns::default(),
2375            });
2376        }
2377    }
2378}
2379
2380fn discover_opencode(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
2381    let mut dbs = Vec::new();
2382    if root.is_file() {
2383        dbs.push(root.to_path_buf());
2384    } else if let Ok(entries) = fs::read_dir(root) {
2385        dbs.extend(entries.flatten().map(|entry| entry.path()).filter(|path| {
2386            path.file_name()
2387                .and_then(|v| v.to_str())
2388                .is_some_and(|name| name.starts_with("opencode") && name.ends_with(".db"))
2389        }));
2390    }
2391    dbs.sort();
2392    for db in dbs {
2393        let Ok(conn) = Connection::open_with_flags(
2394            &db,
2395            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
2396        ) else {
2397            continue;
2398        };
2399        let has_model = conn.prepare("SELECT model FROM session LIMIT 0").is_ok();
2400        let model_column = if has_model { "s.model" } else { "NULL" };
2401        let query = format!(
2402            "SELECT s.id, s.directory, s.title, s.time_updated, {model_column}, COUNT(m.id) \
2403             FROM session s LEFT JOIN message m ON m.session_id = s.id \
2404             GROUP BY s.id ORDER BY s.time_updated DESC"
2405        );
2406        let Ok(mut stmt) = conn.prepare(&query) else {
2407            continue;
2408        };
2409        let Ok(rows) = stmt.query_map([], |row| {
2410            Ok((
2411                row.get::<_, String>(0)?,
2412                row.get::<_, String>(1)?,
2413                row.get::<_, String>(2)?,
2414                row.get::<_, i64>(3)?,
2415                row.get::<_, Option<String>>(4)?,
2416                row.get::<_, i64>(5)?,
2417            ))
2418        }) else {
2419            continue;
2420        };
2421        for row in rows.flatten() {
2422            let (id, cwd, title, updated, model, messages) = row;
2423            let cwd = PathBuf::from(cwd);
2424            if workspace.is_some_and(|wanted| !recorded_cwd_matches(&cwd, wanted)) {
2425                continue;
2426            }
2427            found.push(SessionDescriptor {
2428                locator: SessionLocator {
2429                    harness: HarnessId::from(HarnessId::OPENCODE),
2430                    session_id: id.clone(),
2431                    storage: StorageLocator::Sqlite {
2432                        path: db.clone(),
2433                        selector: id,
2434                    },
2435                },
2436                cwd: Some(cwd),
2437                title: (!title.is_empty()).then_some(title),
2438                preview_candidates: Vec::new(),
2439                latest_message_candidates: Vec::new(),
2440                updated_at_ms: u64::try_from(updated).ok(),
2441                message_count: usize::try_from(messages).ok(),
2442                model,
2443                parent_session_id: None,
2444                child_session_count: 0,
2445                nouns: OrchestrationNouns::default(),
2446            });
2447        }
2448    }
2449}
2450
2451fn discover_goose(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
2452    let db = if root.is_file() {
2453        root.to_path_buf()
2454    } else if root.join("sessions.db").is_file() {
2455        root.join("sessions.db")
2456    } else {
2457        root.join("sessions/sessions.db")
2458    };
2459    let Ok(connection) = Connection::open_with_flags(
2460        &db,
2461        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
2462    ) else {
2463        return;
2464    };
2465    let Ok(mut statement) = connection.prepare(
2466        "SELECT s.id, s.working_dir, s.name, s.updated_at, s.model_config_json, \
2467                COUNT(m.id) \
2468         FROM sessions s LEFT JOIN messages m ON m.session_id = s.id \
2469         WHERE s.archived_at IS NULL \
2470         GROUP BY s.id ORDER BY s.updated_at DESC",
2471    ) else {
2472        return;
2473    };
2474    let Ok(rows) = statement.query_map([], |row| {
2475        Ok((
2476            row.get::<_, String>(0)?,
2477            row.get::<_, String>(1)?,
2478            row.get::<_, String>(2)?,
2479            row.get::<_, String>(3)?,
2480            row.get::<_, Option<String>>(4)?,
2481            row.get::<_, i64>(5)?,
2482        ))
2483    }) else {
2484        return;
2485    };
2486    for row in rows.flatten() {
2487        let (id, cwd, title, updated_at, model_config, message_count) = row;
2488        let cwd = PathBuf::from(cwd);
2489        if workspace.is_some_and(|wanted| !recorded_cwd_matches(&cwd, wanted)) {
2490            continue;
2491        }
2492        let model = model_config
2493            .as_deref()
2494            .and_then(|value| serde_json::from_str::<Value>(value).ok())
2495            .and_then(|value| {
2496                value
2497                    .get("model_name")
2498                    .or_else(|| value.get("modelName"))
2499                    .and_then(Value::as_str)
2500                    .map(str::to_string)
2501            });
2502        let updated_at_ms = crate::sidecar::rfc3339_to_ms(&updated_at)
2503            .or_else(|| {
2504                // SQLite's CURRENT_TIMESTAMP uses `YYYY-MM-DD HH:MM:SS`.
2505                crate::sidecar::rfc3339_to_ms(&format!("{}Z", updated_at.replace(' ', "T")))
2506            })
2507            .and_then(|value| u64::try_from(value).ok());
2508        found.push(SessionDescriptor {
2509            locator: SessionLocator {
2510                harness: HarnessId::from(HarnessId::GOOSE),
2511                session_id: id.clone(),
2512                storage: StorageLocator::Sqlite {
2513                    path: db.clone(),
2514                    selector: id,
2515                },
2516            },
2517            cwd: Some(cwd),
2518            title: (!title.trim().is_empty()).then_some(title),
2519            preview_candidates: Vec::new(),
2520            latest_message_candidates: Vec::new(),
2521            updated_at_ms,
2522            message_count: usize::try_from(message_count).ok(),
2523            model,
2524            parent_session_id: None,
2525            child_session_count: 0,
2526            nouns: OrchestrationNouns::default(),
2527        });
2528    }
2529}
2530
2531const LATEST_PREVIEW_CANDIDATES: usize = 8;
2532const TOPIC_PREVIEW_HEAD_BYTES: u64 = 512 * 1024;
2533const LATEST_PREVIEW_TAIL_BYTES: u64 = 512 * 1024;
2534const LATEST_PREVIEW_MAX_BYTES: u64 = 4 * 1024 * 1024;
2535
2536fn topic_message_candidates(locator: &SessionLocator) -> Result<Vec<SessionPreviewCandidate>> {
2537    match &locator.storage {
2538        StorageLocator::File { path }
2539            if matches!(
2540                locator.harness.as_str(),
2541                HarnessId::CLAUDE_CODE | HarnessId::CODEX
2542            ) =>
2543        {
2544            topic_file_message_candidates(path, locator.harness.as_str())
2545        }
2546        _ => Ok(Vec::new()),
2547    }
2548}
2549
2550fn codex_history_topics(
2551    sessions_root: &Path,
2552    sessions: &[SessionDescriptor],
2553) -> Result<HashMap<String, Vec<SessionPreviewCandidate>>> {
2554    let wanted: HashSet<&str> = sessions
2555        .iter()
2556        .filter(|descriptor| descriptor.locator.harness.as_str() == HarnessId::CODEX)
2557        .map(|descriptor| descriptor.locator.session_id.as_str())
2558        .collect();
2559    if wanted.is_empty() {
2560        return Ok(HashMap::new());
2561    }
2562    let Some(root) = sessions_root.parent() else {
2563        return Ok(HashMap::new());
2564    };
2565    let file = match File::open(root.join("history.jsonl")) {
2566        Ok(file) => file,
2567        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(HashMap::new()),
2568        Err(error) => return Err(error.into()),
2569    };
2570    let mut topics = HashMap::new();
2571    for line in BufReader::new(file).lines() {
2572        let Ok(value) = serde_json::from_str::<Value>(&line?) else {
2573            continue;
2574        };
2575        let Some(session_id) = value.get("session_id").and_then(Value::as_str) else {
2576            continue;
2577        };
2578        if !wanted.contains(session_id) || topics.contains_key(session_id) {
2579            continue;
2580        }
2581        let mut candidates = Vec::new();
2582        push_message_candidate(&mut candidates, "user", value.get("text"), HashMap::new());
2583        if !candidates.is_empty() {
2584            topics.insert(session_id.to_string(), candidates);
2585            if topics.len() == wanted.len() {
2586                break;
2587            }
2588        }
2589    }
2590    Ok(topics)
2591}
2592
2593fn latest_message_candidates(locator: &SessionLocator) -> Result<Vec<SessionPreviewCandidate>> {
2594    match &locator.storage {
2595        // A Hermes locator addresses one session inside the whole-store `state.db`; the file's
2596        // tail is SQLite pages, not transcript lines, so the preview comes from the store by id.
2597        StorageLocator::File { path } | StorageLocator::Sqlite { path, .. }
2598            if locator.harness.as_str() == HarnessId::HERMES =>
2599        {
2600            latest_hermes_message_candidates(path, &locator.session_id)
2601        }
2602        StorageLocator::File { path } => {
2603            latest_file_message_candidates(path, locator.harness.as_str())
2604        }
2605        StorageLocator::Sqlite { path, selector }
2606            if locator.harness.as_str() == HarnessId::OPENCODE =>
2607        {
2608            latest_opencode_message_candidates(path, selector)
2609        }
2610        StorageLocator::Sqlite { path, selector }
2611            if locator.harness.as_str() == HarnessId::GOOSE =>
2612        {
2613            latest_goose_message_candidates(path, selector)
2614        }
2615        StorageLocator::Sqlite { .. } => Ok(Vec::new()),
2616    }
2617}
2618
2619fn topic_file_message_candidates(
2620    path: &Path,
2621    harness: &str,
2622) -> Result<Vec<SessionPreviewCandidate>> {
2623    let mut file = File::open(path)?;
2624    let mut bytes = Vec::with_capacity(TOPIC_PREVIEW_HEAD_BYTES as usize);
2625    file.by_ref()
2626        .take(TOPIC_PREVIEW_HEAD_BYTES)
2627        .read_to_end(&mut bytes)?;
2628    if file.metadata()?.len() > TOPIC_PREVIEW_HEAD_BYTES {
2629        if let Some(newline) = bytes.iter().rposition(|byte| *byte == b'\n') {
2630            bytes.truncate(newline);
2631        }
2632    }
2633    let text = String::from_utf8(bytes).map_err(|_| {
2634        Error::Other(format!(
2635            "{} contains non-UTF-8 data in its topic-preview window",
2636            path.display()
2637        ))
2638    })?;
2639    if harness == HarnessId::CODEX {
2640        return Ok(codex_preview_candidates(text.lines(), false));
2641    }
2642    let mut candidates = Vec::new();
2643    for line in text.lines() {
2644        let Ok(value) = serde_json::from_str::<Value>(line) else {
2645            continue;
2646        };
2647        push_topic_message_candidate(&mut candidates, harness, &value);
2648        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
2649            break;
2650        }
2651    }
2652    Ok(candidates)
2653}
2654
2655fn latest_file_message_candidates(
2656    path: &Path,
2657    harness: &str,
2658) -> Result<Vec<SessionPreviewCandidate>> {
2659    let mut candidates =
2660        latest_file_message_candidates_with_limit(path, harness, LATEST_PREVIEW_TAIL_BYTES)?;
2661    if candidates.is_empty() {
2662        candidates =
2663            latest_file_message_candidates_with_limit(path, harness, LATEST_PREVIEW_MAX_BYTES)?;
2664    }
2665    Ok(candidates)
2666}
2667
2668fn latest_file_message_candidates_with_limit(
2669    path: &Path,
2670    harness: &str,
2671    byte_limit: u64,
2672) -> Result<Vec<SessionPreviewCandidate>> {
2673    let mut file = File::open(path)?;
2674    let file_len = file.metadata()?.len();
2675    let start = file_len.saturating_sub(byte_limit);
2676    file.seek(SeekFrom::Start(start))?;
2677    let mut bytes = Vec::with_capacity((file_len - start) as usize);
2678    file.read_to_end(&mut bytes)?;
2679    if start > 0 {
2680        if let Some(newline) = bytes.iter().position(|byte| *byte == b'\n') {
2681            bytes.drain(..=newline);
2682        } else {
2683            return Ok(Vec::new());
2684        }
2685    }
2686    let text = String::from_utf8(bytes).map_err(|_| {
2687        Error::Other(format!(
2688            "{} contains non-UTF-8 data in its list-preview window",
2689            path.display()
2690        ))
2691    })?;
2692    if harness == HarnessId::CODEX {
2693        return Ok(codex_preview_candidates(text.lines().rev(), true));
2694    }
2695    let mut candidates = Vec::new();
2696    for line in text.lines().rev() {
2697        let Ok(value) = serde_json::from_str::<Value>(line) else {
2698            continue;
2699        };
2700        let (role, content, metadata) = match harness {
2701            HarnessId::CLAUDE_CODE => {
2702                let role = value.get("type").and_then(Value::as_str);
2703                if !matches!(role, Some("user" | "assistant")) {
2704                    continue;
2705                }
2706                let metadata = if role == Some("user") {
2707                    crate::session::claude_user_provenance(&value)
2708                        .into_iter()
2709                        .collect()
2710                } else {
2711                    HashMap::new()
2712                };
2713                (
2714                    role.unwrap_or_default(),
2715                    value
2716                        .get("message")
2717                        .and_then(|message| message.get("content")),
2718                    metadata,
2719                )
2720            }
2721            HarnessId::PI => {
2722                if value.get("type").and_then(Value::as_str) != Some("message") {
2723                    continue;
2724                }
2725                let message = value.get("message").unwrap_or(&Value::Null);
2726                let Some(role @ ("user" | "assistant")) =
2727                    message.get("role").and_then(Value::as_str)
2728                else {
2729                    continue;
2730                };
2731                (role, message.get("content"), HashMap::new())
2732            }
2733            HarnessId::GEMINI => {
2734                let Some(kind @ ("user" | "gemini")) = value.get("type").and_then(Value::as_str)
2735                else {
2736                    continue;
2737                };
2738                (
2739                    if kind == "gemini" {
2740                        "assistant"
2741                    } else {
2742                        "user"
2743                    },
2744                    value.get("content"),
2745                    HashMap::new(),
2746                )
2747            }
2748            HarnessId::GROK => {
2749                let Some(role @ ("user" | "assistant")) = value.get("type").and_then(Value::as_str)
2750                else {
2751                    continue;
2752                };
2753                (role, value.get("content"), HashMap::new())
2754            }
2755            HarnessId::SUPERCODE => {
2756                let Some(role @ ("user" | "assistant")) = value.get("role").and_then(Value::as_str)
2757                else {
2758                    continue;
2759                };
2760                (role, value.get("content"), HashMap::new())
2761            }
2762            _ => continue,
2763        };
2764        let mut metadata = metadata;
2765        if matches!(harness, HarnessId::CLAUDE_CODE | HarnessId::CODEX) {
2766            if let Some(timestamp) = value.get("timestamp").and_then(Value::as_str) {
2767                metadata.insert("timestamp".to_string(), timestamp.to_string());
2768            }
2769        }
2770        push_message_candidate_with_cursor(
2771            &mut candidates,
2772            role,
2773            content,
2774            metadata,
2775            Some(message_candidate_cursor(harness, &value)),
2776        );
2777        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
2778            break;
2779        }
2780    }
2781    Ok(candidates)
2782}
2783
2784struct CodexPreviewRecord {
2785    native: Value,
2786    role: String,
2787    text: String,
2788}
2789
2790// Pair adjacent conversational event/response mirrors one-to-one, within the
2791// bytes already read. This is intentionally narrower than the full codec's
2792// global assistant-text dedup: repeated same-kind records and separate pairs
2793// remain separate turns. Compare full display text BEFORE the 4096-character
2794// cap; prefer the canonical response's cursor/timestamp in either scan direction.
2795fn codex_preview_candidates<'a>(
2796    lines: impl Iterator<Item = &'a str>,
2797    latest: bool,
2798) -> Vec<SessionPreviewCandidate> {
2799    let mut candidates = Vec::new();
2800    let mut pending: Option<CodexPreviewRecord> = None;
2801    for line in lines {
2802        let Ok(native) = serde_json::from_str::<Value>(line) else {
2803            continue;
2804        };
2805        let Some((role, content)) = codex_preview_message(&native) else {
2806            continue;
2807        };
2808        let Some(text) = display_text(content).filter(|text| !text.trim().is_empty()) else {
2809            continue;
2810        };
2811        let current = CodexPreviewRecord {
2812            role: role.to_string(),
2813            text,
2814            native,
2815        };
2816        if let Some(previous) = pending.take() {
2817            if previous.role == current.role
2818                && previous.text == current.text
2819                && previous.native.get("type") != current.native.get("type")
2820            {
2821                let canonical = if previous.native.get("type").and_then(Value::as_str)
2822                    == Some("response_item")
2823                {
2824                    previous
2825                } else {
2826                    current
2827                };
2828                push_codex_preview_candidate(&mut candidates, canonical, latest);
2829            } else {
2830                push_codex_preview_candidate(&mut candidates, previous, latest);
2831                pending = Some(current);
2832            }
2833        } else {
2834            pending = Some(current);
2835        }
2836        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
2837            break;
2838        }
2839    }
2840    if let Some(last) = pending {
2841        push_codex_preview_candidate(&mut candidates, last, latest);
2842    }
2843    candidates
2844}
2845
2846fn push_codex_preview_candidate(
2847    candidates: &mut Vec<SessionPreviewCandidate>,
2848    record: CodexPreviewRecord,
2849    latest: bool,
2850) {
2851    let mut metadata = HashMap::new();
2852    if latest {
2853        if let Some(timestamp) = record.native.get("timestamp").and_then(Value::as_str) {
2854            metadata.insert("timestamp".to_string(), timestamp.to_string());
2855        }
2856    }
2857    let cursor = latest.then(|| message_candidate_cursor(HarnessId::CODEX, &record.native));
2858    push_message_candidate_with_cursor(
2859        candidates,
2860        &record.role,
2861        Some(&Value::String(record.text)),
2862        metadata,
2863        cursor,
2864    );
2865}
2866
2867// Codex collab rollouts can carry narration only as event_msg records.
2868fn codex_preview_message(value: &Value) -> Option<(&str, Option<&Value>)> {
2869    let payload = value.get("payload")?;
2870    match (
2871        value.get("type").and_then(Value::as_str)?,
2872        payload.get("type").and_then(Value::as_str)?,
2873    ) {
2874        ("response_item", "message") => {
2875            let role @ ("user" | "assistant") = payload.get("role").and_then(Value::as_str)? else {
2876                return None;
2877            };
2878            Some((role, payload.get("content")))
2879        }
2880        ("event_msg", "user_message") => Some(("user", payload.get("message"))),
2881        ("event_msg", "agent_message") => Some(("assistant", payload.get("message"))),
2882        _ => None,
2883    }
2884}
2885
2886fn push_topic_message_candidate(
2887    candidates: &mut Vec<SessionPreviewCandidate>,
2888    harness: &str,
2889    value: &Value,
2890) {
2891    let (role, content, metadata) = match harness {
2892        HarnessId::CLAUDE_CODE => {
2893            let role = value.get("type").and_then(Value::as_str);
2894            if !matches!(role, Some("user" | "assistant")) {
2895                return;
2896            }
2897            let metadata = if role == Some("user") {
2898                crate::session::claude_user_provenance(value)
2899                    .into_iter()
2900                    .collect()
2901            } else {
2902                HashMap::new()
2903            };
2904            (
2905                role.unwrap_or_default(),
2906                value
2907                    .get("message")
2908                    .and_then(|message| message.get("content")),
2909                metadata,
2910            )
2911        }
2912        _ => return,
2913    };
2914    push_message_candidate(candidates, role, content, metadata);
2915}
2916
2917fn latest_opencode_message_candidates(
2918    path: &Path,
2919    session_id: &str,
2920) -> Result<Vec<SessionPreviewCandidate>> {
2921    let connection = Connection::open_with_flags(
2922        path,
2923        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
2924    )
2925    .map_err(|error| Error::Other(format!("failed to open list-preview store: {error}")))?;
2926    let mut statement = connection
2927        .prepare(
2928            "SELECT m.data, p.data FROM message m JOIN part p ON p.message_id = m.id \
2929         WHERE m.session_id = ?1 ORDER BY m.time_created DESC, p.time_created DESC LIMIT 32",
2930        )
2931        .map_err(|error| Error::Other(format!("failed to prepare list-preview query: {error}")))?;
2932    let rows = statement
2933        .query_map([session_id], |row| {
2934            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
2935        })
2936        .map_err(|error| Error::Other(format!("failed to read list-preview rows: {error}")))?;
2937    let mut candidates = Vec::new();
2938    for row in rows.flatten() {
2939        let (Ok(message), Ok(part)) = (
2940            serde_json::from_str::<Value>(&row.0),
2941            serde_json::from_str::<Value>(&row.1),
2942        ) else {
2943            continue;
2944        };
2945        let Some(role @ ("user" | "assistant")) = message.get("role").and_then(Value::as_str)
2946        else {
2947            continue;
2948        };
2949        if part.get("type").and_then(Value::as_str) != Some("text") {
2950            continue;
2951        }
2952        push_message_candidate(&mut candidates, role, part.get("text"), HashMap::new());
2953        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
2954            break;
2955        }
2956    }
2957    Ok(candidates)
2958}
2959
2960fn latest_hermes_message_candidates(
2961    path: &Path,
2962    session_id: &str,
2963) -> Result<Vec<SessionPreviewCandidate>> {
2964    let connection = Connection::open_with_flags(
2965        path,
2966        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
2967    )
2968    .map_err(|error| Error::Other(format!("failed to open list-preview store: {error}")))?;
2969    let mut statement = connection
2970        .prepare(
2971            "SELECT role, content FROM messages WHERE session_id = ?1 AND active = 1 \
2972             AND role IN ('user', 'assistant') AND content IS NOT NULL AND content != '' \
2973             ORDER BY timestamp DESC, id DESC LIMIT 32",
2974        )
2975        .map_err(|error| Error::Other(format!("failed to prepare list-preview query: {error}")))?;
2976    let rows = statement
2977        .query_map([session_id], |row| {
2978            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
2979        })
2980        .map_err(|error| Error::Other(format!("failed to read list-preview rows: {error}")))?;
2981    let mut candidates = Vec::new();
2982    for (role, content) in rows.flatten() {
2983        push_message_candidate(
2984            &mut candidates,
2985            &role,
2986            Some(&Value::String(content)),
2987            HashMap::new(),
2988        );
2989        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
2990            break;
2991        }
2992    }
2993    Ok(candidates)
2994}
2995
2996fn latest_goose_message_candidates(
2997    path: &Path,
2998    session_id: &str,
2999) -> Result<Vec<SessionPreviewCandidate>> {
3000    let connection = Connection::open_with_flags(
3001        path,
3002        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
3003    )
3004    .map_err(|error| Error::Other(format!("failed to open list-preview store: {error}")))?;
3005    let mut statement = connection
3006        .prepare(
3007            "SELECT role, content_json FROM messages WHERE session_id = ?1 \
3008         ORDER BY created_timestamp DESC, id DESC LIMIT 16",
3009        )
3010        .map_err(|error| Error::Other(format!("failed to prepare list-preview query: {error}")))?;
3011    let rows = statement
3012        .query_map([session_id], |row| {
3013            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
3014        })
3015        .map_err(|error| Error::Other(format!("failed to read list-preview rows: {error}")))?;
3016    let mut candidates = Vec::new();
3017    for row in rows.flatten() {
3018        let (role, content) = row;
3019        if !matches!(role.as_str(), "user" | "assistant") {
3020            continue;
3021        }
3022        let Ok(content) = serde_json::from_str::<Value>(&content) else {
3023            continue;
3024        };
3025        push_message_candidate(&mut candidates, &role, Some(&content), HashMap::new());
3026        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
3027            break;
3028        }
3029    }
3030    Ok(candidates)
3031}
3032
3033fn push_message_candidate(
3034    candidates: &mut Vec<SessionPreviewCandidate>,
3035    role: &str,
3036    content: Option<&Value>,
3037    metadata: HashMap<String, String>,
3038) {
3039    push_message_candidate_with_cursor(candidates, role, content, metadata, None);
3040}
3041
3042fn push_message_candidate_with_cursor(
3043    candidates: &mut Vec<SessionPreviewCandidate>,
3044    role: &str,
3045    content: Option<&Value>,
3046    metadata: HashMap<String, String>,
3047    cursor: Option<String>,
3048) {
3049    if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
3050        return;
3051    }
3052    let Some(text) = display_text(content).filter(|text| !text.trim().is_empty()) else {
3053        return;
3054    };
3055    const MAX_CHARS: usize = 4_096;
3056    candidates.push(SessionPreviewCandidate {
3057        cursor,
3058        role: role.to_string(),
3059        content: text.chars().take(MAX_CHARS).collect(),
3060        metadata,
3061    });
3062}
3063
3064fn message_candidate_cursor(harness: &str, value: &Value) -> String {
3065    let native_identity = value
3066        .get("uuid")
3067        .or_else(|| value.get("id"))
3068        .or_else(|| value.pointer("/message/id"))
3069        .or_else(|| value.pointer("/payload/id"))
3070        .and_then(Value::as_str)
3071        .or_else(|| value.get("timestamp").and_then(Value::as_str));
3072    let mut hasher = blake3::Hasher::new();
3073    hasher.update(b"supercode.session-preview-cursor.v1\0");
3074    hasher.update(harness.as_bytes());
3075    hasher.update(b"\0");
3076    if let Some(identity) = native_identity {
3077        hasher.update(identity.as_bytes());
3078    } else {
3079        // Some formats do not publish message ids. Hashing the complete native
3080        // record is still stable across discovery refreshes and reveals none
3081        // of the record itself to an untrusted presentation surface.
3082        hasher.update(value.to_string().as_bytes());
3083    }
3084    format!("v1:{}", &hasher.finalize().to_hex()[..24])
3085}
3086
3087fn fill_string(target: &mut Option<String>, value: Option<&Value>) {
3088    if target.is_none() {
3089        *target = value.and_then(Value::as_str).map(str::to_owned);
3090    }
3091}
3092
3093fn fill_path(target: &mut Option<PathBuf>, value: Option<&Value>) {
3094    if target.is_none() {
3095        *target = value.and_then(Value::as_str).map(PathBuf::from);
3096    }
3097}
3098
3099/// What the END of a JSONL transcript says about a session: when it last took
3100/// a turn, and which model took it.
3101#[derive(Default)]
3102struct TailFacts {
3103    /// Newest `timestamp` among the last records, as Unix epoch milliseconds.
3104    last_turn_ms: Option<u64>,
3105    /// Model on the LAST turn that named one.
3106    model: Option<String>,
3107}
3108
3109/// Read [`TailFacts`] from the last records of a JSONL transcript.
3110///
3111/// Both facts have to come from the conversation rather than from cheaper
3112/// stand-ins. Recency is not the file's mtime: harnesses touch a transcript
3113/// without saying anything (see [`SessionDescriptor::updated_at_ms`]). The model
3114/// is not the one in the header either — every dialect here records the model
3115/// per turn, so a mid-session switch (`/model`) leaves the opening record naming
3116/// a model the session has not used for hours.
3117///
3118/// Reading whole files is not an option — one catalog holds thousands of
3119/// sessions and a single transcript runs to tens of megabytes — so this seeks to
3120/// the end and walks backwards over a bounded window, which is where an
3121/// append-only log keeps its newest records. The window grows only while no
3122/// timestamp has been found, and gives up at [`TAIL_SCAN_LIMIT`]; a model the
3123/// window does not reach stays `None` and the caller keeps the header's.
3124fn tail_facts(path: &Path, harness: &str) -> TailFacts {
3125    let mut facts = TailFacts::default();
3126    let Ok(mut file) = File::open(path) else {
3127        return facts;
3128    };
3129    let Ok(len) = file.metadata().map(|meta| meta.len()) else {
3130        return facts;
3131    };
3132    let mut window = TAIL_SCAN_START.min(len);
3133    loop {
3134        if file.seek(SeekFrom::Start(len - window)).is_err() {
3135            return facts;
3136        }
3137        let Ok(size) = usize::try_from(window) else {
3138            return facts;
3139        };
3140        let mut buf = vec![0u8; size];
3141        if file.read_exact(&mut buf).is_err() {
3142            return facts;
3143        }
3144        // A transcript is append-only, so the newest record is the LAST one:
3145        // walk line boundaries backwards from the end and decode one record at a
3146        // time, stopping as soon as both facts are in hand. Decoding the whole
3147        // window instead would put a UTF-8 validation of every byte of every
3148        // transcript in the catalog on the path of one `discover`.
3149        //
3150        // `end` is the exclusive end of the line under inspection; the scan stops
3151        // at `floor`, because a window that starts mid-file almost certainly
3152        // starts mid-record and that partial first line belongs to the next,
3153        // wider window.
3154        let floor = if window < len {
3155            buf.iter().position(|byte| *byte == b'\n').map(|at| at + 1)
3156        } else {
3157            Some(0)
3158        };
3159        if let Some(floor) = floor {
3160            let mut end = buf.len();
3161            while end > floor && !(facts.last_turn_ms.is_some() && facts.model.is_some()) {
3162                let start = buf[floor..end]
3163                    .iter()
3164                    .rposition(|byte| *byte == b'\n')
3165                    .map_or(floor, |at| floor + at + 1);
3166                if let Ok(record) = std::str::from_utf8(&buf[start..end])
3167                    .map_err(|_| ())
3168                    .and_then(|line| serde_json::from_str::<Value>(line).map_err(|_| ()))
3169                {
3170                    if facts.last_turn_ms.is_none() {
3171                        facts.last_turn_ms = record_timestamp(&record, harness)
3172                            .and_then(crate::sidecar::rfc3339_to_ms)
3173                            .and_then(|millis| u64::try_from(millis).ok());
3174                    }
3175                    if facts.model.is_none() {
3176                        facts.model = record_model(&record, harness).map(str::to_owned);
3177                    }
3178                }
3179                end = start.saturating_sub(1);
3180            }
3181        }
3182        if facts.last_turn_ms.is_some() || window >= len || window >= TAIL_SCAN_LIMIT {
3183            return facts;
3184        }
3185        window = (window * 2).min(len).min(TAIL_SCAN_LIMIT);
3186    }
3187}
3188
3189/// When one transcript record was written, in the dialect that wrote it.
3190/// Every dialect discovery reads stamps its records with an RFC3339 UTC string;
3191/// only the key differs.
3192fn record_timestamp<'a>(record: &'a Value, harness: &str) -> Option<&'a str> {
3193    let key = match harness {
3194        HarnessId::SUPERCODE => "ts",
3195        _ => "timestamp",
3196    };
3197    record.get(key)?.as_str()
3198}
3199
3200/// The model one transcript record names, in the dialect that wrote it.
3201/// Mirrors the model arms of [`update_header_meta`], read newest-first.
3202fn record_model<'a>(record: &'a Value, harness: &str) -> Option<&'a str> {
3203    match harness {
3204        HarnessId::CLAUDE_CODE | HarnessId::PI => record.get("message")?.get("model")?.as_str(),
3205        HarnessId::CODEX => {
3206            if record.get("type")?.as_str()? != "turn_context" {
3207                return None;
3208            }
3209            record.get("payload")?.get("model")?.as_str()
3210        }
3211        _ => None,
3212    }
3213}
3214
3215/// Bytes read from a transcript's tail on the first attempt: comfortably more
3216/// than one record in every dialect discovery reads.
3217const TAIL_SCAN_START: u64 = 16 * 1024;
3218
3219/// Where the widening tail scan stops. A transcript whose last mebibyte holds
3220/// no timestamped record is not one whose recency this can honestly report.
3221const TAIL_SCAN_LIMIT: u64 = 1024 * 1024;
3222
3223fn modified_ms(path: &Path) -> Option<u64> {
3224    fs::metadata(path)
3225        .ok()?
3226        .modified()
3227        .ok()?
3228        .duration_since(UNIX_EPOCH)
3229        .ok()
3230        .and_then(|duration| u64::try_from(duration.as_millis()).ok())
3231}
3232
3233/// A workspace filter is satisfiable only by a session whose RECORDED working
3234/// directory is absolute. A relative recorded cwd (OpenCode has shipped
3235/// literal `"."` session rows) carries no information about where the session
3236/// ran; resolving it against the discoverer's own current directory made such
3237/// a session match every workspace discovery happened to run from.
3238fn recorded_cwd_matches(recorded: &Path, wanted: &Path) -> bool {
3239    recorded.is_absolute() && same_path(recorded, wanted)
3240}
3241
3242fn same_path(left: &Path, right: &Path) -> bool {
3243    match (fs::canonicalize(left), fs::canonicalize(right)) {
3244        (Ok(left), Ok(right)) => left == right,
3245        _ => normalize_path(left) == normalize_path(right),
3246    }
3247}
3248
3249fn normalize_path(path: &Path) -> PathBuf {
3250    let absolute = if path.is_absolute() {
3251        path.to_path_buf()
3252    } else {
3253        std::env::current_dir()
3254            .unwrap_or_else(|_| PathBuf::from("."))
3255            .join(path)
3256    };
3257    let mut normalized = PathBuf::new();
3258    for component in absolute.components() {
3259        match component {
3260            Component::CurDir => {}
3261            Component::ParentDir => {
3262                normalized.pop();
3263            }
3264            other => normalized.push(other.as_os_str()),
3265        }
3266    }
3267    normalized
3268}
3269
3270#[cfg(test)]
3271mod tests {
3272    use super::*;
3273    use std::io::Write;
3274    use std::time::{SystemTime, UNIX_EPOCH};
3275
3276    fn temp_dir(label: &str) -> PathBuf {
3277        let nonce = SystemTime::now()
3278            .duration_since(UNIX_EPOCH)
3279            .unwrap()
3280            .as_nanos();
3281        let path = std::env::temp_dir().join(format!(
3282            "supercode-catalog-{label}-{}-{nonce}",
3283            std::process::id()
3284        ));
3285        fs::create_dir_all(&path).unwrap();
3286        path
3287    }
3288
3289    #[test]
3290    fn codex_history_index_reads_appends_and_repairs_replacements() {
3291        let root = temp_dir("codex-history-index");
3292        let sessions = root.join("sessions");
3293        fs::create_dir_all(&sessions).unwrap();
3294        let history = root.join("history.jsonl");
3295        fs::write(
3296            &history,
3297            "{\"session_id\":\"alpha\",\"text\":\"first topic\"}\n",
3298        )
3299        .unwrap();
3300
3301        let mut index = CodexHistoryTopicIndex::new(&sessions);
3302        assert_eq!(index.refresh().unwrap(), BTreeSet::from(["alpha".into()]));
3303        assert_eq!(index.topics["alpha"][0].content, "first topic");
3304        assert!(index.refresh().unwrap().is_empty());
3305
3306        let mut file = fs::OpenOptions::new().append(true).open(&history).unwrap();
3307        write!(
3308            file,
3309            "{{\"session_id\":\"alpha\",\"text\":\"later topic\"}}\n\
3310             {{\"session_id\":\"beta\",\"text\":\"second topic\"}}\n"
3311        )
3312        .unwrap();
3313        file.flush().unwrap();
3314        assert_eq!(index.refresh().unwrap(), BTreeSet::from(["beta".into()]));
3315        assert_eq!(index.topics["alpha"][0].content, "first topic");
3316        assert_eq!(index.topics["beta"][0].content, "second topic");
3317
3318        fs::write(
3319            &history,
3320            "{\"session_id\":\"gamma\",\"text\":\"replacement\"}\n",
3321        )
3322        .unwrap();
3323        assert_eq!(
3324            index.refresh().unwrap(),
3325            BTreeSet::from(["alpha".into(), "beta".into(), "gamma".into()])
3326        );
3327        assert!(!index.topics.contains_key("alpha"));
3328        assert_eq!(index.topics["gamma"][0].content, "replacement");
3329
3330        fs::remove_dir_all(root).ok();
3331    }
3332
3333    #[test]
3334    fn codex_history_index_retains_an_incomplete_appended_record() {
3335        let root = temp_dir("codex-history-partial");
3336        let sessions = root.join("sessions");
3337        fs::create_dir_all(&sessions).unwrap();
3338        let history = root.join("history.jsonl");
3339        fs::write(&history, "{\"session_id\":\"partial\",\"text\":\"hel").unwrap();
3340
3341        let mut index = CodexHistoryTopicIndex::new(&sessions);
3342        assert!(index.refresh().unwrap().is_empty());
3343        let mut file = fs::OpenOptions::new().append(true).open(&history).unwrap();
3344        writeln!(file, "lo\"}}").unwrap();
3345        file.flush().unwrap();
3346
3347        assert_eq!(index.refresh().unwrap(), BTreeSet::from(["partial".into()]));
3348        assert_eq!(index.topics["partial"][0].content, "hello");
3349        fs::remove_dir_all(root).ok();
3350    }
3351
3352    #[test]
3353    fn cached_codex_history_enrichment_matches_stateless_discovery() {
3354        let root = temp_dir("codex-history-parity");
3355        let sessions = root.join("sessions");
3356        let workspace = root.join("workspace");
3357        fs::create_dir_all(&sessions).unwrap();
3358        fs::create_dir_all(&workspace).unwrap();
3359        fs::write(
3360            sessions.join("rollout.jsonl"),
3361            format!(
3362                "{{\"type\":\"session_meta\",\"payload\":{{\"id\":\"alpha\",\"cwd\":{}}}}}\n{{\"type\":\"turn_context\",\"payload\":{{\"cwd\":{},\"model\":\"gpt-test\"}}}}\n{{\"type\":\"response_item\",\"payload\":{{\"type\":\"message\",\"role\":\"user\",\"content\":[{{\"type\":\"input_text\",\"text\":\"transcript fallback\"}}]}}}}\n",
3363                serde_json::to_string(&workspace.to_string_lossy()).unwrap(),
3364                serde_json::to_string(&workspace.to_string_lossy()).unwrap(),
3365            ),
3366        )
3367        .unwrap();
3368        fs::write(
3369            root.join("history.jsonl"),
3370            "{\"session_id\":\"alpha\",\"text\":\"history topic\"}\n",
3371        )
3372        .unwrap();
3373        let query = DiscoveryQuery {
3374            harnesses: vec![HarnessId::from(HarnessId::CODEX)],
3375            homes: HarnessHomes {
3376                codex: sessions.clone(),
3377                ..HarnessHomes::default()
3378            },
3379            include_topic_candidates: true,
3380            ..DiscoveryQuery::default()
3381        };
3382        let catalog = HarnessCatalog::new();
3383        let projected = catalog
3384            .project_index(&query, catalog.discover_raw_index(&query))
3385            .unwrap();
3386        let expected = catalog
3387            .enrich_index_page(&query, projected.clone())
3388            .unwrap();
3389        let mut history = CodexHistoryTopicIndex::new(&sessions);
3390        history.refresh().unwrap();
3391        let actual = catalog
3392            .enrich_index_page_with_codex_history(&query, projected, &history)
3393            .unwrap();
3394
3395        assert_eq!(actual, expected);
3396        assert_eq!(actual[0].preview_candidates[0].content, "history topic");
3397        fs::remove_dir_all(root).ok();
3398    }
3399
3400    #[test]
3401    fn locator_json_round_trip_preserves_sqlite_selector() {
3402        let locator = SessionLocator {
3403            harness: HarnessId::from(HarnessId::OPENCODE),
3404            session_id: "ses_123".into(),
3405            storage: StorageLocator::Sqlite {
3406                path: PathBuf::from("/tmp/opencode-dev.db"),
3407                selector: "ses_123".into(),
3408            },
3409        };
3410        let encoded = serde_json::to_string(&locator).unwrap();
3411        assert_eq!(
3412            serde_json::from_str::<SessionLocator>(&encoded).unwrap(),
3413            locator
3414        );
3415    }
3416
3417    #[test]
3418    fn discovers_filters_loads_and_follows_three_jsonl_harnesses() {
3419        let root = temp_dir("jsonl");
3420        let workspace = root.join("workspace");
3421        let other = root.join("other");
3422        fs::create_dir_all(&workspace).unwrap();
3423        fs::create_dir_all(&other).unwrap();
3424
3425        let claude = root.join("claude");
3426        let codex = root.join("codex");
3427        let pi = root.join("pi");
3428        fs::create_dir_all(&claude).unwrap();
3429        fs::create_dir_all(&codex).unwrap();
3430        fs::create_dir_all(&pi).unwrap();
3431        fs::write(
3432            claude.join("claude.jsonl"),
3433            format!(
3434                "{{\"type\":\"user\",\"sessionId\":\"cc-1\",\"cwd\":{},\"timestamp\":\"2026-01-01T00:00:01Z\",\"message\":{{\"role\":\"user\",\"content\":\"hi\"}}}}\n",
3435                serde_json::to_string(&workspace.to_string_lossy()).unwrap()
3436            ),
3437        )
3438        .unwrap();
3439        fs::write(
3440            codex.join("rollout.jsonl"),
3441            format!(
3442                "{{\"timestamp\":\"2026-01-01T00:00:00Z\",\"type\":\"session_meta\",\"payload\":{{\"id\":\"cx-1\",\"cwd\":{}}}}}\n{{\"timestamp\":\"2026-01-01T00:00:02Z\",\"type\":\"response_item\",\"payload\":{{\"type\":\"message\",\"role\":\"user\",\"content\":[{{\"type\":\"input_text\",\"text\":\"inspect codex\"}}]}}}}\n",
3443                serde_json::to_string(&workspace.to_string_lossy()).unwrap()
3444            ),
3445        )
3446        .unwrap();
3447        fs::write(
3448            pi.join("pi.jsonl"),
3449            format!(
3450                "{{\"type\":\"session\",\"version\":3,\"id\":\"pi-1\",\"timestamp\":\"2026-01-01T00:00:00Z\",\"cwd\":{}}}\n{{\"type\":\"message\",\"message\":{{\"role\":\"user\",\"content\":\"inspect pi\"}}}}\n",
3451                serde_json::to_string(&workspace.to_string_lossy()).unwrap()
3452            ),
3453        )
3454        .unwrap();
3455        fs::write(
3456            pi.join("unrelated.jsonl"),
3457            format!(
3458                "{{\"type\":\"session\",\"version\":3,\"id\":\"pi-2\",\"timestamp\":\"2026-01-01T00:00:00Z\",\"cwd\":{}}}\n",
3459                serde_json::to_string(&other.to_string_lossy()).unwrap()
3460            ),
3461        )
3462        .unwrap();
3463        fs::write(claude.join("partial.jsonl"), "{truncated").unwrap();
3464
3465        let query = DiscoveryQuery {
3466            workspace: Some(workspace),
3467            homes: HarnessHomes {
3468                claude_code: claude,
3469                codex,
3470                pi,
3471                opencode: root.join("missing-opencode"),
3472                grok: root.join("missing-grok"),
3473                gemini: root.join("missing-gemini"),
3474                goose: root.join("missing-goose"),
3475                supercode: root.join("missing-supercode"),
3476                openclaw: root.join("missing-openclaw"),
3477                hermes: root.join("missing-hermes"),
3478                orchestrator: root.join("missing-orchestrator"),
3479            },
3480            ..DiscoveryQuery::default()
3481        };
3482        let catalog = HarnessCatalog::new();
3483        let found = catalog.discover(&query).unwrap();
3484        assert_eq!(found.len(), 3);
3485        assert_eq!(
3486            found
3487                .iter()
3488                .map(|item| item.locator.harness.as_str())
3489                .collect::<HashSet<_>>(),
3490            HashSet::from([HarnessId::CLAUDE_CODE, HarnessId::CODEX, HarnessId::PI])
3491        );
3492        for descriptor in found {
3493            assert!(descriptor.preview_candidates.is_empty());
3494            assert_eq!(descriptor.latest_message_candidates.len(), 1);
3495            assert_eq!(descriptor.latest_message_candidates[0].role, "user");
3496            assert!(descriptor.latest_message_candidates[0].cursor.is_some());
3497            if descriptor.locator.harness.as_str() == HarnessId::CLAUDE_CODE {
3498                assert_eq!(
3499                    descriptor.latest_message_candidates[0]
3500                        .metadata
3501                        .get("timestamp")
3502                        .map(String::as_str),
3503                    Some("2026-01-01T00:00:01Z")
3504                );
3505            } else if descriptor.locator.harness.as_str() == HarnessId::CODEX {
3506                assert_eq!(
3507                    descriptor.latest_message_candidates[0]
3508                        .metadata
3509                        .get("timestamp")
3510                        .map(String::as_str),
3511                    Some("2026-01-01T00:00:02Z")
3512                );
3513            }
3514            let loaded = catalog.load(&descriptor.locator).unwrap();
3515            assert_eq!(
3516                loaded.meta.session_id.as_deref(),
3517                Some(descriptor.locator.session_id.as_str())
3518            );
3519            let mut follower = catalog.follow(&descriptor.locator).unwrap();
3520            assert!(matches!(
3521                follower.poll().unwrap(),
3522                Some(crate::SessionWatchEvent::SessionSnapshot { .. })
3523            ));
3524        }
3525        fs::remove_dir_all(root).ok();
3526    }
3527
3528    #[test]
3529    fn codex_event_messages_supply_bounded_native_order_previews() {
3530        // The list reader must handle the native event-only narration emitted by
3531        // collab sessions, without loading the transcript or widening its budgets.
3532        let root = temp_dir("codex-event-previews");
3533        let path = root.join("rollout.jsonl");
3534        let user = serde_json::json!({
3535            "timestamp": "2026-01-01T00:00:01Z", "type": "event_msg",
3536            "payload": {"type": "user_message", "message": "Investigate the worker"}
3537        });
3538        let answer = serde_json::json!({
3539            "timestamp": "2026-01-01T00:00:02Z", "type": "event_msg",
3540            "payload": {"type": "agent_message", "message": "Worker findings"}
3541        });
3542        let noise = serde_json::json!({
3543            "type": "event_msg", "payload": {"type": "token_count", "message": "not a message"}
3544        });
3545        fs::write(&path, format!("{user}\n{answer}\n{noise}\n{{partial")).unwrap();
3546        let latest = latest_file_message_candidates(&path, HarnessId::CODEX).unwrap();
3547        assert_eq!(latest.len(), 2);
3548        assert_eq!(latest[0].role, "assistant");
3549        assert_eq!(latest[0].content, "Worker findings");
3550        assert_eq!(latest[1].role, "user");
3551        assert_eq!(latest[1].content, "Investigate the worker");
3552        assert_eq!(
3553            latest[0].metadata.get("timestamp").map(String::as_str),
3554            Some("2026-01-01T00:00:02Z")
3555        );
3556        assert_eq!(
3557            latest[0].cursor.as_deref(),
3558            Some(message_candidate_cursor(HarnessId::CODEX, &answer).as_str())
3559        );
3560        let topics = topic_file_message_candidates(&path, HarnessId::CODEX).unwrap();
3561        assert_eq!(topics.len(), 2);
3562        assert_eq!(topics[0].content, "Investigate the worker");
3563
3564        let mut context_pairs = String::new();
3565        for index in 0..5 {
3566            let content = if index < 4 {
3567                format!("# AGENTS.md instructions for /work/{index}\n\n<INSTRUCTIONS>Context</INSTRUCTIONS>")
3568            } else {
3569                "The actual user request".to_string()
3570            };
3571            let event = serde_json::json!({
3572                "type": "event_msg", "payload": {"type": "user_message", "message": content}
3573            });
3574            let response = serde_json::json!({
3575                "type": "response_item", "payload": {"type": "message", "role": "user", "content": content}
3576            });
3577            context_pairs.push_str(&format!("{event}\n{response}\n"));
3578        }
3579        fs::write(&path, context_pairs).unwrap();
3580        let topics = topic_file_message_candidates(&path, HarnessId::CODEX).unwrap();
3581        assert!(topics
3582            .iter()
3583            .any(|candidate| candidate.content == "The actual user request"));
3584
3585        // Mirrors must not halve the previously visible response-item window.
3586        // Either physical order keeps the response's original native cursor.
3587        let mut paired = String::new();
3588        for index in 0..12 {
3589            let content = format!("answer {index}");
3590            let event = serde_json::json!({
3591                "type": "event_msg", "payload": {"type": "agent_message", "message": content}
3592            });
3593            let response = serde_json::json!({
3594                "id": format!("response-{index}"), "type": "response_item",
3595                "payload": {"type": "message", "role": "assistant", "content": content}
3596            });
3597            if index % 2 == 0 {
3598                paired.push_str(&format!("{event}\n{response}\n"));
3599            } else {
3600                paired.push_str(&format!("{response}\n{event}\n"));
3601            }
3602        }
3603        fs::write(&path, paired).unwrap();
3604        let latest = latest_file_message_candidates(&path, HarnessId::CODEX).unwrap();
3605        assert_eq!(latest.len(), LATEST_PREVIEW_CANDIDATES);
3606        assert_eq!(latest[0].content, "answer 11");
3607        assert_eq!(latest[1].content, "answer 10");
3608        assert_eq!(latest[7].content, "answer 4");
3609        for (offset, candidate) in latest.iter().enumerate() {
3610            let response = serde_json::json!({"id": format!("response-{}", 11 - offset)});
3611            assert_eq!(
3612                candidate.cursor.as_deref(),
3613                Some(message_candidate_cursor(HarnessId::CODEX, &response).as_str())
3614            );
3615        }
3616        let topics = topic_file_message_candidates(&path, HarnessId::CODEX).unwrap();
3617        assert_eq!(topics.len(), LATEST_PREVIEW_CANDIDATES);
3618        assert_eq!(topics[7].content, "answer 7");
3619
3620        let event = serde_json::json!({
3621            "type": "event_msg", "payload": {"type": "agent_message", "message": "again"}
3622        });
3623        let response = serde_json::json!({
3624            "type": "response_item", "payload": {"type": "message", "role": "assistant", "content": "again"}
3625        });
3626        for records in [
3627            format!("{event}\n{event}\n"),
3628            format!("{response}\n{response}\n"),
3629            format!("{event}\n{response}\n{event}\n{response}\n"),
3630            format!("{response}\n{event}\n{response}\n{event}\n"),
3631        ] {
3632            fs::write(&path, records).unwrap();
3633            assert_eq!(
3634                latest_file_message_candidates(&path, HarnessId::CODEX)
3635                    .unwrap()
3636                    .len(),
3637                2
3638            );
3639            assert_eq!(
3640                topic_file_message_candidates(&path, HarnessId::CODEX)
3641                    .unwrap()
3642                    .len(),
3643                2
3644            );
3645        }
3646
3647        // Equal truncated prefixes alone are not evidence of a mirrored turn.
3648        let prefix = "x".repeat(4096);
3649        let distinct_event = serde_json::json!({
3650            "type": "event_msg", "payload": {"type": "agent_message", "message": format!("{prefix}A")}
3651        });
3652        let distinct_response = serde_json::json!({
3653            "type": "response_item", "payload": {"type": "message", "role": "assistant", "content": format!("{prefix}B")}
3654        });
3655        fs::write(&path, format!("{distinct_event}\n{distinct_response}\n")).unwrap();
3656        assert_eq!(
3657            latest_file_message_candidates(&path, HarnessId::CODEX)
3658                .unwrap()
3659                .len(),
3660            2
3661        );
3662        assert_eq!(
3663            topic_file_message_candidates(&path, HarnessId::CODEX)
3664                .unwrap()
3665                .len(),
3666            2
3667        );
3668        let long = serde_json::json!({
3669            "type": "event_msg", "payload": {"type": "agent_message", "message": "x".repeat(5000)}
3670        });
3671        fs::write(&path, format!("{long}\n")).unwrap();
3672        let latest = latest_file_message_candidates(&path, HarnessId::CODEX).unwrap();
3673        assert_eq!(latest[0].content.len(), 4096);
3674        fs::remove_dir_all(root).ok();
3675    }
3676
3677    #[test]
3678    fn preview_cursor_tracks_native_boundary_not_growing_text() {
3679        let first = serde_json::json!({
3680            "timestamp": "2026-01-01T00:00:02Z",
3681            "type": "response_item",
3682            "payload": {"type": "message", "role": "assistant", "content": "partial"}
3683        });
3684        let grown = serde_json::json!({
3685            "timestamp": "2026-01-01T00:00:02Z",
3686            "type": "response_item",
3687            "payload": {"type": "message", "role": "assistant", "content": "partial and complete"}
3688        });
3689        let next = serde_json::json!({
3690            "timestamp": "2026-01-01T00:00:03Z",
3691            "type": "response_item",
3692            "payload": {"type": "message", "role": "assistant", "content": "next"}
3693        });
3694
3695        assert_eq!(
3696            message_candidate_cursor(HarnessId::CODEX, &first),
3697            message_candidate_cursor(HarnessId::CODEX, &grown)
3698        );
3699        assert_ne!(
3700            message_candidate_cursor(HarnessId::CODEX, &first),
3701            message_candidate_cursor(HarnessId::CODEX, &next)
3702        );
3703    }
3704
3705    #[test]
3706    fn codex_child_rollouts_roll_into_roots_before_pagination() {
3707        let root = temp_dir("codex-roots");
3708        let codex = root.join("codex");
3709        fs::create_dir_all(&codex).unwrap();
3710        let write_rollout =
3711            |name: &str, payload: Value, modified_seconds: u64| {
3712                let path = codex.join(format!("{name}.jsonl"));
3713                fs::write(
3714                    &path,
3715                    format!(
3716                        "{}\n",
3717                        serde_json::json!({
3718                            "timestamp": "2026-01-01T00:00:00Z",
3719                            "type": "session_meta",
3720                            "payload": payload,
3721                        })
3722                    ),
3723                )
3724                .unwrap();
3725                File::open(&path)
3726                    .unwrap()
3727                    .set_times(fs::FileTimes::new().set_modified(
3728                        UNIX_EPOCH + std::time::Duration::from_secs(modified_seconds),
3729                    ))
3730                    .unwrap();
3731            };
3732        write_rollout(
3733            "parent",
3734            serde_json::json!({"id":"parent","cwd":"/project","source":"cli"}),
3735            100,
3736        );
3737        write_rollout(
3738            "other",
3739            serde_json::json!({"id":"other","cwd":"/project","source":"cli"}),
3740            200,
3741        );
3742        write_rollout(
3743            "child",
3744            serde_json::json!({
3745                "id": "child",
3746                "cwd": "/project",
3747                "parent_thread_id": "parent",
3748                "source": {"subagent":{"thread_spawn":{
3749                    "parent_thread_id":"parent",
3750                    "depth":1,
3751                    "agent_path":"/root/reviewer"
3752                }}}
3753            }),
3754            300,
3755        );
3756
3757        let catalog = HarnessCatalog::new();
3758        let query = DiscoveryQuery {
3759            harnesses: vec![HarnessId::from(HarnessId::CODEX)],
3760            homes: HarnessHomes {
3761                codex: codex.clone(),
3762                ..HarnessHomes::default()
3763            },
3764            limit: Some(1),
3765            ..DiscoveryQuery::default()
3766        };
3767        let roots = catalog.discover(&query).unwrap();
3768        assert_eq!(roots.len(), 1);
3769        assert_eq!(roots[0].locator.session_id, "parent");
3770        assert_eq!(roots[0].updated_at_ms, Some(300_000));
3771        assert_eq!(roots[0].parent_session_id, None);
3772        assert_eq!(roots[0].child_session_count, 1);
3773
3774        let tree = catalog
3775            .discover(&DiscoveryQuery {
3776                limit: None,
3777                include_child_sessions: true,
3778                root_session_id: Some("parent".into()),
3779                ..query
3780            })
3781            .unwrap();
3782        assert_eq!(tree.len(), 2);
3783        assert!(tree
3784            .iter()
3785            .all(|descriptor| descriptor.locator.session_id != "other"));
3786        let child = tree
3787            .iter()
3788            .find(|descriptor| descriptor.locator.session_id == "child")
3789            .unwrap();
3790        assert_eq!(child.parent_session_id.as_deref(), Some("parent"));
3791        fs::remove_dir_all(root).ok();
3792    }
3793
3794    #[test]
3795    fn discovers_loads_and_follows_opencode_sqlite() {
3796        let db = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
3797            .join("../harness/tests/fixtures/opencode_fixture/opencode.db");
3798        let catalog = HarnessCatalog::new();
3799        let found = catalog
3800            .discover(&DiscoveryQuery {
3801                harnesses: vec![HarnessId::from(HarnessId::OPENCODE)],
3802                homes: HarnessHomes {
3803                    opencode: db,
3804                    ..HarnessHomes::default()
3805                },
3806                ..DiscoveryQuery::default()
3807            })
3808            .unwrap();
3809        assert!(!found.is_empty());
3810        for descriptor in found {
3811            assert_eq!(descriptor.locator.harness.as_str(), HarnessId::OPENCODE);
3812            assert_eq!(
3813                catalog.load(&descriptor.locator).unwrap().meta.session_id,
3814                Some(descriptor.locator.session_id.clone())
3815            );
3816            assert!(catalog.follow(&descriptor.locator).is_ok());
3817        }
3818    }
3819
3820    #[test]
3821    fn discovers_loads_and_follows_hermes_sqlite_by_session() {
3822        // A copy of the committed Hermes fixture store, so the test may append to it.
3823        let root = temp_dir("hermes-follow");
3824        fs::create_dir_all(&root).unwrap();
3825        let db = root.join("state.db");
3826        fs::copy(
3827            PathBuf::from(env!("CARGO_MANIFEST_DIR"))
3828                .join("../harness/tests/fixtures/hermes_home/state.db"),
3829            &db,
3830        )
3831        .unwrap();
3832        let catalog = HarnessCatalog::new();
3833        let query = DiscoveryQuery {
3834            harnesses: vec![HarnessId::from(HarnessId::HERMES)],
3835            homes: HarnessHomes {
3836                hermes: db.clone(),
3837                ..HarnessHomes::default()
3838            },
3839            ..DiscoveryQuery::default()
3840        };
3841        let found = catalog.discover(&query).unwrap();
3842        assert!(found.len() >= 2, "{found:#?}");
3843        // Every discovered locator loads AND follows as ITS OWN session (not the store's newest),
3844        // and its list preview is that session's own latest message rather than an empty read of
3845        // the store file's tail.
3846        for descriptor in &found {
3847            assert_eq!(descriptor.locator.harness.as_str(), HarnessId::HERMES);
3848            let loaded = catalog.load(&descriptor.locator).unwrap();
3849            let last_text = loaded.messages.iter().rev().find_map(|message| {
3850                (matches!(message.role, crate::Role::User | crate::Role::Assistant))
3851                    .then(|| message.content.clone())
3852                    .flatten()
3853            });
3854            // Lineage-only fixture rows have no messages and therefore no preview.
3855            assert_eq!(
3856                descriptor
3857                    .latest_message_candidates
3858                    .first()
3859                    .map(|c| c.content.as_str()),
3860                last_text.as_deref(),
3861                "{}",
3862                descriptor.locator.session_id
3863            );
3864            assert_eq!(
3865                catalog.load(&descriptor.locator).unwrap().meta.session_id,
3866                Some(descriptor.locator.session_id.clone())
3867            );
3868            let mut follower = catalog.follow(&descriptor.locator).unwrap();
3869            match follower.poll().unwrap() {
3870                Some(crate::watch::SessionWatchEvent::SessionSnapshot { session, .. }) => {
3871                    assert_eq!(
3872                        session.meta.session_id,
3873                        Some(descriptor.locator.session_id.clone())
3874                    );
3875                }
3876                other => panic!("expected an initial snapshot, got {other:?}"),
3877            }
3878        }
3879        // Append a message to ONE session: only that session's follower wakes, with exactly the new
3880        // message, while a sibling's follower stays quiet.
3881        let target = &found[0].locator;
3882        let sibling = &found[1].locator;
3883        let mut target_follower = catalog.follow(target).unwrap();
3884        let mut sibling_follower = catalog.follow(sibling).unwrap();
3885        target_follower.poll().unwrap();
3886        sibling_follower.poll().unwrap();
3887        std::thread::sleep(std::time::Duration::from_millis(20));
3888        {
3889            let conn = rusqlite::Connection::open(&db).unwrap();
3890            conn.execute(
3891                "INSERT INTO messages (session_id, role, content, timestamp, active) VALUES (?1, 'assistant', 'appended by the follow test', ?2, 1)",
3892                rusqlite::params![target.session_id, 1_800_000_000.0_f64],
3893            )
3894            .unwrap();
3895        }
3896        match target_follower.poll().unwrap() {
3897            Some(crate::watch::SessionWatchEvent::MessagesAppended {
3898                session_id,
3899                messages,
3900                ..
3901            }) => {
3902                assert_eq!(session_id, Some(target.session_id.clone()));
3903                assert_eq!(messages.len(), 1);
3904                assert_eq!(
3905                    messages[0].content.as_deref(),
3906                    Some("appended by the follow test")
3907                );
3908            }
3909            other => panic!("expected messages_appended for the target session, got {other:?}"),
3910        }
3911        assert!(
3912            sibling_follower.poll().unwrap().is_none(),
3913            "the sibling session must not wake"
3914        );
3915        fs::remove_dir_all(&root).ok();
3916    }
3917
3918    #[test]
3919    fn discovers_loads_and_follows_gemini_conversation_records() {
3920        let root = temp_dir("gemini");
3921        let workspace = root.join("workspace");
3922        let chats = root.join("gemini/tmp/demo/chats");
3923        fs::create_dir_all(&workspace).unwrap();
3924        fs::create_dir_all(&chats).unwrap();
3925        fs::write(
3926            root.join("gemini/projects.json"),
3927            serde_json::json!({
3928                "projects": {workspace.to_string_lossy(): "demo"}
3929            })
3930            .to_string(),
3931        )
3932        .unwrap();
3933        let transcript = chats.join("gemini-id.jsonl");
3934        fs::write(
3935            &transcript,
3936            include_str!("../../harness/tests/fixtures/gemini_session.jsonl"),
3937        )
3938        .unwrap();
3939
3940        let catalog = HarnessCatalog::new();
3941        let found = catalog
3942            .discover(&DiscoveryQuery {
3943                harnesses: vec![HarnessId::from(HarnessId::GEMINI)],
3944                homes: HarnessHomes {
3945                    gemini: root.join("gemini"),
3946                    ..HarnessHomes::default()
3947                },
3948                workspace: Some(workspace.clone()),
3949                ..DiscoveryQuery::default()
3950            })
3951            .unwrap();
3952
3953        assert_eq!(found.len(), 1);
3954        assert_eq!(found[0].cwd.as_deref(), Some(workspace.as_path()));
3955        assert_eq!(found[0].message_count, None);
3956        assert_eq!(found[0].model.as_deref(), Some("gemini-2.5-pro"));
3957        assert_eq!(found[0].title, None);
3958        assert!(found[0].preview_candidates.is_empty());
3959        assert_eq!(found[0].latest_message_candidates.len(), 3);
3960        assert_eq!(
3961            found[0].latest_message_candidates[0].content,
3962            "Fixture inspected."
3963        );
3964        let loaded = catalog.load(&found[0].locator).unwrap();
3965        assert_eq!(
3966            loaded.meta.session_id.as_deref(),
3967            Some("11111111-1111-4111-8111-111111111111")
3968        );
3969        assert_eq!(loaded.messages.len(), 4);
3970        assert!(matches!(
3971            catalog.follow(&found[0].locator).unwrap().poll().unwrap(),
3972            Some(crate::SessionWatchEvent::SessionSnapshot { .. })
3973        ));
3974        fs::remove_dir_all(root).ok();
3975    }
3976
3977    #[test]
3978    fn preview_search_filters_before_pagination_without_changing_metadata_search() {
3979        // Search/pagination must compose: filtering only the returned page loses
3980        // matches and gives a false total. Exercise the public catalog door.
3981        let root = temp_dir("preview-search");
3982        for (id, first, last) in [
3983            ("topic-hit", "NEBULA opening", "Finished"),
3984            ("latest-hit", "Ordinary opening", "Found the nebula"),
3985            ("no-hit", "Unrelated opening", "Finished"),
3986        ] {
3987            fs::write(root.join(format!("{id}.jsonl")), format!("{}\n{}\n",
3988                serde_json::json!({"sessionId": id, "cwd": "/work", "type": "user", "message": {"role": "user", "content": first}}),
3989                serde_json::json!({"sessionId": id, "type": "assistant", "message": {"role": "assistant", "content": last}}),
3990            )).unwrap();
3991        }
3992        let query: DiscoveryQuery = serde_json::from_value(serde_json::json!({
3993            "harnesses": ["claude-code"], "homes": {"claude_code": root},
3994            "query": "  nebula  ", "search_previews": true, "limit": 1
3995        }))
3996        .unwrap();
3997        let catalog = HarnessCatalog::new();
3998        let first = catalog.discover_page(&query).unwrap();
3999        assert!(first.receipt.searched_previews);
4000        assert_eq!(first.receipt.total_matched, 2);
4001        assert_eq!(first.sessions.len(), 1);
4002        assert!(first.receipt.truncated);
4003        let second = catalog
4004            .discover_page(&DiscoveryQuery {
4005                cursor: first.next_cursor.clone(),
4006                ..query.clone()
4007            })
4008            .unwrap();
4009        assert_eq!(second.receipt.total_matched, 2);
4010        assert_eq!(second.sessions.len(), 1);
4011        assert_ne!(first.sessions[0].locator, second.sessions[0].locator);
4012        assert!(!second.receipt.truncated);
4013        let mut metadata = serde_json::to_value(&query).unwrap();
4014        metadata["search_previews"] = false.into();
4015        let metadata_page = catalog
4016            .discover_page(&serde_json::from_value(metadata).unwrap())
4017            .unwrap();
4018        assert!(metadata_page.sessions.is_empty());
4019        assert!(serde_json::to_value(&metadata_page.receipt)
4020            .unwrap()
4021            .get("searched_previews")
4022            .is_none());
4023
4024        // Metadata remains part of the union; matching multiple candidates must
4025        // still yield one session, not one row per message.
4026        let all = catalog
4027            .discover_page(&DiscoveryQuery {
4028                query: Some("hit".into()),
4029                limit: None,
4030                ..query.clone()
4031            })
4032            .unwrap();
4033        assert_eq!(all.sessions.len(), 3);
4034        assert_eq!(all.receipt.total_matched, 3);
4035        let elsewhere = catalog
4036            .discover_page(&DiscoveryQuery {
4037                workspace: Some("/elsewhere".into()),
4038                ..query.clone()
4039            })
4040            .unwrap();
4041        assert_eq!(elsewhere.receipt.total_matched, 0);
4042        let excluded_by_time = catalog
4043            .discover_page(&DiscoveryQuery {
4044                updated_after_ms: Some(u64::MAX),
4045                ..query.clone()
4046            })
4047            .unwrap();
4048        assert_eq!(excluded_by_time.receipt.total_matched, 0);
4049        for invalid in [
4050            DiscoveryQuery {
4051                query: None,
4052                ..query.clone()
4053            },
4054            DiscoveryQuery {
4055                query: Some("  ".into()),
4056                ..query.clone()
4057            },
4058            DiscoveryQuery {
4059                limit: Some(0),
4060                ..query.clone()
4061            },
4062            DiscoveryQuery {
4063                cursor: Some("bad-cursor".into()),
4064                ..query.clone()
4065            },
4066            DiscoveryQuery {
4067                cursor: first.next_cursor,
4068                query: Some("absent".into()),
4069                ..query.clone()
4070            },
4071        ] {
4072            assert!(catalog.discover_page(&invalid).is_err());
4073        }
4074        assert!(catalog.project_index_page(&query, Vec::new()).is_err());
4075        fs::remove_dir_all(root).ok();
4076    }
4077
4078    #[test]
4079    fn preview_search_uses_codex_first_history_topic_and_bounded_candidates() {
4080        let root = temp_dir("preview-search-codex");
4081        let sessions = root.join("sessions");
4082        fs::create_dir_all(&sessions).unwrap();
4083        fs::write(
4084            root.join("history.jsonl"),
4085            format!(
4086                "{}\n{}\n",
4087                serde_json::json!({"session_id": "history-hit", "text": "Original nebula topic"}),
4088                serde_json::json!({"session_id": "history-hit", "text": "laterhistoryonly"}),
4089            ),
4090        )
4091        .unwrap();
4092        for id in ["history-hit", "latest-hit", "bounded"] {
4093            let mut content = format!(
4094                "{}\n",
4095                serde_json::json!({
4096                    "type": "session_meta", "payload": {"id": id, "cwd": "/work"}
4097                })
4098            );
4099            for index in 0..20 {
4100                let message = if id == "latest-hit" && index == 19 {
4101                    "Found NEBULA".to_string()
4102                } else if index == 10 {
4103                    "middlehistoryonly".to_string()
4104                } else {
4105                    format!("{}beyondtextcap", "x".repeat(4096))
4106                };
4107                content.push_str(&format!("{}\n", serde_json::json!({
4108                    "type": "event_msg", "payload": {"type": "agent_message", "message": message}
4109                })));
4110            }
4111            fs::write(sessions.join(format!("{id}.jsonl")), content).unwrap();
4112        }
4113        let catalog = HarnessCatalog::new();
4114        let query: DiscoveryQuery = serde_json::from_value(serde_json::json!({
4115            "harnesses": ["codex"], "homes": {"codex": sessions},
4116            "query": "nebula", "search_previews": true
4117        }))
4118        .unwrap();
4119        let page = catalog.discover_page(&query).unwrap();
4120        assert_eq!(page.receipt.total_matched, 2);
4121        for row in &page.sessions {
4122            assert!(row.preview_candidates.len() <= 8);
4123            assert!(row.latest_message_candidates.len() <= 8);
4124            assert!(row
4125                .preview_candidates
4126                .iter()
4127                .chain(&row.latest_message_candidates)
4128                .all(|candidate| candidate.content.chars().count() <= 4096));
4129        }
4130        for text in ["middlehistoryonly", "laterhistoryonly", "beyondtextcap"] {
4131            assert!(
4132                catalog
4133                    .discover_page(&DiscoveryQuery {
4134                        query: Some(text.into()),
4135                        ..query.clone()
4136                    })
4137                    .unwrap()
4138                    .sessions
4139                    .is_empty(),
4140                "not a full-history search: {text}"
4141            );
4142        }
4143        fs::remove_dir_all(root).unwrap();
4144    }
4145
4146    #[test]
4147    fn discovers_native_store_and_pages_search_results() {
4148        let root = temp_dir("supercode");
4149        let store_root = root.join("sessions");
4150        fs::create_dir_all(&store_root).unwrap();
4151        for (name, title) in [
4152            ("alpha", "Alpha planning"),
4153            ("beta", "Beta implementation"),
4154            ("gamma", "Gamma review"),
4155        ] {
4156            fs::write(
4157                store_root.join(format!("{name}.jsonl")),
4158                format!("{{\"role\":\"user\",\"content\":\"{title}\"}}\n"),
4159            )
4160            .unwrap();
4161            fs::write(
4162                store_root.join(format!("{name}.meta.json")),
4163                serde_json::json!({"name": name, "title": title}).to_string(),
4164            )
4165            .unwrap();
4166        }
4167        let catalog = HarnessCatalog::new();
4168        let base = DiscoveryQuery {
4169            harnesses: vec![HarnessId::from(HarnessId::SUPERCODE)],
4170            homes: HarnessHomes {
4171                supercode: store_root,
4172                ..HarnessHomes::default()
4173            },
4174            limit: Some(1),
4175            ..DiscoveryQuery::default()
4176        };
4177
4178        let first = catalog.discover_page(&base).unwrap();
4179        assert_eq!(first.sessions.len(), 1);
4180        assert!(first.next_cursor.is_some());
4181        let second = catalog
4182            .discover_page(&DiscoveryQuery {
4183                cursor: first.next_cursor,
4184                ..base.clone()
4185            })
4186            .unwrap();
4187        assert_eq!(second.sessions.len(), 1);
4188        assert_ne!(
4189            first.sessions[0].locator.session_id,
4190            second.sessions[0].locator.session_id
4191        );
4192        let search = catalog
4193            .discover_page(&DiscoveryQuery {
4194                limit: None,
4195                query: Some("implementation".into()),
4196                ..base
4197            })
4198            .unwrap();
4199        assert_eq!(search.sessions.len(), 1);
4200        assert_eq!(search.sessions[0].locator.session_id, "beta");
4201        assert_eq!(search.sessions[0].message_count, None);
4202        assert_eq!(
4203            catalog
4204                .load(&search.sessions[0].locator)
4205                .unwrap()
4206                .messages
4207                .len(),
4208            1
4209        );
4210        fs::remove_dir_all(root).ok();
4211    }
4212
4213    #[test]
4214    fn native_workspace_discovery_reads_bounded_sidecar_headers() {
4215        let root = temp_dir("supercode-bounded-header");
4216        let store_root = root.join("sessions");
4217        let workspace = root.join("project");
4218        fs::create_dir_all(&store_root).unwrap();
4219        fs::create_dir_all(&workspace).unwrap();
4220        let name = "bounded-native";
4221        fs::write(
4222            store_root.join(format!("{name}.meta.json")),
4223            serde_json::json!({"name": name, "title": "Bounded native"}).to_string(),
4224        )
4225        .unwrap();
4226        fs::write(
4227            store_root.join(format!("{name}.jsonl")),
4228            "{\"role\":\"user\",\"content\":\"projected view\"}\n",
4229        )
4230        .unwrap();
4231        let sidecar = [
4232            serde_json::json!({
4233                "supercode_native": 2,
4234                "source": "claude_code",
4235                "session_id": "native-session"
4236            })
4237            .to_string(),
4238            serde_json::json!({
4239                "type": "user",
4240                "sessionId": "native-session",
4241                "cwd": workspace,
4242                "message": {"role": "user", "content": "hello"}
4243            })
4244            .to_string(),
4245            serde_json::json!({
4246                "type": "assistant",
4247                "sessionId": "native-session",
4248                "cwd": workspace,
4249                "message": {"role": "assistant", "model": "claude-sonnet-5", "content": []}
4250            })
4251            .to_string(),
4252            // A full native-family parse rejects this trailing residue. Header
4253            // discovery must not touch it after it has enough metadata.
4254            "not-json".into(),
4255        ]
4256        .join("\n");
4257        fs::write(
4258            store_root.join(format!("{name}.sidecar.jsonl")),
4259            format!("{sidecar}\n"),
4260        )
4261        .unwrap();
4262
4263        let found = HarnessCatalog::new()
4264            .discover(&DiscoveryQuery {
4265                workspace: Some(workspace.clone()),
4266                harnesses: vec![HarnessId::from(HarnessId::SUPERCODE)],
4267                homes: HarnessHomes {
4268                    supercode: store_root,
4269                    ..HarnessHomes::default()
4270                },
4271                ..DiscoveryQuery::default()
4272            })
4273            .unwrap();
4274
4275        assert_eq!(found.len(), 1);
4276        assert_eq!(found[0].cwd.as_deref(), Some(workspace.as_path()));
4277        assert_eq!(found[0].model.as_deref(), Some("claude-sonnet-5"));
4278        assert_eq!(found[0].message_count, None);
4279        fs::remove_dir_all(root).ok();
4280    }
4281
4282    #[test]
4283    fn discovers_current_opencode_schema_without_a_session_model_column() {
4284        let root = temp_dir("opencode-current");
4285        let db = root.join("opencode.db");
4286        let conn = Connection::open(&db).unwrap();
4287        conn.execute_batch(
4288            "CREATE TABLE session (
4289                id TEXT PRIMARY KEY,
4290                directory TEXT NOT NULL,
4291                title TEXT NOT NULL,
4292                time_updated INTEGER NOT NULL
4293             );
4294             CREATE TABLE message (
4295                id TEXT PRIMARY KEY,
4296                session_id TEXT NOT NULL
4297             );
4298             INSERT INTO session VALUES ('ses_current', '/tmp/work', 'Current', 42);
4299             INSERT INTO message VALUES ('msg_current', 'ses_current');",
4300        )
4301        .unwrap();
4302        drop(conn);
4303
4304        let found = HarnessCatalog::new()
4305            .discover(&DiscoveryQuery {
4306                harnesses: vec![HarnessId::from(HarnessId::OPENCODE)],
4307                homes: HarnessHomes {
4308                    opencode: db,
4309                    ..HarnessHomes::default()
4310                },
4311                ..DiscoveryQuery::default()
4312            })
4313            .unwrap();
4314
4315        assert_eq!(found.len(), 1);
4316        assert_eq!(found[0].locator.session_id, "ses_current");
4317        assert_eq!(found[0].message_count, Some(1));
4318        assert_eq!(found[0].model, None);
4319        fs::remove_dir_all(root).ok();
4320    }
4321
4322    #[test]
4323    fn workspace_filter_never_matches_a_relative_recorded_cwd() {
4324        // OpenCode has shipped session rows whose `directory` is the literal
4325        // ".". Resolving that against the discoverer's own cwd made the
4326        // session match every workspace discovery ran from — the workspace
4327        // here IS the test process cwd, the exact aliasing that leaked.
4328        let root = temp_dir("opencode-relative-cwd");
4329        let db = root.join("opencode.db");
4330        let conn = Connection::open(&db).unwrap();
4331        let here = std::env::current_dir().unwrap();
4332        conn.execute_batch(&format!(
4333            "CREATE TABLE session (
4334                id TEXT PRIMARY KEY,
4335                directory TEXT NOT NULL,
4336                title TEXT NOT NULL,
4337                time_updated INTEGER NOT NULL
4338             );
4339             CREATE TABLE message (
4340                id TEXT PRIMARY KEY,
4341                session_id TEXT NOT NULL
4342             );
4343             INSERT INTO session VALUES ('ses_relative', '.', 'Ghost', 41);
4344             INSERT INTO session VALUES ('ses_here', '{}', 'Real', 42);",
4345            here.display()
4346        ))
4347        .unwrap();
4348        drop(conn);
4349
4350        let found = HarnessCatalog::new()
4351            .discover(&DiscoveryQuery {
4352                workspace: Some(here),
4353                harnesses: vec![HarnessId::from(HarnessId::OPENCODE)],
4354                homes: HarnessHomes {
4355                    opencode: db,
4356                    ..HarnessHomes::default()
4357                },
4358                ..DiscoveryQuery::default()
4359            })
4360            .unwrap();
4361
4362        assert_eq!(found.len(), 1);
4363        assert_eq!(found[0].locator.session_id, "ses_here");
4364        fs::remove_dir_all(root).ok();
4365    }
4366}