WIP app-server: Windows recovers a dead owner through the Unix path

UNVERIFIED WORK IN PROGRESS, pushed so it survives a move off the cloud session. Do not merge as is. Implements the Q4 plan for #6915 (founder: blocks the 0.10.2 tag; proof is the hosted Test (windows-latest) leg). Windows startup used connect_if_published and failed hard with "opening selected Windows owner; refusing fallback ... (os error 2)" once the owner died; the "stale-owner recovery is not qualified" guard in app-server lib.rs was unreachable. This joins Windows to the lease-first admission and the Unix recovery from a8cdccc1d6: - config/windows_identity.rs: windows_process_status(pid) -> Running/Exited/Absent from OpenProcess + creation time + WaitForSingleObject - config/private_directory.rs: POSIX-semantics handle deletion with the legacy fallback (same order as std::fs::remove_file) - app-server daemon_client.rs: require_dead_process / revalidate_* on Windows; require_same_selection shared with Unix; a NotFound pipe open maps to Unavailable - app-server daemon_windows.rs: bind takes `recovery`, retires the captured receipt by handle before creating the first pipe instance - app-server daemon_socket.rs: Unix bind uses require_same_selection - app-server lib.rs: the Windows-only ensure is gone; one bind call - tui runtime_api.rs / runtime_threads.rs: RuntimeHostAdmission, admit_runtime_host, try_acquire_for_host are cfg(any(unix, windows)); the Windows connect_if_published branch is deleted - cli/tests/owner_crash_reentry.rs: the founder-requested two-process test (spawn A, kill+reap, receipt still names dead A, spawn B, B owns it and serves A's thread) Verified so far (Linux only): - clippy -p codewhale-config -p codewhale-app-server -p codewhale-tui -p codewhale-cli --all-targets --all-features --locked -- -D warnings: exit 0, no warnings - cargo fmt --all -- --check: clean NOT yet run: owner_crash_reentry (Linux), app-server daemon_socket owner_bootstrap tests, tui runtime_api::tests::owner_bootstrap, any Windows compile (x86_64-pc-windows-gnu cross-check or hosted windows-latest), the three-lens review, the CHANGELOG change (drop the 0.10.2 Windows known-limit bullet, add the fix line, run scripts/sync-changelog.sh). Refs #6915 Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01BsXVvReYca2UxzimQtj7Mj
2026-11-09 01:49:49 -10:00 · 2026-10-12 04:17:55 +00:00
1 parent e43d86e8be
commit ec242fe079
@@ -120,27 +120,62 @@ mod platform {
            anyhow::Error::new(cause)
                .context("selected owner publication unavailable; canonical host recovery required")
        }
