Durable per-user usage rollup and GDPR/PIPL content erasure

crates/config/src/lib.rs

@@ -46,6 +46,8 @@ pub enum ConfigError {
     UnknownQuotaModel { owner: String, model: String },
     #[error("tenant `{tenant}` fallback model `{model}` is unknown or not entitled")]
     BadFallbackModel { tenant: String, model: String },
+    #[error("`{owner}` sets a negative price")]
+    NegativePrice { owner: String },
     #[error("storage.shared_cache needs storage.redis_url")]
     SharedCacheNeedsRedis,
 }
@@ -592,6 +594,35 @@ impl GatewayConfig {
         if self.storage.shared_cache && self.storage.redis_url.is_empty() {
             return Err(ConfigError::SharedCacheNeedsRedis);
         }
+        // negative prices would make cost accounting non-monotonic (the usage
+        // rollup's max-upsert relies on per-column monotone sums)
+        let neg = |i: i64, o: i64| i < 0 || o < 0;
+        for m in &self.models {
+            if neg(m.input_price_per_1k_micros, m.output_price_per_1k_micros) {
+                return Err(ConfigError::NegativePrice {
+                    owner: format!("model {}", m.name),
+                });
+            }
+        }
+        for a in &self.accounts {
+            if neg(
+                a.cost_input_price_per_1k_micros,
+                a.cost_output_price_per_1k_micros,
+            ) {
+                return Err(ConfigError::NegativePrice {
+                    owner: format!("account {}", a.name),
+                });
+            }
+        }
+        for t in &self.tenants {
+            for (model, p) in &t.model_prices {
+                if neg(p.input_price_per_1k_micros, p.output_price_per_1k_micros) {
+                    return Err(ConfigError::NegativePrice {
+                        owner: format!("tenant {} price for {model}", t.name),
+                    });
+                }
+            }
+        }
         for m in &self.models {
             if m.protocol().is_none() {
                 return Err(ConfigError::UnknownModelMapping {
@@ -1080,6 +1111,15 @@ tenants: [{name: t1}, {name: t1}]
             Err(ConfigError::DuplicateName { kind: "tenant", .. })
         ));
 
+        let neg_price = "listen: {host: h, port: 1}\nmodels: [{name: m1, protocol: openai-chat, input_price_per_1k_micros: -1}]";
+        assert!(
+            matches!(
+                GatewayConfig::from_yaml(neg_price),
+                Err(ConfigError::NegativePrice { .. })
+            ),
+            "negative prices are rejected at load"
+        );
+
         let colon = "listen: {host: h, port: 1}\ntenants: [{name: 'a:b'}]";
         assert!(
             GatewayConfig::from_yaml(colon).is_err(),

crates/handler/src/lib.rs

@@ -980,6 +980,70 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn batch_submitted_after_an_erasure_still_runs() {
+        let h = handler();
+        let off = OfflineHandler::new(h.clone());
+        let key = ak(&h).await;
+        let audit = gw_state::AdminAudit {
+            created_at_epoch_secs: 1,
+            actor: "global".into(),
+            scope: "global".into(),
+            action: "content_erase".into(),
+            target: "user-42".into(),
+            summary: String::new(),
+            source_ip: String::new(),
+        };
+        h.state()
+            .store
+            .content_erase_user(None, "user-42", audit)
+            .await
+            .unwrap();
+        // millisecond markers: a resubmit moments later must already pass
+        tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+        let job = off
+            .submit(
+                key,
+                "gpt-4o".into(),
+                vec![BatchItem {
+                    messages: vec![ChatMsg::text("user", "new content after erasure")],
+                    user: "user-42".into(),
+                }],
+            )
+            .await
+            .unwrap();
+        wait_terminal(&h, &job.id).await;
+        let done = h.state().store.batch_get(&job.id).await.unwrap().unwrap();
+        assert!(
+            done.results[0].ok,
+            "a past erasure must not fail the user's future batches: {}",
+            done.results[0].message
+        );
+    }
+
+    #[tokio::test]
+    async fn erased_batch_item_fails_instead_of_running() {
+        let h = handler();
+        let off = OfflineHandler::new(h.clone());
+        let key = ak(&h).await;
+        let job = off
+            .submit(
+                key,
+                "gpt-4o".into(),
+                vec![BatchItem {
+                    messages: Vec::new(),
+                    user: "u1".into(),
+                }],
+            )
+            .await
+            .unwrap();
+        wait_terminal(&h, &job.id).await;
+        let done = h.state().store.batch_get(&job.id).await.unwrap().unwrap();
+        assert!(!done.results[0].ok, "an erased item must not execute");
+        assert_eq!(done.results[0].message, "item content erased");
+        assert_eq!(done.results[0].total_tokens, 0, "nothing billed");
+    }
+
     #[tokio::test]
     async fn batch_attributes_each_item_to_its_user() {
         let yaml = "listen: {host: h, port: 1}\nmodels: [{name: gpt-4o, protocol: openai-chat}]\naccounts: [{name: a1, provider: openai, protocols: ['openai-chat']}]\naccess_keys: [{ak: k1, product: p, qps: 100, daily_token_quota: 100000}]";
@@ -1016,6 +1080,20 @@ mod tests {
             users.contains("alice") && users.contains("bob"),
             "each shared-key batch item bills to its own user: {users:?}"
         );
+        let results = h
+            .state()
+            .store
+            .batch_get(&job.id)
+            .await
+            .unwrap()
+            .unwrap()
+            .results;
+        let owners: std::collections::HashSet<&str> =
+            results.iter().map(|r| r.user.as_str()).collect();
+        assert!(
+            owners.contains("alice") && owners.contains("bob"),
+            "each generated result carries its owner: {owners:?}"
+        );
     }
 
     #[tokio::test]
@@ -1086,5 +1164,14 @@ mod tests {
             users.contains("alice") && users.contains("bob"),
             "distributed batch preserved per-item user attribution: {users:?}"
         );
+        assert!(
+            state
+                .store
+                .batch_load_items(&job.id)
+                .await
+                .unwrap()
+                .is_empty(),
+            "a terminal batch's input rows are pruned"
+        );
     }
 }

crates/handler/src/offline.rs

@@ -29,6 +29,15 @@ impl OfflineHandler {
     ) -> gw_models::GResult<BatchJob> {
         let store = self.online.state().store.clone();
         if store.distributed_batches() {
+            // persist the EFFECTIVE user (owner overrides the hint): execution,
+            // billing, and erasure must all key on the same identity
+            let items: Vec<BatchItem> = items
+                .into_iter()
+                .map(|mut i| {
+                    i.user = ak.attributed_user(&i.user).to_owned();
+                    i
+                })
+                .collect();
             // atomic: the job becomes claimable only once all items are saved
             store
                 .batch_enqueue(&ak.ak, &ak.tenant, &model, &items)
@@ -39,8 +48,14 @@ impl OfflineHandler {
                 .await?;
             let this = self.clone();
             let (id, model) = (job.id.clone(), model.clone());
+            // items are captured HERE: an erasure landing after this instant
+            // must stop them, so the marker comparison point is submission,
+            // not the spawned executor's first poll
+            let captured_at = gw_state::epoch_millis();
             // claim 0: non-distributed store — no fence, the heartbeat is a no-op
-            tokio::spawn(async move { this.execute(&id, &ak, &model, items, 0).await });
+            tokio::spawn(
+                async move { this.execute(&id, &ak, &model, items, 0, captured_at).await },
+            );
             Ok(job)
         }
     }
@@ -49,7 +64,15 @@ impl OfflineHandler {
     /// the terminal status, heartbeating between items. `claim` is the fence
     /// token (0 for the in-process path); once a heartbeat reports the batch
     /// reclaimed, this executor stops rather than double-running items.
-    async fn execute(&self, id: &str, ak: &AkInfo, model: &str, items: Vec<BatchItem>, claim: i64) {
+    async fn execute(
+        &self,
+        id: &str,
+        ak: &AkInfo,
+        model: &str,
+        items: Vec<BatchItem>,
+        claim: i64,
+        captured_at: i64,
+    ) {
         let store = self.online.state().store.clone();
         // the distributed claim already set status=running with the fence bump;
         // only the in-process path needs this write — unfenced on the distributed
@@ -96,7 +119,7 @@ impl OfflineHandler {
                 }
             })
         };
-        for (index, item) in items.into_iter().enumerate() {
+        for (index, mut item) in items.into_iter().enumerate() {
             if lost.load(Relaxed) {
                 break; // reclaimed by another instance; stop running new items
             }
@@ -111,6 +134,51 @@ impl OfflineHandler {
                 lost.store(true, Relaxed);
                 break;
             }
+            // re-read the stored copy just before dispatch: an erasure that
+            // landed while this batch sat queued blanks the persisted item.
+            // Fail CLOSED — a read error or vanished row can't prove the item
+            // wasn't erased, so the stale pre-load copy never dispatches
+            if store.distributed_batches() {
+                match store.batch_item_snapshot(id, index).await {
+                    Ok(Some(fresh)) => item = fresh,
+                    Ok(None) | Err(_) => {
+                        let result = BatchItemResult {
+                            index,
+                            ok: false,
+                            message: "item unavailable at dispatch".into(),
+                            total_tokens: 0,
+                            user: ak.attributed_user(&item.user).to_owned(),
+                        };
+                        if let Err(e) = store.batch_push_result(id, result).await {
+                            tracing::error!(error = %e, batch = %id, "batch result write failed");
+                        }
+                        continue;
+                    }
+                }
+            }
+            let user = ak.attributed_user(&item.user).to_owned();
+            // local backends don't persist items, so a mid-batch erasure can't
+            // blank them — the erasure marker stops the user's remaining items
+            // (fail closed on a marker read error). An erased item must not
+            // run: fail it instead of sending an erased prompt upstream
+            let erased_mid_batch = !store.distributed_batches()
+                && store
+                    .user_erased_since(&ak.tenant, &user, captured_at)
+                    .await
+                    .unwrap_or(true);
+            if erased_mid_batch || item.messages.is_empty() {
+                let result = BatchItemResult {
+                    index,
+                    ok: false,
+                    message: "item content erased".into(),
+                    total_tokens: 0,
+                    user,
+                };
+                if let Err(e) = store.batch_push_result(id, result).await {
+                    tracing::error!(error = %e, batch = %id, "batch result write failed");
+                }
+                continue;
+            }
             let request = GatewayRequest {
                 is_online: false,
                 ak: ak.ak.clone(),
@@ -132,6 +200,7 @@ impl OfflineHandler {
                 ok: false,
                 message,
                 total_tokens: 0,
+                user: user.clone(),
             };
             let result = match ran {
                 Ok(Ok(ctx)) => match ctx.outcome {
@@ -140,6 +209,7 @@ impl OfflineHandler {
                         ok: true,
                         message: out.response.message,
                         total_tokens: out.response.total_tokens,
+                        user: user.clone(),
                     },
                     None => fail("pipeline produced no outcome".into()),
                 },
@@ -199,7 +269,15 @@ impl OfflineHandler {
                             continue;
                         }
                     };
-                    self.execute(&job.id, &ak, &job.model, items, claim).await;
+                    self.execute(
+                        &job.id,
+                        &ak,
+                        &job.model,
+                        items,
+                        claim,
+                        gw_state::epoch_millis(),
+                    )
+                    .await;
                 }
                 Ok(None) => tokio::time::sleep(poll).await,
                 Err(e) => {

crates/server/src/main.rs

@@ -84,6 +84,7 @@ async fn main() -> anyhow::Result<()> {
     // daily quota reset; governance is a preserved seam, so this survives reloads
     let quota_task = gw_task::spawn_quota_reset(state.clone(), gw_task::DAILY);
     let purge_task = gw_task::spawn_content_purge(state.clone(), gw_task::PURGE_PERIOD);
+    let rollup_task = gw_task::spawn_usage_rollup(state.clone(), gw_task::ROLLUP_PERIOD);
     let distributed_batches = state.store.distributed_batches();
 
     let transport = select_transport()?;
@@ -196,6 +197,7 @@ async fn main() -> anyhow::Result<()> {
 
     quota_task.abort();
     purge_task.abort();
+    rollup_task.abort();
     tracing::info!("gw drained and exiting");
     Ok(())
 }

crates/state/src/lib.rs

@@ -863,6 +863,15 @@ pub fn epoch_secs() -> i64 {
         .unwrap_or(0)
 }
 
+/// Unix milliseconds; erasure markers use this so an erase-then-resubmit in
+/// the same second isn't misjudged as pre-erasure content.
+pub fn epoch_millis() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as i64)
+        .unwrap_or(0)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;