jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 12:40:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(agents): anchor active-turn timer to authoritative start time (#2033)

Browse Source 
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Will Pfleger
2026-07-17 16:21:58 -04:00
committed by GitHub
parent c19e0ce2d4
commit 4be5c6ae32
6 changed files with 798 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
crates/buzz-acp/src/lib.rs
View File

@@ -759,6 +759,7 @@ fn handle_cancel_turn_control(
channel_id: Some(channel_id.to_string()),
session_id: None,
turn_id: None,
started_at: None,
},
serde_json::json!({
"type": "cancel_turn",
@@ -835,6 +836,7 @@ fn handle_switch_model_control(
channel_id: Some(channel_id.to_string()),
session_id: None,
turn_id: None,
started_at: None,
},
serde_json::json!({
"type": "switch_model",
@@ -2697,6 +2699,8 @@ fn dispatch_pending(
// Prompt text is now built inside run_prompt_task (needs async for
// context fetching). Pass None for prompt_text; batch carries the data.
let (control_tx, control_rx) = tokio::sync::oneshot::channel::<ControlSignal>();
let turn_id = Uuid::new_v4().to_string();
let task_turn_id = turn_id.clone();
let abort_handle = pool.join_set.spawn(async move {
pool::run_prompt_task(
@@ -2706,6 +2710,7 @@ fn dispatch_pending(
ctx_clone,
result_tx,
Some(control_rx),
task_turn_id,
)
.await;
});
@@ -2715,6 +2720,7 @@ fn dispatch_pending(
pool::TaskMeta {
agent_index,
channel_id: Some(channel_id),
turn_id,
recoverable_batch,
control_tx: Some(control_tx),
steer_tx,
@@ -2888,6 +2894,7 @@ fn handle_prompt_result(
PromptSource::Channel(ch) => Some(*ch),
PromptSource::Heartbeat => None,
};
let turn_id = result.turn_id.clone();
let emit_turn_error = |error_msg: &str, error_code: Option<i64>| {
if let Some(ref observer) = observer {
let mut payload = serde_json::json!({
@@ -2900,7 +2907,7 @@ fn handle_prompt_result(
observer.emit(
"turn_error",
Some(agent_index),
&observer::context_for(channel_id, None, None),
&observer::context_for(channel_id, None, Some(turn_id.clone())),
payload,
);
}
@@ -3123,7 +3130,7 @@ fn recover_panicked_agent(
observer.emit(
"agent_panic",
Some(i),
&observer::context_for(meta.channel_id, None, None),
&observer::context_for(meta.channel_id, None, Some(meta.turn_id)),
serde_json::json!({
"outcome": "panic",
"error": format!("Agent task panicked: {join_error}"),
@@ -3228,9 +3235,20 @@ fn dispatch_heartbeat(
let result_tx = pool.result_tx();
let ctx_clone = Arc::clone(ctx);
let agent_index = agent.index;
let turn_id = Uuid::new_v4().to_string();
let task_turn_id = turn_id.clone();
let abort_handle = pool.join_set.spawn(async move {
pool::run_prompt_task(agent, None, Some(prompt_text), ctx_clone, result_tx, None).await;
pool::run_prompt_task(
agent,
None,
Some(prompt_text),
ctx_clone,
result_tx,
None,
task_turn_id,
)
.await;
});
pool.task_map_mut().insert(
@@ -3238,6 +3256,7 @@ fn dispatch_heartbeat(
pool::TaskMeta {
agent_index,
channel_id: None,
turn_id,
recoverable_batch: None,
control_tx: None,
steer_tx: None,
@@ -3825,6 +3844,7 @@ mod owner_control_command_tests {
pool::TaskMeta {
agent_index: 0,
channel_id: Some(channel_id),
turn_id: "test-turn-id".to_string(),
recoverable_batch: None,
control_tx: Some(control_tx),
steer_tx: None,
@@ -4027,6 +4047,7 @@ mod observer_chunk_coalescer_tests {
channel_id: Some("channel-1".to_string()),
session_id: Some("session-1".to_string()),
turn_id: Some("turn-1".to_string()),
started_at: None,
payload: serde_json::json!({
"jsonrpc": "2.0",
"method": "session/update",
@@ -4054,6 +4075,7 @@ mod observer_chunk_coalescer_tests {
channel_id: Some("channel-1".to_string()),
session_id: Some("session-1".to_string()),
turn_id: Some("turn-1".to_string()),
started_at: None,
payload: serde_json::json!({ "type": "turn_started" }),
}
}
@@ -4374,6 +4396,7 @@ mod error_outcome_emission_tests {
crate::pool::TaskMeta {
agent_index: 0,
channel_id: None,
turn_id: "test-turn-id".to_string(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
@@ -4396,6 +4419,7 @@ mod error_outcome_emission_tests {
let result = PromptResult {
agent,
source: PromptSource::Channel(Uuid::new_v4()),
turn_id: "test-turn-id".to_string(),
outcome,
batch: None,
};
@@ -4414,11 +4438,18 @@ mod error_outcome_emission_tests {
None,
);
observer
let turn_errors: Vec<_> = observer
.snapshot()
.iter()
.into_iter()
.filter(|e| e.kind == "turn_error")
.count()
.collect();
assert!(
turn_errors
.iter()
.all(|event| event.turn_id.as_deref() == Some("test-turn-id")),
"turn_error must retain the completed turn id"
);
turn_errors.len()
}
#[tokio::test]
@@ -4426,6 +4457,71 @@ mod error_outcome_emission_tests {
assert_eq!(turn_errors_emitted_for(PromptOutcome::AgentExited).await, 1);
}
#[tokio::test]
async fn panic_event_retains_task_turn_id() {
let mut pool = AgentPool::from_slots(vec![]);
let channel_id = Uuid::new_v4();
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
let abort_handle = pool.join_set.spawn(async move {
let _ = started_tx.send(());
std::future::pending::<()>().await;
});
let task_id = abort_handle.id();
pool.task_map_mut().insert(
task_id,
crate::pool::TaskMeta {
agent_index: 0,
channel_id: Some(channel_id),
turn_id: "panic-turn-id".to_string(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
},
);
started_rx.await.unwrap();
abort_handle.abort();
let join_error = pool.join_set.join_next().await.unwrap().unwrap_err();
let mut queue = EventQueue::new(config::DedupMode::Queue);
let config = test_config();
let mut heartbeat_in_flight = false;
let removed_channels = HashSet::new();
let mut typing_channels = HashMap::new();
let mut crash_history = vec![SlotCircuit {
crash_times: Vec::new(),
open_until: None,
respawn_in_flight: false,
}];
let (respawn_tx, _respawn_rx) = mpsc::channel(8);
let mut respawn_tasks = tokio::task::JoinSet::new();
let observer = ObserverHandle::in_process();
recover_panicked_agent(
&mut pool,
&mut queue,
&config,
join_error,
&mut heartbeat_in_flight,
&removed_channels,
&mut typing_channels,
&mut crash_history,
&respawn_tx,
&mut respawn_tasks,
Some(observer.clone()),
);
let panic = observer
.snapshot()
.into_iter()
.find(|event| event.kind == "agent_panic")
.expect("panic recovery emits an observer event");
assert_eq!(
panic.channel_id.as_deref(),
Some(channel_id.to_string().as_str())
);
assert_eq!(panic.turn_id.as_deref(), Some("panic-turn-id"));
}
#[tokio::test]
async fn idle_timeout_emits_exactly_one_feed_event() {
assert_eq!(
@@ -4465,6 +4561,7 @@ mod error_outcome_emission_tests {
crate::pool::TaskMeta {
agent_index: 0,
channel_id: None,
turn_id: "test-turn-id".to_string(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
@@ -4485,6 +4582,7 @@ mod error_outcome_emission_tests {
let result = PromptResult {
agent,
source: PromptSource::Channel(Uuid::new_v4()),
turn_id: "test-turn-id".to_string(),
outcome,
batch: None,
};
@@ -4548,6 +4646,7 @@ mod error_outcome_emission_tests {
crate::pool::TaskMeta {
agent_index: 0,
channel_id: None,
turn_id: "test-turn-id".to_string(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
@@ -4567,6 +4666,7 @@ mod error_outcome_emission_tests {
let result = PromptResult {
agent,
source: PromptSource::Channel(channel_id),
turn_id: "test-turn-id".to_string(),
outcome,
batch: Some(batch),
};
@@ -4658,6 +4758,7 @@ mod error_outcome_emission_tests {
crate::pool::TaskMeta {
agent_index: 0,
channel_id: None,
turn_id: "test-turn-id".to_string(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
@@ -4689,6 +4790,7 @@ mod error_outcome_emission_tests {
let result = PromptResult {
agent,
source: PromptSource::Channel(channel_id),
turn_id: "test-turn-id".to_string(),
outcome: PromptOutcome::CancelDrainTimeout(grace),
batch: Some(batch),
};
@@ -4795,6 +4897,7 @@ mod error_outcome_emission_tests {
crate::pool::TaskMeta {
agent_index: 0,
channel_id: None,
turn_id: "test-turn-id".to_string(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
@@ -4816,6 +4919,7 @@ mod error_outcome_emission_tests {
let result = PromptResult {
agent,
source: PromptSource::Channel(Uuid::new_v4()),
turn_id: "test-turn-id".to_string(),
outcome: PromptOutcome::CancelDrainTimeout(grace),
// Explicit Stop already dropped the batch upstream in
// `classify_control_cancel_failure` — `handle_prompt_result`
@@ -4905,6 +5009,7 @@ mod observer_payload_trim_tests {
channel_id: Some("11111111-1111-1111-1111-111111111111".to_string()),
session_id: Some("sess-1".to_string()),
turn_id: Some("turn-1".to_string()),
started_at: None,
payload,
}
}

22
crates/buzz-acp/src/observer.rs
View File

@@ -26,6 +26,8 @@ pub struct ObserverContext {
pub session_id: Option<String>,
/// Local UUID for one prompt turn.
pub turn_id: Option<String>,
/// RFC3339 timestamp at which the current turn began, when known.
pub started_at: Option<String>,
}
/// Handle used by the harness to publish local observer events.
@@ -69,6 +71,9 @@ pub struct ObserverEvent {
pub session_id: Option<String>,
/// Local UUID for one prompt turn.
pub turn_id: Option<String>,
/// RFC3339 timestamp at which the current turn began, when known.
#[serde(skip_serializing_if = "Option::is_none")]
pub started_at: Option<String>,
/// Raw or semantic event payload.
pub payload: serde_json::Value,
}
@@ -111,6 +116,7 @@ impl ObserverHandle {
channel_id: context.channel_id.clone(),
session_id: context.session_id.clone(),
turn_id: context.turn_id.clone(),
started_at: context.started_at.clone(),
payload,
};
@@ -140,5 +146,21 @@ pub fn context_for(
channel_id: channel_id.map(|id| id.to_string()),
session_id,
turn_id,
started_at: None,
}
}
/// Attach the authoritative start timestamp to every observer frame for a turn.
pub fn context_for_turn(
channel_id: Option<uuid::Uuid>,
session_id: Option<String>,
turn_id: String,
started_at: String,
) -> ObserverContext {
ObserverContext {
channel_id: channel_id.map(|id| id.to_string()),
session_id,
turn_id: Some(turn_id),
started_at: Some(started_at),
}
}

475
crates/buzz-acp/src/pool.rs
View File

@@ -20,11 +20,11 @@
//! `AcpClient` is NOT Clone — ownership moves out on claim and back on return.
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::task::JoinSet;
use tokio::task::{JoinHandle, JoinSet};
use tokio::time::timeout;
use uuid::Uuid;
@@ -47,6 +47,8 @@ use crate::relay::{ChannelInfo, RestClient};
pub struct TaskMeta {
pub agent_index: usize,
pub channel_id: Option<Uuid>,
/// Identifies terminal events when the task panics before returning a result.
pub turn_id: String,
/// Clone of batch for Queue mode panic recovery.
pub recoverable_batch: Option<FlushBatch>,
/// Control signal for the in-flight prompt task.
@@ -215,6 +217,8 @@ pub struct AgentPool {
pub struct PromptResult {
pub agent: OwnedAgent,
pub source: PromptSource,
/// Identifies the completed turn for observer terminal events.
pub turn_id: String,
pub outcome: PromptOutcome,
/// Present on failure in Queue mode, for requeue.
pub batch: Option<FlushBatch>,
@@ -1165,6 +1169,7 @@ fn with_canvas(prompt: Option<String>, canvas: Option<&str>) -> Option<String> {
/// On the happy path the read loop has already called `take()`, so this is a no-op.
fn send_prompt_result(
result_tx: &mpsc::UnboundedSender<PromptResult>,
turn_id: &str,
mut agent: OwnedAgent,
source: PromptSource,
outcome: PromptOutcome,
@@ -1174,6 +1179,7 @@ fn send_prompt_result(
let _ = result_tx.send(PromptResult {
agent,
source,
turn_id: turn_id.to_owned(),
outcome,
batch,
});
@@ -1198,21 +1204,23 @@ pub async fn run_prompt_task(
ctx: Arc<PromptContext>,
result_tx: mpsc::UnboundedSender<PromptResult>,
control_rx: Option<tokio::sync::oneshot::Receiver<ControlSignal>>,
turn_id: String,
) {
// Is this a channel prompt or a heartbeat?
let source = match &batch {
Some(b) => PromptSource::Channel(b.channel_id),
None => PromptSource::Heartbeat,
};
let turn_id = uuid::Uuid::new_v4().to_string();
let observer_channel_id = match &source {
PromptSource::Channel(channel_id) => Some(*channel_id),
PromptSource::Heartbeat => None,
};
agent.acp.set_observer_context(observer::context_for(
let turn_started_at = chrono::Utc::now().to_rfc3339();
agent.acp.set_observer_context(observer::context_for_turn(
observer_channel_id,
None,
Some(turn_id.clone()),
turn_id.clone(),
turn_started_at.clone(),
));
let triggering_event_ids: Vec<String> = batch
.as_ref()
@@ -1229,6 +1237,45 @@ pub async fn run_prompt_task(
}),
);
// Emits `turn_completed` on any exit path. Captures observer handle and
// metadata now, before the agent is moved into PromptResult. It must be
// declared before `liveness_guard`: Rust drops locals in reverse order, so
// liveness is aborted before completion makes the turn terminal.
let _turn_guard = TurnCompletionGuard::new(
agent.acp.observer_handle(),
agent.acp.observer_agent_index(),
observer_channel_id,
turn_id.clone(),
);
// Start liveness with `turn_started`, not the final session/prompt call:
// session creation, context fetches, and an initial message can themselves
// take longer than the desktop's bounded prune pause. This future is pinned
// for the whole task and dropped with the turn on every exit path.
//
// `liveness_state` is shared with `LivenessGuard`: see its docs for why a
// bare `abort()` alone cannot prevent a `turn_liveness` frame emitted after
// `turn_completed`. Once the session resolves below, `set_session_id`
// updates the same shared state so later ticks stop carrying `None`.
let liveness_state = Arc::new(Mutex::new(LivenessState {
closed: false,
session_id: None,
}));
let liveness = run_turn_liveness(
agent.acp.observer_handle(),
agent.acp.observer_agent_index(),
observer::context_for_turn(
observer_channel_id,
None,
turn_id.clone(),
turn_started_at.clone(),
),
ctx.turn_liveness_interval,
Arc::clone(&liveness_state),
);
let liveness_handle = tokio::spawn(liveness);
let liveness_guard = LivenessGuard::new(liveness_handle, liveness_state);
// Collects event IDs up front. On drop (any exit path — normal, early
// return, or panic), spawns best-effort cleanup of both 👀 and 💬.
// See `ReactionGuard` docs for ordering guarantees and known edge cases.
@@ -1238,15 +1285,6 @@ pub async fn run_prompt_task(
.unwrap_or_default();
let _reaction_guard = ReactionGuard::new(ctx.rest_client.clone(), reaction_ids.clone());
// Emits `turn_completed` on any exit path. Captures observer handle and
// metadata now, before the agent is moved into PromptResult.
let _turn_guard = TurnCompletionGuard::new(
agent.acp.observer_handle(),
agent.acp.observer_agent_index(),
observer_channel_id,
turn_id.clone(),
);
//
// Core memory is delivered inside the system prompt the harness already
// builds (system role for protocol >= 2, the `[System]` user-message
@@ -1393,6 +1431,7 @@ pub async fn run_prompt_task(
agent.state.invalidate_all();
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::AgentExited,
@@ -1405,6 +1444,7 @@ pub async fn run_prompt_task(
// so the next retry will re-fetch a fresh revision.
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Error(e),
@@ -1433,6 +1473,7 @@ pub async fn run_prompt_task(
agent.state.invalidate_all();
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::AgentExited,
@@ -1443,6 +1484,7 @@ pub async fn run_prompt_task(
Err(e) => {
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Error(e),
@@ -1454,11 +1496,15 @@ pub async fn run_prompt_task(
}
}
};
agent.acp.set_observer_context(observer::context_for(
agent.acp.set_observer_context(observer::context_for_turn(
observer_channel_id,
Some(session_id.clone()),
Some(turn_id.clone()),
turn_id.clone(),
turn_started_at,
));
// Backfill liveness's shared session ID so ticks after this point carry
// it too, matching every other observer frame for this turn.
liveness_guard.set_session_id(session_id.clone());
agent.acp.observe(
"session_resolved",
serde_json::json!({
@@ -1519,6 +1565,7 @@ pub async fn run_prompt_task(
agent.state.invalidate_all();
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::AgentExited,
@@ -1544,6 +1591,7 @@ pub async fn run_prompt_task(
agent.state.invalidate_all();
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::AgentExited,
@@ -1561,6 +1609,7 @@ pub async fn run_prompt_task(
}
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Timeout(TimeoutKind::Idle),
@@ -1577,6 +1626,7 @@ pub async fn run_prompt_task(
agent.state.invalidate_all();
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Timeout(TimeoutKind::Hard),
@@ -1592,6 +1642,7 @@ pub async fn run_prompt_task(
agent.state.invalidate(&source);
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Error(e),
@@ -1678,6 +1729,7 @@ pub async fn run_prompt_task(
tracing::error!("run_prompt_task: no batch and no prompt_text — returning agent");
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Error(AcpError::Protocol("no batch and no prompt_text".into())),
@@ -1713,22 +1765,6 @@ pub async fn run_prompt_task(
// the main loop can cancel, interrupt, or rotate it. Heartbeats
// (control_rx=None) take the simple await path — they are not controllable.
//
// The liveness future emits `turn_liveness` pings on an interval and never
// resolves; it rides every prompt-await path as a non-winning select arm so
// a turn stays alive on the desktop while it runs. Built from a captured
// observer handle (not `&agent.acp`) because the prompt holds `&mut agent.acp`.
let liveness = run_turn_liveness(
agent.acp.observer_handle(),
agent.acp.observer_agent_index(),
observer::context_for(
observer_channel_id,
Some(session_id.clone()),
Some(turn_id.clone()),
),
ctx.turn_liveness_interval,
);
tokio::pin!(liveness);
let prompt_result = match control_rx {
None => {
// Heartbeat / non-cancellable path.
@@ -1740,7 +1776,6 @@ pub async fn run_prompt_task(
ctx.idle_timeout,
ctx.max_turn_duration,
) => result,
_ = &mut liveness => unreachable!("liveness future never resolves"),
}
}
Some(rx) => {
@@ -1752,7 +1787,6 @@ pub async fn run_prompt_task(
ctx.idle_timeout,
ctx.max_turn_duration,
) => result,
_ = &mut liveness => unreachable!("liveness future never resolves"),
mode = rx => {
let control_signal = mode.unwrap_or(ControlSignal::Cancel);
// Land the model switch before any cancel/requeue work: setting
@@ -1790,6 +1824,7 @@ pub async fn run_prompt_task(
.await;
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Cancelled,
@@ -1825,6 +1860,7 @@ pub async fn run_prompt_task(
.await;
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
failure.outcome,
@@ -1879,6 +1915,7 @@ pub async fn run_prompt_task(
.await;
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Ok(StopReason::EndTurn),
@@ -1941,6 +1978,7 @@ pub async fn run_prompt_task(
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Ok(stop_reason),
@@ -1962,6 +2000,7 @@ pub async fn run_prompt_task(
.await;
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::AgentExited,
@@ -1995,6 +2034,7 @@ pub async fn run_prompt_task(
// session state will be discarded with the old agent.
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Timeout(TimeoutKind::Idle),
@@ -2020,6 +2060,7 @@ pub async fn run_prompt_task(
.await;
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::AgentExited,
@@ -2044,6 +2085,7 @@ pub async fn run_prompt_task(
.await;
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Timeout(TimeoutKind::Idle),
@@ -2071,6 +2113,7 @@ pub async fn run_prompt_task(
.await;
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Timeout(TimeoutKind::Hard),
@@ -2097,6 +2140,7 @@ pub async fn run_prompt_task(
.await;
send_prompt_result(
&result_tx,
&turn_id,
agent,
source,
PromptOutcome::Error(e),
@@ -3046,23 +3090,29 @@ impl Drop for ReactionGuard {
// Periodically emits a `turn_liveness` observer event while a turn is in-flight,
// so the desktop can prune turns whose host died without unwinding (kill -9 /
// crash) far sooner than the no-activity backstop. Runs as a non-resolving
// `select!` arm in `run_prompt_task`: it lives and dies with the prompt future,
// so emission stops on every exit path (complete / cancel / error / panic) with
// no separate teardown to forget.
// crash) far sooner than the no-activity backstop. `run_prompt_task` runs it in
// a background task from `turn_started` until `LivenessGuard` drops, covering
// session setup as well as the final prompt call. When `interval` is zero,
// liveness is disabled and the future parks forever without emitting.
//
// Takes a captured `ObserverHandle` rather than `&agent.acp` because the prompt
// future holds `&mut agent.acp` for its whole duration — a second borrow would
// not compile.
// `state` is the other half of `LivenessGuard`'s shutdown mutex (see its
// docs): held here across the check-then-emit, so a `LivenessGuard::drop`
// racing an in-flight tick either observes `state.closed == true` and skips
// the emit, or is blocked on the same lock until this tick's emit has
// already landed. Either way `turn_completed` cannot pass a live
// `turn_liveness` frame on the wire — the race is closed, not narrowed.
//
// This future never resolves; callers must race it against the prompt and rely
// on drop for teardown. When `interval` is zero, liveness is disabled and the
// future parks forever without emitting.
// `context`'s `session_id` starts `None` (liveness begins before session
// creation) and is filled in from `state.session_id` on each tick — set once
// by `run_prompt_task` after session resolution — so pings emitted for the
// remainder of the turn carry the real session, matching every other
// observer frame for this turn instead of a permanent `None`.
async fn run_turn_liveness(
observer: Option<observer::ObserverHandle>,
agent_index: Option<usize>,
context: observer::ObserverContext,
mut context: observer::ObserverContext,
interval: Duration,
state: Arc<Mutex<LivenessState>>,
) {
let Some(observer) = observer else {
return std::future::pending::<()>().await;
@@ -3077,12 +3127,83 @@ async fn run_turn_liveness(
ticker.tick().await;
loop {
ticker.tick().await;
// Nothing awaitable between the lock and the emit: `LivenessGuard::drop`
// takes this same lock before its `abort()`, so the guard can only ever
// observe this tick fully emitted or not yet started — never mid-emit.
let guard = match state.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
if guard.closed {
return;
}
context.session_id = guard.session_id.clone();
observer.emit(
"turn_liveness",
agent_index,
&context,
serde_json::json!({}),
);
drop(guard);
}
}
/// Shared shutdown/session state between `run_turn_liveness` and its
/// `LivenessGuard`. A single lock covers both fields so a tick's
/// check-session/emit and a guard's set-closed/abort can never interleave.
struct LivenessState {
closed: bool,
session_id: Option<String>,
}
/// Owns the background liveness task for one `run_prompt_task` invocation.
///
/// Dropping the guard aborts the non-resolving task, so liveness covers all
/// pre-prompt setup yet cannot survive a completed, cancelled, or panicked turn.
///
/// `abort()` alone leaves a race: tokio's cooperative cancellation only takes
/// effect at the next `.await` point inside the aborted task, so a tick that
/// has already passed its await and is mid-`observer.emit` when `drop` runs
/// can still complete that emit — a `turn_liveness` frame lands on the wire
/// after `turn_completed`, reviving a finished turn's badge for up to the
/// desktop's bounded prune-pause window. `state` shares a lock with
/// `run_turn_liveness`'s check-then-emit (see its docs): setting `closed`
/// and aborting under the same lock the emitter holds during its tick means
/// `drop` either sees the flag land before that tick's lock is taken (emit
/// skipped) or blocks until the in-flight emit under the lock has finished
/// (then aborts, so there is no next tick) — no interleaving emits a frame
/// after this guard has dropped.
struct LivenessGuard {
handle: JoinHandle<()>,
state: Arc<Mutex<LivenessState>>,
}
impl LivenessGuard {
fn new(handle: JoinHandle<()>, state: Arc<Mutex<LivenessState>>) -> Self {
Self { handle, state }
}
/// Record the turn's session ID once known, so subsequent liveness ticks
/// stamp it on the emitted `turn_liveness` frame instead of `None`.
fn set_session_id(&self, session_id: String) {
let mut guard = match self.state.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
guard.session_id = Some(session_id);
}
}
impl Drop for LivenessGuard {
fn drop(&mut self) {
{
let mut guard = match self.state.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
guard.closed = true;
}
self.handle.abort();
}
}
@@ -4492,9 +4613,6 @@ mod tests {
}
// ── turn liveness emission ───────────────────────────────────────────────
// `run_turn_liveness` is raced against a "prompt" future the same way
// `run_prompt_task` does it: the prompt wins the select and the liveness
// future is dropped. We assert what the observer saw.
fn liveness_count(handle: &observer::ObserverHandle) -> usize {
handle
@@ -4504,47 +4622,174 @@ mod tests {
.count()
}
fn open_liveness_state() -> Arc<Mutex<LivenessState>> {
Arc::new(Mutex::new(LivenessState {
closed: false,
session_id: None,
}))
}
#[tokio::test(start_paused = true)]
async fn test_liveness_fires_while_prompt_pends_then_stops() {
async fn test_liveness_stops_before_completion_frame() {
let observer = observer::ObserverHandle::in_process();
let context = observer::context_for(None, None, Some("t-1".into()));
let liveness = run_turn_liveness(
Some(observer.clone()),
Some(0),
context,
Duration::from_secs(10),
let context =
observer::context_for_turn(None, None, "t-1".into(), "2026-07-14T21:00:00Z".into());
let completion_context = observer::context_for(None, None, Some("t-1".into()));
let completion_observer = observer.clone();
let completion_handle = tokio::spawn(async move {
let state = open_liveness_state();
let _liveness_guard = LivenessGuard::new(
tokio::spawn(run_turn_liveness(
Some(observer.clone()),
Some(0),
context,
Duration::from_secs(10),
Arc::clone(&state),
)),
state,
);
tokio::time::sleep(Duration::from_secs(25)).await;
observer.emit(
"turn_completed",
Some(0),
&completion_context,
serde_json::json!({}),
);
});
tokio::task::yield_now().await;
tokio::time::advance(Duration::from_secs(25)).await;
completion_handle.await.unwrap();
tokio::task::yield_now().await;
let events = completion_observer.snapshot();
let completion_index = events
.iter()
.position(|event| event.kind == "turn_completed")
.expect("turn must complete");
assert!(
events[..completion_index]
.iter()
.all(|event| event.kind != "turn_liveness"
|| event.turn_id.as_deref() == Some("t-1")),
"pre-completion liveness must belong to the active turn"
);
tokio::pin!(liveness);
assert!(
events[completion_index + 1..]
.iter()
.all(|event| event.kind != "turn_liveness"),
"liveness must be aborted before a completion frame is emitted"
);
}
// Prompt pends for 25s, then completes — first liveness tick at 10s,
// second at 20s, so the observer must see exactly two pings.
tokio::select! {
biased;
() = tokio::time::sleep(Duration::from_secs(25)) => {}
_ = &mut liveness => unreachable!("liveness future never resolves"),
}
#[tokio::test(start_paused = true)]
async fn test_liveness_fires_until_guard_drops() {
let observer = observer::ObserverHandle::in_process();
let started_at = "2026-07-14T21:00:00Z".to_string();
let context = observer::context_for_turn(None, None, "t-1".into(), started_at.clone());
let state = open_liveness_state();
let guard = LivenessGuard::new(
tokio::spawn(run_turn_liveness(
Some(observer.clone()),
Some(0),
context,
Duration::from_secs(10),
Arc::clone(&state),
)),
state,
);
tokio::task::yield_now().await;
// First liveness tick at 10s and the second at 20s.
tokio::time::advance(Duration::from_secs(25)).await;
tokio::task::yield_now().await;
assert_eq!(liveness_count(&observer), 2);
// The turn carried the live turn_id on each ping.
let pings: Vec<_> = observer
.snapshot()
.into_iter()
.filter(|e| e.kind == "turn_liveness")
.collect();
assert!(pings.iter().all(|e| e.turn_id.as_deref() == Some("t-1")));
assert!(pings
.iter()
.all(|event| event.turn_id.as_deref() == Some("t-1")));
assert!(pings
.iter()
.all(|event| event.started_at.as_deref() == Some(&started_at)));
assert!(pings
.iter()
.all(|event| event.payload == serde_json::json!({})));
assert_eq!(
serde_json::to_value(&pings[0]).unwrap()["startedAt"],
started_at,
"turn start must serialize in the observer envelope"
);
// After the prompt wins the select, the liveness future is dropped —
// advancing the clock further produces no new pings.
tokio::time::sleep(Duration::from_secs(60)).await;
// The guard is owned by `run_prompt_task`; dropping it aborts liveness
// so completed, cancelled, and errored turns cannot emit late pings.
drop(guard);
tokio::task::yield_now().await;
tokio::time::advance(Duration::from_secs(60)).await;
tokio::task::yield_now().await;
assert_eq!(liveness_count(&observer), 2);
}
#[tokio::test(start_paused = true)]
async fn test_liveness_backfills_session_id_after_resolution() {
let observer = observer::ObserverHandle::in_process();
let context =
observer::context_for_turn(None, None, "t-1".into(), "2026-07-14T21:00:00Z".into());
let state = open_liveness_state();
let guard = LivenessGuard::new(
tokio::spawn(run_turn_liveness(
Some(observer.clone()),
Some(0),
context,
Duration::from_secs(10),
Arc::clone(&state),
)),
state,
);
tokio::task::yield_now().await;
// First tick at 10s fires before the session resolves — must carry
// no session ID, matching every other pre-resolution observer frame.
tokio::time::advance(Duration::from_secs(10)).await;
tokio::task::yield_now().await;
guard.set_session_id("sess-1".to_string());
// Second tick at 20s fires after resolution — must carry it.
tokio::time::advance(Duration::from_secs(10)).await;
tokio::task::yield_now().await;
let pings: Vec<_> = observer
.snapshot()
.into_iter()
.filter(|e| e.kind == "turn_liveness")
.collect();
assert_eq!(pings.len(), 2);
assert_eq!(
pings[0].session_id, None,
"pre-resolution ping must not carry a session ID"
);
assert_eq!(
pings[1].session_id.as_deref(),
Some("sess-1"),
"post-resolution ping must carry the resolved session ID"
);
}
#[tokio::test(start_paused = true)]
async fn test_liveness_disabled_when_interval_zero_emits_nothing() {
let observer = observer::ObserverHandle::in_process();
let context = observer::context_for(None, None, Some("t-1".into()));
let liveness = run_turn_liveness(Some(observer.clone()), Some(0), context, Duration::ZERO);
let liveness = run_turn_liveness(
Some(observer.clone()),
Some(0),
context,
Duration::ZERO,
open_liveness_state(),
);
tokio::pin!(liveness);
tokio::select! {
@@ -4561,7 +4806,13 @@ mod tests {
// A turn that never started has no observer handle — the future must
// park without emitting or panicking.
let context = observer::context_for(None, None, Some("t-1".into()));
let liveness = run_turn_liveness(None, None, context, Duration::from_secs(10));
let liveness = run_turn_liveness(
None,
None,
context,
Duration::from_secs(10),
open_liveness_state(),
);
tokio::pin!(liveness);
tokio::select! {
@@ -4572,6 +4823,88 @@ mod tests {
// No observer to assert against — reaching here without panic is the test.
}
// These two tests pin the shutdown mechanism itself (F1), not timing.
// The existing paused-clock tests above only prove liveness stops
// *eventually* after a guard drop — under `tokio::time::pause`, the
// scheduler never actually interleaves a drop with an in-flight emit, so
// they cannot catch a real cross-thread race between `LivenessGuard::drop`
// and `run_turn_liveness`'s tick. These assert the two halves of the
// contract directly: the check gates the emit with the flag pre-set (no
// `LivenessGuard` involved), and `drop` cannot return while the shared
// lock is held by an in-flight tick (real OS threads, no cooperative
// scheduling to serialize the race away).
#[tokio::test(start_paused = true)]
async fn test_liveness_emits_nothing_once_closed_flag_is_set() {
let observer = observer::ObserverHandle::in_process();
let context =
observer::context_for_turn(None, None, "t-1".into(), "2026-07-14T21:00:00Z".into());
// Set directly, bypassing `LivenessGuard` — isolates the read side of
// the contract: the check under the lock must gate the emit on its own.
let state = Arc::new(Mutex::new(LivenessState {
closed: true,
session_id: None,
}));
let liveness = run_turn_liveness(
Some(observer.clone()),
Some(0),
context,
Duration::from_secs(10),
state,
);
tokio::time::timeout(Duration::from_secs(60), liveness)
.await
.expect("run_turn_liveness must return once closed, not park forever");
assert_eq!(
liveness_count(&observer),
0,
"the pre-set closed flag must suppress every tick's emit"
);
}
#[test]
fn test_liveness_guard_drop_blocks_while_emit_lock_is_held() {
// Standing in for a tick that has already entered its critical
// section: hold the shared lock before the guard drops.
let state = Arc::new(Mutex::new(LivenessState {
closed: false,
session_id: None,
}));
let held = state.lock().unwrap();
let rt = tokio::runtime::Runtime::new().unwrap();
let handle = rt.spawn(std::future::pending::<()>());
let guard = LivenessGuard::new(handle, Arc::clone(&state));
let (tx, rx) = std::sync::mpsc::channel();
let drop_thread = std::thread::spawn(move || {
drop(guard);
tx.send(()).unwrap();
});
// While the emit lock is held, `drop` cannot have completed: it takes
// the same lock before it sets the flag and aborts. A bounded timeout
// proves non-completion by construction of the lock, not the clock —
// `recv_timeout` returning `Timeout` here only holds because the
// mutex is genuinely contended; it cannot pass by scheduling luck.
assert_eq!(
rx.recv_timeout(Duration::from_millis(200)),
Err(std::sync::mpsc::RecvTimeoutError::Timeout),
"drop must block while the tick's emit lock is held"
);
// Release the lock — drop can now acquire it, set the flag, and abort.
drop(held);
rx.recv_timeout(Duration::from_secs(5))
.expect("drop must complete once the emit lock is released");
drop_thread.join().unwrap();
assert!(
state.lock().unwrap().closed,
"closed flag must be set by the time drop has returned"
);
}
// ── steer_rx invariant tests ──────────────────────────────────────────
//
// These pin the `send_prompt_result` invariant: `steer_rx` is always
@@ -4624,6 +4957,7 @@ mod tests {
let source = PromptSource::Heartbeat;
send_prompt_result(
&result_tx,
"test-turn-id",
agent,
source,
PromptOutcome::Error(AcpError::Protocol("simulated session-create error".into())),
@@ -4681,6 +5015,7 @@ mod tests {
let source = PromptSource::Heartbeat;
send_prompt_result(
&result_tx,
"test-turn-id",
agent,
source,
PromptOutcome::Ok(StopReason::EndTurn),

252
desktop/src/features/agents/activeAgentTurnsStore.test.mjs
View File

@@ -293,6 +293,64 @@ describe("activeAgentTurnsStore", () => {
assert.equal(channels.size, 1);
assert.ok(channels.has("c1"));
});
it("agent_panic with an explicit turnId removes only that turn", () => {
syncAgentTurnsFromEvents(AGENT, [
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
makeEvent({ seq: 2, turnId: "t2", channelId: "c1" }),
makeEvent({
seq: 3,
kind: "agent_panic",
turnId: "t2",
channelId: "c1",
}),
]);
assert.equal(
getActiveTurnsForAgent(AGENT).length,
1,
"the explicit panic must preserve the other live turn",
);
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 4,
kind: "turn_completed",
turnId: "t1",
channelId: "c1",
}),
]);
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
});
it("turn_error with an explicit turnId removes only that turn", () => {
syncAgentTurnsFromEvents(AGENT, [
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
makeEvent({ seq: 2, turnId: "t2", channelId: "c1" }),
makeEvent({
seq: 3,
kind: "turn_error",
turnId: "t2",
channelId: "c1",
}),
]);
assert.equal(
getActiveTurnsForAgent(AGENT).length,
1,
"the explicit error must preserve the other live turn",
);
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 4,
kind: "turn_completed",
turnId: "t1",
channelId: "c1",
}),
]);
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
});
});
describe("listener notifications", () => {
@@ -670,11 +728,10 @@ describe("activeAgentTurnsStore", () => {
// to it, so elapsed time is exactly what mock.timers.tick advances.
const EPOCH = Date.parse("2024-01-01T00:00:00Z");
const at = (ms) => new Date(EPOCH + ms).toISOString();
// Mirrors the store's REMOVE_AFTER_MS (LIVENESS_INTERVAL_MS * 2.5) and
// PRUNE_INTERVAL_MS. Not exported — kept in lockstep here so the prune
// bound stays asserted from the consumer's perspective.
const REMOVE_AFTER_MS = 25_000;
// Mirrors the store's private timing constants. Keep these consumer-level
// tests deterministic without exporting implementation details.
const PRUNE_INTERVAL_MS = 5_000;
const PRUNE_PAUSE_MAX_MS = 3 * 60_000;
let unsubscribe;
@@ -716,11 +773,9 @@ describe("activeAgentTurnsStore", () => {
);
});
it("prunes a dead turn at the bound while a live sibling keeps the stream fresh", () => {
// The no-regression case for the all-stale pause: a turn dies (no more
// liveness) but ANOTHER turn keeps refreshing. The pause gates on the MAX
// lastActivityAt, so the live sibling keeps it fresh and the dead turn
// still prunes at 25s — the pause only engages when EVERY turn is stale.
it("prunes a stale turn at the bound when its tracked sibling stays fresh", () => {
// A same-agent tracked sibling keeps the agent's max lastActivityAt
// fresh, so the stale turn is genuinely dead and must still prune at 25s.
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 1,
@@ -756,41 +811,75 @@ describe("activeAgentTurnsStore", () => {
assert.ok(channels.has("c2"), "the live sibling must survive");
});
it("pauses pruning when EVERY tracked turn goes stale at once (relay drop)", () => {
// The "all at once" drop signature: all liveness stops simultaneously.
// No turn refreshes the max, so the pause engages before the 25s prune
// and the badges stay visible through the transient drop.
it("pauses an agent's stale turns while another agent stays fresh", () => {
// Agent B keeps reporting tracked-turn activity, but that must not cause
// agent A's fully stale stream to prune at 25s.
syncAgentTurnsFromEvents(AGENT, [
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
makeEvent({ seq: 2, turnId: "t2", channelId: "c2", timestamp: at(0) }),
makeEvent({
seq: 1,
turnId: "stale",
channelId: "a",
timestamp: at(0),
}),
]);
syncAgentTurnsFromEvents(AGENT_2, [
makeEvent({ seq: 1, turnId: "live", channelId: "b", timestamp: at(0) }),
]);
assert.equal(getActiveTurnsForAgent(AGENT).length, 2);
// Silence past the bound — the pause must hold both badges.
mock.timers.tick(REMOVE_AFTER_MS + PRUNE_INTERVAL_MS);
for (let t = 10_000; t <= 30_000; t += 10_000) {
mock.timers.tick(10_000);
syncAgentTurnsFromEvents(AGENT_2, [
makeEvent({
seq: t / 10_000 + 1,
kind: "turn_liveness",
turnId: "live",
channelId: "b",
timestamp: at(t),
}),
]);
}
assert.equal(
getActiveTurnsForAgent(AGENT).length,
2,
"all-stale-at-once must pause the prune so badges survive a drop",
assert.ok(
channelIdsOf(getActiveTurnsForAgent(AGENT)).has("a"),
"a fully stale agent must stay visible while another agent remains fresh",
);
// Keep B's tracked turn fresh until A passes the bounded 3-minute pause.
for (let t = 40_000; t <= PRUNE_PAUSE_MAX_MS; t += 10_000) {
mock.timers.tick(10_000);
syncAgentTurnsFromEvents(AGENT_2, [
makeEvent({
seq: t / 10_000 + 1,
kind: "turn_liveness",
turnId: "live",
channelId: "b",
timestamp: at(t),
}),
]);
}
mock.timers.tick(PRUNE_INTERVAL_MS);
assert.ok(
!channelIdsOf(getActiveTurnsForAgent(AGENT)).has("a"),
"the stale agent must prune after the 3-minute pause cap",
);
assert.ok(
channelIdsOf(getActiveTurnsForAgent(AGENT_2)).has("b"),
"the fresh agent must stay active",
);
});
it("holds a lone silent turn past the bound until the next frame (residual)", () => {
// The accepted residual: a single turn whose host dies (kill -9) under a
// HEALTHY relay is indistinguishable from a drop with one live turn, so
// its badge lingers past 25s. It clears the instant any frame arrives.
it("prunes a lone silent turn after the 3-minute pause cap", () => {
syncAgentTurnsFromEvents(AGENT, [
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
]);
assert.equal(getActiveTurnsForAgent(AGENT).length, 1);
mock.timers.tick(REMOVE_AFTER_MS + PRUNE_INTERVAL_MS);
mock.timers.tick(PRUNE_PAUSE_MAX_MS);
assert.equal(
getActiveTurnsForAgent(AGENT).length,
1,
"a lone silent turn lingers (pause engages) — accepted residual",
0,
"the pause backstop must clear a dead lone agent at the 3-minute cap",
);
});
@@ -991,7 +1080,6 @@ describe("activeAgentTurnsStore", () => {
describe("resurrection after a prune (A) gated by completion (C)", () => {
const EPOCH = Date.parse("2024-01-01T00:00:00Z");
const at = (ms) => new Date(EPOCH + ms).toISOString();
const REMOVE_AFTER_MS = 25_000;
const PRUNE_INTERVAL_MS = 5_000;
let unsubscribe;
@@ -1006,43 +1094,121 @@ describe("activeAgentTurnsStore", () => {
mock.timers.reset();
});
it("resurrects a lone turn pruned out from under a still-running host", () => {
// The lone-crash residual self-heals: the badge is pruned during silence,
// then a recovered liveness frame for the same turn revives it.
it("resurrects a pruned turn at startedAt from the first recovered ACP frame", () => {
syncAgentTurnsFromEvents(AGENT, [
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
]);
mock.timers.tick(REMOVE_AFTER_MS + PRUNE_INTERVAL_MS);
// A live sibling appears, which lets the prune fire and clear the lone
// stale turn; then a recovered liveness for t1 must revive its badge.
// A fresh tracked sibling permits pruning stale t1 at the normal bound.
mock.timers.tick(30_000);
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 2,
turnId: "t2",
channelId: "c2",
timestamp: at(40_000),
timestamp: at(30_000),
}),
]);
mock.timers.tick(PRUNE_INTERVAL_MS);
assert.ok(
!channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
"the stale lone turn must prune once a live sibling unblocks the sweep",
"the stale turn must prune before its activity recovers",
);
mock.timers.tick(10_000);
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 3,
kind: "turn_liveness",
kind: "acp_read",
turnId: "t1",
channelId: "c1",
timestamp: at(45_000),
startedAt: at(0),
}),
]);
assert.ok(
channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
"a recovered liveness must resurrect the pruned turn's badge",
const resurrected = getActiveTurnsForAgent(AGENT).find(
(turn) => turn.channelId === "c1",
);
assert.equal(
Date.now() - resurrected.anchorAt,
45_000,
"the first recovered ACP frame must preserve elapsed time from the original turn start",
);
});
it("preserves a valid Unix-epoch startedAt envelope timestamp", () => {
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 1,
kind: "turn_liveness",
turnId: "t1",
channelId: "c1",
timestamp: at(0),
startedAt: "1970-01-01T00:00:00.000Z",
}),
]);
const [resurrected] = getActiveTurnsForAgent(AGENT);
assert.equal(
resurrected.anchorAt,
0,
"a valid zero timestamp must not be replaced with the recovery time",
);
});
it("falls back to the frame timestamp when startedAt is absent or invalid", () => {
for (const invalidStartedAt of [
undefined,
null,
"not-a-timestamp",
"future",
]) {
resetActiveAgentTurnsStore();
const startedAtMs = Date.now();
const iso = (offset) => new Date(startedAtMs + offset).toISOString();
const startedAt =
invalidStartedAt === "future" ? iso(46_000) : invalidStartedAt;
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 1,
turnId: "t1",
channelId: "c1",
timestamp: iso(0),
}),
]);
mock.timers.tick(30_000);
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 2,
turnId: "t2",
channelId: "c2",
timestamp: iso(30_000),
}),
]);
mock.timers.tick(PRUNE_INTERVAL_MS);
mock.timers.tick(10_000);
syncAgentTurnsFromEvents(AGENT, [
makeEvent({
seq: 3,
kind: "turn_liveness",
turnId: "t1",
channelId: "c1",
timestamp: iso(45_000),
startedAt,
}),
]);
const resurrected = getActiveTurnsForAgent(AGENT).find(
(turn) => turn.channelId === "c1",
);
assert.equal(
Date.now() - resurrected.anchorAt,
0,
"old or malformed frames must retain the existing recovery anchor",
);
}
});
it("does NOT resurrect a turn whose liveness is older than its completion", () => {

85
desktop/src/features/agents/activeAgentTurnsStore.ts
View File

@@ -16,11 +16,13 @@ const LIVENESS_INTERVAL_MS = 10_000;
* graceful exits clear via turn_completed and working turns refresh on every
* stream event. Derived from the interval so it tracks if the interval changes. */
const REMOVE_AFTER_MS = LIVENESS_INTERVAL_MS * 2.5;
/** Pause pruning once EVERY tracked turn has gone this long without activity —
* the "all at once" signature of a relay drop (flaky VPN), where liveness frames
* stop arriving for all agents simultaneously. Set below REMOVE_AFTER_MS so the
* pause engages before the 25s prune would wipe the badges. */
/** Pause pruning for an agent once ALL of its tracked turns have gone this long
* without activity — the "all at once" signature of that agent's frame stream
* being down. Set below REMOVE_AFTER_MS so the pause engages before the 25s
* prune would wipe badges. */
const FRAME_GAP_PAUSE_MS = LIVENESS_INTERVAL_MS * 2;
/** A silent agent is treated as dead after this bounded prune pause. */
const PRUNE_PAUSE_MAX_MS = 3 * 60_000;
/** Maximum concurrent active turns tracked per agent (matches pool size). */
const MAX_TURNS_PER_AGENT = 4;
/** Cap on per-agent terminal tombstones (A's resurrection guard). Only the
@@ -123,6 +125,11 @@ function sampleClockOffset(agentKey: string, timestamp: string): boolean {
return true;
}
function parseTimestamp(timestamp: string): number | null {
const parsed = Date.parse(timestamp);
return Number.isFinite(parsed) ? parsed : null;
}
function startTurn(
agentPubkey: string,
channelId: string,
@@ -151,7 +158,7 @@ function startTurn(
}
}
const startedAt = Date.parse(timestamp) || Date.now();
const startedAt = parseTimestamp(timestamp) ?? Date.now();
agentTurns.set(turnId, {
turnId,
channelId,
@@ -179,23 +186,31 @@ function recordActivity(agentPubkey: string, turnId: string | null): boolean {
* A recovered liveness/acp frame for a turn no longer in the live map recreates
* it, UNLESS C's tombstone shows the turn already terminally ended at or after
* this frame's time (a stale frame must not revive a completed turn). The frame
* carries no record of the original start, so the badge re-anchors to this
* frame's timestamp — it resumes counting from recovery, which is the honest
* floor for a turn whose true start is unrecoverable. Returns true on revive.
* may carry its original `startedAt` envelope field; when valid and not later
* than the frame, preserve the elapsed timer by anchoring to that timestamp.
* Old, malformed, or impossible future starts fall back to the recovery
* timestamp. Returns true on revive.
*/
function resurrectTurn(agentPubkey: string, event: ObserverEvent): boolean {
if (!event.turnId || !event.channelId) return false;
const key = normalizePubkey(agentPubkey);
const terminalAt = terminalAtByAgent.get(key)?.get(event.turnId);
const frameAt = Date.parse(event.timestamp);
const frameAt = parseTimestamp(event.timestamp);
// Only revive when this frame is strictly newer than the recorded terminal.
if (
terminalAt !== undefined &&
(!Number.isFinite(frameAt) || frameAt <= terminalAt)
) {
if (terminalAt !== undefined && (frameAt === null || frameAt <= terminalAt)) {
return false;
}
startTurn(agentPubkey, event.channelId, event.turnId, event.timestamp);
const startedAt =
typeof event.startedAt === "string" &&
parseTimestamp(event.startedAt) !== null
? event.startedAt
: event.timestamp;
const startedAtMs = parseTimestamp(startedAt);
const safeStartedAt =
frameAt !== null && startedAtMs !== null && startedAtMs <= frameAt
? startedAt
: event.timestamp;
startTurn(agentPubkey, event.channelId, event.turnId, safeStartedAt);
return true;
}
@@ -255,34 +270,34 @@ function endTurn(
invalidateCache(key);
}
/** True when every tracked turn across every agent is simultaneously stale —
* no turn has had activity within FRAME_GAP_PAUSE_MS. With no tracked turns
* there is nothing to prune, so it returns false (never pause). */
function shouldPausePrune(now: number): boolean {
/** True when every tracked turn for one agent is stale, but only until the
* bounded backstop expires. Other agents' activity intentionally has no effect. */
function shouldPausePrune(
agentTurns: Map<string, ActiveTurn>,
now: number,
): boolean {
let maxActivity = 0;
for (const agentTurns of activeTurnsByAgent.values())
for (const turn of agentTurns.values())
if (turn.lastActivityAt > maxActivity) maxActivity = turn.lastActivityAt;
return maxActivity > 0 && now - maxActivity > FRAME_GAP_PAUSE_MS;
for (const turn of agentTurns.values()) {
if (turn.lastActivityAt > maxActivity) maxActivity = turn.lastActivityAt;
}
const silentFor = now - maxActivity;
return (
maxActivity > 0 &&
silentFor > FRAME_GAP_PAUSE_MS &&
silentFor < PRUNE_PAUSE_MAX_MS
);
}
function pruneExpired() {
const now = Date.now();
// Pause pruning when ALL tracked turns are simultaneously stale — the "all
// at once" signature of a relay drop, where every agent's liveness stops in
// the same instant. Gating on the MAX lastActivityAt (not a global frame
// clock) is what keeps this from over-pausing: a single live sibling turn
// keeps the max fresh, so a genuinely dead turn still prunes at 25s — no
// regression for the multi-agent crash case. Residual: a LONE turn kill -9'd
// under a HEALTHY relay (it was the only active turn) keeps its badge until
// the next frame instead of clearing at 25s, since local-only sensing cannot
// distinguish that from a drop. The badge self-heals the instant any frame
// arrives. Accepted tradeoff to keep badges visible through transient drops.
if (shouldPausePrune(now)) {
return;
}
let changed = false;
for (const [agentKey, agentTurns] of activeTurnsByAgent) {
// A single fresh tracked turn for this agent means a stale sibling is
// genuinely dead and must still prune at 25s. Conversely, all of this
// agent's turns going stale together identifies a per-agent frame-stream
// gap, regardless of whether other agents keep reporting activity.
if (shouldPausePrune(agentTurns, now)) continue;
for (const [turnId, turn] of agentTurns) {
if (now - turn.lastActivityAt > REMOVE_AFTER_MS) {
agentTurns.delete(turnId);

1
desktop/src/features/agents/ui/agentSessionTypes.ts
View File

@@ -8,6 +8,7 @@ export type ObserverEvent = {
channelId: string | null;
sessionId: string | null;
turnId: string | null;
startedAt?: string | null;
payload: unknown;
};