Skip to main content

supercode_interchange/
watch.rs

1//! Passive, change-triggered following of local coding-harness sessions.
2//!
3//! This module deliberately observes persisted session state; it does not
4//! attach to, control, or infer the liveness of the process writing it.
5
6use std::fs::Metadata;
7use std::path::{Path, PathBuf};
8use std::time::UNIX_EPOCH;
9
10use serde_json::{json, Value};
11
12use crate::catalog::{SessionLocator, StorageLocator};
13use crate::native_store::load_native_store_family;
14use crate::session::{looks_like_sqlite, Session, SessionSource};
15use crate::{ChatMessage, Error, Fidelity, Result};
16
17/// Why a watcher emitted a complete session snapshot.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum SessionSnapshotReason {
20    /// The first event emitted after opening the follower.
21    Initial,
22    /// Existing normalized history changed, disappeared, or branched.
23    HistoryRewritten,
24    /// Session identity or other non-message state changed.
25    SourceChanged,
26}
27
28impl SessionSnapshotReason {
29    fn as_str(self) -> &'static str {
30        match self {
31            Self::Initial => "initial",
32            Self::HistoryRewritten => "history_rewritten",
33            Self::SourceChanged => "source_changed",
34        }
35    }
36}
37
38/// A normalized event emitted while following a local session.
39#[derive(Debug, Clone)]
40pub enum SessionWatchEvent {
41    /// A complete normalized view of the selected session.
42    SessionSnapshot {
43        /// Monotonically increasing sequence number, starting at one.
44        sequence: u64,
45        /// Why the full snapshot was necessary.
46        reason: SessionSnapshotReason,
47        /// The current normalized session.
48        session: Box<Session>,
49    },
50    /// Messages appended without changing existing normalized history.
51    MessagesAppended {
52        /// Monotonically increasing sequence number.
53        sequence: u64,
54        /// Selected session id, when the source records one.
55        session_id: Option<String>,
56        /// Newly appended normalized messages.
57        messages: Vec<ChatMessage>,
58        /// Total normalized messages in the source-side display projection.
59        total_message_count: usize,
60    },
61    /// A recoverable read or parse problem. The follower remains usable.
62    WatchError {
63        /// Monotonically increasing sequence number.
64        sequence: u64,
65        /// Human-readable description of the problem.
66        message: String,
67    },
68}
69
70impl SessionWatchEvent {
71    /// The event's monotonic sequence number.
72    pub fn sequence(&self) -> u64 {
73        match self {
74            Self::SessionSnapshot { sequence, .. }
75            | Self::MessagesAppended { sequence, .. }
76            | Self::WatchError { sequence, .. } => *sequence,
77        }
78    }
79
80    /// Render this event as one self-contained JSON value suitable for NDJSON.
81    pub fn to_json(&self) -> Value {
82        match self {
83            Self::SessionSnapshot {
84                sequence,
85                reason,
86                session,
87            } => json!({
88                "type": "session_snapshot",
89                "sequence": sequence,
90                "reason": reason.as_str(),
91                "session": normalized_session_json(session),
92            }),
93            Self::MessagesAppended {
94                sequence,
95                session_id,
96                messages,
97                total_message_count,
98            } => json!({
99                "type": "messages_appended",
100                "sequence": sequence,
101                "session_id": session_id,
102                "messages": messages.iter().map(message_json).collect::<Vec<_>>(),
103                "total_message_count": total_message_count,
104            }),
105            Self::WatchError { sequence, message } => json!({
106                "type": "watch_error",
107                "sequence": sequence,
108                "recoverable": true,
109                "message": message,
110            }),
111        }
112    }
113}
114
115/// Poll-based follower for one persisted Claude Code, Codex, Pi, OpenCode, or Grok
116/// session.
117///
118/// Polling first compares cheap filesystem stamps. Full-history Claude followers
119/// can normalize simple appended records after comparing every retained prefix
120/// byte. Complex changes and bounded display-history views retain the canonical
121/// loader. No extra timer, second watcher or approximate history is introduced.
122pub struct SessionFollower {
123    path: PathBuf,
124    opencode_session: Option<String>,
125    store: SqliteStore,
126    fidelity: Fidelity,
127    include_subagents: bool,
128    message_limit: Option<usize>,
129    max_message_chars: Option<usize>,
130    display_history: bool,
131    current: Session,
132    fingerprint: Vec<PathStamp>,
133    initial_pending: bool,
134    next_sequence: u64,
135    claude_append: Option<crate::session::ClaudeAppendState>,
136}
137
138/// Which SQLite-backed store a follower is reading. Each store keeps every session in one file, so
139/// the follower is pinned to a session selector and reloads through that store's own loader.
140#[derive(Debug, Clone, Copy, PartialEq, Eq)]
141pub(crate) enum SqliteStore {
142    OpenCode,
143    Goose,
144    Hermes,
145}
146
147#[derive(Clone, Copy)]
148struct FollowerView {
149    include_subagents: bool,
150    message_limit: Option<usize>,
151    max_message_chars: Option<usize>,
152    display_history: bool,
153}
154
155impl SessionFollower {
156    /// Open a persisted session using its durable catalog locator.
157    pub fn open_locator(locator: &SessionLocator) -> Result<Self> {
158        Self::open_locator_with_fidelity(locator, Fidelity::ByteLossless)
159    }
160
161    /// [`Self::open_locator`] at a declared fidelity.
162    ///
163    /// A read-only mirror follows at [`Fidelity::Semantic`] so a compacted
164    /// transcript keeps streaming instead of turning every poll into a
165    /// `watch_error`. See [`crate::Session::load_with_fidelity`].
166    pub fn open_locator_with_fidelity(
167        locator: &SessionLocator,
168        fidelity: Fidelity,
169    ) -> Result<Self> {
170        Self::open_locator_with_view(locator, fidelity, true, None, None, false)
171    }
172
173    /// Open a frontend-oriented follower whose snapshots contain only the
174    /// selected parent and at most `message_limit` trailing messages.
175    pub fn open_locator_with_view(
176        locator: &SessionLocator,
177        fidelity: Fidelity,
178        include_subagents: bool,
179        message_limit: Option<usize>,
180        max_message_chars: Option<usize>,
181        display_history: bool,
182    ) -> Result<Self> {
183        // A Hermes locator names one session inside a whole-store `state.db` (its discovery emits the
184        // store as a file locator): route by the locator's own session id, the way `load` does, so the
185        // follower never pins the store's newest row instead of the one asked for.
186        if locator.harness.as_str() == crate::HarnessId::HERMES {
187            let path = match &locator.storage {
188                StorageLocator::File { path } | StorageLocator::Sqlite { path, .. } => path,
189            };
190            return Self::open_with_options(
191                path,
192                Some(&locator.session_id),
193                SqliteStore::Hermes,
194                fidelity,
195                FollowerView {
196                    include_subagents,
197                    message_limit,
198                    max_message_chars,
199                    display_history,
200                },
201            );
202        }
203        match &locator.storage {
204            StorageLocator::File { path } => Self::open_with_options(
205                path,
206                None,
207                SqliteStore::OpenCode,
208                fidelity,
209                FollowerView {
210                    include_subagents,
211                    message_limit,
212                    max_message_chars,
213                    display_history,
214                },
215            ),
216            StorageLocator::Sqlite { path, selector } => Self::open_with_options(
217                path,
218                Some(selector),
219                if locator.harness.as_str() == crate::HarnessId::GOOSE {
220                    SqliteStore::Goose
221                } else {
222                    SqliteStore::OpenCode
223                },
224                fidelity,
225                FollowerView {
226                    include_subagents,
227                    message_limit,
228                    max_message_chars,
229                    display_history,
230                },
231            ),
232        }
233    }
234
235    /// Open a local session for passive following.
236    ///
237    /// `opencode_session` is valid only for an OpenCode SQLite store. When it
238    /// is omitted, the initially selected session is pinned for all later
239    /// polls rather than following whichever database row becomes newest.
240    pub fn open(path: impl Into<PathBuf>, opencode_session: Option<&str>) -> Result<Self> {
241        Self::open_with_fidelity(path, opencode_session, Fidelity::ByteLossless)
242    }
243
244    /// [`Self::open`] at a declared fidelity.
245    pub fn open_with_fidelity(
246        path: impl Into<PathBuf>,
247        opencode_session: Option<&str>,
248        fidelity: Fidelity,
249    ) -> Result<Self> {
250        Self::open_with_options(
251            path,
252            opencode_session,
253            SqliteStore::OpenCode,
254            fidelity,
255            FollowerView {
256                include_subagents: true,
257                message_limit: None,
258                max_message_chars: None,
259                display_history: false,
260            },
261        )
262    }
263
264    fn open_with_options(
265        path: impl Into<PathBuf>,
266        opencode_session: Option<&str>,
267        store: SqliteStore,
268        fidelity: Fidelity,
269        view: FollowerView,
270    ) -> Result<Self> {
271        let path = path.into();
272        let sqlite = looks_like_sqlite(&path);
273        if opencode_session.is_some() && !sqlite {
274            return Err(Error::Other(format!(
275                "an OpenCode session selector requires a SQLite store; {} is not one",
276                path.display()
277            )));
278        }
279
280        let mut selected = opencode_session.map(str::to_owned);
281        let mut current = load_selected(&path, selected.as_deref(), store, fidelity, view)?;
282        bound_session_view(&mut current, view.message_limit, view.max_message_chars);
283        if sqlite && selected.is_none() {
284            selected = current.meta.session_id.clone();
285        }
286        let fingerprint =
287            source_fingerprint(&path, &current, selected.as_deref(), view.include_subagents)?;
288        let claude_append = if !view.display_history {
289            crate::session::ClaudeAppendState::new(&current, fidelity)
290                .ok()
291                .flatten()
292        } else {
293            None
294        };
295
296        Ok(Self {
297            path,
298            opencode_session: selected,
299            store,
300            fidelity,
301            include_subagents: view.include_subagents,
302            message_limit: view.message_limit,
303            max_message_chars: view.max_message_chars,
304            display_history: view.display_history,
305            current,
306            fingerprint,
307            initial_pending: true,
308            next_sequence: 1,
309            claude_append,
310        })
311    }
312
313    /// Inspect the filesystem once and return the next event, if any.
314    ///
315    /// The first call always returns an initial snapshot. Later calls return
316    /// `None` while the relevant filesystem stamps are unchanged.
317    pub fn poll(&mut self) -> Result<Option<SessionWatchEvent>> {
318        if self.initial_pending {
319            self.initial_pending = false;
320            return Ok(Some(self.snapshot(SessionSnapshotReason::Initial)));
321        }
322
323        let observed = source_fingerprint(
324            &self.path,
325            &self.current,
326            self.opencode_session.as_deref(),
327            self.include_subagents,
328        )?;
329        if observed == self.fingerprint {
330            return Ok(None);
331        }
332
333        let companions_unchanged = self
334            .fingerprint
335            .iter()
336            .filter(|stamp| stamp.path != self.path)
337            .eq(observed.iter().filter(|stamp| stamp.path != self.path));
338        let appended = if companions_unchanged {
339            self.claude_append.as_ref().and_then(|state| {
340                state
341                    .append(&self.path, &self.current, self.fidelity)
342                    .ok()
343                    .flatten()
344            })
345        } else {
346            None
347        };
348        let (loaded, next_append) = if let Some((session, state)) = appended {
349            (Ok(session), Some(state))
350        } else {
351            (
352                load_selected(
353                    &self.path,
354                    self.opencode_session.as_deref(),
355                    self.store,
356                    self.fidelity,
357                    FollowerView {
358                        include_subagents: self.include_subagents,
359                        message_limit: self.message_limit,
360                        max_message_chars: self.max_message_chars,
361                        display_history: self.display_history,
362                    },
363                ),
364                None,
365            )
366        };
367        self.fingerprint = observed;
368        let next = match loaded {
369            Ok(session) if session.parse_error_lines > 0 => {
370                let count = session.parse_error_lines;
371                Some(self.watch_error(format!(
372                    "{} contains {count} malformed or truncated JSON line(s); retaining the last good snapshot",
373                    self.path.display()
374                )))
375            }
376            Err(error) => Some(self.watch_error(format!(
377                "could not reload {}: {error}; retaining the last good snapshot",
378                self.path.display()
379            ))),
380            Ok(mut session) => {
381                self.claude_append = next_append.or_else(|| {
382                    (!self.display_history)
383                        .then(|| {
384                            crate::session::ClaudeAppendState::new(&session, self.fidelity)
385                                .ok()
386                                .flatten()
387                        })
388                        .flatten()
389                });
390                bound_session_view(&mut session, self.message_limit, self.max_message_chars);
391                self.event_for_session(session)
392            }
393        };
394        Ok(next)
395    }
396
397    fn event_for_session(&mut self, session: Session) -> Option<SessionWatchEvent> {
398        if normalized_session_eq(&self.current, &session) {
399            self.current = session;
400            return None;
401        }
402
403        let identity_same = session_identity_eq(&self.current, &session);
404        let subagents_same = normalized_subagents_eq(&self.current, &session);
405        let append_prefix = if identity_same && subagents_same {
406            append_prefix_len(&self.current.messages, &session.messages)
407        } else {
408            0
409        };
410        if append_prefix > 0 && session.messages.len() > append_prefix {
411            let messages = session.messages[append_prefix..].to_vec();
412            let session_id = session.meta.session_id.clone();
413            let total_message_count = session
414                .imported_message_count
415                .unwrap_or(session.messages.len())
416                .max(session.messages.len());
417            self.current = session;
418            return Some(SessionWatchEvent::MessagesAppended {
419                sequence: self.take_sequence(),
420                session_id,
421                messages,
422                total_message_count,
423            });
424        }
425
426        let reason = if identity_same {
427            SessionSnapshotReason::HistoryRewritten
428        } else {
429            SessionSnapshotReason::SourceChanged
430        };
431        self.current = session;
432        Some(self.snapshot(reason))
433    }
434
435    fn snapshot(&mut self, reason: SessionSnapshotReason) -> SessionWatchEvent {
436        SessionWatchEvent::SessionSnapshot {
437            sequence: self.take_sequence(),
438            reason,
439            session: Box::new(self.current.clone()),
440        }
441    }
442
443    fn watch_error(&mut self, message: String) -> SessionWatchEvent {
444        SessionWatchEvent::WatchError {
445            sequence: self.take_sequence(),
446            message,
447        }
448    }
449
450    fn take_sequence(&mut self) -> u64 {
451        let sequence = self.next_sequence;
452        self.next_sequence += 1;
453        sequence
454    }
455}
456
457fn load_selected(
458    path: &Path,
459    selected: Option<&str>,
460    store: SqliteStore,
461    fidelity: Fidelity,
462    view: FollowerView,
463) -> Result<Session> {
464    let sqlite = looks_like_sqlite(path);
465    if sqlite {
466        match store {
467            SqliteStore::Goose => {
468                let selector = selected.ok_or_else(|| {
469                    Error::Other("a Goose SQLite locator requires a session selector".to_string())
470                })?;
471                Ok(Session::from_goose_sqlite(path, selector)?)
472            }
473            SqliteStore::Hermes => {
474                let selector = selected.ok_or_else(|| {
475                    Error::Other("a Hermes SQLite locator requires a session selector".to_string())
476                })?;
477                Ok(Session::from_hermes_sqlite(path, Some(selector))?)
478            }
479            SqliteStore::OpenCode => Ok(Session::from_opencode_sqlite(path, selected)?),
480        }
481    } else if let Some(session) = load_native_store_family(path)? {
482        Ok(session)
483    } else if view.display_history {
484        Ok(Session::load_display_view(
485            path,
486            fidelity,
487            view.message_limit.unwrap_or(500),
488        )?)
489    } else if view.include_subagents {
490        Ok(Session::load_with_fidelity(path, fidelity)?)
491    } else {
492        Ok(Session::load_parent_with_fidelity(path, fidelity)?)
493    }
494}
495
496#[doc(hidden)]
497pub fn bound_session_view(
498    session: &mut Session,
499    message_limit: Option<usize>,
500    max_message_chars: Option<usize>,
501) {
502    if let Some(limit) = message_limit {
503        if session.messages.len() > limit {
504            session.messages.drain(..session.messages.len() - limit);
505        }
506    }
507
508    let Some(max_chars) = max_message_chars else {
509        return;
510    };
511    for message in &mut session.messages {
512        if let Some(content) = &mut message.content {
513            truncate_utf8(content, max_chars);
514        }
515        if let Some(parts) = &mut message.content_parts {
516            for part in parts {
517                truncate_value_strings(part, max_chars);
518            }
519        }
520        if let Some(tool_calls) = &mut message.tool_calls {
521            for call in tool_calls {
522                truncate_utf8(&mut call.function.arguments, max_chars);
523            }
524        }
525        for value in message.metadata.values_mut() {
526            truncate_utf8(value, max_chars);
527        }
528    }
529}
530
531fn truncate_value_strings(value: &mut Value, max_chars: usize) {
532    match value {
533        Value::String(text) => truncate_utf8(text, max_chars),
534        Value::Array(values) => {
535            for value in values {
536                truncate_value_strings(value, max_chars);
537            }
538        }
539        Value::Object(values) => {
540            for value in values.values_mut() {
541                truncate_value_strings(value, max_chars);
542            }
543        }
544        _ => {}
545    }
546}
547
548fn truncate_utf8(value: &mut String, max_chars: usize) {
549    let Some((byte_index, _)) = value.char_indices().nth(max_chars) else {
550        return;
551    };
552    value.truncate(byte_index);
553    value.push_str("\n…");
554}
555
556fn session_identity_eq(left: &Session, right: &Session) -> bool {
557    left.meta.source == right.meta.source
558        && left.meta.session_id == right.meta.session_id
559        && left.meta.model == right.meta.model
560        && left.meta.cwd == right.meta.cwd
561        && left.meta.system_prompt == right.meta.system_prompt
562        && left.meta.agent_id == right.meta.agent_id
563        && left.meta.parent_tool_use_id == right.meta.parent_tool_use_id
564        && left.meta.lineage == right.meta.lineage
565}
566
567fn normalized_session_eq(left: &Session, right: &Session) -> bool {
568    session_identity_eq(left, right)
569        && left.messages == right.messages
570        && normalized_subagents_eq(left, right)
571        && left.parse_error_lines == right.parse_error_lines
572        && left.load_residue == right.load_residue
573}
574
575fn normalized_subagents_eq(left: &Session, right: &Session) -> bool {
576    left.subagents.len() == right.subagents.len()
577        && left
578            .subagents
579            .iter()
580            .zip(&right.subagents)
581            .all(|(left, right)| normalized_session_eq(left, right))
582}
583
584/// Length of the already-known prefix in `next`. A bounded display window is
585/// either a plain sliding tail, or an anchored tail whose first user row stays
586/// pinned while records immediately after it slide. Both shapes prove that
587/// consumers can append the remaining suffix without replacing visible rows.
588fn append_prefix_len(current: &[ChatMessage], next: &[ChatMessage]) -> usize {
589    let plain = (1..=current.len().min(next.len()))
590        .rev()
591        .find(|&length| current[current.len() - length..] == next[..length])
592        .unwrap_or(0);
593    let anchored = if current.first() == next.first() && next.len() > 1 {
594        (1..=current.len().saturating_sub(1).min(next.len() - 1))
595            .rev()
596            .find(|&length| current[current.len() - length..] == next[1..1 + length])
597            .map(|length| length + 1)
598            .unwrap_or(0)
599    } else {
600        0
601    };
602    plain.max(anchored)
603}
604
605fn source_name(source: SessionSource) -> &'static str {
606    match source {
607        SessionSource::ClaudeCode => "claude_code",
608        SessionSource::Codex => "codex",
609        SessionSource::OpenCode => "opencode",
610        SessionSource::Pi => "pi",
611        SessionSource::Grok => "grok",
612        SessionSource::Gemini => "gemini",
613        SessionSource::Goose => "goose",
614        SessionSource::OpenClaw => "openclaw",
615        SessionSource::Hermes => "hermes",
616        SessionSource::Native => "native",
617    }
618}
619
620#[doc(hidden)]
621pub fn message_json(message: &ChatMessage) -> Value {
622    let mut value = serde_json::to_value(message).unwrap_or_else(|_| json!({}));
623    if let Value::Object(object) = &mut value {
624        object.insert("metadata".to_string(), json!(message.metadata));
625    }
626    value
627}
628
629/// Render a normalized session as a language-neutral JSON value.
630pub fn normalized_session_json(session: &Session) -> Value {
631    json!({
632        "source": source_name(session.meta.source),
633        "session_id": session.meta.session_id,
634        "model": session.meta.model,
635        "cwd": session.meta.cwd,
636        "system_prompt": session.meta.system_prompt,
637        "agent_id": session.meta.agent_id,
638        "parent_tool_use_id": session.meta.parent_tool_use_id,
639        "lineage": session.meta.lineage,
640        // ORCH-6: the ORCH-3 conversation nouns, read straight off the loaded
641        // meta so `sessions.load` and `sessions.discover` report the same
642        // trigger/surface/profile/recurrence/cross-surface/workspace.
643        "trigger": session.meta.trigger_or_default(),
644        "surface": session.meta.surface,
645        "profile": session.meta.profile,
646        "recurrence": session.meta.recurrence,
647        "cross_surface": session.meta.cross_surface,
648        "workspace": session.meta.workspace_ref(),
649        "messages": session.messages.iter().map(message_json).collect::<Vec<_>>(),
650        "subagents": session.subagents.iter().map(normalized_session_json).collect::<Vec<_>>(),
651        "raw_record_count": session.raw.len(),
652        "total_message_count": session.imported_message_count.unwrap_or(session.messages.len()).max(session.messages.len()),
653        "parse_error_lines": session.parse_error_lines,
654        // Same pair `harness.v1.sessions.export` reports for an artifact: the
655        // level reached, and exactly what was given up to reach it. `semantic`
656        // with a non-empty residue means this is a read-only VIEW of a
657        // transcript that cannot be losslessly reconstructed.
658        "fidelity": session.load_fidelity(),
659        "residue": session.load_residue,
660    })
661}
662
663#[derive(Debug, Clone, PartialEq, Eq)]
664struct PathStamp {
665    path: PathBuf,
666    kind: StampKind,
667    len: u64,
668    modified_nanos: Option<u128>,
669}
670
671#[derive(Debug, Clone, Copy, PartialEq, Eq)]
672enum StampKind {
673    Missing,
674    File,
675    Directory,
676    Other,
677}
678
679fn source_fingerprint(
680    path: &Path,
681    session: &Session,
682    selected_session: Option<&str>,
683    include_subagents: bool,
684) -> Result<Vec<PathStamp>> {
685    let mut stamps = vec![path_stamp(path)?];
686    match session.meta.source {
687        SessionSource::ClaudeCode if include_subagents => {
688            if let (Some(parent), Some(stem)) = (path.parent(), path.file_stem()) {
689                collect_tree_stamps(&parent.join(stem).join("subagents"), &mut stamps)?;
690            }
691        }
692        SessionSource::OpenCode if looks_like_sqlite(path) => {
693            stamps.push(path_stamp(&path_with_suffix(path, "-wal"))?);
694            stamps.push(path_stamp(&path_with_suffix(path, "-shm"))?);
695            if let (Some(parent), Some(session_id)) = (path.parent(), selected_session) {
696                stamps.push(path_stamp(
697                    &parent
698                        .join("storage")
699                        .join("session_diff")
700                        .join(format!("{session_id}.json")),
701                )?);
702            }
703        }
704        SessionSource::Hermes if looks_like_sqlite(path) => {
705            // WAL mode: a running Hermes appends to `state.db-wal`; the main file's stamp may not move
706            // until a checkpoint, so the WAL and shm stamps are what wake the follower.
707            stamps.push(path_stamp(&path_with_suffix(path, "-wal"))?);
708            stamps.push(path_stamp(&path_with_suffix(path, "-shm"))?);
709        }
710        SessionSource::Grok => {
711            if let Some(parent) = path.parent() {
712                // `chat_history.jsonl` is the resumable transcript. The
713                // adjacent update stream and summary are cheap companion
714                // stamps that make a running Grok session wake the follower
715                // even while it is between committed transcript turns.
716                stamps.push(path_stamp(&parent.join("updates.jsonl"))?);
717                stamps.push(path_stamp(&parent.join("summary.json"))?);
718            }
719        }
720        SessionSource::Native => {
721            if let (Some(parent), Some(stem)) = (path.parent(), path.file_stem()) {
722                stamps.push(path_stamp(
723                    &parent.join(format!("{}.sidecar.jsonl", stem.to_string_lossy())),
724                )?);
725                stamps.push(path_stamp(
726                    &parent.join(format!("{}.meta.json", stem.to_string_lossy())),
727                )?);
728                collect_tree_stamps(
729                    &parent.join(format!("{}.subagents", stem.to_string_lossy())),
730                    &mut stamps,
731                )?;
732            }
733        }
734        _ => {}
735    }
736    stamps.sort_by(|left, right| left.path.cmp(&right.path));
737    Ok(stamps)
738}
739
740fn collect_tree_stamps(path: &Path, out: &mut Vec<PathStamp>) -> Result<()> {
741    collect_tree_stamps_inner(path, out, true)
742}
743
744fn collect_tree_stamps_inner(path: &Path, out: &mut Vec<PathStamp>, follow: bool) -> Result<()> {
745    let stamp = if follow {
746        path_stamp(path)?
747    } else {
748        path_stamp_no_follow(path)?
749    };
750    let is_directory = stamp.kind == StampKind::Directory;
751    out.push(stamp);
752    if !is_directory {
753        return Ok(());
754    }
755
756    let mut children = std::fs::read_dir(path)?.collect::<std::io::Result<Vec<_>>>()?;
757    children.sort_by_key(|entry| entry.path());
758    for child in children {
759        collect_tree_stamps_inner(&child.path(), out, false)?;
760    }
761    Ok(())
762}
763
764fn path_stamp(path: &Path) -> Result<PathStamp> {
765    path_stamp_with(path, |path| std::fs::metadata(path))
766}
767
768fn path_stamp_no_follow(path: &Path) -> Result<PathStamp> {
769    path_stamp_with(path, |path| std::fs::symlink_metadata(path))
770}
771
772fn path_stamp_with(
773    path: &Path,
774    metadata: impl FnOnce(&Path) -> std::io::Result<Metadata>,
775) -> Result<PathStamp> {
776    match metadata(path) {
777        Ok(metadata) => Ok(stamp_from_metadata(path, &metadata)),
778        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(PathStamp {
779            path: path.to_path_buf(),
780            kind: StampKind::Missing,
781            len: 0,
782            modified_nanos: None,
783        }),
784        Err(error) => Err(error.into()),
785    }
786}
787
788fn path_with_suffix(path: &Path, suffix: &str) -> PathBuf {
789    let mut value = path.as_os_str().to_os_string();
790    value.push(suffix);
791    PathBuf::from(value)
792}
793
794fn stamp_from_metadata(path: &Path, metadata: &Metadata) -> PathStamp {
795    let file_type = metadata.file_type();
796    let kind = if file_type.is_file() {
797        StampKind::File
798    } else if file_type.is_dir() {
799        StampKind::Directory
800    } else {
801        StampKind::Other
802    };
803    PathStamp {
804        path: path.to_path_buf(),
805        kind,
806        len: metadata.len(),
807        modified_nanos: metadata
808            .modified()
809            .ok()
810            .and_then(|time| time.duration_since(UNIX_EPOCH).ok())
811            .map(|duration| duration.as_nanos()),
812    }
813}
814
815#[cfg(test)]
816mod tests {
817    use super::append_prefix_len;
818    use crate::ChatMessage;
819
820    #[test]
821    fn bounded_append_overlap_handles_plain_and_user_anchored_windows() {
822        let user = ChatMessage::user("anchor");
823        let one = ChatMessage::assistant("one");
824        let two = ChatMessage::assistant("two");
825        let three = ChatMessage::assistant("three");
826        let newest = ChatMessage::user("newest");
827
828        assert_eq!(
829            append_prefix_len(
830                &[one.clone(), two.clone(), three.clone()],
831                &[two.clone(), three.clone(), newest.clone()],
832            ),
833            2,
834        );
835        assert_eq!(
836            append_prefix_len(
837                &[user.clone(), one, two.clone(), three.clone()],
838                &[user, two, three, newest],
839            ),
840            3,
841        );
842    }
843}