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