diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 6c272187f3c..3b8d3823446 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -4497,6 +4497,7 @@ fn dispatch_pending( } tracing::debug!(agent = agent.index, channel = %channel_id, scope = %scope.telemetry_label(), affinity_hit, "agent_claimed"); + let triggering = queue::triggering_event_context(&batch); let recoverable_batch = match ctx.dedup_mode { DedupMode::Queue => Some(batch.clone()), DedupMode::Drop => None, @@ -4553,6 +4554,7 @@ fn dispatch_pending( channel_id: Some(channel_id), scope: Some(scope.clone()), turn_id, + triggering, recoverable_batch, control_tx: Some(control_tx), steer_tx, @@ -4647,6 +4649,12 @@ fn handle_prompt_result( ) -> LoopAction { let before = pool.task_map().len(); let agent_index = result.agent.index; + let task_triggering = pool + .task_map() + .values() + .find(|meta| meta.agent_index == agent_index) + .map(|meta| meta.triggering.clone()) + .unwrap_or_default(); let successful_steer_deliveries = pool .task_map() .values() @@ -4673,6 +4681,21 @@ fn handle_prompt_result( } } + // Capture conversation correlation before the batch is moved into retry / + // dead-letter handling. Terminal observer events repeat this context so a + // client can render a failure at its originating message without replaying + // or joining an earlier turn_started frame. + let triggering = result + .batch + .as_ref() + .map(queue::triggering_event_context) + .unwrap_or(task_triggering); + let mut attempt = result + .batch + .as_ref() + .map(|batch| queue.retry_count(batch.scope.clone()) + 1); + let mut disposition = "stopped"; + // The hard-timeout death_message (below) must describe the batch's // *actual* fate, not just the `recently_active` eligibility flag — a // recently-active batch that exhausts the retry budget in queue.requeue() @@ -4711,6 +4734,7 @@ fn handle_prompt_result( // accounting, same as a clean cancel. let reason = batch.cancel_reason.unwrap_or(CancelReason::Steer); queue.requeue_as_cancelled(batch, reason); + disposition = "retrying"; } else if matches!( result.outcome, PromptOutcome::Timeout(TimeoutKind::Hard { @@ -4729,6 +4753,7 @@ fn handle_prompt_result( ); spawn_failure_notice(rest_client, &batch, content); hard_timeout_fate_suffix = Some(" — dead-lettered (no recent activity)"); + disposition = "dead_lettered"; } else if matches!( result.outcome, PromptOutcome::Timeout(TimeoutKind::Hard { @@ -4747,8 +4772,10 @@ fn handle_prompt_result( ); spawn_failure_notice(rest_client, &dead, content); hard_timeout_fate_suffix = Some(" — dead-lettered (retry budget exhausted)"); + disposition = "dead_lettered"; } else { hard_timeout_fate_suffix = Some(" — requeued for retry (recently active)"); + disposition = "retrying"; } } else if matches!( &result.outcome, @@ -4767,6 +4794,7 @@ fn handle_prompt_result( to apply the new configuration, then re-send your request." .to_string(); spawn_failure_notice(rest_client, &batch, content); + disposition = "action_required"; } else if matches!(&result.outcome, PromptOutcome::Error(e) if is_auth_error(e)) { // Auth errors are non-retryable: the token won't self-repair // between retries, so requeueing only wastes attempt slots and @@ -4782,6 +4810,7 @@ fn handle_prompt_result( and then re-send." .to_string(); spawn_failure_notice(rest_client, &batch, content); + disposition = "action_required"; } else if let Some(dead) = queue.requeue(batch) { let reason = match &result.outcome { PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(), @@ -4797,6 +4826,9 @@ 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); + disposition = "dead_lettered"; + } else { + disposition = "retrying"; } } else { tracing::debug!( @@ -4805,6 +4837,8 @@ fn handle_prompt_result( "dropping failed batch for removed channel" ); hard_timeout_fate_suffix = Some(" — batch dropped (channel removed)"); + disposition = "stopped"; + attempt = None; } } @@ -4847,12 +4881,25 @@ fn handle_prompt_result( let channel_id = result.source.channel_id(); let turn_id = result.turn_id.clone(); - let emit_turn_error = |error_msg: &str, error_code: Option| { + let emit_turn_error = |error_msg: &str, + error_code: Option, + respawn_scheduled: Option, + reported_disposition: &str| { if let Some(ref observer) = observer { let mut payload = serde_json::json!({ "outcome": outcome_label, "error": error_msg, + "disposition": reported_disposition, + "triggeringEventIds": triggering.event_ids, + "triggeringRootEventId": triggering.root_event_id, + "triggeringParentEventId": triggering.parent_event_id, }); + if let Some(attempt) = attempt { + payload["attempt"] = serde_json::json!(attempt); + } + if let Some(scheduled) = respawn_scheduled { + payload["respawnScheduled"] = serde_json::json!(scheduled); + } if let Some(code) = error_code { payload["code"] = serde_json::json!(code); } @@ -4898,23 +4945,31 @@ fn handle_prompt_result( } _ => "Agent session timed out due to inactivity".to_string(), }; - emit_turn_error(&death_message, None); - let index = result.agent.index; - let slot_history = &mut crash_history[index]; - if !spawn_respawn_task( + let respawn_scheduled = spawn_respawn_task( result.agent, config, - slot_history, + &mut crash_history[index], respawn_tx, respawn_tasks, observer.clone(), - ) { - // Circuit open — slot stays empty until maintenance refill. - if pool.live_count() == 0 && !any_respawn_in_flight(crash_history) { - tracing::error!("all agents dead — exiting"); - return LoopAction::Exit; - } + ); + let exiting = !respawn_scheduled + && pool.live_count() == 0 + && !any_respawn_in_flight(crash_history); + emit_turn_error( + &death_message, + None, + Some(respawn_scheduled), + if exiting && disposition == "retrying" { + "stopped" + } else { + disposition + }, + ); + if exiting { + tracing::error!("all agents dead — exiting"); + return LoopAction::Exit; } } // Cancel-drain expiry: a control-signal cancel (steer fallback, @@ -4935,28 +4990,41 @@ fn handle_prompt_result( grace = ?grace, "agent_returned — respawning (cancel-drain timeout)" ); - let death_message = format!( - "Agent did not stop within {grace:?} after cancellation; the agent process is being replaced." - ); - emit_turn_error(&death_message, None); - let index = result.agent.index; - let slot_history = &mut crash_history[index]; - if !spawn_respawn_task( + let respawn_scheduled = spawn_respawn_task( result.agent, config, - slot_history, + &mut crash_history[index], respawn_tx, respawn_tasks, observer.clone(), - ) { - // Circuit open — slot stays empty until maintenance refill. - if pool.live_count() == 0 && !any_respawn_in_flight(crash_history) { - tracing::error!("all agents dead — exiting"); - return LoopAction::Exit; - } + ); + let replacement = if respawn_scheduled { + "the agent process is being replaced" + } else { + "the agent restart is delayed by the crash circuit breaker" + }; + let death_message = + format!("Agent did not stop within {grace:?} after cancellation; {replacement}."); + let exiting = !respawn_scheduled + && pool.live_count() == 0 + && !any_respawn_in_flight(crash_history); + emit_turn_error( + &death_message, + None, + Some(respawn_scheduled), + if exiting && disposition == "retrying" { + "stopped" + } else { + disposition + }, + ); + if exiting { + tracing::error!("all agents dead — exiting"); + return LoopAction::Exit; } } + // Errors fall into two categories: // // 1. Transport-class (Io, WriteTimeout, Timeout, Protocol): the stdio @@ -4989,7 +5057,7 @@ fn handle_prompt_result( reason, "agent_returned (local project context indeterminate — pipe intact)" ); - emit_turn_error(&reason, None); + emit_turn_error(&reason, None, None, disposition); pool.return_agent(result.agent); } PromptOutcome::Error(ref e) => { @@ -5013,20 +5081,29 @@ fn handle_prompt_result( error = %e, "transport/protocol error — respawning agent" ); - emit_turn_error(&e.to_string(), error_code); - let index = result.agent.index; - let slot_history = &mut crash_history[index]; - if !spawn_respawn_task( + let respawn_scheduled = spawn_respawn_task( result.agent, config, - slot_history, + &mut crash_history[index], respawn_tx, respawn_tasks, - observer, - ) && pool.live_count() == 0 - && !any_respawn_in_flight(crash_history) - { + observer.clone(), + ); + let exiting = !respawn_scheduled + && pool.live_count() == 0 + && !any_respawn_in_flight(crash_history); + emit_turn_error( + &e.to_string(), + error_code, + Some(respawn_scheduled), + if exiting && disposition == "retrying" { + "stopped" + } else { + disposition + }, + ); + if exiting { tracing::error!("all agents dead — exiting"); return LoopAction::Exit; } @@ -5039,7 +5116,7 @@ fn handle_prompt_result( error = %e, "agent_returned (application error — pipe intact)" ); - emit_turn_error(&e.to_string(), error_code); + emit_turn_error(&e.to_string(), error_code, None, disposition); pool.return_agent(result.agent); } } @@ -5068,14 +5145,27 @@ fn recover_panicked_agent( }; let i = meta.agent_index; + let triggering = meta + .recoverable_batch + .as_ref() + .map(queue::triggering_event_context) + .unwrap_or(meta.triggering); + let attempt = meta + .recoverable_batch + .as_ref() + .map(|batch| queue.retry_count(batch.scope.clone()) + 1); + let mut disposition = "stopped"; + let had_batch = meta.recoverable_batch.is_some(); // Requeue BEFORE mark_complete (same rationale as handle_prompt_result). if let Some(batch) = meta.recoverable_batch { if let Some(ch) = meta.channel_id { 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); - tracing::warn!("requeued batch for panicked agent {i}"); + disposition = if queue.requeue(batch).is_some() { + "dead_lettered" + } else { + "retrying" + }; + tracing::warn!(agent = i, disposition, "recovered panicked batch"); } else { tracing::debug!( channel_id = %ch, @@ -5109,31 +5199,17 @@ fn recover_panicked_agent( tracing::warn!("cleared wedged heartbeat_in_flight from panicked agent {i}"); } - if let Some(ref observer) = observer { - observer.emit( - "agent_panic", - Some(i), - &observer::context_for(meta.channel_id, None, Some(meta.turn_id)), - serde_json::json!({ - "outcome": "panic", - "error": format!("Agent task panicked: {join_error}"), - }), - ); - } - // Panics count as crashes for the circuit breaker. // The panicked task already dropped the AcpClient, so we just need to // check the circuit and spawn a fresh agent in the background. - let slot = &mut crash_history[i]; - - let delay = match slot.record_crash() { + let delay = match crash_history[i].record_crash() { CrashVerdict::CircuitOpen => { tracing::error!(agent = i, "circuit open after panic — not respawning"); - return; + None } CrashVerdict::HalfOpenProbe => { tracing::info!(agent = i, "circuit half-open — probe respawn after panic"); - Duration::ZERO + Some(Duration::ZERO) } CrashVerdict::Respawn(d) => { tracing::info!( @@ -5141,12 +5217,50 @@ fn recover_panicked_agent( delay_ms = d.as_millis(), "respawn backoff after panic" ); - d + Some(d) } }; + if disposition == "retrying" + && delay.is_none() + && pool.live_count() == 0 + && !any_respawn_in_flight(crash_history) + { + // The caller exits in this state; queued memory will not survive. + disposition = "stopped"; + } + if !had_batch + && delay.is_some() + && !meta + .channel_id + .is_some_and(|ch| removed_channels.contains(&ch)) + { + disposition = "respawning"; + } + if let Some(ref observer) = observer { + observer.emit( + "agent_panic", + Some(i), + &observer::context_for(meta.channel_id, None, Some(meta.turn_id)), + serde_json::json!({ + "outcome": "panic", + "error": format!("Agent task panicked: {join_error}"), + "disposition": disposition, + "attempt": attempt, + "respawnScheduled": delay.is_some(), + "triggeringEventIds": triggering.event_ids, + "triggeringRootEventId": triggering.root_event_id, + "triggeringParentEventId": triggering.parent_event_id, + }), + ); + } + + let Some(delay) = delay else { + return; + }; + // Spawn respawn work off the main loop. - slot.respawn_in_flight = true; + crash_history[i].respawn_in_flight = true; let cmd = config.agent_command.clone(); let args = config.agent_args.clone(); let env = config.persona_env_vars.clone(); @@ -5241,6 +5355,7 @@ fn dispatch_heartbeat( channel_id: None, scope: None, turn_id, + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -6057,6 +6172,7 @@ mod owner_control_command_tests { channel_id: Some(channel_id), scope: Some(scope::SessionScope::Conversation { channel_id }), turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: Some(control_tx), steer_tx: None, @@ -6103,6 +6219,7 @@ mod owner_control_command_tests { channel_id: Some(scope.channel_id()), scope: Some(scope), turn_id: "t".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: Some(control_tx), steer_tx: None, @@ -9472,6 +9589,7 @@ mod error_outcome_emission_tests { channel_id: Some(channel_id), scope: Some(scope::SessionScope::Conversation { channel_id }), turn_id: "test-turn-id".into(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -9552,6 +9670,7 @@ mod error_outcome_emission_tests { channel_id: Some(channel_id), scope: Some(scope::SessionScope::Conversation { channel_id }), turn_id: "test-turn-id".into(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -9674,6 +9793,7 @@ mod error_outcome_emission_tests { channel_id: Some(channel_id), scope: Some(scope::SessionScope::Conversation { channel_id }), turn_id: "test-turn-id".into(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -9743,6 +9863,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -9806,6 +9927,466 @@ mod error_outcome_emission_tests { assert_eq!(turn_errors_emitted_for(PromptOutcome::AgentExited).await, 1); } + #[tokio::test] + async fn fatal_failure_reports_queue_fate_and_actual_respawn_decision() { + for outcome_kind in [ + "exited", + "idle", + "hard_active", + "hard_inactive", + "transport", + ] { + for batch_fate in ["retry", "exhausted", "removed"] { + for capacity in ["none", "idle", "busy", "respawning", "scheduled"] { + let case = format!("{outcome_kind}/{batch_fate}/{capacity}"); + let channel_id = Uuid::new_v4(); + let root = "a".repeat(64); + let parent = "b".repeat(64); + let event = EventBuilder::new(Kind::Custom(9), "fatal request") + .tags([ + nostr::Tag::parse(["e", root.as_str(), "", "root"]).unwrap(), + nostr::Tag::parse(["e", parent.as_str(), "", "reply"]).unwrap(), + ]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let scope = scope::SessionScope::Thread { + channel_id, + root_event_id: root.clone(), + }; + let mut queue = EventQueue::new(config::DedupMode::Queue); + queue.push(QueuedEvent { + channel_id, + scope: scope.clone(), + event: event.clone(), + received_at: std::time::Instant::now(), + prompt_tag: "test".into(), + }); + let batch = queue.flush_next().unwrap(); + if batch_fate == "exhausted" { + queue.set_retry_count_for_test(scope.clone(), queue::MAX_RETRIES); + } + let agent = dummy_agent(0).await; + let sibling = if capacity == "idle" { + Some(dummy_agent(1).await) + } else { + None + }; + let mut pool = AgentPool::from_slots(vec![None, sibling]); + let task = pool.join_set.spawn(async {}); + pool.task_map_mut().insert( + task.id(), + pool::TaskMeta { + agent_index: 0, + channel_id: Some(channel_id), + scope: Some(scope.clone()), + turn_id: "fatal-turn".into(), + triggering: queue::triggering_event_context(&batch), + recoverable_batch: Some(batch.clone()), + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::new(), + }, + ); + if capacity == "busy" { + let sibling_task = pool.join_set.spawn(std::future::pending()); + pool.task_map_mut().insert( + sibling_task.id(), + pool::TaskMeta { + agent_index: 1, + channel_id: None, + scope: None, + turn_id: "sibling-turn".into(), + triggering: Default::default(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::new(), + }, + ); + } + let mut crash_history = vec![ + SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: false, + }, + SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: capacity == "respawning", + }, + ]; + // The completing failure itself opens the circuit, while + // the triggering batch can still have retry budget left. + if capacity != "scheduled" { + for _ in 0..CIRCUIT_BREAKER_THRESHOLD - 1 { + assert!(matches!( + crash_history[0].record_crash(), + CrashVerdict::Respawn(_) + )); + } + } + let outcome = match outcome_kind { + "exited" => PromptOutcome::AgentExited, + "idle" => PromptOutcome::Timeout(TimeoutKind::Idle), + "hard_active" => PromptOutcome::Timeout(TimeoutKind::Hard { + recently_active: true, + }), + "hard_inactive" => PromptOutcome::Timeout(TimeoutKind::Hard { + recently_active: false, + }), + "transport" => { + PromptOutcome::Error(AcpError::Protocol("broken pipe".into())) + } + _ => unreachable!(), + }; + let (respawn_tx, _respawn_rx) = mpsc::channel(8); + let mut respawn_tasks = tokio::task::JoinSet::new(); + let observer = ObserverHandle::in_process(); + let removed = batch_fate == "removed"; + let removed_channels = if removed { + HashSet::from([channel_id]) + } else { + HashSet::new() + }; + let action = handle_prompt_result( + &mut pool, + &mut queue, + &test_config(), + PromptResult { + agent, + source: PromptSource::Channel(scope.clone()), + turn_id: "fatal-turn".into(), + outcome, + batch: Some(batch), + }, + &mut false, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + Some(observer.clone()), + None, + ); + let dead_lettered = + !removed && (batch_fate == "exhausted" || outcome_kind == "hard_inactive"); + let exiting = capacity == "none"; + let scheduled = capacity == "scheduled"; + let events = observer.snapshot(); + let failures: Vec<_> = events + .iter() + .filter(|event| event.kind == "turn_error") + .collect(); + assert_eq!(failures.len(), 1, "{case}"); + let failure = failures[0]; + assert_eq!( + failure.payload["disposition"], + if dead_lettered { + "dead_lettered" + } else if removed || exiting { + "stopped" + } else { + "retrying" + }, + "{case}" + ); + assert_eq!(failure.payload["respawnScheduled"], scheduled, "{case}"); + assert_eq!(respawn_tasks.len(), usize::from(scheduled), "{case}"); + assert_eq!(crash_history[0].respawn_in_flight, scheduled, "{case}"); + assert_eq!(action == LoopAction::Exit, exiting, "{case}"); + assert_eq!(failure.turn_id.as_deref(), Some("fatal-turn"), "{case}"); + assert_eq!(failure.payload["triggeringRootEventId"], root, "{case}"); + assert_eq!(failure.payload["triggeringParentEventId"], parent, "{case}"); + assert_eq!( + failure.payload["triggeringEventIds"], + serde_json::json!([event.id.to_hex()]), + "{case}" + ); + assert!(!queue.is_scope_in_flight(&scope), "{case}"); + assert_eq!( + queue.queued_event_count(scope.clone()), + usize::from(!removed && !dead_lettered), + "{case}" + ); + if !removed && !dead_lettered { + assert_eq!(queue.retry_count(scope), 1, "{case}"); + assert_eq!( + queue.drain_channel(channel_id), + vec![event.id.to_hex()], + "{case}" + ); + } + } + } + } + } + + // No turn_started frame is emitted in these tests: terminal context must + // stand alone when an owner subscribes after a long turn has started. + #[tokio::test] + async fn panic_recovery_reports_queue_fate_and_actual_respawn_decision() { + for (recoverable, exhausted, removed, circuit_open, capacity, expected) in [ + (true, false, false, false, "none", "retrying"), + (true, true, false, false, "none", "dead_lettered"), + (true, false, true, false, "none", "stopped"), + (true, false, false, true, "none", "stopped"), + (true, false, false, true, "idle", "retrying"), + (true, false, false, true, "respawning", "retrying"), + (false, false, true, false, "none", "stopped"), + (false, false, false, false, "none", "respawning"), + (false, false, false, true, "none", "stopped"), + ] { + let channel_id = Uuid::new_v4(); + let root = "a".repeat(64); + let parent = "b".repeat(64); + let event = EventBuilder::new(Kind::Custom(9), "panic request") + .tags([ + nostr::Tag::parse(["e", root.as_str(), "", "root"]).unwrap(), + nostr::Tag::parse(["e", parent.as_str(), "", "reply"]).unwrap(), + ]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let scope = scope::SessionScope::Thread { + channel_id, + root_event_id: root.clone(), + }; + let mut queue = EventQueue::new(config::DedupMode::Queue); + queue.push(QueuedEvent { + channel_id, + scope: scope.clone(), + event: event.clone(), + received_at: std::time::Instant::now(), + prompt_tag: "test".into(), + }); + let batch = queue.flush_next().unwrap(); + if exhausted { + queue.set_retry_count_for_test(scope.clone(), queue::MAX_RETRIES); + } + let sibling = if capacity == "idle" { + Some(dummy_agent(1).await) + } else { + None + }; + let mut pool = AgentPool::from_slots(vec![None, sibling]); + let task = pool.join_set.spawn(async { + panic!("test panic"); + }); + pool.task_map_mut().insert( + task.id(), + pool::TaskMeta { + agent_index: 0, + channel_id: Some(channel_id), + scope: Some(scope.clone()), + turn_id: "panic-turn".into(), + triggering: queue::triggering_event_context(&batch), + recoverable_batch: recoverable.then_some(batch), + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::new(), + }, + ); + let join_error = pool.join_set.join_next().await.unwrap().unwrap_err(); + let mut crash_history = vec![SlotCircuit { + crash_times: Vec::new(), + open_until: circuit_open + .then(|| std::time::Instant::now() + Duration::from_secs(60)), + respawn_in_flight: false, + }]; + crash_history.push(SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: capacity == "respawning", + }); + let (respawn_tx, _respawn_rx) = mpsc::channel(8); + let mut respawn_tasks = tokio::task::JoinSet::new(); + let observer = ObserverHandle::in_process(); + let removed_channels = if removed { + HashSet::from([channel_id]) + } else { + HashSet::new() + }; + recover_panicked_agent( + &mut pool, + &mut queue, + &test_config(), + join_error, + &mut false, + &removed_channels, + &mut HashMap::new(), + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + Some(observer.clone()), + ); + let events = observer.snapshot(); + let panic = events + .iter() + .find(|event| event.kind == "agent_panic") + .unwrap(); + assert_eq!(panic.turn_id.as_deref(), Some("panic-turn")); + assert_eq!(panic.payload["disposition"], expected); + assert_eq!( + panic.payload["triggeringEventIds"], + serde_json::json!([event.id.to_hex()]) + ); + assert_eq!(panic.payload["triggeringRootEventId"], root); + assert_eq!(panic.payload["triggeringParentEventId"], parent); + assert_eq!(panic.payload["respawnScheduled"], !circuit_open); + assert_eq!(respawn_tasks.len(), usize::from(!circuit_open)); + assert_eq!( + queue.queued_event_count(scope.clone()), + usize::from(recoverable && !exhausted && !removed) + ); + assert!(!queue.is_scope_in_flight(&scope)); + if recoverable && !exhausted && !removed { + assert_eq!(queue.drain_channel(channel_id), vec![event.id.to_hex()]); + } + if recoverable { + assert_eq!( + panic.payload["attempt"], + if exhausted { queue::MAX_RETRIES + 1 } else { 1 } + ); + } + } + } + + /// Called with the actual production cancellation classifier's output. + pub(super) async fn assert_cancel_failure_metadata( + outcome: PromptOutcome, + batch: Option, + original: FlushBatch, + removed: bool, + circuit_open: bool, + capacity: &str, + ) { + let channel_id = original.channel_id; + let triggering = queue::triggering_event_context(&original); + let preserved = batch.is_some() && !removed; + let agent = dummy_agent(0).await; + let sibling = if capacity == "idle" { + Some(dummy_agent(1).await) + } else { + None + }; + let mut pool = AgentPool::from_slots(vec![None, sibling]); + let task = pool.join_set.spawn(async {}); + pool.task_map_mut().insert( + task.id(), + pool::TaskMeta { + agent_index: 0, + channel_id: Some(channel_id), + scope: Some(original.scope.clone()), + turn_id: "cancel-turn".into(), + triggering: triggering.clone(), + recoverable_batch: Some(original.clone()), + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::new(), + }, + ); + let mut queue = EventQueue::new(config::DedupMode::Queue); + let mut crash_history = vec![SlotCircuit { + crash_times: Vec::new(), + open_until: circuit_open.then(|| std::time::Instant::now() + Duration::from_secs(60)), + respawn_in_flight: false, + }]; + crash_history.push(SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: capacity == "respawning", + }); + let (respawn_tx, _respawn_rx) = mpsc::channel(8); + let mut respawn_tasks = tokio::task::JoinSet::new(); + let observer = ObserverHandle::in_process(); + let removed_channels = if removed { + HashSet::from([channel_id]) + } else { + HashSet::new() + }; + let action = handle_prompt_result( + &mut pool, + &mut queue, + &test_config(), + PromptResult { + agent, + source: PromptSource::Channel(original.scope.clone()), + turn_id: "cancel-turn".into(), + outcome, + batch, + }, + &mut false, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + Some(observer.clone()), + None, + ); + assert!(matches!(action, LoopAction::Exit) == (circuit_open && capacity == "none")); + let events = observer.snapshot(); + let failure = events + .iter() + .find(|event| event.kind == "turn_error") + .unwrap(); + assert_eq!( + failure.payload["disposition"], + if preserved && (!circuit_open || capacity != "none") { + "retrying" + } else { + "stopped" + } + ); + assert_eq!( + failure.payload["triggeringEventIds"], + serde_json::json!(triggering.event_ids) + ); + assert_eq!( + failure.payload["triggeringRootEventId"], + serde_json::json!(triggering.root_event_id) + ); + assert_eq!( + failure.payload["triggeringParentEventId"], + serde_json::json!(triggering.parent_event_id) + ); + assert_eq!(failure.turn_id.as_deref(), Some("cancel-turn")); + assert_eq!(failure.payload["respawnScheduled"], !circuit_open); + assert_eq!(respawn_tasks.len(), usize::from(!circuit_open)); + assert_eq!( + failure.payload["error"] + .as_str() + .unwrap() + .contains("is being replaced"), + !circuit_open + ); + // Seed the new request which causes cancelled work to merge on flush. + let new_event = EventBuilder::new(Kind::Custom(9), "follow-up") + .sign_with_keys(&Keys::generate()) + .unwrap(); + queue.push(QueuedEvent { + channel_id, + scope: original.scope, + event: new_event, + received_at: std::time::Instant::now(), + prompt_tag: "test".into(), + }); + let next = queue.flush_next().unwrap(); + assert_eq!( + next.cancelled_events.len(), + if preserved { original.events.len() } else { 0 } + ); + if preserved { + assert_eq!( + next.cancelled_events[0].event.id, + original.events[0].event.id + ); + } + assert_eq!( + queue.retry_count(channel_id), + 0, + "cancellation never consumes a retry" + ); + } + #[tokio::test] async fn panic_event_retains_task_turn_id() { let mut pool = AgentPool::from_slots(vec![]); @@ -9823,6 +10404,7 @@ mod error_outcome_emission_tests { channel_id: Some(channel_id), scope: Some(scope::SessionScope::Conversation { channel_id }), turn_id: "panic-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -9915,6 +10497,7 @@ mod error_outcome_emission_tests { channel_id: Some(channel_id), scope: Some(scope.clone()), turn_id: "panic-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: Some(batch), control_tx: None, steer_tx: None, @@ -10014,6 +10597,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -10111,6 +10695,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -10219,6 +10804,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -10297,6 +10883,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -10363,6 +10950,16 @@ mod error_outcome_emission_tests { config.max_turn_duration_secs ), ); + assert_eq!(turn_error.payload["disposition"], "retrying"); + assert_eq!(turn_error.payload["attempt"], 1); + assert_eq!( + turn_error.payload["triggeringRootEventId"], + turn_error.payload["triggeringEventIds"][0] + ); + assert_eq!( + turn_error.payload["triggeringParentEventId"], + turn_error.payload["triggeringEventIds"][0] + ); assert_eq!( queue.pending_channels(), 1, @@ -10394,6 +10991,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -10459,6 +11057,8 @@ mod error_outcome_emission_tests { config.max_turn_duration_secs ), ); + assert_eq!(turn_error.payload["disposition"], "dead_lettered"); + assert_eq!(turn_error.payload["attempt"], crate::queue::MAX_RETRIES + 1); assert_eq!( queue.queued_event_count(channel_id), 0, @@ -10514,6 +11114,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -10656,6 +11257,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -10790,6 +11392,7 @@ mod error_outcome_emission_tests { channel_id: Some(channel_id), scope: Some(session_scope.clone()), turn_id: "indeterminate-project".into(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -10945,6 +11548,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -11048,6 +11652,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -11111,6 +11716,13 @@ mod error_outcome_emission_tests { assert_eq!(errors.len(), 1); assert_eq!(errors[0].payload["code"], -32002); assert_eq!(errors[0].payload["error"], expected_error); + assert_eq!(errors[0].payload["disposition"], "action_required"); + assert_eq!(errors[0].payload["attempt"], 1); + assert_eq!(errors[0].payload["triggeringRootEventId"], root.to_hex()); + assert_eq!( + errors[0].payload["triggeringParentEventId"], + parent.to_hex() + ); // Capture the real signed notice sent by handle_prompt_result, without a live relay. let notice: nostr::Event = tokio::time::timeout(Duration::from_secs(3), async { @@ -11209,6 +11821,7 @@ mod error_outcome_emission_tests { channel_id: None, scope: None, turn_id: "test-turn-id".to_string(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 250221badca..73a18878df4 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -67,6 +67,9 @@ pub struct TaskMeta { pub scope: Option, /// Identifies terminal events when the task panics before returning a result. pub turn_id: String, + /// Conversation correlation retained even when cancellation or Drop mode + /// discards the batch. Terminal events must not depend on turn_started replay. + pub triggering: crate::queue::TriggeringEventContext, /// Clone of batch for Queue mode panic recovery. pub recoverable_batch: Option, /// Control signal for the in-flight prompt task. @@ -2260,9 +2263,9 @@ pub async fn run_prompt_task( turn_id.clone(), turn_started_at.clone(), )); - let triggering_event_ids: Vec = batch + let triggering = batch .as_ref() - .map(|b| b.events.iter().map(|be| be.event.id.to_hex()).collect()) + .map(crate::queue::triggering_event_context) .unwrap_or_default(); agent.acp.observe( "turn_started", @@ -2271,7 +2274,9 @@ pub async fn run_prompt_task( PromptSource::Channel(_) => "channel", PromptSource::Heartbeat => "heartbeat", }, - "triggeringEventIds": triggering_event_ids, + "triggeringEventIds": triggering.event_ids, + "triggeringRootEventId": triggering.root_event_id, + "triggeringParentEventId": triggering.parent_event_id, }), ); @@ -7584,6 +7589,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" channel_id: Some(busy_scope.channel_id()), scope: Some(busy_scope), turn_id: "t".into(), + triggering: Default::default(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -8182,6 +8188,43 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" ); } + #[tokio::test] + async fn cancel_failure_observer_reports_production_batch_fate() { + let mut ctx = make_prompt_context_no_owner(); + ctx.dedup_mode = DedupMode::Queue; + for signal in [ + ControlSignal::Steer, + ControlSignal::Interrupt, + ControlSignal::Cancel, + ControlSignal::Rotate, + ] { + for (removed, circuit_open, capacity) in [ + (false, false, "none"), + (true, false, "none"), + (false, true, "none"), + (false, true, "idle"), + (false, true, "respawning"), + ] { + let batch = one_event_batch(Uuid::new_v4()); + let failure = classify_control_cancel_failure( + &ctx, + AcpError::CancelDrainTimeout(CONTROL_CANCEL_GRACE), + signal.clone(), + Some(batch.clone()), + ); + crate::error_outcome_emission_tests::assert_cancel_failure_metadata( + failure.outcome, + failure.retry_batch, + batch, + removed, + circuit_open, + capacity, + ) + .await; + } + } + } + #[test] fn test_classify_control_cancel_failure_crosses_error_outcome_and_batch_fate() { let ctx = { diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index a419b78178d..9ed3d66a411 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -525,6 +525,17 @@ impl EventQueue { } } + /// Return the number of failed attempts already recorded for a scope. + /// + /// Used to annotate owner-only observer failures before `requeue` either + /// preserves the counter or clears it on dead-letter. + pub fn retry_count(&self, scope: K) -> u32 { + self.retry_counts + .get(&scope.into_scope()) + .copied() + .unwrap_or(0) + } + /// Re-queue a batch of events that failed to process. /// /// Events are pushed back to the **front** of the channel's queue so they @@ -1103,6 +1114,43 @@ pub fn parse_thread_tags(event: &Event) -> ThreadTags { } } +/// The observer-facing conversation context for a flushed prompt batch. +/// +/// Top-level events have no NIP-10 root marker, so their own event id becomes +/// the conversation root. Thread replies preserve the root and parent resolved +/// from their tags. Cancelled events come first in merged prompts; the newest +/// regular event remains the reply anchor. Keeping this at the queue boundary +/// gives `turn_started` and `turn_error` one identical correlation contract. +#[derive(Debug, Clone, Default)] +pub struct TriggeringEventContext { + pub event_ids: Vec, + pub root_event_id: Option, + pub parent_event_id: Option, +} + +pub fn triggering_event_context(batch: &FlushBatch) -> TriggeringEventContext { + let mut event_ids = Vec::with_capacity(batch.cancelled_events.len() + batch.events.len()); + event_ids.extend(batch.cancelled_events.iter().map(|be| be.event.id.to_hex())); + event_ids.extend(batch.events.iter().map(|be| be.event.id.to_hex())); + let Some(last) = batch + .events + .last() + .or_else(|| batch.cancelled_events.last()) + else { + return TriggeringEventContext { + event_ids, + ..TriggeringEventContext::default() + }; + }; + let trigger_id = last.event.id.to_hex(); + let thread = parse_thread_tags(&last.event); + TriggeringEventContext { + event_ids, + root_event_id: Some(thread.root_event_id.unwrap_or_else(|| trigger_id.clone())), + parent_event_id: thread.parent_event_id.or(Some(trigger_id)), + } +} + /// Extract a leading slash command from message content. /// /// ACP connectors (claude-agent-acp, codex-acp) detect slash commands by @@ -3915,6 +3963,44 @@ mod tests { assert!(tags.parent_event_id.is_none()); } + #[test] + fn triggering_context_includes_cancelled_events_but_anchors_to_new_work() { + let ch = Uuid::new_v4(); + let cancelled = make_event("cancelled"); + let root = "a".repeat(64); + let parent = "b".repeat(64); + let latest = make_event_with_tags( + "latest", + vec![ + vec!["e".into(), root.clone(), "".into(), "root".into()], + vec!["e".into(), parent.clone(), "".into(), "reply".into()], + ], + ); + let batch = FlushBatch { + channel_id: ch, + scope: thread(ch, &root), + events: vec![BatchEvent { + event: latest.clone(), + prompt_tag: "test".into(), + received_at: Instant::now(), + }], + cancelled_events: vec![BatchEvent { + event: cancelled.clone(), + prompt_tag: "test".into(), + received_at: Instant::now(), + }], + cancel_reason: Some(CancelReason::Steer), + }; + + let context = triggering_event_context(&batch); + assert_eq!( + context.event_ids, + vec![cancelled.id.to_hex(), latest.id.to_hex()] + ); + assert_eq!(context.root_event_id.as_deref(), Some(root.as_str())); + assert_eq!(context.parent_event_id.as_deref(), Some(parent.as_str())); + } + #[test] fn test_format_prompt_with_channel_info() { let ch = Uuid::new_v4(); diff --git a/desktop/src/features/agents/recentAgentTurnFailuresStore.test.mjs b/desktop/src/features/agents/recentAgentTurnFailuresStore.test.mjs new file mode 100644 index 00000000000..3800d6917ff --- /dev/null +++ b/desktop/src/features/agents/recentAgentTurnFailuresStore.test.mjs @@ -0,0 +1,294 @@ +import assert from "node:assert/strict"; +import { beforeEach, describe, it } from "node:test"; + +import { + createRecentAgentTurnFailuresObserverListener, + getRecentAgentTurnFailures, + resetRecentAgentTurnFailuresStore, + syncRecentAgentTurnFailuresFromEvents, +} from "./recentAgentTurnFailuresStore.ts"; + +const AGENT = "a".repeat(64); +const TRIGGER = "1".repeat(64); +const ROOT = "2".repeat(64); +const TOP_LEVEL = "3".repeat(64); + +function event(overrides = {}) { + return { + seq: 1, + timestamp: "2026-09-09T17:49:05Z", + kind: "turn_started", + agentIndex: 0, + channelId: "channel-1", + sessionId: "session-1", + turnId: "turn-1", + payload: { + triggeringEventIds: [TRIGGER], + triggeringRootEventId: ROOT, + triggeringParentEventId: TRIGGER, + }, + ...overrides, + }; +} + +describe("recentAgentTurnFailuresStore", () => { + beforeEach(resetRecentAgentTurnFailuresStore); + + it("retains a friendly retrying failure in its originating conversation", () => { + syncRecentAgentTurnFailuresFromEvents(AGENT, [ + event(), + event({ + seq: 2, + timestamp: "2026-09-09T17:49:09Z", + kind: "turn_error", + payload: { + error: "Agent reported error (code -32603): Internal error", + code: -32603, + disposition: "retrying", + attempt: 1, + triggeringEventIds: [TRIGGER], + triggeringRootEventId: ROOT, + triggeringParentEventId: TRIGGER, + }, + }), + ]); + + const failures = getRecentAgentTurnFailures("channel-1", ROOT); + assert.equal(failures.length, 1); + assert.equal(failures[0].disposition, "retrying"); + assert.equal(failures[0].attempt, 1); + assert.equal(failures[0].rootEventId, ROOT); + assert.match(failures[0].error, /upgrade the adapter/); + assert.equal(getRecentAgentTurnFailures("channel-1", TRIGGER).length, 0); + }); + + it("updates one conversation failure instead of accumulating retry spam", () => { + syncRecentAgentTurnFailuresFromEvents(AGENT, [ + event(), + event({ + seq: 2, + kind: "turn_error", + payload: { + error: "Internal error", + code: -32603, + disposition: "retrying", + attempt: 1, + triggeringRootEventId: ROOT, + triggeringParentEventId: TRIGGER, + }, + }), + event({ + seq: 3, + timestamp: "2026-09-09T17:50:09Z", + kind: "turn_error", + turnId: "turn-2", + payload: { + error: "Internal error", + code: -32603, + disposition: "dead_lettered", + attempt: 11, + triggeringRootEventId: ROOT, + triggeringParentEventId: TRIGGER, + }, + }), + ]); + + const failures = getRecentAgentTurnFailures("channel-1", ROOT); + assert.equal(failures.length, 1); + assert.equal(failures[0].disposition, "dead_lettered"); + assert.equal(failures[0].attempt, 11); + assert.equal(failures[0].turnId, "turn-2"); + }); + + it("clears the retained failure when the conversation starts a new turn", () => { + syncRecentAgentTurnFailuresFromEvents(AGENT, [ + event(), + event({ + seq: 2, + kind: "turn_error", + payload: { + error: "Internal error", + disposition: "retrying", + triggeringRootEventId: ROOT, + triggeringParentEventId: TRIGGER, + }, + }), + ]); + assert.equal(getRecentAgentTurnFailures("channel-1", ROOT).length, 1); + + syncRecentAgentTurnFailuresFromEvents(AGENT, [ + event({ + seq: 3, + timestamp: "2026-09-09T17:50:09Z", + turnId: "turn-2", + }), + ]); + assert.equal(getRecentAgentTurnFailures("channel-1", ROOT).length, 0); + }); + + it("keeps threaded failures out of the main composer scope", () => { + syncRecentAgentTurnFailuresFromEvents(AGENT, [ + event({ + seq: 2, + kind: "turn_error", + payload: { + error: "thread failure", + triggeringEventIds: [TRIGGER], + triggeringRootEventId: ROOT, + triggeringParentEventId: TRIGGER, + }, + }), + event({ + seq: 3, + timestamp: "2026-09-09T17:50:09Z", + kind: "turn_error", + turnId: "turn-2", + payload: { + error: "top-level failure", + triggeringEventIds: [TOP_LEVEL], + triggeringRootEventId: TOP_LEVEL, + triggeringParentEventId: TOP_LEVEL, + }, + }), + ]); + + assert.equal( + getRecentAgentTurnFailures("channel-1")[0]?.error, + "top-level failure", + ); + assert.equal( + getRecentAgentTurnFailures("channel-1", ROOT)[0]?.error, + "thread failure", + ); + }); + + it("processes only observer updates for an active agent", () => { + const listener = createRecentAgentTurnFailuresObserverListener([ + { pubkey: AGENT, status: "running" }, + { pubkey: "b".repeat(64), status: "stopped" }, + ]); + const failureEvent = event({ + seq: 2, + kind: "turn_error", + payload: { + error: "live failure", + triggeringRootEventId: ROOT, + triggeringParentEventId: TRIGGER, + }, + }); + + listener({ agentPubkey: "b".repeat(64), events: [failureEvent] }); + assert.equal(getRecentAgentTurnFailures("channel-1", ROOT).length, 0); + + listener({ agentPubkey: AGENT, events: [event(), failureEvent] }); + assert.equal( + getRecentAgentTurnFailures("channel-1", ROOT)[0]?.error, + "live failure", + ); + }); + + it("keeps actual legacy source-and-ID-only context unknown and visible", () => { + syncRecentAgentTurnFailuresFromEvents(AGENT, [ + event({ payload: { source: "mention", triggeringEventIds: [TRIGGER] } }), + event({ + seq: 2, + kind: "turn_error", + payload: { error: "legacy failure" }, + }), + ]); + + const [failure] = getRecentAgentTurnFailures("channel-1"); + assert.equal(failure.error, "legacy failure"); + assert.equal(failure.rootEventId, null); + assert.equal(failure.parentEventId, null); + assert.equal(failure.disposition, "unknown"); + assert.equal(getRecentAgentTurnFailures("channel-1", ROOT).length, 0); + assert.equal(getRecentAgentTurnFailures("channel-1", TRIGGER).length, 0); + }); +}); + +describe("retry batch coverage through observer listener", () => { + beforeEach(resetRecentAgentTurnFailuresStore); + it("clears A when retrying [A,B] anchored at B, preserving partial and unrelated failures", () => { + const listener = createRecentAgentTurnFailuresObserverListener([ + { pubkey: AGENT, status: "running" }, + ]); + const failed = (seq, ids, root, overrides = {}) => + event({ + seq, + kind: "turn_error", + turnId: `failed-${seq}`, + payload: { + error: "failed", + disposition: "retrying", + triggeringEventIds: ids, + triggeringRootEventId: root, + ...overrides, + }, + }); + listener({ + agentPubkey: AGENT, + events: [ + failed(1, ["A"], "A"), + failed(2, ["A", "C"], "C"), + failed(3, ["D"], "D"), + { ...failed(4, ["A"], "A"), channelId: "other-channel" }, + event({ + seq: 5, + turnId: "retry-ab", + payload: { + triggeringEventIds: ["A", "B"], + triggeringRootEventId: "B", + triggeringParentEventId: "B", + }, + }), + ], + }); + assert.equal(getRecentAgentTurnFailures("channel-1", "A").length, 0); + assert.equal(getRecentAgentTurnFailures("channel-1", "C").length, 1); + assert.equal(getRecentAgentTurnFailures("channel-1", "D").length, 1); + assert.equal(getRecentAgentTurnFailures("other-channel", "A").length, 1); + }); + + it("preserves full started batch when an error only reports its root", () => { + syncRecentAgentTurnFailuresFromEvents(AGENT, [ + event({ + payload: { triggeringEventIds: ["A", "B"], triggeringRootEventId: "B" }, + }), + event({ + seq: 2, + kind: "turn_error", + payload: { error: "failed", triggeringRootEventId: "B" }, + }), + event({ + seq: 3, + turnId: "partial-retry", + payload: { triggeringEventIds: ["B"], triggeringRootEventId: "B" }, + }), + ]); + assert.deepEqual( + getRecentAgentTurnFailures("channel-1", "B")[0].triggeringEventIds, + ["A", "B"], + ); + }); + + it("accepts panic context without a preceding start and preserves reported recovery", () => { + syncRecentAgentTurnFailuresFromEvents(AGENT, [ + event({ + kind: "agent_panic", + payload: { + error: "panic", + triggeringEventIds: [TRIGGER], + triggeringRootEventId: ROOT, + disposition: "retrying", + respawnScheduled: false, + attempt: 2, + }, + }), + ]); + const [failure] = getRecentAgentTurnFailures("channel-1", ROOT); + assert.equal(failure.disposition, "retrying"); + assert.equal(failure.respawnScheduled, false); + assert.equal(failure.attempt, 2); + }); +}); diff --git a/desktop/src/features/agents/recentAgentTurnFailuresStore.ts b/desktop/src/features/agents/recentAgentTurnFailuresStore.ts new file mode 100644 index 00000000000..3e34d895862 --- /dev/null +++ b/desktop/src/features/agents/recentAgentTurnFailuresStore.ts @@ -0,0 +1,349 @@ +import * as React from "react"; + +import { + type AgentObserverStoreUpdate, + compareObserverEvents, + getAgentObserverSnapshot, + subscribeAgentObserverStore, +} from "@/features/agents/observerRelayStore"; +import { friendlyTurnErrorCopy } from "@/features/agents/lib/friendlyAgentLastError"; +import { normalizePubkey } from "@/shared/lib/pubkey"; +import type { ObserverEvent } from "./ui/agentSessionTypes"; + +const MAX_FAILURES_PER_AGENT = 20; +const MAX_TURN_CONTEXTS_PER_AGENT = 3_000; +const EMPTY_FAILURES: RecentAgentTurnFailure[] = []; + +type TurnContext = { + channelId: string; + rootEventId: string | null; + parentEventId: string | null; + triggeringEventIds: string[]; +}; + +export type TurnFailureDisposition = + | "retrying" + | "dead_lettered" + | "action_required" + | "respawning" + | "stopped" + | "unknown"; + +export type RecentAgentTurnFailure = TurnContext & { + agentPubkey: string; + turnId: string; + error: string; + disposition: TurnFailureDisposition; + attempt: number | null; + respawnScheduled?: boolean; + timestamp: string; +}; + +const failuresByAgent = new Map>(); +const turnContextsByAgent = new Map>(); +const lastProcessedByAgent = new Map>(); +const listeners = new Set<() => void>(); +const cachedByScope = new Map(); + +function asRecord(value: unknown): Record { + return value && typeof value === "object" + ? (value as Record) + : {}; +} + +function asString(value: unknown): string | null { + return typeof value === "string" && value.length > 0 ? value : null; +} + +function eventIds(payload: Record): string[] { + return Array.isArray(payload.triggeringEventIds) + ? payload.triggeringEventIds.filter( + (id): id is string => typeof id === "string" && id.length > 0, + ) + : []; +} + +function contextFromEvent(event: ObserverEvent): TurnContext | null { + if (!event.channelId) return null; + const payload = asRecord(event.payload); + const ids = eventIds(payload); + const rootEventId = asString(payload.triggeringRootEventId); + const parentEventId = asString(payload.triggeringParentEventId); + if (ids.length === 0 && !rootEventId && !parentEventId) return null; + return { + channelId: event.channelId, + rootEventId, + parentEventId, + triggeringEventIds: ids, + }; +} + +function failureKey(context: TurnContext): string { + return `${context.channelId}:${context.rootEventId ?? `unknown:${context.triggeringEventIds.join(",")}`}`; +} + +function disposition(value: unknown): TurnFailureDisposition { + switch (value) { + case "retrying": + case "dead_lettered": + case "action_required": + case "respawning": + case "stopped": + return value; + default: + return "unknown"; + } +} + +function invalidate() { + cachedByScope.clear(); +} + +function notify() { + invalidate(); + for (const listener of listeners) listener(); +} + +function rememberTurnContext( + agentKey: string, + turnId: string, + context: TurnContext, +) { + let contexts = turnContextsByAgent.get(agentKey); + if (!contexts) { + contexts = new Map(); + turnContextsByAgent.set(agentKey, contexts); + } + contexts.delete(turnId); + contexts.set(turnId, context); + while (contexts.size > MAX_TURN_CONTEXTS_PER_AGENT) { + const oldest = contexts.keys().next().value; + if (oldest === undefined) break; + contexts.delete(oldest); + } +} + +function removeCoveredFailures( + agentKey: string, + context: TurnContext, +): boolean { + const failures = failuresByAgent.get(agentKey); + if (!failures) return false; + const coveredIds = new Set(context.triggeringEventIds); + let changed = false; + for (const [key, failure] of failures) { + if (failure.channelId !== context.channelId) continue; + const covered = + failure.triggeringEventIds.length > 0 + ? failure.triggeringEventIds.every((id) => coveredIds.has(id)) + : failure.rootEventId !== null && + failure.rootEventId === context.rootEventId; + if (covered) { + failures.delete(key); + changed = true; + } + } + if (failures.size === 0) failuresByAgent.delete(agentKey); + return changed; +} + +function setFailure(agentKey: string, failure: RecentAgentTurnFailure) { + let failures = failuresByAgent.get(agentKey); + if (!failures) { + failures = new Map(); + failuresByAgent.set(agentKey, failures); + } + const key = failureKey(failure); + failures.delete(key); + failures.set(key, failure); + while (failures.size > MAX_FAILURES_PER_AGENT) { + const oldest = failures.keys().next().value; + if (oldest === undefined) break; + failures.delete(oldest); + } +} + +function processEvent(agentPubkey: string, event: ObserverEvent): boolean { + const agentKey = normalizePubkey(agentPubkey); + const channelKey = event.channelId ?? "\u0000null-channel"; + let watermarks = lastProcessedByAgent.get(agentKey); + const prior = watermarks?.get(channelKey); + if (prior && compareObserverEvents(event, prior) <= 0) return false; + if (!watermarks) { + watermarks = new Map(); + lastProcessedByAgent.set(agentKey, watermarks); + } + watermarks.set(channelKey, event); + + const turnId = event.turnId; + if (event.kind === "turn_started" && turnId) { + const context = contextFromEvent(event); + if (!context) return false; + rememberTurnContext(agentKey, turnId, context); + return removeCoveredFailures(agentKey, context); + } + + if (event.kind !== "turn_error" && event.kind !== "agent_panic") { + return false; + } + + const payload = asRecord(event.payload); + const startedContext = turnId + ? turnContextsByAgent.get(agentKey)?.get(turnId) + : null; + const reportedContext = contextFromEvent(event); + const context = + reportedContext && startedContext + ? { + ...reportedContext, + triggeringEventIds: + reportedContext.triggeringEventIds.length > 0 + ? reportedContext.triggeringEventIds + : startedContext.triggeringEventIds, + rootEventId: + reportedContext.rootEventId ?? startedContext.rootEventId, + parentEventId: + reportedContext.parentEventId ?? startedContext.parentEventId, + } + : (reportedContext ?? startedContext); + if (!context || !turnId) return false; + const rawError = asString(payload.error) ?? "Unknown error"; + const numericAttempt = Number(payload.attempt); + setFailure(agentKey, { + ...context, + agentPubkey, + turnId, + error: friendlyTurnErrorCopy(rawError, payload.code), + disposition: disposition(payload.disposition), + respawnScheduled: + typeof payload.respawnScheduled === "boolean" + ? payload.respawnScheduled + : undefined, + attempt: + Number.isInteger(numericAttempt) && numericAttempt > 0 + ? numericAttempt + : null, + timestamp: event.timestamp, + }); + return true; +} + +export function syncRecentAgentTurnFailuresFromEvents( + agentPubkey: string, + events: ObserverEvent[], +) { + let changed = false; + for (const event of events) { + changed = processEvent(agentPubkey, event) || changed; + } + if (changed) notify(); +} + +export function syncRecentAgentTurnFailuresFromObserver( + agents: readonly { pubkey: string; status: string }[], +) { + for (const agent of agents) { + if (agent.status !== "running" && agent.status !== "deployed") continue; + syncRecentAgentTurnFailuresFromEvents( + agent.pubkey, + getAgentObserverSnapshot(agent.pubkey, true).events, + ); + } +} + +export function subscribeRecentAgentTurnFailures(listener: () => void) { + listeners.add(listener); + return () => listeners.delete(listener); +} + +function isTopLevelFailure(failure: RecentAgentTurnFailure): boolean { + return ( + failure.rootEventId !== null && + failure.rootEventId === + failure.triggeringEventIds[failure.triggeringEventIds.length - 1] + ); +} + +export function getRecentAgentTurnFailures( + channelId: string | null | undefined, + rootEventId: string | null = null, +): RecentAgentTurnFailure[] { + if (!channelId) return EMPTY_FAILURES; + const cacheKey = `${channelId}:${rootEventId ?? "*"}`; + const cached = cachedByScope.get(cacheKey); + if (cached) return cached; + + const failures: RecentAgentTurnFailure[] = []; + for (const agentFailures of failuresByAgent.values()) { + for (const failure of agentFailures.values()) { + if ( + failure.channelId === channelId && + (rootEventId === null + ? failure.rootEventId === null || isTopLevelFailure(failure) + : failure.rootEventId === rootEventId) + ) { + failures.push(failure); + } + } + } + failures.sort((left, right) => right.timestamp.localeCompare(left.timestamp)); + const result = failures.length > 0 ? failures : EMPTY_FAILURES; + cachedByScope.set(cacheKey, result); + return result; +} + +export function useRecentAgentTurnFailures( + channelId: string | null | undefined, + rootEventId: string | null = null, +) { + const getSnapshot = React.useCallback( + () => getRecentAgentTurnFailures(channelId, rootEventId), + [channelId, rootEventId], + ); + return React.useSyncExternalStore( + subscribeRecentAgentTurnFailures, + getSnapshot, + ); +} + +export function createRecentAgentTurnFailuresObserverListener( + agents: readonly { pubkey: string; status: string }[], +): (update?: AgentObserverStoreUpdate) => void { + const activeAgentPubkeys = new Set( + agents + .filter( + (agent) => agent.status === "running" || agent.status === "deployed", + ) + .map((agent) => normalizePubkey(agent.pubkey)), + ); + + return (update?: AgentObserverStoreUpdate) => { + if ( + !update || + !activeAgentPubkeys.has(normalizePubkey(update.agentPubkey)) + ) { + return; + } + syncRecentAgentTurnFailuresFromEvents(update.agentPubkey, [ + ...update.events, + ]); + }; +} + +export function useRecentAgentTurnFailuresBridge( + agents: readonly { pubkey: string; status: string }[], +) { + React.useEffect(() => { + syncRecentAgentTurnFailuresFromObserver(agents); + return subscribeAgentObserverStore( + createRecentAgentTurnFailuresObserverListener(agents), + ); + }, [agents]); +} + +export function resetRecentAgentTurnFailuresStore() { + failuresByAgent.clear(); + turnContextsByAgent.clear(); + lastProcessedByAgent.clear(); + notify(); +} diff --git a/desktop/src/features/agents/useAgentObserverIngestion.ts b/desktop/src/features/agents/useAgentObserverIngestion.ts index 386b7621427..078fd98155f 100644 --- a/desktop/src/features/agents/useAgentObserverIngestion.ts +++ b/desktop/src/features/agents/useAgentObserverIngestion.ts @@ -1,6 +1,7 @@ import * as React from "react"; import { useActiveAgentTurnsBridge } from "@/features/agents/activeAgentTurnsStore"; +import { useRecentAgentTurnFailuresBridge } from "@/features/agents/recentAgentTurnFailuresStore"; import { useManagedAgentsQuery, useRelayAgentsQuery, @@ -113,4 +114,5 @@ export function useAgentObserverIngestion() { useManagedAgentObserverBridge(ingestionAgents); useActiveAgentTurnsBridge(ingestionAgents); + useRecentAgentTurnFailuresBridge(ingestionAgents); } diff --git a/desktop/src/features/channels/ui/AgentTurnFailureStatus.test.mjs b/desktop/src/features/channels/ui/AgentTurnFailureStatus.test.mjs new file mode 100644 index 00000000000..91ee0adf2c6 --- /dev/null +++ b/desktop/src/features/channels/ui/AgentTurnFailureStatus.test.mjs @@ -0,0 +1,71 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import React from "react"; +import { renderToStaticMarkup } from "react-dom/server"; + +import { AgentTurnFailureStatus } from "./AgentTurnFailureStatus.tsx"; + +const AGENT = "a".repeat(64); + +function renderFailure(disposition = "retrying", overrides = {}) { + return renderToStaticMarkup( + React.createElement(AgentTurnFailureStatus, { + agents: [{ pubkey: AGENT, name: "Review Bee" }], + failure: { + agentPubkey: AGENT, + turnId: "turn-1", + channelId: "channel-1", + rootEventId: "1".repeat(64), + parentEventId: "2".repeat(64), + triggeringEventIds: ["2".repeat(64)], + error: "Upgrade the adapter and try again.", + disposition, + attempt: 2, + timestamp: "2026-09-09T17:49:09Z", + ...overrides, + }, + onOpenAgentSession() {}, + profiles: {}, + }), + ); +} + +describe("AgentTurnFailureStatus", () => { + it("renders an actionable retry failure in the composer activity rail", () => { + const html = renderFailure(); + + assert.match(html, /data-testid="agent-turn-failure-status"/); + assert.match(html, /Review Bee couldn/); + assert.match(html, /Retrying automatically/); + assert.match(html, /attempt 2/); + assert.match(html, /View activity/); + assert.match(html, /title="Upgrade the adapter and try again\."/); + }); + + it("renders terminal disposition copy", () => { + assert.match( + renderFailure("dead_lettered"), + /Stopped after multiple attempts/, + ); + }); +}); + +it("renders legacy correlation and recovery as unknown", () => { + const html = renderFailure("unknown", { + rootEventId: null, + parentEventId: null, + }); + assert.match(html, /Conversation unknown/); + assert.match(html, /Recovery status unknown/); + assert.doesNotMatch(html, /Stopped/); +}); + +it("distinguishes queued work from a promised runtime restart", () => { + const html = renderFailure("retrying", { respawnScheduled: false }); + assert.match(html, /Queued for retry/); + assert.doesNotMatch(html, /Restarting|restart needed|Retrying automatically/); + assert.match( + renderFailure("respawning", { respawnScheduled: true }), + /Restarting the agent/, + ); +}); diff --git a/desktop/src/features/channels/ui/AgentTurnFailureStatus.tsx b/desktop/src/features/channels/ui/AgentTurnFailureStatus.tsx new file mode 100644 index 00000000000..12c7953e8fb --- /dev/null +++ b/desktop/src/features/channels/ui/AgentTurnFailureStatus.tsx @@ -0,0 +1,72 @@ +import { AlertTriangle } from "lucide-react"; + +import type { RecentAgentTurnFailure } from "@/features/agents/recentAgentTurnFailuresStore"; +import type { UserProfileLookup } from "@/features/profile/lib/identity"; +import type { BotActivityAgent } from "@/features/channels/ui/BotActivityBar"; +import { UserAvatar } from "@/shared/ui/UserAvatar"; + +const dispositionCopy: Record = { + retrying: "Retrying automatically", + dead_lettered: "Stopped after multiple attempts", + action_required: "Action required", + respawning: "Restarting the agent", + stopped: "Stopped", + unknown: "Recovery status unknown", +}; + +type AgentTurnFailureStatusProps = { + agents: BotActivityAgent[]; + failure: RecentAgentTurnFailure; + onOpenAgentSession: (pubkey: string, channelId?: string | null) => void; + profiles?: UserProfileLookup; +}; + +export function AgentTurnFailureStatus({ + agents, + failure, + onOpenAgentSession, + profiles, +}: AgentTurnFailureStatusProps) { + const agent = agents.find( + (candidate) => + candidate.pubkey.toLowerCase() === failure.agentPubkey.toLowerCase(), + ); + const name = agent?.name ?? "Agent"; + + const recoveryCopy = + failure.disposition === "retrying" && failure.respawnScheduled === false + ? "Queued for retry" + : dispositionCopy[failure.disposition]; + const contextCopy = + failure.rootEventId === null ? " · Conversation unknown" : ""; + + return ( + + ); +} diff --git a/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx b/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx index f99888f0112..34772cb695e 100644 --- a/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx +++ b/desktop/src/features/channels/ui/ChannelComposerActivityAccessory.tsx @@ -1,5 +1,8 @@ import type { ComponentProps } from "react"; +import type { RecentAgentTurnFailure } from "@/features/agents/recentAgentTurnFailuresStore"; +import { AgentTurnFailureStatus } from "@/features/channels/ui/AgentTurnFailureStatus"; + import { CardMintComposerChip } from "@/features/agents/ui/CardMintComposerChip"; import { useCardMintJobs } from "@/features/agents/cardMintStore"; import { BotActivityComposerAction } from "@/features/channels/ui/BotActivityBar"; @@ -10,6 +13,7 @@ type ChannelComposerActivityAccessoryProps = { agents: ComponentProps["agents"]; channel: ComponentProps["channel"]; currentPubkey: ComponentProps["currentPubkey"]; + failure: RecentAgentTurnFailure | null; onOpenAgentSession: ComponentProps< typeof BotActivityComposerAction >["onOpenAgentSession"]; @@ -26,6 +30,7 @@ export function ChannelComposerActivityAccessory({ agents, channel, currentPubkey, + failure, onOpenAgentSession, openAgentSessionPubkey, profiles, @@ -42,6 +47,14 @@ export function ChannelComposerActivityAccessory({ >
{cardMintJobs.length > 0 ? : null} + {failure && workingBotPubkeys.length === 0 ? ( + + ) : null} {workingBotPubkeys.length > 0 ? (
0; + const recentTurnFailures = useRecentAgentTurnFailures( + activeChannel?.id ?? null, + ); + const latestTurnFailure = recentTurnFailures[0] ?? null; const hasCardMintActivity = useCardMintJobs().length > 0; const hasComposerBottomActivity = - hasComposerBotActivity || hasTypingActivity || hasCardMintActivity; + hasComposerBotActivity || + Boolean(latestTurnFailure) || + hasTypingActivity || + hasCardMintActivity; const threadComposerBotTypingPubkeys = React.useMemo( () => selectThreadComposerBotTypingPubkeys(botTypingEntries, openThreadHeadId), @@ -347,6 +356,13 @@ export const ChannelPane = React.memo(function ChannelPane({ ); const hasThreadComposerBotActivity = threadComposerBotTypingPubkeys.length > 0; + const threadTurnFailures = useRecentAgentTurnFailures( + activeChannel?.id ?? null, + openThreadHeadId ?? null, + ); + const latestThreadTurnFailure = threadTurnFailures[0] ?? null; + const hasThreadComposerActivity = + hasThreadComposerBotActivity || Boolean(latestThreadTurnFailure); const directMessageIntro = React.useMemo( () => buildDirectMessageIntro({ @@ -800,6 +816,7 @@ export const ChannelPane = React.memo(function ChannelPane({ agents={activityAgents} channel={activeChannel} currentPubkey={currentPubkey} + failure={latestTurnFailure} onOpenAgentSession={onOpenAgentSession} openAgentSessionPubkey={openAgentSessionPubkey} profiles={profiles} @@ -893,7 +910,7 @@ export const ChannelPane = React.memo(function ChannelPane({ )} threadReplyUnreadCounts={threadReplyUnreadCounts} threadTypingPubkeys={threadTypingPubkeys} - activityAccessoryVisible={hasThreadComposerBotActivity} + activityAccessoryVisible={hasThreadComposerActivity} activityAccessoryContent={ hasThreadComposerBotActivity ? ( + ) : latestThreadTurnFailure ? ( + ) : null } /> diff --git a/desktop/src/features/communities/useCommunityInit.ts b/desktop/src/features/communities/useCommunityInit.ts index 4a1ddf9d0b8..c05ecdef5b3 100644 --- a/desktop/src/features/communities/useCommunityInit.ts +++ b/desktop/src/features/communities/useCommunityInit.ts @@ -36,6 +36,7 @@ import { } from "@/features/agents/activeAgentTurnsStore"; import { resetAgentWorkingSignal } from "@/features/agents/agentWorkingSignal"; import { resetAgentObserverStore } from "@/features/agents/observerRelayStore"; +import { resetRecentAgentTurnFailuresStore } from "@/features/agents/recentAgentTurnFailuresStore"; import { resetAvatarPresentations } from "@/features/profile/avatarPresentationStore"; import { resetAvatarProfileSync } from "@/features/profile/avatarProfileSync"; import { resetSidebarRelayConnectionCardState } from "@/features/sidebar/ui/useSidebarRelayConnectionCard"; @@ -67,6 +68,7 @@ async function resetCommunityState({ clearAllDrafts(); resetAgentObserverStore(); resetActiveAgentTurnsStore(); + resetRecentAgentTurnFailuresStore(); resetAgentWorkingSignal(); if (isTauri() && isMacPlatform()) { void clearTrayAgentActivity();