fix(gateway): discard abandoned Pi session files on /new

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

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<String> {
     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(),

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/--<cwd>--/<timestamp>_<uuid>.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,

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<Option<String>> {
         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())