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 let file = match File::open(path) {
197 Ok(f) => f,
198 Err(e) if e.kind() == io::ErrorKind::NotFound => {
199 let meta_path =
203 if let (Some(parent), Some(stem)) = (path.parent(), path.file_stem()) {
204 parent.join(format!("{}.meta.json", stem.to_string_lossy()))
205 } else {
206 path.with_extension("meta.json")
207 };
208 if meta_path.exists() {
209 return Err(SubAgentError::Transcript(format!(
210 "transcript file '{}' is missing but meta sidecar exists — \
211 transcript data may have been deleted",
212 path.display()
213 )));
214 }
215 return Ok(vec![]);
216 }
217 Err(e) => {
218 return Err(SubAgentError::Transcript(format!(
219 "failed to open transcript '{}': {e}",
220 path.display()
221 )));
222 }
223 };
224
225 let reader = BufReader::new(file);
226 let mut messages = Vec::new();
227 for (line_no, line_result) in reader.lines().enumerate() {
228 let line = match line_result {
229 Ok(l) => l,
230 Err(e) => {
231 tracing::warn!(
232 path = %path.display(),
233 line = line_no + 1,
234 error = %e,
235 "failed to read transcript line — skipping"
236 );
237 continue;
238 }
239 };
240 let trimmed = line.trim();
241 if trimmed.is_empty() {
242 continue;
243 }
244 match serde_json::from_str::<TranscriptEntry>(trimmed) {
245 Ok(entry) => messages.push(entry.message),
246 Err(e) => {
247 tracing::warn!(
248 path = %path.display(),
249 line = line_no + 1,
250 error = %e,
251 "malformed transcript entry — skipping"
252 );
253 }
254 }
255 }
256 Ok(messages)
257 }
258
259 pub fn load_meta(dir: &Path, agent_id: &str) -> Result<TranscriptMeta, SubAgentError> {
266 let path = dir.join(format!("{agent_id}.meta.json"));
267 let content = fs::read_to_string(&path).map_err(|e| {
268 if e.kind() == io::ErrorKind::NotFound {
269 SubAgentError::NotFound(agent_id.to_owned())
270 } else {
271 SubAgentError::Transcript(format!("failed to read meta '{}': {e}", path.display()))
272 }
273 })?;
274 serde_json::from_str(&content).map_err(|e| {
275 SubAgentError::Transcript(format!("failed to parse meta '{}': {e}", path.display()))
276 })
277 }
278
279 pub fn find_by_prefix(dir: &Path, prefix: &str) -> Result<String, SubAgentError> {
288 let entries = fs::read_dir(dir).map_err(|e| {
289 SubAgentError::Transcript(format!(
290 "failed to read transcript dir '{}': {e}",
291 dir.display()
292 ))
293 })?;
294
295 let mut matches: Vec<String> = Vec::new();
296 for entry in entries {
297 let entry = entry
298 .map_err(|e| SubAgentError::Transcript(format!("failed to read dir entry: {e}")))?;
299 let name = entry.file_name();
300 let name_str = name.to_string_lossy();
301 if let Some(agent_id) = name_str.strip_suffix(".meta.json")
302 && agent_id.starts_with(prefix)
303 {
304 matches.push(agent_id.to_owned());
305 }
306 }
307
308 match matches.len() {
309 0 => Err(SubAgentError::NotFound(prefix.to_owned())),
310 1 => Ok(matches.remove(0)),
311 n => Err(SubAgentError::AmbiguousId(prefix.to_owned(), n)),
312 }
313 }
314}
315
316pub fn sweep_old_transcripts(dir: &Path, max_files: usize) -> io::Result<usize> {
325 if max_files == 0 {
326 return Ok(0);
327 }
328
329 if !dir.exists() {
331 fs::create_dir_all(dir)?;
332 return Ok(0);
333 }
334
335 let mut jsonl_files: Vec<(PathBuf, std::time::SystemTime)> = Vec::new();
336 for entry in fs::read_dir(dir)? {
337 let entry = entry?;
338 let path = entry.path();
339 if path.extension().and_then(|e| e.to_str()) == Some("jsonl") {
340 let mtime = entry
341 .metadata()
342 .and_then(|m| m.modified())
343 .unwrap_or(std::time::SystemTime::UNIX_EPOCH);
344 jsonl_files.push((path, mtime));
345 }
346 }
347
348 if jsonl_files.len() <= max_files {
349 return Ok(0);
350 }
351
352 jsonl_files.sort_by_key(|(_, mtime)| *mtime);
354
355 let to_delete = jsonl_files.len() - max_files;
356 let mut deleted = 0;
357 for (path, _) in jsonl_files.into_iter().take(to_delete) {
358 let meta = path.with_extension("meta.json");
360 if meta.exists() {
361 let _ = fs::remove_file(&meta);
362 }
363 fs::remove_file(&path)?;
364 deleted += 1;
365 }
366 Ok(deleted)
367}
368
369#[must_use]
371pub(crate) fn utc_now() -> String {
372 let secs = std::time::SystemTime::now()
375 .duration_since(std::time::UNIX_EPOCH)
376 .unwrap_or_default()
377 .as_secs();
378 let (y, mo, d, h, mi, s) = epoch_to_parts(secs);
379 format!("{y:04}-{mo:02}-{d:02}T{h:02}:{mi:02}:{s:02}Z")
380}
381
382fn epoch_to_parts(epoch: u64) -> (u32, u32, u32, u32, u32, u32) {
388 let sec = epoch % 60;
389 let epoch = epoch / 60;
390 let min = epoch % 60;
391 let epoch = epoch / 60;
392 let hour = epoch % 24;
393 let days = epoch / 24;
394
395 let z = days + 719_468;
397 let era = z / 146_097;
398 let doe = z - era * 146_097;
399 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
400 let year = yoe + era * 400;
401 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
402 let mp = (5 * doy + 2) / 153;
403 let day = doy - (153 * mp + 2) / 5 + 1;
404 let month = if mp < 10 { mp + 3 } else { mp - 9 };
405 let year = if month <= 2 { year + 1 } else { year };
406
407 #[allow(clippy::cast_possible_truncation)]
409 (
410 year as u32,
411 month as u32,
412 day as u32,
413 hour as u32,
414 min as u32,
415 sec as u32,
416 )
417}
418
419#[cfg(test)]
420mod tests {
421 use std::assert_matches;
422 use zeph_llm::provider::{Message, MessageMetadata, Role};
423
424 use super::*;
425
426 fn test_message(role: Role, content: &str) -> Message {
427 Message {
428 role,
429 content: content.to_owned(),
430 parts: vec![],
431 metadata: MessageMetadata::default(),
432 }
433 }
434
435 fn test_meta(agent_id: &str) -> TranscriptMeta {
436 TranscriptMeta {
437 agent_id: agent_id.to_owned(),
438 agent_name: "bot".to_owned(),
439 def_name: "bot".to_owned(),
440 status: SubAgentState::Completed,
441 started_at: "2026-01-01T00:00:00Z".to_owned(),
442 finished_at: Some("2026-01-01T00:01:00Z".to_owned()),
443 resumed_from: None,
444 turns_used: 2,
445 mcp_tool_names: Vec::new(),
446 }
447 }
448
449 #[tokio::test]
450 async fn writer_reader_roundtrip() {
451 let dir = tempfile::tempdir().unwrap();
452 let path = dir.path().join("test.jsonl");
453
454 let msg1 = test_message(Role::User, "hello");
455 let msg2 = test_message(Role::Assistant, "world");
456
457 let writer = TranscriptWriter::new(&path).unwrap();
458 writer.append(0, &msg1).await.unwrap();
459 writer.append(1, &msg2).await.unwrap();
460 drop(writer);
461
462 let messages = TranscriptReader::load(&path).unwrap();
463 assert_eq!(messages.len(), 2);
464 assert_eq!(messages[0].content, "hello");
465 assert_eq!(messages[1].content, "world");
466 }
467
468 #[test]
469 fn load_missing_file_no_meta_returns_empty() {
470 let dir = tempfile::tempdir().unwrap();
471 let path = dir.path().join("ghost.jsonl");
472 let messages = TranscriptReader::load(&path).unwrap();
473 assert!(messages.is_empty());
474 }
475
476 #[test]
477 fn load_missing_file_with_meta_returns_error() {
478 let dir = tempfile::tempdir().unwrap();
479 let meta_path = dir.path().join("ghost.meta.json");
480 std::fs::write(&meta_path, "{}").unwrap();
481 let jsonl_path = dir.path().join("ghost.jsonl");
482 let err = TranscriptReader::load(&jsonl_path).unwrap_err();
483 assert_matches!(err, SubAgentError::Transcript(_));
484 }
485
486 #[test]
487 fn load_skips_malformed_lines() {
488 let dir = tempfile::tempdir().unwrap();
489 let path = dir.path().join("mixed.jsonl");
490
491 let good = test_message(Role::User, "good");
492 let entry = TranscriptEntry {
493 seq: 0,
494 timestamp: "2026-01-01T00:00:00Z".to_owned(),
495 message: good.clone(),
496 };
497 let good_line = serde_json::to_string(&entry).unwrap();
498 let content = format!("{good_line}\nnot valid json\n{good_line}\n");
499 std::fs::write(&path, &content).unwrap();
500
501 let messages = TranscriptReader::load(&path).unwrap();
502 assert_eq!(messages.len(), 2);
503 }
504
505 #[test]
506 fn meta_roundtrip() {
507 let dir = tempfile::tempdir().unwrap();
508 let meta = test_meta("abc-123");
509 TranscriptWriter::write_meta(dir.path(), "abc-123", &meta).unwrap();
510 let loaded = TranscriptReader::load_meta(dir.path(), "abc-123").unwrap();
511 assert_eq!(loaded.agent_id, "abc-123");
512 assert_eq!(loaded.turns_used, 2);
513 }
514
515 #[test]
516 fn meta_not_found_returns_not_found_error() {
517 let dir = tempfile::tempdir().unwrap();
518 let err = TranscriptReader::load_meta(dir.path(), "ghost").unwrap_err();
519 assert_matches!(err, SubAgentError::NotFound(_));
520 }
521
522 #[test]
523 fn find_by_prefix_exact() {
524 let dir = tempfile::tempdir().unwrap();
525 let meta = test_meta("abcdef01-0000-0000-0000-000000000000");
526 TranscriptWriter::write_meta(dir.path(), "abcdef01-0000-0000-0000-000000000000", &meta)
527 .unwrap();
528 let id =
529 TranscriptReader::find_by_prefix(dir.path(), "abcdef01-0000-0000-0000-000000000000")
530 .unwrap();
531 assert_eq!(id, "abcdef01-0000-0000-0000-000000000000");
532 }
533
534 #[test]
535 fn find_by_prefix_short_prefix() {
536 let dir = tempfile::tempdir().unwrap();
537 let meta = test_meta("deadbeef-0000-0000-0000-000000000000");
538 TranscriptWriter::write_meta(dir.path(), "deadbeef-0000-0000-0000-000000000000", &meta)
539 .unwrap();
540 let id = TranscriptReader::find_by_prefix(dir.path(), "deadbeef").unwrap();
541 assert_eq!(id, "deadbeef-0000-0000-0000-000000000000");
542 }
543
544 #[test]
545 fn find_by_prefix_not_found() {
546 let dir = tempfile::tempdir().unwrap();
547 let err = TranscriptReader::find_by_prefix(dir.path(), "xxxxxxxx").unwrap_err();
548 assert_matches!(err, SubAgentError::NotFound(_));
549 }
550
551 #[test]
552 fn find_by_prefix_ambiguous() {
553 let dir = tempfile::tempdir().unwrap();
554 TranscriptWriter::write_meta(dir.path(), "aabb0001-x", &test_meta("aabb0001-x")).unwrap();
555 TranscriptWriter::write_meta(dir.path(), "aabb0002-y", &test_meta("aabb0002-y")).unwrap();
556 let err = TranscriptReader::find_by_prefix(dir.path(), "aabb").unwrap_err();
557 assert_matches!(err, SubAgentError::AmbiguousId(_, 2));
558 }
559
560 #[test]
561 fn sweep_old_transcripts_removes_oldest() {
562 let dir = tempfile::tempdir().unwrap();
563
564 for i in 0..5u32 {
565 let path = dir.path().join(format!("file{i:02}.jsonl"));
566 std::fs::write(&path, b"").unwrap();
567 }
572
573 let deleted = sweep_old_transcripts(dir.path(), 3).unwrap();
574 assert_eq!(deleted, 2);
575
576 let remaining: Vec<_> = std::fs::read_dir(dir.path())
577 .unwrap()
578 .filter_map(std::result::Result::ok)
579 .filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("jsonl"))
580 .collect();
581 assert_eq!(remaining.len(), 3);
582 }
583
584 #[test]
585 fn sweep_with_zero_max_does_nothing() {
586 let dir = tempfile::tempdir().unwrap();
587 std::fs::write(dir.path().join("a.jsonl"), b"").unwrap();
588 let deleted = sweep_old_transcripts(dir.path(), 0).unwrap();
589 assert_eq!(deleted, 0);
590 }
591
592 #[test]
593 fn sweep_below_max_does_nothing() {
594 let dir = tempfile::tempdir().unwrap();
595 std::fs::write(dir.path().join("a.jsonl"), b"").unwrap();
596 let deleted = sweep_old_transcripts(dir.path(), 50).unwrap();
597 assert_eq!(deleted, 0);
598 }
599
600 #[test]
601 fn utc_now_format() {
602 let ts = utc_now();
603 assert_eq!(ts.len(), 20);
605 assert!(ts.ends_with('Z'));
606 assert!(ts.contains('T'));
607 }
608
609 #[test]
610 fn load_empty_file_returns_empty() {
611 let dir = tempfile::tempdir().unwrap();
612 let path = dir.path().join("empty.jsonl");
613 std::fs::write(&path, b"").unwrap();
614 let messages = TranscriptReader::load(&path).unwrap();
615 assert!(messages.is_empty());
616 }
617
618 #[test]
619 fn load_meta_invalid_json_returns_transcript_error() {
620 let dir = tempfile::tempdir().unwrap();
621 std::fs::write(dir.path().join("bad.meta.json"), b"not json at all {{{{").unwrap();
622 let err = TranscriptReader::load_meta(dir.path(), "bad").unwrap_err();
623 assert_matches!(err, SubAgentError::Transcript(_));
624 }
625
626 #[test]
627 fn sweep_removes_companion_meta() {
628 let dir = tempfile::tempdir().unwrap();
629 for i in 0..4u32 {
631 let stem = format!("file{i:02}");
632 std::fs::write(dir.path().join(format!("{stem}.jsonl")), b"").unwrap();
633 std::fs::write(dir.path().join(format!("{stem}.meta.json")), b"{}").unwrap();
634 }
635 let deleted = sweep_old_transcripts(dir.path(), 2).unwrap();
636 assert_eq!(deleted, 2);
637 let meta_count = std::fs::read_dir(dir.path())
639 .unwrap()
640 .filter_map(std::result::Result::ok)
641 .filter(|e| e.path().to_string_lossy().ends_with(".meta.json"))
642 .count();
643 assert_eq!(
644 meta_count, 2,
645 "orphaned meta sidecars should have been removed"
646 );
647 }
648
649 #[test]
650 fn data_loss_guard_uses_stem_based_meta_path() {
651 let dir = tempfile::tempdir().unwrap();
654 let agent_id = "deadbeef-0000-0000-0000-000000000000";
655 std::fs::write(dir.path().join(format!("{agent_id}.meta.json")), b"{}").unwrap();
657 let jsonl_path = dir.path().join(format!("{agent_id}.jsonl"));
658 let err = TranscriptReader::load(&jsonl_path).unwrap_err();
659 assert_matches!(err, SubAgentError::Transcript(ref m) if m.contains("missing"));
660 }
661
662 #[test]
663 fn meta_roundtrip_preserves_mcp_tool_names() {
664 let dir = tempfile::tempdir().unwrap();
665 let agent_id = "abc-123";
666 let mut meta = test_meta(agent_id);
667 meta.mcp_tool_names = vec!["search".into(), "write_file".into()];
668 TranscriptWriter::write_meta(dir.path(), agent_id, &meta).unwrap();
669 let loaded = TranscriptReader::load_meta(dir.path(), agent_id).unwrap();
670 assert_eq!(loaded.mcp_tool_names, vec!["search", "write_file"]);
671 }
672}