Move the container exit watcher's Docker calls behind host::watch

container_watch keeps all of its logic (sweep, compare-and-set settle,
backoff, cancellation) but no longer touches bollard: reachability, the
death stream and the positive-exit check come from host::watch::ExitSource
and ContainerTarget::has_exited, implemented for Docker in docker/events.rs.
An AWS host can supply deaths by polling. container_watch.rs leaves the
boundary test's allowlist.

Verified with the live end-to-end watcher test against a real daemon.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/src-tauri/src/container_watch.rs

@@ -8,9 +8,10 @@
 //! disconnected, while the sidebar, Start/Stop and Project Home all still
 //! described a live container.
 //!
-//! One subscription to Docker's event stream, filtered daemon-side to container
-//! `die` events, so an idle daemon costs nothing and the app hears about a death
-//! as it happens. No polling loop. Each `die` naming a project's container is
+//! One subscription to the host's death reports (`host::watch`; for Docker,
+//! the daemon's event stream filtered to container `die` events, so an idle
+//! daemon costs nothing and the app hears about a death as it happens — no
+//! polling loop). Each `die` naming a project's container is
 //! confirmed with one inspect before anything changes (an event can be
 //! replayed, and a restart can reuse the id), and the flip itself is
 //! [`ProjectsStore::mark_exited`]'s compare-and-set, which never touches a
@@ -22,20 +23,17 @@
 //! the only window that replay cannot cover is a daemon restart, which forgets
 //! its event history, and that is exactly when containers go away.
 
-use std::collections::HashMap;
 use std::future::Future;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use bollard::errors::Error as DockerError;
-use bollard::models::{EventMessage, EventMessageTypeEnum};
-use bollard::system::EventsOptions;
-use bollard::Docker;
 use futures_util::StreamExt;
 use serde::Serialize;
 use tauri::{AppHandle, Emitter};
 use tokio::sync::watch;
 
+use crate::host::watch::ExitSource;
+use crate::host::ContainerTarget;
 use crate::models::{Project, ProjectStatus};
 use crate::storage::projects_store::{ExitMark, ProjectsStore};
 
@@ -72,46 +70,28 @@ pub struct ContainerExited<'a> {
     pub reason: ExitReason,
 }
 
-/// The container id a Docker event reports as having died, if it is one.
-pub(crate) fn died_container(msg: &EventMessage) -> Option<&str> {
-    if msg.typ != Some(EventMessageTypeEnum::CONTAINER) || msg.action.as_deref() != Some("die") {
-        return None;
-    }
-    msg.actor.as_ref()?.id.as_deref().filter(|id| !id.is_empty())
-}
-
-/// `true` only when Docker positively says the container is not running: it
-/// inspects as stopped, or it no longer exists. Any other failure is "don't
-/// know" — `docker::is_container_running` maps every error to `false`, which
-/// is right for its callers and wrong here, where it would mark every project
-/// Stopped the moment the daemon hiccups.
-async fn has_exited(docker: &Docker, container_id: &str) -> bool {
-    match docker.inspect_container(container_id, None).await {
-        Ok(info) => !info.state.and_then(|s| s.running).unwrap_or(false),
-        Err(DockerError::DockerResponseServerError { status_code: 404, .. }) => true,
-        Err(_) => false,
-    }
-}
-
 fn project_busy(project_id: &str) -> bool {
     crate::project_lock::held(project_id).is_some()
 }
 
-/// Confirm `container_id` really stopped and, if a project was Running on it,
+/// Confirm `container` really stopped and, if a project was Running on it,
 /// flip that project. Returns what changed.
