pushkin_core/
delivery.rs

1//! Compaction-aware delivered-slice index (spec §7.3): file-backed state
2//! keyed on (session id + cwd) recording contract slices already
3//! delivered IN FULL, so repeats collapse to a pointer line. Unlike the
4//! event log this table is deliberately mutable working state — spec
5//! §7.3 requires clearing it on compaction, and horizon expiry models
6//! content scrolling out of the agent's context window. Invariants:
7//! a truncated first emission is never recorded (the agent never saw
8//! it); clearing one scope leaves every other scope intact.
9
10use std::path::Path;
11
12use rusqlite::Connection;
13use thiserror::Error;
14
15use crate::events::SessionId;
16
17/// Emissions after which a full delivery is treated as scrolled out of
18/// the agent's context and re-emitted in full.
19pub const DELIVERY_HORIZON: u64 = 40;
20
21#[derive(Debug, Error)]
22pub enum DeliveryError {
23    #[error("delivery index storage error: {0}")]
24    Storage(#[from] rusqlite::Error),
25    #[error("delivery index schema error: {0}")]
26    Schema(String),
27}
28
29/// How the caller should emit a slice this time.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub enum Delivery {
32    /// Emit the full slice (never fully delivered in this scope, or the
33    /// prior full delivery has scrolled past the horizon).
34    Full,
35    /// Collapse to a one-line pointer ("already shown — unchanged").
36    Pointer,
37}
38
39/// One emission decision request. `complete` states whether THIS
40/// emission will carry the slice in full — a truncated emission is
41/// never recorded as delivered (the agent never saw it).
42#[derive(Debug)]
43pub struct DeliveryRequest<'a> {
44    pub session: &'a SessionId,
45    pub cwd: &'a str,
46    pub slice_key: &'a str,
47    pub complete: bool,
48}
49
50pub struct DeliveryIndex {
51    conn: Connection,
52}
53
54impl DeliveryIndex {
55    /// Opens (creating if needed) the index at `path`, applying pending
56    /// schema migrations (shared with the event log — same database).
57    ///
58    /// # Errors
59    /// Returns `DeliveryError` on any `SQLite` or migration failure.
60    pub fn open(path: impl AsRef<Path>) -> Result<Self, DeliveryError> {
61        let conn = Connection::open(path)?;
62        crate::events::apply_migrations(&conn)
63            .map_err(|error| DeliveryError::Schema(error.to_string()))?;
64        Ok(Self { conn })
65    }
66
67    /// Advances the scope's emission counter and decides how to emit
68    /// `slice_key`, recording the delivery only for complete emissions.
69    ///
70    /// # Errors
71    /// Returns `DeliveryError::Storage` on any `SQLite` failure.
72    pub fn decide(&self, request: &DeliveryRequest<'_>) -> Result<Delivery, DeliveryError> {
73        let scope = (request.session.as_str(), request.cwd);
74        self.conn.execute(
75            "INSERT INTO delivery_counters (session, cwd, emissions) VALUES (?1, ?2, 1)
76             ON CONFLICT (session, cwd) DO UPDATE SET emissions = emissions + 1",
77            scope,
78        )?;
79        let emission: u64 = self.conn.query_row(
80            "SELECT emissions FROM delivery_counters WHERE session = ?1 AND cwd = ?2",
81            scope,
82            |row| row.get(0),
83        )?;
84        let delivered_at: Option<u64> = self
85            .conn
86            .query_row(
87                "SELECT delivered_at_emission FROM delivered_slices
88                 WHERE session = ?1 AND cwd = ?2 AND slice_key = ?3",
89                (scope.0, scope.1, request.slice_key),
90                |row| row.get(0),
91            )
92            .map(Some)
93            .or_else(ignore_missing_row)?;
94        if let Some(at) = delivered_at {
95            if emission.saturating_sub(at) < DELIVERY_HORIZON {
96                return Ok(Delivery::Pointer);
97            }
98        }
99        if request.complete {
100            self.conn.execute(
101                "INSERT INTO delivered_slices (session, cwd, slice_key, delivered_at_emission)
102                 VALUES (?1, ?2, ?3, ?4)
103                 ON CONFLICT (session, cwd, slice_key)
104                 DO UPDATE SET delivered_at_emission = ?4",
105                (scope.0, scope.1, request.slice_key, emission),
106            )?;
107        }
108        Ok(Delivery::Full)
109    }
110
111    /// Clears one (session, cwd) scope — the compaction hook (spec §7.3:
112    /// the next reference re-emits in full).
113    ///
114    /// # Errors
115    /// Returns `DeliveryError::Storage` on any `SQLite` failure.
116    pub fn clear(&self, session: &SessionId, cwd: &str) -> Result<(), DeliveryError> {
117        let scope = (session.as_str(), cwd);
118        self.conn.execute(
119            "DELETE FROM delivered_slices WHERE session = ?1 AND cwd = ?2",
120            scope,
121        )?;
122        self.conn.execute(
123            "DELETE FROM delivery_counters WHERE session = ?1 AND cwd = ?2",
124            scope,
125        )?;
126        Ok(())
127    }
128}
129
130/// Maps "no row" to `None`; every other storage error stays loud (§13).
131fn ignore_missing_row(error: rusqlite::Error) -> Result<Option<u64>, rusqlite::Error> {
132    if matches!(error, rusqlite::Error::QueryReturnedNoRows) {
133        Ok(None)
134    } else {
135        Err(error)
136    }
137}
pushkin_core/delivery.rs

pushkin_core/
delivery.rs
