crtx-store 0.1.0

//! Episode repository operations.
//!
//! ADR 0015 makes summary-span provenance a first-class column on episodes,
//! and ADR 0026 §2 requires every mutation of an interpreted-episode row to
//! compose through the policy lattice. Every write entry point on
//! [`EpisodeRepo`] now requires the caller to pass a composed
//! [`PolicyDecision`] whose contributing rules name the ADR 0015 / ADR 0027
//! invariants being asserted.
//!
//! Required contributor rule ids:
//!
//! - [`INSERT_SOURCE_EVENT_LINEAGE_RULE_ID`] and
//!   [`INSERT_REDACTION_STATUS_RULE_ID`] for [`EpisodeRepo::insert`] (the v1
//!   write path).
//! - The same two contributors plus [`INSERT_SUMMARY_SPAN_PROOF_RULE_ID`] for
//!   [`EpisodeRepo::insert_with_summary_spans`] (the schema-v2 summary-span
//!   write path). The v2 path is strictly stronger than v1: it adds the
//!   summary-span proof contributor on top of the v1 contributor set so the
//!   ADR 0015 cached-authority invariant is auditable through the engine.
//!
//! Fold decision: the two write methods stay separate. ADR 0026 §4 forbids
//! widening the v1 contract by stealth, and the v2 path composes a strictly
//! larger contributor set. House style in [`crate::repo::memories`] and
//! [`crate::repo::context_packs`] keeps the v1 and v2 insert entry points
//! separate for the same reason: a v1 caller that has not yet wired
//! summary-span provenance must not be coerced into fabricating empty span
//! arrays.

use cortex_core::{
    compose_policy_outcomes, validate_summary_spans, EpisodeId, PolicyContribution, PolicyDecision,
    PolicyOutcome, SourceAuthority, SummarySpan, TraceId,
};
use rusqlite::{params, OptionalExtension, Row};
use serde_json::Value;

use crate::{Pool, StoreError, StoreResult};

/// Required contributor rule id documenting that the caller asserted the
/// episode's `source_events_json` lineage is non-empty and resolvable before
/// the row is persisted (ADR 0026 §5, ADR 0015 §"Source attribution"). The
/// repo refuses orphan episodes whose lineage array is empty.
pub const INSERT_SOURCE_EVENT_LINEAGE_RULE_ID: &str = "episode.insert.source_event_lineage";
/// Required contributor rule id documenting that the caller asserted the
/// episode payload was redacted per ADR 0027 tier rules before the row is
/// persisted. The repo enforces that the contributor is present and that the
/// composed decision is not `Reject` / `Quarantine` — it cannot inspect the
/// payload itself.
pub const INSERT_REDACTION_STATUS_RULE_ID: &str = "episode.insert.redaction_status";
/// Required contributor rule id documenting that the caller composed a
/// summary-span provenance proof per ADR 0015 before the row is persisted.
/// Only applies to the schema-v2 [`EpisodeRepo::insert_with_summary_spans`]
/// write path; the v1 [`EpisodeRepo::insert`] path does not write summary
/// spans and therefore does not require this contributor.
pub const INSERT_SUMMARY_SPAN_PROOF_RULE_ID: &str = "episode.summary_span_proof";

macro_rules! episode_select_sql {
    ($where_clause:literal) => {
        concat!(
            "SELECT id, trace_id, source_events_json, summary, domains_json, entities_json,
                    candidate_meaning, extracted_by_json, confidence, status
             FROM episodes ",
            $where_clause,
            ";"
        )
    };
}

/// Durable episode row matching the current store schema.
#[derive(Debug, Clone, PartialEq)]
pub struct EpisodeRecord {
    /// Stable episode identifier.
    pub id: EpisodeId,
    /// Source trace identifier.
    pub trace_id: TraceId,
    /// Source event lineage as JSON.
    pub source_events_json: Value,
    /// Human-readable episode summary.
    pub summary: String,
    /// Domain tags as JSON.
    pub domains_json: Value,
    /// Entity tags as JSON.
    pub entities_json: Value,
    /// Optional candidate meaning.
    pub candidate_meaning: Option<String>,
    /// Extractor descriptor JSON.
    pub extracted_by_json: Value,
    /// Confidence score in `[0, 1]`.
    pub confidence: f64,
    /// Interpretation lifecycle status.
    pub status: String,
}

