fix(runtime): recover from a killed Host and stop rereading the store for previews

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()

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()
     );

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;
 

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!(

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,

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();

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; },

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] {

products/maka/cli/tests/integration/candidate.rs

@@ -28,6 +28,8 @@ use std::{
     time::{Duration, Instant},
 };
 
+#[cfg(unix)]
+mod crash;
 mod owned;
 mod startup;