Skip to main content

zeph_subagent/
transcript.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! JSONL-based transcript persistence for sub-agent conversations.
5//!
6//! Each sub-agent session writes a `<task_id>.jsonl` file of [`TranscriptEntry`] lines
7//! and a companion `<task_id>.meta.json` sidecar with [`TranscriptMeta`].
8//!
9//! Files are created with `0o600` permissions on Unix to prevent other users from
10//! reading conversation history.
11//!
12//! The [`sweep_old_transcripts`] function prunes the oldest `.jsonl` files when a
13//! configurable maximum count is exceeded.
14
15use 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
31/// Domain-separation tag for this subsystem's hash chain (issue #6360) — distinct from
32/// `zeph-session`'s so a chain from one subsystem can never verify against the other, and folded
33/// into every genesis hash via [`zeph_common::hash_chain::genesis`].
34pub const CHAIN_DOMAIN: &str = "zeph-subagent transcript v1";
35
36/// Process-wide history-chain key ring, configured once at bootstrap by resolving
37/// `ZEPH_HISTORY_KEY` from the vault (see `zeph_core::history_integrity`).
38///
39/// # Why a process-global registry, not a constructor parameter
40///
41/// `TranscriptWriter::new`/[`TranscriptReader::load`] are called from 3+ call sites across
42/// crates outside this feature's ownership (`zeph-core`'s scheduler loop and subagent-plan
43/// tests, in addition to `zeph-subagent::manager::collect`), and `PayloadCipher`-style explicit
44/// `Option<Arc<dyn _>>` injection into every one of those call sites was judged too invasive for
45/// this change (would require touching crates outside this PR's scope during a period other
46/// teammates are also editing them). A `RwLock` (not `OnceLock`) is used deliberately so tests
47/// in this crate and its callers can reconfigure it per-test rather than being limited to a
48/// single process-lifetime value — see `configure_history_integrity`'s doc for the tradeoff this
49/// accepts. Flagged in the implementation handoff for critic/reviewer scrutiny as a deviation
50/// from the codebase's usual per-call dependency injection pattern.
51static HISTORY_INTEGRITY: StdRwLock<Option<Arc<ChainKeyRing>>> = StdRwLock::new(None);
52
53/// Configure (or disable, with `None`) history-chain verification for every
54/// [`TranscriptWriter`]/[`TranscriptReader`] operation in this process from this point forward.
55///
56/// Call once at process bootstrap after resolving `ZEPH_HISTORY_KEY` from the vault (see
57/// `zeph_core::history_integrity::resolve_key_ring`). Passing `None` — the default until this is
58/// called — disables chain computation/verification entirely: writers append unchained entries
59/// (as before this feature existed) and readers treat every file as legacy. This is the
60/// generate-on-first-use / vault-unavailable fallback posture (spec-069 M2): a transient vault
61/// outage degrades to unchained rather than blocking every transcript write.
62///
63/// # Invariant: single-set-at-startup
64///
65/// This is `pub` (not `pub(crate)`) specifically so `src/runner.rs` — a different crate from
66/// this one — can call it once during CLI bootstrap, before any transcript is written or read
67/// (see `configure_history_integrity_from_default_vault` in `src/runner.rs`). It is **not**
68/// meant to be called again later by production code: reconfiguring mid-process cannot make an
69/// already-constructed `TranscriptWriter` less safe (each writer captures `ring` at construction
70/// and is immune to later reconfiguration, and setting `ring = None` only ever makes
71/// *subsequent* reads fail-closed on a chained file, never trust-bypassing), but a caller
72/// reconfiguring after bootstrap without a clear reason is almost certainly a bug, not an
73/// intended feature — no production code path does this today, and none should be added without
74/// updating this doc. Tests are the one legitimate exception, calling this per-test under
75/// `cargo nextest`'s one-process-per-test isolation.
76pub 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
86/// Process-wide vault-anchor store (issue #6449), configured once at bootstrap alongside
87/// [`configure_history_integrity`]. `None` (the default) disables anchor writes/checks entirely —
88/// transcripts behave exactly as they did under #6453 (chain-verified, but not
89/// downgrade-resistant against a whole-file strip).
90static ANCHOR_STORE: StdRwLock<Option<Arc<dyn AnchorStore>>> = StdRwLock::new(None);
91
92/// Configure (or disable, with `None`) the vault-anchor store for every [`TranscriptWriter`]/
93/// [`TranscriptReader`] operation in this process from this point forward. See
94/// [`configure_history_integrity`]'s doc for the single-set-at-startup contract this mirrors.
95pub 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
105/// Derive a transcript file's chain identity from its path (the `task_id`, e.g. `"abc123"` from
106/// `"abc123.jsonl"`) — binds the chain to this one file so a whole-file substitution (swapping
107/// in another task's transcript) breaks at the genesis hash.
108fn 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
115/// Paths already warned about via [`warn_legacy_under_active_key_once`] this process — kept
116/// small (one entry per distinct transcript path actually read while chaining-disabled, not
117/// per-read) so a session's history isn't re-warned every time it's reloaded.
118static WARNED_LEGACY_UNDER_KEY: std::sync::LazyLock<StdRwLock<std::collections::HashSet<PathBuf>>> =
119    std::sync::LazyLock::new(|| StdRwLock::new(std::collections::HashSet::new()));
120
121/// Log a structured `WARN` the first time a given path is found to be pure-legacy (no `chain`
122/// field anywhere) while a history-integrity key ring IS configured (issue #6360, security
123/// review B2 condition (c)).
124///
125/// Deliberately `WARN`, not a hard failure: a chainless file under an active key is *anomalous*
126/// but not distinguishable from genuine pre-upgrade content without the vault anchor (#6449) —
127/// this exists purely to make that anomaly observable instead of silent. Deduplicated per path
128/// (not per read) to avoid alert fatigue on the many genuinely-legacy files that exist right
129/// after upgrading to this feature.
130fn 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; // another thread warned first between the read and write locks
141    }
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/// A single entry in a JSONL transcript file.
153///
154/// Each line in `<task_id>.jsonl` deserializes to a `TranscriptEntry`.
155/// Entries are written in append order; `seq` is a monotonically increasing counter
156/// within a single session.
157#[derive(Debug, Clone, Serialize, Deserialize)]
158pub struct TranscriptEntry {
159    /// Zero-based sequence number within the session.
160    pub seq: u32,
161    /// ISO 8601 UTC timestamp at the time of writing (e.g. `"2026-04-09T12:00:00Z"`).
162    pub timestamp: String,
163    /// The LLM message that was appended at this sequence position.
164    pub message: Message,
165    /// Keyed-BLAKE3 hash chain link (hex-encoded), binding this entry's content and the
166    /// previous entry's hash (issue #6360). `None` on every entry means this transcript
167    /// predates the feature or history-chain verification is disabled for this process
168    /// (legacy, auto-trusted-once per spec-069 FR-006). Additive field: `#[serde(default)]`
169    /// means an older reader/writer that doesn't know this field ignores it, and legacy files
170    /// without it parse unchanged.
171    #[serde(default, skip_serializing_if = "Option::is_none")]
172    pub chain: Option<String>,
173}
174
175/// Sidecar metadata for a transcript, written as `<agent_id>.meta.json`.
176///
177/// The sidecar is written twice: once at spawn time with `status: Submitted` and
178/// again at collection time with the final terminal state and `finished_at`.
179#[derive(Debug, Clone, Serialize, Deserialize)]
180pub struct TranscriptMeta {
181    /// UUID of this sub-agent session.
182    pub agent_id: String,
183    /// Runtime agent name (same as `def_name` for non-resumed sessions).
184    pub agent_name: String,
185    /// Name of the [`SubAgentDef`][crate::SubAgentDef] that was used.
186    pub def_name: String,
187    /// Terminal lifecycle state recorded at collection time.
188    pub status: SubAgentState,
189    /// ISO 8601 UTC timestamp when the session was spawned.
190    pub started_at: String,
191    /// ISO 8601 UTC timestamp when the session finished, if known.
192    #[serde(skip_serializing_if = "Option::is_none")]
193    pub finished_at: Option<String>,
194    /// ID of the original agent session this was resumed from.
195    #[serde(skip_serializing_if = "Option::is_none")]
196    pub resumed_from: Option<String>,
197    /// Number of LLM turns consumed by the session.
198    pub turns_used: u32,
199    /// MCP tool names available when this session was spawned.
200    ///
201    /// Persisted so that a resumed session can restore the same tool name annotations
202    /// in its system prompt without re-connecting MCP servers.
203    #[serde(default)]
204    pub mcp_tool_names: Vec<String>,
205}
206
207/// Appends [`TranscriptEntry`] lines to a JSONL transcript file.
208///
209/// The file handle is kept open for the writer's lifetime to avoid
210/// race conditions from repeated open/close cycles. The handle is wrapped in
211/// `Arc<Mutex<File>>` so the writer can be cheaply cloned and passed to
212/// `tokio::task::spawn_blocking` for non-blocking appends.
213///
214/// # Examples
215///
216/// ```rust,no_run
217/// use std::path::Path;
218/// use zeph_subagent::transcript::TranscriptWriter;
219///
220/// let writer = TranscriptWriter::new(Path::new("/tmp/session.jsonl")).unwrap();
221/// // writer.append(seq, &message) to persist each message.
222/// ```
223struct TranscriptWriteState {
224    file: File,
225    /// Running chain head. `None` until either the first chained append in this writer's
226    /// lifetime (fresh chaining start on a legacy or empty file) or seeded from the file's
227    /// existing chained tail at open time (M3, see [`TranscriptWriter::new`]).
228    prev: Option<ChainHash>,
229    /// Total on-disk entry count (seeded from any pre-existing content at open time,
230    /// incremented on every successful append) — the `count` half of the vault anchor written
231    /// by [`TranscriptWriter::finalize`] (issue #6449).
232    count: u64,
233}
234
235#[derive(Clone)]
236pub struct TranscriptWriter {
237    /// `file` and `prev` share one lock so the chain-link read-modify-write is always atomic
238    /// with the physical write (S2, issue #6360 critic rev2): two concurrent `append` calls via
239    /// `spawn_blocking` can never compute their chain link in one order but land their physical
240    /// writes in another, which would desynchronize on-disk order from chain order and produce
241    /// a false tamper verdict on read.
242    state: Arc<Mutex<TranscriptWriteState>>,
243    file_identity: Vec<u8>,
244    /// Captured once at construction so every `append` on this writer instance uses one
245    /// consistent key ring, even if `configure_history_integrity` is called again concurrently
246    /// (which only affects writers/readers constructed afterward).
247    ring: Option<Arc<ChainKeyRing>>,
248}
249
250impl TranscriptWriter {
251    /// Create (or open) a JSONL transcript file in append mode.
252    ///
253    /// Creates parent directories if they do not already exist. If the file already has content
254    /// and history-chain verification is configured (see [`configure_history_integrity`]), the
255    /// existing content is scanned and its chain verified before the writer is returned (M3
256    /// open-time tail verify/seed) — a writer can never open atop content it hasn't itself
257    /// verified, and the running chain state (`prev`) is seeded from the verified tail so the
258    /// very next append continues the existing chain rather than restarting it.
259    ///
260    /// # Errors
261    ///
262    /// Returns `io::Error` if the directory cannot be created, the file cannot be opened, or
263    /// (per NFR-004) the existing content fails chain verification — a broken chain must never
264    /// be silently opened past.
265    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    /// Append a single message as a JSON line and flush immediately.
299    ///
300    /// `MessagePart::Image` parts are stripped (via [`MessagePart::strip_images`]) from the
301    /// persisted copy before serialization — they are ephemeral, current-turn-only vision input
302    /// (spec-072 §4, C1) and must never reach a transcript file on disk, mirroring the strip point
303    /// already enforced for `Agent::persist_message`'s `SQLite`/Qdrant/durable-JSONL writers. The
304    /// caller's `message` is untouched, so callers that hold onto it for the current turn's
305    /// provider request keep their `Image` parts.
306    ///
307    /// When history-chain verification is configured, the chain-link read-modify-write,
308    /// canonicalization (serialize with `chain: None`, hash, then serialize again with the
309    /// computed hash), physical write, and flush all happen inside the same
310    /// `tokio::task::spawn_blocking` critical section, under the single lock guarding both the
311    /// file handle and the running chain state (S2) — so on-disk order always matches chain
312    /// order even under concurrent `append` calls from a cloned writer.
313    ///
314    /// # Errors
315    ///
316    /// Returns `io::Error` on serialization, write failure, lock poison, or thread-pool panic.
317    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            // Only advance the running chain state after the write+flush succeeded — a failed
363            // write must not desynchronize `prev` from what is actually durable on disk.
364            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    /// Finalize this writer: if a vault-anchor store is configured (issue #6449) and this
375    /// writer's lifetime saw at least one chained append, persist an [`Anchor`] recording the
376    /// final `(epoch, count, head)` — written **last**, after every append is durably flushed,
377    /// so a crash before this point leaves the file present with no anchor, which is always
378    /// benign (never a false tamper signature — see the module-level anchor docs).
379    ///
380    /// A no-op, not an error, when no anchor store is configured or this writer never chained
381    /// (pure legacy for its whole lifetime): there is nothing to anchor.
382    ///
383    /// # Errors
384    ///
385    /// Returns `io::Error` if the configured anchor store's `put` fails (a store-level failure,
386    /// not an absent anchor). Callers should treat this as best-effort and log rather than fail
387    /// the whole collection flow — the transcript file itself is already safely written.
388    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    /// Write the meta sidecar file for an agent.
415    ///
416    /// # Errors
417    ///
418    /// Returns `io::Error` on serialization or write failure.
419    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    /// Async variant of [`write_meta`][Self::write_meta] that offloads the blocking FS write
427    /// to a `spawn_blocking` thread so the Tokio executor is not stalled.
428    ///
429    /// # Errors
430    ///
431    /// Returns `io::Error` on serialization, write failure, or thread-pool panic.
432    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
447/// Reads and reconstructs message history from JSONL transcript files.
448///
449/// `TranscriptReader` is a zero-size marker type with only associated functions.
450/// Use [`TranscriptReader::load`] to reconstruct a message history from a `.jsonl` file,
451/// [`TranscriptReader::load_meta`] to read the companion `.meta.json` sidecar, and
452/// [`TranscriptReader::find_by_prefix`] to resolve a short ID prefix to a full UUID.
453pub struct TranscriptReader;
454
455impl TranscriptReader {
456    /// Load all messages from a JSONL transcript file.
457    ///
458    /// Malformed lines are skipped with a warning. An empty or missing file
459    /// returns an empty `Vec`. If the file does not exist at all but a matching
460    /// `.meta.json` sidecar exists, returns `SubAgentError::Transcript` with a
461    /// clear message so the caller knows the data is gone rather than silently
462    /// degrading to a fresh start.
463    ///
464    /// # Errors
465    ///
466    /// Returns [`SubAgentError::Transcript`] on unrecoverable I/O failures, or
467    /// when the transcript file is missing but meta exists (data-loss guard).
468    pub fn load(path: &Path) -> Result<Vec<Message>, SubAgentError> {
469        Self::load_impl(path, false)
470    }
471
472    /// Load all messages from a JSONL transcript file, failing closed on the first skipped line.
473    ///
474    /// Unlike [`load`][Self::load], which tolerates an unreadable or malformed line by skipping
475    /// it with a warning and returning the surviving entries as `Ok`, `load_strict` returns
476    /// `SubAgentError::Transcript` the moment any line would be skipped. Callers that must be
477    /// able to distinguish a genuinely complete trace from a partial one — e.g. tool-call
478    /// grounding, where a silently dropped `ToolUse` entry would misrepresent a partial read as
479    /// an authoritative "no tool ran" trace — should use this instead of [`load`][Self::load].
480    ///
481    /// # Errors
482    ///
483    /// Returns [`SubAgentError::Transcript`] if any line is unreadable or fails to parse, or if
484    /// the file is missing but a meta sidecar exists (data-loss guard, same as
485    /// [`load`][Self::load]).
486    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            // Check if a meta sidecar exists — if so, data has been lost.
493            // Build meta path from the file stem (e.g. "abc" from "abc.jsonl")
494            // so it is consistent with write_meta which uses format!("{agent_id}.meta.json").
495            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    /// Load the meta sidecar for an agent.
525    ///
526    /// # Errors
527    ///
528    /// Returns [`SubAgentError::NotFound`] if the file does not exist,
529    /// [`SubAgentError::Transcript`] on parse failure.
530    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    /// Find the full agent ID by scanning `dir` for `.meta.json` files whose names
545    /// start with `prefix`.
546    ///
547    /// # Errors
548    ///
549    /// Returns [`SubAgentError::NotFound`] if no match is found,
550    /// [`SubAgentError::AmbiguousId`] if multiple matches are found,
551    /// [`SubAgentError::Transcript`] on I/O failure.
552    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
581/// Open and parse every line of an existing transcript file into [`TranscriptEntry`] values,
582/// applying the same read/parse leniency [`TranscriptReader::load`]/[`TranscriptReader::load_strict`]
583/// use (`strict` fails on the first unreadable/malformed line; lenient warns and skips it).
584///
585/// Assumes `path` exists — callers needing the "missing file" / "meta sidecar exists"
586/// disambiguation must check that first (see [`TranscriptReader::load_impl`]).
587///
588/// Note this is purely JSON-syntax leniency, unrelated to chain verification: chain breaks
589/// always escalate to a hard error in both modes (Q3, see [`verify_and_extract_messages`]).
590///
591/// # Errors
592///
593/// Returns [`SubAgentError::Transcript`] if the file cannot be opened, or if `strict` and any
594/// line is unreadable or fails to parse as JSON.
595fn 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/// Walk a transcript's parsed entries, verifying the hash chain over the chained region
651/// (spec-069 FR-001/FR-002) and returning the trusted messages plus the verified head hash (used
652/// by [`TranscriptWriter::new`]'s open-time seeding, M3).
653///
654/// The **legacy prefix** — entries before the first one carrying a `chain` field — is
655/// auto-trusted-once (FR-006 Q2): best-effort, unverified, exactly as this transcript format
656/// behaved before this feature existed. A file with no `chain` field anywhere is pure legacy;
657/// its messages are returned with `head = None` and no key is required.
658///
659/// Once a `chain` field appears, **every subsequent entry MUST also carry one**: a missing field
660/// after the chained region starts is a partial strip, not a legacy tail, and is a hard tamper
661/// failure (critic C1) — this check runs regardless of the caller's lenient/strict JSON-parsing
662/// mode, because a chain break invalidates trust in everything downstream of it, unlike a single
663/// malformed line (Q3).
664///
665/// # Errors
666///
667/// Returns [`SubAgentError::Integrity`] when: the file carries chain metadata but no
668/// history-integrity key ring is configured (`ring.is_none()`, NFR-004 — never silently treated
669/// as legacy); a partial strip is detected; [`verify_chained_prefix`] reports a definite tamper
670/// ([`ChainError::Mismatch`]) or an unverifiable/possibly-re-keyed chain
671/// ([`ChainError::Unverifiable`]); or `anchor` disagrees with the on-disk content (issue #6449 —
672/// see the read-side decision table in the module-level anchor docs, `zeph_common::anchor`).
673#[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        // Legacy-looking file (no chain field anywhere) + a vault anchor exists for this file's
682        // identity: this IS a tamper signature, unlike the "absent anchor" case below. An anchor
683        // can only exist if this file was previously finalized while chained — a file-write-only
684        // attacker cannot delete a vault entry, so a legacy-looking file with a live anchor means
685        // every `chain` field was deliberately stripped (the whole-strip downgrade attack #6449
686        // closes).
687        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        // Pure legacy file: no chain metadata anywhere, and no anchor either. Auto-trusted per
706        // FR-006 — but if a key ring IS configured, every legitimately-written file since this
707        // process started should carry a chain field, so a chainless file under an active key is
708        // anomalous: either genuine pre-upgrade content, or (absent an anchor to prove otherwise)
709        // indistinguishable from one. Not a hard failure — but it must be observable, not silent
710        // (security review B2 condition, NFR-005).
711        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    // The anchor's `count` is a total on-disk count; the chained region starts at `chain_start`,
756    // so the checkpoint index within `chained` (already sliced from `chain_start`) is
757    // `count - chain_start - 1` (0-based, the position of the anchor's last entry).
758    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
829/// Render a [`ChainError`] as a [`SubAgentError::Integrity`] with operator-actionable wording
830/// that distinguishes a definite tamper verdict from an ambiguous/possibly-re-keyed one (FR-008
831/// — an operator must not be misled into believing a re-keyed transcript was tampered with).
832fn 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
853/// Delete the oldest `.jsonl` files in `dir` when the count exceeds `max_files`, plus each
854/// deleted file's companion `.meta.json` sidecar.
855///
856/// Files are sorted by modification time (oldest first). Returns the number of
857/// files deleted.
858///
859/// # Vault anchors (issue #6449)
860///
861/// This function stays deliberately synchronous (it is called from 2+ sync/`spawn_blocking`
862/// contexts outside this feature's ownership — see `crates/zeph-subagent/src/manager/collect.rs`
863/// — and making it async would force those callers async too, an out-of-scope blast radius).
864/// It therefore does **not** delete a swept file's vault anchor inline. This is safe, not merely
865/// deferred-and-hoped: an anchor whose file no longer exists is an **orphan**, and an orphan
866/// anchor is always benign on read (an anchor is only ever consulted when opening a file that
867/// exists — see the module-level anchor docs, `zeph_common::anchor`) — it never produces a false
868/// TAMPER verdict for anything. Orphans left behind by this sweep are reaped later by the
869/// process-wide reconcile-and-cap sweep (`zeph-core`'s `anchor_store` module), which lists every
870/// `ZEPH_HISTORY_ANCHOR_*` vault key and drops any whose file no longer exists on disk, bounding
871/// vault growth exactly as it already does for the session-anchor LRU cap.
872///
873/// # Errors
874///
875/// Returns `io::Error` if the directory cannot be read or a file cannot be deleted.
876pub fn sweep_old_transcripts(dir: &Path, max_files: usize) -> io::Result<usize> {
877    if max_files == 0 {
878        return Ok(0);
879    }
880
881    // Create the directory if it does not exist yet (first run).
882    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    // Sort oldest first.
905    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        // Also remove the companion .meta.json sidecar if present.
911        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/// Returns the current UTC time as an ISO 8601 string (`"YYYY-MM-DDTHH:MM:SSZ"`).
922#[must_use]
923pub(crate) fn utc_now() -> String {
924    // Use SystemTime for a zero-dependency ISO 8601 timestamp.
925    // Format: 2026-03-05T00:18:16Z
926    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
934/// Convert Unix epoch seconds to (year, month, day, hour, minute, second).
935///
936/// Uses the proleptic Gregorian calendar algorithm (Fliegel-Van Flandern variant).
937/// All values are u64 throughout to avoid truncating casts; the caller knows values
938/// fit in u32 for the ranges used (years 1970–2554, seconds/minutes/hours/days).
939fn 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    // Days since 1970-01-01 → civil calendar (Gregorian).
948    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    // All values are in range for u32 for any timestamp in [1970, 2554].
960    #[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/// RAII guard that resets [`HISTORY_INTEGRITY`] and [`ANCHOR_STORE`] to `None` on drop
972/// (issue #6686). See `zeph_session::log::IntegrityConfigGuard`'s identical doc for the full
973/// rationale — this mirrors it exactly. Every test in this crate that configures either
974/// static, or that constructs a [`TranscriptWriter`]/[`TranscriptReader`] while one could be
975/// configured (e.g. `manager::tests::run_agent_loop_finalizes_transcript_anchor_on_llm_error_exit_path`),
976/// must both construct this guard and carry
977/// `#[serial_test::serial(subagent_transcript_integrity)]`.
978#[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    /// #6305: `MessagePart::Image` must never reach the on-disk transcript — it is ephemeral,
1047    /// current-turn-only vision input (spec-072 §4, C1), mirroring the strip already enforced
1048    /// for `Agent::persist_message`'s `SQLite`/Qdrant/durable-JSONL writers.
1049    #[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        // The caller's own copy keeps the Image part for the current turn's provider request.
1070        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        // The image payload must not appear anywhere in the file on disk either.
1085        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        // Mirrors the `task_msg` / turn-generated-message call sites in `agent_loop.rs`, which
1127        // always pass an empty `parts` vec — the strip must be a no-op for them.
1128        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        // A torn/malformed line sits between two well-formed entries — simulates a sub-agent
1196        // canceled/killed mid-write.
1197        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        // The lenient reader still tolerates the same file, proving the two variants diverge
1204        // only in this failure mode.
1205        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            // Vary mtime by touching the file — not reliable without explicit mtime set,
1308            // but tempdir files get sequential syscall timestamps in practice.
1309            // We set the mtime explicitly via filetime crate... but we have no filetime dep.
1310            // Instead we just verify count is correct.
1311        }
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        // Basic format check: 2026-03-05T00:18:16Z
1347        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        // Create 4 JSONL files each with a companion meta sidecar.
1376        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        // Companion metas for the two deleted files should also be gone.
1384        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        // path.with_extension("meta.json") on "abc.jsonl" should yield "abc.meta.json"
1399        // which matches write_meta's format!("{agent_id}.meta.json") when agent_id == stem.
1400        let dir = tempfile::tempdir().unwrap();
1401        let agent_id = "deadbeef-0000-0000-0000-000000000000";
1402        // Write meta sidecar but not the JSONL file.
1403        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    // --- Hash-chain integrity tests (issue #6360) ---
1422    //
1423    // `configure_history_integrity` mutates process-global state, so these tests rely on
1424    // `cargo nextest`'s one-process-per-test model for isolation (never run this module with
1425    // plain `cargo test`, which shares one process across tests in a binary and could race).
1426
1427    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        // A first, untouched entry so the key epoch resolves cleanly there; tampering the
1475        // *second* entry below then produces a definite Mismatch (not an ambiguous
1476        // Unverifiable, which is what tampering the very first chained entry would produce).
1477        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        // load_strict must fail identically — chain breaks always escalate (Q3), even in modes
1495        // that otherwise differ only on JSON-syntax leniency.
1496        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        // Written with integrity disabled (the pre-feature/legacy shape).
1505        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        // Now integrity comes online for this process (e.g. vault became available).
1522        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        // A legacy file read while a key IS configured must be flagged exactly once per path
1531        // (security review B2 condition (c)) — repeat reads must not re-warn.
1532        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        // Strip the chain field from only the second line, simulating an attacker who deletes
1565        // one line's chain metadata rather than the whole file's (the C1 partial-strip attack).
1566        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        // Simulate the vault becoming unavailable: no key ring configured at read time.
1593        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        // Rotate: new current epoch 1, old epoch 0 retained as the previous window.
1614        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        // Reopen a fresh writer on the same file (M3 open-time tail seed) and append more.
1646        {
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        // The full file, spanning both writer instances, must verify as one continuous chain.
1655        let messages = TranscriptReader::load(&path).unwrap();
1656        assert_eq!(messages.len(), 2);
1657    }
1658
1659    /// S2 regression: concurrent `append` calls via a cloned writer must never desynchronize
1660    /// on-disk physical order from chain-link order. Mirrors
1661    /// `zeph_session::log::tests::test_concurrent_append_preserves_seq_order`.
1662    #[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        // If chain order had diverged from physical write order, this would fail with a
1686        // definite Mismatch tamper verdict even though nothing was actually tampered with.
1687        let messages = TranscriptReader::load(&path).unwrap();
1688        assert_eq!(messages.len(), usize::try_from(N).unwrap());
1689    }
1690
1691    // --- Vault-anchor downgrade-resistance tests (issue #6449) ---
1692
1693    /// In-memory [`AnchorStore`] mock for tests — a simple `Mutex<HashMap>` keyed by
1694    /// [`zeph_common::anchor::anchor_key`], mirroring `zeph_vault::MockVaultProvider`'s role for
1695    /// the history-key tests above.
1696    #[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    /// Regression test for FINDING B / acceptance criterion 2: a pre-anchor chained file (no
1762    /// anchor store configured when it was written) must still open normally when an anchor
1763    /// store comes online later — an absent anchor is never a tamper signature.
1764    #[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        // Written with no anchor store configured (the #6453-only posture).
1773        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        // Now an anchor store comes online, but this file was never anchored.
1781        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    /// Acceptance criterion 1/3: whole-strip of an anchored transcript is TAMPER, and so is
1791    /// truncation below the anchored count.
1792    #[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        // Sanity: with the anchor present and content untouched, the file still opens.
1814        let messages = TranscriptReader::load(&path).unwrap();
1815        assert_eq!(messages.len(), 2);
1816
1817        // Whole-strip: rewrite every line with its `chain` field removed, so the file looks
1818        // pre-feature-legacy — the attack #6449 closes.
1819        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        // Truncate the file to just its first line — content still verifies as a valid (shorter)
1858        // chain, but disagrees with the anchor's recorded count.
1859        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        // No anchor store configured: finalize must succeed as a no-op.
1872        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        // Anchor store configured, but chaining disabled: finalize must still be a no-op (no
1884        // chain head to anchor).
1885        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}