agent_doc_sqlite/
op_log.rs

1//! # Module: op_log
2//!
3//! Durable SQLite op-log writer — phase 1 of the operation-scoped drift model
4//! (`#op-scoped-drift-1`, `tasks/agent-doc/plan-operation-scoped-drift.md`).
5//!
6//! ## Spec
7//! - Persists node-keyed document operations (`agent_doc_core::op_log::DocumentOp`)
8//!   tagged with actor + causal (Lamport / session-origin) clock to a derived
9//!   sqlite database at `.agent-doc/op-log.db`.
10//! - The durable store owns Lamport assignment: each appended op gets the next
11//!   monotonic per-document tick, so the log is totally ordered per document.
12//!   The caller's `clock.lamport` is ignored; `clock.origin_session` is honored.
13//! - Append is idempotent against the most recent op for the same node: a
14//!   repeated preflight pass over the same uncommitted diff does not duplicate
15//!   rows or advance the clock.
16//! - The DB is rebuildable derived state; like the archive index it is best
17//!   effort and may be dropped without losing canonical document history.
18//!
19//! ## Agentic Contracts
20//! - The DB lives under the project root that owns `.agent-doc`.
21//! - `append_ops` opens (creating if needed) the DB, ensures the schema, and
22//!   writes the batch in a single transaction.
23//! - `read_ops` returns rows for a document in Lamport order for inspection and
24//!   for the phase-2/3 readers.
25//!
26//! ## Evals
27//! - append_assigns_monotonic_lamport: sequential appends get increasing ticks
28//! - append_is_idempotent_for_repeated_op: re-appending the same op is a no-op
29//! - lamport_is_per_document: independent documents keep independent clocks
30//! - read_ops_returns_lamport_order: rows come back ordered by Lamport tick
31
32use agent_doc_core::op_log::{CausalClock, DocumentOp, OpActor};
33use anyhow::{Context, Result};
34use rusqlite::{Connection, params};
35use std::collections::HashMap;
36use std::path::{Path, PathBuf};
37
38/// Path to the derived op-log database for a project root.
39pub fn db_path(project_root: &Path) -> PathBuf {
40    project_root.join(".agent-doc").join("op-log.db")
41}
42
43fn ensure_schema(conn: &Connection) -> Result<()> {
44    conn.execute_batch(
45        "CREATE TABLE IF NOT EXISTS op_log (
46             id INTEGER PRIMARY KEY AUTOINCREMENT,
47             document_path TEXT NOT NULL,
48             component TEXT NOT NULL,
49             node_key TEXT NOT NULL,
50             item_id TEXT NOT NULL DEFAULT '',
51             op_kind TEXT NOT NULL,
52             actor TEXT NOT NULL,
53             lamport INTEGER NOT NULL,
54             origin_session TEXT,
55             before_preview TEXT,
56             after_preview TEXT,
57             recorded_at TEXT NOT NULL
58         );
59         CREATE INDEX IF NOT EXISTS idx_op_log_document_lamport
60             ON op_log(document_path, lamport);
61         CREATE INDEX IF NOT EXISTS idx_op_log_node
62             ON op_log(document_path, node_key);",
63    )?;
64    Ok(())
65}
66
67/// Highest Lamport tick recorded for a document, or 0 when the log is empty.
68fn max_lamport(conn: &Connection, document_path: &str) -> Result<u64> {
69    let value: Option<i64> = conn.query_row(
70        "SELECT MAX(lamport) FROM op_log WHERE document_path = ?1",
71        params![document_path],
72        |row| row.get(0),
73    )?;
74    Ok(value.unwrap_or(0).max(0) as u64)
75}
76
77/// The latest persisted op for a node, used to suppress duplicate re-appends.
78fn latest_op_for_node(
79    conn: &Connection,
80    document_path: &str,
81    node_key: &str,
82) -> Result<Option<DocumentOp>> {
83    let mut stmt = conn.prepare(
84        "SELECT document_path, component, node_key, item_id, op_kind, actor, lamport,
85                origin_session, before_preview, after_preview, recorded_at
86         FROM op_log
87         WHERE document_path = ?1 AND node_key = ?2
88         ORDER BY lamport DESC, id DESC
89         LIMIT 1",
90    )?;
91    let mut rows = stmt.query(params![document_path, node_key])?;
92    match rows.next()? {
93        Some(row) => Ok(Some(row_to_op(row)?)),
94        None => Ok(None),
95    }
96}
97
98fn row_to_op(row: &rusqlite::Row<'_>) -> Result<DocumentOp> {
99    let actor_str: String = row.get(5)?;
100    let actor = OpActor::from_str_lenient(&actor_str).unwrap_or(OpActor::User);
101    let lamport: i64 = row.get(6)?;
102    Ok(DocumentOp {
103        document_path: row.get(0)?,
104        component: row.get(1)?,
105        node_key: row.get(2)?,
106        // The durable store does not persist the within-component node index;
107        // it is only needed by the live preflight affectedness classifier, never
108        // by a replayed op (`#loop-guard-exchange-node-granularity`).
109        node_index: None,
110        item_id: row.get(3)?,
111        op_kind: row.get(4)?,
112        actor,
113        clock: CausalClock {
114            lamport: lamport.max(0) as u64,
115            origin_session: row.get(7)?,
116        },
117        before_preview: row.get(8)?,
118        after_preview: row.get(9)?,
119        recorded_at: row.get(10)?,
120    })
121}
122
123/// Append a batch of ops, assigning a monotonic per-document Lamport tick to
124/// each new op. Returns the number of ops actually written (deduped ops are
125/// skipped). Best-effort: callers should log and continue on error rather than
126/// failing the cycle.
127pub fn append_ops(project_root: &Path, ops: &[DocumentOp]) -> Result<usize> {
128    if ops.is_empty() {
129        return Ok(0);
130    }
131    let path = db_path(project_root);
132    if let Some(parent) = path.parent()
133        && !parent.exists()
134    {
135        std::fs::create_dir_all(parent)
136            .with_context(|| format!("failed to create {}", parent.display()))?;
137    }
138    let mut conn =
139        Connection::open(&path).with_context(|| format!("failed to open {}", path.display()))?;
140    ensure_schema(&conn)?;
141
142    let tx = conn.transaction()?;
143    // Lazily seed the next Lamport tick per document from the persisted max so a
144    // batch spanning multiple documents keeps independent clocks.
145    let mut next_lamport: HashMap<String, u64> = HashMap::new();
146    let mut written = 0usize;
147    for op in ops {
148        if let Some(previous) = latest_op_for_node(&tx, &op.document_path, &op.node_key)?
149            && previous.same_mutation(op)
150        {
151            // Repeated observation of the same node mutation — keep the log
152            // idempotent and do not advance the clock.
153            continue;
154        }
155        let base = match next_lamport.get(&op.document_path) {
156            Some(value) => *value,
157            None => max_lamport(&tx, &op.document_path)?,
158        };
159        let lamport = base + 1;
160        next_lamport.insert(op.document_path.clone(), lamport);
161
162        let recorded_at = op
163            .recorded_at
164            .clone()
165            .unwrap_or_else(|| timestamp_now().to_string());
166        tx.execute(
167            "INSERT INTO op_log (
168                 document_path, component, node_key, item_id, op_kind, actor,
169                 lamport, origin_session, before_preview, after_preview, recorded_at
170             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
171            params![
172                op.document_path,
173                op.component,
174                op.node_key,
175                op.item_id,
176                op.op_kind,
177                op.actor.as_str(),
178                lamport as i64,
179                op.clock.origin_session,
180                op.before_preview,
181                op.after_preview,
182                recorded_at,
183            ],
184        )?;
185        written += 1;
186    }
187    tx.commit()?;
188    Ok(written)
189}
190
191/// Read persisted ops for a document in Lamport order (most recent last),
192/// capped at `limit` rows (0 = unlimited).
193pub fn read_ops(project_root: &Path, document_path: &str, limit: usize) -> Result<Vec<DocumentOp>> {
194    let path = db_path(project_root);
195    if !path.exists() {
196        return Ok(Vec::new());
197    }
198    let conn =
199        Connection::open(&path).with_context(|| format!("failed to open {}", path.display()))?;
200    ensure_schema(&conn)?;
201    let sql = "SELECT document_path, component, node_key, item_id, op_kind, actor, lamport,
202                      origin_session, before_preview, after_preview, recorded_at
203               FROM op_log
204               WHERE document_path = ?1
205               ORDER BY lamport ASC, id ASC"
206        .to_string();
207    let sql = if limit > 0 {
208        format!("{sql} LIMIT {limit}")
209    } else {
210        sql
211    };
212    let mut stmt = conn.prepare(&sql)?;
213    let mut rows = stmt.query(params![document_path])?;
214    let mut ops = Vec::new();
215    while let Some(row) = rows.next()? {
216        ops.push(row_to_op(row)?);
217    }
218    Ok(ops)
219}
220
221fn timestamp_now() -> u64 {
222    use std::time::{SystemTime, UNIX_EPOCH};
223    SystemTime::now()
224        .duration_since(UNIX_EPOCH)
225        .map(|d| d.as_secs())
226        .unwrap_or_default()
227}
228
229#[cfg(test)]
230mod tests {
231    use super::*;
232    use tempfile::TempDir;
233
234    fn project_with_agent_doc() -> TempDir {
235        let dir = TempDir::new().unwrap();
236        std::fs::create_dir_all(dir.path().join(".agent-doc")).unwrap();
237        dir
238    }
239
240    fn op(doc: &str, node_key: &str, op_kind: &str, after: &str) -> DocumentOp {
241        DocumentOp {
242            document_path: doc.to_string(),
243            component: "queue".to_string(),
244            node_key: node_key.to_string(),
245            node_index: None,
246            item_id: node_key.to_string(),
247            op_kind: op_kind.to_string(),
248            actor: OpActor::User,
249            clock: CausalClock {
250                lamport: 0,
251                origin_session: Some("sess-1".to_string()),
252            },
253            before_preview: None,
254            after_preview: Some(after.to_string()),
255            recorded_at: Some("100".to_string()),
256        }
257    }
258
259    #[test]
260    fn append_assigns_monotonic_lamport() {
261        let dir = project_with_agent_doc();
262        let written = append_ops(
263            dir.path(),
264            &[
265                op("plan.md", "queue:0:a:0", "insert", "- do [#a]"),
266                op("plan.md", "queue:0:b:0", "insert", "- do [#b]"),
267            ],
268        )
269        .unwrap();
270        assert_eq!(written, 2);
271        let ops = read_ops(dir.path(), "plan.md", 0).unwrap();
272        assert_eq!(ops.len(), 2);
273        assert_eq!(ops[0].clock.lamport, 1);
274        assert_eq!(ops[1].clock.lamport, 2);
275        assert_eq!(ops[0].clock.origin_session.as_deref(), Some("sess-1"));
276
277        // A later append keeps climbing from the persisted max.
278        let written = append_ops(
279            dir.path(),
280            &[op("plan.md", "queue:0:c:0", "insert", "- do [#c]")],
281        )
282        .unwrap();
283        assert_eq!(written, 1);
284        let ops = read_ops(dir.path(), "plan.md", 0).unwrap();
285        assert_eq!(ops.len(), 3);
286        assert_eq!(ops[2].clock.lamport, 3);
287    }
288
289    #[test]
290    fn append_is_idempotent_for_repeated_op() {
291        let dir = project_with_agent_doc();
292        let same = op("plan.md", "queue:0:a:0", "insert", "- do [#a]");
293        assert_eq!(append_ops(dir.path(), &[same.clone()]).unwrap(), 1);
294        // Re-observe the identical node mutation: no new row, clock unchanged.
295        assert_eq!(append_ops(dir.path(), &[same.clone()]).unwrap(), 0);
296        let ops = read_ops(dir.path(), "plan.md", 0).unwrap();
297        assert_eq!(ops.len(), 1);
298        assert_eq!(ops[0].clock.lamport, 1);
299
300        // A genuinely different mutation on the same node is still recorded.
301        let edited = op("plan.md", "queue:0:a:0", "replace", "- do [#a] edited");
302        assert_eq!(append_ops(dir.path(), &[edited]).unwrap(), 1);
303        let ops = read_ops(dir.path(), "plan.md", 0).unwrap();
304        assert_eq!(ops.len(), 2);
305        assert_eq!(ops[1].clock.lamport, 2);
306    }
307
308    #[test]
309    fn lamport_is_per_document() {
310        let dir = project_with_agent_doc();
311        append_ops(
312            dir.path(),
313            &[
314                op("a.md", "queue:0:x:0", "insert", "- do [#x]"),
315                op("b.md", "queue:0:y:0", "insert", "- do [#y]"),
316                op("a.md", "queue:0:z:0", "insert", "- do [#z]"),
317            ],
318        )
319        .unwrap();
320        let a = read_ops(dir.path(), "a.md", 0).unwrap();
321        let b = read_ops(dir.path(), "b.md", 0).unwrap();
322        assert_eq!(
323            a.iter().map(|o| o.clock.lamport).collect::<Vec<_>>(),
324            vec![1, 2]
325        );
326        assert_eq!(
327            b.iter().map(|o| o.clock.lamport).collect::<Vec<_>>(),
328            vec![1]
329        );
330    }
331
332    #[test]
333    fn read_ops_returns_empty_without_db() {
334        let dir = project_with_agent_doc();
335        assert!(read_ops(dir.path(), "missing.md", 0).unwrap().is_empty());
336    }
337}
agent_doc_sqlite/op_log.rs

agent_doc_sqlite/
op_log.rs
