1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
//! Decision Cache Repository (#1648)
//!
//! Storage half of the L1 exact-decision-reuse ring. Keys come from
//! [`crate::decisions::normalize`] and are the row's whole identity:
//! (tier, policy_version, normalizer_version, canonical input) under one
//! sha256. A miss is just "ask the model" — correctness lives at the write
//! gate: a margin below the tier's floor is refused HERE, so a borderline
//! decision can never be frozen no matter what a caller does (PR2's
//! `decide_cached` relies on this being unskippable).
//!
//! Removal path (release-day evaluation, #1648): `[decisions]` off in
//! config stops every read/write; `DROP TABLE IF EXISTS decision_cache`
//! finishes it. Nothing else references this table.
use crate::db::Pool;
use crate::db::database::interact_err;
use anyhow::{Context, Result};
use rusqlite::params;
/// One cached decision, as stored.
#[derive(Debug, Clone)]
pub struct DecisionCacheRow {
pub key: String,
pub tier_id: String,
pub result_json: String,
pub p: Option<f64>,
pub margin: Option<f64>,
pub policy_version: String,
pub normalizer_version: String,
pub created_at: i64,
pub hits: i64,
pub last_used_at: Option<i64>,
}
/// A freshly derived decision ready to be written.
#[derive(Debug, Clone)]
pub struct DecisionPut {
pub key: String,
pub tier_id: String,
/// The decision itself, verbatim JSON from the model's tiny schema.
pub result_json: String,
/// Model-reported confidence, when it reported one.
pub p: Option<f64>,
/// Distance to the runner-up. `None` means not measurable (a
/// deterministic-shaped decision); measurable-but-borderline is
/// refused by the floor gate in [`DecisionCacheRepository::put`].
pub margin: Option<f64>,
pub policy_version: String,
pub normalizer_version: String,
}
#[derive(Clone)]
pub struct DecisionCacheRepository {
pool: Pool,
}
impl DecisionCacheRepository {
pub fn new(pool: Pool) -> Self {
Self { pool }
}
/// Exact-key lookup. `None` = miss = ask the model; there is no fuzzy
/// path here (similarity reuse is a separate, opt-in later thing and
/// never goes through this method).
pub async fn get(&self, key: &str) -> Result<Option<DecisionCacheRow>> {
let key = key.to_string();
let row = self
.pool
.get()
.await
.context("Failed to get connection")?
.interact(move |conn| {
let mut stmt = conn.prepare(
"SELECT key, tier_id, result_json, p, margin, policy_version, \
normalizer_version, created_at, hits, last_used_at \
FROM decision_cache WHERE key = ?1",
)?;
let mapped = stmt
.query_map(params![key], |row| {
Ok(DecisionCacheRow {
key: row.get(0)?,
tier_id: row.get(1)?,
result_json: row.get(2)?,
p: row.get(3)?,
margin: row.get(4)?,
policy_version: row.get(5)?,
normalizer_version: row.get(6)?,
created_at: row.get(7)?,
hits: row.get(8)?,
last_used_at: row.get(9)?,
})
})?
.next()
.transpose()?;
Ok::<_, rusqlite::Error>(mapped)
})
.await
.map_err(interact_err)?
.context("Failed to read decision_cache row")?;
Ok(row)
}
/// Write a freshly derived decision.
///
/// `margin_floor` is the tier's gate: a measurable margin below it is
/// refused (`Ok(false)`) and the caller must treat the decision as
/// live-only, re-asking next time. Refusal is the only loss this table
/// can hand out, and it is always the cheap one.
///
/// Re-put of an existing key refreshes the stored result in place: one
/// row per key, hit counter and last_used preserved.
pub async fn put(&self, input: DecisionPut, margin_floor: f64) -> Result<bool> {
if let Some(m) = input.margin
&& m < margin_floor
{
tracing::debug!(
"decision_cache: refused write for tier {} (margin {m} < floor {margin_floor})",
input.tier_id
);
return Ok(false);
}
self.pool
.get()
.await
.context("Failed to get connection")?
.interact(move |conn| {
conn.execute(
"INSERT INTO decision_cache \
(key, tier_id, result_json, p, margin, policy_version, normalizer_version, created_at, hits) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, strftime('%s','now'), 0) \
ON CONFLICT(key) DO UPDATE SET \
result_json = excluded.result_json, \
p = excluded.p, \
margin = excluded.margin, \
created_at = excluded.created_at",
params![
input.key,
input.tier_id,
input.result_json,
input.p,
input.margin,
input.policy_version,
input.normalizer_version,
],
)
})
.await
.map_err(interact_err)?
.context("Failed to write decision_cache row")?;
Ok(true)
}
/// Record one use: the /usage accounting counter and the TTL signal.
/// A missing key is not an error (the row may have been swept between
/// get and bump); a missed bump under-counts, never corrupts.
pub async fn bump_hits(&self, key: &str) -> Result<()> {
let key = key.to_string();
self.pool
.get()
.await
.context("Failed to get connection")?
.interact(move |conn| {
conn.execute(
"UPDATE decision_cache \
SET hits = hits + 1, last_used_at = strftime('%s','now') \
WHERE key = ?1",
params![key],
)
})
.await
.map_err(interact_err)?
.context("Failed to bump decision_cache hits")?;
Ok(())
}
/// Delete one (tier, policy_version) slice: the stale-policy sweeper and
/// the kill-switch cleanup. Returns how many rows went.
pub async fn delete_by_tier_version(
&self,
tier_id: &str,
policy_version: &str,
) -> Result<usize> {
let (tier_id, policy_version) = (tier_id.to_string(), policy_version.to_string());
let n = self
.pool
.get()
.await
.context("Failed to get connection")?
.interact(move |conn| {
conn.execute(
"DELETE FROM decision_cache WHERE tier_id = ?1 AND policy_version = ?2",
params![tier_id, policy_version],
)
})
.await
.map_err(interact_err)?
.context("Failed to delete decision_cache slice")?;
Ok(n)
}
/// Row count per tier, for the /usage decisions block (#1648 PR3):
/// the cached-answer distribution alongside the call counters.
pub async fn count_by_tier(&self) -> Result<Vec<(String, i64)>> {
let rows = self
.pool
.get()
.await
.context("Failed to get connection")?
.interact(|conn| {
let mut stmt =
conn.prepare("SELECT tier_id, COUNT(*) FROM decision_cache GROUP BY tier_id")?;
let mapped = stmt
.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?)))?
.collect::<std::result::Result<Vec<_>, _>>()?;
Ok::<_, rusqlite::Error>(mapped)
})
.await
.map_err(interact_err)?
.context("Failed to count decision_cache rows by tier")?;
Ok(rows)
}
/// TTL pruning for one tier (#1648 PR3): rows untouched for longer than
/// `ttl_hours` go. Recency is `last_used_at` when a hit has refreshed
/// it, else `created_at`; a row is only ever kept by being useful.
/// Only called for tiers with `ttl_hours` set; `None` in config means
/// "bounded by policy_version only", which is the sweeper's job.
pub async fn prune_expired(&self, tier_id: &str, ttl_hours: i64) -> Result<usize> {
let tier_id = tier_id.to_string();
let n = self
.pool
.get()
.await
.context("Failed to get connection")?
.interact(move |conn| {
conn.execute(
"DELETE FROM decision_cache \
WHERE tier_id = ?1 \
AND COALESCE(last_used_at, created_at) \
< strftime('%s','now') - (?2 * 3600)",
params![tier_id, ttl_hours],
)
})
.await
.map_err(interact_err)?
.context("Failed to prune expired decision_cache rows")?;
Ok(n)
}
/// Delete rows for tiers with no `[decisions.tiers]` entry (#1648 PR3).
/// An unconfigured tier's rows can never be read: lookup starts from
/// config, so these are garbage the kill switch left behind. Passing an
/// empty list (feature fully off) therefore clears the table, which is
/// exactly the removal cheapness the kill rule promises.
pub async fn delete_unknown_tiers(&self, known_tiers: &[String]) -> Result<usize> {
let known = known_tiers.to_vec();
let n = self
.pool
.get()
.await
.context("Failed to get connection")?
.interact(move |conn| {
if known.is_empty() {
return conn.execute("DELETE FROM decision_cache", []);
}
let placeholders = known.iter().map(|_| "?").collect::<Vec<_>>().join(",");
let sql =
format!("DELETE FROM decision_cache WHERE tier_id NOT IN ({placeholders})");
let mut stmt = conn.prepare(&sql)?;
stmt.execute(rusqlite::params_from_iter(known.iter()))
})
.await
.map_err(interact_err)?
.context("Failed to delete orphaned decision_cache rows")?;
Ok(n)
}
}