1use 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#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
28#[serde(tag = "kind", rename_all = "snake_case")]
29pub enum StorageLocator {
30 File {
32 path: PathBuf,
34 },
35 Sqlite {
37 path: PathBuf,
39 selector: String,
41 },
42}
43
44impl StorageLocator {
45 pub fn path(&self) -> &Path {
47 match self {
48 Self::File { path } | Self::Sqlite { path, .. } => path,
49 }
50 }
51}
52
53#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
55pub struct SessionLocator {
56 pub harness: HarnessId,
58 pub session_id: String,
60 pub storage: StorageLocator,
62}
63
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
66pub struct SessionDescriptor {
67 pub locator: SessionLocator,
70 pub cwd: Option<PathBuf>,
72 pub title: Option<String>,
74 #[serde(default, skip_serializing_if = "Vec::is_empty")]
79 pub preview_candidates: Vec<SessionPreviewCandidate>,
80 #[serde(default, skip_serializing_if = "Vec::is_empty")]
85 pub latest_message_candidates: Vec<SessionPreviewCandidate>,
86 pub updated_at_ms: Option<u64>,
88 pub message_count: Option<usize>,
93 pub model: Option<String>,
95 #[serde(default, skip_serializing_if = "Option::is_none")]
99 pub parent_session_id: Option<String>,
100 #[serde(default, skip_serializing_if = "is_zero")]
104 pub child_session_count: usize,
105 #[serde(flatten)]
111 pub nouns: OrchestrationNouns,
112}
113
114#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
116pub struct SessionPreviewCandidate {
117 #[serde(default, skip_serializing_if = "Option::is_none")]
121 pub cursor: Option<String>,
122 pub role: String,
125 pub content: String,
127 #[serde(default, skip_serializing_if = "HashMap::is_empty")]
129 pub metadata: HashMap<String, String>,
130}
131
132#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
134pub struct DiscoveryPage {
135 pub sessions: Vec<SessionDescriptor>,
137 pub next_cursor: Option<String>,
139 #[serde(default)]
142 pub receipt: DiscoveryReceipt,
143}
144
145#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
147#[serde(default)]
148pub struct DiscoveryReceipt {
149 #[serde(skip_serializing_if = "is_false")]
152 pub searched_previews: bool,
153 #[serde(skip_serializing_if = "Option::is_none")]
155 pub requested_after_ms: Option<u64>,
156 #[serde(skip_serializing_if = "Option::is_none")]
158 pub requested_before_ms: Option<u64>,
159 #[serde(skip_serializing_if = "Option::is_none")]
161 pub requested_limit: Option<usize>,
162 #[serde(skip_serializing_if = "Option::is_none")]
164 pub oldest_returned_ms: Option<u64>,
165 #[serde(skip_serializing_if = "Option::is_none")]
167 pub newest_returned_ms: Option<u64>,
168 pub returned: usize,
170 pub total_matched: usize,
172 pub truncated: bool,
176}
177
178#[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 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 pub fn path(&self) -> &Path {
216 &self.path
217 }
218
219 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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
320#[serde(default)]
321pub struct HarnessHomes {
322 pub claude_code: PathBuf,
324 pub codex: PathBuf,
326 pub pi: PathBuf,
328 pub opencode: PathBuf,
330 pub grok: PathBuf,
332 pub gemini: PathBuf,
334 pub goose: PathBuf,
336 pub supercode: PathBuf,
338 pub openclaw: PathBuf,
342 pub hermes: PathBuf,
345 pub orchestrator: PathBuf,
351}
352
353pub 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 .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 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#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
520#[serde(default)]
521pub struct DiscoveryQuery {
522 pub workspace: Option<PathBuf>,
524 pub workspace_family: Option<PathBuf>,
532 pub updated_after_ms: Option<u64>,
534 pub updated_before_ms: Option<u64>,
536 pub harnesses: Vec<HarnessId>,
538 pub homes: HarnessHomes,
540 pub query: Option<String>,
542 pub search_previews: bool,
546 pub cursor: Option<String>,
548 pub limit: Option<usize>,
550 pub include_topic_candidates: bool,
554 pub include_child_sessions: bool,
558 pub root_session_id: Option<String>,
561 pub profile: Option<String>,
566}
567
568#[derive(Debug, Clone, PartialEq, Eq)]
574struct RepoFamily {
575 identity: PathBuf,
578 is_repository: bool,
581 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 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 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
653fn 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#[derive(Debug, Default, Clone, Copy)]
681pub struct HarnessCatalog;
682
683impl HarnessCatalog {
684 pub fn new() -> Self {
686 Self
687 }
688
689 pub fn discover(&self, query: &DiscoveryQuery) -> Result<Vec<SessionDescriptor>> {
693 Ok(self.discover_page(query)?.sessions)
694 }
695
696 pub fn discover_raw_index(&self, query: &DiscoveryQuery) -> Vec<SessionDescriptor> {
701 self.scan_descriptors(query, true)
702 }
703
704 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 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 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 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 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 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 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 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 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 pub fn load(&self, locator: &SessionLocator) -> Result<Session> {
1052 self.load_with_fidelity(locator, Fidelity::ByteLossless)
1053 }
1054
1055 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 #[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 #[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 pub fn follow(&self, locator: &SessionLocator) -> Result<SessionFollower> {
1166 self.follow_with_fidelity(locator, Fidelity::ByteLossless)
1167 }
1168
1169 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 #[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
1200fn 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
1215fn 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
1558fn 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(¤t) 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 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
1825fn 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 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
1938fn discover_orchestrator(
1945 root: &Path,
1946 workspace: Option<&Path>,
1947 found: &mut Vec<SessionDescriptor>,
1948) {
1949 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 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
2058fn 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 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
2185fn 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 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 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
2761fn 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
2838fn 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 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
3080fn 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 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 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 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 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 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 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 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 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 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 "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 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}