Skip to main content

agent_core/core/
context_archive.rs

1//! Durable, disclosure-filtered context segments; never a summary or a session log.
2//!
3//! The trusted host supplies an absolute base directory, scope key and logical
4//! conversation id (never model/tool path arguments). Scope and logical id are
5//! hashed; project scope determines the directory name. Only user/assistant text
6//! and tool-call structure
7//! are retained. System/developer messages, reasoning, unknown blocks, binary,
8//! non-normal sensitivity and restricted/unknown disclosure are omitted.
9//! Arguments/results are WITHHELD unless a trusted redactor is installed. All
10//! retained strings pass the disclosure gate and a conservative credential
11//! screen. This screen is not a universal secret detector: host classification
12//! and the optional redactor remain the authority for unlabelled secrets.
13//!
14//! Each segment contains its hidden, screened note and source indices in ONE
15//! atomic file. Notes are not searched or returned by message fetch. Eligible
16//! text is otherwise exact: no truncation, whitespace rewriting or summarizing.
17//! `fetch_with_provenance` reports indices into the original `seal` slice.
18//!
19//! Fixed slots avoid unbounded directory enumeration/indexes: at most 128
20//! segments (including permanent tombstones), 256 MiB total, 16 MiB per file.
21//! Input is limited to 4096 messages, 16 MiB of eligible strings/structure,
22//! 65536 visited nodes and depth 32; excluded payload descendants are opaque.
23//! Notes are limited to 8 KiB. Fetch returns <=128 messages and <=16 MiB;
24//! search queries <=512 bytes, results <=32 with <=256-byte snippets. Every
25//! operation scans at most 128 files / 256 MiB; the cross-process lock never waits.
26//! Budget exhaustion and corruption fail closed, never silently truncate seal.
27//!
28//! Unix-only private, handle-relative no-symlink filesystem; other platforms
29//! fail Unsupported. Live records are immutable; forget atomically replaces a
30//! record with a content-free id/digest/scope tombstone. The digest covers ONLY
31//! the eligible projection and source indices, not the note or forbidden data.
32//! Identical projections in one logical conversation deduplicate permanently;
33//! a changed note does not create a new segment. Retrying a tombstoned projection
34//! returns PermissionDenied. A successful seal fsyncs data and directory before
35//! returning; callers MUST keep source history on any error (even if publication
36//! happened before a directory-fsync error). Advisory locks serialize cooperating
37//! processes, not malicious processes running as the same OS user.
38//!
39//! Within one process, operations on the same archive directory wait for each
40//! other (not in FIFO order) and are admitted within [`LOCK_WAIT`] or fail as
41//! busy; another process holding the lock still fails closed ("archive busy")
42//! without waiting for it. The lock is released explicitly, so a descriptor
43//! inherited by a concurrently forked child cannot keep it held. An operation
44//! can be abandoned before it is admitted ([`ArchiveStore::with_abort`]); once
45//! admitted it always completes.
46
47use crate::core::disclosure::{gate_for_model, may_persist, DisclosureClass, ModelVisibility};
48use crate::core::stream_types::SharedMessage;
49use serde::{Deserialize, Serialize};
50use serde_json::{json, Value};
51use sha2::{Digest, Sha256};
52use std::collections::HashMap;
53use std::io::{self, Read, Write};
54use std::path::Path;
55use std::sync::atomic::{AtomicBool, Ordering};
56use std::sync::{Arc, Condvar, Mutex, OnceLock, PoisonError, Weak};
57use std::time::{Duration, Instant};
58
59pub const MAX_SEGMENTS: usize = 128;
60pub const MAX_SEGMENT_BYTES: usize = 16 * 1024 * 1024;
61pub const MAX_STORE_BYTES: usize = 256 * 1024 * 1024;
62pub const MAX_INPUT_BYTES: usize = 16 * 1024 * 1024;
63pub const MAX_MESSAGES: usize = 4096;
64pub const MAX_NOTE_BYTES: usize = 8192;
65pub const MAX_FETCH_MESSAGES: usize = 128;
66pub const MAX_SEARCH_RESULTS: usize = 32;
67/// Longest an operation waits for earlier operations on the same archive in
68/// this process before failing closed as busy.
69pub const LOCK_WAIT: Duration = Duration::from_secs(30);
70/// How often a queued operation with an abort flag rechecks it.
71const ABORT_POLL: Duration = Duration::from_millis(20);
72const WITHHELD: &str = "[archive: withheld]";
73const BUSY: &str = "archive busy; retain source and retry";
74const BUSY_EXPORT: &str = "archive busy; retry export";
75
76/// Host-owned sanitizer, never supplied by model input. Must not expand output
77/// beyond the archive budgets. Tool arguments are passed as serialized JSON;
78/// if the returned text is not JSON the entire argument payload is withheld.
79pub type ArchiveRedactor = dyn Fn(&str) -> String + Send + Sync;
80
81#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
82pub struct ArchiveRef {
83    pub id: String,
84    /// Eligible messages, not the length of the source slice.
85    pub message_count: usize,
86    pub source_message_count: usize,
87}
88
89#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
90pub struct ArchiveDescriptor {
91    pub id: String,
92    pub message_count: usize,
93    pub source_message_count: usize,
94    /// A bounded excerpt of eligible source JSON, never the hidden note.
95    pub snippet: String,
96}
97
98#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
99#[serde(deny_unknown_fields)]
100pub struct ArchivedMessage {
101    /// Zero-based offset in the slice passed to seal, including omitted messages.
102    pub source_index: usize,
103    /// Original content-array indices; empty for string content.
104    pub block_indices: Vec<usize>,
105    pub message: SharedMessage,
106}
107
108/// Eligible, screened source evidence without filesystem state. `logical_id`
109/// is the legacy 64-hex hash of the raw host session ID; the digest deliberately
110/// excludes the hidden note and omitted content. Never log this payload.
111#[derive(Clone, PartialEq, Serialize, Deserialize)]
112#[serde(deny_unknown_fields)]
113pub struct ArchiveProjection {
114    pub logical_id: String,
115    pub source_message_count: usize,
116    pub messages: Vec<ArchivedMessage>,
117    pub note: String,
118    pub digest: String,
119}
120
121/// Pure projection: no path discovery, directory creation, or archive I/O.
122/// The trusted host supplies the raw logical session ID and optional sanitizer.
123pub fn project_archive(
124    logical_id: &str,
125    messages: &[SharedMessage],
126    note: &str,
127    redactor: Option<Arc<ArchiveRedactor>>,
128) -> io::Result<ArchiveProjection> {
129    validate_key(logical_id)?;
130    project_hashed(
131        &digest_parts(&[logical_id.as_bytes()]),
132        messages,
133        note,
134        redactor,
135    )
136}
137
138fn project_hashed(
139    logical_id: &str,
140    messages: &[SharedMessage],
141    note: &str,
142    redactor: Option<Arc<ArchiveRedactor>>,
143) -> io::Result<ArchiveProjection> {
144    if messages.len() > MAX_MESSAGES || note.len() > MAX_NOTE_BYTES {
145        return Err(invalid("archive message/note budget exceeded"));
146    }
147    let mut budget = InputBudget { bytes: 0, nodes: 0 };
148    for message in messages {
149        budget.message(message, redactor.is_some())?;
150    }
151    let projector = Projector { redactor };
152    let mut eligible = Vec::new();
153    for (source_index, message) in messages.iter().enumerate() {
154        if let Some((message, block_indices)) = projector.project(message)? {
155            eligible.push(ArchivedMessage {
156                source_index,
157                block_indices,
158                message: Arc::new(message),
159            });
160        }
161    }
162    let digest = projection_digest(logical_id, messages.len(), &eligible)?;
163    let note = projector.screen(note, DisclosureClass::ModelVisible)?;
164    if note.len() > MAX_NOTE_BYTES {
165        return Err(invalid("redacted note exceeds archive budget"));
166    }
167    Ok(ArchiveProjection {
168        logical_id: logical_id.to_owned(),
169        source_message_count: messages.len(),
170        messages: eligible,
171        note,
172        digest,
173    })
174}
175
176struct Projector {
177    redactor: Option<Arc<ArchiveRedactor>>,
178}
179
180/// No Debug implementation: do not accidentally log redactor state or content.
181pub struct ArchiveStore {
182    fs: PrivateDir,
183    scope: String,
184    logical: String,
185    redactor: Option<Arc<ArchiveRedactor>>,
186    abort: Option<Arc<AtomicBool>>,
187    wait: Duration,
188}
189
190/// Process-local mutual exclusion for one archive directory. The advisory
191/// file lock excludes other processes, but each operation opens its own
192/// descriptor, so without this two threads of one process would see each
193/// other as busy. Waiters are not served in FIFO order.
194struct Gate {
195    held: Mutex<bool>,
196    released: Condvar,
197    /// Operations that have waited at least once (tests synchronize on it).
198    #[cfg(test)]
199    waited: std::sync::atomic::AtomicUsize,
200}
201
202/// Field order is drop order: unlock the file, then admit the next operation.
203struct Locked {
204    file: Option<FlockGuard>,
205    _gate: GateGuard,
206}
207
208struct GateGuard(Arc<Gate>);
209
210/// Device and inode of an archive directory.
211type DirIdentity = (u64, u64);
212type Gates = Mutex<HashMap<DirIdentity, Weak<Gate>>>;
213
214impl Gate {
215    /// One gate per directory identity, however the directory was reached.
216    fn for_directory(identity: DirIdentity) -> Arc<Gate> {
217        static GATES: OnceLock<Gates> = OnceLock::new();
218        let mut gates = GATES
219            .get_or_init(Default::default)
220            .lock()
221            .unwrap_or_else(PoisonError::into_inner);
222        if let Some(gate) = gates.get(&identity).and_then(Weak::upgrade) {
223            return gate;
224        }
225        gates.retain(|_, gate| gate.strong_count() > 0);
226        let gate = Arc::new(Gate {
227            held: Mutex::new(false),
228            released: Condvar::new(),
229            #[cfg(test)]
230            waited: Default::default(),
231        });
232        gates.insert(identity, Arc::downgrade(&gate));
233        gate
234    }
235
236    fn enter(
237        gate: Arc<Gate>,
238        wait: Duration,
239        abort: Option<&AtomicBool>,
240        busy: &'static str,
241    ) -> io::Result<GateGuard> {
242        let deadline = Instant::now() + wait;
243        {
244            let mut held = gate.held.lock().unwrap_or_else(PoisonError::into_inner);
245            let mut waited = false;
246            loop {
247                if abort.is_some_and(|abort| abort.load(Ordering::Acquire)) {
248                    return Err(io::Error::new(
249                        io::ErrorKind::Interrupted,
250                        "archive operation canceled before it started",
251                    ));
252                }
253                // An uncontended first attempt is admitted at once; after
254                // waiting, an expired operation is never admitted, even when
255                // the gate has just become free.
256                let now = Instant::now();
257                if waited && now >= deadline {
258                    return Err(io::Error::new(io::ErrorKind::WouldBlock, busy));
259                }
260                if !*held {
261                    *held = true;
262                    break;
263                }
264                if now >= deadline {
265                    return Err(io::Error::new(io::ErrorKind::WouldBlock, busy));
266                }
267                let mut slice = deadline - now;
268                if abort.is_some() {
269                    slice = slice.min(ABORT_POLL);
270                }
271                #[cfg(test)]
272                if !waited {
273                    gate.waited.fetch_add(1, Ordering::AcqRel);
274                }
275                waited = true;
276                held = gate
277                    .released
278                    .wait_timeout(held, slice)
279                    .unwrap_or_else(PoisonError::into_inner)
280                    .0;
281            }
282        }
283        Ok(GateGuard(gate))
284    }
285}
286
287impl Drop for GateGuard {
288    fn drop(&mut self) {
289        *self.0.held.lock().unwrap_or_else(PoisonError::into_inner) = false;
290        // Every waiter rechecks; an aborted or expired one must not absorb the
291        // only wakeup.
292        self.0.released.notify_all();
293    }
294}
295
296/// Unlocks explicitly before closing. A flock belongs to the open file
297/// description, and a child forked by any thread holds a copy of it until it
298/// execs (O_CLOEXEC applies only at exec); closing our copy alone would leave
299/// the archive locked meanwhile.
300struct FlockGuard(std::fs::File);
301
302impl Drop for FlockGuard {
303    fn drop(&mut self) {
304        let _ = fs4::fs_std::FileExt::unlock(&self.0);
305    }
306}
307
308#[derive(Serialize, Deserialize)]
309#[serde(deny_unknown_fields)]
310struct Record {
311    version: u32,
312    scope: String,
313    logical: String,
314    id: String,
315    digest: String,
316    body: Body,
317}
318
319#[derive(Serialize, Deserialize)]
320#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
321enum Body {
322    Live {
323        source_message_count: usize,
324        messages: Vec<ArchivedMessage>,
325        note: String,
326    },
327    Tombstone,
328}
329
330impl Record {
331    fn reference(&self) -> io::Result<ArchiveRef> {
332        match &self.body {
333            Body::Live {
334                source_message_count,
335                messages,
336                ..
337            } => Ok(ArchiveRef {
338                id: self.id.clone(),
339                message_count: messages.len(),
340                source_message_count: *source_message_count,
341            }),
342            Body::Tombstone => Err(not_found()),
343        }
344    }
345}
346
347impl ArchiveStore {
348    pub fn new(base: &Path, scope_key: &str, logical_id: &str) -> io::Result<Self> {
349        for key in [scope_key, logical_id] {
350            validate_key(key)?;
351        }
352        let scope = digest_parts(&[scope_key.as_bytes()]);
353        let logical = digest_parts(&[logical_id.as_bytes()]);
354        let fs = PrivateDir::new(base, &scope)?;
355        Ok(Self {
356            fs,
357            scope,
358            logical,
359            redactor: None,
360            abort: None,
361            wait: LOCK_WAIT,
362        })
363    }
364
365    /// Opt in to screened tool evidence. Install the SAME trusted policy when
366    /// reopening a store; retrieval never restores material withheld at seal.
367    pub fn with_redactor(mut self, redactor: Arc<ArchiveRedactor>) -> Self {
368        self.redactor = Some(redactor);
369        self
370    }
371
372    /// Once `abort` is set, an operation that has not been admitted yet (for
373    /// example one waiting behind another operation in this process) fails
374    /// with `Interrupted` without reading or writing any record. Setup before
375    /// admission (opening the store, projecting a seal) is not undone. An
376    /// admitted operation always completes, so callers must still treat an
377    /// abandoned seal as possibly published.
378    pub fn with_abort(mut self, abort: Arc<AtomicBool>) -> Self {
379        self.abort = Some(abort);
380        self
381    }
382
383    /// Queue behind other operations on this archive in this process, then
384    /// take the cross-process lock without waiting.
385    fn exclusive(&self) -> io::Result<Locked> {
386        let gate = self.enter(BUSY)?;
387        Ok(Locked {
388            file: Some(self.fs.lock()?),
389            _gate: gate,
390        })
391    }
392
393    fn enter(&self, busy: &'static str) -> io::Result<GateGuard> {
394        let gate = Gate::for_directory(self.fs.identity()?);
395        Gate::enter(gate, self.wait, self.abort.as_deref(), busy)
396    }
397
398    pub fn seal(&self, messages: &[SharedMessage], note: &str) -> io::Result<ArchiveRef> {
399        self.seal_with_note(messages, note)
400            .map(|(reference, _)| reference)
401    }
402
403    /// [`seal`](Self::seal) that also returns the screened note stored with the
404    /// segment (the earlier note when this projection was already sealed), in
405    /// the same locked operation.
406    pub fn seal_with_note(
407        &self,
408        messages: &[SharedMessage],
409        note: &str,
410    ) -> io::Result<(ArchiveRef, String)> {
411        let projection = project_hashed(&self.logical, messages, note, self.redactor.clone())?;
412        let ArchiveProjection {
413            digest,
414            source_message_count,
415            messages: eligible,
416            note,
417            ..
418        } = projection;
419        let _lock = self.exclusive()?;
420        let (records, used) = self.scan()?;
421        if let Some((_, record)) = records.iter().find(|(_, r)| r.digest == digest) {
422            let Body::Live { note: stored, .. } = &record.body else {
423                return Err(io::Error::new(
424                    io::ErrorKind::PermissionDenied,
425                    "archive projection was forgotten",
426                ));
427            };
428            // A previous call may have failed AFTER rename but BEFORE fsync.
429            self.fs.sync()?;
430            return Ok((record.reference()?, stored.clone()));
431        }
432        let slot = (0..MAX_SEGMENTS)
433            .find(|slot| !records.iter().any(|(s, _)| s == slot))
434            .ok_or_else(|| invalid("archive segment budget exhausted (tombstones retained)"))?;
435        let record = Record {
436            version: 1,
437            scope: self.scope.clone(),
438            logical: self.logical.clone(),
439            id: uuid::Uuid::new_v4().simple().to_string(),
440            digest,
441            body: Body::Live {
442                source_message_count,
443                messages: eligible,
444                note: note.clone(),
445            },
446        };
447        let bytes = bounded_json(&record, MAX_SEGMENT_BYTES)?;
448        if used + bytes.len() > MAX_STORE_BYTES {
449            return Err(invalid("archive storage byte budget exhausted"));
450        }
451        self.fs.publish(slot, &bytes)?;
452        Ok((record.reference()?, note))
453    }
454
455    pub fn fetch(&self, id: &str, start: usize, limit: usize) -> io::Result<Vec<SharedMessage>> {
456        Ok(self
457            .fetch_with_provenance(id, start, limit)?
458            .into_iter()
459            .map(|m| m.message)
460            .collect())
461    }
462
463    /// `start` indexes ELIGIBLE messages; source_index retains original offsets.
464    /// Oversized limits are rejected rather than clamped; start beyond end is empty.
465    pub fn fetch_with_provenance(
466        &self,
467        id: &str,
468        start: usize,
469        limit: usize,
470    ) -> io::Result<Vec<ArchivedMessage>> {
471        validate_id(id)?;
472        if limit > MAX_FETCH_MESSAGES {
473            return Err(invalid("archive fetch limit exceeds 128"));
474        }
475        let _lock = self.exclusive()?;
476        let (records, _) = self.scan()?;
477        let record = find_record(records, id)?;
478        match record.body {
479            Body::Live { messages, .. } => {
480                Ok(messages.into_iter().skip(start).take(limit).collect())
481            }
482            Body::Tombstone => Err(not_found()),
483        }
484    }
485
486    /// Explicit host-only access to the hidden, screened note. Never include it
487    /// in search descriptors or inject it as system/developer instructions.
488    pub fn fetch_note(&self, id: &str) -> io::Result<String> {
489        validate_id(id)?;
490        let _lock = self.exclusive()?;
491        let (records, _) = self.scan()?;
492        match find_record(records, id)?.body {
493            Body::Live { note, .. } => Ok(note),
494            Body::Tombstone => Err(not_found()),
495        }
496    }
497
498    /// Case-insensitive literal substring search of eligible source JSON, in
499    /// publication-slot order. Empty query lists descriptors. Not an exhaustive
500    /// ranking service: output is bounded by limit, with no hidden-note matches.
501    pub fn search(&self, query: &str, limit: usize) -> io::Result<Vec<ArchiveDescriptor>> {
502        if query.len() > 512 || limit > MAX_SEARCH_RESULTS {
503            return Err(invalid("archive search query/result budget exceeded"));
504        }
505        if limit == 0 {
506            return Ok(Vec::new());
507        }
508        let query = query.to_lowercase();
509        let _lock = self.exclusive()?;
510        let (records, _) = self.scan()?;
511        let mut result = Vec::new();
512        for (_, record) in records {
513            if let Body::Live {
514                messages,
515                source_message_count,
516                ..
517            } = &record.body
518            {
519                let mut snippet = None;
520                for message in messages {
521                    let text = serde_json::to_string(&message.message).map_err(json_error)?;
522                    if query.is_empty() || text.to_lowercase().contains(&query) {
523                        snippet = Some(prefix(&text, 256).to_owned());
524                        break;
525                    }
526                }
527                if snippet.is_some() || (query.is_empty() && messages.is_empty()) {
528                    result.push(ArchiveDescriptor {
529                        id: record.id,
530                        message_count: messages.len(),
531                        source_message_count: *source_message_count,
532                        snippet: snippet.unwrap_or_default(),
533                    });
534                }
535            }
536            if result.len() == limit {
537                break;
538            }
539        }
540        Ok(result)
541    }
542
543    /// Durable, idempotent deletion. Replaces content in the SAME slot with a
544    /// marker; it never frees the slot/digest for reuse. Unknown ids are NotFound.
545    /// This is logical deletion, not secure erasure of filesystem snapshots.
546    pub fn forget(&self, id: &str) -> io::Result<()> {
547        validate_id(id)?;
548        let _lock = self.exclusive()?;
549        let (records, _) = self.scan()?;
550        let (slot, mut record) = records
551            .into_iter()
552            .find(|(_, r)| r.id == id)
553            .ok_or_else(not_found)?;
554        if matches!(record.body, Body::Tombstone) {
555            return self.fs.sync();
556        }
557        record.body = Body::Tombstone;
558        self.fs
559            .publish(slot, &bounded_json(&record, MAX_SEGMENT_BYTES)?)
560    }
561
562    /// Migration-only export of validated eligible evidence and permanent
563    /// tombstones. Includes hidden notes: this is HOST data, never a model tool.
564    /// No files, directories, permissions or archive contents are changed.
565    pub fn export_records(&self) -> io::Result<Vec<Value>> {
566        let gate = self.enter(BUSY_EXPORT)?;
567        let lock = Locked {
568            file: self.fs.lock_existing()?,
569            _gate: gate,
570        };
571        let (records, _) = self.scan_with(true)?;
572        if lock.file.is_none() && !records.is_empty() {
573            return Err(corrupt());
574        }
575        Ok(records
576            .into_iter()
577            .map(|(_, record)| {
578                let (source_message_count, messages, note, tombstone) = match record.body {
579                    Body::Live {
580                        source_message_count,
581                        messages,
582                        note,
583                    } => (source_message_count, messages, note, false),
584                    Body::Tombstone => (0, Vec::new(), String::new(), true),
585                };
586                json!({
587                    "id": record.id, "digest": record.digest, "logical_id": record.logical,
588                    "source_message_count": source_message_count, "messages": messages,
589                    "note": note, "tombstone": tombstone,
590                })
591            })
592            .collect())
593    }
594
595    fn scan(&self) -> io::Result<(Vec<(usize, Record)>, usize)> {
596        self.scan_with(false)
597    }
598
599    fn scan_with(&self, readonly: bool) -> io::Result<(Vec<(usize, Record)>, usize)> {
600        let mut records = Vec::new();
601        let mut used = 0;
602        for slot in 0..MAX_SEGMENTS {
603            let Some(bytes) = self.fs.read_with(slot, MAX_STORE_BYTES - used, readonly)? else {
604                continue;
605            };
606            used += bytes.len();
607            let record: Record = serde_json::from_slice(&bytes).map_err(|_| corrupt())?;
608            if record.version != 1
609                || record.scope != self.scope
610                || record.logical.len() != 64
611                || !record.logical.bytes().all(|b| b.is_ascii_hexdigit())
612                || validate_id(&record.id).is_err()
613                || record.digest.len() != 64
614                || !record.digest.bytes().all(|b| b.is_ascii_hexdigit())
615            {
616                return Err(corrupt());
617            }
618            if let Body::Live {
619                messages,
620                source_message_count,
621                note,
622            } = &record.body
623            {
624                if messages.len() > MAX_MESSAGES
625                    || *source_message_count > MAX_MESSAGES
626                    || note.len() > MAX_NOTE_BYTES
627                    || messages
628                        .iter()
629                        .any(|m| m.source_index >= *source_message_count)
630                    || messages
631                        .windows(2)
632                        .any(|m| m[0].source_index >= m[1].source_index)
633                    || projection_digest(&record.logical, *source_message_count, messages)?
634                        != record.digest
635                {
636                    return Err(corrupt());
637                }
638            }
639            if records
640                .iter()
641                .any(|(_, r): &(usize, Record)| r.id == record.id || r.digest == record.digest)
642            {
643                return Err(corrupt());
644            }
645            records.push((slot, record));
646        }
647        Ok((records, used))
648    }
649}
650
651impl Projector {
652    fn screen(&self, text: &str, class: DisclosureClass) -> io::Result<String> {
653        if class == DisclosureClass::ModelVisibleAfterRedaction && self.redactor.is_none() {
654            return Ok(WITHHELD.to_owned());
655        }
656        let redactor = |text: &str| {
657            let text = self
658                .redactor
659                .as_ref()
660                .map_or_else(|| text.to_owned(), |r| r(text));
661            if credential_like(&text) {
662                WITHHELD.to_owned()
663            } else {
664                text
665            }
666        };
667        // All text is screened, including the ordinary model-visible class.
668        let text = redactor(text);
669        if text.len() > MAX_SEGMENT_BYTES {
670            return Err(invalid("redactor output exceeds archive budget"));
671        }
672        match gate_for_model(class, &text, false, Some(&|s| s.to_owned())) {
673            ModelVisibility::Visible(text) => Ok(text),
674            ModelVisibility::Withheld(_) => Ok(WITHHELD.to_owned()),
675        }
676    }
677
678    fn project(&self, message: &Value) -> io::Result<Option<(Value, Vec<usize>)>> {
679        let Some((class, role, content)) = projection_content(message) else {
680            return Ok(None);
681        };
682        if let Some(text) = content.as_str() {
683            return Ok(Some((
684                json!({"role": role, "content": self.screen(text, class)?}),
685                Vec::new(),
686            )));
687        }
688        let Some(blocks) = content.as_array() else {
689            return Ok(None);
690        };
691        let mut output = Vec::new();
692        let mut indices = Vec::new();
693        for (index, block) in blocks.iter().enumerate() {
694            let Some(mut block_class) = eligible_class(block) else {
695                continue;
696            };
697            if class == DisclosureClass::ModelVisibleAfterRedaction {
698                block_class = class;
699            }
700            let kept = match block.get("type").and_then(Value::as_str) {
701                Some("text") => block
702                    .get("text")
703                    .and_then(Value::as_str)
704                    .map(|text| {
705                        self.screen(text, block_class)
706                            .map(|text| json!({"type":"text", "text":text}))
707                    })
708                    .transpose()?,
709                Some("tool_use") if role == "assistant" => {
710                    if let (Some(id), Some(name)) =
711                        (identifier(block, "id"), identifier(block, "name"))
712                    {
713                        let input = if name == "context_checkpoint" {
714                            // The working note has its own hidden field. Do not
715                            // expose it a second time via tool arguments.
716                            json!({"_archive_withheld":true})
717                        } else if self.redactor.is_some() {
718                            self.tool_input(block.get("input"), block_class)?
719                        } else {
720                            json!({"_archive_withheld":true})
721                        };
722                        Some(json!({"type":"tool_use", "id":id, "name":name, "input":input}))
723                    } else {
724                        None
725                    }
726                }
727                Some("tool_result") if role == "user" => {
728                    if let Some(id) = identifier(block, "tool_use_id") {
729                        let content = if self.redactor.is_some() {
730                            self.tool_result(&block["content"], block_class)?
731                        } else {
732                            Value::String(WITHHELD.to_owned())
733                        };
734                        let mut result =
735                            json!({"type":"tool_result", "tool_use_id":id, "content":content});
736                        if let Some(error) = block.get("is_error").and_then(Value::as_bool) {
737                            result["is_error"] = json!(error);
738                        }
739                        Some(result)
740                    } else {
741                        None
742                    }
743                }
744                _ => None,
745            };
746            if let Some(kept) = kept {
747                output.push(kept);
748                indices.push(index);
749            }
750        }
751        if output.is_empty() {
752            Ok(None)
753        } else {
754            Ok(Some((json!({"role":role,"content":output}), indices)))
755        }
756    }
757
758    fn tool_input(&self, input: Option<&Value>, class: DisclosureClass) -> io::Result<Value> {
759        if let Some(input) = input {
760            // The full preflight already enforced aggregate limits. Reuse its
761            // guarded admission predicate so recursion and withholding cannot
762            // drift. JSON escaping expands at most sixfold (node allowance
763            // covers punctuation/numbers); redactors may shrink that scratch
764            // string below the unchanged final output limit.
765            let mut budget = InputBudget { bytes: 0, nodes: 0 };
766            if let Some(bytes) = budget.input(input, 3)? {
767                budget.charge(bytes)?;
768                let limit = bytes
769                    .checked_mul(6)
770                    .ok_or_else(|| invalid("archive input byte budget exceeded"))?;
771                let encoded = bounded_json(input, limit)?;
772                let text =
773                    self.screen(std::str::from_utf8(&encoded).map_err(|_| corrupt())?, class)?;
774                return Ok(serde_json::from_str(&text)
775                    .unwrap_or_else(|_| json!({"_archive_withheld":true})));
776            }
777        }
778        Ok(json!({"_archive_withheld":true}))
779    }
780
781    fn tool_result(&self, content: &Value, class: DisclosureClass) -> io::Result<Value> {
782        if let Some(text) = content.as_str() {
783            return Ok(json!(self.screen(text, class)?));
784        }
785        let mut output = Vec::new();
786        if let Some(blocks) = content.as_array() {
787            for block in blocks {
788                if let Some(class) = eligible_class(block) {
789                    if block["type"] == "text" {
790                        if let Some(text) = block["text"].as_str() {
791                            output.push(json!({"type":"text", "text":self.screen(text, class)?}));
792                        }
793                    }
794                }
795            }
796        }
797        if output.is_empty() {
798            Ok(json!(WITHHELD))
799        } else {
800            Ok(json!(output))
801        }
802    }
803}
804
805/// Read-only migration preview for an existing project archive. A missing
806/// archive is empty; corrupt, unsafe or locked paths fail closed. In particular,
807/// this never invokes `ArchiveStore::new`, mkdir, chmod, or lock-file creation.
808pub fn export_in(base: &Path, scope_key: &str) -> io::Result<Vec<Value>> {
809    validate_key(scope_key)?;
810    let scope = digest_parts(&[scope_key.as_bytes()]);
811    let fs = match PrivateDir::open_existing(base, &scope) {
812        Ok(fs) => fs,
813        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
814        Err(error) => return Err(error),
815    };
816    ArchiveStore {
817        fs,
818        scope,
819        logical: String::new(), // export scans all logical conversations in this scope
820        redactor: None,
821        abort: None,
822        wait: LOCK_WAIT,
823    }
824    .export_records()
825}
826
827/// Project-wide archive lookup; base and scope are trusted host values, never
828/// model-supplied paths. The same budgets and disclosure policy apply.
829pub fn project_search(
830    base: &Path,
831    scope_key: &str,
832    query: &str,
833    limit: usize,
834) -> io::Result<Vec<ArchiveDescriptor>> {
835    ArchiveStore::new(base, scope_key, "project-reader")?.search(query, limit)
836}
837
838/// Fetch an opaque reference from any logical conversation in this project.
839pub fn project_fetch(
840    base: &Path,
841    scope_key: &str,
842    id: &str,
843    start: usize,
844    limit: usize,
845) -> io::Result<Vec<SharedMessage>> {
846    ArchiveStore::new(base, scope_key, "project-reader")?.fetch(id, start, limit)
847}
848
849fn eligible_class(value: &Value) -> Option<DisclosureClass> {
850    // Provider/UI envelopes sometimes label private reasoning as assistant text.
851    // Do not rely on the role or content-block type alone.
852    if value
853        .get("channel")
854        .is_some_and(|v| !matches!(v.as_str(), Some("final")))
855        || value.get("content_class").is_some_and(|v| {
856            !matches!(
857                v.as_str(),
858                Some("user_text" | "assistant_text" | "tool_calls" | "tool_results")
859            )
860        })
861        || value
862            .get("private")
863            .is_some_and(|v| v != &Value::Bool(false))
864    {
865        return None;
866    }
867    for key in ["sensitivity", "retention", "retention_class"] {
868        if let Some(v) = value.get(key) {
869            if !matches!(v.as_str(), Some("normal" | "standard" | "model_visible")) {
870                return None;
871            }
872        }
873    }
874    let mut class = DisclosureClass::ModelVisible;
875    for key in ["disclosure", "disclosure_class"] {
876        if let Some(v) = value.get(key) {
877            let parsed = DisclosureClass::parse(v.as_str()?)?;
878            if !may_persist(parsed)
879                || !matches!(
880                    parsed,
881                    DisclosureClass::ModelVisible | DisclosureClass::ModelVisibleAfterRedaction
882                )
883            {
884                return None;
885            }
886            class = parsed;
887        }
888    }
889    Some(class)
890}
891
892/// Shared message-envelope selection for preflight and projection. Host
893/// continuation notes and the host's tool-in-flight note are derived
894/// working context, never source evidence (nor the user's words).
895fn projection_content(message: &Value) -> Option<(DisclosureClass, &str, &Value)> {
896    if message.get("_synaps_context").is_some()
897        || crate::core::host_note::is_tool_in_flight_note(message)
898    {
899        return None;
900    }
901    let class = eligible_class(message)?;
902    let role @ ("user" | "assistant") = message.get("role")?.as_str()? else {
903        return None;
904    };
905    Some((class, role, message.get("content")?))
906}
907
908fn identifier<'a>(value: &'a Value, key: &str) -> Option<&'a str> {
909    let text = value.get(key)?.as_str()?;
910    (!text.is_empty()
911        && text.len() <= 128
912        && !credential_like(text)
913        && text
914            .bytes()
915            .all(|b| b.is_ascii_alphanumeric() || b"_-.:".contains(&b)))
916    .then_some(text)
917}
918
919/// Conservatively withhold an entire field rather than risk partial redaction.
920fn credential_like(text: &str) -> bool {
921    let lower = text.to_ascii_lowercase();
922    [
923        "password",
924        "passwd",
925        "api_key",
926        "apikey",
927        "api-key",
928        "access_token",
929        "refresh_token",
930        "secret",
931        "authorization",
932        "bearer ",
933        "private key",
934        "sk-",
935        "ghp_",
936        "github_pat_",
937        "xoxb-",
938        "akia",
939        "eyjh",
940    ]
941    .iter()
942    .any(|needle| lower.contains(needle))
943}
944
945struct InputBudget {
946    bytes: usize,
947    nodes: usize,
948}
949impl InputBudget {
950    fn node(&mut self, depth: usize) -> io::Result<()> {
951        self.nodes += 1;
952        if depth > 32 || self.nodes > 65536 {
953            return Err(invalid("archive input structural budget exceeded"));
954        }
955        Ok(())
956    }
957
958    fn charge(&mut self, bytes: usize) -> io::Result<()> {
959        self.bytes = self.bytes.saturating_add(bytes);
960        if self.bytes > MAX_INPUT_BYTES {
961            return Err(invalid("archive input byte budget exceeded"));
962        }
963        Ok(())
964    }
965
966    // Only selected fields are charged; ignored metadata is opaque. Include the
967    // old per-node/key allowance, never more than the corresponding raw subtree.
968    fn field(&mut self, key: &str, text: &str, depth: usize) -> io::Result<()> {
969        self.node(depth)?;
970        self.charge(key.len().saturating_add(8).saturating_add(text.len()))
971    }
972
973    fn text_field(
974        &mut self,
975        key: &str,
976        text: &str,
977        class: DisclosureClass,
978        redactor: bool,
979        depth: usize,
980    ) -> io::Result<()> {
981        if class == DisclosureClass::ModelVisibleAfterRedaction && !redactor {
982            return self.node(depth); // screen() returns the fixed placeholder
983        }
984        self.field(key, text, depth)
985    }
986
987    /// Allocation-free semantic preflight. Complete for ALL messages before
988    /// invoking a redactor or materializing the projection. Keep the selected
989    /// branches in sync with Projector; skipped payload descendants are opaque.
990    fn message(&mut self, message: &Value, redactor: bool) -> io::Result<()> {
991        self.node(0)?;
992        let Some((class, role, content)) = projection_content(message) else {
993            return Ok(());
994        };
995        self.charge(8)?;
996        self.field("role", role, 1)?;
997        if let Some(text) = content.as_str() {
998            return self.text_field("content", text, class, redactor, 1);
999        }
1000        self.node(1)?;
1001        let Some(blocks) = content.as_array() else {
1002            return Ok(());
1003        };
1004        self.charge("content".len() + 8)?;
1005        for block in blocks {
1006            self.node(2)?;
1007            let Some(mut block_class) = eligible_class(block) else {
1008                continue;
1009            };
1010            if class == DisclosureClass::ModelVisibleAfterRedaction {
1011                block_class = class;
1012            }
1013            match block.get("type").and_then(Value::as_str) {
1014                Some("text") => {
1015                    if let Some(text) = block.get("text").and_then(Value::as_str) {
1016                        self.charge(8)?;
1017                        self.field("type", "text", 3)?;
1018                        self.text_field("text", text, block_class, redactor, 3)?;
1019                    }
1020                }
1021                Some("tool_use") if role == "assistant" => {
1022                    if let (Some(id), Some(name)) =
1023                        (identifier(block, "id"), identifier(block, "name"))
1024                    {
1025                        self.charge(8)?;
1026                        self.field("type", "tool_use", 3)?;
1027                        self.field("id", id, 3)?;
1028                        self.field("name", name, 3)?;
1029                        if name != "context_checkpoint" && redactor {
1030                            if let Some(input) = block.get("input") {
1031                                if let Some(bytes) = self.input(input, 3)? {
1032                                    self.charge("input".len().saturating_add(bytes))?;
1033                                }
1034                            }
1035                        }
1036                    }
1037                }
1038                Some("tool_result") if role == "user" => {
1039                    if let Some(id) = identifier(block, "tool_use_id") {
1040                        self.charge(8)?;
1041                        self.field("type", "tool_result", 3)?;
1042                        self.field("tool_use_id", id, 3)?;
1043                        if block.get("is_error").and_then(Value::as_bool).is_some() {
1044                            self.field("is_error", "", 3)?;
1045                        }
1046                        if redactor {
1047                            if let Some(content) = block.get("content") {
1048                                self.tool_result(content, block_class)?;
1049                            }
1050                        }
1051                    }
1052                }
1053                _ => {}
1054            }
1055        }
1056        Ok(())
1057    }
1058
1059    fn tool_result(&mut self, content: &Value, class: DisclosureClass) -> io::Result<()> {
1060        if let Some(text) = content.as_str() {
1061            return self.text_field("content", text, class, true, 3);
1062        }
1063        self.node(3)?;
1064        if let Some(blocks) = content.as_array() {
1065            self.charge("content".len() + 8)?;
1066            for block in blocks {
1067                self.node(4)?;
1068                if let Some(class) = eligible_class(block) {
1069                    if block["type"] == "text" {
1070                        if let Some(text) = block["text"].as_str() {
1071                            self.charge(8)?;
1072                            self.field("type", "text", 5)?;
1073                            self.text_field("text", text, class, true, 5)?;
1074                        }
1075                    }
1076                }
1077            }
1078        }
1079        Ok(())
1080    }
1081
1082    /// Guarded whole-argument admission, shared with Projector. Byte accounting
1083    /// is provisional until every visited child is eligible: a later excluded
1084    /// child withholds the WHOLE input, regardless of tentative byte cost.
1085    /// Structural visits are never rolled back. Arbitrary tool JSON is not a
1086    /// content-block array; unclassified data/base64 fields remain eligible.
1087    fn input(&mut self, value: &Value, depth: usize) -> io::Result<Option<usize>> {
1088        self.node(depth)?;
1089        if eligible_class(value).is_none() {
1090            return Ok(None);
1091        }
1092        let mut bytes: usize = 8;
1093        match value {
1094            Value::String(text) => bytes = bytes.saturating_add(text.len()),
1095            Value::Array(values) => {
1096                for value in values {
1097                    let Some(cost) = self.input(value, depth + 1)? else {
1098                        return Ok(None);
1099                    };
1100                    bytes = bytes.saturating_add(cost);
1101                }
1102            }
1103            Value::Object(values) => {
1104                for (key, value) in values {
1105                    let Some(cost) = self.input(value, depth + 1)? else {
1106                        return Ok(None);
1107                    };
1108                    bytes = bytes.saturating_add(key.len()).saturating_add(cost);
1109                }
1110            }
1111            _ => {}
1112        }
1113        Ok(Some(bytes))
1114    }
1115}
1116
1117fn projection_digest(
1118    scope: &str,
1119    count: usize,
1120    messages: &[ArchivedMessage],
1121) -> io::Result<String> {
1122    let bytes = bounded_json(messages, MAX_SEGMENT_BYTES)?;
1123    Ok(digest_parts(&[
1124        scope.as_bytes(),
1125        &(count as u64).to_be_bytes(),
1126        &bytes,
1127    ]))
1128}
1129fn digest_parts(parts: &[&[u8]]) -> String {
1130    let mut hash = Sha256::new();
1131    for part in parts {
1132        hash.update((part.len() as u64).to_be_bytes());
1133        hash.update(part);
1134    }
1135    format!("{:x}", hash.finalize())
1136}
1137fn bounded_json<T: Serialize + ?Sized>(value: &T, limit: usize) -> io::Result<Vec<u8>> {
1138    struct Sink {
1139        bytes: Vec<u8>,
1140        limit: usize,
1141    }
1142    impl Write for Sink {
1143        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
1144            if bytes.len() > self.limit - self.bytes.len() {
1145                return Err(invalid("archive serialized byte budget exceeded"));
1146            }
1147            self.bytes.extend_from_slice(bytes);
1148            Ok(bytes.len())
1149        }
1150        fn flush(&mut self) -> io::Result<()> {
1151            Ok(())
1152        }
1153    }
1154    let mut sink = Sink {
1155        bytes: Vec::new(),
1156        limit,
1157    };
1158    serde_json::to_writer(&mut sink, value).map_err(json_error)?;
1159    Ok(sink.bytes)
1160}
1161fn find_record(records: Vec<(usize, Record)>, id: &str) -> io::Result<Record> {
1162    records
1163        .into_iter()
1164        .find(|(_, r)| r.id == id)
1165        .map(|(_, r)| r)
1166        .ok_or_else(not_found)
1167}
1168fn prefix(text: &str, bytes: usize) -> &str {
1169    let mut end = bytes.min(text.len());
1170    while !text.is_char_boundary(end) {
1171        end -= 1;
1172    }
1173    &text[..end]
1174}
1175fn validate_key(key: &str) -> io::Result<()> {
1176    if key.is_empty() || key.len() > 4096 {
1177        return Err(invalid("archive scope/logical id must be 1..4096 bytes"));
1178    }
1179    Ok(())
1180}
1181
1182fn validate_id(id: &str) -> io::Result<()> {
1183    if id.len() != 32
1184        || !id
1185            .bytes()
1186            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
1187    {
1188        Err(invalid("invalid opaque archive id"))
1189    } else {
1190        Ok(())
1191    }
1192}
1193fn invalid(message: &'static str) -> io::Error {
1194    io::Error::new(io::ErrorKind::InvalidInput, message)
1195}
1196fn corrupt() -> io::Error {
1197    io::Error::new(io::ErrorKind::InvalidData, "invalid archive record")
1198}
1199fn not_found() -> io::Error {
1200    io::Error::new(
1201        io::ErrorKind::NotFound,
1202        "archive not available in this scope",
1203    )
1204}
1205fn json_error(_: serde_json::Error) -> io::Error {
1206    invalid("archive JSON serialization failed or exceeded budget")
1207}
1208
1209// ConfinedDir has no public directory fsync or bounded enumeration API. Keep a
1210// private directory fd using the same openat/O_NOFOLLOW policy and modes as
1211// private_fs, and use fixed numbered slots instead of an unbounded index scan.
1212#[cfg(unix)]
1213struct PrivateDir(std::fs::File);
1214#[cfg(unix)]
1215impl PrivateDir {
1216    fn new(base: &Path, scope: &str) -> io::Result<Self> {
1217        use std::os::unix::fs::OpenOptionsExt;
1218        if !base.is_absolute() || base == Path::new("/") {
1219            return Err(invalid("archive base must be an absolute non-root path"));
1220        }
1221        let mut dir = Self(
1222            std::fs::OpenOptions::new()
1223                .read(true)
1224                .custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW | libc::O_CLOEXEC)
1225                .open("/")?,
1226        );
1227        let components: Vec<_> = base.components().collect();
1228        for (index, component) in components.iter().enumerate() {
1229            match component {
1230                std::path::Component::RootDir => {}
1231                std::path::Component::Normal(name) => {
1232                    let name = name
1233                        .to_str()
1234                        .ok_or_else(|| invalid("archive path must be UTF-8"))?;
1235                    dir = dir.child(name, index == components.len() - 1)?;
1236                }
1237                _ => return Err(invalid("archive base contains traversal")),
1238            }
1239        }
1240        dir = dir.child("context-archives", true)?;
1241        dir.child(scope, true)
1242    }
1243    fn open_existing(base: &Path, scope: &str) -> io::Result<Self> {
1244        use std::os::unix::fs::OpenOptionsExt;
1245        if !base.is_absolute() || base == Path::new("/") {
1246            return Err(invalid("archive base must be an absolute non-root path"));
1247        }
1248        let mut dir = Self(
1249            std::fs::OpenOptions::new()
1250                .read(true)
1251                .custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW | libc::O_CLOEXEC)
1252                .open("/")?,
1253        );
1254        for component in base.components() {
1255            match component {
1256                std::path::Component::RootDir => {}
1257                std::path::Component::Normal(name) => {
1258                    let name = c_name(
1259                        name.to_str()
1260                            .ok_or_else(|| invalid("archive path must be UTF-8"))?,
1261                    )?;
1262                    dir = Self(dir.open(&name, libc::O_RDONLY | libc::O_DIRECTORY)?);
1263                }
1264                _ => return Err(invalid("archive base contains traversal")),
1265            }
1266        }
1267        for name in ["context-archives", scope] {
1268            dir = Self(dir.open(&c_name(name)?, libc::O_RDONLY | libc::O_DIRECTORY)?);
1269        }
1270        Ok(dir)
1271    }
1272
1273    fn fd(&self) -> libc::c_int {
1274        use std::os::fd::AsRawFd;
1275        self.0.as_raw_fd()
1276    }
1277    fn child(&self, name: &str, private: bool) -> io::Result<Self> {
1278        use crate::core::private_fs::DIR_MODE;
1279        use std::os::unix::fs::PermissionsExt;
1280        let name = c_name(name)?;
1281        // `mode_t` is u16 on macOS, u32 on Linux; DIR_MODE (0o700) fits both.
1282        let created =
1283            unsafe { libc::mkdirat(self.fd(), name.as_ptr(), DIR_MODE as libc::mode_t) } == 0;
1284        if !created {
1285            let error = io::Error::last_os_error();
1286            if error.raw_os_error() != Some(libc::EEXIST) {
1287                return Err(error);
1288            }
1289        }
1290        let file = self.open(&name, libc::O_RDONLY | libc::O_DIRECTORY)?;
1291        if private || created {
1292            file.set_permissions(std::fs::Permissions::from_mode(DIR_MODE))?;
1293        }
1294        // A previous mkdir may have succeeded but its parent fsync failed.
1295        // EEXIST on retry is not evidence that the directory entry is durable.
1296        self.sync()?;
1297        Ok(Self(file))
1298    }
1299    fn open(&self, name: &std::ffi::CStr, flags: i32) -> io::Result<std::fs::File> {
1300        use std::os::fd::FromRawFd;
1301        let fd = unsafe {
1302            libc::openat(
1303                self.fd(),
1304                name.as_ptr(),
1305                flags | libc::O_NOFOLLOW | libc::O_CLOEXEC | libc::O_NONBLOCK,
1306                crate::core::private_fs::FILE_MODE,
1307            )
1308        };
1309        if fd < 0 {
1310            Err(io::Error::last_os_error())
1311        } else {
1312            Ok(unsafe { std::fs::File::from_raw_fd(fd) })
1313        }
1314    }
1315    fn check_readonly(file: &std::fs::File) -> io::Result<()> {
1316        use std::os::unix::fs::MetadataExt;
1317        let meta = file.metadata()?;
1318        if !meta.is_file() || meta.nlink() != 1 || meta.uid() != unsafe { libc::geteuid() } {
1319            return Err(io::Error::new(
1320                io::ErrorKind::PermissionDenied,
1321                "archive requires owned regular single-link files",
1322            ));
1323        }
1324        Ok(())
1325    }
1326    fn check(file: &std::fs::File) -> io::Result<()> {
1327        use std::os::unix::fs::PermissionsExt;
1328        Self::check_readonly(file)?;
1329        file.set_permissions(std::fs::Permissions::from_mode(
1330            crate::core::private_fs::FILE_MODE,
1331        ))
1332    }
1333    /// Directory identity for the in-process gate.
1334    fn identity(&self) -> io::Result<DirIdentity> {
1335        use std::os::unix::fs::MetadataExt;
1336        let meta = self.0.metadata()?;
1337        Ok((meta.dev(), meta.ino()))
1338    }
1339    fn lock_existing(&self) -> io::Result<Option<FlockGuard>> {
1340        let file = match self.open(&c_name("lock")?, libc::O_RDONLY) {
1341            Ok(file) => file,
1342            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
1343            Err(error) => return Err(error),
1344        };
1345        Self::check_readonly(&file)?;
1346        if !fs4::fs_std::FileExt::try_lock_shared(&file)? {
1347            return Err(io::Error::new(io::ErrorKind::WouldBlock, BUSY_EXPORT));
1348        }
1349        Ok(Some(FlockGuard(file)))
1350    }
1351    /// Never waits: a lock held by another process fails closed at once.
1352    fn lock(&self) -> io::Result<FlockGuard> {
1353        let file = self.open(&c_name("lock")?, libc::O_RDWR | libc::O_CREAT)?;
1354        Self::check(&file)?;
1355        if !fs4::fs_std::FileExt::try_lock_exclusive(&file)? {
1356            return Err(io::Error::new(io::ErrorKind::WouldBlock, BUSY));
1357        }
1358        Ok(FlockGuard(file))
1359    }
1360    fn read_with(
1361        &self,
1362        slot: usize,
1363        remaining: usize,
1364        readonly: bool,
1365    ) -> io::Result<Option<Vec<u8>>> {
1366        let file = match self.open(&c_name(&format!("{slot:03}.json"))?, libc::O_RDONLY) {
1367            Ok(file) => file,
1368            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
1369            Err(error) => return Err(error),
1370        };
1371        if readonly {
1372            Self::check_readonly(&file)?;
1373        } else {
1374            Self::check(&file)?;
1375        }
1376        let limit = remaining.min(MAX_SEGMENT_BYTES);
1377        if file.metadata()?.len() > limit as u64 {
1378            return Err(corrupt());
1379        }
1380        let mut bytes = Vec::new();
1381        file.take(limit as u64 + 1).read_to_end(&mut bytes)?;
1382        if bytes.len() > limit {
1383            return Err(corrupt());
1384        }
1385        Ok(Some(bytes))
1386    }
1387    fn publish(&self, slot: usize, bytes: &[u8]) -> io::Result<()> {
1388        let target = c_name(&format!("{slot:03}.json"))?;
1389        let tmp = c_name("pending.tmp")?;
1390        // Under the lock, an old temporary file is an unpublished failed write.
1391        // unlinkat never follows a link; existing content is never reused.
1392        if unsafe { libc::unlinkat(self.fd(), tmp.as_ptr(), 0) } != 0 {
1393            let error = io::Error::last_os_error();
1394            if error.kind() != io::ErrorKind::NotFound {
1395                return Err(error);
1396            }
1397        }
1398        let result = (|| {
1399            let mut file = self.open(&tmp, libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL)?;
1400            file.write_all(bytes)?;
1401            file.sync_all()?;
1402            // Refuse any target that appeared as a symlink or hardlink.
1403            match self.open(&target, libc::O_RDONLY) {
1404                Ok(existing) => Self::check(&existing)?,
1405                Err(error) if error.kind() == io::ErrorKind::NotFound => {}
1406                Err(error) => return Err(error),
1407            }
1408            if unsafe { libc::renameat(self.fd(), tmp.as_ptr(), self.fd(), target.as_ptr()) } != 0 {
1409                return Err(io::Error::last_os_error());
1410            }
1411            self.sync()
1412        })();
1413        if result.is_err() {
1414            unsafe {
1415                libc::unlinkat(self.fd(), tmp.as_ptr(), 0);
1416            }
1417        }
1418        result
1419    }
1420    fn sync(&self) -> io::Result<()> {
1421        self.0.sync_all()
1422    }
1423}
1424#[cfg(unix)]
1425fn c_name(name: &str) -> io::Result<std::ffi::CString> {
1426    if name.is_empty() || matches!(name, "." | "..") || name.contains(['/', '\\']) {
1427        return Err(invalid("invalid archive component"));
1428    }
1429    std::ffi::CString::new(name).map_err(|_| invalid("invalid archive component"))
1430}
1431
1432#[cfg(not(unix))]
1433struct PrivateDir;
1434#[cfg(not(unix))]
1435impl PrivateDir {
1436    fn unsupported<T>() -> io::Result<T> {
1437        Err(io::Error::new(
1438            io::ErrorKind::Unsupported,
1439            "private context archive requires Unix",
1440        ))
1441    }
1442    fn new(_: &Path, _: &str) -> io::Result<Self> {
1443        Self::unsupported()
1444    }
1445    fn identity(&self) -> io::Result<DirIdentity> {
1446        Self::unsupported()
1447    }
1448    fn lock(&self) -> io::Result<FlockGuard> {
1449        Self::unsupported()
1450    }
1451    fn open_existing(_: &Path, _: &str) -> io::Result<Self> {
1452        Self::unsupported()
1453    }
1454    fn lock_existing(&self) -> io::Result<Option<FlockGuard>> {
1455        Self::unsupported()
1456    }
1457    fn read_with(&self, _: usize, _: usize, _: bool) -> io::Result<Option<Vec<u8>>> {
1458        Self::unsupported()
1459    }
1460    fn publish(&self, _: usize, _: &[u8]) -> io::Result<()> {
1461        Self::unsupported()
1462    }
1463    fn sync(&self) -> io::Result<()> {
1464        Self::unsupported()
1465    }
1466}
1467
1468#[cfg(test)]
1469mod projection_tests {
1470    use super::*;
1471
1472    fn excluded_fixture(blob: &str) -> Vec<SharedMessage> {
1473        vec![
1474            Arc::new(json!({"role":"system","content":blob})),
1475            Arc::new(json!({"role":"developer","content":blob})),
1476            Arc::new(json!({"role":"user","_synaps_context":{"note":blob},"content":blob})),
1477            Arc::new(json!({"role":"user","retention":"local_only","content":blob})),
1478            Arc::new(json!({"role":"assistant","metadata":blob,"content":[
1479                {"type":"thinking","thinking":blob},
1480                {"type":"text","private":true,"text":blob},
1481                {"type":"image","source":{"data":blob}},
1482                {"type":"document","source":{"data":blob}},
1483                {"type":"text","text":" exact evidence αβ\n","metadata":blob},
1484                {"type":"tool_use","id":"cp","name":"context_checkpoint","input":{"note":blob}}
1485            ]})),
1486            Arc::new(json!({"role":"user","content":[
1487                {"type":"tool_result","tool_use_id":"r1","metadata":blob,"content":[
1488                    {"type":"image","source":{"data":blob}},
1489                    {"type":"text","sensitivity":"secret","text":blob},
1490                    {"type":"text","text":"exact tool evidence\n","metadata":blob}
1491                ]}
1492            ]})),
1493        ]
1494    }
1495
1496    #[test]
1497    fn excluded_aggregate_does_not_consume_eligible_budget_or_change_digest() {
1498        // Every blob fits by itself; collectively the raw input exceeds 16MiB.
1499        let messages = excluded_fixture(&"x".repeat(2 * 1024 * 1024));
1500        let source = messages.clone();
1501        for redactor in [None, Some(Arc::new(str::to_owned) as Arc<ArchiveRedactor>)] {
1502            let small =
1503                project_archive("excluded", &excluded_fixture("x"), "note", redactor.clone())
1504                    .unwrap();
1505            let large = project_archive("excluded", &messages, "note", redactor).unwrap();
1506            assert!(small == large);
1507            assert_eq!(large.source_message_count, 6);
1508            assert_eq!(large.messages[0].source_index, 4);
1509            assert_eq!(large.messages[0].block_indices, vec![4, 5]);
1510        }
1511        for (old, new) in source.iter().zip(&messages) {
1512            assert!(Arc::ptr_eq(old, new));
1513        }
1514    }
1515
1516    #[test]
1517    fn eligible_aggregate_refuses_before_any_redactor_call() {
1518        use std::sync::atomic::{AtomicUsize, Ordering};
1519        let calls = Arc::new(AtomicUsize::new(0));
1520        let count = calls.clone();
1521        let redactor: Arc<ArchiveRedactor> = Arc::new(move |_| {
1522            count.fetch_add(1, Ordering::SeqCst);
1523            "small".to_owned()
1524        });
1525        let piece = "x".repeat(MAX_INPUT_BYTES / 2);
1526        for messages in [
1527            vec![
1528                Arc::new(json!({"role":"user","content":piece})),
1529                Arc::new(json!({"role":"assistant","content":piece})),
1530            ],
1531            vec![Arc::new(json!({"role":"user","content":[{
1532                "type":"tool_result","tool_use_id":"r1","content":[
1533                    {"type":"text","text":piece}, {"type":"text","text":piece}
1534                ]
1535            }]}))],
1536        ] {
1537            let before = messages.clone();
1538            let error = project_archive("overflow", &messages, "", Some(redactor.clone()))
1539                .err()
1540                .unwrap();
1541            assert!(error.to_string().contains("input byte budget"));
1542            assert_eq!(calls.load(Ordering::SeqCst), 0);
1543            assert_eq!(before, messages);
1544        }
1545    }
1546
1547    #[test]
1548    fn large_tool_results_exact_with_redactor_opaque_without_it() {
1549        let text = "evidence αβ\n".repeat(250_000);
1550        let make = |text: &str| {
1551            vec![Arc::new(json!({"role":"user","content":[{
1552                "type":"tool_result","tool_use_id":"r1","content":text
1553            }]}))]
1554        };
1555        let p = project_archive("logs", &make(&text), "", Some(Arc::new(str::to_owned))).unwrap();
1556        assert_eq!(p.messages[0].message["content"][0]["content"], text);
1557        let huge = make(&"x".repeat(MAX_INPUT_BYTES + 1));
1558        let p = project_archive("logs", &huge, "", None).unwrap();
1559        assert_eq!(p.messages[0].message["content"][0]["content"], WITHHELD);
1560        assert!(project_archive("logs", &huge, "", Some(Arc::new(str::to_owned))).is_err());
1561    }
1562
1563    #[test]
1564    fn whole_argument_withholding_discards_tentative_bytes_not_structural_guards() {
1565        let huge = "x".repeat(MAX_INPUT_BYTES + 1);
1566        let make = |input: Value| {
1567            vec![Arc::new(json!({"role":"assistant","content":[{
1568                "type":"tool_use","id":"r1","name":"read","input":input
1569            }]}))]
1570        };
1571        // Both orders: the otherwise eligible field can precede the exclusion.
1572        for input in [
1573            json!({"a":huge,"z":{"sensitivity":"secret","body":huge}}),
1574            json!({"a":{"sensitivity":"secret","body":huge},"z":huge}),
1575        ] {
1576            let p =
1577                project_archive("input", &make(input), "", Some(Arc::new(str::to_owned))).unwrap();
1578            assert_eq!(
1579                p.messages[0].message["content"][0]["input"],
1580                json!({"_archive_withheld":true})
1581            );
1582        }
1583        // Arbitrary binary-looking tool arguments are NOT content blocks.
1584        let input = json!({"type":"image","data":"YWJj"});
1585        let p = project_archive(
1586            "input",
1587            &make(input.clone()),
1588            "",
1589            Some(Arc::new(str::to_owned)),
1590        )
1591        .unwrap();
1592        assert_eq!(p.messages[0].message["content"][0]["input"], input);
1593        let large = make(json!({"type":"image","data":huge}));
1594        assert!(project_archive("input", &large, "", Some(Arc::new(str::to_owned))).is_err());
1595        assert!(project_archive("input", &large, "", None).is_ok());
1596    }
1597
1598    #[test]
1599    fn visited_structure_is_bounded_but_excluded_descendants_are_opaque() {
1600        let mut deep = json!("leaf");
1601        for _ in 0..40 {
1602            deep = json!([deep]);
1603        }
1604        let make = |input: Value| {
1605            vec![Arc::new(json!({"role":"assistant","content":[{
1606                "type":"tool_use","id":"r1","name":"read","input":input
1607            }]}))]
1608        };
1609        let identity = Some(Arc::new(str::to_owned) as Arc<ArchiveRedactor>);
1610        assert!(project_archive("depth", &make(deep.clone()), "", identity.clone()).is_err());
1611        assert!(project_archive("depth", &make(deep.clone()), "", None).is_ok());
1612        assert!(project_archive(
1613            "depth",
1614            &make(json!({"private":true,"blob":deep})),
1615            "",
1616            identity.clone()
1617        )
1618        .is_ok());
1619        // A malformed content block isn't recursively interpreted as a block array.
1620        assert!(project_archive(
1621            "depth",
1622            &[Arc::new(json!({"role":"user","content":deep}))],
1623            "",
1624            identity.clone()
1625        )
1626        .is_ok());
1627        let wide = vec![Value::Null; 65536];
1628        assert!(project_archive("nodes", &make(json!(wide)), "", identity.clone()).is_err());
1629        let excluded = vec![Arc::new(json!({"role":"user","content":wide}))];
1630        assert!(project_archive("nodes", &excluded, "", identity).is_err());
1631    }
1632
1633    #[test]
1634    fn withheld_after_redaction_text_is_not_scanned_without_redactor() {
1635        let messages = vec![Arc::new(json!({"role":"user",
1636            "disclosure":"model_visible_after_redaction", "content":"x".repeat(MAX_INPUT_BYTES+1)
1637        }))];
1638        let p = project_archive("withheld", &messages, "", None).unwrap();
1639        assert_eq!(p.messages[0].message["content"], WITHHELD);
1640        assert!(project_archive("withheld", &messages, "", Some(Arc::new(str::to_owned))).is_err());
1641    }
1642
1643    #[test]
1644    fn json_escape_scratch_can_exceed_segment_limit_before_redaction() {
1645        // 3MiB source ->18MiB JSON, accepted by the old raw preflight and then
1646        // reduced by the trusted redactor. Do not introduce a new 16MiB scratch cap.
1647        let input = json!({"text":"\0".repeat(3 * 1024 * 1024)});
1648        let messages = vec![Arc::new(json!({"role":"assistant","content":[{
1649            "type":"tool_use","id":"r1","name":"read","input":input
1650        }]}))];
1651        let redactor: Arc<ArchiveRedactor> = Arc::new(|s| {
1652            if s.starts_with('{') {
1653                assert!(s.len() > MAX_SEGMENT_BYTES);
1654                "{\"text\":\"screened\"}".to_owned()
1655            } else {
1656                s.to_owned()
1657            }
1658        });
1659        let p = project_archive("escaping", &messages, "", Some(redactor)).unwrap();
1660        assert_eq!(
1661            p.messages[0].message["content"][0]["input"],
1662            json!({"text":"screened"})
1663        );
1664        assert!(project_archive("escaping", &messages, "", Some(Arc::new(str::to_owned))).is_err());
1665    }
1666
1667    fn compatibility_fixture() -> Vec<SharedMessage> {
1668        vec![
1669            Arc::new(json!({"role":"system","content":"omitted"})),
1670            Arc::new(json!({"role":"user","content":" exact αβ\n"})),
1671            Arc::new(json!({"role":"assistant","content":[
1672                {"type":"thinking","thinking":"private"},
1673                {"type":"text","text":"classified-value"},
1674                {"type":"tool_use","id":"r1","name":"read","input":{"path":"src/lib.rs","n":7,"binary":{"type":"image","data":"YWJj"}}},
1675                {"type":"tool_use","id":"c1","name":"context_checkpoint","input":{"note":"hidden"}},
1676                {"type":"tool_use","id":"x1","name":"read","input":{"nested":{"disclosure":"local_only","body":"omitted"}}}
1677            ]})),
1678            Arc::new(json!({"role":"user","content":[
1679                {"type":"tool_result","tool_use_id":"r1","is_error":false,"content":[
1680                    {"type":"image","data":"omitted"},
1681                    {"type":"text","text":"exact result\n"},
1682                    {"type":"text","text":"classified-value","disclosure":"model_visible_after_redaction"}
1683                ]}
1684            ]})),
1685            Arc::new(
1686                json!({"role":"user","disclosure":"model_visible_after_redaction","content":"classified-value"}),
1687            ),
1688            Arc::new(json!({"role":"user","_synaps_context":{},"content":"hidden"})),
1689        ]
1690    }
1691
1692    #[test]
1693    fn legacy_projection_golden() {
1694        // Captured against the legacy raw-input preflight before this fix.
1695        for (redactor, digest) in [
1696            (
1697                None,
1698                "aeb544a71629f741def049575eefb1a765117d65aa914bdcc1657fa677622031",
1699            ),
1700            (
1701                Some(
1702                    Arc::new(|s: &str| s.replace("classified-value", "[redacted]"))
1703                        as Arc<ArchiveRedactor>,
1704                ),
1705                "d70b321210ca3d2bc531b31a35cc02ee267eeb572e8a6dd88f1fb62725b082ec",
1706            ),
1707        ] {
1708            let p = project_archive(
1709                "compat-session",
1710                &compatibility_fixture(),
1711                "hidden note",
1712                redactor,
1713            )
1714            .unwrap();
1715            assert_eq!(p.digest, digest);
1716        }
1717    }
1718
1719    #[test]
1720    fn pure_projection_keeps_legacy_logical_digest_and_never_opens_paths() {
1721        let tmp = tempfile::tempdir().unwrap();
1722        let logical = tmp
1723            .path()
1724            .join("not-created/session")
1725            .to_str()
1726            .unwrap()
1727            .to_owned();
1728        let messages = vec![
1729            Arc::new(json!({"role":"system","content":"OMITTED"})),
1730            Arc::new(json!({"role":"assistant","content":[
1731                {"type":"thinking","thinking":"OMITTED"},
1732                {"type":"text","text":" exact αβ evidence\n"}
1733            ]})),
1734        ];
1735        let source = serde_json::to_vec(&messages).unwrap();
1736        let first = project_archive(&logical, &messages, "hidden note", None).unwrap();
1737        assert_eq!(first.logical_id, digest_parts(&[logical.as_bytes()]));
1738        assert_eq!(
1739            first.digest,
1740            projection_digest(&first.logical_id, 2, &first.messages).unwrap()
1741        );
1742        assert_eq!(first.messages[0].source_index, 1);
1743        assert_eq!(first.messages[0].block_indices, vec![1]);
1744        assert_eq!(first.source_message_count, 2);
1745        let mut changed = messages.clone();
1746        changed[0] = Arc::new(json!({"role":"developer","content":"DIFFERENT OMITTED"}));
1747        let second = project_archive(&logical, &changed, "changed note", None).unwrap();
1748        assert_eq!(first.digest, second.digest);
1749        assert_ne!(
1750            first.digest,
1751            project_archive("other-session", &messages, "hidden note", None)
1752                .unwrap()
1753                .digest
1754        );
1755        assert_eq!(serde_json::to_vec(&messages).unwrap(), source);
1756        assert_eq!(std::fs::read_dir(tmp.path()).unwrap().count(), 0);
1757    }
1758
1759    #[test]
1760    fn pure_projection_shares_disclosure_redaction_and_input_limits() {
1761        let redactor: Arc<ArchiveRedactor> =
1762            Arc::new(|s| s.replace("classified-value", "[redacted]"));
1763        let messages = vec![
1764            Arc::new(json!({"role":"assistant","content":[
1765                {"type":"tool_use","id":"read_1","name":"read","input":{"path":"src/lib.rs"}},
1766                {"type":"tool_use","id":"cp","name":"context_checkpoint","input":{"note":"NOTE_ONLY"}}
1767            ]})),
1768            Arc::new(
1769                json!({"role":"user","content":[{"type":"tool_result","tool_use_id":"read_1","content":"classified-value"}]}),
1770            ),
1771            Arc::new(json!({"role":"user","sensitivity":"secret","content":"OMITTED"})),
1772            Arc::new(json!({"role":"user","_synaps_context":{},"content":"NOTE_ONLY"})),
1773            // The host's tool-in-flight note: never archived as user evidence.
1774            Arc::new(json!({"role":"user","content":crate::core::host_note::tool_in_flight_note_text(
1775                "Canceled by user",
1776                &["bash".into()],
1777            )})),
1778        ];
1779        let projected =
1780            project_archive("session", &messages, "classified-value", Some(redactor)).unwrap();
1781        assert_eq!(projected.note, "[redacted]");
1782        let json = serde_json::to_string(&projected.messages).unwrap();
1783        assert!(!json.contains("classified-value"));
1784        assert!(!json.contains("NOTE_ONLY"));
1785        assert!(!json.contains("partially applied"), "the tool note is not archived");
1786        assert!(!json.contains("OMITTED"));
1787        assert!(json.contains("src/lib.rs"));
1788        assert_eq!(projected.messages.len(), 2);
1789        let withheld = project_archive("session", &messages, "password=bad", None).unwrap();
1790        assert_eq!(withheld.note, WITHHELD);
1791        assert!(!serde_json::to_string(&withheld.messages)
1792            .unwrap()
1793            .contains("src/lib.rs"));
1794        assert!(project_archive("", &[], "", None).is_err());
1795        assert!(project_archive(&"l".repeat(4097), &[], "", None).is_err());
1796        assert!(project_archive("s", &[], &"n".repeat(MAX_NOTE_BYTES + 1), None).is_err());
1797        assert!(
1798            project_archive("s", &vec![messages[0].clone(); MAX_MESSAGES + 1], "", None).is_err()
1799        );
1800    }
1801}
1802
1803// Archive locks are non-blocking: a concurrently spawned test process can
1804// briefly hold an inherited lock descriptor until it executes, see
1805// memory::repository tests.
1806#[cfg(all(test, unix))]
1807#[serial_test::parallel(process_spawn)]
1808mod tests {
1809    use super::*;
1810    use std::os::unix::fs::{symlink, PermissionsExt};
1811
1812    fn msg(role: &str, content: Value) -> SharedMessage {
1813        Arc::new(json!({"role":role,"content":content}))
1814    }
1815    fn setup() -> (tempfile::TempDir, ArchiveStore) {
1816        let tmp = tempfile::tempdir().unwrap();
1817        let store =
1818            ArchiveStore::new(tmp.path(), "synthetic-project", "synthetic-logical").unwrap();
1819        (tmp, store)
1820    }
1821    fn directory(tmp: &tempfile::TempDir, store: &ArchiveStore) -> std::path::PathBuf {
1822        tmp.path().join("context-archives").join(&store.scope)
1823    }
1824
1825    #[test]
1826    fn export_preview_is_read_only_and_matches_projection_including_tombstones() {
1827        use std::os::unix::fs::MetadataExt;
1828        let tmp = tempfile::tempdir().unwrap();
1829        assert!(export_in(&tmp.path().join("missing"), "synthetic-project")
1830            .unwrap()
1831            .is_empty());
1832        assert!(export_in(tmp.path(), "synthetic-project")
1833            .unwrap()
1834            .is_empty());
1835        assert_eq!(std::fs::read_dir(tmp.path()).unwrap().count(), 0);
1836        let store =
1837            ArchiveStore::new(tmp.path(), "synthetic-project", "synthetic-logical").unwrap();
1838        let dir = directory(&tmp, &store);
1839        // Even a newly opened empty store has no lock until its first operation.
1840        assert!(export_in(tmp.path(), "synthetic-project")
1841            .unwrap()
1842            .is_empty());
1843        assert!(!dir.join("lock").exists());
1844        let messages = vec![msg("user", json!("migration evidence"))];
1845        let reference = store.seal(&messages, "hidden note").unwrap();
1846        let projection =
1847            project_archive("synthetic-logical", &messages, "hidden note", None).unwrap();
1848        let tombstone = store
1849            .seal(&[msg("assistant", json!("remove me"))], "remove note")
1850            .unwrap();
1851        store.forget(&tombstone.id).unwrap();
1852        let mut paths = vec![
1853            tmp.path().to_path_buf(),
1854            tmp.path().join("context-archives"),
1855            dir.clone(),
1856        ];
1857        paths.extend(std::fs::read_dir(&dir).unwrap().map(|e| e.unwrap().path()));
1858        // Export must not silently repair permissions, unlike normal archive IO.
1859        for path in &paths {
1860            let mode = if path.is_dir() { 0o750 } else { 0o640 };
1861            std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode)).unwrap();
1862        }
1863        let snapshot = || {
1864            paths
1865                .iter()
1866                .map(|p| {
1867                    let m = std::fs::metadata(p).unwrap();
1868                    (
1869                        m.mode(),
1870                        m.mtime(),
1871                        m.mtime_nsec(),
1872                        if m.is_file() {
1873                            std::fs::read(p).unwrap()
1874                        } else {
1875                            vec![]
1876                        },
1877                    )
1878                })
1879                .collect::<Vec<_>>()
1880        };
1881        let before = snapshot();
1882        let rows = export_in(tmp.path(), "synthetic-project").unwrap();
1883        assert_eq!(rows, store.export_records().unwrap());
1884        assert_eq!(before, snapshot());
1885        assert_eq!(
1886            rows[0],
1887            json!({"id":reference.id,"digest":projection.digest,"logical_id":projection.logical_id,
1888            "source_message_count":1,"messages":projection.messages,"note":"hidden note","tombstone":false})
1889        );
1890        assert_eq!(rows[1]["id"], tombstone.id);
1891        assert_eq!(rows[1]["tombstone"], true);
1892        assert_eq!(rows[1]["source_message_count"], 0);
1893        assert_eq!(rows[1]["messages"], json!([]));
1894        assert_eq!(rows[1]["note"], "");
1895        let lock = store.fs.lock().unwrap();
1896        assert_eq!(
1897            export_in(tmp.path(), "synthetic-project")
1898                .unwrap_err()
1899                .kind(),
1900            io::ErrorKind::WouldBlock
1901        );
1902        drop(lock);
1903        std::fs::write(dir.join("000.json"), b"corrupt").unwrap();
1904        assert!(export_in(tmp.path(), "synthetic-project").is_err());
1905    }
1906
1907    #[test]
1908    fn export_preview_refuses_links_and_never_creates_a_missing_lock() {
1909        let (tmp, store) = setup();
1910        let dir = directory(&tmp, &store);
1911        let r = store
1912            .seal(&[msg("user", json!("evidence"))], "note")
1913            .unwrap();
1914        let path = dir.join("000.json");
1915        let bytes = std::fs::read(&path).unwrap();
1916        std::fs::remove_file(dir.join("lock")).unwrap();
1917        assert!(export_in(tmp.path(), "synthetic-project").is_err());
1918        assert!(!dir.join("lock").exists());
1919        // The normal API may recreate its lock, but preview cannot.
1920        assert!(store.fetch(&r.id, 0, 1).is_ok());
1921        std::fs::remove_file(&path).unwrap();
1922        let outside = tmp.path().join("outside");
1923        std::fs::write(&outside, &bytes).unwrap();
1924        symlink(&outside, &path).unwrap();
1925        assert!(export_in(tmp.path(), "synthetic-project").is_err());
1926        std::fs::remove_file(&path).unwrap();
1927        std::fs::hard_link(&outside, &path).unwrap();
1928        assert!(export_in(tmp.path(), "synthetic-project").is_err());
1929        let alias = tmp.path().join("alias");
1930        symlink(tmp.path(), &alias).unwrap();
1931        assert!(export_in(&alias, "synthetic-project").is_err());
1932        assert_eq!(std::fs::read(&outside).unwrap(), bytes);
1933    }
1934
1935    #[test]
1936    fn attachments_remain_outside_eligible_memory_archive() {
1937        let (tmp, store) = setup();
1938        let messages = vec![msg(
1939            "user",
1940            json!([
1941                {"type":"text","text":"Compare the selected files"},
1942                {"type":"image","source":{"type":"base64","media_type":"image/png","data":"ATTACHMENT_IMAGE_SECRET"}},
1943                {"type":"document","title":"ATTACHMENT_FILENAME_SECRET","source":{"type":"base64","media_type":"application/pdf","data":"ATTACHMENT_PDF_SECRET"}},
1944                {"type":"document","source":{"type":"text","media_type":"text/plain","data":"ATTACHMENT_TEXT_SECRET"}}
1945            ]),
1946        )];
1947        let reference = store.seal(&messages, "safe checkpoint").unwrap();
1948        let fetched = store.fetch(&reference.id, 0, 10).unwrap();
1949        let serialized = serde_json::to_string(&fetched).unwrap();
1950        assert!(serialized.contains("Compare the selected files"));
1951        assert!(!serialized.contains("ATTACHMENT_"));
1952        let raw = std::fs::read_to_string(directory(&tmp, &store).join("000.json")).unwrap();
1953        assert!(!raw.contains("ATTACHMENT_"));
1954        assert!(messages[0].to_string().contains("ATTACHMENT_TEXT_SECRET"));
1955    }
1956
1957    #[test]
1958    fn exact_text_source_indices_note_and_restart() {
1959        let (tmp, store) = setup();
1960        let messages = vec![
1961            msg("system", json!("FORBIDDEN SYSTEM")),
1962            msg("user", json!(" exact\n αβ  ")),
1963            msg(
1964                "assistant",
1965                json!([
1966                    {"type":"thinking","thinking":"FORBIDDEN REASONING"},
1967                    {"type":"text","text":"evidence✓", "cache_control":{"type":"ephemeral"}},
1968                    {"type":"image","data":"FORBIDDEN IMAGE"}
1969                ]),
1970            ),
1971            msg("developer", json!("FORBIDDEN DEVELOPER")),
1972        ];
1973        let reference = store.seal(&messages, "hidden routing note").unwrap();
1974        assert_eq!(reference.message_count, 2);
1975        assert_eq!(reference.source_message_count, 4);
1976        let reopened =
1977            ArchiveStore::new(tmp.path(), "synthetic-project", "synthetic-logical").unwrap();
1978        let fetched = reopened
1979            .fetch_with_provenance(&reference.id, 0, 128)
1980            .unwrap();
1981        assert_eq!(fetched[0].message, messages[1]);
1982        assert_eq!(fetched[0].source_index, 1);
1983        assert_eq!(fetched[1].source_index, 2);
1984        assert_eq!(fetched[1].block_indices, vec![1]);
1985        assert_eq!(
1986            reopened.fetch_note(&reference.id).unwrap(),
1987            "hidden routing note"
1988        );
1989        assert!(reopened.search("hidden routing", 10).unwrap().is_empty());
1990        assert_eq!(reopened.search("evidence", 10).unwrap()[0].id, reference.id);
1991        assert!(reopened
1992            .fetch(&reference.id, usize::MAX, 10)
1993            .unwrap()
1994            .is_empty());
1995        let raw = std::fs::read_to_string(directory(&tmp, &store).join("000.json")).unwrap();
1996        assert!(!raw.contains("FORBIDDEN"));
1997        assert!(!raw.contains("cache_control"));
1998        assert_eq!(messages[0]["content"], "FORBIDDEN SYSTEM");
1999    }
2000
2001    #[test]
2002    fn duplicate_retry_ignores_note_but_not_logical_id() {
2003        let (tmp, store) = setup();
2004        let messages = vec![msg("user", json!("same eligible words"))];
2005        let first = store.seal(&messages, "first").unwrap();
2006        assert_eq!(store.seal(&messages, "second").unwrap(), first);
2007        assert_eq!(store.fetch_note(&first.id).unwrap(), "first");
2008        let other = ArchiveStore::new(tmp.path(), "synthetic-project", "other-logical").unwrap();
2009        let second = other.seal(&messages, "").unwrap();
2010        assert_ne!(first.id, second.id);
2011        assert_eq!(
2012            project_search(tmp.path(), "synthetic-project", "", 10)
2013                .unwrap()
2014                .len(),
2015            2
2016        );
2017        assert_eq!(
2018            project_fetch(tmp.path(), "synthetic-project", &first.id, 0, 10).unwrap(),
2019            messages
2020        );
2021        assert_eq!(other.fetch(&first.id, 0, 1).unwrap(), messages);
2022        store.forget(&first.id).unwrap();
2023        assert!(store.seal(&messages, "third").is_err());
2024        assert_eq!(other.fetch(&second.id, 0, 1).unwrap(), messages);
2025        assert_eq!(other.seal(&messages, "third").unwrap(), second);
2026    }
2027
2028    #[test]
2029    fn deletion_is_content_free_durable_and_cannot_resurrect() {
2030        let (tmp, store) = setup();
2031        let messages = vec![msg("user", json!("uniquely removable content"))];
2032        let reference = store.seal(&messages, "removable note").unwrap();
2033        store.forget(&reference.id).unwrap();
2034        store.forget(&reference.id).unwrap();
2035        let raw = std::fs::read_to_string(directory(&tmp, &store).join("000.json")).unwrap();
2036        assert!(raw.contains("tombstone"));
2037        assert!(!raw.contains("removable"));
2038        assert!(!raw.contains("source_message_count"));
2039        let store =
2040            ArchiveStore::new(tmp.path(), "synthetic-project", "synthetic-logical").unwrap();
2041        assert_eq!(
2042            store.fetch(&reference.id, 0, 1).unwrap_err().kind(),
2043            io::ErrorKind::NotFound
2044        );
2045        assert!(store.fetch_note(&reference.id).is_err());
2046        assert!(store.search("", 10).unwrap().is_empty());
2047        assert_eq!(
2048            store.seal(&messages, "changed note").unwrap_err().kind(),
2049            io::ErrorKind::PermissionDenied
2050        );
2051    }
2052
2053    #[test]
2054    fn foreign_scope_copied_records_and_traversal_fail_closed() {
2055        let (tmp, store) = setup();
2056        let reference = store
2057            .seal(&[msg("user", json!("scope evidence"))], "")
2058            .unwrap();
2059        let foreign =
2060            ArchiveStore::new(tmp.path(), "foreign-project", "synthetic-logical").unwrap();
2061        assert!(foreign.fetch(&reference.id, 0, 1).is_err());
2062        for id in [
2063            "../000.json",
2064            "/tmp/file",
2065            "",
2066            "a/../b",
2067            "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
2068        ] {
2069            assert_eq!(
2070                store.fetch(id, 0, 1).unwrap_err().kind(),
2071                io::ErrorKind::InvalidInput
2072            );
2073            assert!(store.forget(id).is_err());
2074        }
2075        std::fs::copy(
2076            directory(&tmp, &store).join("000.json"),
2077            directory(&tmp, &foreign).join("000.json"),
2078        )
2079        .unwrap();
2080        assert_eq!(
2081            foreign.search("", 1).unwrap_err().kind(),
2082            io::ErrorKind::InvalidData
2083        );
2084        assert!(ArchiveStore::new(&tmp.path().join("../escape"), "p", "l").is_err());
2085    }
2086
2087    #[test]
2088    fn disclosure_and_raw_tool_data_are_withheld() {
2089        let (tmp, store) = setup();
2090        let mut messages = vec![
2091            msg(
2092                "assistant",
2093                json!([{"type":"tool_use","id":"call_1","name":"read","input":{"path":"unclassified raw argument"}}]),
2094            ),
2095            msg(
2096                "user",
2097                json!([{"type":"tool_result","tool_use_id":"call_1","content":"unclassified raw output"}]),
2098            ),
2099            msg("user", json!("password=synthetic-credential")),
2100        ];
2101        for class in [
2102            "local_only",
2103            "persist_never_transmit",
2104            "never_persist",
2105            "model_visible_after_consent",
2106            "unknown",
2107        ] {
2108            messages.push(Arc::new(
2109                json!({"role":"user","content":"FORBIDDEN DISCLOSURE","disclosure":class}),
2110            ));
2111        }
2112        messages.push(Arc::new(
2113            json!({"role":"user","content":"FORBIDDEN SENSITIVITY","sensitivity":"secret"}),
2114        ));
2115        let reference = store.seal(&messages, "").unwrap();
2116        assert_eq!(reference.message_count, 3);
2117        let raw = std::fs::read_to_string(directory(&tmp, &store).join("000.json")).unwrap();
2118        for forbidden in ["unclassified raw", "synthetic-credential", "FORBIDDEN"] {
2119            assert!(!raw.contains(forbidden));
2120        }
2121        assert!(raw.contains("call_1"));
2122        assert!(raw.contains("_archive_withheld"));
2123    }
2124
2125    #[test]
2126    fn derived_context_envelope_is_not_searchable_source() {
2127        let (_tmp, store) = setup();
2128        let first = store
2129            .seal(
2130                &[msg("assistant", json!("original evidence"))],
2131                "NOTE_ONLY_SENTINEL",
2132            )
2133            .unwrap();
2134        let second=store.seal(&[
2135            Arc::new(json!({"role":"user","_synaps_context":{"schema":"synaps-context-window/1","archive":first.id},"content":"NOTE_ONLY_SENTINEL"})),
2136            Arc::new(json!({"role":"assistant","content":[{"type":"tool_use","id":"checkpoint-note","name":"context_checkpoint","input":{"phase":"plan","note":"NOTE_ONLY_SENTINEL"}}]})),
2137            msg("assistant",json!("second-window evidence")),
2138        ],"next note").unwrap();
2139        assert!(store.search("NOTE_ONLY_SENTINEL", 10).unwrap().is_empty());
2140        let fetched = store.fetch_with_provenance(&second.id, 0, 10).unwrap();
2141        assert_eq!(fetched.len(), 2);
2142        assert_eq!(fetched[0].source_index, 1);
2143        assert_eq!(fetched[1].source_index, 2);
2144        assert!(!serde_json::to_string(&fetched)
2145            .unwrap()
2146            .contains("NOTE_ONLY_SENTINEL"));
2147    }
2148
2149    #[test]
2150    fn checkpoint_tool_note_stays_hidden_even_with_redactor() {
2151        let (_tmp, store) = setup();
2152        let store = store.with_redactor(Arc::new(str::to_owned));
2153        let reference = store
2154            .seal(
2155                &[msg(
2156                    "assistant",
2157                    json!([
2158                        {"type":"tool_use","id":"checkpoint-note","name":"context_checkpoint",
2159                         "input":{"phase":"plan","note":"NOTE_ARGUMENT_SENTINEL"}}
2160                    ]),
2161                )],
2162                "NOTE_ARGUMENT_SENTINEL",
2163            )
2164            .unwrap();
2165        assert!(store
2166            .search("NOTE_ARGUMENT_SENTINEL", 10)
2167            .unwrap()
2168            .is_empty());
2169        assert!(
2170            !serde_json::to_string(&store.fetch(&reference.id, 0, 10).unwrap())
2171                .unwrap()
2172                .contains("NOTE_ARGUMENT_SENTINEL")
2173        );
2174        assert_eq!(
2175            store.fetch_note(&reference.id).unwrap(),
2176            "NOTE_ARGUMENT_SENTINEL"
2177        );
2178    }
2179
2180    #[test]
2181    fn trusted_redactor_retains_exact_eligible_tool_evidence() {
2182        let (_tmp, store) = setup();
2183        let store = store.with_redactor(Arc::new(|s| s.replace("classified-value", "[redacted]")));
2184        let messages = vec![
2185            msg(
2186                "assistant",
2187                json!([{"type":"tool_use","id":"call_1","name":"read","input":{"path":"src/lib.rs","line":7}}]),
2188            ),
2189            msg(
2190                "user",
2191                json!([{"type":"tool_result","tool_use_id":"call_1","is_error":false,"content":"line 7\n  exact evidence\nclassified-value"}]),
2192            ),
2193        ];
2194        let reference = store.seal(&messages, "").unwrap();
2195        let fetched = store.fetch(&reference.id, 0, 10).unwrap();
2196        assert_eq!(fetched[0], messages[0]);
2197        assert_eq!(
2198            fetched[1]["content"][0]["content"],
2199            "line 7\n  exact evidence\n[redacted]"
2200        );
2201    }
2202
2203    #[test]
2204    fn oversized_input_output_limits_and_depth_leave_source_intact() {
2205        let (_tmp, store) = setup();
2206        let huge = vec![msg("user", json!("x".repeat(MAX_INPUT_BYTES + 1)))];
2207        assert!(store.seal(&huge, "").is_err());
2208        assert_eq!(
2209            huge[0]["content"].as_str().unwrap().len(),
2210            MAX_INPUT_BYTES + 1
2211        );
2212        let file_too_large = vec![msg("user", json!("\n".repeat(MAX_SEGMENT_BYTES / 2)))];
2213        assert!(store.seal(&file_too_large, "").is_err());
2214        assert!(store
2215            .seal(&vec![msg("user", json!("a")); MAX_MESSAGES + 1], "")
2216            .is_err());
2217        assert!(store.seal(&[], &"n".repeat(MAX_NOTE_BYTES + 1)).is_err());
2218        let mut deep = json!("value");
2219        for _ in 0..40 {
2220            deep = json!([deep]);
2221        }
2222        // Unknown content blocks are opaque; admitted tool JSON still has a
2223        // strict recursion limit before serialization or redaction.
2224        let deep = vec![msg(
2225            "assistant",
2226            json!([{
2227                "type":"tool_use", "id":"r1", "name":"read", "input":deep
2228            }]),
2229        )];
2230        let store = store.with_redactor(Arc::new(str::to_owned));
2231        assert!(store.seal(&deep, "").is_err());
2232        assert!(store.search(&"q".repeat(513), 1).is_err());
2233        assert!(store.search("", MAX_SEARCH_RESULTS + 1).is_err());
2234        assert!(store
2235            .fetch(&"a".repeat(32), 0, MAX_FETCH_MESSAGES + 1)
2236            .is_err());
2237        assert!(store.search("", 10).unwrap().is_empty());
2238    }
2239
2240    #[test]
2241    fn torn_final_fails_closed_stale_temporary_is_not_visible() {
2242        let (tmp, store) = setup();
2243        let dir = directory(&tmp, &store);
2244        std::fs::write(dir.join("pending.tmp"), b"unpublished partial").unwrap();
2245        assert!(store.search("", 1).unwrap().is_empty());
2246        let reference = store.seal(&[msg("user", json!("complete"))], "").unwrap();
2247        assert!(!dir.join("pending.tmp").exists());
2248        std::fs::write(dir.join("000.json"), b"{\"version\":1,").unwrap();
2249        assert_eq!(
2250            store.fetch(&reference.id, 0, 1).unwrap_err().kind(),
2251            io::ErrorKind::InvalidData
2252        );
2253        assert!(store.search("", 1).is_err());
2254        assert!(store.seal(&[msg("user", json!("next"))], "").is_err());
2255        assert!(store.forget(&reference.id).is_err());
2256    }
2257
2258    #[test]
2259    fn private_modes_symlink_hardlink_fifo_and_lock_refusal() {
2260        let (tmp, store) = setup();
2261        let dir = directory(&tmp, &store);
2262        let reference = store.seal(&[msg("user", json!("private"))], "").unwrap();
2263        for path in [
2264            tmp.path().to_path_buf(),
2265            tmp.path().join("context-archives"),
2266            dir.clone(),
2267        ] {
2268            assert_eq!(
2269                std::fs::metadata(path).unwrap().permissions().mode() & 0o777,
2270                0o700
2271            );
2272        }
2273        for name in ["000.json", "lock"] {
2274            assert_eq!(
2275                std::fs::metadata(dir.join(name))
2276                    .unwrap()
2277                    .permissions()
2278                    .mode()
2279                    & 0o777,
2280                0o600
2281            );
2282        }
2283        let lock = store.fs.lock().unwrap();
2284        assert_eq!(
2285            store.fetch(&reference.id, 0, 1).unwrap_err().kind(),
2286            io::ErrorKind::WouldBlock
2287        );
2288        drop(lock);
2289        let victim = tmp.path().join("synthetic-victim");
2290        std::fs::write(&victim, "do not change").unwrap();
2291        std::fs::remove_file(dir.join("000.json")).unwrap();
2292        symlink(&victim, dir.join("000.json")).unwrap();
2293        assert!(store.search("", 1).is_err());
2294        assert!(store.seal(&[msg("user", json!("private"))], "").is_err());
2295        assert_eq!(std::fs::read_to_string(&victim).unwrap(), "do not change");
2296        std::fs::remove_file(dir.join("000.json")).unwrap();
2297        std::fs::hard_link(&victim, dir.join("000.json")).unwrap();
2298        assert!(store.search("", 1).is_err());
2299        std::fs::remove_file(dir.join("000.json")).unwrap();
2300        let fifo = std::ffi::CString::new(dir.join("000.json").to_str().unwrap()).unwrap();
2301        assert_eq!(unsafe { libc::mkfifo(fifo.as_ptr(), 0o600) }, 0);
2302        assert!(store.search("", 1).is_err());
2303        let alias = tmp.path().join("alias");
2304        symlink(tmp.path().join("context-archives"), &alias).unwrap();
2305        assert!(ArchiveStore::new(&alias.join("nested"), "p", "l").is_err());
2306        std::fs::remove_file(dir.join("lock")).unwrap();
2307        symlink(&victim, dir.join("lock")).unwrap();
2308        assert!(store.fs.lock().is_err());
2309    }
2310
2311    #[test]
2312    fn multi_mib_context_roundtrips_without_truncation() {
2313        let (_tmp, store) = setup();
2314        let text = "eligible exact evidence\n".repeat(400_000);
2315        assert!(text.len() > 8 * 1024 * 1024);
2316        let messages = vec![msg("user", json!(text))];
2317        let reference = store.seal(&messages, "large source window").unwrap();
2318        assert_eq!(store.fetch(&reference.id, 0, 1).unwrap(), messages);
2319        assert_eq!(store.seal(&messages, "retry").unwrap(), reference);
2320    }
2321
2322    #[test]
2323    fn private_reasoning_disguised_as_assistant_text_is_omitted() {
2324        let (_tmp, store) = setup();
2325        let messages = vec![
2326            Arc::new(json!({"role":"assistant","channel":"analysis","content":"private analysis"})),
2327            Arc::new(
2328                json!({"role":"assistant","content_class":"private_reasoning","content":"private reasoning"}),
2329            ),
2330            Arc::new(json!({"role":"assistant","private":true,"content":"private text"})),
2331            Arc::new(json!({"role":"assistant","channel":"final","content":"public evidence"})),
2332        ];
2333        let reference = store.seal(&messages, "").unwrap();
2334        assert_eq!(reference.message_count, 1);
2335        let fetched = store.fetch_with_provenance(&reference.id, 0, 1).unwrap();
2336        assert_eq!(fetched[0].source_index, 3);
2337        assert_eq!(fetched[0].message["content"], "public evidence");
2338    }
2339
2340    #[test]
2341    fn after_redaction_requires_host_redactor_including_parent_class() {
2342        let (_tmp, store) = setup();
2343        let messages = vec![Arc::new(json!({
2344            "role":"assistant", "disclosure":"model_visible_after_redaction",
2345            "content":[{"type":"text", "text":"sensitive host classified material"}]
2346        }))];
2347        let reference = store.seal(&messages, "").unwrap();
2348        let fetched = store.fetch(&reference.id, 0, 1).unwrap();
2349        assert_eq!(fetched[0]["content"][0]["text"], WITHHELD);
2350    }
2351
2352    #[test]
2353    fn bounded_search_utf8_and_permanent_capacity() {
2354        let (_tmp, store) = setup();
2355        for index in 0..MAX_SEGMENTS {
2356            store
2357                .seal(
2358                    &[msg("user", json!(format!("{index} {}", "é".repeat(200))))],
2359                    "",
2360                )
2361                .unwrap();
2362        }
2363        let descriptors = store.search("", MAX_SEARCH_RESULTS).unwrap();
2364        assert_eq!(descriptors.len(), MAX_SEARCH_RESULTS);
2365        assert!(descriptors.iter().all(|d| d.snippet.len() <= 256));
2366        assert!(store.seal(&[msg("user", json!("overflow"))], "").is_err());
2367        store.forget(&descriptors[0].id).unwrap();
2368        assert!(store.seal(&[msg("user", json!("still full"))], "").is_err());
2369    }
2370}
2371
2372#[cfg(all(test, unix))]
2373mod lock_tests {
2374    use super::*;
2375    use std::time::{Duration, Instant};
2376
2377    fn msg(role: &str, content: Value) -> SharedMessage {
2378        Arc::new(json!({"role":role,"content":content}))
2379    }
2380    fn setup() -> (tempfile::TempDir, ArchiveStore) {
2381        let tmp = tempfile::tempdir().unwrap();
2382        let store =
2383            ArchiveStore::new(tmp.path(), "synthetic-project", "synthetic-logical").unwrap();
2384        (tmp, store)
2385    }
2386
2387    /// Wait (bounded) for a condition another thread establishes.
2388    fn wait_until(condition: impl Fn() -> bool) {
2389        let started = Instant::now();
2390        while !condition() {
2391            assert!(started.elapsed() < Duration::from_secs(30), "timed out");
2392            std::thread::sleep(Duration::from_millis(1));
2393        }
2394    }
2395
2396    /// The process-local gate of `store`'s archive directory.
2397    fn gate_of(store: &ArchiveStore) -> Arc<Gate> {
2398        Gate::for_directory(store.fs.identity().unwrap())
2399    }
2400
2401    /// A flock belongs to the open file description. A child forked by another
2402    /// thread holds copies of every descriptor (O_CLOEXEC applies only at
2403    /// exec), so closing our copy alone does not release the lock. The child
2404    /// stays between fork and exec until the parent has fetched.
2405    #[test]
2406    fn a_descriptor_inherited_by_a_forked_child_does_not_keep_the_archive_locked() {
2407        use std::io::{Read, Write};
2408        use std::os::fd::AsRawFd;
2409        use std::os::unix::process::CommandExt;
2410        use std::process::ExitStatus;
2411        use std::thread::JoinHandle;
2412        /// Releases the child and reaps it on every path, unwinding included.
2413        struct Child {
2414            release: Option<io::PipeWriter>,
2415            spawner: Option<JoinHandle<io::Result<ExitStatus>>>,
2416        }
2417        impl Child {
2418            fn release_and_reap(&mut self) -> Option<std::thread::Result<io::Result<ExitStatus>>> {
2419                // The byte, or failing that EOF, ends the child's wait.
2420                if let Some(mut release) = self.release.take() {
2421                    let _ = release.write_all(b"r");
2422                }
2423                // The spawner returns once the child has exited and is reaped.
2424                self.spawner.take().map(JoinHandle::join)
2425            }
2426        }
2427        impl Drop for Child {
2428            fn drop(&mut self) {
2429                let _ = self.release_and_reap();
2430            }
2431        }
2432        let (_tmp, store) = setup();
2433        let reference = store
2434            .seal(&[msg("user", json!("evidence"))], "note")
2435            .unwrap();
2436        // Close-on-exec, like every descriptor the archive opens.
2437        let (mut forked, signal) = io::pipe().unwrap();
2438        let (hold, release) = io::pipe().unwrap();
2439        let (signal_fd, hold_fd) = (signal.as_raw_fd(), hold.as_raw_fd());
2440        let held = store.fs.lock().unwrap();
2441        let mut child = Child {
2442            release: Some(release),
2443            spawner: Some(std::thread::spawn(move || {
2444                let mut command = std::process::Command::new("true");
2445                // Only async-signal-safe calls between fork and exec.
2446                unsafe {
2447                    command.pre_exec(move || {
2448                        let interrupted =
2449                            || io::Error::last_os_error().raw_os_error() == Some(libc::EINTR);
2450                        loop {
2451                            match libc::write(signal_fd, b"f".as_ptr().cast(), 1) {
2452                                1 => break,
2453                                -1 if interrupted() => {}
2454                                // EOF on the parent's side fails the test.
2455                                _ => libc::_exit(1),
2456                            }
2457                        }
2458                        let mut byte = 0u8;
2459                        while libc::read(hold_fd, (&mut byte as *mut u8).cast(), 1) < 0
2460                            && interrupted()
2461                        {}
2462                        Ok(())
2463                    });
2464                }
2465                let status = command.status();
2466                drop((signal, hold));
2467                status
2468            })),
2469        };
2470        // Bounded: a child that never signals fails the test, and the guard
2471        // releases and reaps it.
2472        let deadline = Instant::now() + Duration::from_secs(30);
2473        loop {
2474            let mut ready = libc::pollfd {
2475                fd: forked.as_raw_fd(),
2476                events: libc::POLLIN,
2477                revents: 0,
2478            };
2479            let left = deadline.saturating_duration_since(Instant::now());
2480            let timeout = left.as_millis().try_into().unwrap_or(i32::MAX);
2481            match unsafe { libc::poll(&mut ready, 1, timeout) } {
2482                1 => break,
2483                0 => panic!("the forked child never signaled"),
2484                _ => assert_eq!(io::Error::last_os_error().raw_os_error(), Some(libc::EINTR)),
2485            }
2486        }
2487        forked.read_exact(&mut [0; 1]).unwrap();
2488        drop(held);
2489        // The child is blocked before exec, holding an inherited copy.
2490        let fetched = store.fetch(&reference.id, 0, 1);
2491        let status = child.release_and_reap().unwrap().unwrap().unwrap();
2492        assert!(status.success());
2493        assert_eq!(fetched.unwrap().len(), 1);
2494    }
2495
2496    /// After waiting, an operation past its deadline is refused even when the
2497    /// gate has become free by the time it rechecks.
2498    #[test]
2499    fn an_expired_waiter_is_not_admitted_when_the_gate_frees_late() {
2500        let gate = Gate::for_directory((u64::MAX, u64::MAX));
2501        let holder = Gate::enter(gate.clone(), Duration::ZERO, None, BUSY).unwrap();
2502        let wait = Duration::from_millis(50);
2503        let waiter = {
2504            let gate = gate.clone();
2505            std::thread::spawn(move || Gate::enter(gate, wait, None, BUSY).map(drop))
2506        };
2507        wait_until(|| gate.waited.load(Ordering::Acquire) == 1);
2508        {
2509            // Free the gate but keep the waiter from rechecking until its
2510            // deadline has certainly passed.
2511            let mut held = gate.held.lock().unwrap();
2512            *held = false;
2513            std::thread::sleep(wait * 2);
2514        }
2515        let error = waiter.join().unwrap().unwrap_err();
2516        assert_eq!(error.kind(), io::ErrorKind::WouldBlock);
2517        assert_eq!(error.to_string(), BUSY);
2518        drop(holder);
2519        // An uncontended first attempt is still admitted at once.
2520        Gate::enter(gate, Duration::ZERO, None, BUSY).unwrap();
2521    }
2522
2523    #[test]
2524    fn concurrent_operations_in_one_process_wait_instead_of_reporting_busy() {
2525        let tmp = tempfile::tempdir().unwrap();
2526        let workers = 8;
2527        let barrier = Arc::new(std::sync::Barrier::new(workers));
2528        let handles: Vec<_> = (0..workers)
2529            .map(|worker| {
2530                let base = tmp.path().to_path_buf();
2531                let barrier = barrier.clone();
2532                std::thread::spawn(move || {
2533                    let logical = format!("synthetic-logical-{worker}");
2534                    let store = ArchiveStore::new(&base, "synthetic-project", &logical)?;
2535                    barrier.wait();
2536                    for round in 0..8 {
2537                        let text = format!("worker {worker} round {round}");
2538                        let reference = store.seal(&[msg("user", json!(text))], "note")?;
2539                        assert_eq!(store.fetch(&reference.id, 0, 1)?.len(), 1);
2540                        assert_eq!(store.fetch_note(&reference.id)?, "note");
2541                        store.search("worker", 4)?;
2542                    }
2543                    Ok::<_, io::Error>(())
2544                })
2545            })
2546            .collect();
2547        let results: Vec<_> = handles.into_iter().map(|h| h.join().unwrap()).collect();
2548        for result in results {
2549            result.unwrap();
2550        }
2551        let store = ArchiveStore::new(tmp.path(), "synthetic-project", "synthetic-reader").unwrap();
2552        assert_eq!(store.scan().unwrap().0.len(), workers * 8);
2553    }
2554
2555    /// A separate open file description is what another process holds.
2556    #[test]
2557    fn a_lock_held_outside_this_process_still_fails_closed_without_waiting() {
2558        let (_tmp, store) = setup();
2559        let reference = store
2560            .seal(&[msg("user", json!("evidence"))], "note")
2561            .unwrap();
2562        let foreign = store.fs.lock().unwrap();
2563        let started = Instant::now();
2564        let results = [
2565            store
2566                .seal(&[msg("user", json!("other evidence"))], "")
2567                .map(drop),
2568            store.fetch(&reference.id, 0, 1).map(drop),
2569            store.fetch_note(&reference.id).map(drop),
2570            store.search("", 1).map(drop),
2571            store.forget(&reference.id),
2572        ];
2573        for result in results {
2574            assert_eq!(result.unwrap_err().kind(), io::ErrorKind::WouldBlock);
2575        }
2576        assert!(started.elapsed() < Duration::from_secs(10));
2577        drop(foreign);
2578        assert_eq!(store.fetch(&reference.id, 0, 1).unwrap().len(), 1);
2579        assert_eq!(store.scan().unwrap().0.len(), 1);
2580    }
2581
2582    fn other(tmp: &tempfile::TempDir) -> ArchiveStore {
2583        ArchiveStore::new(tmp.path(), "synthetic-project", "synthetic-other").unwrap()
2584    }
2585
2586    #[test]
2587    fn an_operation_queues_behind_the_one_in_progress_in_this_process() {
2588        let (tmp, store) = setup();
2589        let held = store.exclusive().unwrap();
2590        let queued = other(&tmp);
2591        let waiter =
2592            std::thread::spawn(move || queued.seal(&[msg("user", json!("queued evidence"))], ""));
2593        let gate = gate_of(&store);
2594        wait_until(|| gate.waited.load(Ordering::Acquire) == 1);
2595        assert!(!waiter.is_finished());
2596        drop(held);
2597        let reference = waiter.join().unwrap().unwrap();
2598        assert_eq!(store.fetch(&reference.id, 0, 1).unwrap().len(), 1);
2599    }
2600
2601    #[test]
2602    fn a_stuck_operation_in_this_process_fails_closed_after_the_bounded_wait() {
2603        let (tmp, store) = setup();
2604        let held = store.exclusive().unwrap();
2605        let mut queued = other(&tmp);
2606        queued.wait = Duration::from_millis(50);
2607        let started = Instant::now();
2608        let error = queued.search("", 1).unwrap_err();
2609        assert_eq!(error.kind(), io::ErrorKind::WouldBlock);
2610        assert_eq!(error.to_string(), BUSY);
2611        assert!(started.elapsed() >= queued.wait);
2612        drop(held);
2613        assert!(queued.search("", 1).unwrap().is_empty());
2614    }
2615
2616    #[test]
2617    fn an_aborted_operation_never_starts_even_while_queued() {
2618        let (tmp, store) = setup();
2619        let abort = Arc::new(AtomicBool::new(true));
2620        let error = other(&tmp)
2621            .with_abort(abort.clone())
2622            .seal(&[msg("user", json!("abandoned before start"))], "")
2623            .unwrap_err();
2624        assert_eq!(error.kind(), io::ErrorKind::Interrupted);
2625
2626        abort.store(false, Ordering::Release);
2627        let held = store.exclusive().unwrap();
2628        let queued = other(&tmp).with_abort(abort.clone());
2629        let waiter = std::thread::spawn(move || {
2630            queued.seal(&[msg("user", json!("abandoned while queued"))], "")
2631        });
2632        let gate = gate_of(&store);
2633        wait_until(|| gate.waited.load(Ordering::Acquire) == 1);
2634        assert!(!waiter.is_finished());
2635        abort.store(true, Ordering::Release);
2636        // Returns while the earlier operation still holds the archive.
2637        let error = waiter.join().unwrap().unwrap_err();
2638        assert_eq!(error.kind(), io::ErrorKind::Interrupted);
2639        drop(held);
2640        assert!(store.scan().unwrap().0.is_empty());
2641        assert!(store.search("", 1).unwrap().is_empty());
2642    }
2643
2644    #[test]
2645    fn seal_with_note_returns_the_note_stored_with_the_segment() {
2646        let (_tmp, store) = setup();
2647        let messages = [msg("user", json!("evidence"))];
2648        let (first, note) = store.seal_with_note(&messages, "first note").unwrap();
2649        assert_eq!(note, "first note");
2650        let (again, note) = store.seal_with_note(&messages, "second note").unwrap();
2651        assert_eq!(again, first);
2652        assert_eq!(note, "first note");
2653        assert_eq!(store.fetch_note(&first.id).unwrap(), "first note");
2654        store.forget(&first.id).unwrap();
2655        let error = store.seal_with_note(&messages, "").unwrap_err();
2656        assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
2657    }
2658}