/// Repository for interpreted episode rows.
#[derive(Debug)]
pub struct EpisodeRepo<'a> {
    pool: &'a Pool,
}

impl<'a> EpisodeRepo<'a> {
    /// Creates an episode repository over an open SQLite connection.
    #[must_use]
    pub const fn new(pool: &'a Pool) -> Self {
        Self { pool }
    }

    /// Inserts one episode row through the ADR 0026 enforcement lattice.
    ///
    /// `policy` is the composed [`PolicyDecision`] for this insertion and MUST
    /// satisfy:
    ///
    /// 1. The final outcome is one of [`PolicyOutcome::Allow`],
    ///    [`PolicyOutcome::Warn`], or [`PolicyOutcome::BreakGlass`]. A
    ///    `Quarantine` or `Reject` decision fails closed and writes nothing.
    /// 2. The composition includes contributors for both
    ///    [`INSERT_SOURCE_EVENT_LINEAGE_RULE_ID`] and
    ///    [`INSERT_REDACTION_STATUS_RULE_ID`]. The repo refuses callers that
    ///    skipped composition.
    ///
    /// Structural fail-closed: even when policy composes `Allow`, the repo
    /// refuses orphan episodes whose `source_events_json` lineage array is
    /// empty. ADR 0015 requires every persisted episode to carry resolvable
    /// source-event lineage.
    pub fn insert(&self, episode: &EpisodeRecord, policy: &PolicyDecision) -> StoreResult<()> {
        require_policy_final_outcome(policy, "episode.insert")?;
        require_contributor_rule(policy, INSERT_SOURCE_EVENT_LINEAGE_RULE_ID)?;
        require_contributor_rule(policy, INSERT_REDACTION_STATUS_RULE_ID)?;

        if json_array_empty(&episode.source_events_json) {
            return Err(StoreError::Validation(
                "episode requires source event lineage".into(),
            ));
        }

        self.pool.execute(
            "INSERT INTO episodes (
                id, trace_id, source_events_json, summary, domains_json, entities_json,
                candidate_meaning, extracted_by_json, confidence, status
             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10);",
            params![
                episode.id.to_string(),
                episode.trace_id.to_string(),
                serde_json::to_string(&episode.source_events_json)?,
                episode.summary,
                serde_json::to_string(&episode.domains_json)?,
                serde_json::to_string(&episode.entities_json)?,
                episode.candidate_meaning,
                serde_json::to_string(&episode.extracted_by_json)?,
                episode.confidence,
                episode.status,
            ],
        )?;

