dbmd_core/log.rs
1//! `log` — the append-only, month-rotating chronological log.
2//!
3//! One logical timeline: the active `log.md` at the store root plus
4//! `log/<YYYY-MM>.md` archives. [`Log::append`] rolls older months into
5//! archives on write so the active file stays current-month. [`Log::tail`] and
6//! [`Log::since`] **reverse-read from EOF**. Both read each file they touch in
7//! full — the on-disk order is not guaranteed monotonic, so neither can
8//! early-stop within a file — and select by timestamp: `tail` keeps the `n`
9//! newest, `since` keeps everything newer than the cutoff. Both cross into
10//! month archives only as far back as the requested window reaches (by the
11//! cutoff's month for `since`, by the current `n`th-newest's month for `tail`)
12//! — never the whole history.
13//!
14//! Append-only contract: there is no rewrite API. Corrective entries go on the
15//! end; out-of-order timestamps are a validate warning (`LOG_OUT_OF_ORDER`),
16//! signalling a probable rewrite.
17
18use std::collections::BTreeMap;
19use std::fs::{self, File};
20use std::io::{Read, Seek, SeekFrom};
21use std::path::{Path, PathBuf};
22
23use chrono::{DateTime, Datelike, FixedOffset, NaiveDateTime, TimeZone, Utc};
24
25use crate::store::Store;
26
27/// The on-disk header timestamp format: `YYYY-MM-DD HH:MM` (minute precision,
28/// no timezone). Parsing reattaches UTC; emitting renders the entry's own
29/// wall-clock, so a read→write→read round-trip is stable at minute precision.
30const TS_FORMAT: &str = "%Y-%m-%d %H:%M";
31
32/// The frontmatter block written when the active `log.md` is created.
33const LOG_FRONTMATTER: &str = "---\ntype: log\n---\n\n# Curator log\n";
34
35/// Block size for the backward (reverse-from-EOF) reader.
36const REVERSE_BLOCK: usize = 8 * 1024;
37
38/// A recognized `log.md` entry kind. Custom kinds are valid in the format
39/// (`dbmd validate` warns on unrecognized via `LOG_UNKNOWN_KIND`); this enum
40/// carries the recognized vocabulary plus a [`LogKind::Custom`] catch-all so an
41/// unknown kind round-trips without loss.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub enum LogKind {
44 /// A source artifact was ingested.
45 Ingest,
46 /// A file was created.
47 Create,
48 /// A file was updated.
49 Update,
50 /// A file was deleted.
51 Delete,
52 /// A file was renamed/moved.
53 Rename,
54 /// A wiki-link was added.
55 Link,
56 /// A validation pass ran.
57 Validate,
58 /// The index was rebuilt.
59 IndexRebuild,
60 /// A contradiction between sources was flagged.
61 Contradiction,
62 /// Any kind outside the recognized vocabulary, preserved verbatim.
63 Custom(String),
64}
65
66impl LogKind {
67 /// The canonical lowercase string for this kind, as it appears in a log
68 /// header (`ingest`, `index-rebuild`, …).
69 pub fn as_str(&self) -> &str {
70 match self {
71 LogKind::Ingest => "ingest",
72 LogKind::Create => "create",
73 LogKind::Update => "update",
74 LogKind::Delete => "delete",
75 LogKind::Rename => "rename",
76 LogKind::Link => "link",
77 LogKind::Validate => "validate",
78 LogKind::IndexRebuild => "index-rebuild",
79 LogKind::Contradiction => "contradiction",
80 LogKind::Custom(s) => s,
81 }
82 }
83
84 /// Parse a kind from its header token; non-canonical tokens become
85 /// [`LogKind::Custom`].
86 pub fn parse(token: &str) -> LogKind {
87 match token {
88 "ingest" => LogKind::Ingest,
89 "create" => LogKind::Create,
90 "update" => LogKind::Update,
91 "delete" => LogKind::Delete,
92 "rename" => LogKind::Rename,
93 "link" => LogKind::Link,
94 "validate" => LogKind::Validate,
95 "index-rebuild" => LogKind::IndexRebuild,
96 "contradiction" => LogKind::Contradiction,
97 other => LogKind::Custom(other.to_string()),
98 }
99 }
100
101 /// True if this is one of the recognized kinds (i.e. not
102 /// [`LogKind::Custom`]).
103 pub fn is_recognized(&self) -> bool {
104 !matches!(self, LogKind::Custom(_))
105 }
106}
107
108/// One parsed `log.md` entry: a header
109/// (`## [YYYY-MM-DD HH:MM] <kind> | <object>`) plus its body.
110#[derive(Debug, Clone, PartialEq, Eq)]
111pub struct LogEntry {
112 /// The entry timestamp from the header.
113 pub timestamp: DateTime<FixedOffset>,
114 /// The entry kind.
115 pub kind: LogKind,
116 /// The object slot — a store-relative path/wiki-link target, or `None` for
117 /// store-wide actions like `validate`.
118 pub object: Option<String>,
119 /// The free-form body (one or more lines) explaining what happened.
120 pub note: String,
121}
122
123impl LogEntry {
124 /// Render this entry as it appears on disk: the `## [...]` header line,
125 /// then the note body, then a trailing blank line so successive entries are
126 /// separated. The note is emitted with header-shaped continuation lines
127 /// **escaped** (see [`escape_note_line`]) so a note line that happens to
128 /// match the entry-header shape (`## [YYYY-MM-DD HH:MM] <kind> | <obj>`) can
129 /// never be mistaken for a real entry header on readback or on the next
130 /// rotation. The escape round-trips exactly through [`unescape_note_line`].
131 fn render(&self) -> String {
132 let ts = self.timestamp.format(TS_FORMAT);
133 let mut out = String::new();
134 match &self.object {
135 Some(obj) => {
136 out.push_str(&format!("## [{}] {} | {}\n", ts, self.kind.as_str(), obj));
137 }
138 None => {
139 out.push_str(&format!("## [{}] {}\n", ts, self.kind.as_str()));
140 }
141 }
142 // Trim only the structural line terminators (`\n`/`\r`) — the trailing
143 // blank line separating entries is appended below, so a note's own
144 // trailing newlines would otherwise stack up and shift on every
145 // re-render. Spaces and tabs are legitimate note *content* and must be
146 // preserved verbatim, so the round-trip is exact: readback
147 // (`parse_entries`) trims the same `['\n', '\r']` set and no more, and a
148 // note ending in a space (`"note 0 "`) must reconstruct unchanged.
149 let note = self.note.trim_end_matches(['\n', '\r']);
150 if !note.is_empty() {
151 // Escape per line: a note line that parses as an entry header is
152 // prefixed so it is no longer at column 0 as `## [` — it stays note
153 // body on readback and on rotation, never a fabricated entry.
154 for (i, line) in note.split('\n').enumerate() {
155 if i > 0 {
156 out.push('\n');
157 }
158 out.push_str(&escape_note_line(line));
159 }
160 out.push('\n');
161 }
162 out.push('\n');
163 out
164 }
165
166 /// The `(year, month)` of this entry's wall-clock timestamp — the rotation
167 /// bucket.
168 fn year_month(&self) -> (i32, u32) {
169 (self.timestamp.year(), self.timestamp.month())
170 }
171}
172
173/// The store's chronological log: a thin handle for the append-only timeline.
174/// All methods take the [`Store`] so they resolve the active `log.md` and the
175/// `log/` archives under the store root.
176#[derive(Debug, Clone)]
177pub struct Log;
178
179impl Log {
180 /// Atomically append `entry` to the active `log.md`, creating it (with
181 /// `type: log` frontmatter) if absent. **If the active log holds entries
182 /// from a prior month, roll those older months into `log/<YYYY-MM>.md`
183 /// first** (atomic move), keeping the active file to the current month.
184 ///
185 /// **Concurrency.** `append` is a read-modify-write of the whole active file
186 /// (`write_atomic` is atomic at the file level, but the read→render→write
187 /// window is not). Two concurrent appenders — the manager and a cron-driven
188 /// background system, say — would otherwise both read the same N-entry
189 /// snapshot and each write N+1 entries, the second rename clobbering the
190 /// first and silently dropping an audit entry. We serialize the whole
191 /// read-modify-write under an advisory file lock (`flock`, held for the
192 /// duration) so concurrent appends queue instead of racing. The lock is
193 /// advisory and process-scoped; it guards the toolkit's own appends, which is
194 /// the realistic contention path.
195 pub fn append(store: &Store, entry: &LogEntry) -> crate::Result<()> {
196 let active = active_log_path(store);
197
198 // Serialize concurrent appends for the whole read-modify-write. Held
199 // until `_lock` drops at function exit (covering both the rotation and
200 // the plain-append paths). A lock failure is non-fatal: we proceed
201 // unlocked rather than refuse to log (best-effort, same posture as the
202 // pre-fix behaviour on platforms without advisory locks).
203 let _lock = AppendLock::acquire(&active);
204
205 // Read the active file's current contents (if any). The "current month"
206 // is the month of the entry being appended (the newest in the timeline);
207 // every existing entry from a strictly-earlier month rolls to archives.
208 let current_ym = entry.year_month();
209
210 if active.exists() {
211 let content = fs::read_to_string(&active)?;
212 let (header, entries) = parse_active(&content);
213
214 // Partition existing entries into prior-month (roll out) and
215 // current-or-later (keep in the active file).
216 let mut by_month: BTreeMap<(i32, u32), Vec<LogEntry>> = BTreeMap::new();
217 let mut keep: Vec<LogEntry> = Vec::new();
218 for e in entries {
219 if e.year_month() < current_ym {
220 by_month.entry(e.year_month()).or_default().push(e);
221 } else {
222 keep.push(e);
223 }
224 }
225
226 if !by_month.is_empty() {
227 // Roll each prior month into its archive (atomic per-file),
228 // appending to any existing archive for that month.
229 let dir = archive_dir(store);
230 fs::create_dir_all(&dir)?;
231 for ((y, m), month_entries) in &by_month {
232 let path = archive_path(store, *y, *m);
233 append_to_archive(&path, month_entries)?;
234 }
235
236 // Rewrite the active file to the kept (current-month) entries
237 // plus the new entry — atomically.
238 let mut body = String::new();
239 for e in &keep {
240 body.push_str(&e.render());
241 }
242 body.push_str(&entry.render());
243 let full = compose_active(&header, &body);
244 crate::fsx::write_atomic(&active, full.as_bytes())?;
245 return Ok(());
246 }
247
248 // No rotation needed: plain atomic append of the rendered entry.
249 let mut full = content;
250 if !full.ends_with('\n') {
251 full.push('\n');
252 }
253 full.push_str(&entry.render());
254 crate::fsx::write_atomic(&active, full.as_bytes())?;
255 Ok(())
256 } else {
257 // Fresh log: frontmatter + the single entry.
258 if let Some(parent) = active.parent() {
259 fs::create_dir_all(parent)?;
260 }
261 let body = entry.render();
262 let full = compose_active(LOG_FRONTMATTER, &body);
263 crate::fsx::write_atomic(&active, full.as_bytes())?;
264 Ok(())
265 }
266 }
267
268 /// The `n` most-recent entries **by timestamp**, returned oldest→newest.
269 ///
270 /// **Out-of-order safety (mirrors [`Log::since`]).** The log is append-only
271 /// but *not* guaranteed to be in non-decreasing timestamp order on disk: a
272 /// corrective entry is appended below the entry it corrects, a
273 /// backdated/clock-skewed write lands physically after newer entries, and a
274 /// `merge=union` clone merge interleaves both sides until a later agent
275 /// reorders. Out-of-order is only a `LOG_OUT_OF_ORDER` warning, never
276 /// rejected. So the last `n` *physical* entries are **not** the `n` newest
277 /// by time — taking them would omit a genuinely-recent entry that sits
278 /// physically before an older one, and the documented curator warm-up
279 /// (`dbmd log tail 20`) would report a stale picture of what was done lately.
280 /// We therefore feed every entry of each file we touch through a bounded
281 /// newest-by-timestamp window and let it select the true top `n`.
282 ///
283 /// Bounded cost: the active `log.md` is kept to the current month by
284 /// rotation, so a full read of it is cheap and is not a whole-store walk.
285 /// Across archives we *can* prune: each `log/<YYYY-MM>.md` holds only entries
286 /// from that month (rotation buckets by the entry's own year-month), so once
287 /// the window is full, an archive whose month is strictly before the
288 /// window-minimum's month cannot contain any entry newer than the current
289 /// `n`th-newest. We cross archives newest-month-first and stop at the first
290 /// such archive.
291 pub fn tail(store: &Store, n: usize) -> crate::Result<Vec<LogEntry>> {
292 if n == 0 {
293 return Ok(Vec::new());
294 }
295
296 // A bounded window of the `n` entries with the largest timestamps. No
297 // within-file early stop: out-of-order entries mean a newer entry can
298 // sit physically before an older one, so each file is read fully.
299 let mut window = NewestWindow::new(n);
300 // Cross-file identity dedup (see `since`): an interrupted rotation can
301 // leave the same entry in both the untrimmed active file and the
302 // archive; without this the duplicate would occupy two window slots and
303 // surface twice. The active copy (scanned first) is the one kept.
304 let mut seen: std::collections::HashSet<EntryKey> = std::collections::HashSet::new();
305
306 // Active file: scan fully (current-month-bounded by rotation).
307 let active = active_log_path(store);
308 if active.exists() {
309 reverse_collect(&active, |e| {
310 if seen.insert(entry_key(&e)) {
311 window.consider(e);
312 }
313 false
314 })?;
315 }
316
317 // Archives, newest-month-first. Once the window is full, an archive
318 // whose month is strictly before the window-minimum's month holds only
319 // entries older than the current cutoff, so it (and every older archive)
320 // is skippable.
321 for archive in list_archives_desc(store)? {
322 if let (true, Some(cutoff_ym), Some(arch_ym)) = (
323 window.is_full(),
324 window.min_year_month(),
325 archive_year_month(&archive),
326 ) {
327 if arch_ym < cutoff_ym {
328 break;
329 }
330 }
331 reverse_collect(&archive, |e| {
332 if seen.insert(entry_key(&e)) {
333 window.consider(e);
334 }
335 false
336 })?;
337 }
338
339 Ok(window.into_sorted())
340 }
341
342 /// Entries strictly newer than `time`, reverse-scanning active → archives.
343 ///
344 /// **No within-file early stop.** The log is append-only but *not*
345 /// guaranteed to be in non-decreasing timestamp order on disk: a corrective
346 /// entry is appended below the entry it corrects (SPEC: "if a finding is
347 /// wrong, append a corrective entry below it"), a backdated/clock-skewed
348 /// write lands physically after newer entries, and a `merge=union` clone
349 /// merge interleaves both sides until a later agent reorders. Out-of-order
350 /// is only a `LOG_OUT_OF_ORDER` warning, never rejected. So a newer entry
351 /// can sit physically *before* an older one; stopping at the first
352 /// older-than-`time` entry would silently drop those — the documented
353 /// curator warm-up (`dbmd log since <ts>`) would miss real recent work.
354 /// We therefore read every entry of each file we touch.
355 ///
356 /// Bounded cost: the active `log.md` is kept to the current month by
357 /// rotation, so a full read of it is cheap (the same read `tail` does for a
358 /// large `n`) and is not a whole-store walk. Across archives we *can* stop:
359 /// each `log/<YYYY-MM>.md` holds only entries from that month (rotation
360 /// buckets by the entry's own year-month), so an archive whose month is
361 /// strictly before `time`'s month cannot contain any entry newer than
362 /// `time`. We cross archives newest-month-first and stop at the first whose
363 /// month is entirely at or before `time`'s.
364 pub fn since(store: &Store, time: DateTime<FixedOffset>) -> crate::Result<Vec<LogEntry>> {
365 let mut collected: Vec<LogEntry> = Vec::new();
366 // Cross-file identity dedup. An interrupted rotation (archive write
367 // committed, active rewrite not) leaves the same entries in BOTH the
368 // untrimmed active file and the archive; without dedup every such entry
369 // comes back twice. Keyed on the full entry identity, so only a
370 // byte-identical duplicate is suppressed (the active copy, scanned first,
371 // is the one kept); two genuinely-distinct entries never collide.
372 let mut seen: std::collections::HashSet<EntryKey> = std::collections::HashSet::new();
373
374 // Active file: scan fully, no early stop (out-of-order safe).
375 let active = active_log_path(store);
376 if active.exists() {
377 reverse_collect(&active, |e| {
378 if e.timestamp > time && seen.insert(entry_key(&e)) {
379 collected.push(e);
380 }
381 false
382 })?;
383 }
384
385 // The cutoff's own (year, month): any archive strictly before it holds
386 // only older entries and is skippable. Archive months are bucketed on
387 // the UTC calendar (on-disk timestamps are offset-free and re-read as
388 // UTC; rotation buckets by the entry's UTC year-month), so the pruning
389 // calendar must be UTC too. A non-UTC `since` offset (advertised in the
390 // CLI hint, e.g. `…T00:30:00+07:00`) whose local month differs from its
391 // UTC month would otherwise prune away an archive holding entries that
392 // are strictly newer than `time` — `time.year()/.month()` read the
393 // offset-LOCAL calendar, not UTC.
394 let cutoff_utc = time.with_timezone(&Utc);
395 let cutoff_ym = (cutoff_utc.year(), cutoff_utc.month());
396
397 for archive in list_archives_desc(store)? {
398 // Archives are newest-month-first; once a month is strictly before
399 // the cutoff's month, every remaining (older) archive is too.
400 if let Some(arch_ym) = archive_year_month(&archive) {
401 if arch_ym < cutoff_ym {
402 break;
403 }
404 }
405 // Scan this archive fully — within a month, entries may still be
406 // out of order, so no within-file early stop.
407 reverse_collect(&archive, |e| {
408 if e.timestamp > time && seen.insert(entry_key(&e)) {
409 collected.push(e);
410 }
411 false
412 })?;
413 }
414
415 collected.reverse();
416 Ok(collected)
417 }
418
419 /// The timestamp of the most recent `validate` entry — the default `since`
420 /// window for working-set validation ([`crate::validate::validate_working_set`]).
421 pub fn last_validate_at(store: &Store) -> crate::Result<Option<DateTime<FixedOffset>>> {
422 let mut found: Option<DateTime<FixedOffset>> = None;
423
424 let active = active_log_path(store);
425 if active.exists() {
426 reverse_collect(&active, |e| {
427 if e.kind == LogKind::Validate {
428 found = Some(e.timestamp);
429 true
430 } else {
431 false
432 }
433 })?;
434 }
435
436 if found.is_none() {
437 for archive in list_archives_desc(store)? {
438 reverse_collect(&archive, |e| {
439 if e.kind == LogKind::Validate {
440 found = Some(e.timestamp);
441 true
442 } else {
443 false
444 }
445 })?;
446 if found.is_some() {
447 break;
448 }
449 }
450 }
451
452 Ok(found)
453 }
454
455 /// Parse a single entry header (`## [YYYY-MM-DD HH:MM] <kind> | <object>`)
456 /// into its timestamp, kind, and object. Returns `None` if the line isn't a
457 /// well-formed entry header.
458 pub fn parse_header(line: &str) -> Option<(DateTime<FixedOffset>, LogKind, Option<String>)> {
459 let line = line.trim_end_matches(['\n', '\r']);
460 let rest = line.strip_prefix("## [")?;
461 let close = rest.find(']')?;
462 let ts_str = &rest[..close];
463 let timestamp = parse_timestamp(ts_str)?;
464
465 // Everything after the closing bracket: ` <kind> | <object>` or
466 // ` <kind>`.
467 let after = rest[close + 1..].trim();
468 if after.is_empty() {
469 return None;
470 }
471
472 let (kind_str, object) = match after.split_once('|') {
473 Some((k, o)) => {
474 let obj = o.trim();
475 let obj = if obj.is_empty() {
476 None
477 } else {
478 Some(obj.to_string())
479 };
480 (k.trim(), obj)
481 }
482 None => (after, None),
483 };
484
485 if kind_str.is_empty() {
486 return None;
487 }
488
489 Some((timestamp, LogKind::parse(kind_str), object))
490 }
491}
492
493// ── Internal helpers ────────────────────────────────────────────────────────
494
495/// A bounded window of the `n` entries with the largest timestamps, fed by a
496/// **reverse (newest-physical-first) scan** and used by [`Log::tail`].
497///
498/// Why this exists: the last `n` *physical* entries are the `n` newest only
499/// when the log is in non-decreasing time order. That's the append-only
500/// contract, not a guarantee — a backdated, clock-skewed, or merge-interleaved
501/// entry violates it (and trips the `LOG_OUT_OF_ORDER` validate warning). The
502/// window decouples `tail` from that assumption: it keeps the `n` largest
503/// timestamps seen regardless of the order they arrive in, so the caller can
504/// read each file fully (no fragile within-file early stop) and still get the
505/// true top `n`.
506///
507/// Tie-break: entries sharing a timestamp at the window boundary are ordered by
508/// **physical recency** — the one appended later (encountered earlier in the
509/// reverse scan, i.e. a smaller `arrival`) wins. "Newest" means most-recently
510/// recorded.
511struct NewestWindow {
512 cap: usize,
513 /// Min-by-(timestamp, then physical-oldest) heap: the root is always the
514 /// next entry to evict once the window is full.
515 heap: std::collections::BinaryHeap<WindowItem>,
516 /// Count of entries fed in, in reverse-scan order, used as the tie-break
517 /// key (0 = newest physical).
518 next_arrival: u64,
519}
520
521impl NewestWindow {
522 fn new(cap: usize) -> Self {
523 NewestWindow {
524 cap,
525 heap: std::collections::BinaryHeap::with_capacity(cap),
526 next_arrival: 0,
527 }
528 }
529
530 /// Offer one entry from the scan. If the window isn't full it's kept; once
531 /// full, it's kept (evicting the current minimum) iff its timestamp is `>=`
532 /// the window minimum. Equal-timestamp boundary entries resolve by physical
533 /// recency (see the type doc).
534 fn consider(&mut self, entry: LogEntry) {
535 let arrival = self.next_arrival;
536 self.next_arrival += 1;
537
538 if self.heap.len() < self.cap {
539 self.heap.push(WindowItem { entry, arrival });
540 return;
541 }
542
543 // Window full. The heap root is the current minimum (oldest-by-
544 // timestamp held; on a tie, the oldest-physical).
545 let root = self.heap.peek().expect("full window has a root");
546 if entry.timestamp > root.entry.timestamp {
547 // Strictly newer than the window minimum: it belongs; evict the min.
548 self.heap.pop();
549 self.heap.push(WindowItem { entry, arrival });
550 }
551 // On `<=` we keep the window as-is. `<` is plainly too old. `==` is the
552 // tie case: the scan is newest-physical-first, so this entry is
553 // physically *older* than the held one of equal timestamp, and the
554 // tie-break keeps the physically-newer (most-recently-recorded) entry —
555 // so the incoming one is dropped.
556 }
557
558 /// Whether the window already holds its full `cap` entries.
559 fn is_full(&self) -> bool {
560 self.heap.len() >= self.cap
561 }
562
563 /// The `(year, month)` of the window's current minimum (oldest kept) entry,
564 /// or `None` when the window is empty. Used to prune older archives: an
565 /// archive month strictly before this can't beat the current cutoff.
566 fn min_year_month(&self) -> Option<(i32, u32)> {
567 self.heap
568 .peek()
569 .map(|item| (item.entry.timestamp.year(), item.entry.timestamp.month()))
570 }
571
572 /// The held entries, oldest→newest (chronological), ties broken
573 /// oldest-physical→newest-physical.
574 fn into_sorted(self) -> Vec<LogEntry> {
575 let mut items: Vec<WindowItem> = self.heap.into_vec();
576 // Ascending by timestamp; on a tie, oldest-physical (larger arrival)
577 // first so the most-recently-recorded entry sorts last.
578 items.sort_by(|a, b| {
579 a.entry
580 .timestamp
581 .cmp(&b.entry.timestamp)
582 .then(b.arrival.cmp(&a.arrival))
583 });
584 items.into_iter().map(|i| i.entry).collect()
585 }
586}
587
588/// One slot in [`NewestWindow`]'s heap. `Ord` is defined so the heap is a
589/// **min-heap on `(timestamp, physical-oldest)`**: `BinaryHeap` is a max-heap,
590/// so the root (max under this `Ord`) is the eviction candidate — the smallest
591/// timestamp, and on a tie the oldest-physical (largest `arrival`).
592struct WindowItem {
593 entry: LogEntry,
594 arrival: u64,
595}
596
597impl PartialEq for WindowItem {
598 fn eq(&self, other: &Self) -> bool {
599 self.entry.timestamp == other.entry.timestamp && self.arrival == other.arrival
600 }
601}
602impl Eq for WindowItem {}
603
604impl Ord for WindowItem {
605 fn cmp(&self, other: &Self) -> std::cmp::Ordering {
606 // Reverse on timestamp so the *smallest* timestamp is the heap max
607 // (eviction candidate). On equal timestamps, the larger `arrival`
608 // (older physical) is the heap max so it is evicted first.
609 other
610 .entry
611 .timestamp
612 .cmp(&self.entry.timestamp)
613 .then(self.arrival.cmp(&other.arrival))
614 }
615}
616impl PartialOrd for WindowItem {
617 fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
618 Some(self.cmp(other))
619 }
620}
621
622/// An advisory, exclusive lock serializing concurrent [`Log::append`] calls.
623///
624/// Held on a dedicated sibling lock file (`<active>.lock`) rather than on
625/// `log.md` itself: `write_atomic` replaces the active file by `rename`, so the
626/// active inode changes under us and a lock on its fd would not cover the new
627/// file. The lock file is stable, so the lock spans the whole read-modify-write.
628///
629/// On Unix this is `flock(LOCK_EX)`, released on drop (or implicitly when the
630/// process exits / the fd closes, so a crash never strands the lock). The
631/// lock file is created if absent and intentionally left on disk between runs
632/// (locking it does not depend on its contents). On non-Unix targets the lock
633/// is a no-op — db.md's append surface is Unix-targeted, and a missing advisory
634/// lock degrades to the pre-fix last-writer-wins, never to incorrectness of a
635/// single writer.
636struct AppendLock {
637 #[cfg(unix)]
638 file: Option<File>,
639}
640
641impl AppendLock {
642 /// Acquire the exclusive append lock for the store whose active log is
643 /// `active`. Best-effort: any failure to open or lock the lock file yields
644 /// an unlocked guard (we log rather than refuse to log). Blocks until the
645 /// lock is granted when another appender holds it.
646 fn acquire(active: &Path) -> AppendLock {
647 #[cfg(unix)]
648 {
649 let file = Self::open_and_lock(active);
650 AppendLock { file }
651 }
652 #[cfg(not(unix))]
653 {
654 let _ = active;
655 AppendLock {}
656 }
657 }
658
659 #[cfg(unix)]
660 fn open_and_lock(active: &Path) -> Option<File> {
661 use std::os::unix::io::AsRawFd;
662
663 // The lock file lives beside the active log; ensure its parent exists
664 // (the fresh-log path may run before `log.md`'s directory is created).
665 if let Some(parent) = active.parent() {
666 let _ = fs::create_dir_all(parent);
667 }
668 let lock_path = lock_path_for(active);
669 let file = std::fs::OpenOptions::new()
670 .create(true)
671 .truncate(false)
672 .write(true)
673 .open(&lock_path)
674 .ok()?;
675
676 // Blocking exclusive advisory lock. `flock` is in libc, which every Rust
677 // binary links, so the bare `extern "C"` declaration needs no crate dep.
678 let rc = unsafe { flock(file.as_raw_fd(), LOCK_EX) };
679 if rc != 0 {
680 // Could not lock (e.g. a filesystem without flock support): proceed
681 // unlocked rather than fail the append.
682 return None;
683 }
684 Some(file)
685 }
686}
687
688#[cfg(unix)]
689impl Drop for AppendLock {
690 fn drop(&mut self) {
691 use std::os::unix::io::AsRawFd;
692 if let Some(file) = &self.file {
693 // Release explicitly; the fd close on drop would also release it.
694 unsafe { flock(file.as_raw_fd(), LOCK_UN) };
695 }
696 }
697}
698
699#[cfg(unix)]
700extern "C" {
701 fn flock(fd: std::os::raw::c_int, operation: std::os::raw::c_int) -> std::os::raw::c_int;
702}
703
704/// `flock` operation: exclusive lock (`LOCK_EX`), blocking.
705#[cfg(unix)]
706const LOCK_EX: std::os::raw::c_int = 2;
707/// `flock` operation: unlock (`LOCK_UN`).
708#[cfg(unix)]
709const LOCK_UN: std::os::raw::c_int = 8;
710
711/// The advisory-lock sibling path for an active log file (`<name>.lock`).
712#[cfg(unix)]
713fn lock_path_for(active: &Path) -> PathBuf {
714 let mut name = active
715 .file_name()
716 .map(|s| s.to_os_string())
717 .unwrap_or_else(|| std::ffi::OsString::from("log.md"));
718 name.push(".lock");
719 match active.parent() {
720 Some(parent) => parent.join(name),
721 None => PathBuf::from(name),
722 }
723}
724
725/// The active `log.md` path under the store root.
726fn active_log_path(store: &Store) -> PathBuf {
727 store.root.join("log.md")
728}
729
730/// The `log/` archive directory under the store root.
731fn archive_dir(store: &Store) -> PathBuf {
732 store.root.join("log")
733}
734
735/// The `log/<YYYY-MM>.md` archive path for a given month.
736fn archive_path(store: &Store, year: i32, month: u32) -> PathBuf {
737 archive_dir(store).join(format!("{:04}-{:02}.md", year, month))
738}
739
740/// Parse a `YYYY-MM-DD HH:MM` header timestamp, reattaching UTC. `None` on any
741/// malformed shape.
742fn parse_timestamp(s: &str) -> Option<DateTime<FixedOffset>> {
743 let naive = NaiveDateTime::parse_from_str(s.trim(), TS_FORMAT).ok()?;
744 let utc = FixedOffset::east_opt(0)?;
745 utc.from_local_datetime(&naive).single()
746}
747
748/// Split a `log.md` / archive file into its leading frontmatter+heading block
749/// (everything up to and including the line before the first `## [` header) and
750/// its parsed entries. If there are no entries, the whole content is the header
751/// block.
752fn parse_active(content: &str) -> (String, Vec<LogEntry>) {
753 match find_first_header(content) {
754 Some(idx) => {
755 let header = content[..idx].to_string();
756 let entries = parse_entries(&content[idx..]);
757 (header, entries)
758 }
759 None => (content.to_string(), Vec::new()),
760 }
761}
762
763/// Byte offset of the first entry header (`## [` at the start of a line), or
764/// `None`.
765fn find_first_header(content: &str) -> Option<usize> {
766 if content.starts_with("## [") {
767 return Some(0);
768 }
769 content.match_indices("\n## [").next().map(|(i, _)| i + 1)
770}
771
772/// Whether `line` is a note line that — left unescaped — could be mistaken for
773/// an entry header. It is *header-ambiguous* when it is a (possibly empty) run
774/// of leading backslashes followed by a string that [`Log::parse_header`]
775/// accepts. The escape (one leading backslash) and only the escape is added to,
776/// or stripped from, such lines, so the transform is fully reversible:
777/// `## [..]` (a real header shape in note text) ⇄ `\## [..]`, and a literal
778/// `\## [..]` a note already contains ⇄ `\\## [..]`.
779fn is_header_ambiguous(line: &str) -> bool {
780 let stripped = line.trim_start_matches('\\');
781 // Only treat it as ambiguous if some backslashes were the *only* prefix and
782 // the remainder is a valid header — a backslash run that does not lead into
783 // a header (e.g. `\not a header`) is ordinary note text, left untouched.
784 Log::parse_header(stripped).is_some()
785}
786
787/// Escape one note line for on-disk emission so it can never be parsed as an
788/// entry header (the [write-path fix] for header-shaped notes corrupting the
789/// append-only log). A header-ambiguous line is prefixed with a single
790/// backslash, moving its `## [` off column 0; every other line is emitted
791/// verbatim. Reversed exactly by [`unescape_note_line`].
792fn escape_note_line(line: &str) -> std::borrow::Cow<'_, str> {
793 if is_header_ambiguous(line) {
794 std::borrow::Cow::Owned(format!("\\{line}"))
795 } else {
796 std::borrow::Cow::Borrowed(line)
797 }
798}
799
800/// Reverse [`escape_note_line`]: strip exactly one leading backslash from a
801/// header-ambiguous on-disk note line, restoring the literal the author wrote.
802/// A line that is not header-ambiguous (including a genuine `\not a header`) is
803/// returned untouched, so the round-trip is lossless for arbitrary note text.
804fn unescape_note_line(line: &str) -> std::borrow::Cow<'_, str> {
805 if let Some(rest) = line.strip_prefix('\\') {
806 if is_header_ambiguous(line) {
807 return std::borrow::Cow::Borrowed(rest);
808 }
809 }
810 std::borrow::Cow::Borrowed(line)
811}
812
813/// Parse every entry in a slice that begins at (or before, header-block
814/// included) a sequence of `## [` headers. Headers that fail to parse are
815/// skipped (their body folds into the previous valid entry's note is avoided —
816/// they simply start no new entry).
817fn parse_entries(text: &str) -> Vec<LogEntry> {
818 let mut entries: Vec<LogEntry> = Vec::new();
819 let mut cur_header: Option<(DateTime<FixedOffset>, LogKind, Option<String>)> = None;
820 let mut cur_note: Vec<&str> = Vec::new();
821
822 let flush = |entries: &mut Vec<LogEntry>,
823 header: &mut Option<(DateTime<FixedOffset>, LogKind, Option<String>)>,
824 note: &mut Vec<&str>| {
825 if let Some((timestamp, kind, object)) = header.take() {
826 // Reverse the per-line header escape `render` applies so an escaped
827 // header-shaped note line round-trips back to its literal form.
828 let joined = note
829 .iter()
830 .map(|line| unescape_note_line(line))
831 .collect::<Vec<_>>()
832 .join("\n");
833 let note_str = joined.trim_matches(['\n', '\r']).to_string();
834 entries.push(LogEntry {
835 timestamp,
836 kind,
837 object,
838 note: note_str,
839 });
840 }
841 note.clear();
842 };
843
844 for line in text.lines() {
845 if line.starts_with("## [") {
846 if let Some(parsed) = Log::parse_header(line) {
847 // Close the previous entry, start a new one.
848 flush(&mut entries, &mut cur_header, &mut cur_note);
849 cur_header = Some(parsed);
850 continue;
851 }
852 // Unparseable `## [` line: treat as body of the current entry.
853 }
854 if cur_header.is_some() {
855 cur_note.push(line);
856 }
857 }
858 flush(&mut entries, &mut cur_header, &mut cur_note);
859 entries
860}
861
862/// Recompose an active/archive file from a header block and an entry body.
863fn compose_active(header: &str, body: &str) -> String {
864 let mut out = String::new();
865 out.push_str(header);
866 if !header.is_empty() && !header.ends_with('\n') {
867 out.push('\n');
868 }
869 // Exactly one blank line between the heading block and the first entry.
870 if !header.is_empty() && !out.ends_with("\n\n") {
871 out.push('\n');
872 }
873 out.push_str(body);
874 out
875}
876
877/// Append entries to a month archive, creating it with `type: log` frontmatter
878/// if absent. Atomic (temp-file rename). Entries are appended in the given
879/// order (callers pass them already chronological within the month).
880///
881/// **Idempotent re-roll.** Rotation in [`Log::append`] is two non-atomic durable
882/// writes — roll prior-month entries into the archive, *then* rewrite the active
883/// file. If the process crashes or the active rewrite errors (e.g. ENOSPC,
884/// permission) *after* the archive write commits, the prior-month entries remain
885/// in the still-untrimmed active file, and `Log::append` surfaces the error so
886/// the agent retries. The retry re-partitions the same prior-month entries and
887/// re-rolls them here — so a naive concatenate would duplicate every entry in
888/// the month archive, amplifying on each retry, with no validate check to detect
889/// or repair it (the log is primary, no-rewrite data). To make the re-roll a
890/// no-op, we skip any incoming entry already present verbatim in the archive,
891/// keyed on the full entry identity `(timestamp, kind, object, note)`.
892fn append_to_archive(path: &Path, entries: &[LogEntry]) -> crate::Result<()> {
893 if path.exists() {
894 let existing = fs::read_to_string(path)?;
895 // Identities already on disk in this archive, so an interrupted-then-
896 // retried rotation re-rolling identical entries adds nothing.
897 let (_header, existing_entries) = parse_active(&existing);
898 let present: std::collections::HashSet<EntryKey> =
899 existing_entries.iter().map(entry_key).collect();
900
901 let mut body = String::new();
902 for e in entries {
903 if present.contains(&entry_key(e)) {
904 continue;
905 }
906 body.push_str(&e.render());
907 }
908 // Nothing new to add (a fully-duplicate re-roll): leave the archive
909 // byte-for-byte untouched (append-only: don't rewrite identical data).
910 if body.is_empty() {
911 return Ok(());
912 }
913
914 let mut full = existing;
915 if !full.ends_with('\n') {
916 full.push('\n');
917 }
918 full.push_str(&body);
919 crate::fsx::write_atomic(path, full.as_bytes())?;
920 } else {
921 let mut body = String::new();
922 for e in entries {
923 body.push_str(&e.render());
924 }
925 if let Some(parent) = path.parent() {
926 fs::create_dir_all(parent)?;
927 }
928 let full = compose_active(LOG_FRONTMATTER, &body);
929 crate::fsx::write_atomic(path, full.as_bytes())?;
930 }
931 Ok(())
932}
933
934/// A hashable identity for a log entry, used to dedup an idempotent archive
935/// re-roll (see [`append_to_archive`]). Two entries are "the same" when their
936/// timestamp, kind, object, and note all match — exactly the fields that
937/// round-trip through `render`/`parse`, so a re-rolled entry compares equal to
938/// the one already archived. Owned (rather than borrowed) so keys from the
939/// existing archive and from the incoming entries share one type regardless of
940/// where they came from; the cost is paid only on the cold rotation path.
941type EntryKey = (DateTime<FixedOffset>, String, Option<String>, String);
942
943/// Derive the dedup key for `e` (see [`EntryKey`]). Keying on `kind.as_str()`
944/// (rather than `LogKind`, which is not `Hash`) is exact: `as_str`/`parse`
945/// round-trips every recognized kind and preserves any `Custom` token.
946fn entry_key(e: &LogEntry) -> EntryKey {
947 (
948 e.timestamp,
949 e.kind.as_str().to_string(),
950 e.object.clone(),
951 e.note.clone(),
952 )
953}
954
955/// Every `log/<YYYY-MM>.md` archive, sorted **newest month first**.
956fn list_archives_desc(store: &Store) -> crate::Result<Vec<PathBuf>> {
957 let dir = archive_dir(store);
958 if !dir.is_dir() {
959 return Ok(Vec::new());
960 }
961 let mut months: Vec<(String, PathBuf)> = Vec::new();
962 for entry in fs::read_dir(&dir)? {
963 let entry = entry?;
964 let path = entry.path();
965 if !path.is_file() {
966 continue;
967 }
968 let name = match path.file_name().and_then(|s| s.to_str()) {
969 Some(n) => n,
970 None => continue,
971 };
972 // Match `YYYY-MM.md`.
973 if let Some(stem) = name.strip_suffix(".md") {
974 if is_year_month(stem) {
975 months.push((stem.to_string(), path.clone()));
976 }
977 }
978 }
979 // `YYYY-MM` strings sort lexically == chronologically; reverse for newest
980 // first.
981 months.sort_by(|a, b| b.0.cmp(&a.0));
982 Ok(months.into_iter().map(|(_, p)| p).collect())
983}
984
985/// The `(year, month)` an archive file represents, parsed from its
986/// `log/<YYYY-MM>.md` name. `None` if the name isn't a well-formed month
987/// archive (in which case the caller scans it rather than risk skipping it).
988fn archive_year_month(path: &Path) -> Option<(i32, u32)> {
989 let stem = path
990 .file_name()
991 .and_then(|s| s.to_str())
992 .and_then(|n| n.strip_suffix(".md"))?;
993 if !is_year_month(stem) {
994 return None;
995 }
996 let year: i32 = stem[..4].parse().ok()?;
997 let month: u32 = stem[5..7].parse().ok()?;
998 // The month must be a real calendar month. A hand-created / externally-
999 // produced `log/2026-00.md` or `log/2026-13.md` parses as two digits but
1000 // names no month; returning `Some((year, 0))` would sort it below every
1001 // legitimate month, so the newest-month-first early-break in `since`/`tail`
1002 // could prune it and silently drop its entries. Out-of-range → `None`, so the
1003 // caller scans the file instead of risk-skipping it (the safe fallback).
1004 if !(1..=12).contains(&month) {
1005 return None;
1006 }
1007 Some((year, month))
1008}
1009
1010/// True if `s` looks like `YYYY-MM` (4 digits, dash, 2 digits).
1011fn is_year_month(s: &str) -> bool {
1012 let bytes = s.as_bytes();
1013 if bytes.len() != 7 {
1014 return false;
1015 }
1016 bytes[..4].iter().all(u8::is_ascii_digit)
1017 && bytes[4] == b'-'
1018 && bytes[5].is_ascii_digit()
1019 && bytes[6].is_ascii_digit()
1020}
1021
1022/// Reverse-read `path` from EOF, parsing entries newest-first and feeding each
1023/// to `take`. `take` returns `true` to stop early (enough collected). The file
1024/// is read backward in blocks; only the tail region needed to satisfy `take`
1025/// is read — the whole file is read only if `take` never returns `true`.
1026fn reverse_collect<F>(path: &Path, mut take: F) -> crate::Result<()>
1027where
1028 F: FnMut(LogEntry) -> bool,
1029{
1030 let mut file = File::open(path)?;
1031 let len = file.metadata()?.len();
1032 if len == 0 {
1033 return Ok(());
1034 }
1035
1036 // Algorithm: grow a tail buffer leftward one block at a time, emitting
1037 // entries strictly newest-first as their left boundary is confirmed, and
1038 // stopping the instant `take` says enough. The whole file is read only if
1039 // `take` never returns `true` (e.g. `tail(n)` with n ≥ entry count).
1040 //
1041 // Invariant: a `## [` line-start anywhere in the buffer is a *complete*
1042 // entry — its header is the entry's first line, and its body lies to the
1043 // right and is therefore already buffered (we read right-to-left). So we
1044 // never split an entry across blocks.
1045 //
1046 // `buf` holds the file's bytes from absolute offset `start` (growing
1047 // leftward toward 0) to EOF. `emitted_abs` records the absolute offsets of
1048 // headers already handed to `take`, so re-visiting a header in a later block
1049 // never double-emits.
1050 let mut buf: Vec<u8> = Vec::new();
1051 let mut start = len;
1052 // O(1) membership: a `Vec` + `.contains()` here would be O(E²) across a large
1053 // single-month file (every header re-checked against all prior emissions).
1054 let mut emitted_abs: std::collections::HashSet<u64> = std::collections::HashSet::new();
1055 // Every header's absolute offset found so far, ascending. Built
1056 // *incrementally*: each block contributes only the markers whose `#` starts
1057 // inside it (all strictly smaller than any already-known offset, so they
1058 // prepend in order). This is the fix for the accidental O(file²) scan — the
1059 // old code re-ran `header_offsets` over the whole accumulated buffer on every
1060 // block (O(file²/block) byte comparisons on the default no-early-stop
1061 // tail/since path); now each byte is scanned for a header exactly once.
1062 let mut headers: Vec<u64> = Vec::new();
1063 let mut stop = false;
1064 // The first backward block has no already-scanned region to its right, so it
1065 // scans exactly `[0, block)`; every later block scans one byte further
1066 // (`block + 1`) to re-classify the prior block's deferred left-edge candidate
1067 // now that its left neighbour is buffered (see the scan call below).
1068 let mut first = true;
1069
1070 while start > 0 && !stop {
1071 let block = std::cmp::min(REVERSE_BLOCK as u64, start);
1072 let new_start = start - block;
1073 file.seek(SeekFrom::Start(new_start))?;
1074 let mut chunk = vec![0u8; block as usize];
1075 file.read_exact(&mut chunk)?;
1076 chunk.extend_from_slice(&buf);
1077 buf = chunk;
1078 start = new_start;
1079
1080 // Scan the freshly-prepended block (buffer indices `[0, block)`) for new
1081 // header markers. A marker straddling the block boundary has its `#` in
1082 // this window and so is still caught (see `header_offsets_range`).
1083 //
1084 // One subtlety the scan must respect: a `## [` whose `#` sits at the
1085 // block's LEFT edge (buffer index 0, absolute offset `start`) cannot have
1086 // its line-start confirmed yet when `start > 0` — the byte at `start - 1`
1087 // is not buffered. Treating index 0 as a line start there fabricates an
1088 // entry from a mid-line `## [` fragment that happens to align with a block
1089 // boundary. So `header_offsets_range` DEFERS the leftmost candidate when
1090 // `base` is not the true file start, and we re-scan one byte further
1091 // right next time: after the first block the buffer carries the previous
1092 // block's left-edge byte at index `block` with its left neighbour now in
1093 // hand, so extending the window to `block + 1` re-classifies that exactly
1094 // once. `first` guards the first block (nothing to re-check on its right).
1095 let base_is_file_start = start == 0;
1096 let scan_hi = if first { block } else { block + 1 } as usize;
1097 let mut new_headers = header_offsets_range(&buf, start, 0, scan_hi, base_is_file_start);
1098 first = false;
1099 if !new_headers.is_empty() {
1100 new_headers.extend_from_slice(&headers);
1101 headers = new_headers;
1102 }
1103
1104 // Process newest (largest offset) → oldest (smallest), emitting any
1105 // header not yet emitted. Hold back only the buffer's *leftmost* header
1106 // while we have not reached file start (`start > 0`): older entries may
1107 // still lie to its left in unread blocks, and newest-first order
1108 // requires we not emit it until we've confirmed it really is the oldest
1109 // (or read enough to bound it on the left). One extra block read at
1110 // most; on the next iteration its left boundary is in-buffer.
1111 for i in (0..headers.len()).rev() {
1112 let abs = headers[i];
1113 if emitted_abs.contains(&abs) {
1114 continue;
1115 }
1116 let is_oldest_in_buf = i == 0;
1117 if is_oldest_in_buf && start > 0 {
1118 continue;
1119 }
1120
1121 let entry_text = entry_text_at(&buf, start, abs, &headers, i);
1122 if let Some(entry) = parse_single_entry(&entry_text) {
1123 emitted_abs.insert(abs);
1124 if take(entry) {
1125 stop = true;
1126 break;
1127 }
1128 } else {
1129 emitted_abs.insert(abs);
1130 }
1131 }
1132 }
1133
1134 // Reached file start (or stopped). If we stopped, done. If we reached
1135 // start, emit any held-back oldest header(s) now (start == 0 means the
1136 // buffer's first header is genuinely the oldest). `headers` already holds
1137 // every offset (the loop scanned down to start == 0), so reuse it.
1138 if !stop && start == 0 {
1139 for i in (0..headers.len()).rev() {
1140 let abs = headers[i];
1141 if emitted_abs.contains(&abs) {
1142 continue;
1143 }
1144 let entry_text = entry_text_at(&buf, start, abs, &headers, i);
1145 if let Some(entry) = parse_single_entry(&entry_text) {
1146 emitted_abs.insert(abs);
1147 if take(entry) {
1148 break;
1149 }
1150 } else {
1151 emitted_abs.insert(abs);
1152 }
1153 }
1154 }
1155
1156 Ok(())
1157}
1158
1159/// Absolute byte offsets of every **valid** entry-header line-start (`## […]`)
1160/// in `buf`, where `buf` begins at absolute offset `base`.
1161///
1162/// Only a `## [` line that [`Log::parse_header`] accepts is an entry boundary,
1163/// mirroring the forward parser ([`parse_entries`]), which folds an unparseable
1164/// `## [` line into the preceding entry's note rather than starting a new entry.
1165/// Without this validity check the reverse reader would split a real entry's
1166/// multi-line note at a continuation line beginning at column 0 with `## [`
1167/// (a shape the SPEC permits — notes are "one or more lines" with no
1168/// restriction), truncating the note and dropping the carved pseudo-entry, so
1169/// `tail`/`since`/`last_validate_at` would return a note diverging from the
1170/// intact on-disk bytes.
1171///
1172/// Whole-buffer convenience wrapper over [`header_offsets_range`]. The runtime
1173/// reverse reader now always scans incrementally (one freshly-prepended window
1174/// per backward block), so this whole-buffer form is retained only as the
1175/// oracle the range-scan tests check the incremental scan against.
1176#[cfg(test)]
1177fn header_offsets(buf: &[u8], base: u64) -> Vec<u64> {
1178 // The whole-buffer oracle treats `base` as the file start iff it is 0, so a
1179 // `## [` at buffer index 0 is a real line-start there.
1180 header_offsets_range(buf, base, 0, buf.len(), base == 0)
1181}
1182
1183/// Like [`header_offsets`] but only reports header *markers whose `#` starts in*
1184/// `buf[scan_lo..scan_hi)`, while still consulting bytes outside that window —
1185/// to the left for the line-start (`buf[i-1] == b'\n'`) check and to the right
1186/// for the header line's content. This is the incremental scan
1187/// [`reverse_collect`] uses: each backward block searches only the freshly-
1188/// prepended region for *new* markers, so total header-scan work is linear in
1189/// the file size, not the O(file²) of re-scanning the whole growing buffer on
1190/// every block.
1191///
1192/// A `## [` marker that *straddles* the boundary (its `#` in the new block, its
1193/// `[` or trailing bytes in the already-scanned region) is still detected here:
1194/// its `#` index is `< scan_hi`, so it falls in this window, and it was never
1195/// reported by an earlier scan (whose window was `[block, …)`, strictly to the
1196/// right of this one) — so each marker is reported exactly once across all
1197/// blocks.
1198///
1199/// **Left-edge line-start safety.** A `## [` whose `#` is at buffer index 0 has
1200/// no buffered left neighbour, so its line-start cannot be confirmed unless
1201/// index 0 really is the file start. `base_is_file_start` says so: when it is
1202/// `false`, an index-0 candidate is DEFERRED (not reported) rather than assumed
1203/// to be at a line start — otherwise a mid-line `## […]` fragment that happens
1204/// to align with a block's left edge would be fabricated into an entry,
1205/// truncating the real entry's note and (after rotation) corrupting the
1206/// append-only archive. The caller re-scans that byte on the next block, once
1207/// its left neighbour is buffered, so a genuine boundary header is still found
1208/// exactly once.
1209fn header_offsets_range(
1210 buf: &[u8],
1211 base: u64,
1212 scan_lo: usize,
1213 scan_hi: usize,
1214 base_is_file_start: bool,
1215) -> Vec<u64> {
1216 const PAT: &[u8] = b"## [";
1217 let mut out = Vec::new();
1218 let n = buf.len();
1219 let hi = scan_hi.min(n);
1220 let mut i = scan_lo;
1221 // A marker's `#` must start strictly before `hi`; the pattern/line content
1222 // may read past `hi` into `buf` (the right neighbour is already buffered).
1223 while i < hi && i + PAT.len() <= n {
1224 if &buf[i..i + PAT.len()] == PAT {
1225 // Index 0 is a line start only when it is the genuine file start;
1226 // otherwise its left neighbour is unbuffered and the candidate is
1227 // deferred to the next block (see the doc comment).
1228 let at_line_start = if i == 0 {
1229 base_is_file_start
1230 } else {
1231 buf[i - 1] == b'\n'
1232 };
1233 if at_line_start && is_valid_header_line(buf, i) {
1234 out.push(base + i as u64);
1235 // skip ahead past this marker
1236 i += PAT.len();
1237 continue;
1238 }
1239 }
1240 i += 1;
1241 }
1242 out
1243}
1244
1245/// Whether the `## [` line starting at byte `i` in `buf` parses as a valid
1246/// entry header. Reads the line up to (but not including) the next `\n` (or
1247/// buffer end) and defers to [`Log::parse_header`] — the same validity gate the
1248/// forward parser applies, keeping the reverse reader's boundary set identical
1249/// to the forward one.
1250fn is_valid_header_line(buf: &[u8], i: usize) -> bool {
1251 let line_end = buf[i..]
1252 .iter()
1253 .position(|&b| b == b'\n')
1254 .map(|p| i + p)
1255 .unwrap_or(buf.len());
1256 let line = String::from_utf8_lossy(&buf[i..line_end]);
1257 Log::parse_header(&line).is_some()
1258}
1259
1260/// Extract the text of the entry whose header is at absolute offset
1261/// `header_abs` (the `headers[idx]` entry), spanning to the next header (or
1262/// buffer end). `buf` begins at absolute offset `base`.
1263fn entry_text_at(buf: &[u8], base: u64, header_abs: u64, headers: &[u64], idx: usize) -> String {
1264 let rel_start = (header_abs - base) as usize;
1265 let rel_end = if idx + 1 < headers.len() {
1266 (headers[idx + 1] - base) as usize
1267 } else {
1268 buf.len()
1269 };
1270 String::from_utf8_lossy(&buf[rel_start..rel_end]).into_owned()
1271}
1272
1273/// Parse a single entry from a text block that begins at its header line.
1274fn parse_single_entry(text: &str) -> Option<LogEntry> {
1275 parse_entries(text).into_iter().next()
1276}
1277
1278#[cfg(test)]
1279mod tests {
1280 use super::*;
1281 use crate::parser::Config;
1282 use std::fs;
1283 use tempfile::TempDir;
1284
1285 /// Build a `Store` rooted at a fresh temp dir with a minimal `DB.md`.
1286 /// Construct the `Store` struct directly so the test stays narrow and never
1287 /// exercises the `Store::open` parser path.
1288 fn temp_store() -> (TempDir, Store) {
1289 let dir = tempfile::tempdir().expect("tempdir");
1290 fs::write(dir.path().join("DB.md"), "---\ntype: db-md\n---\n").expect("write DB.md");
1291 let store = Store {
1292 root: dir.path().to_path_buf(),
1293 config: Config::default(),
1294 };
1295 (dir, store)
1296 }
1297
1298 /// Regression (adversarial review): a hand-created / externally-produced
1299 /// archive with an out-of-range month (`00`, `13`..`99`) must NOT parse as a
1300 /// real month archive — otherwise its `(year, 0)` bucket sorts below every
1301 /// legitimate month and the newest-first early-break in `since`/`tail` can
1302 /// silently prune it. Out-of-range → `None` (the caller scans it instead).
1303 #[test]
1304 fn archive_year_month_rejects_out_of_range_months() {
1305 use std::path::Path;
1306 assert_eq!(
1307 archive_year_month(Path::new("log/2026-05.md")),
1308 Some((2026, 5))
1309 );
1310 assert_eq!(
1311 archive_year_month(Path::new("log/2026-01.md")),
1312 Some((2026, 1))
1313 );
1314 assert_eq!(
1315 archive_year_month(Path::new("log/2026-12.md")),
1316 Some((2026, 12))
1317 );
1318 for bad in ["log/2026-00.md", "log/2026-13.md", "log/2026-99.md"] {
1319 assert_eq!(
1320 archive_year_month(Path::new(bad)),
1321 None,
1322 "{bad} has an out-of-range month and must not parse as an archive"
1323 );
1324 }
1325 }
1326
1327 /// A timestamp at UTC from `YYYY-MM-DD HH:MM` components.
1328 fn ts(y: i32, mo: u32, d: u32, h: u32, mi: u32) -> DateTime<FixedOffset> {
1329 let naive = chrono::NaiveDate::from_ymd_opt(y, mo, d)
1330 .unwrap()
1331 .and_hms_opt(h, mi, 0)
1332 .unwrap();
1333 FixedOffset::east_opt(0)
1334 .unwrap()
1335 .from_local_datetime(&naive)
1336 .single()
1337 .unwrap()
1338 }
1339
1340 #[allow(clippy::too_many_arguments)] // test fixture builder; struct-ifying churns every call site
1341 fn entry(
1342 y: i32,
1343 mo: u32,
1344 d: u32,
1345 h: u32,
1346 mi: u32,
1347 kind: LogKind,
1348 object: Option<&str>,
1349 note: &str,
1350 ) -> LogEntry {
1351 LogEntry {
1352 timestamp: ts(y, mo, d, h, mi),
1353 kind,
1354 object: object.map(|s| s.to_string()),
1355 note: note.to_string(),
1356 }
1357 }
1358
1359 // ── parse_header ────────────────────────────────────────────────────────
1360
1361 #[test]
1362 fn parse_header_with_object() {
1363 let (t, k, o) =
1364 Log::parse_header("## [2026-05-27 10:00] ingest | sources/emails/x.eml").unwrap();
1365 assert_eq!(t, ts(2026, 5, 27, 10, 0));
1366 assert_eq!(k, LogKind::Ingest);
1367 assert_eq!(o.as_deref(), Some("sources/emails/x.eml"));
1368 }
1369
1370 #[test]
1371 fn parse_header_without_object_is_none_object() {
1372 let (t, k, o) = Log::parse_header("## [2026-05-27 10:20] validate").unwrap();
1373 assert_eq!(t, ts(2026, 5, 27, 10, 20));
1374 assert_eq!(k, LogKind::Validate);
1375 assert_eq!(o, None);
1376 }
1377
1378 #[test]
1379 fn parse_header_custom_kind_roundtrips_token() {
1380 let (_, k, o) = Log::parse_header("## [2026-05-27 10:00] proposal | records/x").unwrap();
1381 assert_eq!(k, LogKind::Custom("proposal".to_string()));
1382 assert!(!k.is_recognized());
1383 assert_eq!(o.as_deref(), Some("records/x"));
1384 }
1385
1386 #[test]
1387 fn parse_header_index_rebuild_hyphenated_kind() {
1388 let (_, k, _) = Log::parse_header("## [2026-05-27 10:00] index-rebuild").unwrap();
1389 assert_eq!(k, LogKind::IndexRebuild);
1390 assert_eq!(k.as_str(), "index-rebuild");
1391 }
1392
1393 #[test]
1394 fn parse_header_rejects_non_headers() {
1395 assert!(Log::parse_header("Not a header").is_none());
1396 assert!(Log::parse_header("# Curator log").is_none());
1397 assert!(Log::parse_header("## [garbage] ingest | x").is_none());
1398 assert!(Log::parse_header("## [2026-05-27 10:00]").is_none()); // no kind
1399 // A bracketed but non-timestamp date must be rejected (LOG_BAD_TIMESTAMP territory).
1400 assert!(Log::parse_header("## [2026-13-40 99:99] ingest | x").is_none());
1401 }
1402
1403 // ── kind round-trip ───────────────────────────────────────────────────────
1404
1405 #[test]
1406 fn kind_as_str_parse_roundtrip_for_all_recognized() {
1407 for k in [
1408 LogKind::Ingest,
1409 LogKind::Create,
1410 LogKind::Update,
1411 LogKind::Delete,
1412 LogKind::Rename,
1413 LogKind::Link,
1414 LogKind::Validate,
1415 LogKind::IndexRebuild,
1416 LogKind::Contradiction,
1417 ] {
1418 assert_eq!(LogKind::parse(k.as_str()), k);
1419 assert!(k.is_recognized());
1420 }
1421 }
1422
1423 // ── append: creation + frontmatter ───────────────────────────────────────
1424
1425 #[test]
1426 fn append_creates_log_with_frontmatter_and_entry() {
1427 let (_d, store) = temp_store();
1428 let e = entry(
1429 2026,
1430 5,
1431 27,
1432 10,
1433 0,
1434 LogKind::Ingest,
1435 Some("sources/emails/x.eml"),
1436 "Email received.",
1437 );
1438 Log::append(&store, &e).unwrap();
1439
1440 let content = fs::read_to_string(store.root.join("log.md")).unwrap();
1441 // type: log frontmatter present.
1442 assert!(
1443 content.starts_with("---\ntype: log\n---\n"),
1444 "missing log frontmatter; got:\n{content}"
1445 );
1446 // The entry header is rendered verbatim.
1447 assert!(content.contains("## [2026-05-27 10:00] ingest | sources/emails/x.eml"));
1448 assert!(content.contains("Email received."));
1449 // No archive dir created when nothing rotates.
1450 assert!(!store.root.join("log").exists());
1451 }
1452
1453 // ── append → tail → since round-trip ─────────────────────────────────────
1454
1455 #[test]
1456 fn append_tail_since_roundtrip() {
1457 let (_d, store) = temp_store();
1458 let e1 = entry(2026, 5, 27, 10, 0, LogKind::Ingest, Some("a"), "first");
1459 let e2 = entry(2026, 5, 27, 10, 5, LogKind::Create, Some("b"), "second");
1460 let e3 = entry(2026, 5, 27, 10, 10, LogKind::Update, Some("c"), "third");
1461 Log::append(&store, &e1).unwrap();
1462 Log::append(&store, &e2).unwrap();
1463 Log::append(&store, &e3).unwrap();
1464
1465 // tail(2) returns the two newest, in chronological order.
1466 let tail = Log::tail(&store, 2).unwrap();
1467 assert_eq!(tail.len(), 2);
1468 assert_eq!(tail[0], e2);
1469 assert_eq!(tail[1], e3);
1470
1471 // tail(n) larger than the log returns everything, chronologically.
1472 let all = Log::tail(&store, 99).unwrap();
1473 assert_eq!(all, vec![e1.clone(), e2.clone(), e3.clone()]);
1474
1475 // since(10:05) returns strictly-newer entries (excludes the 10:05 one).
1476 let since = Log::since(&store, ts(2026, 5, 27, 10, 5)).unwrap();
1477 assert_eq!(since, vec![e3.clone()]);
1478
1479 // since before everything returns all.
1480 let since_all = Log::since(&store, ts(2026, 5, 27, 9, 0)).unwrap();
1481 assert_eq!(since_all, vec![e1, e2, e3]);
1482 }
1483
1484 #[test]
1485 fn tail_zero_is_empty() {
1486 let (_d, store) = temp_store();
1487 Log::append(
1488 &store,
1489 &entry(2026, 5, 27, 10, 0, LogKind::Ingest, Some("a"), "x"),
1490 )
1491 .unwrap();
1492 assert!(Log::tail(&store, 0).unwrap().is_empty());
1493 }
1494
1495 #[test]
1496 fn tail_and_since_on_missing_log_are_empty() {
1497 let (_d, store) = temp_store();
1498 assert!(Log::tail(&store, 5).unwrap().is_empty());
1499 assert!(Log::since(&store, ts(2000, 1, 1, 0, 0)).unwrap().is_empty());
1500 assert!(Log::last_validate_at(&store).unwrap().is_none());
1501 }
1502
1503 #[test]
1504 fn since_exact_timestamp_is_exclusive() {
1505 let (_d, store) = temp_store();
1506 let e = entry(2026, 5, 27, 10, 0, LogKind::Validate, None, "PASS");
1507 Log::append(&store, &e).unwrap();
1508 // Equal timestamp must NOT be included (strictly newer).
1509 assert!(Log::since(&store, ts(2026, 5, 27, 10, 0))
1510 .unwrap()
1511 .is_empty());
1512 }
1513
1514 // ── since: out-of-order on disk (append-only correction / merge=union) ────
1515
1516 /// Write a `log.md` at the store root from `entries` in the EXACT given
1517 /// physical order, with the standard `type: log` frontmatter. Unlike
1518 /// [`Log::append`] (which always lands the newest entry at EOF), this lets a
1519 /// test author the non-monotonic on-disk shape the SPEC permits — a
1520 /// backdated corrective entry below the entry it corrects, or a
1521 /// `merge=union` interleave.
1522 fn write_raw_log(store: &Store, entries: &[LogEntry]) {
1523 let mut content = String::from(LOG_FRONTMATTER);
1524 content.push('\n');
1525 for e in entries {
1526 content.push_str(&e.render());
1527 }
1528 fs::write(store.root.join("log.md"), content).expect("write raw log.md");
1529 }
1530
1531 #[test]
1532 fn since_returns_newer_entries_even_when_disk_order_is_non_monotonic() {
1533 // The demonstrated regression: a curator appended a backdated CORRECTIVE
1534 // entry (10:00) below newer entries (10:10, 10:05), so the physical
1535 // on-disk order is 10:10, 10:05, 10:00 — newest-first, not chronological.
1536 // The append-only SPEC explicitly permits this ("append a corrective
1537 // entry below it"; out-of-order is only LOG_OUT_OF_ORDER, a warning).
1538 let (_d, store) = temp_store();
1539 let e_1010 = entry(2026, 5, 27, 10, 10, LogKind::Update, Some("c"), "newest");
1540 let e_1005 = entry(2026, 5, 27, 10, 5, LogKind::Create, Some("b"), "middle");
1541 let e_1000 = entry(
1542 2026,
1543 5,
1544 27,
1545 10,
1546 0,
1547 LogKind::Update,
1548 Some("a"),
1549 "backdated fix",
1550 );
1551 // Physical order on disk: 10:10, 10:05, then the backdated 10:00 LAST.
1552 write_raw_log(&store, &[e_1010, e_1005, e_1000]);
1553
1554 // since 10:02 must return BOTH entries strictly newer than 10:02
1555 // (10:05 and 10:10). The old early-stop hit the physically-last 10:00
1556 // entry (<= 10:02), stopped, and returned EMPTY — silently dropping the
1557 // two newer entries that sit earlier in the file.
1558 let got = Log::since(&store, ts(2026, 5, 27, 10, 2)).unwrap();
1559 let stamps: std::collections::BTreeSet<_> = got.iter().map(|e| e.timestamp).collect();
1560 assert_eq!(
1561 stamps,
1562 [ts(2026, 5, 27, 10, 5), ts(2026, 5, 27, 10, 10)]
1563 .into_iter()
1564 .collect(),
1565 "since(10:02) must include both 10:05 and 10:10 despite the backdated \
1566 10:00 entry sitting physically last, and exclude 10:00; got {got:?}"
1567 );
1568
1569 // A cutoff before everything still returns all three, regardless of the
1570 // scrambled disk order.
1571 let all = Log::since(&store, ts(2026, 5, 27, 9, 0)).unwrap();
1572 let all_stamps: std::collections::BTreeSet<_> = all.iter().map(|e| e.timestamp).collect();
1573 assert_eq!(
1574 all_stamps,
1575 [
1576 ts(2026, 5, 27, 10, 0),
1577 ts(2026, 5, 27, 10, 5),
1578 ts(2026, 5, 27, 10, 10),
1579 ]
1580 .into_iter()
1581 .collect()
1582 );
1583 }
1584
1585 #[test]
1586 fn since_crosses_archive_when_newer_entry_is_out_of_order_inside_it() {
1587 // Out-of-order INSIDE an archive month, with the cutoff landing in that
1588 // month. The April archive is authored newest-physical-first (04-20,
1589 // then a backdated 04-05 last); a naive early-stop on the first
1590 // older-than-cutoff entry would miss the later April entry. The active
1591 // file holds a clean May entry. Cutoff = mid-April.
1592 let (_d, store) = temp_store();
1593
1594 // Active file: one current-month (May) entry.
1595 let may = entry(2026, 5, 2, 8, 0, LogKind::Update, Some("may-a"), "may1");
1596 write_raw_log(&store, &[may]);
1597
1598 // April archive authored out of order: 04-20 first, backdated 04-05 last.
1599 let apr_late = entry(
1600 2026,
1601 4,
1602 20,
1603 9,
1604 0,
1605 LogKind::Create,
1606 Some("apr-b"),
1607 "apr-late",
1608 );
1609 let apr_early = entry(
1610 2026,
1611 4,
1612 5,
1613 9,
1614 0,
1615 LogKind::Ingest,
1616 Some("apr-a"),
1617 "apr-early",
1618 );
1619 let dir = store.root.join("log");
1620 fs::create_dir_all(&dir).unwrap();
1621 let mut arch = String::from(LOG_FRONTMATTER);
1622 arch.push('\n');
1623 arch.push_str(&apr_late.render());
1624 arch.push_str(&apr_early.render());
1625 fs::write(dir.join("2026-04.md"), arch).unwrap();
1626
1627 // since mid-April: the later April entry (04-20) AND the May entry must
1628 // come back; the early April entry (04-05) must not.
1629 let got = Log::since(&store, ts(2026, 4, 15, 0, 0)).unwrap();
1630 let stamps: std::collections::BTreeSet<_> = got.iter().map(|e| e.timestamp).collect();
1631 assert_eq!(
1632 stamps,
1633 [ts(2026, 4, 20, 9, 0), ts(2026, 5, 2, 8, 0)]
1634 .into_iter()
1635 .collect(),
1636 "since(mid-April) must include the out-of-order later April entry \
1637 and the May entry, and exclude the earlier April entry; got {got:?}"
1638 );
1639 }
1640
1641 // ── multi-line notes ──────────────────────────────────────────────────────
1642
1643 #[test]
1644 fn multiline_note_is_preserved() {
1645 let (_d, store) = temp_store();
1646 let e = entry(
1647 2026,
1648 5,
1649 27,
1650 10,
1651 0,
1652 LogKind::Create,
1653 Some("records/x"),
1654 "Line one.\nLine two.\nLine three.",
1655 );
1656 Log::append(&store, &e).unwrap();
1657 let got = Log::tail(&store, 1).unwrap();
1658 assert_eq!(got[0].note, "Line one.\nLine two.\nLine three.");
1659 }
1660
1661 #[test]
1662 fn empty_note_roundtrips_as_empty() {
1663 let (_d, store) = temp_store();
1664 let e = entry(2026, 5, 27, 10, 0, LogKind::Validate, None, "");
1665 Log::append(&store, &e).unwrap();
1666 let got = Log::tail(&store, 1).unwrap();
1667 assert_eq!(got[0], e);
1668 assert_eq!(got[0].note, "");
1669 }
1670
1671 // ── last_validate_at ─────────────────────────────────────────────────────
1672
1673 #[test]
1674 fn last_validate_at_finds_most_recent_validate() {
1675 let (_d, store) = temp_store();
1676 Log::append(
1677 &store,
1678 &entry(2026, 5, 27, 10, 0, LogKind::Validate, None, "first pass"),
1679 )
1680 .unwrap();
1681 Log::append(
1682 &store,
1683 &entry(2026, 5, 27, 10, 5, LogKind::Create, Some("a"), "made a"),
1684 )
1685 .unwrap();
1686 Log::append(
1687 &store,
1688 &entry(2026, 5, 27, 10, 10, LogKind::Validate, None, "second pass"),
1689 )
1690 .unwrap();
1691 Log::append(
1692 &store,
1693 &entry(2026, 5, 27, 10, 15, LogKind::Update, Some("a"), "edit a"),
1694 )
1695 .unwrap();
1696
1697 let last = Log::last_validate_at(&store).unwrap();
1698 assert_eq!(last, Some(ts(2026, 5, 27, 10, 10)));
1699 }
1700
1701 #[test]
1702 fn last_validate_at_none_when_no_validate() {
1703 let (_d, store) = temp_store();
1704 Log::append(
1705 &store,
1706 &entry(2026, 5, 27, 10, 0, LogKind::Create, Some("a"), "x"),
1707 )
1708 .unwrap();
1709 assert_eq!(Log::last_validate_at(&store).unwrap(), None);
1710 }
1711
1712 // ── month-boundary rotation ──────────────────────────────────────────────
1713
1714 #[test]
1715 fn rotation_rolls_prior_months_into_archives() {
1716 let (_d, store) = temp_store();
1717 // Two April entries and one May entry, all written while "current" was
1718 // their own month (append-only chronological order).
1719 let a1 = entry(2026, 4, 10, 9, 0, LogKind::Ingest, Some("apr-a"), "apr one");
1720 let a2 = entry(2026, 4, 20, 9, 0, LogKind::Create, Some("apr-b"), "apr two");
1721 Log::append(&store, &a1).unwrap();
1722 Log::append(&store, &a2).unwrap();
1723
1724 // Before rotation: no archive dir, both April entries in active.
1725 assert!(!store.root.join("log").exists());
1726
1727 // Appending a May entry must roll April into log/2026-04.md.
1728 let m1 = entry(2026, 5, 2, 8, 0, LogKind::Update, Some("may-a"), "may one");
1729 Log::append(&store, &m1).unwrap();
1730
1731 // Archive exists and holds both April entries with frontmatter.
1732 let arch_path = store.root.join("log").join("2026-04.md");
1733 assert!(arch_path.exists(), "expected April archive to be created");
1734 let arch = fs::read_to_string(&arch_path).unwrap();
1735 assert!(arch.starts_with("---\ntype: log\n---\n"));
1736 assert!(arch.contains("## [2026-04-10 09:00] ingest | apr-a"));
1737 assert!(arch.contains("## [2026-04-20 09:00] create | apr-b"));
1738 assert!(arch.contains("apr one"));
1739 assert!(arch.contains("apr two"));
1740
1741 // Active file now holds ONLY the May entry (no April entries).
1742 let active = fs::read_to_string(store.root.join("log.md")).unwrap();
1743 assert!(active.contains("## [2026-05-02 08:00] update | may-a"));
1744 assert!(
1745 !active.contains("apr-a") && !active.contains("apr-b"),
1746 "April entries must be gone from the active file; got:\n{active}"
1747 );
1748
1749 // The full timeline (archives ++ active) is intact and chronological.
1750 let all = Log::tail(&store, 99).unwrap();
1751 assert_eq!(all, vec![a1, a2, m1]);
1752 }
1753
1754 #[test]
1755 fn rotation_groups_distinct_prior_months_into_separate_archives() {
1756 let (_d, store) = temp_store();
1757 // March + April entries accumulate, then a May append rolls BOTH prior
1758 // months into their own archive files.
1759 let mar = entry(2026, 3, 5, 9, 0, LogKind::Ingest, Some("mar"), "march");
1760 let apr = entry(2026, 4, 5, 9, 0, LogKind::Create, Some("apr"), "april");
1761 Log::append(&store, &mar).unwrap();
1762 Log::append(&store, &apr).unwrap();
1763 // At this point April is current, March already rolled into its archive.
1764 assert!(store.root.join("log").join("2026-03.md").exists());
1765
1766 let may = entry(2026, 5, 5, 9, 0, LogKind::Update, Some("may"), "may");
1767 Log::append(&store, &may).unwrap();
1768
1769 assert!(store.root.join("log").join("2026-03.md").exists());
1770 assert!(store.root.join("log").join("2026-04.md").exists());
1771
1772 // Each archive holds only its own month.
1773 let mar_arch = fs::read_to_string(store.root.join("log").join("2026-03.md")).unwrap();
1774 let apr_arch = fs::read_to_string(store.root.join("log").join("2026-04.md")).unwrap();
1775 assert!(mar_arch.contains("mar") && !mar_arch.contains("apr"));
1776 assert!(apr_arch.contains("apr") && !apr_arch.contains("mar"));
1777
1778 // Active holds only May.
1779 let active = fs::read_to_string(store.root.join("log.md")).unwrap();
1780 assert!(active.contains("may") && !active.contains("mar") && !active.contains("apr"));
1781
1782 // Timeline intact and ordered across both archives + active.
1783 let all = Log::tail(&store, 99).unwrap();
1784 assert_eq!(all, vec![mar, apr, may]);
1785 }
1786
1787 #[test]
1788 fn tail_crosses_into_archive_when_n_spans_month_boundary() {
1789 let (_d, store) = temp_store();
1790 let a1 = entry(2026, 4, 10, 9, 0, LogKind::Ingest, Some("apr-a"), "apr1");
1791 let a2 = entry(2026, 4, 20, 9, 0, LogKind::Create, Some("apr-b"), "apr2");
1792 let m1 = entry(2026, 5, 2, 8, 0, LogKind::Update, Some("may-a"), "may1");
1793 let m2 = entry(2026, 5, 3, 8, 0, LogKind::Update, Some("may-b"), "may2");
1794 for e in [&a1, &a2, &m1, &m2] {
1795 Log::append(&store, e).unwrap();
1796 }
1797 // April is now archived; active holds only May. tail(3) must reach back
1798 // into the archive for the third-newest entry.
1799 let tail3 = Log::tail(&store, 3).unwrap();
1800 assert_eq!(tail3, vec![a2.clone(), m1.clone(), m2.clone()]);
1801
1802 // tail within the active month does NOT need the archive but is still
1803 // correct.
1804 let tail2 = Log::tail(&store, 2).unwrap();
1805 assert_eq!(tail2, vec![m1, m2]);
1806 }
1807
1808 #[test]
1809 fn since_crosses_into_archive_and_early_stops() {
1810 let (_d, store) = temp_store();
1811 let a1 = entry(2026, 4, 10, 9, 0, LogKind::Ingest, Some("apr-a"), "apr1");
1812 let a2 = entry(2026, 4, 20, 9, 0, LogKind::Create, Some("apr-b"), "apr2");
1813 let m1 = entry(2026, 5, 2, 8, 0, LogKind::Update, Some("may-a"), "may1");
1814 for e in [&a1, &a2, &m1] {
1815 Log::append(&store, e).unwrap();
1816 }
1817 // since a mid-April time: must include the later April entry (from the
1818 // archive) and the May entry, but not the earlier April one.
1819 let got = Log::since(&store, ts(2026, 4, 15, 0, 0)).unwrap();
1820 assert_eq!(got, vec![a2, m1]);
1821 }
1822
1823 #[test]
1824 fn last_validate_at_crosses_into_archive() {
1825 let (_d, store) = temp_store();
1826 // A validate in April, then non-validate work that rolls April away.
1827 Log::append(
1828 &store,
1829 &entry(2026, 4, 10, 9, 0, LogKind::Validate, None, "apr validate"),
1830 )
1831 .unwrap();
1832 Log::append(
1833 &store,
1834 &entry(2026, 5, 2, 8, 0, LogKind::Update, Some("may-a"), "may work"),
1835 )
1836 .unwrap();
1837 // Active has only the May update; the most-recent validate lives in the
1838 // April archive and must still be found.
1839 let last = Log::last_validate_at(&store).unwrap();
1840 assert_eq!(last, Some(ts(2026, 4, 10, 9, 0)));
1841 }
1842
1843 // ── reverse-read correctness on a large (multi-block) log ────────────────
1844
1845 #[test]
1846 fn reverse_read_correct_on_large_single_month_log() {
1847 let (_d, store) = temp_store();
1848 // Append many same-month entries with chunky multi-line notes so the
1849 // file spans well past one REVERSE_BLOCK (8 KiB). Timestamps are
1850 // strictly increasing (a real append-only log is monotonic): each entry
1851 // is 3 minutes after the previous, all within June, so physical order
1852 // equals chronological order and the last-k-physical ARE the k-newest.
1853 let n = 400usize;
1854 let mut expected: Vec<LogEntry> = Vec::new();
1855 for i in 0..n {
1856 let total_min = (i as u32) * 3;
1857 let day = 1 + total_min / (24 * 60);
1858 let hour = (total_min / 60) % 24;
1859 let min = total_min % 60;
1860 // Unique, multi-line note to bulk up the file and detect mis-parses.
1861 let note = format!(
1862 "entry number {i}\nbody line A for {i}\nbody line B for {i} with padding {}",
1863 "x".repeat(40)
1864 );
1865 let e = entry(
1866 2026,
1867 6,
1868 day,
1869 hour,
1870 min,
1871 LogKind::Update,
1872 Some(&format!("records/item-{i:04}")),
1873 ¬e,
1874 );
1875 Log::append(&store, &e).unwrap();
1876 expected.push(e);
1877 }
1878
1879 // File must actually be multi-block to exercise the backward reader.
1880 let size = fs::metadata(store.root.join("log.md")).unwrap().len();
1881 assert!(
1882 size > (REVERSE_BLOCK as u64) * 2,
1883 "test log not large enough ({size} bytes) to exercise multi-block reverse-read"
1884 );
1885
1886 // tail(5) must equal the 5 newest, exactly.
1887 let tail5 = Log::tail(&store, 5).unwrap();
1888 assert_eq!(tail5, expected[n - 5..].to_vec());
1889
1890 // tail(50) must equal the 50 newest.
1891 let tail50 = Log::tail(&store, 50).unwrap();
1892 assert_eq!(tail50, expected[n - 50..].to_vec());
1893
1894 // tail(all) must reconstruct the whole timeline in order.
1895 let all = Log::tail(&store, n + 10).unwrap();
1896 assert_eq!(all.len(), n);
1897 assert_eq!(all, expected);
1898 }
1899
1900 // ── tail on OUT-OF-ORDER logs (newest-by-timestamp, not last-physical) ────
1901 //
1902 // The append-only contract is non-decreasing time order, but it's only a
1903 // `LOG_OUT_OF_ORDER` warning when violated (corrective entries land below
1904 // the entry they correct; backdated / clock-skewed writes; `merge=union`
1905 // clone merges). `tail N` must return the N newest *by timestamp*, never the
1906 // last N *physical* entries.
1907
1908 /// Write `log.md` verbatim from rendered entries in the given **physical
1909 /// (file) order**, bypassing `Log::append` so the test controls on-disk
1910 /// order exactly (append never reorders within a month, but this is the
1911 /// clearest way to pin a specific physical layout).
1912 fn write_log_physical(store: &Store, entries: &[LogEntry]) {
1913 let mut body = String::new();
1914 for e in entries {
1915 body.push_str(&e.render());
1916 }
1917 let full = compose_active(LOG_FRONTMATTER, &body);
1918 fs::write(store.root.join("log.md"), full).expect("write log.md");
1919 }
1920
1921 #[test]
1922 fn tail_returns_newest_by_timestamp_on_demonstrated_out_of_order_log() {
1923 // The exact case from the review finding: physical order 10:10, 10:05,
1924 // 10:00 (a backdated entry tail). The OLD code returned the last two
1925 // physical entries {10:05, 10:00}; the correct answer is the two newest
1926 // by time {10:05, 10:10}.
1927 let (_d, store) = temp_store();
1928 let e_1010 = entry(2026, 5, 27, 10, 10, LogKind::Update, Some("c"), "ten-ten");
1929 let e_1005 = entry(
1930 2026,
1931 5,
1932 27,
1933 10,
1934 5,
1935 LogKind::Create,
1936 Some("b"),
1937 "ten-oh-five",
1938 );
1939 let e_1000 = entry(2026, 5, 27, 10, 0, LogKind::Ingest, Some("a"), "ten-oh-oh");
1940 // Physical order: newest first, then the two older ones — out of order.
1941 write_log_physical(&store, &[e_1010.clone(), e_1005.clone(), e_1000.clone()]);
1942
1943 let tail2 = Log::tail(&store, 2).unwrap();
1944 assert_eq!(
1945 tail2,
1946 vec![e_1005.clone(), e_1010.clone()],
1947 "tail(2) must be the two NEWEST by timestamp (chronological), \
1948 not the last two physical entries"
1949 );
1950 // The newest entry must be present and the oldest absent.
1951 assert!(tail2.contains(&e_1010), "newest (10:10) must be included");
1952 assert!(!tail2.contains(&e_1000), "oldest (10:00) must be excluded");
1953
1954 // tail(1) is just the single newest.
1955 assert_eq!(Log::tail(&store, 1).unwrap(), vec![e_1010.clone()]);
1956 // tail(all) is the full set in chronological order.
1957 assert_eq!(Log::tail(&store, 99).unwrap(), vec![e_1000, e_1005, e_1010]);
1958 }
1959
1960 #[test]
1961 fn tail_no_early_stop_when_newer_entry_sits_before_an_older_one() {
1962 // Guards the unsound within-file early stop: a newer entry (10:50) sits
1963 // PHYSICALLY BEFORE a much older one (10:00). Reading newest-physical-
1964 // first, the scan meets 10:00 before 10:50; any "stop at the first entry
1965 // below the window minimum" rule would bail and drop 10:50.
1966 //
1967 // Physical (top→bottom): 10:55, 10:10, 10:50, 10:00.
1968 // Reverse-scan order: 10:00, 10:50, 10:10, 10:55.
1969 let (_d, store) = temp_store();
1970 let e55 = entry(2026, 5, 27, 10, 55, LogKind::Update, Some("x55"), "55");
1971 let e10 = entry(2026, 5, 27, 10, 10, LogKind::Update, Some("x10"), "10");
1972 let e50 = entry(2026, 5, 27, 10, 50, LogKind::Update, Some("x50"), "50");
1973 let e00 = entry(2026, 5, 27, 10, 0, LogKind::Update, Some("x00"), "00");
1974 write_log_physical(
1975 &store,
1976 &[e55.clone(), e10.clone(), e50.clone(), e00.clone()],
1977 );
1978
1979 // The two newest by timestamp are 10:55 and 10:50 — NOT the early-stop
1980 // victim 10:10, and NOT the last-physical 10:00.
1981 let tail2 = Log::tail(&store, 2).unwrap();
1982 assert_eq!(tail2, vec![e50.clone(), e55.clone()]);
1983
1984 let tail3 = Log::tail(&store, 3).unwrap();
1985 assert_eq!(tail3, vec![e10.clone(), e50.clone(), e55.clone()]);
1986 }
1987
1988 #[test]
1989 fn tail_orders_equal_timestamps_by_physical_recency() {
1990 // Three entries share 10:00; one is at 09:59. tail(2) must keep both
1991 // 10:00 entries, and among the equal pair the one appended LATER
1992 // (physically last) sorts last ("newest" = most-recently recorded).
1993 let (_d, store) = temp_store();
1994 let early = entry(2026, 5, 27, 9, 59, LogKind::Create, Some("early"), "before");
1995 let tie_a = entry(
1996 2026,
1997 5,
1998 27,
1999 10,
2000 0,
2001 LogKind::Update,
2002 Some("tie-a"),
2003 "first 10:00",
2004 );
2005 let tie_b = entry(
2006 2026,
2007 5,
2008 27,
2009 10,
2010 0,
2011 LogKind::Update,
2012 Some("tie-b"),
2013 "second 10:00",
2014 );
2015 // Physical append order: early, tie_a, tie_b.
2016 write_log_physical(&store, &[early.clone(), tie_a.clone(), tie_b.clone()]);
2017
2018 let tail2 = Log::tail(&store, 2).unwrap();
2019 assert_eq!(
2020 tail2,
2021 vec![tie_a.clone(), tie_b.clone()],
2022 "both 10:00 entries kept, physically-later one (tie_b) last; 09:59 dropped"
2023 );
2024 // tail(1) keeps only the most-recently-recorded of the equal pair.
2025 assert_eq!(Log::tail(&store, 1).unwrap(), vec![tie_b]);
2026 }
2027
2028 #[test]
2029 fn tail_finds_newest_across_a_backdated_entry_spanning_the_month_boundary() {
2030 // A backdated entry can land physically after newer entries even across
2031 // a rotation: append May entries, then a June entry (rolls May to its
2032 // archive), then append a May-dated correction — it goes into the ACTIVE
2033 // file, physically after June. tail must still rank by timestamp, so the
2034 // June entry stays newest and the backdated May entry is not mistaken
2035 // for the tail.
2036 let (_d, store) = temp_store();
2037 let may1 = entry(2026, 5, 10, 9, 0, LogKind::Ingest, Some("may-1"), "may one");
2038 let may2 = entry(2026, 5, 20, 9, 0, LogKind::Create, Some("may-2"), "may two");
2039 let jun1 = entry(2026, 6, 2, 8, 0, LogKind::Update, Some("jun-1"), "jun one");
2040 Log::append(&store, &may1).unwrap();
2041 Log::append(&store, &may2).unwrap();
2042 Log::append(&store, &jun1).unwrap(); // rotates May -> log/2026-05.md
2043 assert!(store.root.join("log").join("2026-05.md").exists());
2044
2045 // A backdated May correction, appended now: it lands in the active file
2046 // (its month May is not strictly before the active month June), so the
2047 // active file is physically [jun1, may_corr] — out of order.
2048 let may_corr = entry(
2049 2026,
2050 5,
2051 25,
2052 9,
2053 0,
2054 LogKind::Update,
2055 Some("may-2"),
2056 "may correction",
2057 );
2058 Log::append(&store, &may_corr).unwrap();
2059 let active = fs::read_to_string(store.root.join("log.md")).unwrap();
2060 assert!(
2061 active.contains("jun-1") && active.contains("may correction"),
2062 "backdated May entry should be in the active file alongside June; got:\n{active}"
2063 );
2064
2065 // The single newest by timestamp is the June entry, even though the
2066 // backdated May entry is physically last.
2067 assert_eq!(Log::tail(&store, 1).unwrap(), vec![jun1.clone()]);
2068
2069 // tail(2): the two newest by time are may_corr (05-25) and jun1 (06-02).
2070 let tail2 = Log::tail(&store, 2).unwrap();
2071 assert_eq!(tail2, vec![may_corr.clone(), jun1.clone()]);
2072
2073 // tail(3) must reach into the May archive for the third-newest (may2,
2074 // 05-20), proving archive crossing still works on an out-of-order store.
2075 let tail3 = Log::tail(&store, 3).unwrap();
2076 assert_eq!(tail3, vec![may2.clone(), may_corr.clone(), jun1.clone()]);
2077
2078 // tail(all) reconstructs the whole timeline in chronological order.
2079 let all = Log::tail(&store, 99).unwrap();
2080 assert_eq!(all, vec![may1, may2, may_corr, jun1]);
2081 }
2082
2083 #[test]
2084 fn parse_entries_skips_unparseable_header_folding_into_body() {
2085 // A `## [` line that is NOT a valid header should not start a new entry;
2086 // it folds into the preceding entry's note. This guards the
2087 // parse_entries header-validation branch.
2088 let text = "\
2089## [2026-05-27 10:00] create | records/x
2090Body mentions a literal: ## [not a real header here]
2091More body.
2092
2093## [2026-05-27 10:05] update | records/y
2094Second.
2095";
2096 let entries = parse_entries(text);
2097 assert_eq!(entries.len(), 2);
2098 assert_eq!(entries[0].kind, LogKind::Create);
2099 assert!(entries[0].note.contains("## [not a real header here]"));
2100 assert!(entries[0].note.contains("More body."));
2101 assert_eq!(entries[1].kind, LogKind::Update);
2102 assert_eq!(entries[1].note, "Second.");
2103 }
2104
2105 // ── append-only: corrective entries go on the end ─────────────────────────
2106
2107 #[test]
2108 fn append_only_corrective_entry_goes_on_end_without_rewriting() {
2109 let (_d, store) = temp_store();
2110 let original = entry(
2111 2026,
2112 5,
2113 27,
2114 10,
2115 0,
2116 LogKind::Update,
2117 Some("records/northstar"),
2118 "Seat count 120 -> 175.",
2119 );
2120 Log::append(&store, &original).unwrap();
2121 let after_first = fs::read_to_string(store.root.join("log.md")).unwrap();
2122
2123 // A correction is a NEW entry appended on the end; the original text is
2124 // left byte-for-byte intact (append-only contract: no rewrite API).
2125 let correction = entry(
2126 2026,
2127 5,
2128 27,
2129 11,
2130 0,
2131 LogKind::Update,
2132 Some("records/northstar"),
2133 "Correction: seat count is 165, not 175.",
2134 );
2135 Log::append(&store, &correction).unwrap();
2136 let after_second = fs::read_to_string(store.root.join("log.md")).unwrap();
2137
2138 assert!(
2139 after_second.starts_with(&after_first),
2140 "appending must not rewrite earlier bytes"
2141 );
2142 assert!(after_second.contains("Correction: seat count is 165, not 175."));
2143
2144 // Both entries are readable, in order.
2145 let all = Log::tail(&store, 99).unwrap();
2146 assert_eq!(all, vec![original, correction]);
2147 }
2148
2149 // ── concurrent append safety (atomic via temp-file rename) ────────────────
2150
2151 #[test]
2152 fn concurrent_appends_are_atomic_and_total() {
2153 use std::sync::{Arc, Barrier};
2154 use std::thread;
2155
2156 let (_d, store) = temp_store();
2157 // Seed the file so all threads take the read-modify-write path.
2158 Log::append(
2159 &store,
2160 &entry(2026, 7, 1, 0, 0, LogKind::Create, Some("seed"), "seed"),
2161 )
2162 .unwrap();
2163
2164 let threads = 8usize;
2165 let per = 25usize;
2166 let barrier = Arc::new(Barrier::new(threads));
2167 let store = Arc::new(store);
2168
2169 let mut handles = Vec::new();
2170 for tnum in 0..threads {
2171 let b = Arc::clone(&barrier);
2172 let s = Arc::clone(&store);
2173 handles.push(thread::spawn(move || {
2174 b.wait();
2175 for i in 0..per {
2176 let e = entry(
2177 2026,
2178 7,
2179 1,
2180 (tnum % 24) as u32,
2181 (i % 60) as u32,
2182 LogKind::Update,
2183 Some(&format!("t{tnum}-i{i}")),
2184 &format!("thread {tnum} item {i}"),
2185 );
2186 Log::append(&s, &e).unwrap();
2187 }
2188 }));
2189 }
2190 for h in handles {
2191 h.join().unwrap();
2192 }
2193
2194 // The atomic temp-file-rename write means no append truncates or
2195 // corrupts another: the file must remain parseable and every line of
2196 // every entry header must be well-formed. Crucially, no entry should be
2197 // lost to a torn write of the *content already on disk* — though
2198 // interleaved read-modify-write WILL drop some appends (last-writer-
2199 // wins on the snapshot). We therefore assert integrity + that the file
2200 // never went empty / corrupt, not an exact count.
2201 let content = fs::read_to_string(store.root.join("log.md")).unwrap();
2202 assert!(content.starts_with("---\ntype: log\n---\n"));
2203
2204 // Every `## [` line must parse as a valid header (no half-written line).
2205 for line in content.lines() {
2206 if line.starts_with("## [") {
2207 assert!(
2208 Log::parse_header(line).is_some(),
2209 "corrupt/torn header line on disk: {line:?}"
2210 );
2211 }
2212 }
2213
2214 // The seed entry must survive (it was written before the race and
2215 // every snapshot included it).
2216 assert!(content.contains("## [2026-07-01 00:00] create | seed"));
2217
2218 // The reverse reader must still produce a clean, fully-parseable view.
2219 let all = Log::tail(&store, 10_000).unwrap();
2220 assert!(!all.is_empty());
2221 // No duplicate adjacent identical headers from a torn write: every
2222 // returned entry must have a recognized-or-custom kind and a parseable
2223 // timestamp (already guaranteed by parse), and the list must be
2224 // internally consistent (re-render → re-parse identity for each).
2225 for e in &all {
2226 let rendered = e.render();
2227 let reparsed = parse_single_entry(&rendered).unwrap();
2228 assert_eq!(&reparsed, e);
2229 }
2230 }
2231
2232 // ── render/parse identity ────────────────────────────────────────────────
2233
2234 #[test]
2235 fn render_then_parse_is_identity() {
2236 let cases = vec![
2237 entry(
2238 2026,
2239 1,
2240 2,
2241 3,
2242 4,
2243 LogKind::Ingest,
2244 Some("sources/a.eml"),
2245 "n",
2246 ),
2247 entry(
2248 2026,
2249 12,
2250 31,
2251 23,
2252 59,
2253 LogKind::Validate,
2254 None,
2255 "PASS - 0 errors",
2256 ),
2257 entry(
2258 2026,
2259 6,
2260 15,
2261 12,
2262 30,
2263 LogKind::Custom("proposal".to_string()),
2264 Some("records/p"),
2265 "multi\nline\nnote",
2266 ),
2267 entry(2026, 6, 15, 12, 30, LogKind::Contradiction, Some("obj"), ""),
2268 ];
2269 for e in cases {
2270 let rendered = e.render();
2271 let parsed = parse_single_entry(&rendered).unwrap_or_else(|| {
2272 panic!("failed to reparse rendered entry:\n{rendered}");
2273 });
2274 assert_eq!(parsed, e, "round-trip mismatch for {e:?}");
2275 }
2276 }
2277
2278 // ── regression: rotation re-roll must not duplicate archive entries (#3) ──
2279
2280 /// Count occurrences of `needle` in `haystack` (non-overlapping).
2281 fn count_occurrences(haystack: &str, needle: &str) -> usize {
2282 haystack.matches(needle).count()
2283 }
2284
2285 #[test]
2286 fn regression_archive_reroll_is_idempotent_after_interrupted_rotation() {
2287 // Reconstructs the finding's exact failure window: rotation is two
2288 // non-atomic durable writes — (1) roll prior-month entries into the
2289 // archive, then (2) trim the active file. If the process crashes or the
2290 // active rewrite errors AFTER step (1) commits, the prior-month entries
2291 // stay in the untrimmed active file, the agent retries, and the retry
2292 // re-rolls the SAME entries into the archive a second time. The
2293 // mechanism is precisely a second `append_to_archive` of identical
2294 // entries onto an archive that already holds them.
2295 let (_d, store) = temp_store();
2296 let dir = archive_dir(&store);
2297 let arch = archive_path(&store, 2026, 4);
2298
2299 let apr1 = entry(2026, 4, 10, 9, 0, LogKind::Ingest, Some("apr-a"), "apr one");
2300 let apr2 = entry(2026, 4, 20, 9, 0, LogKind::Create, Some("apr-b"), "apr two");
2301 let month = [apr1.clone(), apr2.clone()];
2302
2303 // First roll (the committed step-(1) write before the crash).
2304 fs::create_dir_all(&dir).unwrap();
2305 append_to_archive(&arch, &month).unwrap();
2306
2307 // The retry re-rolls the identical prior-month entries. Pre-fix this
2308 // blindly concatenated, doubling every entry; do it twice to prove the
2309 // amplification a real retry loop would cause is fully suppressed.
2310 append_to_archive(&arch, &month).unwrap();
2311 append_to_archive(&arch, &month).unwrap();
2312
2313 let archived = fs::read_to_string(&arch).unwrap();
2314 // Each entry header must appear EXACTLY once despite the re-rolls.
2315 assert_eq!(
2316 count_occurrences(&archived, "## [2026-04-10 09:00] ingest | apr-a"),
2317 1,
2318 "re-rolled archive duplicated the first April entry; got:\n{archived}"
2319 );
2320 assert_eq!(
2321 count_occurrences(&archived, "## [2026-04-20 09:00] create | apr-b"),
2322 1,
2323 "re-rolled archive duplicated the second April entry; got:\n{archived}"
2324 );
2325
2326 // And the reader surface (`since`) must return each entry once, not the
2327 // duplicated set the pre-fix archive would have yielded.
2328 let got = Log::since(&store, ts(2026, 4, 1, 0, 0)).unwrap();
2329 assert_eq!(
2330 got,
2331 vec![apr1, apr2],
2332 "since over the re-rolled archive must return each April entry once"
2333 );
2334 }
2335
2336 #[test]
2337 fn regression_rotation_reroll_after_active_untrimmed_does_not_duplicate() {
2338 // End-to-end variant driving the real `Log::append` rotation path. We
2339 // rotate April into its archive via a May append, then SIMULATE the
2340 // partial failure by restoring the pre-trim active file (April + May)
2341 // and re-running `append` — exactly the state a crash-between-the-two-
2342 // writes / failed-active-rewrite + agent-retry produces. The archive
2343 // must still hold each April entry once.
2344 let (_d, store) = temp_store();
2345 let apr1 = entry(2026, 4, 10, 9, 0, LogKind::Ingest, Some("apr-a"), "apr one");
2346 let apr2 = entry(2026, 4, 20, 9, 0, LogKind::Create, Some("apr-b"), "apr two");
2347 Log::append(&store, &apr1).unwrap();
2348 Log::append(&store, &apr2).unwrap();
2349
2350 // Snapshot the active file holding both April entries (this is what is
2351 // still on disk if the post-rotation active rewrite never lands).
2352 let active_path = active_log_path(&store);
2353 let pre_rotation_active = fs::read_to_string(&active_path).unwrap();
2354
2355 // A May append rotates April out and trims the active file.
2356 let may = entry(2026, 5, 2, 8, 0, LogKind::Update, Some("may-a"), "may one");
2357 Log::append(&store, &may).unwrap();
2358 let arch = archive_path(&store, 2026, 4);
2359 assert!(arch.exists(), "April should have rotated to its archive");
2360
2361 // Simulate the crash/error: the active rewrite never persisted, so the
2362 // active file still contains the (now also archived) April entries.
2363 fs::write(&active_path, &pre_rotation_active).unwrap();
2364
2365 // The agent retries the append. Re-partitioning sees April as prior
2366 // months again and re-rolls them — which must NOT duplicate the archive.
2367 let may2 = entry(2026, 5, 3, 8, 0, LogKind::Update, Some("may-b"), "may two");
2368 Log::append(&store, &may2).unwrap();
2369
2370 let archived = fs::read_to_string(&arch).unwrap();
2371 assert_eq!(
2372 count_occurrences(&archived, "## [2026-04-10 09:00] ingest | apr-a"),
2373 1,
2374 "retried rotation duplicated an April entry in the archive; got:\n{archived}"
2375 );
2376 assert_eq!(
2377 count_occurrences(&archived, "## [2026-04-20 09:00] create | apr-b"),
2378 1,
2379 "retried rotation duplicated an April entry in the archive; got:\n{archived}"
2380 );
2381 }
2382
2383 // ── regression: reverse reader keeps a `## [` continuation note line (#10) ─
2384
2385 #[test]
2386 fn regression_reverse_reader_preserves_note_line_starting_with_bracket_header() {
2387 // SPEC permits a note of "one or more lines" with no restriction on a
2388 // continuation line starting at column 0 with `## [`. The forward parser
2389 // folds such an unparseable `## [` line into the note; the reverse
2390 // reader (tail/since/last_validate_at) must agree, not split on it.
2391 let (_d, store) = temp_store();
2392 let multi = "First line.\n## [draft outline] more\nThird line.";
2393 let e = entry(
2394 2026,
2395 5,
2396 27,
2397 10,
2398 0,
2399 LogKind::Update,
2400 Some("records/x"),
2401 multi,
2402 );
2403 // Author the log verbatim (render writes the note as-is); this is the
2404 // on-disk shape a hand-written / appended multi-line note produces.
2405 write_raw_log(&store, std::slice::from_ref(&e));
2406
2407 // Pre-fix: header_offsets treated `## [draft outline] more` as a second
2408 // entry boundary, truncating the note to "First line." and dropping the
2409 // carved (non-header) fragment. Post-fix: the full note survives.
2410 let got = Log::tail(&store, 1).unwrap();
2411 assert_eq!(got.len(), 1, "the single entry must be returned");
2412 assert_eq!(
2413 got[0].note, multi,
2414 "reverse reader truncated the note at the `## [` continuation line; \
2415 got {:?}",
2416 got[0].note
2417 );
2418 assert_eq!(got[0], e, "the whole entry must round-trip through tail");
2419
2420 // `since` (the other reverse-reading surface) must agree.
2421 let since = Log::since(&store, ts(2026, 5, 27, 9, 0)).unwrap();
2422 assert_eq!(since, vec![e]);
2423 }
2424
2425 // ── regression: `since` archive pruning uses the UTC month, not local (#11) ─
2426
2427 /// A `DateTime<FixedOffset>` at the given fixed offset (hours east of UTC).
2428 fn ts_offset(
2429 y: i32,
2430 mo: u32,
2431 d: u32,
2432 h: u32,
2433 mi: u32,
2434 offset_hours: i32,
2435 ) -> DateTime<FixedOffset> {
2436 let naive = chrono::NaiveDate::from_ymd_opt(y, mo, d)
2437 .unwrap()
2438 .and_hms_opt(h, mi, 0)
2439 .unwrap();
2440 FixedOffset::east_opt(offset_hours * 3600)
2441 .unwrap()
2442 .from_local_datetime(&naive)
2443 .single()
2444 .unwrap()
2445 }
2446
2447 #[test]
2448 fn regression_since_prunes_archives_on_utc_month_not_local_offset_month() {
2449 // Archive months are bucketed on the UTC calendar. A `since` cutoff with
2450 // a non-UTC offset near a month boundary must not prune an archive whose
2451 // UTC month equals the cutoff's UTC month just because the cutoff's
2452 // LOCAL month is later.
2453 let (_d, store) = temp_store();
2454
2455 // April archive: an entry late on 2026-04-30 at 18:00 UTC.
2456 let apr = entry(
2457 2026,
2458 4,
2459 30,
2460 18,
2461 0,
2462 LogKind::Update,
2463 Some("apr-late"),
2464 "april late",
2465 );
2466 let dir = archive_dir(&store);
2467 fs::create_dir_all(&dir).unwrap();
2468 let mut arch = String::from(LOG_FRONTMATTER);
2469 arch.push('\n');
2470 arch.push_str(&apr.render());
2471 fs::write(archive_path(&store, 2026, 4), arch).unwrap();
2472
2473 // Active file: a clean May entry, so an archive scan is actually needed.
2474 let may = entry(2026, 5, 5, 8, 0, LogKind::Update, Some("may-a"), "may one");
2475 write_raw_log(&store, std::slice::from_ref(&may));
2476
2477 // Cutoff 2026-05-01T00:30:00+07:00 == 2026-04-30T17:30:00Z. The April
2478 // 18:00 UTC entry is strictly newer than this instant.
2479 let cutoff = ts_offset(2026, 5, 1, 0, 30, 7);
2480 // Sanity: the cutoff's UTC month is April, its local month is May.
2481 assert_eq!((cutoff.year(), cutoff.month()), (2026, 5));
2482 assert_eq!(
2483 (
2484 cutoff.with_timezone(&Utc).year(),
2485 cutoff.with_timezone(&Utc).month()
2486 ),
2487 (2026, 4)
2488 );
2489
2490 // Pre-fix: cutoff_ym = (2026, 5) from local fields, so the (2026, 4)
2491 // archive was pruned and the genuinely-newer 18:00 UTC entry was dropped
2492 // — `since` returned only the May entry. Post-fix: cutoff_ym is UTC
2493 // (2026, 4), the April archive is scanned, and both come back.
2494 let got = Log::since(&store, cutoff).unwrap();
2495 let stamps: std::collections::BTreeSet<_> = got.iter().map(|e| e.timestamp).collect();
2496 assert_eq!(
2497 stamps,
2498 [ts(2026, 4, 30, 18, 0), ts(2026, 5, 5, 8, 0)]
2499 .into_iter()
2500 .collect(),
2501 "since(non-UTC cutoff near a month boundary) must include the April \
2502 archive entry newer than the cutoff instant; got {got:?}"
2503 );
2504 }
2505
2506 // ── regression: header-shaped note line corrupts the append-only log (#critical)
2507
2508 #[test]
2509 fn note_line_shaped_like_a_header_is_escaped_and_round_trips() {
2510 // A `contradiction` note quoting an earlier entry header is the
2511 // demonstrated corruption: the verbatim `## [2020-01-01 00:00] delete |
2512 // …` line was parsed as a REAL entry on readback (fabricated entry, real
2513 // note truncated). With write-path escaping it stays note body.
2514 let (_d, store) = temp_store();
2515 let note = "quoting earlier entry:\n## [2020-01-01 00:00] delete | records/contacts/jane.md\nend of quote";
2516 let e = entry(
2517 2026,
2518 6,
2519 11,
2520 4,
2521 41,
2522 LogKind::Contradiction,
2523 Some("records/contacts/jane.md"),
2524 note,
2525 );
2526 Log::append(&store, &e).unwrap();
2527
2528 // On disk: the header-shaped note line must NOT sit at column 0 as a
2529 // `## [` header — `grep "^## \["` must see exactly the one real header.
2530 let raw = fs::read_to_string(store.root.join("log.md")).unwrap();
2531 let header_lines = raw.lines().filter(|l| l.starts_with("## [")).count();
2532 assert_eq!(
2533 header_lines, 1,
2534 "exactly one real entry header may sit at column 0; got:\n{raw}"
2535 );
2536
2537 // Readback returns ONE entry, with the full note intact (no fabricated
2538 // 2020 entry, no truncation).
2539 let got = Log::tail(&store, 10).unwrap();
2540 assert_eq!(got.len(), 1, "exactly one entry; got {got:?}");
2541 assert_eq!(got[0].note, note, "note must round-trip verbatim");
2542 assert_eq!(got[0], e);
2543 let since = Log::since(&store, ts(2026, 1, 1, 0, 0)).unwrap();
2544 assert_eq!(since, vec![e.clone()]);
2545 }
2546
2547 #[test]
2548 fn header_shaped_note_survives_a_later_rotation_uncorrupted() {
2549 // Physical corruption: pre-fix, the fabricated past-dated pseudo-entry
2550 // (year 2020 < current) was rolled into an archive on the NEXT append,
2551 // splitting the real note. With escaping the line is note text, so a
2552 // later append never sees a phantom prior-month entry to roll out.
2553 let (_d, store) = temp_store();
2554 let note = "see\n## [2020-01-01 00:00] delete | records/x.md\nbelow";
2555 let first = entry(
2556 2026,
2557 6,
2558 11,
2559 4,
2560 41,
2561 LogKind::Contradiction,
2562 Some("records/x.md"),
2563 note,
2564 );
2565 Log::append(&store, &first).unwrap();
2566
2567 // Append another current-month entry — the path that re-parses + may
2568 // rotate. No 2020 archive must be created and the first note stays whole.
2569 let second = entry(
2570 2026,
2571 6,
2572 11,
2573 5,
2574 0,
2575 LogKind::Update,
2576 Some("records/y.md"),
2577 "y",
2578 );
2579 Log::append(&store, &second).unwrap();
2580
2581 assert!(
2582 !store.root.join("log").join("2020-01.md").exists(),
2583 "a header-shaped note line must not fabricate a 2020 archive"
2584 );
2585 let got = Log::tail(&store, 10).unwrap();
2586 assert_eq!(got.len(), 2, "two real entries only; got {got:?}");
2587 let first_back = got
2588 .iter()
2589 .find(|e| e.object.as_deref() == Some("records/x.md"));
2590 assert_eq!(
2591 first_back.map(|e| e.note.as_str()),
2592 Some(note),
2593 "the header-shaped note must survive the rotation pass intact"
2594 );
2595 }
2596
2597 #[test]
2598 fn escape_unescape_note_line_round_trips_including_literal_backslash() {
2599 // The escape must be lossless for arbitrary note lines, including a line
2600 // the author genuinely wrote starting with `\` before a header shape.
2601 let valid_header = "## [2020-01-01 00:00] delete | x";
2602 // A real header shape: escaped on write, restored on read.
2603 assert_eq!(
2604 &*escape_note_line(valid_header),
2605 &format!("\\{valid_header}")
2606 );
2607 let escaped = escape_note_line(valid_header).into_owned();
2608 assert_eq!(&*unescape_note_line(&escaped), valid_header);
2609 // An already-`\`-prefixed header-shape line escapes to two backslashes
2610 // and restores to one (never collapses to a bare header).
2611 let pre = format!("\\{valid_header}");
2612 assert_eq!(&*escape_note_line(&pre), &format!("\\{pre}"));
2613 let pre_escaped = escape_note_line(&pre).into_owned();
2614 assert_eq!(&*unescape_note_line(&pre_escaped), &pre);
2615 // Ordinary text (including a `\` that does NOT lead into a header) is
2616 // untouched both ways.
2617 for plain in ["plain note", "## [not a header]", "\\not a header", ""] {
2618 assert_eq!(&*escape_note_line(plain), plain);
2619 assert_eq!(&*unescape_note_line(plain), plain);
2620 }
2621 }
2622
2623 // ── regression: reverse reader scans each block once (no O(file²)) (#perf) ──
2624
2625 #[test]
2626 fn reverse_read_correct_with_header_straddling_a_block_boundary() {
2627 // The incremental per-block header scan must still catch a `## [` marker
2628 // whose `#` falls in one block but whose bytes extend into the already-
2629 // scanned region. Build a log whose total size crosses several blocks and
2630 // verify a full read reconstructs every entry — the straddle case is hit
2631 // by construction across the many block boundaries.
2632 let (_d, store) = temp_store();
2633 let n = 600usize;
2634 let mut expected: Vec<LogEntry> = Vec::new();
2635 for i in 0..n {
2636 let total_min = (i as u32) * 2;
2637 let day = 1 + total_min / (24 * 60);
2638 let hour = (total_min / 60) % 24;
2639 let min = total_min % 60;
2640 // Vary note length so headers land at many offsets relative to the
2641 // fixed 8 KiB block grid, exercising boundary straddles.
2642 let note = format!("note {i} {}", "y".repeat(i % 97));
2643 let e = entry(
2644 2026,
2645 6,
2646 day,
2647 hour,
2648 min,
2649 LogKind::Update,
2650 Some(&format!("records/item-{i:05}")),
2651 ¬e,
2652 );
2653 Log::append(&store, &e).unwrap();
2654 expected.push(e);
2655 }
2656 let size = fs::metadata(store.root.join("log.md")).unwrap().len();
2657 assert!(
2658 size > (REVERSE_BLOCK as u64) * 3,
2659 "test log not large enough ({size} bytes) to cross several blocks"
2660 );
2661 let all = Log::tail(&store, n + 10).unwrap();
2662 assert_eq!(all, expected, "every entry must reconstruct across blocks");
2663 // A small tail must also be exact (the n-newest by timestamp).
2664 assert_eq!(Log::tail(&store, 7).unwrap(), expected[n - 7..].to_vec());
2665 }
2666
2667 #[test]
2668 fn header_offsets_range_finds_boundary_straddling_marker_once() {
2669 // Two headers; `header_offsets` (whole-buffer) finds both. The range
2670 // scan with a window that splits the buffer between them must report the
2671 // one in its window exactly once, consulting the left neighbour for the
2672 // line-start check.
2673 let buf =
2674 b"## [2026-06-01 00:00] update | a\nnote a\n## [2026-06-01 00:01] update | b\nnote b\n";
2675 let full = header_offsets(buf, 0);
2676 assert_eq!(full.len(), 2, "both headers found over the whole buffer");
2677 let second = full[1] as usize;
2678 // A window covering only the SECOND header's `#` reports just it. Its `#`
2679 // is not at index 0, so `base_is_file_start` is irrelevant here.
2680 let only_second = header_offsets_range(buf, 0, second, second + 1, false);
2681 assert_eq!(only_second, vec![full[1]]);
2682 // A window covering only the FIRST reports just it (right content read
2683 // past the window into the buffer). `base == 0` is the true file start,
2684 // so the index-0 candidate is a real line start.
2685 let only_first = header_offsets_range(buf, 0, 0, 1, true);
2686 assert_eq!(only_first, vec![full[0]]);
2687 // Disjoint windows partition the markers with no double-count.
2688 let mut combined = header_offsets_range(buf, 0, 0, second, true);
2689 combined.extend(header_offsets_range(buf, 0, second, buf.len(), false));
2690 assert_eq!(combined, full);
2691 }
2692
2693 /// CRITICAL regression: a MID-LINE `## [<valid header>]` fragment inside a
2694 /// real entry's note that happens to align with a reverse-read block boundary
2695 /// must NOT be fabricated into an entry. The incremental backward scan reads
2696 /// each block's left edge before its left neighbour is buffered; treating
2697 /// buffer index 0 as a line start there would carve a phantom entry from the
2698 /// fragment and truncate the real entry's note. The fix defers the left-edge
2699 /// candidate until its neighbour is read, so the fragment is correctly seen
2700 /// as note body (its `#` is not at a line start).
2701 #[test]
2702 fn reverse_read_does_not_fabricate_entry_from_midline_header_at_block_boundary() {
2703 let (_d, store) = temp_store();
2704
2705 // A single real entry. Its note carries a mid-line `## [` fragment that
2706 // is a *valid* header shape but is NOT at column 0 (so the writer's
2707 // column-0 escape correctly leaves it verbatim — it is the trigger).
2708 let fragment = "see ## [2020-01-01 00:00] delete | records/x.md";
2709 let hash_in_fragment = fragment.find("##").expect("fragment has `##`");
2710
2711 // Build the raw active log by hand so the fragment's `#` lands at the
2712 // FIRST backward block's left edge: the reverse reader anchors its blocks
2713 // at EOF (`new_start = len - REVERSE_BLOCK` on the first block), so the
2714 // `#` must sit exactly `REVERSE_BLOCK` bytes before EOF. We append note
2715 // padding AFTER the fragment to push EOF out to that distance.
2716 //
2717 // Layout (one entry):
2718 // <frontmatter>\n## [<header>] | records/real.md\nlead\n<fragment><tail>\n\n
2719 let header_line = "## [2026-06-14 10:00] update | records/real.md\n";
2720 let mut head = String::from(LOG_FRONTMATTER);
2721 head.push('\n');
2722 head.push_str(header_line);
2723 head.push_str("lead\n");
2724 head.push_str(fragment); // fragment opens the second note line
2725
2726 // Absolute offset of the fragment's `#`.
2727 let hash_off = head.len() - fragment.len() + hash_in_fragment;
2728 // We append `<tail>\n\n`. Bytes after `#` = (head.len() - hash_off) +
2729 // tail_len + 2. Need that == REVERSE_BLOCK so `#` is at `len -
2730 // REVERSE_BLOCK` (the first block's left edge).
2731 let after_hash_in_head = head.len() - hash_off;
2732 let tail_len = REVERSE_BLOCK
2733 .checked_sub(after_hash_in_head + 2)
2734 .expect("REVERSE_BLOCK comfortably exceeds the post-`#` head bytes");
2735 let mut body = head;
2736 body.push_str(&"z".repeat(tail_len)); // valid note bytes on the fragment line
2737 body.push('\n');
2738 body.push('\n');
2739 fs::write(store.root.join("log.md"), &body).unwrap();
2740
2741 // The file must be large enough to cross at least one block boundary.
2742 assert!(
2743 body.len() as u64 > REVERSE_BLOCK as u64,
2744 "test log must span >1 block (len {})",
2745 body.len()
2746 );
2747 // And the fragment's `#` sits exactly at the first block's left edge.
2748 let real_hash_off = body.find("see ##").unwrap() + hash_in_fragment;
2749 assert_eq!(
2750 real_hash_off,
2751 body.len() - REVERSE_BLOCK,
2752 "fragment `#` must land on the first backward block's left edge to exercise the bug"
2753 );
2754
2755 // Reverse read must return EXACTLY ONE entry — the real one — and never a
2756 // fabricated `2020-01-01 delete records/x.md` carved from the fragment.
2757 let got = Log::tail(&store, 10).unwrap();
2758 assert_eq!(
2759 got.len(),
2760 1,
2761 "exactly the one real entry; got {} (a fabricated entry means the boundary `#` was mis-read as a header): {got:#?}",
2762 got.len()
2763 );
2764 let only = &got[0];
2765 assert_eq!(only.object.as_deref(), Some("records/real.md"));
2766 assert_eq!(only.timestamp, ts(2026, 6, 14, 10, 0));
2767 // The note is intact end-to-end (not truncated at the fragment): both the
2768 // lead and the verbatim fragment survive.
2769 assert!(
2770 only.note.contains("lead"),
2771 "note keeps its lead; got {:?}",
2772 only.note
2773 );
2774 assert!(
2775 only.note.contains(fragment),
2776 "note keeps the verbatim mid-line fragment (not truncated); got {:?}",
2777 only.note
2778 );
2779 }
2780
2781 // ── regression: tail/since dedup across active+archive on interrupted rotation
2782
2783 #[test]
2784 fn tail_and_since_dedup_entries_present_in_both_active_and_archive() {
2785 // Reconstructs the finding's crash window: the archive write committed
2786 // but the active rewrite never trimmed, so the same April entries live in
2787 // BOTH the untrimmed active file and `log/2026-04.md`. Readers must
2788 // return each entry ONCE, not twice.
2789 let (_d, store) = temp_store();
2790 let apr_a = entry(2026, 4, 10, 9, 0, LogKind::Ingest, Some("apr-a"), "apr one");
2791 let apr_b = entry(2026, 4, 20, 9, 0, LogKind::Create, Some("apr-b"), "apr two");
2792
2793 // Active file still holds both April entries (the un-trimmed state).
2794 write_raw_log(&store, &[apr_a.clone(), apr_b.clone()]);
2795 // The committed step-1 archive holds the same two entries.
2796 let dir = archive_dir(&store);
2797 fs::create_dir_all(&dir).unwrap();
2798 let mut arch = String::from(LOG_FRONTMATTER);
2799 arch.push('\n');
2800 arch.push_str(&apr_a.render());
2801 arch.push_str(&apr_b.render());
2802 fs::write(archive_path(&store, 2026, 4), arch).unwrap();
2803
2804 // `since` must return each April entry exactly once.
2805 let since = Log::since(&store, ts(2026, 4, 1, 0, 0)).unwrap();
2806 assert_eq!(
2807 since,
2808 vec![apr_a.clone(), apr_b.clone()],
2809 "since must dedup the doubly-present entries; got {since:?}"
2810 );
2811
2812 // `tail` must too — no duplicate window slots.
2813 let tail = Log::tail(&store, 10).unwrap();
2814 assert_eq!(
2815 tail,
2816 vec![apr_a, apr_b],
2817 "tail must dedup the doubly-present entries; got {tail:?}"
2818 );
2819 }
2820}