fix(acp): persist accepted work across restarts

crates/buzz-acp/src/lib.rs

@@ -5,6 +5,7 @@ mod config;
 mod engram_fetch;
 mod filter;
 mod observer;
+mod pending_store;
 mod pool;
 mod pool_lifecycle;
 mod queue;
@@ -1553,8 +1554,18 @@ async fn tokio_main() -> Result<()> {
 
     let runtime_start_nonce = std::env::var("BUZZ_MANAGED_AGENT_START_NONCE").unwrap_or_default();
     let dedup_mode = config.dedup_mode;
-    let mut queue =
-        EventQueue::new(dedup_mode).with_in_flight_deadline(config.max_turn_duration_secs);
+    let pending_store = pending_store::PendingStore::open(&pubkey_hex)
+        .map_err(|error| anyhow::anyhow!("open durable pending-work journal: {error}"))?;
+    let mut queue = EventQueue::new(dedup_mode)
+        .with_in_flight_deadline(config.max_turn_duration_secs)
+        .with_pending_store(pending_store);
+    let restored_event_ids = queue.pending_event_ids();
+    if !restored_event_ids.is_empty() {
+        let rest = relay.rest_client();
+        for event_id in &restored_event_ids {
+            pool::reaction_add(&rest, event_id, "👀").await;
+        }
+    }
 
     // Online means the harness can receive work, not merely that its socket is
     // connected. Publishing after channel subscriptions gives desktop callers
@@ -2731,6 +2742,15 @@ async fn tokio_main() -> Result<()> {
     // explicitly shut them down here to reap child processes. If the grace
     // period expires, remaining tasks are aborted and fall back to
     // AcpClient::Drop (start_kill + try_wait — best-effort, not guaranteed).
+    let shutdown_batches: HashMap<usize, FlushBatch> = pool
+        .task_map()
+        .values()
+        .filter_map(|meta| {
+            meta.recoverable_batch
+                .clone()
+                .map(|batch| (meta.agent_index, batch))
+        })
+        .collect();
     let (rx_ref, js_ref) = pool.rx_and_join_set();
     let shutdown_result = tokio::time::timeout(grace, async {
         loop {
@@ -2745,6 +2765,11 @@ async fn tokio_main() -> Result<()> {
                 maybe_result = rx_ref.recv() => {
                     if let Some(mut pr) = maybe_result {
                         let idx = pr.agent.index;
+                        if matches!(pr.outcome, PromptOutcome::Ok(_)) {
+                            if let Some(batch) = shutdown_batches.get(&idx) {
+                                queue.finish_batch(batch);
+                            }
+                        }
                         pr.agent.acp.shutdown().await;
                         tracing::debug!(agent = idx, "reaped checked-out agent on shutdown");
                     }
@@ -2762,6 +2787,11 @@ async fn tokio_main() -> Result<()> {
     // before tasks were aborted.
     while let Ok(mut pr) = pool.result_rx_try_recv() {
         let idx = pr.agent.index;
+        if matches!(pr.outcome, PromptOutcome::Ok(_)) {
+            if let Some(batch) = shutdown_batches.get(&idx) {
+                queue.finish_batch(batch);
+            }
+        }
         pr.agent.acp.shutdown().await;
         tracing::debug!(agent = idx, "reaped late-arriving agent on shutdown");
     }
@@ -3166,6 +3196,12 @@ fn handle_prompt_result(
     observer: Option<observer::ObserverHandle>,
     rest_client: Option<&relay::RestClient>,
 ) -> LoopAction {
+    let had_retry_batch = result.batch.is_some();
+    let recoverable_batch = pool
+        .task_map()
+        .values()
+        .find(|meta| meta.agent_index == result.agent.index)
+        .and_then(|meta| meta.recoverable_batch.clone());
     let before = pool.task_map().len();
     let agent_index = result.agent.index;
     pool.task_map_mut()
@@ -3227,6 +3263,7 @@ fn handle_prompt_result(
                     config.max_turn_duration_secs
                 );
                 spawn_failure_notice(rest_client, &batch, content);
+                queue.finish_batch(&batch);
                 hard_timeout_fate_suffix = Some(" — dead-lettered (no recent activity)");
             } else if matches!(
                 result.outcome,
@@ -3245,6 +3282,7 @@ fn handle_prompt_result(
                         config.max_turn_duration_secs
                     );
                     spawn_failure_notice(rest_client, &dead, content);
+                    queue.finish_batch(&dead);
                     hard_timeout_fate_suffix = Some(" — dead-lettered (retry budget exhausted)");
                 } else {
                     hard_timeout_fate_suffix = Some(" — requeued for retry (recently active)");
@@ -3264,6 +3302,7 @@ fn handle_prompt_result(
                     and then re-send."
                     .to_string();
                 spawn_failure_notice(rest_client, &batch, content);
+                queue.finish_batch(&batch);
             } else if let Some(dead) = queue.requeue(batch) {
                 let reason = match &result.outcome {
                     PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(),
@@ -3278,6 +3317,7 @@ fn handle_prompt_result(
                     "⚠️ I couldn't process the last request after multiple retries ({reason}). Please re-send if it's still needed."
                 );
                 spawn_failure_notice(rest_client, &dead, content);
+                queue.finish_batch(&dead);
             }
         } else {
             tracing::debug!(
@@ -3286,6 +3326,18 @@ fn handle_prompt_result(
                 "dropping failed batch for removed channel"
             );
             hard_timeout_fate_suffix = Some(" — batch dropped (channel removed)");
+            queue.finish_batch(&batch);
+        }
+    }
+
+    if matches!(result.outcome, PromptOutcome::Ok(_))
+        || (matches!(
+            result.outcome,
+            PromptOutcome::Cancelled | PromptOutcome::CancelDrainTimeout(_)
+        ) && !had_retry_batch)
+    {
+        if let Some(batch) = recoverable_batch.as_ref() {
+            queue.finish_batch(batch);
         }
     }
 
@@ -3547,7 +3599,9 @@ fn recover_panicked_agent(
             if !removed_channels.contains(&ch) {
                 // Dead-letter on exhaustion is logged inside requeue(); a
                 // panic path has no outcome to report, so no notice here.
-                let _ = queue.requeue(batch);
+                if let Some(dead) = queue.requeue(batch) {
+                    queue.finish_batch(&dead);
+                }
                 tracing::warn!("requeued batch for panicked agent {i}");
             } else {
                 tracing::debug!(

crates/buzz-acp/src/pending_store.rs

@@ -0,0 +1,195 @@
+//! Durable journal for channel events accepted by the ACP queue.
+//!
+//! The relay subscription watermark intentionally starts at process startup, so
+//! an event accepted immediately before a service restart is not replayed by
+//! the relay. This journal closes that gap: an event is written atomically
+//! before the harness publishes its "seen" reaction and is removed only after
+//! the turn succeeds or the user receives a terminal failure.
+
+use nostr::Event;
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+use std::io;
+use std::path::{Path, PathBuf};
+use uuid::Uuid;
+
+const STORE_VERSION: u32 = 1;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct StoredPendingEvent {
+    pub channel_id: Uuid,
+    pub event: Event,
+    pub prompt_tag: String,
+    pub accepted_at_nanos: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct StoreFile {
+    version: u32,
+    events: BTreeMap<String, StoredPendingEvent>,
+}
+
+pub(crate) struct PendingStore {
+    path: PathBuf,
+    events: BTreeMap<String, StoredPendingEvent>,
+}
+
+impl PendingStore {
+    pub(crate) fn open(agent_pubkey: &str) -> io::Result<Self> {
+        let path = pending_store_path(agent_pubkey)?;
+        Self::open_path(path)
+    }
+
+    pub(crate) fn open_path(path: PathBuf) -> io::Result<Self> {
+        let events = match std::fs::read(&path) {
+            Ok(bytes) => {
+                let stored: StoreFile = serde_json::from_slice(&bytes).map_err(|error| {
+                    io::Error::new(
+                        io::ErrorKind::InvalidData,
+                        format!("invalid pending-work journal {}: {error}", path.display()),
+                    )
+                })?;
+                if stored.version != STORE_VERSION {
+                    return Err(io::Error::new(
+                        io::ErrorKind::InvalidData,
+                        format!(
+                            "unsupported pending-work journal version {} in {}",
+                            stored.version,
+                            path.display()
+                        ),
+                    ));
+                }
+                stored.events
+            }
+            Err(error) if error.kind() == io::ErrorKind::NotFound => BTreeMap::new(),
+            Err(error) => return Err(error),
+        };
+        Ok(Self { path, events })
+    }
+
+    pub(crate) fn restored(&self) -> Vec<StoredPendingEvent> {
+        let mut events = self.events.values().cloned().collect::<Vec<_>>();
+        events.sort_by_key(|event| event.accepted_at_nanos);
+        events
+    }
+
+    pub(crate) fn record(&mut self, event: StoredPendingEvent) -> io::Result<()> {
+        let id = event.event.id.to_hex();
+        let previous = self.events.insert(id.clone(), event);
+        if let Err(error) = self.persist() {
+            match previous {
+                Some(previous) => {
+                    self.events.insert(id, previous);
+                }
+                None => {
+                    self.events.remove(&id);
+                }
+            }
+            return Err(error);
+        }
+        Ok(())
+    }
+
+    pub(crate) fn remove<'a>(&mut self, ids: impl IntoIterator<Item = &'a str>) -> io::Result<()> {
+        let removed: Vec<(String, StoredPendingEvent)> = ids
+            .into_iter()
+            .filter_map(|id| self.events.remove_entry(id))
+            .collect();
+        if removed.is_empty() {
+            return Ok(());
+        }
+        if let Err(error) = self.persist() {
+            self.events.extend(removed);
+            return Err(error);
+        }
+        Ok(())
+    }
+
+    fn persist(&self) -> io::Result<()> {
+        let parent = self.path.parent().ok_or_else(|| {
+            io::Error::new(io::ErrorKind::InvalidInput, "pending journal has no parent")
+        })?;
+        std::fs::create_dir_all(parent)?;
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt;
+            std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700))?;
+        }
+        let bytes = serde_json::to_vec(&StoreFile {
+            version: STORE_VERSION,
+            events: self.events.clone(),
+        })
+        .map_err(io::Error::other)?;
+        let temp = self.path.with_extension("json.tmp");
+        let mut options = std::fs::OpenOptions::new();
+        options.create(true).truncate(true).write(true);
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::OpenOptionsExt;
+            options.mode(0o600);
+        }
+        let mut file = options.open(&temp)?;
+        std::io::Write::write_all(&mut file, &bytes)?;
+        file.sync_all()?;
+        std::fs::rename(temp, &self.path)
+    }
+}
+
+fn pending_store_path(agent_pubkey: &str) -> io::Result<PathBuf> {
+    if let Some(path) = std::env::var_os("BUZZ_ACP_PENDING_STORE") {
+        return Ok(PathBuf::from(path));
+    }
+    let home = std::env::var_os("HOME").ok_or_else(|| {
+        io::Error::new(
+            io::ErrorKind::NotFound,
+            "HOME is unset and BUZZ_ACP_PENDING_STORE was not provided",
+        )
+    })?;
+    Ok(Path::new(&home)
+        .join(".local/state/buzz-acp")
+        .join(format!("pending-{agent_pubkey}.json")))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nostr::{EventBuilder, Keys};
+
+    fn event(content: &str) -> Event {
+        EventBuilder::text_note(content)
+            .sign_with_keys(&Keys::generate())
+            .expect("sign event")
+    }
+
+    #[test]
+    fn journal_round_trips_and_removes_atomically() {
+        let dir = std::env::temp_dir().join(format!("buzz-acp-pending-{}", Uuid::new_v4()));
+        std::fs::create_dir_all(&dir).expect("create tempdir");
+        let path = dir.join("pending.json");
+        let mut store = PendingStore::open_path(path.clone()).expect("open");
+        let event = event("survive restart");
+        let id = event.id.to_hex();
+        store
+            .record(StoredPendingEvent {
+                channel_id: Uuid::new_v4(),
+                event,
+                prompt_tag: "@mention".into(),
+                accepted_at_nanos: 1,
+            })
+            .expect("record");
+        drop(store);
+
+        let mut restored = PendingStore::open_path(path.clone()).expect("reopen");
+        assert_eq!(restored.restored().len(), 1);
+        restored.remove([id.as_str()]).expect("remove");
+        drop(restored);
+        assert_eq!(
+            PendingStore::open_path(path)
+                .expect("final reopen")
+                .restored()
+                .len(),
+            0
+        );
+        std::fs::remove_dir_all(dir).expect("remove tempdir");
+    }
+}