kcode_session_control_state/
lib.rs

1//! Durable typed state for one `.session-control` journal.
2//!
3//! Pure record projection and compaction semantics live in
4//! `kcode-session-control-records`. Filesystem ownership, durable append,
5//! replacement, repair observation, and deletion remain here.
6
7use std::{
8    fs::File,
9    path::{Path, PathBuf},
10};
11
12use anyhow::Context as _;
13use kcode_session_control_journal::Journal;
14use kcode_session_control_records::{compact_records, encode_update, project_records};
15
16pub use kcode_session_control_records::{
17    ControlProjection, ControlUpdate, SessionCommand, SessionRecord, SessionStopRequest,
18};
19
20const CONTROL_EXTENSION: &str = "session-control";
21
22#[derive(Clone, Copy, Debug, Eq, PartialEq)]
23pub enum OpenMode {
24    CreateNew,
25    OpenOrCreate,
26    ExistingOnly,
27}
28
29pub struct SessionControl {
30    directory: PathBuf,
31    path: PathBuf,
32    journal: Journal,
33}
34
35impl SessionControl {
36    pub fn open(
37        directory: impl AsRef<Path>,
38        session_id: &str,
39        mode: OpenMode,
40    ) -> anyhow::Result<Option<Self>> {
41        let directory = directory.as_ref().to_path_buf();
42        let path = control_path(&directory, session_id);
43        let journal = match mode {
44            OpenMode::CreateNew => Journal::create(path.clone())?,
45            OpenMode::OpenOrCreate => match Journal::open(path.clone())? {
46                Some(journal) => journal,
47                None => Journal::create(path.clone())?,
48            },
49            OpenMode::ExistingOnly => {
50                let Some(journal) = Journal::open(path.clone())? else {
51                    return Ok(None);
52                };
53                journal
54            }
55        };
56        Ok(Some(Self {
57            directory,
58            path,
59            journal,
60        }))
61    }
62
63    pub fn projection(&self) -> ControlProjection {
64        project_records(self.journal.records())
65    }
66
67    pub fn append(
68        &mut self,
69        recorded_at: impl Into<String>,
70        update: ControlUpdate,
71    ) -> anyhow::Result<ControlUpdate> {
72        let (projected, kind, value) = encode_update(update)?;
73        self.journal.append(kind, recorded_at, value)?;
74        Ok(projected)
75    }
76
77    pub fn delete(self) -> anyhow::Result<()> {
78        let Self {
79            directory,
80            path,
81            journal,
82        } = self;
83        drop(journal);
84        if path.exists() {
85            std::fs::remove_file(&path).with_context(|| format!("removing {}", path.display()))?;
86            sync_directory(&directory)?;
87        }
88        Ok(())
89    }
90
91    pub fn compact_directory(directory: impl AsRef<Path>) -> anyhow::Result<()> {
92        let directory = directory.as_ref();
93        let mut paths = std::fs::read_dir(directory)?
94            .filter_map(Result::ok)
95            .map(|entry| entry.path())
96            .filter(|path| {
97                path.extension().and_then(|value| value.to_str()) == Some(CONTROL_EXTENSION)
98            })
99            .collect::<Vec<_>>();
100        paths.sort();
101        for path in paths {
102            compact_journal(&path)?;
103        }
104        Ok(())
105    }
106}
107
108fn control_path(directory: &Path, session_id: &str) -> PathBuf {
109    directory.join(format!("{session_id}.{CONTROL_EXTENSION}"))
110}
111
112fn compact_journal(path: &Path) -> anyhow::Result<()> {
113    let original_bytes = std::fs::metadata(path)
114        .with_context(|| format!("reading metadata for {}", path.display()))?
115        .len();
116    if original_bytes >= 16 * 1024 * 1024 {
117        tracing::info!(
118            path = %path.display(),
119            original_bytes,
120            "Compacting legacy Session History control journal"
121        );
122    }
123
124    let mut journal = Journal::open(path.to_path_buf())?
125        .with_context(|| format!("session-control journal {} disappeared", path.display()))?;
126    let repaired_bytes = std::fs::metadata(path)?.len();
127    let tail_repaired = repaired_bytes != original_bytes;
128    let compacted = compact_records(journal.records())?;
129    let rewritten = compacted.is_some();
130
131    if let Some(records) = compacted {
132        journal.replace(records)?;
133    }
134
135    if rewritten || tail_repaired {
136        tracing::info!(
137            path = %path.display(),
138            original_bytes,
139            compacted_bytes = std::fs::metadata(path)?.len(),
140            "Compacted Session History control journal"
141        );
142    }
143    Ok(())
144}
145
146fn sync_directory(path: &Path) -> anyhow::Result<()> {
147    File::open(path)
148        .with_context(|| format!("opening directory {} for sync", path.display()))?
149        .sync_all()
150        .with_context(|| format!("syncing directory {}", path.display()))
151}
152
153#[cfg(test)]
154mod tests {
155    use std::{
156        fs::{self, OpenOptions},
157        io::Write as _,
158        sync::atomic::{AtomicU64, Ordering},
159        time::{SystemTime, UNIX_EPOCH},
160    };
161
162    use kcode_session_control_journal::Journal;
163    use serde_json::{Value, json};
164
165    use super::*;
166
167    const LIFECYCLE_SIDEBAND: &str = "session_lifecycle";
168    const COMMAND_SIDEBAND: &str = "session_command";
169    const STOP_SIDEBAND: &str = "session_stop";
170
171    static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);
172
173    fn root(label: &str) -> PathBuf {
174        let path = std::env::temp_dir().join(format!(
175            "kcode-session-control-state-{label}-{}-{}-{}",
176            std::process::id(),
177            SystemTime::now()
178                .duration_since(UNIX_EPOCH)
179                .unwrap()
180                .as_nanos(),
181            NEXT_ROOT.fetch_add(1, Ordering::Relaxed),
182        ));
183        fs::create_dir(&path).unwrap();
184        path
185    }
186
187    fn lifecycle(id: &str, version: i64, state: Value) -> SessionRecord {
188        SessionRecord {
189            id: id.into(),
190            phase: "active".into(),
191            started_at: "2026-08-14T00:00:00Z".into(),
192            updated_at: format!("2026-08-14T00:00:0{version}Z"),
193            state,
194            provenance_id: None,
195            version,
196            last_user_message_at: None,
197            ended_at: None,
198            ingress_failure_count: 0,
199            ingress_failures: json!([]),
200            ingress_next_attempt_at: None,
201            summary: false,
202        }
203    }
204
205    fn command(id: &str, status: &str, sequence: i64) -> SessionCommand {
206        SessionCommand {
207            id: id.into(),
208            conversation_id: "session-1".into(),
209            sequence,
210            kind: "message".into(),
211            payload: json!({"text":"hello"}),
212            status: status.into(),
213            cancel_requested: false,
214            outcome: None,
215            created_at: "2026-08-14T00:00:00Z".into(),
216            processing_started_at: None,
217            completed_at: None,
218            idempotency_id: format!("command-{id}"),
219        }
220    }
221
222    fn stop(id: &str, status: &str) -> SessionStopRequest {
223        SessionStopRequest {
224            id: id.into(),
225            session_id: "session-1".into(),
226            scope: "turn".into(),
227            status: status.into(),
228            outcome: None,
229            requested_at: "2026-08-14T00:00:00Z".into(),
230            completed_at: None,
231            idempotency_id: format!("stop-{id}"),
232        }
233    }
234
235    #[test]
236    fn opening_modes_preserve_create_and_absence_distinctions() {
237        let root = root("open-modes");
238        assert!(
239            SessionControl::open(&root, "missing", OpenMode::ExistingOnly)
240                .unwrap()
241                .is_none()
242        );
243
244        let created = SessionControl::open(&root, "new", OpenMode::CreateNew)
245            .unwrap()
246            .unwrap();
247        assert!(control_path(&root, "new").is_file());
248        assert!(SessionControl::open(&root, "new", OpenMode::CreateNew).is_err());
249        drop(created);
250
251        let opened = SessionControl::open(&root, "new", OpenMode::OpenOrCreate)
252            .unwrap()
253            .unwrap();
254        drop(opened);
255        let created_on_absence = SessionControl::open(&root, "other", OpenMode::OpenOrCreate)
256            .unwrap()
257            .unwrap();
258        assert!(control_path(&root, "other").is_file());
259        drop(created_on_absence);
260        fs::remove_dir_all(root).unwrap();
261    }
262
263    #[test]
264    fn append_return_and_reopen_retain_launch_identity() {
265        let root = root("launch-retention");
266        let mut control = SessionControl::open(&root, "session-1", OpenMode::CreateNew)
267            .unwrap()
268            .unwrap();
269        let update = ControlUpdate::Lifecycle(lifecycle(
270            "session-1",
271            1,
272            json!({
273                "sessionId":"session-1",
274                "sessionType":"conversation",
275                "launchContextNodeIds":["A1234567","B1234567"],
276                "launchProvenance":{"syntheticBootstrap":true},
277                "chatendText":"discard"
278            }),
279        ));
280
281        let returned = control.append("t1", update).unwrap();
282        let ControlUpdate::Lifecycle(returned) = returned else {
283            panic!("append changed update kind");
284        };
285        assert_eq!(
286            returned.state["launchContextNodeIds"],
287            json!(["A1234567", "B1234567"])
288        );
289        assert_eq!(
290            returned.state["launchProvenance"],
291            json!({"syntheticBootstrap":true})
292        );
293        assert!(returned.state.get("chatendText").is_none());
294
295        drop(control);
296        let reopened = SessionControl::open(&root, "session-1", OpenMode::ExistingOnly)
297            .unwrap()
298            .unwrap();
299        let restored = reopened.projection().lifecycle.unwrap();
300        assert_eq!(
301            restored.state["launchContextNodeIds"],
302            json!(["A1234567", "B1234567"])
303        );
304        assert_eq!(
305            restored.state["launchProvenance"],
306            json!({"syntheticBootstrap":true})
307        );
308        assert!(restored.state.get("chatendText").is_none());
309        drop(reopened);
310        fs::remove_dir_all(root).unwrap();
311    }
312
313    #[test]
314    fn projection_retains_latest_command_and_stop_values() {
315        let root = root("typed-projection");
316        let mut control = SessionControl::open(&root, "session-1", OpenMode::CreateNew)
317            .unwrap()
318            .unwrap();
319        control
320            .append(
321                "t1",
322                ControlUpdate::Lifecycle(lifecycle(
323                    "session-1",
324                    1,
325                    json!({"sessionType":"conversation"}),
326                )),
327            )
328            .unwrap();
329        control
330            .append("t2", ControlUpdate::Command(command("a", "pending", 1)))
331            .unwrap();
332        control
333            .append("t3", ControlUpdate::Command(command("a", "complete", 1)))
334            .unwrap();
335        control
336            .append("t4", ControlUpdate::Command(command("b", "pending", 2)))
337            .unwrap();
338        control
339            .append("t5", ControlUpdate::StopRequest(stop("s", "pending")))
340            .unwrap();
341        control
342            .append("t6", ControlUpdate::StopRequest(stop("s", "complete")))
343            .unwrap();
344
345        let projection = control.projection();
346        assert_eq!(projection.lifecycle.unwrap().version, 1);
347        assert_eq!(projection.commands.len(), 2);
348        assert_eq!(projection.commands["a"].status, "complete");
349        assert_eq!(projection.commands["b"].status, "pending");
350        assert_eq!(projection.stop_requests["s"].status, "complete");
351        drop(control);
352        fs::remove_dir_all(root).unwrap();
353    }
354
355    #[test]
356    fn directory_compaction_retains_launch_identity_and_is_idempotent() {
357        let root = root("compaction");
358        let path = control_path(&root, "session-1");
359        let mut journal = Journal::create(path.clone()).unwrap();
360        journal
361            .append("unknown-first", "t0", json!({"value":0}))
362            .unwrap();
363        journal
364            .append(
365                LIFECYCLE_SIDEBAND,
366                "t1",
367                serde_json::to_value(lifecycle(
368                    "session-1",
369                    1,
370                    json!({"sessionType":"conversation","chatendText":"old"}),
371                ))
372                .unwrap(),
373            )
374            .unwrap();
375        journal
376            .append(
377                COMMAND_SIDEBAND,
378                "t2",
379                serde_json::to_value(command("a", "pending", 1)).unwrap(),
380            )
381            .unwrap();
382        journal
383            .append(
384                LIFECYCLE_SIDEBAND,
385                "t3",
386                serde_json::to_value(lifecycle(
387                    "session-1",
388                    2,
389                    json!({
390                        "sessionType":"conversation",
391                        "launchContextNodeIds":[],
392                        "launchProvenance":{"syntheticBootstrap":true},
393                        "chatendText":"discard"
394                    }),
395                ))
396                .unwrap(),
397            )
398            .unwrap();
399        journal
400            .append(
401                COMMAND_SIDEBAND,
402                "t4",
403                serde_json::to_value(command("a", "complete", 1)).unwrap(),
404            )
405            .unwrap();
406        journal
407            .append(
408                STOP_SIDEBAND,
409                "t5",
410                serde_json::to_value(stop("s", "pending")).unwrap(),
411            )
412            .unwrap();
413        journal
414            .append("unknown-last", "t6", json!({"value":6}))
415            .unwrap();
416        journal
417            .append(
418                STOP_SIDEBAND,
419                "t7",
420                serde_json::to_value(stop("s", "complete")).unwrap(),
421            )
422            .unwrap();
423        drop(journal);
424
425        SessionControl::compact_directory(&root).unwrap();
426        let after_first = fs::read(&path).unwrap();
427
428        let compacted = SessionControl::open(&root, "session-1", OpenMode::ExistingOnly)
429            .unwrap()
430            .unwrap();
431        let projection = compacted.projection();
432        let lifecycle = projection.lifecycle.unwrap();
433        assert_eq!(lifecycle.version, 2);
434        assert_eq!(lifecycle.state["launchContextNodeIds"], json!([]));
435        assert_eq!(
436            lifecycle.state["launchProvenance"],
437            json!({"syntheticBootstrap":true})
438        );
439        assert!(lifecycle.state.get("chatendText").is_none());
440        assert_eq!(projection.commands["a"].status, "complete");
441        assert_eq!(projection.stop_requests["s"].status, "complete");
442        drop(compacted);
443
444        SessionControl::compact_directory(&root).unwrap();
445        assert_eq!(fs::read(&path).unwrap(), after_first);
446        fs::remove_dir_all(root).unwrap();
447    }
448
449    #[test]
450    fn malformed_typed_records_keep_existing_tolerance_and_errors() {
451        let root = root("malformed");
452        let path = control_path(&root, "session-1");
453        let mut journal = Journal::create(path).unwrap();
454        journal
455            .append(
456                LIFECYCLE_SIDEBAND,
457                "t1",
458                serde_json::to_value(lifecycle("session-1", 1, json!({}))).unwrap(),
459            )
460            .unwrap();
461        journal
462            .append(LIFECYCLE_SIDEBAND, "t2", json!({"not":"a lifecycle"}))
463            .unwrap();
464        journal
465            .append(COMMAND_SIDEBAND, "t3", json!({"id":"partial"}))
466            .unwrap();
467        drop(journal);
468
469        let control = SessionControl::open(&root, "session-1", OpenMode::ExistingOnly)
470            .unwrap()
471            .unwrap();
472        let projection = control.projection();
473        assert!(projection.lifecycle.is_none());
474        assert!(projection.commands.is_empty());
475        drop(control);
476        assert!(SessionControl::compact_directory(&root).is_ok());
477
478        let malformed_path = control_path(&root, "missing-id");
479        let mut malformed = Journal::create(malformed_path).unwrap();
480        malformed
481            .append(COMMAND_SIDEBAND, "t1", json!({"status":"pending"}))
482            .unwrap();
483        drop(malformed);
484        assert!(
485            SessionControl::compact_directory(&root)
486                .unwrap_err()
487                .to_string()
488                .contains("session command record has no ID")
489        );
490        fs::remove_dir_all(root).unwrap();
491    }
492
493    #[test]
494    fn opening_repairs_incomplete_tail_but_rejects_complete_corruption() {
495        let root = root("integrity");
496        let path = control_path(&root, "tail");
497        let mut control = SessionControl::open(&root, "tail", OpenMode::CreateNew)
498            .unwrap()
499            .unwrap();
500        control
501            .append(
502                "t1",
503                ControlUpdate::Lifecycle(lifecycle("tail", 1, json!({}))),
504            )
505            .unwrap();
506        drop(control);
507        let complete = fs::read(&path).unwrap();
508        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
509        file.write_all(b"incomplete tail").unwrap();
510        file.sync_all().unwrap();
511        drop(file);
512        let repaired = SessionControl::open(&root, "tail", OpenMode::ExistingOnly)
513            .unwrap()
514            .unwrap();
515        assert_eq!(repaired.projection().lifecycle.unwrap().version, 1);
516        drop(repaired);
517        assert_eq!(fs::read(&path).unwrap(), complete);
518
519        let corrupt_path = control_path(&root, "corrupt");
520        let mut corrupt = SessionControl::open(&root, "corrupt", OpenMode::CreateNew)
521            .unwrap()
522            .unwrap();
523        corrupt
524            .append(
525                "t1",
526                ControlUpdate::Lifecycle(lifecycle("corrupt", 1, json!({}))),
527            )
528            .unwrap();
529        drop(corrupt);
530        let mut bytes = fs::read(&corrupt_path).unwrap();
531        bytes[0] = if bytes[0] == b'0' { b'1' } else { b'0' };
532        fs::write(&corrupt_path, bytes).unwrap();
533        assert!(SessionControl::open(&root, "corrupt", OpenMode::ExistingOnly).is_err());
534        fs::remove_dir_all(root).unwrap();
535    }
536
537    #[test]
538    fn delete_removes_only_the_control_file() {
539        let root = root("delete");
540        let unrelated = root.join("keep.session-log");
541        fs::write(&unrelated, b"log").unwrap();
542        let control = SessionControl::open(&root, "session-1", OpenMode::CreateNew)
543            .unwrap()
544            .unwrap();
545        let path = control_path(&root, "session-1");
546        control.delete().unwrap();
547        assert!(!path.exists());
548        assert_eq!(fs::read(unrelated).unwrap(), b"log");
549        fs::remove_dir_all(root).unwrap();
550    }
551}
kcode_session_control_state/lib.rs

kcode_session_control_state/
lib.rs
