agd_core/write/
journal.rs

1//! The write journal, `journal.jsonl`: `JournalRecord`s appended before and
2//! after each write, the latest record of each entry, review, and settling
3//! interrupted writes from the file's bytes.
4//!
5//! A pending record is synced before its file changes, so every interruption
6//! can be settled: a file holding the post-image was committed, one holding the
7//! pre-image was never changed, and anything else is a conflict that blocks
8//! writes to that file until `agd setup check --resolve`. The journal is the
9//! user's alone (mode 0600 in a 0700 directory).
10
11use std::collections::BTreeMap;
12use std::fs::{self, OpenOptions};
13use std::io::Write;
14use std::path::{Path, PathBuf};
15
16use chrono::Local;
17use serde::{Deserialize, Serialize};
18
19use crate::document::sha256_hex;
20use crate::{AgdError, ErrorCode, JOURNAL_SCHEMA};
21
22use super::engine::{JournalImage, WriteContext, WriteEngine};
23use super::file::{
24    atomic_replace, create_private_directory, read_org_file, replace_all_or_restore,
25};
26use super::patch::StoredPatch;
27
28#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
29#[serde(rename_all = "snake_case")]
30pub enum JournalStatus {
31    Pending,
32    Committed,
33    Unapplied,
34    Conflict,
35    Reviewed,
36    Undone,
37    /// A conflict the user settled with `agd setup check --resolve`; the
38    /// file was kept as it was.
39    Resolved,
40}
41
42/// What a journaled write did, so every surface names a change the same
43/// way without reading its diff. A review, undo or resolve keeps the
44/// operation of the change it settles.
45#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
46#[serde(rename_all = "snake_case")]
47pub enum JournalOperation {
48    Capture,
49    /// A TODO state change.
50    State,
51    /// SCHEDULED set or removed.
52    Schedule,
53    /// DEADLINE set or removed.
54    Deadline,
55    Priority,
56    Tags,
57    /// Checklist boxes checked or unchecked.
58    Checklist,
59    /// Any other edit, or one that changes several of the above.
60    Edit,
61    Refile,
62}
63
64impl JournalOperation {
65    /// The change in words, as `undid capture` or an Agents card says it.
66    pub fn label(self) -> &'static str {
67        match self {
68            Self::Capture => "capture",
69            Self::State => "state change",
70            Self::Schedule => "schedule",
71            Self::Deadline => "deadline",
72            Self::Priority => "priority",
73            Self::Tags => "tags",
74            Self::Checklist => "checklist",
75            Self::Edit => "edit",
76            Self::Refile => "refile",
77        }
78    }
79}
80
81#[derive(Clone, Debug, Deserialize, Serialize)]
82pub struct JournalRecord {
83    pub schema: String,
84    pub id: String,
85    pub status: JournalStatus,
86    pub timestamp: String,
87    pub actor: String,
88    pub project: String,
89    pub file: String,
90    pub item_ref: Option<String>,
91    #[serde(default, skip_serializing_if = "Option::is_none")]
92    pub transaction: Option<String>,
93    /// What the write did; absent from records agd wrote before it
94    /// journaled operations.
95    #[serde(default, skip_serializing_if = "Option::is_none")]
96    pub operation: Option<JournalOperation>,
97    pub pre_hash: String,
98    pub post_hash: String,
99    pub patches: Vec<StoredPatch>,
100    pub diff: String,
101}
102
103impl JournalRecord {
104    /// The agent that made the change, `codex` for `mcp:codex`; None for a
105    /// person's change from the CLI or TUI.
106    pub fn agent(&self) -> Option<&str> {
107        super::engine::agent_name(&self.actor)
108    }
109
110    /// Who made the change as agd names them: the agent (`codex`), or
111    /// `you` for the CLI and TUI.
112    pub fn actor_name(&self) -> &str {
113        self.agent().unwrap_or("you")
114    }
115
116    /// What the change did in words (JournalOperation::label); `change`
117    /// for a record without an operation.
118    pub fn operation_label(&self) -> &'static str {
119        self.operation.map_or("change", JournalOperation::label)
120    }
121}
122
123impl WriteEngine {
124    pub fn review(&self, journal_id: &str) -> Result<(), AgdError> {
125        let records = self.latest_records()?;
126        let mut record = records.get(journal_id).cloned().ok_or_else(|| {
127            AgdError::new(
128                ErrorCode::JournalNotFound,
129                format!("journal entry not found: {journal_id}"),
130            )
131        })?;
132        if record.status != JournalStatus::Committed {
133            return Err(AgdError::new(
134                ErrorCode::InvalidJournalState,
135                format!("journal entry {journal_id} is not awaiting review"),
136            ));
137        }
138        record.status = JournalStatus::Reviewed;
139        record.timestamp = Local::now().to_rfc3339();
140        self.append_record(&record)
141    }
142
143    pub fn journal(&self) -> Result<Vec<JournalRecord>, AgdError> {
144        let mut records: Vec<JournalRecord> = self.latest_records()?.into_values().collect();
145        records.sort_by(|left, right| right.timestamp.cmp(&left.timestamp));
146        Ok(records)
147    }
148
149    /// Writes several files as one journal transaction, in the order given:
150    /// every pending record first, then each file; when a file cannot be
151    /// written the ones before it are restored. Returns the committed
152    /// records in the same order.
153    pub(super) fn commit_transaction(
154        &self,
155        images: &[JournalImage<'_>],
156        operation: JournalOperation,
157        context: &WriteContext,
158    ) -> Result<Vec<JournalRecord>, AgdError> {
159        let transaction = journal_id();
160        let mut records: Vec<JournalRecord> = images
161            .iter()
162            .enumerate()
163            .map(|(position, image)| {
164                pending_record(
165                    format!("{transaction}:{position}"),
166                    Some(transaction.clone()),
167                    image,
168                    operation,
169                    context,
170                )
171            })
172            .collect();
173        for record in &records {
174            self.append_record(record)?;
175        }
176        replace_all_or_restore(images)?;
177        let timestamp = Local::now().to_rfc3339();
178        for record in &mut records {
179            record.status = JournalStatus::Committed;
180            record.timestamp = timestamp.clone();
181            self.append_record(record)?;
182        }
183        Ok(records)
184    }
185
186    pub(super) fn commit_image(
187        &self,
188        image: JournalImage<'_>,
189        operation: JournalOperation,
190        context: &WriteContext,
191    ) -> Result<JournalRecord, AgdError> {
192        let mut record = pending_record(journal_id(), None, &image, operation, context);
193        self.append_record(&record)?;
194        atomic_replace(image.path, image.post_image.as_bytes())?;
195        record.status = JournalStatus::Committed;
196        record.timestamp = Local::now().to_rfc3339();
197        self.append_record(&record)?;
198        Ok(record)
199    }
200
201    pub(super) fn append_record(&self, record: &JournalRecord) -> Result<(), AgdError> {
202        if let Some(parent) = self.journal_path.parent() {
203            create_private_directory(parent)?;
204        }
205        let serialized = serde_json::to_vec(record).map_err(|error| {
206            AgdError::new(
207                ErrorCode::JournalFailed,
208                format!("cannot serialize journal record: {error}"),
209            )
210        })?;
211        let mut options = OpenOptions::new();
212        options.create(true).append(true);
213        #[cfg(unix)]
214        {
215            use std::os::unix::fs::OpenOptionsExt;
216            options.mode(0o600);
217        }
218        let mut file = options.open(&self.journal_path).map_err(AgdError::from)?;
219        file.write_all(&serialized).map_err(AgdError::from)?;
220        file.write_all(b"\n").map_err(AgdError::from)?;
221        file.sync_all().map_err(AgdError::from)
222    }
223
224    pub(super) fn latest_records(&self) -> Result<BTreeMap<String, JournalRecord>, AgdError> {
225        let source = match fs::read_to_string(&self.journal_path) {
226            Ok(source) => source,
227            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
228                return Ok(BTreeMap::new());
229            }
230            Err(error) => return Err(AgdError::from(error)),
231        };
232        let mut records = BTreeMap::new();
233        for (line, raw) in source.lines().enumerate() {
234            let record: JournalRecord = serde_json::from_str(raw).map_err(|error| {
235                AgdError::new(
236                    ErrorCode::InvalidJournal,
237                    format!("invalid journal JSON at line {}: {error}", line + 1),
238                )
239            })?;
240            records.insert(record.id.clone(), record);
241        }
242        Ok(records)
243    }
244
245    /// Settles interrupted writes to `paths`, which the caller holds
246    /// locked: a pending entry whose file holds its post-image was
247    /// committed, one whose file still holds its pre-image never applied.
248    /// Any other pending entry is a conflict only the user can settle, so
249    /// the write is refused with the command that settles it.
250    pub(super) fn recover_files(&self, paths: &[PathBuf]) -> Result<(), AgdError> {
251        for record in self.unsettled_records()? {
252            if !paths.iter().any(|path| Path::new(&record.file) == path) {
253                continue;
254            }
255            let record = self.settle(record)?;
256            if record.status == JournalStatus::Conflict {
257                return Err(journal_conflict_error(&record));
258            }
259        }
260        Ok(())
261    }
262
263    /// Interrupted writes whose file now holds neither their pre- nor their
264    /// post-image. Reads the journal and the files only.
265    pub fn journal_conflicts(&self) -> Result<Vec<JournalRecord>, AgdError> {
266        let mut conflicts = Vec::new();
267        for record in self.unsettled_records()? {
268            if settled_status(&record)? == JournalStatus::Conflict {
269                conflicts.push(record);
270            }
271        }
272        Ok(conflicts)
273    }
274
275    /// Settles every interrupted write: recoverable ones as committed or
276    /// unapplied, and conflicts as resolved, keeping each file exactly as
277    /// it is on disk. Returns the entries marked resolved.
278    pub fn resolve_conflicts(
279        &self,
280        context: &WriteContext,
281    ) -> Result<Vec<JournalRecord>, AgdError> {
282        let mut resolved = Vec::new();
283        for record in self.unsettled_records()? {
284            let _lock = self.lock_target(Path::new(&record.file))?;
285            let mut record = self.settle(record)?;
286            if record.status != JournalStatus::Conflict {
287                continue;
288            }
289            record.status = JournalStatus::Resolved;
290            record.timestamp = Local::now().to_rfc3339();
291            record.actor = context.actor.clone();
292            record.project = context.project.clone();
293            self.append_record(&record)?;
294            resolved.push(record);
295        }
296        Ok(resolved)
297    }
298
299    /// Latest records still pending or in conflict, oldest first.
300    fn unsettled_records(&self) -> Result<Vec<JournalRecord>, AgdError> {
301        let mut records: Vec<JournalRecord> = self
302            .latest_records()?
303            .into_values()
304            .filter(|record| {
305                matches!(
306                    record.status,
307                    JournalStatus::Pending | JournalStatus::Conflict
308                )
309            })
310            .collect();
311        records.sort_by(|left, right| left.timestamp.cmp(&right.timestamp));
312        Ok(records)
313    }
314
315    /// Records the status the file's bytes give an unsettled record.
316    fn settle(&self, mut record: JournalRecord) -> Result<JournalRecord, AgdError> {
317        let status = settled_status(&record)?;
318        if status != record.status {
319            record.status = status;
320            record.timestamp = Local::now().to_rfc3339();
321            self.append_record(&record)?;
322        }
323        Ok(record)
324    }
325}
326
327/// What the file on disk says about an interrupted write: its post-image
328/// means committed, its pre-image unapplied, anything else a conflict.
329fn settled_status(record: &JournalRecord) -> Result<JournalStatus, AgdError> {
330    let current = content_hash(&read_org_file(Path::new(&record.file))?);
331    Ok(if current == record.post_hash {
332        JournalStatus::Committed
333    } else if current == record.pre_hash {
334        JournalStatus::Unapplied
335    } else {
336        JournalStatus::Conflict
337    })
338}
339
340fn journal_conflict_error(record: &JournalRecord) -> AgdError {
341    AgdError::new(
342        ErrorCode::JournalConflict,
343        format!(
344            "{} changed after an interrupted agd write (journal entry {}); check the file, then run `agd setup check --resolve` to keep it as it is",
345            record.file, record.id
346        ),
347    )
348}
349
350fn pending_record(
351    id: String,
352    transaction: Option<String>,
353    image: &JournalImage<'_>,
354    operation: JournalOperation,
355    context: &WriteContext,
356) -> JournalRecord {
357    JournalRecord {
358        schema: JOURNAL_SCHEMA.to_string(),
359        id,
360        status: JournalStatus::Pending,
361        timestamp: Local::now().to_rfc3339(),
362        actor: context.actor.clone(),
363        project: context.project.clone(),
364        file: image.path.to_string_lossy().into_owned(),
365        item_ref: image.item_ref.map(str::to_string),
366        transaction,
367        operation: Some(operation),
368        pre_hash: content_hash(image.pre_image),
369        post_hash: content_hash(image.post_image),
370        patches: image.patches.to_vec(),
371        diff: image.diff.to_string(),
372    }
373}
374
375pub(super) fn content_hash(source: &str) -> String {
376    sha256_hex(source.as_bytes())
377}
378
379fn journal_id() -> String {
380    format!(
381        "{}-{}",
382        Local::now().format("%Y%m%dT%H%M%S%.9f"),
383        std::process::id()
384    )
385}
386
387#[cfg(test)]
388mod tests {
389    use super::*;
390    use crate::write::test_support::{context, first_ref, refs_by_title, settings, state_change};
391
392    /// Appends a pending journal record for `file`, as a write interrupted
393    /// between journaling and committing leaves it.
394    fn append_interrupted_write(journal: &Path, file: &Path, pre: &str, post: &str) -> String {
395        let record = JournalRecord {
396            schema: JOURNAL_SCHEMA.to_string(),
397            id: "20261016T090000.000000000-1".to_string(),
398            status: JournalStatus::Pending,
399            timestamp: "2026-10-16T09:00:00+00:00".to_string(),
400            actor: "test".to_string(),
401            project: "fixture".to_string(),
402            file: fs::canonicalize(file)
403                .unwrap()
404                .to_string_lossy()
405                .into_owned(),
406            item_ref: None,
407            transaction: None,
408            operation: Some(JournalOperation::State),
409            pre_hash: content_hash(pre),
410            post_hash: content_hash(post),
411            patches: Vec::new(),
412            diff: String::new(),
413        };
414        let mut line = serde_json::to_string(&record).unwrap();
415        line.push('\n');
416        let mut existing = fs::read_to_string(journal).unwrap_or_default();
417        existing.push_str(&line);
418        fs::write(journal, existing).unwrap();
419        record.id
420    }
421
422    #[test]
423    fn an_unresolvable_interrupted_write_blocks_only_its_file_until_resolved() {
424        let temp = tempfile::tempdir().unwrap();
425        let journal = temp.path().join("journal.jsonl");
426        let conflicted = temp.path().join("conflicted.org");
427        let other = temp.path().join("other.org");
428        let conflicted_bytes = "* TODO Edited by hand since\n";
429        fs::write(&conflicted, conflicted_bytes).unwrap();
430        fs::write(&other, "* TODO Other\n").unwrap();
431        let id =
432            append_interrupted_write(&journal, &conflicted, "* TODO Before\n", "* DONE Before\n");
433
434        let settings = settings(temp.path());
435        let engine = WriteEngine::new(settings.clone(), journal.clone()).unwrap();
436        assert_eq!(engine.journal().unwrap().len(), 1);
437        let conflicts = engine.journal_conflicts().unwrap();
438        assert_eq!(conflicts.len(), 1);
439        assert_eq!(conflicts[0].id, id);
440
441        let refs = refs_by_title(&settings);
442        let done = state_change("DONE");
443        engine
444            .edit(&refs["Other"], &done, false, &context())
445            .unwrap();
446        let error = engine
447            .edit(&refs["Edited by hand since"], &done, false, &context())
448            .unwrap_err();
449        assert_eq!(error.code().as_str(), "journal_conflict");
450        assert!(
451            error.to_string().contains("agd setup check --resolve"),
452            "{error}"
453        );
454        assert_eq!(fs::read_to_string(&conflicted).unwrap(), conflicted_bytes);
455
456        let resolved = engine.resolve_conflicts(&context()).unwrap();
457        assert_eq!(resolved.len(), 1);
458        assert_eq!(resolved[0].status, JournalStatus::Resolved);
459        assert_eq!(fs::read_to_string(&conflicted).unwrap(), conflicted_bytes);
460        assert!(engine.journal_conflicts().unwrap().is_empty());
461        engine
462            .edit(&refs["Edited by hand since"], &done, false, &context())
463            .unwrap();
464    }
465
466    #[test]
467    fn an_interrupted_write_that_reached_the_file_settles_as_committed() {
468        let temp = tempfile::tempdir().unwrap();
469        let journal = temp.path().join("journal.jsonl");
470        let path = temp.path().join("tasks.org");
471        fs::write(&path, "* DONE Landed\n").unwrap();
472        let id = append_interrupted_write(&journal, &path, "* TODO Landed\n", "* DONE Landed\n");
473        let settings = settings(temp.path());
474        let item_ref = first_ref(&settings);
475        let engine = WriteEngine::new(settings, journal).unwrap();
476        assert!(engine.journal_conflicts().unwrap().is_empty());
477        let status = |engine: &WriteEngine| {
478            engine
479                .journal()
480                .unwrap()
481                .into_iter()
482                .find(|record| record.id == id)
483                .unwrap()
484                .status
485        };
486        assert_eq!(status(&engine), JournalStatus::Pending);
487        let todo = state_change("TODO");
488        engine.edit(&item_ref, &todo, false, &context()).unwrap();
489        assert_eq!(status(&engine), JournalStatus::Committed);
490    }
491}
agd_core/write/journal.rs

agd_core/write/
journal.rs
