worker: fence controller during repair

crates/worker/src/controller.rs

@@ -1489,6 +1489,44 @@ where
     Ok(workdir_for_view)
 }
 
+fn reject_method_while_attention_locked(
+    method: &Method,
+    working_event_tx: &broadcast::Sender<Event>,
+    message: &str,
+) -> bool {
+    match method {
+        Method::Shutdown { .. } | Method::ListPendingSubmissions => false,
+        Method::Submit {
+            submission_request_id,
+            ..
+        }
+        | Method::SubmitTracked {
+            submission_request_id,
+            ..
+        } => {
+            let _ = working_event_tx.send(Event::SubmissionRejected {
+                submission_request_id: submission_request_id.clone(),
+                message: message.to_owned(),
+            });
+            true
+        }
+        Method::Notify { .. } | Method::NotifyTracked { .. } => {
+            let _ = working_event_tx.send(Event::Error {
+                code: ErrorCode::InvalidRequest,
+                message: message.to_owned(),
+            });
+            true
+        }
+        _ => {
+            let _ = working_event_tx.send(Event::Error {
+                code: ErrorCode::InvalidRequest,
+                message: message.to_owned(),
+            });
+            true
+        }
+    }
+}
+
 /// Idle/Paused event loop. Each iteration either fires a staged
 /// `PendingRun` (delegating to [`drive_turn`] for the Running phase) or
 /// waits for the next `Method`. Method handlers stop at "update state +
