1use std::{
9 collections::HashMap,
10 path::{Path, PathBuf},
11 sync::{Arc, Mutex, OnceLock, Weak},
12};
13
14use anyhow::{Context as _, bail, ensure};
15use kcode_checksummed_frame_log::{
16 AppendDurability, IncompleteTailPolicy, append as append_frame, create as create_frame_log,
17 visit_frames,
18};
19use kcode_pending_object_store::{PendingObjectStore, StoredPendingObject};
20use serde::{Deserialize, Serialize};
21
22pub const FORMAT_VERSION: &str = "0.2.1";
23
24const SESSION_MAGIC: &[u8] = b"KSESSIONLOG\n";
25const HEADER_FRAME: u8 = 1;
26const EVENT_FRAME: u8 = 2;
27const SEALED_FRAME: u8 = 3;
28
29#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
30#[serde(rename_all = "camelCase")]
31pub struct SessionHeader {
32 pub format_version: String,
33 pub session_id: String,
34 pub created_at: String,
35}
36
37#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
38#[serde(rename_all = "kebab-case")]
39pub enum Role {
40 SystemMessage,
41 SystemError,
42 UserMessage,
43 KennedyMessage,
44 KennedyToolCall,
45 ToolResult,
46 ToolError,
47 Object,
48 PendingObject,
49}
50
51#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
52pub struct SessionEvent {
53 pub role: Role,
54 pub text: String,
55}
56
57#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
58pub struct SessionLog {
59 pub header: SessionHeader,
60 pub events: Vec<SessionEvent>,
61}
62
63#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
64pub struct EventPosition(pub u64);
65
66impl EventPosition {
67 pub fn index(self) -> u64 {
68 self.0
69 }
70}
71
72impl std::fmt::Display for EventPosition {
73 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
74 self.0.fmt(formatter)
75 }
76}
77
78#[derive(Clone, Debug, Eq, PartialEq)]
79pub struct PendingObject {
80 pub event_position: EventPosition,
81 pub text: String,
82 pub file_name: String,
83 pub media_type: String,
84 pub bytes: Vec<u8>,
85}
86
87#[derive(Clone, Debug, Eq, PartialEq)]
88pub struct SealedSession {
89 log: SessionLog,
90 directory: PathBuf,
91}
92
93impl SealedSession {
94 pub fn list(&self) -> &SessionLog {
95 &self.log
96 }
97
98 pub fn pending_objects(&self) -> anyhow::Result<Vec<PendingObject>> {
99 let store = pending_object_store(&self.directory, &self.log.header.session_id)?;
100 self.log
101 .events
102 .iter()
103 .enumerate()
104 .filter(|(_, event)| event.role == Role::PendingObject)
105 .map(|(position, event)| {
106 read_pending_object(&store, EventPosition(position as u64), event.text.clone())
107 })
108 .collect()
109 }
110}
111
112#[derive(Clone, Debug)]
113pub struct SessionStore {
114 directory: PathBuf,
115}
116
117impl SessionStore {
118 pub fn new(directory: impl Into<PathBuf>) -> Self {
119 Self {
120 directory: directory.into(),
121 }
122 }
123
124 pub fn directory(&self) -> &Path {
125 &self.directory
126 }
127
128 pub fn create_session(
129 &self,
130 session_id: impl Into<String>,
131 created_at: impl Into<String>,
132 ) -> anyhow::Result<Session> {
133 let session_id = session_id.into();
134 let created_at = created_at.into();
135 validate_session_id(&session_id)?;
136 ensure!(
137 !created_at.trim().is_empty(),
138 "session creation time cannot be empty"
139 );
140 std::fs::create_dir_all(&self.directory)
141 .with_context(|| format!("creating {}", self.directory.display()))?;
142 let path = session_path(&self.directory, &session_id);
143 let append_lock = session_lock(&path);
144 let _guard = append_lock
145 .lock()
146 .map_err(|_| anyhow::anyhow!("session-log append lock is poisoned"))?;
147 let header = SessionHeader {
148 format_version: FORMAT_VERSION.into(),
149 session_id,
150 created_at,
151 };
152 create_frame_log(
153 &path,
154 SESSION_MAGIC,
155 HEADER_FRAME,
156 &serde_json::to_vec(&header)?,
157 )?;
158 drop(_guard);
159 Ok(Session {
160 directory: self.directory.clone(),
161 path,
162 log: SessionLog {
163 header,
164 events: Vec::new(),
165 },
166 sealed: false,
167 append_lock,
168 })
169 }
170
171 pub fn open_session(&self, session_id: &str) -> anyhow::Result<Session> {
172 validate_session_id(session_id)?;
173 let path = session_path(&self.directory, session_id);
174 let append_lock = session_lock(&path);
175 let _guard = append_lock
176 .lock()
177 .map_err(|_| anyhow::anyhow!("session-log append lock is poisoned"))?;
178 let loaded = load_session_file(&path, true)?;
179 ensure!(
180 loaded.log.header.session_id == session_id,
181 "session filename and header identity differ"
182 );
183 let object_store = pending_object_store(&self.directory, &loaded.log.header.session_id)?;
184 object_store.reconcile(&referenced_pending_positions(&loaded.log))?;
185 drop(_guard);
186 Ok(Session {
187 directory: self.directory.clone(),
188 path,
189 log: loaded.log,
190 sealed: loaded.sealed,
191 append_lock,
192 })
193 }
194
195 pub fn session_ids(&self) -> anyhow::Result<Vec<String>> {
196 if !self.directory.exists() {
197 return Ok(Vec::new());
198 }
199 let mut ids = std::fs::read_dir(&self.directory)?
200 .filter_map(Result::ok)
201 .filter_map(|entry| {
202 entry
203 .file_name()
204 .to_str()
205 .and_then(|name| name.strip_suffix(".session-log"))
206 .map(str::to_owned)
207 })
208 .filter(|id| validate_session_id(id).is_ok())
209 .collect::<Vec<_>>();
210 ids.sort();
211 Ok(ids)
212 }
213}
214
215pub struct Session {
216 directory: PathBuf,
217 path: PathBuf,
218 log: SessionLog,
219 sealed: bool,
220 append_lock: Arc<Mutex<()>>,
221}
222
223impl Session {
224 pub fn path(&self) -> &Path {
225 &self.path
226 }
227
228 pub fn list(&self) -> SessionLog {
229 self.log.clone()
230 }
231
232 pub fn is_sealed(&self) -> bool {
233 self.sealed
234 }
235
236 pub fn add_event(
237 &mut self,
238 role: Role,
239 text: impl Into<String>,
240 ) -> anyhow::Result<EventPosition> {
241 ensure!(
242 role != Role::PendingObject,
243 "pending objects must be added through add_pending_object"
244 );
245 let event = SessionEvent {
246 role,
247 text: text.into(),
248 };
249 let append_lock = self.append_lock.clone();
250 let _guard = append_lock
251 .lock()
252 .map_err(|_| anyhow::anyhow!("session-log append lock is poisoned"))?;
253 self.refresh_locked()?;
254 ensure!(!self.sealed, "session is sealed");
255 let position = EventPosition(self.log.events.len() as u64);
256 append_event_file(&self.path, &event)?;
257 self.log.events.push(event);
258 Ok(position)
259 }
260
261 pub fn add_pending_object(
262 &mut self,
263 text: impl Into<String>,
264 file_name: impl Into<String>,
265 media_type: impl Into<String>,
266 bytes: &[u8],
267 ) -> anyhow::Result<EventPosition> {
268 let text = text.into();
269 let file_name = file_name.into();
270 let media_type = media_type.into();
271 ensure!(
272 !file_name.trim().is_empty(),
273 "object filename cannot be empty"
274 );
275 ensure!(
276 !media_type.trim().is_empty(),
277 "object media type cannot be empty"
278 );
279 let append_lock = self.append_lock.clone();
280 let _guard = append_lock
281 .lock()
282 .map_err(|_| anyhow::anyhow!("session-log append lock is poisoned"))?;
283 self.refresh_locked()?;
284 ensure!(!self.sealed, "session is sealed");
285 let position = EventPosition(self.log.events.len() as u64);
286 let object_store = pending_object_store(&self.directory, &self.log.header.session_id)?;
287 object_store.install(position.0, &file_name, &media_type, bytes)?;
288 let event = SessionEvent {
289 role: Role::PendingObject,
290 text,
291 };
292 append_event_file(&self.path, &event)?;
293 self.log.events.push(event);
294 Ok(position)
295 }
296
297 pub fn read_pending_object(&self, position: EventPosition) -> anyhow::Result<PendingObject> {
298 let index = usize::try_from(position.0).context("event position does not fit memory")?;
299 let event = self
300 .log
301 .events
302 .get(index)
303 .with_context(|| format!("event {position} does not exist"))?;
304 ensure!(
305 event.role == Role::PendingObject,
306 "event {position} is not a pending object"
307 );
308 let store = pending_object_store(&self.directory, &self.log.header.session_id)?;
309 read_pending_object(&store, position, event.text.clone())
310 }
311
312 pub fn seal(&mut self) -> anyhow::Result<SealedSession> {
313 let append_lock = self.append_lock.clone();
314 let _guard = append_lock
315 .lock()
316 .map_err(|_| anyhow::anyhow!("session-log append lock is poisoned"))?;
317 self.refresh_locked()?;
318 if !self.sealed {
319 verify_referenced_objects(&self.directory, &self.log)?;
320 append_frame(
321 &self.path,
322 SEALED_FRAME,
323 &[],
324 AppendDurability::FileAndParentDirectory,
325 )?;
326 self.sealed = true;
327 }
328 Ok(SealedSession {
329 log: self.log.clone(),
330 directory: self.directory.clone(),
331 })
332 }
333
334 pub fn delete_committed(self) -> anyhow::Result<()> {
335 self.delete_files()
336 }
337
338 pub fn delete_abandoned(self) -> anyhow::Result<()> {
339 self.delete_files()
340 }
341
342 fn refresh_locked(&mut self) -> anyhow::Result<()> {
343 let loaded = load_session_file(&self.path, true)?;
344 ensure!(
345 loaded.log.header.session_id == self.log.header.session_id,
346 "session identity changed on disk"
347 );
348 self.log = loaded.log;
349 self.sealed = loaded.sealed;
350 Ok(())
351 }
352
353 fn delete_files(self) -> anyhow::Result<()> {
354 let append_lock = self.append_lock.clone();
355 let _guard = append_lock
356 .lock()
357 .map_err(|_| anyhow::anyhow!("session-log append lock is poisoned"))?;
358 let object_store = pending_object_store(&self.directory, &self.log.header.session_id)?;
359 if self.path.exists() {
360 std::fs::remove_file(&self.path)
361 .with_context(|| format!("removing {}", self.path.display()))?;
362 }
363 object_store.delete_all()
364 }
365}
366
367struct LoadedSession {
368 log: SessionLog,
369 sealed: bool,
370}
371
372struct SessionFrameVisitor {
373 header: Option<SessionHeader>,
374 events: Vec<SessionEvent>,
375 sealed: bool,
376 frame_count: u64,
377}
378
379impl SessionFrameVisitor {
380 fn new() -> Self {
381 Self {
382 header: None,
383 events: Vec::new(),
384 sealed: false,
385 frame_count: 0,
386 }
387 }
388
389 fn visit(&mut self, kind: u8, payload: &[u8]) -> anyhow::Result<()> {
390 match kind {
391 HEADER_FRAME => {
392 ensure!(self.frame_count == 0, "duplicate session header");
393 let value: SessionHeader = serde_json::from_slice(payload)?;
394 ensure!(
395 value.format_version == FORMAT_VERSION,
396 "unsupported session-log format {}",
397 value.format_version
398 );
399 validate_session_id(&value.session_id)?;
400 ensure!(
401 !value.created_at.trim().is_empty(),
402 "session creation time cannot be empty"
403 );
404 self.header = Some(value);
405 }
406 EVENT_FRAME => {
407 ensure!(self.header.is_some(), "session event precedes header");
408 ensure!(!self.sealed, "session event follows sealed footer");
409 self.events.push(serde_json::from_slice(payload)?);
410 }
411 SEALED_FRAME => {
412 ensure!(self.header.is_some(), "sealed footer precedes header");
413 ensure!(payload.is_empty(), "sealed footer payload must be empty");
414 ensure!(!self.sealed, "duplicate sealed footer");
415 self.sealed = true;
416 }
417 other => bail!("unknown complete session-log frame kind {other}"),
418 }
419 self.frame_count += 1;
420 Ok(())
421 }
422
423 fn finish(self) -> anyhow::Result<LoadedSession> {
424 let header = self.header.context("session log has no header")?;
425 Ok(LoadedSession {
426 log: SessionLog {
427 header,
428 events: self.events,
429 },
430 sealed: self.sealed,
431 })
432 }
433}
434
435fn validate_session_id(session_id: &str) -> anyhow::Result<()> {
436 ensure!(!session_id.is_empty(), "session ID cannot be empty");
437 ensure!(session_id.len() <= 255, "session ID exceeds 255 characters");
438 ensure!(
439 session_id
440 .bytes()
441 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')),
442 "session ID contains characters that are unsafe in filenames"
443 );
444 Ok(())
445}
446
447fn session_path(directory: &Path, session_id: &str) -> PathBuf {
448 directory.join(format!("{session_id}.session-log"))
449}
450
451fn append_event_file(path: &Path, event: &SessionEvent) -> anyhow::Result<()> {
452 append_frame(
453 path,
454 EVENT_FRAME,
455 &serde_json::to_vec(event)?,
456 AppendDurability::FileOnly,
457 )
458}
459
460fn load_session_file(path: &Path, repair_tail: bool) -> anyhow::Result<LoadedSession> {
461 let tail_policy = if repair_tail {
462 IncompleteTailPolicy::TruncateAndSync
463 } else {
464 IncompleteTailPolicy::Reject
465 };
466 let mut visitor = SessionFrameVisitor::new();
467 visit_frames(path, SESSION_MAGIC, tail_policy, |kind, payload| {
468 visitor.visit(kind, payload)
469 })?;
470 visitor.finish()
471}
472
473fn pending_object_store(directory: &Path, session_id: &str) -> anyhow::Result<PendingObjectStore> {
474 PendingObjectStore::new(directory, session_id, FORMAT_VERSION)
475}
476
477fn read_pending_object(
478 store: &PendingObjectStore,
479 position: EventPosition,
480 text: String,
481) -> anyhow::Result<PendingObject> {
482 let StoredPendingObject {
483 file_name,
484 media_type,
485 bytes,
486 } = store.read(position.0)?;
487 Ok(PendingObject {
488 event_position: position,
489 text,
490 file_name,
491 media_type,
492 bytes,
493 })
494}
495
496fn referenced_pending_positions(log: &SessionLog) -> Vec<u64> {
497 log.events
498 .iter()
499 .enumerate()
500 .filter_map(|(position, event)| {
501 (event.role == Role::PendingObject).then_some(position as u64)
502 })
503 .collect()
504}
505
506fn verify_referenced_objects(directory: &Path, log: &SessionLog) -> anyhow::Result<()> {
507 pending_object_store(directory, &log.header.session_id)?
508 .verify_all(&referenced_pending_positions(log))
509}
510
511fn session_lock(path: &Path) -> Arc<Mutex<()>> {
512 static LOCKS: OnceLock<Mutex<HashMap<PathBuf, Weak<Mutex<()>>>>> = OnceLock::new();
513 let locks = LOCKS.get_or_init(|| Mutex::new(HashMap::new()));
514 let mut locks = locks.lock().expect("session-log lock registry is poisoned");
515 locks.retain(|_, lock| lock.strong_count() > 0);
516 if let Some(lock) = locks.get(path).and_then(Weak::upgrade) {
517 return lock;
518 }
519 let lock = Arc::new(Mutex::new(()));
520 locks.insert(path.to_path_buf(), Arc::downgrade(&lock));
521 lock
522}
523
524#[cfg(test)]
525mod tests {
526 use std::{
527 fs::OpenOptions,
528 io::Write,
529 time::{SystemTime, UNIX_EPOCH},
530 };
531
532 use super::*;
533
534 const FRAME_HEADER_LEN: usize = 1 + 8 + 32;
535
536 fn directory(label: &str) -> PathBuf {
537 std::env::temp_dir().join(format!(
538 "session-log-{label}-{}-{}",
539 std::process::id(),
540 SystemTime::now()
541 .duration_since(UNIX_EPOCH)
542 .unwrap()
543 .as_nanos()
544 ))
545 }
546
547 #[test]
548 fn events_are_an_ordered_role_and_text_array_without_serialized_ids() {
549 let directory = directory("ordered");
550 let store = SessionStore::new(&directory);
551 let mut session = store
552 .create_session("session-1", "2026-07-24T00:00:00Z")
553 .unwrap();
554 assert_eq!(
555 session.add_event(Role::SystemMessage, "system").unwrap(),
556 EventPosition(0)
557 );
558 assert_eq!(
559 session.add_event(Role::UserMessage, "hello").unwrap(),
560 EventPosition(1)
561 );
562 drop(session);
563
564 let reopened = store.open_session("session-1").unwrap();
565 assert_eq!(
566 reopened.list(),
567 SessionLog {
568 header: SessionHeader {
569 format_version: "0.2.1".into(),
570 session_id: "session-1".into(),
571 created_at: "2026-07-24T00:00:00Z".into(),
572 },
573 events: vec![
574 SessionEvent {
575 role: Role::SystemMessage,
576 text: "system".into(),
577 },
578 SessionEvent {
579 role: Role::UserMessage,
580 text: "hello".into(),
581 },
582 ],
583 }
584 );
585 let serialized = serde_json::to_string(&reopened.list().events).unwrap();
586 assert!(!serialized.contains("\"id\""));
587 std::fs::remove_dir_all(directory).unwrap();
588 }
589
590 #[test]
591 fn pending_object_is_durable_before_its_event_and_uses_event_position() {
592 let directory = directory("object");
593 let store = SessionStore::new(&directory);
594 let mut session = store
595 .create_session("object-session", "2026-07-24T00:00:00Z")
596 .unwrap();
597 session.add_event(Role::UserMessage, "upload").unwrap();
598 let position = session
599 .add_pending_object("notes.txt", "notes.txt", "text/plain", b"durable bytes")
600 .unwrap();
601 assert_eq!(position, EventPosition(1));
602 assert!(directory.join("object-session-1.pending-object").exists());
603 let object = session.read_pending_object(position).unwrap();
604 assert_eq!(object.file_name, "notes.txt");
605 assert_eq!(object.media_type, "text/plain");
606 assert_eq!(object.bytes, b"durable bytes");
607 std::fs::remove_dir_all(directory).unwrap();
608 }
609
610 #[test]
611 fn open_removes_unreferenced_final_and_temporary_objects() {
612 let directory = directory("orphans");
613 let store = SessionStore::new(&directory);
614 let session = store
615 .create_session("orphan-session", "2026-07-24T00:00:00Z")
616 .unwrap();
617 drop(session);
618 std::fs::write(directory.join("orphan-session-0.pending-object"), b"orphan").unwrap();
619 std::fs::write(
620 directory.join("orphan-session-1.pending-object.tmp"),
621 b"temporary",
622 )
623 .unwrap();
624 let leading_zero = directory.join("orphan-session-00.pending-object");
625 let similar = directory.join("orphan-session-other-2.pending-object");
626 std::fs::write(&leading_zero, b"keep").unwrap();
627 std::fs::write(&similar, b"keep").unwrap();
628
629 store.open_session("orphan-session").unwrap();
630
631 assert!(!directory.join("orphan-session-0.pending-object").exists());
632 assert!(
633 !directory
634 .join("orphan-session-1.pending-object.tmp")
635 .exists()
636 );
637 assert!(leading_zero.exists());
638 assert!(similar.exists());
639 std::fs::remove_dir_all(directory).unwrap();
640 }
641
642 #[test]
643 fn incomplete_event_tail_is_discarded() {
644 let directory = directory("tail");
645 let store = SessionStore::new(&directory);
646 let mut session = store
647 .create_session("tail-session", "2026-07-24T00:00:00Z")
648 .unwrap();
649 session.add_event(Role::UserMessage, "complete").unwrap();
650 let path = session.path().to_path_buf();
651 drop(session);
652 let valid_len = std::fs::metadata(&path).unwrap().len();
653 OpenOptions::new()
654 .append(true)
655 .open(&path)
656 .unwrap()
657 .write_all(&[EVENT_FRAME, 20, 0, 0])
658 .unwrap();
659 let reopened = store.open_session("tail-session").unwrap();
660 assert_eq!(reopened.list().events.len(), 1);
661 assert_eq!(std::fs::metadata(&path).unwrap().len(), valid_len);
662 std::fs::remove_dir_all(directory).unwrap();
663 }
664
665 #[test]
666 fn checksum_invalid_complete_frame_is_corruption_not_a_recoverable_tail() {
667 let directory = directory("checksum");
668 let store = SessionStore::new(&directory);
669 let mut session = store
670 .create_session("checksum-session", "2026-07-24T00:00:00Z")
671 .unwrap();
672 session.add_event(Role::UserMessage, "complete").unwrap();
673 let path = session.path().to_path_buf();
674 drop(session);
675 let mut bytes = std::fs::read(&path).unwrap();
676 let payload_byte = SESSION_MAGIC.len() + FRAME_HEADER_LEN;
677 bytes[payload_byte] ^= 0xff;
678 std::fs::write(&path, &bytes).unwrap();
679 assert!(store.open_session("checksum-session").is_err());
680 assert_eq!(std::fs::read(&path).unwrap(), bytes);
681 std::fs::remove_dir_all(directory).unwrap();
682 }
683
684 #[test]
685 fn seal_is_durable_idempotent_and_rejects_later_events() {
686 let directory = directory("seal");
687 let store = SessionStore::new(&directory);
688 let mut session = store
689 .create_session("sealed-session", "2026-07-24T00:00:00Z")
690 .unwrap();
691 session.add_event(Role::UserMessage, "hello").unwrap();
692 session.seal().unwrap();
693 session.seal().unwrap();
694 assert!(session.add_event(Role::KennedyMessage, "too late").is_err());
695 drop(session);
696 assert!(store.open_session("sealed-session").unwrap().is_sealed());
697 std::fs::remove_dir_all(directory).unwrap();
698 }
699
700 #[test]
701 fn deletion_matches_only_exact_session_object_names() {
702 let directory = directory("delete");
703 let store = SessionStore::new(&directory);
704 let session = store.create_session("abc", "2026-07-24T00:00:00Z").unwrap();
705 let final_object = directory.join("abc-0.pending-object");
706 let temporary_object = directory.join("abc-1.pending-object.tmp");
707 let similar = directory.join("abc-other-0.pending-object");
708 let malformed = directory.join("abc-not-a-number.pending-object");
709 let leading_zero = directory.join("abc-01.pending-object");
710 std::fs::write(&final_object, b"remove").unwrap();
711 std::fs::write(&temporary_object, b"remove").unwrap();
712 std::fs::write(&similar, b"keep").unwrap();
713 std::fs::write(&malformed, b"keep").unwrap();
714 std::fs::write(&leading_zero, b"keep").unwrap();
715
716 session.delete_abandoned().unwrap();
717
718 assert!(!final_object.exists());
719 assert!(!temporary_object.exists());
720 assert!(similar.exists());
721 assert!(malformed.exists());
722 assert!(leading_zero.exists());
723 std::fs::remove_dir_all(directory).unwrap();
724 }
725
726 #[test]
727 fn frame_kinds_and_json_payloads_remain_legacy_compatible() {
728 let directory = directory("frame-mapping");
729 let store = SessionStore::new(&directory);
730 let mut session = store
731 .create_session("mapping-session", "2026-07-24T00:00:00Z")
732 .unwrap();
733 session.add_event(Role::ToolResult, "ok").unwrap();
734 session
735 .add_pending_object(
736 "pending text",
737 "blob.bin",
738 "application/octet-stream",
739 b"\0\xff",
740 )
741 .unwrap();
742 let path = session.path().to_path_buf();
743 session.seal().unwrap();
744
745 let mut frames = Vec::new();
746 visit_frames(
747 &path,
748 SESSION_MAGIC,
749 IncompleteTailPolicy::Reject,
750 |kind, payload| {
751 frames.push((kind, payload.to_vec()));
752 Ok(())
753 },
754 )
755 .unwrap();
756
757 assert_eq!(
758 frames,
759 vec![
760 (
761 HEADER_FRAME,
762 br#"{"formatVersion":"0.2.1","sessionId":"mapping-session","createdAt":"2026-07-24T00:00:00Z"}"#
763 .to_vec(),
764 ),
765 (
766 EVENT_FRAME,
767 br#"{"role":"tool-result","text":"ok"}"#.to_vec(),
768 ),
769 (
770 EVENT_FRAME,
771 br#"{"role":"pending-object","text":"pending text"}"#.to_vec(),
772 ),
773 (SEALED_FRAME, Vec::new()),
774 ]
775 );
776 std::fs::remove_dir_all(directory).unwrap();
777 }
778
779 #[test]
780 fn pending_object_sidecar_mapping_matches_extracted_store_exactly() {
781 let directory = directory("sidecar-mapping");
782 let store = SessionStore::new(&directory);
783 let mut session = store
784 .create_session("sidecar-session", "2026-07-24T00:00:00Z")
785 .unwrap();
786 let position = session
787 .add_pending_object("display text", "notes.txt", "text/plain", b"durable bytes")
788 .unwrap();
789
790 let expected_directory = directory.join("expected");
791 std::fs::create_dir(&expected_directory).unwrap();
792 PendingObjectStore::new(&expected_directory, "sidecar-session", "0.2.1")
793 .unwrap()
794 .install(0, "notes.txt", "text/plain", b"durable bytes")
795 .unwrap();
796
797 let actual_path = directory.join("sidecar-session-0.pending-object");
798 let expected_path = expected_directory.join("sidecar-session-0.pending-object");
799 assert_eq!(
800 std::fs::read(&actual_path).unwrap(),
801 std::fs::read(&expected_path).unwrap()
802 );
803 assert!(
804 !directory
805 .join("sidecar-session-0.pending-object.tmp")
806 .exists()
807 );
808 assert_eq!(
809 session.read_pending_object(position).unwrap(),
810 PendingObject {
811 event_position: EventPosition(0),
812 text: "display text".into(),
813 file_name: "notes.txt".into(),
814 media_type: "text/plain".into(),
815 bytes: b"durable bytes".to_vec(),
816 }
817 );
818 std::fs::remove_dir_all(directory).unwrap();
819 }
820
821 #[test]
822 fn opens_legacy_0_2_1_log_and_sidecar_bytes() {
823 let directory = directory("legacy");
824 std::fs::create_dir_all(&directory).unwrap();
825 let path = directory.join("legacy-session.session-log");
826 create_frame_log(
827 &path,
828 SESSION_MAGIC,
829 HEADER_FRAME,
830 br#"{"formatVersion":"0.2.1","sessionId":"legacy-session","createdAt":"2026-07-24T00:00:00Z"}"#,
831 )
832 .unwrap();
833 append_frame(
834 &path,
835 EVENT_FRAME,
836 br#"{"role":"user-message","text":"legacy event"}"#,
837 AppendDurability::FileOnly,
838 )
839 .unwrap();
840
841 let object_store = PendingObjectStore::new(&directory, "legacy-session", "0.2.1").unwrap();
842 object_store
843 .install(1, "legacy.bin", "application/octet-stream", b"legacy bytes")
844 .unwrap();
845 append_frame(
846 &path,
847 EVENT_FRAME,
848 br#"{"role":"pending-object","text":"legacy object"}"#,
849 AppendDurability::FileOnly,
850 )
851 .unwrap();
852 append_frame(
853 &path,
854 SEALED_FRAME,
855 &[],
856 AppendDurability::FileAndParentDirectory,
857 )
858 .unwrap();
859
860 let store = SessionStore::new(&directory);
861 let reopened = store.open_session("legacy-session").unwrap();
862 assert!(reopened.is_sealed());
863 assert_eq!(reopened.list().header.format_version, "0.2.1");
864 assert_eq!(
865 reopened.list().events,
866 vec![
867 SessionEvent {
868 role: Role::UserMessage,
869 text: "legacy event".into(),
870 },
871 SessionEvent {
872 role: Role::PendingObject,
873 text: "legacy object".into(),
874 },
875 ]
876 );
877 assert_eq!(
878 reopened
879 .read_pending_object(EventPosition(1))
880 .unwrap()
881 .bytes,
882 b"legacy bytes"
883 );
884 assert!(directory.join("legacy-session.session-log").exists());
885 assert!(directory.join("legacy-session-1.pending-object").exists());
886 std::fs::remove_dir_all(directory).unwrap();
887 }
888}