innate_core/storage/

traces.rs

use super::*;

impl Storage {
    #[allow(clippy::too_many_arguments)]
    pub fn insert_usage_trace(
        &self,
        trace_id: &str,
        chunk_id: Option<&str>,
        event: &str,
        strength: f64,
        similarity: Option<f64>,
        refine_mode: Option<&str>,
        tokens: Option<i64>,
        rank: Option<i64>,
        attribution: Option<&str>,
        source: &str,
        ts: &str,
    ) -> Result<usize> {
        let mut stmt = self.conn.prepare_cached(
            "INSERT OR IGNORE INTO usage_trace
             (trace_id, chunk_id, event, strength, similarity, refine_mode, tokens, rank, attribution, source, ts)
             VALUES (?,?,?,?,?,?,?,?,?,?,?)",
        )?;
        Ok(stmt.execute(params![
            trace_id,
            chunk_id,
            event,
            strength,
            similarity,
            refine_mode,
            tokens,
            rank,
            attribution,
            source,
            ts
        ])?)
    }

    pub fn replace_used_trace(
        &self,
        trace_id: &str,
        used_ids: &[String],
        strength: f64,
        attribution: &str,
        source: &str,
        ts: &str,
    ) -> Result<()> {
        self.conn.execute(
            "DELETE FROM usage_trace WHERE trace_id=? AND event='used'",
            [trace_id],
        )?;
        for chunk_id in used_ids {
            self.insert_usage_trace(
                trace_id,
                Some(chunk_id),
                "used",
                strength,
                None,
                None,
                None,
                None,
                Some(attribution),
                source,
                ts,
            )?;
        }
        Ok(())
    }

    pub fn merge_used_trace(
        &self,
        trace_id: &str,
        used_ids: &[String],
        strength: f64,
        attribution: &str,
        source: &str,
        ts: &str,
    ) -> Result<()> {
        if used_ids.is_empty() {
            return Ok(());
        }
        let attribution_rank = |value: &str| match value {
            "explicit" => 3,
            "cited" => 2,
            "inferred" => 1,
            _ => 0,
        };

        // Batch-fetch all existing 'used' rows for this trace in one query
        // instead of one SELECT per chunk id.
        let placeholders = used_ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
        let sql = format!(
            "SELECT chunk_id, attribution FROM usage_trace
             WHERE trace_id=? AND event='used' AND chunk_id IN ({placeholders})"
        );
        let mut qparams: Vec<&str> = Vec::with_capacity(used_ids.len() + 1);
        qparams.push(trace_id);
        qparams.extend(used_ids.iter().map(String::as_str));
        let existing: HashMap<String, String> = {
            let mut stmt = self.conn.prepare(&sql)?;
            let rows = stmt.query_map(rusqlite::params_from_iter(qparams.iter()), |r| {
                let id: String = r.get(0)?;
                let attr: Option<String> = r.get(1)?;
                Ok((id, attr.unwrap_or_else(|| "inferred".to_string())))
            })?;
            rows.collect::<rusqlite::Result<HashMap<_, _>>>()?
        };

        for chunk_id in used_ids {
            match existing.get(chunk_id) {
                Some(existing_attribution) => {
                    if attribution_rank(attribution) > attribution_rank(existing_attribution) {
                        self.conn.execute(
                            "UPDATE usage_trace
                             SET strength=?, attribution=?, source=?, ts=?
                             WHERE trace_id=? AND chunk_id=? AND event='used'",
                            params![strength, attribution, source, ts, trace_id, chunk_id],
                        )?;
                    }
                }
                None => {
                    self.insert_usage_trace(
                        trace_id,
                        Some(chunk_id),
                        "used",
                        strength,
                        None,
                        None,
                        None,
                        None,
                        Some(attribution),
                        source,
                        ts,
                    )?;
                }
            }
        }
        Ok(())
    }

