use std::{
fs::File,
path::{Path, PathBuf},
};
use anyhow::Context as _;
use kcode_session_control_journal::Journal;
use kcode_session_control_records::{compact_records, encode_update, project_records};
pub use kcode_session_control_records::{
ControlProjection, ControlUpdate, SessionCommand, SessionRecord, SessionStopRequest,
};
const CONTROL_EXTENSION: &str = "session-control";
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum OpenMode {
CreateNew,
OpenOrCreate,
ExistingOnly,
}
pub struct SessionControl {
directory: PathBuf,
path: PathBuf,
journal: Journal,
}
impl SessionControl {
pub fn open(
directory: impl AsRef<Path>,
session_id: &str,
mode: OpenMode,
) -> anyhow::Result<Option<Self>> {
let directory = directory.as_ref().to_path_buf();
let path = control_path(&directory, session_id);
let journal = match mode {
OpenMode::CreateNew => Journal::create(path.clone())?,
OpenMode::OpenOrCreate => match Journal::open(path.clone())? {
Some(journal) => journal,
None => Journal::create(path.clone())?,
},
OpenMode::ExistingOnly => {
let Some(journal) = Journal::open(path.clone())? else {
return Ok(None);
};
journal
}
};
Ok(Some(Self {
directory,
path,
journal,
}))
}
pub fn projection(&self) -> ControlProjection {
project_records(self.journal.records())
}
pub fn append(
&mut self,
recorded_at: impl Into<String>,
update: ControlUpdate,
) -> anyhow::Result<ControlUpdate> {
let (projected, kind, value) = encode_update(update)?;
self.journal.append(kind, recorded_at, value)?;
Ok(projected)
}
pub fn delete(self) -> anyhow::Result<()> {
let Self {
directory,
path,
journal,
} = self;
drop(journal);
if path.exists() {
std::fs::remove_file(&path).with_context(|| format!("removing {}", path.display()))?;
sync_directory(&directory)?;
}
Ok(())
}
pub fn compact_directory(directory: impl AsRef<Path>) -> anyhow::Result<()> {
let directory = directory.as_ref();
let mut paths = std::fs::read_dir(directory)?
.filter_map(Result::ok)
.map(|entry| entry.path())
.filter(|path| {
path.extension().and_then(|value| value.to_str()) == Some(CONTROL_EXTENSION)
})
.collect::<Vec<_>>();
paths.sort();
for path in paths {
compact_journal(&path)?;
}
Ok(())
}
}
fn control_path(directory: &Path, session_id: &str) -> PathBuf {
directory.join(format!("{session_id}.{CONTROL_EXTENSION}"))
}
fn compact_journal(path: &Path) -> anyhow::Result<()> {
let original_bytes = std::fs::metadata(path)
.with_context(|| format!("reading metadata for {}", path.display()))?
.len();
if original_bytes >= 16 * 1024 * 1024 {
tracing::info!(
path = %path.display(),
original_bytes,
"Compacting legacy Session History control journal"
);
}
let mut journal = Journal::open(path.to_path_buf())?
.with_context(|| format!("session-control journal {} disappeared", path.display()))?;
let repaired_bytes = std::fs::metadata(path)?.len();
let tail_repaired = repaired_bytes != original_bytes;
let compacted = compact_records(journal.records())?;
let rewritten = compacted.is_some();
if let Some(records) = compacted {
journal.replace(records)?;
}
if rewritten || tail_repaired {
tracing::info!(
path = %path.display(),
original_bytes,
compacted_bytes = std::fs::metadata(path)?.len(),
"Compacted Session History control journal"
);
}
Ok(())
}
fn sync_directory(path: &Path) -> anyhow::Result<()> {
File::open(path)
.with_context(|| format!("opening directory {} for sync", path.display()))?
.sync_all()
.with_context(|| format!("syncing directory {}", path.display()))
}
#[cfg(test)]
mod tests {
use std::{
fs::{self, OpenOptions},
io::Write as _,
sync::atomic::{AtomicU64, Ordering},
time::{SystemTime, UNIX_EPOCH},
};
use kcode_session_control_journal::Journal;
use serde_json::{Value, json};
use super::*;
const LIFECYCLE_SIDEBAND: &str = "session_lifecycle";
const COMMAND_SIDEBAND: &str = "session_command";
const STOP_SIDEBAND: &str = "session_stop";
static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);
fn root(label: &str) -> PathBuf {
let path = std::env::temp_dir().join(format!(
"kcode-session-control-state-{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-14T00:00:00Z".into(),
updated_at: format!("2026-08-14T00: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-14T00: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-14T00:00:00Z".into(),
completed_at: None,
idempotency_id: format!("stop-{id}"),
}
}
#[test]
fn opening_modes_preserve_create_and_absence_distinctions() {
let root = root("open-modes");
assert!(
SessionControl::open(&root, "missing", OpenMode::ExistingOnly)
.unwrap()
.is_none()
);
let created = SessionControl::open(&root, "new", OpenMode::CreateNew)
.unwrap()
.unwrap();
assert!(control_path(&root, "new").is_file());
assert!(SessionControl::open(&root, "new", OpenMode::CreateNew).is_err());
drop(created);
let opened = SessionControl::open(&root, "new", OpenMode::OpenOrCreate)
.unwrap()
.unwrap();
drop(opened);
let created_on_absence = SessionControl::open(&root, "other", OpenMode::OpenOrCreate)
.unwrap()
.unwrap();
assert!(control_path(&root, "other").is_file());
drop(created_on_absence);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn append_return_and_reopen_retain_launch_identity() {
let root = root("launch-retention");
let mut control = SessionControl::open(&root, "session-1", OpenMode::CreateNew)
.unwrap()
.unwrap();
let update = ControlUpdate::Lifecycle(lifecycle(
"session-1",
1,
json!({
"sessionId":"session-1",
"sessionType":"conversation",
"launchContextNodeIds":["A1234567","B1234567"],
"launchProvenance":{"syntheticBootstrap":true},
"chatendText":"discard"
}),
));
let returned = control.append("t1", update).unwrap();
let ControlUpdate::Lifecycle(returned) = returned else {
panic!("append changed update kind");
};
assert_eq!(
returned.state["launchContextNodeIds"],
json!(["A1234567", "B1234567"])
);
assert_eq!(
returned.state["launchProvenance"],
json!({"syntheticBootstrap":true})
);
assert!(returned.state.get("chatendText").is_none());
drop(control);
let reopened = SessionControl::open(&root, "session-1", OpenMode::ExistingOnly)
.unwrap()
.unwrap();
let restored = reopened.projection().lifecycle.unwrap();
assert_eq!(
restored.state["launchContextNodeIds"],
json!(["A1234567", "B1234567"])
);
assert_eq!(
restored.state["launchProvenance"],
json!({"syntheticBootstrap":true})
);
assert!(restored.state.get("chatendText").is_none());
drop(reopened);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn projection_retains_latest_command_and_stop_values() {
let root = root("typed-projection");
let mut control = SessionControl::open(&root, "session-1", OpenMode::CreateNew)
.unwrap()
.unwrap();
control
.append(
"t1",
ControlUpdate::Lifecycle(lifecycle(
"session-1",
1,
json!({"sessionType":"conversation"}),
)),
)
.unwrap();
control
.append("t2", ControlUpdate::Command(command("a", "pending", 1)))
.unwrap();
control
.append("t3", ControlUpdate::Command(command("a", "complete", 1)))
.unwrap();
control
.append("t4", ControlUpdate::Command(command("b", "pending", 2)))
.unwrap();
control
.append("t5", ControlUpdate::StopRequest(stop("s", "pending")))
.unwrap();
control
.append("t6", ControlUpdate::StopRequest(stop("s", "complete")))
.unwrap();
let projection = control.projection();
assert_eq!(projection.lifecycle.unwrap().version, 1);
assert_eq!(projection.commands.len(), 2);
assert_eq!(projection.commands["a"].status, "complete");
assert_eq!(projection.commands["b"].status, "pending");
assert_eq!(projection.stop_requests["s"].status, "complete");
drop(control);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn directory_compaction_retains_launch_identity_and_is_idempotent() {
let root = root("compaction");
let path = control_path(&root, "session-1");
let mut journal = Journal::create(path.clone()).unwrap();
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(
LIFECYCLE_SIDEBAND,
"t3",
serde_json::to_value(lifecycle(
"session-1",
2,
json!({
"sessionType":"conversation",
"launchContextNodeIds":[],
"launchProvenance":{"syntheticBootstrap":true},
"chatendText":"discard"
}),
))
.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("unknown-last", "t6", json!({"value":6}))
.unwrap();
journal
.append(
STOP_SIDEBAND,
"t7",
serde_json::to_value(stop("s", "complete")).unwrap(),
)
.unwrap();
drop(journal);
SessionControl::compact_directory(&root).unwrap();
let after_first = fs::read(&path).unwrap();
let compacted = SessionControl::open(&root, "session-1", OpenMode::ExistingOnly)
.unwrap()
.unwrap();
let projection = compacted.projection();
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(compacted);
SessionControl::compact_directory(&root).unwrap();
assert_eq!(fs::read(&path).unwrap(), after_first);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn malformed_typed_records_keep_existing_tolerance_and_errors() {
let root = root("malformed");
let path = control_path(&root, "session-1");
let mut journal = Journal::create(path).unwrap();
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();
drop(journal);
let control = SessionControl::open(&root, "session-1", OpenMode::ExistingOnly)
.unwrap()
.unwrap();
let projection = control.projection();
assert!(projection.lifecycle.is_none());
assert!(projection.commands.is_empty());
drop(control);
assert!(SessionControl::compact_directory(&root).is_ok());
let malformed_path = control_path(&root, "missing-id");
let mut malformed = Journal::create(malformed_path).unwrap();
malformed
.append(COMMAND_SIDEBAND, "t1", json!({"status":"pending"}))
.unwrap();
drop(malformed);
assert!(
SessionControl::compact_directory(&root)
.unwrap_err()
.to_string()
.contains("session command record has no ID")
);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn opening_repairs_incomplete_tail_but_rejects_complete_corruption() {
let root = root("integrity");
let path = control_path(&root, "tail");
let mut control = SessionControl::open(&root, "tail", OpenMode::CreateNew)
.unwrap()
.unwrap();
control
.append(
"t1",
ControlUpdate::Lifecycle(lifecycle("tail", 1, json!({}))),
)
.unwrap();
drop(control);
let complete = fs::read(&path).unwrap();
let mut file = OpenOptions::new().append(true).open(&path).unwrap();
file.write_all(b"incomplete tail").unwrap();
file.sync_all().unwrap();
drop(file);
let repaired = SessionControl::open(&root, "tail", OpenMode::ExistingOnly)
.unwrap()
.unwrap();
assert_eq!(repaired.projection().lifecycle.unwrap().version, 1);
drop(repaired);
assert_eq!(fs::read(&path).unwrap(), complete);
let corrupt_path = control_path(&root, "corrupt");
let mut corrupt = SessionControl::open(&root, "corrupt", OpenMode::CreateNew)
.unwrap()
.unwrap();
corrupt
.append(
"t1",
ControlUpdate::Lifecycle(lifecycle("corrupt", 1, json!({}))),
)
.unwrap();
drop(corrupt);
let mut bytes = fs::read(&corrupt_path).unwrap();
bytes[0] = if bytes[0] == b'0' { b'1' } else { b'0' };
fs::write(&corrupt_path, bytes).unwrap();
assert!(SessionControl::open(&root, "corrupt", OpenMode::ExistingOnly).is_err());
fs::remove_dir_all(root).unwrap();
}
#[test]
fn delete_removes_only_the_control_file() {
let root = root("delete");
let unrelated = root.join("keep.session-log");
fs::write(&unrelated, b"log").unwrap();
let control = SessionControl::open(&root, "session-1", OpenMode::CreateNew)
.unwrap()
.unwrap();
let path = control_path(&root, "session-1");
control.delete().unwrap();
assert!(!path.exists());
assert_eq!(fs::read(unrelated).unwrap(), b"log");
fs::remove_dir_all(root).unwrap();
}
}