Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

Merge remote-tracking branch 'origin/develop' into work/T-668-subjektiv-memory-lifecycle

Browse Source
This commit is contained in:
Hare
2026-10-02 06:00:49 +09:00
parent 6091fafcac d10783c0d6
commit 9a7bfba507
53 changed files with 5108 additions and 892 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+189 -20
View File
@@ -65,6 +65,14 @@ pub struct WorkerHandle {
impl WorkerHandle {
pub async fn send(&self, method: Method) -> Result<(), mpsc::error::SendError<Method>> {
// Reject known-busy sends before channel dispatch, including maintenance
// that cannot receive methods until it finishes. The controller checks
// again at admission to cover an Idle -> busy race after this snapshot.
if self.shared_state.snapshot().state != WorkerState::Idle
&& reject_idle_only_submission(&method, &self.working_event_tx)
{
return Ok(());
}
self.method_tx.send(method).await
}
@@ -1551,6 +1559,21 @@ where
}
}
fn reject_idle_only_submission(method: &Method, events: &broadcast::Sender<Event>) -> bool {
let Method::SubmitIfIdle {
submission_request_id,
..
} = method
else {
return false;
};
let _ = events.send(Event::SubmissionRejected {
submission_request_id: submission_request_id.clone(),
message: "WorkerSendInput requires an Idle Worker and never queues input. Use WorkerNotify for information about work already in progress.".to_string(),
});
true
}
fn durably_accept_method_while_busy<St>(
method: Method,
pending_submissions: &PendingSubmissionHandle<St>,
@@ -1559,6 +1582,9 @@ fn durably_accept_method_while_busy<St>(
where
St: session_store::Store + Clone,
{
if reject_idle_only_submission(&method, working_event_tx) {
return None;
}
match method {
Method::Submit {
submission_request_id,
@@ -1668,6 +1694,10 @@ fn reject_method_while_attention_locked(
| Method::SubmitTracked {
submission_request_id,
..
}
| Method::SubmitIfIdle {
submission_request_id,
..
} => {
let _ = working_event_tx.send(Event::SubmissionRejected {
submission_request_id: submission_request_id.clone(),
@@ -1974,6 +2004,12 @@ async fn controller_loop<C, St>(
}
};
if shared_state.snapshot().state != WorkerState::Idle
&& reject_idle_only_submission(&method, &working_event_tx)
{
continue;
}
if worker.has_pending_compaction_cleanup() {
match worker.retry_pending_compaction_cleanup().await {
Ok(()) => {
@@ -2060,6 +2096,11 @@ async fn controller_loop<C, St>(
submission_request_id,
input,
source,
}
| Method::SubmitIfIdle {
submission_request_id,
input,
source,
} => {
let request_id = submission_request_id.clone();
let (source_namespace, provenance) =
@@ -2446,7 +2487,9 @@ async fn controller_loop<C, St>(
)
.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(),
@@ -2870,11 +2913,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
@@ -2883,20 +2922,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)
}
@@ -3045,6 +3086,9 @@ where
}
}
}
Some(method @ Method::SubmitIfIdle { .. }) => {
reject_idle_only_submission(&method, working_event_tx);
}
Some(Method::SubmitTracked {
submission_request_id,
input,
@@ -3359,8 +3403,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,
@@ -3586,6 +3644,36 @@ mod tests {
tokio::time::timeout(timeout, accept).await.ok().flatten()
}
#[tokio::test]
async fn idle_only_input_is_rejected_in_busy_maintenance_admission() {
let env = make_env().await;
let mut events = env.working_event_tx.subscribe();
let method = Method::SubmitIfIdle {
submission_request_id: "maintenance-input".to_string(),
input: vec![protocol::Segment::text("must not queue")],
source: protocol::AuthenticatedInputSource::Backend {
operation_id: "maintenance-input".to_string(),
},
};
assert!(
durably_accept_method_while_busy(
method,
&env.pending_submissions,
&env.working_event_tx
)
.is_none()
);
let event = events.try_recv().unwrap();
assert!(
matches!(event, Event::SubmissionRejected { submission_request_id, message }
if submission_request_id == "maintenance-input" && message.contains("WorkerNotify"))
);
assert!(
events.try_recv().is_err(),
"must not emit durable acceptance or queue changes"
);
}
#[tokio::test]
async fn parent_originated_finished_fires_turn_ended() {
let mut env = make_env().await;
@@ -3939,6 +4027,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::<WorkerRunResult, _>(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::<WorkerRunResult, _>(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;
crates/worker/src/feature/builtin/manage_worker.rs
+17 -6
View File
@@ -245,11 +245,11 @@ impl WorkerControlService for WorkspaceWorkerControlService {
"unknown Worker or permission not granted".to_string(),
)
})?;
record
.session
.send(content)
.await
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
record.session.send(content).await.map_err(|error| {
WorkspaceClientError::Request(format!(
"WorkerSendInput requires an Idle SubWorker and never queues input: {error}"
))
})?;
Ok(WorkspaceResponse {
status: 200,
body: serde_json::json!({
@@ -855,7 +855,7 @@ impl WorkerOperation {
"Spawn a Backend/Runtime Worker session in one existing Workspace Workdir. The model-facing input remains singular and the shared lifecycle wraps it as the canonical attachment collection under the stable `workdir` alias. The Workdir id is authority; filesystem paths and Runtime URLs are not accepted. `initial_submit` carries the normal typed user submission. After the Orchestrator has committed a Ticket to `inprogress`, set `ticket_id` with a Flow segment in `initial_submit` to atomically assign the new Coder Worker; the operation id is derived from the durable tool call rather than model input."
}
Self::SendInput => {
"Send user input to a known Runtime Worker. Idle Workers can start a fresh turn; Running or Paused Workers durably queue it for a later turn. Use WorkerNotify for advisory information during work already in progress. Direct SubWorkers require Idle."
"Start a fresh turn for an Idle Worker. Non-Idle Workers reject WorkerSendInput without queueing. Use WorkerNotify for advisory information during work already in progress. Direct SubWorkers also require Idle."
}
Self::Notify => {
"Send an advisory notification to a known Runtime Worker so information can be incorporated into work already in progress without creating a queued Submit."
@@ -1536,6 +1536,17 @@ mod tests {
assert_eq!(client.requests.lock().unwrap().len(), 1);
}
#[test]
fn send_input_rejection_is_a_tool_error_with_notify_guidance() {
let error = tool_output(WorkerOperation::SendInput, WorkspaceResponse {
status: 400,
body: "WorkerSendInput requires an Idle Worker. Use WorkerNotify for work already in progress.".to_string(),
}).unwrap_err();
assert!(matches!(error, ToolError::ExecutionFailed(_)));
assert!(error.to_string().contains("WorkerNotify"));
assert!(!error.to_string().contains("completed"));
}
#[tokio::test]
async fn control_service_routes_by_subject_without_cross_provider_fallback() {
let client = Arc::new(RecordingWorkspaceClient::default());
crates/worker/src/internal_worker.rs
+83 -14
View File
@@ -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<WorkerRunResult, WorkerError>,
) -> (InternalWorkerSessionStatus, Option<String>) {
) -> (
InternalWorkerSessionStatus,
Option<String>,
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<Event>,
visibility: InternalWorkerVisibility,
last_error: Arc<Mutex<Option<String>>>,
last_outcome: Arc<Mutex<Option<InternalWorkerTurnOutcome>>>,
child_registry: Option<Arc<SpawnedWorkerRegistry>>,
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<InternalWorkerTurnOutcome> {
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]
crates/worker/src/prompt/catalog.rs
+2 -2
View File
@@ -937,8 +937,8 @@ mod tests {
"missing canonical tool {name}"
);
}
assert!(orchestration.contains("durably queue it for a later turn"));
assert!(orchestration.contains("Direct SubWorkers still require Idle"));
assert!(orchestration.contains("requires an Idle Worker and never queues input"));
assert!(orchestration.contains("Direct SubWorkers also require Idle"));
assert!(orchestration.contains("use `WorkerNotify` instead"));
assert!(orchestration.contains("do not create a queued Submit"));
for alias in ["SubWorkerList", "SubWorkerSend", "SubWorkerStop"] {
crates/worker/src/worker.rs
+265 -88
View File
@@ -4949,11 +4949,19 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
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();
}
@@ -5007,33 +5015,33 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
}
}
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),
},
}
}
@@ -5086,9 +5094,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
}
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
}
@@ -5110,14 +5116,11 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
}
// 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))
}
}
})
@@ -5296,6 +5299,10 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
info!(new_segment_id = %new_segment_id, "Manual compaction succeeded");
Ok(ManualCompactResult::Compacted { new_segment_id })
}
Err(WorkerError::CompactCancelled) => {
info!("Manual compaction was cancelled");
Err(WorkerError::CompactCancelled)
}
Err(e) => {
warn!(error = %e, "Manual compaction failed");
self.alert(
@@ -5959,25 +5966,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// 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
@@ -6011,20 +6000,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.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();
@@ -6043,20 +6019,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.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
@@ -7912,6 +7875,37 @@ fn restored_flow_runtime_state(
.transpose()
}
fn ensure_compaction_internal_turn_succeeded(
outcome: Option<crate::internal_worker::InternalWorkerTurnOutcome>,
) -> 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,
@@ -8027,6 +8021,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<WorkerError>,
},
/// 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<WorkerError>,
persist_error: Box<WorkerError>,
},
#[error("compact summary too large: {tokens} tokens exceeds max {max}")]
CompactSummaryTooLarge { tokens: u64, max: u64 },
@@ -9112,6 +9126,83 @@ mod build_summary_prompt_tests {
}
}
#[derive(Clone)]
struct FailRunTerminalStore {
inner: session_store::FsStore,
fail_run_terminal: Arc<AtomicBool>,
}
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<Vec<LogEntry>, StoreError> {
self.inner.read_all(session_id, segment_id)
}
fn list_sessions(&self) -> Result<Vec<SessionId>, StoreError> {
self.inner.list_sessions()
}
fn list_segments(&self, session_id: SessionId) -> Result<Vec<SegmentId>, StoreError> {
self.inner.list_segments(session_id)
}
fn lookup_session_of(
&self,
segment_id: SegmentId,
) -> Result<Option<SessionId>, 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<bool, StoreError> {
self.inner.exists(session_id, segment_id)
}
fn read_entry_count(
&self,
session_id: SessionId,
segment_id: SegmentId,
) -> Result<usize, StoreError> {
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;
@@ -9322,9 +9413,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())
@@ -9356,6 +9447,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();
crates/worker/tests/compact_events_test.rs
+118 -3
View File
@@ -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,11 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() {
Event::CompactionProgress { compaction: None } => {
saw_interrupted = true;
}
Event::Error { .. }
| Event::Alert(protocol::Alert {
level: protocol::AlertLevel::Error,
..
}) => saw_cancel_error = true,
Event::WorkerState { snapshot }
if snapshot.catalog_status() == protocol::WorkerStatus::Idle =>
{
@@ -1642,6 +1753,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
crates/worker/tests/controller_test.rs
+91
View File
@@ -2895,6 +2895,97 @@ async fn drain_until<F: FnMut(&Event) -> bool>(
}
}
fn idle_only_input(request_id: &str) -> Method {
Method::SubmitIfIdle {
submission_request_id: request_id.to_string(),
input: vec![protocol::Segment::text(request_id)],
source: protocol::AuthenticatedInputSource::Backend {
operation_id: request_id.to_string(),
},
}
}
#[tokio::test]
async fn idle_only_submit_rejects_admission_race_and_paused_without_persisting() {
let client = MockClient::sequential(vec![MockResponse::Hang(vec![
LlmEvent::text_block_start(0),
LlmEvent::text_delta(0, "partial"),
])]);
let handle = spawn_controller(make_worker(client.clone()).await).await;
let mut rx = handle.subscribe();
assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Idle);
// Both callers can observe Idle. Admission of the first request must make
// the second fail rather than silently scheduling a later turn.
handle.send(idle_only_input("first-idle")).await.unwrap();
handle.send(idle_only_input("racing-input")).await.unwrap();
let mut accepted_first = false;
assert!(
drain_until(&mut rx, std::time::Duration::from_secs(2), |event| {
match event {
Event::SubmissionAccepted {
submission_request_id,
..
} => {
assert_eq!(submission_request_id, "first-idle");
accepted_first = true;
false
}
Event::SubmissionRejected {
submission_request_id,
message,
} => {
assert_eq!(submission_request_id, "racing-input");
assert!(message.contains("WorkerNotify"));
true
}
_ => false,
}
})
.await
);
assert!(accepted_first);
wait_for_status(&handle, WorkerStatus::Running).await;
handle
.send(Method::Pause {
command: worker_command(&handle),
})
.await
.unwrap();
wait_for_status(&handle, WorkerStatus::Paused).await;
handle.send(idle_only_input("paused-input")).await.unwrap();
assert!(
drain_until(&mut rx, std::time::Duration::from_secs(2), |event| {
matches!(event, Event::SubmissionRejected { submission_request_id, message }
if submission_request_id == "paused-input" && message.contains("WorkerNotify"))
})
.await
);
assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Paused);
handle.send(Method::ListPendingSubmissions).await.unwrap();
assert!(
drain_until(&mut rx, std::time::Duration::from_secs(2), |event| {
if let Event::PendingSubmissionsChanged { pending } = event {
assert!(pending.submissions.is_empty());
true
} else {
false
}
})
.await
);
assert_eq!(client.captured_requests().len(), 1);
let entries = serde_json::to_string(&handle.committed_entries()).unwrap();
assert!(!entries.contains("racing-input"));
assert!(!entries.contains("paused-input"));
handle
.send(Method::Cancel {
command: worker_command(&handle),
})
.await
.unwrap();
wait_for_status(&handle, WorkerStatus::Idle).await;
}
/// Paused → Running → Idle. A notification accepted while Paused must not
/// auto-resume the Worker, and explicit Resume must inject it while preserving
/// the interrupted turn's history consistency.