1use 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;
67pub const LOCK_WAIT: Duration = Duration::from_secs(30);
70const 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
76pub 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 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 pub snippet: String,
96}
97
98#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
99#[serde(deny_unknown_fields)]
100pub struct ArchivedMessage {
101 pub source_index: usize,
103 pub block_indices: Vec<usize>,
105 pub message: SharedMessage,
106}
107
108#[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
121pub 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
180pub 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
190struct Gate {
195 held: Mutex<bool>,
196 released: Condvar,
197 #[cfg(test)]
199 waited: std::sync::atomic::AtomicUsize,
200}
201
202struct Locked {
204 file: Option<FlockGuard>,
205 _gate: GateGuard,
206}
207
208struct GateGuard(Arc<Gate>);
209
210type DirIdentity = (u64, u64);
212type Gates = Mutex<HashMap<DirIdentity, Weak<Gate>>>;
213
214impl Gate {
215 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 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 self.0.released.notify_all();
293 }
294}
295
296struct 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 pub fn with_redactor(mut self, redactor: Arc<ArchiveRedactor>) -> Self {
368 self.redactor = Some(redactor);
369 self
370 }
371
372 pub fn with_abort(mut self, abort: Arc<AtomicBool>) -> Self {
379 self.abort = Some(abort);
380 self
381 }
382
383 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 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 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 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 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 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 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 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 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 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 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
805pub 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(), redactor: None,
821 abort: None,
822 wait: LOCK_WAIT,
823 }
824 .export_records()
825}
826
827pub 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
838pub 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 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
892fn 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
919fn 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 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); }
984 self.field(key, text, depth)
985 }
986
987 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 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#[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 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 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 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 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 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 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 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 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 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 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 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 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 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#[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 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 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 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 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 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 fn gate_of(store: &ArchiveStore) -> Arc<Gate> {
2398 Gate::for_directory(store.fs.identity().unwrap())
2399 }
2400
2401 #[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 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 if let Some(mut release) = self.release.take() {
2421 let _ = release.write_all(b"r");
2422 }
2423 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 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 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 _ => 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 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 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 #[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 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 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 #[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 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}