fix(acp): make accepted work durable and controllable

crates/buzz-acp/README.md

@@ -100,7 +100,7 @@ treats both Claude ACP command names as the same zero-arg runtime.
 
 ## Configuration
 
-All configuration is via environment variables (or CLI flags — every env var has a matching flag).
+Configuration is via environment variables and CLI flags. Runtime queue-journal overrides are environment-only because they are operational recovery controls, not normal launch policy.
 
 ### Core
 
@@ -113,6 +113,8 @@ All configuration is via environment variables (or CLI flags — every env var h
 | `BUZZ_ACP_MCP_COMMAND` | no | `""` (empty) | Path to an optional MCP server binary to provide to the agent subprocess. |
 | `BUZZ_ACP_IDLE_TIMEOUT` | no | `620` | Idle timeout: max seconds of silence before cancelling a turn. Resets on any agent stdout activity. |
 | `BUZZ_ACP_MAX_TURN_DURATION` | no | `7200` | Absolute wall-clock cap per turn (safety valve). |
+| `BUZZ_ACP_PENDING_STORE` | no | `$HOME/.local/state/buzz-acp/pending-<agent-pubkey>.json` | Override the per-agent durable pending/dead-letter journal path. |
+| `BUZZ_ACP_REPLAY_DEAD_LETTERS` | no | `false` | Move all retained dead letters back to pending work at startup. Enable for a deliberate replay, then unset it. |
 | `BUZZ_API_TOKEN` | no | — | API token (required if relay enforces token auth). |
 
 **Note:** `BUZZ_ACP_AGENT_ARGS` splits on commas. For args with values, use: `-c,key="value"`.
@@ -152,10 +154,11 @@ The gate applies to **all** inbound events — @mentions, DMs, thread replies, a
 | Command | Effect |
 |---------|--------|
 | `!shutdown` | Gracefully exits the harness. |
-| `!cancel` | Cancels the current in-flight turn for that channel, if any. |
+| `!status`, `status`, or `status?` | Reports whether a turn is active and how many follow-ups are queued without disturbing the active turn. A single leading display mention is accepted. |
+| `!cancel`, `!stop`, or `stop` | Cancels the current in-flight turn for that channel, if any. A single leading display mention is accepted. |
 | `!rotate` | Rotates the ACP session for that channel. If a turn is in-flight, it is cancelled and the channel session is invalidated when the task returns; otherwise the cached idle session is invalidated immediately. The next queued/received event starts a fresh session. |
 
-Use `!cancel` to stop only the current turn; it is a no-op when the channel is idle. Use `!rotate` when you want the next turn in the channel to start from a fresh ACP session, even if the channel is currently idle.
+Use `!cancel` or `!stop` to stop only the current turn; it is a no-op when the channel is idle. Use `!rotate` when you want the next turn in the channel to start from a fresh ACP session, even if the channel is currently idle.
 
 Owner control commands must be kind:9 stream messages from the owner, must mention this agent with a `p` tag, and are consumed by the harness instead of being forwarded to the agent.
 
@@ -255,11 +258,11 @@ Forum event kinds:
 3. **Event loop** — Listens for @mention events (kind 9 with the agent's pubkey in a `#p` tag). Events queue per channel.
 4. **Prompting** — When events are pending and no prompt is in flight for that channel, drains all queued events for the oldest channel into a single batched prompt via ACP `session/prompt`.
 5. **Agent response** — The agent processes the prompt and uses the Buzz CLI (`send_message`, `get_messages`, etc.) to interact with Buzz.
-6. **Recovery** — If the agent crashes, the harness respawns it. If the relay disconnects, the harness reconnects with a `since` filter to avoid missing events.
+6. **Recovery** — Accepted events are atomically journaled before the harness acknowledges them. A process restart restores unfinished work from that journal. Terminal failures remain in durable dead-letter storage for audit or deliberate replay. If the relay disconnects, the harness reconnects with a `since` filter to avoid missing events that were not yet accepted.
 
 Each channel has at most one prompt in flight. Multiple channels can be processed concurrently when agents > 1.
 
-> **Note:** On startup, the harness replays all unprocessed @mentions since the last run. Expect a burst of activity if there are stale events in the channel.
+> **Note:** On startup, the harness replays unfinished accepted work from its per-agent journal. Set `BUZZ_ACP_REPLAY_DEAD_LETTERS=1` only for a deliberate dead-letter replay, then unset it to avoid replaying future failures after every restart.
 
 ## Bring Your Own Harness (BYOH)
 

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;
@@ -2135,8 +2136,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
@@ -2746,19 +2757,72 @@ async fn tokio_main() -> Result<()> {
                                 // contain "!shutdown" from a non-owner.
                             }
 
+                            // Owner-only, mode-independent status lane. Exact
+                            // commands are answered by the harness so checking
+                            // progress never cancels, steers, or queues behind
+                            // the active model turn.
+                            let is_status = ["!status", "status", "status?"]
+                                .iter()
+                                .any(|command| {
+                                    is_owner_control_phrase(
+                                        &buzz_event.event,
+                                        kind_u32,
+                                        command,
+                                        &pubkey_hex,
+                                    )
+                                });
+                            if is_status {
+                                if let Some(owner) = owner_cache.get() {
+                                    if buzz_event.event.pubkey.to_hex() == *owner {
+                                        let active = queue
+                                            .is_channel_in_flight(buzz_event.channel_id);
+                                        let queued =
+                                            queue.queued_event_count(&buzz_event.channel_id);
+                                        let content = if active {
+                                            format!(
+                                                "Status: the current task is still running; {queued} follow-up request(s) are durably queued. Send !stop to stop the active task."
+                                            )
+                                        } else {
+                                            format!(
+                                                "Status: no task is currently running; {queued} request(s) are durably queued."
+                                            )
+                                        };
+                                        let thread_tags =
+                                            queue::parse_thread_tags(&buzz_event.event);
+                                        let rest = ctx.rest_client.clone();
+                                        let channel_id = buzz_event.channel_id;
+                                        tokio::spawn(async move {
+                                            pool::post_failure_notice(
+                                                &rest,
+                                                channel_id,
+                                                &thread_tags,
+                                                &content,
+                                            )
+                                            .await;
+                                        });
+                                        continue;
+                                    }
+                                }
+                                // Not from owner — fall through to normal prompt handling.
+                            }
+
                             // Mirrors !shutdown: kind:9, content "!cancel", from
                             // owner, mentions THIS agent. Must be BEFORE
                             // queue.push() — the event content is moved by push.
                             //
                             // Mode-independent: !cancel fires regardless of
                             // --multiple-event-handling. It is explicit user
                             // intent, not an automatic policy decision.
-                            let is_cancel = is_owner_control_command(
-                                &buzz_event.event,
-                                kind_u32,
-                                "!cancel",
-                                &pubkey_hex,
-                            );
+                            let is_cancel = ["!cancel", "!stop", "stop"]
+                                .iter()
+                                .any(|command| {
+                                    is_owner_control_phrase(
+                                        &buzz_event.event,
+                                        kind_u32,
+                                        command,
+                                        &pubkey_hex,
+                                    )
+                                });
                             if is_cancel {
                                 if let Some(owner) = owner_cache.get() {
                                     if buzz_event.event.pubkey.to_hex() == *owner {
@@ -2770,7 +2834,7 @@ async fn tokio_main() -> Result<()> {
                                         if !fired {
                                             tracing::warn!(
                                                 channel_id = %buzz_event.channel_id,
-                                                "!cancel received but no in-flight task — no-op"
+                                                "stop command received but no in-flight task — no-op"
                                             );
                                         }
                                         continue; // consume event — do NOT push to queue
@@ -3413,6 +3477,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 {
@@ -3427,6 +3500,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");
                     }
@@ -3444,6 +3522,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");
     }
@@ -3528,6 +3611,27 @@ fn is_owner_control_command(
         && event_mentions_agent(event, agent_pubkey_hex)
 }
 
+fn is_owner_control_phrase(
+    event: &nostr::Event,
+    kind_u32: u32,
+    command: &str,
+    agent_pubkey_hex: &str,
+) -> bool {
+    if kind_u32 != KIND_STREAM_MESSAGE || !event_mentions_agent(event, agent_pubkey_hex) {
+        return false;
+    }
+    let content = event.content.trim();
+    if content.eq_ignore_ascii_case(command) {
+        return true;
+    }
+    let mut words = content.split_whitespace();
+    matches!(
+        (words.next(), words.next(), words.next()),
+        (Some(mention), Some(candidate), None)
+            if mention.starts_with('@') && candidate.eq_ignore_ascii_case(command)
+    )
+}
+
 // ── signal_in_flight_task ─────────────────────────────────────────────────────
 
 /// Decide which [`ControlSignal`] (if any) to send to an in-flight turn when a
@@ -3849,6 +3953,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;
     let successful_steer_deliveries = pool
@@ -3923,14 +4033,15 @@ fn handle_prompt_result(
                 tracing::error!(
                     channel_id = %batch.channel_id,
                     events = batch.events.len(),
-                    "dead-lettering batch after hard-cap timeout (no recent activity) — discarding {} events",
+                    "dead-lettering batch after hard-cap timeout (no recent activity) — retaining {} events in durable dead-letter storage",
                     batch.events.len(),
                 );
                 let content = format!(
                     "⚠️ I couldn't process the last request (the turn exceeded the maximum duration ({}s)). Please re-send if it's still needed.",
                     config.max_turn_duration_secs
                 );
                 spawn_failure_notice(rest_client, &batch, content);
+                queue.dead_letter_batch(&batch, "hard-cap timeout without recent activity");
                 hard_timeout_fate_suffix = Some(" — dead-lettered (no recent activity)");
             } else if matches!(
                 result.outcome,
@@ -3949,6 +4060,7 @@ fn handle_prompt_result(
                         config.max_turn_duration_secs
                     );
                     spawn_failure_notice(rest_client, &dead, content);
+                    queue.dead_letter_batch(&dead, "retry budget exhausted after hard-cap timeout");
                     hard_timeout_fate_suffix = Some(" — dead-lettered (retry budget exhausted)");
                 } else {
                     hard_timeout_fate_suffix = Some(" — requeued for retry (recently active)");
@@ -3968,6 +4080,7 @@ fn handle_prompt_result(
                     and then re-send."
                     .to_string();
                 spawn_failure_notice(rest_client, &batch, content);
+                queue.dead_letter_batch(&batch, "non-retryable authentication error");
             } else if let Some(dead) = queue.requeue(batch) {
                 let reason = match &result.outcome {
                     PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(),
@@ -3982,6 +4095,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.dead_letter_batch(&dead, "retry budget exhausted");
             }
         } else {
             tracing::debug!(
@@ -3990,6 +4104,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);
         }
     }
 
@@ -4251,7 +4377,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!(
@@ -5273,6 +5401,26 @@ mod owner_control_command_tests {
         );
         assert!(unaddressed.is_err());
     }
+
+    #[test]
+    fn owner_control_phrase_accepts_a_single_display_mention_prefix() {
+        let agent = "ab".repeat(32);
+        let prefixed = make_event(KIND_STREAM_MESSAGE, "@Jarvis status?", Some(&agent));
+        assert!(is_owner_control_phrase(
+            &prefixed,
+            KIND_STREAM_MESSAGE,
+            "status?",
+            &agent
+        ));
+
+        let conversational = make_event(KIND_STREAM_MESSAGE, "@Jarvis please stop", Some(&agent));
+        assert!(!is_owner_control_phrase(
+            &conversational,
+            KIND_STREAM_MESSAGE,
+            "stop",
+            &agent
+        ));
+    }
 }
 
 #[cfg(test)]