From 88338229a1d409443083d76438e7c066b707a3aa Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 2 Oct 2026 05:30:17 +0900 Subject: [PATCH] fix: deduplicate compaction terminal failures --- crates/worker/src/controller.rs | 135 ++++++- crates/worker/src/internal_worker.rs | 97 ++++- crates/worker/src/worker.rs | 349 +++++++++++++----- crates/worker/tests/compact_events_test.rs | 117 +++++- .../src/lib/workspace/console/model.test.ts | 46 ++- .../src/lib/workspace/console/model.ts | 5 +- 6 files changed, 616 insertions(+), 133 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 18d72a2d8..e07aef804 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -2459,7 +2459,9 @@ async fn controller_loop( ) .await; } - if let Err(error) = result { + if let Err(error) = result + && !matches!(&error, WorkerError::CompactCancelled) + { let _ = working_event_tx.send(Event::Error { code: worker_error_code(&error), message: error.to_string(), @@ -2883,11 +2885,7 @@ where (WorkerStatus::Paused, shutdown_requested, false) } Err(e) => { - let status = if matches!( - &e, - WorkerError::SegmentActivationIncomplete { .. } - | WorkerError::CompactionCleanupPending { .. } - ) { + let status = if worker_error_requires_attention(&e) { // Metadata has already committed the replacement, but // one live authority failed to publish. Keep the // controller visibly busy/fail-closed; publishing Idle @@ -2896,20 +2894,22 @@ where } else { WorkerStatus::Idle }; - let code = worker_error_code(&e); - let message = e.to_string(); - let _ = working_event_tx.send(Event::Error { - code, - message: message.clone(), - }); - if parent_originated { - crate::ipc::event::fire_and_forget( - parent_socket.cloned(), - protocol::WorkerEvent::Errored { - worker_name: self_name.to_string(), - message, - }, - ); + if !matches!(&e, WorkerError::RunFailureAlreadyDelivered { .. }) { + let code = worker_error_code(&e); + let message = e.to_string(); + let _ = working_event_tx.send(Event::Error { + code, + message: message.clone(), + }); + if parent_originated { + crate::ipc::event::fire_and_forget( + parent_socket.cloned(), + protocol::WorkerEvent::Errored { + worker_name: self_name.to_string(), + message, + }, + ); + } } (status, shutdown_requested, false) } @@ -3375,8 +3375,22 @@ where } } +fn worker_error_requires_attention(error: &WorkerError) -> bool { + match error { + WorkerError::SegmentActivationIncomplete { .. } + | WorkerError::CompactionCleanupPending { .. } => true, + WorkerError::RunFailureAlreadyDelivered { source } + | WorkerError::RunTerminalPersistence { + run_failure: source, + .. + } => worker_error_requires_attention(source), + _ => false, + } +} + fn worker_error_code(e: &WorkerError) -> ErrorCode { match e { + WorkerError::RunFailureAlreadyDelivered { source } => worker_error_code(source), WorkerError::Engine(we) => match we { EngineError::Tool(_) => ErrorCode::ToolError, EngineError::Client(_) => ErrorCode::ProviderError, @@ -3985,6 +3999,87 @@ mod tests { } } + #[tokio::test] + async fn delivered_run_failure_is_not_reported_again_by_controller() { + let mut env = make_env().await; + let listener = UnixListener::bind(&env.parent_socket_path).expect("bind listener"); + let mut events = env.working_event_tx.subscribe(); + let worker_future = async { + Err::(WorkerError::RunFailureAlreadyDelivered { + source: Box::new(WorkerError::CompactSummaryMissing), + }) + }; + + let (status, _, may_drain_pending) = drive_turn( + worker_future, + &mut env.method_rx, + &env.working_event_tx, + &env.cancel_tx, + &env.pause_tx, + &env.shared_state, + &env.runtime_dir, + None, + &env.notify_buffer, + &env.pending_submissions, + Some(&env.parent_socket_path), + "child-worker", + &env.spawned_registry, + true, + ) + .await; + + assert_eq!(status, WorkerStatus::Idle); + assert!(!may_drain_pending, "failed runs must not become successful"); + assert!(events.try_recv().is_err(), "must not emit a generic Error"); + assert!( + tokio::time::timeout(Duration::from_millis(200), listener.accept()) + .await + .is_err(), + "must not send a second parent Errored event" + ); + } + + #[tokio::test] + async fn terminal_persistence_failure_reports_both_causes_once() { + let mut env = make_env().await; + let mut events = env.working_event_tx.subscribe(); + let worker_future = async { + Err::(WorkerError::RunTerminalPersistence { + run_failure: Box::new(WorkerError::CompactSummaryMissing), + persist_error: Box::new(WorkerError::Store(session_store::StoreError::Io( + std::io::Error::other("terminal append failed"), + ))), + }) + }; + + let (status, _, may_drain_pending) = drive_turn( + worker_future, + &mut env.method_rx, + &env.working_event_tx, + &env.cancel_tx, + &env.pause_tx, + &env.shared_state, + &env.runtime_dir, + None, + &env.notify_buffer, + &env.pending_submissions, + None, + "child-worker", + &env.spawned_registry, + false, + ) + .await; + + assert_eq!(status, WorkerStatus::Idle); + assert!(!may_drain_pending); + let Event::Error { message, .. } = events.try_recv().expect("one error event") else { + panic!("expected Error event"); + }; + assert!(message.contains("compact worker did not produce a summary")); + assert!(message.contains("terminal append failed")); + assert!(events.try_recv().is_err(), "must report once"); + } + #[tokio::test] async fn non_parent_originated_worker_error_stays_silent() { let mut env = make_env().await; diff --git a/crates/worker/src/internal_worker.rs b/crates/worker/src/internal_worker.rs index 6167c0050..21ff16e03 100644 --- a/crates/worker/src/internal_worker.rs +++ b/crates/worker/src/internal_worker.rs @@ -271,6 +271,17 @@ pub(crate) enum InternalWorkerSessionStatus { Failed, } +/// Typed result of the most recent reusable Internal Worker turn. +/// +/// Session status intentionally remains a compact presentation state. Owners that +/// make domain decisions (such as Compaction) must use this result instead of +/// guessing a stop cause from `Stopped` or parsing the presentation error string. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum InternalWorkerTurnOutcome { + Completed(WorkerRunResult), + Failed { message: String }, +} + impl InternalWorkerSessionStatus { fn encode(self) -> u8 { match self { @@ -323,26 +334,53 @@ fn send_internal_worker_state( fn classify_internal_turn_result( result: Result, -) -> (InternalWorkerSessionStatus, Option) { +) -> ( + InternalWorkerSessionStatus, + Option, + InternalWorkerTurnOutcome, +) { match result { - Ok(WorkerRunResult::Finished) => (InternalWorkerSessionStatus::Idle, None), - Ok(WorkerRunResult::Paused) => (InternalWorkerSessionStatus::Paused, None), - Ok(WorkerRunResult::LimitReached) => ( - InternalWorkerSessionStatus::Stopped, - Some("internal Worker reached its turn limit".to_string()), + Ok(result @ WorkerRunResult::Finished) => ( + InternalWorkerSessionStatus::Idle, + None, + InternalWorkerTurnOutcome::Completed(result), ), - Ok(WorkerRunResult::Interrupted { message, .. }) => { - (InternalWorkerSessionStatus::Stopped, Some(message)) + Ok(result @ WorkerRunResult::Paused) => ( + InternalWorkerSessionStatus::Paused, + None, + InternalWorkerTurnOutcome::Completed(result), + ), + Ok(result @ WorkerRunResult::LimitReached) => { + let message = "internal Worker reached its turn limit".to_string(); + ( + InternalWorkerSessionStatus::Stopped, + Some(message), + InternalWorkerTurnOutcome::Completed(result), + ) } - Ok(WorkerRunResult::Cancelled) => ( + Ok(WorkerRunResult::Interrupted { code, message }) => ( + InternalWorkerSessionStatus::Stopped, + Some(message.clone()), + InternalWorkerTurnOutcome::Completed(WorkerRunResult::Interrupted { code, message }), + ), + Ok(result @ WorkerRunResult::Cancelled) => ( InternalWorkerSessionStatus::Stopped, Some("internal Worker run was cancelled".to_string()), + InternalWorkerTurnOutcome::Completed(result), ), - Ok(WorkerRunResult::RolledBack) => ( + Ok(result @ WorkerRunResult::RolledBack) => ( InternalWorkerSessionStatus::Stopped, Some("internal Worker run was cancelled before AI output".to_string()), + InternalWorkerTurnOutcome::Completed(result), ), - Err(error) => (InternalWorkerSessionStatus::Failed, Some(error.to_string())), + Err(error) => { + let message = error.to_string(); + ( + InternalWorkerSessionStatus::Failed, + Some(message.clone()), + InternalWorkerTurnOutcome::Failed { message }, + ) + } } } @@ -389,6 +427,7 @@ pub(crate) struct InternalWorkerSessionHandle { event_tx: broadcast::Sender, visibility: InternalWorkerVisibility, last_error: Arc>>, + last_outcome: Arc>>, child_registry: Option>, sink: SegmentLogSink, #[cfg(test)] @@ -488,6 +527,8 @@ impl InternalWorkerSessionHandle { InternalWorkerSessionStatus::Idle => InternalWorkerSessionError::Unavailable, }, )?; + *self.last_error.lock().unwrap() = None; + *self.last_outcome.lock().unwrap() = None; if self .command_tx .send(InternalWorkerSessionCommand::Run(input.into())) @@ -501,6 +542,9 @@ impl InternalWorkerSessionHandle { self.state_changed.notify_waiters(); let message = "internal Worker session actor is unavailable".to_owned(); *self.last_error.lock().unwrap() = Some(message.clone()); + *self.last_outcome.lock().unwrap() = Some(InternalWorkerTurnOutcome::Failed { + message: message.clone(), + }); let _ = self.event_tx.send(Event::Error { code: protocol::ErrorCode::Internal, message, @@ -524,6 +568,15 @@ impl InternalWorkerSessionHandle { } } + /// Wait for the active turn and return its typed outcome. + /// + /// `None` means the actor stopped without publishing a turn result; callers + /// must treat that as unknown rather than manufacturing a cancellation. + pub(crate) async fn wait_for_turn_outcome(&self) -> Option { + self.wait_until_idle().await; + self.last_outcome.lock().unwrap().clone() + } + #[cfg(test)] pub(crate) fn force_status(&self, status: InternalWorkerSessionStatus) { self.status @@ -808,6 +861,7 @@ pub(crate) async fn prepare_internal_worker_session( )); let state_changed = Arc::new(tokio::sync::Notify::new()); let last_error = Arc::new(Mutex::new(None)); + let last_outcome = Arc::new(Mutex::new(None)); let handle = InternalWorkerSessionHandle { command_tx, status: status.clone(), @@ -819,6 +873,7 @@ pub(crate) async fn prepare_internal_worker_session( event_tx: event_tx.clone(), visibility, last_error: last_error.clone(), + last_outcome: last_outcome.clone(), child_registry, sink, #[cfg(test)] @@ -836,8 +891,10 @@ pub(crate) async fn prepare_internal_worker_session( loop { tokio::select! { result = &mut run => { - let (turn_status, error) = classify_internal_turn_result(result); + let (turn_status, error, outcome) = + classify_internal_turn_result(result); actor_in_flight.clear(); + *last_outcome.lock().unwrap() = Some(outcome); status.store(turn_status.encode(), std::sync::atomic::Ordering::Release); if let Some(message) = error { *last_error.lock().unwrap() = Some(message.clone()); @@ -1148,6 +1205,7 @@ pub(crate) fn test_internal_worker_session( event_tx: event_tx.clone(), visibility, last_error: Arc::new(Mutex::new(None)), + last_outcome: Arc::new(Mutex::new(None)), child_registry: None, sink, fail_stop: Arc::new(std::sync::atomic::AtomicBool::new(false)), @@ -1354,6 +1412,11 @@ permission = "write" InternalWorkerSessionStatus::Stopped, true, ), + ( + WorkerRunResult::Cancelled, + InternalWorkerSessionStatus::Stopped, + true, + ), ( WorkerRunResult::RolledBack, InternalWorkerSessionStatus::Stopped, @@ -1362,16 +1425,22 @@ permission = "write" ]; for (result, expected_status, expects_error) in cases { - let (status, error) = classify_internal_turn_result(Ok(result)); + let expected_outcome = InternalWorkerTurnOutcome::Completed(result.clone()); + let (status, error, outcome) = classify_internal_turn_result(Ok(result)); assert_eq!(status, expected_status); assert_eq!(error.is_some(), expects_error); + assert_eq!(outcome, expected_outcome); } - let (status, error) = classify_internal_turn_result(Err(WorkerError::Engine( + let (status, error, outcome) = classify_internal_turn_result(Err(WorkerError::Engine( EngineError::Aborted("fatal".to_string()), ))); assert_eq!(status, InternalWorkerSessionStatus::Failed); assert!(error.is_some_and(|message| message.contains("fatal"))); + assert!(matches!( + outcome, + InternalWorkerTurnOutcome::Failed { message } if message.contains("fatal") + )); } #[tokio::test] diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 4313a184f..18c440419 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -4941,11 +4941,19 @@ impl Worker { if let Some(block) = request_block { let error = automatic_compact_block_error(block); - if matches!(result, EngineRunExit::Yielded) { - self.persist_compaction_run_failure( + let error = if matches!(result, EngineRunExit::Yielded) { + self.terminalize_compaction_run_failure( request_block_message.expect("blocked request has a terminal reason"), - ); - } + error, + ) + } else { + // `persist_turn` above already committed this non-yielded Run's + // compaction terminal. Preserve Err semantics while preventing a + // second Controller Error for the same durable entry. + WorkerError::RunFailureAlreadyDelivered { + source: Box::new(error), + } + }; if let Some(state) = &self.compact_state { state.finish_logical_run(); } @@ -4999,33 +5007,33 @@ impl Worker { } } - fn persist_compaction_run_failure(&mut self, message: String) { - // Keep the original compaction/preparation error as controller authority. - // A terminal append failure is diagnosed separately and cannot publish a - // false persisted result through the session-log sink. - match self.commit_entry(LogEntry::RunErrored { + fn persist_compaction_run_failure(&mut self, message: String) -> Result<(), WorkerError> { + // The Store is the terminal authority. Only a successful append can be + // marked as already delivered to clients through the session-log sink. + self.commit_entry(LogEntry::RunErrored { ts: segment_log::now_millis(), interrupted: false, message, failure: Some(RunFailureKind::Compaction), - }) { - Ok(()) => { - self.last_run_interrupted = false; - self.engine_mut().set_active_run_turn_count(None); - } - Err(persist_error) => { - warn!( - error = %persist_error, - "failed to persist mid-run compaction terminal record" - ); - self.alert( - AlertLevel::Error, - AlertSource::Worker, - format!( - "mid-run compaction failed and its Run terminal record could not be saved: {persist_error}" - ), - ); - } + })?; + self.last_run_interrupted = false; + self.engine_mut().set_active_run_turn_count(None); + Ok(()) + } + + fn terminalize_compaction_run_failure( + &mut self, + message: String, + run_failure: WorkerError, + ) -> WorkerError { + match self.persist_compaction_run_failure(message) { + Ok(()) => WorkerError::RunFailureAlreadyDelivered { + source: Box::new(run_failure), + }, + Err(persist_error) => WorkerError::RunTerminalPersistence { + run_failure: Box::new(run_failure), + persist_error: Box::new(persist_error), + }, } } @@ -5078,9 +5086,7 @@ impl Worker { } let message = format!("mid-run resume preparation failed after compaction: {error}"); - self.persist_compaction_run_failure(message.clone()); - self.alert(AlertLevel::Error, AlertSource::Compactor, message); - return Err(error); + return Err(self.terminalize_compaction_run_failure(message, error)); } self.resume_prepared(RunResumeSource::Compaction).await } @@ -5102,14 +5108,11 @@ impl Worker { } // The yielded logical Run is no longer resumable. Commit its - // durable failure before any log-derived live event can claim - // the terminal transition. If storage itself is unavailable, - // the helper retains the original compaction error (and its - // fail-closed controller classification) while diagnosing the - // missing terminal record separately. - self.persist_compaction_run_failure(message.clone()); - self.alert(AlertLevel::Error, AlertSource::Compactor, message); - Err(e) + // durable failure before returning a typed handled error. The + // Controller suppresses only that exact handled variant; if + // persistence fails it instead reports both failures through + // the non-durable error channel. + Err(self.terminalize_compaction_run_failure(message, e)) } } }) @@ -5951,25 +5954,7 @@ impl Worker { // outer compaction boundary performs the single stop attempt. return Err(WorkerError::InvalidState(error.to_string())); } - match handle.wait_until_idle().await { - crate::internal_worker::InternalWorkerSessionStatus::Idle => {} - crate::internal_worker::InternalWorkerSessionStatus::Stopped => { - return Err(WorkerError::CompactCancelled); - } - crate::internal_worker::InternalWorkerSessionStatus::Failed => { - return Err(WorkerError::InvalidState( - handle - .protocol_snapshot() - .error - .unwrap_or_else(|| "compactor Internal Worker failed".into()), - )); - } - status => { - return Err(WorkerError::InvalidState(format!( - "compactor Internal Worker ended in {status:?}" - ))); - } - } + ensure_compaction_internal_turn_succeeded(handle.wait_for_turn_outcome().await)?; // Guard: nudge the worker once more if the expected outputs // (summary, and any auto-read nominations when default refs @@ -6003,20 +5988,7 @@ impl Worker { .send(prompt) .await .map_err(|error| WorkerError::InvalidState(error.to_string()))?; - match handle.wait_until_idle().await { - crate::internal_worker::InternalWorkerSessionStatus::Idle => {} - crate::internal_worker::InternalWorkerSessionStatus::Stopped => { - return Err(WorkerError::CompactCancelled); - } - _ => { - return Err(WorkerError::InvalidState( - handle - .protocol_snapshot() - .error - .unwrap_or_else(|| "compactor Internal Worker failed".into()), - )); - } - } + ensure_compaction_internal_turn_succeeded(handle.wait_for_turn_outcome().await)?; } let mut final_ctx = ctx.lock().expect("compact ctx poisoned").clone(); @@ -6035,20 +6007,7 @@ impl Worker { .send(prompt) .await .map_err(|error| WorkerError::InvalidState(error.to_string()))?; - match handle.wait_until_idle().await { - crate::internal_worker::InternalWorkerSessionStatus::Idle => {} - crate::internal_worker::InternalWorkerSessionStatus::Stopped => { - return Err(WorkerError::CompactCancelled); - } - _ => { - return Err(WorkerError::InvalidState( - handle - .protocol_snapshot() - .error - .unwrap_or_else(|| "compactor Internal Worker failed".into()), - )); - } - } + ensure_compaction_internal_turn_succeeded(handle.wait_for_turn_outcome().await)?; final_ctx = ctx.lock().expect("compact ctx poisoned").clone(); summary_text = final_ctx .summary @@ -7904,6 +7863,37 @@ fn restored_flow_runtime_state( .transpose() } +fn ensure_compaction_internal_turn_succeeded( + outcome: Option, +) -> Result<(), WorkerError> { + use crate::internal_worker::InternalWorkerTurnOutcome; + + match outcome { + Some(InternalWorkerTurnOutcome::Completed(WorkerRunResult::Finished)) => Ok(()), + Some(InternalWorkerTurnOutcome::Completed(WorkerRunResult::LimitReached)) => { + Err(WorkerError::Engine(EngineError::Aborted( + "compactor Internal Worker reached its turn limit".to_string(), + ))) + } + Some(InternalWorkerTurnOutcome::Completed(WorkerRunResult::Interrupted { + message, + .. + })) => Err(WorkerError::Engine(EngineError::Aborted(message))), + Some(InternalWorkerTurnOutcome::Completed( + WorkerRunResult::Cancelled | WorkerRunResult::RolledBack, + )) => Err(WorkerError::CompactCancelled), + Some(InternalWorkerTurnOutcome::Completed(WorkerRunResult::Paused)) => Err( + WorkerError::InvalidState("compactor Internal Worker paused unexpectedly".to_string()), + ), + Some(InternalWorkerTurnOutcome::Failed { message }) => { + Err(WorkerError::InvalidState(message)) + } + None => Err(WorkerError::InvalidState( + "compactor Internal Worker stopped without a typed turn outcome".to_string(), + )), + } +} + fn automatic_compact_block_error(block: AutomaticCompactBlock) -> WorkerError { match block { AutomaticCompactBlock::Thrash => WorkerError::CompactThrash, @@ -8019,6 +8009,26 @@ pub enum WorkerError { #[error("compaction was cancelled")] CompactCancelled, + /// The logical Run failed, and its terminal entry was already committed and + /// published through the session-log authority. Controllers must preserve the + /// error outcome without emitting a second generic error event. + #[error("{source}")] + RunFailureAlreadyDelivered { + #[source] + source: Box, + }, + + /// The logical Run failed and its terminal entry could not be committed. + /// This remains unhandled so the Controller reports the run and persistence + /// failures together through its non-durable error channel. + #[error( + "run failed ({run_failure}) and its terminal record could not be persisted: {persist_error}" + )] + RunTerminalPersistence { + run_failure: Box, + persist_error: Box, + }, + #[error("compact summary too large: {tokens} tokens exceeds max {max}")] CompactSummaryTooLarge { tokens: u64, max: u64 }, @@ -9104,6 +9114,83 @@ mod build_summary_prompt_tests { } } + #[derive(Clone)] + struct FailRunTerminalStore { + inner: session_store::FsStore, + fail_run_terminal: Arc, + } + + impl Store for FailRunTerminalStore { + fn append( + &self, + session_id: SessionId, + segment_id: SegmentId, + entry: &LogEntry, + ) -> Result<(), StoreError> { + if matches!(entry, LogEntry::RunErrored { .. }) + && self.fail_run_terminal.swap(false, Ordering::SeqCst) + { + return Err(StoreError::Io(std::io::Error::other( + "synthetic terminal append failure", + ))); + } + self.inner.append(session_id, segment_id, entry) + } + + fn read_all( + &self, + session_id: SessionId, + segment_id: SegmentId, + ) -> Result, StoreError> { + self.inner.read_all(session_id, segment_id) + } + + fn list_sessions(&self) -> Result, StoreError> { + self.inner.list_sessions() + } + + fn list_segments(&self, session_id: SessionId) -> Result, StoreError> { + self.inner.list_segments(session_id) + } + + fn lookup_session_of( + &self, + segment_id: SegmentId, + ) -> Result, StoreError> { + self.inner.lookup_session_of(segment_id) + } + + fn create_segment( + &self, + session_id: SessionId, + segment_id: SegmentId, + entries: &[LogEntry], + ) -> Result<(), StoreError> { + self.inner.create_segment(session_id, segment_id, entries) + } + + fn exists(&self, session_id: SessionId, segment_id: SegmentId) -> Result { + self.inner.exists(session_id, segment_id) + } + + fn read_entry_count( + &self, + session_id: SessionId, + segment_id: SegmentId, + ) -> Result { + self.inner.read_entry_count(session_id, segment_id) + } + + fn append_trace( + &self, + session_id: SessionId, + segment_id: SegmentId, + entry: &session_store::TraceEntry, + ) -> Result<(), StoreError> { + self.inner.append_trace(session_id, segment_id, entry) + } + } + #[derive(Clone)] struct NoopClient; @@ -9314,9 +9401,9 @@ mod build_summary_prompt_tests { .unwrap(); worker.last_run_interrupted = true; - worker.persist_compaction_run_failure( - "mid-run compaction failed: summary unavailable".into(), - ); + worker + .persist_compaction_run_failure("mid-run compaction failed: summary unavailable".into()) + .unwrap(); let entries = store .read_all(worker.session_id(), worker.segment_id()) @@ -9348,6 +9435,92 @@ mod build_summary_prompt_tests { assert_eq!(worker.engine().active_run_turn_count(), None); } + #[tokio::test] + async fn failed_compaction_terminal_append_stays_unhandled_and_unpublished() { + let dir = tempfile::tempdir().unwrap(); + let store = FailRunTerminalStore { + inner: session_store::FsStore::new(dir.path().join("sessions")).unwrap(), + fail_run_terminal: Arc::new(AtomicBool::new(false)), + }; + let mut worker = Worker::new( + minimal_manifest(), + Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient), + store.clone(), + WorkerWorkspaceContext::no_workspace(), + WorkerFilesystemAuthority::None, + Scope::empty(), + ) + .await + .unwrap(); + worker.ensure_segment_head().await.unwrap(); + store.fail_run_terminal.store(true, Ordering::SeqCst); + + let error = worker.terminalize_compaction_run_failure( + "mid-run compaction failed: summary unavailable".into(), + WorkerError::CompactSummaryMissing, + ); + + assert!(matches!( + error, + WorkerError::RunTerminalPersistence { + run_failure, + persist_error, + } if matches!(*run_failure, WorkerError::CompactSummaryMissing) + && persist_error.to_string().contains("synthetic terminal append failure") + )); + assert!( + store + .read_all(worker.session_id(), worker.segment_id()) + .unwrap() + .iter() + .all(|entry| !matches!(entry, LogEntry::RunErrored { .. })) + ); + assert!( + worker + .sink() + .subscribe_with_snapshot() + .0 + .iter() + .all(|entry| !matches!(entry, LogEntry::RunErrored { .. })), + "a failed append must not be published as durable" + ); + } + + #[test] + fn compaction_internal_turn_preserves_typed_stop_causes() { + use crate::internal_worker::InternalWorkerTurnOutcome; + + let limit = ensure_compaction_internal_turn_succeeded(Some( + InternalWorkerTurnOutcome::Completed(WorkerRunResult::LimitReached), + )) + .unwrap_err(); + assert!( + matches!(limit, WorkerError::Engine(EngineError::Aborted(message)) if message.contains("turn limit")) + ); + + let interrupted = ensure_compaction_internal_turn_succeeded(Some( + InternalWorkerTurnOutcome::Completed(WorkerRunResult::Interrupted { + code: ErrorCode::ProviderError, + message: "summary provider failed".to_string(), + }), + )) + .unwrap_err(); + assert!( + matches!(interrupted, WorkerError::Engine(EngineError::Aborted(message)) if message == "summary provider failed") + ); + + let cancelled = ensure_compaction_internal_turn_succeeded(Some( + InternalWorkerTurnOutcome::Completed(WorkerRunResult::Cancelled), + )) + .unwrap_err(); + assert!(matches!(cancelled, WorkerError::CompactCancelled)); + + let unknown = ensure_compaction_internal_turn_succeeded(None).unwrap_err(); + assert!( + matches!(unknown, WorkerError::InvalidState(message) if message.contains("without a typed turn outcome")) + ); + } + #[tokio::test] async fn intentional_cancel_commits_a_non_failure_terminal() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index 3785e800d..68842c5c9 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -991,12 +991,16 @@ async fn compact_failure_and_cancellation_emit_bounded_categories() { let encoded = serde_json::to_string(&failure.metric).unwrap(); assert!(!encoded.contains("missing summary")); - let (_cancel_tx, cancel_rx) = tokio::sync::watch::channel(true); + let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false); + cancel_tx.send(true).unwrap(); let error = worker .manual_compact_with_cancel(cancel_rx) .await .unwrap_err(); - assert!(matches!(error, worker::WorkerError::CompactCancelled)); + assert!( + matches!(error, worker::WorkerError::CompactCancelled), + "unexpected cancellation error: {error:?}" + ); let metrics = session_metrics::read_session_metrics(worker.store(), session_id).unwrap(); let cancelled = metrics .iter() @@ -1503,7 +1507,14 @@ async fn compacted_context_above_request_threshold_fails_before_provider_request .await .expect_err("unsafe compacted context must fail closed"); - assert!(matches!(error, WorkerError::CompactThrash)); + assert!( + matches!( + error, + WorkerError::RunFailureAlreadyDelivered { ref source } + if matches!(**source, WorkerError::CompactThrash) + ), + "unexpected error: {error:?}" + ); assert_eq!( call_count.load(Ordering::SeqCst), 3, @@ -1526,6 +1537,100 @@ async fn compacted_context_above_request_threshold_fails_before_provider_request })); } +#[tokio::test] +async fn controller_does_not_repeat_a_persisted_mid_run_compaction_failure() { + let client = MockClient::new(vec![ + grow_tool_use_events("grow-before-compaction", 1_000), + single_text_events("missing summary"), + single_text_events("still missing summary"), + ]); + let (worker, store) = make_worker_with_manifest_and_store(MID_TURN_MANIFEST_TOML, client).await; + let session_id = worker.session_id(); + let runtime_tmp = tempfile::tempdir().unwrap(); + let bash_output_dir = runtime_tmp.path().join("bash-output"); + let (handle, shutdown_receiver) = + WorkerController::spawn(worker, runtime_tmp.path(), &bash_output_dir) + .await + .unwrap(); + let mut rx = handle.subscribe(); + + handle + .send(Method::submit_text( + protocol::new_submission_request_id(), + "trigger failing compaction", + )) + .await + .unwrap(); + let mut observed = Vec::new(); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + let event = rx.recv().await.expect("event"); + let idle = matches!( + event, + Event::WorkerState { + snapshot: protocol::WorkerStateSnapshot { + state: protocol::WorkerState::Idle, + .. + } + } + ); + observed.push(event); + if idle { + break; + } + } + }) + .await + .expect("timeout waiting for failed compaction and Idle"); + tokio::task::yield_now().await; + while let Ok(event) = rx.try_recv() { + observed.push(event); + } + + let delivered = handle + .committed_entries() + .into_iter() + .filter(|entry| matches!(entry, LogEntry::RunErrored { .. })) + .count(); + assert_eq!(delivered, 1, "the durable terminal is the only delivery"); + assert!( + observed + .iter() + .all(|event| !matches!(event, Event::Error { .. })), + "Controller must not add a generic Error after durable delivery" + ); + assert!( + observed.iter().all(|event| !matches!( + event, + Event::Alert(protocol::Alert { + level: protocol::AlertLevel::Error, + .. + }) + )), + "mid-run terminal failure must not add an Error Alert" + ); + + let terminal_count = store + .list_segments(session_id) + .unwrap() + .into_iter() + .flat_map(|segment_id| store.read_all(session_id, segment_id).unwrap()) + .filter(|entry| matches!(entry, LogEntry::RunErrored { .. })) + .count(); + assert_eq!(terminal_count, 1); + + handle + .send(Method::Shutdown { + command: protocol::WorkerCommandEnvelope::new(1), + }) + .await + .unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(2), shutdown_receiver) + .await + .expect("controller shutdown timeout") + .expect("shutdown confirmation"); +} + #[tokio::test] async fn pre_run_compact_failure_clears_runtime_progress() { // Only the first run has a response. Compaction will run the @@ -1621,6 +1726,7 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() { .expect("send compact cancel"); let mut saw_interrupted = false; let mut saw_idle = false; + let mut saw_cancel_error = false; while !(saw_interrupted && saw_idle) { match tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await @@ -1630,6 +1736,7 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() { Event::CompactionProgress { compaction: None } => { saw_interrupted = true; } + Event::Error { .. } => saw_cancel_error = true, Event::WorkerState { snapshot } if snapshot.catalog_status() == protocol::WorkerStatus::Idle => { @@ -1642,6 +1749,10 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() { _ => {} } } + assert!( + !saw_cancel_error, + "intentional manual cancellation must not be reported as an Error" + ); let compact = protocol::WorkerCommandEnvelope::new(3); handle diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index c9c938adb..b831db902 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -338,6 +338,13 @@ Deno.test("durable compaction failure reconciles live alert and error in both vi }, }, }, + { + eventId: "controller-error", + event: { + event: "error", + data: { code: "internal", message: "summary unavailable" }, + }, + }, { eventId: "failed", event: transition("failed-entry", { @@ -346,13 +353,6 @@ Deno.test("durable compaction failure reconciles live alert and error in both vi failure: "compaction", }), }, - { - eventId: "controller-error", - event: { - event: "error", - data: { code: "internal", message: "summary unavailable" }, - }, - }, ]); const errors = projection.lines.filter((line) => line.kind === "error"); @@ -362,6 +362,18 @@ Deno.test("durable compaction failure reconciles live alert and error in both vi assertEquals(errors[0].entryId, "failed-entry"); assertEquals(projectConsoleLines(projection.lines, "normal"), errors); assertEquals(projectConsoleLines(projection.lines, "overview"), errors); + + const history = projectSessionHistoryEntries([{ + entry_id: "failed-entry", + timestamp: 10, + provenance: "legacy_unknown", + kind: "run_error", + message, + failure: "compaction", + }], null); + const merged = mergeCommittedHistoryLines(history, projection.lines); + assertEquals(merged.filter((line) => line.kind === "error").length, 1); + assertEquals(merged.find((line) => line.kind === "error")?.entryId, "failed-entry"); }); Deno.test("snapshot restores compaction terminal without reviving transition progress", () => { @@ -3232,6 +3244,26 @@ Deno.test("committed history reconciles overlapping Read groups before counting } }); +Deno.test("committed history keeps identical run failures with distinct entry identities", () => { + const message = "the same failure happened again"; + const runError = (entryId: string): SessionSnapshotEntry => ({ + kind: "run_error", + entry_id: entryId, + timestamp: 1, + provenance: "legacy_unknown", + message, + failure: "compaction", + }); + const retained = projectSessionHistoryEntries([runError("run-error-1")], null); + const current = projectSessionHistoryEntries([runError("run-error-2")], null); + + const merged = mergeCommittedHistoryLines(retained, current); + assertEquals(merged.map((line) => line.entryId), ["run-error-1", "run-error-2"]); + assertEquals(merged.map((line) => line.body), [message, message]); + assertEquals(projectConsoleLines(merged, "normal").length, 2); + assertEquals(projectConsoleLines(merged, "overview").length, 2); +}); + Deno.test("committed history matches tool calls across live and durable result identities", () => { const retained: ConsoleLine = { ...consoleLine("tool-call-a", "tool"), entryId: "call-entry-a", diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index 4763ada3e..1471417aa 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -2208,7 +2208,10 @@ function applySessionEntry( const entryId = stringField(value, "entry_id"); if (entryId) { for ( - let index = Math.max(0, lineCountBefore - 1); + let index = Math.max( + 0, + Math.min(lineCountBefore, projection.lines.length - 1), + ); index < projection.lines.length; index += 1 ) {