diff --git a/docs/prd.md b/docs/prd.md index f503a57..9ebe9f1 100644 --- a/docs/prd.md +++ b/docs/prd.md @@ -169,6 +169,8 @@ user prompt. ## Control Commands - `/clear`, `/new`, `/reset`: rotate the current backend session. + For Pi, also delete the abandoned session JSONL under `~/.pi/agent/sessions` + (best-effort; missing files are ignored). - `/help`: show available commands. ## Acceptance Criteria diff --git a/src/gateway/worker.rs b/src/gateway/worker.rs index 458184f..050872a 100644 --- a/src/gateway/worker.rs +++ b/src/gateway/worker.rs @@ -747,17 +747,32 @@ fn complete_job(ctx: &Ctx, job: &Job, reason: &str) { /// Handles gateway-level slash commands before anything reaches the agent. fn command(ctx: &Ctx, job: &Job) -> Option { match job.text.trim().to_lowercase().as_str() { - "/clear" | "/new" | "/reset" => match ctx.store.lock().unwrap().rotate( - &job.thread, - job.backend.as_str(), - ctx.runners - .get(&job.backend) - .map(|r| r.initial_session_id()) - .unwrap_or_default(), - ) { - Ok(()) => Some("Started a fresh conversation.".to_string()), - Err(_) => Some("Couldn't reset the conversation.".to_string()), - }, + "/clear" | "/new" | "/reset" => { + match ctx.store.lock().unwrap().rotate( + &job.thread, + job.backend.as_str(), + ctx.runners + .get(&job.backend) + .map(|r| r.initial_session_id()) + .unwrap_or_default(), + ) { + Ok(previous) => { + if job.backend == crate::config::AgentBackend::Pi { + if let Some(session_id) = previous { + let removed = crate::pi::discard_session(&session_id); + if removed > 0 { + info!( + "[{}] discarded {removed} abandoned Pi session file(s)", + job.thread + ); + } + } + } + Some("Started a fresh conversation.".to_string()) + } + Err(_) => Some("Couldn't reset the conversation.".to_string()), + } + } "/help" => Some( "Commands:\n/clear - start a fresh conversation\n/stop - stop the active request\n/help - this message" .to_string(), diff --git a/src/pi.rs b/src/pi.rs index 4ec2f3a..0eec661 100644 --- a/src/pi.rs +++ b/src/pi.rs @@ -228,6 +228,68 @@ fn missing_resume_error(message: &str) -> bool { .contains("no session found matching") } +/// Best-effort delete of abandoned Pi session JSONL files after `/new`. +/// +/// Pi stores sessions as `~/.pi/agent/sessions/----/_.jsonl` +/// (or under `$PI_CODING_AGENT_DIR/sessions`). Push only keeps one active id per +/// thread; orphaned files otherwise accumulate on disk and never cost tokens. +/// +/// ponytail: walk project session dirs for `*_{id}.jsonl`. Upgrade path: call a +/// Pi delete API if one is added. +pub fn discard_session(session_id: &str) -> usize { + let Some(id) = safe_session_file_id(session_id) else { + return 0; + }; + discard_session_under(&sessions_root(), id) +} + +fn safe_session_file_id(session_id: &str) -> Option<&str> { + let id = session_id.trim(); + if id.len() < 8 || id.contains('/') || id.contains('\\') || id.contains("..") { + return None; + } + Some(id) +} + +fn sessions_root() -> std::path::PathBuf { + if let Ok(dir) = std::env::var("PI_CODING_AGENT_DIR") { + if !dir.trim().is_empty() { + return std::path::PathBuf::from(dir).join("sessions"); + } + } + match std::env::var_os("HOME") { + Some(home) => std::path::PathBuf::from(home).join(".pi/agent/sessions"), + None => std::path::PathBuf::from(".pi/agent/sessions"), + } +} + +fn discard_session_under(root: &std::path::Path, id: &str) -> usize { + let suffix = format!("_{id}.jsonl"); + let Ok(projects) = std::fs::read_dir(root) else { + return 0; + }; + let mut removed = 0; + for project in projects.flatten() { + let project_path = project.path(); + if !project_path.is_dir() { + continue; + } + let Ok(files) = std::fs::read_dir(&project_path) else { + continue; + }; + for file in files.flatten() { + let path = file.path(); + let Some(name) = path.file_name().and_then(|name| name.to_str()) else { + continue; + }; + if name.ends_with(&suffix) && std::fs::remove_file(&path).is_ok() { + removed += 1; + } + } + } + removed +} + #[cfg(test)] mod tests { use super::*; @@ -236,6 +298,24 @@ mod tests { ContractRequest, ContractRunner, FakeCli, RunnerContract, }; + #[test] + fn discards_matching_session_jsonl_only() { + let root = temp_dir("pi-discard-sessions"); + let project = root.join("--tmp-project--"); + std::fs::create_dir_all(&project).unwrap(); + let keep = project.join("2026-01-01T00-00-00-000Z_keep-session-aaaaaaaa.jsonl"); + let drop = project.join("2026-01-01T00-00-00-000Z_drop-session-bbbbbbbb.jsonl"); + std::fs::write(&keep, "{}\n").unwrap(); + std::fs::write(&drop, "{}\n").unwrap(); + + assert_eq!(discard_session_under(&root, "drop-session-bbbbbbbb"), 1); + assert!(keep.exists()); + assert!(!drop.exists()); + assert_eq!(discard_session_under(&root, "drop-session-bbbbbbbb"), 0); + assert_eq!(discard_session("../etc/passwd"), 0); + assert_eq!(safe_session_file_id("short"), None); + } + impl ContractRunner for Runner { fn run<'a>( &'a self, diff --git a/src/store.rs b/src/store.rs index dcab2a8..7b85c54 100644 --- a/src/store.rs +++ b/src/store.rs @@ -285,24 +285,53 @@ impl Store { } /// Assigns a fresh backend session to a thread (the `/clear` behavior). - pub fn rotate(&mut self, thread: &str, backend: &str, initial_id: String) -> Result<()> { + /// + /// Returns the previous `session_id` when one existed, so callers can discard + /// abandoned backend artifacts (for example Pi session JSONL files). + pub fn rotate( + &mut self, + thread: &str, + backend: &str, + initial_id: String, + ) -> Result> { validate_backend(backend)?; let (channel, thread_key) = split_thread(thread)?; self.fail_session_write_for_test()?; - self.conn - .execute( - "INSERT INTO backend_sessions ( - channel, thread_key, backend, session_id, started - ) VALUES (?1, ?2, ?3, ?4, 0) - ON CONFLICT(channel, thread_key) DO UPDATE SET - backend = excluded.backend, - session_id = excluded.session_id, - started = 0, - updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", - params![channel, thread_key, backend, initial_id], + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .with_context(|| { + format!( + "begin session rotate transaction in {}", + self.database_path.display() + ) + })?; + let previous = tx + .query_row( + "SELECT session_id + FROM backend_sessions + WHERE channel = ?1 AND thread_key = ?2", + params![channel, thread_key], + |row| row.get::<_, String>(0), ) - .with_context(|| format!("rotate session for {thread:?}"))?; - Ok(()) + .optional() + .with_context(|| format!("read session before rotate for {thread:?}"))? + .and_then(|id| non_empty_session_id(&id).map(str::to_string)); + tx.execute( + "INSERT INTO backend_sessions ( + channel, thread_key, backend, session_id, started + ) VALUES (?1, ?2, ?3, ?4, 0) + ON CONFLICT(channel, thread_key) DO UPDATE SET + backend = excluded.backend, + session_id = excluded.session_id, + started = 0, + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", + params![channel, thread_key, backend, initial_id], + ) + .with_context(|| format!("rotate session for {thread:?}"))?; + tx.commit() + .with_context(|| format!("commit session rotate for {thread:?}"))?; + Ok(previous) } fn migrate_legacy_state(&mut self, _fail_before_commit: bool) -> Result<()> { @@ -802,9 +831,12 @@ mod tests { store.mark_started(thread, None).unwrap(); } - store - .rotate("telegram:dm:7", "codex", "telegram-new".into()) - .unwrap(); + assert_eq!( + store + .rotate("telegram:dm:7", "codex", "telegram-new".into()) + .unwrap(), + Some("telegram:dm:7-old".into()) + ); assert_eq!( store .session_for("telegram:dm:7", "codex", "unused".into())