        Ok(())
    }

    /// Inserts one episode row with explicit schema-v2 summary-span provenance
    /// through the ADR 0026 enforcement lattice.
    ///
    /// This opt-in API is only valid after the S2 expand/backfill shape has
    /// added `episodes.summary_spans_json`. The default v1 [`Self::insert`]
    /// path remains unchanged.
    ///
    /// `policy` is the composed [`PolicyDecision`] for this insertion and MUST
    /// satisfy:
    ///
    /// 1. The final outcome is one of [`PolicyOutcome::Allow`],
    ///    [`PolicyOutcome::Warn`], or [`PolicyOutcome::BreakGlass`]. A
    ///    `Quarantine` or `Reject` decision fails closed and writes nothing.
    /// 2. The composition includes contributors for
    ///    [`INSERT_SOURCE_EVENT_LINEAGE_RULE_ID`],
    ///    [`INSERT_REDACTION_STATUS_RULE_ID`], and
    ///    [`INSERT_SUMMARY_SPAN_PROOF_RULE_ID`]. The schema-v2 write path is
    ///    strictly stronger than the v1 [`Self::insert`] path: it carries the
    ///    v1 contributor set plus the summary-span proof contributor required
    ///    by ADR 0015 §"Cached authority invariant".
    ///
    /// Structural fail-closed: even when policy composes `Allow`, the repo
    /// refuses orphan episodes whose `source_events_json` lineage array is
    /// empty, and refuses summary spans that fail
    /// [`validate_summary_spans`] (UTF-8 alignment, ordering, coverage, and
    /// cached-authority fold per ADR 0015). The repo cannot resolve event ids
    /// to authority tiers from the store boundary, so the cached-authority
    /// fold passes `SourceAuthority::Derived` as the conservative tier; a
    /// span whose cached `max_source_authority` is `User` / `Agent` therefore
    /// fails the invariant and is rejected. Composing
    /// [`INSERT_SUMMARY_SPAN_PROOF_RULE_ID`] is the caller's contract that
    /// the proper authority fold ran upstream.
    pub fn insert_with_summary_spans(
        &self,
        episode: &EpisodeRecord,
        summary_spans: &[SummarySpan],
        policy: &PolicyDecision,
    ) -> StoreResult<()> {
        require_policy_final_outcome(policy, "episode.insert_with_summary_spans")?;
        require_contributor_rule(policy, INSERT_SOURCE_EVENT_LINEAGE_RULE_ID)?;
        require_contributor_rule(policy, INSERT_REDACTION_STATUS_RULE_ID)?;
        require_contributor_rule(policy, INSERT_SUMMARY_SPAN_PROOF_RULE_ID)?;

        if json_array_empty(&episode.source_events_json) {
            return Err(StoreError::Validation(
                "episode requires source event lineage".into(),
            ));
        }
        validate_summary_spans(&episode.summary, summary_spans, |_| {
            SourceAuthority::Derived
        })
        .map_err(|err| {
            StoreError::Validation(format!(
                "episode summary_spans_json failed {}",
                err.invariant()
            ))
        })?;

        self.pool.execute(
            "INSERT INTO episodes (
                id, trace_id, source_events_json, summary, domains_json, entities_json,
                candidate_meaning, extracted_by_json, confidence, status, summary_spans_json
             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11);",
            params![
                episode.id.to_string(),
                episode.trace_id.to_string(),
                serde_json::to_string(&episode.source_events_json)?,
                episode.summary,
                serde_json::to_string(&episode.domains_json)?,
                serde_json::to_string(&episode.entities_json)?,
                episode.candidate_meaning,
                serde_json::to_string(&episode.extracted_by_json)?,
                episode.confidence,
                episode.status,
                serde_json::to_string(summary_spans)?,
            ],
        )?;

        Ok(())
    }

    /// Fetches an episode by id.
    pub fn get_by_id(&self, id: &EpisodeId) -> StoreResult<Option<EpisodeRecord>> {
        let row = self
            .pool
            .query_row(
                episode_select_sql!("WHERE id = ?1"),
                params![id.to_string()],
                episode_row,
            )
            .optional()?;

        row.map(TryInto::try_into).transpose()
    }

    /// Lists episodes for a trace in deterministic id order.
    pub fn list_by_trace(&self, trace_id: &TraceId) -> StoreResult<Vec<EpisodeRecord>> {
        let mut stmt = self
            .pool
            .prepare(episode_select_sql!("WHERE trace_id = ?1 ORDER BY id"))?;
        let rows = stmt.query_map(params![trace_id.to_string()], episode_row)?;

        let mut episodes = Vec::new();
        for row in rows {
            episodes.push(row?.try_into()?);
        }
        Ok(episodes)
    }
}

#[derive(Debug)]
struct EpisodeRow {
    id: String,
    trace_id: String,
    source_events_json: String,
    summary: String,
    domains_json: String,
    entities_json: String,
    candidate_meaning: Option<String>,
    extracted_by_json: String,
    confidence: f64,
    status: String,
}

fn episode_row(row: &Row<'_>) -> rusqlite::Result<EpisodeRow> {
    Ok(EpisodeRow {
        id: row.get(0)?,
        trace_id: row.get(1)?,
        source_events_json: row.get(2)?,
        summary: row.get(3)?,
        domains_json: row.get(4)?,
        entities_json: row.get(5)?,
        candidate_meaning: row.get(6)?,
        extracted_by_json: row.get(7)?,
        confidence: row.get(8)?,
        status: row.get(9)?,
    })
}

