mod jsonl;
mod mapping;
use std::path::PathBuf;
use serde::{Deserialize, Serialize};
use serde_json::Value;
pub use jsonl::JsonlWriter;
pub const EVENT_SCHEMA_VERSION: u32 = 2;
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct EventLine {
pub seq: u64,
#[serde(flatten)]
pub event: Event,
}
impl EventLine {
pub fn new(seq: u64, event: Event) -> Self {
Self { seq, event }
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Mutability {
ReadOnly,
Mutating,
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PermissionOutcome {
Allowed,
Denied,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RuleScope {
Session,
Project,
Global,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NoticeSeverity {
#[default]
Info,
Warning,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TaskKind {
Subagent,
BackgroundTask,
Teammate,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TaskStatus {
Spawned,
Running,
Finished,
Failed,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum ToolResultContentKind {
Text,
Structured,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum ToolResultElisionAction {
Preview,
Marker,
Omitted,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum RequestToolResultElisionPolicy {
KeepRecent {
configured_keep_recent_tool_results: usize,
},
ByteBudget {
configured_max_bytes: usize,
configured_prioritize_recent_results: usize,
configured_max_preview_bytes: usize,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ElidedToolResult {
pub tool_call_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub tool_name: Option<String>,
pub is_error: bool,
pub canonical_content_kind: ToolResultContentKind,
pub action: ToolResultElisionAction,
pub canonical_content_bytes: usize,
pub projected_content_bytes: usize,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "snake_case")]
#[non_exhaustive]
pub enum RunOutcome {
Ok,
Error { message: String },
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SkillSummary {
pub name: String,
pub description: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TemplateSummary {
pub name: String,
pub description: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub argument_hint: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ContextFile {
pub path: PathBuf,
pub scope: String,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
#[non_exhaustive]
pub enum Event {
RunStarted {
schema: u32,
basis: String,
session_id: String,
workspace: PathBuf,
model: String,
provider: String,
context_files: Vec<ContextFile>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
skills_dirs: Vec<PathBuf>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
skills: Vec<SkillSummary>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
templates_dirs: Vec<PathBuf>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
templates: Vec<TemplateSummary>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
mcp_files: Vec<ContextFile>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
mcp_servers: Vec<String>,
},
UserMessage {
text: String,
#[serde(default, skip_serializing_if = "is_zero")]
image_count: usize,
},
AssistantDelta {
text: String,
},
AssistantReasoningDelta {
text: String,
},
AssistantMessage {
text: String,
},
ToolQueued {
tool_call_id: String,
tool_name: String,
summary: String,
mutability: Mutability,
input: Value,
},
ToolStarted {
tool_call_id: String,
tool_name: String,
},
ToolProgress {
tool_call_id: String,
tool_name: String,
progress: String,
},
ToolCompleted {
tool_call_id: String,
tool_name: String,
summary: String,
is_error: bool,
},
PermissionRequested {
request_id: String,
tool_call_id: String,
tool_name: String,
description: String,
preview: Value,
},
PermissionResolved {
request_id: String,
tool_call_id: String,
tool_name: String,
outcome: PermissionOutcome,
#[serde(skip_serializing_if = "Option::is_none")]
rule_scope: Option<RuleScope>,
},
TaskUpdated {
task_id: String,
kind: TaskKind,
status: TaskStatus,
title: String,
#[serde(skip_serializing_if = "Option::is_none")]
detail: Option<String>,
},
CompactionStarted {
agent_id: String,
},
CompactionCompleted {
agent_id: String,
replaced_items: usize,
preserved_items: usize,
transcript_len: usize,
extracted_facts: usize,
summary_preview: String,
},
RequestToolResultsElided {
agent_id: String,
policy: RequestToolResultElisionPolicy,
canonical_tool_result_content_bytes: usize,
projected_tool_result_content_bytes: usize,
results: Vec<ElidedToolResult>,
},
MemoryUpdated {
agent_id: String,
stored_records: usize,
},
Usage {
agent_id: String,
input_tokens: u64,
output_tokens: u64,
cache_read_tokens: u64,
cache_creation_tokens: u64,
#[serde(default)]
reasoning_tokens: u64,
#[serde(default)]
thoughts_tokens: u64,
},
Notice {
#[serde(default)]
severity: NoticeSeverity,
message: String,
},
Retry {
agent_id: String,
error: String,
attempt: u32,
max_attempts: u32,
next_delay_ms: u64,
},
Error {
message: String,
recoverable: bool,
},
Branched {
entry_id: String,
abandoned_entries: usize,
},
RunFinished {
#[serde(flatten)]
outcome: RunOutcome,
#[serde(skip_serializing_if = "Option::is_none", default)]
stopped_by: Option<crate::run::Bound>,
#[serde(skip_serializing_if = "Option::is_none", default)]
usage: Option<crate::run::RunUsage>,
},
}
impl Event {
pub fn from_session_event(event: &mentra::SessionEvent) -> Option<Self> {
mapping::from_session_event(event)
}
pub fn type_tag(&self) -> &'static str {
match self {
Event::RunStarted { .. } => "run_started",
Event::UserMessage { .. } => "user_message",
Event::AssistantDelta { .. } => "assistant_delta",
Event::AssistantReasoningDelta { .. } => "assistant_reasoning_delta",
Event::AssistantMessage { .. } => "assistant_message",
Event::ToolQueued { .. } => "tool_queued",
Event::ToolStarted { .. } => "tool_started",
Event::ToolProgress { .. } => "tool_progress",
Event::ToolCompleted { .. } => "tool_completed",
Event::PermissionRequested { .. } => "permission_requested",
Event::PermissionResolved { .. } => "permission_resolved",
Event::TaskUpdated { .. } => "task_updated",
Event::CompactionStarted { .. } => "compaction_started",
Event::CompactionCompleted { .. } => "compaction_completed",
Event::RequestToolResultsElided { .. } => "request_tool_results_elided",
Event::MemoryUpdated { .. } => "memory_updated",
Event::Usage { .. } => "usage",
Event::Notice { .. } => "notice",
Event::Retry { .. } => "retry",
Event::Error { .. } => "error",
Event::Branched { .. } => "branched",
Event::RunFinished { .. } => "run_finished",
}
}
}
fn is_zero(count: &usize) -> bool {
*count == 0
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_type_tag_is_the_tag_serde_writes() {
for event in [
Event::AssistantDelta {
text: "hi".to_string(),
},
Event::Notice {
severity: NoticeSeverity::Info,
message: "m".to_string(),
},
Event::RequestToolResultsElided {
agent_id: "a-1".to_string(),
policy: RequestToolResultElisionPolicy::KeepRecent {
configured_keep_recent_tool_results: 3,
},
canonical_tool_result_content_bytes: 1_024,
projected_tool_result_content_bytes: 128,
results: Vec::new(),
},
Event::RunFinished {
outcome: RunOutcome::Ok,
stopped_by: None,
usage: None,
},
] {
let written = serde_json::to_value(&event).expect("serializes");
assert_eq!(written["type"].as_str().expect("tagged"), event.type_tag());
}
}
#[test]
fn a_line_is_one_flat_object() {
let line = EventLine::new(
7,
Event::AssistantDelta {
text: "hi".to_string(),
},
);
let json = serde_json::to_value(&line).expect("serializes");
assert_eq!(json["seq"], 7);
assert_eq!(json["type"], "assistant_delta");
assert_eq!(json["text"], "hi");
assert!(json.get("event").is_none(), "envelope must stay flat");
}
#[test]
fn the_header_carries_the_schema_version() {
let line = EventLine::new(
0,
Event::RunStarted {
schema: EVENT_SCHEMA_VERSION,
basis: "0.1.0".to_string(),
session_id: "s1".to_string(),
workspace: PathBuf::from("/repo"),
model: "gpt-5".to_string(),
provider: "openai".to_string(),
context_files: vec![ContextFile {
path: PathBuf::from("/repo/AGENTS.md"),
scope: "workspace".to_string(),
}],
skills_dirs: Vec::new(),
skills: Vec::new(),
templates_dirs: Vec::new(),
templates: Vec::new(),
mcp_files: Vec::new(),
mcp_servers: Vec::new(),
},
);
let json = serde_json::to_value(&line).expect("serializes");
assert_eq!(json["type"], "run_started");
assert_eq!(json["schema"], 2, "the new event tag requires schema 2");
assert_eq!(json["context_files"][0]["scope"], "workspace");
assert!(
json.get("skills_dirs").is_none() && json.get("skills").is_none(),
"a run without skills must not mention them"
);
}
#[test]
fn request_tool_result_elision_json_is_typed_and_body_free() {
let line = EventLine::new(
9,
Event::RequestToolResultsElided {
agent_id: "agent-1".to_string(),
policy: RequestToolResultElisionPolicy::ByteBudget {
configured_max_bytes: 4_096,
configured_prioritize_recent_results: 2,
configured_max_preview_bytes: 512,
},
canonical_tool_result_content_bytes: 8_192,
projected_tool_result_content_bytes: 4_096,
results: vec![ElidedToolResult {
tool_call_id: "call-1".to_string(),
tool_name: Some("grep".to_string()),
is_error: false,
canonical_content_kind: ToolResultContentKind::Text,
action: ToolResultElisionAction::Preview,
canonical_content_bytes: 8_192,
projected_content_bytes: 4_096,
}],
},
);
let json = serde_json::to_value(&line).expect("serializes");
assert_eq!(json["seq"], 9);
assert_eq!(json["type"], "request_tool_results_elided");
assert_eq!(json["policy"]["kind"], "byte_budget");
assert_eq!(json["policy"]["configured_max_bytes"], 4_096);
assert_eq!(json["results"][0]["canonical_content_kind"], "text");
assert_eq!(json["results"][0]["action"], "preview");
assert!(
json["results"][0].get("content").is_none(),
"the event must not duplicate the canonical or projected body"
);
let restored: EventLine = serde_json::from_value(json).expect("round trips");
assert_eq!(restored, line);
}
#[test]
fn skills_are_reported_when_there_are_any() {
let line = EventLine::new(
0,
Event::RunStarted {
schema: EVENT_SCHEMA_VERSION,
basis: "0.1.0".to_string(),
session_id: "s1".to_string(),
workspace: PathBuf::from("/repo"),
model: "gpt-5".to_string(),
provider: "openai".to_string(),
context_files: Vec::new(),
skills_dirs: vec![PathBuf::from("/repo/.basis/skills")],
skills: vec![SkillSummary {
name: "review".to_string(),
description: "house review style".to_string(),
}],
templates_dirs: Vec::new(),
templates: Vec::new(),
mcp_files: Vec::new(),
mcp_servers: Vec::new(),
},
);
let json = serde_json::to_value(&line).expect("serializes");
assert_eq!(json["skills_dirs"][0], "/repo/.basis/skills");
assert_eq!(json["skills"][0]["name"], "review");
assert!(
!json["skills"][0]
.as_object()
.expect("an object")
.contains_key("path"),
"the stream carries what the model can load, not where it lives on this machine"
);
}
#[test]
fn run_outcome_flattens_into_the_finish_line() {
let ok = serde_json::to_value(EventLine::new(
3,
Event::RunFinished {
outcome: RunOutcome::Ok,
stopped_by: None,
usage: None,
},
))
.expect("serializes");
assert_eq!(ok["type"], "run_finished");
assert_eq!(ok["status"], "ok");
assert!(
!ok.as_object()
.expect("an object")
.contains_key("stopped_by"),
"an unbounded finish is byte-identical to what a schema-1 consumer already reads"
);
let failed = serde_json::to_value(EventLine::new(
3,
Event::RunFinished {
outcome: RunOutcome::Error {
message: "boom".to_string(),
},
stopped_by: None,
usage: None,
},
))
.expect("serializes");
assert_eq!(failed["status"], "error");
assert_eq!(failed["message"], "boom");
}
#[test]
fn a_finish_line_reports_what_the_run_spent() {
let line = serde_json::to_value(EventLine::new(
4,
Event::RunFinished {
outcome: RunOutcome::Ok,
stopped_by: None,
usage: Some(crate::RunUsage {
input_tokens: 12_300,
output_tokens: 1_200,
cache_read_tokens: 40,
cache_creation_tokens: 5,
reasoning_tokens: 300,
thoughts_tokens: 0,
model_responses: 2,
}),
},
))
.expect("serializes");
assert_eq!(line["usage"]["input_tokens"], 12_300);
assert_eq!(line["usage"]["output_tokens"], 1_200);
assert_eq!(line["usage"]["cache_read_tokens"], 40);
assert_eq!(line["usage"]["cache_creation_tokens"], 5);
let unreported = serde_json::to_value(EventLine::new(
4,
Event::RunFinished {
outcome: RunOutcome::Ok,
stopped_by: None,
usage: None,
},
))
.expect("serializes");
assert!(
!unreported
.as_object()
.expect("an object")
.contains_key("usage"),
"a producer that reported nothing says nothing, and the line keeps its schema-1 shape"
);
let read_back: EventLine =
serde_json::from_value(unreported).expect("a line without usage still parses");
assert!(matches!(
read_back.event,
Event::RunFinished { usage: None, .. }
));
}
#[test]
fn a_bounded_finish_names_its_bound_on_the_stream() {
let line = serde_json::to_value(EventLine::new(
2,
Event::RunFinished {
outcome: RunOutcome::Ok,
stopped_by: Some(crate::run::Bound::TokenBudget),
usage: None,
},
))
.expect("serializes");
assert_eq!(line["type"], "run_finished");
assert_eq!(line["status"], "ok");
assert_eq!(line["stopped_by"], "token_budget");
}
#[test]
fn the_header_names_mcp_files_and_servers_but_never_their_configuration() {
let line = EventLine::new(
0,
Event::RunStarted {
schema: EVENT_SCHEMA_VERSION,
basis: "0.1.0".to_string(),
session_id: "s1".to_string(),
workspace: PathBuf::from("/repo"),
model: "gpt-5".to_string(),
provider: "openai".to_string(),
context_files: Vec::new(),
skills_dirs: Vec::new(),
skills: Vec::new(),
templates_dirs: Vec::new(),
templates: Vec::new(),
mcp_files: vec![ContextFile {
path: PathBuf::from("/repo/.mcp.json"),
scope: "workspace".to_string(),
}],
mcp_servers: vec!["github".to_string()],
},
);
let text = serde_json::to_string(&line).expect("serializes");
assert!(text.contains("/repo/.mcp.json"), "the file must be named");
assert!(text.contains("github"), "so must the server");
for leak in ["command", "args", "env", "npx", "token"] {
assert!(
!text.contains(leak),
"the header must not carry MCP configuration, found {leak}: {text}"
);
}
}
#[test]
fn absent_optionals_are_omitted_not_null() {
let json = serde_json::to_value(EventLine::new(
1,
Event::TaskUpdated {
task_id: "t1".to_string(),
kind: TaskKind::Subagent,
status: TaskStatus::Running,
title: "work".to_string(),
detail: None,
},
))
.expect("serializes");
assert!(json.get("detail").is_none());
}
#[test]
fn lines_round_trip() {
let line = EventLine::new(
2,
Event::ToolCompleted {
tool_call_id: "c1".to_string(),
tool_name: "shell".to_string(),
summary: "ok".to_string(),
is_error: false,
},
);
let text = serde_json::to_string(&line).expect("serializes");
let back: EventLine = serde_json::from_str(&text).expect("deserializes");
assert_eq!(line, back);
}
}