fix(tunnel): keep embedded tunnels alive through heartbeat write failures

.github/workflows/ci.yml

@@ -797,6 +797,12 @@ jobs:
           if ($LASTEXITCODE -ne 0) { throw "Computer helper ownership tests failed" }
           cargo test --locked -p webcodex-environment --lib windows_helper_task_scheduler_roundtrip_is_owned -- --ignored
           if ($LASTEXITCODE -ne 0) { throw "Computer helper Task Scheduler roundtrip failed" }
+      - name: Test Windows Tunnel heartbeat contention and recovery
+        run: |
+          cargo test --locked -p webcodex-environment --lib tunnel::health_tests
+          if ($LASTEXITCODE -ne 0) { throw "Windows Tunnel health file tests failed" }
+          cargo test --locked -p webcodex --lib server_tunnels
+          if ($LASTEXITCODE -ne 0) { throw "Windows Tunnel heartbeat supervision tests failed" }
       - name: Test Windows unified installer routing and Desktop build freshness
         run: python -m unittest scripts.tests.test_prepare_windows_unified_installer scripts.tests.test_desktop_runtime_manifest scripts.tests.test_desktop_frontend_build
       - name: Test optional Codex observation adapter on Windows

crates/webcodex-environment/src/tunnel.rs

@@ -1893,7 +1893,7 @@ pub(crate) fn read_tunnel_health(
     })
 }
 
-#[cfg(all(test, unix))]
+#[cfg(test)]
 mod health_tests;
 
 #[cfg(test)]

crates/webcodex-environment/src/tunnel/health_tests.rs

@@ -1,6 +1,7 @@
 use super::*;
 
 #[test]
+#[cfg(unix)]
 fn health_locks_release_without_waiting_for_inherited_handles() {
     for exclusive in [false, true] {
         for fail_after_lock in [false, true] {
@@ -53,3 +54,45 @@ fn health_locks_release_without_waiting_for_inherited_handles() {
         }
     }
 }
+
+#[test]
+fn missed_heartbeats_expire_during_contention_and_recover_after_unlock() {
+    let temp = crate::test_tempdir().unwrap();
+    let store = EnvironmentStore::open(temp.path().join("environment")).unwrap();
+    let path = store.root().join("readiness.json");
+    let stale = serde_json::json!({
+        "schema_version": 1,
+        "observed_at_ms": 0,
+        "service_pid": std::process::id(),
+        "profile_revision": 7,
+        "state": "running",
+        "tunnel_ready": true,
+        "local_mcp_ready": true,
+    });
+    atomic_private_write(&path, stale.to_string().as_bytes()).unwrap();
+    let file = std::fs::File::open(&path).unwrap();
+    fs2::FileExt::try_lock_shared(&file).unwrap();
+    let reader = TunnelHealthFileLock { file };
+    for _ in 0..4 {
+        assert!(write_embedded_tunnel_observation(
+            &path,
+            7,
+            TunnelState::Running,
+            true,
+            true,
+            None,
+        )
+        .is_err());
+        let health = read_tunnel_health(&path).unwrap();
+        assert!(!health.fresh);
+        assert!(!health.tunnel_ready && !health.local_mcp_ready);
+        assert_eq!(health.state, Some(TunnelState::Running));
+        assert_eq!(health.profile_revision, Some(7));
+    }
+    drop(reader);
+    write_embedded_tunnel_observation(&path, 7, TunnelState::Running, true, true, None).unwrap();
+    let health = read_tunnel_health(&path).unwrap();
+    assert!(health.fresh && health.tunnel_ready && health.local_mcp_ready);
+    assert_eq!(health.service_pid, Some(std::process::id()));
+    assert_eq!(health.profile_revision, Some(7));
+}

docs/architecture/server-owned-tunnels.md

@@ -106,6 +106,16 @@ existing per-profile service for standalone. Readiness files keep the existing
 freshness contract. Transport readiness is not evidence of a successful client
 operation, and process liveness is not either readiness bit.
 
+The initial status write must succeed before a profile launches. Once its Tunnel
+task is owned, a heartbeat publication error (including a concurrent reader's
+shared file lock) does not end or restart that task. Runtime status keeps the live
+readiness bits and reports `tunnel_health_unavailable` until publication succeeds
+at a later normal two-second health tick. Only failure/recovery transitions are
+logged. A missed write does not refresh the saved observation: readers still
+reject it after seven seconds. The writer retains the precreated file, its private
+ACL and its nonblocking lock; it never replaces the file or bypasses validation.
+Task failure, startup deadlines, shutdown and restart-fence semantics are unchanged.
+
 Server termination runs these phases in order:
 
 1. Stop new Tunnel polling/admission.

src/server_tunnels.rs

@@ -259,11 +259,7 @@ async fn run_profile(
                     state.state = if reached_ready { TunnelState::Running } else { TunnelState::Starting };
                     state.tunnel_ready = ready; state.local_mcp_ready = local;
                 });
