use std::collections::{BTreeSet, HashMap, HashSet};
use std::fs::{self, File};
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
use std::path::{Component, Path, PathBuf};
use std::time::UNIX_EPOCH;
use rusqlite::Connection;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::native_store::load_native_store_family;
use crate::ontology::{Binding, OrchestratorBindingRow};
use crate::session::{
hermes_capture_nouns, openclaw_agent_id_from_path, openclaw_capture_header_nouns,
percent_decode_path, OrchestrationNouns, SessionMeta, SessionSource,
};
use crate::{Error, Fidelity, Result, Session, SessionFollower};
pub use crate::ontology::HarnessId;
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum StorageLocator {
File {
path: PathBuf,
},
Sqlite {
path: PathBuf,
selector: String,
},
}
impl StorageLocator {
pub fn path(&self) -> &Path {
match self {
Self::File { path } | Self::Sqlite { path, .. } => path,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct SessionLocator {
pub harness: HarnessId,
pub session_id: String,
pub storage: StorageLocator,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionDescriptor {
pub locator: SessionLocator,
pub cwd: Option<PathBuf>,
pub title: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub preview_candidates: Vec<SessionPreviewCandidate>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub latest_message_candidates: Vec<SessionPreviewCandidate>,
pub updated_at_ms: Option<u64>,
pub message_count: Option<usize>,
pub model: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub parent_session_id: Option<String>,
#[serde(default, skip_serializing_if = "is_zero")]
pub child_session_count: usize,
#[serde(flatten)]
pub nouns: OrchestrationNouns,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionPreviewCandidate {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cursor: Option<String>,
pub role: String,
pub content: String,
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub metadata: HashMap<String, String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DiscoveryPage {
pub sessions: Vec<SessionDescriptor>,
pub next_cursor: Option<String>,
#[serde(default)]
pub receipt: DiscoveryReceipt,
}
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(default)]
pub struct DiscoveryReceipt {
#[serde(skip_serializing_if = "Option::is_none")]
pub requested_after_ms: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub requested_before_ms: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub requested_limit: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
pub oldest_returned_ms: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub newest_returned_ms: Option<u64>,
pub returned: usize,
pub total_matched: usize,
pub truncated: bool,
}
#[derive(Debug)]
pub struct CodexHistoryTopicIndex {
path: PathBuf,
fingerprint: Option<CodexHistoryFingerprint>,
offset: u64,
trailing: Vec<u8>,
topics: HashMap<String, Vec<SessionPreviewCandidate>>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct CodexHistoryFingerprint {
len: u64,
modified_ns: u128,
identity: u128,
}
impl CodexHistoryTopicIndex {
pub fn new(sessions_root: &Path) -> Self {
let root = sessions_root.parent().unwrap_or(sessions_root);
Self {
path: root.join("history.jsonl"),
fingerprint: None,
offset: 0,
trailing: Vec::new(),
topics: HashMap::new(),
}
}
pub fn path(&self) -> &Path {
&self.path
}
pub fn refresh(&mut self) -> Result<BTreeSet<String>> {
let metadata = match fs::metadata(&self.path) {
Ok(metadata) => metadata,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
let changed = self.topics.keys().cloned().collect();
self.fingerprint = None;
self.offset = 0;
self.trailing.clear();
self.topics.clear();
return Ok(changed);
}
Err(error) => return Err(error.into()),
};
let fingerprint = codex_history_fingerprint(&metadata)?;
if self.fingerprint == Some(fingerprint) {
return Ok(BTreeSet::new());
}
let is_append = self.fingerprint.is_some_and(|previous| {
previous.identity == fingerprint.identity
&& previous.len < fingerprint.len
&& self.offset <= previous.len
});
if is_append {
let mut file = File::open(&self.path)?;
file.seek(SeekFrom::Start(self.offset))?;
let mut bytes = Vec::with_capacity(
usize::try_from(fingerprint.len.saturating_sub(self.offset)).unwrap_or(0),
);
file.read_to_end(&mut bytes)?;
self.offset = file.stream_position()?;
let changed = self.ingest(bytes);
self.fingerprint = Some(CodexHistoryFingerprint {
len: self.offset,
..fingerprint
});
return Ok(changed);
}
let previous = std::mem::take(&mut self.topics);
let mut file = File::open(&self.path)?;
let mut bytes = Vec::with_capacity(usize::try_from(fingerprint.len).unwrap_or(0));
file.read_to_end(&mut bytes)?;
self.offset = file.stream_position()?;
self.trailing.clear();
self.ingest(bytes);
self.fingerprint = Some(CodexHistoryFingerprint {
len: self.offset,
..fingerprint
});
Ok(changed_topic_ids(&previous, &self.topics))
}
fn ingest(&mut self, bytes: Vec<u8>) -> BTreeSet<String> {
let mut input = std::mem::take(&mut self.trailing);
input.extend(bytes);
let complete_len = input
.iter()
.rposition(|byte| *byte == b'\n')
.map_or(0, |index| index + 1);
let mut changed = BTreeSet::new();
for line in input[..complete_len].split(|byte| *byte == b'\n') {
self.ingest_line(line, &mut changed);
}
self.trailing.extend_from_slice(&input[complete_len..]);
if !self.trailing.is_empty() {
let trailing = self.trailing.clone();
if self.ingest_line(&trailing, &mut changed) {
self.trailing.clear();
}
}
changed
}
fn ingest_line(&mut self, line: &[u8], changed: &mut BTreeSet<String>) -> bool {
let Ok(value) = serde_json::from_slice::<Value>(line) else {
return false;
};
let Some(session_id) = value.get("session_id").and_then(Value::as_str) else {
return true;
};
if self.topics.contains_key(session_id) {
return true;
}
let mut candidates = Vec::new();
push_message_candidate(&mut candidates, "user", value.get("text"), HashMap::new());
if !candidates.is_empty() {
self.topics.insert(session_id.to_string(), candidates);
changed.insert(session_id.to_string());
}
true
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct HarnessHomes {
pub claude_code: PathBuf,
pub codex: PathBuf,
pub pi: PathBuf,
pub opencode: PathBuf,
pub grok: PathBuf,
pub gemini: PathBuf,
pub goose: PathBuf,
pub supercode: PathBuf,
pub openclaw: PathBuf,
pub hermes: PathBuf,
pub orchestrator: PathBuf,
}
pub fn orchestrator_profile_dirs(root: &Path) -> Vec<(String, PathBuf)> {
if !root.is_dir() {
return Vec::new();
}
let mut dirs = vec![("default".to_string(), root.to_path_buf())];
if let Ok(entries) = fs::read_dir(root.join("profiles")) {
let mut named: Vec<(String, PathBuf)> = entries
.flatten()
.filter(|entry| entry.path().is_dir())
.filter_map(|entry| {
entry
.file_name()
.into_string()
.ok()
.map(|name| (name, entry.path()))
})
.filter(|(name, _)| !name.starts_with('.'))
.collect();
named.sort();
dirs.extend(named);
}
dirs
}
impl Default for HarnessHomes {
fn default() -> Self {
let home = std::env::var_os("HOME")
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from("."));
let claude_root = std::env::var_os("CLAUDE_CONFIG_DIR")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".claude"));
let codex_root = std::env::var_os("CODEX_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".codex"));
let pi = std::env::var_os("PI_CODING_AGENT_SESSION_DIR")
.map(PathBuf::from)
.unwrap_or_else(|| {
std::env::var_os("PI_CODING_AGENT_DIR")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".pi/agent"))
.join("sessions")
});
let opencode = std::env::var_os("OPENCODE_DB")
.map(PathBuf::from)
.unwrap_or_else(|| {
std::env::var_os("XDG_DATA_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".local/share"))
.join("opencode")
});
let grok = std::env::var_os("GROK_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".grok"))
.join("sessions");
let gemini = std::env::var_os("GEMINI_CLI_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".gemini"));
let openclaw = std::env::var_os("OPENCLAW_STATE_DIR")
.map(PathBuf::from)
.or_else(|| {
std::env::var_os("OPENCLAW_HOME").map(|root| PathBuf::from(root).join(".openclaw"))
})
.unwrap_or_else(|| home.join(".openclaw"));
let orchestrator = std::env::var_os("SUPERCODE_ORCHESTRATOR_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".supercode/orchestrator"));
let hermes = std::env::var_os("HERMES_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".hermes"))
.join("state.db");
let goose = std::env::var_os("GOOSE_PATH_ROOT")
.map(PathBuf::from)
.map(|root| root.join("data/sessions/sessions.db"))
.unwrap_or_else(|| {
#[cfg(target_os = "macos")]
{
home.join("Library/Application Support/Block/goose/sessions/sessions.db")
}
#[cfg(target_os = "windows")]
{
std::env::var_os("APPDATA")
.map(PathBuf::from)
.unwrap_or_else(|| home.join("AppData/Roaming"))
.join("Block/goose/sessions/sessions.db")
}
#[cfg(not(any(target_os = "macos", target_os = "windows")))]
{
std::env::var_os("XDG_DATA_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".local/share"))
.join("goose/sessions/sessions.db")
}
});
let supercode = std::env::var_os("SUPERCODE_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| {
std::env::var_os("XDG_CONFIG_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home.join(".config"))
.join("supercode")
})
.join("sessions");
Self {
claude_code: claude_root.join("projects"),
codex: codex_root.join("sessions"),
gemini,
goose,
supercode,
openclaw,
hermes,
orchestrator,
pi,
opencode,
grok,
}
}
}
fn codex_history_fingerprint(metadata: &fs::Metadata) -> Result<CodexHistoryFingerprint> {
let modified_ns = metadata
.modified()?
.duration_since(UNIX_EPOCH)
.map_err(|error| Error::Other(format!("history timestamp predates Unix epoch: {error}")))?
.as_nanos();
#[cfg(unix)]
let identity = {
use std::os::unix::fs::MetadataExt;
(u128::from(metadata.dev()) << 64) | u128::from(metadata.ino())
};
#[cfg(not(unix))]
let identity = 0;
Ok(CodexHistoryFingerprint {
len: metadata.len(),
modified_ns,
identity,
})
}
fn changed_topic_ids(
before: &HashMap<String, Vec<SessionPreviewCandidate>>,
after: &HashMap<String, Vec<SessionPreviewCandidate>>,
) -> BTreeSet<String> {
before
.keys()
.chain(after.keys())
.filter(|session_id| before.get(*session_id) != after.get(*session_id))
.cloned()
.collect()
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct DiscoveryQuery {
pub workspace: Option<PathBuf>,
pub workspace_family: Option<PathBuf>,
pub updated_after_ms: Option<u64>,
pub updated_before_ms: Option<u64>,
pub harnesses: Vec<HarnessId>,
pub homes: HarnessHomes,
pub query: Option<String>,
pub cursor: Option<String>,
pub limit: Option<usize>,
pub include_topic_candidates: bool,
pub include_child_sessions: bool,
pub root_session_id: Option<String>,
pub profile: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct RepoFamily {
identity: PathBuf,
is_repository: bool,
origin_url: Option<String>,
}
impl RepoFamily {
fn of(path: &Path) -> Self {
let start = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
let mut current = Some(start.as_path());
while let Some(dir) = current {
let dot_git = dir.join(".git");
if dot_git.is_dir() {
let identity = std::fs::canonicalize(&dot_git).unwrap_or_else(|_| dot_git.clone());
let origin_url = read_origin_url(&identity);
return Self {
identity,
is_repository: true,
origin_url,
};
}
if dot_git.is_file() {
if let Ok(text) = std::fs::read_to_string(&dot_git) {
if let Some(gitdir) = text
.lines()
.find_map(|line| line.trim().strip_prefix("gitdir:"))
{
let gitdir = PathBuf::from(gitdir.trim());
let gitdir = if gitdir.is_absolute() {
gitdir
} else {
dir.join(gitdir)
};
let common = gitdir
.parent()
.filter(|parent| parent.ends_with("worktrees"))
.and_then(Path::parent)
.map(Path::to_path_buf)
.unwrap_or(gitdir);
let identity = std::fs::canonicalize(&common).unwrap_or(common);
let origin_url = read_origin_url(&identity);
return Self {
identity,
is_repository: true,
origin_url,
};
}
}
}
current = dir.parent();
}
Self {
identity: start,
is_repository: false,
origin_url: None,
}
}
fn joins(&self, other: &Self) -> bool {
if self.identity == other.identity {
return true;
}
self.is_repository
&& other.is_repository
&& matches!((&self.origin_url, &other.origin_url), (Some(a), Some(b)) if a == b)
}
}
fn read_origin_url(common_dir: &Path) -> Option<String> {
let text = std::fs::read_to_string(common_dir.join("config")).ok()?;
let mut in_origin = false;
for line in text.lines() {
let line = line.trim();
if line.starts_with('[') {
in_origin = line.starts_with("[remote \"origin\"]");
continue;
}
if in_origin {
if let Some(value) = line.strip_prefix("url") {
let value = value.trim_start();
if let Some(url) = value.strip_prefix('=') {
let url = url.trim();
if !url.is_empty() {
return Some(url.to_string());
}
}
}
}
}
None
}
#[derive(Debug, Default, Clone, Copy)]
pub struct HarnessCatalog;
impl HarnessCatalog {
pub fn new() -> Self {
Self
}
pub fn discover(&self, query: &DiscoveryQuery) -> Result<Vec<SessionDescriptor>> {
Ok(self.discover_page(query)?.sessions)
}
pub fn discover_raw_index(&self, query: &DiscoveryQuery) -> Vec<SessionDescriptor> {
self.scan_descriptors(query, true)
}
pub fn project_index(
&self,
query: &DiscoveryQuery,
descriptors: impl IntoIterator<Item = SessionDescriptor>,
) -> Result<Vec<SessionDescriptor>> {
let mut found = descriptors.into_iter().collect::<Vec<_>>();
project_descriptors(query, &mut found);
let (sessions, _) = paginate_descriptors(query, found)?;
Ok(sessions)
}
pub fn enrich_index_page(
&self,
query: &DiscoveryQuery,
mut sessions: Vec<SessionDescriptor>,
) -> Result<Vec<SessionDescriptor>> {
enrich_descriptors(query, &mut sessions, None)?;
Ok(sessions)
}
pub fn enrich_index_page_with_codex_history(
&self,
query: &DiscoveryQuery,
mut sessions: Vec<SessionDescriptor>,
codex_history: &CodexHistoryTopicIndex,
) -> Result<Vec<SessionDescriptor>> {
enrich_descriptors(query, &mut sessions, Some(codex_history))?;
Ok(sessions)
}
pub fn discover_page(&self, query: &DiscoveryQuery) -> Result<DiscoveryPage> {
let mut found = self.scan_descriptors(query, query.include_child_sessions);
project_descriptors(query, &mut found);
let total_matched = found.len();
let (mut sessions, next_cursor) = paginate_descriptors(query, found)?;
enrich_descriptors(query, &mut sessions, None)?;
let receipt = DiscoveryReceipt {
requested_after_ms: query.updated_after_ms,
requested_before_ms: query.updated_before_ms,
requested_limit: query.limit,
oldest_returned_ms: sessions.iter().filter_map(|s| s.updated_at_ms).min(),
newest_returned_ms: sessions.iter().filter_map(|s| s.updated_at_ms).max(),
returned: sessions.len(),
total_matched,
truncated: next_cursor.is_some(),
};
Ok(DiscoveryPage {
sessions,
next_cursor,
receipt,
})
}
fn scan_descriptors(
&self,
query: &DiscoveryQuery,
include_child_sessions: bool,
) -> Vec<SessionDescriptor> {
let selected: HashSet<&str> = if query.harnesses.is_empty() {
[
HarnessId::CLAUDE_CODE,
HarnessId::CODEX,
HarnessId::PI,
HarnessId::OPENCODE,
HarnessId::GROK,
HarnessId::GEMINI,
HarnessId::GOOSE,
HarnessId::SUPERCODE,
HarnessId::OPENCLAW,
HarnessId::HERMES,
HarnessId::ORCHESTRATOR,
]
.into_iter()
.collect()
} else {
query.harnesses.iter().map(HarnessId::as_str).collect()
};
let mut found = Vec::new();
if selected.contains(HarnessId::CLAUDE_CODE) {
discover_jsonl(
&query.homes.claude_code,
HarnessId::CLAUDE_CODE,
query.workspace.as_deref(),
include_child_sessions,
&mut found,
);
}
if selected.contains(HarnessId::CODEX) {
discover_jsonl(
&query.homes.codex,
HarnessId::CODEX,
query.workspace.as_deref(),
include_child_sessions,
&mut found,
);
}
if selected.contains(HarnessId::PI) {
discover_jsonl(
&query.homes.pi,
HarnessId::PI,
query.workspace.as_deref(),
include_child_sessions,
&mut found,
);
}
if selected.contains(HarnessId::OPENCODE) {
discover_opencode(
&query.homes.opencode,
query.workspace.as_deref(),
&mut found,
);
}
if selected.contains(HarnessId::GROK) {
discover_grok(&query.homes.grok, query.workspace.as_deref(), &mut found);
}
if selected.contains(HarnessId::GEMINI) {
discover_gemini(&query.homes.gemini, query.workspace.as_deref(), &mut found);
}
if selected.contains(HarnessId::GOOSE) {
discover_goose(&query.homes.goose, query.workspace.as_deref(), &mut found);
}
if selected.contains(HarnessId::OPENCLAW) {
discover_openclaw(
&query.homes.openclaw,
query.workspace.as_deref(),
&mut found,
);
}
if selected.contains(HarnessId::HERMES) {
discover_hermes(&query.homes.hermes, query.workspace.as_deref(), &mut found);
}
if selected.contains(HarnessId::ORCHESTRATOR) {
discover_orchestrator(
&query.homes.orchestrator,
query.workspace.as_deref(),
&mut found,
);
}
if selected.contains(HarnessId::SUPERCODE) {
discover_supercode(
&query.homes.supercode,
query.workspace.as_deref(),
&mut found,
);
}
for descriptor in &mut found {
finalize_nouns(descriptor);
}
found
}
pub fn refresh_file_descriptor(
&self,
locator: &SessionLocator,
workspace: Option<&Path>,
include_topic_candidates: bool,
) -> Result<Option<SessionDescriptor>> {
let Some(mut descriptor) = self.refresh_file_index_descriptor(locator, workspace)? else {
return Ok(None);
};
if include_topic_candidates {
descriptor.preview_candidates =
topic_message_candidates(&descriptor.locator).unwrap_or_default();
}
descriptor.latest_message_candidates =
latest_message_candidates(&descriptor.locator).unwrap_or_default();
Ok(Some(descriptor))
}
pub fn refresh_file_index_descriptor(
&self,
locator: &SessionLocator,
workspace: Option<&Path>,
) -> Result<Option<SessionDescriptor>> {
let StorageLocator::File { path } = &locator.storage else {
return Ok(None);
};
if !matches!(
locator.harness.as_str(),
HarnessId::CLAUDE_CODE | HarnessId::CODEX
) {
return Ok(None);
}
if !path.is_file() {
return Ok(None);
}
let Ok(meta) = read_header(path, locator.harness.as_str()) else {
return Ok(None);
};
if workspace.is_some_and(|wanted| {
meta.cwd
.as_deref()
.is_none_or(|cwd| !recorded_cwd_matches(cwd, wanted))
}) {
return Ok(None);
}
let parent_session_id = meta.parent_session_id.or_else(|| {
(locator.harness.as_str() == HarnessId::CLAUDE_CODE)
.then(|| claude_subagent_parent_id(path))
.flatten()
});
let descriptor = SessionDescriptor {
locator: SessionLocator {
harness: locator.harness.clone(),
session_id: meta
.session_id
.unwrap_or_else(|| locator.session_id.clone()),
storage: StorageLocator::File { path: path.clone() },
},
cwd: meta.cwd,
title: meta.title,
preview_candidates: Vec::new(),
latest_message_candidates: Vec::new(),
updated_at_ms: modified_ms(path),
message_count: None,
model: meta.model,
parent_session_id,
child_session_count: 0,
nouns: OrchestrationNouns::default(),
};
Ok(Some(descriptor))
}
pub fn load(&self, locator: &SessionLocator) -> Result<Session> {
self.load_with_fidelity(locator, Fidelity::ByteLossless)
}
pub fn load_with_fidelity(
&self,
locator: &SessionLocator,
fidelity: Fidelity,
) -> Result<Session> {
if let Some(session) = load_hermes_locator(locator) {
return session;
}
match &locator.storage {
StorageLocator::File { path } => {
if let Some(session) = load_native_store_family(path)? {
Ok(session)
} else {
Ok(Session::load_with_fidelity(path, fidelity)?)
}
}
StorageLocator::Sqlite { path, selector } => {
if locator.harness.as_str() == HarnessId::GOOSE {
Ok(Session::from_goose_sqlite(path, selector)?)
} else {
Ok(Session::from_opencode_sqlite(path, Some(selector))?)
}
}
}
}
#[doc(hidden)]
pub fn load_parent_with_fidelity(
&self,
locator: &SessionLocator,
fidelity: Fidelity,
) -> Result<Session> {
if let Some(session) = load_hermes_locator(locator) {
return session;
}
match &locator.storage {
StorageLocator::File { path } => {
if let Some(session) = load_native_store_family(path)? {
Ok(session)
} else {
Ok(Session::load_parent_with_fidelity(path, fidelity)?)
}
}
StorageLocator::Sqlite { path, selector } => {
if locator.harness.as_str() == HarnessId::GOOSE {
Ok(Session::from_goose_sqlite(path, selector)?)
} else {
Ok(Session::from_opencode_sqlite(path, Some(selector))?)
}
}
}
}
#[doc(hidden)]
pub fn load_display_view(
&self,
locator: &SessionLocator,
fidelity: Fidelity,
message_limit: usize,
) -> Result<Session> {
if let Some(session) = load_hermes_locator(locator) {
let mut session = session?;
if session.messages.len() > message_limit.max(1) {
session
.messages
.drain(..session.messages.len() - message_limit.max(1));
}
return Ok(session);
}
match &locator.storage {
StorageLocator::File { path } => {
if let Some(mut session) = load_native_store_family(path)? {
if session.messages.len() > message_limit.max(1) {
session
.messages
.drain(..session.messages.len() - message_limit.max(1));
}
Ok(session)
} else {
Ok(Session::load_display_view(path, fidelity, message_limit)?)
}
}
StorageLocator::Sqlite { path, selector } => {
let mut session = if locator.harness.as_str() == HarnessId::GOOSE {
Session::from_goose_sqlite_display(path, selector, message_limit)?
} else {
Session::from_opencode_sqlite(path, Some(selector))?
};
if session.messages.len() > message_limit.max(1) {
session
.messages
.drain(..session.messages.len() - message_limit.max(1));
}
Ok(session)
}
}
}
pub fn follow(&self, locator: &SessionLocator) -> Result<SessionFollower> {
self.follow_with_fidelity(locator, Fidelity::ByteLossless)
}
pub fn follow_with_fidelity(
&self,
locator: &SessionLocator,
fidelity: Fidelity,
) -> Result<SessionFollower> {
SessionFollower::open_locator_with_fidelity(locator, fidelity)
}
#[doc(hidden)]
pub fn follow_read_view(
&self,
locator: &SessionLocator,
fidelity: Fidelity,
include_subagents: bool,
message_limit: Option<usize>,
max_message_chars: Option<usize>,
display_history: bool,
) -> Result<SessionFollower> {
SessionFollower::open_locator_with_view(
locator,
fidelity,
include_subagents,
message_limit,
max_message_chars,
display_history,
)
}
}
fn load_hermes_locator(locator: &SessionLocator) -> Option<Result<Session>> {
if locator.harness.as_str() != HarnessId::HERMES {
return None;
}
let StorageLocator::File { path } = &locator.storage else {
return None;
};
Some(Session::from_hermes_sqlite(path, Some(&locator.session_id)))
}
fn finalize_nouns(descriptor: &mut SessionDescriptor) {
let mut meta = SessionMeta::new(SessionSource::Native);
meta.cwd = descriptor.cwd.clone();
meta.trigger = descriptor.nouns.trigger;
meta.surface = descriptor.nouns.surface.clone();
meta.profile = descriptor.nouns.profile.clone();
meta.recurrence = descriptor.nouns.recurrence.clone();
meta.cross_surface = descriptor.nouns.cross_surface.clone();
descriptor.nouns = OrchestrationNouns::from_meta(&meta);
}
fn project_descriptors(query: &DiscoveryQuery, found: &mut Vec<SessionDescriptor>) {
roll_up_session_children(found, query.include_child_sessions);
if let Some(root_session_id) = query.root_session_id.as_deref() {
retain_session_family(found, root_session_id);
}
if let Some(family_path) = query.workspace_family.as_deref() {
let family = RepoFamily::of(family_path);
let mut cache: HashMap<PathBuf, bool> = HashMap::new();
found.retain(|descriptor| {
let Some(cwd) = descriptor.cwd.as_deref() else {
return false;
};
*cache
.entry(cwd.to_path_buf())
.or_insert_with(|| RepoFamily::of(cwd).joins(&family))
});
}
if let Some(profile) = query
.profile
.as_deref()
.map(str::trim)
.filter(|p| !p.is_empty())
{
found.retain(|descriptor| descriptor.nouns.profile.as_deref() == Some(profile));
}
if let Some(after) = query.updated_after_ms {
found.retain(|descriptor| descriptor.updated_at_ms.is_some_and(|at| at >= after));
}
if let Some(before) = query.updated_before_ms {
found.retain(|descriptor| descriptor.updated_at_ms.is_some_and(|at| at <= before));
}
found.sort_by(|a, b| {
b.updated_at_ms
.cmp(&a.updated_at_ms)
.then_with(|| a.locator.harness.cmp(&b.locator.harness))
.then_with(|| a.locator.session_id.cmp(&b.locator.session_id))
});
if let Some(search) = query
.query
.as_deref()
.map(str::trim)
.filter(|query| !query.is_empty())
{
let search = search.to_lowercase();
found.retain(|descriptor| descriptor_matches(descriptor, &search));
}
}
fn paginate_descriptors(
query: &DiscoveryQuery,
found: Vec<SessionDescriptor>,
) -> Result<(Vec<SessionDescriptor>, Option<String>)> {
let start = match query.cursor.as_deref() {
Some(cursor) => {
let key = decode_cursor(cursor)?;
found
.iter()
.position(|descriptor| descriptor_cursor_key(descriptor) == key)
.map(|index| index + 1)
.ok_or_else(|| Error::Other("discovery cursor is stale or invalid".into()))?
}
None => 0,
};
let end = query
.limit
.map(|limit| start.saturating_add(limit).min(found.len()))
.unwrap_or(found.len());
let sessions = found[start.min(found.len())..end].to_vec();
let next_cursor = (end < found.len())
.then(|| sessions.last().map(encode_cursor))
.flatten();
Ok((sessions, next_cursor))
}
fn enrich_descriptors(
query: &DiscoveryQuery,
sessions: &mut [SessionDescriptor],
codex_history: Option<&CodexHistoryTopicIndex>,
) -> Result<()> {
let codex_topics = if query.include_topic_candidates && codex_history.is_none() {
codex_history_topics(&query.homes.codex, sessions).unwrap_or_default()
} else {
HashMap::new()
};
for descriptor in sessions {
if query.include_topic_candidates {
descriptor.preview_candidates =
if descriptor.locator.harness.as_str() == HarnessId::CODEX {
codex_history
.and_then(|history| history.topics.get(&descriptor.locator.session_id))
.or_else(|| codex_topics.get(&descriptor.locator.session_id))
.cloned()
.unwrap_or_else(|| {
topic_message_candidates(&descriptor.locator).unwrap_or_default()
})
} else {
topic_message_candidates(&descriptor.locator).unwrap_or_default()
};
}
descriptor.latest_message_candidates =
latest_message_candidates(&descriptor.locator).unwrap_or_default();
}
Ok(())
}
fn descriptor_matches(descriptor: &SessionDescriptor, search: &str) -> bool {
[
Some(descriptor.locator.harness.as_str()),
Some(descriptor.locator.session_id.as_str()),
descriptor.title.as_deref(),
descriptor.cwd.as_ref().and_then(|path| path.to_str()),
descriptor.model.as_deref(),
]
.into_iter()
.flatten()
.any(|value| value.to_lowercase().contains(search))
}
fn descriptor_cursor_key(descriptor: &SessionDescriptor) -> (Option<u64>, String, String) {
(
descriptor.updated_at_ms,
descriptor.locator.harness.as_str().to_string(),
descriptor.locator.session_id.clone(),
)
}
fn encode_cursor(descriptor: &SessionDescriptor) -> String {
let json = serde_json::to_vec(&descriptor_cursor_key(descriptor)).unwrap_or_default();
let mut encoded = String::with_capacity(json.len() * 2);
for byte in json {
use std::fmt::Write;
let _ = write!(&mut encoded, "{byte:02x}");
}
encoded
}
fn decode_cursor(cursor: &str) -> Result<(Option<u64>, String, String)> {
if cursor.len() % 2 != 0 {
return Err(Error::Other("discovery cursor is invalid".into()));
}
let bytes = (0..cursor.len())
.step_by(2)
.map(|index| u8::from_str_radix(&cursor[index..index + 2], 16))
.collect::<std::result::Result<Vec<_>, _>>()
.map_err(|_| Error::Other("discovery cursor is invalid".into()))?;
serde_json::from_slice(&bytes).map_err(|_| Error::Other("discovery cursor is invalid".into()))
}
#[derive(Default)]
struct HeaderMeta {
session_id: Option<String>,
cwd: Option<PathBuf>,
title: Option<String>,
model: Option<String>,
parent_session_id: Option<String>,
}
fn discover_jsonl(
root: &Path,
harness: &str,
workspace: Option<&Path>,
include_child_sessions: bool,
found: &mut Vec<SessionDescriptor>,
) {
let mut files = Vec::new();
collect_jsonl(root, harness, include_child_sessions, &mut files);
for path in files {
let Ok(meta) = read_header(&path, harness) else {
continue;
};
if workspace.is_some_and(|wanted| {
meta.cwd
.as_deref()
.is_none_or(|cwd| !recorded_cwd_matches(cwd, wanted))
}) {
continue;
}
let session_id = meta.session_id.unwrap_or_else(|| {
path.file_stem()
.and_then(|value| value.to_str())
.unwrap_or("unknown")
.to_string()
});
let parent_session_id = meta.parent_session_id.or_else(|| {
(harness == HarnessId::CLAUDE_CODE)
.then(|| claude_subagent_parent_id(&path))
.flatten()
});
found.push(SessionDescriptor {
locator: SessionLocator {
harness: HarnessId::new(harness),
session_id,
storage: StorageLocator::File { path: path.clone() },
},
cwd: meta.cwd,
title: meta.title,
preview_candidates: Vec::new(),
latest_message_candidates: Vec::new(),
updated_at_ms: modified_ms(&path),
message_count: None,
model: meta.model,
parent_session_id,
child_session_count: if harness == HarnessId::CLAUDE_CODE && !include_child_sessions {
count_claude_subagents(&path)
} else {
0
},
nouns: OrchestrationNouns::default(),
});
}
}
fn collect_jsonl(root: &Path, harness: &str, include_child_sessions: bool, out: &mut Vec<PathBuf>) {
let Ok(entries) = fs::read_dir(root) else {
return;
};
for entry in entries.flatten() {
let Ok(kind) = entry.file_type() else {
continue;
};
let path = entry.path();
if kind.is_dir() {
if harness == HarnessId::CLAUDE_CODE
&& path.file_name().and_then(|v| v.to_str()) == Some("subagents")
&& !include_child_sessions
{
continue;
}
collect_jsonl(&path, harness, include_child_sessions, out);
} else if kind.is_file() && path.extension().and_then(|v| v.to_str()) == Some("jsonl") {
out.push(path);
}
}
}
fn read_header(path: &Path, harness: &str) -> Result<HeaderMeta> {
let file = File::open(path)?;
let mut result = HeaderMeta::default();
let mut bytes = 0usize;
for line in BufReader::new(file).lines().take(32) {
let line = line?;
bytes += line.len();
if bytes > 256 * 1024 {
break;
}
let Ok(value) = serde_json::from_str::<Value>(&line) else {
continue;
};
update_header_meta(&mut result, &value, harness);
if result.session_id.is_some() && result.cwd.is_some() && result.model.is_some() {
break;
}
}
if result.session_id.is_none() && result.cwd.is_none() {
return Err(Error::Other(format!(
"{} has no recognizable {harness} session header",
path.display()
)));
}
Ok(result)
}
fn update_header_meta(result: &mut HeaderMeta, value: &Value, harness: &str) {
match harness {
HarnessId::CLAUDE_CODE => {
fill_string(&mut result.session_id, value.get("sessionId"));
fill_path(&mut result.cwd, value.get("cwd"));
fill_string(
&mut result.model,
value.get("message").and_then(|v| v.get("model")),
);
}
HarnessId::CODEX => {
let payload = value.get("payload").unwrap_or(&Value::Null);
if value.get("type").and_then(Value::as_str) == Some("session_meta") {
fill_string(&mut result.session_id, payload.get("id"));
fill_path(&mut result.cwd, payload.get("cwd"));
fill_string(&mut result.title, payload.get("thread_name"));
fill_string(&mut result.title, payload.get("title"));
fill_string(
&mut result.parent_session_id,
payload.get("parent_thread_id"),
);
if let Some(parent) = payload
.pointer("/source/subagent/thread_spawn/parent_thread_id")
.and_then(Value::as_str)
{
result.parent_session_id = Some(parent.to_string());
}
if result.title.is_none() {
result.title = payload
.pointer("/source/subagent/thread_spawn/agent_path")
.and_then(Value::as_str)
.and_then(|path| path.rsplit('/').find(|part| !part.is_empty()))
.map(humanize_topic);
}
}
if value.get("type").and_then(Value::as_str) == Some("turn_context") {
fill_path(&mut result.cwd, payload.get("cwd"));
fill_string(&mut result.model, payload.get("model"));
}
}
HarnessId::PI => {
if value.get("type").and_then(Value::as_str) == Some("session") {
fill_string(&mut result.session_id, value.get("id"));
fill_path(&mut result.cwd, value.get("cwd"));
}
fill_string(
&mut result.model,
value.get("message").and_then(|v| v.get("model")),
);
}
_ => {}
}
}
fn roll_up_session_children(found: &mut Vec<SessionDescriptor>, include_children: bool) {
let by_id = found
.iter()
.enumerate()
.map(|(index, descriptor)| {
(
(
descriptor.locator.harness.as_str().to_string(),
descriptor.locator.session_id.clone(),
),
index,
)
})
.collect::<HashMap<_, _>>();
let mut root_updates = HashMap::<usize, u64>::new();
let mut root_child_counts = HashMap::<usize, usize>::new();
for descriptor in found.iter() {
let Some(mut parent_id) = descriptor.parent_session_id.as_deref() else {
continue;
};
let harness = descriptor.locator.harness.as_str();
let mut root = None;
let mut visited = HashSet::new();
while visited.insert(parent_id.to_string()) {
let Some(&parent_index) = by_id.get(&(harness.to_string(), parent_id.to_string()))
else {
break;
};
root = Some(parent_index);
let Some(next_parent) = found[parent_index].parent_session_id.as_deref() else {
break;
};
parent_id = next_parent;
}
if let (Some(root), Some(updated_at_ms)) = (root, descriptor.updated_at_ms) {
root_updates
.entry(root)
.and_modify(|current| *current = (*current).max(updated_at_ms))
.or_insert(updated_at_ms);
}
if let Some(root) = root {
*root_child_counts.entry(root).or_default() += 1;
}
}
for (root, child_updated_at_ms) in root_updates {
found[root].updated_at_ms = Some(
found[root]
.updated_at_ms
.unwrap_or_default()
.max(child_updated_at_ms),
);
}
for (root, child_count) in root_child_counts {
found[root].child_session_count = child_count;
}
if !include_children {
found.retain(|descriptor| descriptor.parent_session_id.is_none());
}
}
fn retain_session_family(found: &mut Vec<SessionDescriptor>, root_session_id: &str) {
let parent_by_id = found
.iter()
.map(|descriptor| {
(
descriptor.locator.session_id.clone(),
descriptor.parent_session_id.clone(),
)
})
.collect::<HashMap<_, _>>();
found.retain(|descriptor| {
let mut current = descriptor.locator.session_id.clone();
let mut visited = HashSet::new();
while visited.insert(current.clone()) {
if current == root_session_id {
return true;
}
let Some(Some(parent)) = parent_by_id.get(¤t) else {
return false;
};
current = parent.clone();
}
false
});
}
fn claude_subagent_parent_id(path: &Path) -> Option<String> {
let subagents = path.parent()?;
if subagents.file_name()?.to_str()? != "subagents" {
return None;
}
subagents
.parent()?
.file_name()?
.to_str()
.map(str::to_string)
}
fn count_claude_subagents(parent_path: &Path) -> usize {
let Some(parent) = parent_path.parent() else {
return 0;
};
let Some(stem) = parent_path.file_stem() else {
return 0;
};
let root = parent.join(stem).join("subagents");
let mut files = Vec::new();
collect_jsonl(&root, HarnessId::CLAUDE_CODE, true, &mut files);
files.len()
}
fn is_zero(value: &usize) -> bool {
*value == 0
}
fn humanize_topic(value: &str) -> String {
let text = value.replace(['_', '-'], " ");
let mut characters = text.chars();
match characters.next() {
Some(first) => first.to_uppercase().collect::<String>() + characters.as_str(),
None => text,
}
}
fn discover_gemini(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
let slug_to_cwd = std::fs::read_to_string(root.join("projects.json"))
.ok()
.and_then(|text| serde_json::from_str::<Value>(&text).ok())
.and_then(|value| value.get("projects").and_then(Value::as_object).cloned())
.map(|projects| {
projects
.into_iter()
.filter_map(|(cwd, slug)| Some((slug.as_str()?.to_string(), PathBuf::from(cwd))))
.collect::<HashMap<_, _>>()
})
.unwrap_or_default();
let mut files = Vec::new();
collect_jsonl(&root.join("tmp"), HarnessId::GEMINI, false, &mut files);
let worker_count = std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(4)
.clamp(1, 8)
.min(files.len().max(1));
let chunk_size = files.len().max(1).div_ceil(worker_count);
let discovered = std::thread::scope(|scope| {
files
.chunks(chunk_size)
.map(|paths| {
scope.spawn(|| {
paths
.iter()
.filter_map(|path| gemini_descriptor(path, &slug_to_cwd, workspace))
.collect::<Vec<_>>()
})
})
.collect::<Vec<_>>()
.into_iter()
.flat_map(|worker| {
worker
.join()
.expect("Gemini discovery worker must not panic")
})
.collect::<Vec<_>>()
});
found.extend(discovered);
}
fn gemini_descriptor(
path: &Path,
slug_to_cwd: &HashMap<String, PathBuf>,
workspace: Option<&Path>,
) -> Option<SessionDescriptor> {
if path
.parent()
.and_then(Path::file_name)
.and_then(|name| name.to_str())
!= Some("chats")
{
return None;
}
let slug = path
.parent()
.and_then(Path::parent)
.and_then(Path::file_name)
.and_then(|name| name.to_str());
let cwd = slug.and_then(|slug| slug_to_cwd.get(slug)).cloned();
if workspace.is_some_and(|wanted| {
cwd.as_deref()
.is_none_or(|actual| !recorded_cwd_matches(actual, wanted))
}) {
return None;
}
let file = File::open(path).ok()?;
let mut reader = BufReader::new(file.take(64 * 1024));
let mut header = String::new();
reader.read_line(&mut header).ok()?;
let header = serde_json::from_str::<Value>(&header).ok()?;
let session_id = header.get("sessionId")?.as_str()?.to_string();
let mut model = None;
for line in reader
.take(4 * 1024)
.lines()
.map_while(std::result::Result::ok)
{
let Ok(value) = serde_json::from_str::<Value>(&line) else {
continue;
};
let kind = value.get("type").and_then(Value::as_str);
if kind != Some("user") && kind != Some("gemini") {
continue;
}
if model.is_none() {
model = value
.get("model")
.and_then(Value::as_str)
.map(str::to_string);
}
if model.is_some() {
break;
}
}
Some(SessionDescriptor {
locator: SessionLocator {
harness: HarnessId::from(HarnessId::GEMINI),
session_id,
storage: StorageLocator::File {
path: path.to_path_buf(),
},
},
cwd,
title: None,
preview_candidates: Vec::new(),
latest_message_candidates: Vec::new(),
updated_at_ms: modified_ms(path),
message_count: None,
model,
parent_session_id: None,
child_session_count: 0,
nouns: OrchestrationNouns::default(),
})
}
fn display_text(content: Option<&Value>) -> Option<String> {
match content? {
Value::String(text) => Some(text.clone()),
Value::Array(parts) => Some(
parts
.iter()
.filter_map(|part| part.get("text").and_then(Value::as_str))
.collect::<Vec<_>>()
.join(" ")
.trim()
.to_string(),
),
_ => None,
}
}
fn discover_hermes(db_path: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
if !db_path.is_file() {
return;
}
let Ok(conn) = Connection::open_with_flags(
db_path,
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
) else {
return;
};
let fingerprint_ok = ["sessions", "messages", "schema_version"].iter().all(|t| {
conn.query_row(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1",
[t],
|_| Ok(()),
)
.is_ok()
});
if !fingerprint_ok {
return;
}
let Ok(mut statement) = conn.prepare(
"SELECT id, cwd, title, model, message_count, started_at, ended_at, parent_session_id, \
source, model_config FROM sessions ORDER BY started_at DESC",
) else {
return;
};
let Ok(rows) = statement.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Option<String>>(1)?,
row.get::<_, Option<String>>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<i64>>(4)?,
row.get::<_, Option<f64>>(5)?,
row.get::<_, Option<f64>>(6)?,
row.get::<_, Option<String>>(7)?,
row.get::<_, Option<String>>(8)?,
row.get::<_, Option<String>>(9)?,
))
}) else {
return;
};
for row in rows.flatten() {
let (
id,
cwd,
title,
model,
message_count,
started_at,
ended_at,
parent,
source,
model_config,
) = row;
let cwd = cwd.map(PathBuf::from);
if let Some(filter) = workspace {
if cwd.as_deref() != Some(filter) {
continue;
}
}
let updated_at_ms = ended_at
.or(started_at)
.map(|seconds| (seconds * 1000.0) as u64);
let mut meta = SessionMeta::new(SessionSource::Hermes);
meta.cwd = cwd.clone();
if let Some(hermes_source) = source.filter(|value| !value.is_empty()) {
meta.lineage
.insert("hermes_source".to_string(), hermes_source);
}
if let Some(parent_id) = parent.as_deref() {
meta.lineage.insert(
"hermes_lineage_kind".to_string(),
crate::session::hermes_lineage_kind(
&conn,
parent_id,
model_config.as_deref(),
started_at,
)
.to_string(),
);
}
hermes_capture_nouns(&conn, &id, &mut meta);
found.push(SessionDescriptor {
locator: SessionLocator {
harness: HarnessId::new(HarnessId::HERMES),
session_id: id,
storage: StorageLocator::File {
path: db_path.to_path_buf(),
},
},
cwd,
title: title.filter(|t| !t.is_empty()),
preview_candidates: Vec::new(),
latest_message_candidates: Vec::new(),
updated_at_ms,
message_count: message_count.map(|count| count.max(0) as usize),
model,
parent_session_id: parent,
child_session_count: 0,
nouns: OrchestrationNouns::from_meta(&meta),
});
}
}
fn discover_orchestrator(
root: &Path,
workspace: Option<&Path>,
found: &mut Vec<SessionDescriptor>,
) {
if workspace.is_some() {
return;
}
for (profile, dir) in orchestrator_profile_dirs(root) {
let db_path = dir.join("state.db");
if !db_path.is_file() {
continue;
}
let Ok(conn) = Connection::open_with_flags(
&db_path,
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
) else {
continue;
};
let Ok(mut statement) = conn.prepare(
"SELECT platform, chat_type, chat_id, thread_id, participant_id, worker_harness, \
worker_session_id, worker_locator, started_at, last_activity_at, ended_at, \
end_reason, handoff_to, handoff_state, handoff_error, recurrence_job_id, \
CAST(strftime('%s', last_activity_at) AS INTEGER) \
FROM bindings ORDER BY last_activity_at DESC, started_at DESC",
) else {
continue;
};
let Ok(rows) = statement.query_map([], |row| {
let text = |index: usize| -> rusqlite::Result<Option<String>> {
Ok(row
.get::<_, Option<String>>(index)?
.filter(|value| !value.is_empty()))
};
Ok((
OrchestratorBindingRow {
platform: text(0)?.unwrap_or_default(),
chat_type: text(1)?.unwrap_or_default(),
chat_id: text(2)?,
thread_id: text(3)?,
participant_id: text(4)?,
worker_harness: text(5)?.unwrap_or_default(),
worker_session_id: text(6)?.filter(|s| !s.is_empty()),
worker_locator: text(7)?,
started_at: text(8)?,
last_activity_at: text(9)?,
ended_at: text(10)?,
end_reason: text(11)?,
handoff_to: text(12)?,
handoff_state: text(13)?,
handoff_error: text(14)?,
recurrence_job_id: text(15)?,
},
row.get::<_, Option<i64>>(16)?,
))
}) else {
continue;
};
for (row, last_activity_epoch) in rows.flatten() {
found.push(orchestrator_descriptor(
&db_path,
&profile,
&row,
last_activity_epoch,
));
}
}
}
fn orchestrator_descriptor(
db_path: &Path,
profile: &str,
row: &OrchestratorBindingRow,
last_activity_epoch: Option<i64>,
) -> SessionDescriptor {
let binding = Binding::from_orchestrator_row(profile, row);
let nouns = binding.nouns();
let mut title = format!(
"{} {}",
row.worker_harness,
row.worker_session_id
.as_deref()
.unwrap_or("(no worker session yet)")
);
if let Some(reason) = row.end_reason.as_deref().filter(|_| row.ended_at.is_some()) {
title.push_str(&format!(" (ended: {reason})"));
}
SessionDescriptor {
locator: SessionLocator {
harness: HarnessId::new(HarnessId::ORCHESTRATOR),
session_id: row.worker_session_id.clone().unwrap_or_default(),
storage: StorageLocator::File {
path: row
.worker_locator
.clone()
.map_or_else(|| db_path.to_path_buf(), PathBuf::from),
},
},
cwd: None,
title: Some(title),
preview_candidates: Vec::new(),
latest_message_candidates: Vec::new(),
updated_at_ms: last_activity_epoch.map(|seconds| (seconds.max(0) as u64) * 1000),
message_count: None,
model: None,
parent_session_id: None,
child_session_count: 0,
nouns,
}
}
fn discover_openclaw(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
let agents = root.join("agents");
let Ok(agent_dirs) = std::fs::read_dir(&agents) else {
return;
};
for agent_dir in agent_dirs.flatten() {
let sessions = agent_dir.path().join("sessions");
let Ok(files) = std::fs::read_dir(&sessions) else {
continue;
};
for file in files.flatten() {
let path = file.path();
let name = file.file_name();
let name = name.to_string_lossy();
if !name.ends_with(".jsonl") || name.ends_with(".trajectory.jsonl") {
continue;
}
let Ok(text) = std::fs::read_to_string(&path) else {
continue;
};
let Some(header_line) = text.lines().find(|line| !line.trim().is_empty()) else {
continue;
};
let Ok(header) = serde_json::from_str::<serde_json::Value>(header_line) else {
continue;
};
if header.get("type").and_then(serde_json::Value::as_str) != Some("session") {
continue;
}
let session_id = header
.get("id")
.and_then(serde_json::Value::as_str)
.unwrap_or_else(|| name.trim_end_matches(".jsonl"))
.to_string();
let cwd = header
.get("cwd")
.and_then(serde_json::Value::as_str)
.map(PathBuf::from);
if let Some(filter) = workspace {
if cwd.as_deref() != Some(filter) {
continue;
}
}
let updated_at_ms = file
.metadata()
.ok()
.and_then(|metadata| metadata.modified().ok())
.and_then(|modified| modified.duration_since(std::time::UNIX_EPOCH).ok())
.map(|elapsed| elapsed.as_millis() as u64);
let message_count = text
.lines()
.filter(|line| line.contains("\"type\":\"message\""))
.count();
let mut meta = SessionMeta::new(SessionSource::OpenClaw);
meta.cwd = cwd.clone();
openclaw_capture_header_nouns(&header, &mut meta);
if meta.profile.is_none() {
meta.profile = openclaw_agent_id_from_path(&path);
}
found.push(SessionDescriptor {
locator: SessionLocator {
harness: HarnessId::new(HarnessId::OPENCLAW),
session_id,
storage: StorageLocator::File { path },
},
cwd,
title: None,
preview_candidates: Vec::new(),
latest_message_candidates: Vec::new(),
updated_at_ms,
message_count: Some(message_count),
model: None,
parent_session_id: None,
child_session_count: 0,
nouns: OrchestrationNouns::from_meta(&meta),
});
}
}
}
fn discover_supercode(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
for info in list_native_store(root) {
let path = if info.archived {
root.join("archived").join(format!("{}.jsonl", info.name))
} else {
root.join(format!("{}.jsonl", info.name))
};
let header = read_native_store_header(&path);
if workspace.is_some_and(|wanted| {
header
.as_ref()
.and_then(|meta| meta.cwd.as_deref())
.is_none_or(|cwd| !recorded_cwd_matches(cwd, wanted))
}) {
continue;
}
let title = (!info.title.trim().is_empty()).then_some(info.title);
let updated_at_ms =
modified_ms(&path).or_else(|| modified_ms(&path.with_extension("sidecar.jsonl")));
found.push(SessionDescriptor {
locator: SessionLocator {
harness: HarnessId::from(HarnessId::SUPERCODE),
session_id: info.name,
storage: StorageLocator::File { path: path.clone() },
},
cwd: header.as_ref().and_then(|meta| meta.cwd.clone()),
title,
preview_candidates: Vec::new(),
latest_message_candidates: Vec::new(),
updated_at_ms,
message_count: None,
model: header.and_then(|meta| meta.model),
parent_session_id: None,
child_session_count: 0,
nouns: OrchestrationNouns::default(),
});
}
}
fn read_native_store_header(path: &Path) -> Option<HeaderMeta> {
let name = path.file_stem()?.to_str()?;
let sidecar = path.with_file_name(format!("{name}.sidecar.jsonl"));
let source_path = if sidecar.is_file() {
sidecar
} else {
path.to_path_buf()
};
let file = File::open(source_path).ok()?;
let mut result = HeaderMeta::default();
let mut source = None;
let mut bytes = 0usize;
for line in BufReader::new(file).lines().take(32) {
let line = line.ok()?;
bytes += line.len();
if bytes > 256 * 1024 {
break;
}
let Ok(value) = serde_json::from_str::<Value>(&line) else {
continue;
};
if source.is_none() {
source = value.get("source").and_then(Value::as_str).map(|source| {
if source == "claude_code" {
HarnessId::CLAUDE_CODE.to_string()
} else {
source.to_string()
}
});
fill_string(&mut result.session_id, value.get("session_id"));
}
if let Some(harness) = source.as_deref() {
update_header_meta(&mut result, &value, harness);
}
if result.cwd.is_some() && result.model.is_some() {
break;
}
}
Some(result)
}
#[derive(Deserialize)]
struct NativeStoreInfo {
name: String,
#[serde(default)]
title: String,
#[serde(skip)]
archived: bool,
}
fn list_native_store(root: &Path) -> Vec<NativeStoreInfo> {
let mut sessions = Vec::new();
for archived in [false, true] {
let directory = if archived {
root.join("archived")
} else {
root.to_path_buf()
};
let Ok(entries) = fs::read_dir(directory) else {
continue;
};
for entry in entries.flatten() {
let path = entry.path();
if !path.to_string_lossy().ends_with(".meta.json") {
continue;
}
let Ok(text) = fs::read_to_string(path) else {
continue;
};
let Ok(mut info) = serde_json::from_str::<NativeStoreInfo>(&text) else {
continue;
};
info.archived = archived;
sessions.push(info);
}
}
sessions.sort_by(|left, right| left.name.cmp(&right.name));
sessions
}
fn discover_grok(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
let Ok(workspaces) = fs::read_dir(root) else {
return;
};
for workspace_entry in workspaces.flatten() {
let encoded = workspace_entry.file_name();
let Some(cwd) = encoded
.to_str()
.and_then(percent_decode_path)
.map(PathBuf::from)
else {
continue;
};
if workspace.is_some_and(|wanted| !recorded_cwd_matches(&cwd, wanted)) {
continue;
}
let Ok(sessions) = fs::read_dir(workspace_entry.path()) else {
continue;
};
for session_entry in sessions.flatten() {
let session_dir = session_entry.path();
if !session_dir.is_dir() {
continue;
}
let transcript = session_dir.join("chat_history.jsonl");
if !transcript.is_file() {
continue;
}
let Some(session_id) = session_dir
.file_name()
.and_then(|name| name.to_str())
.map(str::to_string)
else {
continue;
};
let summary = fs::read_to_string(session_dir.join("summary.json"))
.ok()
.and_then(|text| serde_json::from_str::<Value>(&text).ok());
let title = summary
.as_ref()
.and_then(|value| value.get("generated_title"))
.and_then(Value::as_str)
.filter(|title| !title.is_empty())
.map(str::to_string);
let model = summary
.as_ref()
.and_then(|value| value.get("current_model_id"))
.and_then(Value::as_str)
.map(str::to_string);
let message_count = summary
.as_ref()
.and_then(|value| value.get("num_chat_messages"))
.and_then(Value::as_u64)
.and_then(|count| usize::try_from(count).ok());
let updated_at_ms = summary
.as_ref()
.and_then(|value| value.get("updated_at"))
.and_then(Value::as_str)
.and_then(crate::sidecar::rfc3339_to_ms)
.and_then(|millis| u64::try_from(millis).ok())
.or_else(|| modified_ms(&transcript));
found.push(SessionDescriptor {
locator: SessionLocator {
harness: HarnessId::from(HarnessId::GROK),
session_id,
storage: StorageLocator::File { path: transcript },
},
cwd: Some(cwd.clone()),
title,
preview_candidates: Vec::new(),
latest_message_candidates: Vec::new(),
updated_at_ms,
message_count,
model,
parent_session_id: None,
child_session_count: 0,
nouns: OrchestrationNouns::default(),
});
}
}
}
fn discover_opencode(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
let mut dbs = Vec::new();
if root.is_file() {
dbs.push(root.to_path_buf());
} else if let Ok(entries) = fs::read_dir(root) {
dbs.extend(entries.flatten().map(|entry| entry.path()).filter(|path| {
path.file_name()
.and_then(|v| v.to_str())
.is_some_and(|name| name.starts_with("opencode") && name.ends_with(".db"))
}));
}
dbs.sort();
for db in dbs {
let Ok(conn) = Connection::open_with_flags(
&db,
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
) else {
continue;
};
let has_model = conn.prepare("SELECT model FROM session LIMIT 0").is_ok();
let model_column = if has_model { "s.model" } else { "NULL" };
let query = format!(
"SELECT s.id, s.directory, s.title, s.time_updated, {model_column}, COUNT(m.id) \
FROM session s LEFT JOIN message m ON m.session_id = s.id \
GROUP BY s.id ORDER BY s.time_updated DESC"
);
let Ok(mut stmt) = conn.prepare(&query) else {
continue;
};
let Ok(rows) = stmt.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, i64>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, i64>(5)?,
))
}) else {
continue;
};
for row in rows.flatten() {
let (id, cwd, title, updated, model, messages) = row;
let cwd = PathBuf::from(cwd);
if workspace.is_some_and(|wanted| !recorded_cwd_matches(&cwd, wanted)) {
continue;
}
found.push(SessionDescriptor {
locator: SessionLocator {
harness: HarnessId::from(HarnessId::OPENCODE),
session_id: id.clone(),
storage: StorageLocator::Sqlite {
path: db.clone(),
selector: id,
},
},
cwd: Some(cwd),
title: (!title.is_empty()).then_some(title),
preview_candidates: Vec::new(),
latest_message_candidates: Vec::new(),
updated_at_ms: u64::try_from(updated).ok(),
message_count: usize::try_from(messages).ok(),
model,
parent_session_id: None,
child_session_count: 0,
nouns: OrchestrationNouns::default(),
});
}
}
}
fn discover_goose(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
let db = if root.is_file() {
root.to_path_buf()
} else if root.join("sessions.db").is_file() {
root.join("sessions.db")
} else {
root.join("sessions/sessions.db")
};
let Ok(connection) = Connection::open_with_flags(
&db,
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
) else {
return;
};
let Ok(mut statement) = connection.prepare(
"SELECT s.id, s.working_dir, s.name, s.updated_at, s.model_config_json, \
COUNT(m.id) \
FROM sessions s LEFT JOIN messages m ON m.session_id = s.id \
WHERE s.archived_at IS NULL \
GROUP BY s.id ORDER BY s.updated_at DESC",
) else {
return;
};
let Ok(rows) = statement.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, i64>(5)?,
))
}) else {
return;
};
for row in rows.flatten() {
let (id, cwd, title, updated_at, model_config, message_count) = row;
let cwd = PathBuf::from(cwd);
if workspace.is_some_and(|wanted| !recorded_cwd_matches(&cwd, wanted)) {
continue;
}
let model = model_config
.as_deref()
.and_then(|value| serde_json::from_str::<Value>(value).ok())
.and_then(|value| {
value
.get("model_name")
.or_else(|| value.get("modelName"))
.and_then(Value::as_str)
.map(str::to_string)
});
let updated_at_ms = crate::sidecar::rfc3339_to_ms(&updated_at)
.or_else(|| {
crate::sidecar::rfc3339_to_ms(&format!("{}Z", updated_at.replace(' ', "T")))
})
.and_then(|value| u64::try_from(value).ok());
found.push(SessionDescriptor {
locator: SessionLocator {
harness: HarnessId::from(HarnessId::GOOSE),
session_id: id.clone(),
storage: StorageLocator::Sqlite {
path: db.clone(),
selector: id,
},
},
cwd: Some(cwd),
title: (!title.trim().is_empty()).then_some(title),
preview_candidates: Vec::new(),
latest_message_candidates: Vec::new(),
updated_at_ms,
message_count: usize::try_from(message_count).ok(),
model,
parent_session_id: None,
child_session_count: 0,
nouns: OrchestrationNouns::default(),
});
}
}
const LATEST_PREVIEW_CANDIDATES: usize = 8;
const TOPIC_PREVIEW_HEAD_BYTES: u64 = 512 * 1024;
const LATEST_PREVIEW_TAIL_BYTES: u64 = 512 * 1024;
const LATEST_PREVIEW_MAX_BYTES: u64 = 4 * 1024 * 1024;
fn topic_message_candidates(locator: &SessionLocator) -> Result<Vec<SessionPreviewCandidate>> {
match &locator.storage {
StorageLocator::File { path }
if matches!(
locator.harness.as_str(),
HarnessId::CLAUDE_CODE | HarnessId::CODEX
) =>
{
topic_file_message_candidates(path, locator.harness.as_str())
}
_ => Ok(Vec::new()),
}
}
fn codex_history_topics(
sessions_root: &Path,
sessions: &[SessionDescriptor],
) -> Result<HashMap<String, Vec<SessionPreviewCandidate>>> {
let wanted: HashSet<&str> = sessions
.iter()
.filter(|descriptor| descriptor.locator.harness.as_str() == HarnessId::CODEX)
.map(|descriptor| descriptor.locator.session_id.as_str())
.collect();
if wanted.is_empty() {
return Ok(HashMap::new());
}
let Some(root) = sessions_root.parent() else {
return Ok(HashMap::new());
};
let file = match File::open(root.join("history.jsonl")) {
Ok(file) => file,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(HashMap::new()),
Err(error) => return Err(error.into()),
};
let mut topics = HashMap::new();
for line in BufReader::new(file).lines() {
let Ok(value) = serde_json::from_str::<Value>(&line?) else {
continue;
};
let Some(session_id) = value.get("session_id").and_then(Value::as_str) else {
continue;
};
if !wanted.contains(session_id) || topics.contains_key(session_id) {
continue;
}
let mut candidates = Vec::new();
push_message_candidate(&mut candidates, "user", value.get("text"), HashMap::new());
if !candidates.is_empty() {
topics.insert(session_id.to_string(), candidates);
if topics.len() == wanted.len() {
break;
}
}
}
Ok(topics)
}
fn latest_message_candidates(locator: &SessionLocator) -> Result<Vec<SessionPreviewCandidate>> {
match &locator.storage {
StorageLocator::File { path } | StorageLocator::Sqlite { path, .. }
if locator.harness.as_str() == HarnessId::HERMES =>
{
latest_hermes_message_candidates(path, &locator.session_id)
}
StorageLocator::File { path } => {
latest_file_message_candidates(path, locator.harness.as_str())
}
StorageLocator::Sqlite { path, selector }
if locator.harness.as_str() == HarnessId::OPENCODE =>
{
latest_opencode_message_candidates(path, selector)
}
StorageLocator::Sqlite { path, selector }
if locator.harness.as_str() == HarnessId::GOOSE =>
{
latest_goose_message_candidates(path, selector)
}
StorageLocator::Sqlite { .. } => Ok(Vec::new()),
}
}
fn topic_file_message_candidates(
path: &Path,
harness: &str,
) -> Result<Vec<SessionPreviewCandidate>> {
let mut file = File::open(path)?;
let mut bytes = Vec::with_capacity(TOPIC_PREVIEW_HEAD_BYTES as usize);
file.by_ref()
.take(TOPIC_PREVIEW_HEAD_BYTES)
.read_to_end(&mut bytes)?;
if file.metadata()?.len() > TOPIC_PREVIEW_HEAD_BYTES {
if let Some(newline) = bytes.iter().rposition(|byte| *byte == b'\n') {
bytes.truncate(newline);
}
}
let text = String::from_utf8(bytes).map_err(|_| {
Error::Other(format!(
"{} contains non-UTF-8 data in its topic-preview window",
path.display()
))
})?;
let mut candidates = Vec::new();
for line in text.lines() {
let Ok(value) = serde_json::from_str::<Value>(line) else {
continue;
};
push_topic_message_candidate(&mut candidates, harness, &value);
if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
break;
}
}
Ok(candidates)
}
fn latest_file_message_candidates(
path: &Path,
harness: &str,
) -> Result<Vec<SessionPreviewCandidate>> {
let mut candidates =
latest_file_message_candidates_with_limit(path, harness, LATEST_PREVIEW_TAIL_BYTES)?;
if candidates.is_empty() {
candidates =
latest_file_message_candidates_with_limit(path, harness, LATEST_PREVIEW_MAX_BYTES)?;
}
Ok(candidates)
}
fn latest_file_message_candidates_with_limit(
path: &Path,
harness: &str,
byte_limit: u64,
) -> Result<Vec<SessionPreviewCandidate>> {
let mut file = File::open(path)?;
let file_len = file.metadata()?.len();
let start = file_len.saturating_sub(byte_limit);
file.seek(SeekFrom::Start(start))?;
let mut bytes = Vec::with_capacity((file_len - start) as usize);
file.read_to_end(&mut bytes)?;
if start > 0 {
if let Some(newline) = bytes.iter().position(|byte| *byte == b'\n') {
bytes.drain(..=newline);
} else {
return Ok(Vec::new());
}
}
let text = String::from_utf8(bytes).map_err(|_| {
Error::Other(format!(
"{} contains non-UTF-8 data in its list-preview window",
path.display()
))
})?;
let mut candidates = Vec::new();
for line in text.lines().rev() {
let Ok(value) = serde_json::from_str::<Value>(line) else {
continue;
};
let (role, content, metadata) = match harness {
HarnessId::CLAUDE_CODE => {
let role = value.get("type").and_then(Value::as_str);
if !matches!(role, Some("user" | "assistant")) {
continue;
}
let metadata = if role == Some("user") {
crate::session::claude_user_provenance(&value)
.into_iter()
.collect()
} else {
HashMap::new()
};
(
role.unwrap_or_default(),
value
.get("message")
.and_then(|message| message.get("content")),
metadata,
)
}
HarnessId::CODEX => {
let payload = value.get("payload").unwrap_or(&Value::Null);
if value.get("type").and_then(Value::as_str) != Some("response_item")
|| payload.get("type").and_then(Value::as_str) != Some("message")
{
continue;
}
let Some(role @ ("user" | "assistant")) =
payload.get("role").and_then(Value::as_str)
else {
continue;
};
(role, payload.get("content"), HashMap::new())
}
HarnessId::PI => {
if value.get("type").and_then(Value::as_str) != Some("message") {
continue;
}
let message = value.get("message").unwrap_or(&Value::Null);
let Some(role @ ("user" | "assistant")) =
message.get("role").and_then(Value::as_str)
else {
continue;
};
(role, message.get("content"), HashMap::new())
}
HarnessId::GEMINI => {
let Some(kind @ ("user" | "gemini")) = value.get("type").and_then(Value::as_str)
else {
continue;
};
(
if kind == "gemini" {
"assistant"
} else {
"user"
},
value.get("content"),
HashMap::new(),
)
}
HarnessId::GROK => {
let Some(role @ ("user" | "assistant")) = value.get("type").and_then(Value::as_str)
else {
continue;
};
(role, value.get("content"), HashMap::new())
}
HarnessId::SUPERCODE => {
let Some(role @ ("user" | "assistant")) = value.get("role").and_then(Value::as_str)
else {
continue;
};
(role, value.get("content"), HashMap::new())
}
_ => continue,
};
let mut metadata = metadata;
if matches!(harness, HarnessId::CLAUDE_CODE | HarnessId::CODEX) {
if let Some(timestamp) = value.get("timestamp").and_then(Value::as_str) {
metadata.insert("timestamp".to_string(), timestamp.to_string());
}
}
push_message_candidate_with_cursor(
&mut candidates,
role,
content,
metadata,
Some(message_candidate_cursor(harness, &value)),
);
if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
break;
}
}
Ok(candidates)
}
fn push_topic_message_candidate(
candidates: &mut Vec<SessionPreviewCandidate>,
harness: &str,
value: &Value,
) {
let (role, content, metadata) = match harness {
HarnessId::CLAUDE_CODE => {
let role = value.get("type").and_then(Value::as_str);
if !matches!(role, Some("user" | "assistant")) {
return;
}
let metadata = if role == Some("user") {
crate::session::claude_user_provenance(value)
.into_iter()
.collect()
} else {
HashMap::new()
};
(
role.unwrap_or_default(),
value
.get("message")
.and_then(|message| message.get("content")),
metadata,
)
}
HarnessId::CODEX => {
let payload = value.get("payload").unwrap_or(&Value::Null);
if value.get("type").and_then(Value::as_str) != Some("response_item")
|| payload.get("type").and_then(Value::as_str) != Some("message")
{
return;
}
let Some(role @ ("user" | "assistant")) = payload.get("role").and_then(Value::as_str)
else {
return;
};
(role, payload.get("content"), HashMap::new())
}
_ => return,
};
push_message_candidate(candidates, role, content, metadata);
}
fn latest_opencode_message_candidates(
path: &Path,
session_id: &str,
) -> Result<Vec<SessionPreviewCandidate>> {
let connection = Connection::open_with_flags(
path,
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
)
.map_err(|error| Error::Other(format!("failed to open list-preview store: {error}")))?;
let mut statement = connection
.prepare(
"SELECT m.data, p.data FROM message m JOIN part p ON p.message_id = m.id \
WHERE m.session_id = ?1 ORDER BY m.time_created DESC, p.time_created DESC LIMIT 32",
)
.map_err(|error| Error::Other(format!("failed to prepare list-preview query: {error}")))?;
let rows = statement
.query_map([session_id], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.map_err(|error| Error::Other(format!("failed to read list-preview rows: {error}")))?;
let mut candidates = Vec::new();
for row in rows.flatten() {
let (Ok(message), Ok(part)) = (
serde_json::from_str::<Value>(&row.0),
serde_json::from_str::<Value>(&row.1),
) else {
continue;
};
let Some(role @ ("user" | "assistant")) = message.get("role").and_then(Value::as_str)
else {
continue;
};
if part.get("type").and_then(Value::as_str) != Some("text") {
continue;
}
push_message_candidate(&mut candidates, role, part.get("text"), HashMap::new());
if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
break;
}
}
Ok(candidates)
}
fn latest_hermes_message_candidates(
path: &Path,
session_id: &str,
) -> Result<Vec<SessionPreviewCandidate>> {
let connection = Connection::open_with_flags(
path,
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
)
.map_err(|error| Error::Other(format!("failed to open list-preview store: {error}")))?;
let mut statement = connection
.prepare(
"SELECT role, content FROM messages WHERE session_id = ?1 AND active = 1 \
AND role IN ('user', 'assistant') AND content IS NOT NULL AND content != '' \
ORDER BY timestamp DESC, id DESC LIMIT 32",
)
.map_err(|error| Error::Other(format!("failed to prepare list-preview query: {error}")))?;
let rows = statement
.query_map([session_id], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.map_err(|error| Error::Other(format!("failed to read list-preview rows: {error}")))?;
let mut candidates = Vec::new();
for (role, content) in rows.flatten() {
push_message_candidate(
&mut candidates,
&role,
Some(&Value::String(content)),
HashMap::new(),
);
if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
break;
}
}
Ok(candidates)
}
fn latest_goose_message_candidates(
path: &Path,
session_id: &str,
) -> Result<Vec<SessionPreviewCandidate>> {
let connection = Connection::open_with_flags(
path,
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
)
.map_err(|error| Error::Other(format!("failed to open list-preview store: {error}")))?;
let mut statement = connection
.prepare(
"SELECT role, content_json FROM messages WHERE session_id = ?1 \
ORDER BY created_timestamp DESC, id DESC LIMIT 16",
)
.map_err(|error| Error::Other(format!("failed to prepare list-preview query: {error}")))?;
let rows = statement
.query_map([session_id], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.map_err(|error| Error::Other(format!("failed to read list-preview rows: {error}")))?;
let mut candidates = Vec::new();
for row in rows.flatten() {
let (role, content) = row;
if !matches!(role.as_str(), "user" | "assistant") {
continue;
}
let Ok(content) = serde_json::from_str::<Value>(&content) else {
continue;
};
push_message_candidate(&mut candidates, &role, Some(&content), HashMap::new());
if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
break;
}
}
Ok(candidates)
}
fn push_message_candidate(
candidates: &mut Vec<SessionPreviewCandidate>,
role: &str,
content: Option<&Value>,
metadata: HashMap<String, String>,
) {
push_message_candidate_with_cursor(candidates, role, content, metadata, None);
}
fn push_message_candidate_with_cursor(
candidates: &mut Vec<SessionPreviewCandidate>,
role: &str,
content: Option<&Value>,
metadata: HashMap<String, String>,
cursor: Option<String>,
) {
if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
return;
}
let Some(text) = display_text(content).filter(|text| !text.trim().is_empty()) else {
return;
};
const MAX_CHARS: usize = 4_096;
candidates.push(SessionPreviewCandidate {
cursor,
role: role.to_string(),
content: text.chars().take(MAX_CHARS).collect(),
metadata,
});
}
fn message_candidate_cursor(harness: &str, value: &Value) -> String {
let native_identity = value
.get("uuid")
.or_else(|| value.get("id"))
.or_else(|| value.pointer("/message/id"))
.or_else(|| value.pointer("/payload/id"))
.and_then(Value::as_str)
.or_else(|| value.get("timestamp").and_then(Value::as_str));
let mut hasher = blake3::Hasher::new();
hasher.update(b"supercode.session-preview-cursor.v1\0");
hasher.update(harness.as_bytes());
hasher.update(b"\0");
if let Some(identity) = native_identity {
hasher.update(identity.as_bytes());
} else {
hasher.update(value.to_string().as_bytes());
}
format!("v1:{}", &hasher.finalize().to_hex()[..24])
}
fn fill_string(target: &mut Option<String>, value: Option<&Value>) {
if target.is_none() {
*target = value.and_then(Value::as_str).map(str::to_owned);
}
}
fn fill_path(target: &mut Option<PathBuf>, value: Option<&Value>) {
if target.is_none() {
*target = value.and_then(Value::as_str).map(PathBuf::from);
}
}
fn modified_ms(path: &Path) -> Option<u64> {
fs::metadata(path)
.ok()?
.modified()
.ok()?
.duration_since(UNIX_EPOCH)
.ok()
.and_then(|duration| u64::try_from(duration.as_millis()).ok())
}
fn recorded_cwd_matches(recorded: &Path, wanted: &Path) -> bool {
recorded.is_absolute() && same_path(recorded, wanted)
}
fn same_path(left: &Path, right: &Path) -> bool {
match (fs::canonicalize(left), fs::canonicalize(right)) {
(Ok(left), Ok(right)) => left == right,
_ => normalize_path(left) == normalize_path(right),
}
}
fn normalize_path(path: &Path) -> PathBuf {
let absolute = if path.is_absolute() {
path.to_path_buf()
} else {
std::env::current_dir()
.unwrap_or_else(|_| PathBuf::from("."))
.join(path)
};
let mut normalized = PathBuf::new();
for component in absolute.components() {
match component {
Component::CurDir => {}
Component::ParentDir => {
normalized.pop();
}
other => normalized.push(other.as_os_str()),
}
}
normalized
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Write;
use std::time::{SystemTime, UNIX_EPOCH};
fn temp_dir(label: &str) -> PathBuf {
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos();
let path = std::env::temp_dir().join(format!(
"supercode-catalog-{label}-{}-{nonce}",
std::process::id()
));
fs::create_dir_all(&path).unwrap();
path
}
#[test]
fn codex_history_index_reads_appends_and_repairs_replacements() {
let root = temp_dir("codex-history-index");
let sessions = root.join("sessions");
fs::create_dir_all(&sessions).unwrap();
let history = root.join("history.jsonl");
fs::write(
&history,
"{\"session_id\":\"alpha\",\"text\":\"first topic\"}\n",
)
.unwrap();
let mut index = CodexHistoryTopicIndex::new(&sessions);
assert_eq!(index.refresh().unwrap(), BTreeSet::from(["alpha".into()]));
assert_eq!(index.topics["alpha"][0].content, "first topic");
assert!(index.refresh().unwrap().is_empty());
let mut file = fs::OpenOptions::new().append(true).open(&history).unwrap();
write!(
file,
"{{\"session_id\":\"alpha\",\"text\":\"later topic\"}}\n\
{{\"session_id\":\"beta\",\"text\":\"second topic\"}}\n"
)
.unwrap();
file.flush().unwrap();
assert_eq!(index.refresh().unwrap(), BTreeSet::from(["beta".into()]));
assert_eq!(index.topics["alpha"][0].content, "first topic");
assert_eq!(index.topics["beta"][0].content, "second topic");
fs::write(
&history,
"{\"session_id\":\"gamma\",\"text\":\"replacement\"}\n",
)
.unwrap();
assert_eq!(
index.refresh().unwrap(),
BTreeSet::from(["alpha".into(), "beta".into(), "gamma".into()])
);
assert!(!index.topics.contains_key("alpha"));
assert_eq!(index.topics["gamma"][0].content, "replacement");
fs::remove_dir_all(root).ok();
}
#[test]
fn codex_history_index_retains_an_incomplete_appended_record() {
let root = temp_dir("codex-history-partial");
let sessions = root.join("sessions");
fs::create_dir_all(&sessions).unwrap();
let history = root.join("history.jsonl");
fs::write(&history, "{\"session_id\":\"partial\",\"text\":\"hel").unwrap();
let mut index = CodexHistoryTopicIndex::new(&sessions);
assert!(index.refresh().unwrap().is_empty());
let mut file = fs::OpenOptions::new().append(true).open(&history).unwrap();
writeln!(file, "lo\"}}").unwrap();
file.flush().unwrap();
assert_eq!(index.refresh().unwrap(), BTreeSet::from(["partial".into()]));
assert_eq!(index.topics["partial"][0].content, "hello");
fs::remove_dir_all(root).ok();
}
#[test]
fn cached_codex_history_enrichment_matches_stateless_discovery() {
let root = temp_dir("codex-history-parity");
let sessions = root.join("sessions");
let workspace = root.join("workspace");
fs::create_dir_all(&sessions).unwrap();
fs::create_dir_all(&workspace).unwrap();
fs::write(
sessions.join("rollout.jsonl"),
format!(
"{{\"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",
serde_json::to_string(&workspace.to_string_lossy()).unwrap(),
serde_json::to_string(&workspace.to_string_lossy()).unwrap(),
),
)
.unwrap();
fs::write(
root.join("history.jsonl"),
"{\"session_id\":\"alpha\",\"text\":\"history topic\"}\n",
)
.unwrap();
let query = DiscoveryQuery {
harnesses: vec![HarnessId::from(HarnessId::CODEX)],
homes: HarnessHomes {
codex: sessions.clone(),
..HarnessHomes::default()
},
include_topic_candidates: true,
..DiscoveryQuery::default()
};
let catalog = HarnessCatalog::new();
let projected = catalog
.project_index(&query, catalog.discover_raw_index(&query))
.unwrap();
let expected = catalog
.enrich_index_page(&query, projected.clone())
.unwrap();
let mut history = CodexHistoryTopicIndex::new(&sessions);
history.refresh().unwrap();
let actual = catalog
.enrich_index_page_with_codex_history(&query, projected, &history)
.unwrap();
assert_eq!(actual, expected);
assert_eq!(actual[0].preview_candidates[0].content, "history topic");
fs::remove_dir_all(root).ok();
}
#[test]
fn locator_json_round_trip_preserves_sqlite_selector() {
let locator = SessionLocator {
harness: HarnessId::from(HarnessId::OPENCODE),
session_id: "ses_123".into(),
storage: StorageLocator::Sqlite {
path: PathBuf::from("/tmp/opencode-dev.db"),
selector: "ses_123".into(),
},
};
let encoded = serde_json::to_string(&locator).unwrap();
assert_eq!(
serde_json::from_str::<SessionLocator>(&encoded).unwrap(),
locator
);
}
#[test]
fn discovers_filters_loads_and_follows_three_jsonl_harnesses() {
let root = temp_dir("jsonl");
let workspace = root.join("workspace");
let other = root.join("other");
fs::create_dir_all(&workspace).unwrap();
fs::create_dir_all(&other).unwrap();
let claude = root.join("claude");
let codex = root.join("codex");
let pi = root.join("pi");
fs::create_dir_all(&claude).unwrap();
fs::create_dir_all(&codex).unwrap();
fs::create_dir_all(&pi).unwrap();
fs::write(
claude.join("claude.jsonl"),
format!(
"{{\"type\":\"user\",\"sessionId\":\"cc-1\",\"cwd\":{},\"timestamp\":\"2026-01-01T00:00:01Z\",\"message\":{{\"role\":\"user\",\"content\":\"hi\"}}}}\n",
serde_json::to_string(&workspace.to_string_lossy()).unwrap()
),
)
.unwrap();
fs::write(
codex.join("rollout.jsonl"),
format!(
"{{\"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",
serde_json::to_string(&workspace.to_string_lossy()).unwrap()
),
)
.unwrap();
fs::write(
pi.join("pi.jsonl"),
format!(
"{{\"type\":\"session\",\"version\":3,\"id\":\"pi-1\",\"timestamp\":\"2026-01-01T00:00:00Z\",\"cwd\":{}}}\n{{\"type\":\"message\",\"message\":{{\"role\":\"user\",\"content\":\"inspect pi\"}}}}\n",
serde_json::to_string(&workspace.to_string_lossy()).unwrap()
),
)
.unwrap();
fs::write(
pi.join("unrelated.jsonl"),
format!(
"{{\"type\":\"session\",\"version\":3,\"id\":\"pi-2\",\"timestamp\":\"2026-01-01T00:00:00Z\",\"cwd\":{}}}\n",
serde_json::to_string(&other.to_string_lossy()).unwrap()
),
)
.unwrap();
fs::write(claude.join("partial.jsonl"), "{truncated").unwrap();
let query = DiscoveryQuery {
workspace: Some(workspace),
homes: HarnessHomes {
claude_code: claude,
codex,
pi,
opencode: root.join("missing-opencode"),
grok: root.join("missing-grok"),
gemini: root.join("missing-gemini"),
goose: root.join("missing-goose"),
supercode: root.join("missing-supercode"),
openclaw: root.join("missing-openclaw"),
hermes: root.join("missing-hermes"),
orchestrator: root.join("missing-orchestrator"),
},
..DiscoveryQuery::default()
};
let catalog = HarnessCatalog::new();
let found = catalog.discover(&query).unwrap();
assert_eq!(found.len(), 3);
assert_eq!(
found
.iter()
.map(|item| item.locator.harness.as_str())
.collect::<HashSet<_>>(),
HashSet::from([HarnessId::CLAUDE_CODE, HarnessId::CODEX, HarnessId::PI])
);
for descriptor in found {
assert!(descriptor.preview_candidates.is_empty());
assert_eq!(descriptor.latest_message_candidates.len(), 1);
assert_eq!(descriptor.latest_message_candidates[0].role, "user");
assert!(descriptor.latest_message_candidates[0].cursor.is_some());
if descriptor.locator.harness.as_str() == HarnessId::CLAUDE_CODE {
assert_eq!(
descriptor.latest_message_candidates[0]
.metadata
.get("timestamp")
.map(String::as_str),
Some("2026-01-01T00:00:01Z")
);
} else if descriptor.locator.harness.as_str() == HarnessId::CODEX {
assert_eq!(
descriptor.latest_message_candidates[0]
.metadata
.get("timestamp")
.map(String::as_str),
Some("2026-01-01T00:00:02Z")
);
}
let loaded = catalog.load(&descriptor.locator).unwrap();
assert_eq!(
loaded.meta.session_id.as_deref(),
Some(descriptor.locator.session_id.as_str())
);
let mut follower = catalog.follow(&descriptor.locator).unwrap();
assert!(matches!(
follower.poll().unwrap(),
Some(crate::SessionWatchEvent::SessionSnapshot { .. })
));
}
fs::remove_dir_all(root).ok();
}
#[test]
fn preview_cursor_tracks_native_boundary_not_growing_text() {
let first = serde_json::json!({
"timestamp": "2026-01-01T00:00:02Z",
"type": "response_item",
"payload": {"type": "message", "role": "assistant", "content": "partial"}
});
let grown = serde_json::json!({
"timestamp": "2026-01-01T00:00:02Z",
"type": "response_item",
"payload": {"type": "message", "role": "assistant", "content": "partial and complete"}
});
let next = serde_json::json!({
"timestamp": "2026-01-01T00:00:03Z",
"type": "response_item",
"payload": {"type": "message", "role": "assistant", "content": "next"}
});
assert_eq!(
message_candidate_cursor(HarnessId::CODEX, &first),
message_candidate_cursor(HarnessId::CODEX, &grown)
);
assert_ne!(
message_candidate_cursor(HarnessId::CODEX, &first),
message_candidate_cursor(HarnessId::CODEX, &next)
);
}
#[test]
fn codex_child_rollouts_roll_into_roots_before_pagination() {
let root = temp_dir("codex-roots");
let codex = root.join("codex");
fs::create_dir_all(&codex).unwrap();
let write_rollout =
|name: &str, payload: Value, modified_seconds: u64| {
let path = codex.join(format!("{name}.jsonl"));
fs::write(
&path,
format!(
"{}\n",
serde_json::json!({
"timestamp": "2026-01-01T00:00:00Z",
"type": "session_meta",
"payload": payload,
})
),
)
.unwrap();
File::open(&path)
.unwrap()
.set_times(fs::FileTimes::new().set_modified(
UNIX_EPOCH + std::time::Duration::from_secs(modified_seconds),
))
.unwrap();
};
write_rollout(
"parent",
serde_json::json!({"id":"parent","cwd":"/project","source":"cli"}),
100,
);
write_rollout(
"other",
serde_json::json!({"id":"other","cwd":"/project","source":"cli"}),
200,
);
write_rollout(
"child",
serde_json::json!({
"id": "child",
"cwd": "/project",
"parent_thread_id": "parent",
"source": {"subagent":{"thread_spawn":{
"parent_thread_id":"parent",
"depth":1,
"agent_path":"/root/reviewer"
}}}
}),
300,
);
let catalog = HarnessCatalog::new();
let query = DiscoveryQuery {
harnesses: vec![HarnessId::from(HarnessId::CODEX)],
homes: HarnessHomes {
codex: codex.clone(),
..HarnessHomes::default()
},
limit: Some(1),
..DiscoveryQuery::default()
};
let roots = catalog.discover(&query).unwrap();
assert_eq!(roots.len(), 1);
assert_eq!(roots[0].locator.session_id, "parent");
assert_eq!(roots[0].updated_at_ms, Some(300_000));
assert_eq!(roots[0].parent_session_id, None);
assert_eq!(roots[0].child_session_count, 1);
let tree = catalog
.discover(&DiscoveryQuery {
limit: None,
include_child_sessions: true,
root_session_id: Some("parent".into()),
..query
})
.unwrap();
assert_eq!(tree.len(), 2);
assert!(tree
.iter()
.all(|descriptor| descriptor.locator.session_id != "other"));
let child = tree
.iter()
.find(|descriptor| descriptor.locator.session_id == "child")
.unwrap();
assert_eq!(child.parent_session_id.as_deref(), Some("parent"));
fs::remove_dir_all(root).ok();
}
#[test]
fn discovers_loads_and_follows_opencode_sqlite() {
let db = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../harness/tests/fixtures/opencode_fixture/opencode.db");
let catalog = HarnessCatalog::new();
let found = catalog
.discover(&DiscoveryQuery {
harnesses: vec![HarnessId::from(HarnessId::OPENCODE)],
homes: HarnessHomes {
opencode: db,
..HarnessHomes::default()
},
..DiscoveryQuery::default()
})
.unwrap();
assert!(!found.is_empty());
for descriptor in found {
assert_eq!(descriptor.locator.harness.as_str(), HarnessId::OPENCODE);
assert_eq!(
catalog.load(&descriptor.locator).unwrap().meta.session_id,
Some(descriptor.locator.session_id.clone())
);
assert!(catalog.follow(&descriptor.locator).is_ok());
}
}
#[test]
fn discovers_loads_and_follows_hermes_sqlite_by_session() {
let root = temp_dir("hermes-follow");
fs::create_dir_all(&root).unwrap();
let db = root.join("state.db");
fs::copy(
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../harness/tests/fixtures/hermes_home/state.db"),
&db,
)
.unwrap();
let catalog = HarnessCatalog::new();
let query = DiscoveryQuery {
harnesses: vec![HarnessId::from(HarnessId::HERMES)],
homes: HarnessHomes {
hermes: db.clone(),
..HarnessHomes::default()
},
..DiscoveryQuery::default()
};
let found = catalog.discover(&query).unwrap();
assert!(found.len() >= 2, "{found:#?}");
for descriptor in &found {
assert_eq!(descriptor.locator.harness.as_str(), HarnessId::HERMES);
let loaded = catalog.load(&descriptor.locator).unwrap();
let last_text = loaded.messages.iter().rev().find_map(|message| {
(matches!(message.role, crate::Role::User | crate::Role::Assistant))
.then(|| message.content.clone())
.flatten()
});
assert_eq!(
descriptor
.latest_message_candidates
.first()
.map(|c| c.content.as_str()),
last_text.as_deref(),
"{}",
descriptor.locator.session_id
);
assert_eq!(
catalog.load(&descriptor.locator).unwrap().meta.session_id,
Some(descriptor.locator.session_id.clone())
);
let mut follower = catalog.follow(&descriptor.locator).unwrap();
match follower.poll().unwrap() {
Some(crate::watch::SessionWatchEvent::SessionSnapshot { session, .. }) => {
assert_eq!(
session.meta.session_id,
Some(descriptor.locator.session_id.clone())
);
}
other => panic!("expected an initial snapshot, got {other:?}"),
}
}
let target = &found[0].locator;
let sibling = &found[1].locator;
let mut target_follower = catalog.follow(target).unwrap();
let mut sibling_follower = catalog.follow(sibling).unwrap();
target_follower.poll().unwrap();
sibling_follower.poll().unwrap();
std::thread::sleep(std::time::Duration::from_millis(20));
{
let conn = rusqlite::Connection::open(&db).unwrap();
conn.execute(
"INSERT INTO messages (session_id, role, content, timestamp, active) VALUES (?1, 'assistant', 'appended by the follow test', ?2, 1)",
rusqlite::params![target.session_id, 1_800_000_000.0_f64],
)
.unwrap();
}
match target_follower.poll().unwrap() {
Some(crate::watch::SessionWatchEvent::MessagesAppended {
session_id,
messages,
..
}) => {
assert_eq!(session_id, Some(target.session_id.clone()));
assert_eq!(messages.len(), 1);
assert_eq!(
messages[0].content.as_deref(),
Some("appended by the follow test")
);
}
other => panic!("expected messages_appended for the target session, got {other:?}"),
}
assert!(
sibling_follower.poll().unwrap().is_none(),
"the sibling session must not wake"
);
fs::remove_dir_all(&root).ok();
}
#[test]
fn discovers_loads_and_follows_gemini_conversation_records() {
let root = temp_dir("gemini");
let workspace = root.join("workspace");
let chats = root.join("gemini/tmp/demo/chats");
fs::create_dir_all(&workspace).unwrap();
fs::create_dir_all(&chats).unwrap();
fs::write(
root.join("gemini/projects.json"),
serde_json::json!({
"projects": {workspace.to_string_lossy(): "demo"}
})
.to_string(),
)
.unwrap();
let transcript = chats.join("gemini-id.jsonl");
fs::write(
&transcript,
include_str!("../../harness/tests/fixtures/gemini_session.jsonl"),
)
.unwrap();
let catalog = HarnessCatalog::new();
let found = catalog
.discover(&DiscoveryQuery {
harnesses: vec![HarnessId::from(HarnessId::GEMINI)],
homes: HarnessHomes {
gemini: root.join("gemini"),
..HarnessHomes::default()
},
workspace: Some(workspace.clone()),
..DiscoveryQuery::default()
})
.unwrap();
assert_eq!(found.len(), 1);
assert_eq!(found[0].cwd.as_deref(), Some(workspace.as_path()));
assert_eq!(found[0].message_count, None);
assert_eq!(found[0].model.as_deref(), Some("gemini-2.5-pro"));
assert_eq!(found[0].title, None);
assert!(found[0].preview_candidates.is_empty());
assert_eq!(found[0].latest_message_candidates.len(), 3);
assert_eq!(
found[0].latest_message_candidates[0].content,
"Fixture inspected."
);
let loaded = catalog.load(&found[0].locator).unwrap();
assert_eq!(
loaded.meta.session_id.as_deref(),
Some("11111111-1111-4111-8111-111111111111")
);
assert_eq!(loaded.messages.len(), 4);
assert!(matches!(
catalog.follow(&found[0].locator).unwrap().poll().unwrap(),
Some(crate::SessionWatchEvent::SessionSnapshot { .. })
));
fs::remove_dir_all(root).ok();
}
#[test]
fn discovers_native_store_and_pages_search_results() {
let root = temp_dir("supercode");
let store_root = root.join("sessions");
fs::create_dir_all(&store_root).unwrap();
for (name, title) in [
("alpha", "Alpha planning"),
("beta", "Beta implementation"),
("gamma", "Gamma review"),
] {
fs::write(
store_root.join(format!("{name}.jsonl")),
format!("{{\"role\":\"user\",\"content\":\"{title}\"}}\n"),
)
.unwrap();
fs::write(
store_root.join(format!("{name}.meta.json")),
serde_json::json!({"name": name, "title": title}).to_string(),
)
.unwrap();
}
let catalog = HarnessCatalog::new();
let base = DiscoveryQuery {
harnesses: vec![HarnessId::from(HarnessId::SUPERCODE)],
homes: HarnessHomes {
supercode: store_root,
..HarnessHomes::default()
},
limit: Some(1),
..DiscoveryQuery::default()
};
let first = catalog.discover_page(&base).unwrap();
assert_eq!(first.sessions.len(), 1);
assert!(first.next_cursor.is_some());
let second = catalog
.discover_page(&DiscoveryQuery {
cursor: first.next_cursor,
..base.clone()
})
.unwrap();
assert_eq!(second.sessions.len(), 1);
assert_ne!(
first.sessions[0].locator.session_id,
second.sessions[0].locator.session_id
);
let search = catalog
.discover_page(&DiscoveryQuery {
limit: None,
query: Some("implementation".into()),
..base
})
.unwrap();
assert_eq!(search.sessions.len(), 1);
assert_eq!(search.sessions[0].locator.session_id, "beta");
assert_eq!(search.sessions[0].message_count, None);
assert_eq!(
catalog
.load(&search.sessions[0].locator)
.unwrap()
.messages
.len(),
1
);
fs::remove_dir_all(root).ok();
}
#[test]
fn native_workspace_discovery_reads_bounded_sidecar_headers() {
let root = temp_dir("supercode-bounded-header");
let store_root = root.join("sessions");
let workspace = root.join("project");
fs::create_dir_all(&store_root).unwrap();
fs::create_dir_all(&workspace).unwrap();
let name = "bounded-native";
fs::write(
store_root.join(format!("{name}.meta.json")),
serde_json::json!({"name": name, "title": "Bounded native"}).to_string(),
)
.unwrap();
fs::write(
store_root.join(format!("{name}.jsonl")),
"{\"role\":\"user\",\"content\":\"projected view\"}\n",
)
.unwrap();
let sidecar = [
serde_json::json!({
"supercode_native": 2,
"source": "claude_code",
"session_id": "native-session"
})
.to_string(),
serde_json::json!({
"type": "user",
"sessionId": "native-session",
"cwd": workspace,
"message": {"role": "user", "content": "hello"}
})
.to_string(),
serde_json::json!({
"type": "assistant",
"sessionId": "native-session",
"cwd": workspace,
"message": {"role": "assistant", "model": "claude-sonnet-5", "content": []}
})
.to_string(),
"not-json".into(),
]
.join("\n");
fs::write(
store_root.join(format!("{name}.sidecar.jsonl")),
format!("{sidecar}\n"),
)
.unwrap();
let found = HarnessCatalog::new()
.discover(&DiscoveryQuery {
workspace: Some(workspace.clone()),
harnesses: vec![HarnessId::from(HarnessId::SUPERCODE)],
homes: HarnessHomes {
supercode: store_root,
..HarnessHomes::default()
},
..DiscoveryQuery::default()
})
.unwrap();
assert_eq!(found.len(), 1);
assert_eq!(found[0].cwd.as_deref(), Some(workspace.as_path()));
assert_eq!(found[0].model.as_deref(), Some("claude-sonnet-5"));
assert_eq!(found[0].message_count, None);
fs::remove_dir_all(root).ok();
}
#[test]
fn discovers_current_opencode_schema_without_a_session_model_column() {
let root = temp_dir("opencode-current");
let db = root.join("opencode.db");
let conn = Connection::open(&db).unwrap();
conn.execute_batch(
"CREATE TABLE session (
id TEXT PRIMARY KEY,
directory TEXT NOT NULL,
title TEXT NOT NULL,
time_updated INTEGER NOT NULL
);
CREATE TABLE message (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL
);
INSERT INTO session VALUES ('ses_current', '/tmp/work', 'Current', 42);
INSERT INTO message VALUES ('msg_current', 'ses_current');",
)
.unwrap();
drop(conn);
let found = HarnessCatalog::new()
.discover(&DiscoveryQuery {
harnesses: vec![HarnessId::from(HarnessId::OPENCODE)],
homes: HarnessHomes {
opencode: db,
..HarnessHomes::default()
},
..DiscoveryQuery::default()
})
.unwrap();
assert_eq!(found.len(), 1);
assert_eq!(found[0].locator.session_id, "ses_current");
assert_eq!(found[0].message_count, Some(1));
assert_eq!(found[0].model, None);
fs::remove_dir_all(root).ok();
}
#[test]
fn workspace_filter_never_matches_a_relative_recorded_cwd() {
let root = temp_dir("opencode-relative-cwd");
let db = root.join("opencode.db");
let conn = Connection::open(&db).unwrap();
let here = std::env::current_dir().unwrap();
conn.execute_batch(&format!(
"CREATE TABLE session (
id TEXT PRIMARY KEY,
directory TEXT NOT NULL,
title TEXT NOT NULL,
time_updated INTEGER NOT NULL
);
CREATE TABLE message (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL
);
INSERT INTO session VALUES ('ses_relative', '.', 'Ghost', 41);
INSERT INTO session VALUES ('ses_here', '{}', 'Real', 42);",
here.display()
))
.unwrap();
drop(conn);
let found = HarnessCatalog::new()
.discover(&DiscoveryQuery {
workspace: Some(here),
harnesses: vec![HarnessId::from(HarnessId::OPENCODE)],
homes: HarnessHomes {
opencode: db,
..HarnessHomes::default()
},
..DiscoveryQuery::default()
})
.unwrap();
assert_eq!(found.len(), 1);
assert_eq!(found[0].locator.session_id, "ses_here");
fs::remove_dir_all(root).ok();
}
}