diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 3fe23dee2..afa94952e 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1384,6 +1384,7 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { #[derive(Default)] struct OrchestratorAttentionState { sending: bool, + restore_recheck: bool, accepted: Option, pending: Option, } @@ -1395,11 +1396,24 @@ struct OrchestratorAttentionDelivery { content: String, } -struct OrchestratorAttentionReservation(Arc>); +struct OrchestratorAttentionReservation(Option>>); + +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 can reserve its own check; one arriving before is ours to drain. + state.sending = false; + std::mem::take(&mut state.restore_recheck) + } +} impl Drop for OrchestratorAttentionReservation { fn drop(&mut self) { - self.0.lock().unwrap_or_else(|p| p.into_inner()).sending = false; + if let Some(state) = &self.0 { + state.lock().unwrap_or_else(|p| p.into_inner()).sending = false; + } } } @@ -16877,17 +16891,48 @@ fn dispatch_orchestrator_queue_attention( api: &WorkspaceApi, worker: &RuntimeWorkerRef, retry_only: bool, +) { + with_orchestrator_attention_reservation(api, worker, retry_only, |state, pending| { + check_orchestrator_queue_attention(api, worker, state, pending); + }); +} + +fn with_orchestrator_attention_reservation( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + mut retry_only: bool, + mut check: impl FnMut( + &Arc>, + Option, + ), ) { let state = orchestrator_attention_state(api, worker); - let pending = { - let mut state = state.lock().unwrap_or_else(|p| p.into_inner()); - if state.sending || (retry_only && state.pending.is_none()) { + loop { + let pending = { + let mut state = state.lock().unwrap_or_else(|p| p.into_inner()); + if state.sending || (retry_only && state.pending.is_none()) { + return; + } + state.sending = true; + state.pending.clone() + }; + let reservation = OrchestratorAttentionReservation(Some(state.clone())); + check(&state, pending); + if !reservation.complete() { return; } - state.sending = true; - state.pending.clone() - }; - let _reservation = OrchestratorAttentionReservation(state.clone()); + // Accepted Restore is an explicit recheck even when the current check + // was a no-op or an already-online retry. Never infer it from Idle. + retry_only = false; + } +} + +fn check_orchestrator_queue_attention( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + state: &Arc>, + pending: Option, +) { // Unknown outcomes keep the original immutable payload and ID, even if the // backlog or prompt projection changed before this retry. if let Some(delivery) = pending { @@ -22322,12 +22367,16 @@ async fn scoped_start_workspace_orchestrator( let mut attention = attention.lock().unwrap_or_else(|p| p.into_inner()); // Retire only pre-Restore success. Concurrent reservations, unknown // outcomes and newly accepted deliveries must survive Restore. - if !attention.sending - && attention.pending.is_none() + if attention.pending.is_none() && attention.accepted.as_ref().map(|d| &d.request_id) == before_restore.as_ref() { 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_queue_attention(&api, &existing.worker, false); @@ -48297,6 +48346,140 @@ mod tests { assert!(state.lock().unwrap().accepted.is_some()); } + #[tokio::test] + async fn orchestrator_attention_restore_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; + let worker = RuntimeWorkerRef::new( + EMBEDDED_WORKER_RUNTIME_ID, + started.worker.unwrap().worker_id, + ); + let state = orchestrator_attention_state(&api, &worker); + let old_id = state + .lock() + .unwrap() + .accepted + .as_ref() + .unwrap() + .request_id + .clone(); + execution.take_inputs(); + execution.take_input_request_ids(); + api.runtime + .stop_worker( + &worker, + WorkerLifecycleRequest { + reason: None, + ticket_assignment: None, + }, + ) + .unwrap(); + execution.accept_restores(); + let (entered_tx, entered_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + std::thread::scope(|scope| { + let api_for_caller = api.clone(); + let worker_for_caller = worker.clone(); + let caller = scope.spawn(move || { + let mut first = true; + with_orchestrator_attention_reservation( + &api_for_caller, + &worker_for_caller, + false, + |state, pending| { + if first && pause_before_comparison { + entered_tx.send(()).unwrap(); + release_rx + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + } + check_orchestrator_queue_attention( + &api_for_caller, + &worker_for_caller, + state, + pending, + ); + if first && !pause_before_comparison { + // Old fingerprint matched: check returned without sending, + // but the exact same production reservation still exists. + assert_eq!( + state.lock().unwrap().accepted.as_ref().unwrap().request_id, + old_id + ); + entered_tx.send(()).unwrap(); + release_rx + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + } + first = false; + }, + ); + }); + entered_rx + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + 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 old_retired = state.lock().unwrap().accepted.is_none(); + let inputs_before_release = execution.take_inputs(); + release_tx.send(()).unwrap(); + caller.join().unwrap(); + assert!(deferred); + assert!(old_retired); + assert!(inputs_before_release.is_empty()); + }); + assert_eq!(execution.take_inputs().len(), 1); + assert_eq!(execution.take_input_request_ids().len(), 1); + let state = state.lock().unwrap(); + assert!(state.accepted.is_some()); + assert!(!state.sending); + assert!(!state.restore_recheck); + } + } + + #[tokio::test] + async fn orchestrator_attention_restore_keeps_pending_unknown_delivery_and_id() { + let (_dir, api, execution) = attention_test_api().await; + execution.reject_inputs("unknown acceptance response"); + let started = start_attention_test_orchestrator(&api).await; + let worker = RuntimeWorkerRef::new( + EMBEDDED_WORKER_RUNTIME_ID, + started.worker.unwrap().worker_id, + ); + let state = orchestrator_attention_state(&api, &worker); + let pending = state.lock().unwrap().pending.clone().unwrap(); + execution.take_inputs(); + execution.take_input_request_ids(); + api.runtime + .stop_worker( + &worker, + WorkerLifecycleRequest { + reason: None, + ticket_assignment: None, + }, + ) + .unwrap(); + execution.accept_restores(); + *execution.input_failure.lock().unwrap() = None; + let restored = start_attention_test_orchestrator(&api).await; + assert_eq!(restored.disposition, "restored"); + let attempts = execution.take_inputs(); + assert_eq!(attempts.len(), 1); + assert_eq!(attempts[0].1, pending.content); + assert_eq!( + execution.take_input_request_ids(), + vec![Some(pending.request_id.clone())] + ); + assert!(state.lock().unwrap().pending.is_none()); + assert_eq!( + state.lock().unwrap().accepted.as_ref().unwrap().request_id, + pending.request_id + ); + } + #[tokio::test] async fn orchestrator_attention_unknown_response_retry_keeps_id_and_payload() { let (_dir, api, execution) = attention_test_api().await; diff --git a/docs/report/2026-10-05-review-verdict-schema-rejection.md b/docs/report/2026-10-05-review-verdict-schema-rejection.md new file mode 100644 index 000000000..7d5e9ce2e --- /dev/null +++ b/docs/report/2026-10-05-review-verdict-schema-rejection.md @@ -0,0 +1,30 @@ +# Reviewer verdict のvalidation rejectionとone-call制約 + +## 確認した事実 + +T-703のMR `01a10b3b-87ec-7093-b0c4-a00a78095695`、source +`3174142bd1314fe34c5de7b6a66654ff1c2a8af2` の独立ReviewerはRestoreとno-op +attention checkの競合を見つけ、`ReviewMergeRequest`でrequest_changesを提出しようとした。 +severityに`medium`を指定したためHTTP 422となった。Backendの許容値は +`blocker` / `major` / `minor` / `note` だった。 + +親の`ShowMergeRequest`再読でthreadはReviewRequested seq2までであり、verdictが +永続化されていないことを確認した。Reviewerは「ReviewMergeRequestを正確に一度だけ +呼ぶ」という指示に従い、validation rejection後も再提出を行わなかった。親のfollow-upでも +同じ境界を報告したため、再提出を強制せずReviewerを停止した。 + +## 障壁 + +正しいBackend拒否を迂回すべきではない一方、mutationが行われていない入力validation失敗が +一回限りの提出枠を消費する扱いだと、レビュー済みの指摘がMR authorityに残らずpendingになる。 +セッションでの報告をverdictや承認の代わりにはできない。 + +## 改善案(このTicketの実装対象ではない) + +- Reviewer-facing tool schemaにseverityのBackend enumを明示し、不正値を入力時に防ぐ。 +- one-call指示について「一度のdurable verdict」と「validationで拒否された呼出」を区別できるか + 設計上判断する。無制限再提出や記録済みverdictの変更を認める提案ではない。 +- mutationなし/retry可否をtyped errorとして返せると、親とReviewerの双方が推測なしに扱える。 + +T-703では元のMR/source selectorを保持し、指摘を修正したcommitを通常pushした後、fresh +exact-source reviewで修正内容とその証拠をMRに記録する。未記録のverdictを承認扱いしない。