| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109 |
- use std::collections::BTreeMap;
- use std::fmt::{Display, Formatter};
- use std::fs::{self, OpenOptions};
- use std::io::Write;
- use std::path::{Path, PathBuf};
- use std::sync::atomic::{AtomicU64, Ordering};
- use std::time::{SystemTime, UNIX_EPOCH};
- use crate::json::{JsonError, JsonValue};
- use crate::usage::TokenUsage;
- const SESSION_VERSION: u32 = 1;
- const ROTATE_AFTER_BYTES: u64 = 256 * 1024;
- const MAX_ROTATED_FILES: usize = 3;
- static SESSION_ID_COUNTER: AtomicU64 = AtomicU64::new(0);
- #[derive(Debug, Clone, Copy, PartialEq, Eq)]
- pub enum MessageRole {
- System,
- User,
- Assistant,
- Tool,
- }
- #[derive(Debug, Clone, PartialEq, Eq)]
- pub enum ContentBlock {
- Text {
- text: String,
- },
- ToolUse {
- id: String,
- name: String,
- input: String,
- },
- ToolResult {
- tool_use_id: String,
- tool_name: String,
- output: String,
- is_error: bool,
- },
- }
- #[derive(Debug, Clone, PartialEq, Eq)]
- pub struct ConversationMessage {
- pub role: MessageRole,
- pub blocks: Vec<ContentBlock>,
- pub usage: Option<TokenUsage>,
- }
- #[derive(Debug, Clone, PartialEq, Eq)]
- pub struct SessionCompaction {
- pub count: u32,
- pub removed_message_count: usize,
- pub summary: String,
- }
- #[derive(Debug, Clone, PartialEq, Eq)]
- pub struct SessionFork {
- pub parent_session_id: String,
- pub branch_name: Option<String>,
- }
- #[derive(Debug, Clone, PartialEq, Eq)]
- struct SessionPersistence {
- path: PathBuf,
- }
- #[derive(Debug, Clone)]
- pub struct Session {
- pub version: u32,
- pub session_id: String,
- pub created_at_ms: u64,
- pub updated_at_ms: u64,
- pub messages: Vec<ConversationMessage>,
- pub compaction: Option<SessionCompaction>,
- pub fork: Option<SessionFork>,
- persistence: Option<SessionPersistence>,
- }
- impl PartialEq for Session {
- fn eq(&self, other: &Self) -> bool {
- self.version == other.version
- && self.session_id == other.session_id
- && self.created_at_ms == other.created_at_ms
- && self.updated_at_ms == other.updated_at_ms
- && self.messages == other.messages
- && self.compaction == other.compaction
- && self.fork == other.fork
- }
- }
- impl Eq for Session {}
- #[derive(Debug)]
- pub enum SessionError {
- Io(std::io::Error),
- Json(JsonError),
- Format(String),
- }
- impl Display for SessionError {
- fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- match self {
- Self::Io(error) => write!(f, "{error}"),
- Self::Json(error) => write!(f, "{error}"),
- Self::Format(error) => write!(f, "{error}"),
- }
- }
- }
- impl std::error::Error for SessionError {}
- impl From<std::io::Error> for SessionError {
- fn from(value: std::io::Error) -> Self {
- Self::Io(value)
- }
- }
- impl From<JsonError> for SessionError {
- fn from(value: JsonError) -> Self {
- Self::Json(value)
- }
- }
- impl Session {
- #[must_use]
- pub fn new() -> Self {
- let now = current_time_millis();
- Self {
- version: SESSION_VERSION,
- session_id: generate_session_id(),
- created_at_ms: now,
- updated_at_ms: now,
- messages: Vec::new(),
- compaction: None,
- fork: None,
- persistence: None,
- }
- }
- #[must_use]
- pub fn with_persistence_path(mut self, path: impl Into<PathBuf>) -> Self {
- self.persistence = Some(SessionPersistence { path: path.into() });
- self
- }
- #[must_use]
- pub fn persistence_path(&self) -> Option<&Path> {
- self.persistence.as_ref().map(|value| value.path.as_path())
- }
- pub fn save_to_path(&self, path: impl AsRef<Path>) -> Result<(), SessionError> {
- let path = path.as_ref();
- rotate_session_file_if_needed(path)?;
- write_atomic(path, &self.render_jsonl_snapshot())?;
- cleanup_rotated_logs(path)?;
- Ok(())
- }
- pub fn load_from_path(path: impl AsRef<Path>) -> Result<Self, SessionError> {
- let path = path.as_ref();
- let contents = fs::read_to_string(path)?;
- let session = match JsonValue::parse(&contents) {
- Ok(value)
- if value
- .as_object()
- .is_some_and(|object| object.contains_key("messages")) =>
- {
- Self::from_json(&value)?
- }
- Err(_) => Self::from_jsonl(&contents)?,
- Ok(_) => Self::from_jsonl(&contents)?,
- };
- Ok(session.with_persistence_path(path.to_path_buf()))
- }
- pub fn push_message(&mut self, message: ConversationMessage) -> Result<(), SessionError> {
- self.touch();
- self.messages.push(message.clone());
- self.append_persisted_message(&message)
- }
- pub fn push_user_text(&mut self, text: impl Into<String>) -> Result<(), SessionError> {
- self.push_message(ConversationMessage::user_text(text))
- }
- pub fn record_compaction(&mut self, summary: impl Into<String>, removed_message_count: usize) {
- self.touch();
- let count = self.compaction.as_ref().map_or(1, |value| value.count + 1);
- self.compaction = Some(SessionCompaction {
- count,
- removed_message_count,
- summary: summary.into(),
- });
- }
- #[must_use]
- pub fn fork(&self, branch_name: Option<String>) -> Self {
- let now = current_time_millis();
- Self {
- version: self.version,
- session_id: generate_session_id(),
- created_at_ms: now,
- updated_at_ms: now,
- messages: self.messages.clone(),
- compaction: self.compaction.clone(),
- fork: Some(SessionFork {
- parent_session_id: self.session_id.clone(),
- branch_name: normalize_optional_string(branch_name),
- }),
- persistence: None,
- }
- }
- #[must_use]
- pub fn to_json(&self) -> JsonValue {
- let mut object = BTreeMap::new();
- object.insert(
- "version".to_string(),
- JsonValue::Number(i64::from(self.version)),
- );
- object.insert(
- "session_id".to_string(),
- JsonValue::String(self.session_id.clone()),
- );
- object.insert(
- "created_at_ms".to_string(),
- JsonValue::Number(i64_from_u64(self.created_at_ms, "created_at_ms")),
- );
- object.insert(
- "updated_at_ms".to_string(),
- JsonValue::Number(i64_from_u64(self.updated_at_ms, "updated_at_ms")),
- );
- object.insert(
- "messages".to_string(),
- JsonValue::Array(
- self.messages
- .iter()
- .map(ConversationMessage::to_json)
- .collect(),
- ),
- );
- if let Some(compaction) = &self.compaction {
- object.insert("compaction".to_string(), compaction.to_json());
- }
- if let Some(fork) = &self.fork {
- object.insert("fork".to_string(), fork.to_json());
- }
- JsonValue::Object(object)
- }
- pub fn from_json(value: &JsonValue) -> Result<Self, SessionError> {
- let object = value
- .as_object()
- .ok_or_else(|| SessionError::Format("session must be an object".to_string()))?;
- let version = object
- .get("version")
- .and_then(JsonValue::as_i64)
- .ok_or_else(|| SessionError::Format("missing version".to_string()))?;
- let version = u32::try_from(version)
- .map_err(|_| SessionError::Format("version out of range".to_string()))?;
- let messages = object
- .get("messages")
- .and_then(JsonValue::as_array)
- .ok_or_else(|| SessionError::Format("missing messages".to_string()))?
- .iter()
- .map(ConversationMessage::from_json)
- .collect::<Result<Vec<_>, _>>()?;
- let now = current_time_millis();
- let session_id = object
- .get("session_id")
- .and_then(JsonValue::as_str)
- .map(ToOwned::to_owned)
- .unwrap_or_else(generate_session_id);
- let created_at_ms = object
- .get("created_at_ms")
- .map(|value| required_u64_from_value(value, "created_at_ms"))
- .transpose()?
- .unwrap_or(now);
- let updated_at_ms = object
- .get("updated_at_ms")
- .map(|value| required_u64_from_value(value, "updated_at_ms"))
- .transpose()?
- .unwrap_or(created_at_ms);
- let compaction = object
- .get("compaction")
- .map(SessionCompaction::from_json)
- .transpose()?;
- let fork = object.get("fork").map(SessionFork::from_json).transpose()?;
- Ok(Self {
- version,
- session_id,
- created_at_ms,
- updated_at_ms,
- messages,
- compaction,
- fork,
- persistence: None,
- })
- }
- fn from_jsonl(contents: &str) -> Result<Self, SessionError> {
- let mut version = SESSION_VERSION;
- let mut session_id = None;
- let mut created_at_ms = None;
- let mut updated_at_ms = None;
- let mut messages = Vec::new();
- let mut compaction = None;
- let mut fork = None;
- for (line_number, raw_line) in contents.lines().enumerate() {
- let line = raw_line.trim();
- if line.is_empty() {
- continue;
- }
- let value = JsonValue::parse(line).map_err(|error| {
- SessionError::Format(format!(
- "invalid JSONL record at line {}: {}",
- line_number + 1,
- error
- ))
- })?;
- let object = value.as_object().ok_or_else(|| {
- SessionError::Format(format!(
- "JSONL record at line {} must be an object",
- line_number + 1
- ))
- })?;
- match object
- .get("type")
- .and_then(JsonValue::as_str)
- .ok_or_else(|| {
- SessionError::Format(format!(
- "JSONL record at line {} missing type",
- line_number + 1
- ))
- })? {
- "session_meta" => {
- version = required_u32(object, "version")?;
- session_id = Some(required_string(object, "session_id")?);
- created_at_ms = Some(required_u64(object, "created_at_ms")?);
- updated_at_ms = Some(required_u64(object, "updated_at_ms")?);
- fork = object.get("fork").map(SessionFork::from_json).transpose()?;
- }
- "message" => {
- let message_value = object.get("message").ok_or_else(|| {
- SessionError::Format(format!(
- "JSONL record at line {} missing message",
- line_number + 1
- ))
- })?;
- messages.push(ConversationMessage::from_json(message_value)?);
- }
- "compaction" => {
- compaction = Some(SessionCompaction::from_json(&JsonValue::Object(
- object.clone(),
- ))?);
- }
- other => {
- return Err(SessionError::Format(format!(
- "unsupported JSONL record type at line {}: {other}",
- line_number + 1
- )))
- }
- }
- }
- let now = current_time_millis();
- Ok(Self {
- version,
- session_id: session_id.unwrap_or_else(generate_session_id),
- created_at_ms: created_at_ms.unwrap_or(now),
- updated_at_ms: updated_at_ms.unwrap_or(created_at_ms.unwrap_or(now)),
- messages,
- compaction,
- fork,
- persistence: None,
- })
- }
- fn render_jsonl_snapshot(&self) -> String {
- let mut lines = vec![self.meta_record().render()];
- if let Some(compaction) = &self.compaction {
- lines.push(compaction.to_jsonl_record().render());
- }
- lines.extend(
- self.messages
- .iter()
- .map(|message| message_record(message).render()),
- );
- let mut rendered = lines.join("\n");
- rendered.push('\n');
- rendered
- }
- fn append_persisted_message(&self, message: &ConversationMessage) -> Result<(), SessionError> {
- let Some(path) = self.persistence_path() else {
- return Ok(());
- };
- let needs_bootstrap = !path.exists() || fs::metadata(path)?.len() == 0;
- if needs_bootstrap {
- self.save_to_path(path)?;
- return Ok(());
- }
- let mut file = OpenOptions::new().append(true).open(path)?;
- writeln!(file, "{}", message_record(message).render())?;
- Ok(())
- }
- fn meta_record(&self) -> JsonValue {
- let mut object = BTreeMap::new();
- object.insert(
- "type".to_string(),
- JsonValue::String("session_meta".to_string()),
- );
- object.insert(
- "version".to_string(),
- JsonValue::Number(i64::from(self.version)),
- );
- object.insert(
- "session_id".to_string(),
- JsonValue::String(self.session_id.clone()),
- );
- object.insert(
- "created_at_ms".to_string(),
- JsonValue::Number(i64_from_u64(self.created_at_ms, "created_at_ms")),
- );
- object.insert(
- "updated_at_ms".to_string(),
- JsonValue::Number(i64_from_u64(self.updated_at_ms, "updated_at_ms")),
- );
- if let Some(fork) = &self.fork {
- object.insert("fork".to_string(), fork.to_json());
- }
- JsonValue::Object(object)
- }
- fn touch(&mut self) {
- self.updated_at_ms = current_time_millis();
- }
- }
- impl Default for Session {
- fn default() -> Self {
- Self::new()
- }
- }
- impl ConversationMessage {
- #[must_use]
- pub fn user_text(text: impl Into<String>) -> Self {
- Self {
- role: MessageRole::User,
- blocks: vec![ContentBlock::Text { text: text.into() }],
- usage: None,
- }
- }
- #[must_use]
- pub fn assistant(blocks: Vec<ContentBlock>) -> Self {
- Self {
- role: MessageRole::Assistant,
- blocks,
- usage: None,
- }
- }
- #[must_use]
- pub fn assistant_with_usage(blocks: Vec<ContentBlock>, usage: Option<TokenUsage>) -> Self {
- Self {
- role: MessageRole::Assistant,
- blocks,
- usage,
- }
- }
- #[must_use]
- pub fn tool_result(
- tool_use_id: impl Into<String>,
- tool_name: impl Into<String>,
- output: impl Into<String>,
- is_error: bool,
- ) -> Self {
- Self {
- role: MessageRole::Tool,
- blocks: vec![ContentBlock::ToolResult {
- tool_use_id: tool_use_id.into(),
- tool_name: tool_name.into(),
- output: output.into(),
- is_error,
- }],
- usage: None,
- }
- }
- #[must_use]
- pub fn to_json(&self) -> JsonValue {
- let mut object = BTreeMap::new();
- object.insert(
- "role".to_string(),
- JsonValue::String(
- match self.role {
- MessageRole::System => "system",
- MessageRole::User => "user",
- MessageRole::Assistant => "assistant",
- MessageRole::Tool => "tool",
- }
- .to_string(),
- ),
- );
- object.insert(
- "blocks".to_string(),
- JsonValue::Array(self.blocks.iter().map(ContentBlock::to_json).collect()),
- );
- if let Some(usage) = self.usage {
- object.insert("usage".to_string(), usage_to_json(usage));
- }
- JsonValue::Object(object)
- }
- fn from_json(value: &JsonValue) -> Result<Self, SessionError> {
- let object = value
- .as_object()
- .ok_or_else(|| SessionError::Format("message must be an object".to_string()))?;
- let role = match object
- .get("role")
- .and_then(JsonValue::as_str)
- .ok_or_else(|| SessionError::Format("missing role".to_string()))?
- {
- "system" => MessageRole::System,
- "user" => MessageRole::User,
- "assistant" => MessageRole::Assistant,
- "tool" => MessageRole::Tool,
- other => {
- return Err(SessionError::Format(format!(
- "unsupported message role: {other}"
- )))
- }
- };
- let blocks = object
- .get("blocks")
- .and_then(JsonValue::as_array)
- .ok_or_else(|| SessionError::Format("missing blocks".to_string()))?
- .iter()
- .map(ContentBlock::from_json)
- .collect::<Result<Vec<_>, _>>()?;
- let usage = object.get("usage").map(usage_from_json).transpose()?;
- Ok(Self {
- role,
- blocks,
- usage,
- })
- }
- }
- impl ContentBlock {
- #[must_use]
- pub fn to_json(&self) -> JsonValue {
- let mut object = BTreeMap::new();
- match self {
- Self::Text { text } => {
- object.insert("type".to_string(), JsonValue::String("text".to_string()));
- object.insert("text".to_string(), JsonValue::String(text.clone()));
- }
- Self::ToolUse { id, name, input } => {
- object.insert(
- "type".to_string(),
- JsonValue::String("tool_use".to_string()),
- );
- object.insert("id".to_string(), JsonValue::String(id.clone()));
- object.insert("name".to_string(), JsonValue::String(name.clone()));
- object.insert("input".to_string(), JsonValue::String(input.clone()));
- }
- Self::ToolResult {
- tool_use_id,
- tool_name,
- output,
- is_error,
- } => {
- object.insert(
- "type".to_string(),
- JsonValue::String("tool_result".to_string()),
- );
- object.insert(
- "tool_use_id".to_string(),
- JsonValue::String(tool_use_id.clone()),
- );
- object.insert(
- "tool_name".to_string(),
- JsonValue::String(tool_name.clone()),
- );
- object.insert("output".to_string(), JsonValue::String(output.clone()));
- object.insert("is_error".to_string(), JsonValue::Bool(*is_error));
- }
- }
- JsonValue::Object(object)
- }
- fn from_json(value: &JsonValue) -> Result<Self, SessionError> {
- let object = value
- .as_object()
- .ok_or_else(|| SessionError::Format("block must be an object".to_string()))?;
- match object
- .get("type")
- .and_then(JsonValue::as_str)
- .ok_or_else(|| SessionError::Format("missing block type".to_string()))?
- {
- "text" => Ok(Self::Text {
- text: required_string(object, "text")?,
- }),
- "tool_use" => Ok(Self::ToolUse {
- id: required_string(object, "id")?,
- name: required_string(object, "name")?,
- input: required_string(object, "input")?,
- }),
- "tool_result" => Ok(Self::ToolResult {
- tool_use_id: required_string(object, "tool_use_id")?,
- tool_name: required_string(object, "tool_name")?,
- output: required_string(object, "output")?,
- is_error: object
- .get("is_error")
- .and_then(JsonValue::as_bool)
- .ok_or_else(|| SessionError::Format("missing is_error".to_string()))?,
- }),
- other => Err(SessionError::Format(format!(
- "unsupported block type: {other}"
- ))),
- }
- }
- }
- impl SessionCompaction {
- #[must_use]
- pub fn to_json(&self) -> JsonValue {
- let mut object = BTreeMap::new();
- object.insert(
- "count".to_string(),
- JsonValue::Number(i64::from(self.count)),
- );
- object.insert(
- "removed_message_count".to_string(),
- JsonValue::Number(i64_from_usize(
- self.removed_message_count,
- "removed_message_count",
- )),
- );
- object.insert(
- "summary".to_string(),
- JsonValue::String(self.summary.clone()),
- );
- JsonValue::Object(object)
- }
- #[must_use]
- pub fn to_jsonl_record(&self) -> JsonValue {
- let mut object = self
- .to_json()
- .as_object()
- .cloned()
- .expect("compaction should render to object");
- object.insert(
- "type".to_string(),
- JsonValue::String("compaction".to_string()),
- );
- JsonValue::Object(object)
- }
- fn from_json(value: &JsonValue) -> Result<Self, SessionError> {
- let object = value
- .as_object()
- .ok_or_else(|| SessionError::Format("compaction must be an object".to_string()))?;
- Ok(Self {
- count: required_u32(object, "count")?,
- removed_message_count: required_usize(object, "removed_message_count")?,
- summary: required_string(object, "summary")?,
- })
- }
- }
- impl SessionFork {
- #[must_use]
- pub fn to_json(&self) -> JsonValue {
- let mut object = BTreeMap::new();
- object.insert(
- "parent_session_id".to_string(),
- JsonValue::String(self.parent_session_id.clone()),
- );
- if let Some(branch_name) = &self.branch_name {
- object.insert(
- "branch_name".to_string(),
- JsonValue::String(branch_name.clone()),
- );
- }
- JsonValue::Object(object)
- }
- fn from_json(value: &JsonValue) -> Result<Self, SessionError> {
- let object = value
- .as_object()
- .ok_or_else(|| SessionError::Format("fork metadata must be an object".to_string()))?;
- Ok(Self {
- parent_session_id: required_string(object, "parent_session_id")?,
- branch_name: object
- .get("branch_name")
- .and_then(JsonValue::as_str)
- .map(ToOwned::to_owned),
- })
- }
- }
- fn message_record(message: &ConversationMessage) -> JsonValue {
- let mut object = BTreeMap::new();
- object.insert("type".to_string(), JsonValue::String("message".to_string()));
- object.insert("message".to_string(), message.to_json());
- JsonValue::Object(object)
- }
- fn usage_to_json(usage: TokenUsage) -> JsonValue {
- let mut object = BTreeMap::new();
- object.insert(
- "input_tokens".to_string(),
- JsonValue::Number(i64::from(usage.input_tokens)),
- );
- object.insert(
- "output_tokens".to_string(),
- JsonValue::Number(i64::from(usage.output_tokens)),
- );
- object.insert(
- "cache_creation_input_tokens".to_string(),
- JsonValue::Number(i64::from(usage.cache_creation_input_tokens)),
- );
- object.insert(
- "cache_read_input_tokens".to_string(),
- JsonValue::Number(i64::from(usage.cache_read_input_tokens)),
- );
- JsonValue::Object(object)
- }
- fn usage_from_json(value: &JsonValue) -> Result<TokenUsage, SessionError> {
- let object = value
- .as_object()
- .ok_or_else(|| SessionError::Format("usage must be an object".to_string()))?;
- Ok(TokenUsage {
- input_tokens: required_u32(object, "input_tokens")?,
- output_tokens: required_u32(object, "output_tokens")?,
- cache_creation_input_tokens: required_u32(object, "cache_creation_input_tokens")?,
- cache_read_input_tokens: required_u32(object, "cache_read_input_tokens")?,
- })
- }
- fn required_string(
- object: &BTreeMap<String, JsonValue>,
- key: &str,
- ) -> Result<String, SessionError> {
- object
- .get(key)
- .and_then(JsonValue::as_str)
- .map(ToOwned::to_owned)
- .ok_or_else(|| SessionError::Format(format!("missing {key}")))
- }
- fn required_u32(object: &BTreeMap<String, JsonValue>, key: &str) -> Result<u32, SessionError> {
- let value = object
- .get(key)
- .and_then(JsonValue::as_i64)
- .ok_or_else(|| SessionError::Format(format!("missing {key}")))?;
- u32::try_from(value).map_err(|_| SessionError::Format(format!("{key} out of range")))
- }
- fn required_u64(object: &BTreeMap<String, JsonValue>, key: &str) -> Result<u64, SessionError> {
- let value = object
- .get(key)
- .ok_or_else(|| SessionError::Format(format!("missing {key}")))?;
- required_u64_from_value(value, key)
- }
- fn required_u64_from_value(value: &JsonValue, key: &str) -> Result<u64, SessionError> {
- let value = value
- .as_i64()
- .ok_or_else(|| SessionError::Format(format!("missing {key}")))?;
- u64::try_from(value).map_err(|_| SessionError::Format(format!("{key} out of range")))
- }
- fn required_usize(object: &BTreeMap<String, JsonValue>, key: &str) -> Result<usize, SessionError> {
- let value = object
- .get(key)
- .and_then(JsonValue::as_i64)
- .ok_or_else(|| SessionError::Format(format!("missing {key}")))?;
- usize::try_from(value).map_err(|_| SessionError::Format(format!("{key} out of range")))
- }
- fn i64_from_u64(value: u64, key: &str) -> i64 {
- i64::try_from(value).unwrap_or_else(|_| panic!("{key} out of range for JSON number"))
- }
- fn i64_from_usize(value: usize, key: &str) -> i64 {
- i64::try_from(value).unwrap_or_else(|_| panic!("{key} out of range for JSON number"))
- }
- fn normalize_optional_string(value: Option<String>) -> Option<String> {
- value.and_then(|value| {
- let trimmed = value.trim();
- if trimmed.is_empty() {
- None
- } else {
- Some(trimmed.to_string())
- }
- })
- }
- fn current_time_millis() -> u64 {
- SystemTime::now()
- .duration_since(UNIX_EPOCH)
- .map(|duration| duration.as_millis() as u64)
- .unwrap_or_default()
- }
- fn generate_session_id() -> String {
- let millis = current_time_millis();
- let counter = SESSION_ID_COUNTER.fetch_add(1, Ordering::Relaxed);
- format!("session-{millis}-{counter}")
- }
- fn write_atomic(path: &Path, contents: &str) -> Result<(), SessionError> {
- if let Some(parent) = path.parent() {
- fs::create_dir_all(parent)?;
- }
- let temp_path = temporary_path_for(path);
- fs::write(&temp_path, contents)?;
- fs::rename(temp_path, path)?;
- Ok(())
- }
- fn temporary_path_for(path: &Path) -> PathBuf {
- let file_name = path
- .file_name()
- .and_then(|value| value.to_str())
- .unwrap_or("session");
- path.with_file_name(format!(
- "{file_name}.tmp-{}-{}",
- current_time_millis(),
- SESSION_ID_COUNTER.fetch_add(1, Ordering::Relaxed)
- ))
- }
- fn rotate_session_file_if_needed(path: &Path) -> Result<(), SessionError> {
- let Ok(metadata) = fs::metadata(path) else {
- return Ok(());
- };
- if metadata.len() < ROTATE_AFTER_BYTES {
- return Ok(());
- }
- let rotated_path = rotated_log_path(path);
- fs::rename(path, rotated_path)?;
- Ok(())
- }
- fn rotated_log_path(path: &Path) -> PathBuf {
- let stem = path
- .file_stem()
- .and_then(|value| value.to_str())
- .unwrap_or("session");
- path.with_file_name(format!("{stem}.rot-{}.jsonl", current_time_millis()))
- }
- fn cleanup_rotated_logs(path: &Path) -> Result<(), SessionError> {
- let Some(parent) = path.parent() else {
- return Ok(());
- };
- let stem = path
- .file_stem()
- .and_then(|value| value.to_str())
- .unwrap_or("session");
- let prefix = format!("{stem}.rot-");
- let mut rotated_paths = fs::read_dir(parent)?
- .filter_map(Result::ok)
- .map(|entry| entry.path())
- .filter(|entry_path| {
- entry_path
- .file_name()
- .and_then(|value| value.to_str())
- .is_some_and(|name| name.starts_with(&prefix) && name.ends_with(".jsonl"))
- })
- .collect::<Vec<_>>();
- rotated_paths.sort_by_key(|entry_path| {
- fs::metadata(entry_path)
- .and_then(|metadata| metadata.modified())
- .unwrap_or(UNIX_EPOCH)
- });
- let remove_count = rotated_paths.len().saturating_sub(MAX_ROTATED_FILES);
- for stale_path in rotated_paths.into_iter().take(remove_count) {
- fs::remove_file(stale_path)?;
- }
- Ok(())
- }
- #[cfg(test)]
- mod tests {
- use super::{
- cleanup_rotated_logs, rotate_session_file_if_needed, ContentBlock, ConversationMessage,
- MessageRole, Session, SessionFork,
- };
- use crate::json::JsonValue;
- use crate::usage::TokenUsage;
- use std::fs;
- use std::path::PathBuf;
- use std::time::{SystemTime, UNIX_EPOCH};
- #[test]
- fn persists_and_restores_session_jsonl() {
- let mut session = Session::new();
- session
- .push_user_text("hello")
- .expect("user message should append");
- session
- .push_message(ConversationMessage::assistant_with_usage(
- vec![
- ContentBlock::Text {
- text: "thinking".to_string(),
- },
- ContentBlock::ToolUse {
- id: "tool-1".to_string(),
- name: "bash".to_string(),
- input: "echo hi".to_string(),
- },
- ],
- Some(TokenUsage {
- input_tokens: 10,
- output_tokens: 4,
- cache_creation_input_tokens: 1,
- cache_read_input_tokens: 2,
- }),
- ))
- .expect("assistant message should append");
- session
- .push_message(ConversationMessage::tool_result(
- "tool-1", "bash", "hi", false,
- ))
- .expect("tool result should append");
- let path = temp_session_path("jsonl");
- session.save_to_path(&path).expect("session should save");
- let restored = Session::load_from_path(&path).expect("session should load");
- fs::remove_file(&path).expect("temp file should be removable");
- assert_eq!(restored, session);
- assert_eq!(restored.messages[2].role, MessageRole::Tool);
- assert_eq!(
- restored.messages[1].usage.expect("usage").total_tokens(),
- 17
- );
- assert_eq!(restored.session_id, session.session_id);
- }
- #[test]
- fn loads_legacy_session_json_object() {
- let path = temp_session_path("legacy");
- let legacy = JsonValue::Object(
- [
- ("version".to_string(), JsonValue::Number(1)),
- (
- "messages".to_string(),
- JsonValue::Array(vec![ConversationMessage::user_text("legacy").to_json()]),
- ),
- ]
- .into_iter()
- .collect(),
- );
- fs::write(&path, legacy.render()).expect("legacy file should write");
- let restored = Session::load_from_path(&path).expect("legacy session should load");
- fs::remove_file(&path).expect("temp file should be removable");
- assert_eq!(restored.messages.len(), 1);
- assert_eq!(
- restored.messages[0],
- ConversationMessage::user_text("legacy")
- );
- assert!(!restored.session_id.is_empty());
- }
- #[test]
- fn appends_messages_to_persisted_jsonl_session() {
- let path = temp_session_path("append");
- let mut session = Session::new().with_persistence_path(path.clone());
- session
- .save_to_path(&path)
- .expect("initial save should succeed");
- session
- .push_user_text("hi")
- .expect("user append should succeed");
- session
- .push_message(ConversationMessage::assistant(vec![ContentBlock::Text {
- text: "hello".to_string(),
- }]))
- .expect("assistant append should succeed");
- let restored = Session::load_from_path(&path).expect("session should replay from jsonl");
- fs::remove_file(&path).expect("temp file should be removable");
- assert_eq!(restored.messages.len(), 2);
- assert_eq!(restored.messages[0], ConversationMessage::user_text("hi"));
- }
- #[test]
- fn persists_compaction_metadata() {
- let path = temp_session_path("compaction");
- let mut session = Session::new();
- session
- .push_user_text("before")
- .expect("message should append");
- session.record_compaction("summarized earlier work", 4);
- session.save_to_path(&path).expect("session should save");
- let restored = Session::load_from_path(&path).expect("session should load");
- fs::remove_file(&path).expect("temp file should be removable");
- let compaction = restored.compaction.expect("compaction metadata");
- assert_eq!(compaction.count, 1);
- assert_eq!(compaction.removed_message_count, 4);
- assert!(compaction.summary.contains("summarized"));
- }
- #[test]
- fn forks_sessions_with_branch_metadata_and_persists_it() {
- let path = temp_session_path("fork");
- let mut session = Session::new();
- session
- .push_user_text("before fork")
- .expect("message should append");
- let forked = session
- .fork(Some("investigation".to_string()))
- .with_persistence_path(path.clone());
- forked
- .save_to_path(&path)
- .expect("forked session should save");
- let restored = Session::load_from_path(&path).expect("forked session should load");
- fs::remove_file(&path).expect("temp file should be removable");
- assert_ne!(restored.session_id, session.session_id);
- assert_eq!(
- restored.fork,
- Some(SessionFork {
- parent_session_id: session.session_id,
- branch_name: Some("investigation".to_string()),
- })
- );
- assert_eq!(restored.messages, forked.messages);
- }
- #[test]
- fn rotates_and_cleans_up_large_session_logs() {
- let path = temp_session_path("rotation");
- fs::write(&path, "x".repeat((super::ROTATE_AFTER_BYTES + 10) as usize))
- .expect("oversized file should write");
- rotate_session_file_if_needed(&path).expect("rotation should succeed");
- assert!(
- !path.exists(),
- "original path should be rotated away before rewrite"
- );
- for _ in 0..5 {
- let rotated = super::rotated_log_path(&path);
- fs::write(&rotated, "old").expect("rotated file should write");
- }
- cleanup_rotated_logs(&path).expect("cleanup should succeed");
- let rotated_count = rotation_files(&path).len();
- assert!(rotated_count <= super::MAX_ROTATED_FILES);
- for rotated in rotation_files(&path) {
- fs::remove_file(rotated).expect("rotated file should be removable");
- }
- }
- fn temp_session_path(label: &str) -> PathBuf {
- let nanos = SystemTime::now()
- .duration_since(UNIX_EPOCH)
- .expect("system time should be after epoch")
- .as_nanos();
- std::env::temp_dir().join(format!("runtime-session-{label}-{nanos}.json"))
- }
- fn rotation_files(path: &PathBuf) -> Vec<PathBuf> {
- let stem = path
- .file_stem()
- .and_then(|value| value.to_str())
- .expect("temp path should have file stem")
- .to_string();
- fs::read_dir(path.parent().expect("temp path should have parent"))
- .expect("temp dir should read")
- .filter_map(Result::ok)
- .map(|entry| entry.path())
- .filter(|entry_path| {
- entry_path
- .file_name()
- .and_then(|value| value.to_str())
- .is_some_and(|name| {
- name.starts_with(&format!("{stem}.rot-")) && name.ends_with(".jsonl")
- })
- })
- .collect()
- }
- }
|