1use std::fs::{self, File};
16use std::io::{self, BufRead, BufReader, Write as _};
17use std::path::{Path, PathBuf};
18use std::sync::{Arc, Mutex, RwLock as StdRwLock};
19
20use serde::{Deserialize, Serialize};
21use zeph_common::anchor::{Anchor, AnchorStore, AnchorSubsystem};
22use zeph_common::hash_chain::{
23 ChainHash, ChainKeyRing, KeyResolution, chain_next, genesis,
24 verify_chained_prefix_with_checkpoint,
25};
26use zeph_llm::provider::{Message, MessagePart};
27
28use super::error::SubAgentError;
29use super::state::SubAgentState;
30
31pub const CHAIN_DOMAIN: &str = "zeph-subagent transcript v1";
35
36static HISTORY_INTEGRITY: StdRwLock<Option<Arc<ChainKeyRing>>> = StdRwLock::new(None);
52
53pub fn configure_history_integrity(ring: Option<Arc<ChainKeyRing>>) {
77 if let Ok(mut guard) = HISTORY_INTEGRITY.write() {
78 *guard = ring;
79 }
80}
81
82fn history_integrity() -> Option<Arc<ChainKeyRing>> {
83 HISTORY_INTEGRITY.read().ok().and_then(|g| g.clone())
84}
85
86static ANCHOR_STORE: StdRwLock<Option<Arc<dyn AnchorStore>>> = StdRwLock::new(None);
91
92pub fn configure_anchor_store(store: Option<Arc<dyn AnchorStore>>) {
96 if let Ok(mut guard) = ANCHOR_STORE.write() {
97 *guard = store;
98 }
99}
100
101fn anchor_store() -> Option<Arc<dyn AnchorStore>> {
102 ANCHOR_STORE.read().ok().and_then(|g| g.clone())
103}
104
105fn file_identity(path: &Path) -> Vec<u8> {
109 path.file_stem()
110 .map(|s| s.to_string_lossy().into_owned())
111 .unwrap_or_default()
112 .into_bytes()
113}
114
115static WARNED_LEGACY_UNDER_KEY: std::sync::LazyLock<StdRwLock<std::collections::HashSet<PathBuf>>> =
119 std::sync::LazyLock::new(|| StdRwLock::new(std::collections::HashSet::new()));
120
121fn warn_legacy_under_active_key_once(path: &Path) {
131 let already_warned = WARNED_LEGACY_UNDER_KEY
132 .read()
133 .is_ok_and(|set| set.contains(path));
134 if already_warned {
135 return;
136 }
137 if let Ok(mut set) = WARNED_LEGACY_UNDER_KEY.write()
138 && !set.insert(path.to_path_buf())
139 {
140 return; }
142 tracing::warn!(
143 path = %path.display(),
144 "history-chain integrity: transcript classifies as legacy (no chain field anywhere) \
145 while a history-integrity key IS configured for this process — this is expected for \
146 genuine pre-upgrade content, but is also the signature of a full chain-strip downgrade \
147 attack (issue #6449, the vault-anchor gap); accepted per FR-006, flagged for operator \
148 visibility"
149 );
150}
151
152#[derive(Debug, Clone, Serialize, Deserialize)]
158pub struct TranscriptEntry {
159 pub seq: u32,
161 pub timestamp: String,
163 pub message: Message,
165 #[serde(default, skip_serializing_if = "Option::is_none")]
172 pub chain: Option<String>,
173}
174
175#[derive(Debug, Clone, Serialize, Deserialize)]
180pub struct TranscriptMeta {
181 pub agent_id: String,
183 pub agent_name: String,
185 pub def_name: String,
187 pub status: SubAgentState,
189 pub started_at: String,
191 #[serde(skip_serializing_if = "Option::is_none")]
193 pub finished_at: Option<String>,
194 #[serde(skip_serializing_if = "Option::is_none")]
196 pub resumed_from: Option<String>,
197 pub turns_used: u32,
199 #[serde(default)]
204 pub mcp_tool_names: Vec<String>,
205}
206
207struct TranscriptWriteState {
224 file: File,
225 prev: Option<ChainHash>,
229 count: u64,
233}
234
235#[derive(Clone)]
236pub struct TranscriptWriter {
237 state: Arc<Mutex<TranscriptWriteState>>,
243 file_identity: Vec<u8>,
244 ring: Option<Arc<ChainKeyRing>>,
248}
249
250impl TranscriptWriter {
251 pub fn new(path: &Path) -> io::Result<Self> {
266 if let Some(parent) = path.parent() {
267 fs::create_dir_all(parent)?;
268 }
269 let ring = history_integrity();
270 let identity = file_identity(path);
271
272 let (prev, count) = if path.exists() {
273 let entries =
274 parse_entries(path, false).map_err(|e| io::Error::other(e.to_string()))?;
275 let count = u64::try_from(entries.len()).unwrap_or(u64::MAX);
276 let anchor = match anchor_store() {
277 Some(store) => store
278 .get_sync(AnchorSubsystem::SubagentTranscript, &identity)
279 .map_err(|e| io::Error::other(format!("anchor lookup failed: {e}")))?,
280 None => None,
281 };
282 let (_messages, head) =
283 verify_and_extract_messages(path, entries, ring.as_deref(), anchor.as_ref())
284 .map_err(|e| io::Error::other(e.to_string()))?;
285 (head, count)
286 } else {
287 (None, 0)
288 };
289
290 let file = zeph_common::fs_secure::append_private(path)?;
291 Ok(Self {
292 state: Arc::new(Mutex::new(TranscriptWriteState { file, prev, count })),
293 file_identity: identity,
294 ring,
295 })
296 }
297
298 pub async fn append(&self, seq: u32, message: &Message) -> io::Result<()> {
318 let mut persisted_message = message.clone();
319 persisted_message.parts = MessagePart::strip_images(&persisted_message.parts);
320 let timestamp = utc_now();
321 let state = Arc::clone(&self.state);
322 let ring = self.ring.clone();
323 let identity = self.file_identity.clone();
324
325 tokio::task::spawn_blocking(move || {
326 let mut guard = state
327 .lock()
328 .map_err(|_| io::Error::other("transcript writer lock poisoned"))?;
329
330 let mut entry = TranscriptEntry {
331 seq,
332 timestamp,
333 message: persisted_message,
334 chain: None,
335 };
336
337 let new_head = match ring.as_deref() {
338 Some(ring) => {
339 let content = serde_json::to_vec(&entry)
340 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
341 let base = guard.prev.unwrap_or_else(|| {
342 genesis(
343 &ring.current_key(),
344 CHAIN_DOMAIN,
345 &identity,
346 ring.current_epoch(),
347 )
348 });
349 let h = chain_next(&ring.current_key(), &base, &content);
350 entry.chain = Some(h.to_hex());
351 Some(h)
352 }
353 None => None,
354 };
355
356 let line = serde_json::to_string(&entry)
357 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
358 guard.file.write_all(line.as_bytes())?;
359 guard.file.write_all(b"\n")?;
360 guard.file.flush()?;
361
362 if let Some(h) = new_head {
365 guard.prev = Some(h);
366 }
367 guard.count += 1;
368 Ok(())
369 })
370 .await
371 .map_err(|e| io::Error::other(format!("spawn_blocking panicked: {e}")))?
372 }
373
374 pub async fn finalize(self) -> io::Result<()> {
389 let Some(store) = anchor_store() else {
390 return Ok(());
391 };
392 let (head, count) = {
393 let guard = self
394 .state
395 .lock()
396 .map_err(|_| io::Error::other("transcript writer lock poisoned"))?;
397 let Some(head) = guard.prev else {
398 return Ok(());
399 };
400 (head, guard.count)
401 };
402 let epoch = self.ring.as_ref().map_or(0, |r| r.current_epoch());
403 let anchor = Anchor::new(epoch, count, head);
404 store
405 .put(
406 AnchorSubsystem::SubagentTranscript,
407 &self.file_identity,
408 anchor,
409 )
410 .await
411 .map_err(|e| io::Error::other(format!("anchor put failed: {e}")))
412 }
413
414 pub fn write_meta(dir: &Path, agent_id: &str, meta: &TranscriptMeta) -> io::Result<()> {
420 let path = dir.join(format!("{agent_id}.meta.json"));
421 let content = serde_json::to_string_pretty(meta)
422 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
423 zeph_common::fs_secure::write_private(&path, content.as_bytes())
424 }
425
426 pub async fn write_meta_async(
433 dir: &Path,
434 agent_id: &str,
435 meta: &TranscriptMeta,
436 ) -> io::Result<()> {
437 let path = dir.join(format!("{agent_id}.meta.json"));
438 let content = serde_json::to_string_pretty(meta)
439 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
440 let bytes = content.into_bytes();
441 tokio::task::spawn_blocking(move || zeph_common::fs_secure::write_private(&path, &bytes))
442 .await
443 .map_err(|e| io::Error::other(format!("spawn_blocking panicked: {e}")))?
444 }
445}
446
447pub struct TranscriptReader;
454
455impl TranscriptReader {
456 pub fn load(path: &Path) -> Result<Vec<Message>, SubAgentError> {
469 Self::load_impl(path, false)
470 }
471
472 pub fn load_strict(path: &Path) -> Result<Vec<Message>, SubAgentError> {
487 Self::load_impl(path, true)
488 }
489
490 fn load_impl(path: &Path, strict: bool) -> Result<Vec<Message>, SubAgentError> {
491 if !path.exists() {
492 let meta_path = if let (Some(parent), Some(stem)) = (path.parent(), path.file_stem()) {
496 parent.join(format!("{}.meta.json", stem.to_string_lossy()))
497 } else {
498 path.with_extension("meta.json")
499 };
500 if meta_path.exists() {
501 return Err(SubAgentError::Transcript(format!(
502 "transcript file '{}' is missing but meta sidecar exists — \
503 transcript data may have been deleted",
504 path.display()
505 )));
506 }
507 return Ok(vec![]);
508 }
509
510 let entries = parse_entries(path, strict)?;
511 let ring = history_integrity();
512 let identity = file_identity(path);
513 let anchor = match anchor_store() {
514 Some(store) => store
515 .get_sync(AnchorSubsystem::SubagentTranscript, &identity)
516 .map_err(|e| SubAgentError::Integrity(format!("anchor lookup failed: {e}")))?,
517 None => None,
518 };
519 let (messages, _head) =
520 verify_and_extract_messages(path, entries, ring.as_deref(), anchor.as_ref())?;
521 Ok(messages)
522 }
523
524 pub fn load_meta(dir: &Path, agent_id: &str) -> Result<TranscriptMeta, SubAgentError> {
531 let path = dir.join(format!("{agent_id}.meta.json"));
532 let content = fs::read_to_string(&path).map_err(|e| {
533 if e.kind() == io::ErrorKind::NotFound {
534 SubAgentError::NotFound(agent_id.to_owned())
535 } else {
536 SubAgentError::Transcript(format!("failed to read meta '{}': {e}", path.display()))
537 }
538 })?;
539 serde_json::from_str(&content).map_err(|e| {
540 SubAgentError::Transcript(format!("failed to parse meta '{}': {e}", path.display()))
541 })
542 }
543
544 pub fn find_by_prefix(dir: &Path, prefix: &str) -> Result<String, SubAgentError> {
553 let entries = fs::read_dir(dir).map_err(|e| {
554 SubAgentError::Transcript(format!(
555 "failed to read transcript dir '{}': {e}",
556 dir.display()
557 ))
558 })?;
559
560 let mut matches: Vec<String> = Vec::new();
561 for entry in entries {
562 let entry = entry
563 .map_err(|e| SubAgentError::Transcript(format!("failed to read dir entry: {e}")))?;
564 let name = entry.file_name();
565 let name_str = name.to_string_lossy();
566 if let Some(agent_id) = name_str.strip_suffix(".meta.json")
567 && agent_id.starts_with(prefix)
568 {
569 matches.push(agent_id.to_owned());
570 }
571 }
572
573 match matches.len() {
574 0 => Err(SubAgentError::NotFound(prefix.to_owned())),
575 1 => Ok(matches.remove(0)),
576 n => Err(SubAgentError::AmbiguousId(prefix.to_owned(), n)),
577 }
578 }
579}
580
581fn parse_entries(path: &Path, strict: bool) -> Result<Vec<TranscriptEntry>, SubAgentError> {
596 let file = File::open(path).map_err(|e| {
597 SubAgentError::Transcript(format!(
598 "failed to open transcript '{}': {e}",
599 path.display()
600 ))
601 })?;
602 let reader = BufReader::new(file);
603 let mut entries = Vec::new();
604 for (line_no, line_result) in reader.lines().enumerate() {
605 let line = match line_result {
606 Ok(l) => l,
607 Err(e) => {
608 if strict {
609 return Err(SubAgentError::Transcript(format!(
610 "failed to read transcript '{}' line {}: {e}",
611 path.display(),
612 line_no + 1
613 )));
614 }
615 tracing::warn!(
616 path = %path.display(),
617 line = line_no + 1,
618 error = %e,
619 "failed to read transcript line — skipping"
620 );
621 continue;
622 }
623 };
624 let trimmed = line.trim();
625 if trimmed.is_empty() {
626 continue;
627 }
628 match serde_json::from_str::<TranscriptEntry>(trimmed) {
629 Ok(entry) => entries.push(entry),
630 Err(e) => {
631 if strict {
632 return Err(SubAgentError::Transcript(format!(
633 "malformed transcript entry in '{}' line {}: {e}",
634 path.display(),
635 line_no + 1
636 )));
637 }
638 tracing::warn!(
639 path = %path.display(),
640 line = line_no + 1,
641 error = %e,
642 "malformed transcript entry — skipping"
643 );
644 }
645 }
646 }
647 Ok(entries)
648}
649
650#[allow(clippy::too_many_lines)]
674fn verify_and_extract_messages(
675 path: &Path,
676 entries: Vec<TranscriptEntry>,
677 ring: Option<&ChainKeyRing>,
678 anchor: Option<&Anchor>,
679) -> Result<(Vec<Message>, Option<ChainHash>), SubAgentError> {
680 let Some(chain_start) = entries.iter().position(|e| e.chain.is_some()) else {
681 if let Some(anchor) = anchor {
688 tracing::error!(
689 audit_event = "history_integrity_tamper",
690 subsystem = "subagent_transcript",
691 reason = "whole_strip_legacy_with_anchor",
692 path = %path.display(),
693 anchored_count = anchor.count,
694 "TAMPER DETECTED: transcript is legacy-looking but a vault anchor exists for it \
695 (issue #6449)"
696 );
697 return Err(SubAgentError::Integrity(format!(
698 "TAMPER DETECTED in transcript '{}': file has no chain metadata (legacy-looking) \
699 but a vault anchor exists for it (anchored at count={}) — this file was \
700 previously chained and its chain fields have been stripped",
701 path.display(),
702 anchor.count
703 )));
704 }
705 if ring.is_some() {
712 warn_legacy_under_active_key_once(path);
713 }
714 return Ok((entries.into_iter().map(|e| e.message).collect(), None));
715 };
716
717 for (offset, entry) in entries[chain_start..].iter().enumerate() {
718 if entry.chain.is_none() {
719 return Err(SubAgentError::Integrity(format!(
720 "transcript '{}' entry at chained-region position {offset} is missing its \
721 chain field while earlier entries in this file are chained — partial strip \
722 detected, TAMPER DETECTED",
723 path.display()
724 )));
725 }
726 }
727
728 let Some(ring) = ring else {
729 return Err(SubAgentError::Integrity(format!(
730 "transcript '{}' carries chain metadata but no history-integrity key is configured \
731 for this process — refusing to trust it unverified (NFR-004)",
732 path.display()
733 )));
734 };
735
736 let mut chained: Vec<(Vec<u8>, ChainHash)> = Vec::with_capacity(entries.len() - chain_start);
737 for entry in &entries[chain_start..] {
738 let stored_hex = entry.chain.as_deref().unwrap_or_default();
739 let stored = ChainHash::from_hex(stored_hex).map_err(|_| {
740 SubAgentError::Integrity(format!(
741 "transcript '{}' has a malformed chain hash",
742 path.display()
743 ))
744 })?;
745 let mut stripped = entry.clone();
746 stripped.chain = None;
747 let content = serde_json::to_vec(&stripped).map_err(|e| {
748 SubAgentError::Transcript(format!("failed to canonicalize transcript entry: {e}"))
749 })?;
750 chained.push((content, stored));
751 }
752
753 let identity = file_identity(path);
754 let on_disk_count = u64::try_from(entries.len()).unwrap_or(u64::MAX);
755 let checkpoint_index = anchor.and_then(|a| {
759 a.count
760 .checked_sub(u64::try_from(chain_start).unwrap_or(u64::MAX) + 1)
761 });
762 let (head, checkpoint_head, resolution) = verify_chained_prefix_with_checkpoint(
763 ring,
764 CHAIN_DOMAIN,
765 &identity,
766 &chained,
767 checkpoint_index.unwrap_or(u64::MAX),
768 )
769 .map_err(|e| describe_chain_error(path, &e))?;
770
771 if let KeyResolution::Rekeyed(epoch) = resolution {
772 tracing::info!(
773 path = %path.display(),
774 epoch,
775 "transcript verified under a previous key epoch (re-keyed, not tampered)"
776 );
777 }
778
779 if let Some(anchor) = anchor {
780 if on_disk_count < anchor.count {
781 tracing::error!(
782 audit_event = "history_integrity_tamper",
783 subsystem = "subagent_transcript",
784 reason = "truncated_below_anchor_count",
785 path = %path.display(),
786 on_disk_count,
787 anchored_count = anchor.count,
788 "TAMPER DETECTED: transcript truncated below its anchored count (issue #6449)"
789 );
790 return Err(SubAgentError::Integrity(format!(
791 "TAMPER DETECTED in transcript '{}': on-disk entry count ({on_disk_count}) is \
792 below the anchored count ({}) — the file was truncated after being anchored",
793 path.display(),
794 anchor.count
795 )));
796 }
797 let anchor_head = anchor.head().map_err(|e| {
798 SubAgentError::Integrity(format!(
799 "transcript '{}' anchor is malformed: {e}",
800 path.display()
801 ))
802 })?;
803 match checkpoint_head {
804 Some(h) if h == anchor_head => {}
805 _ => {
806 tracing::error!(
807 audit_event = "history_integrity_tamper",
808 subsystem = "subagent_transcript",
809 reason = "anchor_head_mismatch",
810 path = %path.display(),
811 anchored_count = anchor.count,
812 "TAMPER DETECTED: transcript chain head at the anchored count does not match \
813 the stored vault anchor (issue #6449)"
814 );
815 return Err(SubAgentError::Integrity(format!(
816 "TAMPER DETECTED in transcript '{}': chain head at the anchored count ({}) \
817 does not match the stored vault anchor",
818 path.display(),
819 anchor.count
820 )));
821 }
822 }
823 }
824
825 let messages = entries.into_iter().map(|e| e.message).collect();
826 Ok((messages, Some(head)))
827}
828
829fn describe_chain_error(path: &Path, err: &zeph_common::hash_chain::ChainError) -> SubAgentError {
833 use zeph_common::hash_chain::ChainError;
834 match err {
835 ChainError::Unverifiable => SubAgentError::Integrity(format!(
836 "transcript '{}' is unverifiable: no known key epoch (current or previous rotation \
837 window) produces a valid chain — possibly re-keyed past the rotation window, or \
838 tampered; this is fail-closed by design (NFR-004) and cannot be auto-recovered",
839 path.display()
840 )),
841 ChainError::Mismatch { index } => SubAgentError::Integrity(format!(
842 "TAMPER DETECTED in transcript '{}': chain hash mismatch at chained-entry index \
843 {index} — content was modified, reordered, or deleted after being written",
844 path.display()
845 )),
846 other => SubAgentError::Integrity(format!(
847 "transcript '{}' failed chain verification: {other}",
848 path.display()
849 )),
850 }
851}
852
853pub fn sweep_old_transcripts(dir: &Path, max_files: usize) -> io::Result<usize> {
877 if max_files == 0 {
878 return Ok(0);
879 }
880
881 if !dir.exists() {
883 fs::create_dir_all(dir)?;
884 return Ok(0);
885 }
886
887 let mut jsonl_files: Vec<(PathBuf, std::time::SystemTime)> = Vec::new();
888 for entry in fs::read_dir(dir)? {
889 let entry = entry?;
890 let path = entry.path();
891 if path.extension().and_then(|e| e.to_str()) == Some("jsonl") {
892 let mtime = entry
893 .metadata()
894 .and_then(|m| m.modified())
895 .unwrap_or(std::time::SystemTime::UNIX_EPOCH);
896 jsonl_files.push((path, mtime));
897 }
898 }
899
900 if jsonl_files.len() <= max_files {
901 return Ok(0);
902 }
903
904 jsonl_files.sort_by_key(|(_, mtime)| *mtime);
906
907 let to_delete = jsonl_files.len() - max_files;
908 let mut deleted = 0;
909 for (path, _) in jsonl_files.into_iter().take(to_delete) {
910 let meta = path.with_extension("meta.json");
912 if meta.exists() {
913 let _ = fs::remove_file(&meta);
914 }
915 fs::remove_file(&path)?;
916 deleted += 1;
917 }
918 Ok(deleted)
919}
920
921#[must_use]
923pub(crate) fn utc_now() -> String {
924 let secs = std::time::SystemTime::now()
927 .duration_since(std::time::UNIX_EPOCH)
928 .unwrap_or_default()
929 .as_secs();
930 let (y, mo, d, h, mi, s) = epoch_to_parts(secs);
931 format!("{y:04}-{mo:02}-{d:02}T{h:02}:{mi:02}:{s:02}Z")
932}
933
934fn epoch_to_parts(epoch: u64) -> (u32, u32, u32, u32, u32, u32) {
940 let sec = epoch % 60;
941 let epoch = epoch / 60;
942 let min = epoch % 60;
943 let epoch = epoch / 60;
944 let hour = epoch % 24;
945 let days = epoch / 24;
946
947 let z = days + 719_468;
949 let era = z / 146_097;
950 let doe = z - era * 146_097;
951 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
952 let year = yoe + era * 400;
953 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
954 let mp = (5 * doy + 2) / 153;
955 let day = doy - (153 * mp + 2) / 5 + 1;
956 let month = if mp < 10 { mp + 3 } else { mp - 9 };
957 let year = if month <= 2 { year + 1 } else { year };
958
959 #[allow(clippy::cast_possible_truncation)]
961 (
962 year as u32,
963 month as u32,
964 day as u32,
965 hour as u32,
966 min as u32,
967 sec as u32,
968 )
969}
970
971#[cfg(test)]
979pub(crate) struct IntegrityConfigGuard(());
980
981#[cfg(test)]
982impl IntegrityConfigGuard {
983 pub(crate) fn new() -> Self {
984 Self(())
985 }
986}
987
988#[cfg(test)]
989impl Drop for IntegrityConfigGuard {
990 fn drop(&mut self) {
991 configure_history_integrity(None);
992 configure_anchor_store(None);
993 }
994}
995
996#[cfg(test)]
997mod tests {
998 use std::assert_matches;
999 use zeph_llm::provider::{ImageData, Message, MessageMetadata, MessagePart, Role};
1000
1001 use super::*;
1002
1003 fn test_message(role: Role, content: &str) -> Message {
1004 Message {
1005 role,
1006 content: content.to_owned(),
1007 parts: vec![],
1008 metadata: MessageMetadata::default(),
1009 }
1010 }
1011
1012 fn test_meta(agent_id: &str) -> TranscriptMeta {
1013 TranscriptMeta {
1014 agent_id: agent_id.to_owned(),
1015 agent_name: "bot".to_owned(),
1016 def_name: "bot".to_owned(),
1017 status: SubAgentState::Completed,
1018 started_at: "2026-01-01T00:00:00Z".to_owned(),
1019 finished_at: Some("2026-01-01T00:01:00Z".to_owned()),
1020 resumed_from: None,
1021 turns_used: 2,
1022 mcp_tool_names: Vec::new(),
1023 }
1024 }
1025
1026 #[tokio::test]
1027 #[serial_test::serial(subagent_transcript_integrity)]
1028 async fn writer_reader_roundtrip() {
1029 let dir = tempfile::tempdir().unwrap();
1030 let path = dir.path().join("test.jsonl");
1031
1032 let msg1 = test_message(Role::User, "hello");
1033 let msg2 = test_message(Role::Assistant, "world");
1034
1035 let writer = TranscriptWriter::new(&path).unwrap();
1036 writer.append(0, &msg1).await.unwrap();
1037 writer.append(1, &msg2).await.unwrap();
1038 drop(writer);
1039
1040 let messages = TranscriptReader::load(&path).unwrap();
1041 assert_eq!(messages.len(), 2);
1042 assert_eq!(messages[0].content, "hello");
1043 assert_eq!(messages[1].content, "world");
1044 }
1045
1046 #[tokio::test]
1050 #[serial_test::serial(subagent_transcript_integrity)]
1051 async fn append_strips_image_parts() {
1052 let dir = tempfile::tempdir().unwrap();
1053 let path = dir.path().join("test.jsonl");
1054
1055 let mut msg = test_message(Role::User, "look at this");
1056 msg.parts = vec![
1057 MessagePart::Text {
1058 text: "look at this".to_owned(),
1059 },
1060 MessagePart::Image(Box::new(ImageData {
1061 data: vec![0xFFu8, 0xD8, 0xFF, 0xE0],
1062 mime_type: "image/jpeg".to_owned(),
1063 })),
1064 ];
1065
1066 let writer = TranscriptWriter::new(&path).unwrap();
1067 writer.append(0, &msg).await.unwrap();
1068
1069 assert_eq!(msg.parts.len(), 2);
1071
1072 let messages = TranscriptReader::load(&path).unwrap();
1073 assert_eq!(messages.len(), 1);
1074 assert_eq!(messages[0].parts.len(), 1);
1075 assert!(matches!(messages[0].parts[0], MessagePart::Text { .. }));
1076 assert!(
1077 !messages[0]
1078 .parts
1079 .iter()
1080 .any(|p| matches!(p, MessagePart::Image(_))),
1081 "transcript must not retain Image parts"
1082 );
1083
1084 let raw = std::fs::read_to_string(&path).unwrap();
1086 assert!(
1087 !raw.contains("mime_type") && !raw.contains("image/jpeg"),
1088 "raw image payload leaked into transcript file"
1089 );
1090 }
1091
1092 #[tokio::test]
1093 #[serial_test::serial(subagent_transcript_integrity)]
1094 async fn append_preserves_non_image_parts() {
1095 let dir = tempfile::tempdir().unwrap();
1096 let path = dir.path().join("test.jsonl");
1097
1098 let mut msg = test_message(Role::Assistant, "used a tool");
1099 msg.parts = vec![
1100 MessagePart::Text {
1101 text: "used a tool".to_owned(),
1102 },
1103 MessagePart::ToolUse {
1104 id: "call-1".to_owned(),
1105 name: "search".to_owned(),
1106 input: serde_json::json!({"query": "rust"}),
1107 },
1108 ];
1109
1110 let writer = TranscriptWriter::new(&path).unwrap();
1111 writer.append(0, &msg).await.unwrap();
1112
1113 let messages = TranscriptReader::load(&path).unwrap();
1114 assert_eq!(messages.len(), 1);
1115 assert_eq!(messages[0].parts.len(), 2);
1116 assert!(matches!(messages[0].parts[0], MessagePart::Text { .. }));
1117 assert!(matches!(messages[0].parts[1], MessagePart::ToolUse { .. }));
1118 }
1119
1120 #[tokio::test]
1121 #[serial_test::serial(subagent_transcript_integrity)]
1122 async fn append_empty_parts_unchanged() {
1123 let dir = tempfile::tempdir().unwrap();
1124 let path = dir.path().join("test.jsonl");
1125
1126 let msg = test_message(Role::User, "plain task message");
1129 assert!(msg.parts.is_empty());
1130
1131 let writer = TranscriptWriter::new(&path).unwrap();
1132 writer.append(0, &msg).await.unwrap();
1133
1134 let messages = TranscriptReader::load(&path).unwrap();
1135 assert_eq!(messages.len(), 1);
1136 assert!(messages[0].parts.is_empty());
1137 assert_eq!(messages[0].content, "plain task message");
1138 }
1139
1140 #[test]
1141 #[serial_test::serial(subagent_transcript_integrity)]
1142 fn load_missing_file_no_meta_returns_empty() {
1143 let dir = tempfile::tempdir().unwrap();
1144 let path = dir.path().join("ghost.jsonl");
1145 let messages = TranscriptReader::load(&path).unwrap();
1146 assert!(messages.is_empty());
1147 }
1148
1149 #[test]
1150 #[serial_test::serial(subagent_transcript_integrity)]
1151 fn load_missing_file_with_meta_returns_error() {
1152 let dir = tempfile::tempdir().unwrap();
1153 let meta_path = dir.path().join("ghost.meta.json");
1154 std::fs::write(&meta_path, "{}").unwrap();
1155 let jsonl_path = dir.path().join("ghost.jsonl");
1156 let err = TranscriptReader::load(&jsonl_path).unwrap_err();
1157 assert_matches!(err, SubAgentError::Transcript(_));
1158 }
1159
1160 #[test]
1161 #[serial_test::serial(subagent_transcript_integrity)]
1162 fn load_skips_malformed_lines() {
1163 let dir = tempfile::tempdir().unwrap();
1164 let path = dir.path().join("mixed.jsonl");
1165
1166 let good = test_message(Role::User, "good");
1167 let entry = TranscriptEntry {
1168 seq: 0,
1169 timestamp: "2026-01-01T00:00:00Z".to_owned(),
1170 message: good.clone(),
1171 chain: None,
1172 };
1173 let good_line = serde_json::to_string(&entry).unwrap();
1174 let content = format!("{good_line}\nnot valid json\n{good_line}\n");
1175 std::fs::write(&path, &content).unwrap();
1176
1177 let messages = TranscriptReader::load(&path).unwrap();
1178 assert_eq!(messages.len(), 2);
1179 }
1180
1181 #[test]
1182 #[serial_test::serial(subagent_transcript_integrity)]
1183 fn load_strict_fails_on_first_malformed_line() {
1184 let dir = tempfile::tempdir().unwrap();
1185 let path = dir.path().join("mixed.jsonl");
1186
1187 let good = test_message(Role::User, "good");
1188 let entry = TranscriptEntry {
1189 seq: 0,
1190 timestamp: "2026-01-01T00:00:00Z".to_owned(),
1191 message: good.clone(),
1192 chain: None,
1193 };
1194 let good_line = serde_json::to_string(&entry).unwrap();
1195 let content = format!("{good_line}\nnot valid json\n{good_line}\n");
1198 std::fs::write(&path, &content).unwrap();
1199
1200 let err = TranscriptReader::load_strict(&path).unwrap_err();
1201 assert_matches!(err, SubAgentError::Transcript(_));
1202
1203 let messages = TranscriptReader::load(&path).unwrap();
1206 assert_eq!(messages.len(), 2);
1207 }
1208
1209 #[test]
1210 #[serial_test::serial(subagent_transcript_integrity)]
1211 fn load_strict_succeeds_on_intact_file() {
1212 let dir = tempfile::tempdir().unwrap();
1213 let path = dir.path().join("clean.jsonl");
1214
1215 let good = test_message(Role::User, "good");
1216 let entry = TranscriptEntry {
1217 seq: 0,
1218 timestamp: "2026-01-01T00:00:00Z".to_owned(),
1219 message: good,
1220 chain: None,
1221 };
1222 let good_line = serde_json::to_string(&entry).unwrap();
1223 std::fs::write(&path, format!("{good_line}\n")).unwrap();
1224
1225 let messages = TranscriptReader::load_strict(&path).unwrap();
1226 assert_eq!(messages.len(), 1);
1227 }
1228
1229 #[test]
1230 #[serial_test::serial(subagent_transcript_integrity)]
1231 fn load_strict_missing_file_no_meta_returns_empty() {
1232 let dir = tempfile::tempdir().unwrap();
1233 let path = dir.path().join("ghost.jsonl");
1234 let messages = TranscriptReader::load_strict(&path).unwrap();
1235 assert!(messages.is_empty());
1236 }
1237
1238 #[test]
1239 #[serial_test::serial(subagent_transcript_integrity)]
1240 fn meta_roundtrip() {
1241 let dir = tempfile::tempdir().unwrap();
1242 let meta = test_meta("abc-123");
1243 TranscriptWriter::write_meta(dir.path(), "abc-123", &meta).unwrap();
1244 let loaded = TranscriptReader::load_meta(dir.path(), "abc-123").unwrap();
1245 assert_eq!(loaded.agent_id, "abc-123");
1246 assert_eq!(loaded.turns_used, 2);
1247 }
1248
1249 #[test]
1250 #[serial_test::serial(subagent_transcript_integrity)]
1251 fn meta_not_found_returns_not_found_error() {
1252 let dir = tempfile::tempdir().unwrap();
1253 let err = TranscriptReader::load_meta(dir.path(), "ghost").unwrap_err();
1254 assert_matches!(err, SubAgentError::NotFound(_));
1255 }
1256
1257 #[test]
1258 #[serial_test::serial(subagent_transcript_integrity)]
1259 fn find_by_prefix_exact() {
1260 let dir = tempfile::tempdir().unwrap();
1261 let meta = test_meta("abcdef01-0000-0000-0000-000000000000");
1262 TranscriptWriter::write_meta(dir.path(), "abcdef01-0000-0000-0000-000000000000", &meta)
1263 .unwrap();
1264 let id =
1265 TranscriptReader::find_by_prefix(dir.path(), "abcdef01-0000-0000-0000-000000000000")
1266 .unwrap();
1267 assert_eq!(id, "abcdef01-0000-0000-0000-000000000000");
1268 }
1269
1270 #[test]
1271 #[serial_test::serial(subagent_transcript_integrity)]
1272 fn find_by_prefix_short_prefix() {
1273 let dir = tempfile::tempdir().unwrap();
1274 let meta = test_meta("deadbeef-0000-0000-0000-000000000000");
1275 TranscriptWriter::write_meta(dir.path(), "deadbeef-0000-0000-0000-000000000000", &meta)
1276 .unwrap();
1277 let id = TranscriptReader::find_by_prefix(dir.path(), "deadbeef").unwrap();
1278 assert_eq!(id, "deadbeef-0000-0000-0000-000000000000");
1279 }
1280
1281 #[test]
1282 #[serial_test::serial(subagent_transcript_integrity)]
1283 fn find_by_prefix_not_found() {
1284 let dir = tempfile::tempdir().unwrap();
1285 let err = TranscriptReader::find_by_prefix(dir.path(), "xxxxxxxx").unwrap_err();
1286 assert_matches!(err, SubAgentError::NotFound(_));
1287 }
1288
1289 #[test]
1290 #[serial_test::serial(subagent_transcript_integrity)]
1291 fn find_by_prefix_ambiguous() {
1292 let dir = tempfile::tempdir().unwrap();
1293 TranscriptWriter::write_meta(dir.path(), "aabb0001-x", &test_meta("aabb0001-x")).unwrap();
1294 TranscriptWriter::write_meta(dir.path(), "aabb0002-y", &test_meta("aabb0002-y")).unwrap();
1295 let err = TranscriptReader::find_by_prefix(dir.path(), "aabb").unwrap_err();
1296 assert_matches!(err, SubAgentError::AmbiguousId(_, 2));
1297 }
1298
1299 #[test]
1300 #[serial_test::serial(subagent_transcript_integrity)]
1301 fn sweep_old_transcripts_removes_oldest() {
1302 let dir = tempfile::tempdir().unwrap();
1303
1304 for i in 0..5u32 {
1305 let path = dir.path().join(format!("file{i:02}.jsonl"));
1306 std::fs::write(&path, b"").unwrap();
1307 }
1312
1313 let deleted = sweep_old_transcripts(dir.path(), 3).unwrap();
1314 assert_eq!(deleted, 2);
1315
1316 let remaining: Vec<_> = std::fs::read_dir(dir.path())
1317 .unwrap()
1318 .filter_map(std::result::Result::ok)
1319 .filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("jsonl"))
1320 .collect();
1321 assert_eq!(remaining.len(), 3);
1322 }
1323
1324 #[test]
1325 #[serial_test::serial(subagent_transcript_integrity)]
1326 fn sweep_with_zero_max_does_nothing() {
1327 let dir = tempfile::tempdir().unwrap();
1328 std::fs::write(dir.path().join("a.jsonl"), b"").unwrap();
1329 let deleted = sweep_old_transcripts(dir.path(), 0).unwrap();
1330 assert_eq!(deleted, 0);
1331 }
1332
1333 #[test]
1334 #[serial_test::serial(subagent_transcript_integrity)]
1335 fn sweep_below_max_does_nothing() {
1336 let dir = tempfile::tempdir().unwrap();
1337 std::fs::write(dir.path().join("a.jsonl"), b"").unwrap();
1338 let deleted = sweep_old_transcripts(dir.path(), 50).unwrap();
1339 assert_eq!(deleted, 0);
1340 }
1341
1342 #[test]
1343 #[serial_test::serial(subagent_transcript_integrity)]
1344 fn utc_now_format() {
1345 let ts = utc_now();
1346 assert_eq!(ts.len(), 20);
1348 assert!(ts.ends_with('Z'));
1349 assert!(ts.contains('T'));
1350 }
1351
1352 #[test]
1353 #[serial_test::serial(subagent_transcript_integrity)]
1354 fn load_empty_file_returns_empty() {
1355 let dir = tempfile::tempdir().unwrap();
1356 let path = dir.path().join("empty.jsonl");
1357 std::fs::write(&path, b"").unwrap();
1358 let messages = TranscriptReader::load(&path).unwrap();
1359 assert!(messages.is_empty());
1360 }
1361
1362 #[test]
1363 #[serial_test::serial(subagent_transcript_integrity)]
1364 fn load_meta_invalid_json_returns_transcript_error() {
1365 let dir = tempfile::tempdir().unwrap();
1366 std::fs::write(dir.path().join("bad.meta.json"), b"not json at all {{{{").unwrap();
1367 let err = TranscriptReader::load_meta(dir.path(), "bad").unwrap_err();
1368 assert_matches!(err, SubAgentError::Transcript(_));
1369 }
1370
1371 #[test]
1372 #[serial_test::serial(subagent_transcript_integrity)]
1373 fn sweep_removes_companion_meta() {
1374 let dir = tempfile::tempdir().unwrap();
1375 for i in 0..4u32 {
1377 let stem = format!("file{i:02}");
1378 std::fs::write(dir.path().join(format!("{stem}.jsonl")), b"").unwrap();
1379 std::fs::write(dir.path().join(format!("{stem}.meta.json")), b"{}").unwrap();
1380 }
1381 let deleted = sweep_old_transcripts(dir.path(), 2).unwrap();
1382 assert_eq!(deleted, 2);
1383 let meta_count = std::fs::read_dir(dir.path())
1385 .unwrap()
1386 .filter_map(std::result::Result::ok)
1387 .filter(|e| e.path().to_string_lossy().ends_with(".meta.json"))
1388 .count();
1389 assert_eq!(
1390 meta_count, 2,
1391 "orphaned meta sidecars should have been removed"
1392 );
1393 }
1394
1395 #[test]
1396 #[serial_test::serial(subagent_transcript_integrity)]
1397 fn data_loss_guard_uses_stem_based_meta_path() {
1398 let dir = tempfile::tempdir().unwrap();
1401 let agent_id = "deadbeef-0000-0000-0000-000000000000";
1402 std::fs::write(dir.path().join(format!("{agent_id}.meta.json")), b"{}").unwrap();
1404 let jsonl_path = dir.path().join(format!("{agent_id}.jsonl"));
1405 let err = TranscriptReader::load(&jsonl_path).unwrap_err();
1406 assert_matches!(err, SubAgentError::Transcript(ref m) if m.contains("missing"));
1407 }
1408
1409 #[test]
1410 #[serial_test::serial(subagent_transcript_integrity)]
1411 fn meta_roundtrip_preserves_mcp_tool_names() {
1412 let dir = tempfile::tempdir().unwrap();
1413 let agent_id = "abc-123";
1414 let mut meta = test_meta(agent_id);
1415 meta.mcp_tool_names = vec!["search".into(), "write_file".into()];
1416 TranscriptWriter::write_meta(dir.path(), agent_id, &meta).unwrap();
1417 let loaded = TranscriptReader::load_meta(dir.path(), agent_id).unwrap();
1418 assert_eq!(loaded.mcp_tool_names, vec!["search", "write_file"]);
1419 }
1420
1421 fn test_ring(epoch: u32, byte: u8) -> Arc<ChainKeyRing> {
1428 Arc::new(ChainKeyRing::new(
1429 epoch,
1430 zeph_common::hash_chain::ChainKey::new([byte; 32]),
1431 ))
1432 }
1433
1434 #[tokio::test]
1435 #[serial_test::serial(subagent_transcript_integrity)]
1436 async fn chained_writer_reader_roundtrip() {
1437 let _guard = IntegrityConfigGuard::new();
1438 configure_history_integrity(Some(test_ring(0, 1)));
1439 let dir = tempfile::tempdir().unwrap();
1440 let path = dir.path().join("abc.jsonl");
1441
1442 let writer = TranscriptWriter::new(&path).unwrap();
1443 writer
1444 .append(0, &test_message(Role::User, "hello"))
1445 .await
1446 .unwrap();
1447 writer
1448 .append(1, &test_message(Role::Assistant, "world"))
1449 .await
1450 .unwrap();
1451 drop(writer);
1452
1453 let raw = std::fs::read_to_string(&path).unwrap();
1454 assert!(
1455 raw.lines().all(|l| l.contains("\"chain\":")),
1456 "every line must carry a chain field once integrity is configured"
1457 );
1458
1459 let messages = TranscriptReader::load(&path).unwrap();
1460 assert_eq!(messages.len(), 2);
1461 assert_eq!(messages[0].content, "hello");
1462 assert_eq!(messages[1].content, "world");
1463 }
1464
1465 #[tokio::test]
1466 #[serial_test::serial(subagent_transcript_integrity)]
1467 async fn tamper_in_place_edit_is_detected() {
1468 let _guard = IntegrityConfigGuard::new();
1469 configure_history_integrity(Some(test_ring(0, 2)));
1470 let dir = tempfile::tempdir().unwrap();
1471 let path = dir.path().join("abc.jsonl");
1472
1473 let writer = TranscriptWriter::new(&path).unwrap();
1474 writer
1478 .append(0, &test_message(Role::User, "untouched"))
1479 .await
1480 .unwrap();
1481 writer
1482 .append(1, &test_message(Role::Assistant, "original"))
1483 .await
1484 .unwrap();
1485 drop(writer);
1486
1487 let raw = std::fs::read_to_string(&path).unwrap();
1488 let tampered = raw.replace("original", "forged-approval");
1489 assert_ne!(raw, tampered);
1490 std::fs::write(&path, tampered).unwrap();
1491
1492 let err = TranscriptReader::load(&path).unwrap_err();
1493 assert_matches!(err, SubAgentError::Integrity(ref m) if m.contains("TAMPER"));
1494 let err = TranscriptReader::load_strict(&path).unwrap_err();
1497 assert_matches!(err, SubAgentError::Integrity(_));
1498 }
1499
1500 #[tokio::test]
1501 #[serial_test::serial(subagent_transcript_integrity)]
1502 async fn legacy_file_is_auto_trusted_once_when_integrity_configured_later() {
1503 let _guard = IntegrityConfigGuard::new();
1504 configure_history_integrity(None);
1506 let dir = tempfile::tempdir().unwrap();
1507 let path = dir.path().join("legacy.jsonl");
1508 let writer = TranscriptWriter::new(&path).unwrap();
1509 writer
1510 .append(0, &test_message(Role::User, "pre-feature message"))
1511 .await
1512 .unwrap();
1513 drop(writer);
1514
1515 let raw = std::fs::read_to_string(&path).unwrap();
1516 assert!(
1517 !raw.contains("\"chain\":"),
1518 "legacy file must carry no chain field"
1519 );
1520
1521 configure_history_integrity(Some(test_ring(0, 3)));
1523 let messages = TranscriptReader::load(&path).unwrap();
1524 assert_eq!(
1525 messages.len(),
1526 1,
1527 "legacy content must be auto-trusted, not rejected"
1528 );
1529
1530 assert!(
1533 WARNED_LEGACY_UNDER_KEY.read().unwrap().contains(&path),
1534 "path must be recorded as warned after the first legacy-under-active-key read"
1535 );
1536 let warned_count_before = WARNED_LEGACY_UNDER_KEY.read().unwrap().len();
1537 let _ = TranscriptReader::load(&path).unwrap();
1538 assert_eq!(
1539 WARNED_LEGACY_UNDER_KEY.read().unwrap().len(),
1540 warned_count_before,
1541 "a second read of the same path must not add a second warned-set entry"
1542 );
1543 }
1544
1545 #[tokio::test]
1546 #[serial_test::serial(subagent_transcript_integrity)]
1547 async fn partial_strip_of_chain_field_is_detected_as_tamper() {
1548 let _guard = IntegrityConfigGuard::new();
1549 configure_history_integrity(Some(test_ring(0, 4)));
1550 let dir = tempfile::tempdir().unwrap();
1551 let path = dir.path().join("abc.jsonl");
1552
1553 let writer = TranscriptWriter::new(&path).unwrap();
1554 writer
1555 .append(0, &test_message(Role::User, "one"))
1556 .await
1557 .unwrap();
1558 writer
1559 .append(1, &test_message(Role::Assistant, "two"))
1560 .await
1561 .unwrap();
1562 drop(writer);
1563
1564 let raw = std::fs::read_to_string(&path).unwrap();
1567 let lines: Vec<&str> = raw.lines().collect();
1568 assert_eq!(lines.len(), 2);
1569 let mut second: serde_json::Value = serde_json::from_str(lines[1]).unwrap();
1570 second.as_object_mut().unwrap().remove("chain");
1571 let stripped = format!("{}\n{}\n", lines[0], second);
1572 std::fs::write(&path, stripped).unwrap();
1573
1574 let err = TranscriptReader::load(&path).unwrap_err();
1575 assert_matches!(err, SubAgentError::Integrity(ref m) if m.contains("partial strip"));
1576 }
1577
1578 #[tokio::test]
1579 #[serial_test::serial(subagent_transcript_integrity)]
1580 async fn key_unavailable_on_chained_file_fails_closed_not_legacy() {
1581 let _guard = IntegrityConfigGuard::new();
1582 configure_history_integrity(Some(test_ring(0, 5)));
1583 let dir = tempfile::tempdir().unwrap();
1584 let path = dir.path().join("abc.jsonl");
1585 let writer = TranscriptWriter::new(&path).unwrap();
1586 writer
1587 .append(0, &test_message(Role::User, "chained"))
1588 .await
1589 .unwrap();
1590 drop(writer);
1591
1592 configure_history_integrity(None);
1594 let err = TranscriptReader::load(&path).unwrap_err();
1595 assert_matches!(err, SubAgentError::Integrity(ref m) if m.contains("NFR-004") || m.contains("no history-integrity key"));
1596 }
1597
1598 #[tokio::test]
1599 #[serial_test::serial(subagent_transcript_integrity)]
1600 async fn rotated_key_epoch_verifies_as_rekeyed_not_tampered() {
1601 let _guard = IntegrityConfigGuard::new();
1602 let old_key_byte = 6u8;
1603 configure_history_integrity(Some(test_ring(0, old_key_byte)));
1604 let dir = tempfile::tempdir().unwrap();
1605 let path = dir.path().join("abc.jsonl");
1606 let writer = TranscriptWriter::new(&path).unwrap();
1607 writer
1608 .append(0, &test_message(Role::User, "written before rotation"))
1609 .await
1610 .unwrap();
1611 drop(writer);
1612
1613 let ring = Arc::new(
1615 ChainKeyRing::new(1, zeph_common::hash_chain::ChainKey::new([9u8; 32])).with_previous(
1616 0,
1617 zeph_common::hash_chain::ChainKey::new([old_key_byte; 32]),
1618 ),
1619 );
1620 configure_history_integrity(Some(ring));
1621
1622 let messages = TranscriptReader::load(&path).unwrap();
1623 assert_eq!(
1624 messages.len(),
1625 1,
1626 "a legitimately re-keyed file must still verify"
1627 );
1628 }
1629
1630 #[tokio::test]
1631 #[serial_test::serial(subagent_transcript_integrity)]
1632 async fn writer_reopen_seeds_chain_from_existing_tail() {
1633 let _guard = IntegrityConfigGuard::new();
1634 configure_history_integrity(Some(test_ring(0, 7)));
1635 let dir = tempfile::tempdir().unwrap();
1636 let path = dir.path().join("abc.jsonl");
1637
1638 {
1639 let writer = TranscriptWriter::new(&path).unwrap();
1640 writer
1641 .append(0, &test_message(Role::User, "first session"))
1642 .await
1643 .unwrap();
1644 }
1645 {
1647 let writer = TranscriptWriter::new(&path).unwrap();
1648 writer
1649 .append(1, &test_message(Role::Assistant, "second session"))
1650 .await
1651 .unwrap();
1652 }
1653
1654 let messages = TranscriptReader::load(&path).unwrap();
1656 assert_eq!(messages.len(), 2);
1657 }
1658
1659 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
1663 #[serial_test::serial(subagent_transcript_integrity)]
1664 async fn concurrent_append_preserves_chain_order() {
1665 const N: u32 = 50;
1666 let _guard = IntegrityConfigGuard::new();
1667 configure_history_integrity(Some(test_ring(0, 8)));
1668 let dir = tempfile::tempdir().unwrap();
1669 let path = dir.path().join("abc.jsonl");
1670 let writer = TranscriptWriter::new(&path).unwrap();
1671
1672 let mut tasks = tokio::task::JoinSet::new();
1673 for i in 0..N {
1674 let writer = writer.clone();
1675 tasks.spawn(async move {
1676 writer
1677 .append(i, &test_message(Role::User, &format!("msg-{i}")))
1678 .await
1679 .unwrap();
1680 });
1681 }
1682 while tasks.join_next().await.is_some() {}
1683 drop(writer);
1684
1685 let messages = TranscriptReader::load(&path).unwrap();
1688 assert_eq!(messages.len(), usize::try_from(N).unwrap());
1689 }
1690
1691 #[derive(Default)]
1697 struct MockAnchorStore {
1698 map: std::sync::Mutex<std::collections::HashMap<String, Anchor>>,
1699 }
1700
1701 impl AnchorStore for MockAnchorStore {
1702 fn get(
1703 &self,
1704 subsystem: AnchorSubsystem,
1705 file_id: &[u8],
1706 ) -> std::pin::Pin<
1707 Box<
1708 dyn std::future::Future<
1709 Output = Result<Option<Anchor>, zeph_common::anchor::AnchorError>,
1710 > + Send
1711 + '_,
1712 >,
1713 > {
1714 let result = self.get_sync(subsystem, file_id);
1715 Box::pin(async move { result })
1716 }
1717
1718 fn get_sync(
1719 &self,
1720 subsystem: AnchorSubsystem,
1721 file_id: &[u8],
1722 ) -> Result<Option<Anchor>, zeph_common::anchor::AnchorError> {
1723 let key = zeph_common::anchor::anchor_key(subsystem, file_id);
1724 Ok(self.map.lock().unwrap().get(&key).cloned())
1725 }
1726
1727 fn put(
1728 &self,
1729 subsystem: AnchorSubsystem,
1730 file_id: &[u8],
1731 anchor: Anchor,
1732 ) -> std::pin::Pin<
1733 Box<
1734 dyn std::future::Future<Output = Result<(), zeph_common::anchor::AnchorError>>
1735 + Send
1736 + '_,
1737 >,
1738 > {
1739 let key = zeph_common::anchor::anchor_key(subsystem, file_id);
1740 self.map.lock().unwrap().insert(key, anchor);
1741 Box::pin(async { Ok(()) })
1742 }
1743
1744 fn delete(
1745 &self,
1746 subsystem: AnchorSubsystem,
1747 file_id: &[u8],
1748 ) -> std::pin::Pin<
1749 Box<
1750 dyn std::future::Future<Output = Result<(), zeph_common::anchor::AnchorError>>
1751 + Send
1752 + '_,
1753 >,
1754 > {
1755 let key = zeph_common::anchor::anchor_key(subsystem, file_id);
1756 self.map.lock().unwrap().remove(&key);
1757 Box::pin(async { Ok(()) })
1758 }
1759 }
1760
1761 #[tokio::test]
1765 #[serial_test::serial(subagent_transcript_integrity)]
1766 async fn pre_anchor_chained_file_still_opens_with_anchor_store_online() {
1767 let _guard = IntegrityConfigGuard::new();
1768 configure_history_integrity(Some(test_ring(0, 20)));
1769 let dir = tempfile::tempdir().unwrap();
1770 let path = dir.path().join("abc.jsonl");
1771
1772 let writer = TranscriptWriter::new(&path).unwrap();
1774 writer
1775 .append(0, &test_message(Role::User, "pre-anchor"))
1776 .await
1777 .unwrap();
1778 drop(writer);
1779
1780 configure_anchor_store(Some(Arc::new(MockAnchorStore::default())));
1782 let messages = TranscriptReader::load(&path).unwrap();
1783 assert_eq!(
1784 messages.len(),
1785 1,
1786 "absent anchor must never brick a legacy-chained file"
1787 );
1788 }
1789
1790 #[tokio::test]
1793 #[serial_test::serial(subagent_transcript_integrity)]
1794 async fn whole_strip_of_anchored_transcript_is_tamper() {
1795 let _guard = IntegrityConfigGuard::new();
1796 configure_history_integrity(Some(test_ring(0, 21)));
1797 let store: Arc<dyn AnchorStore> = Arc::new(MockAnchorStore::default());
1798 configure_anchor_store(Some(Arc::clone(&store)));
1799
1800 let dir = tempfile::tempdir().unwrap();
1801 let path = dir.path().join("abc.jsonl");
1802 let writer = TranscriptWriter::new(&path).unwrap();
1803 writer
1804 .append(0, &test_message(Role::User, "one"))
1805 .await
1806 .unwrap();
1807 writer
1808 .append(1, &test_message(Role::Assistant, "two"))
1809 .await
1810 .unwrap();
1811 writer.finalize().await.unwrap();
1812
1813 let messages = TranscriptReader::load(&path).unwrap();
1815 assert_eq!(messages.len(), 2);
1816
1817 let raw = std::fs::read_to_string(&path).unwrap();
1820 let stripped: String = raw
1821 .lines()
1822 .map(|line| {
1823 let mut value: serde_json::Value = serde_json::from_str(line).unwrap();
1824 value.as_object_mut().unwrap().remove("chain");
1825 value.to_string()
1826 })
1827 .collect::<Vec<_>>()
1828 .join("\n")
1829 + "\n";
1830 std::fs::write(&path, stripped).unwrap();
1831
1832 let err = TranscriptReader::load(&path).unwrap_err();
1833 assert_matches!(err, SubAgentError::Integrity(ref m) if m.contains("TAMPER") && m.contains("vault anchor"));
1834 }
1835
1836 #[tokio::test]
1837 #[serial_test::serial(subagent_transcript_integrity)]
1838 async fn truncation_below_anchored_count_is_tamper() {
1839 let _guard = IntegrityConfigGuard::new();
1840 configure_history_integrity(Some(test_ring(0, 22)));
1841 let store: Arc<dyn AnchorStore> = Arc::new(MockAnchorStore::default());
1842 configure_anchor_store(Some(Arc::clone(&store)));
1843
1844 let dir = tempfile::tempdir().unwrap();
1845 let path = dir.path().join("abc.jsonl");
1846 let writer = TranscriptWriter::new(&path).unwrap();
1847 writer
1848 .append(0, &test_message(Role::User, "one"))
1849 .await
1850 .unwrap();
1851 writer
1852 .append(1, &test_message(Role::Assistant, "two"))
1853 .await
1854 .unwrap();
1855 writer.finalize().await.unwrap();
1856
1857 let raw = std::fs::read_to_string(&path).unwrap();
1860 let first_line = raw.lines().next().unwrap();
1861 std::fs::write(&path, format!("{first_line}\n")).unwrap();
1862
1863 let err = TranscriptReader::load(&path).unwrap_err();
1864 assert_matches!(err, SubAgentError::Integrity(ref m) if m.contains("TAMPER") && m.contains("truncated"));
1865 }
1866
1867 #[tokio::test]
1868 #[serial_test::serial(subagent_transcript_integrity)]
1869 async fn finalize_is_noop_without_anchor_store_or_without_chaining() {
1870 let _guard = IntegrityConfigGuard::new();
1871 configure_history_integrity(Some(test_ring(0, 23)));
1873 let dir = tempfile::tempdir().unwrap();
1874 let path = dir.path().join("abc.jsonl");
1875 let writer = TranscriptWriter::new(&path).unwrap();
1876 writer
1877 .append(0, &test_message(Role::User, "x"))
1878 .await
1879 .unwrap();
1880 writer.finalize().await.unwrap();
1881 configure_history_integrity(None);
1882
1883 let store: Arc<dyn AnchorStore> = Arc::new(MockAnchorStore::default());
1886 configure_anchor_store(Some(Arc::clone(&store)));
1887 let path2 = dir.path().join("legacy.jsonl");
1888 let writer2 = TranscriptWriter::new(&path2).unwrap();
1889 writer2
1890 .append(0, &test_message(Role::User, "legacy"))
1891 .await
1892 .unwrap();
1893 writer2.finalize().await.unwrap();
1894 assert!(
1895 store
1896 .get_sync(AnchorSubsystem::SubagentTranscript, b"legacy")
1897 .unwrap()
1898 .is_none(),
1899 "no anchor should be written for an unchained writer"
1900 );
1901 }
1902}