    pub fn refresh_chunk_last_used(&self, chunk_id: &str, now: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE chunks
             SET last_used_at=COALESCE(
                   (SELECT MAX(ts) FROM usage_trace
                    WHERE chunk_id=? AND event='used'
                      AND ts > COALESCE(chunks.evidence_cutoff_at, '')),
                   last_used_base
                 ),
                 updated_at=?
             WHERE id=?",
            params![chunk_id, now, chunk_id],
        )?;
        Ok(())
    }

    pub fn get_outcome_for_trace(&self, trace_id: &str) -> Result<Option<String>> {
        let row = self.conn.query_row(
            "SELECT event FROM usage_trace
             WHERE trace_id=? AND event IN ('task_ok','task_fail') AND chunk_id IS NULL
             LIMIT 1",
            [trace_id],
            |r| r.get::<_, String>(0),
        );
        match row {
            Ok(v) => Ok(Some(v)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    pub fn purge_usage_trace(&self, before_ts: &str) -> Result<usize> {
        // Preserve compact attribution facts. They are required to replay corrections.
        let n = self.conn.execute(
            "DELETE FROM usage_trace
             WHERE ts < ?
             AND event IN ('retrieved','refined')
             AND NOT (event = 'retrieved'
                      AND chunk_id IN (SELECT id FROM chunks WHERE origin='spark'))",
            [before_ts],
        )?;
        Ok(n)
    }

    // ------------------------------------------------------------------
    // Episodic log
    // ------------------------------------------------------------------

    /// Insert an episodic_log row **if its trace_id does not already exist**.
    ///
    /// All callers (recall, appraise, record's fresh-insert branch) pass a newly
    /// generated trace_id, so this is an insert in practice. It is `INSERT OR
    /// IGNORE` rather than `INSERT OR REPLACE` on purpose: REPLACE deletes and
    /// re-inserts the whole row, which would silently wipe lifecycle state
    /// (`distill_state`, `distill_attempts` → 0, `distill_last_failed_at` → NULL,
    /// `outcome`, `usage_state`). An accidental re-upsert of an existing trace
    /// must never destroy that progress — mutate existing rows through
    /// `update_episodic_log_state` / `patch_episodic_log_content` instead.
    pub fn upsert_episodic_log(&self, log: &EpisodicLogRow) -> Result<()> {
        self.conn.execute(
            "INSERT OR IGNORE INTO episodic_log
             (id, trace_id, lib_id, ts, query, recall_snapshot, output,
              output_summary, outcome, event_source, task_state, completed_at,
              usage_state, used_ids, used_attribution, used_complete, context_key, nomination, priority,
              distill_state, distill_note, distill_attempts, distill_last_failed_at, agent)
             VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,0,NULL,?22)",
            params![
                log.id,
                log.trace_id,
                log.lib_id,
                log.ts,
                log.query,
                log.recall_snapshot,
                log.output,
                log.output_summary,
                log.outcome,
                log.event_source,
                log.task_state,
                log.completed_at,
                log.usage_state,
                log.used_ids,
                log.used_attribution,
                i64::from(log.used_complete),
                log.context_key,
                log.nomination,
                log.priority,
                log.distill_state,
                log.distill_note,
                log.agent
            ],
        )?;
        Ok(())
    }

    pub fn get_episodic_log(&self, trace_id: &str) -> Result<Option<Value>> {
        let row = self.conn.query_row(
            "SELECT * FROM episodic_log WHERE trace_id=?",
            [trace_id],
            row_to_json,
        );
        match row {
            Ok(v) => Ok(Some(v)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    /// Recent recall→record trace timeline (newest first) for the Web "Sessions"
    /// view. Projects a compact, UI-friendly subset of `episodic_log` rather than
    /// `SELECT *` so the endpoint stays cheap and stable. Read-only.
    pub fn recent_episodic_logs(&self, limit: usize) -> Result<Vec<Value>> {
        let mut stmt = self.conn.prepare_cached(
            "SELECT trace_id, ts, query, outcome, event_source, task_state,
                    usage_state, distill_state, agent, output_summary
             FROM episodic_log
             ORDER BY ts DESC
             LIMIT ?1",
        )?;
        let rows = stmt.query_map([limit as i64], |r| {
            Ok(serde_json::json!({
                "trace_id": r.get::<_, String>(0)?,
                "ts": r.get::<_, String>(1)?,
                "query": r.get::<_, Option<String>>(2)?,
                "outcome": r.get::<_, Option<String>>(3)?,
                "event_source": r.get::<_, Option<String>>(4)?,
                "task_state": r.get::<_, Option<String>>(5)?,
                "usage_state": r.get::<_, Option<String>>(6)?,
                "distill_state": r.get::<_, Option<String>>(7)?,
                "agent": r.get::<_, Option<String>>(8)?,
                "output_summary": r.get::<_, Option<String>>(9)?,
            }))
        })?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    /// Provenance timeline for a single chunk, powering the Web chunk-detail
    /// Provenance panel. Read-only projection across `usage_trace`,
    /// `feedback_events`, `chunk_success_traces`, and (for distilled chunks) the
    /// source `episodic_log`. Returns:
    ///   - `events`: newest-first merge of per-trace outcomes (task_ok/task_fail,
    ///     derived from the trace's chunk-less outcome row) and feedback_up/down
    ///     events, capped at `limit`.
    ///   - `source`: the originating episodic_log (query + output_summary) when the
    ///     chunk is `distilled` (`chunks.distilled_from` → `episodic_log.id`), else null.
    ///   - `stats` + `explanation`: success/failure/feedback counts, current EMA
    ///     confidence, and a natural-language confidence summary.
    pub fn chunk_provenance(&self, chunk_id: &str, limit: usize) -> Result<Value> {
        // Per-trace outcome timeline: each trace where this chunk was `used`, joined
        // to that trace's outcome row (chunk_id IS NULL, task_ok/task_fail).
        let mut stmt = self.conn.prepare_cached(
            "SELECT u.trace_id, u.source, u.ts AS used_ts, o.event AS outcome, o.ts AS outcome_ts
             FROM usage_trace u
             LEFT JOIN usage_trace o
               ON o.trace_id = u.trace_id AND o.chunk_id IS NULL
                  AND o.event IN ('task_ok','task_fail')
             WHERE u.chunk_id = ?1 AND u.event = 'used'",
        )?;
        let outcome_rows = stmt
            .query_map([chunk_id], |r| {
                let trace_id: String = r.get(0)?;
                let source: Option<String> = r.get(1)?;
                let used_ts: String = r.get(2)?;
                let outcome: Option<String> = r.get(3)?;
                let outcome_ts: Option<String> = r.get(4)?;
                let event = outcome.clone().unwrap_or_else(|| "used".to_string());
                let ts = outcome_ts.unwrap_or(used_ts);
                Ok(serde_json::json!({
                    "event": event,
                    "ts": ts,
                    "trace_id": trace_id,
                    "source": source,
                    "reason": Value::Null,
                }))
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?;

        // Feedback timeline: up/down signals carrying their own chunk_id.
        let mut fstmt = self.conn.prepare_cached(
            "SELECT signal, ts, trace_id, source, reason
             FROM feedback_events WHERE chunk_id = ?1",
        )?;
        let feedback_rows = fstmt
            .query_map([chunk_id], |r| {
                let signal: String = r.get(0)?;
                Ok(serde_json::json!({
                    "event": format!("feedback_{signal}"),
                    "ts": r.get::<_, String>(1)?,
                    "trace_id": r.get::<_, String>(2)?,
                    "source": r.get::<_, Option<String>>(3)?,
                    "reason": r.get::<_, Option<String>>(4)?,
                }))
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?;

        // Merge, newest-first, cap at `limit`. `ts` is the lexicographically
        // sortable utc_now_iso() format, so string compare == chronological.
        let mut events: Vec<Value> = outcome_rows.into_iter().chain(feedback_rows).collect();
        events.sort_by(|a, b| {
            let ta = a.get("ts").and_then(Value::as_str).unwrap_or("");
            let tb = b.get("ts").and_then(Value::as_str).unwrap_or("");
            tb.cmp(ta)
        });
        events.truncate(limit);

        // Aggregate stats counted live from usage_trace (independent of the curate
        // roll-up, so the panel is accurate before chunk_success_traces is built):
        // a success/failure is a trace where this chunk was `used` and that trace's
        // chunk-less outcome row is task_ok / task_fail respectively.
        let count_outcome = |outcome: &str| -> Result<i64> {
            Ok(self.conn.query_row(
                "SELECT COUNT(DISTINCT u.trace_id)
                 FROM usage_trace u
                 JOIN usage_trace o
                   ON o.trace_id = u.trace_id AND o.chunk_id IS NULL AND o.event = ?2
                 WHERE u.chunk_id = ?1 AND u.event = 'used'",
                params![chunk_id, outcome],
                |r| r.get(0),
            )?)
        };
        let successes = count_outcome("task_ok")?;
        let failures = count_outcome("task_fail")?;
        let (feedback_up, feedback_down): (i64, i64) = self.conn.query_row(
            "SELECT
               COALESCE(SUM(signal = 'up'), 0),
               COALESCE(SUM(signal = 'down'), 0)
             FROM feedback_events WHERE chunk_id = ?1",
            [chunk_id],
            |r| Ok((r.get(0)?, r.get(1)?)),
        )?;
        let (confidence, distilled_from): (Option<f64>, Option<String>) = self.conn.query_row(
            "SELECT confidence, distilled_from FROM chunks WHERE id = ?1",
            [chunk_id],
            |r| Ok((r.get(0)?, r.get(1)?)),
        )?;
        // Most recent negative signal: feedback_down or a task_fail outcome.
        let last_negative_at: Option<String> = self.conn.query_row(
            "SELECT MAX(ts) FROM (
               SELECT ts FROM feedback_events WHERE chunk_id = ?1 AND signal = 'down'
               UNION ALL
               SELECT o.ts FROM usage_trace u
                 JOIN usage_trace o ON o.trace_id = u.trace_id
                   AND o.chunk_id IS NULL AND o.event = 'task_fail'
                 WHERE u.chunk_id = ?1 AND u.event = 'used'
             )",
            [chunk_id],
            |r| r.get(0),
        )?;

        // Source episodic_log for distilled chunks (distilled_from = episodic_log.id).
        let source = match distilled_from.as_deref() {
            Some(log_id) if !log_id.is_empty() => self
                .conn
                .query_row(
                    "SELECT id, trace_id, query, output_summary, ts
                     FROM episodic_log WHERE id = ?1",
                    [log_id],
                    |r| {
                        Ok(serde_json::json!({
                            "log_id": r.get::<_, String>(0)?,
                            "trace_id": r.get::<_, String>(1)?,
                            "query": r.get::<_, Option<String>>(2)?,
                            "output_summary": r.get::<_, Option<String>>(3)?,
                            "ts": r.get::<_, String>(4)?,
                        }))
                    },
                )
                .optional()?
                .unwrap_or(Value::Null),
            _ => Value::Null,
        };

        let explanation = confidence_explanation(
            successes,
            failures,
            confidence,
            last_negative_at.as_deref(),
        );

        Ok(serde_json::json!({
            "chunk_id": chunk_id,
            "events": events,
            "source": source,
            "stats": {
                "successes": successes,
                "failures": failures,
                "feedback_up": feedback_up,
                "feedback_down": feedback_down,
                "confidence": confidence,
                "last_negative_at": last_negative_at,
            },
            "explanation": explanation,
        }))
    }

    pub fn update_episodic_log_state(
        &self,
        trace_id: &str,
        state: &str,
        note: Option<&str>,
        outcome: Option<&str>,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE episodic_log
             SET distill_state=?, distill_note=COALESCE(?,distill_note),
                 outcome=COALESCE(?,outcome),
                 distill_run_id=NULL, distill_locked_at=NULL
             WHERE trace_id=?",
            params![state, note, outcome, trace_id],
        )?;
        Ok(())
    }

    /// Patch content fields on an existing episodic_log row (補写: output_summary, nomination, etc.)
    pub fn patch_episodic_log_content(
        &self,
        trace_id: &str,
        query: Option<&str>,
        output: Option<&str>,
        output_summary: Option<&str>,
        nomination: Option<&str>,
        priority: i64,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE episodic_log
             SET output_summary = COALESCE(?, output_summary),
                 nomination     = COALESCE(?, nomination),
                 output         = COALESCE(?, output),
                 query          = COALESCE(?, query),
                 priority       = MAX(priority, ?)
             WHERE trace_id = ?",
            params![
                output_summary,
                nomination,
                output,
                query,
                priority,
                trace_id
            ],
        )?;
        Ok(())
    }

    #[allow(clippy::too_many_arguments)]
    pub fn update_trace_lifecycle(
        &self,
        trace_id: &str,
        task_state: &str,
        completed_at: Option<&str>,
        usage_state: Option<&str>,
        used_ids: Option<&str>,
        used_attribution: Option<&str>,
        used_complete: Option<bool>,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE episodic_log
             SET task_state=?,
                 completed_at=COALESCE(?, completed_at),
                 usage_state=COALESCE(?, usage_state),
                 used_ids=COALESCE(?, used_ids),
                 used_attribution=COALESCE(?, used_attribution),
                 used_complete=COALESCE(?, used_complete)
             WHERE trace_id=?",
            params![
                task_state,
                completed_at,
                usage_state,
                used_ids,
                used_attribution,
                used_complete.map(i64::from),
                trace_id
            ],
        )?;
        Ok(())
    }

    #[allow(clippy::too_many_arguments)]
    pub fn upsert_confidence_evidence(
        &self,
        id: &str,
        trace_id: Option<&str>,
        chunk_id: &str,
        kind: &str,
        target: f64,
        alpha: f64,
        reason: &str,
        context_key: Option<&str>,
        ts: &str,
        provenance: &str,
    ) -> Result<()> {
        self.conn.execute(
            "INSERT INTO confidence_evidence
             (id, trace_id, chunk_id, kind, target, alpha, reason, context_key, ts, provenance)
             VALUES (?,?,?,?,?,?,?,?,?,?)
             ON CONFLICT(trace_id, chunk_id, kind) WHERE trace_id IS NOT NULL
             DO UPDATE SET target=excluded.target, alpha=excluded.alpha,
                           reason=excluded.reason, context_key=excluded.context_key,
                           provenance=excluded.provenance",
            params![
                id,
                trace_id,
                chunk_id,
                kind,
                target,
                alpha,
                reason,
                context_key,
                ts,
                provenance
            ],
        )?;
        Ok(())
    }

    /// 方案 C / 门3:某 chunk 实际观测到的结果数(只数 provenance='observed' 的
    /// outcome 证据)。供 appraise 门3「证据充分性」判断邻居是否有观测历史。
    pub fn observed_outcome_count(&self, chunk_id: &str) -> Result<i64> {
        let n = self.conn.query_row(
            "SELECT COUNT(*) FROM confidence_evidence
             WHERE chunk_id=? AND provenance='observed'
               AND kind IN ('outcome_ok','outcome_fail')",
            params![chunk_id],
            |r| r.get::<_, i64>(0),
        )?;
        Ok(n)
    }

    /// 方案 F 门2:返回在给定 context_key(coarse signature 桶)下**有校准历史**的
    /// chunk 集合。rich 嵌入说「近」的邻居里,有多少在 signature 通道也「近」(有该
    /// 情境类的观测),低 = rich 嵌入在撒谎(疑似假共振)。
    pub fn context_stat_present_batch(
        &self,
        chunk_ids: &[&str],
        context_key: &str,
    ) -> Result<std::collections::HashSet<String>> {
        let mut set = std::collections::HashSet::new();
        if chunk_ids.is_empty() {
            return Ok(set);
        }
        let placeholders = chunk_ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
        let sql = format!(
            "SELECT chunk_id FROM chunk_context_stats
             WHERE context_key=? AND chunk_id IN ({placeholders})"
        );
        let mut params: Vec<&str> = Vec::with_capacity(chunk_ids.len() + 1);
        params.push(context_key);
        params.extend_from_slice(chunk_ids);
        let mut stmt = self.conn.prepare(&sql)?;
        let rows = stmt.query_map(rusqlite::params_from_iter(params.iter()), |r| {
            r.get::<_, String>(0)
        })?;
        for row in rows {
            set.insert(row?);
        }
        Ok(set)
    }

    /// 方案 B:写一条 verdict_log(emit 时)。表态填 valence/conf/strength/tier,
    /// 弃权填 abstain_reason(其余 NULL)。outcome 列留空,等 record 回填。
    #[allow(clippy::too_many_arguments)]
    pub fn insert_verdict_log(
        &self,
        verdict_id: &str,
        trace_id: &str,
        situation_sig: &str,
        emitted_valence: Option<&str>,
        emitted_conf: Option<f64>,
        emitted_strength: f64,
        emitted_tier: Option<&str>,
        abstain_reason: Option<&str>,
        emitted_at: &str,
    ) -> Result<()> {
        // verdict_id is a freshly minted UUID, so there is exactly one row per
        // appraise — a plain INSERT documents that invariant (no silent OR IGNORE).
        self.conn.execute(
            "INSERT INTO verdict_log
             (verdict_id, trace_id, situation_sig, emitted_valence, emitted_conf,
              emitted_strength, emitted_tier, abstain_reason, emitted_at)
             VALUES (?,?,?,?,?,?,?,?,?)",
            params![
                verdict_id,
                trace_id,
                situation_sig,
                emitted_valence,
                emitted_conf,
                emitted_strength,
                emitted_tier,
                abstain_reason,
                emitted_at
            ],
        )?;
        Ok(())
    }

    /// 方案 B/H:用 record 的实际结果回填 verdict_log。`provenance` 区分
    /// 'observed'(真实采取动作并观测到结果,计入校准)与
    /// 'counterfactual_censored'(因警告回避了动作,**不计入校准**,见原则 3)。
    pub fn backfill_verdict_outcome(
        &self,
        trace_id: &str,
        observed_outcome: f64,
        provenance: &str,
        observed_at: &str,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE verdict_log
                SET observed_outcome=?, outcome_observed_at=?, outcome_provenance=?
              WHERE trace_id=? AND outcome_observed_at IS NULL",
            params![observed_outcome, observed_at, provenance, trace_id],
        )?;
        Ok(())
    }

    /// 方案 E:加载校准映射(分桶查表)。返回 (claimed_lo, claimed_hi, observed_rate)。
    pub fn load_calibration_map(&self) -> Result<Vec<(f64, f64, f64)>> {
        let mut stmt = self.conn.prepare(
            "SELECT claimed_lo, claimed_hi, observed_rate FROM calibration_map ORDER BY bucket",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, f64>(0)?,
                r.get::<_, f64>(1)?,
                r.get::<_, f64>(2)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            out.push(row?);
        }
        Ok(out)
    }

    /// 方案 E/B:取所有「observed」回填的 (emitted_strength, emitted_conf, hit) 三元组,
    /// 供 curate 重算校准映射(按 **strength** 分桶,因为 emit 时 `calibrate_confidence`
    /// 正是用原始 strength 查表)与 inspect 算 ECE(按 **conf** 分桶,衡量声称置信度的
    /// 真实兑现率)。两者域不同,故同时返回,调用方各取所需。
    ///
    /// `hit` = verdict 的关切是否兑现,只对**方向性** verdict 有良定义:
    ///   affirm → 命中=结果 ok(observed_outcome<0);caution → 命中=结果 fail。
    /// neutral(无信号)与 mixed(方向歧义)不参与校准 —— 否则把「没表态」误记成
    /// 「预测失败」,污染校准映射与 ECE。
    pub fn verdict_calibration_samples(&self) -> Result<Vec<(f64, f64, f64)>> {
        let mut stmt = self.conn.prepare(
            "SELECT emitted_strength, emitted_conf,
                    CASE WHEN emitted_valence='affirm'
                         THEN (CASE WHEN observed_outcome < 0 THEN 1.0 ELSE 0.0 END)
                         ELSE (CASE WHEN observed_outcome > 0 THEN 1.0 ELSE 0.0 END) END
               FROM verdict_log
              WHERE outcome_provenance='observed'
                AND emitted_conf IS NOT NULL AND emitted_strength IS NOT NULL
                AND observed_outcome IS NOT NULL
                AND emitted_valence IN ('affirm','caution')",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, f64>(0)?,
                r.get::<_, f64>(1)?,
                r.get::<_, f64>(2)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            out.push(row?);
        }
        Ok(out)
    }

    /// 方案 B:verdict_log 概览 (total, abstained, with_observed_outcome)。供 inspect 仪表盘。
    pub fn verdict_log_overview(&self) -> Result<(i64, i64, i64)> {
        let total: i64 = self
            .conn
            .query_row("SELECT COUNT(*) FROM verdict_log", [], |r| r.get(0))?;
        let abstained: i64 = self.conn.query_row(
            "SELECT COUNT(*) FROM verdict_log WHERE abstain_reason IS NOT NULL",
            [],
            |r| r.get(0),
        )?;
        let observed: i64 = self.conn.query_row(
            "SELECT COUNT(*) FROM verdict_log WHERE outcome_provenance='observed'",
            [],
            |r| r.get(0),
        )?;
        Ok((total, abstained, observed))
    }

    /// 方案 E:重写 calibration_map(curate 调用)。`buckets` = (lo, hi, rate, n)。
    pub fn replace_calibration_map(
        &self,
        buckets: &[(f64, f64, f64, i64)],
        now: &str,
    ) -> Result<()> {
        self.conn.execute("DELETE FROM calibration_map", [])?;
        for (i, (lo, hi, rate, n)) in buckets.iter().enumerate() {
            self.conn.execute(
                "INSERT INTO calibration_map
                 (bucket, claimed_lo, claimed_hi, observed_rate, n, updated_at)
                 VALUES (?,?,?,?,?,?)",
                params![i as i64, lo, hi, rate, n, now],
            )?;
        }
        Ok(())
    }

    pub fn delete_trace_confidence_evidence(&self, trace_id: &str, kinds: &[&str]) -> Result<()> {
        if kinds.is_empty() {
            return Ok(());
        }
        let placeholders = kinds.iter().map(|_| "?").collect::<Vec<_>>().join(",");
        let sql = format!(
            "DELETE FROM confidence_evidence WHERE trace_id=? AND kind IN ({placeholders})"
        );
        let mut params: Vec<&str> = Vec::with_capacity(kinds.len() + 1);
        params.push(trace_id);
        params.extend_from_slice(kinds);
        self.conn
            .execute(&sql, rusqlite::params_from_iter(params.iter()))?;
        Ok(())
    }

    pub fn delete_chunk_trace_confidence_evidence(
        &self,
        trace_id: &str,
        chunk_id: &str,
        kind: &str,
    ) -> Result<()> {
        self.conn.execute(
            "DELETE FROM confidence_evidence
             WHERE trace_id=? AND chunk_id=? AND kind=?",
            params![trace_id, chunk_id, kind],
        )?;
        Ok(())
    }

    pub fn confidence_evidence_for_chunk(&self, chunk_id: &str) -> Result<Vec<Value>> {
        // 方案 C:只让观测结果驱动置信度;verdict_derived 证据留痕但不参与重算。
        self.query_json(
            "SELECT target, alpha, reason, ts, id
             FROM confidence_evidence WHERE chunk_id=? AND provenance='observed'
             ORDER BY ts ASC,
                      CASE kind
                        WHEN 'outcome_ok' THEN 1
                        WHEN 'outcome_fail' THEN 1
                        WHEN 'selected_unused' THEN 2
                        WHEN 'feedback_up' THEN 3
                        WHEN 'feedback_down' THEN 3
                        WHEN 'decay' THEN 4
                        ELSE 5
                      END ASC,
                      kind ASC, id ASC",
            [chunk_id],
        )
    }

    #[allow(clippy::too_many_arguments)]
    pub fn insert_feedback_event(
        &self,
        id: &str,
        trace_id: &str,
        chunk_id: &str,
        signal: &str,
        strength: f64,
        source: &str,
        actor: Option<&str>,
        reason: Option<&str>,
        context_key: Option<&str>,
        ts: &str,
    ) -> Result<usize> {
        Ok(self.conn.execute(
            "INSERT OR IGNORE INTO feedback_events
             (id, trace_id, chunk_id, signal, strength, source, actor, reason, context_key, ts)
             VALUES (?,?,?,?,?,?,?,?,?,?)",
            params![
                id,
                trace_id,
                chunk_id,
                signal,
                strength,
                source,
                actor,
                reason,
                context_key,
                ts
            ],
        )?)
    }

    pub fn delete_feedback_event(
        &self,
        trace_id: &str,
        chunk_id: &str,
        signal: &str,
    ) -> Result<usize> {
        Ok(self.conn.execute(
            "DELETE FROM feedback_events
             WHERE trace_id=? AND chunk_id=? AND signal=?",
            params![trace_id, chunk_id, signal],
        )?)
    }

    pub fn update_chunk_last_decayed_at(&self, id: &str, now: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE chunks SET last_decayed_at=?, updated_at=? WHERE id=?",
            params![now, now, id],
        )?;
        Ok(())
    }

    #[allow(clippy::too_many_arguments)]
    pub fn update_context_stat(
        &self,
        chunk_id: &str,
        context_key: &str,
        success: i64,
        failure: i64,
        positive: i64,
        negative: i64,
        now: &str,
    ) -> Result<()> {
        self.conn.execute(
            "INSERT INTO chunk_context_stats
             (chunk_id, context_key, success_count, failure_count,
              positive_feedback, negative_feedback, last_updated_at)
             VALUES (?,?,?,?,?,?,?)
             ON CONFLICT(chunk_id, context_key) DO UPDATE SET
               success_count=success_count+excluded.success_count,
               failure_count=failure_count+excluded.failure_count,
               positive_feedback=positive_feedback+excluded.positive_feedback,
               negative_feedback=negative_feedback+excluded.negative_feedback,
               last_updated_at=excluded.last_updated_at",
            params![
                chunk_id,
                context_key,
                success,
                failure,
                positive,
                negative,
                now
            ],
        )?;
        Ok(())
    }

    pub fn context_score(
        &self,
        chunk_id: &str,
        context_key: &str,
        prior_m: f64,
        base_rate: f64,
    ) -> Result<f64> {
        let mut stmt = self.conn.prepare_cached(
            "SELECT success_count, failure_count, positive_feedback, negative_feedback
             FROM chunk_context_stats WHERE chunk_id=? AND context_key=?",
        )?;
        let row = stmt
            .query_row(params![chunk_id, context_key], |row| {
                Ok((
                    row.get::<_, i64>(0)?,
                    row.get::<_, i64>(1)?,
                    row.get::<_, i64>(2)?,
                    row.get::<_, i64>(3)?,
                ))
            })
            .optional()?;
        let Some((success, failure, positive, negative)) = row else {
            return Ok(0.0);
        };
        Ok(context_score_from_counts(
            success, failure, positive, negative, prior_m, base_rate,
        ))
    }

    /// Batch variant of `context_score`: one query for many chunk ids under a
    /// single context key. Chunks with no stats are absent from the map (score 0).
    pub fn context_scores_batch(
        &self,
        chunk_ids: &[&str],
        context_key: &str,
        prior_m: f64,
        base_rate: f64,
    ) -> Result<HashMap<String, f64>> {
        if chunk_ids.is_empty() {
            return Ok(HashMap::new());
        }
        let placeholders = chunk_ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
        let sql = format!(
            "SELECT chunk_id, success_count, failure_count, positive_feedback, negative_feedback
             FROM chunk_context_stats
             WHERE context_key=? AND chunk_id IN ({placeholders})"
        );
        let mut params: Vec<&str> = Vec::with_capacity(chunk_ids.len() + 1);
        params.push(context_key);
        params.extend_from_slice(chunk_ids);
        let mut stmt = self.conn.prepare(&sql)?;
        let rows = stmt.query_map(rusqlite::params_from_iter(params.iter()), |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, i64>(1)?,
                r.get::<_, i64>(2)?,
                r.get::<_, i64>(3)?,
                r.get::<_, i64>(4)?,
            ))
        })?;
        let mut map = HashMap::new();
        for row in rows {
            let (id, success, failure, positive, negative) = row?;
            map.insert(
                id,
                context_score_from_counts(success, failure, positive, negative, prior_m, base_rate),
            );
        }
        Ok(map)
    }
}

