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 {
122 path: PathBuf,
123 opencode_session: Option<String>,
124 store: SqliteStore,
125 fidelity: Fidelity,
126 include_subagents: bool,
127 message_limit: Option<usize>,
128 max_message_chars: Option<usize>,
129 display_history: bool,
130 current: Session,
131 fingerprint: Vec<PathStamp>,
132 initial_pending: bool,
133 next_sequence: u64,
134}
135
136#[derive(Debug, Clone, Copy, PartialEq, Eq)]
139pub(crate) enum SqliteStore {
140 OpenCode,
141 Goose,
142 Hermes,
143}
144
145#[derive(Clone, Copy)]
146struct FollowerView {
147 include_subagents: bool,
148 message_limit: Option<usize>,
149 max_message_chars: Option<usize>,
150 display_history: bool,
151}
152
153impl SessionFollower {
154 pub fn open_locator(locator: &SessionLocator) -> Result<Self> {
156 Self::open_locator_with_fidelity(locator, Fidelity::ByteLossless)
157 }
158
159 pub fn open_locator_with_fidelity(
165 locator: &SessionLocator,
166 fidelity: Fidelity,
167 ) -> Result<Self> {
168 Self::open_locator_with_view(locator, fidelity, true, None, None, false)
169 }
170
171 pub fn open_locator_with_view(
174 locator: &SessionLocator,
175 fidelity: Fidelity,
176 include_subagents: bool,
177 message_limit: Option<usize>,
178 max_message_chars: Option<usize>,
179 display_history: bool,
180 ) -> Result<Self> {
181 if locator.harness.as_str() == crate::HarnessId::HERMES {
185 let path = match &locator.storage {
186 StorageLocator::File { path } | StorageLocator::Sqlite { path, .. } => path,
187 };
188 return Self::open_with_options(
189 path,
190 Some(&locator.session_id),
191 SqliteStore::Hermes,
192 fidelity,
193 FollowerView {
194 include_subagents,
195 message_limit,
196 max_message_chars,
197 display_history,
198 },
199 );
200 }
201 match &locator.storage {
202 StorageLocator::File { path } => Self::open_with_options(
203 path,
204 None,
205 SqliteStore::OpenCode,
206 fidelity,
207 FollowerView {
208 include_subagents,
209 message_limit,
210 max_message_chars,
211 display_history,
212 },
213 ),
214 StorageLocator::Sqlite { path, selector } => Self::open_with_options(
215 path,
216 Some(selector),
217 if locator.harness.as_str() == crate::HarnessId::GOOSE {
218 SqliteStore::Goose
219 } else {
220 SqliteStore::OpenCode
221 },
222 fidelity,
223 FollowerView {
224 include_subagents,
225 message_limit,
226 max_message_chars,
227 display_history,
228 },
229 ),
230 }
231 }
232
233 pub fn open(path: impl Into<PathBuf>, opencode_session: Option<&str>) -> Result<Self> {
239 Self::open_with_fidelity(path, opencode_session, Fidelity::ByteLossless)
240 }
241
242 pub fn open_with_fidelity(
244 path: impl Into<PathBuf>,
245 opencode_session: Option<&str>,
246 fidelity: Fidelity,
247 ) -> Result<Self> {
248 Self::open_with_options(
249 path,
250 opencode_session,
251 SqliteStore::OpenCode,
252 fidelity,
253 FollowerView {
254 include_subagents: true,
255 message_limit: None,
256 max_message_chars: None,
257 display_history: false,
258 },
259 )
260 }
261
262 fn open_with_options(
263 path: impl Into<PathBuf>,
264 opencode_session: Option<&str>,
265 store: SqliteStore,
266 fidelity: Fidelity,
267 view: FollowerView,
268 ) -> Result<Self> {
269 let path = path.into();
270 let sqlite = looks_like_sqlite(&path);
271 if opencode_session.is_some() && !sqlite {
272 return Err(Error::Other(format!(
273 "an OpenCode session selector requires a SQLite store; {} is not one",
274 path.display()
275 )));
276 }
277
278 let mut selected = opencode_session.map(str::to_owned);
279 let mut current = load_selected(&path, selected.as_deref(), store, fidelity, view)?;
280 bound_session_view(&mut current, view.message_limit, view.max_message_chars);
281 if sqlite && selected.is_none() {
282 selected = current.meta.session_id.clone();
283 }
284 let fingerprint =
285 source_fingerprint(&path, ¤t, selected.as_deref(), view.include_subagents)?;
286
287 Ok(Self {
288 path,
289 opencode_session: selected,
290 store,
291 fidelity,
292 include_subagents: view.include_subagents,
293 message_limit: view.message_limit,
294 max_message_chars: view.max_message_chars,
295 display_history: view.display_history,
296 current,
297 fingerprint,
298 initial_pending: true,
299 next_sequence: 1,
300 })
301 }
302
303 pub fn poll(&mut self) -> Result<Option<SessionWatchEvent>> {
308 if self.initial_pending {
309 self.initial_pending = false;
310 return Ok(Some(self.snapshot(SessionSnapshotReason::Initial)));
311 }
312
313 let observed = source_fingerprint(
314 &self.path,
315 &self.current,
316 self.opencode_session.as_deref(),
317 self.include_subagents,
318 )?;
319 if observed == self.fingerprint {
320 return Ok(None);
321 }
322
323 let loaded = load_selected(
324 &self.path,
325 self.opencode_session.as_deref(),
326 self.store,
327 self.fidelity,
328 FollowerView {
329 include_subagents: self.include_subagents,
330 message_limit: self.message_limit,
331 max_message_chars: self.max_message_chars,
332 display_history: self.display_history,
333 },
334 );
335 self.fingerprint = observed;
336 let next = match loaded {
337 Ok(session) if session.parse_error_lines > 0 => {
338 let count = session.parse_error_lines;
339 Some(self.watch_error(format!(
340 "{} contains {count} malformed or truncated JSON line(s); retaining the last good snapshot",
341 self.path.display()
342 )))
343 }
344 Err(error) => Some(self.watch_error(format!(
345 "could not reload {}: {error}; retaining the last good snapshot",
346 self.path.display()
347 ))),
348 Ok(mut session) => {
349 bound_session_view(&mut session, self.message_limit, self.max_message_chars);
350 self.event_for_session(session)
351 }
352 };
353 Ok(next)
354 }
355
356 fn event_for_session(&mut self, session: Session) -> Option<SessionWatchEvent> {
357 if normalized_session_eq(&self.current, &session) {
358 self.current = session;
359 return None;
360 }
361
362 let identity_same = session_identity_eq(&self.current, &session);
363 let subagents_same = normalized_subagents_eq(&self.current, &session);
364 let append_prefix = if identity_same && subagents_same {
365 append_prefix_len(&self.current.messages, &session.messages)
366 } else {
367 0
368 };
369 if append_prefix > 0 && session.messages.len() > append_prefix {
370 let messages = session.messages[append_prefix..].to_vec();
371 let session_id = session.meta.session_id.clone();
372 let total_message_count = session
373 .imported_message_count
374 .unwrap_or(session.messages.len())
375 .max(session.messages.len());
376 self.current = session;
377 return Some(SessionWatchEvent::MessagesAppended {
378 sequence: self.take_sequence(),
379 session_id,
380 messages,
381 total_message_count,
382 });
383 }
384
385 let reason = if identity_same {
386 SessionSnapshotReason::HistoryRewritten
387 } else {
388 SessionSnapshotReason::SourceChanged
389 };
390 self.current = session;
391 Some(self.snapshot(reason))
392 }
393
394 fn snapshot(&mut self, reason: SessionSnapshotReason) -> SessionWatchEvent {
395 SessionWatchEvent::SessionSnapshot {
396 sequence: self.take_sequence(),
397 reason,
398 session: Box::new(self.current.clone()),
399 }
400 }
401
402 fn watch_error(&mut self, message: String) -> SessionWatchEvent {
403 SessionWatchEvent::WatchError {
404 sequence: self.take_sequence(),
405 message,
406 }
407 }
408
409 fn take_sequence(&mut self) -> u64 {
410 let sequence = self.next_sequence;
411 self.next_sequence += 1;
412 sequence
413 }
414}
415
416fn load_selected(
417 path: &Path,
418 selected: Option<&str>,
419 store: SqliteStore,
420 fidelity: Fidelity,
421 view: FollowerView,
422) -> Result<Session> {
423 let sqlite = looks_like_sqlite(path);
424 if sqlite {
425 match store {
426 SqliteStore::Goose => {
427 let selector = selected.ok_or_else(|| {
428 Error::Other("a Goose SQLite locator requires a session selector".to_string())
429 })?;
430 Ok(Session::from_goose_sqlite(path, selector)?)
431 }
432 SqliteStore::Hermes => {
433 let selector = selected.ok_or_else(|| {
434 Error::Other("a Hermes SQLite locator requires a session selector".to_string())
435 })?;
436 Ok(Session::from_hermes_sqlite(path, Some(selector))?)
437 }
438 SqliteStore::OpenCode => Ok(Session::from_opencode_sqlite(path, selected)?),
439 }
440 } else if let Some(session) = load_native_store_family(path)? {
441 Ok(session)
442 } else if view.display_history {
443 Ok(Session::load_display_view(
444 path,
445 fidelity,
446 view.message_limit.unwrap_or(500),
447 )?)
448 } else if view.include_subagents {
449 Ok(Session::load_with_fidelity(path, fidelity)?)
450 } else {
451 Ok(Session::load_parent_with_fidelity(path, fidelity)?)
452 }
453}
454
455#[doc(hidden)]
456pub fn bound_session_view(
457 session: &mut Session,
458 message_limit: Option<usize>,
459 max_message_chars: Option<usize>,
460) {
461 if let Some(limit) = message_limit {
462 if session.messages.len() > limit {
463 session.messages.drain(..session.messages.len() - limit);
464 }
465 }
466
467 let Some(max_chars) = max_message_chars else {
468 return;
469 };
470 for message in &mut session.messages {
471 if let Some(content) = &mut message.content {
472 truncate_utf8(content, max_chars);
473 }
474 if let Some(parts) = &mut message.content_parts {
475 for part in parts {
476 truncate_value_strings(part, max_chars);
477 }
478 }
479 if let Some(tool_calls) = &mut message.tool_calls {
480 for call in tool_calls {
481 truncate_utf8(&mut call.function.arguments, max_chars);
482 }
483 }
484 for value in message.metadata.values_mut() {
485 truncate_utf8(value, max_chars);
486 }
487 }
488}
489
490fn truncate_value_strings(value: &mut Value, max_chars: usize) {
491 match value {
492 Value::String(text) => truncate_utf8(text, max_chars),
493 Value::Array(values) => {
494 for value in values {
495 truncate_value_strings(value, max_chars);
496 }
497 }
498 Value::Object(values) => {
499 for value in values.values_mut() {
500 truncate_value_strings(value, max_chars);
501 }
502 }
503 _ => {}
504 }
505}
506
507fn truncate_utf8(value: &mut String, max_chars: usize) {
508 let Some((byte_index, _)) = value.char_indices().nth(max_chars) else {
509 return;
510 };
511 value.truncate(byte_index);
512 value.push_str("\n…");
513}
514
515fn session_identity_eq(left: &Session, right: &Session) -> bool {
516 left.meta.source == right.meta.source
517 && left.meta.session_id == right.meta.session_id
518 && left.meta.model == right.meta.model
519 && left.meta.cwd == right.meta.cwd
520 && left.meta.system_prompt == right.meta.system_prompt
521 && left.meta.agent_id == right.meta.agent_id
522 && left.meta.parent_tool_use_id == right.meta.parent_tool_use_id
523 && left.meta.lineage == right.meta.lineage
524}
525
526fn normalized_session_eq(left: &Session, right: &Session) -> bool {
527 session_identity_eq(left, right)
528 && left.messages == right.messages
529 && normalized_subagents_eq(left, right)
530 && left.parse_error_lines == right.parse_error_lines
531 && left.load_residue == right.load_residue
532}
533
534fn normalized_subagents_eq(left: &Session, right: &Session) -> bool {
535 left.subagents.len() == right.subagents.len()
536 && left
537 .subagents
538 .iter()
539 .zip(&right.subagents)
540 .all(|(left, right)| normalized_session_eq(left, right))
541}
542
543fn append_prefix_len(current: &[ChatMessage], next: &[ChatMessage]) -> usize {
548 let plain = (1..=current.len().min(next.len()))
549 .rev()
550 .find(|&length| current[current.len() - length..] == next[..length])
551 .unwrap_or(0);
552 let anchored = if current.first() == next.first() && next.len() > 1 {
553 (1..=current.len().saturating_sub(1).min(next.len() - 1))
554 .rev()
555 .find(|&length| current[current.len() - length..] == next[1..1 + length])
556 .map(|length| length + 1)
557 .unwrap_or(0)
558 } else {
559 0
560 };
561 plain.max(anchored)
562}
563
564fn source_name(source: SessionSource) -> &'static str {
565 match source {
566 SessionSource::ClaudeCode => "claude_code",
567 SessionSource::Codex => "codex",
568 SessionSource::OpenCode => "opencode",
569 SessionSource::Pi => "pi",
570 SessionSource::Grok => "grok",
571 SessionSource::Gemini => "gemini",
572 SessionSource::Goose => "goose",
573 SessionSource::OpenClaw => "openclaw",
574 SessionSource::Hermes => "hermes",
575 SessionSource::Native => "native",
576 }
577}
578
579#[doc(hidden)]
580pub fn message_json(message: &ChatMessage) -> Value {
581 let mut value = serde_json::to_value(message).unwrap_or_else(|_| json!({}));
582 if let Value::Object(object) = &mut value {
583 object.insert("metadata".to_string(), json!(message.metadata));
584 }
585 value
586}
587
588pub fn normalized_session_json(session: &Session) -> Value {
590 json!({
591 "source": source_name(session.meta.source),
592 "session_id": session.meta.session_id,
593 "model": session.meta.model,
594 "cwd": session.meta.cwd,
595 "system_prompt": session.meta.system_prompt,
596 "agent_id": session.meta.agent_id,
597 "parent_tool_use_id": session.meta.parent_tool_use_id,
598 "lineage": session.meta.lineage,
599 "trigger": session.meta.trigger_or_default(),
603 "surface": session.meta.surface,
604 "profile": session.meta.profile,
605 "recurrence": session.meta.recurrence,
606 "cross_surface": session.meta.cross_surface,
607 "workspace": session.meta.workspace_ref(),
608 "messages": session.messages.iter().map(message_json).collect::<Vec<_>>(),
609 "subagents": session.subagents.iter().map(normalized_session_json).collect::<Vec<_>>(),
610 "raw_record_count": session.raw.len(),
611 "total_message_count": session.imported_message_count.unwrap_or(session.messages.len()).max(session.messages.len()),
612 "parse_error_lines": session.parse_error_lines,
613 "fidelity": session.load_fidelity(),
618 "residue": session.load_residue,
619 })
620}
621
622#[derive(Debug, Clone, PartialEq, Eq)]
623struct PathStamp {
624 path: PathBuf,
625 kind: StampKind,
626 len: u64,
627 modified_nanos: Option<u128>,
628}
629
630#[derive(Debug, Clone, Copy, PartialEq, Eq)]
631enum StampKind {
632 Missing,
633 File,
634 Directory,
635 Other,
636}
637
638fn source_fingerprint(
639 path: &Path,
640 session: &Session,
641 selected_session: Option<&str>,
642 include_subagents: bool,
643) -> Result<Vec<PathStamp>> {
644 let mut stamps = vec![path_stamp(path)?];
645 match session.meta.source {
646 SessionSource::ClaudeCode if include_subagents => {
647 if let (Some(parent), Some(stem)) = (path.parent(), path.file_stem()) {
648 collect_tree_stamps(&parent.join(stem).join("subagents"), &mut stamps)?;
649 }
650 }
651 SessionSource::OpenCode if looks_like_sqlite(path) => {
652 stamps.push(path_stamp(&path_with_suffix(path, "-wal"))?);
653 stamps.push(path_stamp(&path_with_suffix(path, "-shm"))?);
654 if let (Some(parent), Some(session_id)) = (path.parent(), selected_session) {
655 stamps.push(path_stamp(
656 &parent
657 .join("storage")
658 .join("session_diff")
659 .join(format!("{session_id}.json")),
660 )?);
661 }
662 }
663 SessionSource::Hermes if looks_like_sqlite(path) => {
664 stamps.push(path_stamp(&path_with_suffix(path, "-wal"))?);
667 stamps.push(path_stamp(&path_with_suffix(path, "-shm"))?);
668 }
669 SessionSource::Grok => {
670 if let Some(parent) = path.parent() {
671 stamps.push(path_stamp(&parent.join("updates.jsonl"))?);
676 stamps.push(path_stamp(&parent.join("summary.json"))?);
677 }
678 }
679 SessionSource::Native => {
680 if let (Some(parent), Some(stem)) = (path.parent(), path.file_stem()) {
681 stamps.push(path_stamp(
682 &parent.join(format!("{}.sidecar.jsonl", stem.to_string_lossy())),
683 )?);
684 stamps.push(path_stamp(
685 &parent.join(format!("{}.meta.json", stem.to_string_lossy())),
686 )?);
687 collect_tree_stamps(
688 &parent.join(format!("{}.subagents", stem.to_string_lossy())),
689 &mut stamps,
690 )?;
691 }
692 }
693 _ => {}
694 }
695 stamps.sort_by(|left, right| left.path.cmp(&right.path));
696 Ok(stamps)
697}
698
699fn collect_tree_stamps(path: &Path, out: &mut Vec<PathStamp>) -> Result<()> {
700 collect_tree_stamps_inner(path, out, true)
701}
702
703fn collect_tree_stamps_inner(path: &Path, out: &mut Vec<PathStamp>, follow: bool) -> Result<()> {
704 let stamp = if follow {
705 path_stamp(path)?
706 } else {
707 path_stamp_no_follow(path)?
708 };
709 let is_directory = stamp.kind == StampKind::Directory;
710 out.push(stamp);
711 if !is_directory {
712 return Ok(());
713 }
714
715 let mut children = std::fs::read_dir(path)?.collect::<std::io::Result<Vec<_>>>()?;
716 children.sort_by_key(|entry| entry.path());
717 for child in children {
718 collect_tree_stamps_inner(&child.path(), out, false)?;
719 }
720 Ok(())
721}
722
723fn path_stamp(path: &Path) -> Result<PathStamp> {
724 path_stamp_with(path, |path| std::fs::metadata(path))
725}
726
727fn path_stamp_no_follow(path: &Path) -> Result<PathStamp> {
728 path_stamp_with(path, |path| std::fs::symlink_metadata(path))
729}
730
731fn path_stamp_with(
732 path: &Path,
733 metadata: impl FnOnce(&Path) -> std::io::Result<Metadata>,
734) -> Result<PathStamp> {
735 match metadata(path) {
736 Ok(metadata) => Ok(stamp_from_metadata(path, &metadata)),
737 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(PathStamp {
738 path: path.to_path_buf(),
739 kind: StampKind::Missing,
740 len: 0,
741 modified_nanos: None,
742 }),
743 Err(error) => Err(error.into()),
744 }
745}
746
747fn path_with_suffix(path: &Path, suffix: &str) -> PathBuf {
748 let mut value = path.as_os_str().to_os_string();
749 value.push(suffix);
750 PathBuf::from(value)
751}
752
753fn stamp_from_metadata(path: &Path, metadata: &Metadata) -> PathStamp {
754 let file_type = metadata.file_type();
755 let kind = if file_type.is_file() {
756 StampKind::File
757 } else if file_type.is_dir() {
758 StampKind::Directory
759 } else {
760 StampKind::Other
761 };
762 PathStamp {
763 path: path.to_path_buf(),
764 kind,
765 len: metadata.len(),
766 modified_nanos: metadata
767 .modified()
768 .ok()
769 .and_then(|time| time.duration_since(UNIX_EPOCH).ok())
770 .map(|duration| duration.as_nanos()),
771 }
772}
773
774#[cfg(test)]
775mod tests {
776 use super::append_prefix_len;
777 use crate::ChatMessage;
778
779 #[test]
780 fn bounded_append_overlap_handles_plain_and_user_anchored_windows() {
781 let user = ChatMessage::user("anchor");
782 let one = ChatMessage::assistant("one");
783 let two = ChatMessage::assistant("two");
784 let three = ChatMessage::assistant("three");
785 let newest = ChatMessage::user("newest");
786
787 assert_eq!(
788 append_prefix_len(
789 &[one.clone(), two.clone(), three.clone()],
790 &[two.clone(), three.clone(), newest.clone()],
791 ),
792 2,
793 );
794 assert_eq!(
795 append_prefix_len(
796 &[user.clone(), one, two.clone(), three.clone()],
797 &[user, two, three, newest],
798 ),
799 3,
800 );
801 }
802}