kcode-session-control-records 0.1.0

//! Pure typed meaning for `.session-control` journal records.
//!
//! Filesystem durability, append synchronization, replacement, locking, and
//! directory lifecycle remain with the composing state package.

use std::collections::BTreeMap;

use anyhow::Context as _;
use kcode_session_control_journal::Record;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};

const LIFECYCLE_SIDEBAND: &str = "session_lifecycle";
const COMMAND_SIDEBAND: &str = "session_command";
const STOP_SIDEBAND: &str = "session_stop";

#[derive(Clone, Debug, Default)]
pub struct ControlProjection {
    pub lifecycle: Option<SessionRecord>,
    pub commands: BTreeMap<String, SessionCommand>,
    pub stop_requests: BTreeMap<String, SessionStopRequest>,
}

#[derive(Clone, Debug)]
pub enum ControlUpdate {
    Lifecycle(SessionRecord),
    Command(SessionCommand),
    StopRequest(SessionStopRequest),
}

impl ControlUpdate {
    /// Projects lifecycle state to authoritative control fields.
    ///
    /// Command and stop updates are returned unchanged.
    pub fn projected(mut self) -> Self {
        if let Self::Lifecycle(record) = &mut self {
            record.state = control_state(&record.state);
        }
        self
    }
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct SessionRecord {
    pub id: String,
    pub phase: String,
    pub started_at: String,
    pub updated_at: String,
    pub state: Value,
    pub provenance_id: Option<String>,
    pub version: i64,
    pub last_user_message_at: Option<String>,
    pub ended_at: Option<String>,
    pub ingress_failure_count: i64,
    pub ingress_failures: Value,
    pub ingress_next_attempt_at: Option<String>,
    #[serde(default, skip_serializing_if = "is_false")]
    pub summary: bool,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionCommand {
    pub id: String,
    pub conversation_id: String,
    pub sequence: i64,
    pub kind: String,
    pub payload: Value,
    pub status: String,
    pub cancel_requested: bool,
    pub outcome: Option<Value>,
    pub created_at: String,
    pub processing_started_at: Option<String>,
    pub completed_at: Option<String>,
    pub idempotency_id: String,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionStopRequest {
    pub id: String,
    pub session_id: String,
    pub scope: String,
    pub status: String,
    pub outcome: Option<Value>,
    pub requested_at: String,
    pub completed_at: Option<String>,
    pub idempotency_id: String,
}

pub fn encode_update(
    update: ControlUpdate,
) -> anyhow::Result<(ControlUpdate, &'static str, Value)> {
    let update = update.projected();
    let (kind, value) = match &update {
        ControlUpdate::Lifecycle(record) => (
            LIFECYCLE_SIDEBAND,
            serde_json::to_value(record).context("encoding session lifecycle record")?,
        ),
        ControlUpdate::Command(command) => (
            COMMAND_SIDEBAND,
            serde_json::to_value(command).context("encoding session command record")?,
        ),
        ControlUpdate::StopRequest(request) => (
            STOP_SIDEBAND,
            serde_json::to_value(request).context("encoding session stop record")?,
        ),
    };
    Ok((update, kind, value))
}

pub fn project_records(records: &[Record]) -> ControlProjection {
    let latest_lifecycle = records
        .iter()
        .rev()
        .find(|record| record.kind == LIFECYCLE_SIDEBAND)
        .and_then(|record| serde_json::from_value(record.value.clone()).ok());
    let mut commands = BTreeMap::new();
    let mut stop_requests = BTreeMap::new();
    for record in records {
        match record.kind.as_str() {
            COMMAND_SIDEBAND => {
                if let Ok(command) = serde_json::from_value::<SessionCommand>(record.value.clone())
                {
                    commands.insert(command.id.clone(), command);
                }
            }
            STOP_SIDEBAND => {
                if let Ok(request) =
                    serde_json::from_value::<SessionStopRequest>(record.value.clone())
                {
                    stop_requests.insert(request.id.clone(), request);
                }
            }
            _ => {}
        }
    }
    ControlProjection {
        lifecycle: latest_lifecycle,
        commands,
        stop_requests,
    }
}

pub fn compact_records(records: &[Record]) -> anyhow::Result<Option<Vec<Record>>> {
    let mut latest_lifecycle = None;
    let mut latest_commands = BTreeMap::<String, (u64, Record)>::new();
    let mut latest_stop_requests = BTreeMap::<String, (u64, Record)>::new();
    let mut retained_other = Vec::new();
    let mut needs_rewrite = false;

    for (sequence, mut record) in records.iter().cloned().enumerate() {
        let sequence = sequence as u64;
        match record.kind.as_str() {
            LIFECYCLE_SIDEBAND => {
                if let Some(state) = record.value.get_mut("state") {
                    let projected = control_state(state);
                    if *state != projected {
                        *state = projected;
                        needs_rewrite = true;
                    }
                }
                if latest_lifecycle.replace((sequence, record)).is_some() {
                    needs_rewrite = true;
                }
            }
            COMMAND_SIDEBAND => {
                let id = record
                    .value
                    .get("id")
                    .and_then(Value::as_str)
                    .context("session command record has no ID")?
                    .to_owned();
                if latest_commands.insert(id, (sequence, record)).is_some() {
                    needs_rewrite = true;
                }
            }
            STOP_SIDEBAND => {
                let id = record
                    .value
                    .get("id")
                    .and_then(Value::as_str)
                    .context("session stop record has no ID")?
                    .to_owned();
                if latest_stop_requests
                    .insert(id, (sequence, record))
                    .is_some()
                {
                    needs_rewrite = true;
                }
            }
            _ => retained_other.push((sequence, record)),
        }
    }

    if !needs_rewrite {
        return Ok(None);
    }

    let mut retained = retained_other;
    retained.extend(latest_lifecycle);
    retained.extend(latest_commands.into_values());
    retained.extend(latest_stop_requests.into_values());
    retained.sort_by_key(|(sequence, _)| *sequence);
    Ok(Some(
        retained.into_iter().map(|(_, record)| record).collect(),
    ))
}

fn is_false(value: &bool) -> bool {
    !*value
}

fn control_state(value: &Value) -> Value {
    const KEYS: &[&str] = &[
        "format",
        "version",
        "stateVersion",
        "sessionId",
        "sessionType",
        "sourceSessionType",
        "channel",
        "freeTime",
        "selfTimeIntent",
        "orchestration",
        "provenanceId",
        "rustLibSessionId",
        "rootNodeIds",
        "referenceRootNodeIds",
        "startedAt",
        "pendingTurn",
        "pendingExternalEventId",
        "roundsUsed",
        "providerAffinity",
        "nextThreadResetReason",
        "completed",
        "sessionObjectId",
        "commitReceipt",
        "commitAuthor",
        "providerModel",
        "kwebPlan",
        "startIdempotencyId",
        "ingressSource",
        "firstUserMessage",
        "boxCount",
        "eventCount",
        "chatendMetadata",
        "sessionStatus",
        "launchContextNodeIds",
        "launchProvenance",
        "historyIngress",
    ];
    let mut output = Map::new();
    for key in KEYS {
        if let Some(item) = value.get(*key) {
            if *key == "commitReceipt" && item.is_null() {
                continue;
            }
            let item = if *key == "historyIngress" {
                control_state(item)
            } else {
                item.clone()
            };
            output.insert((*key).into(), item);
        }
    }
    Value::Object(output)
}

#[cfg(test)]
mod tests {
    use std::{
        fs,
        path::PathBuf,
        sync::atomic::{AtomicU64, Ordering},
        time::{SystemTime, UNIX_EPOCH},
    };

    use kcode_session_control_journal::Journal;
    use serde_json::json;

    use super::*;

    static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);

    fn root(label: &str) -> PathBuf {
        let path = std::env::temp_dir().join(format!(
            "kcode-session-control-records-{label}-{}-{}-{}",
            std::process::id(),
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos(),
            NEXT_ROOT.fetch_add(1, Ordering::Relaxed),
        ));
        fs::create_dir(&path).unwrap();
        path
    }

    fn lifecycle(id: &str, version: i64, state: Value) -> SessionRecord {
        SessionRecord {
            id: id.into(),
            phase: "active".into(),
            started_at: "2026-08-02T00:00:00Z".into(),
            updated_at: format!("2026-08-02T00:00:0{version}Z"),
            state,
            provenance_id: None,
            version,
            last_user_message_at: None,
            ended_at: None,
            ingress_failure_count: 0,
            ingress_failures: json!([]),
            ingress_next_attempt_at: None,
            summary: false,
        }
    }

    fn command(id: &str, status: &str, sequence: i64) -> SessionCommand {
        SessionCommand {
            id: id.into(),
            conversation_id: "session-1".into(),
            sequence,
            kind: "message".into(),
            payload: json!({"text":"hello"}),
            status: status.into(),
            cancel_requested: false,
            outcome: None,
            created_at: "2026-08-02T00:00:00Z".into(),
            processing_started_at: None,
            completed_at: None,
            idempotency_id: format!("command-{id}"),
        }
    }

    fn stop(id: &str, status: &str) -> SessionStopRequest {
        SessionStopRequest {
            id: id.into(),
            session_id: "session-1".into(),
            scope: "turn".into(),
            status: status.into(),
            outcome: None,
            requested_at: "2026-08-02T00:00:00Z".into(),
            completed_at: None,
            idempotency_id: format!("stop-{id}"),
        }
    }

    fn journal(label: &str) -> (PathBuf, Journal) {
        let root = root(label);
        let journal = Journal::create(root.join("records.session-control")).unwrap();
        (root, journal)
    }

    #[test]
    fn lifecycle_projection_retains_launch_identity_and_discards_presentation() {
        let update = ControlUpdate::Lifecycle(lifecycle(
            "session-1",
            1,
            json!({
                "sessionType":"conversation",
                "launchContextNodeIds":["A1234567","B1234567"],
                "launchProvenance":{"syntheticBootstrap":true},
                "chatendText":"discard",
                "commitReceipt":null,
                "historyIngress":{
                    "launchContextNodeIds":["C1234567"],
                    "launchProvenance":{"syntheticBootstrap":false},
                    "completed":true,
                    "commitReceipt":null,
                    "boxes":{"1":{"text":"discard"}}
                }
            }),
        ))
        .projected();
        let ControlUpdate::Lifecycle(record) = update else {
            panic!("projection changed update kind");
        };
        assert_eq!(
            record.state["launchContextNodeIds"],
            json!(["A1234567", "B1234567"])
        );
        assert_eq!(
            record.state["launchProvenance"],
            json!({"syntheticBootstrap":true})
        );
        assert!(record.state.get("chatendText").is_none());
        assert!(record.state.get("commitReceipt").is_none());
        assert_eq!(
            record.state["historyIngress"]["launchContextNodeIds"],
            json!(["C1234567"])
        );
        assert_eq!(
            record.state["historyIngress"]["launchProvenance"],
            json!({"syntheticBootstrap":false})
        );
        assert!(
            record.state["historyIngress"]
                .get("commitReceipt")
                .is_none()
        );
        assert!(record.state["historyIngress"].get("boxes").is_none());
    }

    #[test]
    fn encoding_projects_lifecycle_and_preserves_command_and_stop_shapes() {
        let lifecycle = ControlUpdate::Lifecycle(lifecycle(
            "session-1",
            1,
            json!({
                "sessionType":"conversation",
                "launchContextNodeIds":[],
                "chatendText":"discard"
            }),
        ));
        let (projected, kind, value) = encode_update(lifecycle).unwrap();
        assert_eq!(kind, LIFECYCLE_SIDEBAND);
        assert_eq!(value["state"]["launchContextNodeIds"], json!([]));
        assert!(value["state"].get("chatendText").is_none());
        assert!(matches!(projected, ControlUpdate::Lifecycle(_)));

        let command = command("a", "pending", 1);
        let (_, kind, value) = encode_update(ControlUpdate::Command(command.clone())).unwrap();
        assert_eq!(kind, COMMAND_SIDEBAND);
        assert_eq!(value["conversationId"], command.conversation_id);

        let stop = stop("s", "pending");
        let (_, kind, value) = encode_update(ControlUpdate::StopRequest(stop.clone())).unwrap();
        assert_eq!(kind, STOP_SIDEBAND);
        assert_eq!(value["sessionId"], stop.session_id);
    }

    #[test]
    fn projection_selects_latest_valid_typed_values() {
        let (root, mut journal) = journal("projection");
        journal
            .append(
                LIFECYCLE_SIDEBAND,
                "t1",
                serde_json::to_value(lifecycle("session-1", 1, json!({}))).unwrap(),
            )
            .unwrap();
        journal
            .append(
                LIFECYCLE_SIDEBAND,
                "t2",
                serde_json::to_value(lifecycle("session-1", 2, json!({}))).unwrap(),
            )
            .unwrap();
        journal
            .append(
                COMMAND_SIDEBAND,
                "t3",
                serde_json::to_value(command("a", "pending", 1)).unwrap(),
            )
            .unwrap();
        journal
            .append(
                COMMAND_SIDEBAND,
                "t4",
                serde_json::to_value(command("a", "complete", 1)).unwrap(),
            )
            .unwrap();
        journal
            .append(
                STOP_SIDEBAND,
                "t5",
                serde_json::to_value(stop("s", "pending")).unwrap(),
            )
            .unwrap();
        journal
            .append(
                STOP_SIDEBAND,
                "t6",
                serde_json::to_value(stop("s", "complete")).unwrap(),
            )
            .unwrap();

        let projection = project_records(journal.records());
        assert_eq!(projection.lifecycle.unwrap().version, 2);
        assert_eq!(projection.commands["a"].status, "complete");
        assert_eq!(projection.stop_requests["s"].status, "complete");
        drop(journal);
        fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn malformed_records_retain_existing_projection_tolerance() {
        let (root, mut journal) = journal("malformed");
        journal
            .append(
                LIFECYCLE_SIDEBAND,
                "t1",
                serde_json::to_value(lifecycle("session-1", 1, json!({}))).unwrap(),
            )
            .unwrap();
        journal
            .append(LIFECYCLE_SIDEBAND, "t2", json!({"not":"a lifecycle"}))
            .unwrap();
        journal
            .append(COMMAND_SIDEBAND, "t3", json!({"id":"partial"}))
            .unwrap();
        journal
            .append(STOP_SIDEBAND, "t4", json!({"id":"partial"}))
            .unwrap();

        let projection = project_records(journal.records());
        assert!(projection.lifecycle.is_none());
        assert!(projection.commands.is_empty());
        assert!(projection.stop_requests.is_empty());
        drop(journal);
        fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn compaction_preserves_survivor_order_unknown_kinds_and_launch_fields() {
        let (root, mut journal) = journal("compaction");
        journal
            .append("unknown-first", "t0", json!({"value":0}))
            .unwrap();
        journal
            .append(
                LIFECYCLE_SIDEBAND,
                "t1",
                serde_json::to_value(lifecycle(
                    "session-1",
                    1,
                    json!({"sessionType":"conversation","chatendText":"old"}),
                ))
                .unwrap(),
            )
            .unwrap();
        journal
            .append(
                COMMAND_SIDEBAND,
                "t2",
                serde_json::to_value(command("a", "pending", 1)).unwrap(),
            )
            .unwrap();
        journal
            .append("unknown-middle", "t3", json!({"value":3}))
            .unwrap();
        journal
            .append(
                LIFECYCLE_SIDEBAND,
                "t4",
                serde_json::to_value(lifecycle(
                    "session-1",
                    2,
                    json!({
                        "sessionType":"conversation",
                        "launchContextNodeIds":[],
                        "launchProvenance":{"syntheticBootstrap":true},
                        "chatendText":"discard"
                    }),
                ))
                .unwrap(),
            )
            .unwrap();
        journal
            .append(
                STOP_SIDEBAND,
                "t5",
                serde_json::to_value(stop("s", "pending")).unwrap(),
            )
            .unwrap();
        journal
            .append(
                COMMAND_SIDEBAND,
                "t6",
                serde_json::to_value(command("a", "complete", 1)).unwrap(),
            )
            .unwrap();
        journal
            .append("unknown-last", "t7", json!({"value":7}))
            .unwrap();
        journal
            .append(
                STOP_SIDEBAND,
                "t8",
                serde_json::to_value(stop("s", "complete")).unwrap(),
            )
            .unwrap();

        let compacted = compact_records(journal.records()).unwrap().unwrap();
        let kinds = compacted
            .iter()
            .map(|record| record.kind.as_str())
            .collect::<Vec<_>>();
        assert_eq!(
            kinds,
            [
                "unknown-first",
                "unknown-middle",
                LIFECYCLE_SIDEBAND,
                COMMAND_SIDEBAND,
                "unknown-last",
                STOP_SIDEBAND,
            ]
        );
        let projection = project_records(&compacted);
        let lifecycle = projection.lifecycle.unwrap();
        assert_eq!(lifecycle.version, 2);
        assert_eq!(lifecycle.state["launchContextNodeIds"], json!([]));
        assert_eq!(
            lifecycle.state["launchProvenance"],
            json!({"syntheticBootstrap":true})
        );
        assert!(lifecycle.state.get("chatendText").is_none());
        assert_eq!(projection.commands["a"].status, "complete");
        assert_eq!(projection.stop_requests["s"].status, "complete");
        drop(journal);
        fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn compaction_reports_missing_typed_ids_and_noop() {
        let (root, mut no_op_journal) = journal("no-op");
        no_op_journal
            .append("unknown", "t0", json!({"value":0}))
            .unwrap();
        assert!(compact_records(no_op_journal.records()).unwrap().is_none());
        no_op_journal
            .append(COMMAND_SIDEBAND, "t1", json!({"status":"pending"}))
            .unwrap();
        assert!(
            compact_records(no_op_journal.records())
                .unwrap_err()
                .to_string()
                .contains("session command record has no ID")
        );
        drop(no_op_journal);
        fs::remove_dir_all(root).unwrap();

        let (root, mut journal) = journal("missing-stop-id");
        journal
            .append(STOP_SIDEBAND, "t1", json!({"status":"pending"}))
            .unwrap();
        assert!(
            compact_records(journal.records())
                .unwrap_err()
                .to_string()
                .contains("session stop record has no ID")
        );
        drop(journal);
        fs::remove_dir_all(root).unwrap();
    }
}