-async fn settle(docker: &Docker, store: &ProjectsStore, container_id: &str) -> Option<Project> {
+async fn settle(store: &ProjectsStore, container: &ContainerTarget) -> Option<Project> {
+    let container_id = container.id();
     for _ in 0..SETTLE_ATTEMPTS {
         // Cheap and allocation-free: most containers that die are not ours
         // (probe and scratch containers, other software on the daemon), and
         // those cost no inspect at all.
         let seen = store.running_on(container_id)?;
-        if !has_exited(docker, container_id).await {
+        // Positive exit only: a host that can't answer must not mark every
+        // project Stopped.
+        if !container.has_exited().await {
             return None;
         }
         match store.mark_exited(container_id, &seen, project_busy) {
             ExitMark::Flipped(p) => return Some(*p),
             ExitMark::NotApplicable => return None,
-            // The store moved while Docker was answering — possibly a stop and
+            // The store moved while the host was answering — possibly a stop and
             // start of this very container. The inspect is stale; ask again.
             ExitMark::Changed => continue,
         }
@@ -120,27 +100,20 @@ async fn settle(docker: &Docker, store: &ProjectsStore, container_id: &str) -> O
 }
 
 /// Re-check every Running project once. Used on each (re)subscription.
-async fn sweep(docker: &Docker, store: &ProjectsStore) -> Vec<Project> {
+async fn sweep(store: &ProjectsStore) -> Vec<Project> {
     let mut changed = Vec::new();
     for p in store.list() {
         if p.status != ProjectStatus::Running {
             continue;
         }
-        let Some(cid) = p.container_id.as_deref() else { continue };
-        if let Some(c) = settle(docker, store, cid).await {
+        let Some(container) = ContainerTarget::for_project(&p) else { continue };
+        if let Some(c) = settle(store, &container).await {
             changed.push(c);
         }
     }
     changed
 }
 
-fn die_filter() -> HashMap<String, Vec<String>> {
-    HashMap::from([
-        ("type".to_string(), vec!["container".to_string()]),
-        ("event".to_string(), vec!["die".to_string()]),
-    ])
-}
-
 /// The wait before connection attempt `n + 1`, after `n` consecutive
 /// unproductive ones: 5 s, 10 s, 20 s, 40 s, then 60 s for good.
 fn retry_delay(consecutive_failures: u32) -> Duration {
@@ -148,7 +121,7 @@ fn retry_delay(consecutive_failures: u32) -> Duration {
     (RETRY_MIN * 2u32.pow(doublings)).min(RETRY_MAX)
 }
 
-/// `fut`, unless the app starts shutting down first. Every Docker call in here
+/// `fut`, unless the app starts shutting down first. Every host call in here
 /// goes through this: a wedged daemon must not hold shutdown for its full
 /// budget.
 async fn unless_cancelled<T>(
@@ -174,32 +147,26 @@ async fn watch_once(
     sweep_reason: ExitReason,
     on_change: &impl Fn(&Project, ExitReason),
 ) -> Option<bool> {
-    let Ok(docker) = crate::docker::get_docker() else { return Some(false) };
-    if unless_cancelled(cancel, docker.ping()).await?.is_err() {
+    let source = ExitSource::Docker;
+    if !unless_cancelled(cancel, source.reachable()).await? {
         return Some(false);
     }
-    // A second's overlap: the daemon replays from `since`, so a death during
+    // A second's overlap: the host replays from `since`, so a death during
     // the sweep below is still delivered. Duplicates are harmless — `settle`
     // confirms with an inspect every time.
-    let since = (chrono::Utc::now().timestamp() - 1).to_string();
-    for p in unless_cancelled(cancel, sweep(&docker, store)).await? {
+    let since = chrono::Utc::now().timestamp() - 1;
+    for p in unless_cancelled(cancel, sweep(store)).await? {
         on_change(&p, sweep_reason);
     }
-    let mut events = docker.events(Some(EventsOptions::<String> {
-        since: Some(since),
-        until: None,
-        filters: die_filter(),
-    }));
+    let mut deaths = source.deaths(since);
     let connected = Instant::now();
     let mut delivered = false;
     loop {
-        match unless_cancelled(cancel, events.next()).await? {
-            Some(Ok(msg)) => {
+        match unless_cancelled(cancel, deaths.next()).await? {
+            Some(Ok(container)) => {
                 delivered = true;
-                if let Some(cid) = died_container(&msg) {
-                    if let Some(p) = unless_cancelled(cancel, settle(&docker, store, cid)).await? {
-                        on_change(&p, ExitReason::Exited);
-                    }
+                if let Some(p) = unless_cancelled(cancel, settle(store, &container)).await? {
+                    on_change(&p, ExitReason::Exited);
                 }
             }
             Some(Err(e)) => {
@@ -265,41 +232,6 @@ pub fn spawn(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use bollard::models::EventActor;
-
-    fn event(typ: EventMessageTypeEnum, action: &str, id: Option<&str>) -> EventMessage {
-        EventMessage {
-            typ: Some(typ),
-            action: Some(action.to_string()),
-            actor: Some(EventActor { id: id.map(str::to_string), attributes: None }),
-            ..Default::default()
-        }
-    }
-
-    #[test]
-    fn only_a_container_die_names_a_container() {
-        let die = event(EventMessageTypeEnum::CONTAINER, "die", Some("abc"));
-        assert_eq!(died_container(&die), Some("abc"));
-
-        for (typ, action, id) in [
-            (EventMessageTypeEnum::CONTAINER, "stop", Some("abc")),
-            (EventMessageTypeEnum::CONTAINER, "start", Some("abc")),
-            (EventMessageTypeEnum::CONTAINER, "exec_die", Some("abc")),
-            (EventMessageTypeEnum::NETWORK, "die", Some("abc")),
-            (EventMessageTypeEnum::CONTAINER, "die", None),
-            (EventMessageTypeEnum::CONTAINER, "die", Some("")),
-        ] {
-            assert_eq!(died_container(&event(typ, action, id)), None, "{typ:?} {action} {id:?}");
-        }
-    }
-
-    #[test]
-    fn the_subscription_is_filtered_to_container_deaths_daemon_side() {
-        let f = die_filter();
-        assert_eq!(f.get("type"), Some(&vec!["container".to_string()]));
-        assert_eq!(f.get("event"), Some(&vec!["die".to_string()]));
-        assert_eq!(f.len(), 2);
-    }
 
     #[test]
     fn retries_back_off_to_a_minute_and_stay_there() {
@@ -321,7 +253,7 @@ mod tests {
         assert_eq!(CONTAINER_EXITED_EVENT, "project-container-exited");
     }
 
-    /// The shutdown path: a Docker call that never answers (a wedged daemon)
+    /// The shutdown path: a host call that never answers (a wedged daemon)
     /// must give way to cancellation at once, not hold teardown.
     #[tokio::test]
     async fn a_hung_docker_call_yields_to_cancellation() {

app/src-tauri/src/docker/events.rs

@@ -0,0 +1,114 @@
+//! The Docker half of `container_watch`: reachability, the daemon's container
+//! `die` event stream, and the "has it positively exited" check. Everything
+//! about *what to do* with a death (confirm, compare-and-set the store, tell
+//! the frontend, back off) is host-neutral and stays in `container_watch.rs`,
+//! which reaches these through `host::watch`.
+
+use std::collections::HashMap;
+use std::pin::Pin;
+
+use bollard::errors::Error as DockerError;
+use bollard::models::{EventMessage, EventMessageTypeEnum};
+use bollard::system::EventsOptions;
+use futures_util::{Stream, StreamExt};
+
+use super::client::get_docker;
+
+/// Whether the daemon answers at all.
+pub async fn ping() -> bool {
+    match get_docker() {
+        Ok(docker) => docker.ping().await.is_ok(),
+        Err(_) => false,
+    }
+}
+
+/// The container id a Docker event reports as having died, if it is one.
+pub(crate) fn died_container(msg: &EventMessage) -> Option<&str> {
+    if msg.typ != Some(EventMessageTypeEnum::CONTAINER) || msg.action.as_deref() != Some("die") {
+        return None;
+    }
+    msg.actor.as_ref()?.id.as_deref().filter(|id| !id.is_empty())
+}
+
+fn die_filter() -> HashMap<String, Vec<String>> {
+    HashMap::from([
+        ("type".to_string(), vec!["container".to_string()]),
+        ("event".to_string(), vec!["die".to_string()]),
+    ])
+}
+
+/// Container ids as Docker reports them dying, replayed from `since` (unix
+/// seconds). Filtered daemon-side to container `die` events, so an idle daemon
+/// costs nothing. Anything else that comes through is dropped here. An `Err`
+/// item, or the stream ending, means the connection is gone.
+pub fn container_deaths(since: i64) -> Pin<Box<dyn Stream<Item = Result<String, String>> + Send>> {
+    let docker = match get_docker() {
+        Ok(d) => d,
+        Err(e) => return Box::pin(futures_util::stream::once(async move { Err(e) })),
+    };
+    let events = docker.events(Some(EventsOptions::<String> {
+        since: Some(since.to_string()),
+        until: None,
+        filters: die_filter(),
+    }));
+    Box::pin(events.filter_map(|item| async move {
+        match item {
+            Ok(msg) => died_container(&msg).map(|id| Ok(id.to_string())),
+            Err(e) => Some(Err(e.to_string())),
+        }
+    }))
+}
+
+/// `true` only when Docker positively says the container is not running: it
+/// inspects as stopped, or it no longer exists. Any other failure is "don't
+/// know" — `is_container_running` maps every error to `false`, which is right
+/// for its callers and wrong for the watcher, where it would mark every project
+/// Stopped the moment the daemon hiccups.
+pub async fn has_exited(container_id: &str) -> bool {
+    let Ok(docker) = get_docker() else { return false };
+    match docker.inspect_container(container_id, None).await {
+        Ok(info) => !info.state.and_then(|s| s.running).unwrap_or(false),
+        Err(DockerError::DockerResponseServerError { status_code: 404, .. }) => true,
+        Err(_) => false,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use bollard::models::EventActor;
+
+    fn event(typ: EventMessageTypeEnum, action: &str, id: Option<&str>) -> EventMessage {
+        EventMessage {
+            typ: Some(typ),
+            action: Some(action.to_string()),
+            actor: Some(EventActor { id: id.map(str::to_string), attributes: None }),
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn only_a_container_die_names_a_container() {
+        let die = event(EventMessageTypeEnum::CONTAINER, "die", Some("abc"));
+        assert_eq!(died_container(&die), Some("abc"));
+
+        for (typ, action, id) in [
+            (EventMessageTypeEnum::CONTAINER, "stop", Some("abc")),
+            (EventMessageTypeEnum::CONTAINER, "start", Some("abc")),
+            (EventMessageTypeEnum::CONTAINER, "exec_die", Some("abc")),
+            (EventMessageTypeEnum::NETWORK, "die", Some("abc")),
+            (EventMessageTypeEnum::CONTAINER, "die", None),
+            (EventMessageTypeEnum::CONTAINER, "die", Some("")),
+        ] {
+            assert_eq!(died_container(&event(typ, action, id)), None, "{typ:?} {action} {id:?}");
+        }
+    }
+
+    #[test]
+    fn the_subscription_is_filtered_to_container_deaths_daemon_side() {
+        let f = die_filter();
+        assert_eq!(f.get("type"), Some(&vec!["container".to_string()]));
+        assert_eq!(f.get("event"), Some(&vec!["die".to_string()]));
+        assert_eq!(f.len(), 2);
+    }
+}

app/src-tauri/src/docker/mod.rs

@@ -1,5 +1,6 @@
 pub mod ca_certs;
 pub mod client;
+pub mod events;
 pub mod container;
 pub mod image;
 pub mod exec;

app/src-tauri/src/host/mod.rs

@@ -27,6 +27,7 @@
 //! may still do so.
 
 pub mod sessions;
+pub mod watch;
 
 pub use crate::docker::exec::{
     MAX_DROP_BYTES, OUTPUT_LIMIT_MARKER, PROC_NET_OUTPUT_LIMIT,
@@ -432,8 +433,6 @@ mod tests {
         const ALLOWED: &[&str] = &[
             // Docker-only: base-image migration replays apt / restores a payload.
             "commands/migration_commands.rs",
-            // Phase 1b: lifecycle and the `die` event stream.
-            "container_watch.rs",
         ];
         let src = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
         let mut offenders = Vec::new();

app/src-tauri/src/host/watch.rs

@@ -0,0 +1,55 @@
+//! Where `container_watch` learns that a project container stopped on its
+//! own. Host-neutral shape: a reachability check, a stream of containers
+//! reported dead (replayable from a point in time), and a positive-exit check
+//! used to confirm each report before anything changes.
+//!
+//! Docker pushes `die` events. An AWS host will have no push stream from the
+//! app's point of view; its [`deaths`] will be a `DescribeTasks` poll that
+//! yields stopped tasks, which `container_watch` handles the same way because
+//! every report is confirmed with [`ContainerTarget::has_exited`] anyway.
+
+use futures_util::{Stream, StreamExt};
+use std::pin::Pin;
+
+use super::ContainerTarget;
+use crate::docker;
+
+/// Which host's deaths to watch. One variant today.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ExitSource {
+    Docker,
+}
+
+pub type DeathStream = Pin<Box<dyn Stream<Item = Result<ContainerTarget, String>> + Send>>;
+
+impl ExitSource {
+    /// Whether the host answers at all. A `false` here is a failed connection
+    /// attempt for the watcher's backoff.
+    pub async fn reachable(&self) -> bool {
+        match self {
+            ExitSource::Docker => docker::events::ping().await,
+        }
+    }
+
+    /// Containers reported dead, replayed from `since` (unix seconds). An `Err`
+    /// item, or the stream ending, means the connection is gone.
+    pub fn deaths(&self, since: i64) -> DeathStream {
+        match self {
+            ExitSource::Docker => Box::pin(
+                docker::events::container_deaths(since).map(|r| r.map(ContainerTarget::docker)),
+            ),
+        }
+    }
+}
+
+impl ContainerTarget {
+    /// `true` only when the host positively says the container is not running
+    /// (stopped, or gone). Any failure to find out is `false` — unlike
+    /// [`ContainerTarget::is_running`], whose callers want errors to read as
+    /// "not running".
+    pub async fn has_exited(&self) -> bool {
+        match self {
+            ContainerTarget::Docker { container_id } => docker::events::has_exited(container_id).await,
+        }
+    }
+}