@@ -1551,6 +1589,12 @@ async fn controller_loop<C, St>(
     let mut deferred_methods = VecDeque::new();
 
     'controller: loop {
+        if worker.has_failed_segment_activation() || worker.has_pending_compaction_cleanup() {
+            // A queued activation may have been accepted while the failing turn
+            // was still running. Preserve its durable queue entry, but never
+            // start it while repair/cleanup attention is required.
+            pending = None;
+        }
         // here so the status flip → drive_turn → finish sequence lives
         // in one place, regardless of which Method caused it.
         if let Some(run) = pending.take() {
@@ -1700,7 +1744,15 @@ async fn controller_loop<C, St>(
             tokio::select! {
                 method = method_rx.recv() => match method {
                     Some(method) => method,
-                    None => break,
+                    None => {
+                        while worker.has_pending_compaction_cleanup() {
+                            if let Err(error) = worker.retry_pending_compaction_cleanup().await {
+                                tracing::warn!(error = %error, "shutdown waiting for compaction cleanup");
+                                tokio::time::sleep(Duration::from_millis(50)).await;
+                            }
+                        }
+                        break 'controller;
+                    }
                 },
                 _ = notify_buffer.notified() => continue,
                 _ = tokio::time::sleep(remaining) => {
@@ -1726,12 +1778,63 @@ async fn controller_loop<C, St>(
             tokio::select! {
                 method = method_rx.recv() => match method {
                     Some(method) => method,
-                    None => break,
+                    None => {
+                        while worker.has_pending_compaction_cleanup() {
+                            if let Err(error) = worker.retry_pending_compaction_cleanup().await {
+                                tracing::warn!(error = %error, "shutdown waiting for compaction cleanup");
+                                tokio::time::sleep(Duration::from_millis(50)).await;
+                            }
+                        }
+                        break 'controller;
+                    }
                 },
                 _ = notify_buffer.notified() => continue,
             }
         };
 
+        if worker.has_pending_compaction_cleanup() {
+            match worker.retry_pending_compaction_cleanup().await {
+                Ok(()) => {
+                    set_controller_state(
+                        &shared_state,
+                        &runtime_dir,
+                        &working_event_tx,
+                        WorkerState::Idle,
+                    )
+                    .await;
+                }
+                Err(error) => {
+                    let message = error.to_string();
+                    let _ = working_event_tx.send(Event::Error {
+                        code: worker_error_code(&error),
+                        message: message.clone(),
+                    });
+                    if reject_method_while_attention_locked(
+                        &method,
+                        &working_event_tx,
+                        &message,
+                    ) {
+                        continue;
+                    }
+                    // Shutdown is an explicit cleanup barrier: it is not allowed
+                    // to complete while the service remains unconfirmed.
+                    if matches!(&method, Method::Shutdown { .. }) {
+                        continue;
+                    }
+                }
+            }
+        }
+
+        if worker.has_failed_segment_activation()
+            && reject_method_while_attention_locked(
+                &method,
+                &working_event_tx,
+                "Segment activation is incomplete; restore or shut down the Worker",
+            )
+        {
+            continue;
+        }
+
         match method {
             Method::Submit {
                 submission_request_id,
@@ -2165,6 +2268,15 @@ async fn controller_loop<C, St>(
                     });
                 }
                 if shutdown_after_compaction {
+                    while worker.has_pending_compaction_cleanup() {
+                        if let Err(error) = worker.retry_pending_compaction_cleanup().await {
+                            let _ = working_event_tx.send(Event::Error {
+                                code: worker_error_code(&error),
+                                message: error.to_string(),
+                            });
+                            tokio::time::sleep(Duration::from_millis(50)).await;
+                        }
+                    }
                     let _ = working_event_tx.send(Event::Shutdown);
                     break 'controller;
                 }
@@ -3333,6 +3445,28 @@ mod tests {
         }
     }
 
+    #[test]
+    fn attention_gate_rejects_submit_before_dispatch() {
+        let (event_tx, mut event_rx) = broadcast::channel(4);
+        let method = Method::Submit {
+            submission_request_id: "request-1".into(),
+            input: Vec::new(),
+        };
+
+        assert!(reject_method_while_attention_locked(
+            &method,
+            &event_tx,
+            "repair required",
+        ));
+        assert!(matches!(
+            event_rx.try_recv().unwrap(),
+            Event::SubmissionRejected {
+                submission_request_id,
+                message,
+            } if submission_request_id == "request-1" && message == "repair required"
+        ));
+    }
+
     #[tokio::test]
     async fn segment_activation_failure_from_run_never_publishes_idle() {
         let mut env = make_env().await;

crates/worker/src/worker.rs

@@ -5178,6 +5178,21 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
         }
     }
 
+    pub(crate) fn has_pending_compaction_cleanup(&self) -> bool {
+        self.pending_compaction_cleanup
+            .lock()
+            .expect("pending compaction cleanup mutex poisoned")
+            .is_some()
+    }
+
+    pub(crate) fn has_failed_segment_activation(&self) -> bool {
+        self.segment_state.activation_failed.load(Ordering::Acquire)
+    }
+
+    pub(crate) async fn retry_pending_compaction_cleanup(&self) -> Result<(), WorkerError> {
+        self.release_pending_compaction_service().await
+    }
+
     async fn release_pending_compaction_service(&self) -> Result<(), WorkerError> {
         let cleanup = self
             .pending_compaction_cleanup
@@ -5480,7 +5495,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
         }));
 
         if let Err(error) = handle.send(summary_input.text).await {
-            let _ = registry.remove_service(&handle.session_id_string());
+            // Keep the registry record and Worker-owned cleanup authority. The
+            // outer compaction boundary performs the single stop attempt.
             return Err(WorkerError::InvalidState(error.to_string()));
         }
         match handle.wait_until_idle().await {

docs/design/compaction.md

@@ -62,7 +62,10 @@ is pending. Cleanup authority is stored on the Worker (not only on one compactio
 future). A failed stop returns a typed cleanup-pending outcome without clearing that
 authority; the next compaction boundary retries it. The authority and lifecycle
 reference are cleared exactly once after the registry record is gone, before
-activation or a terminal idle-capable result.
+activation or a terminal idle-capable result. While either activation repair or
+cleanup attention is outstanding, the controller fences Submit/Notify dispatch and
+does not start queued work. Shutdown is itself a cleanup barrier: it is not emitted
+until the retained service record has been stopped successfully.
 
 ## Metrics and comparison procedure