diff --git a/crates/llm-engine/src/engine.rs b/crates/llm-engine/src/engine.rs index d41d8631..4efd53f6 100644 --- a/crates/llm-engine/src/engine.rs +++ b/crates/llm-engine/src/engine.rs @@ -50,6 +50,9 @@ pub enum EngineError { /// Config warnings (unsupported options) #[error("Config warnings: {}", .0.iter().map(|w| w.to_string()).collect::>().join(", "))] ConfigWarnings(Vec), + /// A durable-history observer rejected an item before it entered history. + #[error("History append failed: {0}")] + HistoryAppend(String), } /// Tool registration error @@ -222,10 +225,10 @@ pub struct Engine { /// truncation have been applied — i.e. on the same data that /// enters history. tool_result_cbs: Vec>, - /// History-append callbacks. Invoked for non-streamed items when they - /// are appended to persistent engine history, so upper layers can - /// broadcast those items using history itself as the source of truth. - history_append_cbs: Vec>, + /// History-append callbacks. Invoked before non-streamed items enter + /// engine history. An error rejects the item and aborts the turn, allowing + /// upper layers to make durable storage the commit gate. + history_append_cbs: Vec Result<(), String> + Send + Sync>>, /// Request configuration (max_tokens, temperature, etc.) request_config: RequestConfig, /// Whether the previous run was interrupted @@ -498,23 +501,31 @@ impl Engine { } } - /// Register a callback invoked for items appended directly to engine - /// history outside streaming timeline callbacks. - pub fn on_history_append(&mut self, callback: impl Fn(&Item) + Send + Sync + 'static) { + /// Register a fallible callback invoked before an item enters engine + /// history. Returning an error rejects that item and aborts the turn. + pub fn on_history_append( + &mut self, + callback: impl Fn(&Item) -> Result<(), String> + Send + Sync + 'static, + ) { self.history_append_cbs.push(Box::new(callback)); } - fn emit_history_append(&self, item: &Item) { + fn emit_history_append(&self, item: &Item) -> Result<(), EngineError> { for cb in &self.history_append_cbs { - cb(item); + cb(item).map_err(EngineError::HistoryAppend)?; } + Ok(()) } - fn append_history_items(&mut self, items: impl IntoIterator) { + fn append_history_items( + &mut self, + items: impl IntoIterator, + ) -> Result<(), EngineError> { for item in items { - self.emit_history_append(&item); + self.emit_history_append(&item)?; self.history.push(item); } + Ok(()) } fn request_trace_payload(&self, request: &Request) -> Value { @@ -1125,9 +1136,13 @@ impl Engine { // These are committed *before* the per-request clone so they // participate in the LLM request below and get persisted by // the caller that owns durable history. - let pending = self.interceptor.pending_history_appends().await; + let pending = self + .interceptor + .pending_history_appends() + .await + .map_err(EngineError::HistoryAppend)?; if !pending.is_empty() { - self.append_history_items(pending); + self.append_history_items(pending)?; } // Clone the history into a per-request context. Everything @@ -1202,7 +1217,7 @@ impl Engine { return Err(EngineError::Aborted(reason)); } PreRequestAction::YieldWith(items) => { - self.append_history_items(items.clone()); + self.append_history_items(items.clone())?; request_context.extend(items); info!("Yielded by interceptor after pre-request history append"); for cb in &self.turn_end_cbs { @@ -1220,7 +1235,7 @@ impl Engine { return Ok(EngineResult::Yielded); } PreRequestAction::ContinueWith(items) => { - self.append_history_items(items.clone()); + self.append_history_items(items.clone())?; request_context.extend(items); } PreRequestAction::Continue => {} @@ -1280,7 +1295,7 @@ impl Engine { let assistant_items = self.build_assistant_items(&reasoning_items, &text_blocks, &[]); if !assistant_items.is_empty() { - self.append_history_items(assistant_items); + self.append_history_items(assistant_items)?; } self.emit_llm_continuation( current_llm_call, @@ -1307,7 +1322,7 @@ impl Engine { let tool_calls = self.tool_call_collector.take_collected(); let assistant_items = self.build_assistant_items(&reasoning_items, &text_blocks, &tool_calls); - self.append_history_items(assistant_items); + self.append_history_items(assistant_items)?; if tool_calls.is_empty() { match self.interceptor.on_turn_end(&self.history).await { @@ -1316,7 +1331,7 @@ impl Engine { return Ok(EngineResult::Finished); } TurnEndAction::ContinueWithMessages(additional) => { - self.append_history_items(additional); + self.append_history_items(additional)?; continue; } TurnEndAction::Pause => { @@ -1610,7 +1625,7 @@ impl Engine { result.is_error, ) }); - self.append_history_items(items); + self.append_history_items(items)?; Ok(None) } Err(err) => { @@ -1815,12 +1830,15 @@ impl Engine { self.history = items; } - /// Append items to history and notify history-append observers for each - /// item before it lands. This is the only public Mutable-state API for - /// growing engine history; callers that need session-log persistence must - /// install [`on_history_append`](Self::on_history_append) before calling it. - pub fn append_history(&mut self, items: impl IntoIterator) { - self.append_history_items(items); + /// Append items to history after every history-append observer accepts the + /// item. This is the only public Mutable-state API for growing engine + /// history; callers that need session-log persistence must install + /// [`on_history_append`](Self::on_history_append) before calling it. + pub fn append_history( + &mut self, + items: impl IntoIterator, + ) -> Result<(), EngineError> { + self.append_history_items(items) } /// Truncate history without emitting append callbacks. @@ -1969,9 +1987,9 @@ impl Engine { PromptAction::Continue => Vec::new(), PromptAction::ContinueWith(items) => items, }; - self.append_history_items(std::iter::once(user_item)); + self.append_history_items(std::iter::once(user_item))?; if !extras.is_empty() { - self.append_history_items(extras); + self.append_history_items(extras)?; } let result = self.run_turn_loop().await; self.finalize_interruption(result).await diff --git a/crates/llm-engine/src/interceptor.rs b/crates/llm-engine/src/interceptor.rs index 3e088ced..5c5a307f 100644 --- a/crates/llm-engine/src/interceptor.rs +++ b/crates/llm-engine/src/interceptor.rs @@ -158,8 +158,8 @@ pub trait Interceptor: Send + Sync { /// reproducible per-request transformations (pruning, content /// trimming, cache anchors) that depend only on the existing /// history. - async fn pending_history_appends(&self) -> Vec { - Vec::new() + async fn pending_history_appends(&self) -> Result, String> { + Ok(Vec::new()) } /// Called before each LLM request. The context starts as a clone diff --git a/crates/llm-engine/tests/engine_state_test.rs b/crates/llm-engine/tests/engine_state_test.rs index 12faac64..0ce02700 100644 --- a/crates/llm-engine/tests/engine_state_test.rs +++ b/crates/llm-engine/tests/engine_state_test.rs @@ -44,12 +44,18 @@ fn test_mutable_history_manipulation() { assert!(engine.history().is_empty()); // Add to history - engine.append_history(vec![Item::user_message("Hello")]); - engine.append_history(vec![Item::assistant_message("Hi there!")]); + engine + .append_history(vec![Item::user_message("Hello")]) + .unwrap(); + engine + .append_history(vec![Item::assistant_message("Hi there!")]) + .unwrap(); assert_eq!(engine.history().len(), 2); // Append to history via the callback-aware API. - engine.append_history(vec![Item::user_message("How are you?")]); + engine + .append_history(vec![Item::user_message("How are you?")]) + .unwrap(); assert_eq!(engine.history().len(), 3); // Clear history @@ -86,15 +92,20 @@ fn test_mutable_append_history() { if let Some(text) = item.as_text() { observed_for_callback.lock().unwrap().push(text.to_string()); } + Ok(()) }); - engine.append_history(vec![Item::user_message("First")]); + engine + .append_history(vec![Item::user_message("First")]) + .unwrap(); - engine.append_history(vec![ - Item::assistant_message("Response 1"), - Item::user_message("Second"), - Item::assistant_message("Response 2"), - ]); + engine + .append_history(vec![ + Item::assistant_message("Response 1"), + Item::user_message("Second"), + Item::assistant_message("Response 2"), + ]) + .unwrap(); assert_eq!(engine.history().len(), 4); assert_eq!( @@ -157,6 +168,40 @@ fn test_mutable_can_register_tool() { engine.register_tool(tool.definition()); } +/// A durable-history failure on a tool call must stop the turn before the +/// tool can produce an external side effect. +#[tokio::test] +async fn history_append_failure_stops_before_tool_execution() { + let client = MockLlmClient::new(vec![ + Event::tool_use_start(0, "call_1", "count_tool"), + Event::tool_input_delta(0, r#"{}"#), + Event::tool_use_stop(0), + Event::Status(StatusEvent { + status: ResponseStatus::Completed, + }), + ]); + let tool = CountingTool::new("count_tool"); + let mut engine = Engine::new(client); + engine.register_tool(tool.definition()); + engine.on_history_append(|item| { + if item.is_tool_call() { + Err("simulated ENOSPC".to_string()) + } else { + Ok(()) + } + }); + + let mut engine = engine.lock(); + let error = engine.run("use the tool").await.unwrap_err(); + + assert!( + matches!(error, EngineError::HistoryAppend(ref message) if message == "simulated ENOSPC") + ); + assert_eq!(tool.call_count(), 0); + assert_eq!(engine.history().len(), 1); + assert_eq!(engine.history()[0].as_text(), Some("use the tool")); +} + // ============================================================================= // State Transition Tests // ============================================================================= @@ -168,8 +213,12 @@ fn test_lock_transition() { let mut engine = Engine::new(client); engine.set_system_prompt("System"); - engine.append_history(vec![Item::user_message("Hello")]); - engine.append_history(vec![Item::assistant_message("Hi")]); + engine + .append_history(vec![Item::user_message("Hello")]) + .unwrap(); + engine + .append_history(vec![Item::assistant_message("Hi")]) + .unwrap(); // Lock let locked_engine = engine.lock(); @@ -186,14 +235,18 @@ fn test_unlock_transition() { let client = MockLlmClient::new(vec![]); let mut engine = Engine::new(client); - engine.append_history(vec![Item::user_message("Hello")]); + engine + .append_history(vec![Item::user_message("Hello")]) + .unwrap(); let locked_engine = engine.lock(); // Unlock let mut engine = locked_engine.unlock(); // History operations are available again in Mutable state - engine.append_history(vec![Item::assistant_message("Hi")]); + engine + .append_history(vec![Item::assistant_message("Hi")]) + .unwrap(); engine.clear_history(); assert!(engine.history().is_empty()); } @@ -316,8 +369,12 @@ async fn test_locked_prefix_len_tracking() { let mut engine = Engine::new(client); // Add items beforehand - engine.append_history(vec![Item::user_message("Pre-existing message 1")]); - engine.append_history(vec![Item::assistant_message("Pre-existing response 1")]); + engine + .append_history(vec![Item::user_message("Pre-existing message 1")]) + .unwrap(); + engine + .append_history(vec![Item::assistant_message("Pre-existing response 1")]) + .unwrap(); assert_eq!(engine.history().len(), 2); @@ -387,10 +444,12 @@ async fn test_unlock_edit_relock() { ]]); let mut engine = Engine::new(client); - engine.append_history(vec![ - Item::user_message("Hello"), - Item::assistant_message("Hi"), - ]); + engine + .append_history(vec![ + Item::user_message("Hello"), + Item::assistant_message("Hi"), + ]) + .unwrap(); // Lock -> Unlock let locked = engine.lock(); @@ -400,7 +459,9 @@ async fn test_unlock_edit_relock() { // Edit history unlocked.clear_history(); - unlocked.append_history(vec![Item::user_message("Fresh start")]); + unlocked + .append_history(vec![Item::user_message("Fresh start")]) + .unwrap(); // Re-lock let relocked = unlocked.lock(); diff --git a/crates/session-store/src/fs_store.rs b/crates/session-store/src/fs_store.rs index 1339e7c0..cf0edab0 100644 --- a/crates/session-store/src/fs_store.rs +++ b/crates/session-store/src/fs_store.rs @@ -20,17 +20,23 @@ use crate::segment_log::LogEntry; use crate::store::{Store, StoreError}; use crate::{SegmentId, SessionId}; use std::fs; -use std::io::Write; +use std::io::{Read, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; use std::time::SystemTime; /// Filesystem-backed JSONL store. /// /// Each segment is stored as a single `.jsonl` file with one [`LogEntry`] -/// per line. Writes use append mode for crash safety. +/// per line. A trailing line is committed only once its newline has been +/// written; readers ignore an unterminated tail and the next append removes it. #[derive(Clone)] pub struct FsStore { root: PathBuf, + /// Serialises append repair + write + rollback across clones. A failed + /// `write_all` may have extended the file, so rollback is safe only while + /// no sibling writer can append behind it. + append_lock: Arc>, } impl FsStore { @@ -39,7 +45,10 @@ impl FsStore { pub fn new(root: impl Into) -> Result { let root = root.into(); fs::create_dir_all(&root)?; - Ok(Self { root }) + Ok(Self { + root, + append_lock: Arc::new(Mutex::new(())), + }) } /// Return the filesystem root used by this store. @@ -101,22 +110,62 @@ impl FsStore { } fn append_line(&self, path: &Path, line: &str) -> Result<(), StoreError> { + let _guard = self + .append_lock + .lock() + .map_err(|_| std::io::Error::other("session store append lock was poisoned"))?; if let Some(parent) = path.parent() { fs::create_dir_all(parent)?; } let mut file = fs::OpenOptions::new() .create(true) + .read(true) + .write(true) .append(true) .open(path)?; - file.write_all(line.as_bytes())?; - file.write_all(b"\n")?; - // Append-mode write is the durability boundary; an explicit - // `sync_all` here would multiply latency by ~10× for no gain - // since the kernel already orders concurrent `O_APPEND` writes. + let committed_len = Self::truncate_uncommitted_tail(&mut file)?; + let mut record = Vec::with_capacity(line.len() + 1); + record.extend_from_slice(line.as_bytes()); + record.push(b'\n'); + + if let Err(write_error) = file.write_all(&record) { + return match file.set_len(committed_len) { + Ok(()) => Err(write_error.into()), + Err(rollback_error) => Err(std::io::Error::new( + rollback_error.kind(), + format!( + "session append failed ({write_error}) and partial-write rollback failed: {rollback_error}" + ), + ) + .into()), + }; + } Ok(()) } - fn parse_jsonl(content: &str) -> Result, StoreError> { + /// Return only newline-terminated records. A process interruption or + /// ENOSPC can leave the final UTF-8 code point / JSON object incomplete; + /// without a newline that record never crossed the commit boundary. + fn complete_jsonl_prefix(content: &[u8]) -> &[u8] { + if content.last() == Some(&b'\n') { + return content; + } + match content.iter().rposition(|byte| *byte == b'\n') { + Some(index) => &content[..=index], + None => &[], + } + } + + fn parse_jsonl(content: &[u8]) -> Result, StoreError> { + let complete = Self::complete_jsonl_prefix(content); + let content = std::str::from_utf8(complete).map_err(|error| StoreError::Corrupt { + line: complete[..error.valid_up_to()] + .iter() + .filter(|byte| **byte == b'\n') + .count() + + 1, + message: error.to_string(), + })?; let mut entries = Vec::new(); for (i, line) in content.lines().enumerate() { if line.trim().is_empty() { @@ -130,6 +179,43 @@ impl FsStore { } Ok(entries) } + + /// Remove a prior unterminated record and return the committed file size. + /// Scans backwards in bounded chunks so repairing a large session does not + /// require loading it into memory. + fn truncate_uncommitted_tail(file: &mut fs::File) -> std::io::Result { + const SCAN_BYTES: usize = 8 * 1024; + + let len = file.metadata()?.len(); + if len == 0 { + return Ok(0); + } + + file.seek(SeekFrom::End(-1))?; + let mut last = [0_u8; 1]; + file.read_exact(&mut last)?; + if last[0] == b'\n' { + return Ok(len); + } + + let mut end = len; + let mut buffer = [0_u8; SCAN_BYTES]; + while end > 0 { + let start = end.saturating_sub(SCAN_BYTES as u64); + let chunk_len = (end - start) as usize; + file.seek(SeekFrom::Start(start))?; + file.read_exact(&mut buffer[..chunk_len])?; + if let Some(index) = buffer[..chunk_len].iter().rposition(|byte| *byte == b'\n') { + let committed_len = start + index as u64 + 1; + file.set_len(committed_len)?; + return Ok(committed_len); + } + end = start; + } + + file.set_len(0)?; + Ok(0) + } } impl Store for FsStore { @@ -152,7 +238,7 @@ impl Store for FsStore { if !path.exists() { return Err(StoreError::NotFound(segment_id)); } - let content = fs::read_to_string(&path)?; + let content = fs::read(&path)?; Self::parse_jsonl(&content) } @@ -251,8 +337,17 @@ impl Store for FsStore { if !path.exists() { return Err(StoreError::NotFound(segment_id)); } - let content = fs::read_to_string(&path)?; - Ok(content.lines().filter(|l| !l.trim().is_empty()).count()) + let content = fs::read(&path)?; + let complete = Self::complete_jsonl_prefix(&content); + let complete = std::str::from_utf8(complete).map_err(|error| StoreError::Corrupt { + line: complete[..error.valid_up_to()] + .iter() + .filter(|byte| **byte == b'\n') + .count() + + 1, + message: error.to_string(), + })?; + Ok(complete.lines().filter(|l| !l.trim().is_empty()).count()) } fn append_trace( diff --git a/crates/session-store/src/segment_log.rs b/crates/session-store/src/segment_log.rs index ace151a3..58e62008 100644 --- a/crates/session-store/src/segment_log.rs +++ b/crates/session-store/src/segment_log.rs @@ -69,7 +69,10 @@ pub enum LogEntry { /// Field name is `trigger` (not `kind`) because the LogEntry /// serde tag already occupies `"kind"`. /// - /// Marker only — replay does not mutate `RestoredState`. + /// Replay marks the run interrupted until a terminal `RunCompleted`, + /// `RunErrored`, or `PausedTurnAbandoned` entry proves how it ended. This + /// makes a process/disk failure between Invoke and its terminal record + /// restore conservatively instead of re-running a dangling tool call. Invoke { ts: u64, trigger: InvokeKind }, /// User input accepted at submit time. Carries the original typed @@ -236,9 +239,9 @@ pub fn collect_state(entries: &[LogEntry]) -> RestoredState { state.history = history.iter().cloned().map(Item::from).collect(); } LogEntry::Invoke { .. } => { - // Marker only; no state mutation. The trailing - // UserInput / SystemItem / TurnEnd entries carry all - // replay-relevant data. + // A terminal run record below clears or refines this. If the + // log ends first, restore must treat the turn as interrupted. + state.last_run_interrupted = true; } LogEntry::UserInput { segments, .. } => { let text = Segment::flatten_to_text(segments); @@ -368,6 +371,35 @@ mod tests { assert!(!state.last_run_interrupted); } + #[test] + fn replay_incomplete_invoke_is_interrupted() { + let state = collect_state(&[ + LogEntry::SegmentStart { + ts: 1000, + session_id: uuid::Uuid::nil(), + system_prompt: None, + config: RequestConfig::default(), + history: vec![], + forked_from: None, + compacted_from: None, + }, + LogEntry::Invoke { + ts: 2000, + trigger: InvokeKind::UserSend, + }, + LogEntry::UserInput { + ts: 2001, + segments: vec![Segment::text("run a tool")], + }, + LogEntry::AssistantItem { + ts: 3000, + item: Item::tool_call("call_1", "side_effect", "{}").into(), + }, + ]); + + assert!(state.last_run_interrupted); + } + #[test] fn replay_with_tool_calls() { let state = collect_state(&[ @@ -546,7 +578,7 @@ mod tests { } #[test] - fn replay_invoke_marker_does_not_mutate_state() { + fn replay_invoke_marker_only_mutates_interrupted_state() { let state = collect_state(&[ LogEntry::SegmentStart { ts: 0, @@ -576,6 +608,7 @@ mod tests { ]); assert_eq!(state.history.len(), 1); assert_eq!(state.turn_count, 1); + assert!(state.last_run_interrupted); } #[test] diff --git a/crates/session-store/src/store.rs b/crates/session-store/src/store.rs index 5937e4af..7510d225 100644 --- a/crates/session-store/src/store.rs +++ b/crates/session-store/src/store.rs @@ -40,8 +40,8 @@ pub enum StoreError { pub trait Store: Send + Sync { /// Append a single log entry to the segment log. /// - /// One line per call. The kernel orders concurrent `O_APPEND` writes - /// for lines < `PIPE_BUF`, so user-space serialization is unnecessary. + /// One committed line per successful call. Implementations must not expose + /// a failed call's partial record as committed data on later reads. fn append( &self, session_id: SessionId, diff --git a/crates/session-store/tests/fs_store_test.rs b/crates/session-store/tests/fs_store_test.rs index 1b940c05..9ae6f1c7 100644 --- a/crates/session-store/tests/fs_store_test.rs +++ b/crates/session-store/tests/fs_store_test.rs @@ -3,6 +3,7 @@ use llm_engine::llm_client::types::{Item, RequestConfig}; use session_store::{ FsStore, LogEntry, Store, TraceEntry, collect_state, new_segment_id, new_session_id, }; +use std::io::Write; fn nil_session_start(ts: u64, session_id: uuid::Uuid) -> LogEntry { LogEntry::SegmentStart { @@ -224,6 +225,71 @@ fn read_entry_count_matches_append_tally() { assert_eq!(store.read_entry_count(sid, segid).unwrap(), entries.len()); } +#[test] +fn unterminated_utf8_tail_is_ignored_and_replaced_on_append() { + let dir = tempfile::tempdir().unwrap(); + let store = FsStore::new(dir.path()).unwrap(); + let sid = new_session_id(); + let segid = new_segment_id(); + let path = dir + .path() + .join(sid.to_string()) + .join(format!("{segid}.jsonl")); + + store + .append(sid, segid, &nil_session_start(1, sid)) + .unwrap(); + std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap() + // First byte of a three-byte UTF-8 code point, matching an ENOSPC + // partial write observed in a real session log. + .write_all(&[0xe3]) + .unwrap(); + + assert_eq!(store.read_all(sid, segid).unwrap().len(), 1); + assert_eq!(store.read_entry_count(sid, segid).unwrap(), 1); + + let next = LogEntry::UserInput { + ts: 2, + segments: vec![protocol::Segment::text("recovered")], + }; + store.append(sid, segid, &next).unwrap(); + + let bytes = std::fs::read(&path).unwrap(); + assert!(std::str::from_utf8(&bytes).is_ok()); + assert_eq!(store.read_all(sid, segid).unwrap().len(), 2); + assert_eq!(store.read_entry_count(sid, segid).unwrap(), 2); +} + +#[test] +fn newline_terminated_invalid_utf8_is_reported_as_corruption() { + let dir = tempfile::tempdir().unwrap(); + let store = FsStore::new(dir.path()).unwrap(); + let sid = new_session_id(); + let segid = new_segment_id(); + let path = dir + .path() + .join(sid.to_string()) + .join(format!("{segid}.jsonl")); + + store + .append(sid, segid, &nil_session_start(1, sid)) + .unwrap(); + std::fs::OpenOptions::new() + .append(true) + .open(path) + .unwrap() + .write_all(&[0xe3, b'\n']) + .unwrap(); + + assert!(matches!( + store.read_all(sid, segid), + Err(session_store::StoreError::Corrupt { line: 2, .. }) + )); +} + #[test] fn lookup_session_of_finds_owning_session() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/worker/src/ipc/interceptor.rs b/crates/worker/src/ipc/interceptor.rs index 8eca9147..ca4ca06b 100644 --- a/crates/worker/src/ipc/interceptor.rs +++ b/crates/worker/src/ipc/interceptor.rs @@ -112,13 +112,14 @@ impl WorkerInterceptor { /// `Item::system_message`s reach the worker via /// `ContinueWith` / `pending_history_appends`, so on-disk order /// matches worker-history order. - fn commit_system_items(&self, items: &[SystemItem]) { + fn commit_system_items(&self, items: &[SystemItem]) -> Result<(), session_store::StoreError> { let Some(writer) = self.log_writer.as_ref() else { - return; + return Ok(()); }; for item in items { - writer.commit_system_item(item.clone()); + writer.commit_system_item(item.clone())?; } + Ok(()) } fn current_turn_index(&self) -> usize { @@ -194,15 +195,17 @@ impl Interceptor for WorkerInterceptor { // `Item::system_message`s, so on-disk order matches // worker-history order. let items: Vec = extras.iter().map(SystemItem::to_history_item).collect(); - self.commit_system_items(&extras); - PromptAction::ContinueWith(items) + match self.commit_system_items(&extras) { + Ok(()) => PromptAction::ContinueWith(items), + Err(error) => PromptAction::Cancel(format!("session persistence failed: {error}")), + } } } - async fn pending_history_appends(&self) -> Vec { + async fn pending_history_appends(&self) -> Result, String> { let drained = self.pending_notifies.drain(); if drained.is_empty() { - return Vec::new(); + return Ok(Vec::new()); } let mut system_items: Vec = Vec::with_capacity(drained.len()); @@ -231,8 +234,9 @@ impl Interceptor for WorkerInterceptor { } } } - self.commit_system_items(&system_items); - items + self.commit_system_items(&system_items) + .map_err(|error| format!("session persistence failed: {error}"))?; + Ok(items) } async fn pre_llm_request(&self, context: &mut Vec) -> PreRequestAction { @@ -278,7 +282,9 @@ impl Interceptor for WorkerInterceptor { let current_tokens = self.estimated_tokens(effective_context.as_ref()); if self.request_threshold_exceeded(current_tokens, effective_context.as_ref()) { - self.commit_system_items(&system_items); + if let Err(error) = self.commit_system_items(&system_items) { + return PreRequestAction::Cancel(format!("session persistence failed: {error}")); + } return if appended_items.is_empty() { PreRequestAction::Yield } else { @@ -292,8 +298,10 @@ impl Interceptor for WorkerInterceptor { if system_items.is_empty() { return PreRequestAction::Continue; } - self.commit_system_items(&system_items); - PreRequestAction::ContinueWith(appended_items) + match self.commit_system_items(&system_items) { + Ok(()) => PreRequestAction::ContinueWith(appended_items), + Err(error) => PreRequestAction::Cancel(format!("session persistence failed: {error}")), + } } async fn pre_tool_call(&self, info: &mut ToolCallInfo) -> PreToolAction { @@ -451,13 +459,17 @@ mod tests { } impl SystemItemCommitter for RecordingSystemItemCommitter { - fn commit_log_entry(&self, entry: session_store::LogEntry) { + fn commit_log_entry( + &self, + entry: session_store::LogEntry, + ) -> Result<(), session_store::StoreError> { if let session_store::LogEntry::SystemItem { item, .. } = entry { self.committed .lock() .expect("committed system-item list poisoned") .push(item); } + Ok(()) } } @@ -1034,7 +1046,7 @@ mod tests { None, ); - let items = interceptor.pending_history_appends().await; + let items = interceptor.pending_history_appends().await.unwrap(); assert_eq!(items.len(), 2); let first = items[0].as_text().unwrap_or_default(); let second = items[1].as_text().unwrap_or_default(); @@ -1048,7 +1060,7 @@ mod tests { ); // Empty buffer → empty Vec (no synthesised items). - let again = interceptor.pending_history_appends().await; + let again = interceptor.pending_history_appends().await.unwrap(); assert!(again.is_empty()); } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index c051ec91..d1a191cc 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -568,9 +568,8 @@ where St: Store + Clone, { /// Append `entry` to the log: disk write → counter bump → in-memory - /// mirror push → broadcast. The kernel orders concurrent `O_APPEND` - /// writes for `< PIPE_BUF` lines, so no user-space serialization is - /// needed across appenders. + /// mirror push → broadcast. The Store owns physical write ordering and + /// partial-write recovery; publication happens only after it returns Ok. pub fn append_entry(&self, entry: LogEntry) -> Result<(), StoreError> { let loc = self.state.location(); self.store.append(loc.session_id, loc.segment_id, &entry)?; @@ -602,13 +601,13 @@ where /// interceptor commit `SystemItem`s without being generic over the /// concrete `Store` type. pub trait SystemItemCommitter: Send + Sync { - fn commit_log_entry(&self, entry: LogEntry); + fn commit_log_entry(&self, entry: LogEntry) -> Result<(), StoreError>; - fn commit_system_item(&self, item: SystemItem) { + fn commit_system_item(&self, item: SystemItem) -> Result<(), StoreError> { self.commit_log_entry(LogEntry::SystemItem { ts: segment_log::now_millis(), item, - }); + }) } } @@ -616,10 +615,8 @@ impl SystemItemCommitter for LogWriterHandle where St: Store + Clone + Send + Sync + 'static, { - fn commit_log_entry(&self, entry: LogEntry) { - if let Err(err) = self.append_entry(entry) { - warn!(error = %err, "session log entry commit failed; dropping"); - } + fn commit_log_entry(&self, entry: LogEntry) -> Result<(), StoreError> { + self.append_entry(entry) } } @@ -914,7 +911,7 @@ impl Worker let writer = self.log_writer_handle(); self.engine_mut().on_history_append(move |item| { if item.is_user_message() { - return; + return Ok(()); } if matches!( item, @@ -923,12 +920,12 @@ impl Worker .. } ) { - return; + return Ok(()); } let entry = session_store::classify_history_item(item, segment_log::now_millis()); - if let Err(err) = writer.append_entry(entry) { - warn!(error = %err, "history append commit failed; dropping"); - } + writer + .append_entry(entry) + .map_err(|error| error.to_string()) }); if self.manifest.session.record_event_trace { let writer = self.log_writer_handle(); @@ -1206,7 +1203,7 @@ impl Worker { }, })?; self.engine_mut() - .append_history(std::iter::once(llm_engine::Item::system_message(body))); + .append_history(std::iter::once(llm_engine::Item::system_message(body)))?; Ok(activation) } @@ -1251,9 +1248,8 @@ impl Worker { } /// Append `entry` to the session log AND publish it through the - /// broadcast sink. No user-space serialization is needed across - /// concurrent appenders — the kernel orders `O_APPEND` writes for - /// lines smaller than `PIPE_BUF`. + /// broadcast sink. The Store is the commit boundary: a failed write is + /// never counted or published. pub(crate) fn commit_entry(&self, entry: LogEntry) -> Result<(), StoreError> { let loc = self.segment_state.location(); self.store.append(loc.session_id, loc.segment_id, &entry)?; @@ -2083,7 +2079,7 @@ impl Worker { &tool_result_summary, ); if !closures.is_empty() { - self.engine_mut().append_history(closures); + self.engine_mut().append_history(closures)?; } self.commit_entry(LogEntry::SystemItem { ts: segment_log::now_millis(), @@ -2094,7 +2090,7 @@ impl Worker { self.engine_mut() .append_history(std::iter::once(llm_engine::Item::system_message( system_note, - ))); + )))?; Ok(()) } @@ -2164,6 +2160,12 @@ impl Worker { ), "run_for_notification expects a non-UserSend InvokeKind; got {kind:?}" ); + // This is a fresh Invoke, not an explicit resume of the interrupted + // turn. Close any dangling tool calls before an auto-run notification + // can enter `Engine::resume` and execute them again after a crash. + if self.engine.as_ref().unwrap().last_run_interrupted() { + self.apply_interrupt_prep()?; + } self.prepare_for_run().await?; // IDLE → active marker for the buffered notification / worker-event @@ -5914,6 +5916,57 @@ mod build_summary_prompt_tests { assert_eq!(interrupt_system_count, 1); } + #[tokio::test] + async fn notification_run_closes_interrupted_tool_call_before_engine_resume() { + let dir = tempfile::tempdir().unwrap(); + let manifest = minimal_manifest(); + let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap(); + let cwd = dir.path().join("workspace"); + std::fs::create_dir_all(&cwd).unwrap(); + let scope = Scope::writable(&cwd).unwrap(); + let authority = WorkerFilesystemAuthority::local(cwd.clone(), cwd.clone()); + let mut worker = Worker::new( + manifest, + Engine::new(NoopClient), + store, + WorkerWorkspaceContext::local_filesystem(None), + authority, + scope, + ) + .await + .unwrap(); + + worker.ensure_segment_head().unwrap(); + worker.wire_history_persistence(); + let dangling_call = Item::tool_call("call-1", "SideEffect", "{}"); + worker + .commit_entry(LogEntry::AssistantItem { + ts: segment_log::now_millis(), + item: dangling_call.clone().into(), + }) + .unwrap(); + worker.engine_mut().set_history(vec![dangling_call]); + worker.engine_mut().set_last_run_interrupted(true); + + worker + .run_for_notification(protocol::InvokeKind::Notify) + .await + .unwrap(); + + let history = worker.engine().history(); + assert!(matches!( + history.get(1), + Some(Item::ToolResult { call_id, .. }) if call_id == "call-1" + )); + assert!(matches!( + history.get(2), + Some(Item::Message { + role: Role::System, + .. + }) + )); + } + #[derive(Clone, Copy)] struct ResidentInjectionGates { summary: bool, diff --git a/docs/report/2026-08-05-session-enospc-partial-jsonl.md b/docs/report/2026-08-05-session-enospc-partial-jsonl.md new file mode 100644 index 00000000..690d640f --- /dev/null +++ b/docs/report/2026-08-05-session-enospc-partial-jsonl.md @@ -0,0 +1,46 @@ +# ENOSPC で Session JSONL の末尾が UTF-8 途中切れになる + +## 観測 + +`Companion1` (`worker-runtime-30`) の直近 Segment +`019fce34-eb04-7420-9cc5-0e7e12f5bd67` は、Workdir の Edit tool が +`no storage space` で失敗した後、会話ログ `.jsonl` の末尾が `0xe3` 1 byte +だけで終わっていた。これは 3 byte UTF-8 文字の先頭 byte であり、実際の末尾は +「Workdir のストレー」の次の文字の途中で切れていた。 + +同 Segment の `.trace.jsonl` は valid UTF-8 だった。再開時の +`stream did not contain valid UTF-8` は provider stream ではなく、Session log を +`fs::read_to_string` した際のエラーだった。 + +## 原因 + +`FsStore::append_line` は JSON 本文と改行を別々に `write_all` し、ENOSPC で +部分書き込みになっても元の file length へ戻していなかった。reader も file 全体を +UTF-8 String として読むため、newline に到達していない未コミット末尾だけで Segment +全体を復元不能にしていた。 + +さらに Engine の history append callback と SystemItem committer は、Store error を +warning にして drop していた。このため disk 上の history を更新できなくても memory +上の history と tool loop が先へ進み得た。 + +## 改善 + +- newline を JSONL record の commit marker とする。 +- reader は newline 未到達の末尾を未コミット record として無視する。 +- 次回 append 前に未コミット末尾を最後の newline まで truncate する。 +- append の partial write は append 開始時の file length へ rollback する。 +- repair/write/rollback は `FsStore` clone 間で直列化する。 +- Engine history append を fallible にし、Store write 成功前には item を memory history + に入れない。 +- tool call の永続化に失敗した turn は tool 実行前に停止する。 +- SystemItem の commit failure も transient context injection にせず turn error にする。 +- `Invoke` 後に terminal run record が無い Segment は restore 時に interrupted とする。 + これにより crash/ENOSPC 後の dangling tool call は新しい user turn の前に閉じられ、 + side effect を無条件に再実行しない。 + +## 境界 + +この修正は process interruption と ENOSPC による trailing partial record を対象にする。 +newline 済み record 内部の破損は silent recovery せず `StoreError::Corrupt` のまま扱う。 +また append ごとの `fsync` は追加していないため、突然の電源断に対する block-level +durability まで保証するものではない。