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 = "is_false")]
pub searched_previews: bool,
#[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
}
pub fn hermes_session_stores(db_path: &Path) -> Vec<PathBuf> {
let mut stores = vec![db_path.to_path_buf()];
let (Some(root), Some(name)) = (db_path.parent(), db_path.file_name()) else {
return stores;
};
for (_, dir) in orchestrator_profile_dirs(root).into_iter().skip(1) {
stores.push(dir.join(name));
}
stores
}
fn claude_live_named(query: &DiscoveryQuery) -> Option<(String, String)> {
let wanted = query
.query
.as_deref()
.map(str::trim)
.filter(|q| !q.is_empty())?;
if !query.harnesses.is_empty()
&& !query
.harnesses
.iter()
.any(|harness| harness.as_str() == HarnessId::CLAUDE_CODE)
{
return None;
}
let registry = query.homes.claude_code.parent()?.join("sessions");
std::fs::read_dir(registry)
.ok()?
.flatten()
.find_map(|entry| {
let record: Value = serde_json::from_slice(&std::fs::read(entry.path()).ok()?).ok()?;
let name = record.get("name")?.as_str()?;
let id = record.get("sessionId")?.as_str()?;
(name == wanted).then(|| (name.to_string(), id.to_string()))
})
}
impl Default for HarnessHomes {
fn default() -> Self {
let home = crate::user_home()
.map(std::path::PathBuf::into_os_string)
.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>,
#[serde(skip_serializing_if = "std::ops::Not::not")]
pub workspace_subtree: bool,
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 search_previews: bool,
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>> {
Ok(self.project_index_page(query, descriptors)?.sessions)
}
pub fn project_index_page(
&self,
query: &DiscoveryQuery,
descriptors: impl IntoIterator<Item = SessionDescriptor>,
) -> Result<DiscoveryPage> {
if query.search_previews {
return Err(Error::Other(
"preview search requires discover_page, not a metadata-only index projection"
.into(),
));
}
let mut found = descriptors.into_iter().collect::<Vec<_>>();
project_descriptors(query, &mut found);
let total_matched = found.len();
let (sessions, next_cursor) = paginate_descriptors(query, found)?;
let receipt = DiscoveryReceipt {
searched_previews: false,
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,
})
}
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> {
if query.search_previews {
return self.discover_preview_page(query);
}
let mut page = self.project_index_page(
query,
self.scan_descriptors(query, query.include_child_sessions),
)?;
if query.cursor.is_none() {
if let Some((name, id)) = claude_live_named(query) {
if let Some(at) = page
.sessions
.iter()
.position(|session| session.locator.session_id == id)
{
let mut found = page.sessions.swap_remove(at);
found.title.get_or_insert(name);
page.sessions = vec![found];
page.next_cursor = None;
}
}
}
if let Some(id) = query.query.as_deref().map(str::trim) {
if page
.sessions
.iter()
.any(|session| session.locator.session_id == id)
{
page.sessions
.retain(|session| session.locator.session_id == id);
page.next_cursor = None;
}
}
if page.sessions.is_empty() && query.cursor.is_none() {
if let Some(named) = self.claude_session_named(query)? {
page.sessions.push(named);
}
}
enrich_descriptors(query, &mut page.sessions, None)?;
Ok(page)
}
fn claude_session_named(&self, query: &DiscoveryQuery) -> Result<Option<SessionDescriptor>> {
let Some(name) = query
.query
.as_deref()
.map(str::trim)
.filter(|name| !name.is_empty())
else {
return Ok(None);
};
if !query.harnesses.is_empty()
&& !query
.harnesses
.iter()
.any(|harness| harness.as_str() == HarnessId::CLAUDE_CODE)
{
return Ok(None);
}
let mut best: Option<(std::time::SystemTime, String)> = None;
let Ok(projects) = std::fs::read_dir(&query.homes.claude_code) else {
return Ok(None);
};
for project in projects.flatten() {
let Ok(files) = std::fs::read_dir(project.path()) else {
continue;
};
for file in files.flatten() {
let path = file.path();
if path.extension().and_then(|value| value.to_str()) != Some("jsonl") {
continue;
}
if !claude_custom_titles(&path)
.iter()
.any(|title| title == name)
{
continue;
}
let modified = file
.metadata()
.and_then(|meta| meta.modified())
.unwrap_or(std::time::UNIX_EPOCH);
let Some(id) = path
.file_stem()
.map(|stem| stem.to_string_lossy().into_owned())
else {
continue;
};
if best.as_ref().is_none_or(|(time, _)| modified > *time) {
best = Some((modified, id));
}
}
}
let Some((_, id)) = best else {
return Ok(None);
};
let mut unnamed = query.clone();
unnamed.query = None;
unnamed.harnesses = vec![HarnessId::new(HarnessId::CLAUDE_CODE)];
unnamed.limit = None;
let found = self
.project_index_page(
&unnamed,
self.scan_descriptors(&unnamed, unnamed.include_child_sessions),
)?
.sessions
.into_iter()
.find(|descriptor| descriptor.locator.session_id == id);
Ok(found.map(|mut descriptor| {
descriptor.title = Some(name.to_string());
descriptor
}))
}
fn discover_preview_page(&self, query: &DiscoveryQuery) -> Result<DiscoveryPage> {
let search = query
.query
.as_deref()
.map(str::trim)
.filter(|text| !text.is_empty())
.ok_or_else(|| Error::Other("search_previews requires a nonempty query".into()))?
.to_lowercase();
if query.limit == Some(0) {
return Err(Error::Other("preview search limit must be positive".into()));
}
let cursor = query.cursor.as_deref().map(decode_cursor).transpose()?;
let mut eligible_query = query.clone();
eligible_query.query = None;
eligible_query.search_previews = false;
eligible_query.include_topic_candidates = true;
let mut eligible = self.scan_descriptors(query, query.include_child_sessions);
project_descriptors(&eligible_query, &mut eligible);
let topics = codex_history_topics(&query.homes.codex, &eligible).unwrap_or_default();
let mut sessions = Vec::new();
let mut total_matched = 0;
let mut cursor_seen = cursor.is_none();
let mut more = false;
for mut descriptor in eligible {
let metadata_match = descriptor_matches(&descriptor, &search);
if !metadata_match {
let topic = topics
.get(&descriptor.locator.session_id)
.filter(|_| descriptor.locator.harness.as_str() == HarnessId::CODEX);
enrich_descriptor(&eligible_query, &mut descriptor, topic);
if !descriptor
.preview_candidates
.iter()
.chain(&descriptor.latest_message_candidates)
.any(|candidate| candidate.content.to_lowercase().contains(&search))
{
continue;
}
}
total_matched += 1;
if !cursor_seen {
cursor_seen = cursor.as_ref() == Some(&descriptor_cursor_key(&descriptor));
continue;
}
if sessions.len() < query.limit.unwrap_or(usize::MAX) {
if metadata_match {
let topic = topics
.get(&descriptor.locator.session_id)
.filter(|_| descriptor.locator.harness.as_str() == HarnessId::CODEX);
enrich_descriptor(&eligible_query, &mut descriptor, topic);
}
sessions.push(descriptor);
} else {
more = true;
}
}
if !cursor_seen {
return Err(Error::Other("discovery cursor is stale or invalid".into()));
}
let next_cursor = more.then(|| sessions.last().map(encode_cursor)).flatten();
let receipt = DiscoveryReceipt {
searched_previews: true,
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: more,
};
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 scope = WorkspaceScope::of(query);
let mut found = Vec::new();
if selected.contains(HarnessId::CLAUDE_CODE) {
discover_jsonl(
&query.homes.claude_code,
HarnessId::CLAUDE_CODE,
scope.as_ref(),
include_child_sessions,
&mut found,
);
}
if selected.contains(HarnessId::CODEX) {
discover_jsonl(
&query.homes.codex,
HarnessId::CODEX,
scope.as_ref(),
include_child_sessions,
&mut found,
);
}
if selected.contains(HarnessId::PI) {
discover_jsonl(
&query.homes.pi,
HarnessId::PI,
scope.as_ref(),
include_child_sessions,
&mut found,
);
}
if selected.contains(HarnessId::OPENCODE) {
discover_opencode(&query.homes.opencode, scope.as_ref(), &mut found);
}
if selected.contains(HarnessId::GROK) {
discover_grok(&query.homes.grok, scope.as_ref(), &mut found);
}
if selected.contains(HarnessId::GEMINI) {
discover_gemini(&query.homes.gemini, scope.as_ref(), &mut found);
}
if selected.contains(HarnessId::GOOSE) {
discover_goose(&query.homes.goose, scope.as_ref(), &mut found);
}
if selected.contains(HarnessId::OPENCLAW) {
discover_openclaw(&query.homes.openclaw, scope.as_ref(), &mut found);
}
if selected.contains(HarnessId::HERMES) {
for store in hermes_session_stores(&query.homes.hermes) {
discover_hermes(&store, scope.as_ref(), &selected, &mut found);
}
}
if selected.contains(HarnessId::ORCHESTRATOR) {
discover_orchestrator(&query.homes.orchestrator, scope.as_ref(), &mut found);
}
if selected.contains(HarnessId::SUPERCODE) {
discover_supercode(&query.homes.supercode, scope.as_ref(), &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 scope = workspace.map(WorkspaceScope::exact);
self.refresh_file_index_descriptor_scoped(locator, scope.as_ref())
}
pub fn refresh_file_index_descriptor_for(
&self,
locator: &SessionLocator,
query: &DiscoveryQuery,
) -> Result<Option<SessionDescriptor>> {
let scope = WorkspaceScope::of(query);
self.refresh_file_index_descriptor_scoped(locator, scope.as_ref())
}
fn refresh_file_index_descriptor_scoped(
&self,
locator: &SessionLocator,
workspace: Option<&WorkspaceScope>,
) -> 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| !wanted.admits(cwd)))
{
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 tail = tail_facts(path, locator.harness.as_str());
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: tail.last_turn_ms.or_else(|| modified_ms(path)),
message_count: None,
model: tail.model.or(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 {
let topic = (descriptor.locator.harness.as_str() == HarnessId::CODEX)
.then(|| {
codex_history
.and_then(|history| history.topics.get(&descriptor.locator.session_id))
.or_else(|| codex_topics.get(&descriptor.locator.session_id))
})
.flatten();
enrich_descriptor(query, descriptor, topic);
}
Ok(())
}
fn enrich_descriptor(
query: &DiscoveryQuery,
descriptor: &mut SessionDescriptor,
codex_topic: Option<&Vec<SessionPreviewCandidate>>,
) {
if query.include_topic_candidates {
descriptor.preview_candidates = codex_topic
.cloned()
.unwrap_or_else(|| topic_message_candidates(&descriptor.locator).unwrap_or_default());
}
descriptor.latest_message_candidates =
latest_message_candidates(&descriptor.locator).unwrap_or_default();
}
fn is_false(value: &bool) -> bool {
!value
}
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<&WorkspaceScope>,
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| !wanted.admits(cwd)))
{
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()
});
let tail = tail_facts(&path, harness);
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: tail.last_turn_ms.or_else(|| modified_ms(&path)),
message_count: None,
model: tail.model.or(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 mut walked = HashSet::new();
collect_jsonl_in(root, harness, include_child_sessions, out, &mut walked);
}
fn collect_jsonl_in(
root: &Path,
harness: &str,
include_child_sessions: bool,
out: &mut Vec<PathBuf>,
walked: &mut HashSet<PathBuf>,
) {
if !walked.insert(fs::canonicalize(root).unwrap_or_else(|_| root.to_path_buf())) {
return;
}
let Ok(entries) = fs::read_dir(root) else {
return;
};
for entry in entries.flatten() {
let Ok(mut kind) = entry.file_type() else {
continue;
};
let path = entry.path();
if kind.is_symlink() {
let Ok(target) = fs::metadata(&path) else {
continue;
};
kind = target.file_type();
}
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_in(&path, harness, include_child_sessions, out, walked);
} else if kind.is_file()
&& (path.extension().and_then(|v| v.to_str()) == Some("jsonl")
|| (harness == HarnessId::CODEX
&& path
.file_name()
.and_then(|v| v.to_str())
.is_some_and(|name| name.ends_with(".jsonl.zst"))))
{
out.push(path);
}
}
}
fn read_header(path: &Path, harness: &str) -> Result<HeaderMeta> {
let file = crate::session::open_session_reader(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.title, value.get("customTitle"));
fill_string(&mut result.title, value.get("agentName"));
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<&WorkspaceScope>,
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<&WorkspaceScope>,
) -> 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| !wanted.admits(actual))) {
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: tail_facts(path, HarnessId::GEMINI)
.last_turn_ms
.or_else(|| 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<&WorkspaceScope>,
selected: &HashSet<&str>,
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 mirror = model_config
.as_deref()
.and_then(|c| serde_json::from_str::<serde_json::Value>(c).ok())
.and_then(|c| c.get("_supercode_mirror").cloned());
if let Some(mirror) = mirror {
let worker_read = mirror
.get("harness")
.and_then(serde_json::Value::as_str)
.is_some_and(|h| selected.contains(h));
let mirrored = mirror.get("messages").and_then(serde_json::Value::as_i64);
if worker_read && mirrored.is_none_or(|m| message_count.unwrap_or(0) <= m) {
continue;
}
if mirror.get("continued_as").is_some()
&& mirrored.is_none_or(|m| message_count.unwrap_or(0) <= m)
{
continue;
}
}
let cwd = cwd.map(PathBuf::from);
if let Some(filter) = workspace {
if !filter.admits_literal(cwd.as_deref()) {
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<&WorkspaceScope>,
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 sessions = dir.join("sessions");
let scope = std::fs::canonicalize(&sessions)
.unwrap_or(sessions)
.display()
.to_string();
let Ok(mut statement) = conn.prepare(
"SELECT json_extract(entry_json, '$.metadata.supercode.binding'), \
CAST(strftime('%s', json_extract(entry_json, '$.metadata.supercode.binding.last_activity_at')) AS INTEGER) \
FROM gateway_routing WHERE scope = ?1 AND json_extract(entry_json, '$.metadata.supercode') IS NOT NULL \
ORDER BY json_extract(entry_json, '$.metadata.supercode.binding.last_activity_at') DESC",
) else {
continue;
};
let Ok(rows) = statement.query_map([&scope], |row| {
Ok((
row.get::<_, Option<String>>(0)?,
row.get::<_, Option<i64>>(1)?,
))
}) else {
continue;
};
let rows = rows.flatten().filter_map(|(json, epoch)| {
let b: Binding = serde_json::from_str(&json?).ok()?;
let text = |v: &Option<String>| v.clone().filter(|s| !s.is_empty());
Some((
OrchestratorBindingRow {
platform: b.key.platform.clone().unwrap_or_default(),
chat_type: b.key.kind.clone().unwrap_or_default(),
chat_id: text(&b.key.chat_id),
thread_id: text(&b.key.thread_id),
participant_id: text(&b.key.participant_id),
worker_harness: b.worker.harness.as_str().to_string(),
worker_session_id: text(&b.worker.session_id),
worker_locator: text(&b.worker.locator),
started_at: b.started_at.clone(),
last_activity_at: b.last_activity_at.clone(),
ended_at: b.ended_at.clone(),
end_reason: b.end_reason.map(|r| r.as_str().to_string()),
handoff_to: b.handoff.as_ref().and_then(|h| h.to.clone()),
handoff_state: b.handoff.as_ref().map(|h| h.state.clone()),
handoff_error: b.handoff.as_ref().and_then(|h| h.error.clone()),
recurrence_job_id: b.recurrence.as_ref().map(|r| r.job_id.clone()),
},
epoch,
))
});
for (row, last_activity_epoch) in rows {
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<&WorkspaceScope>,
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 !filter.admits_literal(cwd.as_deref()) {
continue;
}
}
let updated_at_ms = tail_facts(&path, HarnessId::OPENCLAW)
.last_turn_ms
.or_else(|| {
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<&WorkspaceScope>,
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 sidecar = path.with_extension("sidecar.jsonl");
let path = if path.is_file() {
path
} else {
sidecar.clone()
};
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| !wanted.admits(cwd))
}) {
continue;
}
let title = (!info.title.trim().is_empty()).then_some(info.title);
let updated_at_ms = tail_facts(&sidecar, HarnessId::SUPERCODE)
.last_turn_ms
.or_else(|| tail_facts(&path, HarnessId::SUPERCODE).last_turn_ms)
.or_else(|| modified_ms(&path))
.or_else(|| modified_ms(&sidecar));
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<&WorkspaceScope>,
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| !wanted.admits(&cwd)) {
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<&WorkspaceScope>,
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| !wanted.admits(&cwd)) {
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<&WorkspaceScope>,
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| !wanted.admits(&cwd)) {
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()
))
})?;
if harness == HarnessId::CODEX {
return Ok(codex_preview_candidates(text.lines(), false));
}
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()
))
})?;
if harness == HarnessId::CODEX {
return Ok(codex_preview_candidates(text.lines().rev(), true));
}
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::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)
}
struct CodexPreviewRecord {
native: Value,
role: String,
text: String,
}
fn codex_preview_candidates<'a>(
lines: impl Iterator<Item = &'a str>,
latest: bool,
) -> Vec<SessionPreviewCandidate> {
let mut candidates = Vec::new();
let mut pending: Option<CodexPreviewRecord> = None;
for line in lines {
let Ok(native) = serde_json::from_str::<Value>(line) else {
continue;
};
let Some((role, content)) = codex_preview_message(&native) else {
continue;
};
let Some(text) = display_text(content).filter(|text| !text.trim().is_empty()) else {
continue;
};
let current = CodexPreviewRecord {
role: role.to_string(),
text,
native,
};
if let Some(previous) = pending.take() {
if previous.role == current.role
&& previous.text == current.text
&& previous.native.get("type") != current.native.get("type")
{
let canonical = if previous.native.get("type").and_then(Value::as_str)
== Some("response_item")
{
previous
} else {
current
};
push_codex_preview_candidate(&mut candidates, canonical, latest);
} else {
push_codex_preview_candidate(&mut candidates, previous, latest);
pending = Some(current);
}
} else {
pending = Some(current);
}
if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
break;
}
}
if let Some(last) = pending {
push_codex_preview_candidate(&mut candidates, last, latest);
}
candidates
}
fn push_codex_preview_candidate(
candidates: &mut Vec<SessionPreviewCandidate>,
record: CodexPreviewRecord,
latest: bool,
) {
let mut metadata = HashMap::new();
if latest {
if let Some(timestamp) = record.native.get("timestamp").and_then(Value::as_str) {
metadata.insert("timestamp".to_string(), timestamp.to_string());
}
}
let cursor = latest.then(|| message_candidate_cursor(HarnessId::CODEX, &record.native));
push_message_candidate_with_cursor(
candidates,
&record.role,
Some(&Value::String(record.text)),
metadata,
cursor,
);
}
fn codex_preview_message(value: &Value) -> Option<(&str, Option<&Value>)> {
let payload = value.get("payload")?;
match (
value.get("type").and_then(Value::as_str)?,
payload.get("type").and_then(Value::as_str)?,
) {
("response_item", "message") => {
let role @ ("user" | "assistant") = payload.get("role").and_then(Value::as_str)? else {
return None;
};
Some((role, payload.get("content")))
}
("event_msg", "user_message") => Some(("user", payload.get("message"))),
("event_msg", "agent_message") => Some(("assistant", payload.get("message"))),
_ => None,
}
}
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,
)
}
_ => 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);
}
}
#[derive(Default)]
struct TailFacts {
last_turn_ms: Option<u64>,
model: Option<String>,
}
fn tail_facts(path: &Path, harness: &str) -> TailFacts {
let mut facts = TailFacts::default();
let Ok(mut file) = File::open(path) else {
return facts;
};
let Ok(len) = file.metadata().map(|meta| meta.len()) else {
return facts;
};
let mut window = TAIL_SCAN_START.min(len);
loop {
if file.seek(SeekFrom::Start(len - window)).is_err() {
return facts;
}
let Ok(size) = usize::try_from(window) else {
return facts;
};
let mut buf = vec![0u8; size];
if file.read_exact(&mut buf).is_err() {
return facts;
}
let floor = if window < len {
buf.iter().position(|byte| *byte == b'\n').map(|at| at + 1)
} else {
Some(0)
};
if let Some(floor) = floor {
let mut end = buf.len();
while end > floor && !(facts.last_turn_ms.is_some() && facts.model.is_some()) {
let start = buf[floor..end]
.iter()
.rposition(|byte| *byte == b'\n')
.map_or(floor, |at| floor + at + 1);
if let Ok(record) = std::str::from_utf8(&buf[start..end])
.map_err(|_| ())
.and_then(|line| serde_json::from_str::<Value>(line).map_err(|_| ()))
{
if facts.last_turn_ms.is_none() {
facts.last_turn_ms = record_timestamp(&record, harness)
.and_then(crate::sidecar::rfc3339_to_ms)
.and_then(|millis| u64::try_from(millis).ok());
}
if facts.model.is_none() {
facts.model = record_model(&record, harness).map(str::to_owned);
}
}
end = start.saturating_sub(1);
}
}
if facts.last_turn_ms.is_some() || window >= len || window >= TAIL_SCAN_LIMIT {
return facts;
}
window = (window * 2).min(len).min(TAIL_SCAN_LIMIT);
}
}
fn record_timestamp<'a>(record: &'a Value, harness: &str) -> Option<&'a str> {
let key = match harness {
HarnessId::SUPERCODE => "ts",
_ => "timestamp",
};
record.get(key)?.as_str()
}
fn record_model<'a>(record: &'a Value, harness: &str) -> Option<&'a str> {
match harness {
HarnessId::CLAUDE_CODE | HarnessId::PI => record.get("message")?.get("model")?.as_str(),
HarnessId::CODEX => {
if record.get("type")?.as_str()? != "turn_context" {
return None;
}
record.get("payload")?.get("model")?.as_str()
}
_ => None,
}
}
const TAIL_SCAN_START: u64 = 16 * 1024;
const TAIL_SCAN_LIMIT: u64 = 1024 * 1024;
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())
}
pub struct WorkspaceScope {
wanted: PathBuf,
roots: Vec<PathBuf>,
subtree: bool,
judged: std::sync::Mutex<HashMap<PathBuf, bool>>,
}
impl WorkspaceScope {
pub fn exact(wanted: &Path) -> Self {
Self {
wanted: wanted.to_path_buf(),
roots: Vec::new(),
subtree: false,
judged: std::sync::Mutex::new(HashMap::new()),
}
}
pub fn subtree(root: &Path) -> Self {
Self {
wanted: root.to_path_buf(),
roots: path_comparison_keys(root),
subtree: true,
judged: std::sync::Mutex::new(HashMap::new()),
}
}
pub fn of(query: &DiscoveryQuery) -> Option<Self> {
query.workspace.as_deref().map(|wanted| {
if query.workspace_subtree {
Self::subtree(wanted)
} else {
Self::exact(wanted)
}
})
}
pub fn admits(&self, recorded: &Path) -> bool {
if !recorded.is_absolute() {
return false;
}
if !self.subtree {
return same_path(recorded, &self.wanted);
}
let mut judged = self
.judged
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
*judged.entry(recorded.to_path_buf()).or_insert_with(|| {
path_comparison_keys(recorded)
.iter()
.any(|key| self.roots.iter().any(|root| key.starts_with(root)))
})
}
fn admits_literal(&self, recorded: Option<&Path>) -> bool {
if self.subtree {
recorded.is_some_and(|cwd| self.admits(cwd))
} else {
recorded == Some(self.wanted.as_path())
}
}
}
fn path_comparison_keys(path: &Path) -> Vec<PathBuf> {
let mut keys = Vec::with_capacity(2);
if let Ok(canonical) = fs::canonicalize(path) {
keys.push(comparison_key(&canonical));
}
let lexical = comparison_key(&normalize_path(path));
if !keys.contains(&lexical) {
keys.push(lexical);
}
keys
}
#[cfg(windows)]
fn comparison_key(path: &Path) -> PathBuf {
let text = path.to_string_lossy();
let text = if let Some(unc) = text.strip_prefix(r"\\?\UNC\") {
format!(r"\\{unc}")
} else if let Some(local) = text.strip_prefix(r"\\?\") {
local.to_string()
} else {
text.into_owned()
};
PathBuf::from(text.to_lowercase())
}
#[cfg(not(windows))]
fn comparison_key(path: &Path) -> PathBuf {
path.to_path_buf()
}
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
}
fn claude_custom_titles(path: &std::path::Path) -> Vec<String> {
use std::io::{Read, Seek, SeekFrom};
use std::sync::{Mutex, OnceLock};
static CACHE: OnceLock<
Mutex<std::collections::HashMap<std::path::PathBuf, (u64, Vec<String>)>>,
> = OnceLock::new();
const KEY: &[u8] = b"\"customTitle\":";
let Ok(len) = std::fs::metadata(path).map(|meta| meta.len()) else {
return Vec::new();
};
let cache = CACHE.get_or_init(Default::default);
let known = cache.lock().ok().and_then(|cache| cache.get(path).cloned());
let (from, mut titles) = match known {
Some((seen, titles)) if seen == len => return titles,
Some((seen, titles)) if seen < len => (seen.saturating_sub(4096), titles),
_ => (0, Vec::new()),
};
let mut bytes = Vec::new();
let read = std::fs::File::open(path).and_then(|mut file| {
file.seek(SeekFrom::Start(from))?;
file.read_to_end(&mut bytes)
});
if read.is_err() {
return titles;
}
let mut at = 0;
while let Some(offset) = bytes[at..]
.windows(KEY.len())
.position(|window| window == KEY)
{
let start = at + offset + KEY.len();
if let Some(Ok(title)) = serde_json::Deserializer::from_slice(&bytes[start..])
.into_iter::<String>()
.next()
{
if !titles.contains(&title) {
titles.push(title);
}
}
at = start;
}
if let Ok(mut cache) = cache.lock() {
cache.insert(path.to_path_buf(), (len, titles.clone()));
}
titles
}