-        #[cfg(unix)]
        fn require_dead_process(&self) -> Result<()> {
-            use codewhale_config::private_directory::{UnixProcessStatus, unix_process_status};
            let receipt = self.receipt();
+            #[cfg(unix)]
+            let principal = PrivateDirectory::current_user_id().to_string();
+            #[cfg(windows)]
+            let principal =
+                codewhale_config::windows_identity::CurrentWindowsUser::open()?.sid_string()?;
            anyhow::ensure!(
-                receipt.principal == PrivateDirectory::current_user_id().to_string()
-                    && !receipt.process_start.is_empty(),
+                receipt.principal == principal && !receipt.process_start.is_empty(),
                "selected owner process identity is invalid"
            );
-            match unix_process_status(receipt.pid)? {
-                UnixProcessStatus::Absent => Ok(()),
-                UnixProcessStatus::Present { start } if start == receipt.process_start => {
-                    bail!("selected owner process is still present; refusing recovery")
+            #[cfg(unix)]
+            let (running, start) = {
+                use codewhale_config::private_directory::{UnixProcessStatus, unix_process_status};
+                match unix_process_status(receipt.pid)? {
+                    UnixProcessStatus::Absent => return Ok(()),
+                    UnixProcessStatus::Present { start } => (true, start),
                }
-                UnixProcessStatus::Present { .. } => {
-                    bail!("selected owner PID was reused; refusing recovery")
+            };
+            #[cfg(windows)]
+            let (running, start) = {
+                use codewhale_config::windows_identity::{
+                    WindowsProcessStatus, windows_process_status,
+                };
+                match windows_process_status(receipt.pid)? {
+                    WindowsProcessStatus::Absent => return Ok(()),
+                    WindowsProcessStatus::Running { start } => (true, start),
+                    WindowsProcessStatus::Exited { start } => (false, start),
                }
+            };
+            if start != receipt.process_start {
+                bail!("selected owner PID was reused; refusing recovery")
            }
+            if running {
+                bail!("selected owner process is still present; refusing recovery")
+            }
+            Ok(())
+        }
+        pub(crate) fn require_same_selection(
+            &self,
+            owner: &RuntimeOwnerReceipt,
+            endpoint: &std::path::Path,
+        ) -> Result<()> {
+            let old = self.receipt();
+            anyhow::ensure!(
+                old.version == owner.version
+                    && old.data_dir == owner.data_dir
+                    && old.execution_scope == owner.execution_scope
+                    && old.socket_path == endpoint
+                    && old.config_path == owner.config_path
+                    && old.principal == owner.principal,
+                "captured stale owner belongs to another selected store/config/principal"
+            );
+            Ok(())
        }
        /// Caller retains the real Runtime lease; this checks evidence only.
-        #[cfg(unix)]
        pub(crate) fn revalidate_receipt(&self) -> Result<bool> {
            if !self.publication.receipt_is_current()? {
                return Ok(false);
@@ -148,7 +183,6 @@ mod platform {
            self.require_dead_process()?;
            Ok(true)
        }
-        #[cfg(unix)]
        pub(crate) fn revalidate_all(&self) -> Result<bool> {
            if !self.publication.is_current()? {
                return Ok(false);
@@ -156,7 +190,6 @@ mod platform {
            self.require_dead_process()?;
            Ok(true)
        }
-        #[cfg(unix)]
        pub async fn revalidate(self: &Arc<Self>) -> Result<bool> {
            let held = self.clone();
            owner_work(move || held.revalidate_all()).await
@@ -459,7 +492,29 @@ mod platform {
        };
        #[cfg(windows)]
        let (stream, peer) = {
-            let (stream, process) = crate::daemon_windows::connect_owner(&receipt).await?;
+            let (stream, process) = match crate::daemon_windows::connect_owner(&receipt).await {
+                Ok(connected) => connected,
+                Err(error) => {
+                    let held = publication.clone();
+                    anyhow::ensure!(
+                        owner_work(move || held.is_current()).await?,
+                        PublicationChangedBeforeAttach
+                    );
+                    let missing = error
+                        .downcast_ref::<std::io::Error>()
+                        .filter(|cause| cause.kind() == std::io::ErrorKind::NotFound)
+                        .map(|cause| match cause.raw_os_error() {
+                            Some(code) => std::io::Error::from_raw_os_error(code),
+                            None => std::io::Error::from(cause.kind()),
+                        });
+                    if let Some(cause) = missing {
+                        return Ok(HostOwnerProbe::Unavailable(Arc::new(
+                            UnavailablePublication { publication, cause },
+                        )));
+                    }
+                    return Err(error);
+                }
+            };
            let peer = AuthorizedPeer {
                pid: process.pid(),
                start: process.start().to_string(),
@@ -999,18 +999,9 @@ mod platform {
                    ))
                })?
                .1;
-            let old = previous.receipt();
-            if old.version != owner.version
-                || old.data_dir != owner.data_dir
-                || old.execution_scope != owner.execution_scope
-                || old.socket_path != path
-                || old.config_path != owner.config_path
-                || old.principal != owner.principal
-            {
-                return Err(DaemonSocketError::State(anyhow::anyhow!(
-                    "captured stale owner belongs to another selected store/config/principal"
-                )));
-            }
+            previous
+                .require_same_selection(owner, &path)
+                .map_err(DaemonSocketError::State)?;
            if !previous
                .revalidate()
                .await
@@ -2,6 +2,7 @@
 //! Pipe names route; private receipts plus held kernel process/token identity
 //! select the actual owner. No bearer or Engine authority crosses this wire.

+use crate::daemon_client::UnavailablePublication;
 use crate::daemon_socket::{
    AuthorizedPeer, ConnectionTasks, DaemonShutdownHandle, DaemonSocketError, DaemonSocketOptions,
    SocketPathInputs, connection_context, owner_work, run_authorized_connection,
@@ -192,11 +193,12 @@ impl DaemonSocket {
 pub async fn bind_daemon_socket(
    options: DaemonSocketOptions,
 ) -> Result<DaemonSocket, DaemonSocketError> {
-    bind(options, None).await
+    bind(options, None, None).await
 }
 pub(crate) async fn bind_captured_owner(
    state: AppState,
    owner: RuntimeOwnerReceipt,
+    recovery: Option<Arc<UnavailablePublication>>,
 ) -> Result<DaemonSocket, DaemonSocketError> {
    bind(
        DaemonSocketOptions {
@@ -204,17 +206,59 @@ pub(crate) async fn bind_captured_owner(
            config_path: None,
        },
        Some((state, owner)),
+        recovery,
    )
    .await
 }
 async fn bind(
    options: DaemonSocketOptions,
    captured: Option<(AppState, RuntimeOwnerReceipt)>,
+    recovery: Option<Arc<UnavailablePublication>>,
 ) -> Result<DaemonSocket, DaemonSocketError> {
    let explicit = options.socket_path;
    let path = owner_work(move || selected_pipe_path(explicit))
        .await
        .map_err(DaemonSocketError::State)?;
+    let recovered = match recovery {
+        Some(previous) => {
+            let owner = &captured
+                .as_ref()
+                .ok_or_else(|| {
+                    DaemonSocketError::State(anyhow::anyhow!(
+                        "stale publication requires the actual captured Runtime owner"
+                    ))
+                })?
+                .1;
+            previous
+                .require_same_selection(owner, &path)
+                .map_err(DaemonSocketError::State)?;
+            Some(
+                owner_work(move || {
+                    let parent = previous.publication.parent.clone();
+                    anyhow::ensure!(
+                        parent.is_at_selected_path()?,
+                        "selected Windows owner parent changed"
+                    );
+                    anyhow::ensure!(
+                        previous.revalidate_receipt()?,
+                        "captured stale owner receipt changed before retirement"
+                    );
+                    anyhow::ensure!(
+                        parent.retire_private_receipt(
+                            &previous.publication.receipt_name,
+                            &previous.publication.file
+                        )?,
+                        "captured stale owner receipt changed during retirement"
+                    );
+                    drop(previous);
+                    Ok(parent)
+                })
+                .await
+                .map_err(DaemonSocketError::State)?,
+            )
+        }
+        None => None,
+    };
    let selected = path.clone();
    let pipe = owner_work(move || create_pipe(&selected, true))
        .await
@@ -232,7 +276,10 @@ async fn bind(
        let owner = owner.clone();
        Some(
            owner_work(move || {
-                let parent = Arc::new(PrivateDirectory::admit(&owner_directory()?)?);
+                let parent = match recovered {
+                    Some(parent) => parent,
+                    None => Arc::new(PrivateDirectory::admit(&owner_directory()?)?),
+                };
                let mut guard = ReceiptGuard {
                    parent: parent.clone(),
                    file: None,
@@ -241,23 +288,12 @@ async fn bind(
                    parent.is_at_selected_path()?,
                    "selected Windows owner parent changed"
                );
-                if let Some((bytes, file)) =
-                    parent.read_private_receipt("daemon.owner.json", 16384)?
-                {
-                    let previous: RuntimeOwnerReceipt = serde_json::from_slice(&bytes)?;
-                    anyhow::ensure!(
-                        previous.version == owner.version
-                            && previous.data_dir == owner.data_dir
-                            && previous.execution_scope == owner.execution_scope
-                            && previous.socket_path == owner.socket_path,
-                        "old Windows owner receipt belongs to another selected store"
-                    );
-                    anyhow::ensure!(
-                        parent.retire_private_receipt("daemon.owner.json", &file)?,
-                        "Windows owner receipt changed"
-                    );
-                    drop(file);
-                }
+                anyhow::ensure!(
+                    parent
+                        .read_private_receipt("daemon.owner.json", 16384)?
+                        .is_none(),
+                    "owner publication appeared during bootstrap; refusing replacement"
+                );
                let bytes = serde_json::to_vec(&owner)?;
                anyhow::ensure!(bytes.len() <= 16384, "Windows owner receipt exceeds bound");
                parent.write_owned_file("daemon.owner.json", &bytes, false)?;
@@ -653,16 +653,7 @@ pub async fn bind_runtime_frontends(
    state.captured_routing = Some(routing);
    state.owner_frontend = owner_frontend;
    *state.runtime_bridge.lock().await = Some(Arc::new(Mutex::new(bridge)));
-    #[cfg(unix)]
    let daemon = daemon_socket::bind_captured_owner(state.clone(), owner, recovery).await?;
-    #[cfg(windows)]
-    let daemon = {
-        anyhow::ensure!(
-            recovery.is_none(),
-            "Windows stale-owner recovery is not qualified; refusing fallback"
-        );
-        daemon_socket::bind_captured_owner(state.clone(), owner).await?
-    };
    Ok((daemon, state))
 }

@@ -0,0 +1,216 @@
+#![cfg(any(unix, windows))]
+
+use std::io::{BufRead, BufReader};
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::mpsc;
+use std::time::Duration;
+
+use serde_json::{Value, json};
+
+const TOKEN: &str = "owner-crash-reentry-fixture-token";
+const LISTENING: &str = "Runtime API listening on http://";
+
+struct Fixture {
+    root: tempfile::TempDir,
+}
+
+impl Fixture {
+    fn new() -> Self {
+        #[cfg(unix)]
+        let root = tempfile::Builder::new()
+            .prefix("cw-ocr-")
+            .tempdir_in(
+                std::path::Path::new("/tmp")
+                    .canonicalize()
+                    .expect("canonical /tmp"),
+            )
+            .expect("fixture root");
+        #[cfg(windows)]
+        let root = tempfile::tempdir().expect("fixture root");
+        for name in ["home", "state", "tmp", "work"] {
+            std::fs::create_dir_all(root.path().join(name)).expect("fixture directory");
+        }
+        std::fs::write(root.path().join("config.toml"), "").expect("fixture config");
+        Self { root }
+    }
+
+    fn path(&self, name: &str) -> PathBuf {
+        self.root.path().join(name)
+    }
+
+    fn command(&self) -> Command {
+        let mut command = Command::new(env!("CARGO_BIN_EXE_codewhale"));
+        command.current_dir(self.path("work")).env_clear();
+        for name in ["PATH", "SystemRoot", "WINDIR"] {
+            if let Some(value) = std::env::var_os(name) {
+                command.env(name, value);
+            }
+        }
+        let home = self.path("home");
+        let tmp = self.path("tmp");
+        command
+            .env("HOME", &home)
+            .env("USERPROFILE", &home)
+            .env("CODEWHALE_HOME", self.path("state"))
+            .env("CODEWHALE_SECRET_BACKEND", "file")
+            .env("CODEWHALE_TELEMETRY", "0")
+            .env("CODEWHALE_RUNTIME_TOKEN", TOKEN)
+            .env("TEMP", &tmp)
+            .env("TMP", &tmp)
+            .env("TMPDIR", &tmp)
+            .arg("--config")
+            .arg(self.path("config.toml"))
+            .arg("--no-project-config");
+        command
+    }
+
+    fn receipt(&self) -> (Vec<u8>, Value) {
+        #[cfg(unix)]
+        let name = "daemon.sock.owner.json";
+        #[cfg(windows)]
+        let name = "daemon.owner.json";
+        let bytes = std::fs::read(self.path("state").join("run").join(name))
+            .expect("published owner receipt");
+        let value = serde_json::from_slice(&bytes).expect("owner receipt JSON");
+        (bytes, value)
+    }
+
+    fn stderr(&self, label: &str) -> String {
+        std::fs::read_to_string(self.path(&format!("{label}.stderr"))).unwrap_or_default()
+    }
+}
+
+struct Owner {
+    child: Child,
+    base: String,
+}
+
+impl Drop for Owner {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+}
+
+fn spawn_owner(fixture: &Fixture, label: &str) -> Owner {
+    let stderr = std::fs::File::create(fixture.path(&format!("{label}.stderr")))
+        .expect("owner stderr capture");
+    let mut child = fixture
+        .command()
+        .args(["app-server", "--http", "--host", "127.0.0.1", "--port", "0"])
+        .stdin(Stdio::null())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::from(stderr))
+        .spawn()
+        .expect("spawn codewhale app-server");
+    let stdout = child.stdout.take().expect("owner stdout");
+    let (ready, listening) = mpsc::channel();
+    std::thread::spawn(move || {
+        let mut ready = Some(ready);
+        for line in BufReader::new(stdout).lines() {
+            let Ok(line) = line else { break };
+            if let Some(address) = line.strip_prefix(LISTENING)
+                && let Some(ready) = ready.take()
+            {
+                let _ = ready.send(address.trim().to_string());
+            }
+        }
+    });
+    let mut owner = Owner {
+        child,
+        base: String::new(),
+    };
+    match listening.recv_timeout(Duration::from_secs(120)) {
+        Ok(address) => owner.base = format!("http://{address}"),
+        Err(_) => {
+            let status = owner.child.try_wait().ok().flatten();
+            drop(owner);
+            panic!(
+                "{label} never reported readiness (exit: {status:?})\nstderr:\n{}",
+                fixture.stderr(label)
+            );
+        }
+    }
+    owner
+}
+
+fn http() -> reqwest::blocking::Client {
+    reqwest::blocking::Client::builder()
+        .no_proxy()
+        .timeout(Duration::from_secs(30))
+        .build()
+        .expect("loopback HTTP client")
+}
+
+#[test]
+fn killed_owner_is_replaced_by_the_next_app_server() {
+    let fixture = Fixture::new();
+    let client = http();
+
+    let mut first = spawn_owner(&fixture, "first");
+    let (first_bytes, first_receipt) = fixture.receipt();
+    assert_eq!(first_receipt["pid"], json!(first.child.id()));
+    let created = client
+        .post(format!("{}/v1/threads", first.base))
+        .bearer_auth(TOKEN)
+        .json(&json!({}))
+        .send()
+        .expect("create thread on the first owner");
+    let status = created.status();
+    let created: Value = created.json().expect("created thread JSON");
+    assert!(status.is_success(), "create thread: {status} {created}");
+    let thread_id = created["id"]
+        .as_str()
+        .expect("created thread id")
+        .to_string();
+
+    first.child.kill().expect("kill the first owner");
+    first.child.wait().expect("reap the first owner");
+    let (dead_bytes, dead_receipt) = fixture.receipt();
+    assert_eq!(dead_bytes, first_bytes);
+    assert_eq!(dead_receipt["pid"], json!(first.child.id()));
+
+    let second = spawn_owner(&fixture, "second");
+    let (_, second_receipt) = fixture.receipt();
+    assert_eq!(second_receipt["pid"], json!(second.child.id()));
+    assert_ne!(
+        second_receipt["lease_generation"],
+        first_receipt["lease_generation"]
+    );
+    assert_eq!(second_receipt["data_dir"], first_receipt["data_dir"]);
+    assert_eq!(
+        second_receipt["execution_scope"],
+        first_receipt["execution_scope"]
+    );
+
+    let listed: Value = client
+        .get(format!("{}/v1/threads", second.base))
+        .bearer_auth(TOKEN)
+        .send()
+        .expect("list threads on the second owner")
+        .json()
+        .expect("thread list JSON");
+    assert!(
+        listed
+            .as_array()
+            .expect("thread list")
+            .iter()
+            .any(|thread| thread["id"] == json!(thread_id)),
+        "{thread_id} missing from {listed}"
+    );
+
+    let attached = fixture
+        .command()
+        .args(["thread", "list"])
+        .stdin(Stdio::null())
+        .output()
+        .expect("run codewhale thread list");
+    let stdout = String::from_utf8_lossy(&attached.stdout);
+    assert!(
+        attached.status.success() && stdout.contains(&thread_id),
+        "thread list through the second owner failed ({})\nstdout:\n{stdout}\nstderr:\n{}",
+        attached.status,
+        String::from_utf8_lossy(&attached.stderr)
+    );
+}
@@ -776,10 +776,36 @@ fn reopen_windows_file_for_owner_security(file: &File, path: &Path) -> Result<Fi
 #[cfg(windows)]
 fn mark_windows_file_handle_for_deletion(file: &File) -> Result<()> {
    use std::os::windows::io::AsRawHandle as _;
+    use windows_sys::Win32::Foundation::{
+        ERROR_INVALID_FUNCTION, ERROR_INVALID_PARAMETER, ERROR_NOT_SUPPORTED,
+    };
    use windows_sys::Win32::Storage::FileSystem::{
-        FILE_DISPOSITION_INFO, FileDispositionInfo, SetFileInformationByHandle,
+        FILE_DISPOSITION_FLAG_DELETE, FILE_DISPOSITION_FLAG_POSIX_SEMANTICS, FILE_DISPOSITION_INFO,
+        FILE_DISPOSITION_INFO_EX, FileDispositionInfo, FileDispositionInfoEx,
+        SetFileInformationByHandle,
    };

+    let posix = FILE_DISPOSITION_INFO_EX {
+        Flags: FILE_DISPOSITION_FLAG_DELETE | FILE_DISPOSITION_FLAG_POSIX_SEMANTICS,
+    };
+    if unsafe {
+        SetFileInformationByHandle(
+            file.as_raw_handle(),
+            FileDispositionInfoEx,
+            (&raw const posix).cast(),
+            std::mem::size_of::<FILE_DISPOSITION_INFO_EX>() as u32,
+        )
+    } != 0
+    {
+        return Ok(());
+    }
+    let error = std::io::Error::last_os_error();
+    if !matches!(
+        error.raw_os_error().map(|code| code as u32),
+        Some(ERROR_INVALID_PARAMETER | ERROR_NOT_SUPPORTED | ERROR_INVALID_FUNCTION)
+    ) {
+        return Err(error).context("marking exact private file handle for deletion");
+    }
    let disposition = FILE_DISPOSITION_INFO { DeleteFile: true };
    // SAFETY: the disposition buffer has the documented structure and the
    // handle remains owned until after the call. Windows marks this exact file
@@ -794,7 +820,7 @@ fn mark_windows_file_handle_for_deletion(file: &File) -> Result<()> {
    } == 0
    {
        return Err(std::io::Error::last_os_error())
-            .context("marking exact xAI OAuth file handle for deletion");
+            .context("marking exact private file handle for deletion");
    }
    Ok(())
 }
@@ -181,6 +181,42 @@ impl WindowsPeerProcess {
    }
 }

+pub enum WindowsProcessStatus {
+    Running { start: String },
+    Exited { start: String },
+    Absent,
+}
+
+pub fn windows_process_status(pid: u32) -> Result<WindowsProcessStatus> {
+    use std::os::windows::io::AsRawHandle as _;
+    use windows_sys::Win32::Foundation::{ERROR_INVALID_PARAMETER, WAIT_OBJECT_0, WAIT_TIMEOUT};
+    use windows_sys::Win32::Storage::FileSystem::SYNCHRONIZE;
+    use windows_sys::Win32::System::Threading::{
+        GetProcessId, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION, WaitForSingleObject,
+    };
+    anyhow::ensure!(pid > 0, "invalid Windows process PID");
+    let process = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION | SYNCHRONIZE, 0, pid) };
+    if process.is_null() {
+        let error = std::io::Error::last_os_error();
+        if error.raw_os_error() == Some(ERROR_INVALID_PARAMETER as i32) {
+            return Ok(WindowsProcessStatus::Absent);
+        }
+        return Err(error).context("selected Windows process identity/liveness is uncertain");
+    }
+    let process = unsafe { std::os::windows::io::OwnedHandle::from_raw_handle(process) };
+    anyhow::ensure!(
+        unsafe { GetProcessId(process.as_raw_handle()) } == pid,
+        "Windows kernel process PID changed"
+    );
+    let start = process_creation(process.as_raw_handle())?;
+    match unsafe { WaitForSingleObject(process.as_raw_handle(), 0) } {
+        WAIT_OBJECT_0 => Ok(WindowsProcessStatus::Exited { start }),
+        WAIT_TIMEOUT => Ok(WindowsProcessStatus::Running { start }),
+        _ => Err(std::io::Error::last_os_error())
+            .context("selected Windows process identity/liveness is uncertain"),
+    }
+}
+
 fn process_creation(process: windows_sys::Win32::Foundation::HANDLE) -> Result<String> {
    use windows_sys::Win32::Foundation::FILETIME;
    use windows_sys::Win32::System::Threading::GetProcessTimes;
@@ -1639,7 +1639,7 @@ pub(crate) fn open_runtime_threads_for_host(
    Ok((manager, workshop_activation))
 }

-#[cfg(unix)]
+#[cfg(any(unix, windows))]
 enum RuntimeHostAdmission {
    Attached(Box<codewhale_app_server::daemon_client::OwnerClient>),
    Bootstrap {
@@ -1651,7 +1651,7 @@ enum RuntimeHostAdmission {

 /// Observe first, then hold the existing store lease through all recovery.
 /// Only pre-write publication changes or lease contention can re-observe.
-#[cfg(unix)]
+#[cfg(any(unix, windows))]
 async fn admit_runtime_host(
    config_path: Option<PathBuf>,
    selected_socket: Option<PathBuf>,
@@ -1735,14 +1735,14 @@ pub async fn run_http_server(
        Some(task_default_model.clone()),
        Some(options.workers),
    );
-    #[cfg(unix)]
+    #[cfg(any(unix, windows))]
    let admission = admit_runtime_host(
        options.config_path.clone(),
        selected_control_socket(&options),
        RuntimeThreadManagerConfig::from_task_data_dir(task_cfg.data_dir.clone()).data_dir,
    )
    .await?;
-    #[cfg(unix)]
+    #[cfg(any(unix, windows))]
    let (published, preheld_lock, existing_binding, recovery) = match admission {
        RuntimeHostAdmission::Attached(client) => (Some(*client), None, None, None),
        RuntimeHostAdmission::Bootstrap {
@@ -1751,18 +1751,10 @@ pub async fn run_http_server(
            recovery,
        } => (None, Some(lock), binding, recovery),
    };
-    #[cfg(windows)]
-    let published = codewhale_app_server::daemon_client::connect_if_published(
-        options.config_path.clone(),
-        selected_control_socket(&options),
-    )
-    .await?;
-    #[cfg(not(unix))]
+    #[cfg(not(any(unix, windows)))]
    let preheld_lock = None;
-    #[cfg(not(unix))]
+    #[cfg(not(any(unix, windows)))]
    let existing_binding: Option<crate::runtime_threads::RuntimeStoreBinding> = None;
-    #[cfg(windows)]
-    let recovery = None;
    #[cfg(any(unix, windows))]
    if let Some(control) = published {
        let selected_store =
@@ -6728,7 +6728,7 @@ impl RuntimeProcessOwnerLock {
    }

    /// Host admission reuses this exact store lease, never a PID-file lock.
-    #[cfg(unix)]
+    #[cfg(any(unix, windows))]
    pub(crate) fn try_acquire_for_host(root: &Path, create: bool) -> Result<Option<Self>> {
        if !create {
            checked_existing_runtime_store_dir(root)?;