Views step D: verified bind, view token and steady-state cost token (4v-4, 4v-5)

benchmarks/aft-search/slice-descriptors/land-views-d-verified-bind.json

@@ -0,0 +1,13 @@
+{
+  "slice_class": "engine_unwired",
+  "targeted_mechanism": "none",
+  "kind": "harness",
+  "fixtures": [
+    "crates/aft/tests/integration/tool_call_parity_test.rs",
+    "crates/aft/src/lib.rs"
+  ],
+  "slices": [
+    "views-d-verified-bind"
+  ],
+  "note": "The only ranking fence hit is crates/aft/src/lib.rs, which exports the git index reader module brought in from 4v-1. Views stay default-off and unrouted in this step; no search, routing or ranking code changes, so ranked rows must remain byte-identical."
+}

crates/aft/src/blob_store/v2.rs

@@ -363,6 +363,7 @@ pub type StoreResult<T> = Result<T, StoreError>;
 pub struct StoreWriteAccess {
     storage: PathBuf,
     family: String,
+    legacy_allowed: bool,
 }
 
 impl StoreWriteAccess {
@@ -371,12 +372,20 @@ impl StoreWriteAccess {
         Self {
             storage: storage.to_path_buf(),
             family: family.to_owned(),
+            legacy_allowed: true,
         }
     }
 
     pub fn storage(&self) -> &Path {
         &self.storage
     }
+    pub(crate) fn v3_only(storage: &Path, family: &str) -> Self {
+        Self {
+            storage: storage.to_path_buf(),
+            family: family.to_owned(),
+            legacy_allowed: false,
+        }
+    }
 
     pub fn family(&self) -> &str {
         &self.family
@@ -442,6 +451,11 @@ pub struct FamilyStoreReader {
 impl FamilyStore {
     /// Opens (creating when absent) `<storage>/blobs/v2/<family>/<plane>.sqlite`.
     pub fn open(access: &StoreWriteAccess, plane: FamilyPlane) -> StoreResult<Self> {
+        if !access.legacy_allowed {
+            return Err(StoreError::InvalidFamily(
+                "v3-only registration cannot open a v2 writer".into(),
+            ));
+        }
         let path = plane_path(&access.storage, &access.family, plane)?;
         if let Some(parent) = path.parent() {
             crate::private_storage::open_dir(&access.storage, parent)?;

crates/aft/src/blob_store/v3.rs

@@ -52,6 +52,7 @@ const META: &str = "CREATE TABLE format_meta (singleton INTEGER PRIMARY KEY CHEC
 /// [`RegistryWriter::upgrade_to_base_refs`], only by a writer that raised the
 /// registry reader floor to 2 first.
 pub const REGISTRY_BASE_REFS_VERSION: u32 = 2;
+pub const REGISTRY_VERIFICATION_VERSION: u32 = 3;
 macro_rules! registry_v1_schema {
     () => {
         "
@@ -97,6 +98,21 @@ CREATE TABLE sweep_state (singleton INTEGER PRIMARY KEY CHECK(singleton=1),
 const REGISTRY_SCHEMA: &str = registry_v1_schema!();
 const BASE_REFS_SCHEMA: &str = base_refs_schema!();
 const REGISTRY_SCHEMA_V2: &str = concat!(registry_v1_schema!(), base_refs_schema!());
+macro_rules! view_tokens_schema {
+    () => {
+        "CREATE TABLE view_tokens (nonce TEXT NOT NULL PRIMARY KEY, root BLOB NOT NULL,
+ expected_tree TEXT NOT NULL, minted_ms INTEGER NOT NULL, expires_ms INTEGER NOT NULL,
+ bound_root BLOB, bound_gitdir BLOB, consumed_ms INTEGER,
+ CHECK((bound_root IS NULL AND bound_gitdir IS NULL AND consumed_ms IS NULL) OR
+       (bound_root IS NOT NULL AND bound_gitdir IS NOT NULL AND consumed_ms IS NOT NULL)));"
+    };
+}
+const VIEW_TOKENS_SCHEMA: &str = view_tokens_schema!();
+const REGISTRY_SCHEMA_V3: &str = concat!(
+    registry_v1_schema!(),
+    base_refs_schema!(),
+    view_tokens_schema!()
+);
 const PLANE_SCHEMA: &str = "
 CREATE TABLE store_meta (singleton INTEGER PRIMARY KEY CHECK(singleton=1), plane TEXT NOT NULL, gc_epoch INTEGER NOT NULL);
 CREATE TABLE blob_payloads (full_key BLOB NOT NULL PRIMARY KEY CHECK(length(full_key)=32),
@@ -487,6 +503,7 @@ fn validate(c: &Connection, path: &Path, kind: Kind) -> Result<()> {
         // A registry of the base-reference protocol carries its own exact
         // text; any other version/schema pair falls through to the drift check.
         (Kind::Registry, 2, 2) if layout == 0 => Some(REGISTRY_SCHEMA_V2),
+        (Kind::Registry, 3, 3) if layout == 0 => Some(REGISTRY_SCHEMA_V3),
         _ => kind.schema_for(layout),
     };
     let Some(compiled_schema) = compiled_schema else {
@@ -508,6 +525,11 @@ fn validate(c: &Connection, path: &Path, kind: Kind) -> Result<()> {
                         u64::from(REGISTRY_BASE_REFS_VERSION),
                         REGISTRY_BASE_REFS_VERSION,
                     )
+                || (version, schema)
+                    == (
+                        u64::from(REGISTRY_VERIFICATION_VERSION),
+                        REGISTRY_VERIFICATION_VERSION,
+                    )
         }
         Kind::Plane(_) => version == u64::from(FORMAT_VERSION) && schema == SCHEMA_VERSION,
     };
@@ -906,6 +928,8 @@ impl<'a> RegistryWriter<'a> {
             )?;
             if (version, schema) == (REGISTRY_BASE_REFS_VERSION, REGISTRY_BASE_REFS_VERSION)
                 && text == format!("{META}{REGISTRY_SCHEMA_V2}")
+                || (version, schema) == (REGISTRY_VERIFICATION_VERSION, REGISTRY_VERIFICATION_VERSION)
+                    && text == format!("{META}{REGISTRY_SCHEMA_V3}")
             {
                 return Ok(());
             }
@@ -931,6 +955,19 @@ impl<'a> RegistryWriter<'a> {
             Ok(())
         })
     }
+    /// Adds only the registered v3 token protocol; plane stores and legacy
+    /// registries retain their own versions and schema.
+    pub fn upgrade_to_verification(&mut self) -> Result<()> {
+        self.upgrade_to_base_refs()?;
+        self.with_barrier(|tx| {
+            let version: u32 = tx.query_row("SELECT format_version FROM format_meta WHERE singleton=1", [], |r| r.get(0))?;
+            if version == REGISTRY_VERIFICATION_VERSION { return Ok(()); }
+            tx.execute_batch(VIEW_TOKENS_SCHEMA)?;
+            tx.execute("UPDATE format_meta SET format_version=?1,schema_version=?1,schema_text=?2 WHERE singleton=1",
+                params![REGISTRY_VERIFICATION_VERSION, format!("{META}{REGISTRY_SCHEMA_V3}")])?;
+            Ok(())
+        })
+    }
     fn connection(&self) -> Result<connections::Guard<'_>> {
         self.database.connection()
     }
@@ -1437,6 +1474,25 @@ pub fn open_base_refs_registry<'a>(
     )
     .map_err(|e| Error::Io(std::io::Error::other(e)))?
 }
+/// Raises the verification floor before any token or stat-evidence bytes exist.
+pub fn open_verification_registry<'a>(
+    storage: &Path,
+    family: &str,
+    lock: &'a ReservationLock,
+) -> Result<RegistryWriter<'a>> {
+    crate::reader_floor::publish_v3_with_floor(
+        storage,
+        REGISTRY_VERIFICATION_VERSION,
+        FORMAT_VERSION,
+        || {
+            let mut registry = open_base_refs_registry(storage, family, lock, None)?;
+            registry.upgrade_to_verification()?;
+            Ok(registry)
+        },
+    )
+    .map_err(|e| Error::Io(std::io::Error::other(e)))?
+}
+
 fn publish_under_lock<T>(
     storage: &Path,
     family: &str,

crates/aft/src/blob_store/v3/tests.rs

@@ -859,15 +859,15 @@ fn newer_format_keeps_artifact_refusal_and_malformed_file_reports_found_zero() {
     let root = tempdir().unwrap();
     seed(root.path(), A);
     let path = db(root.path(), A);
+    let future = REGISTRY_VERIFICATION_VERSION + 1;
     Connection::open(&path)
         .unwrap()
-        // Registry 2 is the base-reference protocol this build reads; the
-        // first newer version is 3.
-        .execute("UPDATE format_meta SET format_version=3", [])
+        // Keep testing a future registry, not drift within a supported version.
+        .execute("UPDATE format_meta SET format_version=?1", [future])
         .unwrap();
     match LedgerReader::open_existing(root.path(), A).err().unwrap() {
         Error::Incompatible(error) => {
-            assert_eq!(error.found, 3);
+            assert_eq!(error.found, u64::from(future));
             assert_eq!(error.source, persisted_format::RefusalSource::Artifact);
         }
         error => panic!("{error}"),

crates/aft/src/commands/mod.rs

@@ -79,6 +79,7 @@ pub mod undo;
 pub mod untrust_filter_project;
 pub(crate) mod url_output;
 mod url_resolution;
+pub mod view_token;
 pub mod write;
 pub mod writes_census;
 pub mod zoom;

crates/aft/src/commands/view_token.rs

@@ -0,0 +1,70 @@
+//! Internal provisioning request, deliberately not an agent tool. Tokens are
+//! advisory and are never a substitute for a driver's checkout verification.
+
+use crate::{
+    context::AppContext,
+    protocol::{RawRequest, Response},
+};
+
+pub fn handle_view_token(request: &RawRequest, context: &AppContext) -> Response {
+    if !context.config().views.enabled {
+        return Response::error(
+            &request.id,
+            "VIEW_TOKEN",
+            "view tokens require views to be enabled",
+        );
+    }
+    let Some(root) = request.params.get("root").and_then(|v| v.as_str()) else {
+        return Response::error(&request.id, "VIEW_TOKEN", "root is required");
+    };
+    let Some(tree) = request.params.get("expected_tree").and_then(|v| v.as_str()) else {
+        return Response::error(&request.id, "VIEW_TOKEN", "expected_tree is required");
+    };
+    let Some(project) = context.canonical_cache_root_opt() else {
+        return Response::error(
+            &request.id,
+            "VIEW_TOKEN",
+            "view token requires a configured repository",
+        );
+    };
+    let storage = context.storage_paths().artifacts;
+    let family = match crate::search_index::artifact_cache_key_read_only(&project, &storage) {
+        Ok(family) => family,
+        Err(error) => return Response::error(&request.id, "VIEW_TOKEN", &error.to_string()),
+    };
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as u64;
+    match crate::views::registry::mint_view_token(
+        &storage,
+        &family,
+        std::path::Path::new(root),
+        tree,
+        now,
+    ) {
+        Ok(token) => Response::success(
+            &request.id,
+            serde_json::json!({ "token": token.token, "expires_ms": token.expires_ms, "advisory": true }),
+        ),
+        Err(error) => Response::error(&request.id, "VIEW_TOKEN", &error.to_string()),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn view_token_request_with_views_off_creates_no_v3_files() {
+        let storage = tempfile::tempdir().unwrap();
+        let context = crate::context::AppContext::new(
+            Box::new(crate::parser::TreeSitterProvider::new()),
+            crate::config::Config {
+                storage_dir: Some(storage.path().to_path_buf()),
+                ..Default::default()
+            },
+        );
+        let request = serde_json::from_value(serde_json::json!({ "id": "token", "command": "view_token", "root": "/future", "expected_tree": "a".repeat(40) })).unwrap();
+        assert!(!super::handle_view_token(&request, &context).success);
+        assert!(!storage.path().join("blobs/v3").exists());
+    }
+}

crates/aft/src/context.rs

@@ -1720,6 +1720,9 @@ pub(crate) struct WatcherCounters {
     pub(crate) bash_flush: Arc<crate::watcher_backend::flush::FlushControl>,
     pub(crate) bash_changes: std::sync::OnceLock<Arc<crate::bash_changed_files::Journal>>,
     overflow_ignore_evidence: AtomicU8,
+    /// Legacy rescan work coalesces repeated drops, but a newly installed views
+    /// snapshot must still learn that delivery was lost during that work.
+    checkout_watcher_gap_pending: AtomicBool,
     fsevents_stream_creations_total: AtomicU64,
     raw_events_total: AtomicU64,
     raw_events_since_last_rescan: AtomicU64,
@@ -1891,6 +1894,8 @@ impl WatcherCounters {
                 .is_ok()
             {
                 self.overflows_during_rescan.fetch_add(1, Ordering::Relaxed);
+                self.checkout_watcher_gap_pending
+                    .store(true, Ordering::Release);
                 return true;
             }
         }
@@ -10515,6 +10520,31 @@ impl AppContext {
         self.checkout_semantic.wake();
     }
 
+    pub(crate) fn record_checkout_watcher_overflow(&self) {
+        self.reset_view_publication_retry();
+        if let Some(driver) = self.checkout_driver.get() {
+            driver.record_watcher_overflow();
+        }
+        self.checkout_semantic.wake();
+    }
+
+    pub(crate) fn record_checkout_watcher_cookie(&self, path: &Path) {
+        self.record_checkout_pending_watcher_gap();
+        if let Some(driver) = self.checkout_driver.get() {
+            driver.record_watcher_cookie(path);
+        }
+    }
+
+    pub(crate) fn record_checkout_pending_watcher_gap(&self) {
+        if self
+            .watcher_counters()
+            .checkout_watcher_gap_pending
+            .swap(false, Ordering::AcqRel)
+        {
+            self.record_checkout_watcher_overflow();
+        }
+    }
+
     pub(crate) fn view_publication_retry(
         &self,
     ) -> &parking_lot::Mutex<crate::executor::view_publication::PublicationRetry> {