kcode_session_control_state/
lib.rs

1//! Typed state and startup compaction for one `.session-control` journal.
2//!
3//! Application policy and coordination remain with the composing session owner.
4
5use std::{
6    collections::BTreeMap,
7    fs::File,
8    path::{Path, PathBuf},
9};
10
11use anyhow::Context as _;
12use kcode_session_control_journal::{Journal, Record};
13use serde::{Deserialize, Serialize};
14use serde_json::{Map, Value};
15
16const LIFECYCLE_SIDEBAND: &str = "session_lifecycle";
17const COMMAND_SIDEBAND: &str = "session_command";
18const STOP_SIDEBAND: &str = "session_stop";
19const CONTROL_EXTENSION: &str = "session-control";
20
21#[derive(Clone, Copy, Debug, Eq, PartialEq)]
22pub enum OpenMode {
23    CreateNew,
24    OpenOrCreate,
25    ExistingOnly,
26}
27
28#[derive(Clone, Debug, Default)]
29pub struct ControlProjection {
30    pub lifecycle: Option<SessionRecord>,
31    pub commands: BTreeMap<String, SessionCommand>,
32    pub stop_requests: BTreeMap<String, SessionStopRequest>,
33}
34
35#[derive(Clone, Debug)]
36pub enum ControlUpdate {
37    Lifecycle(SessionRecord),
38    Command(SessionCommand),
39    StopRequest(SessionStopRequest),
40}
41
42impl ControlUpdate {
43    /// Projects lifecycle state to the authoritative control fields.
44    ///
45    /// Command and stop updates are returned unchanged.
46    pub fn projected(mut self) -> Self {
47        if let Self::Lifecycle(record) = &mut self {
48            record.state = control_state(&record.state);
49        }
50        self
51    }
52}
53
54#[derive(Clone, Debug, Deserialize, Serialize)]
55pub struct SessionRecord {
56    pub id: String,
57    pub phase: String,
58    pub started_at: String,
59    pub updated_at: String,
60    pub state: Value,
61    pub provenance_id: Option<String>,
62    pub version: i64,
63    pub last_user_message_at: Option<String>,
64    pub ended_at: Option<String>,
65    pub ingress_failure_count: i64,
66    pub ingress_failures: Value,
67    pub ingress_next_attempt_at: Option<String>,
68    #[serde(default, skip_serializing_if = "is_false")]
69    pub summary: bool,
70}
71
72#[derive(Clone, Debug, Deserialize, Serialize)]
73#[serde(rename_all = "camelCase")]
74pub struct SessionCommand {
75    pub id: String,
76    pub conversation_id: String,
77    pub sequence: i64,
78    pub kind: String,
79    pub payload: Value,
80    pub status: String,
81    pub cancel_requested: bool,
82    pub outcome: Option<Value>,
83    pub created_at: String,
84    pub processing_started_at: Option<String>,
85    pub completed_at: Option<String>,
86    pub idempotency_id: String,
87}
88
89#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
90#[serde(rename_all = "camelCase")]
91pub struct SessionStopRequest {
92    pub id: String,
93    pub session_id: String,
94    pub scope: String,
95    pub status: String,
96    pub outcome: Option<Value>,
97    pub requested_at: String,
98    pub completed_at: Option<String>,
99    pub idempotency_id: String,
100}
101
102pub struct SessionControl {
103    directory: PathBuf,
104    path: PathBuf,
105    journal: Journal,
106}
107
108impl SessionControl {
109    pub fn open(
110        directory: impl AsRef<Path>,
111        session_id: &str,
112        mode: OpenMode,
113    ) -> anyhow::Result<Option<Self>> {
114        let directory = directory.as_ref().to_path_buf();
115        let path = control_path(&directory, session_id);
116        let journal = match mode {
117            OpenMode::CreateNew => Journal::create(path.clone())?,
118            OpenMode::OpenOrCreate => match Journal::open(path.clone())? {
119                Some(journal) => journal,
120                None => Journal::create(path.clone())?,
121            },
122            OpenMode::ExistingOnly => {
123                let Some(journal) = Journal::open(path.clone())? else {
124                    return Ok(None);
125                };
126                journal
127            }
128        };
129        Ok(Some(Self {
130            directory,
131            path,
132            journal,
133        }))
134    }
135
136    pub fn projection(&self) -> ControlProjection {
137        project_records(self.journal.records())
138    }
139
140    pub fn append(
141        &mut self,
142        recorded_at: impl Into<String>,
143        update: ControlUpdate,
144    ) -> anyhow::Result<ControlUpdate> {
145        let update = update.projected();
146        let (kind, value) = match &update {
147            ControlUpdate::Lifecycle(record) => (
148                LIFECYCLE_SIDEBAND,
149                serde_json::to_value(record).context("encoding session lifecycle record")?,
150            ),
151            ControlUpdate::Command(command) => (
152                COMMAND_SIDEBAND,
153                serde_json::to_value(command).context("encoding session command record")?,
154            ),
155            ControlUpdate::StopRequest(request) => (
156                STOP_SIDEBAND,
157                serde_json::to_value(request).context("encoding session stop record")?,
158            ),
159        };
160        self.journal.append(kind, recorded_at, value)?;
161        Ok(update)
162    }
163
164    pub fn delete(self) -> anyhow::Result<()> {
165        let Self {
166            directory,
167            path,
168            journal,
169        } = self;
170        drop(journal);
171        if path.exists() {
172            std::fs::remove_file(&path).with_context(|| format!("removing {}", path.display()))?;
173            sync_directory(&directory)?;
174        }
175        Ok(())
176    }
177
178    pub fn compact_directory(directory: impl AsRef<Path>) -> anyhow::Result<()> {
179        let directory = directory.as_ref();
180        let mut paths = std::fs::read_dir(directory)?
181            .filter_map(Result::ok)
182            .map(|entry| entry.path())
183            .filter(|path| {
184                path.extension().and_then(|value| value.to_str()) == Some(CONTROL_EXTENSION)
185            })
186            .collect::<Vec<_>>();
187        paths.sort();
188        for path in paths {
189            compact_journal(&path)?;
190        }
191        Ok(())
192    }
193}
194
195fn is_false(value: &bool) -> bool {
196    !*value
197}
198
199fn control_path(directory: &Path, session_id: &str) -> PathBuf {
200    directory.join(format!("{session_id}.{CONTROL_EXTENSION}"))
201}
202
203fn project_records(records: &[Record]) -> ControlProjection {
204    let latest_lifecycle = records
205        .iter()
206        .rev()
207        .find(|record| record.kind == LIFECYCLE_SIDEBAND)
208        .and_then(|record| serde_json::from_value(record.value.clone()).ok());
209    let mut commands = BTreeMap::new();
210    let mut stop_requests = BTreeMap::new();
211    for record in records {
212        match record.kind.as_str() {
213            COMMAND_SIDEBAND => {
214                if let Ok(command) = serde_json::from_value::<SessionCommand>(record.value.clone())
215                {
216                    commands.insert(command.id.clone(), command);
217                }
218            }
219            STOP_SIDEBAND => {
220                if let Ok(request) =
221                    serde_json::from_value::<SessionStopRequest>(record.value.clone())
222                {
223                    stop_requests.insert(request.id.clone(), request);
224                }
225            }
226            _ => {}
227        }
228    }
229    ControlProjection {
230        lifecycle: latest_lifecycle,
231        commands,
232        stop_requests,
233    }
234}
235
236fn control_state(value: &Value) -> Value {
237    const KEYS: &[&str] = &[
238        "format",
239        "version",
240        "stateVersion",
241        "sessionId",
242        "sessionType",
243        "sourceSessionType",
244        "channel",
245        "freeTime",
246        "selfTimeIntent",
247        "orchestration",
248        "provenanceId",
249        "rustLibSessionId",
250        "rootNodeIds",
251        "referenceRootNodeIds",
252        "startedAt",
253        "pendingTurn",
254        "pendingExternalEventId",
255        "roundsUsed",
256        "providerAffinity",
257        "nextThreadResetReason",
258        "completed",
259        "sessionObjectId",
260        "commitReceipt",
261        "commitAuthor",
262        "providerModel",
263        "kwebPlan",
264        "startIdempotencyId",
265        "ingressSource",
266        "firstUserMessage",
267        "boxCount",
268        "eventCount",
269        "chatendMetadata",
270        "sessionStatus",
271        "historyIngress",
272    ];
273    let mut output = Map::new();
274    for key in KEYS {
275        if let Some(item) = value.get(*key) {
276            if *key == "commitReceipt" && item.is_null() {
277                continue;
278            }
279            let item = if *key == "historyIngress" {
280                control_state(item)
281            } else {
282                item.clone()
283            };
284            output.insert((*key).into(), item);
285        }
286    }
287    Value::Object(output)
288}
289
290fn compact_journal(path: &Path) -> anyhow::Result<()> {
291    let original_bytes = std::fs::metadata(path)
292        .with_context(|| format!("reading metadata for {}", path.display()))?
293        .len();
294    if original_bytes >= 16 * 1024 * 1024 {
295        tracing::info!(
296            path = %path.display(),
297            original_bytes,
298            "Compacting legacy Session History control journal"
299        );
300    }
301
302    let mut journal = Journal::open(path.to_path_buf())?
303        .with_context(|| format!("session-control journal {} disappeared", path.display()))?;
304    let repaired_bytes = std::fs::metadata(path)?.len();
305    let tail_repaired = repaired_bytes != original_bytes;
306    let mut latest_lifecycle = None;
307    let mut latest_commands = BTreeMap::<String, (u64, Record)>::new();
308    let mut latest_stop_requests = BTreeMap::<String, (u64, Record)>::new();
309    let mut retained_other = Vec::new();
310    let mut needs_rewrite = false;
311
312    for (sequence, mut record) in journal.records().iter().cloned().enumerate() {
313        let sequence = sequence as u64;
314        match record.kind.as_str() {
315            LIFECYCLE_SIDEBAND => {
316                if let Some(state) = record.value.get_mut("state") {
317                    let projected = control_state(state);
318                    if *state != projected {
319                        *state = projected;
320                        needs_rewrite = true;
321                    }
322                }
323                if latest_lifecycle.replace((sequence, record)).is_some() {
324                    needs_rewrite = true;
325                }
326            }
327            COMMAND_SIDEBAND => {
328                let id = record
329                    .value
330                    .get("id")
331                    .and_then(Value::as_str)
332                    .context("session command record has no ID")?
333                    .to_owned();
334                if latest_commands.insert(id, (sequence, record)).is_some() {
335                    needs_rewrite = true;
336                }
337            }
338            STOP_SIDEBAND => {
339                let id = record
340                    .value
341                    .get("id")
342                    .and_then(Value::as_str)
343                    .context("session stop record has no ID")?
344                    .to_owned();
345                if latest_stop_requests
346                    .insert(id, (sequence, record))
347                    .is_some()
348                {
349                    needs_rewrite = true;
350                }
351            }
352            _ => retained_other.push((sequence, record)),
353        }
354    }
355
356    if needs_rewrite {
357        let mut retained = retained_other;
358        retained.extend(latest_lifecycle);
359        retained.extend(latest_commands.into_values());
360        retained.extend(latest_stop_requests.into_values());
361        retained.sort_by_key(|(sequence, _)| *sequence);
362        journal.replace(retained.into_iter().map(|(_, record)| record))?;
363    }
364
365    if needs_rewrite || tail_repaired {
366        tracing::info!(
367            path = %path.display(),
368            original_bytes,
369            compacted_bytes = std::fs::metadata(path)?.len(),
370            "Compacted Session History control journal"
371        );
372    }
373    Ok(())
374}
375
376fn sync_directory(path: &Path) -> anyhow::Result<()> {
377    File::open(path)
378        .with_context(|| format!("opening directory {} for sync", path.display()))?
379        .sync_all()
380        .with_context(|| format!("syncing directory {}", path.display()))
381}
382
383#[cfg(test)]
384mod tests {
385    use std::{
386        fs::{self, OpenOptions},
387        io::Write as _,
388        sync::atomic::{AtomicU64, Ordering},
389        time::{SystemTime, UNIX_EPOCH},
390    };
391
392    use serde_json::json;
393
394    use super::*;
395
396    static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);
397
398    fn root(label: &str) -> PathBuf {
399        let path = std::env::temp_dir().join(format!(
400            "kcode-session-control-state-{label}-{}-{}-{}",
401            std::process::id(),
402            SystemTime::now()
403                .duration_since(UNIX_EPOCH)
404                .unwrap()
405                .as_nanos(),
406            NEXT_ROOT.fetch_add(1, Ordering::Relaxed),
407        ));
408        fs::create_dir(&path).unwrap();
409        path
410    }
411
412    fn lifecycle(id: &str, version: i64, state: Value) -> SessionRecord {
413        SessionRecord {
414            id: id.into(),
415            phase: "active".into(),
416            started_at: "2026-08-02T00:00:00Z".into(),
417            updated_at: format!("2026-08-02T00:00:0{version}Z"),
418            state,
419            provenance_id: None,
420            version,
421            last_user_message_at: None,
422            ended_at: None,
423            ingress_failure_count: 0,
424            ingress_failures: json!([]),
425            ingress_next_attempt_at: None,
426            summary: false,
427        }
428    }
429
430    fn command(id: &str, status: &str, sequence: i64) -> SessionCommand {
431        SessionCommand {
432            id: id.into(),
433            conversation_id: "session-1".into(),
434            sequence,
435            kind: "message".into(),
436            payload: json!({"text":"hello"}),
437            status: status.into(),
438            cancel_requested: false,
439            outcome: None,
440            created_at: "2026-08-02T00:00:00Z".into(),
441            processing_started_at: None,
442            completed_at: None,
443            idempotency_id: format!("command-{id}"),
444        }
445    }
446
447    fn stop(id: &str, status: &str) -> SessionStopRequest {
448        SessionStopRequest {
449            id: id.into(),
450            session_id: "session-1".into(),
451            scope: "turn".into(),
452            status: status.into(),
453            outcome: None,
454            requested_at: "2026-08-02T00:00:00Z".into(),
455            completed_at: None,
456            idempotency_id: format!("stop-{id}"),
457        }
458    }
459
460    #[test]
461    fn opening_modes_preserve_create_and_absence_distinctions() {
462        let root = root("open-modes");
463        assert!(
464            SessionControl::open(&root, "missing", OpenMode::ExistingOnly)
465                .unwrap()
466                .is_none()
467        );
468
469        let created = SessionControl::open(&root, "new", OpenMode::CreateNew)
470            .unwrap()
471            .unwrap();
472        assert!(control_path(&root, "new").is_file());
473        assert!(SessionControl::open(&root, "new", OpenMode::CreateNew).is_err());
474        drop(created);
475
476        let opened = SessionControl::open(&root, "new", OpenMode::OpenOrCreate)
477            .unwrap()
478            .unwrap();
479        drop(opened);
480        let created_on_absence = SessionControl::open(&root, "other", OpenMode::OpenOrCreate)
481            .unwrap()
482            .unwrap();
483        assert!(control_path(&root, "other").is_file());
484        drop(created_on_absence);
485        fs::remove_dir_all(root).unwrap();
486    }
487
488    #[test]
489    fn typed_append_projects_recursive_lifecycle_state_and_latest_values() {
490        let root = root("typed-projection");
491        let mut control = SessionControl::open(&root, "session-1", OpenMode::CreateNew)
492            .unwrap()
493            .unwrap();
494        let provider_affinity = json!({
495            "continuation": {"thread_id": "thread-1"},
496            "synchronized_event_id": 42,
497            "material_fingerprint": "material-1"
498        });
499        let persisted = control
500            .append(
501                "t1",
502                ControlUpdate::Lifecycle(lifecycle(
503                    "session-1",
504                    1,
505                    json!({
506                        "sessionType":"conversation",
507                        "firstUserMessage":"hello",
508                        "boxCount":2,
509                        "eventCount":3,
510                        "providerAffinity":provider_affinity.clone(),
511                        "nextThreadResetReason":"provider_material_changed",
512                        "chatendText":"discard",
513                        "commitReceipt":null,
514                        "historyIngress":{
515                            "completed":true,
516                            "commitReceipt":{"sessionObjectId":"A1234567"},
517                            "boxes":{"1":{"text":"discard"}},
518                            "chatendText":"discard"
519                        }
520                    }),
521                )),
522            )
523            .unwrap();
524        let ControlUpdate::Lifecycle(persisted) = persisted else {
525            panic!("append changed update kind");
526        };
527        assert_eq!(persisted.state["sessionType"], "conversation");
528        assert_eq!(persisted.state["firstUserMessage"], "hello");
529        assert_eq!(persisted.state["boxCount"], 2);
530        assert_eq!(persisted.state["eventCount"], 3);
531        assert_eq!(persisted.state["providerAffinity"], provider_affinity);
532        assert_eq!(
533            persisted.state["nextThreadResetReason"],
534            "provider_material_changed"
535        );
536        assert!(persisted.state.get("chatendText").is_none());
537        assert!(persisted.state.get("commitReceipt").is_none());
538        assert_eq!(
539            persisted.state["historyIngress"]["commitReceipt"]["sessionObjectId"],
540            "A1234567"
541        );
542        assert!(
543            persisted.state["historyIngress"]
544                .get("chatendText")
545                .is_none()
546        );
547        assert!(persisted.state["historyIngress"].get("boxes").is_none());
548
549        drop(control);
550        let mut control = SessionControl::open(&root, "session-1", OpenMode::ExistingOnly)
551            .unwrap()
552            .unwrap();
553        let restored = control.projection().lifecycle.unwrap();
554        assert_eq!(restored.state["providerAffinity"], provider_affinity);
555        assert_eq!(
556            restored.state["nextThreadResetReason"],
557            "provider_material_changed"
558        );
559
560        control
561            .append(
562                "t2",
563                ControlUpdate::Lifecycle(lifecycle(
564                    "session-1",
565                    2,
566                    json!({"sessionType":"conversation","pendingTurn":true}),
567                )),
568            )
569            .unwrap();
570        control
571            .append("t3", ControlUpdate::Command(command("a", "pending", 1)))
572            .unwrap();
573        control
574            .append("t4", ControlUpdate::Command(command("a", "complete", 1)))
575            .unwrap();
576        control
577            .append("t5", ControlUpdate::Command(command("b", "pending", 2)))
578            .unwrap();
579        control
580            .append("t6", ControlUpdate::StopRequest(stop("s", "pending")))
581            .unwrap();
582        control
583            .append("t7", ControlUpdate::StopRequest(stop("s", "complete")))
584            .unwrap();
585
586        let projection = control.projection();
587        assert_eq!(projection.lifecycle.unwrap().version, 2);
588        assert_eq!(projection.commands.len(), 2);
589        assert_eq!(projection.commands["a"].status, "complete");
590        assert_eq!(projection.commands["b"].status, "pending");
591        assert_eq!(projection.stop_requests.len(), 1);
592        assert_eq!(projection.stop_requests["s"].status, "complete");
593        drop(control);
594
595        let reopened = SessionControl::open(&root, "session-1", OpenMode::ExistingOnly)
596            .unwrap()
597            .unwrap();
598        assert_eq!(reopened.projection().lifecycle.unwrap().version, 2);
599        drop(reopened);
600        fs::remove_dir_all(root).unwrap();
601    }
602
603    #[test]
604    fn startup_compaction_retains_unknown_kinds_and_survivor_order() {
605        let root = root("compaction-order");
606        let path = control_path(&root, "session-1");
607        let mut journal = Journal::create(path.clone()).unwrap();
608        journal
609            .append("unknown-first", "t0", json!({"value":0}))
610            .unwrap();
611        journal
612            .append(
613                LIFECYCLE_SIDEBAND,
614                "t1",
615                serde_json::to_value(lifecycle(
616                    "session-1",
617                    1,
618                    json!({"sessionType":"conversation","chatendText":"old"}),
619                ))
620                .unwrap(),
621            )
622            .unwrap();
623        journal
624            .append(
625                COMMAND_SIDEBAND,
626                "t2",
627                serde_json::to_value(command("a", "pending", 1)).unwrap(),
628            )
629            .unwrap();
630        journal
631            .append("unknown-middle", "t3", json!({"value":3}))
632            .unwrap();
633        journal
634            .append(
635                LIFECYCLE_SIDEBAND,
636                "t4",
637                serde_json::to_value(lifecycle(
638                    "session-1",
639                    2,
640                    json!({
641                        "sessionType":"conversation",
642                        "pendingTurn":true,
643                        "historyIngress":{"chatendText":"discard","completed":true}
644                    }),
645                ))
646                .unwrap(),
647            )
648            .unwrap();
649        journal
650            .append(
651                STOP_SIDEBAND,
652                "t5",
653                serde_json::to_value(stop("s", "pending")).unwrap(),
654            )
655            .unwrap();
656        journal
657            .append(
658                COMMAND_SIDEBAND,
659                "t6",
660                serde_json::to_value(command("a", "complete", 1)).unwrap(),
661            )
662            .unwrap();
663        journal
664            .append("unknown-last", "t7", json!({"value":7}))
665            .unwrap();
666        journal
667            .append(
668                STOP_SIDEBAND,
669                "t8",
670                serde_json::to_value(stop("s", "complete")).unwrap(),
671            )
672            .unwrap();
673        drop(journal);
674
675        SessionControl::compact_directory(&root).unwrap();
676
677        let compacted = Journal::open(path).unwrap().unwrap();
678        let kinds = compacted
679            .records()
680            .iter()
681            .map(|record| record.kind.as_str())
682            .collect::<Vec<_>>();
683        assert_eq!(
684            kinds,
685            [
686                "unknown-first",
687                "unknown-middle",
688                LIFECYCLE_SIDEBAND,
689                COMMAND_SIDEBAND,
690                "unknown-last",
691                STOP_SIDEBAND,
692            ]
693        );
694        let projected = project_records(compacted.records());
695        assert_eq!(projected.lifecycle.as_ref().unwrap().version, 2);
696        assert_eq!(
697            projected.lifecycle.unwrap().state["historyIngress"]["completed"],
698            true
699        );
700        assert_eq!(projected.commands["a"].status, "complete");
701        assert_eq!(projected.stop_requests["s"].status, "complete");
702        drop(compacted);
703        fs::remove_dir_all(root).unwrap();
704    }
705
706    #[test]
707    fn malformed_typed_records_keep_the_existing_tolerance_and_errors() {
708        let root = root("malformed");
709        let path = control_path(&root, "session-1");
710        let mut journal = Journal::create(path.clone()).unwrap();
711        journal
712            .append(
713                LIFECYCLE_SIDEBAND,
714                "t1",
715                serde_json::to_value(lifecycle("session-1", 1, json!({}))).unwrap(),
716            )
717            .unwrap();
718        journal
719            .append(LIFECYCLE_SIDEBAND, "t2", json!({"not":"a lifecycle"}))
720            .unwrap();
721        journal
722            .append(COMMAND_SIDEBAND, "t3", json!({"id":"partial"}))
723            .unwrap();
724        drop(journal);
725
726        let control = SessionControl::open(&root, "session-1", OpenMode::ExistingOnly)
727            .unwrap()
728            .unwrap();
729        let projection = control.projection();
730        assert!(projection.lifecycle.is_none());
731        assert!(projection.commands.is_empty());
732        drop(control);
733
734        assert!(SessionControl::compact_directory(&root).is_ok());
735
736        let malformed_path = control_path(&root, "missing-id");
737        let mut malformed = Journal::create(malformed_path).unwrap();
738        malformed
739            .append(COMMAND_SIDEBAND, "t1", json!({"status":"pending"}))
740            .unwrap();
741        drop(malformed);
742        assert!(
743            SessionControl::compact_directory(&root)
744                .unwrap_err()
745                .to_string()
746                .contains("session command record has no ID")
747        );
748        fs::remove_dir_all(root).unwrap();
749    }
750
751    #[test]
752    fn opening_repairs_incomplete_tail_but_rejects_complete_corruption() {
753        let root = root("integrity");
754        let path = control_path(&root, "tail");
755        let mut control = SessionControl::open(&root, "tail", OpenMode::CreateNew)
756            .unwrap()
757            .unwrap();
758        control
759            .append(
760                "t1",
761                ControlUpdate::Lifecycle(lifecycle("tail", 1, json!({}))),
762            )
763            .unwrap();
764        drop(control);
765        let complete = fs::read(&path).unwrap();
766        let mut file = OpenOptions::new().append(true).open(&path).unwrap();
767        file.write_all(b"incomplete tail").unwrap();
768        file.sync_all().unwrap();
769        drop(file);
770        let repaired = SessionControl::open(&root, "tail", OpenMode::ExistingOnly)
771            .unwrap()
772            .unwrap();
773        assert_eq!(repaired.projection().lifecycle.unwrap().version, 1);
774        drop(repaired);
775        assert_eq!(fs::read(&path).unwrap(), complete);
776
777        let corrupt_path = control_path(&root, "corrupt");
778        let mut corrupt = SessionControl::open(&root, "corrupt", OpenMode::CreateNew)
779            .unwrap()
780            .unwrap();
781        corrupt
782            .append(
783                "t1",
784                ControlUpdate::Lifecycle(lifecycle("corrupt", 1, json!({}))),
785            )
786            .unwrap();
787        drop(corrupt);
788        let mut bytes = fs::read(&corrupt_path).unwrap();
789        bytes[0] = if bytes[0] == b'0' { b'1' } else { b'0' };
790        fs::write(&corrupt_path, bytes).unwrap();
791        assert!(SessionControl::open(&root, "corrupt", OpenMode::ExistingOnly).is_err());
792        fs::remove_dir_all(root).unwrap();
793    }
794
795    #[test]
796    fn delete_removes_only_the_control_file() {
797        let root = root("delete");
798        let unrelated = root.join("keep.session-log");
799        fs::write(&unrelated, b"log").unwrap();
800        let control = SessionControl::open(&root, "session-1", OpenMode::CreateNew)
801            .unwrap()
802            .unwrap();
803        let path = control_path(&root, "session-1");
804        control.delete().unwrap();
805        assert!(!path.exists());
806        assert_eq!(fs::read(unrelated).unwrap(), b"log");
807        fs::remove_dir_all(root).unwrap();
808    }
809}
kcode_session_control_state/lib.rs

kcode_session_control_state/
lib.rs
