fix: drain concurrent assignment inputs and discard stale pending retries

crates/workspace-server/src/server.rs

@@ -1528,7 +1528,7 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession {
 #[derive(Default)]
 struct OrchestratorAttentionState {
     sending: bool,
-    restore_recheck: bool,
+    input_recheck: bool,
     accepted: Option<OrchestratorAttentionDelivery>,
     pending: Option<OrchestratorAttentionDelivery>,
 }
@@ -1547,10 +1547,11 @@ impl OrchestratorAttentionReservation {
     fn complete(mut self) -> bool {
         let state = self.0.take().expect("active attention reservation");
         let mut state = state.lock().unwrap_or_else(|p| p.into_inner());
-        // Release and take deferred work atomically. A Restore arriving after
+        // Release and take deferred input checks atomically. A committed input or
-        // release can reserve its own check; one arriving before is ours to drain.
+        // Restore arriving before release is ours to drain; one after release
+        // can reserve its own check.
         state.sending = false;
-        std::mem::take(&mut state.restore_recheck)
+        std::mem::take(&mut state.input_recheck)
     }
 }
 
@@ -18865,12 +18866,22 @@ fn maybe_dispatch_orchestrator_turn_end(
     if orchestrator.worker.runtime_id == EMBEDDED_WORKER_RUNTIME_ID
         && orchestrator.worker.worker_id == worker_id
     {
-        dispatch_orchestrator_assignment_attention(api, &orchestrator.worker);
+        // Idle can be a reentrant consequence of this very Runtime send. It is
+        // not evidence of a new input and must not retry a failed send forever.
+        // Concurrent committed-input notifications independently defer a check.
+        with_orchestrator_attention_reservation(
+            api,
+            &orchestrator.worker,
+            false,
+            |state, pending| {
+                check_orchestrator_assignment_attention(api, &orchestrator.worker, state, pending);
+            },
+        );
     }
 }
 
 fn dispatch_orchestrator_assignment_attention(api: &WorkspaceApi, worker: &RuntimeWorkerRef) {
-    with_orchestrator_attention_reservation(api, worker, |state, pending| {
+    with_orchestrator_attention_reservation(api, worker, true, |state, pending| {
         check_orchestrator_assignment_attention(api, worker, state, pending);
     });
 }
@@ -18878,6 +18889,7 @@ fn dispatch_orchestrator_assignment_attention(api: &WorkspaceApi, worker: &Runti
 fn with_orchestrator_attention_reservation(
     api: &WorkspaceApi,
     worker: &RuntimeWorkerRef,
+    defer_input_check_if_reserved: bool,
     mut check: impl FnMut(
         &Arc<Mutex<OrchestratorAttentionState>>,
         Option<OrchestratorAttentionDelivery>,
@@ -18888,6 +18900,9 @@ fn with_orchestrator_attention_reservation(
         let pending = {
             let mut state = state.lock().unwrap_or_else(|p| p.into_inner());
             if state.sending {
+                // A no-op scan may already have read its empty snapshot. Retain
+                // committed input/startup/Restore triggers until its completion.
+                state.input_recheck |= defer_input_check_if_reserved;
                 return;
             }
             state.sending = true;
@@ -18898,8 +18913,8 @@ fn with_orchestrator_attention_reservation(
         if !reservation.complete() {
             return;
         }
-        // Accepted Restore is an explicit durable-input recheck even when the
+        // Drain coalesced input triggers even when the first scan was a no-op.
-        // current check was a no-op. A local cache is never backlog authority.
+        // No recheck is created by a failed send or its own idle observation.
     }
 }
 
@@ -18951,10 +18966,19 @@ fn check_orchestrator_assignment_attention(
             .is_ok()
         {
             send_orchestrator_attention(api, worker, state, delivery);
+            if state
+                .lock()
+                .unwrap_or_else(|p| p.into_inner())
+                .pending
+                .is_some()
+            {
+                return; // Unknown/failed acceptance needs an explicit later trigger.
+            }
         } else {
             state.lock().unwrap_or_else(|p| p.into_inner()).pending = None;
         }
-        return;
+        // Discarding or acknowledging an old retry must not consume this new-
+        // input trigger. Continue the common scan in the same reservation.
     }
     let Ok(inputs) = api
         .store
@@ -22969,11 +22993,6 @@ async fn scoped_start_workspace_orchestrator(
                 {
                     attention.accepted = None;
                 }
-                // A reserved caller may already have decided not to send. Carry
-                // the explicit Restore check through its completion in that case.
-                if attention.sending {
-                    attention.restore_recheck = true;
-                }
             }
             drop(_guard);
             dispatch_orchestrator_assignment_attention(&api, &existing.worker);

crates/workspace-server/src/server_assignment_attention_tests.rs

@@ -154,7 +154,7 @@ async fn orchestrator_attention_barrier_callers_share_unsent_reservation() {
 }
 
 #[tokio::test]
-async fn orchestrator_attention_restore_rechecks_reserved_noop_before_and_after_comparison() {
+async fn orchestrator_attention_input_rechecks_reserved_noop_before_and_after_comparison() {
     for pause_before_comparison in [true, false] {
         let (_dir, api, execution) = attention_test_api().await;
         let started = start_attention_test_orchestrator(&api).await;
@@ -187,6 +187,7 @@ async fn orchestrator_attention_restore_rechecks_reserved_noop_before_and_after_
                 with_orchestrator_attention_reservation(
                     &api_for_caller,
                     &worker_for_caller,
+                    true,
                     |state, pending| {
                         if checks == 0 && pause_before_comparison {
                             entered_tx.send(()).unwrap();
@@ -224,7 +225,7 @@ async fn orchestrator_attention_restore_rechecks_reserved_noop_before_and_after_
             assert!(state.lock().unwrap().sending);
             let restored = futures::executor::block_on(start_attention_test_orchestrator(&api));
             assert_eq!(restored.disposition, "restored");
-            let deferred = state.lock().unwrap().restore_recheck;
+            let deferred = state.lock().unwrap().input_recheck;
             let old_retired = state.lock().unwrap().accepted.is_none();
             let inputs_before_release = execution.take_inputs();
             release_tx.send(()).unwrap();
@@ -239,7 +240,7 @@ async fn orchestrator_attention_restore_rechecks_reserved_noop_before_and_after_
         let state = state.lock().unwrap();
         assert!(state.accepted.is_none());
         assert!(!state.sending);
-        assert!(!state.restore_recheck);
+        assert!(!state.input_recheck);
     }
 }
 
@@ -308,19 +309,19 @@ async fn orchestrator_attention_unknown_response_retry_keeps_id_and_payload() {
         "existing"
     );
     let retry = execution.take_inputs();
-    assert_eq!(retry.len(), 1);
+    assert_eq!(retry.len(), 2, "acknowledged retry must also drain the changed backlog");
     assert_eq!(retry[0].1, pending.content);
-    assert_eq!(
+    assert!(retry[1].1.contains("Changed backlog"));
-        execution.take_input_request_ids(),
+    let ids = execution.take_input_request_ids();
-        vec![Some(pending.request_id)]
+    assert_eq!(ids.len(), 2);
-    );
+    assert_eq!(ids[0], Some(pending.request_id));
+    assert_ne!(ids[1], ids[0]);
     assert!(state.lock().unwrap().pending.is_none());
     assert!(state.lock().unwrap().accepted.is_some());
+    assert!(attention_pending(&api).is_empty());
     use protocol::subscription::SubscriptionWorkerState as S;
     maybe_dispatch_orchestrator_turn_end(&api, &worker.worker_id, Some(S::Running), S::Idle);
-    let changed = execution.take_inputs();
+    assert!(execution.take_inputs().is_empty());
-    assert_eq!(changed.len(), 1);
-    assert!(changed[0].1.contains("Changed backlog"));
 }
 
 #[tokio::test]

crates/workspace-server/src/server_assignment_delivery_tests.rs

@@ -36,7 +36,11 @@ async fn delivery_http(
         "{method} {path}: {}",
         String::from_utf8_lossy(&bytes)
     );
-    serde_json::from_slice(&bytes).unwrap()
+    if bytes.is_empty() {
+        Value::Null
+    } else {
+        serde_json::from_slice(&bytes).unwrap()
+    }
 }
 
 async fn delivery_assign(api: &WorkspaceApi, ticket: &str, key: &str) -> Value {
@@ -330,65 +334,157 @@ async fn assignment_done_historical_inputs_do_not_restart_work_but_reissue_is_ne
 
 #[tokio::test]
 async fn assignment_clear_and_reassign_discard_old_unknown_delivery_without_stopping_worker() {
+    for clear_first in [true, false] {
+        let dir = tempfile::tempdir().unwrap();
+        let (api, execution, worker, _) = delivery_recording_fixture(dir.path()).await;
+        let ticket = delivery_ticket(&api, "Fence old delivery");
+        execution.reject_inputs("unknown acceptance response");
+        let assigned = delivery_assign(&api, &ticket.id, "assign-old").await;
+        let assignment = assigned["assignment"]["assignment_id"].as_str().unwrap();
+        let old_attempts = execution.take_inputs();
+        let old_ids = execution.take_input_request_ids();
+        assert!(!old_attempts.is_empty());
+        assert!(old_ids.iter().all(|id| id == &old_ids[0]));
+        if clear_first {
+            delivery_http(
+                &api,
+                "DELETE",
+                &format!(
+                    "{}?assignment_id={assignment}&operation_id=clear-old",
+                    delivery_assignment_path(&ticket.id)
+                ),
+                None,
+                StatusCode::OK,
+            )
+            .await;
+        }
+        *execution.input_failure.lock().unwrap() = None;
+        // No extra scan, old-operation retry, or synthetic turn end between
+        // clearing/reissuing the failed input and this committed new Assign.
+        let reassigned = delivery_assign(&api, &ticket.id, "assign-new").await;
+        let new_assignment = reassigned["assignment"]["assignment_id"].as_str().unwrap();
+        assert_ne!(new_assignment, assignment);
+        let inputs = execution.take_inputs();
+        assert_eq!(inputs.len(), 1, "new Assign must not need another scan");
+        assert!(inputs[0].1.contains(new_assignment));
+        let ids = execution.take_input_request_ids();
+        assert_eq!(ids.len(), 1);
+        assert_ne!(ids[0], old_ids[0], "new assignment reused stale input identity");
+        assert_eq!(
+            delivery_assign(&api, &ticket.id, "assign-old").await,
+            assigned
+        );
+        assert!(execution.take_inputs().is_empty(), "old retry revived work");
+        let old = delivery_tracking(&api, &ticket.id, assignment);
+        assert!(!old.current);
+        assert_eq!(
+            old.tracking.unwrap().status,
+            ticket::TicketAssignmentTrackingStatus::Waiting,
+            "clear/reissue must not rewrite the old status to done"
+        );
+        assert_eq!(
+            find_workspace_orchestrator(&api).unwrap().worker,
+            worker,
+            "clear/reassign are not Worker stop or replacement"
+        );
+        delivery_stop(&api, &worker);
+    }
+}
+
+#[tokio::test]
+async fn assignment_or_answer_committed_during_empty_scan_is_delivered_without_another_trigger() {
+    for answer in [false, true] {
+        let dir = tempfile::tempdir().unwrap();
+        let (api, execution, worker, identity) = delivery_recording_fixture(dir.path()).await;
+        let ticket = delivery_ticket(&api, "Input racing an empty scan");
+        if answer {
+            let assigned = delivery_assign(&api, &ticket.id, "initial-assign").await;
+            let current = delivery_tracking(
+                &api,
+                &ticket.id,
+                assigned["assignment"]["assignment_id"].as_str().unwrap(),
+            );
+            delivery_self_report(&api, &identity, &worker, &current, "waiting", "wait-answer").await;
+            execution.take_inputs();
+            execution.take_input_request_ids();
+        }
+        assert!(attention_pending(&api).is_empty());
+        let (entered_tx, entered_rx) = std::sync::mpsc::channel();
+        let (release_tx, release_rx) = std::sync::mpsc::channel();
+        std::thread::scope(|scope| {
+            let caller_api = api.clone();
+            let caller_worker = worker.clone();
+            let caller = scope.spawn(move || {
+                let mut checks = 0;
+                with_orchestrator_attention_reservation(
+                    &caller_api,
+                    &caller_worker,
+                    true,
+                    |state, pending| {
+                        check_orchestrator_assignment_attention(
+                            &caller_api, &caller_worker, state, pending,
+                        );
+                        if checks == 0 {
+                            // The production scan has read an empty snapshot but
+                            // still owns its reservation. Race a real HTTP commit.
+                            entered_tx.send(()).unwrap();
+                            release_rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
+                        }
+                        checks += 1;
+                    },
+                );
+                assert_eq!(checks, 2, "committed input must defer a scan");
+            });
+            entered_rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
+            if answer {
+                futures::executor::block_on(delivery_http(
+                    &api,
+                    "POST",
+                    &format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{}/thread-events", ticket.id),
+                    Some(serde_json::to_value(NewTicketEvent::new(
+                        TicketEventKind::Comment, "Human answer: go ahead",
+                    )).unwrap()),
+                    StatusCode::NO_CONTENT,
+                ));
+            } else {
+                futures::executor::block_on(delivery_assign(&api, &ticket.id, "racing-assign"));
+            }
+            // No manual dispatch, Restore, or synthetic Running -> Idle.
+            assert!(execution.take_inputs().is_empty());
+            release_tx.send(()).unwrap();
+            caller.join().unwrap();
+        });
+        let inputs = execution.take_inputs();
+        assert_eq!(inputs.len(), 1, "idle Orchestrator was stranded");
+        assert!(inputs[0].1.contains(ticket.resource_key.as_deref().unwrap()));
+        if answer {
+            assert!(inputs[0].1.contains("Need the human's choice"));
+        }
+        assert_eq!(execution.take_input_request_ids().len(), 1);
+        assert!(attention_pending(&api).is_empty());
+        delivery_stop(&api, &worker);
+    }
+}
+
+#[tokio::test]
+async fn assignment_failed_delivery_and_its_reentrant_idle_do_not_schedule_retry_loop() {
     let dir = tempfile::tempdir().unwrap();
     let (api, execution, worker, _) = delivery_recording_fixture(dir.path()).await;
-    let ticket = delivery_ticket(&api, "Fence old delivery");
+    let ticket = delivery_ticket(&api, "No failed-delivery busy loop");
-    execution.reject_inputs("unknown acceptance response");
+    execution.reject_inputs("Runtime currently cannot accept notifications");
-    let assigned = delivery_assign(&api, &ticket.id, "assign-old").await;
+    let callback_api = api.clone();
-    let assignment = assigned["assignment"]["assignment_id"].as_str().unwrap();
+    execution.before_input_returns(move |worker| {
-    let old_attempts = execution.take_inputs();
+        use protocol::subscription::SubscriptionWorkerState as S;
-    let old_ids = execution.take_input_request_ids();
+        maybe_dispatch_orchestrator_turn_end(
-    assert!(!old_attempts.is_empty());
+            &callback_api, &worker.worker_id.to_string(), Some(S::Running), S::Idle,
-    assert!(old_ids.iter().all(|id| id == &old_ids[0]));
+        );
-    delivery_http(
+    });
-        &api,
+    delivery_assign(&api, &ticket.id, "assign-failed").await;
-        "DELETE",
+    assert_eq!(execution.take_inputs().len(), 2, "idle must not add retries beyond the bounded send");
-        &format!(
-            "{}?assignment_id={assignment}&operation_id=clear-old",
-            delivery_assignment_path(&ticket.id)
-        ),
-        None,
-        StatusCode::OK,
-    )
-    .await;
-    *execution.input_failure.lock().unwrap() = None;
-    notify_ticket_recipients(&api, TEST_WORKSPACE_ID, &ticket.id, "open", "open", None);
-    delivery_scan(&api, &worker);
-    assert!(
-        execution.take_inputs().is_empty(),
-        "cleared reservation was sent as new work"
-    );
-    assert_eq!(
-        delivery_assign(&api, &ticket.id, "assign-old").await,
-        assigned
-    );
-    delivery_scan(&api, &worker);
-    assert!(
-        execution.take_inputs().is_empty(),
-        "old operation retry revived cleared work"
-    );
-    let reassigned = delivery_assign(&api, &ticket.id, "assign-new").await;
-    let new_assignment = reassigned["assignment"]["assignment_id"].as_str().unwrap();
-    assert_ne!(new_assignment, assignment);
-    assert_eq!(execution.take_inputs().len(), 1);
     let ids = execution.take_input_request_ids();
-    assert_eq!(ids.len(), 1);
+    assert_eq!(ids.len(), 2);
-    assert_ne!(
+    assert_eq!(ids[0], ids[1]);
-        ids[0], old_ids[0],
+    assert_eq!(attention_pending(&api).len(), 1, "failure must retain durable input");
-        "new assignment reused stale input identity"
-    );
-    let old = delivery_tracking(&api, &ticket.id, assignment);
-    assert!(!old.current);
-    assert_eq!(
-        old.tracking.unwrap().status,
-        ticket::TicketAssignmentTrackingStatus::Waiting,
-        "clear must not rewrite the old status to done"
-    );
-    assert_eq!(
-        find_workspace_orchestrator(&api).unwrap().worker,
-        worker,
-        "clear/reassign are not Worker stop or replacement"
-    );
     delivery_stop(&api, &worker);
 }
 

docs/development/ticket-assignment-delivery.md

@@ -24,7 +24,7 @@ Delivery order:
 4. Send the saved content with the saved `request_id`, outside the transaction. Response loss retains the same request/payload. Concurrent reservations converge on the same identity, and Runtime request receipts deduplicate actual acceptance.
 5. Acknowledge accepted delivery under another IMMEDIATE transaction after rechecking current authority. The acknowledgement never changes handling.
 
-Process-local attention locks prevent concurrent/reentrant send loops; they are not backlog authority. Losing that cache does not lose a delegation or resend acknowledged input. Replacement does not create assignment rows; only pending, unacknowledged inputs are recovered. Accepted history is not replayed into new work merely because a Worker was exchanged.
+Process-local attention locks prevent concurrent/reentrant send loops; they are not backlog authority. A committed-input/startup/Restore trigger arriving during a reserved scan is coalesced into an input recheck and drained on reservation release, including when that scan read an empty snapshot. Discarding or acknowledging a pending retry continues the common scan in the same reservation; a successful retry cannot consume the trigger for newer input. Reentrant idle observations caused by a send do not schedule another retry of a failed send. These liveness repairs change no DB transaction or public API boundary. Losing that cache does not lose a delegation or resend acknowledged input. Replacement does not create assignment rows; only pending, unacknowledged inputs are recovered. Accepted history is not replayed into new work merely because a Worker was exchanged.
 
 ## Public start/action handoff for T-743