diff --git a/platform/event-log/src/root_store.rs b/platform/event-log/src/root_store.rs index 0dcbea68a4a..9fe058ef23b 100644 --- a/platform/event-log/src/root_store.rs +++ b/platform/event-log/src/root_store.rs @@ -223,10 +223,20 @@ impl RootStore { // observer before recovery can retain a failed SQL job or stage. store.recover_creations().await?; store.recover_checkpoints().await?; - // No admission or retirement can observe abandoned cross-partition - // auxiliary work as live after a process restart. + // No admission or retirement can observe work abandoned by the + // previous owner as live after a process restart. store.recover_host_effects().await?; store.recover_auxiliary_models().await?; + store + .recover_invocations() + .await + .map_err(|error| match error { + crate::recovery::RecoveryError::Store(error) => error, + error => StoreError::InvalidTransition(error.to_string()), + })?; + store + .recover_shell_runs(crate::native_approvals::now_millis()?) + .await?; store .lifecycle .as_mut() diff --git a/platform/event-log/src/root_store/computer_apps/tests.rs b/platform/event-log/src/root_store/computer_apps/tests.rs index f555fdc453e..1441f31a2a0 100644 --- a/platform/event-log/src/root_store/computer_apps/tests.rs +++ b/platform/event-log/src/root_store/computer_apps/tests.rs @@ -667,9 +667,16 @@ async fn always_belongs_to_selected_root_profile_across_workspaces_and_restart() drop(root); let reopened = RootStore::open(owner).await.unwrap(); let other = reopened.session("other").await.unwrap(); + let next = Invocation { + turn_id: "next-turn".into(), + run_id: "next-run".into(), + invocation_id: "next-invocation".into(), + ..other_invocation.clone() + }; + open(&other, &next).await; assert!( other - .computer_app_granted(&other_invocation, "post-restart-effect", &app, &other_basis) + .computer_app_granted(&next, "post-restart-effect", &app, &other_basis) .await .unwrap() ); diff --git a/platform/host-runtime/src/lib.rs b/platform/host-runtime/src/lib.rs index 8e8432d6fe6..43c7d7870f9 100644 --- a/platform/host-runtime/src/lib.rs +++ b/platform/host-runtime/src/lib.rs @@ -22,7 +22,6 @@ mod directories; pub mod http; pub mod ipc; pub mod logs; -pub mod recovery; pub mod sandbox; pub mod shell; diff --git a/platform/host-runtime/src/recovery.rs b/platform/host-runtime/src/recovery.rs deleted file mode 100644 index 3b8e257b06f..00000000000 --- a/platform/host-runtime/src/recovery.rs +++ /dev/null @@ -1,47 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -//! Reconcile abandoned execution facts before any Root workers or activations. - -use crate::HostError; -use maka_event_log::root_store::RootStore; - -/// Requires the original exclusive Root owner and no live execution workers. -/// This settles abandoned facts without replaying model, tool or shell effects. -/// Call once at Root startup, never from a plugin activation or live repair. -/// On failure, wait for the original SQL owner to close before returning Err. -pub async fn recover(log: &RootStore) -> Result<(), HostError> { - let result = async { - log.recover_invocations().await?; - let recovered_at = u64::try_from( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH)? - .as_millis(), - )?; - log.recover_shell_runs(recovered_at).await?; - log.recover_host_effects().await?; - log.recover_auxiliary_models().await?; - Ok(()) - } - .await; - if result.is_err() { - log.drain().await; - } - result -} diff --git a/plugins/agent/engine/tests/integration/startup_recovery.rs b/plugins/agent/engine/tests/integration/startup_recovery.rs index a8264ec891d..cb69d345545 100644 --- a/plugins/agent/engine/tests/integration/startup_recovery.rs +++ b/plugins/agent/engine/tests/integration/startup_recovery.rs @@ -70,7 +70,7 @@ async fn unknown_dispatch_stays_unknown_and_blocks_next_admission_after_reopen() .await .unwrap(), ); - assert_eq!(log.root().recover_invocations().await.unwrap(), 1); + assert_eq!(log.root().recover_invocations().await.unwrap(), 0); let prefix = log.prefix(100, 128 * 1024).await.unwrap(); assert_eq!(prefix.events.len(), 5); assert!( diff --git a/plugins/agent/native/src/owner.rs b/plugins/agent/native/src/owner.rs index 88940997993..23b40ce2877 100644 --- a/plugins/agent/native/src/owner.rs +++ b/plugins/agent/native/src/owner.rs @@ -1440,7 +1440,10 @@ impl Owner { }); } Err(maka_event_log::StoreError::RevisionConflict { .. }) => continue, - Err(_) => return Err("agent_admission_failed".into()), + Err(error) => { + tracing::warn!(%error, "queued message admission failed"); + return Err("agent_admission_failed".into()); + } } } } @@ -1454,7 +1457,10 @@ impl Owner { .map_err(|error| error.to_string())? .admit_message(pending) .await - .map_err(|_| "agent_admission_failed")?; + .map_err(|error| { + tracing::warn!(%error, "message admission failed"); + "agent_admission_failed" + })?; let cancellation = self.stop.child_token(); let running = match self.start(input, cancellation.clone()).await { Ok(running) => running, diff --git a/plugins/agent/native/src/owner/tests/todo_tests.rs b/plugins/agent/native/src/owner/tests/todo_tests.rs index 1a7a8d3b472..1a55eaa453e 100644 --- a/plugins/agent/native/src/owner/tests/todo_tests.rs +++ b/plugins/agent/native/src/owner/tests/todo_tests.rs @@ -194,7 +194,7 @@ async fn user_messages_after_abandoned_tool_dispatch_preserve_unknown_without_re db.execute_batch("DROP TRIGGER reject_settlement").unwrap(); drop(db); // Root recovery runs only after every physical worker has stopped. - maka_host_runtime::recovery::recover(&f.store).await.unwrap(); + f.store.recover_invocations().await.unwrap(); assert!(f.session(&session).await.check_manual_message_history(&session).await.unwrap()); reopen(&mut f).await; f.mutation(2, "send", json!({"sessionId":session,"message":"What happened to the checklist?","model":f.selection().await}), Uuid::new_v4()).await.unwrap(); @@ -205,7 +205,7 @@ async fn user_messages_after_abandoned_tool_dispatch_preserve_unknown_without_re while !f.query("observe", json!({"sessionId":session})).await.unwrap()["run"].is_null() { tokio::task::yield_now().await; } assert!(f.session(&session).await.pending_messages(&session).await.unwrap().is_empty()); f.app.shutdown().await.unwrap(); - maka_host_runtime::recovery::recover(&f.store).await.unwrap(); + f.store.recover_invocations().await.unwrap(); reopen(&mut f).await; f.send(&session).await.unwrap(); let (mut socket, _) = listener.accept().await.unwrap(); diff --git a/plugins/agent/native/src/service.rs b/plugins/agent/native/src/service.rs index efffb130905..94398f8d307 100644 --- a/plugins/agent/native/src/service.rs +++ b/plugins/agent/native/src/service.rs @@ -1021,43 +1021,50 @@ impl Handler for Agent { let mut queue_revision = None; let mut interactions_delivered = None; let mut computer_apps_delivered = None; + // Previews live in memory: the store is read only for a commit + // notice or for committed events not delivered yet. + let mut stale = true; loop { if context.scope().cancellation.is_cancelled() { return Err("agent_observation_cancelled".into()); } - // Queue edits need not advance the transcript cursor. Deliver - // them fairly even while catching up a growing transcript. - let revision = owner - .scoped_store(context.scope(), &input.session_id) - .await? - .message_queue_revision(&input.session_id) - .await - .map_err(|_| "agent_queue_unavailable")?; - if queue_revision != Some(revision) { - let queue = owner.queue(&input.session_id).await?; - queue_revision = Some(queue.revision); - let update = encode(DeliveryPage::Queue { queue })?; - maka_plugin_core::json::exact_numbers(&update) + if stale { + // Queue edits need not advance the transcript cursor. Deliver + // them fairly even while catching up a growing transcript. + let revision = owner + .scoped_store(context.scope(), &input.session_id) + .await? + .message_queue_revision(&input.session_id) + .await .map_err(|_| "agent_queue_unavailable")?; - context.emit(update).await?; - } - let pending = owner.pending_interactions(&input.session_id).await?; - if interactions_delivered.as_ref() != Some(&pending) { - context - .emit(encode(DeliveryPage::Interactions { - pending: pending.clone(), - })?) - .await?; - interactions_delivered = Some(pending); - } - let apps = owner.computer_apps(&input.session_id).await?; - if computer_apps_delivered.as_ref() != Some(&apps) { - context - .emit(encode(DeliveryPage::ComputerApps { apps: apps.clone() })?) - .await?; - computer_apps_delivered = Some(apps); + if queue_revision != Some(revision) { + let queue = owner.queue(&input.session_id).await?; + queue_revision = Some(queue.revision); + let update = encode(DeliveryPage::Queue { queue })?; + maka_plugin_core::json::exact_numbers(&update) + .map_err(|_| "agent_queue_unavailable")?; + context.emit(update).await?; + } + let pending = owner.pending_interactions(&input.session_id).await?; + if interactions_delivered.as_ref() != Some(&pending) { + context + .emit(encode(DeliveryPage::Interactions { + pending: pending.clone(), + })?) + .await?; + interactions_delivered = Some(pending); + } + let apps = owner.computer_apps(&input.session_id).await?; + if computer_apps_delivered.as_ref() != Some(&apps) { + context + .emit(encode(DeliveryPage::ComputerApps { + apps: apps.clone(), + })?) + .await?; + computer_apps_delivered = Some(apps); + } + owner.require_session(&input.session_id).await?; } - owner.require_session(&input.session_id).await?; let mut reader = owner.preview_reader(&input.session_id).await; let current = reader .as_mut() @@ -1069,46 +1076,57 @@ impl Handler for Agent { { reader = None; } - let through = owner - .scoped_store(context.scope(), &input.session_id) - .await? - .committed_sequence() - .await - .map_err(|error| error.to_string())?; - if after > through { - return Err("agent_cursor_invalid".into()); + if stale { + let committed = owner + .scoped_store(context.scope(), &input.session_id) + .await? + .committed_sequence() + .await + .map_err(|error| error.to_string())?; + if after > committed { + return Err("agent_cursor_invalid".into()); + } + let through = preview + .initial_fence() + .map_or(committed, |fence| committed.min(fence.max(after))); + let page = owner + .scoped_store(context.scope(), &input.session_id) + .await? + .session_stream_events( + &input.session_id, + after, + through, + 32, + 48 * 1024, + ) + .await + .map_err(|_| "agent_observation_unavailable")?; + preview.committed(&page.events); + after = page.next_after.unwrap_or(page.through_sequence); + stale = page.next_after.is_some() || through < committed; + context + .emit(encode(DeliveryPage::Events { + through: page.through_sequence, + next_after: after, + events: page.events.into_iter().map(delivery).collect(), + })?) + .await?; } - let through = preview - .initial_fence() - .map_or(through, |fence| through.min(fence.max(after))); - let page = owner - .scoped_store(context.scope(), &input.session_id) - .await? - .session_stream_events(&input.session_id, after, through, 32, 48 * 1024) - .await - .map_err(|_| "agent_observation_unavailable")?; - preview.committed(&page.events); - after = page.next_after.unwrap_or(page.through_sequence); - context - .emit(encode(DeliveryPage::Events { - through: page.through_sequence, - next_after: after, - events: page.events.into_iter().map(delivery).collect(), - })?) - .await?; + let mut ready = stale; if let Some(update) = preview.next(after)? { context .emit(encode(DeliveryPage::Preview { page: update })?) .await?; - continue; - } - if page.next_after.is_some() { - continue; + ready = true; } tokio::select! { biased; _ = context.scope().cancellation.cancelled() => return Err("agent_observation_cancelled".into()), - result = commits.changed() => result.map_err(|_| "agent_observation_closed")?, + result = commits.changed() => { + result.map_err(|_| "agent_observation_closed")?; + stale = true; + }, + () = std::future::ready(()), if ready => {}, _ = async { match reader.as_mut() { Some(reader) => { let _ = reader.changed().await; }, diff --git a/products/maka/cli/src/code.rs b/products/maka/cli/src/code.rs index a1b9d61b748..25797817eb9 100644 --- a/products/maka/cli/src/code.rs +++ b/products/maka/cli/src/code.rs @@ -40,7 +40,6 @@ use maka_code_runtime::{ CellAbort, CellContext, CellDiagnosticKind, CellLimits, CellResult, CellStore, ToolMetadata, }; use maka_event_log::root::{RootNamespaces, RootOwner}; -use maka_event_log::root_store::RootStore; use maka_fs_tools::{MutationExecutor, ReadExecutor, ReadScope, WriteCoordinator, WriteScope}; use maka_js_runtime::CodeExecutor; use maka_sandbox::{ @@ -132,8 +131,7 @@ pub(super) async fn run(args: Args) -> Result<(), maka_host_runtime::HostError> } else { RootOwner::create(&args.root, &namespaces)? }; - let root = Arc::new(RootStore::open(Arc::new(owner)).await?); - maka_host_runtime::recovery::recover(&root).await?; + let root = Arc::new(maka_transport::open_store(Arc::new(owner)).await?); let (mut sandbox, _) = maka_fs_tools::workspace::permissions::resolve(sandbox_mode, &cwd)?; if let Sandbox::Managed { filesystem, .. } = &mut sandbox { for path in [&args.root, &namespaces.ownership, &namespaces.control] { diff --git a/products/maka/cli/tests/integration/candidate.rs b/products/maka/cli/tests/integration/candidate.rs index f2e5bba435f..dd18b1d23ce 100644 --- a/products/maka/cli/tests/integration/candidate.rs +++ b/products/maka/cli/tests/integration/candidate.rs @@ -28,6 +28,8 @@ use std::{ time::{Duration, Instant}, }; +#[cfg(unix)] +mod crash; mod owned; mod startup; diff --git a/products/maka/cli/tests/integration/candidate/crash.rs b/products/maka/cli/tests/integration/candidate/crash.rs new file mode 100644 index 00000000000..50f4b9cbc7e --- /dev/null +++ b/products/maka/cli/tests/integration/candidate/crash.rs @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +use super::*; +use crate::terminal::support::{ModelRequest, configure, mutate}; +use maka_app_client::{Client as AppClient, Journal}; +use maka_host_runtime::application::host::startup; +use serde_json::json; +use std::sync::{Arc, atomic::AtomicUsize}; +use tokio::{net::TcpListener, sync::mpsc}; + +async fn activity(client: &AppClient, session: &str) -> Value { + client + .query("agent", "observe", json!({"sessionId":session})) + .await + .unwrap()["session"]["activity"] + .clone() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn killed_host_leaves_an_ended_turn_and_a_session_that_takes_messages() { + let mut fixture = RootStartupFixture::new(); + let root = fixture.root.clone(); + let executable = Path::new(env!("CARGO_BIN_EXE_maka")); + let journal = fixture.directory().join("client.db"); + let owner = Arc::new( + RootOwner::create(&root, &RootNamespaces::for_current_account().unwrap()).unwrap(), + ); + fixture.adopt(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let selection = configure( + owner.clone(), + &format!("http://{}/v1", listener.local_addr().unwrap()), + false, + ) + .await; + drop(owner); + let (send, mut requests) = mpsc::channel(8); + tokio::spawn(async move { + let count = Arc::new(AtomicUsize::new(0)); + loop { + let (socket, _) = listener.accept().await.unwrap(); + tokio::spawn(ModelRequest::serve(socket, send.clone(), count.clone())); + } + }); + tokio::time::timeout(Duration::from_secs(90), async { + let host = startup::connect_root(executable, &root).await.unwrap(); + let client = AppClient::new( + host, + Journal::open(&journal).await.unwrap(), + "receipts".into(), + ) + .unwrap(); + let roots = client + .query("agent", "directories", json!({"kind":"roots"})) + .await + .unwrap(); + let created = mutate(&client, "create", json!({"title":"Crash","labels":[],"workspace":{"kind":"directory","selection":roots["roots"][0]["selection"]}})).await; + let session = created["session"]["id"].as_str().unwrap().to_owned(); + mutate(&client, "send", json!({"sessionId":session,"message":"BEFORE_THE_CRASH","model":selection})).await; + let held = requests.recv().await.unwrap(); + assert_ne!(activity(&client, &session).await, "idle"); + + let pid = discovery::read(&root).unwrap().pid.get(); + assert_eq!(unsafe { libc::kill(pid as libc::pid_t, libc::SIGKILL) }, 0); + drop(held); + drop(client); + + let host = startup::connect_root(executable, &root).await.unwrap(); + assert_ne!(discovery::read(&root).unwrap().pid.get(), pid); + let client = AppClient::new( + host, + Journal::open(&journal).await.unwrap(), + "receipts".into(), + ) + .unwrap(); + assert_eq!(activity(&client, &session).await, "idle"); + mutate(&client, "send", json!({"sessionId":session,"message":"AFTER_THE_CRASH","model":selection})).await; + let mut answered = false; + while activity(&client, &session).await != "idle" || !answered { + let Ok(request) = + tokio::time::timeout(Duration::from_millis(50), requests.recv()).await + else { + continue; + }; + let mut request = request.unwrap(); + answered |= request.body["messages"].to_string().contains("AFTER_THE_CRASH"); + request.delta("ANSWER").await; + request.finish().await; + } + let history = client + .query("agent", "history", json!({"sessionId":session,"position":{"kind":"start"}})) + .await + .unwrap(); + let chunks = history["chunks"].as_array().unwrap(); + let interrupted = chunks + .iter() + .find(|chunk| chunk.to_string().contains("BEFORE_THE_CRASH")) + .unwrap(); + assert_eq!(interrupted["timing"]["finished"]["outcome"], "failed"); + assert!(chunks.iter().any(|chunk| chunk.to_string().contains("ANSWER"))); + client.close().await.unwrap(); + }) + .await + .expect("recovery after a killed Host timed out"); +} diff --git a/products/maka/terminal-shell/tests/real_host/mod.rs b/products/maka/terminal-shell/tests/real_host/mod.rs index f957a8586aa..319753c6f7e 100644 --- a/products/maka/terminal-shell/tests/real_host/mod.rs +++ b/products/maka/terminal-shell/tests/real_host/mod.rs @@ -26,7 +26,7 @@ mod desktop_new_session; mod keyboard; mod resumable; mod send_hold; -mod support; +pub(crate) mod support; mod ui_plugins; use crossterm::event::{Event, KeyCode, KeyEvent, KeyModifiers}; diff --git a/products/maka/terminal-shell/tests/real_host/support.rs b/products/maka/terminal-shell/tests/real_host/support.rs index 752b53100ee..dcea21c0241 100644 --- a/products/maka/terminal-shell/tests/real_host/support.rs +++ b/products/maka/terminal-shell/tests/real_host/support.rs @@ -374,7 +374,7 @@ impl Fixture { assert!(self.provider.await.unwrap_err().is_cancelled()); } } -async fn configure(owner: Arc, base: &str, vision: bool) -> Value { +pub(crate) async fn configure(owner: Arc, base: &str, vision: bool) -> Value { use maka_agent_facts::configuration::CatalogMutationResult; let config = Arc::new( maka_config::ConfigurationStore::for_root(owner) @@ -457,7 +457,11 @@ fn model_http_error(body: Value) -> Response { } impl ModelRequest { - async fn serve(socket: TcpStream, requests: mpsc::Sender, count: Arc) { + pub(crate) async fn serve( + socket: TcpStream, + requests: mpsc::Sender, + count: Arc, + ) { let completed = Arc::new(Notify::new()); let notification = completed.clone(); let service = service_fn(move |request: Request| {