use std::path::{Path, PathBuf};
use std::sync::Mutex;
use tokio::sync::{broadcast, watch};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;
use crate::event::{Event, EventSink, FlowRunId, TurnId};
use crate::event_log::reader::{find_last_seq, replay_context_snapshot_from};
use crate::event_writer::EventWriter;
use crate::injection::{Injection, InjectionId, InjectionState};
use crate::message::{Message, MessageRole};
use crate::projection::message_window::{
TranscriptEntry, replay_all_messages_with_seq, replay_messages_from, replay_messages_with_seq,
replay_transcript_from,
};
use crate::stream::StreamFrame;
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct SessionId(pub Uuid);
fn is_auto_name_threshold(mut count: u64) -> bool {
if count < 3 {
return false;
}
while count % 3 == 0 {
count /= 3;
}
count == 1
}
impl SessionId {
pub fn now() -> Self {
Self(Uuid::new_v4())
}
pub fn parse(s: &str) -> Result<Self, uuid::Error> {
Uuid::parse_str(s).map(Self)
}
}
impl std::fmt::Display for SessionId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.0.fmt(f)
}
}
type WatchKeepalive = (
watch::Receiver<ContextSnapshot>,
watch::Receiver<Option<String>>,
watch::Receiver<usize>,
watch::Receiver<Vec<crate::memory::todo::Todo>>,
watch::Receiver<Vec<crate::memory::plan::Plan>>,
);
#[derive(Debug)]
pub struct TurnState {
pub current_turn: Mutex<Option<TurnId>>,
pub flow_cancel: Mutex<CancellationToken>,
pub streamed: std::sync::atomic::AtomicBool,
}
impl TurnState {
fn new() -> Self {
Self {
current_turn: Mutex::new(None),
flow_cancel: Mutex::new(CancellationToken::new()),
streamed: std::sync::atomic::AtomicBool::new(false),
}
}
}
pub struct WatchHub {
pub stream_tx: broadcast::Sender<StreamFrame>,
pub context: watch::Sender<ContextSnapshot>,
pub goal: watch::Sender<Option<String>>,
pub attach: watch::Sender<usize>,
pub todos: watch::Sender<Vec<crate::memory::todo::Todo>>,
pub plans: watch::Sender<Vec<crate::memory::plan::Plan>>,
_keepalive: WatchKeepalive,
}
pub struct CompactionState {
pub manual_pending: std::sync::atomic::AtomicBool,
pub last_input_tokens: std::sync::atomic::AtomicU64,
pub review_mode: Mutex<CompactReviewMode>,
pub lock: std::sync::Arc<tokio::sync::Mutex<()>>,
}
impl CompactionState {
fn new() -> Self {
Self {
manual_pending: std::sync::atomic::AtomicBool::new(false),
last_input_tokens: std::sync::atomic::AtomicU64::new(0),
review_mode: Mutex::new(CompactReviewMode::default()),
lock: std::sync::Arc::new(tokio::sync::Mutex::new(())),
}
}
}
pub struct InteractionServices {
pub approval: std::sync::Arc<ApprovalRegistry>,
pub compact_reviews: std::sync::Arc<CompactReviewRegistry>,
pub forms: std::sync::Arc<FormRegistry>,
}
impl InteractionServices {
fn new() -> Self {
Self {
approval: std::sync::Arc::new(ApprovalRegistry::new()),
compact_reviews: std::sync::Arc::new(CompactReviewRegistry::new()),
forms: std::sync::Arc::new(FormRegistry::new()),
}
}
}
pub struct Session {
id: SessionId,
dir: PathBuf,
writer: std::sync::Mutex<Option<EventWriter>>,
sink: EventSink,
message_stream: crate::message_stream::MessageStream,
messages: std::sync::Arc<std::sync::Mutex<Vec<Message>>>,
pub turn: TurnState,
pub watch: WatchHub,
pub watch_hub: std::sync::Arc<crate::watch::WatchHub>,
pub flow_registry: std::sync::Arc<crate::tools::agent_ctrl::FlowRegistry>,
current_root: std::sync::Mutex<Option<String>>,
successful_flow_count: std::sync::atomic::AtomicU64,
pub compaction: CompactionState,
pub interactions: InteractionServices,
injection_queue: Mutex<Vec<Injection>>,
injection_tx: broadcast::Sender<Injection>,
last_image_user_msg: Mutex<Option<LastImageUserMsg>>,
read_files: std::sync::Arc<std::sync::Mutex<std::collections::HashSet<std::path::PathBuf>>>,
output_store: std::sync::Arc<crate::tools::tool_output::OutputStore>,
fs_access_mode: Mutex<Option<crate::fs_access::FsAccessMode>>,
project_index: Option<std::sync::Arc<crate::index::AnchorIndex>>,
}
#[derive(Debug, Clone)]
pub struct PendingCompactReview {
pub review_id: String,
pub summary: String,
pub slice_preview: String,
pub slice_count: usize,
pub range_start: usize,
pub range_end: usize,
pub tokens_before: u64,
pub emitted_at: chrono::DateTime<chrono::Utc>,
}
#[derive(Debug, Clone)]
pub enum CompactReviewDecision {
AcceptAsIs,
AcceptEdited { summary: String },
Reject,
}
pub struct CompactReviewRegistry {
entry: std::sync::Mutex<Option<CompactReviewEntry>>,
watch_tx: watch::Sender<Option<PendingCompactReview>>,
}
struct CompactReviewEntry {
pending: PendingCompactReview,
responder: tokio::sync::oneshot::Sender<CompactReviewDecision>,
}
impl Default for CompactReviewRegistry {
fn default() -> Self {
Self::new()
}
}
impl CompactReviewRegistry {
pub fn new() -> Self {
let (watch_tx, _) = watch::channel(None);
Self {
entry: std::sync::Mutex::new(None),
watch_tx,
}
}
pub fn subscribe(&self) -> watch::Receiver<Option<PendingCompactReview>> {
self.watch_tx.subscribe()
}
pub fn list_pending(&self) -> Option<PendingCompactReview> {
self.entry
.lock()
.unwrap()
.as_ref()
.map(|e| e.pending.clone())
}
pub fn subscriber_count(&self) -> usize {
self.watch_tx.receiver_count()
}
pub fn request(
&self,
pending: PendingCompactReview,
) -> tokio::sync::oneshot::Receiver<CompactReviewDecision> {
let (tx, rx) = tokio::sync::oneshot::channel();
if self.watch_tx.receiver_count() == 0 {
let _ = tx.send(CompactReviewDecision::AcceptAsIs);
return rx;
}
{
let mut slot = self.entry.lock().unwrap();
if let Some(prev) = slot.take() {
let _ = prev.responder.send(CompactReviewDecision::Reject);
}
*slot = Some(CompactReviewEntry {
pending: pending.clone(),
responder: tx,
});
}
let _ = self.watch_tx.send(Some(pending));
rx
}
pub fn decide(&self, review_id: &str, decision: CompactReviewDecision) -> bool {
let entry = {
let mut slot = self.entry.lock().unwrap();
match slot.as_ref() {
Some(e) if e.pending.review_id == review_id => slot.take(),
_ => None,
}
};
match entry {
Some(e) => {
let _ = e.responder.send(decision);
let _ = self.watch_tx.send(None);
true
}
None => false,
}
}
}
#[derive(Debug, Clone)]
pub struct PendingApproval {
pub tool_use_id: String,
pub tool_name: String,
pub args_preview: String,
pub preview: Option<String>,
pub level: crate::tool::ApprovalLevel,
pub run_id: FlowRunId,
pub emitted_at: chrono::DateTime<chrono::Utc>,
pub bypass_auto_ceiling: bool,
}
#[derive(Debug, Clone)]
pub enum ApprovalDecision {
Approve,
Deny { reason: String },
}
pub struct FormRegistry {
entries: std::sync::Mutex<Vec<FormEntry>>,
watch_tx: watch::Sender<Vec<crate::form::PendingForm>>,
}
struct FormEntry {
pending: crate::form::PendingForm,
responder: tokio::sync::oneshot::Sender<crate::form::FormAnswer>,
}
impl Default for FormRegistry {
fn default() -> Self {
Self::new()
}
}
impl FormRegistry {
pub fn new() -> Self {
let (watch_tx, _) = watch::channel(Vec::new());
Self {
entries: std::sync::Mutex::new(Vec::new()),
watch_tx,
}
}
pub fn subscribe(&self) -> watch::Receiver<Vec<crate::form::PendingForm>> {
self.watch_tx.subscribe()
}
pub fn list_pending(&self) -> Vec<crate::form::PendingForm> {
self.entries
.lock()
.unwrap()
.iter()
.map(|e| e.pending.clone())
.collect()
}
pub fn subscriber_count(&self) -> usize {
self.watch_tx.receiver_count()
}
pub fn request(
&self,
pending: crate::form::PendingForm,
) -> tokio::sync::oneshot::Receiver<crate::form::FormAnswer> {
let (tx, rx) = tokio::sync::oneshot::channel();
if self.watch_tx.receiver_count() == 0 {
let _ = tx.send(crate::form::FormAnswer::Cancelled);
return rx;
}
{
let mut entries = self.entries.lock().unwrap();
entries.push(FormEntry {
pending: pending.clone(),
responder: tx,
});
}
self.broadcast_snapshot();
rx
}
pub fn submit(&self, form_id: &str, answer: crate::form::FormAnswer) -> bool {
let entry = {
let mut entries = self.entries.lock().unwrap();
let pos = entries.iter().position(|e| e.pending.form_id == form_id);
pos.map(|p| entries.remove(p))
};
match entry {
Some(e) => {
let _ = e.responder.send(answer);
self.broadcast_snapshot();
true
}
None => false,
}
}
pub fn cancel_all(&self) {
let drained: Vec<FormEntry> = {
let mut entries = self.entries.lock().unwrap();
std::mem::take(&mut *entries)
};
for e in drained {
let _ = e.responder.send(crate::form::FormAnswer::Cancelled);
}
self.broadcast_snapshot();
}
pub fn promote(&self, form_id: &str) {
let mut entries = self.entries.lock().unwrap();
if let Some(pos) = entries.iter().position(|e| e.pending.form_id == form_id) {
if pos == 0 {
return;
}
let entry = entries.remove(pos);
entries.insert(0, entry);
}
drop(entries);
self.broadcast_snapshot();
}
fn broadcast_snapshot(&self) {
let snap = self
.entries
.lock()
.unwrap()
.iter()
.map(|e| e.pending.clone())
.collect();
let _ = self.watch_tx.send(snap);
}
}
pub struct ApprovalRegistry {
entries: std::sync::Mutex<Vec<ApprovalEntry>>,
auto_ceiling: std::sync::Mutex<crate::tool::ApprovalLevel>,
watch_tx: watch::Sender<Vec<PendingApproval>>,
}
struct ApprovalEntry {
pending: PendingApproval,
responder: tokio::sync::oneshot::Sender<ApprovalDecision>,
}
impl Default for ApprovalRegistry {
fn default() -> Self {
Self::new()
}
}
impl ApprovalRegistry {
pub fn new() -> Self {
let (watch_tx, _) = watch::channel(Vec::new());
Self {
entries: std::sync::Mutex::new(Vec::new()),
auto_ceiling: std::sync::Mutex::new(crate::tool::ApprovalLevel::Approve),
watch_tx,
}
}
pub fn subscribe(&self) -> watch::Receiver<Vec<PendingApproval>> {
self.watch_tx.subscribe()
}
pub fn list_pending(&self) -> Vec<PendingApproval> {
self.entries
.lock()
.unwrap()
.iter()
.map(|e| e.pending.clone())
.collect()
}
pub fn set_auto_ceiling(&self, level: crate::tool::ApprovalLevel) {
*self.auto_ceiling.lock().unwrap() = level;
}
pub fn request(
&self,
pending: PendingApproval,
) -> tokio::sync::oneshot::Receiver<ApprovalDecision> {
let (tx, rx) = tokio::sync::oneshot::channel();
if !pending.bypass_auto_ceiling && pending.level <= *self.auto_ceiling.lock().unwrap() {
let _ = tx.send(ApprovalDecision::Approve);
return rx;
}
{
let mut entries = self.entries.lock().unwrap();
entries.push(ApprovalEntry {
pending,
responder: tx,
});
}
self.broadcast_snapshot();
rx
}
pub fn decide(&self, tool_use_id: &str, decision: ApprovalDecision) -> bool {
let mut entries = self.entries.lock().unwrap();
if let Some(pos) = entries
.iter()
.position(|e| e.pending.tool_use_id == tool_use_id)
{
let entry = entries.remove(pos);
let _ = entry.responder.send(decision);
drop(entries);
self.broadcast_snapshot();
true
} else {
false
}
}
pub fn decide_all(&self, decision: ApprovalDecision) -> usize {
let mut entries = self.entries.lock().unwrap();
let count = entries.len();
for entry in entries.drain(..) {
let _ = entry.responder.send(decision.clone());
}
drop(entries);
self.broadcast_snapshot();
count
}
fn broadcast_snapshot(&self) {
let snapshot = self
.entries
.lock()
.unwrap()
.iter()
.map(|e| e.pending.clone())
.collect();
let _ = self.watch_tx.send(snapshot);
}
}
type ImagePart = (usize, String);
#[derive(Debug, Clone)]
struct LastImageUserMsg {
message_seq: u64,
images: Vec<ImagePart>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum CompactReviewMode {
Always,
#[default]
ManualOnly,
Never,
}
impl CompactReviewMode {
pub fn parse(s: &str) -> Option<Self> {
match s.trim() {
"always" => Some(Self::Always),
"manual-only" | "manual_only" => Some(Self::ManualOnly),
"never" => Some(Self::Never),
_ => None,
}
}
pub fn should_review(self, forced: bool) -> bool {
match self {
Self::Always => true,
Self::ManualOnly => forced,
Self::Never => false,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CompactResult {
pub before_tokens: u64,
pub after_tokens: u64,
pub compacted_start: usize,
pub compacted_end: usize,
}
#[derive(Debug, Clone, Default, PartialEq)]
pub struct ContextSnapshot {
pub model: String,
pub tokens_in: u64,
pub tokens_out: u64,
pub cost_usd: f64,
pub mcp_servers: Vec<crate::mcp::McpServerStatus>,
pub memory_recent_count: u16,
pub window_tokens: u64,
pub window_budget: u64,
pub cache_read: u64,
pub cache_write: u64,
pub last_ttft_ms: u64,
pub last_tokens_per_sec: f64,
}
#[derive(Debug, thiserror::Error)]
pub enum SessionOpenError {
#[error("invalid session id `{sid}` (want a UUID)")]
InvalidId { sid: String },
#[error("session `{sid}` not found at {}", dir.display())]
NotFound { sid: String, dir: PathBuf },
#[error("session writer init: {0}")]
WriterInit(#[source] std::io::Error),
#[error("replay {}: {source}", path.display())]
Replay {
path: PathBuf,
#[source]
source: std::io::Error,
},
}
fn load_goal(dir: &Path) -> Option<String> {
if dir.as_os_str().is_empty() {
return None;
}
let store = crate::memory::goal::GoalStore::at(dir);
match store.get() {
Ok(s) if !s.is_empty() => Some(s),
_ => None,
}
}
#[derive(serde::Serialize, serde::Deserialize, Default)]
struct PersistedContextState {
#[serde(default)]
model: String,
#[serde(default)]
window_tokens: u64,
#[serde(default)]
window_budget: u64,
}
impl PersistedContextState {
fn path(dir: &Path) -> PathBuf {
dir.join("context_state.json")
}
fn load(dir: &Path) -> Self {
match std::fs::read_to_string(Self::path(dir)) {
Ok(text) => serde_json::from_str(&text).unwrap_or_default(),
Err(_) => Self::default(),
}
}
fn save(&self, dir: &Path) {
if dir.as_os_str().is_empty() {
return;
}
if let Ok(json) = serde_json::to_string_pretty(self) {
let _ = std::fs::write(Self::path(dir), &json);
}
}
}
fn default_project_index(root: &Path) -> Option<std::sync::Arc<crate::index::AnchorIndex>> {
match crate::index::AnchorIndex::open_project(root) {
Ok(idx) => Some(std::sync::Arc::new(idx)),
Err(e) => {
crate::notify!(
warn,
"project index unavailable at {} — history search disabled: {e}",
root.display()
);
None
}
}
}
impl Session {
pub fn open(root: impl AsRef<Path>) -> std::io::Result<Self> {
Self::open_with_redactor(root, None)
}
pub fn open_with_redactor(
root: impl AsRef<Path>,
redactor: Option<std::sync::Arc<crate::redact::Redactor>>,
) -> std::io::Result<Self> {
let root_ref = root.as_ref();
let project_index = default_project_index(root_ref);
Self::open_with_context(root_ref, redactor, project_index)
}
pub fn open_with_context(
root: impl AsRef<Path>,
redactor: Option<std::sync::Arc<crate::redact::Redactor>>,
project_index: Option<std::sync::Arc<crate::index::AnchorIndex>>,
) -> std::io::Result<Self> {
let id = SessionId::now();
let dir = root.as_ref().join("sessions").join(id.to_string());
if let Some(ls) = crate::notify::log_sink() {
ls.set_session_id(Some(id.to_string()));
}
let writer = EventWriter::spawn_full(
&dir,
redactor.clone(),
project_index.clone(),
Some(id.to_string()),
)?;
if let Err(e) = crate::session_meta::SessionMeta::from_cwd().save(&dir) {
crate::notify!(error, "session meta write failed: {e}");
}
let mut sink = EventSink::new().with_forwarder(writer.sender());
if let Some(r) = redactor {
sink = sink.with_redactor(r);
}
let (injection_tx, _) = broadcast::channel(32);
let (stream_tx, _) = broadcast::channel(2048);
let (context_watch, context_rx) = watch::channel(ContextSnapshot::default());
let (goal_watch, goal_rx) = watch::channel(None);
let (attach_watch, attach_rx) = watch::channel(0);
let (todos_watch, todos_rx) = watch::channel(Vec::new());
let (plans_watch, plans_rx) = watch::channel(Vec::new());
let events_handle = sink.events_handle();
let output_store = std::sync::Arc::new(crate::tools::tool_output::OutputStore::at(&dir));
Ok(Self {
id,
dir,
writer: std::sync::Mutex::new(Some(writer)),
sink,
message_stream: crate::message_stream::MessageStream::new(events_handle),
messages: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
output_store: output_store.clone(),
turn: TurnState::new(),
watch: WatchHub {
stream_tx,
context: context_watch,
goal: goal_watch,
attach: attach_watch,
todos: todos_watch,
plans: plans_watch,
_keepalive: (context_rx, goal_rx, attach_rx, todos_rx, plans_rx),
},
watch_hub: std::sync::Arc::new(crate::watch::WatchHub::new()),
flow_registry: std::sync::Arc::new(crate::tools::agent_ctrl::FlowRegistry::new()),
current_root: std::sync::Mutex::new(None),
successful_flow_count: std::sync::atomic::AtomicU64::new(0),
compaction: CompactionState::new(),
interactions: InteractionServices::new(),
injection_queue: Mutex::new(Vec::new()),
injection_tx,
last_image_user_msg: Mutex::new(None),
read_files: std::sync::Arc::new(
std::sync::Mutex::new(std::collections::HashSet::new()),
),
fs_access_mode: Mutex::new(None),
project_index,
})
}
pub fn open_existing(root: impl AsRef<Path>, sid: &str) -> Result<Self, SessionOpenError> {
Self::open_existing_with_redactor(root, sid, None)
}
pub fn open_existing_with_redactor(
root: impl AsRef<Path>,
sid: &str,
redactor: Option<std::sync::Arc<crate::redact::Redactor>>,
) -> Result<Self, SessionOpenError> {
let project_index = default_project_index(root.as_ref());
Self::open_existing_with_context(root, sid, redactor, project_index)
}
pub fn open_existing_with_context(
root: impl AsRef<Path>,
sid: &str,
redactor: Option<std::sync::Arc<crate::redact::Redactor>>,
project_index: Option<std::sync::Arc<crate::index::AnchorIndex>>,
) -> Result<Self, SessionOpenError> {
let id = SessionId::parse(sid).map_err(|_| SessionOpenError::InvalidId {
sid: sid.to_string(),
})?;
let dir = root.as_ref().join("sessions").join(id.to_string());
if let Some(ls) = crate::notify::log_sink() {
ls.set_session_id(Some(id.to_string()));
}
if !dir.exists() {
return Err(SessionOpenError::NotFound {
sid: sid.to_string(),
dir: dir.clone(),
});
}
let writer = EventWriter::spawn_full(
&dir,
redactor.clone(),
project_index.clone(),
Some(id.to_string()),
)
.map_err(SessionOpenError::WriterInit)?;
let mut sink = EventSink::new().with_forwarder(writer.sender());
if let Some(r) = redactor {
sink = sink.with_redactor(r);
}
let events_path = dir.join("events.jsonl");
let messages = replay_messages_from(&events_path)?;
let initial_msgs = replay_messages_with_seq(&events_path)?;
let all_msgs = replay_all_messages_with_seq(&events_path)?;
if let Some(last_seq) = find_last_seq(&events_path)? {
sink.restore_seq(last_seq);
}
let mut initial_context = replay_context_snapshot_from(&events_path);
let persisted = PersistedContextState::load(&dir);
if !persisted.model.is_empty() {
initial_context.model = persisted.model;
}
initial_context.window_tokens = persisted.window_tokens;
initial_context.window_budget = persisted.window_budget;
let initial_goal = load_goal(&dir);
let (injection_tx, _) = broadcast::channel(32);
let (stream_tx, _) = broadcast::channel(2048);
let (context_watch, context_rx) = watch::channel(initial_context);
let (goal_watch, goal_rx) = watch::channel(initial_goal);
let (attach_watch, attach_rx) = watch::channel(0);
let (todos_watch, todos_rx) = watch::channel(Vec::new());
let (plans_watch, plans_rx) = watch::channel(Vec::new());
let events_handle = sink.events_handle();
let output_store = std::sync::Arc::new(crate::tools::tool_output::OutputStore::at(&dir));
Ok(Self {
id,
dir,
writer: std::sync::Mutex::new(Some(writer)),
sink,
message_stream: crate::message_stream::MessageStream::with_initial(
events_handle,
initial_msgs,
all_msgs,
),
messages: std::sync::Arc::new(std::sync::Mutex::new(messages)),
output_store: output_store.clone(),
turn: TurnState::new(),
watch: WatchHub {
stream_tx,
context: context_watch,
goal: goal_watch,
attach: attach_watch,
todos: todos_watch,
plans: plans_watch,
_keepalive: (context_rx, goal_rx, attach_rx, todos_rx, plans_rx),
},
watch_hub: std::sync::Arc::new(crate::watch::WatchHub::new()),
flow_registry: std::sync::Arc::new(crate::tools::agent_ctrl::FlowRegistry::new()),
current_root: std::sync::Mutex::new(None),
successful_flow_count: std::sync::atomic::AtomicU64::new(0),
compaction: {
let c = CompactionState::new();
if persisted.window_tokens > 0 {
c.last_input_tokens.store(
persisted.window_tokens,
std::sync::atomic::Ordering::Relaxed,
);
}
c
},
interactions: InteractionServices::new(),
injection_queue: Mutex::new(Vec::new()),
injection_tx,
last_image_user_msg: Mutex::new(None),
read_files: std::sync::Arc::new(
std::sync::Mutex::new(std::collections::HashSet::new()),
),
fs_access_mode: Mutex::new(None),
project_index,
})
}
pub fn open_ephemeral() -> Self {
let (injection_tx, _) = broadcast::channel(32);
let (stream_tx, _) = broadcast::channel(2048);
let (context_watch, context_rx) = watch::channel(ContextSnapshot::default());
let (goal_watch, goal_rx) = watch::channel(None);
let (attach_watch, attach_rx) = watch::channel(0);
let (todos_watch, todos_rx) = watch::channel(Vec::new());
let (plans_watch, plans_rx) = watch::channel(Vec::new());
let sink = EventSink::new();
let events_handle = sink.events_handle();
let output_store = std::sync::Arc::new(crate::tools::tool_output::OutputStore::default());
Self {
id: SessionId::now(),
dir: PathBuf::new(),
writer: std::sync::Mutex::new(None),
sink,
message_stream: crate::message_stream::MessageStream::new(events_handle),
messages: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
output_store: output_store.clone(),
turn: TurnState::new(),
watch: WatchHub {
stream_tx,
context: context_watch,
goal: goal_watch,
attach: attach_watch,
todos: todos_watch,
plans: plans_watch,
_keepalive: (context_rx, goal_rx, attach_rx, todos_rx, plans_rx),
},
watch_hub: std::sync::Arc::new(crate::watch::WatchHub::new()),
flow_registry: std::sync::Arc::new(crate::tools::agent_ctrl::FlowRegistry::new()),
current_root: std::sync::Mutex::new(None),
successful_flow_count: std::sync::atomic::AtomicU64::new(0),
compaction: CompactionState::new(),
interactions: InteractionServices::new(),
injection_queue: Mutex::new(Vec::new()),
injection_tx,
last_image_user_msg: Mutex::new(None),
read_files: std::sync::Arc::new(
std::sync::Mutex::new(std::collections::HashSet::new()),
),
fs_access_mode: Mutex::new(None),
project_index: None,
}
}
pub fn project_index(&self) -> Option<std::sync::Arc<crate::index::AnchorIndex>> {
self.project_index.clone()
}
pub fn approval(&self) -> std::sync::Arc<ApprovalRegistry> {
self.interactions.approval.clone()
}
pub fn compact_reviews(&self) -> std::sync::Arc<CompactReviewRegistry> {
self.interactions.compact_reviews.clone()
}
pub fn forms(&self) -> std::sync::Arc<FormRegistry> {
self.interactions.forms.clone()
}
pub fn fs_access_mode(&self) -> Option<crate::fs_access::FsAccessMode> {
*self.fs_access_mode.lock().unwrap()
}
pub fn set_fs_access_mode(&self, mode: crate::fs_access::FsAccessMode) {
*self.fs_access_mode.lock().unwrap() = Some(mode);
}
pub fn compact_review_mode(&self) -> CompactReviewMode {
*self.compaction.review_mode.lock().unwrap()
}
pub fn set_compact_review_mode(&self, mode: CompactReviewMode) {
*self.compaction.review_mode.lock().unwrap() = mode;
}
pub fn read_files(
&self,
) -> std::sync::Arc<std::sync::Mutex<std::collections::HashSet<std::path::PathBuf>>> {
self.read_files.clone()
}
pub fn output_store(&self) -> std::sync::Arc<crate::tools::tool_output::OutputStore> {
self.output_store.clone()
}
pub fn mark_file_read(&self, path: &std::path::Path) {
if let Ok(mut set) = self.read_files.lock() {
set.insert(path.to_path_buf());
if let Ok(canonical) = std::fs::canonicalize(path) {
set.insert(canonical);
}
}
}
pub fn stream_tx(&self) -> broadcast::Sender<StreamFrame> {
self.watch.stream_tx.clone()
}
pub fn set_current_root(&self, handle: String) {
*self.current_root.lock().unwrap() = Some(handle);
}
pub fn current_root(&self) -> Option<String> {
self.current_root.lock().unwrap().clone()
}
pub fn clear_current_root(&self) {
*self.current_root.lock().unwrap() = None;
}
pub fn record_successful_flow(&self) -> Option<u64> {
let count = self
.successful_flow_count
.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
+ 1;
is_auto_name_threshold(count).then_some(count)
}
pub fn successful_flow_count(&self) -> u64 {
self.successful_flow_count
.load(std::sync::atomic::Ordering::Relaxed)
}
pub fn stream_subscribe(&self) -> broadcast::Receiver<StreamFrame> {
self.watch.stream_tx.subscribe()
}
pub fn id(&self) -> &SessionId {
&self.id
}
pub fn dir(&self) -> &Path {
&self.dir
}
pub fn transcript_replay(&self) -> Vec<TranscriptEntry> {
let Some(path) = self.events_path() else {
return Vec::new();
};
replay_transcript_from(&path).unwrap_or_default()
}
pub fn events_path(&self) -> Option<std::path::PathBuf> {
self.writer
.lock()
.unwrap()
.as_ref()
.map(|w| w.events_path().to_path_buf())
}
pub async fn plan_system_prompt(&self) -> Option<String> {
let store = crate::memory::plan::PlanStore::at(&self.dir);
let plan = store.latest().await.ok().flatten()?;
Some(crate::tools::plan::render_plan(&plan))
}
pub fn goal(&self) -> Option<String> {
if let Some(cached) = self.watch.goal.borrow().clone() {
return Some(cached);
}
load_goal(&self.dir)
}
pub fn subscribe_goal(&self) -> watch::Receiver<Option<String>> {
self.watch.goal.subscribe()
}
pub fn goal_watch(&self) -> &watch::Sender<Option<String>> {
&self.watch.goal
}
pub fn subscribe_context(&self) -> watch::Receiver<ContextSnapshot> {
self.watch.context.subscribe()
}
pub fn subscribe_attach(&self) -> watch::Receiver<usize> {
self.watch.attach.subscribe()
}
pub fn subscribe_pending_approvals(&self) -> watch::Receiver<Vec<PendingApproval>> {
self.interactions.approval.subscribe()
}
pub fn meta(&self) -> Option<crate::session_meta::SessionMeta> {
crate::session_meta::SessionMeta::load(&self.dir)
}
pub fn request_manual_compact(&self) {
self.compaction
.manual_pending
.store(true, std::sync::atomic::Ordering::SeqCst);
}
pub fn take_manual_compact_request(&self) -> bool {
self.compaction
.manual_pending
.swap(false, std::sync::atomic::Ordering::SeqCst)
}
pub fn set_goal(&self, goal: Option<String>) {
let _ = self.watch.goal.send(goal);
}
pub fn set_attach_count(&self, count: usize) {
let _ = self.watch.attach.send(count);
}
#[allow(clippy::too_many_arguments)]
pub fn record_llm_call(
&self,
model: &str,
tokens_in: u64,
tokens_out: u64,
cache_read: u64,
cache_write: u64,
ttft_ms: Option<u64>,
tokens_per_sec: Option<f64>,
) {
if tokens_in > 0 {
self.compaction
.last_input_tokens
.store(tokens_in, std::sync::atomic::Ordering::Relaxed);
}
self.watch.context.send_modify(|snap| {
snap.model = model.to_string();
snap.tokens_in = snap.tokens_in.saturating_add(tokens_in);
snap.tokens_out = snap.tokens_out.saturating_add(tokens_out);
snap.cache_read = snap.cache_read.saturating_add(cache_read);
snap.cache_write = snap.cache_write.saturating_add(cache_write);
snap.last_ttft_ms = ttft_ms.unwrap_or(0);
snap.last_tokens_per_sec = tokens_per_sec.unwrap_or(0.0);
});
self.refresh_window_snapshot();
}
pub fn last_input_tokens(&self) -> u64 {
self.compaction
.last_input_tokens
.load(std::sync::atomic::Ordering::Relaxed)
}
pub async fn acquire_compact_lock(&self) -> tokio::sync::MutexGuard<'_, ()> {
self.compaction.lock.lock().await
}
pub async fn acquire_compact_lock_owned(&self) -> tokio::sync::OwnedMutexGuard<()> {
self.compaction.lock.clone().lock_owned().await
}
pub fn compact_lock_handle(&self) -> std::sync::Arc<tokio::sync::Mutex<()>> {
self.compaction.lock.clone()
}
pub fn refresh_window_snapshot(&self) {
let provider_tokens = self.last_input_tokens();
let estimated = crate::compaction::estimate_tokens_for_messages(&self.messages());
let window = if provider_tokens > 0 {
provider_tokens
} else {
estimated
};
let model = self.last_model();
let budget = crate::model_registry::model_info(&model).context_budget;
self.watch.context.send_modify(|snap| {
snap.window_tokens = window;
if budget > 0 {
snap.window_budget = budget;
}
});
let snap = self.watch.context.borrow();
PersistedContextState {
model,
window_tokens: snap.window_tokens,
window_budget: snap.window_budget,
}
.save(&self.dir);
}
pub fn cumulative_input_tokens(&self) -> u64 {
self.watch.context.borrow().tokens_in
}
pub fn reset_input_tokens_to(&self, tokens: u64) {
self.watch.context.send_modify(|snap| {
snap.tokens_in = tokens;
});
}
pub fn last_model(&self) -> String {
self.watch.context.borrow().model.clone()
}
pub fn set_current_model(&self, model: impl Into<String>) {
let model = model.into();
let budget = crate::model_registry::model_info(&model).context_budget;
self.watch.context.send_modify(|snap| {
snap.model = model.clone();
if budget > 0 {
snap.window_budget = budget;
}
});
let snap = self.watch.context.borrow();
PersistedContextState {
model,
window_tokens: snap.window_tokens,
window_budget: snap.window_budget,
}
.save(&self.dir);
}
pub fn update_mcp_server(&self, status: crate::mcp::McpServerStatus) {
self.watch.context.send_modify(|snap| {
if let Some(existing) = snap.mcp_servers.iter_mut().find(|s| s.name == status.name) {
*existing = status;
} else {
snap.mcp_servers.push(status);
}
});
}
pub fn set_memory_recent_count(&self, count: u16) {
self.watch.context.send_modify(|snap| {
snap.memory_recent_count = count;
});
}
pub fn subscribe_todos(&self) -> watch::Receiver<Vec<crate::memory::todo::Todo>> {
self.watch.todos.subscribe()
}
pub fn todos_watch(&self) -> &watch::Sender<Vec<crate::memory::todo::Todo>> {
&self.watch.todos
}
pub fn subscribe_plans(&self) -> watch::Receiver<Vec<crate::memory::plan::Plan>> {
self.watch.plans.subscribe()
}
pub fn plans_watch(&self) -> &watch::Sender<Vec<crate::memory::plan::Plan>> {
&self.watch.plans
}
pub async fn refresh_plans_from_store_async(&self) {
if self.dir.as_os_str().is_empty() {
return;
}
let store = crate::memory::plan::PlanStore::at(&self.dir);
match store.list().await {
Ok(list) => {
let _ = self.watch.plans.send(list);
}
Err(e) => {
crate::notify!(
warn,
location = Log,
stack = dedupe("memory.refresh_plans_async", 60_000),
"refresh_plans_from_store_async: {e}"
);
}
}
}
pub fn refresh_todos_from_store(&self) {
if self.dir.as_os_str().is_empty() {
return;
}
let store = crate::memory::todo::TodoStore::at(&self.dir);
match tokio::task::block_in_place(|| {
tokio::runtime::Handle::try_current()
.ok()
.map(|h| h.block_on(store.list()))
}) {
Some(Ok(list)) => {
let _ = self.watch.todos.send(list);
}
Some(Err(e)) => {
crate::notify!(
warn,
location = Log,
stack = dedupe("memory.refresh_todos", 60_000),
"refresh_todos_from_store: {e}"
);
}
None => {}
}
}
pub async fn refresh_todos_from_store_async(&self) {
if self.dir.as_os_str().is_empty() {
return;
}
let store = crate::memory::todo::TodoStore::at(&self.dir);
match store.list().await {
Ok(list) => {
let _ = self.watch.todos.send(list);
}
Err(e) => {
crate::notify!(
warn,
location = Log,
stack = dedupe("memory.refresh_todos_async", 60_000),
"refresh_todos_from_store_async: {e}"
);
}
}
}
pub fn sink(&self) -> &EventSink {
&self.sink
}
pub fn append_message(&self, msg: Message, flow_run_id: Option<FlowRunId>) {
AppendMessageCommand { msg, flow_run_id }.execute(self);
}
pub fn emit_attachment_degrade(
&self,
message_seq: u64,
part_index: usize,
file_basename: String,
reason: String,
) {
self.sink.emit(Event::AttachmentDegraded {
turn_id: None,
flow_run_id: None,
message_seq,
part_index,
file_basename,
reason,
});
}
pub fn record_attachment_degrade(&self, reason: &str) -> usize {
let target = self.last_image_user_msg.lock().unwrap().take();
let Some(entry) = target else {
return 0;
};
let turn_id = self.turn.current_turn.lock().unwrap().clone();
for (part_index, basename) in &entry.images {
self.sink.emit(Event::AttachmentDegraded {
turn_id: turn_id.clone(),
flow_run_id: None,
message_seq: entry.message_seq,
part_index: *part_index,
file_basename: basename.clone(),
reason: reason.into(),
});
}
if let Ok(mut msgs) = self.messages.lock() {
for m in msgs.iter_mut() {
for (part_index, basename) in &entry.images {
if let Some(part) = m.parts.get_mut(*part_index)
&& matches!(part, crate::message::MessagePart::Image { .. })
{
*part = crate::message::MessagePart::Text {
text: format!("[attachment unavailable: {basename} — {reason}]"),
};
}
}
}
}
entry.images.len()
}
pub fn messages(&self) -> crate::message_stream::MessageWindow {
self.message_stream.window()
}
pub fn messages_full(&self) -> std::sync::Arc<Vec<Message>> {
self.message_stream.full_messages()
}
pub fn messages_handle(&self) -> std::sync::Arc<std::sync::Mutex<Vec<Message>>> {
self.messages.clone()
}
pub fn message_count(&self) -> usize {
self.messages().len()
}
pub fn user_message_count(&self) -> usize {
self.messages()
.iter()
.filter(|m| matches!(m.role, MessageRole::User))
.count()
}
pub fn push_system_note(&self, text: String) {
let _ = self
.watch
.stream_tx
.send(crate::stream::StreamFrame::Note(text));
}
pub fn approval_cooldown_ok_for_compact(&self) -> bool {
self.sink.last_compact_ago_seconds().is_none_or(|s| s >= 60)
}
pub fn emit_compact_warning(
&self,
model: &str,
current_tokens: u64,
threshold: u64,
budget: u64,
reason: &str,
) {
let message = format!(
"context {current_tokens} > threshold {threshold} (budget {budget}, model {model}); skipping compaction: {reason}"
);
self.sink.emit(Event::WatchWarn {
turn_id: self.turn.current_turn.lock().unwrap().clone(),
flow_run_id: None,
target: "context.compaction".into(),
trigger: "auto_compact".into(),
message,
});
self.push_system_note(format!("[warn] compaction skipped: {reason}"));
}
pub fn compact_messages_auto(&self, summary: String) -> Option<CompactResult> {
let msgs = self.messages();
let tokens = crate::compaction::estimate_tokens_for_messages(&msgs);
let info = crate::model_registry::model_info(&self.last_model());
let target = info.compaction_target_after();
let range = crate::compaction::find_compact_range(&msgs, target)?;
self.compact_messages(summary, range, tokens)
}
pub fn commit_rewritten_window(
&self,
replacement: Vec<Message>,
before_tokens: u64,
before_window_tokens: u64,
rewritten_count: usize,
) -> Option<CompactResult> {
let after_tokens = crate::compaction::estimate_tokens_for_messages(&replacement);
if rewritten_count == 0 || after_tokens >= before_window_tokens {
return None;
}
let summary = format!(
"[atman: persistently compacted output from {rewritten_count} retained messages]"
);
self.sink.mark_compacted();
self.sink.emit(Event::ContextCompact {
session_id: self.id.to_string(),
before_tokens,
after_tokens,
compacted_range_start: 0,
compacted_range_end: 0,
summary_text: Some(summary.clone()),
replacement_msg_seq: None,
});
self.sink.emit(Event::CompactionSummary {
session_id: self.id.to_string(),
range_start: 0,
range_end: 0,
compacted_count: rewritten_count,
before_tokens,
after_tokens,
summary: summary.clone(),
});
let _ = self
.watch
.stream_tx
.send(crate::stream::StreamFrame::CompactionSummary {
phase: crate::stream::CompactionPhase::Finished,
range_start: 0,
range_end: 0,
summary,
before_tokens,
after_tokens,
compacted_count: rewritten_count,
});
self.compaction
.last_input_tokens
.store(after_tokens, std::sync::atomic::Ordering::Relaxed);
if let Ok(mut messages) = self.messages.lock() {
*messages = replacement.clone();
}
self.sink.emit(Event::Checkpoint {
session_id: self.id.to_string(),
messages: replacement,
window_tokens: after_tokens,
});
self.refresh_window_snapshot();
Some(CompactResult {
before_tokens,
after_tokens,
compacted_start: 0,
compacted_end: 0,
})
}
pub fn commit_compacted_window(
&self,
summary: String,
replacement: Vec<Message>,
range: crate::compaction::CompactRange,
before_tokens: u64,
before_window_tokens: u64,
) -> Option<CompactResult> {
let after_tokens = crate::compaction::estimate_tokens_for_messages(&replacement);
if after_tokens >= before_window_tokens {
self.push_system_note(format!(
"compaction skipped: replacement would not shrink transcript ({} >= {} tokens)",
after_tokens, before_window_tokens
));
return None;
}
self.sink.mark_compacted();
self.sink.emit(Event::ContextCompact {
session_id: self.id.to_string(),
before_tokens,
after_tokens,
compacted_range_start: range.start as u64,
compacted_range_end: range.end.saturating_sub(1) as u64,
summary_text: Some(summary.clone()),
replacement_msg_seq: None,
});
self.sink.emit(Event::CompactionSummary {
session_id: self.id.to_string(),
range_start: range.start as u64,
range_end: range.end.saturating_sub(1) as u64,
compacted_count: range.end - range.start,
before_tokens,
after_tokens,
summary: summary.clone(),
});
let _ = self
.watch
.stream_tx
.send(crate::stream::StreamFrame::CompactionSummary {
phase: crate::stream::CompactionPhase::Finished,
range_start: range.start,
range_end: range.end.saturating_sub(1),
summary,
before_tokens,
after_tokens,
compacted_count: range.end - range.start,
});
self.compaction
.last_input_tokens
.store(after_tokens, std::sync::atomic::Ordering::Relaxed);
if let Ok(mut messages) = self.messages.lock() {
*messages = replacement.clone();
}
self.sink.emit(Event::Checkpoint {
session_id: self.id.to_string(),
messages: replacement,
window_tokens: after_tokens,
});
self.refresh_window_snapshot();
Some(CompactResult {
before_tokens,
after_tokens,
compacted_start: range.start,
compacted_end: range.end,
})
}
pub fn compact_messages(
&self,
summary: String,
range: crate::compaction::CompactRange,
before_tokens: u64,
) -> Option<CompactResult> {
use crate::compaction::{estimate_tokens_for_messages, replace_range_with_summary};
let msgs = self.messages();
let turn_id = msgs
.get(range.start)
.map(|m| m.turn_id.clone())
.unwrap_or_else(TurnId::now);
let after = replace_range_with_summary(&msgs, &range, summary.clone(), turn_id.clone());
let after_tokens = estimate_tokens_for_messages(&after);
if after_tokens >= before_tokens {
self.push_system_note(format!(
"compaction skipped: summary would not shrink transcript ({} >= {} tokens)",
after_tokens, before_tokens
));
return None;
}
let replacement_msg = after.first().cloned().unwrap_or_else(|| {
Message::system_compact_summary(
turn_id.clone(),
summary.clone(),
range.start as u64,
range.end.saturating_sub(1) as u64,
range.end - range.start,
)
});
self.sink.mark_compacted();
let replacement_seq = self.sink.next_seq_peek();
self.sink.emit(Event::SystemMsg {
turn_id: turn_id.clone(),
message: replacement_msg,
});
self.sink.emit(Event::ContextCompact {
session_id: self.id.to_string(),
before_tokens,
after_tokens,
compacted_range_start: range.start as u64,
compacted_range_end: range.end.saturating_sub(1) as u64,
summary_text: Some(summary.clone()),
replacement_msg_seq: Some(replacement_seq),
});
self.sink.emit(Event::CompactionSummary {
session_id: self.id.to_string(),
range_start: range.start as u64,
range_end: range.end.saturating_sub(1) as u64,
compacted_count: range.end - range.start,
before_tokens,
after_tokens,
summary: summary.clone(),
});
let _ = self
.watch
.stream_tx
.send(crate::stream::StreamFrame::CompactionSummary {
phase: crate::stream::CompactionPhase::Finished,
range_start: range.start,
range_end: range.end.saturating_sub(1),
summary,
before_tokens,
after_tokens,
compacted_count: range.end - range.start,
});
let checkpoint_messages = self.messages();
let window_tokens = estimate_tokens_for_messages(&checkpoint_messages);
self.compaction
.last_input_tokens
.store(window_tokens, std::sync::atomic::Ordering::Relaxed);
let window_owned = checkpoint_messages.to_vec();
if let Ok(mut vec) = self.messages.lock() {
*vec = window_owned;
}
self.refresh_window_snapshot();
self.sink.emit(Event::Checkpoint {
session_id: self.id.to_string(),
messages: checkpoint_messages.to_vec(),
window_tokens,
});
Some(CompactResult {
before_tokens,
after_tokens,
compacted_start: range.start,
compacted_end: range.end,
})
}
pub fn begin_turn(&self, user_msg: Message) -> TurnId {
BeginTurnCommand { user_msg }.execute(self)
}
pub fn mark_streamed(&self) {
self.turn
.streamed
.store(true, std::sync::atomic::Ordering::Relaxed);
}
pub fn take_streamed_flag(&self) -> bool {
self.turn
.streamed
.swap(false, std::sync::atomic::Ordering::Relaxed)
}
pub fn end_turn(&self) {
self.turn
.streamed
.store(false, std::sync::atomic::Ordering::Relaxed);
let turn_id = self.turn.current_turn.lock().unwrap().take();
if let Some(turn_id) = turn_id {
let mut q = self.injection_queue.lock().unwrap();
for inj in q.iter_mut() {
if inj.state == InjectionState::Pending && inj.turn_id == turn_id {
inj.state = InjectionState::Cancelled;
let _ = self.injection_tx.send(inj.clone());
}
}
drop(q);
self.sink.emit(Event::TurnEnd { turn_id });
}
}
pub fn current_turn(&self) -> Option<TurnId> {
self.turn.current_turn.lock().unwrap().clone()
}
pub fn enqueue_injection(&self, text: impl Into<String>) -> Result<InjectionId, EnqueueError> {
self.enqueue_injection_with_level(text, crate::injection::InjectionLevel::L1Nudge, None)
}
pub fn enqueue_injection_with_level(
&self,
text: impl Into<String>,
level: crate::injection::InjectionLevel,
redirect_target: Option<String>,
) -> Result<InjectionId, EnqueueError> {
let turn_id = self
.turn
.current_turn
.lock()
.unwrap()
.clone()
.ok_or(EnqueueError::NoActiveTurn)?;
let inj = Injection::with_level(turn_id.clone(), text, level, redirect_target);
let id = inj.id.clone();
self.sink.emit(Event::UserInject {
turn_id,
injection: inj.clone(),
});
self.injection_queue.lock().unwrap().push(inj.clone());
let _ = self.injection_tx.send(inj);
Ok(id)
}
pub fn subscribe_injections(&self) -> broadcast::Receiver<Injection> {
self.injection_tx.subscribe()
}
pub fn mark_injection_consumed(&self, id: &InjectionId) {
let mut q = self.injection_queue.lock().unwrap();
for inj in q.iter_mut() {
if inj.id == *id && inj.state == InjectionState::Pending {
inj.state = InjectionState::Injected;
let _ = self.injection_tx.send(inj.clone());
return;
}
}
}
pub fn peek_pending_l2_or_higher(&self, turn_id: &TurnId) -> Option<Injection> {
let q = self.injection_queue.lock().unwrap();
q.iter()
.find(|i| {
i.state == InjectionState::Pending
&& i.turn_id == *turn_id
&& !matches!(i.level, crate::injection::InjectionLevel::L1Nudge)
})
.cloned()
}
pub fn drain_injections(&self, turn_id: &TurnId) -> Vec<Injection> {
let mut q = self.injection_queue.lock().unwrap();
let mut out = Vec::new();
for inj in q.iter_mut() {
if inj.state == InjectionState::Pending && inj.turn_id == *turn_id {
inj.state = InjectionState::Injected;
let _ = self.injection_tx.send(inj.clone());
out.push(inj.clone());
}
}
out
}
pub fn list_pending_injections(&self) -> Vec<Injection> {
self.injection_queue
.lock()
.unwrap()
.iter()
.filter(|i| i.state == InjectionState::Pending)
.cloned()
.collect()
}
pub fn cancel_flow(&self) {
self.turn.flow_cancel.lock().unwrap().cancel();
}
pub fn flow_cancel_token(&self) -> CancellationToken {
self.turn.flow_cancel.lock().unwrap().clone()
}
pub async fn shutdown(&self) {
let writer = self.writer.lock().unwrap().take();
if let Some(writer) = writer {
writer.shutdown().await;
}
}
#[allow(clippy::await_holding_lock)]
pub async fn flush_writer(&self) {
let guard = self.writer.lock().unwrap();
let Some(ref writer) = *guard else {
return;
};
writer.flush().await;
}
}
#[derive(Debug, thiserror::Error)]
pub enum EnqueueError {
#[error("enqueue_injection called with no active turn")]
NoActiveTurn,
}
pub struct AppendMessageCommand {
pub msg: Message,
pub flow_run_id: Option<FlowRunId>,
}
impl AppendMessageCommand {
pub fn execute(&self, session: &Session) -> u64 {
let flow_run_id_str = self.flow_run_id.as_ref().map(|r| r.0.to_string());
let msg = crate::tools::tool_output::maybe_truncate_tool_message(
&self.msg,
Some(&session.output_store),
);
let is_internal = msg.origin == crate::message::MessageOrigin::Internal;
let event =
match msg.role {
MessageRole::User => Event::UserMsg {
turn_id: msg.turn_id.clone(),
flow_run_id: self.flow_run_id.clone(),
message: msg.clone(),
},
MessageRole::Assistant => {
if !is_internal {
let _ = session.watch.stream_tx.send(
crate::stream::StreamFrame::AssistantMsg {
flow_run_id: flow_run_id_str.clone(),
message: msg.clone(),
},
);
for source in extract_mermaid_blocks(&msg) {
let _ = session.watch.stream_tx.send(
crate::stream::StreamFrame::MermaidDiagram {
source: source.clone(),
},
);
session
.sink
.emit(crate::event::Event::MermaidDiagram { source });
}
}
Event::AssistantMsg {
turn_id: msg.turn_id.clone(),
flow_run_id: self.flow_run_id.clone(),
message: msg.clone(),
}
}
MessageRole::Tool => {
if !is_internal {
let _ = session.watch.stream_tx.send(
crate::stream::StreamFrame::ToolResultMsg {
flow_run_id: flow_run_id_str.clone(),
message: msg.clone(),
},
);
}
Event::ToolResultMsg {
turn_id: msg.turn_id.clone(),
flow_run_id: self.flow_run_id.clone(),
message: msg.clone(),
}
}
MessageRole::System => Event::SystemMsg {
turn_id: msg.turn_id.clone(),
message: msg.clone(),
},
};
let seq = session.sink.emit_returning_seq(event);
if matches!(msg.role, MessageRole::User) {
let images: Vec<(usize, String)> = msg
.parts
.iter()
.enumerate()
.filter_map(|(i, p)| match p {
crate::message::MessagePart::Image { source } => {
let basename = match &source.data {
crate::message::ImageData::Path { path } => path
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("unknown")
.to_string(),
crate::message::ImageData::Base64 { .. } => "base64".into(),
};
Some((i, basename))
}
_ => None,
})
.collect();
if !images.is_empty() {
*session.last_image_user_msg.lock().unwrap() = Some(LastImageUserMsg {
message_seq: seq,
images,
});
}
}
session.messages.lock().unwrap().push(msg.clone());
seq
}
}
pub struct BeginTurnCommand {
pub user_msg: Message,
}
impl BeginTurnCommand {
pub fn execute(&self, session: &Session) -> TurnId {
let turn_id = self.user_msg.turn_id.clone();
*session.turn.current_turn.lock().unwrap() = Some(turn_id.clone());
*session.turn.flow_cancel.lock().unwrap() = tokio_util::sync::CancellationToken::new();
session.sink.emit(Event::TurnStart {
turn_id: turn_id.clone(),
});
AppendMessageCommand {
msg: self.user_msg.clone(),
flow_run_id: None,
}
.execute(session);
turn_id
}
}
fn extract_mermaid_blocks(msg: &crate::message::Message) -> Vec<String> {
let text = msg.text_concat();
let mut blocks = Vec::new();
let mut lines = text.lines().peekable();
while let Some(line) = lines.next() {
let trimmed = line.trim();
if trimmed.starts_with("```") {
let lang = trimmed.trim_start_matches("```").trim();
if lang == "mermaid" {
let mut source = String::new();
for inner in lines.by_ref() {
if inner.trim() == "```" {
break;
}
if !source.is_empty() {
source.push('\n');
}
source.push_str(inner);
}
if !source.is_empty() {
blocks.push(source);
}
} else {
for inner in lines.by_ref() {
if inner.trim() == "```" {
break;
}
}
}
}
}
blocks
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
fn write_events(dir: &Path, lines: &[&str]) {
let path = dir.join("events.jsonl");
std::fs::write(&path, lines.join("\n") + "\n").unwrap();
}
#[test]
fn successful_flow_count_triggers_at_powers_of_three_and_resets_per_session() {
let session = Session::open_ephemeral();
let mut hits = Vec::new();
for _ in 0..27 {
if let Some(count) = session.record_successful_flow() {
hits.push(count);
}
}
assert_eq!(hits, vec![3, 9, 27]);
assert_eq!(session.successful_flow_count(), 27);
assert_eq!(Session::open_ephemeral().successful_flow_count(), 0);
}
#[test]
fn commit_rewritten_window_uses_checkpoint_without_legacy_range_replay() {
let session = Session::open_ephemeral();
let original = vec![
Message::user_text(TurnId::now(), "first user"),
Message::assistant_text(TurnId::now(), "large output".repeat(2_000)),
Message::user_text(TurnId::now(), "current user"),
];
for message in original.clone() {
session.append_message(message, None);
}
let replacement = vec![
original[0].clone(),
Message::assistant_text(TurnId::now(), "persisted omission"),
original[2].clone(),
];
let before_tokens = crate::compaction::estimate_tokens_for_messages(&original);
session
.commit_rewritten_window(replacement.clone(), before_tokens, before_tokens, 1)
.expect("rewrite commit");
assert_eq!(session.messages().as_ref(), replacement.as_slice());
let events = session.sink().snapshot();
assert!(events.iter().any(|event| matches!(
event,
Event::ContextCompact {
replacement_msg_seq: None,
..
}
)));
assert!(
!events
.iter()
.any(|event| matches!(event, Event::SystemMsg { .. }))
);
let checkpoint = events
.into_iter()
.find(|event| matches!(event, Event::Checkpoint { .. }))
.expect("checkpoint");
let replay = crate::message_stream::MessageStream::new(std::sync::Arc::new(
std::sync::Mutex::new(vec![crate::event::EventEnvelope::new(1, checkpoint)]),
));
assert_eq!(&*replay.window(), replacement.as_slice());
}
#[test]
fn commit_compacted_window_updates_live_handle_and_checkpoint_replay() {
let session = Session::open_ephemeral();
let old = vec![
Message::user_text(TurnId::now(), "old user".repeat(2_000)),
Message::assistant_text(TurnId::now(), "old assistant".repeat(2_000)),
Message::user_text(TurnId::now(), "current user"),
];
for message in old.clone() {
session.append_message(message, None);
}
let replacement = vec![
Message::system_compact_summary(TurnId::now(), "anchor", 0, 1, 2),
old[2].clone(),
Message::assistant_text(TurnId::now(), "persisted omission"),
];
let before_tokens = crate::compaction::estimate_tokens_for_messages(&old);
let range = crate::compaction::CompactRange {
start: 0,
end: 2,
tokens_saved_estimate: before_tokens,
};
session
.commit_compacted_window(
"anchor".into(),
replacement.clone(),
range,
before_tokens,
before_tokens,
)
.expect("commit");
assert_eq!(session.messages().as_ref(), replacement.as_slice());
assert_eq!(
session.messages_handle().lock().unwrap().as_slice(),
replacement.as_slice()
);
let checkpoint = session
.sink()
.snapshot()
.into_iter()
.find(|event| matches!(event, Event::Checkpoint { .. }))
.expect("checkpoint");
let replay = crate::message_stream::MessageStream::new(std::sync::Arc::new(
std::sync::Mutex::new(vec![crate::event::EventEnvelope::new(1, checkpoint)]),
));
assert_eq!(&*replay.window(), replacement.as_slice());
}
#[test]
fn replay_applies_attachment_degraded_patch() {
let dir = TempDir::new().unwrap();
let user_msg = r#"{"type":"user_msg","seq":5,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"image","source":{"media_type":"image/png","data":{"kind":"path","path":"/tmp/photo.png"}}},{"type":"text","text":"describe"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
let degrade = r#"{"type":"attachment_degraded","seq":6,"turn_id":null,"flow_run_id":null,"message_seq":5,"part_index":0,"file_basename":"photo.png","reason":"image_too_large","ts":"2026-07-07T00:00:01Z"}"#;
write_events(dir.path(), &[user_msg, degrade]);
let entries = replay_transcript_from(&dir.path().join("events.jsonl")).unwrap();
let msg = entries
.into_iter()
.find_map(|e| match e {
TranscriptEntry::Message { message, .. } => Some(message),
_ => None,
})
.unwrap();
assert_eq!(msg.parts.len(), 2);
match &msg.parts[0] {
crate::message::MessagePart::Text { text } => {
assert!(text.contains("photo.png"), "expected basename: {text}");
assert!(text.contains("image_too_large"), "expected reason: {text}");
assert!(text.starts_with("[attachment unavailable"));
}
other => panic!("expected Text stub, got {other:?}"),
}
assert!(matches!(
msg.parts[1],
crate::message::MessagePart::Text { .. }
));
}
#[test]
fn approval_registry_auto_approves_when_level_leq_ceiling() {
let reg = ApprovalRegistry::new();
reg.set_auto_ceiling(crate::tool::ApprovalLevel::Approve);
let pending = PendingApproval {
tool_use_id: "tu1".into(),
tool_name: "fs.read".into(),
args_preview: "{}".into(),
preview: None,
level: crate::tool::ApprovalLevel::Auto,
run_id: FlowRunId::now(),
emitted_at: chrono::Utc::now(),
bypass_auto_ceiling: false,
};
let rx = reg.request(pending);
let got = rx.blocking_recv().unwrap();
assert!(matches!(got, ApprovalDecision::Approve));
assert!(reg.list_pending().is_empty());
}
#[test]
fn approval_registry_queues_when_level_above_ceiling() {
let reg = std::sync::Arc::new(ApprovalRegistry::new());
reg.set_auto_ceiling(crate::tool::ApprovalLevel::Auto);
let pending = PendingApproval {
tool_use_id: "tu42".into(),
tool_name: "fs.write".into(),
args_preview: "{}".into(),
preview: None,
level: crate::tool::ApprovalLevel::Approve,
run_id: FlowRunId::now(),
emitted_at: chrono::Utc::now(),
bypass_auto_ceiling: false,
};
let mut rx = reg.request(pending);
assert_eq!(reg.list_pending().len(), 1);
assert!(rx.try_recv().is_err(), "should still be queued");
assert!(reg.decide("tu42", ApprovalDecision::Approve));
let got = rx.blocking_recv().unwrap();
assert!(matches!(got, ApprovalDecision::Approve));
assert!(reg.list_pending().is_empty());
}
#[test]
fn approval_registry_decide_all_flushes_queue() {
let reg = ApprovalRegistry::new();
reg.set_auto_ceiling(crate::tool::ApprovalLevel::Auto);
let mut rxs = Vec::new();
for i in 0..3 {
rxs.push(reg.request(PendingApproval {
tool_use_id: format!("tu{i}"),
tool_name: "bash.exec".into(),
args_preview: "{}".into(),
preview: None,
level: crate::tool::ApprovalLevel::Dangerous,
run_id: FlowRunId::now(),
emitted_at: chrono::Utc::now(),
bypass_auto_ceiling: false,
}));
}
assert_eq!(reg.list_pending().len(), 3);
assert_eq!(
reg.decide_all(ApprovalDecision::Deny {
reason: "user cancelled".into()
}),
3
);
assert!(reg.list_pending().is_empty());
}
#[test]
fn compact_review_registry_auto_accepts_when_no_subscriber() {
let reg = CompactReviewRegistry::new();
let pending = PendingCompactReview {
review_id: "r1".into(),
summary: "gist".into(),
slice_preview: String::new(),
slice_count: 0,
range_start: 0,
range_end: 0,
tokens_before: 0,
emitted_at: chrono::Utc::now(),
};
let rx = reg.request(pending);
let got = rx.blocking_recv().unwrap();
assert!(matches!(got, CompactReviewDecision::AcceptAsIs));
assert!(reg.list_pending().is_none());
}
#[test]
fn compact_review_registry_holds_pending_and_decides() {
let reg = std::sync::Arc::new(CompactReviewRegistry::new());
let _sub = reg.subscribe();
let pending = PendingCompactReview {
review_id: "r2".into(),
summary: "old".into(),
slice_preview: "slice".into(),
slice_count: 3,
range_start: 1,
range_end: 4,
tokens_before: 500,
emitted_at: chrono::Utc::now(),
};
let mut rx = reg.request(pending);
assert!(rx.try_recv().is_err(), "should be queued");
assert!(reg.list_pending().is_some());
assert!(reg.decide(
"r2",
CompactReviewDecision::AcceptEdited {
summary: "new".into()
}
));
let got = rx.blocking_recv().unwrap();
match got {
CompactReviewDecision::AcceptEdited { summary } => assert_eq!(summary, "new"),
other => panic!("unexpected decision: {other:?}"),
}
assert!(reg.list_pending().is_none());
}
#[test]
fn compact_review_registry_reject_flushes() {
let reg = std::sync::Arc::new(CompactReviewRegistry::new());
let _sub = reg.subscribe();
let rx = reg.request(PendingCompactReview {
review_id: "r3".into(),
summary: String::new(),
slice_preview: String::new(),
slice_count: 0,
range_start: 0,
range_end: 0,
tokens_before: 0,
emitted_at: chrono::Utc::now(),
});
assert!(reg.decide("r3", CompactReviewDecision::Reject));
let got = rx.blocking_recv().unwrap();
assert!(matches!(got, CompactReviewDecision::Reject));
}
#[test]
fn replay_context_snapshot_accumulates_llm_call_usage() {
let dir = TempDir::new().unwrap();
let events = [
r#"{"type":"llm_call","seq":1,"model":"anthropic/claude-4","provider":"anthropic","usage":{"input":100,"cached_input":10,"output":50,"cache_write":0},"wallclock_ms":1000,"status":{"kind":"ok"},"run_id":"019f0000-0000-7000-0000-000000000099","ts":"2026-07-08T00:00:00Z"}"#,
r#"{"type":"user_msg","seq":2,"turn_id":"019f0000-0000-7000-0000-000000000002","message":{"role":"user","parts":[{"type":"text","text":"hi"}],"turn_id":"019f0000-0000-7000-0000-000000000002"},"ts":"2026-07-08T00:00:00Z"}"#,
r#"{"type":"llm_call","seq":3,"model":"anthropic/claude-4","provider":"anthropic","usage":{"input":200,"cached_input":0,"output":80,"cache_write":0},"wallclock_ms":1000,"status":{"kind":"ok"},"run_id":"019f0000-0000-7000-0000-000000000099","ts":"2026-07-08T00:00:01Z"}"#,
];
write_events(dir.path(), &events);
let snap = replay_context_snapshot_from(&dir.path().join("events.jsonl"));
assert_eq!(snap.model, "anthropic/claude-4");
assert_eq!(snap.tokens_in, 310);
assert_eq!(snap.tokens_out, 130);
}
#[test]
fn replay_context_snapshot_skips_subagent_llm_calls() {
let dir = TempDir::new().unwrap();
let events = [
r#"{"type":"llm_call","seq":1,"model":"zhipuai/glm-5.2","provider":"zhipu","usage":{"input":100,"cached_input":0,"output":50,"cache_write":0},"wallclock_ms":1000,"status":{"kind":"ok"},"run_id":"019f0000-0000-7000-0000-000000000099","ts":"2026-07-08T00:00:00Z"}"#,
r#"{"type":"llm_call","seq":2,"model":"gpt-4o-mini","provider":"openai","usage":{"input":200,"cached_input":0,"output":80,"cache_write":0},"wallclock_ms":1000,"status":{"kind":"ok"},"run_id":null,"ts":"2026-07-08T00:00:01Z"}"#,
];
write_events(dir.path(), &events);
let snap = replay_context_snapshot_from(&dir.path().join("events.jsonl"));
assert_eq!(snap.model, "zhipuai/glm-5.2");
assert_eq!(snap.tokens_in, 100);
assert_eq!(snap.tokens_out, 50);
}
#[test]
fn compact_review_mode_parses_all_variants() {
assert_eq!(
CompactReviewMode::parse("always"),
Some(CompactReviewMode::Always)
);
assert_eq!(
CompactReviewMode::parse("manual-only"),
Some(CompactReviewMode::ManualOnly)
);
assert_eq!(
CompactReviewMode::parse("manual_only"),
Some(CompactReviewMode::ManualOnly)
);
assert_eq!(
CompactReviewMode::parse("never"),
Some(CompactReviewMode::Never)
);
assert_eq!(CompactReviewMode::parse(" bogus "), None);
}
#[test]
fn compact_review_mode_should_review_matrix() {
assert!(CompactReviewMode::Always.should_review(false));
assert!(CompactReviewMode::Always.should_review(true));
assert!(!CompactReviewMode::ManualOnly.should_review(false));
assert!(CompactReviewMode::ManualOnly.should_review(true));
assert!(!CompactReviewMode::Never.should_review(false));
assert!(!CompactReviewMode::Never.should_review(true));
}
#[test]
fn compact_review_registry_new_request_rejects_previous() {
let reg = std::sync::Arc::new(CompactReviewRegistry::new());
let _sub = reg.subscribe();
let rx_a = reg.request(PendingCompactReview {
review_id: "rA".into(),
summary: String::new(),
slice_preview: String::new(),
slice_count: 0,
range_start: 0,
range_end: 0,
tokens_before: 0,
emitted_at: chrono::Utc::now(),
});
let _rx_b = reg.request(PendingCompactReview {
review_id: "rB".into(),
summary: String::new(),
slice_preview: String::new(),
slice_count: 0,
range_start: 0,
range_end: 0,
tokens_before: 0,
emitted_at: chrono::Utc::now(),
});
let got = rx_a.blocking_recv().unwrap();
assert!(matches!(got, CompactReviewDecision::Reject));
}
fn mk_form(form_id: &str, prompt: &str) -> crate::form::PendingForm {
crate::form::PendingForm {
form_id: form_id.into(),
run_id: crate::event::FlowRunId::now(),
tool_use_id: "tu".into(),
kind: crate::form::FormKind::Confirm {
prompt: prompt.into(),
},
emitted_at: chrono::Utc::now(),
}
}
#[test]
fn form_registry_auto_cancels_without_subscriber() {
let reg = FormRegistry::new();
let rx = reg.request(mk_form("f1", "sure?"));
let got = rx.blocking_recv().unwrap();
assert_eq!(got, crate::form::FormAnswer::Cancelled);
assert!(reg.list_pending().is_empty());
}
#[test]
fn form_registry_delivers_answer_by_form_id() {
let reg = std::sync::Arc::new(FormRegistry::new());
let _sub = reg.subscribe();
let rx = reg.request(mk_form("fA", "?"));
assert_eq!(reg.list_pending().len(), 1);
let ok = reg.submit("fA", crate::form::FormAnswer::Confirmed { value: true });
assert!(ok);
let got = rx.blocking_recv().unwrap();
assert_eq!(got, crate::form::FormAnswer::Confirmed { value: true });
assert!(reg.list_pending().is_empty());
}
#[test]
fn form_registry_submit_unknown_id_is_noop() {
let reg = std::sync::Arc::new(FormRegistry::new());
let _sub = reg.subscribe();
let _rx = reg.request(mk_form("real", "?"));
assert!(!reg.submit("ghost", crate::form::FormAnswer::Cancelled));
assert_eq!(reg.list_pending().len(), 1);
}
#[test]
fn form_registry_cancel_all_flushes_pending() {
let reg = std::sync::Arc::new(FormRegistry::new());
let _sub = reg.subscribe();
let rx_a = reg.request(mk_form("a", "?"));
let rx_b = reg.request(mk_form("b", "?"));
reg.cancel_all();
assert_eq!(
rx_a.blocking_recv().unwrap(),
crate::form::FormAnswer::Cancelled
);
assert_eq!(
rx_b.blocking_recv().unwrap(),
crate::form::FormAnswer::Cancelled
);
assert!(reg.list_pending().is_empty());
}
#[test]
fn form_registry_queues_multiple_pending() {
let reg = std::sync::Arc::new(FormRegistry::new());
let _sub = reg.subscribe();
let _rx1 = reg.request(mk_form("1", "?"));
let _rx2 = reg.request(mk_form("2", "?"));
let pending = reg.list_pending();
assert_eq!(pending.len(), 2);
assert_eq!(pending[0].form_id, "1");
assert_eq!(pending[1].form_id, "2");
}
#[test]
fn replay_without_degraded_events_preserves_image_parts() {
let dir = TempDir::new().unwrap();
let user_msg = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000002","message":{"role":"user","parts":[{"type":"image","source":{"media_type":"image/png","data":{"kind":"path","path":"/tmp/x.png"}}}],"turn_id":"019f0000-0000-7000-0000-000000000002"},"ts":"2026-07-07T00:00:00Z"}"#;
write_events(dir.path(), &[user_msg]);
let entries = replay_transcript_from(&dir.path().join("events.jsonl")).unwrap();
let msg = entries
.into_iter()
.find_map(|e| match e {
TranscriptEntry::Message { message, .. } => Some(message),
_ => None,
})
.unwrap();
assert!(matches!(
msg.parts[0],
crate::message::MessagePart::Image { .. }
));
}
#[test]
fn replay_messages_from_old_format_no_seq_no_ts() {
let dir = TempDir::new().unwrap();
let user_json = r#"{"type":"user_msg","turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"text","text":"hello"}],"turn_id":"019f0000-0000-7000-0000-000000000001"}}"#;
let asst_json = r#"{"type":"assistant_msg","turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"assistant","parts":[{"type":"text","text":"hi there"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"flow_run_id":null}"#;
write_events(dir.path(), &[user_json, asst_json]);
let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
assert_eq!(msgs.len(), 2, "should load both messages from old format");
assert_eq!(msgs[0].text_concat(), "hello");
assert_eq!(msgs[1].text_concat(), "hi there");
}
#[test]
fn replay_messages_from_old_format_with_null_fields() {
let dir = TempDir::new().unwrap();
let sys_json = r#"{"type":"system_msg","turn_id":"019f0000-0000-7000-0000-000000000003","message":{"role":"system","parts":[{"type":"text","text":"note"}],"turn_id":"019f0000-0000-7000-0000-000000000003"}}"#;
write_events(dir.path(), &[sys_json]);
let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
assert_eq!(msgs.len(), 1);
assert_eq!(msgs[0].text_concat(), "note");
}
#[test]
fn replay_messages_from_applies_attachment_degrade() {
let dir = TempDir::new().unwrap();
let user_msg = r#"{"type":"user_msg","seq":5,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"image","source":{"media_type":"image/png","data":{"kind":"path","path":"/tmp/photo.png"}}},{"type":"text","text":"describe"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
let degrade = r#"{"type":"attachment_degraded","seq":6,"turn_id":null,"flow_run_id":null,"message_seq":5,"part_index":0,"file_basename":"photo.png","reason":"image_too_large","ts":"2026-07-07T00:00:01Z"}"#;
write_events(dir.path(), &[user_msg, degrade]);
let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
assert_eq!(msgs.len(), 1, "only the user message (patched)");
assert_eq!(msgs[0].parts.len(), 2);
match &msgs[0].parts[0] {
crate::message::MessagePart::Text { text } => {
assert!(text.contains("photo.png"), "expected basename: {text}");
assert!(text.contains("image_too_large"), "expected reason: {text}");
assert!(
text.starts_with("[attachment unavailable"),
"expected stub prefix: {text}"
);
}
other => panic!("expected Text stub, got {other:?}"),
}
assert!(
matches!(msgs[0].parts[1], crate::message::MessagePart::Text { .. }),
"second part should remain text"
);
}
#[test]
fn replay_messages_from_degrade_before_message_is_noop() {
let dir = TempDir::new().unwrap();
let degrade = r#"{"type":"attachment_degraded","seq":1,"turn_id":null,"flow_run_id":null,"message_seq":99,"part_index":0,"file_basename":"x.png","reason":"test","ts":"2026-07-07T00:00:00Z"}"#;
let user_msg = r#"{"type":"user_msg","seq":2,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"image","source":{"media_type":"image/png","data":{"kind":"path","path":"/tmp/x.png"}}}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:01Z"}"#;
write_events(dir.path(), &[degrade, user_msg]);
let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
assert_eq!(msgs.len(), 1);
assert!(
matches!(msgs[0].parts[0], crate::message::MessagePart::Image { .. }),
"image should remain when degrade targets unknown seq"
);
}
#[test]
fn replay_messages_from_degrade_wrong_seq_leaves_image() {
let dir = TempDir::new().unwrap();
let user_msg = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"image","source":{"media_type":"image/png","data":{"kind":"path","path":"/tmp/x.png"}}}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
let degrade = r#"{"type":"attachment_degraded","seq":2,"turn_id":null,"flow_run_id":null,"message_seq":2,"part_index":0,"file_basename":"x.png","reason":"test","ts":"2026-07-07T00:00:01Z"}"#;
write_events(dir.path(), &[user_msg, degrade]);
let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
assert_eq!(msgs.len(), 1);
assert!(
matches!(msgs[0].parts[0], crate::message::MessagePart::Image { .. }),
"image should remain when degrade targets wrong seq"
);
}
#[test]
fn replay_messages_from_applies_context_compact() {
let dir = TempDir::new().unwrap();
let user1 = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"text","text":"old u1"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
let asst1 = r#"{"type":"assistant_msg","seq":2,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"assistant","parts":[{"type":"text","text":"old a1"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"flow_run_id":null,"ts":"2026-07-07T00:00:01Z"}"#;
let user2 = r#"{"type":"user_msg","seq":3,"turn_id":"019f0000-0000-7000-0000-000000000002","message":{"role":"user","parts":[{"type":"text","text":"old u2"}],"turn_id":"019f0000-0000-7000-0000-000000000002"},"ts":"2026-07-07T00:00:02Z"}"#;
let summary = r#"{"type":"system_msg","seq":4,"turn_id":"019f0000-0000-7000-0000-000000000002","message":{"role":"system","parts":[{"type":"compact_summary","summary":"two messages compacted","seq_start":0,"seq_end":1,"count":2}],"turn_id":"019f0000-0000-7000-0000-000000000002"},"ts":"2026-07-07T00:00:03Z"}"#;
let compact = r#"{"type":"context_compact","seq":5,"session_id":"sess","before_tokens":200,"after_tokens":50,"compacted_range_start":0,"compacted_range_end":1,"summary_text":"two messages compacted","replacement_msg_seq":4,"ts":"2026-07-07T00:00:04Z"}"#;
let after = r#"{"type":"user_msg","seq":6,"turn_id":"019f0000-0000-7000-0000-000000000003","message":{"role":"user","parts":[{"type":"text","text":"after compact"}],"turn_id":"019f0000-0000-7000-0000-000000000003"},"ts":"2026-07-07T00:00:05Z"}"#;
write_events(dir.path(), &[user1, asst1, user2, summary, compact, after]);
let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
assert_eq!(msgs.len(), 3, "compact summary + user2 + after compact");
assert!(
matches!(
msgs[0].parts[0],
crate::message::MessagePart::CompactSummary { .. }
),
"first should be compact summary"
);
if let crate::message::MessagePart::CompactSummary { summary, .. } = &msgs[0].parts[0] {
assert_eq!(summary, "two messages compacted");
}
assert_eq!(msgs[1].text_concat(), "old u2");
assert_eq!(msgs[2].text_concat(), "after compact");
}
#[test]
fn replay_messages_from_no_replacement_seq_ignores_compact() {
let dir = TempDir::new().unwrap();
let user1 = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"text","text":"hello"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
let compact = r#"{"type":"context_compact","seq":2,"session_id":"sess","before_tokens":200,"after_tokens":50,"compacted_range_start":0,"compacted_range_end":0,"summary_text":"ignored","replacement_msg_seq":null,"ts":"2026-07-07T00:00:01Z"}"#;
write_events(dir.path(), &[user1, compact]);
let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
assert_eq!(msgs.len(), 1, "compact without replacement seq is ignored");
assert_eq!(msgs[0].text_concat(), "hello");
}
#[test]
fn replay_messages_from_compact_after_no_change_ignored() {
let dir = TempDir::new().unwrap();
let user1 = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"text","text":"hello"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
let summary = r#"{"type":"system_msg","seq":2,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"system","parts":[{"type":"compact_summary","summary":"no change","seq_start":0,"seq_end":0,"count":1}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:01Z"}"#;
let compact = r#"{"type":"context_compact","seq":3,"session_id":"sess","before_tokens":50,"after_tokens":100,"compacted_range_start":0,"compacted_range_end":0,"summary_text":"no change","replacement_msg_seq":2,"ts":"2026-07-07T00:00:02Z"}"#;
write_events(dir.path(), &[user1, summary, compact]);
let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
assert_eq!(msgs.len(), 2, "compact with after>=before is ignored");
}
#[test]
fn replay_messages_from_missing_file_returns_empty() {
let dir = TempDir::new().unwrap();
let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
assert!(msgs.is_empty());
}
#[test]
fn replay_messages_from_empty_file_returns_empty() {
let dir = TempDir::new().unwrap();
write_events(dir.path(), &[]);
let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
assert!(msgs.is_empty());
}
#[test]
fn replay_all_messages_with_seq_includes_compacted() {
let dir = TempDir::new().unwrap();
let user1 = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"text","text":"old"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
let summary = r#"{"type":"system_msg","seq":4,"turn_id":"019f0000-0000-7000-0000-000000000002","message":{"role":"system","parts":[{"type":"compact_summary","summary":"s","seq_start":0,"seq_end":0,"count":1}],"turn_id":"019f0000-0000-7000-0000-000000000002"},"ts":"2026-07-07T00:00:01Z"}"#;
let compact = r#"{"type":"context_compact","seq":5,"session_id":"sess","before_tokens":200,"after_tokens":50,"compacted_range_start":0,"compacted_range_end":0,"summary_text":"s","replacement_msg_seq":4,"ts":"2026-07-07T00:00:02Z"}"#;
write_events(dir.path(), &[user1, summary, compact]);
let all = replay_all_messages_with_seq(&dir.path().join("events.jsonl")).unwrap();
assert_eq!(all.len(), 2, "all messages preserved (no compaction)");
assert_eq!(all[0].1.text_concat(), "old");
}
}