1use std::fs::{self, File};
16use std::io::{self, BufRead, BufReader, Write as _};
17use std::path::{Path, PathBuf};
18use std::sync::{Arc, Mutex};
19
20use serde::{Deserialize, Serialize};
21use zeph_llm::provider::Message;
22
23use super::error::SubAgentError;
24use super::state::SubAgentState;
25
26#[derive(Debug, Clone, Serialize, Deserialize)]
32pub struct TranscriptEntry {
33 pub seq: u32,
35 pub timestamp: String,
37 pub message: Message,
39}
40
41#[derive(Debug, Clone, Serialize, Deserialize)]
46pub struct TranscriptMeta {
47 pub agent_id: String,
49 pub agent_name: String,
51 pub def_name: String,
53 pub status: SubAgentState,
55 pub started_at: String,
57 #[serde(skip_serializing_if = "Option::is_none")]
59 pub finished_at: Option<String>,
60 #[serde(skip_serializing_if = "Option::is_none")]
62 pub resumed_from: Option<String>,
63 pub turns_used: u32,
65 #[serde(default)]
70 pub mcp_tool_names: Vec<String>,
71}
72
73#[derive(Clone)]
90pub struct TranscriptWriter {
91 file: Arc<Mutex<File>>,
92}
93
94impl TranscriptWriter {
95 pub fn new(path: &Path) -> io::Result<Self> {
103 if let Some(parent) = path.parent() {
104 fs::create_dir_all(parent)?;
105 }
106 let file = zeph_common::fs_secure::append_private(path)?;
107 Ok(Self {
108 file: Arc::new(Mutex::new(file)),
109 })
110 }
111
112 pub async fn append(&self, seq: u32, message: &Message) -> io::Result<()> {
121 let entry = TranscriptEntry {
122 seq,
123 timestamp: utc_now(),
124 message: message.clone(),
125 };
126 let line = serde_json::to_string(&entry)
127 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
128 let file = Arc::clone(&self.file);
129 tokio::task::spawn_blocking(move || {
130 let mut guard = file
131 .lock()
132 .map_err(|_| io::Error::other("transcript writer lock poisoned"))?;
133 guard.write_all(line.as_bytes())?;
134 guard.write_all(b"\n")?;
135 guard.flush()
136 })
137 .await
138 .map_err(|e| io::Error::other(format!("spawn_blocking panicked: {e}")))?
139 }
140
141 pub fn write_meta(dir: &Path, agent_id: &str, meta: &TranscriptMeta) -> io::Result<()> {
147 let path = dir.join(format!("{agent_id}.meta.json"));
148 let content = serde_json::to_string_pretty(meta)
149 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
150 zeph_common::fs_secure::write_private(&path, content.as_bytes())
151 }
152
153 pub async fn write_meta_async(
160 dir: &Path,
161 agent_id: &str,
162 meta: &TranscriptMeta,
163 ) -> io::Result<()> {
164 let path = dir.join(format!("{agent_id}.meta.json"));
165 let content = serde_json::to_string_pretty(meta)
166 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
167 let bytes = content.into_bytes();
168 tokio::task::spawn_blocking(move || zeph_common::fs_secure::write_private(&path, &bytes))
169 .await
170 .map_err(|e| io::Error::other(format!("spawn_blocking panicked: {e}")))?
171 }
172}
173
174pub struct TranscriptReader;
181
182impl TranscriptReader {
183 pub fn load(path: &Path) -> Result<Vec<Message>, SubAgentError> {
196 Self::load_impl(path, false)
197 }
198
199 pub fn load_strict(path: &Path) -> Result<Vec<Message>, SubAgentError> {
214 Self::load_impl(path, true)
215 }
216
217 fn load_impl(path: &Path, strict: bool) -> Result<Vec<Message>, SubAgentError> {
218 let file = match File::open(path) {
219 Ok(f) => f,
220 Err(e) if e.kind() == io::ErrorKind::NotFound => {
221 let meta_path =
225 if let (Some(parent), Some(stem)) = (path.parent(), path.file_stem()) {
226 parent.join(format!("{}.meta.json", stem.to_string_lossy()))
227 } else {
228 path.with_extension("meta.json")
229 };
230 if meta_path.exists() {
231 return Err(SubAgentError::Transcript(format!(
232 "transcript file '{}' is missing but meta sidecar exists — \
233 transcript data may have been deleted",
234 path.display()
235 )));
236 }
237 return Ok(vec![]);
238 }
239 Err(e) => {
240 return Err(SubAgentError::Transcript(format!(
241 "failed to open transcript '{}': {e}",
242 path.display()
243 )));
244 }
245 };
246
247 let reader = BufReader::new(file);
248 let mut messages = Vec::new();
249 for (line_no, line_result) in reader.lines().enumerate() {
250 let line = match line_result {
251 Ok(l) => l,
252 Err(e) => {
253 if strict {
254 return Err(SubAgentError::Transcript(format!(
255 "failed to read transcript '{}' line {}: {e}",
256 path.display(),
257 line_no + 1
258 )));
259 }
260 tracing::warn!(
261 path = %path.display(),
262 line = line_no + 1,
263 error = %e,
264 "failed to read transcript line — skipping"
265 );
266 continue;
267 }
268 };
269 let trimmed = line.trim();
270 if trimmed.is_empty() {
271 continue;
272 }
273 match serde_json::from_str::<TranscriptEntry>(trimmed) {
274 Ok(entry) => messages.push(entry.message),
275 Err(e) => {
276 if strict {
277 return Err(SubAgentError::Transcript(format!(
278 "malformed transcript entry in '{}' line {}: {e}",
279 path.display(),
280 line_no + 1
281 )));
282 }
283 tracing::warn!(
284 path = %path.display(),
285 line = line_no + 1,
286 error = %e,
287 "malformed transcript entry — skipping"
288 );
289 }
290 }
291 }
292 Ok(messages)
293 }
294
295 pub fn load_meta(dir: &Path, agent_id: &str) -> Result<TranscriptMeta, SubAgentError> {
302 let path = dir.join(format!("{agent_id}.meta.json"));
303 let content = fs::read_to_string(&path).map_err(|e| {
304 if e.kind() == io::ErrorKind::NotFound {
305 SubAgentError::NotFound(agent_id.to_owned())
306 } else {
307 SubAgentError::Transcript(format!("failed to read meta '{}': {e}", path.display()))
308 }
309 })?;
310 serde_json::from_str(&content).map_err(|e| {
311 SubAgentError::Transcript(format!("failed to parse meta '{}': {e}", path.display()))
312 })
313 }
314
315 pub fn find_by_prefix(dir: &Path, prefix: &str) -> Result<String, SubAgentError> {
324 let entries = fs::read_dir(dir).map_err(|e| {
325 SubAgentError::Transcript(format!(
326 "failed to read transcript dir '{}': {e}",
327 dir.display()
328 ))
329 })?;
330
331 let mut matches: Vec<String> = Vec::new();
332 for entry in entries {
333 let entry = entry
334 .map_err(|e| SubAgentError::Transcript(format!("failed to read dir entry: {e}")))?;
335 let name = entry.file_name();
336 let name_str = name.to_string_lossy();
337 if let Some(agent_id) = name_str.strip_suffix(".meta.json")
338 && agent_id.starts_with(prefix)
339 {
340 matches.push(agent_id.to_owned());
341 }
342 }
343
344 match matches.len() {
345 0 => Err(SubAgentError::NotFound(prefix.to_owned())),
346 1 => Ok(matches.remove(0)),
347 n => Err(SubAgentError::AmbiguousId(prefix.to_owned(), n)),
348 }
349 }
350}
351
352pub fn sweep_old_transcripts(dir: &Path, max_files: usize) -> io::Result<usize> {
361 if max_files == 0 {
362 return Ok(0);
363 }
364
365 if !dir.exists() {
367 fs::create_dir_all(dir)?;
368 return Ok(0);
369 }
370
371 let mut jsonl_files: Vec<(PathBuf, std::time::SystemTime)> = Vec::new();
372 for entry in fs::read_dir(dir)? {
373 let entry = entry?;
374 let path = entry.path();
375 if path.extension().and_then(|e| e.to_str()) == Some("jsonl") {
376 let mtime = entry
377 .metadata()
378 .and_then(|m| m.modified())
379 .unwrap_or(std::time::SystemTime::UNIX_EPOCH);
380 jsonl_files.push((path, mtime));
381 }
382 }
383
384 if jsonl_files.len() <= max_files {
385 return Ok(0);
386 }
387
388 jsonl_files.sort_by_key(|(_, mtime)| *mtime);
390
391 let to_delete = jsonl_files.len() - max_files;
392 let mut deleted = 0;
393 for (path, _) in jsonl_files.into_iter().take(to_delete) {
394 let meta = path.with_extension("meta.json");
396 if meta.exists() {
397 let _ = fs::remove_file(&meta);
398 }
399 fs::remove_file(&path)?;
400 deleted += 1;
401 }
402 Ok(deleted)
403}
404
405#[must_use]
407pub(crate) fn utc_now() -> String {
408 let secs = std::time::SystemTime::now()
411 .duration_since(std::time::UNIX_EPOCH)
412 .unwrap_or_default()
413 .as_secs();
414 let (y, mo, d, h, mi, s) = epoch_to_parts(secs);
415 format!("{y:04}-{mo:02}-{d:02}T{h:02}:{mi:02}:{s:02}Z")
416}
417
418fn epoch_to_parts(epoch: u64) -> (u32, u32, u32, u32, u32, u32) {
424 let sec = epoch % 60;
425 let epoch = epoch / 60;
426 let min = epoch % 60;
427 let epoch = epoch / 60;
428 let hour = epoch % 24;
429 let days = epoch / 24;
430
431 let z = days + 719_468;
433 let era = z / 146_097;
434 let doe = z - era * 146_097;
435 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
436 let year = yoe + era * 400;
437 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
438 let mp = (5 * doy + 2) / 153;
439 let day = doy - (153 * mp + 2) / 5 + 1;
440 let month = if mp < 10 { mp + 3 } else { mp - 9 };
441 let year = if month <= 2 { year + 1 } else { year };
442
443 #[allow(clippy::cast_possible_truncation)]
445 (
446 year as u32,
447 month as u32,
448 day as u32,
449 hour as u32,
450 min as u32,
451 sec as u32,
452 )
453}
454
455#[cfg(test)]
456mod tests {
457 use std::assert_matches;
458 use zeph_llm::provider::{Message, MessageMetadata, Role};
459
460 use super::*;
461
462 fn test_message(role: Role, content: &str) -> Message {
463 Message {
464 role,
465 content: content.to_owned(),
466 parts: vec![],
467 metadata: MessageMetadata::default(),
468 }
469 }
470
471 fn test_meta(agent_id: &str) -> TranscriptMeta {
472 TranscriptMeta {
473 agent_id: agent_id.to_owned(),
474 agent_name: "bot".to_owned(),
475 def_name: "bot".to_owned(),
476 status: SubAgentState::Completed,
477 started_at: "2026-01-01T00:00:00Z".to_owned(),
478 finished_at: Some("2026-01-01T00:01:00Z".to_owned()),
479 resumed_from: None,
480 turns_used: 2,
481 mcp_tool_names: Vec::new(),
482 }
483 }
484
485 #[tokio::test]
486 async fn writer_reader_roundtrip() {
487 let dir = tempfile::tempdir().unwrap();
488 let path = dir.path().join("test.jsonl");
489
490 let msg1 = test_message(Role::User, "hello");
491 let msg2 = test_message(Role::Assistant, "world");
492
493 let writer = TranscriptWriter::new(&path).unwrap();
494 writer.append(0, &msg1).await.unwrap();
495 writer.append(1, &msg2).await.unwrap();
496 drop(writer);
497
498 let messages = TranscriptReader::load(&path).unwrap();
499 assert_eq!(messages.len(), 2);
500 assert_eq!(messages[0].content, "hello");
501 assert_eq!(messages[1].content, "world");
502 }
503
504 #[test]
505 fn load_missing_file_no_meta_returns_empty() {
506 let dir = tempfile::tempdir().unwrap();
507 let path = dir.path().join("ghost.jsonl");
508 let messages = TranscriptReader::load(&path).unwrap();
509 assert!(messages.is_empty());
510 }
511
512 #[test]
513 fn load_missing_file_with_meta_returns_error() {
514 let dir = tempfile::tempdir().unwrap();
515 let meta_path = dir.path().join("ghost.meta.json");
516 std::fs::write(&meta_path, "{}").unwrap();
517 let jsonl_path = dir.path().join("ghost.jsonl");
518 let err = TranscriptReader::load(&jsonl_path).unwrap_err();
519 assert_matches!(err, SubAgentError::Transcript(_));
520 }
521
522 #[test]
523 fn load_skips_malformed_lines() {
524 let dir = tempfile::tempdir().unwrap();
525 let path = dir.path().join("mixed.jsonl");
526
527 let good = test_message(Role::User, "good");
528 let entry = TranscriptEntry {
529 seq: 0,
530 timestamp: "2026-01-01T00:00:00Z".to_owned(),
531 message: good.clone(),
532 };
533 let good_line = serde_json::to_string(&entry).unwrap();
534 let content = format!("{good_line}\nnot valid json\n{good_line}\n");
535 std::fs::write(&path, &content).unwrap();
536
537 let messages = TranscriptReader::load(&path).unwrap();
538 assert_eq!(messages.len(), 2);
539 }
540
541 #[test]
542 fn load_strict_fails_on_first_malformed_line() {
543 let dir = tempfile::tempdir().unwrap();
544 let path = dir.path().join("mixed.jsonl");
545
546 let good = test_message(Role::User, "good");
547 let entry = TranscriptEntry {
548 seq: 0,
549 timestamp: "2026-01-01T00:00:00Z".to_owned(),
550 message: good.clone(),
551 };
552 let good_line = serde_json::to_string(&entry).unwrap();
553 let content = format!("{good_line}\nnot valid json\n{good_line}\n");
556 std::fs::write(&path, &content).unwrap();
557
558 let err = TranscriptReader::load_strict(&path).unwrap_err();
559 assert_matches!(err, SubAgentError::Transcript(_));
560
561 let messages = TranscriptReader::load(&path).unwrap();
564 assert_eq!(messages.len(), 2);
565 }
566
567 #[test]
568 fn load_strict_succeeds_on_intact_file() {
569 let dir = tempfile::tempdir().unwrap();
570 let path = dir.path().join("clean.jsonl");
571
572 let good = test_message(Role::User, "good");
573 let entry = TranscriptEntry {
574 seq: 0,
575 timestamp: "2026-01-01T00:00:00Z".to_owned(),
576 message: good,
577 };
578 let good_line = serde_json::to_string(&entry).unwrap();
579 std::fs::write(&path, format!("{good_line}\n")).unwrap();
580
581 let messages = TranscriptReader::load_strict(&path).unwrap();
582 assert_eq!(messages.len(), 1);
583 }
584
585 #[test]
586 fn load_strict_missing_file_no_meta_returns_empty() {
587 let dir = tempfile::tempdir().unwrap();
588 let path = dir.path().join("ghost.jsonl");
589 let messages = TranscriptReader::load_strict(&path).unwrap();
590 assert!(messages.is_empty());
591 }
592
593 #[test]
594 fn meta_roundtrip() {
595 let dir = tempfile::tempdir().unwrap();
596 let meta = test_meta("abc-123");
597 TranscriptWriter::write_meta(dir.path(), "abc-123", &meta).unwrap();
598 let loaded = TranscriptReader::load_meta(dir.path(), "abc-123").unwrap();
599 assert_eq!(loaded.agent_id, "abc-123");
600 assert_eq!(loaded.turns_used, 2);
601 }
602
603 #[test]
604 fn meta_not_found_returns_not_found_error() {
605 let dir = tempfile::tempdir().unwrap();
606 let err = TranscriptReader::load_meta(dir.path(), "ghost").unwrap_err();
607 assert_matches!(err, SubAgentError::NotFound(_));
608 }
609
610 #[test]
611 fn find_by_prefix_exact() {
612 let dir = tempfile::tempdir().unwrap();
613 let meta = test_meta("abcdef01-0000-0000-0000-000000000000");
614 TranscriptWriter::write_meta(dir.path(), "abcdef01-0000-0000-0000-000000000000", &meta)
615 .unwrap();
616 let id =
617 TranscriptReader::find_by_prefix(dir.path(), "abcdef01-0000-0000-0000-000000000000")
618 .unwrap();
619 assert_eq!(id, "abcdef01-0000-0000-0000-000000000000");
620 }
621
622 #[test]
623 fn find_by_prefix_short_prefix() {
624 let dir = tempfile::tempdir().unwrap();
625 let meta = test_meta("deadbeef-0000-0000-0000-000000000000");
626 TranscriptWriter::write_meta(dir.path(), "deadbeef-0000-0000-0000-000000000000", &meta)
627 .unwrap();
628 let id = TranscriptReader::find_by_prefix(dir.path(), "deadbeef").unwrap();
629 assert_eq!(id, "deadbeef-0000-0000-0000-000000000000");
630 }
631
632 #[test]
633 fn find_by_prefix_not_found() {
634 let dir = tempfile::tempdir().unwrap();
635 let err = TranscriptReader::find_by_prefix(dir.path(), "xxxxxxxx").unwrap_err();
636 assert_matches!(err, SubAgentError::NotFound(_));
637 }
638
639 #[test]
640 fn find_by_prefix_ambiguous() {
641 let dir = tempfile::tempdir().unwrap();
642 TranscriptWriter::write_meta(dir.path(), "aabb0001-x", &test_meta("aabb0001-x")).unwrap();
643 TranscriptWriter::write_meta(dir.path(), "aabb0002-y", &test_meta("aabb0002-y")).unwrap();
644 let err = TranscriptReader::find_by_prefix(dir.path(), "aabb").unwrap_err();
645 assert_matches!(err, SubAgentError::AmbiguousId(_, 2));
646 }
647
648 #[test]
649 fn sweep_old_transcripts_removes_oldest() {
650 let dir = tempfile::tempdir().unwrap();
651
652 for i in 0..5u32 {
653 let path = dir.path().join(format!("file{i:02}.jsonl"));
654 std::fs::write(&path, b"").unwrap();
655 }
660
661 let deleted = sweep_old_transcripts(dir.path(), 3).unwrap();
662 assert_eq!(deleted, 2);
663
664 let remaining: Vec<_> = std::fs::read_dir(dir.path())
665 .unwrap()
666 .filter_map(std::result::Result::ok)
667 .filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("jsonl"))
668 .collect();
669 assert_eq!(remaining.len(), 3);
670 }
671
672 #[test]
673 fn sweep_with_zero_max_does_nothing() {
674 let dir = tempfile::tempdir().unwrap();
675 std::fs::write(dir.path().join("a.jsonl"), b"").unwrap();
676 let deleted = sweep_old_transcripts(dir.path(), 0).unwrap();
677 assert_eq!(deleted, 0);
678 }
679
680 #[test]
681 fn sweep_below_max_does_nothing() {
682 let dir = tempfile::tempdir().unwrap();
683 std::fs::write(dir.path().join("a.jsonl"), b"").unwrap();
684 let deleted = sweep_old_transcripts(dir.path(), 50).unwrap();
685 assert_eq!(deleted, 0);
686 }
687
688 #[test]
689 fn utc_now_format() {
690 let ts = utc_now();
691 assert_eq!(ts.len(), 20);
693 assert!(ts.ends_with('Z'));
694 assert!(ts.contains('T'));
695 }
696
697 #[test]
698 fn load_empty_file_returns_empty() {
699 let dir = tempfile::tempdir().unwrap();
700 let path = dir.path().join("empty.jsonl");
701 std::fs::write(&path, b"").unwrap();
702 let messages = TranscriptReader::load(&path).unwrap();
703 assert!(messages.is_empty());
704 }
705
706 #[test]
707 fn load_meta_invalid_json_returns_transcript_error() {
708 let dir = tempfile::tempdir().unwrap();
709 std::fs::write(dir.path().join("bad.meta.json"), b"not json at all {{{{").unwrap();
710 let err = TranscriptReader::load_meta(dir.path(), "bad").unwrap_err();
711 assert_matches!(err, SubAgentError::Transcript(_));
712 }
713
714 #[test]
715 fn sweep_removes_companion_meta() {
716 let dir = tempfile::tempdir().unwrap();
717 for i in 0..4u32 {
719 let stem = format!("file{i:02}");
720 std::fs::write(dir.path().join(format!("{stem}.jsonl")), b"").unwrap();
721 std::fs::write(dir.path().join(format!("{stem}.meta.json")), b"{}").unwrap();
722 }
723 let deleted = sweep_old_transcripts(dir.path(), 2).unwrap();
724 assert_eq!(deleted, 2);
725 let meta_count = std::fs::read_dir(dir.path())
727 .unwrap()
728 .filter_map(std::result::Result::ok)
729 .filter(|e| e.path().to_string_lossy().ends_with(".meta.json"))
730 .count();
731 assert_eq!(
732 meta_count, 2,
733 "orphaned meta sidecars should have been removed"
734 );
735 }
736
737 #[test]
738 fn data_loss_guard_uses_stem_based_meta_path() {
739 let dir = tempfile::tempdir().unwrap();
742 let agent_id = "deadbeef-0000-0000-0000-000000000000";
743 std::fs::write(dir.path().join(format!("{agent_id}.meta.json")), b"{}").unwrap();
745 let jsonl_path = dir.path().join(format!("{agent_id}.jsonl"));
746 let err = TranscriptReader::load(&jsonl_path).unwrap_err();
747 assert_matches!(err, SubAgentError::Transcript(ref m) if m.contains("missing"));
748 }
749
750 #[test]
751 fn meta_roundtrip_preserves_mcp_tool_names() {
752 let dir = tempfile::tempdir().unwrap();
753 let agent_id = "abc-123";
754 let mut meta = test_meta(agent_id);
755 meta.mcp_tool_names = vec!["search".into(), "write_file".into()];
756 TranscriptWriter::write_meta(dir.path(), agent_id, &meta).unwrap();
757 let loaded = TranscriptReader::load_meta(dir.path(), agent_id).unwrap();
758 assert_eq!(loaded.mcp_tool_names, vec!["search", "write_file"]);
759 }
760}