From 42908201de343edcb284f4a825b2388ee6bc49cf Mon Sep 17 00:00:00 2001 From: Beaten0123 Date: Sun, 15 Nov 2026 17:41:21 +0800 Subject: [PATCH 1/5] fix(agent): recover stale uncertain attention wake without starving tasks --- .../src/agent_attention_tests.rs | 287 ++++++++++++++++++ crates/webcodex-store/src/agent_wake.rs | 62 +++- .../agent/mcp-app-continuation-experiments.md | 2 + 3 files changed, 347 insertions(+), 4 deletions(-) diff --git a/crates/webcodex-store/src/agent_attention_tests.rs b/crates/webcodex-store/src/agent_attention_tests.rs index e106f164e..c4c862e6b 100644 --- a/crates/webcodex-store/src/agent_attention_tests.rs +++ b/crates/webcodex-store/src/agent_attention_tests.rs @@ -2,6 +2,7 @@ use super::agent_attention::{ attention_events_for_attempt, AGENT_ATTENTION_EVENT_KIND_AGENT_TASK_TERMINAL, }; use super::agent_task::{AgentTaskState, NewAgentTask}; +use super::agent_wake::STALE_ATTENTION_DELIVERY_UNKNOWN_MS; use super::agent_wake::{AgentWakeState, WAKE_TRIGGER_ATTENTION_EVENT}; use super::communication::{NewAgentEndpoint, NewAgentIdentity}; use super::goal::{GoalCorrelationKind, GoalLifecycle, GoalPatch, NewGoal, MAX_GOAL_CORRELATIONS}; @@ -1193,3 +1194,289 @@ fn goal_workflow_attention_reopen_preserves_task_fact_and_exact_wake_carrier() { AgentWakeState::Consumed ); } + +#[test] +fn stale_unknown_attention_cannot_hold_a_fresh_agent_task_or_reuse_its_token() { + let temp = tempfile::tempdir().unwrap(); + let db = Database::open(&temp.path().join("stale-attention-fence.db")).unwrap(); + let owner = principal('d'); + let worker = create_agent(&db, &owner, "stale-attention-worker"); + let controller = create_agent(&db, &owner, "stale-attention-controller"); + let (task_id, attempt_id, fence) = + create_task_and_attempt(&db, &owner, &worker, "stale-attention", T0); + create_goal_and_link_with_controller( + &db, + &owner, + &task_id, + Some(&controller), + "stale-attention", + T0 + 2, + ); + db.complete_agent_task_attempt_at( + &owner, + &task_id, + &attempt_id, + &worker, + &fence, + 1, + AgentTaskState::Succeeded, + Some("terminal notification"), + None, + "stale-attention-complete", + T0 + 4, + ) + .unwrap(); + let old_wake_id = wake_id_for_event(&db, &events(&db, &attempt_id)[0].event_id); + let endpoint = db + .attach_agent_endpoint( + &owner, + NewAgentEndpoint { + agent_id: controller.clone(), + host: "ChatGPT".to_string(), + client_attachment_id: Some("stale-attention-window".to_string()), + wake_capable: true, + idempotency_key: "stale-attention-endpoint".to_string(), + }, + ) + .unwrap() + .endpoint; + let old = db + .claim_next_agent_wake( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app", + ) + .unwrap() + .unwrap(); + assert_eq!(old.wake.wake_id, old_wake_id); + db.prepare_agent_wake_dispatch( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old_wake_id, + &old.attempt.attempt_id, + &old.claim_fence, + &old.consume_token, + ) + .unwrap(); + db.mark_agent_wake_delivery_unknown( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old_wake_id, + &old.attempt.attempt_id, + &old.claim_fence, + ) + .unwrap(); + + assert!( + db.bootstrap_agent_conversation( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + None, + Some(&old_wake_id), + ) + .unwrap() + .wake + .is_some(), + "fresh unknown Wake must remain visible and fenced" + ); + + // A second historical notification remains pending. A time-sensitive + // AgentTask should take priority without discarding that notification. + let (notice_task, notice_attempt, notice_fence) = + create_task_and_attempt(&db, &owner, &worker, "queued-attention", T0 + 100); + create_goal_and_link_with_controller( + &db, + &owner, + ¬ice_task, + Some(&controller), + "queued-attention", + T0 + 102, + ); + db.complete_agent_task_attempt_at( + &owner, + ¬ice_task, + ¬ice_attempt, + &worker, + ¬ice_fence, + 1, + AgentTaskState::Succeeded, + Some("second notification"), + None, + "queued-attention-complete", + T0 + 104, + ) + .unwrap(); + let queued_notification = wake_id_for_event(&db, &events(&db, ¬ice_attempt)[0].event_id); + + // The next real Task gets a new, independent Wake while the Host's + // previous notification is uncertain. Do not claim either too early. + let active_task = db + .create_agent_task( + &owner, + NewAgentTask { + title: "Continue real Action Registry work".to_string(), + instruction: "Use the current Goal checkpoint".to_string(), + assignee_agent_id: Some(controller.clone()), + source_conversation_id: None, + source_message_id: None, + referenced_project_id: None, + idempotency_key: "after-stale-attention-task".to_string(), + }, + ) + .unwrap() + .task + .summary + .task_id; + let started = db + .start_agent_task_attempt( + &owner, + &active_task, + &controller, + "after-stale-attention-attempt", + ) + .unwrap(); + let execution = db + .start_agent_task_endpoint_continuation( + &owner, + &active_task, + &started.attempt.attempt_id, + &controller, + &started.attempt_fence, + started.attempt.attempt_controller_generation, + ) + .unwrap(); + let new_wake_id = execution.execution.wake_id; + assert!(db + .claim_next_agent_wake( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app", + ) + .unwrap() + .is_none()); + + // Test the timer boundary without sleeping or relying on a machine clock. + // Only the *same* uncertain notification and its exact dispatch attempt + // are aged in this isolated database. + let expired_at = + super::store_primitives::now_unix_ms() - STALE_ATTENTION_DELIVERY_UNKNOWN_MS - 1; + db.conn_for_tests() + .execute( + "UPDATE wc_agent_wakes SET updated_at_unix_ms = ?2 WHERE wake_id = ?1", + params![old_wake_id, expired_at], + ) + .unwrap(); + db.conn_for_tests() + .execute( + "UPDATE wc_agent_wake_attempts + SET delivery_unknown_at_unix_ms = ?2 WHERE wake_id = ?1", + params![old_wake_id, expired_at], + ) + .unwrap(); + + // The Model-facing bootstrap is read-only but must not keep a cached + // MCP App controller pinned to an abandoned exact notification. The + // subsequent claim performs the atomic retirement and Task selection. + assert!(db + .bootstrap_agent_conversation( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + None, + Some(&old_wake_id), + ) + .unwrap() + .wake + .is_none()); + assert_eq!( + db.bootstrap_agent_conversation( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + None, + None, + ) + .unwrap() + .wake + .unwrap() + .wake_id, + new_wake_id + ); + + let next = db + .claim_next_agent_wake( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app", + ) + .unwrap() + .unwrap(); + assert_eq!(next.wake.wake_id, new_wake_id); + assert_eq!( + db.agent_wake(&old_wake_id).unwrap().unwrap().state, + AgentWakeState::Retired + ); + assert_eq!( + db.agent_wake_attempts(&old_wake_id).unwrap()[0].state, + super::agent_wake::AgentWakeAttemptState::Revoked, + ); + // A delayed Host turn cannot use the abandoned notification's token. + assert!(db + .consume_agent_wake( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old_wake_id, + &old.consume_token, + ) + .is_err()); + + db.prepare_agent_wake_dispatch( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + &new_wake_id, + &next.attempt.attempt_id, + &next.claim_fence, + &next.consume_token, + ) + .unwrap(); + let consumed = db + .consume_agent_wake( + &owner, + &controller, + &endpoint.endpoint_id, + endpoint.controller_generation, + &new_wake_id, + &next.consume_token, + ) + .unwrap(); + assert_eq!(consumed.state, AgentWakeState::Consumed); + assert_eq!( + db.agent_wake(&queued_notification).unwrap().unwrap().state, + AgentWakeState::Pending, + "the older pending notification must remain durable, not vanish", + ); + assert!(db + .agent_wake(&old_wake_id) + .unwrap() + .unwrap() + .consumed_at_unix_ms + .is_none()); +} diff --git a/crates/webcodex-store/src/agent_wake.rs b/crates/webcodex-store/src/agent_wake.rs index 4ae88346c..96b66e048 100644 --- a/crates/webcodex-store/src/agent_wake.rs +++ b/crates/webcodex-store/src/agent_wake.rs @@ -32,6 +32,12 @@ pub(crate) const AGENT_WAKE_CLAIM_FENCE_PREFIX: &str = "wc_wake_claim_"; pub const AGENT_WAKE_CONSUME_TOKEN_PREFIX: &str = "wc_wake_consume_"; const DEFAULT_WAKE_CLAIM_LEASE_MS: i64 = 30_000; +// A Host dispatch with no outcome must never be blindly retried. However, +// an abandoned *notification* cannot fence unrelated future AgentTask work +// forever. Once the bounded Host scheduling grace has elapsed, retire that +// exact attention Wake and revoke its consume token; preserve the source +// event and immutable attempt history for subsequent Goal recovery. +pub(crate) const STALE_ATTENTION_DELIVERY_UNKNOWN_MS: i64 = 15 * 60_000; const MAX_ADAPTER_KIND_CHARS: usize = 64; const WAKE_TRIGGER_INBOX_CHANGED: &str = "inbox_changed"; pub(crate) const WAKE_TRIGGER_AGENT_TASK_ATTEMPT: &str = "agent_task_attempt"; @@ -542,6 +548,7 @@ impl Database { } release_expired_claims_for_agent(&transaction, agent_id, now)?; retire_stale_agent_task_wakes_for_agent(&transaction, agent_id, now)?; + retire_stale_unknown_attention_wakes_for_agent(&transaction, agent_id, now)?; let blocked: bool = transaction .query_row( "SELECT EXISTS( @@ -561,7 +568,8 @@ impl Database { .query_row( "SELECT wake_id FROM wc_agent_wakes WHERE target_agent_id = ?1 AND state = 'pending' - ORDER BY created_at_unix_ms, wake_id LIMIT 1", + ORDER BY CASE WHEN trigger_kind = 'agent_task_attempt' THEN 0 ELSE 1 END, + created_at_unix_ms, wake_id LIMIT 1", params![agent_id], |row| row.get(0), ) @@ -1478,20 +1486,29 @@ impl Database { (!matches!( wake.state, AgentWakeState::Consumed | AgentWakeState::Retired - )) + ) && !(wake.trigger_kind == WAKE_TRIGGER_ATTENTION_EVENT + && wake.state == AgentWakeState::DeliveryUnknown + && wake.updated_at_unix_ms + <= now_unix_ms().saturating_sub(STALE_ATTENTION_DELIVERY_UNKNOWN_MS))) .then_some(wake) } else { let selected_wake_id: Option = conn .query_row( "SELECT wake_id FROM wc_agent_wakes WHERE target_agent_id = ?1 AND state NOT IN ('consumed', 'retired') + AND NOT (trigger_kind = 'attention_event' + AND state = 'delivery_unknown' AND updated_at_unix_ms <= ?2) ORDER BY CASE WHEN state IN ('prepared', 'delivered', 'delivery_unknown') THEN 0 WHEN state = 'claimed' THEN 1 ELSE 2 - END, created_at_unix_ms, wake_id + END, CASE WHEN trigger_kind = 'agent_task_attempt' THEN 0 ELSE 1 END, + created_at_unix_ms, wake_id LIMIT 1", - params![agent_id], + params![ + agent_id, + now_unix_ms().saturating_sub(STALE_ATTENTION_DELIVERY_UNKNOWN_MS) + ], |row| row.get(0), ) .optional() @@ -2001,6 +2018,43 @@ fn materialize_agent_task_wake_expiry( Ok(()) } +/// Resolve only old, outcome-unknown *attention notifications*. This never +/// resends a Host message or touches pending inbox, AgentWait, or executable +/// AgentTask wakes. The SQLite transaction serializes a racing exact consume: +/// consumed wins unchanged, or retirement wins and late consume fails closed. +fn retire_stale_unknown_attention_wakes_for_agent( + transaction: &Transaction<'_>, + agent_id: &str, + now: i64, +) -> Result<(), CommunicationStoreError> { + let cutoff = now.saturating_sub(STALE_ATTENTION_DELIVERY_UNKNOWN_MS); + transaction + .execute( + "UPDATE wc_agent_wake_attempts + SET state = 'revoked', revoked_at_unix_ms = COALESCE(revoked_at_unix_ms, ?3) + WHERE state = 'delivery_unknown' AND wake_id IN ( + SELECT wake_id FROM wc_agent_wakes + WHERE target_agent_id = ?1 AND trigger_kind = 'attention_event' + AND state = 'delivery_unknown' AND updated_at_unix_ms <= ?2 + )", + params![agent_id, cutoff, now], + ) + .map_err(store_error)?; + transaction + .execute( + "UPDATE wc_agent_wakes + SET state = 'retired', revision = revision + 1, + updated_at_unix_ms = MAX(updated_at_unix_ms, ?3), + claimed_attempt_id = NULL, claimed_endpoint_id = NULL, + claimed_controller_generation = NULL, claim_lease_expires_at_unix_ms = NULL + WHERE target_agent_id = ?1 AND trigger_kind = 'attention_event' + AND state = 'delivery_unknown' AND updated_at_unix_ms <= ?2", + params![agent_id, cutoff, now], + ) + .map_err(store_error)?; + Ok(()) +} + fn retire_agent_task_wake_pre_dispatch( transaction: &Transaction<'_>, wake_id: &str, diff --git a/docs/agent/mcp-app-continuation-experiments.md b/docs/agent/mcp-app-continuation-experiments.md index 0302523a7..9ff0675a4 100644 --- a/docs/agent/mcp-app-continuation-experiments.md +++ b/docs/agent/mcp-app-continuation-experiments.md @@ -220,6 +220,8 @@ The View binding is process-local fencing, not durable authority. Every App-only The App keeps claim fences entirely Server-side. The bounded automatic message is returned only after prepare and contains exact `agent_id`, `endpoint_id`, `controller_generation`, `wake_id`, and `consume_token`; it contains no Conversation Message body, transcript, Agent private description/specialty labels, credential, principal digest, claim fence, Project authority, or Workflow Session authority. The View generates a stable secure random binding fence and sends it in bind input; same-current-View retries renew without replacing its claim or dispatch phase. The automatic message uses the app-only standard `structuredContent.output.app_protocol` result channel. Neither value enters ordinary model-visible projections, typed audit/session projections, or forensic tool-request payload capture. Continuation correctness does not depend on custom ToolResult `_meta`. `ui/message` success is recorded only as `dispatch_accepted`. Timeout, reload, View loss, or any post-fence outcome that cannot prove non-delivery becomes `delivery_unknown`; the App never automatically sends a second `ui/message` for that Attempt. If the new model turn starts before the Host ACK is recorded, exact `consume_agent_wake` may win first; the later ACK is idempotent and cannot move the Wake back from `consumed`. Only exact consume is production evidence of `continuation_consumed`. +A long-abandoned **attention-event notification** in `delivery_unknown` must also not fence unrelated work forever. The Store gives the exact uncertain Host dispatch a **15-minute** scheduling/consume grace. On the next authorized Wake acquire after that bound, it atomically retires only that notification and revokes its exact Wake Attempt; its source attention event and forensic timestamps remain durable. A delayed Host continuation cannot consume the revoked token. The Controller's read-only exact bootstrap excludes such stale notifications so an already-mounted MCP App drops its obsolete cached claim and re-enters the one canonical acquire path. A still-valid AgentTask Attempt Wake is selected before older pending notifications, since task execution has a short lease; deferred notifications remain queued. **No old `ui/message` is resent**, and prepared/delivered Wakes, inbox messages, AgentWait Wakes and execution-bound AgentTask Wakes are not silently retired. This is a bounded recovery mechanism, not proof the Host scheduled a new model turn; `consume_agent_wake` remains the acceptance boundary. + Production v16 applies the already-proven background-carrier policy directly: a hidden/background View keeps the slower bounded heartbeat cadence but may observe, acquire, prepare, send the exact `ui/message`, and finish the Host outcome without waiting for foreground. Visibility transitions themselves do not create `delivery_unknown`; only the existing prepare/Host-response uncertainty, binding loss, or teardown/replacement paths do. Pagehide, beforeunload, and `ui/resource-teardown` still stop polling and attempt exact best-effort unbind. Correctness never depends on reliable teardown, browser memory, timer liveness, or localStorage. From a460d3b9ebaeffadd202737e1ed9e5275c231916 Mon Sep 17 00:00:00 2001 From: Beaten0123 Date: Sun, 15 Nov 2026 19:40:10 +0800 Subject: [PATCH 2/5] fix(agent): retire abandoned uncertain Task Wake only after its Attempt expires --- crates/webcodex-store/src/agent_task_tests.rs | 198 ++++++++++++++++++ crates/webcodex-store/src/agent_wake.rs | 105 +++++++++- .../agent/mcp-app-continuation-experiments.md | 2 + 3 files changed, 299 insertions(+), 6 deletions(-) diff --git a/crates/webcodex-store/src/agent_task_tests.rs b/crates/webcodex-store/src/agent_task_tests.rs index 6db19321d..70a024467 100644 --- a/crates/webcodex-store/src/agent_task_tests.rs +++ b/crates/webcodex-store/src/agent_task_tests.rs @@ -5247,3 +5247,201 @@ fn endpoint_continuation_consumed_wake_stays_consumed_after_task_completion_but_ Some(AgentTaskExecutionStatus::Terminal) ); } + +#[test] +fn expired_uncertain_task_wake_fences_old_token_and_allows_new_host_takeover() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("expired-unknown-task-wake.db"); + let db = Database::open(&path).unwrap(); + let owner = principal('b'); + let start_at = wall_now_ms(); + let fixture = + prepared_endpoint_takeover_fixture(&db, &owner, "unknown-task-recovery", start_at); + db.mark_agent_wake_delivery_unknown_at( + &owner, + &fixture.assignee, + &fixture.endpoint_id, + fixture.endpoint_controller_generation, + &fixture.wake_id, + &fixture.wake_attempt_id, + &fixture.wake_claim_fence, + start_at + 2, + ) + .unwrap(); + assert_eq!( + db.agent_wake(&fixture.wake_id).unwrap().unwrap().state, + AgentWakeState::DeliveryUnknown, + ); + + // Even if an uncertain Host dispatch looks old, the source execution + // remains authoritative until its own Attempt lease actually expires. + let stale_at = wall_now_ms() - super::agent_wake::STALE_TASK_DELIVERY_UNKNOWN_MS - 1_000; + db.conn_for_tests() + .execute( + "UPDATE wc_agent_wakes SET updated_at_unix_ms = ?2 WHERE wake_id = ?1", + rusqlite::params![fixture.wake_id, stale_at], + ) + .unwrap(); + assert!(db + .bootstrap_agent_conversation( + &owner, + &fixture.assignee, + &fixture.endpoint_id, + fixture.endpoint_controller_generation, + None, + Some(&fixture.wake_id), + ) + .unwrap() + .wake + .is_some()); + assert!(db + .claim_next_agent_wake( + &owner, + &fixture.assignee, + &fixture.endpoint_id, + fixture.endpoint_controller_generation, + "mcp_app", + ) + .unwrap() + .is_none()); + assert_eq!( + db.agent_wake(&fixture.wake_id).unwrap().unwrap().state, + AgentWakeState::DeliveryUnknown, + ); + + // Model a lost Host turn whose 5-minute one-shot dispatch grace has + // expired. This does not assert that an unknown dispatch was delivered. + db.conn_for_tests() + .execute( + "UPDATE wc_agent_task_attempts + SET lease_expires_at_unix_ms = ?2 + WHERE attempt_id = ?1 AND state = 'active'", + rusqlite::params![fixture.task_attempt_id, wall_now_ms() - 1], + ) + .unwrap(); + // The read-only Host bootstrap must stop showing the obsolete Wake even + // before the canonical claim path performs transactional retirement. + assert!(db + .bootstrap_agent_conversation( + &owner, + &fixture.assignee, + &fixture.endpoint_id, + fixture.endpoint_controller_generation, + None, + Some(&fixture.wake_id), + ) + .unwrap() + .wake + .is_none()); + assert!(db + .bootstrap_agent_conversation( + &owner, + &fixture.assignee, + &fixture.endpoint_id, + fixture.endpoint_controller_generation, + None, + None, + ) + .unwrap() + .wake + .is_none()); + assert!(db + .claim_next_agent_wake( + &owner, + &fixture.assignee, + &fixture.endpoint_id, + fixture.endpoint_controller_generation, + "mcp_app", + ) + .unwrap() + .is_none()); + + assert_eq!( + db.agent_wake(&fixture.wake_id).unwrap().unwrap().state, + AgentWakeState::Retired, + ); + assert_eq!( + db.agent_wake_attempts(&fixture.wake_id).unwrap()[0].state, + super::agent_wake::AgentWakeAttemptState::Revoked, + ); + assert_eq!( + db.read_agent_task(&owner, &fixture.task_id) + .unwrap() + .summary + .state, + AgentTaskState::Ready, + ); + // Retired exact Host tokens must never ACK or revive an expired Attempt. + assert!(db + .consume_agent_wake( + &owner, + &fixture.assignee, + &fixture.endpoint_id, + fixture.endpoint_controller_generation, + &fixture.wake_id, + &fixture.consume_token, + ) + .is_err()); + + // A new, separately fenced Attempt can now be delivered and exactly + // consumed; this is not a replay of the old outcome-unknown Host message. + let next = start( + &db, + &owner, + &fixture.task_id, + &fixture.assignee, + "unknown-task-recovery-successor", + wall_now_ms(), + ); + assert_eq!(next.attempt.attempt_number, 2); + let execution = db + .start_agent_task_endpoint_continuation_at( + &owner, + &fixture.task_id, + &next.attempt.attempt_id, + &fixture.assignee, + &next.attempt_fence, + next.attempt.attempt_controller_generation, + wall_now_ms(), + ) + .unwrap(); + let claim = db + .claim_next_agent_wake( + &owner, + &fixture.assignee, + &fixture.endpoint_id, + fixture.endpoint_controller_generation, + "mcp_app", + ) + .unwrap() + .expect("new Task Wake must no longer be starved"); + assert_eq!(claim.wake.wake_id, execution.execution.wake_id); + db.prepare_agent_wake_dispatch( + &owner, + &fixture.assignee, + &fixture.endpoint_id, + fixture.endpoint_controller_generation, + &claim.wake.wake_id, + &claim.attempt.attempt_id, + &claim.claim_fence, + &claim.consume_token, + ) + .unwrap(); + db.consume_agent_wake( + &owner, + &fixture.assignee, + &fixture.endpoint_id, + fixture.endpoint_controller_generation, + &claim.wake.wake_id, + &claim.consume_token, + ) + .unwrap(); + assert_eq!( + db.agent_wake(&claim.wake.wake_id).unwrap().unwrap().state, + AgentWakeState::Consumed, + ); + assert_eq!( + db.agent_wake(&fixture.wake_id).unwrap().unwrap().state, + AgentWakeState::Retired, + ); +} diff --git a/crates/webcodex-store/src/agent_wake.rs b/crates/webcodex-store/src/agent_wake.rs index 96b66e048..44aecd637 100644 --- a/crates/webcodex-store/src/agent_wake.rs +++ b/crates/webcodex-store/src/agent_wake.rs @@ -38,6 +38,9 @@ const DEFAULT_WAKE_CLAIM_LEASE_MS: i64 = 30_000; // exact attention Wake and revoke its consume token; preserve the source // event and immutable attempt history for subsequent Goal recovery. pub(crate) const STALE_ATTENTION_DELIVERY_UNKNOWN_MS: i64 = 15 * 60_000; +// A Task Wake needs both a bounded Host scheduling window and an expired +// exact source Attempt before its uncertain delivery may be retired. +pub(crate) const STALE_TASK_DELIVERY_UNKNOWN_MS: i64 = 15 * 60_000; const MAX_ADAPTER_KIND_CHARS: usize = 64; const WAKE_TRIGGER_INBOX_CHANGED: &str = "inbox_changed"; pub(crate) const WAKE_TRIGGER_AGENT_TASK_ATTEMPT: &str = "agent_task_attempt"; @@ -549,6 +552,7 @@ impl Database { release_expired_claims_for_agent(&transaction, agent_id, now)?; retire_stale_agent_task_wakes_for_agent(&transaction, agent_id, now)?; retire_stale_unknown_attention_wakes_for_agent(&transaction, agent_id, now)?; + retire_stale_unknown_task_wakes_for_agent(&transaction, agent_id, now)?; let blocked: bool = transaction .query_row( "SELECT EXISTS( @@ -1483,14 +1487,22 @@ impl Database { .ok_or_else(|| { CommunicationStoreError::new("wake_not_found", "Agent Wake does not exist") })?; + let observed_at = now_unix_ms(); + let stale_task = wake.trigger_kind == WAKE_TRIGGER_AGENT_TASK_ATTEMPT + && wake.state == AgentWakeState::DeliveryUnknown + && wake.updated_at_unix_ms + <= observed_at.saturating_sub(STALE_TASK_DELIVERY_UNKNOWN_MS) + && agent_task_wake_source_terminal(&conn, &wake, observed_at)?; + let stale_attention = wake.trigger_kind == WAKE_TRIGGER_ATTENTION_EVENT + && wake.state == AgentWakeState::DeliveryUnknown + && wake.updated_at_unix_ms + <= observed_at.saturating_sub(STALE_ATTENTION_DELIVERY_UNKNOWN_MS); (!matches!( wake.state, AgentWakeState::Consumed | AgentWakeState::Retired - ) && !(wake.trigger_kind == WAKE_TRIGGER_ATTENTION_EVENT - && wake.state == AgentWakeState::DeliveryUnknown - && wake.updated_at_unix_ms - <= now_unix_ms().saturating_sub(STALE_ATTENTION_DELIVERY_UNKNOWN_MS))) - .then_some(wake) + ) && !stale_attention + && !stale_task) + .then_some(wake) } else { let selected_wake_id: Option = conn .query_row( @@ -1498,6 +1510,13 @@ impl Database { WHERE target_agent_id = ?1 AND state NOT IN ('consumed', 'retired') AND NOT (trigger_kind = 'attention_event' AND state = 'delivery_unknown' AND updated_at_unix_ms <= ?2) + AND NOT (trigger_kind = 'agent_task_attempt' + AND state = 'delivery_unknown' AND updated_at_unix_ms <= ?3 + AND EXISTS ( + SELECT 1 FROM wc_agent_task_attempts a + WHERE a.attempt_id = source_task_attempt_id + AND (a.state != 'active' OR a.lease_expires_at_unix_ms <= ?4) + )) ORDER BY CASE WHEN state IN ('prepared', 'delivered', 'delivery_unknown') THEN 0 WHEN state = 'claimed' THEN 1 @@ -1507,7 +1526,9 @@ impl Database { LIMIT 1", params![ agent_id, - now_unix_ms().saturating_sub(STALE_ATTENTION_DELIVERY_UNKNOWN_MS) + now_unix_ms().saturating_sub(STALE_ATTENTION_DELIVERY_UNKNOWN_MS), + now_unix_ms().saturating_sub(STALE_TASK_DELIVERY_UNKNOWN_MS), + now_unix_ms(), ], |row| row.get(0), ) @@ -2055,6 +2076,78 @@ fn retire_stale_unknown_attention_wakes_for_agent( Ok(()) } +/// A stale Task-origin dispatch must never be retried or silently promoted. +/// Retire only after the old exact Attempt is terminal or its lease expired; +/// a racing exact consume wins or fails closed in the same SQLite transaction. +fn agent_task_wake_source_terminal( + conn: &Connection, + wake: &AgentWakeRecord, + now: i64, +) -> Result { + let Some(attempt_id) = wake.source_task_attempt_id.as_deref() else { + return Ok(false); + }; + conn.query_row( + "SELECT EXISTS(SELECT 1 FROM wc_agent_task_attempts + WHERE attempt_id = ?1 AND (state != 'active' OR lease_expires_at_unix_ms <= ?2))", + params![attempt_id, now], + |row| row.get(0), + ) + .map_err(store_error) +} + +fn retire_stale_unknown_task_wakes_for_agent( + transaction: &Transaction<'_>, + agent_id: &str, + now: i64, +) -> Result<(), CommunicationStoreError> { + let cutoff = now.saturating_sub(STALE_TASK_DELIVERY_UNKNOWN_MS); + let wake_ids = { + let mut statement = transaction + .prepare( + "SELECT w.wake_id FROM wc_agent_wakes w + JOIN wc_agent_task_attempts a ON a.attempt_id = w.source_task_attempt_id + WHERE w.target_agent_id = ?1 AND w.trigger_kind = 'agent_task_attempt' + AND w.state = 'delivery_unknown' AND w.updated_at_unix_ms <= ?2 + AND (a.state != 'active' OR a.lease_expires_at_unix_ms <= ?3)", + ) + .map_err(store_error)?; + let rows = statement + .query_map(params![agent_id, cutoff, now], |row| { + row.get::<_, String>(0) + }) + .map_err(store_error)? + .collect::, _>>() + .map_err(store_error)?; + rows + }; + for wake_id in wake_ids { + let wake = load_wake(transaction, &wake_id)?.expect("selected Task Wake must exist"); + materialize_agent_task_wake_expiry(transaction, &wake, now)?; + transaction + .execute( + "UPDATE wc_agent_wake_attempts + SET state = 'revoked', revoked_at_unix_ms = COALESCE(revoked_at_unix_ms, ?2) + WHERE wake_id = ?1 AND state = 'delivery_unknown'", + params![wake_id, now], + ) + .map_err(store_error)?; + clear_agent_task_wake_carrier(transaction, &wake_id, now)?; + transaction + .execute( + "UPDATE wc_agent_wakes + SET state = 'retired', revision = revision + 1, + updated_at_unix_ms = MAX(updated_at_unix_ms, ?2), + claimed_attempt_id = NULL, claimed_endpoint_id = NULL, + claimed_controller_generation = NULL, claim_lease_expires_at_unix_ms = NULL + WHERE wake_id = ?1 AND state = 'delivery_unknown'", + params![wake_id, now], + ) + .map_err(store_error)?; + } + Ok(()) +} + fn retire_agent_task_wake_pre_dispatch( transaction: &Transaction<'_>, wake_id: &str, diff --git a/docs/agent/mcp-app-continuation-experiments.md b/docs/agent/mcp-app-continuation-experiments.md index 9ff0675a4..b3ac77d88 100644 --- a/docs/agent/mcp-app-continuation-experiments.md +++ b/docs/agent/mcp-app-continuation-experiments.md @@ -221,6 +221,8 @@ The App keeps claim fences entirely Server-side. The bounded automatic message i `ui/message` success is recorded only as `dispatch_accepted`. Timeout, reload, View loss, or any post-fence outcome that cannot prove non-delivery becomes `delivery_unknown`; the App never automatically sends a second `ui/message` for that Attempt. If the new model turn starts before the Host ACK is recorded, exact `consume_agent_wake` may win first; the later ACK is idempotent and cannot move the Wake back from `consumed`. Only exact consume is production evidence of `continuation_consumed`. A long-abandoned **attention-event notification** in `delivery_unknown` must also not fence unrelated work forever. The Store gives the exact uncertain Host dispatch a **15-minute** scheduling/consume grace. On the next authorized Wake acquire after that bound, it atomically retires only that notification and revokes its exact Wake Attempt; its source attention event and forensic timestamps remain durable. A delayed Host continuation cannot consume the revoked token. The Controller's read-only exact bootstrap excludes such stale notifications so an already-mounted MCP App drops its obsolete cached claim and re-enters the one canonical acquire path. A still-valid AgentTask Attempt Wake is selected before older pending notifications, since task execution has a short lease; deferred notifications remain queued. **No old `ui/message` is resent**, and prepared/delivered Wakes, inbox messages, AgentWait Wakes and execution-bound AgentTask Wakes are not silently retired. This is a bounded recovery mechanism, not proof the Host scheduled a new model turn; `consume_agent_wake` remains the acceptance boundary. +An outcome-unknown **AgentTask Attempt Wake** has a distinct recovery rule: retain the exact dispatch fence throughout the **15-minute** Host scheduling window **and** for as long as the source Attempt is authoritative with an unexpired lease. Only once *both* conditions are satisfied may the next authorized Store acquire atomically retire that specific `delivery_unknown` Task Wake, revoke its dispatch/consume capability and release its stale execution carrier. Original Task, Attempt and execution records remain auditable; the stale Attempt is materialized as expired, never revived. The read-only Controller bootstrap likewise suppresses only such abandoned Task Wakes, letting the mounted App reach canonical acquire and a separately fenced successor. A later old Host turn cannot replay the revoked Token; a new Task Attempt is required and can only be started explicitly after inspecting durable Task/Goal state. This prevents permanent queue starvation, **not** an automatic claim that `ui/message` launched a model turn. The acceptance test must still observe the new Task Wake's actual `consume_agent_wake` and valid execution lease. + Production v16 applies the already-proven background-carrier policy directly: a hidden/background View keeps the slower bounded heartbeat cadence but may observe, acquire, prepare, send the exact `ui/message`, and finish the Host outcome without waiting for foreground. Visibility transitions themselves do not create `delivery_unknown`; only the existing prepare/Host-response uncertainty, binding loss, or teardown/replacement paths do. Pagehide, beforeunload, and `ui/resource-teardown` still stop polling and attempt exact best-effort unbind. Correctness never depends on reliable teardown, browser memory, timer liveness, or localStorage. From 9967bfca9b88e287d65e74ec9647bbaa7852632b Mon Sep 17 00:00:00 2001 From: Beaten0123 Date: Sun, 15 Nov 2026 19:53:18 +0800 Subject: [PATCH 3/5] fix(agent): unblock new Wakes behind stale uncertain Inbox dispatch --- crates/webcodex-store/src/agent_wake.rs | 76 +++- crates/webcodex-store/src/agent_wake_tests.rs | 363 ++++++++++++++++++ 2 files changed, 438 insertions(+), 1 deletion(-) diff --git a/crates/webcodex-store/src/agent_wake.rs b/crates/webcodex-store/src/agent_wake.rs index 44aecd637..8d637f27c 100644 --- a/crates/webcodex-store/src/agent_wake.rs +++ b/crates/webcodex-store/src/agent_wake.rs @@ -38,6 +38,9 @@ const DEFAULT_WAKE_CLAIM_LEASE_MS: i64 = 30_000; // exact attention Wake and revoke its consume token; preserve the source // event and immutable attempt history for subsequent Goal recovery. pub(crate) const STALE_ATTENTION_DELIVERY_UNKNOWN_MS: i64 = 15 * 60_000; +// An uncertain Inbox wake has no live execution lease, but a newer distinct +// pending Wake is required before it can safely yield its dispatch fence. +pub(crate) const STALE_INBOX_DELIVERY_UNKNOWN_MS: i64 = 15 * 60_000; // A Task Wake needs both a bounded Host scheduling window and an expired // exact source Attempt before its uncertain delivery may be retired. pub(crate) const STALE_TASK_DELIVERY_UNKNOWN_MS: i64 = 15 * 60_000; @@ -553,6 +556,7 @@ impl Database { retire_stale_agent_task_wakes_for_agent(&transaction, agent_id, now)?; retire_stale_unknown_attention_wakes_for_agent(&transaction, agent_id, now)?; retire_stale_unknown_task_wakes_for_agent(&transaction, agent_id, now)?; + retire_stale_unknown_inbox_wakes_for_agent(&transaction, agent_id, now)?; let blocked: bool = transaction .query_row( "SELECT EXISTS( @@ -1493,6 +1497,18 @@ impl Database { && wake.updated_at_unix_ms <= observed_at.saturating_sub(STALE_TASK_DELIVERY_UNKNOWN_MS) && agent_task_wake_source_terminal(&conn, &wake, observed_at)?; + let stale_inbox = wake.trigger_kind == WAKE_TRIGGER_INBOX_CHANGED + && wake.state == AgentWakeState::DeliveryUnknown + && wake.updated_at_unix_ms + <= observed_at.saturating_sub(STALE_INBOX_DELIVERY_UNKNOWN_MS) + && conn + .query_row( + "SELECT EXISTS(SELECT 1 FROM wc_agent_wakes + WHERE target_agent_id = ?1 AND wake_id != ?2 AND state = 'pending')", + params![agent_id, wake.wake_id], + |row| row.get::<_, bool>(0), + ) + .map_err(store_error)?; let stale_attention = wake.trigger_kind == WAKE_TRIGGER_ATTENTION_EVENT && wake.state == AgentWakeState::DeliveryUnknown && wake.updated_at_unix_ms @@ -1501,7 +1517,8 @@ impl Database { wake.state, AgentWakeState::Consumed | AgentWakeState::Retired ) && !stale_attention - && !stale_task) + && !stale_task + && !stale_inbox) .then_some(wake) } else { let selected_wake_id: Option = conn @@ -1510,6 +1527,14 @@ impl Database { WHERE target_agent_id = ?1 AND state NOT IN ('consumed', 'retired') AND NOT (trigger_kind = 'attention_event' AND state = 'delivery_unknown' AND updated_at_unix_ms <= ?2) + AND NOT (trigger_kind = 'inbox_changed' + AND state = 'delivery_unknown' AND updated_at_unix_ms <= ?5 + AND EXISTS ( + SELECT 1 FROM wc_agent_wakes successor + WHERE successor.target_agent_id = wc_agent_wakes.target_agent_id + AND successor.wake_id != wc_agent_wakes.wake_id + AND successor.state = 'pending' + )) AND NOT (trigger_kind = 'agent_task_attempt' AND state = 'delivery_unknown' AND updated_at_unix_ms <= ?3 AND EXISTS ( @@ -1529,6 +1554,7 @@ impl Database { now_unix_ms().saturating_sub(STALE_ATTENTION_DELIVERY_UNKNOWN_MS), now_unix_ms().saturating_sub(STALE_TASK_DELIVERY_UNKNOWN_MS), now_unix_ms(), + now_unix_ms().saturating_sub(STALE_INBOX_DELIVERY_UNKNOWN_MS), ], |row| row.get(0), ) @@ -2039,6 +2065,54 @@ fn materialize_agent_task_wake_expiry( Ok(()) } +/// An uncertain Inbox notification must not permanently block *new* work. +/// Only after Host scheduling grace and only with a distinct pending successor +/// may the old dispatch fence be retired. No old Host message is resent and no +/// queued Inbox Delivery is consumed; a late old token is revoked atomically. +fn retire_stale_unknown_inbox_wakes_for_agent( + transaction: &Transaction<'_>, + agent_id: &str, + now: i64, +) -> Result<(), CommunicationStoreError> { + let cutoff = now.saturating_sub(STALE_INBOX_DELIVERY_UNKNOWN_MS); + transaction + .execute( + "UPDATE wc_agent_wake_attempts + SET state = 'revoked', revoked_at_unix_ms = COALESCE(revoked_at_unix_ms, ?3) + WHERE state = 'delivery_unknown' AND wake_id IN ( + SELECT old.wake_id FROM wc_agent_wakes old + WHERE old.target_agent_id = ?1 AND old.trigger_kind = 'inbox_changed' + AND old.state = 'delivery_unknown' AND old.updated_at_unix_ms <= ?2 + AND EXISTS ( + SELECT 1 FROM wc_agent_wakes successor + WHERE successor.target_agent_id = old.target_agent_id + AND successor.wake_id != old.wake_id AND successor.state = 'pending' + ) + )", + params![agent_id, cutoff, now], + ) + .map_err(store_error)?; + transaction + .execute( + "UPDATE wc_agent_wakes + SET state = 'retired', revision = revision + 1, + updated_at_unix_ms = MAX(updated_at_unix_ms, ?3), + claimed_attempt_id = NULL, claimed_endpoint_id = NULL, + claimed_controller_generation = NULL, claim_lease_expires_at_unix_ms = NULL + WHERE target_agent_id = ?1 AND trigger_kind = 'inbox_changed' + AND state = 'delivery_unknown' AND updated_at_unix_ms <= ?2 + AND EXISTS ( + SELECT 1 FROM wc_agent_wakes successor + WHERE successor.target_agent_id = wc_agent_wakes.target_agent_id + AND successor.wake_id != wc_agent_wakes.wake_id + AND successor.state = 'pending' + )", + params![agent_id, cutoff, now], + ) + .map_err(store_error)?; + Ok(()) +} + /// Resolve only old, outcome-unknown *attention notifications*. This never /// resends a Host message or touches pending inbox, AgentWait, or executable /// AgentTask wakes. The SQLite transaction serializes a racing exact consume: diff --git a/crates/webcodex-store/src/agent_wake_tests.rs b/crates/webcodex-store/src/agent_wake_tests.rs index 8d631fa1a..24b523760 100644 --- a/crates/webcodex-store/src/agent_wake_tests.rs +++ b/crates/webcodex-store/src/agent_wake_tests.rs @@ -1,8 +1,10 @@ use super::agent_task::NewAgentTask; use super::agent_wake::{AgentWakeAttemptState, AgentWakeState}; +const STALE_INBOX_DELIVERY_UNKNOWN_MS: i64 = 15 * 60_000; use super::communication::{ ConversationAccess, NewAgentEndpoint, NewAgentIdentity, NewConversation, NewConversationMessage, }; +use super::store_primitives::now_unix_ms; use super::store_primitives::CommunicationPrincipal; use super::Database; use rusqlite::params; @@ -268,6 +270,367 @@ fn inbox_wake_consume_does_not_modify_task_attempt_lease() { ); } +#[test] +fn stale_unknown_inbox_wake_yields_to_new_durable_work_without_resending_old_host_message() { + let temp = tempfile::tempdir().unwrap(); + let db = Database::open(&temp.path().join("inbox-host-wake-recovery.db")).unwrap(); + let fixture = create_fixture(&db, '8'); + post_to_receiver( + &db, + &fixture, + "first durable request", + "inbox-unknown-first", + ); + let endpoint = attach_wake_endpoint(&db, &fixture, "ChatGPT", "inbox-unknown-endpoint"); + let old = db + .claim_next_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app", + ) + .unwrap() + .unwrap(); + db.prepare_agent_wake_dispatch( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old.wake.wake_id, + &old.attempt.attempt_id, + &old.claim_fence, + &old.consume_token, + ) + .unwrap(); + db.mark_agent_wake_delivery_unknown( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old.wake.wake_id, + &old.attempt.attempt_id, + &old.claim_fence, + ) + .unwrap(); + + // A fresh human request is a separate pending Wake, not an implicit retry. + post_to_receiver(&db, &fixture, "new durable request", "inbox-unknown-second"); + assert_eq!( + queued_delivery_ids(&db, &fixture.receiver_agent_id).len(), + 2 + ); + assert_eq!( + db.agent_wake(&old.wake.wake_id).unwrap().unwrap().state, + AgentWakeState::DeliveryUnknown + ); + + let current = now_unix_ms(); + db.conn_for_tests() + .execute( + "UPDATE wc_agent_wakes SET updated_at_unix_ms = ?2 WHERE wake_id = ?1", + params![ + old.wake.wake_id, + current - STALE_INBOX_DELIVERY_UNKNOWN_MS + 10_000 + ], + ) + .unwrap(); + assert!( + db.claim_next_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app", + ) + .unwrap() + .is_none(), + "uncertain Host delivery must retain its grace and never be resent" + ); + + db.conn_for_tests() + .execute( + "UPDATE wc_agent_wakes SET updated_at_unix_ms = ?2 WHERE wake_id = ?1", + params![ + old.wake.wake_id, + current - STALE_INBOX_DELIVERY_UNKNOWN_MS - 10_000 + ], + ) + .unwrap(); + let bootstrap = db + .bootstrap_agent_conversation( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + None, + None, + ) + .unwrap(); + assert_ne!(bootstrap.wake.unwrap().wake_id, old.wake.wake_id); + let stale_explicit = db + .bootstrap_agent_conversation( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + None, + Some(&old.wake.wake_id), + ) + .unwrap(); + assert!( + stale_explicit.wake.is_none(), + "expired old Host projection must not take priority" + ); + + let successor = db + .claim_next_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app", + ) + .unwrap() + .expect("new pending work must be able to proceed after bounded grace"); + assert_eq!(successor.wake.trigger_kind, "inbox_changed"); + assert_ne!(successor.wake.wake_id, old.wake.wake_id); + assert_eq!( + db.agent_wake(&old.wake.wake_id).unwrap().unwrap().state, + AgentWakeState::Retired + ); + assert_eq!( + db.agent_wake_attempts(&old.wake.wake_id).unwrap()[0].state, + AgentWakeAttemptState::Revoked + ); + assert!( + db.consume_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old.wake.wake_id, + &old.consume_token, + ) + .is_err(), + "late old Host turn cannot consume a retired Wake" + ); + assert_eq!( + queued_delivery_ids(&db, &fixture.receiver_agent_id).len(), + 2, + "no Inbox message may be dropped or silently consumed" + ); +} + +#[test] +fn stale_unknown_inbox_wake_without_successor_keeps_old_host_token_and_queued_delivery() { + let temp = tempfile::tempdir().unwrap(); + let db = Database::open(&temp.path().join("inbox-host-unknown-no-successor.db")).unwrap(); + let fixture = create_fixture(&db, '9'); + post_to_receiver(&db, &fixture, "no duplicate message", "inbox-only-first"); + let endpoint = attach_wake_endpoint(&db, &fixture, "ChatGPT", "inbox-only-endpoint"); + let old = db + .claim_next_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app", + ) + .unwrap() + .unwrap(); + db.prepare_agent_wake_dispatch( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old.wake.wake_id, + &old.attempt.attempt_id, + &old.claim_fence, + &old.consume_token, + ) + .unwrap(); + db.mark_agent_wake_delivery_unknown( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old.wake.wake_id, + &old.attempt.attempt_id, + &old.claim_fence, + ) + .unwrap(); + db.conn_for_tests() + .execute( + "UPDATE wc_agent_wakes SET updated_at_unix_ms = ?2 WHERE wake_id = ?1", + params![ + old.wake.wake_id, + now_unix_ms() - STALE_INBOX_DELIVERY_UNKNOWN_MS - 10_000 + ], + ) + .unwrap(); + assert!( + db.claim_next_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app", + ) + .unwrap() + .is_none(), + "no successor means the uncertain Host dispatch cannot be retried" + ); + assert_eq!( + db.agent_wake(&old.wake.wake_id).unwrap().unwrap().state, + AgentWakeState::DeliveryUnknown + ); + assert_eq!( + queued_delivery_ids(&db, &fixture.receiver_agent_id).len(), + 1 + ); + db.consume_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old.wake.wake_id, + &old.consume_token, + ) + .expect("a valid late exact Host turn retains its consume authority"); +} + +#[test] +fn old_unknown_inbox_dispatch_cannot_starve_new_endpoint_agent_task_after_grace() { + let temp = tempfile::tempdir().unwrap(); + let db = Database::open(&temp.path().join("inbox-blocks-task.db")).unwrap(); + let fixture = create_fixture(&db, 'a'); + post_to_receiver( + &db, + &fixture, + "original unknown host turn", + "inbox-before-task", + ); + let endpoint = attach_wake_endpoint(&db, &fixture, "ChatGPT", "inbox-before-task-endpoint"); + let old = db + .claim_next_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app", + ) + .unwrap() + .unwrap(); + db.prepare_agent_wake_dispatch( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old.wake.wake_id, + &old.attempt.attempt_id, + &old.claim_fence, + &old.consume_token, + ) + .unwrap(); + db.mark_agent_wake_delivery_unknown( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old.wake.wake_id, + &old.attempt.attempt_id, + &old.claim_fence, + ) + .unwrap(); + + let task = db + .create_agent_task( + &fixture.owner, + NewAgentTask { + title: "Unblocked endpoint work".to_string(), + instruction: "Execute a real pending agent task".to_string(), + assignee_agent_id: Some(fixture.receiver_agent_id.clone()), + source_conversation_id: None, + source_message_id: None, + referenced_project_id: None, + idempotency_key: "after-unknown-inbox-task".to_string(), + }, + ) + .unwrap() + .task + .summary + .task_id; + let started = db + .start_agent_task_attempt( + &fixture.owner, + &task, + &fixture.receiver_agent_id, + "after-unknown-inbox-attempt", + ) + .unwrap(); + let execution = db + .start_agent_task_endpoint_continuation_at( + &fixture.owner, + &task, + &started.attempt.attempt_id, + &fixture.receiver_agent_id, + &started.attempt_fence, + 1, + now_unix_ms(), + ) + .unwrap(); + db.conn_for_tests() + .execute( + "UPDATE wc_agent_wakes SET updated_at_unix_ms = ?2 WHERE wake_id = ?1", + params![ + old.wake.wake_id, + now_unix_ms() - STALE_INBOX_DELIVERY_UNKNOWN_MS - 10_000 + ], + ) + .unwrap(); + let recovered = db + .claim_next_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app", + ) + .unwrap() + .expect("new executable Task Wake must outrank abandoned Inbox delivery"); + assert_eq!(recovered.wake.wake_id, execution.execution.wake_id); + assert_eq!(recovered.wake.trigger_kind, "agent_task_attempt"); + assert_eq!( + db.agent_wake(&old.wake.wake_id).unwrap().unwrap().state, + AgentWakeState::Retired, + ); + assert_eq!( + queued_delivery_ids(&db, &fixture.receiver_agent_id).len(), + 1 + ); + db.prepare_agent_wake_dispatch( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &recovered.wake.wake_id, + &recovered.attempt.attempt_id, + &recovered.claim_fence, + &recovered.consume_token, + ) + .unwrap(); + db.consume_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &recovered.wake.wake_id, + &recovered.consume_token, + ) + .expect("new Task's Host continuation can consume its own exact Wake"); +} + #[test] fn withdrawing_wake_capability_reconciles_exact_endpoint_attempts() { let temp = tempfile::tempdir().unwrap(); From c6c0576461140b95e4ac7ec74573e616aa8a471a Mon Sep 17 00:00:00 2001 From: Beaten0123 Date: Sun, 15 Nov 2026 21:10:42 +0800 Subject: [PATCH 4/5] fix(host): classify uncertain MCP App Wake rejection safely --- .../mcp-app-auto-resume-restoration.md | 13 ++++++-- src/mcp_agent_continuation_app.html | 32 ++++++++++++++++--- src/mcp_tests/agent_continuation_app.test.mjs | 27 ++++++++++++++++ 3 files changed, 65 insertions(+), 7 deletions(-) diff --git a/docs/implementation/mcp-app-auto-resume-restoration.md b/docs/implementation/mcp-app-auto-resume-restoration.md index 405fcff34..7d4779589 100644 --- a/docs/implementation/mcp-app-auto-resume-restoration.md +++ b/docs/implementation/mcp-app-auto-resume-restoration.md @@ -61,9 +61,16 @@ its exact durable terminal wait and present that still-waiting wait. Handle a Job that already terminated in the current turn instead of presenting it. After the presentation tool result, output a normal assistant final response -and ensure the original turn has finished. Do not click, send another user -message or use a current-turn wait to manufacture the later turn. Leave the -conversation visible for foreground cases; switch away for background cases. +and ensure the original turn has finished **before creating the test Wake**. +Do not create a Task Attempt and call `start_agent_task_endpoint_continuation` +in the still-streaming setup turn: the App can prepare and issue `ui/message` +before the Host is ready to accept a new chat turn, producing an inconclusive +post-fence `delivery_unknown`. This is a suspected Host scheduling race, not +proof of the particular cause without a redacted Host result. Trigger test +work from a genuinely delayed, independent event, never a self-sent in-turn +message. Do not click or send another user message to manufacture the next +turn. Leave the conversation visible for foreground cases; switch away for +background cases. | Card | Delay | Visibility | Required observation | | --- | --- | --- | --- | diff --git a/src/mcp_agent_continuation_app.html b/src/mcp_agent_continuation_app.html index 79cc5fe19..ff333e15c 100644 --- a/src/mcp_agent_continuation_app.html +++ b/src/mcp_agent_continuation_app.html @@ -62,7 +62,17 @@ "projection-wake-state-invalid", "projection-recovery-invalid", "host-binding-missing", "host-binding-invalid", "host-bound-not-true", "unexpected", ]); - const SAFE_HOST_FAILURE_KINDS = new Set(["bridge-error", "bridge-timeout", "bridge-cancelled", "malformed-result"]); + const SAFE_HOST_FAILURE_KINDS = new Set(["bridge-error", "bridge-timeout", "bridge-cancelled", "malformed-result", "host-rejected"]); + // Closed diagnostic labels only: never retain Host response bodies or a Wake payload. + const HOST_DISPATCH_ISSUES = Object.freeze({ + host_rejected: "Host reported rejection", + timeout: "Host reply timed out", + cancelled: "Host cancelled the request", + invalid_response: "Host reply had an invalid shape", + unsupported: "Host does not recognize ui/message", + invalid_params: "Host rejected message parameters", + transport: "Host transport error", + }); const CALL_TIMEOUT_MS = 10000; const MAX_BIND_ATTEMPTS = 3; const MAX_RECOVERY_REBINDS = 1; @@ -95,6 +105,7 @@ let dispatchStarted = null; let prepareUncertain = null; let pendingFinish = null; + let lastHostDispatchIssue = null; let waitId = null; let waitLifecycle = null; let waitRevision = 0; @@ -588,7 +599,10 @@ if (projection.dispatch_observation === "continuation_consumed") { el("status").textContent = "Work resumed successfully"; } else if (projection.dispatch_observation === "dispatch_unknown") { - el("status").textContent = "Host delivery unconfirmed · no automatic retry"; + const issue = HOST_DISPATCH_ISSUES[lastHostDispatchIssue]; + el("status").textContent = issue + ? `Host delivery unconfirmed · ${issue} · no automatic retry` + : "Host delivery unconfirmed · no automatic retry"; } else if (projection.dispatch_observation === "dispatch_accepted") { el("status").textContent = "Host accepted follow-up · awaiting a fresh turn"; } else if (document.hidden) { @@ -749,6 +763,7 @@ dispatchStarted = wake.attempt_id; const hostDispatchStartedAt = performance.now(); let outcome = "delivery_unknown"; + lastHostDispatchIssue = null; timing("host_dispatch_start"); try { const hostMessageResult = await request("ui/message", { @@ -757,6 +772,8 @@ }); // A Host can resolve ui/message with isError:true instead of rejecting. // Acceptance is not a new model turn and never licenses a blind retry. + if (hostMessageResult && typeof hostMessageResult === "object" && !Array.isArray(hostMessageResult) + && hostMessageResult.isError === true) throw hostFailure("host-rejected"); if (!hostMessageResult || typeof hostMessageResult !== "object" || Array.isArray(hostMessageResult) || ![undefined, false].includes(hostMessageResult.isError)) { throw hostFailure("malformed-result"); @@ -764,11 +781,18 @@ outcome = "dispatch_accepted"; timing("host_dispatch_complete", { outcome, duration_ms: performance.now() - hostDispatchStartedAt }); } catch (error) { - // Closed labels only: never retain the automatic message, tokens or raw error. + // Preserve only a closed cause category in this live View; never capture + // raw Host errors, message bodies, Wake tokens or principal identifiers. const failureKind = tornDown ? "bridge-cancelled" : SAFE_HOST_FAILURE_KINDS.has(error && error.hostFailureKind) ? error.hostFailureKind : "bridge-error"; + lastHostDispatchIssue = failureKind === "host-rejected" ? "host_rejected" + : failureKind === "bridge-timeout" ? "timeout" + : failureKind === "bridge-cancelled" ? "cancelled" + : failureKind === "malformed-result" ? "invalid_response" + : error && error.rpcCode === -32601 ? "unsupported" + : error && error.rpcCode === -32602 ? "invalid_params" : "transport"; timing("host_dispatch_error", { outcome, failure_kind: failureKind, duration_ms: performance.now() - hostDispatchStartedAt }); - // After prepare, Host rejection/timeout/response loss is conservatively unknown. + // Post-fence failure is always delivery_unknown: no automatic resend. } // Observation only: acceptance is not proof of a new turn or Wake consumption. // Do not log the prepared message, private Attempt ids, or Host errors. diff --git a/src/mcp_tests/agent_continuation_app.test.mjs b/src/mcp_tests/agent_continuation_app.test.mjs index d40be2d2b..4bbabf62d 100644 --- a/src/mcp_tests/agent_continuation_app.test.mjs +++ b/src/mcp_tests/agent_continuation_app.test.mjs @@ -1028,6 +1028,33 @@ for (const hostReply of [ }); } +for (const { label, reply, error, expectedIssue } of [ + { label: "explicit isError rejection", reply: { isError: true }, expectedIssue: "Host reported rejection" }, + { label: "unsupported ui/message", error: { code: -32601, message: "PRIVATE_HOST_ERROR" }, expectedIssue: "Host does not recognize ui/message" }, + { label: "unknown RPC transport error", error: { code: -32000, message: "PRIVATE_HOST_ERROR" }, expectedIssue: "Host transport error" }, +]) { + test(`unknown dispatch shows only a closed diagnostic for ${label}, never resending`, async () => { + const view = await boundView(); + await view.reply(view.calls("get_agent_continuation_state").at(-1), toolResult({ agent_continuation: projection })); + await view.reply(view.calls("acquire_agent_continuation_wake").at(-1), toolResult({ wake })); + await view.reply(view.calls("prepare_agent_continuation_wake").at(-1), prepared(wake, "PRIVATE_WAKE_MESSAGE")); + const hostCall = hostMessages(view)[0]; + if (error) await view.reject(hostCall, error); + else await view.reply(hostCall, reply); + const finish = view.calls("finish_agent_continuation_wake").at(-1); + assert.equal(finish.params.arguments.outcome, "delivery_unknown"); + await view.reply(finish, toolResult({})); + await view.fireTimers(3000); + await view.reply(view.calls("get_agent_continuation_state").at(-1), + toolResult({ agent_continuation: { ...projection, dispatch_observation: "dispatch_unknown" } })); + assert.ok(view.nodes.status.textContent.includes(expectedIssue)); + assert.ok(view.nodes.status.textContent.includes("no automatic retry")); + assert.ok(!view.nodes.status.textContent.includes("PRIVATE_")); + assert.ok(!view.nodes.status.textContent.includes(wake.wake_id)); + assert.equal(hostMessages(view).length, 1); + }); +} + test("Host acceptance and uncertain delivery are distinct, without duplicate follow-up", async () => { const view = await boundView(); await view.reply(view.calls("get_agent_continuation_state").at(-1), toolResult({ agent_continuation: projection })); From 482aec044edec101cad0eecd7459f86187adacbb Mon Sep 17 00:00:00 2001 From: Beaten0123 Date: Sun, 15 Nov 2026 21:43:41 +0800 Subject: [PATCH 5/5] fix(agent): preserve queued Host Wake task lease until dispatch --- crates/webcodex-store/src/agent_task.rs | 6 +- .../src/agent_task/operations.rs | 28 ++++ crates/webcodex-store/src/agent_task_tests.rs | 139 ++++++++++++++++++ crates/webcodex-store/src/agent_wake.rs | 15 ++ crates/webcodex-store/src/agent_wake_tests.rs | 96 ++++++++++++ docs/architecture/durable-agent-runtime.md | 8 +- 6 files changed, 289 insertions(+), 3 deletions(-) diff --git a/crates/webcodex-store/src/agent_task.rs b/crates/webcodex-store/src/agent_task.rs index f060c7575..884e46988 100644 --- a/crates/webcodex-store/src/agent_task.rs +++ b/crates/webcodex-store/src/agent_task.rs @@ -37,8 +37,12 @@ pub(crate) const MAX_AGENT_TASK_PROJECT_REF_CHARS: usize = 256; pub const MAX_AGENT_TASK_LIST_LIMIT: usize = 100; pub(crate) const DEFAULT_AGENT_TASK_ATTEMPT_LEASE_MS: i64 = 60_000; pub(crate) const AGENT_TASK_ENDPOINT_DISPATCH_GRACE_MS: i64 = 5 * 60_000; +// A Task Wake queued behind an uncertain Host dispatch is not executing yet. +// Cover the 15-minute delivery safety window, 5-minute possible prior +// dispatch grace, and a final 60-second margin. This applies only while +// another post-dispatch Wake holds the Agent's single-flight slot. +pub(crate) const AGENT_TASK_ENDPOINT_QUEUED_WAKE_LEASE_MS: i64 = 21 * 60_000; pub(crate) const AGENT_TASK_ENDPOINT_TAKEOVER_LEASE_MS: i64 = 30 * 60_000; - const OP_CREATE_AGENT_TASK: &str = "create_agent_task"; const OP_START_AGENT_TASK_ATTEMPT: &str = "start_agent_task_attempt"; const OP_COMPLETE_AGENT_TASK_ATTEMPT: &str = "complete_agent_task_attempt"; diff --git a/crates/webcodex-store/src/agent_task/operations.rs b/crates/webcodex-store/src/agent_task/operations.rs index cdc5af5b4..b8663a50b 100644 --- a/crates/webcodex-store/src/agent_task/operations.rs +++ b/crates/webcodex-store/src/agent_task/operations.rs @@ -900,6 +900,34 @@ impl Database { )); } + // A new TaskAttempt's usual 60-second lease must not expire merely + // because an older uncertain Host Wake occupies the dispatch slot. + // Grant a one-shot bounded queue reservation in the same transaction + // as the new pending Wake. Exact replay cannot refresh this deadline. + let predecessor_blocks_dispatch: bool = transaction + .query_row( + "SELECT EXISTS(SELECT 1 FROM wc_agent_wakes + WHERE target_agent_id = ?1 + AND state IN ('prepared', 'delivered', 'delivery_unknown'))", + params![assignee_agent_id], + |row| row.get(0), + ) + .map_err(store_error)?; + if predecessor_blocks_dispatch { + transaction + .execute( + "UPDATE wc_agent_task_attempts + SET lease_expires_at_unix_ms = MAX(lease_expires_at_unix_ms, ?3) + WHERE attempt_id = ?1 AND task_id = ?2 AND state = 'active'", + params![ + attempt_id, + task_id, + now.saturating_add(AGENT_TASK_ENDPOINT_QUEUED_WAKE_LEASE_MS), + ], + ) + .map_err(store_error)?; + } + let wake_id = allocate_identity( &transaction, AGENT_WAKE_ID_PREFIX, diff --git a/crates/webcodex-store/src/agent_task_tests.rs b/crates/webcodex-store/src/agent_task_tests.rs index 70a024467..5bbb1d978 100644 --- a/crates/webcodex-store/src/agent_task_tests.rs +++ b/crates/webcodex-store/src/agent_task_tests.rs @@ -5445,3 +5445,142 @@ fn expired_uncertain_task_wake_fences_old_token_and_allows_new_host_takeover() { AgentWakeState::Retired, ); } + +/// A stale uncertain Task dispatch cannot consume the next Attempt's 60s +/// lease while it owns the single-flight Host slot. Exact old consume is +/// revoked after its source Attempt expires and the safety window elapses. +#[test] +fn queued_task_attempt_survives_older_uncertain_task_wake_then_consumes_once() { + let temp = tempfile::tempdir().unwrap(); + let db = Database::open(&temp.path().join("queued-behind-task-unknown.db")).unwrap(); + let owner = principal('f'); + let now = wall_now_ms(); + let old = prepared_endpoint_takeover_fixture(&db, &owner, "old-task-slot", now); + db.mark_agent_wake_delivery_unknown_at( + &owner, + &old.assignee, + &old.endpoint_id, + old.endpoint_controller_generation, + &old.wake_id, + &old.wake_attempt_id, + &old.wake_claim_fence, + now + 2, + ) + .unwrap(); + + let task_id = create_assigned_task(&db, &owner, &old.assignee, "next-task-after-unknown"); + let started = start( + &db, + &owner, + &task_id, + &old.assignee, + "next-task-queued", + now + 3, + ); + let next = db + .start_agent_task_endpoint_continuation_at( + &owner, + &task_id, + &started.attempt.attempt_id, + &old.assignee, + &started.attempt_fence, + started.attempt.attempt_controller_generation, + now + 4, + ) + .unwrap(); + let queued_lease = attempt_lease_expires_at(&db, &started.attempt.attempt_id); + assert!(queued_lease > now + 20 * 60_000); + assert!(db + .claim_next_agent_wake( + &owner, + &old.assignee, + &old.endpoint_id, + old.endpoint_controller_generation, + "mcp_app" + ) + .unwrap() + .is_none()); + assert_eq!( + db.agent_wake(&next.execution.wake_id) + .unwrap() + .unwrap() + .state, + AgentWakeState::Pending + ); + + // Existing execution is terminal before safety retirement, not + // silently promoted by this successor or a fabricated Host ACK. + db.conn_for_tests() + .execute( + "UPDATE wc_agent_task_attempts SET lease_expires_at_unix_ms = ?2 + WHERE attempt_id = ?1 AND state = 'active'", + params![old.task_attempt_id, now - 1], + ) + .unwrap(); + db.conn_for_tests() + .execute( + "UPDATE wc_agent_wakes SET updated_at_unix_ms = ?2 WHERE wake_id = ?1", + params![ + old.wake_id, + now - super::agent_wake::STALE_TASK_DELIVERY_UNKNOWN_MS - 1_000 + ], + ) + .unwrap(); + let claimed = db + .claim_next_agent_wake( + &owner, + &old.assignee, + &old.endpoint_id, + old.endpoint_controller_generation, + "mcp_app", + ) + .unwrap() + .expect("new valid Task Wake must take over after safe retirement"); + assert_eq!(claimed.wake.wake_id, next.execution.wake_id); + assert_eq!( + db.agent_wake(&old.wake_id).unwrap().unwrap().state, + AgentWakeState::Retired + ); + assert!( + db.consume_agent_wake( + &owner, + &old.assignee, + &old.endpoint_id, + old.endpoint_controller_generation, + &old.wake_id, + &old.consume_token + ) + .is_err(), + "late old uncertain Host turn cannot consume the revoked fence" + ); + let dispatch_lease = attempt_lease_expires_at(&db, &started.attempt.attempt_id); + assert!(dispatch_lease < queued_lease); + assert!(dispatch_lease > wall_now_ms() + 4 * 60_000); + db.prepare_agent_wake_dispatch( + &owner, + &old.assignee, + &old.endpoint_id, + old.endpoint_controller_generation, + &claimed.wake.wake_id, + &claimed.attempt.attempt_id, + &claimed.claim_fence, + &claimed.consume_token, + ) + .unwrap(); + db.consume_agent_wake( + &owner, + &old.assignee, + &old.endpoint_id, + old.endpoint_controller_generation, + &claimed.wake.wake_id, + &claimed.consume_token, + ) + .unwrap(); + assert_eq!( + db.agent_wake(&claimed.wake.wake_id).unwrap().unwrap().state, + AgentWakeState::Consumed + ); + assert!( + attempt_lease_expires_at(&db, &started.attempt.attempt_id) > wall_now_ms() + 29 * 60_000 + ); +} diff --git a/crates/webcodex-store/src/agent_wake.rs b/crates/webcodex-store/src/agent_wake.rs index 8d637f27c..4b33539bd 100644 --- a/crates/webcodex-store/src/agent_wake.rs +++ b/crates/webcodex-store/src/agent_wake.rs @@ -2364,6 +2364,21 @@ fn bind_agent_task_wake_carrier( attempt_controller_generation, )?; replace_agent_task_attempt_controller_in_transaction(transaction, principal, authority, now)?; + // This pending Wake has claimed the single-flight Host carrier. Drop + // its long queue-only reservation to the ordinary pre-consume dispatch + // window; only the exact new model-turn consume grants active-turn time. + transaction + .execute( + "UPDATE wc_agent_task_attempts + SET lease_expires_at_unix_ms = MIN(lease_expires_at_unix_ms, ?3) + WHERE attempt_id = ?1 AND task_id = ?2 AND state = 'active'", + params![ + task_attempt_id, + task_id, + now.saturating_add(AGENT_TASK_ENDPOINT_DISPATCH_GRACE_MS), + ], + ) + .map_err(store_error)?; let changed = transaction .execute( "UPDATE wc_agent_task_endpoint_executions diff --git a/crates/webcodex-store/src/agent_wake_tests.rs b/crates/webcodex-store/src/agent_wake_tests.rs index 24b523760..ec9dea813 100644 --- a/crates/webcodex-store/src/agent_wake_tests.rs +++ b/crates/webcodex-store/src/agent_wake_tests.rs @@ -580,6 +580,63 @@ fn old_unknown_inbox_dispatch_cannot_starve_new_endpoint_agent_task_after_grace( now_unix_ms(), ) .unwrap(); + // The old Host dispatch is outcome-unknown: no successor model turn may + // start, but the successor must survive its 60-second default lease. + let queued_lease: i64 = db + .conn_for_tests() + .query_row( + "SELECT lease_expires_at_unix_ms FROM wc_agent_task_attempts WHERE attempt_id = ?1", + [&started.attempt.attempt_id], + |row| row.get(0), + ) + .unwrap(); + assert!( + queued_lease >= now_unix_ms() + STALE_INBOX_DELIVERY_UNKNOWN_MS + 5 * 60_000, + "queued Task Wake must outlive the predecessor's Host uncertainty window" + ); + assert!( + db.claim_next_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + "mcp_app" + ) + .unwrap() + .is_none(), + "the predecessor still owns exclusive Host dispatch" + ); + assert_eq!( + db.agent_wake(&execution.execution.wake_id) + .unwrap() + .unwrap() + .state, + AgentWakeState::Pending + ); + let replay = db + .start_agent_task_endpoint_continuation_at( + &fixture.owner, + &task, + &started.attempt.attempt_id, + &fixture.receiver_agent_id, + &started.attempt_fence, + 1, + now_unix_ms() + 90_000, + ) + .unwrap(); + assert!(replay.replayed); + assert_eq!( + queued_lease, + db.conn_for_tests() + .query_row( + "SELECT lease_expires_at_unix_ms FROM wc_agent_task_attempts WHERE attempt_id = ?1", + [&started.attempt.attempt_id], + |row| row.get::<_, i64>(0) + ) + .unwrap(), + "exact queue replay must never slide its one-shot deadline" + ); + db.conn_for_tests() .execute( "UPDATE wc_agent_wakes SET updated_at_unix_ms = ?2 WHERE wake_id = ?1", @@ -601,6 +658,22 @@ fn old_unknown_inbox_dispatch_cannot_starve_new_endpoint_agent_task_after_grace( .expect("new executable Task Wake must outrank abandoned Inbox delivery"); assert_eq!(recovered.wake.wake_id, execution.execution.wake_id); assert_eq!(recovered.wake.trigger_kind, "agent_task_attempt"); + let dispatch_lease: i64 = db + .conn_for_tests() + .query_row( + "SELECT lease_expires_at_unix_ms FROM wc_agent_task_attempts WHERE attempt_id = ?1", + [&started.attempt.attempt_id], + |row| row.get(0), + ) + .unwrap(); + assert!( + dispatch_lease < queued_lease, + "claim must drop the long queue-only lease" + ); + assert!( + dispatch_lease >= now_unix_ms() + 4 * 60_000, + "claim retains the ordinary bounded Host dispatch window" + ); assert_eq!( db.agent_wake(&old.wake.wake_id).unwrap().unwrap().state, AgentWakeState::Retired, @@ -629,6 +702,29 @@ fn old_unknown_inbox_dispatch_cannot_starve_new_endpoint_agent_task_after_grace( &recovered.consume_token, ) .expect("new Task's Host continuation can consume its own exact Wake"); + assert!( + db.conn_for_tests() + .query_row( + "SELECT lease_expires_at_unix_ms FROM wc_agent_task_attempts WHERE attempt_id = ?1", + [&started.attempt.attempt_id], + |row| row.get::<_, i64>(0), + ) + .unwrap() + > now_unix_ms() + 29 * 60_000, + "only exact consume promotes the Attempt to the active model-turn lease" + ); + assert!( + db.consume_agent_wake( + &fixture.owner, + &fixture.receiver_agent_id, + &endpoint.endpoint_id, + endpoint.controller_generation, + &old.wake.wake_id, + &old.consume_token + ) + .is_err(), + "retired old Host token must never execute again" + ); } #[test] diff --git a/docs/architecture/durable-agent-runtime.md b/docs/architecture/durable-agent-runtime.md index 78cf80e44..15423832d 100644 --- a/docs/architecture/durable-agent-runtime.md +++ b/docs/architecture/durable-agent-runtime.md @@ -850,8 +850,10 @@ A4b intentionally separates a short pre-takeover lease from renewable bounded ac ```text TaskAttempt start -> short 60-second setup/pre-dispatch lease -Endpoint carrier claim / prepare - -> still short pre-dispatch semantics +Endpoint continuation created while older Wake is still post-dispatch + -> one bounded 21-minute queue-only reservation (no Host execution authority) +older Wake safely retires or is exactly consumed; new Wake claims Host carrier + -> long queue lease reduced to at most 5-minute pre-consume dispatch window first durable Host dispatch outcome (dispatch_accepted | delivery_unknown) -> bounded 5-minute Host scheduling grace exact model turn bootstrap + first exact consume during that grace @@ -868,6 +870,8 @@ abnormal/stalled model turn -> a new Attempt requires an explicit authorized start ``` +The 21-minute queue-only lease is created **once**, during the exact Endpoint-continuation start, only when an older `prepared`, `delivered`, or `delivery_unknown` Wake already owns the Agent's single Host dispatch slot. It covers the 15-minute conservative retirement window, a possible 5-minute predecessor scheduling grace, and 60 seconds for the successor; without a blocker the default 60-second lease is unchanged. Exact start replay cannot slide the reservation. Acquiring the successor Wake in a serializable Store transaction shortens any longer queue lease to at most `now + 5 minutes` before dispatch; this is not an actual model turn, does not extend the older Wake, and cannot promote to the 30-minute active-turn lease. Safety retirement and old-token revocation are unchanged. A blocked queue with no available Host or a predecessor that has not reached safe retirement is not claimed or redispatched; its bounded reservation eventually expires and requires a fresh authorized Attempt. This bounded mitigation is not a guarantee of real-time Host delivery. + Crossing the durable Host dispatch fence now creates a separate scheduling reservation only for an exact Task-origin Wake. The first `prepared -> dispatch_accepted | delivery_unknown` transition atomically rechecks the owned latest active/unexpired Attempt, exact assignee, exact Wake, and exact bound Endpoint/generation, then uses `max(existing_lease, now + 5 minutes)`. Delivery uncertainty gets the same reservation because the Host may already have accepted the message and exactly-once delivery forbids a blind resend. Same-generation process-local Host-binding loss or Server takeover that conservatively records an already-prepared Wake as `delivery_unknown` establishes that same one-shot grace before carrier capability is cleared; durable Endpoint replacement/expiry/detach fences the old generation instead and grants it no new grace. Dispatch replay, `delivery_unknown -> delivered` reconciliation, stale Endpoint generations, wrong Wakes, non-Task Wakes, polling, Endpoint heartbeat, and ClientWindow activity do not slide this reservation. A dispatch outcome arriving after the Attempt already expired or was superseded is still durable Wake evidence but never revives that Attempt. The first successful exact consume of an `agent_task_attempt` Wake promotes only the same latest, active, unexpired, exact-assignee Attempt whose durable A4b Endpoint execution still matches that Wake and Endpoint generation. Promotion is atomic with Wake consumption and uses `max(existing_lease, now + 30 minutes)`. Consume replay does not slide the lease. An expired, terminal, superseded, or carrier-mismatched Attempt is never revived; its already-dispatched Wake can still be consumed/ACKed without lease promotion. `inbox_changed`, `attention_event`, and `agent_wait_events` consumption never changes a TaskAttempt lease.