fix(approval): recheck reusable grants and shrink judge caches

Co-authored-by: Medulla <medulla@tinyhumans.ai>

Author: senamakel

crates/tinysecurity-approval/src/binding.rs

@@ -166,7 +166,8 @@ pub(crate) fn original(
 }
 pub(crate) fn envelope(row: &ApprovalRow2) -> Result<ApprovalRecord2> {
     let json = serde_json::to_string(row).map_err(|_| Error::Invalid)?;
-    if json.len() > 65536
+    if row.created_at_ms >= row.expires_at_ms
+        || json.len() > 65536
         || !hex(&row.binding.call_sha256, 64)
         || !hex(&row.binding.content_sha256, 64)
     {

crates/tinysecurity-approval/src/engine.rs

@@ -193,25 +193,11 @@ impl Engine {
         }
         Self::correlation(&request.audit, reg.generation)?;
         let rows = self.rows(&reg).await?;
-        // Durable identities replace retry identities once confirmed by exact
-        // load. Retain every indeterminate creation not present in this snapshot.
-        self.creations
-            .lock()
-            .map_err(|_| Error::Unavailable)?
-            .retain(|_, pending| {
-                !rows.iter().any(|row| {
-                    row.request_id == pending.request_id && row.binding == pending.binding
-                })
-            });
+        self.reconcile_creations(&rows)?;
         if let Some(row) = rows.iter().find(|r| r.binding == bound) {
             return self.resume(&reg, row, now).await;
         }
-        let reusable = reusable_grant(&rows, &bound, now);
-        if let Some(source) = reusable {
-            // A decision CAS can succeed before its required audit fails. Reuse
-            // must reconcile that exact phase before a derived permission exists.
-            self.ports.audit(source.clone()).await?;
-        }
+        let reusable = self.live_reusable(&reg, &bound, &rows, now).await?;
         let automatic = rule == tinysecurity_auto::rules::Rule::Automatic;
         let judge = if reusable.is_none() && !automatic {
             self.score(&request, &bound, external_effect).await?
@@ -226,6 +212,12 @@ impl Engine {
             return Err(Error::Denied);
         }
         let now = self.now()?;
+        if reusable
+            .as_ref()
+            .is_some_and(|source| source.expires_at_ms <= now)
+        {
+            return Err(Error::Denied);
+        }
         let approved = reusable.is_some() || automatic || qualified;
         let expiry = now.checked_add(ttl).ok_or(Error::Invalid)?;
         let row = ApprovalRow2 {
@@ -243,7 +235,9 @@ impl Engine {
             args_redacted,
             summary: binding::text(&self.redactor()?, &request.summary, RedactMode::Text)?,
             created_at_ms: now,
-            expires_at_ms: reusable.map_or(expiry, |r| expiry.min(r.expires_at_ms)),
+            expires_at_ms: reusable
+                .as_ref()
+                .map_or(expiry, |r| expiry.min(r.expires_at_ms)),
             state: if approved {
                 ApprovalLifecycle2::Approved
             } else {
@@ -276,6 +270,40 @@ impl Engine {
         };
         self.finish_begin(&reg, &row, now).await
     }
+    fn reconcile_creations(&self, rows: &[ApprovalRow2]) -> Result<()> {
+        // Durable identities replace retry identities once confirmed by exact
+        // load. Retain every indeterminate creation not present in this snapshot.
+        self.creations
+            .lock()
+            .map_err(|_| Error::Unavailable)?
+            .retain(|_, pending| {
+                !rows.iter().any(|row| {
+                    row.request_id == pending.request_id && row.binding == pending.binding
+                })
+            });
+        Ok(())
+    }
+    async fn live_reusable(
+        &self,
+        reg: &ApprovalScope2,
+        bound: &tinysecurity_bus::ApprovalBinding2,
+        rows: &[ApprovalRow2],
+        now: u64,
+    ) -> Result<Option<ApprovalRow2>> {
+        let Some(source) = reusable_grant(rows, bound, now) else {
+            return Ok(None);
+        };
+        // Reconcile the exact source audit, then reload mutable state and time.
+        // No callback lock is held, and an expired source never creates a row.
+        self.ports.audit(source.clone()).await?;
+        let current = self.rows(reg).await?;
+        let now = self.now()?;
+        let live = reusable_grant(&current, bound, now)
+            .filter(|row| row.request_id == source.request_id)
+            .cloned()
+            .ok_or(Error::Denied)?;
+        Ok(Some(live))
+    }
     async fn score(
         &self,
         request: &EvaluateApprovalRequest2,

crates/tinysecurity-approval/src/engine_tests.rs

@@ -34,6 +34,7 @@ struct Fixture {
     reentrant: Mutex<Option<std::sync::Weak<Engine>>>,
     swallow_decision_audit_failure: AtomicU64,
     audits: Mutex<Vec<ApprovalRow2>>,
+    advance_source_audit: Mutex<Option<(Arc<TestClock>, u64)>>,
     audit_receipts: Mutex<std::collections::BTreeMap<(String, u64), ApprovalRow2>>,
     barrier: Mutex<Option<Arc<tokio::sync::Barrier>>>,
 }
@@ -149,6 +150,15 @@ impl Ports for Fixture {
         })
     }
     async fn audit(&self, _r: ApprovalRow2) -> Result<()> {
+        if _r.decision_ticket_sha256.is_some()
+            && let Some((clock, now)) = self
+                .advance_source_audit
+                .lock()
+                .map_err(|_| Error::Unavailable)?
+                .take()
+        {
+            clock.wall.store(now, Ordering::SeqCst);
+        }
         self.order
             .lock()
             .map_err(|_| Error::Unavailable)?
@@ -1013,3 +1023,34 @@ async fn fresh_and_resumed_prompt_reconcile_a_lost_decision_audit_acknowledgemen
 
 #[path = "automatic_tests.rs"]
 mod automatic_tests;
+
+#[tokio::test]
+async fn source_expiring_during_audit_never_poison_durable_subject_rows() -> Result<()> {
+    for choice in [1, 2] {
+        let f = Arc::new(Fixture::default());
+        f.choice.store(choice, Ordering::SeqCst);
+        let clock = Arc::new(TestClock::default());
+        let owner = engine(f.clone(), clock.clone())?;
+        let mut input = request();
+        input.request.caller.flow = Some("reviewed-flow".into());
+        let pending = owner.begin(input.clone()).await?;
+        let source = owner.decide(decision(&pending)).await?;
+        *f.advance_source_audit
+            .lock()
+            .map_err(|_| Error::Unavailable)? = Some((clock, source.expires_at_ms));
+        input.request.caller.call_id = "expired-source-execution".into();
+        input.audit = audit(3);
+        assert!(owner.begin(input.clone()).await.is_err());
+        assert_eq!(
+            f.rows.lock().map_err(|_| Error::Unavailable)?.len(),
+            1,
+            "no invalid derived CAS"
+        );
+        input.request.caller.call_id = "future-independent-execution".into();
+        input.request.call.tool = "independent_tool".into();
+        let future = owner.begin(input).await?;
+        assert_eq!(future.state, ApprovalLifecycle2::Pending);
+        assert!(future.created_at_ms < future.expires_at_ms);
+    }
+    Ok(())
+}

crates/tinysecurity-auto/src/cache.rs

@@ -3,28 +3,36 @@ use std::collections::BTreeMap;
 #[derive(Debug, Default)]
 pub(crate) struct Cache(BTreeMap<String, (u64, [f64; 3])>);
 impl Cache {
+    pub(crate) fn resize(&mut self, bound: u16) {
+        while self.0.len() > usize::from(bound) {
+            let oldest = self
+                .0
+                .iter()
+                .min_by_key(|(_, (expiry, _))| *expiry)
+                .map(|(key, _)| key.clone());
+            if let Some(oldest) = oldest {
+                self.0.remove(&oldest);
+            }
+        }
+    }
+    #[cfg(test)]
+    pub(crate) fn len(&self) -> usize {
+        self.0.len()
+    }
     pub(crate) fn get(&mut self, key: &str, now: u64) -> Option<[f64; 3]> {
         self.0.retain(|_, (expires, _)| *expires > now);
         self.0.get(key).map(|(_, scores)| *scores)
     }
     pub(crate) fn put(&mut self, key: String, now: u64, ttl: u32, bound: u16, scores: [f64; 3]) {
+        self.resize(bound);
         if ttl == 0 || bound == 0 {
             return;
         }
         let Some(expires) = now.checked_add(u64::from(ttl)) else {
             return;
         };
         self.0.retain(|_, (expires, _)| *expires > now);
-        if self.0.len() >= usize::from(bound) {
-            let oldest = self
-                .0
-                .iter()
-                .min_by_key(|(_, (expiry, _))| *expiry)
-                .map(|(k, _)| k.clone());
-            if let Some(oldest) = oldest {
-                self.0.remove(&oldest);
-            }
-        }
+        self.resize(bound - 1);
         self.0.insert(key, (expires, scores));
     }
 }

crates/tinysecurity-auto/src/cache_tests.rs

@@ -13,3 +13,15 @@ fn cache_expires_at_exact_ttl_and_evicts_bounded_scores() {
     cache.put("a".into(), u64::MAX, 10, 1, [1.0; 3]);
     assert_eq!(cache.get("a", 0), None);
 }
+
+#[test]
+fn shrinking_capacity_trims_existing_inventory_before_insertion() {
+    let mut cache = Cache::default();
+    for key in ["a", "b", "c"] {
+        cache.put(key.into(), 0, 100, 3, [1.0; 3]);
+    }
+    cache.put("d".into(), 1, 100, 1, [0.0; 3]);
+    assert_eq!(cache.0.len(), 1);
+    cache.put("e".into(), 2, 100, 0, [0.0; 3]);
+    assert!(cache.0.is_empty());
+}

crates/tinysecurity-auto/src/judge.rs

@@ -132,6 +132,12 @@ impl Engine {
     pub fn reconfigured(&self, config: JudgeConfig) -> Result<Self> {
         let mut candidate = Self::new(config, Arc::clone(&self.clock))?;
         candidate.state = Arc::clone(&self.state);
+        candidate
+            .state
+            .lock()
+            .map_err(|_| Error::Budget)?
+            .1
+            .resize(candidate.config.cache_entries);
         Ok(candidate)
     }
     /// Score only an already verified pending approval, with fresh policy gates outside.
@@ -175,6 +181,7 @@ impl Engine {
             {
                 return Ok(Some(self.failure(JudgeOutcome2::Budget, thresholds)));
             }
+            state.1.resize(self.config.cache_entries);
             state.1.get(&key, now)
         };
         if let Some(scores) = cached {

crates/tinysecurity-auto/src/judge_tests.rs

@@ -508,3 +508,32 @@ async fn class_origin_thresholds_invalid_config_and_reconfiguration_cannot_reuse
     );
     Ok(())
 }
+
+#[tokio::test]
+async fn legitimate_reconfiguration_shrinks_shared_cache_and_zero_clears_it() -> Result<()> {
+    let mut cfg = config();
+    cfg.max_requests = 10;
+    let engine = Engine::with_client(
+        cfg.clone(),
+        fake([1.0, 1.0, 0.0]),
+        Arc::new(TestClock::default()),
+    )?;
+    for call in ["a", "b", "c"] {
+        let mut request = input();
+        request.binding.caller.call_id = call.into();
+        engine
+            .evaluate(&request, &registry().map_err(|_| Error::Invalid)?)
+            .await?;
+    }
+    assert_eq!(engine.state.lock().map_err(|_| Error::Budget)?.1.len(), 3);
+    cfg.cache_entries = 1;
+    let candidate = engine.reconfigured(cfg.clone())?;
+    assert_eq!(
+        candidate.state.lock().map_err(|_| Error::Budget)?.1.len(),
+        1
+    );
+    cfg.cache_entries = 0;
+    let disabled = candidate.reconfigured(cfg)?;
+    assert_eq!(disabled.state.lock().map_err(|_| Error::Budget)?.1.len(), 0);
+    Ok(())
+}

crates/tinysecurity-module/examples/judge_native_contract.rs

@@ -61,7 +61,7 @@ fn judge(endpoint: String) -> JudgeConfig {
         thresholds: vec![],
         timeout_ms: 1000,
         origins: vec![Origin::Operator],
-        max_requests: 2,
+        max_requests: 3,
         budget_window_ms: 60_000,
         cache_entries: 10,
         cache_ttl_ms: 1000,
@@ -120,6 +120,7 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
     verify_fallback(&proxy, &stub, &store).await?;
     let mut invalid = configuration.clone();
     invalid.judge.as_mut().ok_or("judge")?.endpoint = "http://not-loopback.invalid/jev".into();
+    invalid.judge.as_mut().ok_or("judge")?.credential = Secret("invalid-candidate-key".into());
     assert!(
         lifecycle
             .reinitialize(&loaded.name, serde_json::to_value(invalid)?)
@@ -128,10 +129,11 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
     );
     let info: PolicyInfo = proxy.call(names::methods::POLICY_INFO, ()).await?;
     assert_eq!(info.generation, 2);
+    verify_retained_scoring(&proxy, &stub).await?;
     prompt_failure.store(true, Ordering::SeqCst);
     assert!(
         proxy
-            .call::<EvaluateApprovalResponse2>(names::methods::EVALUATE_APPROVAL2, (begin(7, 8),))
+            .call::<EvaluateApprovalResponse2>(names::methods::EVALUATE_APPROVAL2, (begin(7, 10),))
             .await
             .is_err()
     );
@@ -161,6 +163,64 @@ async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
     Ok(())
 }
 
+async fn verify_retained_scoring(
+    proxy: &tinybus::Proxy,
+    stub: &stub::Stub,
+) -> std::result::Result<(), Box<dyn std::error::Error>> {
+    *stub.response.lock().map_err(|_| "stub")? = (200, stub::response([1.0, 1.0, 0.0]));
+    let retained: EvaluateApprovalResponse2 = proxy
+        .call(names::methods::EVALUATE_APPROVAL2, (begin(9, 7),))
+        .await?;
+    assert_eq!(retained.decision.verdict, Verdict::Allow);
+    assert_eq!(retained.record.attribution.as_deref(), Some("auto:jev"));
+    assert!(retained.record.decision_ticket_sha256.is_none());
+    let evidence = retained
+        .record
+        .judge
+        .as_ref()
+        .ok_or("retained scorer evidence")?;
+    assert_eq!(evidence.outcome, JudgeOutcome2::Qualified);
+    assert_eq!(evidence.scores_millionths, Some([1_000_000, 1_000_000, 0]));
+    {
+        let captured = stub.requests.lock().map_err(|_| "stub")?;
+        assert_eq!(
+            captured.len(),
+            3,
+            "fresh identity must use retained SDK endpoint and credential"
+        );
+        assert_eq!(
+            captured.last().ok_or("retained SDK request")?["model"],
+            "jev-latest"
+        );
+    }
+    let consumed: ApprovalRow2 = proxy
+        .call(
+            names::methods::CONSUME_APPROVAL2,
+            (ApprovalActionRequest2 {
+                scope_id: "a".repeat(32),
+                request_id: retained.record.request_id,
+                request: begin(9, 8).request,
+                audit: audit(8),
+            },),
+        )
+        .await?;
+    assert_eq!(consumed.state, ApprovalLifecycle2::Reserved);
+    let budget: EvaluateApprovalResponse2 = proxy
+        .call(names::methods::EVALUATE_APPROVAL2, (begin(6, 9),))
+        .await?;
+    assert_eq!(
+        budget
+            .record
+            .judge
+            .as_ref()
+            .ok_or("budget evidence")?
+            .outcome,
+        JudgeOutcome2::Budget
+    );
+    assert_eq!(stub.requests.lock().map_err(|_| "stub")?.len(), 3);
+    Ok(())
+}
+
 fn configuration(
     host: &Connection,
     root: &std::path::Path,
@@ -297,18 +357,6 @@ async fn verify_fallback(
         .await?;
     assert_eq!(human.state, ApprovalLifecycle2::Approved);
     assert!(human.decision_ticket_sha256.is_some());
-    let budget: EvaluateApprovalResponse2 = proxy
-        .call(names::methods::EVALUATE_APPROVAL2, (begin(6, 7),))
-        .await?;
-    assert_eq!(
-        budget
-            .record
-            .judge
-            .as_ref()
-            .ok_or("budget evidence")?
-            .outcome,
-        JudgeOutcome2::Budget
-    );
     Ok(())
 }