#891: route binds read config off the frame loop and acknowledge on their own lane

.alfonso/release-notes/next/a0-control-egress.md

@@ -0,0 +1 @@
+- When the disk is slow, AFT no longer holds up new sessions connecting to it while it saves earlier results, and a request sent while a session is still connecting now waits for the connection instead of failing.

.alfonso/release-notes/next/a0-prepared-config.md

@@ -0,0 +1 @@
+- Bind configuration, source tracking and legacy-location notices now use one bounded local snapshot for the bind's selected harness, without rereading config during configure retries.

.alfonso/release-notes/next/subc-exit-stderr-flush.md

@@ -0,0 +1 @@
+- Supervised shutdowns now give final stderr messages a short, bounded flush, so the exit reason is not left queued when the process stops.

benchmarks/aft-search/slice-descriptors/land-frame-loop-a0.json

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

crates/aft/src/commands/configure.rs

@@ -3561,6 +3561,7 @@ fn handle_configure_prepared(
     // refused connect that left it in place.
     let active_before = prepared.is_none().then(|| ctx.config_unpinned());
     if prepared.is_none() {
+        ctx.config_live().clear_applied_bind_snapshot();
         ctx.note_connect_started();
     }
     let _sources_scope = crate::config_live::configure_sources_scope();
@@ -3687,9 +3688,26 @@ fn handle_configure_inner(
     // already holds this request's resolved tiers, so nothing is re-resolved
     // and the recorded config sources are left as they are.
     let recovery = prepared.is_some();
+    let bind_config = if recovery {
+        None
+    } else {
+        crate::subc_config::current_prepared_bind_config()
+    };
+    if bind_config
+        .as_ref()
+        .is_some_and(|snapshot| !snapshot.matches(&root_path, &harness))
+    {
+        return Response::error(
+            &req.id,
+            "config_unknown",
+            "prepared local configuration does not match this bind's root and harness",
+        );
+    }
     ctx.begin_configure_ack_step("tier_file_reads");
     let (tiers, user_config_migration) = if recovery {
         (Vec::new(), None)
+    } else if let Some(snapshot) = bind_config.as_ref() {
+        (snapshot.tiers().to_vec(), snapshot.migration())
     } else {
         match resolve_config_tiers_for_configure(params, &root_path, Some(&harness)) {
             Ok(read) => read,
@@ -3721,14 +3739,17 @@ fn handle_configure_inner(
     }
     ctx.begin_configure_ack_step("source_tracking");
     if !recovery {
-        crate::config_live::stage_configure_sources(
+        let sources = if let Some(snapshot) = bind_config.as_ref() {
+            snapshot.sources(&next_config)
+        } else {
             crate::config_live::ConfigSources::from_configure(
                 parse_cortexkit_user_config_path(params).ok().flatten(),
                 &root_path,
                 &tiers,
                 &next_config,
-            ),
-        );
+            )
+        };
+        crate::config_live::stage_configure_sources(sources);
     }
     ctx.begin_configure_ack_step("validate_params");
     let mut configure_warnings = config_diagnostics
@@ -3759,22 +3780,27 @@ fn handle_configure_inner(
             "message": migration.notice(),
         }));
     }
