diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 8689e90a4..44ac187ea 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -13290,6 +13290,86 @@ permission = "write" assert_eq!(state.pending_notifications[1].provenance, account_b); } + #[test] + fn notification_lost_acceptance_retry_after_restore_commits_once_not_by_body() { + let temp = tempfile::tempdir().unwrap(); + let handle = PendingSubmissionHandle::for_test(temp.path()); + let source = "backend:orchestrator-attention"; + let accept = |handle: &PendingSubmissionHandle, id: &str| { + handle + .accept_notification_from_source( + id.into(), + "Queued Tickets require attention".into(), + source.into(), + WorkerHistoryProvenance::BackendInstruction { operation_id: None }, + ) + .unwrap() + }; + // The sender lost this acceptance response; the receipt is already durable. + assert!(accept(&handle, "attention-1")); + let restored_state: PendingActivationState = handle + .persisted_entries_for_test() + .iter() + .rev() + .find_map(|entry| match entry { + LogEntry::Extension { + domain, payload, .. + } if domain == SESSION_PENDING_ACTIVATIONS_EXTENSION_DOMAIN => { + Some(serde_json::from_value(payload.clone()).unwrap()) + } + _ => None, + }) + .unwrap(); + assert_eq!(restored_state.notification_receipts.len(), 1); + let handle = PendingSubmissionHandle { + state: Arc::new(Mutex::new(restored_state)), + writer: handle.writer.clone(), + }; + assert!(!accept(&handle, "attention-1")); + let commit_batch = |handle: &PendingSubmissionHandle| { + let batch = handle.prepare_notification_batch(); + let count = batch.len(); + for (notification, extension) in batch { + handle + .writer + .commit_system_item_with_extensions( + SystemItem::Notification { + body: notification.message.clone(), + message: notification.message, + prompt_provenance: None, + }, + vec![extension], + Some(notification.provenance), + ) + .unwrap(); + } + handle.finish_notification_batch(); + count + }; + assert_eq!(commit_batch(&handle), 1); + assert!(!accept(&handle, "attention-1")); + assert_eq!(commit_batch(&handle), 0); + let notification_count = |handle: &PendingSubmissionHandle| { + handle + .persisted_entries_for_test() + .iter() + .filter(|entry| { + matches!(entry, + LogEntry::AnnotatedSystemItem { entry, .. } + if matches!(entry.item, SystemItem::Notification { .. }) + ) + }) + .count() + }; + assert_eq!(notification_count(&handle), 1); + assert_eq!(handle.state.lock().unwrap().notification_receipts.len(), 1); + // Same body with a new logical ID is legitimate, not content-deduped. + assert!(accept(&handle, "attention-2")); + assert_eq!(commit_batch(&handle), 1); + assert_eq!(notification_count(&handle), 2); + assert_eq!(handle.state.lock().unwrap().notification_receipts.len(), 2); + } + #[test] fn notification_batch_preserves_fifo_order_and_notification_dedupes() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 39e0163ba..3fe23dee2 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1379,6 +1379,42 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { } } +// Per-recipient reservation. No lock is held across Runtime calls: subscription +// callbacks may reenter the attention path while send_input runs. +#[derive(Default)] +struct OrchestratorAttentionState { + sending: bool, + accepted: Option, + pending: Option, +} + +#[derive(Clone)] +struct OrchestratorAttentionDelivery { + fingerprint: String, + request_id: String, + content: String, +} + +struct OrchestratorAttentionReservation(Arc>); + +impl Drop for OrchestratorAttentionReservation { + fn drop(&mut self) { + self.0.lock().unwrap_or_else(|p| p.into_inner()).sending = false; + } +} + +fn orchestrator_attention_state( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, +) -> Arc> { + api.orchestrator_attention + .lock() + .unwrap_or_else(|p| p.into_inner()) + .entry(worker.clone()) + .or_default() + .clone() +} + #[derive(Clone)] pub struct WorkspaceApi { pub(crate) config: ServerConfig, @@ -1396,7 +1432,8 @@ pub struct WorkspaceApi { runtime_binding_expectations: Arc>>, companion: Arc, orchestrator_spawn_lock: Arc>, - orchestrator_attention_fingerprint: Arc>>, + orchestrator_attention: + Arc>>>>, backend_job_drain_requests: Arc, observation_proxy: BackendObservationProxy, runtime_subscription_broker: RuntimeSubscriptionBroker, @@ -3247,7 +3284,7 @@ impl WorkspaceApi { runtime_binding_expectations: Arc::new(RwLock::new(HashMap::new())), companion, orchestrator_spawn_lock: Arc::new(std::sync::Mutex::new(())), - orchestrator_attention_fingerprint: Arc::new(Mutex::new(None)), + orchestrator_attention: Arc::new(Mutex::new(HashMap::new())), backend_job_drain_requests: Arc::new(AtomicU64::new(0)), observation_proxy, runtime_subscription_broker, @@ -16765,48 +16802,54 @@ async fn run_orchestrator_turn_end_hook(api: WorkspaceApi) { }; let mut worker_states = HashMap::new(); while let Some(update) = subscription.recv().await { - match update { - crate::runtime_subscription::BrokerSubscriptionEvent::Snapshot { snapshot, .. } => { - if let protocol::subscription::SubscriptionSnapshot::Workers { workers } = snapshot - { - worker_states.clear(); - for worker in workers { - let worker_id = worker.worker_id.to_string(); - maybe_dispatch_orchestrator_turn_end(&api, &worker_id, None, worker.state); - worker_states.insert(worker_id, worker.state); - } - } - } - crate::runtime_subscription::BrokerSubscriptionEvent::Event { payload, .. } => { - match payload { - protocol::subscription::SubscriptionEventPayload::WorkerUpserted { worker } => { - let worker_id = worker.worker_id.to_string(); - let previous = worker_states.insert(worker_id.clone(), worker.state); - maybe_dispatch_orchestrator_turn_end( - &api, - &worker_id, - previous, - worker.state, - ); - } - protocol::subscription::SubscriptionEventPayload::WorkerRemoved { - worker_id, - .. - } => { - worker_states.remove(worker_id.as_str()); - } - _ => {} - } - } - crate::runtime_subscription::BrokerSubscriptionEvent::Disconnected { .. } => { - worker_states.clear(); - } - crate::runtime_subscription::BrokerSubscriptionEvent::Rejected { .. } - | crate::runtime_subscription::BrokerSubscriptionEvent::Closed { .. } => return, + if !observe_orchestrator_turn_end(&api, &mut worker_states, update) { + return; } } } +// A subscription snapshot establishes an observation baseline, not a lifecycle +// event. Reconnection deliberately discards the old baseline. +fn observe_orchestrator_turn_end( + api: &WorkspaceApi, + worker_states: &mut HashMap, + update: crate::runtime_subscription::BrokerSubscriptionEvent, +) -> bool { + match update { + crate::runtime_subscription::BrokerSubscriptionEvent::Snapshot { snapshot, .. } => { + if let protocol::subscription::SubscriptionSnapshot::Workers { workers } = snapshot { + worker_states.clear(); + for worker in workers { + let worker_id = worker.worker_id.to_string(); + worker_states.insert(worker_id, worker.state); + } + } + } + crate::runtime_subscription::BrokerSubscriptionEvent::Event { payload, .. } => { + match payload { + protocol::subscription::SubscriptionEventPayload::WorkerUpserted { worker } => { + let worker_id = worker.worker_id.to_string(); + let previous = worker_states.insert(worker_id.clone(), worker.state); + maybe_dispatch_orchestrator_turn_end(api, &worker_id, previous, worker.state); + } + protocol::subscription::SubscriptionEventPayload::WorkerRemoved { + worker_id, + .. + } => { + worker_states.remove(worker_id.as_str()); + } + _ => {} + } + } + crate::runtime_subscription::BrokerSubscriptionEvent::Disconnected { .. } => { + worker_states.clear(); + } + crate::runtime_subscription::BrokerSubscriptionEvent::Rejected { .. } + | crate::runtime_subscription::BrokerSubscriptionEvent::Closed { .. } => return false, + } + true +} + fn maybe_dispatch_orchestrator_turn_end( api: &WorkspaceApi, worker_id: &str, @@ -16816,12 +16859,7 @@ fn maybe_dispatch_orchestrator_turn_end( use protocol::subscription::SubscriptionWorkerState; if current != SubscriptionWorkerState::Idle - || !matches!( - previous, - None | Some(SubscriptionWorkerState::Running) - | Some(SubscriptionWorkerState::Stopped) - | Some(SubscriptionWorkerState::Paused) - ) + || previous != Some(SubscriptionWorkerState::Running) { return; } @@ -16831,14 +16869,31 @@ fn maybe_dispatch_orchestrator_turn_end( if orchestrator.worker.runtime_id == EMBEDDED_WORKER_RUNTIME_ID && orchestrator.worker.worker_id == worker_id { - dispatch_orchestrator_queue_attention(api); + dispatch_orchestrator_queue_attention(api, &orchestrator.worker, false); } } -fn dispatch_orchestrator_queue_attention(api: &WorkspaceApi) { - let Some(orchestrator) = find_workspace_orchestrator(api) else { - return; +fn dispatch_orchestrator_queue_attention( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + retry_only: bool, +) { + let state = orchestrator_attention_state(api, worker); + let pending = { + let mut state = state.lock().unwrap_or_else(|p| p.into_inner()); + if state.sending || (retry_only && state.pending.is_none()) { + return; + } + state.sending = true; + state.pending.clone() }; + let _reservation = OrchestratorAttentionReservation(state.clone()); + // Unknown outcomes keep the original immutable payload and ID, even if the + // backlog or prompt projection changed before this retry. + if let Some(delivery) = pending { + send_orchestrator_attention(api, worker, &state, delivery); + return; + } let Ok(backend) = browser_ticket_backend(api) else { return; }; @@ -16867,9 +16922,7 @@ fn dispatch_orchestrator_queue_attention(api: &WorkspaceApi) { queued.extend(inprogress); queued.sort_by(|left, right| left.id.cmp(&right.id)); if queued.is_empty() { - *api.orchestrator_attention_fingerprint - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; + state.lock().unwrap_or_else(|p| p.into_inner()).accepted = None; return; } let fingerprint = queued @@ -16877,11 +16930,12 @@ fn dispatch_orchestrator_queue_attention(api: &WorkspaceApi) { .map(|ticket| ticket.id.as_str()) .collect::>() .join("|"); - if api - .orchestrator_attention_fingerprint + if state .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - .as_deref() + .unwrap_or_else(|p| p.into_inner()) + .accepted + .as_ref() + .map(|delivery| delivery.fingerprint.as_str()) == Some(fingerprint.as_str()) { return; @@ -16932,22 +16986,42 @@ fn dispatch_orchestrator_queue_attention(api: &WorkspaceApi) { return; } }; - let accepted = api - .runtime - .send_input( - &orchestrator.worker, - WorkerInputRequest { - kind: WorkerInputKind::Notify, - content, - submission_request_id: None, - segments: None, - }, - ) - .is_ok_and(|result| result.state == InternalWorkerOperationState::Accepted); - if accepted { - *api.orchestrator_attention_fingerprint - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(fingerprint); + let delivery = OrchestratorAttentionDelivery { + fingerprint, + request_id: uuid::Uuid::now_v7().to_string(), + content, + }; + state.lock().unwrap_or_else(|p| p.into_inner()).pending = Some(delivery.clone()); + send_orchestrator_attention(api, worker, &state, delivery); +} + +fn send_orchestrator_attention( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + state: &Mutex, + delivery: OrchestratorAttentionDelivery, +) { + // Bounded delivery-side retry, not an Idle-snapshot retry. Rejection and + // transport ambiguity leave pending available for the next attempt. + for _ in 0..2 { + let accepted = api + .runtime + .send_input( + worker, + WorkerInputRequest { + kind: WorkerInputKind::Notify, + content: delivery.content.clone(), + submission_request_id: Some(delivery.request_id.clone()), + segments: None, + }, + ) + .is_ok_and(|result| result.state == InternalWorkerOperationState::Accepted); + if accepted { + let mut state = state.lock().unwrap_or_else(|p| p.into_inner()); + state.accepted = Some(delivery); + state.pending = None; + return; + } } } @@ -22230,14 +22304,33 @@ async fn scoped_start_workspace_orchestrator( let mut disposition = "created"; if let Some(existing) = find_workspace_orchestrator(&api) { if workspace_orchestrator_is_online(&existing) { + drop(_guard); + // Already-online is not a new startup. Only retry outstanding delivery. + dispatch_orchestrator_queue_attention(&api, &existing.worker, true); return Ok(Json(workspace_orchestrator_response(&api, "existing"))); } + let attention = orchestrator_attention_state(&api, &existing.worker); + let before_restore = attention + .lock() + .unwrap_or_else(|p| p.into_inner()) + .accepted + .as_ref() + .map(|delivery| delivery.request_id.clone()); let restored = api.restore_workspace_worker(&existing.worker)?; if restored.state == server_api::WorkerRestoreState::Accepted { - *api.orchestrator_attention_fingerprint - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; - dispatch_orchestrator_queue_attention(&api); + { + let mut attention = attention.lock().unwrap_or_else(|p| p.into_inner()); + // Retire only pre-Restore success. Concurrent reservations, unknown + // outcomes and newly accepted deliveries must survive Restore. + if !attention.sending + && attention.pending.is_none() + && attention.accepted.as_ref().map(|d| &d.request_id) == before_restore.as_ref() + { + attention.accepted = None; + } + } + drop(_guard); + dispatch_orchestrator_queue_attention(&api, &existing.worker, false); return Ok(Json(workspace_orchestrator_response(&api, "restored"))); } if !diagnostics_indicate_unrecoverable_pending_workspace_restore(&restored.diagnostics) { @@ -22316,10 +22409,8 @@ async fn scoped_start_workspace_orchestrator( Some("builtin:orchestrator".to_string()), WorkerRegistryDisplayNamePolicy::UseProvided, )?; - *api.orchestrator_attention_fingerprint - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; - dispatch_orchestrator_queue_attention(&api); + drop(_guard); + dispatch_orchestrator_queue_attention(&api, &worker.worker, false); Ok(Json(workspace_orchestrator_response(&api, disposition))) } @@ -48063,6 +48154,688 @@ mod tests { assert_eq!(attempts[0].0.worker_id.to_string(), orchestrator.worker_id); } + async fn attention_test_api() -> ( + tempfile::TempDir, + WorkspaceApi, + Arc, + ) { + let dir = tempfile::tempdir().unwrap(); + init_clean_git_workspace(dir.path()); + let (api, execution) = test_api_with_recording_backend(dir.path()).await; + let mut input = ticket::NewTicket::new("Attention backlog"); + input.workflow_state = Some(TicketWorkflowState::Queued); + set_test_ticket_target(&mut input, "test-repository", "HEAD"); + let ticket = browser_ticket_backend(&api).unwrap().create(input).unwrap(); + assign_test_orchestrator(&api, &ticket.id); + (dir, api, execution) + } + + async fn start_attention_test_orchestrator( + api: &WorkspaceApi, + ) -> BrowserWorkspaceOrchestratorResponse { + scoped_start_workspace_orchestrator( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + }), + ) + .await + .unwrap() + .0 + } + + #[tokio::test] + async fn orchestrator_attention_parallel_start_and_idle_observation_send_once() { + let (_dir, api, execution) = attention_test_api().await; + let reentrant_api = api.clone(); + execution.before_input_returns(move |worker| { + use protocol::subscription::SubscriptionWorkerState as S; + // The callback is inside Runtime send_input. Neither initial Idle nor + // a real turn end may deadlock or issue another in-flight delivery. + maybe_dispatch_orchestrator_turn_end( + &reentrant_api, + &worker.worker_id.to_string(), + None, + S::Idle, + ); + maybe_dispatch_orchestrator_turn_end( + &reentrant_api, + &worker.worker_id.to_string(), + Some(S::Running), + S::Idle, + ); + }); + let barrier = Arc::new(std::sync::Barrier::new(8)); + let results = std::thread::scope(|scope| { + let joins = (0..8) + .map(|_| { + let api = api.clone(); + let barrier = barrier.clone(); + scope.spawn(move || { + barrier.wait(); + futures::executor::block_on(start_attention_test_orchestrator(&api)) + }) + }) + .collect::>(); + joins + .into_iter() + .map(|j| j.join().unwrap()) + .collect::>() + }); + assert_eq!( + results + .iter() + .filter(|r| r.disposition == "created") + .count(), + 1 + ); + assert_eq!( + results + .iter() + .filter(|r| r.disposition == "existing") + .count(), + 7 + ); + assert_eq!(execution.take_inputs().len(), 1); + assert_eq!(execution.take_input_request_ids().len(), 1); + assert_eq!( + start_attention_test_orchestrator(&api).await.disposition, + "existing" + ); + assert!(execution.take_inputs().is_empty()); + } + + #[tokio::test] + async fn orchestrator_attention_barrier_callers_share_unsent_reservation() { + let (_dir, api, execution) = attention_test_api().await; + let started = start_attention_test_orchestrator(&api).await; + let worker = RuntimeWorkerRef::new( + EMBEDDED_WORKER_RUNTIME_ID, + started.worker.unwrap().worker_id, + ); + let state = orchestrator_attention_state(&api, &worker); + state.lock().unwrap().accepted = None; + execution.take_inputs(); + execution.take_input_request_ids(); + let barrier = Arc::new(std::sync::Barrier::new(8)); + let (entered_tx, entered_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + execution.before_input_returns(move |_| { + entered_tx.send(()).unwrap(); + release_rx + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + }); + std::thread::scope(|scope| { + let joins = (0..8) + .map(|_| { + let api = api.clone(); + let worker = worker.clone(); + let barrier = barrier.clone(); + scope.spawn(move || { + barrier.wait(); + dispatch_orchestrator_queue_attention(&api, &worker, false); + }) + }) + .collect::>(); + entered_rx + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + assert!(state.lock().unwrap().sending); + assert!(state.lock().unwrap().accepted.is_none()); + // The success record does not exist yet. A ninth caller must still + // return immediately rather than compare/send twice or wait on Runtime. + dispatch_orchestrator_queue_attention(&api, &worker, false); + release_tx.send(()).unwrap(); + for join in joins { + join.join().unwrap(); + } + }); + assert_eq!(execution.take_inputs().len(), 1); + assert_eq!(execution.take_input_request_ids().len(), 1); + assert!(!state.lock().unwrap().sending); + assert!(state.lock().unwrap().accepted.is_some()); + } + + #[tokio::test] + async fn orchestrator_attention_unknown_response_retry_keeps_id_and_payload() { + let (_dir, api, execution) = attention_test_api().await; + execution.reject_inputs("accepted response lost (unknown transport outcome)"); + let started = start_attention_test_orchestrator(&api).await; + let worker = RuntimeWorkerRef::new( + EMBEDDED_WORKER_RUNTIME_ID, + started.worker.unwrap().worker_id, + ); + let state = orchestrator_attention_state(&api, &worker); + assert!(state.lock().unwrap().accepted.is_none()); + let pending = state.lock().unwrap().pending.clone().unwrap(); + let attempts = execution.take_inputs(); + let ids = execution.take_input_request_ids(); + assert_eq!(attempts.len(), 2); + assert_eq!(ids, vec![Some(pending.request_id.clone()); 2]); + // Change the backlog before retry: uncertain acceptance must never reuse + // the ID for a newly rendered payload. + let mut next = ticket::NewTicket::new("Changed backlog"); + next.workflow_state = Some(TicketWorkflowState::Queued); + set_test_ticket_target(&mut next, "test-repository", "HEAD"); + let next = browser_ticket_backend(&api).unwrap().create(next).unwrap(); + assign_test_orchestrator(&api, &next.id); + *execution.input_failure.lock().unwrap() = None; + assert_eq!( + start_attention_test_orchestrator(&api).await.disposition, + "existing" + ); + let retry = execution.take_inputs(); + assert_eq!(retry.len(), 1); + assert_eq!(retry[0].1, pending.content); + assert_eq!( + execution.take_input_request_ids(), + vec![Some(pending.request_id)] + ); + assert!(state.lock().unwrap().pending.is_none()); + assert!(state.lock().unwrap().accepted.is_some()); + use protocol::subscription::SubscriptionWorkerState as S; + maybe_dispatch_orchestrator_turn_end(&api, &worker.worker_id, Some(S::Running), S::Idle); + let changed = execution.take_inputs(); + assert_eq!(changed.len(), 1); + assert!(changed[0].1.contains("Changed backlog")); + } + + #[tokio::test] + async fn orchestrator_attention_subscription_snapshots_only_establish_baselines() { + use crate::runtime_subscription::BrokerSubscriptionEvent as B; + use protocol::subscription::{ + SubscriptionEventPayload, SubscriptionSnapshot, SubscriptionWorker, + SubscriptionWorkerState as S, + }; + let (_dir, api, execution) = attention_test_api().await; + let started = start_attention_test_orchestrator(&api).await; + let worker = RuntimeWorkerRef::new( + EMBEDDED_WORKER_RUNTIME_ID, + started.worker.unwrap().worker_id, + ); + let attention = orchestrator_attention_state(&api, &worker); + attention.lock().unwrap().accepted = None; + execution.take_inputs(); + let summary = |state: S| -> SubscriptionWorker { + serde_json::from_value(serde_json::json!({ + "worker_id": worker.worker_id, "state": state, + })) + .unwrap() + }; + let snapshot = |state| B::Snapshot { + snapshot: SubscriptionSnapshot::Workers { + workers: vec![summary(state)], + }, + }; + let event = |state| B::Event { + payload: SubscriptionEventPayload::WorkerUpserted { + worker: summary(state), + }, + }; + let mut states = HashMap::new(); + assert!(observe_orchestrator_turn_end( + &api, + &mut states, + snapshot(S::Idle) + )); + assert!(observe_orchestrator_turn_end( + &api, + &mut states, + event(S::Idle) + )); + observe_orchestrator_turn_end(&api, &mut states, event(S::Running)); + // Neither snapshot itself nor an Idle event after reconnect is turn-end evidence. + observe_orchestrator_turn_end( + &api, + &mut states, + B::Disconnected { + message: "fixture reconnect".into(), + }, + ); + observe_orchestrator_turn_end(&api, &mut states, snapshot(S::Idle)); + observe_orchestrator_turn_end(&api, &mut states, event(S::Idle)); + observe_orchestrator_turn_end(&api, &mut states, snapshot(S::Running)); + observe_orchestrator_turn_end(&api, &mut states, snapshot(S::Idle)); + assert!(attention.lock().unwrap().accepted.is_none()); + assert!(execution.take_inputs().is_empty()); + observe_orchestrator_turn_end(&api, &mut states, event(S::Running)); + observe_orchestrator_turn_end(&api, &mut states, event(S::Idle)); + assert_eq!(execution.take_inputs().len(), 1); + } + + // Real Controller and canonical Session storage with isolated local config. + // A closed loopback model endpoint prevents any production-provider access; + // notifications commit at the model boundary before that request can fail. + struct AttentionWorkerFactory { + root: PathBuf, + } + + impl AttentionWorkerFactory { + async fn controller( + &self, + worker_id: String, + restore: bool, + ) -> std::result::Result + { + let name = format!("attention-{worker_id}"); + let root = self.root.join("workers").join(&worker_id); + let store = session_store::CombinedStore::new( + session_store::WorkerSessionStore::new(root.join("session")) + .map_err(|e| e.to_string())?, + session_store::WorkerAggregateStore::new(&root, &name) + .map_err(|e| e.to_string())?, + ); + let manifest = manifest::WorkerManifest::from_toml(&format!( + r#" + [worker] + name = "{name}" + pwd = "./" + [model] + scheme = "anthropic" + model_id = "attention-test" + base_url = "http://127.0.0.1:1" + auth = {{ kind = "none" }} + [engine] + max_tokens = 10 + [scope] + allow = [] + "# + )) + .map_err(|e| e.to_string())?; + let context = worker::WorkerWorkspaceContext::local_filesystem(None); + let workspace_client = context.client_handle(); + let loader = worker::PromptCatalogSource::builtins_only(); + let worker = if restore { + worker::Worker::restore_from_worker_metadata_with_context( + &name, + manifest, + store, + loader, + context, + worker::WorkerFilesystemAuthority::None, + ) + .await + } else { + worker::Worker::from_manifest_with_context( + manifest, + store, + loader, + context, + worker::WorkerFilesystemAuthority::None, + ) + .await + } + .map_err(|e| e.to_string())?; + let started = worker::PreparedWorker::new( + worker.with_notification_coalesce_delay(std::time::Duration::from_millis(10)), + worker::WorkerBootstrapLayout::RuntimeManagedRun { + run_dir: root.join(uuid::Uuid::now_v7().to_string()), + bash_output_dir: root.join("bash-output"), + }, + worker::WorkerControllerTransport::InProcess, + ) + .start() + .await + .map_err(|e| e.to_string())?; + Ok(worker_runtime::worker_backend::RuntimeWorkerController { + handle: started.handle, + shutdown: Arc::new(tokio::sync::Mutex::new(Some(started.shutdown))), + controller_task: started.controller_task, + workspace_client, + }) + } + } + + #[async_trait::async_trait] + impl worker_runtime::worker_backend::RuntimeWorkerFactory for AttentionWorkerFactory { + async fn spawn_controller( + &self, + request: worker_runtime::execution::WorkerExecutionSpawnRequest, + ) -> std::result::Result + { + self.controller(request.worker_ref.worker_id.to_string(), false) + .await + } + async fn restore_controller( + &self, + request: worker_runtime::execution::WorkerExecutionRestoreRequest, + ) -> std::result::Result + { + self.controller(request.worker_ref.worker_id.to_string(), true) + .await + } + } + + use worker_runtime::execution as attention_execution; + + // Fault injection occurs AFTER the real Controller durably accepted input. + struct AttentionResponseLossBackend { + inner: Arc, + lose_next: std::sync::atomic::AtomicBool, + attempts: Mutex>, + } + + impl attention_execution::WorkerExecutionBackend for AttentionResponseLossBackend { + fn backend_id(&self) -> &str { + self.inner.backend_id() + } + fn spawn_worker( + &self, + r: attention_execution::WorkerExecutionSpawnRequest, + ) -> attention_execution::WorkerExecutionSpawnResult { + self.inner.spawn_worker(r) + } + fn preflight_restore( + &self, + r: &attention_execution::WorkerExecutionRestoreRequest, + ) -> std::result::Result<(), attention_execution::WorkerExecutionResult> { + self.inner.preflight_restore(r) + } + fn reconcile_restore( + &self, + r: attention_execution::WorkerExecutionRestoreRequest, + ) -> attention_execution::WorkerExecutionSpawnResult { + self.inner.reconcile_restore(r) + } + fn restore_worker( + &self, + r: attention_execution::WorkerExecutionRestoreRequest, + ) -> attention_execution::WorkerExecutionSpawnResult { + self.inner.restore_worker(r) + } + fn activate_restored_worker( + &self, + id: attention_execution::WorkerLifecycleOperationId, + handle: &attention_execution::WorkerExecutionHandle, + ) -> std::result::Result<(), String> { + self.inner.activate_restored_worker(id, handle) + } + fn dispatch_input( + &self, + handle: &attention_execution::WorkerExecutionHandle, + input: worker_runtime::interaction::WorkerInput, + ) -> attention_execution::WorkerExecutionResult { + self.attempts + .lock() + .unwrap() + .push(input.submission_request_id.clone().unwrap()); + let result = self.inner.dispatch_input(handle, input); + assert_eq!( + result.outcome, + attention_execution::WorkerExecutionOutcome::Accepted + ); + if self + .lose_next + .swap(false, std::sync::atomic::Ordering::SeqCst) + { + attention_execution::WorkerExecutionResult::errored( + attention_execution::WorkerExecutionOperation::Input, + "injected loss of accepted response", + ) + } else { + result + } + } + fn dispatch_method( + &self, + handle: &attention_execution::WorkerExecutionHandle, + method: protocol::Method, + ) -> attention_execution::WorkerExecutionResult { + self.inner.dispatch_method(handle, method) + } + fn stop_worker_operation( + &self, + r: attention_execution::WorkerExecutionStopRequest, + ) -> attention_execution::WorkerExecutionResult { + self.inner.stop_worker_operation(r) + } + fn worker_snapshot( + &self, + handle: &attention_execution::WorkerExecutionHandle, + ) -> Option { + self.inner.worker_snapshot(handle) + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn orchestrator_attention_real_worker_start_restore_accept_and_commit_once_each() { + real_attention_worker_start_restore(false).await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn orchestrator_attention_real_worker_lost_response_retry_commits_once() { + real_attention_worker_start_restore(true).await; + } + + async fn real_attention_worker_start_restore(lose_response: bool) { + use session_store::{LogEntry, Store, WorkerSessionStore}; + let dir = tempfile::tempdir().unwrap(); + init_clean_git_workspace(dir.path()); + let config = test_server_config(dir.path()); + let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap()); + seed_test_registered_workspace(store.as_ref(), &config) + .await + .unwrap(); + // Real Worker Runtime adapter/controller/storage, isolated from profile + // workspace callbacks and real LLM credentials. + let execution = Arc::new( + worker_runtime::worker_backend::WorkerRuntimeExecutionBackend::new( + AttentionWorkerFactory { + root: config.embedded_runtime_store_root.clone(), + }, + ) + .unwrap(), + ); + let execution = Arc::new(AttentionResponseLossBackend { + inner: execution, + lose_next: std::sync::atomic::AtomicBool::new(lose_response), + attempts: Mutex::new(Vec::new()), + }); + let api = WorkspaceApi::new_with_execution_backend(config, store, execution.clone()) + .await + .unwrap(); + let mut input = ticket::NewTicket::new("Real notification backlog"); + input.workflow_state = Some(TicketWorkflowState::Queued); + set_test_ticket_target(&mut input, "test-repository", "HEAD"); + let ticket = browser_ticket_backend(&api).unwrap().create(input).unwrap(); + assign_test_orchestrator(&api, &ticket.id); + let started = start_attention_test_orchestrator(&api).await; + let worker = RuntimeWorkerRef::new( + EMBEDDED_WORKER_RUNTIME_ID, + started.worker.unwrap().worker_id, + ); + let session = WorkerSessionStore::open_read_only( + api.config + .embedded_runtime_store_root + .join("workers") + .join(&worker.worker_id) + .join("session"), + ) + .unwrap(); + let entries = || { + let id = session.session_id().unwrap().unwrap(); + session + .list_segments(id) + .unwrap() + .into_iter() + .flat_map(|segment| session.read_all_read_only(id, segment).unwrap()) + .collect::>() + }; + let receipts = || { + entries() + .iter() + .rev() + .find_map(|entry| match entry { + LogEntry::Extension { + domain, payload, .. + } if domain == "worker.pending_activations.v1" => { + Some(payload["notification_receipts"].as_array().unwrap().clone()) + } + LogEntry::AnnotatedSystemItem { extensions, .. } => extensions + .iter() + .find(|e| e.domain == "worker.pending_activations.v1") + .map(|e| { + e.payload["notification_receipts"] + .as_array() + .unwrap() + .clone() + }), + _ => None, + }) + .unwrap() + }; + let commits = || { + entries().iter().filter(|entry| matches!(entry, + LogEntry::AnnotatedSystemItem { entry, .. } + if matches!(&entry.item, session_store::SystemItem::Notification { message, .. } + if message.starts_with("Queued Tickets require attention:")) + )).count() + }; + async fn wait_for_commits usize>(count: F, expected: usize) { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let actual = count(); + assert!(actual <= expected, "duplicate canonical notifications"); + if actual == expected { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("controlled Worker notification commit"); + } + assert_eq!(receipts().len(), 1); + wait_for_commits(&commits, 1).await; + assert_eq!( + execution.attempts.lock().unwrap().len(), + if lose_response { 2 } else { 1 } + ); + let first_id = orchestrator_attention_state(&api, &worker) + .lock() + .unwrap() + .accepted + .as_ref() + .unwrap() + .request_id + .clone(); + use protocol::subscription::SubscriptionWorkerState as S; + for previous in [ + None, + Some(S::Idle), + Some(S::Stopped), + Some(S::Paused), + Some(S::Running), + ] { + maybe_dispatch_orchestrator_turn_end(&api, &worker.worker_id, previous, S::Idle); + } + assert_eq!( + start_attention_test_orchestrator(&api).await.disposition, + "existing" + ); + assert_eq!(receipts().len(), 1); + api.runtime + .stop_worker( + &worker, + WorkerLifecycleRequest { + reason: None, + ticket_assignment: None, + }, + ) + .unwrap(); + let restored = start_attention_test_orchestrator(&api).await; + assert_eq!(restored.disposition, "restored"); + let second_id = orchestrator_attention_state(&api, &worker) + .lock() + .unwrap() + .accepted + .as_ref() + .unwrap() + .request_id + .clone(); + assert_ne!(first_id, second_id); + wait_for_commits(&commits, 2).await; + let attempts = execution.attempts.lock().unwrap(); + assert_eq!(attempts.len(), if lose_response { 3 } else { 2 }); + assert_eq!( + attempts.iter().filter(|id| **id == first_id).count(), + if lose_response { 2 } else { 1 } + ); + assert_eq!(attempts.iter().filter(|id| **id == second_id).count(), 1); + drop(attempts); + let receipts = receipts(); + assert_eq!(receipts.len(), 2); + assert_eq!( + receipts + .iter() + .filter(|r| r["notification_request_id"] == first_id) + .count(), + 1 + ); + assert_eq!( + receipts + .iter() + .filter(|r| r["notification_request_id"] == second_id) + .count(), + 1 + ); + // Stop only this isolated test fixture; no live dogfooding Worker is touched. + api.runtime + .stop_worker( + &worker, + WorkerLifecycleRequest { + reason: None, + ticket_assignment: None, + }, + ) + .unwrap(); + } + + #[tokio::test] + async fn orchestrator_attention_restore_and_replacement_have_first_delivery() { + let (_dir, api, execution) = attention_test_api().await; + let started = start_attention_test_orchestrator(&api).await; + let worker = RuntimeWorkerRef::new( + EMBEDDED_WORKER_RUNTIME_ID, + started.worker.unwrap().worker_id, + ); + assert_eq!(execution.take_inputs().len(), 1); + api.runtime + .stop_worker( + &worker, + WorkerLifecycleRequest { + reason: None, + ticket_assignment: None, + }, + ) + .unwrap(); + execution.accept_restores(); + let restored = start_attention_test_orchestrator(&api).await; + assert_eq!(restored.disposition, "restored"); + assert_eq!(restored.worker.unwrap().worker_id, worker.worker_id); + assert_eq!(execution.take_inputs().len(), 1); + use protocol::subscription::SubscriptionWorkerState as S; + maybe_dispatch_orchestrator_turn_end(&api, &worker.worker_id, None, S::Idle); + assert_eq!( + start_attention_test_orchestrator(&api).await.disposition, + "existing" + ); + assert!(execution.take_inputs().is_empty()); + // This is a fixture Worker, not the running dogfooding Orchestrator. + api.runtime.delete_worker(&worker).unwrap(); + api.worker_projection + .publish_ordered(|| { + api.store + .delete_worker_registry(TEST_WORKSPACE_ID, &worker) + .map(|commit| ((), commit)) + }) + .unwrap(); + let replacement = start_attention_test_orchestrator(&api).await; + assert_ne!(replacement.worker.unwrap().worker_id, worker.worker_id); + assert_eq!(execution.take_inputs().len(), 1); + } + #[tokio::test] async fn orchestrator_running_to_idle_recovers_queued_ticket_without_notification_memory() { let dir = tempfile::tempdir().unwrap(); @@ -48074,7 +48847,6 @@ mod tests { set_test_ticket_target(&mut input, "test-repository", "HEAD"); let ticket_ref = backend.create(input).unwrap(); assign_test_orchestrator(&api, &ticket_ref.id); - *api.orchestrator_attention_fingerprint.lock().unwrap() = Some(ticket_ref.id.clone()); let Json(started) = scoped_start_workspace_orchestrator( State(api.clone()), @@ -48087,41 +48859,39 @@ mod tests { assert!(started.online); let startup_inputs = execution.take_inputs(); assert_eq!(startup_inputs.len(), 1); - assert_eq!( - api.orchestrator_attention_fingerprint - .lock() - .unwrap() - .as_deref(), - Some(ticket_ref.id.as_str()) - ); - *api.orchestrator_attention_fingerprint.lock().unwrap() = None; let worker_id = started.worker.as_ref().unwrap().worker_id.clone(); - maybe_dispatch_orchestrator_turn_end( - &api, - &worker_id, - Some(protocol::subscription::SubscriptionWorkerState::Idle), - protocol::subscription::SubscriptionWorkerState::Idle, - ); - assert!( - api.orchestrator_attention_fingerprint - .lock() - .unwrap() - .is_none() - ); - maybe_dispatch_orchestrator_turn_end( - &api, - &worker_id, - Some(protocol::subscription::SubscriptionWorkerState::Running), - protocol::subscription::SubscriptionWorkerState::Idle, - ); - + let worker = RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, &worker_id); + let attention = orchestrator_attention_state(&api, &worker); assert_eq!( - api.orchestrator_attention_fingerprint + attention .lock() .unwrap() - .as_deref(), - Some(ticket_ref.id.as_str()) + .accepted + .as_ref() + .unwrap() + .fingerprint, + ticket_ref.id ); + attention.lock().unwrap().accepted = None; + use protocol::subscription::SubscriptionWorkerState as S; + // Snapshot/reconnection and non-turn transitions cannot stand in for startup. + for previous in [None, Some(S::Idle), Some(S::Stopped), Some(S::Paused)] { + maybe_dispatch_orchestrator_turn_end(&api, &worker_id, previous, S::Idle); + assert!(attention.lock().unwrap().accepted.is_none()); + assert!(execution.take_inputs().is_empty()); + } + maybe_dispatch_orchestrator_turn_end(&api, &worker_id, Some(S::Running), S::Idle); + assert_eq!( + attention + .lock() + .unwrap() + .accepted + .as_ref() + .unwrap() + .fingerprint, + ticket_ref.id + ); + maybe_dispatch_orchestrator_turn_end(&api, &worker_id, Some(S::Running), S::Idle); let notifications = execution.take_inputs(); assert_eq!(notifications.len(), 1); assert_eq!(notifications[0].0.worker_id.to_string(), worker_id);