fix(windows): retry a briefly contended plugin-state replace; make the OAuth test fixture read a whole request

Two Windows failures kept recurring on hosted CI today (PRs #6882, #6884,
#6887 and #6905), each costing an 80-minute re-run:

- plugins::tests::stale_concurrent_registries_do_not_lose_updates:
  "failed to atomically and durably persist ...plugin-state.json: Access is
  denied. (0x80070005)". This is a product gap, not only a test problem. The
  state lock serialises writers, but another registry re-applies the state
  directory ACL before taking that lock, and a scanner can hold the file, so
  MoveFileExW is briefly refused. persist_plugin_state now retries only the
  rename, on the bounded schedule the other atomic writers already share
  (utils::windows_publish_retry_delay: access denied, sharing or lock
  violation; six attempts, about 310 ms). The hardened temporary and the stable
  file are untouched on every attempt. New Windows test holds the destination
  without delete sharing for 50 ms. The blocking-call budget records the one
  new sleep, as fleet/files.rs does for the same pattern.

- oauth::tests::orcarouter_connect_adapter_runs_authorize_callback_exchange_end_to_end:
  "received unexpected message from connection". Test fixture only: the
  accepted socket inherits the listener's non-blocking mode, a single read
  could return WouldBlock, and the fixture answered 404 before the request
  arrived. The fixture now reads a complete request. Reproduced in isolation on
  macOS (which shares the inheritance): 162 of 300 requests succeeded before,
  300 of 300 after.

Evidence:
- macOS, RUSTFLAGS=-Dwarnings, cargo nextest oauth:: and plugins:: filters:
  "173 tests run: 173 passed, 15150 skipped". cargo fmt --check clean.
  Blocking-call budget: 535 sites across 165 files, within budget.
- NOT executed on Windows locally: the Windows-only retry and its test were
  type-checked for x86_64-pc-windows-msvc in an isolated crate with clippy
  -D warnings by two independent reviewers. Hosted Test (windows-latest) is the
  first real run. The new retry test is timing-based, like
  write_atomic_retries_windows_replace_contention.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

crates/tui/src/oauth.rs

@@ -8912,10 +8912,7 @@ consent_version = 1
                 while std::time::Instant::now() < deadline {
                     match listener.accept() {
                         Ok((mut stream, _)) => {
-                            let _ = stream.set_read_timeout(Some(Duration::from_secs(2)));
-                            let mut raw = [0u8; 8192];
-                            let read = stream.read(&mut raw).unwrap_or(0);
-                            let head = String::from_utf8_lossy(&raw[..read]).to_string();
+                            let head = read_fixture_request(&mut stream);
                             seen.lock().unwrap().push(head.clone());
                             let request_line = head.lines().next().unwrap_or_default();
                             let mut parts = request_line.split_whitespace();
@@ -8985,6 +8982,37 @@ consent_version = 1
         }
     }
 
+    /// Read one whole request for the fake auth server: the head, then the
+    /// `Content-Length` bytes of body. The fixture's listener is non-blocking
+    /// and Windows and BSD sockets hand that mode to the accepted stream, so a
+    /// single `read` can return `WouldBlock` before the client has written a
+    /// byte. Answering that empty read puts a response on the wire ahead of
+    /// the request, which the client reports as "received unexpected message
+    /// from connection"; closing with the body still unread resets the
+    /// connection under the reply.
+    fn read_fixture_request(stream: &mut TcpStream) -> String {
+        let _ = stream.set_nonblocking(false);
+        let _ = stream.set_read_timeout(Some(Duration::from_secs(5)));
+        let mut raw = Vec::new();
+        let mut chunk = [0u8; 4096];
+        while raw.len() < 64 * 1024 {
+            if let Some(head_end) = raw.windows(4).position(|window| window == b"\r\n\r\n") {
+                let head = String::from_utf8_lossy(&raw[..head_end]);
+                let body_len = header_value(&head, "content-length")
+                    .and_then(|value| value.parse::<usize>().ok())
+                    .unwrap_or(0);
+                if raw.len() >= head_end + 4 + body_len {
+                    break;
+                }
+            }
+            match stream.read(&mut chunk) {
+                Ok(0) | Err(_) => break,
+                Ok(read) => raw.extend_from_slice(&chunk[..read]),
+            }
+        }
+        String::from_utf8_lossy(&raw).to_string()
+    }
+
     /// The full connect adapter over a local fake auth server: the adapter
     /// binds its own loopback callback, the fake server redirects the code to
     /// exactly that callback with the adapter's own state, and the exchange
@@ -9050,6 +9078,31 @@ consent_version = 1
         })
     }
 
