pushkin-core 0.2.1

//! Compaction-aware delivered-slice index (spec §7.3): file-backed state
//! keyed on (session id + cwd) recording contract slices already
//! delivered IN FULL, so repeats collapse to a pointer line. Unlike the
//! event log this table is deliberately mutable working state — spec
//! §7.3 requires clearing it on compaction, and horizon expiry models
//! content scrolling out of the agent's context window. Invariants:
//! a truncated first emission is never recorded (the agent never saw
//! it); clearing one scope leaves every other scope intact.

use std::path::Path;

use rusqlite::Connection;
use thiserror::Error;

use crate::events::SessionId;

/// Emissions after which a full delivery is treated as scrolled out of
/// the agent's context and re-emitted in full.
pub const DELIVERY_HORIZON: u64 = 40;

#[derive(Debug, Error)]
pub enum DeliveryError {
    #[error("delivery index storage error: {0}")]
    Storage(#[from] rusqlite::Error),
    #[error("delivery index schema error: {0}")]
    Schema(String),
}

/// How the caller should emit a slice this time.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Delivery {
    /// Emit the full slice (never fully delivered in this scope, or the
    /// prior full delivery has scrolled past the horizon).
    Full,
    /// Collapse to a one-line pointer ("already shown — unchanged").
    Pointer,
}

/// One emission decision request. `complete` states whether THIS
/// emission will carry the slice in full — a truncated emission is
/// never recorded as delivered (the agent never saw it).
#[derive(Debug)]
pub struct DeliveryRequest<'a> {
    pub session: &'a SessionId,
    pub cwd: &'a str,
    pub slice_key: &'a str,
    pub complete: bool,
}

pub struct DeliveryIndex {
    conn: Connection,
}

impl DeliveryIndex {
    /// Opens (creating if needed) the index at `path`, applying pending
    /// schema migrations (shared with the event log — same database).
    ///
    /// # Errors
    /// Returns `DeliveryError` on any `SQLite` or migration failure.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, DeliveryError> {
        let conn = Connection::open(path)?;
        crate::events::apply_migrations(&conn)
            .map_err(|error| DeliveryError::Schema(error.to_string()))?;
        Ok(Self { conn })
    }

    /// Advances the scope's emission counter and decides how to emit
    /// `slice_key`, recording the delivery only for complete emissions.
    ///
    /// # Errors
    /// Returns `DeliveryError::Storage` on any `SQLite` failure.
    pub fn decide(&self, request: &DeliveryRequest<'_>) -> Result<Delivery, DeliveryError> {
        let scope = (request.session.as_str(), request.cwd);
        self.conn.execute(
            "INSERT INTO delivery_counters (session, cwd, emissions) VALUES (?1, ?2, 1)
             ON CONFLICT (session, cwd) DO UPDATE SET emissions = emissions + 1",
            scope,
        )?;
        let emission: u64 = self.conn.query_row(
            "SELECT emissions FROM delivery_counters WHERE session = ?1 AND cwd = ?2",
            scope,
            |row| row.get(0),
        )?;
        let delivered_at: Option<u64> = self
            .conn
            .query_row(
                "SELECT delivered_at_emission FROM delivered_slices
                 WHERE session = ?1 AND cwd = ?2 AND slice_key = ?3",
                (scope.0, scope.1, request.slice_key),
                |row| row.get(0),
            )
            .map(Some)
            .or_else(ignore_missing_row)?;
        if let Some(at) = delivered_at {
            if emission.saturating_sub(at) < DELIVERY_HORIZON {
                return Ok(Delivery::Pointer);
            }
        }
        if request.complete {
            self.conn.execute(
                "INSERT INTO delivered_slices (session, cwd, slice_key, delivered_at_emission)
                 VALUES (?1, ?2, ?3, ?4)
                 ON CONFLICT (session, cwd, slice_key)
                 DO UPDATE SET delivered_at_emission = ?4",
                (scope.0, scope.1, request.slice_key, emission),
            )?;
        }
        Ok(Delivery::Full)
    }

    /// Clears one (session, cwd) scope — the compaction hook (spec §7.3:
    /// the next reference re-emits in full).
    ///
    /// # Errors
    /// Returns `DeliveryError::Storage` on any `SQLite` failure.
    pub fn clear(&self, session: &SessionId, cwd: &str) -> Result<(), DeliveryError> {
        let scope = (session.as_str(), cwd);
        self.conn.execute(
            "DELETE FROM delivered_slices WHERE session = ?1 AND cwd = ?2",
            scope,
        )?;
        self.conn.execute(
            "DELETE FROM delivery_counters WHERE session = ?1 AND cwd = ?2",
            scope,
        )?;
        Ok(())
    }
}

/// Maps "no row" to `None`; every other storage error stays loud (§13).
fn ignore_missing_row(error: rusqlite::Error) -> Result<Option<u64>, rusqlite::Error> {
    if matches!(error, rusqlite::Error::QueryReturnedNoRows) {
        Ok(None)
    } else {
        Err(error)
    }
}