From 891f0a3b41f709faf3a9fe1ea1ea7c56babf1dc0 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:36:24 +0100 Subject: [PATCH 1/4] mason: fresh bind actors use the module storage root; only the subc module recovers the call ledger Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> (cherry picked from commit a296566b075417657083f926e34504f7f296f9d8) --- crates/aft/src/db/call_ledger.rs | 33 ++++++ crates/aft/src/subc/bind_log.rs | 2 + crates/aft/src/subc/mod.rs | 19 +++- crates/aft/src/subc/scope_workdir_tests.rs | 2 + crates/aft/tests/integration/main.rs | 1 + .../integration/shared_db_call_ledger_test.rs | 102 ++++++++++++++++++ .../aft/tests/integration/subc_bridge_test.rs | 79 ++++++++++++++ 7 files changed, 237 insertions(+), 1 deletion(-) create mode 100644 crates/aft/tests/integration/shared_db_call_ledger_test.rs diff --git a/crates/aft/src/db/call_ledger.rs b/crates/aft/src/db/call_ledger.rs index d0e7397c61..29d743125b 100644 --- a/crates/aft/src/db/call_ledger.rs +++ b/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>, ) -> rusqlite::Result<()> { + if !is_provider_process() { + return Ok(()); + } type Pool = std::sync::Mutex; static RECOVERED: std::sync::LazyLock>>> = std::sync::LazyLock::new(Default::default); diff --git a/crates/aft/src/subc/bind_log.rs b/crates/aft/src/subc/bind_log.rs index de1774ddf1..f9eccae218 100644 --- a/crates/aft/src/subc/bind_log.rs +++ b/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(); diff --git a/crates/aft/src/subc/mod.rs b/crates/aft/src/subc/mod.rs index b709cee33f..c1f1245ed0 100644 --- a/crates/aft/src/subc/mod.rs +++ b/crates/aft/src/subc/mod.rs @@ -4129,6 +4129,9 @@ fn run_subc_mode_inner( tool_response_body_limit: usize, lifecycle_probe: Option, ) -> 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,6 +7043,12 @@ 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, executor: &Arc, @@ -7046,6 +7056,7 @@ fn register_actor_for_bind( 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::(&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"); diff --git a/crates/aft/src/subc/scope_workdir_tests.rs b/crates/aft/src/subc/scope_workdir_tests.rs index c18a3f2347..3a16c39880 100644 --- a/crates/aft/src/subc/scope_workdir_tests.rs +++ b/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(); diff --git a/crates/aft/tests/integration/main.rs b/crates/aft/tests/integration/main.rs index f231b5bbb1..28053ac8de 100644 --- a/crates/aft/tests/integration/main.rs +++ b/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; diff --git a/crates/aft/tests/integration/shared_db_call_ledger_test.rs b/crates/aft/tests/integration/shared_db_call_ledger_test.rs new file mode 100644 index 0000000000..0290dd008d --- /dev/null +++ b/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); +} diff --git a/crates/aft/tests/integration/subc_bridge_test.rs b/crates/aft/tests/integration/subc_bridge_test.rs index 2e4a555d7f..d2de6f5694 100644 --- a/crates/aft/tests/integration/subc_bridge_test.rs +++ b/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> = 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( From c1cb01dd9a68502a65eda371a5fc2d02433cb242 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:56:12 +0100 Subject: [PATCH 2/4] mason: release note for standalone sessions leaving the daemon's in-flight calls alone Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> (cherry picked from commit 543bc0342423ac6061ad3048cc81bee412df7dff) --- .alfonso/release-notes/next/actor-storage-propagation.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 .alfonso/release-notes/next/actor-storage-propagation.md diff --git a/.alfonso/release-notes/next/actor-storage-propagation.md b/.alfonso/release-notes/next/actor-storage-propagation.md new file mode 100644 index 0000000000..de214355e7 --- /dev/null +++ b/.alfonso/release-notes/next/actor-storage-propagation.md @@ -0,0 +1 @@ +- Starting an OpenCode or Pi session that runs 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. From 357b71a6d844ffa5147b3cfb6e03de981dd190e9 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:57:59 +0100 Subject: [PATCH 3/4] Release note wording and search-quality descriptor for actor-storage-ledger-recovery Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- .alfonso/release-notes/next/actor-storage-propagation.md | 2 +- .../land-actor-storage-ledger-recovery.json | 8 ++++++++ 2 files changed, 9 insertions(+), 1 deletion(-) create mode 100644 benchmarks/aft-search/slice-descriptors/land-actor-storage-ledger-recovery.json diff --git a/.alfonso/release-notes/next/actor-storage-propagation.md b/.alfonso/release-notes/next/actor-storage-propagation.md index de214355e7..1274469f53 100644 --- a/.alfonso/release-notes/next/actor-storage-propagation.md +++ b/.alfonso/release-notes/next/actor-storage-propagation.md @@ -1 +1 @@ -- Starting an OpenCode or Pi session that runs 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. +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. diff --git a/benchmarks/aft-search/slice-descriptors/land-actor-storage-ledger-recovery.json b/benchmarks/aft-search/slice-descriptors/land-actor-storage-ledger-recovery.json new file mode 100644 index 0000000000..ebd574711e --- /dev/null +++ b/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" + ] +} From 087e6c584c5f7f03358b9e8059aaecfdca821a81 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:11:20 +0100 Subject: [PATCH 4/4] Pass the storage root to handle_control_request in the bind tests added by #984 Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/subc/bind_capacity_tests.rs | 2 ++ crates/aft/src/subc/prepared_bind_tests.rs | 1 + 2 files changed, 3 insertions(+) diff --git a/crates/aft/src/subc/bind_capacity_tests.rs b/crates/aft/src/subc/bind_capacity_tests.rs index b16b39212b..0f7bfe15ba 100644 --- a/crates/aft/src/subc/bind_capacity_tests.rs +++ b/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(); diff --git a/crates/aft/src/subc/prepared_bind_tests.rs b/crates/aft/src/subc/prepared_bind_tests.rs index 97b390f9ad..b135fd97c5 100644 --- a/crates/aft/src/subc/prepared_bind_tests.rs +++ b/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();