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