kcode_session_control_records/
lib.rs

1//! Pure typed meaning for `.session-control` journal records.
2//!
3//! Filesystem durability, append synchronization, replacement, locking, and
4//! directory lifecycle remain with the composing state package.
5
6use std::collections::BTreeMap;
7
8use anyhow::Context as _;
9use kcode_session_control_journal::Record;
10use serde::{Deserialize, Serialize};
11use serde_json::{Map, Value};
12
13const LIFECYCLE_SIDEBAND: &str = "session_lifecycle";
14const COMMAND_SIDEBAND: &str = "session_command";
15const STOP_SIDEBAND: &str = "session_stop";
16
17#[derive(Clone, Debug, Default)]
18pub struct ControlProjection {
19    pub lifecycle: Option<SessionRecord>,
20    pub commands: BTreeMap<String, SessionCommand>,
21    pub stop_requests: BTreeMap<String, SessionStopRequest>,
22}
23
24#[derive(Clone, Debug)]
25pub enum ControlUpdate {
26    Lifecycle(SessionRecord),
27    Command(SessionCommand),
28    StopRequest(SessionStopRequest),
29}
30
31impl ControlUpdate {
32    /// Projects lifecycle state to authoritative control fields.
33    ///
34    /// Command and stop updates are returned unchanged.
35    pub fn projected(mut self) -> Self {
36        if let Self::Lifecycle(record) = &mut self {
37            record.state = control_state(&record.state);
38        }
39        self
40    }
41}
42
43#[derive(Clone, Debug, Deserialize, Serialize)]
44pub struct SessionRecord {
45    pub id: String,
46    pub phase: String,
47    pub started_at: String,
48    pub updated_at: String,
49    pub state: Value,
50    pub provenance_id: Option<String>,
51    pub version: i64,
52    pub last_user_message_at: Option<String>,
53    pub ended_at: Option<String>,
54    pub ingress_failure_count: i64,
55    pub ingress_failures: Value,
56    pub ingress_next_attempt_at: Option<String>,
57    #[serde(default, skip_serializing_if = "is_false")]
58    pub summary: bool,
59}
60
61#[derive(Clone, Debug, Deserialize, Serialize)]
62#[serde(rename_all = "camelCase")]
63pub struct SessionCommand {
64    pub id: String,
65    pub conversation_id: String,
66    pub sequence: i64,
67    pub kind: String,
68    pub payload: Value,
69    pub status: String,
70    pub cancel_requested: bool,
71    pub outcome: Option<Value>,
72    pub created_at: String,
73    pub processing_started_at: Option<String>,
74    pub completed_at: Option<String>,
75    pub idempotency_id: String,
76}
77
78#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
79#[serde(rename_all = "camelCase")]
80pub struct SessionStopRequest {
81    pub id: String,
82    pub session_id: String,
83    pub scope: String,
84    pub status: String,
85    pub outcome: Option<Value>,
86    pub requested_at: String,
87    pub completed_at: Option<String>,
88    pub idempotency_id: String,
89}
90
91pub fn encode_update(
92    update: ControlUpdate,
93) -> anyhow::Result<(ControlUpdate, &'static str, Value)> {
94    let update = update.projected();
95    let (kind, value) = match &update {
96        ControlUpdate::Lifecycle(record) => (
97            LIFECYCLE_SIDEBAND,
98            serde_json::to_value(record).context("encoding session lifecycle record")?,
99        ),
100        ControlUpdate::Command(command) => (
101            COMMAND_SIDEBAND,
102            serde_json::to_value(command).context("encoding session command record")?,
103        ),
104        ControlUpdate::StopRequest(request) => (
105            STOP_SIDEBAND,
106            serde_json::to_value(request).context("encoding session stop record")?,
107        ),
108    };
109    Ok((update, kind, value))
110}
111
112pub fn project_records(records: &[Record]) -> ControlProjection {
113    let latest_lifecycle = records
114        .iter()
115        .rev()
116        .find(|record| record.kind == LIFECYCLE_SIDEBAND)
117        .and_then(|record| serde_json::from_value(record.value.clone()).ok());
118    let mut commands = BTreeMap::new();
119    let mut stop_requests = BTreeMap::new();
120    for record in records {
121        match record.kind.as_str() {
122            COMMAND_SIDEBAND => {
123                if let Ok(command) = serde_json::from_value::<SessionCommand>(record.value.clone())
124                {
125                    commands.insert(command.id.clone(), command);
126                }
127            }
128            STOP_SIDEBAND => {
129                if let Ok(request) =
130                    serde_json::from_value::<SessionStopRequest>(record.value.clone())
131                {
132                    stop_requests.insert(request.id.clone(), request);
133                }
134            }
135            _ => {}
136        }
137    }
138    ControlProjection {
139        lifecycle: latest_lifecycle,
140        commands,
141        stop_requests,
142    }
143}
144
145pub fn compact_records(records: &[Record]) -> anyhow::Result<Option<Vec<Record>>> {
146    let mut latest_lifecycle = None;
147    let mut latest_commands = BTreeMap::<String, (u64, Record)>::new();
148    let mut latest_stop_requests = BTreeMap::<String, (u64, Record)>::new();
149    let mut retained_other = Vec::new();
150    let mut needs_rewrite = false;
151
152    for (sequence, mut record) in records.iter().cloned().enumerate() {
153        let sequence = sequence as u64;
154        match record.kind.as_str() {
155            LIFECYCLE_SIDEBAND => {
156                if let Some(state) = record.value.get_mut("state") {
157                    let projected = control_state(state);
158                    if *state != projected {
159                        *state = projected;
160                        needs_rewrite = true;
161                    }
162                }
163                if latest_lifecycle.replace((sequence, record)).is_some() {
164                    needs_rewrite = true;
165                }
166            }
167            COMMAND_SIDEBAND => {
168                let id = record
169                    .value
170                    .get("id")
171                    .and_then(Value::as_str)
172                    .context("session command record has no ID")?
173                    .to_owned();
174                if latest_commands.insert(id, (sequence, record)).is_some() {
175                    needs_rewrite = true;
176                }
177            }
178            STOP_SIDEBAND => {
179                let id = record
180                    .value
181                    .get("id")
182                    .and_then(Value::as_str)
183                    .context("session stop record has no ID")?
184                    .to_owned();
185                if latest_stop_requests
186                    .insert(id, (sequence, record))
187                    .is_some()
188                {
189                    needs_rewrite = true;
190                }
191            }
192            _ => retained_other.push((sequence, record)),
193        }
194    }
195
196    if !needs_rewrite {
197        return Ok(None);
198    }
199
200    let mut retained = retained_other;
201    retained.extend(latest_lifecycle);
202    retained.extend(latest_commands.into_values());
203    retained.extend(latest_stop_requests.into_values());
204    retained.sort_by_key(|(sequence, _)| *sequence);
205    Ok(Some(
206        retained.into_iter().map(|(_, record)| record).collect(),
207    ))
208}
209
210fn is_false(value: &bool) -> bool {
211    !*value
212}
213
214fn control_state(value: &Value) -> Value {
215    const KEYS: &[&str] = &[
216        "format",
217        "version",
218        "stateVersion",
219        "sessionId",
220        "sessionType",
221        "sourceSessionType",
222        "channel",
223        "freeTime",
224        "selfTimeIntent",
225        "orchestration",
226        "provenanceId",
227        "rustLibSessionId",
228        "rootNodeIds",
229        "referenceRootNodeIds",
230        "startedAt",
231        "pendingTurn",
232        "pendingExternalEventId",
233        "roundsUsed",
234        "providerAffinity",
235        "nextThreadResetReason",
236        "completed",
237        "sessionObjectId",
238        "commitReceipt",
239        "commitAuthor",
240        "providerModel",
241        "kwebPlan",
242        "startIdempotencyId",
243        "ingressSource",
244        "firstUserMessage",
245        "boxCount",
246        "eventCount",
247        "chatendMetadata",
248        "sessionStatus",
249        "launchContextNodeIds",
250        "launchProvenance",
251        "historyIngress",
252    ];
253    let mut output = Map::new();
254    for key in KEYS {
255        if let Some(item) = value.get(*key) {
256            if *key == "commitReceipt" && item.is_null() {
257                continue;
258            }
259            let item = if *key == "historyIngress" {
260                control_state(item)
261            } else {
262                item.clone()
263            };
264            output.insert((*key).into(), item);
265        }
266    }
267    Value::Object(output)
268}
269
270#[cfg(test)]
271mod tests {
272    use std::{
273        fs,
274        path::PathBuf,
275        sync::atomic::{AtomicU64, Ordering},
276        time::{SystemTime, UNIX_EPOCH},
277    };
278
279    use kcode_session_control_journal::Journal;
280    use serde_json::json;
281
282    use super::*;
283
284    static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);
285
286    fn root(label: &str) -> PathBuf {
287        let path = std::env::temp_dir().join(format!(
288            "kcode-session-control-records-{label}-{}-{}-{}",
289            std::process::id(),
290            SystemTime::now()
291                .duration_since(UNIX_EPOCH)
292                .unwrap()
293                .as_nanos(),
294            NEXT_ROOT.fetch_add(1, Ordering::Relaxed),
295        ));
296        fs::create_dir(&path).unwrap();
297        path
298    }
299
300    fn lifecycle(id: &str, version: i64, state: Value) -> SessionRecord {
301        SessionRecord {
302            id: id.into(),
303            phase: "active".into(),
304            started_at: "2026-08-02T00:00:00Z".into(),
305            updated_at: format!("2026-08-02T00:00:0{version}Z"),
306            state,
307            provenance_id: None,
308            version,
309            last_user_message_at: None,
310            ended_at: None,
311            ingress_failure_count: 0,
312            ingress_failures: json!([]),
313            ingress_next_attempt_at: None,
314            summary: false,
315        }
316    }
317
318    fn command(id: &str, status: &str, sequence: i64) -> SessionCommand {
319        SessionCommand {
320            id: id.into(),
321            conversation_id: "session-1".into(),
322            sequence,
323            kind: "message".into(),
324            payload: json!({"text":"hello"}),
325            status: status.into(),
326            cancel_requested: false,
327            outcome: None,
328            created_at: "2026-08-02T00:00:00Z".into(),
329            processing_started_at: None,
330            completed_at: None,
331            idempotency_id: format!("command-{id}"),
332        }
333    }
334
335    fn stop(id: &str, status: &str) -> SessionStopRequest {
336        SessionStopRequest {
337            id: id.into(),
338            session_id: "session-1".into(),
339            scope: "turn".into(),
340            status: status.into(),
341            outcome: None,
342            requested_at: "2026-08-02T00:00:00Z".into(),
343            completed_at: None,
344            idempotency_id: format!("stop-{id}"),
345        }
346    }
347
348    fn journal(label: &str) -> (PathBuf, Journal) {
349        let root = root(label);
350        let journal = Journal::create(root.join("records.session-control")).unwrap();
351        (root, journal)
352    }
353
354    #[test]
355    fn lifecycle_projection_retains_launch_identity_and_discards_presentation() {
356        let update = ControlUpdate::Lifecycle(lifecycle(
357            "session-1",
358            1,
359            json!({
360                "sessionType":"conversation",
361                "launchContextNodeIds":["A1234567","B1234567"],
362                "launchProvenance":{"syntheticBootstrap":true},
363                "chatendText":"discard",
364                "commitReceipt":null,
365                "historyIngress":{
366                    "launchContextNodeIds":["C1234567"],
367                    "launchProvenance":{"syntheticBootstrap":false},
368                    "completed":true,
369                    "commitReceipt":null,
370                    "boxes":{"1":{"text":"discard"}}
371                }
372            }),
373        ))
374        .projected();
375        let ControlUpdate::Lifecycle(record) = update else {
376            panic!("projection changed update kind");
377        };
378        assert_eq!(
379            record.state["launchContextNodeIds"],
380            json!(["A1234567", "B1234567"])
381        );
382        assert_eq!(
383            record.state["launchProvenance"],
384            json!({"syntheticBootstrap":true})
385        );
386        assert!(record.state.get("chatendText").is_none());
387        assert!(record.state.get("commitReceipt").is_none());
388        assert_eq!(
389            record.state["historyIngress"]["launchContextNodeIds"],
390            json!(["C1234567"])
391        );
392        assert_eq!(
393            record.state["historyIngress"]["launchProvenance"],
394            json!({"syntheticBootstrap":false})
395        );
396        assert!(
397            record.state["historyIngress"]
398                .get("commitReceipt")
399                .is_none()
400        );
401        assert!(record.state["historyIngress"].get("boxes").is_none());
402    }
403
404    #[test]
405    fn encoding_projects_lifecycle_and_preserves_command_and_stop_shapes() {
406        let lifecycle = ControlUpdate::Lifecycle(lifecycle(
407            "session-1",
408            1,
409            json!({
410                "sessionType":"conversation",
411                "launchContextNodeIds":[],
412                "chatendText":"discard"
413            }),
414        ));
415        let (projected, kind, value) = encode_update(lifecycle).unwrap();
416        assert_eq!(kind, LIFECYCLE_SIDEBAND);
417        assert_eq!(value["state"]["launchContextNodeIds"], json!([]));
418        assert!(value["state"].get("chatendText").is_none());
419        assert!(matches!(projected, ControlUpdate::Lifecycle(_)));
420
421        let command = command("a", "pending", 1);
422        let (_, kind, value) = encode_update(ControlUpdate::Command(command.clone())).unwrap();
423        assert_eq!(kind, COMMAND_SIDEBAND);
424        assert_eq!(value["conversationId"], command.conversation_id);
425
426        let stop = stop("s", "pending");
427        let (_, kind, value) = encode_update(ControlUpdate::StopRequest(stop.clone())).unwrap();
428        assert_eq!(kind, STOP_SIDEBAND);
429        assert_eq!(value["sessionId"], stop.session_id);
430    }
431
432    #[test]
433    fn projection_selects_latest_valid_typed_values() {
434        let (root, mut journal) = journal("projection");
435        journal
436            .append(
437                LIFECYCLE_SIDEBAND,
438                "t1",
439                serde_json::to_value(lifecycle("session-1", 1, json!({}))).unwrap(),
440            )
441            .unwrap();
442        journal
443            .append(
444                LIFECYCLE_SIDEBAND,
445                "t2",
446                serde_json::to_value(lifecycle("session-1", 2, json!({}))).unwrap(),
447            )
448            .unwrap();
449        journal
450            .append(
451                COMMAND_SIDEBAND,
452                "t3",
453                serde_json::to_value(command("a", "pending", 1)).unwrap(),
454            )
455            .unwrap();
456        journal
457            .append(
458                COMMAND_SIDEBAND,
459                "t4",
460                serde_json::to_value(command("a", "complete", 1)).unwrap(),
461            )
462            .unwrap();
463        journal
464            .append(
465                STOP_SIDEBAND,
466                "t5",
467                serde_json::to_value(stop("s", "pending")).unwrap(),
468            )
469            .unwrap();
470        journal
471            .append(
472                STOP_SIDEBAND,
473                "t6",
474                serde_json::to_value(stop("s", "complete")).unwrap(),
475            )
476            .unwrap();
477
478        let projection = project_records(journal.records());
479        assert_eq!(projection.lifecycle.unwrap().version, 2);
480        assert_eq!(projection.commands["a"].status, "complete");
481        assert_eq!(projection.stop_requests["s"].status, "complete");
482        drop(journal);
483        fs::remove_dir_all(root).unwrap();
484    }
485
486    #[test]
487    fn malformed_records_retain_existing_projection_tolerance() {
488        let (root, mut journal) = journal("malformed");
489        journal
490            .append(
491                LIFECYCLE_SIDEBAND,
492                "t1",
493                serde_json::to_value(lifecycle("session-1", 1, json!({}))).unwrap(),
494            )
495            .unwrap();
496        journal
497            .append(LIFECYCLE_SIDEBAND, "t2", json!({"not":"a lifecycle"}))
498            .unwrap();
499        journal
500            .append(COMMAND_SIDEBAND, "t3", json!({"id":"partial"}))
501            .unwrap();
502        journal
503            .append(STOP_SIDEBAND, "t4", json!({"id":"partial"}))
504            .unwrap();
505
506        let projection = project_records(journal.records());
507        assert!(projection.lifecycle.is_none());
508        assert!(projection.commands.is_empty());
509        assert!(projection.stop_requests.is_empty());
510        drop(journal);
511        fs::remove_dir_all(root).unwrap();
512    }
513
514    #[test]
515    fn compaction_preserves_survivor_order_unknown_kinds_and_launch_fields() {
516        let (root, mut journal) = journal("compaction");
517        journal
518            .append("unknown-first", "t0", json!({"value":0}))
519            .unwrap();
520        journal
521            .append(
522                LIFECYCLE_SIDEBAND,
523                "t1",
524                serde_json::to_value(lifecycle(
525                    "session-1",
526                    1,
527                    json!({"sessionType":"conversation","chatendText":"old"}),
528                ))
529                .unwrap(),
530            )
531            .unwrap();
532        journal
533            .append(
534                COMMAND_SIDEBAND,
535                "t2",
536                serde_json::to_value(command("a", "pending", 1)).unwrap(),
537            )
538            .unwrap();
539        journal
540            .append("unknown-middle", "t3", json!({"value":3}))
541            .unwrap();
542        journal
543            .append(
544                LIFECYCLE_SIDEBAND,
545                "t4",
546                serde_json::to_value(lifecycle(
547                    "session-1",
548                    2,
549                    json!({
550                        "sessionType":"conversation",
551                        "launchContextNodeIds":[],
552                        "launchProvenance":{"syntheticBootstrap":true},
553                        "chatendText":"discard"
554                    }),
555                ))
556                .unwrap(),
557            )
558            .unwrap();
559        journal
560            .append(
561                STOP_SIDEBAND,
562                "t5",
563                serde_json::to_value(stop("s", "pending")).unwrap(),
564            )
565            .unwrap();
566        journal
567            .append(
568                COMMAND_SIDEBAND,
569                "t6",
570                serde_json::to_value(command("a", "complete", 1)).unwrap(),
571            )
572            .unwrap();
573        journal
574            .append("unknown-last", "t7", json!({"value":7}))
575            .unwrap();
576        journal
577            .append(
578                STOP_SIDEBAND,
579                "t8",
580                serde_json::to_value(stop("s", "complete")).unwrap(),
581            )
582            .unwrap();
583
584        let compacted = compact_records(journal.records()).unwrap().unwrap();
585        let kinds = compacted
586            .iter()
587            .map(|record| record.kind.as_str())
588            .collect::<Vec<_>>();
589        assert_eq!(
590            kinds,
591            [
592                "unknown-first",
593                "unknown-middle",
594                LIFECYCLE_SIDEBAND,
595                COMMAND_SIDEBAND,
596                "unknown-last",
597                STOP_SIDEBAND,
598            ]
599        );
600        let projection = project_records(&compacted);
601        let lifecycle = projection.lifecycle.unwrap();
602        assert_eq!(lifecycle.version, 2);
603        assert_eq!(lifecycle.state["launchContextNodeIds"], json!([]));
604        assert_eq!(
605            lifecycle.state["launchProvenance"],
606            json!({"syntheticBootstrap":true})
607        );
608        assert!(lifecycle.state.get("chatendText").is_none());
609        assert_eq!(projection.commands["a"].status, "complete");
610        assert_eq!(projection.stop_requests["s"].status, "complete");
611        drop(journal);
612        fs::remove_dir_all(root).unwrap();
613    }
614
615    #[test]
616    fn compaction_reports_missing_typed_ids_and_noop() {
617        let (root, mut no_op_journal) = journal("no-op");
618        no_op_journal
619            .append("unknown", "t0", json!({"value":0}))
620            .unwrap();
621        assert!(compact_records(no_op_journal.records()).unwrap().is_none());
622        no_op_journal
623            .append(COMMAND_SIDEBAND, "t1", json!({"status":"pending"}))
624            .unwrap();
625        assert!(
626            compact_records(no_op_journal.records())
627                .unwrap_err()
628                .to_string()
629                .contains("session command record has no ID")
630        );
631        drop(no_op_journal);
632        fs::remove_dir_all(root).unwrap();
633
634        let (root, mut journal) = journal("missing-stop-id");
635        journal
636            .append(STOP_SIDEBAND, "t1", json!({"status":"pending"}))
637            .unwrap();
638        assert!(
639            compact_records(journal.records())
640                .unwrap_err()
641                .to_string()
642                .contains("session stop record has no ID")
643        );
644        drop(journal);
645        fs::remove_dir_all(root).unwrap();
646    }
647}
kcode_session_control_records/lib.rs

kcode_session_control_records/
lib.rs
