innate_core/storage/
evolution.rs

1use super::*;
2
3impl Storage {
4    #[allow(clippy::too_many_arguments)]
5    pub fn upsert_governance_proposal(
6        &self,
7        id: &str,
8        chunk_id: &str,
9        proposal_type: &str,
10        reason: &str,
11        evidence_count: i64,
12        evidence_score: f64,
13        actor_count: i64,
14        now: &str,
15    ) -> Result<()> {
16        self.conn.execute(
17            "INSERT INTO governance_proposals
18             (id, chunk_id, proposal_type, reason, evidence_count,
19              evidence_score, actor_count, state, created_at, updated_at)
20             VALUES (?,?,?,?,?,?,?,'pending',?,?)
21             ON CONFLICT(chunk_id, proposal_type) WHERE state='pending'
22             DO UPDATE SET reason=excluded.reason,
23                           evidence_count=excluded.evidence_count,
24                           evidence_score=excluded.evidence_score,
25                           actor_count=excluded.actor_count,
26                           updated_at=excluded.updated_at",
27            params![
28                id,
29                chunk_id,
30                proposal_type,
31                reason,
32                evidence_count,
33                evidence_score,
34                actor_count,
35                now,
36                now
37            ],
38        )?;
39        Ok(())
40    }
41
42    /// Read-only listing of governance proposals for the web review queue,
43    /// joined with a compact chunk projection so a reviewer sees *what* was
44    /// flagged and *why* (evidence_score / actor_count) in one round-trip.
45    /// Ordered by evidence_score DESC so the strongest cases surface first.
46    pub fn list_governance_proposals(&self, state: &str, limit: usize) -> Result<Vec<Value>> {
47        let mut stmt = self.conn.prepare(
48            "SELECT g.id, g.chunk_id, g.proposal_type, g.reason,
49                    g.evidence_count, g.evidence_score, g.actor_count,
50                    g.state, g.created_at, g.updated_at,
51                    c.skill_name, c.seq, c.origin, c.state AS chunk_state,
52                    c.confidence, substr(c.content, 1, 280) AS content_preview
53             FROM governance_proposals g
54             LEFT JOIN chunks c ON c.id = g.chunk_id
55             WHERE g.state = :state
56             ORDER BY g.evidence_score DESC, g.updated_at DESC
57             LIMIT :limit",
58        )?;
59        let names: Vec<String> = stmt.column_names().into_iter().map(String::from).collect();
60        let limit_i = limit as i64;
61        let rows = stmt.query_map(
62            rusqlite::named_params! { ":state": state, ":limit": limit_i },
63            |r| row_to_json_with_names(r, &names),
64        )?;
65        let mut out = Vec::new();
66        for row in rows {
67            out.push(row?);
68        }
69        Ok(out)
70    }
71
72    pub fn request_evolve(&self, id: &str, reason: &str, now: &str) -> Result<()> {
73        self.request_evolve_at(id, reason, now, None)
74    }
75
76    pub fn request_evolve_at(
77        &self,
78        id: &str,
79        reason: &str,
80        now: &str,
81        next_retry_at: Option<&str>,
82    ) -> Result<()> {
83        let priority = match reason {
84            "governance_ready" => 100,
85            "governance" => 80,
86            "threshold" => 60,
87            "distill_retry" => 50,
88            "batch_continue" => 40,
89            _ => 20,
90        };
91        self.conn.execute(
92            "INSERT INTO evolve_requests(
93               id, reason, state, requested_at, priority, next_retry_at
94             )
95             VALUES (?,?,'pending',?,?,?)
96             ON CONFLICT(reason) WHERE state='pending'
97             DO UPDATE SET
98               priority=MAX(priority, excluded.priority),
99               next_retry_at=CASE
100                 WHEN excluded.next_retry_at IS NULL THEN NULL
101                 WHEN evolve_requests.next_retry_at IS NULL THEN NULL
102                 ELSE MIN(evolve_requests.next_retry_at, excluded.next_retry_at)
103               END",
104            params![id, reason, now, priority, next_retry_at],
105        )?;
106        Ok(())
107    }
108
109    pub fn claim_evolve_request_with_reason(
110        &self,
111        now: &str,
112        stale_before: &str,
113    ) -> Result<Option<EvolveRequestClaim>> {
114        self.conn.execute(
115            "UPDATE evolve_requests
116             SET state='pending', leased_at=NULL, note='lease_recovered'
117             WHERE state='running' AND leased_at < ?",
118            [stale_before],
119        )?;
120        self.conn.execute(
121            "UPDATE evolve_requests
122             SET state='pending', leased_at=NULL, note='retry_failed'
123             WHERE state='failed' AND attempts < 3
124               AND COALESCE(next_retry_at, completed_at) < ?",
125            [now],
126        )?;
127        Ok(self
128            .conn
129            .query_row(
130                "UPDATE evolve_requests
131             SET state='running', leased_at=?, attempts=attempts+1
132             WHERE id=(
133               SELECT id FROM evolve_requests
134               WHERE state='pending'
135                 AND (next_retry_at IS NULL OR next_retry_at <= ?)
136               ORDER BY priority DESC, requested_at ASC LIMIT 1
137             ) AND state='pending'
138             RETURNING id, reason",
139                params![now, now],
140                |row| {
141                    Ok(EvolveRequestClaim {
142                        id: row.get(0)?,
143                        reason: row.get(1)?,
144                    })
145                },
146            )
147            .optional()?)
148    }
149
150    pub fn defer_evolve_request(&self, id: &str, note: &str, next_retry_at: &str) -> Result<()> {
151        self.conn.execute(
152            "DELETE FROM evolve_requests
153             WHERE state='pending'
154               AND id!=?
155               AND reason=(SELECT reason FROM evolve_requests WHERE id=?)",
156            params![id, id],
157        )?;
158        self.conn.execute(
159            "UPDATE evolve_requests
160             SET state='pending', leased_at=NULL, completed_at=NULL,
161                 note=?, next_retry_at=?
162             WHERE id=?",
163            params![note, next_retry_at, id],
164        )?;
165        Ok(())
166    }
167
168    pub fn finish_evolve_request(
169        &self,
170        id: &str,
171        state: &str,
172        note: Option<&str>,
173        now: &str,
174    ) -> Result<()> {
175        self.conn.execute(
176            "UPDATE evolve_requests
177             SET state=?, completed_at=?, note=?,
178                 last_failed_at=CASE
179                   WHEN ?='failed' THEN ?
180                   ELSE last_failed_at
181                 END,
182                 next_retry_at=CASE WHEN ?='failed'
183                   THEN strftime('%Y-%m-%dT%H:%M:%fZ', ?, '+5 minutes')
184                   ELSE NULL END
185             WHERE id=?",
186            params![state, now, note, state, now, state, now, id],
187        )?;
188        Ok(())
189    }
190
191    /// After a successful generic distill+curate run, bulk-complete the OTHER
192    /// pending requests queued before this run started — they are genuinely
193    /// subsumed by it (`distill_batch` drains every eligible episodic log; curate
194    /// runs globally).
195    ///
196    /// `governance` / `governance_ready` requests are deliberately **excluded**:
197    /// they carry per-request semantics and are special-cased on every other path
198    /// (claim priority, `curate_only` completion in the skip branches), so a
199    /// generic distill run must not silently mark them done across the reason
200    /// boundary. They stay pending and are serviced on their own claim turn.
201    pub fn finish_covered_evolve_requests(&self, requested_before: &str, now: &str) -> Result<()> {
202        self.conn.execute(
203            "UPDATE evolve_requests
204             SET state='completed', completed_at=?, note='covered_by_evolve', next_retry_at=NULL
205             WHERE state='pending' AND requested_at <= ?
206               AND reason NOT IN ('governance','governance_ready')",
207            params![now, requested_before],
208        )?;
209        Ok(())
210    }
211
212    /// Update by primary-key id (used after distill where we have the row id, not trace_id).
213    pub fn update_episodic_log_state_by_id(
214        &self,
215        id: &str,
216        state: &str,
217        note: Option<&str>,
218        outcome: Option<&str>,
219    ) -> Result<()> {
220        self.conn.execute(
221            "UPDATE episodic_log
222             SET distill_state=?, distill_note=COALESCE(?,distill_note),
223                 outcome=COALESCE(?,outcome),
224                 distill_run_id=NULL, distill_locked_at=NULL
225             WHERE id=?",
226            params![state, note, outcome, id],
227        )?;
228        Ok(())
229    }
230
231    #[allow(clippy::too_many_arguments)]
232    pub fn finish_distill_log(
233        &self,
234        id: &str,
235        state: &str,
236        note: Option<&str>,
237        prompt_tokens: i64,
238        completion_tokens: i64,
239        accounted_at: &str,
240    ) -> Result<()> {
241        self.conn.execute(
242            "INSERT INTO distill_token_usage(
243               log_id, prompt_tokens, completion_tokens, outcome, accounted_at
244             ) VALUES (?,?,?,?,?)",
245            params![id, prompt_tokens, completion_tokens, state, accounted_at],
246        )?;
247        self.conn.execute(
248            "UPDATE episodic_log
249             SET distill_state=?, distill_note=?,
250                 distill_prompt_tokens=?, distill_completion_tokens=?,
251                 distill_accounted_at=?,
252                 distill_attempts=distill_attempts
253                   + CASE WHEN ?='failed' THEN 1 ELSE 0 END,
254                 distill_last_failed_at=CASE
255                   WHEN ?='failed' THEN ?
256                   ELSE distill_last_failed_at
257                 END,
258                 distill_run_id=NULL, distill_locked_at=NULL
259             WHERE id=?",
260            params![
261                state,
262                note,
263                prompt_tokens,
264                completion_tokens,
265                accounted_at,
266                state,
267                state,
268                accounted_at,
269                id
270            ],
271        )?;
272        Ok(())
273    }
274
275    /// Claim a batch of 'new' logs for distillation: mark them 'screening' atomically.
276    /// Returns the claimed rows (with distill_run_id set to run_id).
277    pub fn claim_distill_batch(
278        &self,
279        run_id: &str,
280        limit: usize,
281        locked_at: &str,
282    ) -> Result<Vec<Value>> {
283        // BEGIN IMMEDIATE must be held by caller; this is called inside a transaction.
284        self.conn.execute(
285            "UPDATE episodic_log
286             SET distill_state='screening', distill_run_id=?, distill_locked_at=?
287             WHERE id IN (
288               SELECT id FROM episodic_log
289               WHERE distill_state='new'
290               ORDER BY priority DESC, ts ASC
291               LIMIT ?
292             )",
293            params![run_id, locked_at, limit as i64],
294        )?;
295        self.query_json(
296            "SELECT * FROM episodic_log WHERE distill_run_id=? AND distill_state='screening'",
297            params![run_id],
298        )
299    }
300
301    pub fn query_episodic_logs_open(&self, limit: usize) -> Result<Vec<Value>> {
302        self.query_json(
303            "SELECT * FROM episodic_log WHERE distill_state='new' ORDER BY priority DESC, ts ASC LIMIT ?",
304            params![limit as i64],
305        )
306    }
307}
innate_core/storage/evolution.rs

innate_core/storage/
evolution.rs
