pub(super) mod incremental;
use super::{sessions::validate_cwds, startup::RuntimeContext};
use crate::{
cancellation::AgentCancellation,
sessions::{BoundedReadError, SessionManager},
};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use std::time::{Duration, Instant};
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
pub(super) struct ListParams {
pub(super) revision: Option<String>,
pub(super) cursor: Option<String>,
#[serde(default = "default_limit")]
pub(super) limit: usize,
}
fn default_limit() -> usize {
100
}
impl ListParams {
pub(super) fn parse(params: &serde_json::value::RawValue) -> Result<Self, &'static str> {
let params: Self = serde_json::from_str(params.get()).map_err(|_| "invalid_params")?;
if params.limit == 0
|| params.limit > 100
|| params
.cursor
.as_deref()
.is_some_and(|c| !super::valid_id(c))
|| params
.revision
.as_deref()
.is_some_and(|r| !super::valid_id(r))
|| params.cursor.is_some() != params.revision.is_some()
{
return Err("invalid_params");
}
Ok(params)
}
}
#[derive(Serialize)]
struct Entry {
session_id: String,
title: Option<String>,
updated_at: Option<chrono::DateTime<chrono::Utc>>,
eligibility: &'static str,
notices: Vec<&'static str>,
}
pub(super) struct Snapshot {
revision: String,
fingerprint: [u8; 32],
entries: Vec<Entry>,
notices: Vec<&'static str>,
invalidated: bool,
}
impl RuntimeContext {
pub(super) fn list_sessions(
&mut self,
params: ListParams,
cancellation: &AgentCancellation,
) -> Result<Value, &'static str> {
let snapshot = self.scan_sessions(cancellation);
if params.cursor.is_some() {
let stored = self.discovery.as_mut().ok_or("stale_revision")?;
let scanned = match snapshot {
Ok(scanned) => scanned,
Err("operation_cancelled") => return Err("operation_cancelled"),
Err(_) => {
stored.invalidated = true;
return Err("stale_revision");
}
};
if stored.invalidated
|| Some(&stored.revision) != params.revision.as_ref()
|| stored.fingerprint != scanned.fingerprint
{
stored.invalidated = true;
return Err("stale_revision");
}
} else {
self.discovery = None;
self.discovery = Some(snapshot?);
}
let stored = self.discovery.as_ref().ok_or("stale_revision")?;
let start = match params.cursor {
None => 0,
Some(cursor) => cursor
.strip_prefix(&format!("{}.", stored.revision))
.and_then(|offset| offset.parse::<usize>().ok())
.ok_or("stale_revision")?,
};
if start > stored.entries.len() {
return Err("stale_revision");
}
let end = (start + params.limit).min(stored.entries.len());
let next = (end < stored.entries.len()).then(|| format!("{}.{}", stored.revision, end));
let result = json!({"revision":stored.revision, "sessions":&stored.entries[start..end],
"next_cursor":next, "complete":next.is_none(), "notices":stored.notices});
if serde_json::to_vec(&result)
.map_err(|_| "discovery_limit")?
.len()
> 32_768
{
return Err("discovery_limit");
}
Ok(result)
}
fn scan_sessions(&self, cancellation: &AgentCancellation) -> Result<Snapshot, &'static str> {
let deadline = Instant::now() + Duration::from_secs(2);
let manager = SessionManager::new(self.launch.config.paths.sessions.clone());
let (sessions, inventory) = manager.discovery_inventory(cancellation)?;
let mut digest = Sha256::new();
digest.update(inventory);
let mut entries = Vec::new();
let mut notices = Vec::new();
let mut remaining_bytes = 64 * 1024 * 1024;
let mut remaining_lines = 100_000;
for session in sessions {
cancellation.check().map_err(|_| "operation_cancelled")?;
if Instant::now() >= deadline {
return Err("discovery_limit");
}
let read = session.read_events_tolerant_bounded_with_stats(
remaining_lines.min(10_000),
remaining_bytes.min(8 * 1024 * 1024),
);
let (read, stats) = match read {
Ok(read) => read,
Err(error) if error.downcast_ref::<BoundedReadError>().is_some() => {
return Err("discovery_limit");
}
Err(_) => {
remaining_bytes -= remaining_bytes.min(8 * 1024 * 1024);
remaining_lines -= remaining_lines.min(10_000);
if !notices.contains(&"sessions_omitted_invalid") {
notices.push("sessions_omitted_invalid");
}
continue;
}
};
remaining_bytes -= stats.bytes_read;
remaining_lines -= stats.lines_read;
digest.update(stats.content_digest);
if std::fs::symlink_metadata(session.path()).is_err() {
return Err("stale_revision");
}
if !read.diagnostics.is_empty()
|| read.events.iter().any(|e| e.session_id != session.id())
{
if !notices.contains(&"sessions_omitted_invalid") {
notices.push("sessions_omitted_invalid");
}
continue;
}
let verified = match validate_cwds(&read.events, &self.launch.cwd, cancellation) {
Ok(verified) => verified,
Err("cwd_mismatch") => continue,
Err(code) => return Err(code),
};
let attached = self
.attachment
.as_ref()
.is_some_and(|a| a.session.id() == session.id());
let busy = !attached
&& ["writer", "active", "jsonl"].iter().any(|extension| {
std::fs::symlink_metadata(crate::persistence::lock_path(
&session.path().with_extension(extension),
))
.is_ok()
});
digest.update([u8::from(busy), u8::from(attached)]);
let title = read.events.iter().rev().find_map(|e| {
let title = match e.kind() {
Some(crate::sessions::SessionEventKind::SessionTitle) => e.payload.get("title"),
Some(crate::sessions::SessionEventKind::Compaction) => {
e.payload.get("aggregate")?.get("latest_title")
}
_ => None,
};
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)
});
entries.push(Entry {
session_id: session.id().to_owned(),
title,
updated_at: read.events.iter().map(|e| e.timestamp).max(),
eligibility: if busy {
"busy"
} else if verified {
"available"
} else {
"cwd_unverified"
},
notices: if verified {
Vec::new()
} else {
vec!["cwd_unverified"]
},
});
}
if manager.discovery_inventory(cancellation)?.1 != inventory {
return Err("stale_revision");
}
cancellation.check().map_err(|_| "operation_cancelled")?;
if Instant::now() >= deadline {
return Err("discovery_limit");
}
Ok(Snapshot {
revision: uuid::Uuid::new_v4().to_string(),
fingerprint: digest.finalize().into(),
entries,
notices,
invalidated: false,
})
}
}