use super::{Entry, ListParams};
use crate::{
cancellation::AgentCancellation,
local_agent::{sessions::validate_cwds, startup::RuntimeContext},
sessions::{SessionEvent, SessionManager},
};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use std::{
fs::{self, File, ReadDir},
io::{BufRead, BufReader},
os::unix::{ffi::OsStrExt, fs::MetadataExt},
path::{Path, PathBuf},
time::{Duration, Instant},
};
const STEP_NAMES: usize = 256;
const STEP_LINES: usize = 10_000;
const STEP_BYTES: usize = 8 * 1024 * 1024;
const RECORD_BYTES: usize = 8 * 1024 * 1024;
#[derive(Default, PartialEq, Eq)]
struct Inventory {
sum: [u64; 4],
count: u64,
}
impl Inventory {
fn add(&mut self, signature: [u8; 32]) {
for (sum, bytes) in self.sum.iter_mut().zip(signature.as_chunks::<8>().0) {
*sum = sum.wrapping_add(u64::from_le_bytes(*bytes));
}
self.count = self.count.wrapping_add(1);
}
}
fn signature(path: &Path) -> Result<[u8; 32], &'static str> {
let metadata = fs::symlink_metadata(path).map_err(|_| "stale_revision")?;
let lock = path
.extension()
.is_some_and(|extension| extension == "lock");
let mut digest = Sha256::new();
digest.update(path.file_name().ok_or("discovery_unavailable")?.as_bytes());
for value in [
metadata.dev(),
metadata.ino(),
metadata.mode() as u64,
metadata.uid() as u64,
metadata.nlink(),
if lock { 0 } else { metadata.len() },
if lock { 0 } else { metadata.mtime() as u64 },
if lock {
0
} else {
metadata.mtime_nsec() as u64
},
if lock { 0 } else { metadata.ctime() as u64 },
if lock {
0
} else {
metadata.ctime_nsec() as u64
},
] {
digest.update(value.to_le_bytes());
}
Ok(digest.finalize().into())
}
struct History {
path: PathBuf,
signature: [u8; 32],
reader: BufReader<File>,
line: Vec<u8>,
entry: Entry,
verified: bool,
last_cwd: Option<PathBuf>,
}
pub(in crate::local_agent) struct Scan {
revision: String,
cursor: String,
root_signature: [u8; 32],
attached: Option<String>,
directory: ReadDir,
inventory: Inventory,
validation: Option<Inventory>,
history: Option<History>,
pending: Option<Entry>,
notices: Vec<&'static str>,
}
impl Scan {
fn consume_record(
&mut self,
cwd: &Path,
cancellation: &AgentCancellation,
) -> Result<bool, &'static str> {
match self
.history
.as_mut()
.expect("active history")
.record(cwd, cancellation)
{
Ok(true) => Ok(true),
Ok(false) => {
self.omitted();
self.history = None;
Ok(false)
}
Err("excluded_workspace") => {
self.history = None;
Ok(false)
}
Err(code) => Err(code),
}
}
fn new(root: &Path, attached: Option<String>) -> Result<Self, &'static str> {
SessionManager::new(root.to_owned()).validate_discovery_root()?;
Ok(Self {
revision: uuid::Uuid::new_v4().to_string(),
cursor: uuid::Uuid::new_v4().to_string(),
root_signature: signature(root)?,
attached,
directory: fs::read_dir(root).map_err(|_| "discovery_unavailable")?,
inventory: Inventory::default(),
validation: None,
history: None,
pending: None,
notices: Vec::new(),
})
}
fn omitted(&mut self) {
if self.notices.is_empty() {
self.notices.push("sessions_omitted_invalid");
}
}
fn check_revision(&self, root: &Path, attached: &Option<String>) -> Result<(), &'static str> {
if self.root_signature != signature(root)? || &self.attached != attached {
return Err("stale_revision");
}
if let Some(history) = &self.history
&& history.signature != signature(&history.path)?
{
return Err("stale_revision");
}
Ok(())
}
fn step(
&mut self,
root: &Path,
cwd: &Path,
limit: usize,
cancellation: &AgentCancellation,
) -> Result<Value, &'static str> {
let deadline = Instant::now() + Duration::from_secs(2);
let mut names = 0;
let mut lines = 0;
let mut bytes = 0;
let mut entries = Vec::new();
let mut page_bytes = 1024;
let mut complete = false;
let mut progress = "directory_work";
while entries.len() < limit
&& names < STEP_NAMES
&& lines < STEP_LINES
&& bytes < STEP_BYTES
{
cancellation.check().map_err(|_| "operation_cancelled")?;
if Instant::now() >= deadline {
progress = "deadline";
break;
}
if let Some(entry) = self.pending.take() {
let cost = serde_json::to_vec(&entry)
.map_err(|_| "discovery_page_bytes")?
.len()
+ 1;
if page_bytes + cost > 32_768 {
self.pending = Some(entry);
progress = "page_bytes";
break;
}
page_bytes += cost;
entries.push(entry);
continue;
}
if let Some(history) = self.history.as_mut() {
progress = "history_work";
let buffer = history
.reader
.fill_buf()
.map_err(|_| "discovery_unavailable")?;
if buffer.is_empty() {
if !history.line.is_empty() {
lines += 1;
if !self.consume_record(cwd, cancellation)? {
continue;
}
}
let mut history = self.history.take().expect("active history");
if history.signature != signature(&history.path)? {
return Err("stale_revision");
}
let busy = self.attached.as_deref() != Some(&history.entry.session_id)
&& ["writer", "active", "jsonl"].iter().any(|extension| {
fs::symlink_metadata(crate::persistence::lock_path(
&history.path.with_extension(extension),
))
.is_ok()
});
history.entry.eligibility = if busy {
"busy"
} else if history.verified {
"available"
} else {
"cwd_unverified"
};
if !history.verified {
history.entry.notices.push("cwd_unverified");
}
self.pending = Some(history.entry);
continue;
}
let take = buffer
.iter()
.position(|byte| *byte == b'\n')
.map_or(buffer.len(), |index| index + 1)
.min(STEP_BYTES - bytes);
if history.line.len() + take > RECORD_BYTES {
return Err("discovery_record_bytes");
}
history.line.extend_from_slice(&buffer[..take]);
history.reader.consume(take);
bytes += take;
if history.line.last() == Some(&b'\n') {
lines += 1;
self.consume_record(cwd, cancellation)?;
}
continue;
}
let Some(item) = self.directory.next() else {
if let Some(validation) = &self.validation {
if validation != &self.inventory {
return Err("stale_revision");
}
complete = true;
progress = "complete";
break;
}
self.validation = Some(Inventory::default());
self.directory = fs::read_dir(root).map_err(|_| "discovery_unavailable")?;
continue;
};
names += 1;
let item = item.map_err(|_| "discovery_unavailable")?;
let path = item.path();
let item_signature = signature(&path)?;
if let Some(validation) = self.validation.as_mut() {
progress = "inventory_validation";
validation.add(item_signature);
continue;
}
self.inventory.add(item_signature);
let name = item.file_name();
let Some(id) = name.to_str().and_then(|name| name.strip_suffix(".jsonl")) else {
continue;
};
if id.len() > 64 || crate::sessions::validate_session_id(id.to_owned()).is_err() {
return Err("discovery_unavailable");
}
let session = SessionManager::new(root.to_owned())
.open(id)
.map_err(|_| "discovery_unavailable")?;
let file = match session.open_discovery_file() {
Ok(file) => file,
Err(_) => {
self.omitted();
continue;
}
};
if item_signature != signature(&path)? {
return Err("stale_revision");
}
self.history = Some(History {
path,
signature: item_signature,
reader: BufReader::new(file),
line: Vec::new(),
entry: Entry {
session_id: id.to_owned(),
title: None,
updated_at: None,
eligibility: "cwd_unverified",
notices: Vec::new(),
},
verified: false,
last_cwd: None,
});
}
if names == STEP_NAMES {
progress = if self.validation.is_some() {
"inventory_validation"
} else {
"directory_work"
};
}
if lines == STEP_LINES {
progress = "record_work";
}
if bytes == STEP_BYTES {
progress = "byte_work";
}
if entries.len() == limit {
progress = "page_entries";
}
self.cursor = uuid::Uuid::new_v4().to_string();
let result = json!({"revision":self.revision,"sessions":entries,
"next_cursor":if complete { None } else { Some(&self.cursor) },
"complete":complete,"notices":self.notices,
"consistency":"provisional_until_complete","progress":progress,
"recovery":if complete { "none" } else { "continue_cursor" }});
if serde_json::to_vec(&result)
.map_err(|_| "discovery_page_bytes")?
.len()
> 32_768
{
return Err("discovery_page_bytes");
}
Ok(result)
}
}
impl History {
fn record(
&mut self,
cwd: &Path,
cancellation: &AgentCancellation,
) -> Result<bool, &'static str> {
let event = match serde_json::from_slice::<SessionEvent>(&self.line) {
Ok(event) if event.session_id == self.entry.session_id => event,
_ => return Ok(false),
};
self.line.clear();
if !event.cwd.as_os_str().is_empty() && self.last_cwd.as_ref() != Some(&event.cwd) {
match validate_cwds(std::slice::from_ref(&event), cwd, cancellation) {
Ok(verified) => self.verified |= verified,
Err("cwd_mismatch") => return Err("excluded_workspace"),
Err(code) => return Err(code),
}
self.last_cwd = Some(event.cwd.clone());
}
self.entry.updated_at = self.entry.updated_at.max(Some(event.timestamp));
let title = match event.kind() {
Some(crate::sessions::SessionEventKind::SessionTitle) => event.payload.get("title"),
Some(crate::sessions::SessionEventKind::Compaction) => event
.payload
.get("aggregate")
.and_then(|aggregate| aggregate.get("latest_title")),
_ => None,
};
if let Some(title) = title
.and_then(Value::as_str)
.filter(|title| {
title.len() <= 256
&& !title
.split_whitespace()
.any(crate::config::looks_like_secret_value)
})
.and_then(crate::sessions::sanitize_session_title)
{
self.entry.title = Some(title);
}
Ok(true)
}
}
impl RuntimeContext {
pub(in crate::local_agent) fn discover_sessions(
&mut self,
params: ListParams,
cancellation: &AgentCancellation,
) -> Result<Value, &'static str> {
let root = &self.launch.config.paths.sessions;
let attached = self
.attachment
.as_ref()
.map(|attachment| attachment.session.id().to_owned());
let mut scan = if let Some(cursor) = ¶ms.cursor {
let stored = self.incremental_discovery.take().ok_or("stale_revision")?;
if params.revision.as_ref() != Some(&stored.revision) || cursor != &stored.cursor {
return Err("stale_revision");
}
stored
} else {
self.incremental_discovery = None;
Scan::new(root, attached.clone())?
};
scan.check_revision(root, &attached)?;
let result = scan.step(root, &self.launch.cwd, params.limit, cancellation)?;
scan.check_revision(root, &attached)?;
cancellation.check().map_err(|_| "operation_cancelled")?;
if result["complete"] != true {
self.incremental_discovery = Some(scan);
}
Ok(result)
}
}