1use 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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum SessionSnapshotReason {
20 Initial,
22 HistoryRewritten,
24 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#[derive(Debug, Clone)]
40pub enum SessionWatchEvent {
41 SessionSnapshot {
43 sequence: u64,
45 reason: SessionSnapshotReason,
47 session: Box<Session>,
49 },
50 MessagesAppended {
52 sequence: u64,
54 session_id: Option<String>,
56 messages: Vec<ChatMessage>,
58 total_message_count: usize,
60 },
61 WatchError {
63 sequence: u64,
65 message: String,
67 },
68}
69
70impl SessionWatchEvent {
71 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 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
115pub 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#[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 pub fn open_locator(locator: &SessionLocator) -> Result<Self> {
158 Self::open_locator_with_fidelity(locator, Fidelity::ByteLossless)
159 }
160
161 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 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 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 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 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, ¤t, selected.as_deref(), view.include_subagents)?;
288 let claude_append = if !view.display_history {
289 crate::session::ClaudeAppendState::new(¤t, 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 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
584fn 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
629pub 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 "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 "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 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 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}