mod contracts;
#[cfg(test)]
mod tests;
use std::collections::VecDeque;
use std::fs::File;
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use crate::acp::permissions::PendingPermission;
use crate::app::{Entry, ToolStatus};
use crate::artifacts::{self, ArtifactMetadata};
use crate::context::ContextSource;
use crate::prompt::{EnvironmentMetadata, HistoryReuse, PromptBundle};
use crate::skills::{SkillActivation, SkillReferenceMeta};
use crate::tools::{WriteOp, shell};
use crate::{datetime, internals, tools};
use thndrs_agent::ProviderRequestAccounting;
use thndrs_agent::context::{ContextItem, ContextLedger};
pub use contracts::{
AcpPermissionOptionRecord, AcpSessionMetadata, ContextDiagnosticMeta, ContextItemMeta, ContextLedgerMeta,
ContextSourceMeta, McpToolSessionMeta, SessionConfigFile, SessionConfigMeta,
};
pub const SCHEMA_VERSION: u32 = 1;
pub const MAX_LOG_OUTPUT_BYTES: usize = 32 * 1024;
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum SessionRecord {
#[serde(rename = "session_meta")]
SessionMeta {
schema_version: u32,
seq: u64,
time: String,
session_id: String,
cwd: String,
title: String,
provider: String,
model: String,
websearch: String,
app_version: String,
#[serde(skip_serializing_if = "Option::is_none")]
config: Option<SessionConfigMeta>,
},
#[serde(rename = "context")]
Context {
schema_version: u32,
seq: u64,
time: String,
sources: Vec<ContextSourceMeta>,
},
#[serde(rename = "context_ledger")]
ContextLedger {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
ledger: ContextLedgerMeta,
},
#[serde(rename = "context_pin")]
ContextPin {
schema_version: u32,
seq: u64,
time: String,
item: ContextItemMeta,
reason: String,
},
#[serde(rename = "context_drop")]
ContextDrop {
schema_version: u32,
seq: u64,
time: String,
item: ContextItemMeta,
reason: String,
},
#[serde(rename = "context_recovery")]
ContextRecovery {
schema_version: u32,
seq: u64,
time: String,
item: ContextItemMeta,
reason: String,
},
#[serde(rename = "compaction")]
Compaction {
schema_version: u32,
seq: u64,
time: String,
audit: CompactionAudit,
},
#[serde(rename = "compaction_review")]
CompactionReview {
schema_version: u32,
seq: u64,
time: String,
recovery_handle: String,
review: CompactionReviewResult,
},
#[serde(rename = "user")]
User {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
text: String,
},
#[serde(rename = "prompt_metadata")]
PromptMetadata {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
metadata: PromptMetadata,
},
#[serde(rename = "assistant_finished")]
AssistantFinished {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
text: String,
},
#[serde(rename = "reasoning_finished")]
ReasoningFinished {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
text: String,
},
#[serde(rename = "usage")]
Usage {
schema_version: u32,
seq: u64,
time: String,
input_tokens: u64,
output_tokens: u64,
},
#[serde(rename = "request_accounting")]
RequestAccounting {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
accounting: ProviderRequestAccounting,
},
#[serde(rename = "tool_started")]
ToolStarted {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
call_id: String,
name: String,
arguments: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
mcp: Option<McpToolSessionMeta>,
},
#[serde(rename = "tool_finished")]
ToolFinished {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
call_id: String,
status: ToolStatus,
output: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
artifact: Option<ArtifactMetadata>,
#[serde(default, skip_serializing_if = "Option::is_none")]
mcp: Option<McpToolSessionMeta>,
},
#[serde(rename = "cancelled")]
Cancelled {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
reason: String,
},
#[serde(rename = "failed")]
Failed {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
error: String,
},
#[serde(rename = "acp_session")]
AcpSession {
schema_version: u32,
seq: u64,
time: String,
local_session_id: String,
agent_name: String,
acp_session_id: String,
command: String,
protocol_version: String,
#[serde(skip_serializing_if = "Option::is_none")]
agent_info_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
agent_info_version: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
client_info_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
client_info_version: Option<String>,
},
#[serde(rename = "session_renamed")]
SessionRenamed {
schema_version: u32,
seq: u64,
time: String,
title: String,
},
#[serde(rename = "file_write")]
FileWrite {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
op: WriteOp,
path: String,
before_hash: Option<u64>,
before_bytes: Option<usize>,
after_hash: u64,
after_bytes: usize,
status: ToolStatus,
},
#[serde(rename = "mcp_config_changed")]
McpConfigChanged {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
previous_files: Vec<SessionConfigFile>,
current_files: Vec<SessionConfigFile>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
diagnostics: Vec<String>,
},
#[serde(rename = "shell_exec")]
ShellExec {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
process_id: Option<u64>,
command: String,
cwd: String,
process_status: String,
exit_code: Option<i32>,
elapsed_ms: u64,
kind: String,
},
#[serde(rename = "skill_activated")]
SkillActivated {
schema_version: u32,
seq: u64,
time: String,
name: String,
path: String,
content_hash: u64,
byte_count: usize,
#[serde(default)]
rendered_content_hash: u64,
#[serde(default)]
rendered_byte_count: usize,
loaded_references: Vec<SkillReferenceRecord>,
},
#[serde(rename = "queued_input")]
QueuedInput {
schema_version: u32,
seq: u64,
time: String,
kind: String,
text: String,
},
#[serde(rename = "acp_permission_request")]
AcpPermissionRequest {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
tool_call_id: String,
title: String,
options: Vec<AcpPermissionOptionRecord>,
},
#[serde(rename = "acp_permission_outcome")]
AcpPermissionOutcome {
schema_version: u32,
seq: u64,
time: String,
turn_id: String,
tool_call_id: String,
outcome: String,
},
}
impl SessionRecord {
pub fn seq(&self) -> u64 {
match self {
SessionRecord::SessionMeta { seq, .. }
| SessionRecord::Context { seq, .. }
| SessionRecord::ContextLedger { seq, .. }
| SessionRecord::ContextPin { seq, .. }
| SessionRecord::ContextDrop { seq, .. }
| SessionRecord::ContextRecovery { seq, .. }
| SessionRecord::Compaction { seq, .. }
| SessionRecord::CompactionReview { seq, .. }
| SessionRecord::User { seq, .. }
| SessionRecord::PromptMetadata { seq, .. }
| SessionRecord::AssistantFinished { seq, .. }
| SessionRecord::ReasoningFinished { seq, .. }
| SessionRecord::Usage { seq, .. }
| SessionRecord::RequestAccounting { seq, .. }
| SessionRecord::ToolStarted { seq, .. }
| SessionRecord::ToolFinished { seq, .. }
| SessionRecord::Cancelled { seq, .. }
| SessionRecord::Failed { seq, .. }
| SessionRecord::AcpSession { seq, .. }
| SessionRecord::SessionRenamed { seq, .. }
| SessionRecord::FileWrite { seq, .. }
| SessionRecord::McpConfigChanged { seq, .. }
| SessionRecord::ShellExec { seq, .. }
| SessionRecord::SkillActivated { seq, .. }
| SessionRecord::QueuedInput { seq, .. }
| SessionRecord::AcpPermissionRequest { seq, .. }
| SessionRecord::AcpPermissionOutcome { seq, .. } => *seq,
}
}
pub fn to_json(&self) -> Result<String, serde_json::Error> {
serde_json::to_string(self)
}
pub fn from_json(json: &str) -> Result<Self, serde_json::Error> {
serde_json::from_str(json)
}
pub fn from_entry(entry: &Entry, seq: u64, time: &str, turn_id: &str) -> Option<SessionRecord> {
Self::from_entry_with_artifact(entry, seq, time, turn_id, None)
}
pub fn from_entry_with_artifact(
entry: &Entry, seq: u64, time: &str, turn_id: &str, artifact: Option<ArtifactMetadata>,
) -> Option<SessionRecord> {
match entry {
Entry::User { text } => Some(SessionRecord::User {
schema_version: SCHEMA_VERSION,
seq,
time: time.to_string(),
turn_id: turn_id.to_string(),
text: text.clone(),
}),
Entry::Agent { text, streaming: false } => Some(SessionRecord::AssistantFinished {
schema_version: SCHEMA_VERSION,
seq,
time: time.to_string(),
turn_id: turn_id.to_string(),
text: text.clone(),
}),
Entry::Reasoning { text, streaming: false } => Some(SessionRecord::ReasoningFinished {
schema_version: SCHEMA_VERSION,
seq,
time: time.to_string(),
turn_id: turn_id.to_string(),
text: text.clone(),
}),
Entry::Error { text } => Some(SessionRecord::Failed {
schema_version: SCHEMA_VERSION,
seq,
time: time.to_string(),
turn_id: turn_id.to_string(),
error: text.clone(),
}),
Entry::Tool { name, arguments: _, status, output } if *status != ToolStatus::Running => {
let (tool_name, call_id) = split_tool_name_id(name);
Some(SessionRecord::ToolFinished {
schema_version: SCHEMA_VERSION,
seq,
time: time.to_string(),
turn_id: turn_id.to_string(),
call_id,
status: *status,
output: artifacts::bounded_redacted_lines(output, artifacts::DEFAULT_MAX_ARTIFACT_BYTES),
artifact,
mcp: mcp_tool_session_meta(&tool_name),
})
.map(|r| (r, tool_name))
.map(|(r, _)| r)
}
_ => None,
}
}
pub fn to_entry(&self) -> Option<Entry> {
match self {
SessionRecord::User { text, .. } => Some(Entry::User { text: text.clone() }),
SessionRecord::PromptMetadata { .. } => None,
SessionRecord::AssistantFinished { text, .. } => {
Some(Entry::Agent { text: text.clone(), streaming: false })
}
SessionRecord::ReasoningFinished { text, .. } => {
Some(Entry::Reasoning { text: text.clone(), streaming: false })
}
SessionRecord::ToolFinished { call_id, status, output, .. } => Some(Entry::Tool {
name: format!("#{call_id}"),
arguments: String::new(),
status: *status,
output: output.clone(),
}),
SessionRecord::Cancelled { reason, .. } => Some(Entry::Status { text: reason.clone() }),
SessionRecord::Failed { error, .. } => Some(Entry::Error { text: error.clone() }),
SessionRecord::AcpSession { agent_name, acp_session_id, client_info_name, .. } => {
let client = client_info_name
.as_ref()
.map(|name| format!(" client {name}"))
.unwrap_or_default();
Some(Entry::Status { text: format!("acp session {agent_name}: {acp_session_id}{client}") })
}
SessionRecord::FileWrite { op, path, status, .. } => {
Some(Entry::Status { text: format!("{} {}: {path}", status.icon(), op.label()) })
}
SessionRecord::McpConfigChanged { .. } => None,
SessionRecord::ShellExec { command, process_status, process_id, elapsed_ms, .. } => {
let id = process_id.map_or_else(String::new, |id| format!(" [{id}]"));
Some(Entry::Status { text: format!("shell{id} {process_status}: {command} ({elapsed_ms}ms)") })
}
SessionRecord::SkillActivated { name, path, .. } => {
Some(Entry::Status { text: format!("skill activated: {name} ({path})") })
}
SessionRecord::AcpPermissionRequest { tool_call_id, title, .. } => {
Some(Entry::Status { text: format!("acp permission requested: {title} ({tool_call_id})") })
}
SessionRecord::AcpPermissionOutcome { tool_call_id, outcome, .. } => {
Some(Entry::Status { text: format!("acp permission {tool_call_id}: {outcome}") })
}
_ => None,
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct PromptMetadata {
pub model: String,
#[serde(default)]
pub provider: String,
pub search_mode: String,
#[serde(default)]
pub renderer_mode: String,
pub date: String,
pub cwd: String,
#[serde(default)]
pub prompt_fragments: Vec<String>,
#[serde(default)]
pub docs_map: Vec<String>,
pub context_sources: Vec<ContextSourceMeta>,
pub tool_catalog_size: usize,
#[serde(default)]
pub tool_names: Vec<String>,
pub skill_catalog_size: usize,
#[serde(default)]
pub skill_names: Vec<String>,
#[serde(default)]
pub diagnostics: Vec<String>,
pub history_reuse: bool,
pub prev_context_hash: Option<u64>,
pub transcript_tail_size: usize,
pub has_user_turn: bool,
}
impl PromptMetadata {
pub fn from_bundle(bundle: &PromptBundle) -> Self {
let environment: &EnvironmentMetadata = &bundle.environment;
let snapshot: internals::SelfKnowledgeSnapshot = bundle.into();
PromptMetadata {
model: environment.model.clone(),
provider: snapshot.runtime.provider.provider,
search_mode: environment.search_mode.label().to_string(),
renderer_mode: snapshot.runtime.renderer_mode,
date: environment.date.clone(),
cwd: environment.cwd.clone(),
prompt_fragments: snapshot.inventory.prompt_context.prompt_fragments,
docs_map: snapshot
.inventory
.references
.docs
.iter()
.map(|doc| doc.path.to_string())
.collect(),
context_sources: bundle
.project_context
.iter()
.map(ContextSourceMeta::from_source)
.collect(),
tool_catalog_size: bundle.tool_catalog.len(),
tool_names: snapshot.runtime.tools,
skill_catalog_size: bundle.available_skills.len(),
skill_names: snapshot
.inventory
.references
.skills
.into_iter()
.map(|skill| skill.name)
.collect(),
diagnostics: snapshot.diagnostics,
history_reuse: bundle.history_reuse == HistoryReuse::Available,
prev_context_hash: bundle.prev_context_hash,
transcript_tail_size: bundle.transcript_tail.len(),
has_user_turn: !bundle.user_turn.is_empty(),
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CompactionTrigger {
Manual,
Automatic,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CompactionRisk {
Low,
High,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CompactionReviewResult {
NotRequired,
Pending,
Approved,
Rejected,
}
impl CompactionReviewResult {
pub fn label(self) -> &'static str {
match self {
CompactionReviewResult::NotRequired => "not-required",
CompactionReviewResult::Pending => "pending",
CompactionReviewResult::Approved => "approved",
CompactionReviewResult::Rejected => "rejected",
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct CompactionSourceHash {
pub id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub content_hash: Option<u64>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct CompactionTokenUsage {
pub input_tokens: u64,
pub output_tokens: u64,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct CompactionAudit {
pub summary: String,
pub covered_start_seq: u64,
pub covered_end_seq: u64,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub source_hashes: Vec<CompactionSourceHash>,
pub trigger: CompactionTrigger,
pub risk: CompactionRisk,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub review: Option<CompactionReviewResult>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub recovery_handles: Vec<String>,
pub model: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub usage: Option<CompactionTokenUsage>,
}
impl CompactionAudit {
fn redacted(&self) -> Self {
CompactionAudit {
summary: tools::shell::redact_secrets(&self.summary),
covered_start_seq: self.covered_start_seq,
covered_end_seq: self.covered_end_seq,
source_hashes: self.source_hashes.clone(),
trigger: self.trigger,
risk: self.risk,
review: self.review,
recovery_handles: self.recovery_handles.clone(),
model: self.model.clone(),
usage: self.usage,
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct SkillReferenceRecord {
pub path: String,
pub content_hash: u64,
pub byte_count: usize,
pub truncated: bool,
}
impl From<&SkillReferenceMeta> for SkillReferenceRecord {
fn from(reference: &SkillReferenceMeta) -> Self {
SkillReferenceRecord {
path: reference.path.display().to_string(),
content_hash: reference.content_hash,
byte_count: reference.byte_count,
truncated: reference.truncated,
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ContextActionKind {
Pin,
Drop,
Recovery,
}
#[derive(Debug)]
pub struct SessionWriter {
path: PathBuf,
seq: u64,
session_id: String,
lock_path: PathBuf,
_lock: File,
}
impl SessionWriter {
#[expect(
clippy::too_many_arguments,
reason = "session metadata is written as a flat JSONL record"
)]
pub fn create(
dir: &Path, session_id: &str, cwd: &str, title: &str, provider: &str, model: &str, websearch: &str,
app_version: &str, config: Option<SessionConfigMeta>,
) -> std::io::Result<Self> {
std::fs::create_dir_all(dir)?;
let path = dir.join(format!("{session_id}.jsonl"));
let (lock_path, lock) = acquire_writer_lock(&path)?;
let record = SessionRecord::SessionMeta {
schema_version: SCHEMA_VERSION,
seq: 0,
time: datetime::now_iso8601(),
session_id: session_id.to_string(),
cwd: cwd.to_string(),
title: title.to_string(),
provider: provider.to_string(),
model: model.to_string(),
websearch: websearch.to_string(),
app_version: app_version.to_string(),
config,
};
let write_result = (|| -> std::io::Result<()> {
use std::io::Write;
let mut file = std::fs::OpenOptions::new().write(true).create_new(true).open(&path)?;
writeln!(file, "{}", record.to_json().map_err(io_err)?)
})();
if let Err(error) = write_result {
let _ = std::fs::remove_file(&lock_path);
return Err(error);
}
Ok(SessionWriter { path, seq: 1, session_id: session_id.to_string(), lock_path, _lock: lock })
}
pub fn next_sequence(&self) -> u64 {
self.seq
}
pub fn resume(path: &Path, session_id: &str) -> std::io::Result<Self> {
let (lock_path, lock) = acquire_writer_lock(path)?;
let records = SessionReader::read_records(path);
let seq = records
.iter()
.map(SessionRecord::seq)
.max()
.map_or(0, |max_seq| max_seq.saturating_add(1));
if let Err(error) = std::fs::OpenOptions::new().append(true).open(path) {
let _ = std::fs::remove_file(&lock_path);
return Err(error);
}
Ok(SessionWriter { path: path.to_path_buf(), seq, session_id: session_id.to_string(), lock_path, _lock: lock })
}
pub fn append(&mut self, mut record: SessionRecord) -> std::io::Result<()> {
let seq = self.seq;
self.seq += 1;
set_seq(&mut record, seq);
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_entry(&mut self, entry: &Entry, turn_id: &str) -> std::io::Result<()> {
if let Some(record) = SessionRecord::from_entry(entry, self.seq, &datetime::now_iso8601(), turn_id) {
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
}
Ok(())
}
pub fn append_entry_with_artifact(
&mut self, entry: &Entry, turn_id: &str, artifact: Option<ArtifactMetadata>,
) -> std::io::Result<()> {
if let Some(record) =
SessionRecord::from_entry_with_artifact(entry, self.seq, &datetime::now_iso8601(), turn_id, artifact)
{
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
}
Ok(())
}
pub fn append_tool_started(&mut self, turn_id: &str, call_id: &str, name: &str, args: &str) -> std::io::Result<()> {
let record = SessionRecord::ToolStarted {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
turn_id: turn_id.to_string(),
call_id: call_id.to_string(),
name: name.to_string(),
arguments: args.to_string(),
mcp: mcp_tool_session_meta(name),
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_context(&mut self, sources: &[ContextSource]) -> std::io::Result<()> {
let metas: Vec<ContextSourceMeta> = sources.iter().map(ContextSourceMeta::from_source).collect();
let record = SessionRecord::Context {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
sources: metas,
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_context_ledger(&mut self, turn_id: &str, ledger: &ContextLedger) -> std::io::Result<()> {
self.append(SessionRecord::ContextLedger {
schema_version: SCHEMA_VERSION,
seq: 0,
time: datetime::now_iso8601(),
turn_id: turn_id.to_string(),
ledger: ContextLedgerMeta::from(ledger),
})
}
pub fn append_context_pin(&mut self, item: &ContextItem, reason: &str) -> std::io::Result<()> {
self.append_context_action(item, reason, ContextActionKind::Pin)
}
pub fn append_context_drop(&mut self, item: &ContextItem, reason: &str) -> std::io::Result<()> {
self.append_context_action(item, reason, ContextActionKind::Drop)
}
pub fn append_context_recovery(&mut self, item: &ContextItem, reason: &str) -> std::io::Result<()> {
self.append_context_action(item, reason, ContextActionKind::Recovery)
}
pub fn append_compaction(&mut self, audit: &CompactionAudit) -> std::io::Result<()> {
self.append(SessionRecord::Compaction {
schema_version: SCHEMA_VERSION,
seq: 0,
time: datetime::now_iso8601(),
audit: audit.redacted(),
})
}
pub fn append_compaction_review(
&mut self, recovery_handle: &str, review: CompactionReviewResult,
) -> std::io::Result<()> {
self.append(SessionRecord::CompactionReview {
schema_version: SCHEMA_VERSION,
seq: 0,
time: datetime::now_iso8601(),
recovery_handle: tools::shell::redact_secrets(recovery_handle),
review,
})
}
fn append_context_action(
&mut self, item: &ContextItem, reason: &str, action: ContextActionKind,
) -> std::io::Result<()> {
let item = ContextItemMeta::from(item);
let reason = tools::shell::redact_secrets(reason);
let record = match action {
ContextActionKind::Pin => SessionRecord::ContextPin {
schema_version: SCHEMA_VERSION,
seq: 0,
time: datetime::now_iso8601(),
item,
reason,
},
ContextActionKind::Drop => SessionRecord::ContextDrop {
schema_version: SCHEMA_VERSION,
seq: 0,
time: datetime::now_iso8601(),
item,
reason,
},
ContextActionKind::Recovery => SessionRecord::ContextRecovery {
schema_version: SCHEMA_VERSION,
seq: 0,
time: datetime::now_iso8601(),
item,
reason,
},
};
self.append(record)
}
pub fn append_prompt_metadata(&mut self, turn_id: &str, metadata: &PromptMetadata) -> std::io::Result<()> {
let record = SessionRecord::PromptMetadata {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
turn_id: turn_id.to_string(),
metadata: metadata.clone(),
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_usage(&mut self, input_tokens: u64, output_tokens: u64) -> std::io::Result<()> {
if input_tokens == 0 && output_tokens == 0 {
return Ok(());
}
let record = SessionRecord::Usage {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
input_tokens,
output_tokens,
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_request_accounting(
&mut self, turn_id: &str, accounting: &ProviderRequestAccounting,
) -> std::io::Result<()> {
self.append(SessionRecord::RequestAccounting {
schema_version: SCHEMA_VERSION,
seq: 0,
time: datetime::now_iso8601(),
turn_id: turn_id.to_string(),
accounting: accounting.clone(),
})
}
pub fn append_file_write(
&mut self, turn_id: &str, result: &tools::WriteResult, status: ToolStatus,
) -> std::io::Result<()> {
let record = SessionRecord::FileWrite {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
turn_id: turn_id.to_string(),
op: result.op,
path: result.path.display().to_string(),
before_hash: result.before_hash,
before_bytes: result.before_bytes,
after_hash: result.after_hash,
after_bytes: result.after_bytes,
status,
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_mcp_config_changed(
&mut self, turn_id: &str, previous_files: Vec<SessionConfigFile>, current_files: Vec<SessionConfigFile>,
diagnostics: Vec<String>,
) -> std::io::Result<()> {
let record = SessionRecord::McpConfigChanged {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
turn_id: turn_id.to_string(),
previous_files,
current_files,
diagnostics,
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_shell_exec(&mut self, turn_id: &str, result: &shell::ProcessResult) -> std::io::Result<()> {
let record = SessionRecord::ShellExec {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
turn_id: turn_id.to_string(),
process_id: result.process_id,
command: shell::redact_secrets(&result.command.join(" ")),
cwd: result.cwd.display().to_string(),
process_status: result.status.label().to_string(),
exit_code: result.exit_code,
elapsed_ms: result.elapsed.as_millis() as u64,
kind: result.kind.label().to_string(),
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_skill_activation(&mut self, activation: &SkillActivation) -> std::io::Result<()> {
let record = SessionRecord::SkillActivated {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
name: activation.name.clone(),
path: activation.path.display().to_string(),
content_hash: activation.content_hash,
byte_count: activation.byte_count,
rendered_content_hash: activation.rendered_content_hash,
rendered_byte_count: activation.rendered_byte_count,
loaded_references: activation
.loaded_references
.iter()
.map(SkillReferenceRecord::from)
.collect(),
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_queued(&mut self, kind: &str, text: &str) -> std::io::Result<()> {
let record = SessionRecord::QueuedInput {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
kind: kind.to_string(),
text: text.to_string(),
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_acp_permission_request(&mut self, turn_id: &str, request: &PendingPermission) -> std::io::Result<()> {
let record = SessionRecord::AcpPermissionRequest {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
turn_id: turn_id.to_string(),
tool_call_id: request.tool_call_id.clone(),
title: request.title.clone(),
options: request
.options
.iter()
.map(|option| AcpPermissionOptionRecord {
id: option.id.clone(),
name: option.name.clone(),
kind: option.kind.label().to_string(),
})
.collect(),
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_acp_session(&mut self, metadata: &AcpSessionMetadata) -> std::io::Result<()> {
let record = SessionRecord::AcpSession {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
local_session_id: self.session_id.clone(),
agent_name: metadata.agent_name.clone(),
acp_session_id: metadata.acp_session_id.clone(),
command: metadata.command.clone(),
protocol_version: metadata.protocol_version.clone(),
agent_info_name: metadata.agent_info_name.clone(),
agent_info_version: metadata.agent_info_version.clone(),
client_info_name: metadata.client_info_name.clone(),
client_info_version: metadata.client_info_version.clone(),
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn append_acp_permission_outcome(
&mut self, turn_id: &str, tool_call_id: &str, outcome: &str,
) -> std::io::Result<()> {
let record = SessionRecord::AcpPermissionOutcome {
schema_version: SCHEMA_VERSION,
seq: self.seq,
time: datetime::now_iso8601(),
turn_id: turn_id.to_string(),
tool_call_id: tool_call_id.to_string(),
outcome: outcome.to_string(),
};
self.seq += 1;
let line = record.to_json().map_err(io_err)?;
use std::io::Write;
let mut file = std::fs::OpenOptions::new().append(true).open(&self.path)?;
writeln!(file, "{line}")?;
Ok(())
}
pub fn path(&self) -> &Path {
&self.path
}
pub fn session_id(&self) -> &str {
&self.session_id
}
}
impl Drop for SessionWriter {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.lock_path);
}
}
pub struct SessionReader;
impl SessionReader {
pub fn read_records(path: &Path) -> Vec<SessionRecord> {
let Ok(content) = std::fs::read_to_string(path) else {
return Vec::new();
};
content
.lines()
.filter(|line| !line.is_empty())
.filter_map(|line| SessionRecord::from_json(line).ok())
.collect()
}
pub fn read_records_from_tail(path: &Path, max_bytes: usize) -> Vec<SessionRecord> {
use std::io::{Read, Seek};
if max_bytes == 0 {
return Vec::new();
}
let Ok(mut file) = std::fs::File::open(path) else {
return Vec::new();
};
let Ok(file_len) = file.metadata().map(|metadata| metadata.len()) else {
return Vec::new();
};
let start = file_len.saturating_sub(max_bytes as u64);
let read_start = start.saturating_sub(1);
if file.seek(std::io::SeekFrom::Start(read_start)).is_err() {
return Vec::new();
}
let mut bytes = Vec::new();
if file.read_to_end(&mut bytes).is_err() {
return Vec::new();
}
let content = String::from_utf8_lossy(&bytes);
let complete_records = if start == 0 {
content.as_ref()
} else {
content.split_once('\n').map_or("", |(_, records)| records)
};
complete_records
.lines()
.filter(|line| !line.is_empty())
.filter_map(|line| SessionRecord::from_json(line).ok())
.collect()
}
pub fn read_transcript(path: &Path) -> Vec<Entry> {
Self::read_records(path)
.into_iter()
.filter_map(|r| r.to_entry())
.collect()
}
pub fn read_title(path: &Path) -> String {
let records = Self::read_records(path);
for r in records.iter().rev() {
if let SessionRecord::SessionRenamed { title, .. } = r {
return title.clone();
}
}
for r in &records {
if let SessionRecord::SessionMeta { title, .. } = r {
return title.clone();
}
}
path.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("session")
.to_string()
}
pub fn read_summary(path: &Path) -> SessionSummary {
let records = Self::read_records(path);
let mut title = path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("session")
.to_string();
let mut model = String::from("unknown");
let mut input_tokens = 0u64;
let mut output_tokens = 0u64;
let mut accounted_input_tokens = 0u64;
let mut accounted_output_tokens = 0u64;
let mut has_request_accounting = false;
for record in records {
match record {
SessionRecord::SessionMeta { title: t, model: m, .. } => {
title = t;
model = m;
}
SessionRecord::SessionRenamed { title: t, .. } => title = t,
SessionRecord::Usage { input_tokens: i, output_tokens: o, .. } => {
input_tokens = input_tokens.saturating_add(i);
output_tokens = output_tokens.saturating_add(o);
}
SessionRecord::RequestAccounting { accounting, .. } => {
has_request_accounting = true;
if let Some(usage) = accounting.provider_usage {
accounted_input_tokens =
accounted_input_tokens.saturating_add(usage.components.input_tokens.unwrap_or(0));
accounted_output_tokens =
accounted_output_tokens.saturating_add(usage.components.output_tokens.unwrap_or(0));
}
}
_ => {}
}
}
if has_request_accounting {
input_tokens = accounted_input_tokens;
output_tokens = accounted_output_tokens;
}
SessionSummary { title, model, input_tokens, output_tokens }
}
pub fn read_redacted_records(path: &Path) -> Vec<serde_json::Value> {
Self::read_records(path)
.into_iter()
.filter_map(|record| serde_json::to_value(record).ok())
.map(redact_json_value)
.collect()
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SessionSummary {
pub title: String,
pub model: String,
pub input_tokens: u64,
pub output_tokens: u64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SessionLookupError {
NotFound { query: String },
Ambiguous { query: String, matches: Vec<String> },
}
impl std::fmt::Display for SessionLookupError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::NotFound { query } => write!(formatter, "session `{query}` is not found"),
Self::Ambiguous { query, matches } => {
write!(
formatter,
"session prefix `{query}` is ambiguous; matches: {}",
matches.join(", ")
)
}
}
}
}
impl std::error::Error for SessionLookupError {}
impl SessionSummary {
pub fn sidebar_label(&self) -> String {
format!("{}\nin {} out {}", self.model, self.input_tokens, self.output_tokens)
}
}
pub fn sessions_dir(workspace_root: &Path) -> PathBuf {
workspace_root.join(".thndrs").join("sessions")
}
pub fn list_session_files(dir: &Path) -> Vec<PathBuf> {
let Ok(entries) = std::fs::read_dir(dir) else {
return Vec::new();
};
let mut files: Vec<_> = entries
.filter_map(|e| e.ok())
.filter_map(|e| {
let path = e.path();
if path.extension().is_some_and(|ext| ext == "jsonl") { Some(path) } else { None }
})
.collect();
files.sort_by(|a, b| {
let mtime_a = std::fs::metadata(a).and_then(|m| m.modified()).ok();
let mtime_b = std::fs::metadata(b).and_then(|m| m.modified()).ok();
mtime_b.cmp(&mtime_a).then_with(|| b.cmp(a))
});
files
}
pub fn resolve_session_file(dir: &Path, query: &str) -> Result<PathBuf, SessionLookupError> {
let files = list_session_files(dir);
if let Some(path) = files.iter().find(|path| session_id_from_path(path) == Some(query)) {
return Ok(path.clone());
}
let matches: Vec<PathBuf> = files
.into_iter()
.filter(|path| session_id_from_path(path).is_some_and(|id| id.starts_with(query)))
.collect();
match matches.as_slice() {
[] => Err(SessionLookupError::NotFound { query: query.to_string() }),
[path] => Ok(path.clone()),
_ => Err(SessionLookupError::Ambiguous {
query: query.to_string(),
matches: matches
.iter()
.filter_map(|path| session_id_from_path(path).map(ToString::to_string))
.collect(),
}),
}
}
pub fn list_session_titles(dir: &Path) -> Vec<String> {
list_session_files(dir)
.into_iter()
.map(|p| SessionReader::read_title(&p))
.collect()
}
pub fn list_session_summaries(dir: &Path) -> Vec<SessionSummary> {
list_session_files(dir)
.into_iter()
.map(|p| SessionReader::read_summary(&p))
.collect()
}
pub fn latest_session_file(dir: &Path) -> Option<PathBuf> {
list_session_files(dir).into_iter().next()
}
pub fn generate_session_id() -> String {
let secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
let days = secs / 86_400;
let remainder = secs % 86_400;
let hour = remainder / 3600;
let minute = (remainder % 3600) / 60;
let second = remainder % 60;
let date = datetime::date_from_days(days);
let date_compact = date.replace('-', "");
format!("session-{date_compact}-{hour:02}{minute:02}{second:02}")
}
fn io_err(e: serde_json::Error) -> std::io::Error {
std::io::Error::new(std::io::ErrorKind::InvalidData, e)
}
fn acquire_writer_lock(path: &Path) -> std::io::Result<(PathBuf, File)> {
let file_name = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("session.jsonl");
let lock_path = path.with_file_name(format!("{file_name}.lock"));
let lock = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&lock_path)
.map_err(|error| {
if error.kind() == std::io::ErrorKind::AlreadyExists {
std::io::Error::new(
std::io::ErrorKind::WouldBlock,
format!("session `{}` already has an active writer", path.display()),
)
} else {
error
}
})?;
Ok((lock_path, lock))
}
fn session_id_from_path(path: &Path) -> Option<&str> {
path.file_stem().and_then(|stem| stem.to_str())
}
fn redact_json_value(value: serde_json::Value) -> serde_json::Value {
match value {
serde_json::Value::String(text) => serde_json::Value::String(shell::redact_secrets(&text)),
serde_json::Value::Array(items) => serde_json::Value::Array(items.into_iter().map(redact_json_value).collect()),
serde_json::Value::Object(items) => serde_json::Value::Object(
items
.into_iter()
.map(|(key, value)| (key, redact_json_value(value)))
.collect(),
),
value => value,
}
}
pub fn read_redacted_log_tail(path: &Path, max_lines: usize) -> Vec<String> {
if max_lines == 0 {
return Vec::new();
}
let Ok(file) = File::open(path) else {
return Vec::new();
};
let mut lines = VecDeque::with_capacity(max_lines);
for line in BufReader::new(file).lines().map_while(Result::ok) {
if lines.len() == max_lines {
lines.pop_front();
}
lines.push_back(shell::redact_secrets(&line));
}
let mut bytes = 0usize;
let mut output = Vec::new();
for line in lines.into_iter().rev() {
let line_bytes = line.len().saturating_add(1);
if bytes.saturating_add(line_bytes) > MAX_LOG_OUTPUT_BYTES {
break;
}
bytes = bytes.saturating_add(line_bytes);
output.push(line);
}
output.reverse();
output
}
fn split_tool_name_id(name: &str) -> (String, String) {
match name.rsplit_once('#') {
Some((n, id)) => (n.to_string(), id.to_string()),
None => (name.to_string(), "?".to_string()),
}
}
fn mcp_tool_session_meta(name: &str) -> Option<McpToolSessionMeta> {
let rest = name.strip_prefix("mcp__")?;
let (server_name, original_tool_name) = rest.split_once("__")?;
if server_name.is_empty() || original_tool_name.is_empty() {
return None;
}
Some(McpToolSessionMeta {
server_name: server_name.to_string(),
original_tool_name: original_tool_name.to_string(),
})
}
fn set_seq(record: &mut SessionRecord, seq: u64) {
match record {
SessionRecord::SessionMeta { seq: s, .. }
| SessionRecord::Context { seq: s, .. }
| SessionRecord::ContextLedger { seq: s, .. }
| SessionRecord::ContextPin { seq: s, .. }
| SessionRecord::ContextDrop { seq: s, .. }
| SessionRecord::ContextRecovery { seq: s, .. }
| SessionRecord::Compaction { seq: s, .. }
| SessionRecord::CompactionReview { seq: s, .. }
| SessionRecord::User { seq: s, .. }
| SessionRecord::PromptMetadata { seq: s, .. }
| SessionRecord::AssistantFinished { seq: s, .. }
| SessionRecord::ReasoningFinished { seq: s, .. }
| SessionRecord::Usage { seq: s, .. }
| SessionRecord::RequestAccounting { seq: s, .. }
| SessionRecord::ToolStarted { seq: s, .. }
| SessionRecord::ToolFinished { seq: s, .. }
| SessionRecord::Cancelled { seq: s, .. }
| SessionRecord::Failed { seq: s, .. }
| SessionRecord::AcpSession { seq: s, .. }
| SessionRecord::SessionRenamed { seq: s, .. }
| SessionRecord::FileWrite { seq: s, .. }
| SessionRecord::McpConfigChanged { seq: s, .. }
| SessionRecord::ShellExec { seq: s, .. }
| SessionRecord::SkillActivated { seq: s, .. }
| SessionRecord::QueuedInput { seq: s, .. }
| SessionRecord::AcpPermissionRequest { seq: s, .. }
| SessionRecord::AcpPermissionOutcome { seq: s, .. } => *s = seq,
}
}