Bind actors use the module's storage root; only the daemon module recovers the call ledger

.alfonso/release-notes/next/actor-storage-propagation.md

@@ -0,0 +1 @@
+An OpenCode or Pi session running AFT without the daemon no longer marks a call the daemon is still running as "not started" or "outcome unknown" when both use the same AFT storage.

benchmarks/aft-search/slice-descriptors/land-actor-storage-ledger-recovery.json

@@ -0,0 +1,8 @@
+{
+  "targeted_mechanism": "none",
+  "kind": "harness",
+  "slice_class": "non_ranking",
+  "fixtures": [
+    "harness-goldens"
+  ]
+}

crates/aft/src/db/call_ledger.rs

@@ -899,8 +899,34 @@ pub fn ack_through(
     ack(conn, custodian, through.seq).map(Ok)
 }
 
+/// Whether this process admits keyed tool-provider calls and therefore owns
+/// restart recovery of the call ledger. Only the subc module admits ledger
+/// rows; standalone sessions and CLI commands share the same aft.db but never
+/// write a row, so they must not settle the module's in-flight calls. Unit
+/// tests run as a provider so their direct ledger fixtures keep recovering.
+static LEDGER_PROVIDER_PROCESS: std::sync::atomic::AtomicBool =
+    std::sync::atomic::AtomicBool::new(cfg!(test));
+
+/// Marks this process as the call ledger's provider. The subc module calls it
+/// before it can open a database, so its first pool runs restart recovery.
+pub fn mark_provider_process() {
+    LEDGER_PROVIDER_PROCESS.store(true, std::sync::atomic::Ordering::SeqCst);
+}
+
+fn is_provider_process() -> bool {
+    LEDGER_PROVIDER_PROCESS.load(std::sync::atomic::Ordering::SeqCst)
+}
+
 /// Called once when a process installs its database, before admitting calls.
 /// A dispatch with a durable task is observed, never re-sent.