-                if persist_status(&profile, &status, index)
-                .is_err()
-                {
-                    break Some("tunnel_health_unavailable".into());
-                }
+                persist_heartbeat(&profile, &status, index);
             }
         }
     };
@@ -294,6 +290,33 @@ fn persist_status(
     index: usize,
 ) -> webcodex_environment::SetupResultValue<()> {
     let state = &status.snapshot()[index];
+    write_status(profile, state)
+}
+
+/// A saved heartbeat is an observation, not the owned task's execution state.
+/// Retry publication at the next normal tick without restarting or replaying work.
+/// Readers still reject stale observations using the existing freshness bound.
+fn persist_heartbeat(profile: &EmbeddedTunnelProfile, status: &TunnelStatus, index: usize) {
+    let mut state = status.snapshot()[index].clone();
+    let was_unavailable = state.diagnostic.as_deref() == Some("tunnel_health_unavailable");
+    state.diagnostic = None;
+    let unavailable = write_status(profile, &state).is_err();
+    status.update(index, |state| {
+        state.diagnostic = unavailable.then(|| "tunnel_health_unavailable".into());
+    });
+    // Log transitions, not every failed tick. Never include filesystem errors or
+    // private profile bindings in diagnostics.
+    if unavailable && !was_unavailable {
+        tracing::warn!(profile = %profile.profile_id, "Tunnel heartbeat unavailable; retrying at the next health tick");
+    } else if !unavailable && was_unavailable {
+        tracing::info!(profile = %profile.profile_id, "Tunnel heartbeat publication recovered");
+    }
+}
+
+fn write_status(
+    profile: &EmbeddedTunnelProfile,
+    state: &ProfileStatus,
+) -> webcodex_environment::SetupResultValue<()> {
     write_embedded_tunnel_observation(
         &profile.readiness_path,
         profile.runtime_revision,
@@ -303,3 +326,6 @@ fn persist_status(
         state.diagnostic.as_deref(),
     )
 }
+
+#[cfg(test)]
+mod tests;

src/server_tunnels/tests.rs

@@ -0,0 +1,145 @@
+use super::*;
+use webcodex_environment::{EnvironmentStore, Secret, TunnelCredentials};
+
+fn fixture() -> (tempfile::TempDir, EmbeddedTunnelProfile, TunnelStatus) {
+    let temp = tempfile::tempdir().unwrap();
+    let root = temp.path().canonicalize().unwrap().join("environment");
+    let store = EnvironmentStore::open(root).unwrap();
+    let path = store.root().join("readiness.json");
+    create_health_file(&path);
+    let profile = EmbeddedTunnelProfile {
+        profile_id: "primary".into(),
+        autostart: true,
+        runtime_revision: 7,
+        credentials: TunnelCredentials {
+            tunnel_id: Secret::new("tunnel_0123456789abcdef0123456789abcdef".into()),
+            api_key: Secret::new("fixture-control".into()),
+        },
+        local_token: Secret::new("fixture-local".into()),
+        proxy: None,
+        readiness_path: path,
+    };
+    let status = TunnelStatus(Arc::new(Mutex::new(vec![ProfileStatus {
+        profile_id: profile.profile_id.clone(),
+        host_mode: "embedded",
+        state: TunnelState::Running,
+        tunnel_ready: true,
+        local_mcp_ready: true,
+        shutdown_outcome: None,
+        diagnostic: None,
+    }])));
+    (temp, profile, status)
+}
+
+fn create_health_file(path: &Path) {
+    let mut options = std::fs::OpenOptions::new();
+    options.write(true).create_new(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt;
+        options.mode(0o600);
+    }
+    // On Windows the file inherits the EnvironmentStore directory's private ACL.
+    options.open(path).unwrap();
+}
+
+#[test]
+fn reader_contention_preserves_live_status_and_publication_recovers() {
+    let (_temp, profile, status) = fixture();
+    persist_status(&profile, &status, 0).unwrap();
+    let previous = std::fs::read(&profile.readiness_path).unwrap();
+    let reader = std::fs::File::open(&profile.readiness_path).unwrap();
+    reader.try_lock_shared().unwrap();
+    // A separately opened handle exercises the same lock conflict as a Desktop
+    // or CLI reader, deterministically and without timing or external services.
+    assert!(persist_status(&profile, &status, 0).is_err());
+    for _ in 0..4 {
+        persist_heartbeat(&profile, &status, 0);
+        let state = &status.snapshot()[0];
+        assert_eq!(state.state, TunnelState::Running);
+        assert!(state.tunnel_ready && state.local_mcp_ready);
+        assert_eq!(state.shutdown_outcome, None);
+        assert_eq!(
+            state.diagnostic.as_deref(),
+            Some("tunnel_health_unavailable")
+        );
+        assert_eq!(std::fs::read(&profile.readiness_path).unwrap(), previous);
+    }
+
+    // Live observations continue changing while publication is unavailable.
+    status.update(0, |state| state.local_mcp_ready = false);
+    persist_heartbeat(&profile, &status, 0);
+    assert!(!status.snapshot()[0].local_mcp_ready);
+    reader.unlock().unwrap();
+    persist_heartbeat(&profile, &status, 0);
+    let recovered: serde_json::Value =
+        serde_json::from_slice(&std::fs::read(&profile.readiness_path).unwrap()).unwrap();
+    assert_eq!(recovered["state"], "running");
+    assert_eq!(recovered["service_pid"], std::process::id());
+    assert_eq!(recovered["profile_revision"], 7);
+    assert_eq!(recovered["tunnel_ready"], true);
+    assert_eq!(recovered["local_mcp_ready"], false);
+    assert!(recovered["diagnostic"].is_null());
+    assert!(status.snapshot()[0].diagnostic.is_none());
+}
+
+#[test]
+fn missing_heartbeat_file_preserves_live_status_without_recreating_it() {
+    let (_temp, profile, status) = fixture();
+    std::fs::remove_file(&profile.readiness_path).unwrap();
+    persist_heartbeat(&profile, &status, 0);
+    assert_eq!(status.snapshot()[0].state, TunnelState::Running);
+    assert_eq!(
+        status.snapshot()[0].diagnostic.as_deref(),
+        Some("tunnel_health_unavailable")
+    );
+    assert!(!profile.readiness_path.exists());
+    // Only repair of the precreated private file allows publication to recover.
+    create_health_file(&profile.readiness_path);
+    persist_heartbeat(&profile, &status, 0);
+    assert!(status.snapshot()[0].diagnostic.is_none());
+
+    // A real terminal failure still owns the final diagnostic and lifecycle.
+    status.update(0, |state| {
+        state.state = TunnelState::Failed;
+        state.tunnel_ready = false;
+        state.local_mcp_ready = false;
+        state.shutdown_outcome = Some("uncertain");
+        state.diagnostic = Some("tunnel_restart_uncertain".into());
+    });
+    persist_status(&profile, &status, 0).unwrap();
+    let terminal: serde_json::Value =
+        serde_json::from_slice(&std::fs::read(&profile.readiness_path).unwrap()).unwrap();
+    assert_eq!(terminal["state"], "failed");
+    assert_eq!(terminal["diagnostic"], "tunnel_restart_uncertain");
+    assert_eq!(terminal["tunnel_ready"], false);
+    assert_eq!(terminal["local_mcp_ready"], false);
+}
+
+#[tokio::test]
+async fn unavailable_initial_status_still_blocks_profile_launch() {
+    let (_temp, profile, status) = fixture();
+    std::fs::remove_file(&profile.readiness_path).unwrap();
+    status.update(0, |state| {
+        state.state = TunnelState::Starting;
+        state.tunnel_ready = false;
+        state.local_mcp_ready = false;
+    });
+    let (_stop, rx) = watch::channel(false);
+    run_profile(
+        0,
+        profile,
+        "http://127.0.0.1:9/mcp".into(),
+        status.clone(),
+        rx,
+    )
+    .await;
+    let state = &status.snapshot()[0];
+    assert_eq!(state.state, TunnelState::Failed);
+    assert_eq!(
+        state.diagnostic.as_deref(),
+        Some("tunnel_health_unavailable")
+    );
+    assert!(!state.tunnel_ready && !state.local_mcp_ready);
+    assert_eq!(state.shutdown_outcome, None);
+}