-    if let Some(project) = find_config_tier(&tiers, "project") {
-        if let Some(notice) = crate::subc_config::legacy_project_config_location_notice(
-            &root_path,
-            Path::new(&project.source),
-        )
-        .filter(|_| {
-            crate::subc_config::project_config_read_path(&root_path, Some(&harness))
-                == Path::new(&project.source)
-        }) {
-            configure_warnings.push(json!({
-                "code": "legacy_project_config_location",
-                "tier": "project",
-                "hint": notice,
-                "message": notice,
-            }));
-        }
+    let legacy_notice = if let Some(snapshot) = bind_config.as_ref() {
+        snapshot.legacy_notice()
+    } else {
+        find_config_tier(&tiers, "project").and_then(|project| {
+            crate::subc_config::legacy_project_config_location_notice(
+                &root_path,
+                Path::new(&project.source),
+            )
+            .filter(|_| {
+                crate::subc_config::project_config_read_path(&root_path, Some(&harness))
+                    == Path::new(&project.source)
+            })
+        })
+    };
+    if let Some(notice) = legacy_notice {
+        configure_warnings.push(json!({
+            "code": "legacy_project_config_location",
+            "tier": "project",
+            "hint": notice,
+            "message": notice,
+        }));
     }
 
     // NO configure-time SSRF guard on semantic.base_url — deliberate (config
@@ -16619,6 +16645,86 @@ mod tests {
         assert_eq!(fs::read_to_string(&project_path).unwrap(), project_text);
     }
 
+    #[test]
+    fn prepared_bind_branches_do_not_reselect_config_files() {
+        let source = include_str!("configure.rs");
+        for marker in [
+            "else if let Some(snapshot) = bind_config.as_ref() {",
+            "let sources = if let Some(snapshot) = bind_config.as_ref() {",
+            "let legacy_notice = if let Some(snapshot) = bind_config.as_ref() {",
+        ] {
+            let body = source
+                .split_once(marker)
+                .expect("prepared branch exists")
+                .1
+                .split_once("} else")
+                .unwrap()
+                .0;
+            assert!(body.contains("snapshot."));
+            for forbidden in [
+                "resolve_config_tiers_for_configure",
+                "from_configure",
+                "project_config_read_path",
+                ".exists(",
+                "std::fs::",
+            ] {
+                assert!(
+                    !body.contains(forbidden),
+                    "prepared configure re-observes files: {forbidden}"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn prepared_bind_config_keeps_content_sources_and_notice_across_attempts() {
+        let temp = tempfile::tempdir().unwrap();
+        let ctx = test_context();
+        let user = temp.path().join("user.jsonc");
+        let project = temp.path().join(".pi/aft.jsonc");
+        let canonical = temp.path().join(".cortexkit/aft.jsonc");
+        write_config(&user, r#"{"url_fetch_allow_private":true}"#);
+        write_config(&project, r#"{"callgraph_chunk_size":7}"#);
+        write_config(
+            &temp.path().join(".opencode/aft.jsonc"),
+            r#"{"callgraph_chunk_size":8}"#,
+        );
+        let snapshot = crate::subc_config::PreparedBindConfig::read(
+            Some(&user),
+            temp.path(),
+            &crate::harness::Harness::Pi,
+        )
+        .unwrap();
+        write_config(&user, r#"{"url_fetch_allow_private":false}"#);
+        write_config(&canonical, r#"{"callgraph_chunk_size":9}"#);
+        fs::remove_file(&project).unwrap();
+        let request = configure_request_with_params(json!({
+            "project_root": temp.path(), "harness":"pi", "cortexkit_user_config_path":user,
+        }));
+        for _ in 0..2 {
+            let response =
+                crate::subc_config::with_prepared_bind_config(Arc::clone(&snapshot), || {
+                    handle_configure_for_test(&request, &ctx)
+                });
+            assert!(response.success, "{:?}", response.data);
+            assert_eq!(ctx.config().callgraph_chunk_size, 7);
+            assert!(ctx.config().url_fetch_allow_private);
+            let sources = ctx.config_live().sources().expect("applied sources");
+            assert_eq!(sources.project.path.as_deref(), Some(project.as_path()));
+            assert!(sources.project.from_file);
+            assert_eq!(
+                sources.project.text.as_deref(),
+                Some(r#"{"callgraph_chunk_size":7}"#)
+            );
+            assert!(response.data["warnings"]
+                .as_array()
+                .unwrap()
+                .iter()
+                .any(|warning| warning["code"] == "legacy_project_config_location"));
+        }
+        assert!(crate::subc_config::current_prepared_bind_config().is_none());
+    }
+
     #[test]
     fn configure_legacy_project_file_is_read_in_place_and_reports_its_location() {
         for harness in ["opencode", "pi"] {
@@ -17345,6 +17451,57 @@ mod tests {
         );
     }
 
+    #[test]
+    fn prepared_bind_identity_replay_retains_the_original_observation() {
+        let _git_env = crate::test_env::hermetic_git_env_guard();
+        let (_temp, root, storage, lock) = busy_bind_fixture();
+        let ctx = test_context();
+        let user = root.join("user.jsonc");
+        let legacy = root.join(".pi/aft.jsonc");
+        write_config(
+            &user,
+            r#"{"indexes":{"trigram":true,"semantic":false,"callgraph":false},"views":{"enabled":false},"url_fetch_allow_private":true}"#,
+        );
+        write_config(&legacy, r#"{"callgraph_chunk_size":7}"#);
+        let snapshot = crate::subc_config::PreparedBindConfig::read(
+            Some(&user),
+            &root,
+            &crate::harness::Harness::Pi,
+        )
+        .unwrap();
+        let request = configure_request_with_params(
+            json!({"project_root":root,"storage_dir":storage,"harness":"pi","cortexkit_user_config_path":user}),
+        );
+        crate::provisional_identity_retry::with_timing_for_test(
+            Duration::ZERO,
+            Duration::from_secs(600),
+            || {
+                let response = crate::subc_config::with_prepared_bind_config(snapshot, || {
+                    super::handle_configure(&request, &ctx)
+                });
+                assert!(response.success, "{}", response.data);
+                super::drain_deferred_configure_maintenance(&ctx);
+                assert!(ctx.provisional_identity_retry().is_armed());
+                write_config(&user, r#"{"url_fetch_allow_private":false}"#);
+                write_config(
+                    &root.join(".cortexkit/aft.jsonc"),
+                    r#"{"callgraph_chunk_size":9}"#,
+                );
+                fs::remove_file(&legacy).unwrap();
+                lock.execute_batch("ROLLBACK").unwrap();
+                drop(lock);
+                super::drain_deferred_configure_maintenance(&ctx);
+                assert!(!ctx.provisional_identity_retry().is_armed());
+                assert_eq!(ctx.config().callgraph_chunk_size, 7);
+                assert!(ctx.config().url_fetch_allow_private);
+                assert_eq!(
+                    ctx.config_live().sources().unwrap().project.path.as_deref(),
+                    Some(legacy.as_path())
+                );
+            },
+        );
+    }
+
     /// A replay can resolve and memoize the settled key and still be refused
     /// before its commit (for example by a storage error at its retention or
     /// artifact-owner step). The root then stays provisional with the settled

crates/aft/src/config_fix.rs

@@ -600,7 +600,40 @@ fn auto_migrate_user_config_with_writer(
     path: &Path,
     write: impl FnOnce(&Path, &str) -> std::io::Result<Option<std::io::Error>>,
 ) -> Option<UserConfigMigration> {
-    let text = std::fs::read_to_string(path).ok()?;
+    auto_migrate_user_config_with_reader(path, write, |path| std::fs::read_to_string(path))
+}
+
+pub(crate) fn auto_migrate_user_config_bounded(
+    path: &Path,
+    limit: usize,
+) -> std::io::Result<Option<UserConfigMigration>> {
+    let mut failure = None;
+    let result = auto_migrate_user_config_with_reader(
+        path,
+        jsonc_edit::write_atomic_with_sync_warning,
+        |path| {
+            let read = (|| {
+                let file = std::fs::File::open(path)?;
+                crate::subc_config::read_bounded_config(file, limit)
+            })();
+            if let Err(error) = &read {
+                failure = Some(std::io::Error::new(error.kind(), error.to_string()));
+            }
+            read
+        },
+    );
+    match failure {
+        Some(error) => Err(error),
+        None => Ok(result),
+    }
+}
+
+fn auto_migrate_user_config_with_reader(
+    path: &Path,
+    write: impl FnOnce(&Path, &str) -> std::io::Result<Option<std::io::Error>>,
+    mut read: impl FnMut(&Path) -> std::io::Result<String>,
+) -> Option<UserConfigMigration> {
+    let text = read(path).ok()?;
     let translation = retired_key_translation(&text)?;
     let display = path.display().to_string();
     let not_migrated = |reason: String, translation: &feature_config::DocumentTranslation| {
@@ -632,7 +665,7 @@ fn auto_migrate_user_config_with_writer(
     };
 
     // Re-read under the lock: another process may have finished meanwhile.
-    let text = match std::fs::read_to_string(&target) {
+    let text = match read(&target) {
         Ok(text) => text,
         Err(error) => return not_migrated(format!("could not read it: {error}"), &translation),
     };

crates/aft/src/config_live.rs

@@ -113,6 +113,7 @@ pub struct TierSource {
 /// the files changed and which changed keys are still waiting for a connect.
 #[derive(Debug, Clone)]
 pub struct ConfigSources {
+    pub(crate) prepared_bind_snapshot_id: Option<u64>,
     pub user: TierSource,
     pub project: TierSource,
     /// The resolver output of the last configure, before configure's own
@@ -133,10 +134,21 @@ impl ConfigSources {
         tiers: &[ConfigTier],
         connected: &Config,
     ) -> Self {
-        let owns_user_watch = user_config_path.is_some() && process_user_config_path().is_none();
-        let user_path = user_config_path.or_else(process_user_config_path);
         let project_path =
             crate::subc_config::project_config_read_path(project_root, connected.harness.as_ref());
+        Self::from_observed_paths(user_config_path, project_path, tiers, connected)
+    }
+
+    /// Record the source paths selected by a trusted local observation without
+    /// probing them again. Reload must watch the same files that supplied tiers.
+    pub(crate) fn from_observed_paths(
+        user_config_path: Option<PathBuf>,
+        project_path: PathBuf,
+        tiers: &[ConfigTier],
+        connected: &Config,
+    ) -> Self {
+        let owns_user_watch = user_config_path.is_some() && process_user_config_path().is_none();
+        let user_path = user_config_path.or_else(process_user_config_path);
         let source_for = |tier: &'static str, path: Option<PathBuf>| {
             let used = tiers.iter().find(|candidate| candidate.tier == tier);
             let from_file = match (used, path.as_ref()) {
@@ -152,6 +164,7 @@ impl ConfigSources {
             }
         };
         Self {
+            prepared_bind_snapshot_id: None,
             user: source_for("user", user_path),
             project: source_for("project", Some(project_path)),
             connected: connected.clone(),
@@ -244,7 +257,11 @@ pub fn finish_configure(ctx: &AppContext, success: bool) {
     ctx.begin_configure_ack_step("config_sources_commit");
     {
         let _reload = ctx.config_live().reload_lock.lock();
+        let snapshot_id = sources.prepared_bind_snapshot_id.unwrap_or(0);
         *ctx.config_live().sources.lock() = Some(sources);
+        ctx.config_live()
+            .applied_bind_snapshot
+            .store(snapshot_id, Ordering::Release);
         // A connect resolves the files afresh; nothing it applied is held.
         ctx.config_live()
             .project_hold_active
@@ -257,6 +274,7 @@ pub fn finish_configure(ctx: &AppContext, success: bool) {
 /// Per-root live reload state, owned by the root's `AppContext`.
 #[derive(Default)]
 pub struct ConfigLiveState {
+    applied_bind_snapshot: AtomicU64,
     signal: Arc<ConfigReloadSignal>,
     sources: parking_lot::Mutex<Option<ConfigSources>>,
     watches: parking_lot::Mutex<ContextWatches>,
@@ -284,6 +302,16 @@ struct ContextWatches {
 }
 
 impl ConfigLiveState {
+    pub(crate) fn applied_bind_snapshot(&self) -> Option<u64> {
+        match self.applied_bind_snapshot.load(Ordering::Acquire) {
+            0 => None,
+            id => Some(id),
+        }
+    }
+
+    pub(crate) fn clear_applied_bind_snapshot(&self) {
+        self.applied_bind_snapshot.store(0, Ordering::Release);
+    }
     pub fn signal(&self) -> Arc<ConfigReloadSignal> {
         Arc::clone(&self.signal)
     }
@@ -647,7 +675,9 @@ fn reload_config_inner(ctx: &AppContext) -> ReloadOutcome {
 
     {
         let mut record = state.sources.lock();
+        state.clear_applied_bind_snapshot();
         if let Some(record) = record.as_mut() {
+            record.prepared_bind_snapshot_id = None;
             record.user = user;
             // While a project loosening is held, keep the project text the
             // held values came from as the reference, so later reloads keep