+    /// The fake auth server must answer the request, not the accept: a client
+    /// that connects, pauses, and then sends its head and body in separate
+    /// segments still gets the exchange reply, and the server saw the body.
+    #[test]
+    fn fake_orca_auth_server_waits_for_a_late_fragmented_request() {
+        let auth = FakeOrcaAuthServer::start();
+        let mut stream = TcpStream::connect(("127.0.0.1", auth.port())).expect("connect");
+        // Longer than the fixture's accept poll, so it accepts an empty socket.
+        std::thread::sleep(Duration::from_millis(100));
+        let body = r#"{"code":"late-fragment"}"#;
+        write!(
+            stream,
+            "POST {ORCAROUTER_EXCHANGE_PATH} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+            body.len()
+        )
+        .expect("write head");
+        std::thread::sleep(Duration::from_millis(100));
+        stream.write_all(body.as_bytes()).expect("write body");
+        let mut response = String::new();
+        stream.read_to_string(&mut response).expect("read response");
+        assert!(response.starts_with("HTTP/1.1 200 OK"), "{response}");
+        let recorded = auth.requests.lock().unwrap().join("\n");
+        assert!(recorded.contains("late-fragment"), "{recorded}");
+    }
+
     #[test]
     fn orcarouter_connect_adapter_runs_authorize_callback_exchange_end_to_end() {
         let auth = FakeOrcaAuthServer::start();

crates/tui/src/plugins/registry.rs

@@ -964,6 +964,7 @@ fn persist_plugin_state_with_directory_sync(
 #[cfg(windows)]
 fn persist_plugin_state(mut temporary: tempfile::TempPath, path: &Path) -> Result<(), String> {
     use std::os::windows::ffi::OsStrExt as _;
+    use windows::Win32::Foundation::WIN32_ERROR;
     use windows::Win32::Storage::FileSystem::{
         FILE_ATTRIBUTE_NORMAL, FILE_ATTRIBUTE_TEMPORARY, MOVEFILE_REPLACE_EXISTING,
         MOVEFILE_WRITE_THROUGH, MoveFileExW, SetFileAttributesW,
@@ -991,14 +992,35 @@ fn persist_plugin_state(mut temporary: tempfile::TempPath, path: &Path) -> Resul
         format!("failed to prepare private plugin state temp file for publication: {error}")
     })?;
 
+    // The state lock serializes writers, but not everything that opens the
+    // stable file: another registry re-applies the state directory's
+    // inheritable ACL before it takes that lock, and Defender or the indexer
+    // can scan the file. Each holds the destination only briefly, and
+    // MoveFileExW then reports a refusal that clears on its own. Re-attempt
+    // only the rename, on the schedule the other atomic writers share; the
+    // hardened temporary and the stable file are both left untouched.
+    let mut attempt = 0;
     // SAFETY: both paths are NUL-terminated and live.
-    if let Err(error) = unsafe {
+    while let Err(error) = unsafe {
         MoveFileExW(
             PCWSTR::from_raw(temporary_wide.as_ptr()),
             PCWSTR::from_raw(destination_wide.as_ptr()),
             MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
         )
     } {
+        // MoveFileExW reports an HRESULT; the shared classifier reads the
+        // Win32 code it wraps.
+        let backoff = WIN32_ERROR::from_error(&error).and_then(|code| {
+            crate::utils::windows_publish_retry_delay(
+                &std::io::Error::from_raw_os_error(code.0 as i32),
+                attempt,
+            )
+        });
+        if let Some(backoff) = backoff {
+            std::thread::sleep(backoff);
+            attempt += 1;
+            continue;
+        }
         // Restore tempfile's cleanup hint on the still-private source. The
         // stable state path remains untouched when MoveFileExW fails.
         // SAFETY: `temporary_wide` is NUL-terminated and live.
@@ -2878,6 +2900,42 @@ mod windows_acl_tests {
         drop(retained);
     }
 
+    #[test]
+    fn transient_state_replacement_contention_is_retried() {
+        use std::os::windows::fs::OpenOptionsExt as _;
+
+        let directory = tempfile::tempdir().unwrap();
+        let state_path = directory.path().join("state.json");
+        std::fs::write(&state_path, b"old-authoritative-state").unwrap();
+        // FILE_SHARE_READ alone omits delete sharing: the short-lived handle a
+        // concurrent ACL pass or scanner holds on the stable file.
+        let retained = std::fs::OpenOptions::new()
+            .read(true)
+            .share_mode(0x0000_0001)
+            .open(&state_path)
+            .unwrap();
+        let release = std::thread::spawn(move || {
+            std::thread::sleep(std::time::Duration::from_millis(50));
+            drop(retained);
+        });
+
+        save_state_with_hardener(
+            &state_path,
+            &PluginStateFile::default(),
+            harden_plugin_state_file,
+        )
+        .expect("retry contended plugin state replacement");
+
+        release.join().unwrap();
+        let published = std::fs::read_to_string(&state_path).unwrap();
+        assert!(published.contains("\"schema_version\": 1"));
+        let entries = std::fs::read_dir(directory.path())
+            .unwrap()
+            .map(|entry| entry.unwrap().file_name())
+            .collect::<Vec<_>>();
+        assert_eq!(entries, [std::ffi::OsString::from("state.json")]);
+    }
+
     #[test]
     fn owner_only_runtime_acl_is_protected_and_has_one_full_access_ace() {
         let directory = tempfile::tempdir().unwrap();

scripts/check-blocking-calls-budget.json

@@ -270,7 +270,8 @@
     "std_fs": 1
   },
   "crates/tui/src/plugins/registry.rs": {
-    "path_canonicalize": 6
+    "path_canonicalize": 6,
+    "thread_sleep": 1
   },
   "crates/tui/src/plugins/runtime.rs": {
     "path_canonicalize": 2