impl TryFrom<EpisodeRow> for EpisodeRecord {
    type Error = StoreError;

    fn try_from(row: EpisodeRow) -> StoreResult<Self> {
        Ok(Self {
            id: row.id.parse()?,
            trace_id: row.trace_id.parse()?,
            source_events_json: serde_json::from_str(&row.source_events_json)?,
            summary: row.summary,
            domains_json: serde_json::from_str(&row.domains_json)?,
            entities_json: serde_json::from_str(&row.entities_json)?,
            candidate_meaning: row.candidate_meaning,
            extracted_by_json: serde_json::from_str(&row.extracted_by_json)?,
            confidence: row.confidence,
            status: row.status,
        })
    }
}

fn json_array_empty(value: &Value) -> bool {
    value.as_array().is_some_and(Vec::is_empty)
}

fn require_policy_final_outcome(policy: &PolicyDecision, surface: &str) -> StoreResult<()> {
    match policy.final_outcome {
        PolicyOutcome::Allow | PolicyOutcome::Warn | PolicyOutcome::BreakGlass => Ok(()),
        PolicyOutcome::Quarantine | PolicyOutcome::Reject => Err(StoreError::Validation(format!(
            "{surface} preflight: composed policy outcome {:?} blocks episode mutation",
            policy.final_outcome,
        ))),
    }
}

fn require_contributor_rule(policy: &PolicyDecision, rule_id: &str) -> StoreResult<()> {
    let contains_rule = policy
        .contributing
        .iter()
        .chain(policy.discarded.iter())
        .any(|contribution| contribution.rule_id.as_str() == rule_id);
    if contains_rule {
        Ok(())
    } else {
        Err(StoreError::Validation(format!(
            "policy decision missing required contributor `{rule_id}`; caller skipped ADR 0026 composition",
        )))
    }
}

/// Build a [`PolicyDecision`] that satisfies [`EpisodeRepo::insert`] inputs
/// for the happy path. Intended for tests and fixtures only.
///
/// Production callers MUST compose [`INSERT_SOURCE_EVENT_LINEAGE_RULE_ID`]
/// and [`INSERT_REDACTION_STATUS_RULE_ID`] from real ADR 0015 / ADR 0027
/// evidence. This helper is exposed unconditionally because integration test
/// crates outside `cortex-store` need the same fixture shape; the
/// `_test_allow` suffix is the contract that documents intent.
#[must_use]
pub fn insert_policy_decision_test_allow() -> PolicyDecision {
    compose_policy_outcomes(
        vec![
            PolicyContribution::new(
                INSERT_SOURCE_EVENT_LINEAGE_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: episode source-event lineage validated",
            )
            .expect("static test contribution is valid"),
            PolicyContribution::new(
                INSERT_REDACTION_STATUS_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: episode redaction status asserted",
            )
            .expect("static test contribution is valid"),
        ],
        None,
    )
}

/// Build a [`PolicyDecision`] that satisfies
/// [`EpisodeRepo::insert_with_summary_spans`] inputs for the happy path.
/// Intended for tests and fixtures only; see
/// [`insert_policy_decision_test_allow`] for the production-caller contract.
///
/// Production callers MUST additionally compose
/// [`INSERT_SUMMARY_SPAN_PROOF_RULE_ID`] from a real ADR 0015 span proof.
#[must_use]
pub fn insert_with_summary_spans_policy_decision_test_allow() -> PolicyDecision {
    compose_policy_outcomes(
        vec![
            PolicyContribution::new(
                INSERT_SOURCE_EVENT_LINEAGE_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: episode source-event lineage validated",
            )
            .expect("static test contribution is valid"),
            PolicyContribution::new(
                INSERT_REDACTION_STATUS_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: episode redaction status asserted",
            )
            .expect("static test contribution is valid"),
            PolicyContribution::new(
                INSERT_SUMMARY_SPAN_PROOF_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: episode summary-span proof composed",
            )
            .expect("static test contribution is valid"),
        ],
        None,
    )
}