+///
+/// Recovery settles every unsettled row in the file. Rows record no owning
+/// process, so this is correct only while one module process owns a given
+/// aft.db's call ledger: a second module sharing the file would have its
+/// in-flight calls settled here as not_started or unknown. A change that lets
+/// two modules share one storage root must first add an owner column and limit
+/// recovery to rows whose owner is gone. Processes that never admit calls skip
+/// recovery entirely (see [`recover_pool_once`]).
 pub fn recover(conn: &Connection) -> rusqlite::Result<()> {
     let mut stmt =
         conn.prepare("SELECT carrier,call_key FROM call_ledger WHERE state!='Settled'")?;
@@ -953,9 +979,16 @@ pub fn recover(conn: &Connection) -> rusqlite::Result<()> {
 /// Project actors share the tracked connection. Attaching a second registry
 /// to that connection is not a provider restart. Weak references also avoid
 /// keeping closed databases alive or confusing a reused allocation address.
+///
+/// Does nothing outside the provider process: a standalone session opening
+/// the shared aft.db has no ledger rows of its own, and every unsettled row
+/// there may belong to a module that is still running the call.
 pub fn recover_pool_once(
     pool: &std::sync::Arc<std::sync::Mutex<super::TrackedConnection>>,
 ) -> rusqlite::Result<()> {
+    if !is_provider_process() {
+        return Ok(());
+    }
     type Pool = std::sync::Mutex<super::TrackedConnection>;
     static RECOVERED: std::sync::LazyLock<std::sync::Mutex<Vec<std::sync::Weak<Pool>>>> =
         std::sync::LazyLock::new(Default::default);

crates/aft/src/subc/bind_capacity_tests.rs

@@ -74,6 +74,7 @@ async fn exhausted_bind(tool: bool, completion_capacity: bool) {
     let mut live = HashMap::new();
     let mut pending = HashMap::new();
     let mut installed = HashMap::new();
+    let storage = tempfile::tempdir().unwrap();
     handle_control_request(
         &writer,
         &writer,
@@ -107,6 +108,7 @@ async fn exhausted_bind(tool: bool, completion_capacity: bool) {
         Some(&user),
         usize::MAX,
         None,
+        storage.path(),
     )
     .await
     .unwrap();

crates/aft/src/subc/bind_log.rs

@@ -98,6 +98,8 @@ mod tests {
                 None,
                 usize::MAX,
                 None,
+                // A management bind creates no actor, so no storage is opened.
+                Path::new("unused-management-storage"),
             )
             .await
             .unwrap();

crates/aft/src/subc/mod.rs

@@ -4129,6 +4129,9 @@ fn run_subc_mode_inner(
     tool_response_body_limit: usize,
     lifecycle_probe: Option<SubcTestLifecycleProbe>,
 ) -> Result<(), SubcError> {
+    // Before any database opens: this process admits keyed calls, so it is
+    // the one that recovers the call ledger after a restart.
+    crate::db::call_ledger::mark_provider_process();
     let _daemon_storage_lease = crate::root_cache::acquire_daemon_storage_lease(&ctx.storage_dir())
         .map_err(SubcError::StorageLease)?;
     let runtime = tokio::runtime::Builder::new_current_thread()
@@ -5675,6 +5678,7 @@ where
                             user_config_path.as_deref(),
                             tool_response_body_limit,
                             Some(&run_scopes),
+                            storage_dir,
                         )
                         .await
                         {
@@ -7039,13 +7043,20 @@ async fn close_daemon_connection(
     }
 }
 
+/// `storage_dir` is the storage root the module selected at startup, the one
+/// its daemon storage lease covers. A new actor starts from it so that the
+/// bind's configure opens aft.db and the actor's other durable state in that
+/// same root. Starting from an empty `storage_dir` would let the actor fall
+/// back to the default data home and split the module's data in two whenever
+/// the module was started with an explicit storage path.
 fn register_actor_for_bind(
     shared_app: &Arc<App>,
     executor: &Arc<Executor>,
     push_senders: &PushSenders,
     bind_root_id: &ProjectRootId,
     route_channel: u16,
     root_was_live: bool,
+    storage_dir: &Path,
 ) -> bool {
     if executor.actor_registered(bind_root_id) {
         log::debug!(
@@ -7066,7 +7077,10 @@ fn register_actor_for_bind(
 
     let actor_ctx = Arc::new(AppContext::from_app(
         Arc::clone(shared_app),
-        Config::default(),
+        Config {
+            storage_dir: Some(storage_dir.to_path_buf()),
+            ..Config::default()
+        },
     ));
     install_bash_compressor(&actor_ctx);
     actor_ctx.install_fleet_status_client(Some(push_senders.fleet_status_client.clone()));
@@ -7842,6 +7856,7 @@ async fn handle_control_request(
     user_config_path: Option<&Path>,
     tool_response_body_limit: usize,
     run_scopes: Option<&run_scope::RunScopes>,
+    storage_dir: &Path,
 ) -> Result<(), SubcError> {
     metrics.mark_frame_loop_entry("handle_control_request", frame);
     let request = match serde_json::from_slice::<ModuleControlRequest>(&frame.body) {
@@ -8258,6 +8273,7 @@ async fn handle_control_request(
                 &bind_root_id,
                 route_channel,
                 root_was_live,
+                storage_dir,
             );
 
             sync_bg_live_delivery_sessions(executor, routes, Some(&bind_root_id), Some(metrics));
@@ -14806,6 +14822,7 @@ mod tests {
                 Some(&test_support::bind_config_without_indexes(dir.path())),
                 usize::MAX,
                 None,
+                &ctx.storage_dir(),
             )
             .await
             .expect("admit route bind");

crates/aft/src/subc/prepared_bind_tests.rs

@@ -173,6 +173,7 @@ async fn real_bind_dispatch_scope_survives_config_edits_before_executor_admissio
         Some(&user),
         usize::MAX,
         None,
+        &dir.path().join("storage"),
     )
     .await;
     std::fs::write(&user, r#"{"url_fetch_allow_private":false}"#).unwrap();

crates/aft/src/subc/scope_workdir_tests.rs

@@ -77,6 +77,7 @@ fn bind_observed(
     let (reply, lines) = crate::logging::capture_log_lines(|| {
         runtime.block_on(async {
             let app = App::default_shared();
+            let storage = tempfile::tempdir().unwrap();
             let executor = Arc::new(Executor::new());
             let metrics = Arc::new(DispatchPathMetrics::new());
             let (writer_tx, mut writer_rx) = mpsc::channel(8);
@@ -160,6 +161,7 @@ fn bind_observed(
                 Some(&test_support::bind_config_without_indexes(&root)),
                 usize::MAX,
                 None,
+                storage.path(),
             )
             .await
             .unwrap();

crates/aft/tests/integration/main.rs

@@ -165,6 +165,7 @@ mod score_floor_drop_test;
 mod search_index_warm_restart_test;
 mod security_test;
 mod semantic_disk_test;
+mod shared_db_call_ledger_test;
 mod shared_db_contention_test;
 mod shared_index_scope_test;
 mod stage_card_floor_test;

crates/aft/tests/integration/shared_db_call_ledger_test.rs

@@ -0,0 +1,102 @@
+//! A standalone `aft` process (a plugin bridge session, for example) can share
+//! one `aft.db` with a running subc module. Only the module admits keyed
+//! tool-provider calls, so only the module may run the call ledger's restart
+//! recovery: a standalone process that recovered on opening the database
+//! would settle the module's in-flight calls as not started or unknown.
+
+use std::time::{Duration, Instant};
+
+use aft::db::call_ledger::{self as ledger, Admission, Key, State};
+use serde_json::json;
+
+use super::helpers::AftProcess;
+
+fn key(call_key: &str) -> Key {
+    Key {
+        carrier: "reserved:carrier".into(),
+        call_key: call_key.into(),
+    }
+}
+
+fn state_of(conn: &rusqlite::Connection, call_key: &str) -> State {
+    ledger::get(conn, &key(call_key))
+        .expect("read ledger row")
+        .expect("ledger row exists")
+        .state
+}
+
+#[test]
+fn standalone_configure_leaves_a_running_module_s_unsettled_calls_alone() {
+    let project = tempfile::tempdir().expect("project dir");
+    let storage = tempfile::tempdir().expect("storage dir");
+    let db_path = storage.path().join("aft.db");
+
+    // Stand in for a module that is still running these calls: this test
+    // process admits them and stays alive throughout.
+    let module = aft::db::open(&db_path).expect("create shared aft.db");
+    for (call_key, state) in [
+        ("held-before-dispatch", State::Prepared),
+        ("dispatch-in-flight", State::DispatchStarted),
+    ] {
+        let admission = ledger::admit(&module, &key(call_key), "digest", None, state)
+            .expect("admit the module's call");
+        assert!(matches!(admission, Admission::New));
+    }
+    drop(module);
+
+    let mut aft = AftProcess::spawn();
+    let response = aft.send(
+        &json!({
+            "id": "cfg",
+            "command": "configure",
+            "harness": "opencode",
+            "project_root": project.path(),
+            "storage_dir": storage.path(),
+        })
+        .to_string(),
+    );
+    assert_eq!(response["success"], true, "configure failed: {response:?}");
+    // A persistence-requiring command succeeds only after the standalone
+    // process has published its database, which is when it would recover.
+    let deadline = Instant::now() + Duration::from_secs(30);
+    loop {
+        let read = aft.send(
+            &json!({
+                "id": "probe",
+                "command": "db_get_state",
+                "params": { "key": "ledger-recovery-probe" },
+            })
+            .to_string(),
+        );
+        if read["success"] == true {
+            break;
+        }
+        assert_eq!(read["code"], "database_initializing", "{read:?}");
+        assert!(
+            Instant::now() < deadline,
+            "standalone database never opened: {read:?}"
+        );
+        std::thread::sleep(Duration::from_millis(50));
+    }
+
+    let check = aft::db::open(&db_path).expect("reopen shared aft.db");
+    assert_eq!(state_of(&check, "held-before-dispatch"), State::Prepared);
+    assert_eq!(
+        state_of(&check, "dispatch-in-flight"),
+        State::DispatchStarted
+    );
+    drop(check);
+    let status = aft.shutdown();
+    assert!(status.success(), "aft exited with {status}");
+
+    // The module process itself still recovers after a restart: once this
+    // process is marked as the provider, a fresh pool settles both rows.
+    ledger::mark_provider_process();
+    let pool = std::sync::Arc::new(std::sync::Mutex::new(
+        aft::db::open(&db_path).expect("open as the restarted module"),
+    ));
+    ledger::recover_pool_once(&pool).expect("module recovery");
+    let conn = pool.lock().unwrap();
+    assert_eq!(state_of(&conn, "held-before-dispatch"), State::Settled);
+    assert_eq!(state_of(&conn, "dispatch-in-flight"), State::Settled);
+}

crates/aft/tests/integration/subc_bridge_test.rs

@@ -2354,6 +2354,85 @@ fn subc_bridge_same_root_rebind_is_acked_while_a_read_is_held() {
     );
 }
 
+/// The SQLite main file the first bind's actor opened, recorded by the driver
+/// and checked once the run has finished.
+static FRESH_ACTOR_DB_PATH: Mutex<Option<std::path::PathBuf>> = Mutex::new(None);
+
+/// The fixture starts the module with an explicit storage path. The actor a
+/// first bind creates must open its aft.db there, not in the default data
+/// home, so the module and every actor keep one storage root.
+#[test]
+fn subc_bridge_fresh_bind_actor_opens_aft_db_in_the_module_storage_root() {
+    *FRESH_ACTOR_DB_PATH.lock().unwrap() = None;
+    run_subc_bridge_test_inner(
+        "subc_bridge_fresh_bind_actor_opens_aft_db_in_the_module_storage_root",
+        Duration::from_secs(45),
+        // AFT_STORAGE_DIR outranks an explicit path for the module and its
+        // actors alike, which would hide where the actor's own choice lands.
+        || vec![remove_test_env("AFT_STORAGE_DIR")],
+        drive_fresh_actor_database_daemon,
+        |_, _, roots| {
+            let opened = FRESH_ACTOR_DB_PATH
+                .lock()
+                .unwrap()
+                .take()
+                .expect("driver recorded the actor's database");
+            let expected = roots.storage.path().join("aft.db");
+            assert_eq!(
+                std::fs::canonicalize(&opened).expect("opened database exists"),
+                std::fs::canonicalize(&expected).expect("module storage database exists"),
+                "actor opened {} instead of the module's {}",
+                opened.display(),
+                expected.display()
+            );
+        },
+        true,
+        None,
+        real_configure_bridge_dispatch,
+        bridge_executor_config(),
+        false,
+        None,
+        None,
+    );
+}
+
+async fn drive_fresh_actor_database_daemon(input: FakeDaemonInput) {
+    let FakeDaemonSession {
+        mut stream,
+        root1,
+        executor,
+        ..
+    } = open_fake_daemon_session(input).await;
+    bind_route1(&mut stream, &root1).await;
+    let root_id = ProjectRootId::from_path(&root1).expect("root1 id");
+    let actor = executor.actor_context(&root_id).expect("bound actor");
+    // The database opens after the bind is acknowledged; wait until the actor
+    // would admit a persistence-requiring tool.
+    let deadline = Instant::now() + Duration::from_secs(30);
+    let db = loop {
+        if actor.database_runtime_refusal("probe", "bash").is_none() {
+            if let Some(db) = actor.db() {
+                break db;
+            }
+        }
+        assert!(
+            Instant::now() < deadline,
+            "the bound actor never opened its database"
+        );
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    };
+    let opened = db
+        .lock()
+        .unwrap()
+        .path()
+        .map(std::path::PathBuf::from)
+        .expect("aft.db is a file-backed database");
+    *FRESH_ACTOR_DB_PATH.lock().unwrap() = Some(opened);
+    drop(db);
+    drop(actor);
+    send_connection_goodbye(&mut stream).await;
+}
+
 #[test]
 fn subc_bridge_configure_warning_pushes_are_session_scoped() {
     run_subc_bridge_test(