/// Shared scoring math for `context_score` / `context_scores_batch`.
///
/// 方案 D —— 基率锚定先验:prior = Beta(α0, β0),α0 = m·g0,β0 = m·(1-g0),
/// 其中 g0 是全局「好结果」基率、m 是伪观测数(谦逊度旋钮)。证据稀疏时后验回归
/// 到 g0 而非 0.5。`m=2, g0=0.5` 与旧 Laplace `(wins+1)/(evidence+2)` 完全等价。
fn context_score_from_counts(
    success: i64,
    failure: i64,
    positive: i64,
    negative: i64,
    prior_m: f64,
    base_rate: f64,
) -> f64 {
    let wins = success as f64 + positive as f64 * 2.0;
    let losses = failure as f64 + negative as f64 * 2.0;
    let evidence = wins + losses;
    let alpha0 = prior_m * base_rate;
    let beta0 = prior_m * (1.0 - base_rate);
    let posterior = (wins + alpha0) / (evidence + alpha0 + beta0);
    let evidence_weight = (evidence / 5.0).min(1.0);
    (posterior - 0.5) * 2.0 * evidence_weight
}

/// Natural-language confidence summary for the Provenance panel, e.g.
/// `"5 successes, 2 failures → EMA 0.62; last negative feedback 3d ago"`.
/// `last_negative_at` is an `utc_now_iso()` timestamp; the relative suffix is
/// omitted when there has never been a negative signal.
fn confidence_explanation(
    successes: i64,
    failures: i64,
    confidence: Option<f64>,
    last_negative_at: Option<&str>,
) -> String {
    let s_word = if successes == 1 { "success" } else { "successes" };
    let f_word = if failures == 1 { "failure" } else { "failures" };
    let ema = confidence
        .map(|c| format!("{c:.2}"))
        .unwrap_or_else(|| "n/a".to_string());
    let mut out = format!("{successes} {s_word}, {failures} {f_word} → EMA {ema}");
    if let Some(ts) = last_negative_at {
        if let Some(ago) = relative_ago(ts) {
            out.push_str(&format!("; last negative feedback {ago}"));
        }
    }
    out
}

/// Humanize an `utc_now_iso()` timestamp relative to now, e.g. `"3d ago"`,
/// `"5h ago"`, `"2m ago"`, `"just now"`. Returns `None` if the timestamp can't
/// be parsed, so callers can drop the clause rather than render garbage.
fn relative_ago(ts: &str) -> Option<String> {
    let then = chrono::DateTime::parse_from_rfc3339(ts)
        .ok()?
        .with_timezone(&chrono::Utc);
    let secs = (chrono::Utc::now() - then).num_seconds();
    if secs < 0 {
        return Some("just now".to_string());
    }
    let out = if secs < 60 {
        "just now".to_string()
    } else if secs < 3600 {
        format!("{}m ago", secs / 60)
    } else if secs < 86_400 {
        format!("{}h ago", secs / 3600)
    } else {
        format!("{}d ago", secs / 86_400)
    };
    Some(out)
}