feat(console): drive resident timers and TCP streams

Add a revision-scoped ResidentSupervisor that serializes Host events through ResidentSession while Tokio tasks own timers, TCP listeners, and stream sockets.

Enforce deny-by-default endpoint policy, bounded event and stream queues, startup rollback, deterministic service shutdown, stale Actor replacement, and deployment restoration across Runtime restarts.

Add timer and real TCP integration coverage for port conflicts, multi-chunk echo traffic, service restart, and full Runtime recovery.

Cargo.toml

@@ -33,7 +33,7 @@ sha2 = "0.10.9"
 thiserror = "2.0.17"
 toasty = "=0.9.0"
 toasty-driver-turso = "=0.9.0"
-tokio = { version = "1.53.1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] }
+tokio = { version = "1.53.1", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] }
 toml = "0.9.8"
 tower-http = { version = "0.7.0", features = ["cors", "trace"] }
 tracing = "0.1.44"

components/resident-probe/src/lib.rs

@@ -70,9 +70,12 @@ impl ResidentGuest for ResidentProbe {
                 command: source.kind,
                 payload: source.payload,
             })],
+            // A TCP FIN is a read-half close. This fixture has no additional
+            // response to send, so it explicitly asks the Host to close the
+            // remaining write half and release the stream resource.
+            Event::StreamHalfClosed(stream_id) => vec![Effect::CloseStream(stream_id)],
             Event::StreamOpened(_)
             | Event::StreamWritable(_)
-            | Event::StreamHalfClosed(_)
             | Event::StreamClosed(_)
             | Event::Shutdown => Vec::new(),
         })

crates/wasmeld-console/src/api.rs

@@ -585,7 +585,8 @@ impl IntoResponse for GatewayError {
                     | ConsoleError::InvalidDatabaseData(_)
                     | ConsoleError::Runtime(_)
                     | ConsoleError::Join(_)
-                    | ConsoleError::LockPoisoned => (
+                    | ConsoleError::LockPoisoned
+                    | ConsoleError::ResidentSupervisor(_) => (
                         StatusCode::INTERNAL_SERVER_ERROR,
                         "internal_error",
                         "internal gateway error".to_owned(),
@@ -680,7 +681,10 @@ impl IntoResponse for ApiError {
             | ConsoleError::WitRegistry(WitRegistryError::Storage { .. })
             | ConsoleError::WitRegistry(WitRegistryError::LockPoisoned)
             | ConsoleError::Join(_)
-            | ConsoleError::LockPoisoned => (StatusCode::INTERNAL_SERVER_ERROR, "internal_error"),
+            | ConsoleError::LockPoisoned
+            | ConsoleError::ResidentSupervisor(_) => {
+                (StatusCode::INTERNAL_SERVER_ERROR, "internal_error")
+            }
         };
 
         (

crates/wasmeld-console/src/lib.rs

@@ -36,11 +36,12 @@ mod api;
 mod invocation_persistence;
 mod kv_backend;
 mod persistence;
+mod resident_supervisor;
 mod web;
 mod wit_registry;
 
 use std::{
-    collections::{BTreeMap, VecDeque},
+    collections::{BTreeMap, HashMap, VecDeque},
     fs, io,
     path::{Path, PathBuf},
     sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard},
@@ -54,14 +55,19 @@ use wasmeld_package::{
     ComponentPackage, PackageError, read_package, wit_package::WitPackageMetadata,
 };
 use wasmeld_runtime::{
-    CapabilityDescriptor, ResourceLimits, Runtime, RuntimeConfig, RuntimeError, ServiceKey,
-    ServiceManifest,
+    ActorHandle, CapabilityDescriptor, ComponentExecution, ResourceLimits, Runtime, RuntimeConfig,
+    RuntimeError, ServiceKey, ServiceManifest,
 };
 
 pub use api::{app, gateway_app};
 use invocation_persistence::InvocationPersistence;
 use kv_backend::ToastyKvBackend;
 use persistence::{InvocationUpdate, Persistence, StoredDeployment, StoredEvent, StoredService};
+pub use resident_supervisor::{
+    ResidentServiceConfig, ResidentTcpListenerConfig, ResidentTimerConfig,
+};
+use resident_supervisor::{ResidentSupervisor, ResidentSupervisorError};
+pub use wasmeld_runtime::{NetworkScope, ResidentPolicy};
 use wit_registry::WitRegistry;
 pub use wit_registry::WitRegistryError;
 
@@ -87,6 +93,12 @@ pub struct ConsoleConfig {
     pub registration_limits: ResourceLimits,
     /// Wasmtime Engine and epoch settings.
     pub runtime: RuntimeConfig,
+    /// Host-owned resident resources keyed by stable service ID.
+    ///
+    /// Each started revision receives an independent copy of this
+    /// configuration and therefore never shares resource IDs, socket handles,
+    /// or driver tasks with another revision.
+    pub resident_services: BTreeMap<String, ResidentServiceConfig>,
 }
 
 impl Default for ConsoleConfig {
@@ -99,6 +111,7 @@ impl Default for ConsoleConfig {
             max_wit_package_bytes: DEFAULT_MAX_WIT_PACKAGE_BYTES,
             registration_limits: ResourceLimits::default(),
             runtime: RuntimeConfig::default(),
+            resident_services: BTreeMap::new(),
         }
     }
 }
@@ -115,6 +128,9 @@ pub struct Console {
     // to avoid deadlocks between HTTP control operations.
     runtime: RwLock<Option<Runtime>>,
     runtime_config: RuntimeConfig,
+    async_runtime: tokio::runtime::Handle,
+    resident_services: BTreeMap<String, ResidentServiceConfig>,
+    resident_supervisors: Mutex<HashMap<ServiceKey, ResidentSupervisor>>,
     artifact_dir: PathBuf,
     wit_registry: WitRegistry,
     max_artifact_bytes: usize,
@@ -486,6 +502,10 @@ pub enum ConsoleError {
     /// An internal lifecycle, state, or Runtime lock was poisoned.
     #[error("console state lock was poisoned")]
     LockPoisoned,
+
+    /// A resident Host driver could not start, dispatch, or drain.
+    #[error("resident supervisor failed: {0}")]
+    ResidentSupervisor(String),
 }
 
 impl Console {
@@ -495,6 +515,12 @@ impl Console {
     /// configuration. Their previous Store and linear memory are not restored.
     pub async fn new(config: ConsoleConfig) -> Result<Self, ConsoleError> {
         config.registration_limits.validate()?;
+        for (service_id, resident) in &config.resident_services {
+            validate_path_segment("resident service id", service_id)?;
+            resident
+                .validate()
+                .map_err(ConsoleError::ResidentSupervisor)?;
+        }
         fs::create_dir_all(&config.artifact_dir).map_err(|source| ConsoleError::Storage {
             path: config.artifact_dir.clone(),
             source,
@@ -547,6 +573,9 @@ impl Console {
         let console = Self {
             runtime: RwLock::new(Some(runtime)),
             runtime_config,
+            async_runtime: tokio::runtime::Handle::current(),
+            resident_services: config.resident_services,
+            resident_supervisors: Mutex::new(HashMap::new()),
             artifact_dir,
             wit_registry,
             max_artifact_bytes: config.max_artifact_bytes,
@@ -671,6 +700,16 @@ impl Console {
             Ok(started) => {
                 *runtime = Some(started);
                 drop(runtime);
+                if let Err(error) = self.restore_deployments() {
+                    self.rollback_runtime_start();
+                    self.mark_runtime_stopped()?;
+                    self.push_event(
+                        EventKind::Failed,
+                        None,
+                        format!("Wasmeld Runtime failed to restore deployments: {error}"),
+                    )?;
+                    return Err(error);
+                }
                 self.mark_runtime_started()?;
                 self.push_event(
                     EventKind::Started,
@@ -704,6 +743,7 @@ impl Console {
             drop(runtime);
             return self.runtime_view();
         };
+        self.shutdown_all_resident_supervisors();
         let stopped = running.stop_all();
         drop(runtime);
         self.mark_runtime_stopped()?;
@@ -738,23 +778,34 @@ impl Console {
             .write()
             .map_err(|_| ConsoleError::LockPoisoned)?;
 
-        if let Some(running) = runtime.take()
-            && let Err(error) = running.stop_all()
-        {
-            drop(runtime);
-            self.mark_runtime_stopped()?;
-            self.push_event(
-                EventKind::Failed,
-                None,
-                format!("Wasmeld Runtime failed to restart while stopping actors: {error}"),
-            )?;
-            return Err(error.into());
+        if let Some(running) = runtime.take() {
+            self.shutdown_all_resident_supervisors();
+            if let Err(error) = running.stop_all() {
+                drop(runtime);
+                self.mark_runtime_stopped()?;
+                self.push_event(
+                    EventKind::Failed,
+                    None,
+                    format!("Wasmeld Runtime failed to restart while stopping actors: {error}"),
+                )?;
+                return Err(error.into());
+            }
         }
 
         match self.build_runtime() {
             Ok(started) => {
                 *runtime = Some(started);
                 drop(runtime);
+                if let Err(error) = self.restore_deployments() {
+                    self.rollback_runtime_start();
+                    self.mark_runtime_stopped()?;
+                    self.push_event(
+                        EventKind::Failed,
+                        None,
+                        format!("Wasmeld Runtime failed to restore deployments: {error}"),
+                    )?;
+                    return Err(error);
+                }
                 self.mark_runtime_started()?;
                 self.push_event(
                     EventKind::Started,
@@ -783,25 +834,120 @@ impl Console {
             .values()
             .map(|service| service.manifest.clone())
             .collect::<Vec<_>>();
-        let deployments = state
-            .deployments
-            .iter()
-            .map(|(service_id, deployment)| {
-                ServiceKey::new(service_id.clone(), deployment.active_revision.clone())
-            })
-            .collect::<Result<Vec<_>, _>>()?;
         drop(state);
 
         let runtime = Runtime::new(self.runtime_config.clone())?;
         for manifest in manifests {
             runtime.register_from_file(manifest)?;
         }
-        for key in deployments {
-            runtime.start(&key, Vec::new())?;
-        }
         Ok(runtime)
     }
 
+    fn start_actor(
+        &self,
+        runtime: &Runtime,
+        key: &ServiceKey,
+        init_config: Vec<u8>,
+    ) -> Result<ActorHandle, ConsoleError> {
+        let actor = runtime.start(key, init_config)?;
+        if let Err(error) = self.attach_resident_supervisor(actor.clone()) {
+            // A resident Actor without its Host driver would be running in a
+            // misleading half-state. Roll it back before surfacing the error.
+            let _ = runtime.stop(key);
+            return Err(error);
+        }
+        Ok(actor)
+    }
+
+    fn attach_resident_supervisor(&self, actor: ActorHandle) -> Result<(), ConsoleError> {
+        if actor.execution() != ComponentExecution::Resident {
+            return Ok(());
+        }
+        let key = actor.key().clone();
+        let mut supervisors = self
+            .resident_supervisors
+            .lock()
+            .map_err(|_| ConsoleError::LockPoisoned)?;
+        if supervisors
+            .get(&key)
+            .is_some_and(|supervisor| supervisor.owns_actor(&actor))
+        {
+            return Ok(());
+        }
+        if let Some(mut stale) = supervisors.remove(&key)
+            && let Err(error) = stale.shutdown()
+        {
+            tracing::warn!(
+                service = %key,
+                %error,
+                "stale resident supervisor did not drain cleanly before Actor replacement"
+            );
+        }
+        let config = self
+            .resident_services
+            .get(key.id())
+            .cloned()
+            .unwrap_or_default();
+        let command_capacity = actor.mailbox_capacity();
+        let supervisor =
+            ResidentSupervisor::start(actor, config, self.async_runtime.clone(), command_capacity)
+                .map_err(supervisor_error)?;
+        supervisors.insert(key, supervisor);
+        Ok(())
+    }
+
+    fn shutdown_resident_supervisor(&self, key: &ServiceKey) {
+        let supervisor = self
+            .resident_supervisors
+            .lock()
+            .map(|mut supervisors| supervisors.remove(key));
+        match supervisor {
+            Ok(Some(mut supervisor)) => {
+                if let Err(error) = supervisor.shutdown() {
+                    tracing::warn!(service = %key, %error, "resident supervisor did not drain cleanly");
+                }
+            }
+            Ok(None) => {}
+            Err(_) => tracing::warn!(service = %key, "resident supervisor lock was poisoned"),
+        }
+    }
+
+    fn shutdown_all_resident_supervisors(&self) {
+        let supervisors = self
+            .resident_supervisors
+            .lock()
+            .map(|mut supervisors| std::mem::take(&mut *supervisors));
+        match supervisors {
+            Ok(supervisors) => {
+                for (key, mut supervisor) in supervisors {
+                    if let Err(error) = supervisor.shutdown() {
+                        tracing::warn!(service = %key, %error, "resident supervisor did not drain cleanly");
+                    }
+                }
+            }
+            Err(_) => tracing::warn!("resident supervisor lock was poisoned"),
+        }
+    }
+
+    fn stop_actor(&self, runtime: &Runtime, key: &ServiceKey) -> Result<(), RuntimeError> {
+        self.shutdown_resident_supervisor(key);
+        runtime.stop(key)
+    }
+
+    fn rollback_runtime_start(&self) {
+        self.shutdown_all_resident_supervisors();
+        match self.runtime.write() {
+            Ok(mut runtime) => {
+                if let Some(runtime) = runtime.take()
+                    && let Err(error) = runtime.stop_all()
+                {
+                    tracing::warn!(%error, "failed to stop Runtime during start rollback");
+                }
+            }
+            Err(_) => tracing::warn!("Runtime lock was poisoned during start rollback"),
+        }
+    }
+
     fn mark_runtime_started(&self) -> Result<(), ConsoleError> {
         let mut state = self.state()?;
         let deployed = state
@@ -931,6 +1077,7 @@ impl Console {
 
         let runtime = self.runtime()?;
         let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
+        self.shutdown_resident_supervisor(key);
         runtime.unregister(key)?;
         self.state()?.services.remove(key);
 
@@ -956,7 +1103,7 @@ impl Console {
         let runtime = self.runtime()?;
         let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
         self.ensure_managed(key)?;
-        runtime.start(key, init_config)?;
+        self.start_actor(runtime, key, init_config)?;
         let view = self.update_status(key, ServiceStatus::Running)?;
         self.push_event(EventKind::Started, Some(key), "actor started".to_owned())?;
         Ok(view)
@@ -966,7 +1113,7 @@ impl Console {
         let runtime = self.runtime()?;
         let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
         self.ensure_managed(key)?;
-        runtime.stop(key)?;
+        self.stop_actor(runtime, key)?;
         let view = self.update_status(key, ServiceStatus::Stopped)?;
         self.push_event(EventKind::Stopped, Some(key), "actor stopped".to_owned())?;
         Ok(view)
@@ -976,11 +1123,11 @@ impl Console {
         let runtime = self.runtime()?;
         let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
         self.ensure_managed(key)?;
-        match runtime.stop(key) {
+        match self.stop_actor(runtime, key) {
             Ok(()) | Err(RuntimeError::ActorUnavailable(_)) => {}
             Err(error) => return Err(error.into()),
         }
-        runtime.start(key, init_config)?;
+        self.start_actor(runtime, key, init_config)?;
         let view = self.update_status(key, ServiceStatus::Running)?;
         self.push_event(EventKind::Started, Some(key), "actor restarted".to_owned())?;
         Ok(view)
@@ -1006,7 +1153,7 @@ impl Console {
         // Starting is idempotent. The old revision stays resident so calls
         // that resolved immediately before this switch can finish safely and
         // rollback does not require recompilation.
-        runtime.start(&key, init_config)?;
+        self.start_actor(runtime, &key, init_config)?;
 
         let updated_at_ms = unix_time_ms();
         let view = {
@@ -1221,7 +1368,7 @@ impl Console {
             let runtime = self.runtime()?;
             let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
             for key in &deployments {
-                runtime.start(key, Vec::new())?;
+                self.start_actor(runtime, key, Vec::new())?;
             }
         }
 
@@ -1362,6 +1509,22 @@ impl Console {
     }
 }
 
+impl Drop for Console {
+    fn drop(&mut self) {
+        self.shutdown_all_resident_supervisors();
+        if let Ok(runtime) = self.runtime.get_mut()
+            && let Some(runtime) = runtime.take()
+            && let Err(error) = runtime.stop_all()
+        {
+            tracing::warn!(%error, "failed to stop Runtime while dropping Console");
+        }
+    }
+}
+
+fn supervisor_error(error: ResidentSupervisorError) -> ConsoleError {
+    ConsoleError::ResidentSupervisor(error.to_string())
+}
+
 fn runtime_capabilities(
     runtime: &Runtime,
     key: &ServiceKey,

crates/wasmeld-console/src/resident_supervisor.rs

@@ -0,0 +1,877 @@
+//! Revision-scoped execution of Host-owned resident event sources.
+//!
+//! The Wasmtime Actor API is intentionally synchronous: every call waits for a
+//! bounded mailbox response. Running [`ResidentSession`] directly from a Tokio
+//! task would therefore block an async worker for up to the Component deadline.
+//! This module keeps one session on a dedicated supervisor thread. Async
+//! drivers only wait for external activity and use a bounded channel to submit
+//! facts such as a timer deadline, accepted TCP connection, or stream chunk.
+//!
+//! ```text
+//! timer / listener / stream task --> bounded queue --> supervisor thread
+//!          ^                                      |
+//!          |-------- validated operations --------|
+//! ```
+//!
+//! The supervisor is scoped to one immutable service revision. Shutdown first
+//! prevents new driver events, then delivers the resident `shutdown` callback,
+//! aborts every Host task, closes socket handles, releases Host resource IDs,
+//! and only then allows the Console to stop the Actor.
+
+mod tcp;
+
+use std::{
+    collections::{HashMap, VecDeque},
+    io,
+    net::SocketAddr,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+        mpsc::{self, Receiver, SyncSender, TrySendError},
+    },
+    thread::{self, JoinHandle as ThreadJoinHandle},
+    time::{Duration, Instant},
+};
+
+use thiserror::Error;
+use tokio::{net::TcpStream, runtime::Handle, task::JoinHandle as TaskJoinHandle};
+use wasmeld_runtime::{
+    ActorHandle, ResidentEndpoint, ResidentHostError, ResidentOperation, ResidentPolicy,
+    ResidentSession, ResourceId, ServiceKey, StreamCloseReason,
+};
+
+pub use tcp::ResidentTcpListenerConfig;
+use tcp::{TcpDriverError, TcpEndpointDriver, TcpStreamCommand, TcpStreamDriver};
+
+/// Platform configuration for Host resources attached to one service ID.
+///
+/// A configuration is inherited by every started revision of that service, but
+/// each revision receives independent Host handles and resource IDs.
+#[derive(Clone, Debug, Default)]
+pub struct ResidentServiceConfig {
+    /// Defense-in-depth policy checked before any Host endpoint is bound.
+    ///
+    /// The default denies every network endpoint. Adding a TCP listener also
+    /// requires setting `tcp_listen` to `loopback` or `any`.
+    pub policy: ResidentPolicy,
+    /// Host-owned TCP listeners registered when a resident Actor starts.
+    pub tcp_listeners: Vec<ResidentTcpListenerConfig>,
+    /// Host-owned timers registered when a resident Actor starts.
+    pub timers: Vec<ResidentTimerConfig>,
+}
+
+impl ResidentServiceConfig {
+    pub(crate) fn validate(&self) -> Result<(), String> {
+        let mut names = std::collections::BTreeSet::new();
+        for listener in &self.tcp_listeners {
+            listener.validate()?;
+            self.policy
+                .validate_endpoint(&ResidentEndpoint::Tcp {
+                    name: listener.name.clone(),
+                    bind: listener.bind,
+                })
+                .map_err(|error| error.to_string())?;
+            if !names.insert(listener.name.as_str()) {
+                return Err(format!(
+                    "resident resource name {:?} is duplicated",
+                    listener.name
+                ));
+            }
+        }
+        for timer in &self.timers {
+            timer.validate()?;
+            if !names.insert(timer.name.as_str()) {
+                return Err(format!(
+                    "resident resource name {:?} is duplicated",
+                    timer.name
+                ));
+            }
+        }
+        Ok(())
+    }
+}
+
+/// One management-owned timer source for a resident Component.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ResidentTimerConfig {
+    /// Stable operator-facing name, unique within one service configuration.
+    pub name: String,
+    /// Delay before the first event after the Actor starts.
+    pub initial_delay_ms: u64,
+    /// Optional Host-managed repeat period.
+    ///
+    /// A Component may replace this schedule by returning `arm-timer` or stop
+    /// it with `cancel-timer`.
+    pub interval_ms: Option<u64>,
+}
+
+impl ResidentTimerConfig {
+    fn validate(&self) -> Result<(), String> {
+        if self.name.trim().is_empty() {
+            return Err("resident timer name must not be empty".to_owned());
+        }
+        if self.initial_delay_ms == 0 {
+            return Err(format!(
+                "resident timer {:?} initial_delay_ms must be greater than zero",
+                self.name
+            ));
+        }
+        if self.interval_ms == Some(0) {
+            return Err(format!(
+                "resident timer {:?} interval_ms must be greater than zero",
+                self.name
+            ));
+        }
+        Ok(())
+    }
+}
+
+/// Failure while creating, driving, or draining a resident session.
+#[derive(Debug, Error)]
+pub(crate) enum ResidentSupervisorError {
+    #[error(transparent)]
+    Host(#[from] ResidentHostError),
+
+    #[error("resident driver queue capacity must be greater than zero")]
+    EmptyQueue,
+
+    #[error("failed to start resident supervisor thread: {0}")]
+    ThreadStart(#[source] io::Error),
+
+    #[error("resident supervisor for {0} stopped accepting commands")]
+    ChannelClosed(ServiceKey),
+
+    #[error("resident supervisor for {0} panicked")]
+    WorkerPanicked(ServiceKey),
+
+    #[error("resident timer operation referenced unknown resource {0}")]
+    UnknownTimer(u64),
+
+    #[error("resident stream operation referenced unknown resource {0}")]
+    UnknownStream(u64),
+
+    #[error(transparent)]
+    Tcp(#[from] TcpDriverError),
+
+    #[error("resident driver cannot apply {0} before its driver is installed")]
+    UnsupportedOperation(&'static str),
+}
+
+/// Clone-free owner for one revision's resident session and driver tasks.
+pub(crate) struct ResidentSupervisor {
+    key: ServiceKey,
+    actor: ActorHandle,
+    sender: SyncSender<SupervisorCommand>,
+    accepting_events: Arc<AtomicBool>,
+    worker: Option<ThreadJoinHandle<()>>,
+}
+
+impl ResidentSupervisor {
+    pub(crate) fn start(
+        actor: ActorHandle,
+        config: ResidentServiceConfig,
+        async_runtime: Handle,
+        command_capacity: usize,
+    ) -> Result<Self, ResidentSupervisorError> {
+        if command_capacity == 0 {
+            return Err(ResidentSupervisorError::EmptyQueue);
+        }
+        let key = actor.key().clone();
+        let session = ResidentSession::new(actor.clone(), config.policy.clone())?;
+        let (sender, receiver) = mpsc::sync_channel(command_capacity);
+        let accepting_events = Arc::new(AtomicBool::new(true));
+        let core = SupervisorCore::new(
+            session,
+            config,
+            async_runtime,
+            sender.clone(),
+            accepting_events.clone(),
+            command_capacity,
+            Instant::now(),
+        )?;
+        let thread_name = format!("wasmeld-resident-{key}");
+        let worker_key = key.clone();
+        let worker = thread::Builder::new()
+            .name(thread_name)
+            .spawn(move || run_supervisor(worker_key, core, receiver))
+            .map_err(ResidentSupervisorError::ThreadStart)?;
+        Ok(Self {
+            key,
+            actor,
+            sender,
+            accepting_events,
+            worker: Some(worker),
+        })
+    }
+
+    pub(crate) fn owns_actor(&self, actor: &ActorHandle) -> bool {
+        self.actor.is_same_instance(actor)
+    }
+
+    /// Drains this revision and cancels all Host listener, stream, and timer tasks.
+    pub(crate) fn shutdown(&mut self) -> Result<(), ResidentSupervisorError> {
+        let Some(worker) = self.worker.take() else {
+            return Ok(());
+        };
+        // Async event producers observe this before retrying a full bounded
+        // queue, so the control-plane shutdown command cannot be starved by
+        // network traffic.
+        self.accepting_events.store(false, Ordering::Release);
+        let (response_sender, response_receiver) = mpsc::sync_channel(1);
+        let send_result = self
+            .sender
+            .send(SupervisorCommand::Shutdown { response_sender });
+        let response = match send_result {
+            Ok(()) => response_receiver
+                .recv()
+                .map_err(|_| ResidentSupervisorError::ChannelClosed(self.key.clone()))?,
+            Err(_) => Err(ResidentSupervisorError::ChannelClosed(self.key.clone())),
+        };
+        worker
+            .join()
+            .map_err(|_| ResidentSupervisorError::WorkerPanicked(self.key.clone()))?;
+        response
+    }
+}
+
+impl Drop for ResidentSupervisor {
+    fn drop(&mut self) {
+        if let Err(error) = self.shutdown() {
+            tracing::warn!(service = %self.key, %error, "resident supervisor drop could not drain cleanly");
+        }
+    }
+}
+
+enum SupervisorCommand {
+    TcpAccepted {
+        endpoint_id: ResourceId,
+        stream: TcpStream,
+        peer: SocketAddr,
+    },
+    StreamData {
+        stream_id: ResourceId,
+        bytes: Vec<u8>,
+    },
+    StreamHalfClosed {
+        stream_id: ResourceId,
+    },
+    StreamClosed {
+        stream_id: ResourceId,
+        reason: StreamCloseReason,
+    },
+    TimerFired {
+        timer_id: ResourceId,
+        generation: u64,
+        scheduled_at_ns: u64,
+    },
+    Shutdown {
+        response_sender: SyncSender<Result<(), ResidentSupervisorError>>,
+    },
+}
+
+struct SupervisorCore {
+    session: ResidentSession,
+    async_runtime: Handle,
+    sender: SyncSender<SupervisorCommand>,
+    accepting_events: Arc<AtomicBool>,
+    stream_command_capacity: usize,
+    stream_read_bytes: usize,
+    origin: Instant,
+    tcp_endpoints: HashMap<ResourceId, TcpEndpointDriver>,
+    streams: HashMap<ResourceId, TcpStreamDriver>,
+    timers: HashMap<ResourceId, TimerDriver>,
+}
+
+impl SupervisorCore {
+    fn new(
+        session: ResidentSession,
+        config: ResidentServiceConfig,
+        async_runtime: Handle,
+        sender: SyncSender<SupervisorCommand>,
+        accepting_events: Arc<AtomicBool>,
+        stream_command_capacity: usize,
+        origin: Instant,
+    ) -> Result<Self, ResidentSupervisorError> {
+        let stream_read_bytes = session.max_stream_chunk_bytes();
+        let mut core = Self {
+            session,
+            async_runtime,
+            sender,
+            accepting_events,
+            stream_command_capacity,
+            stream_read_bytes,
+            origin,
+            tcp_endpoints: HashMap::new(),
+            streams: HashMap::new(),
+            timers: HashMap::new(),
+        };
+        for listener in config.tcp_listeners {
+            core.register_tcp_listener(listener)?;
+        }
+        for timer in config.timers {
+            let timer_id = core.session.register_timer(timer.name)?;
+            core.timers.insert(timer_id, TimerDriver::default());
+            core.arm_timer(timer_id, timer.initial_delay_ms, timer.interval_ms)?;
+        }
+        Ok(core)
+    }
+
+    fn register_tcp_listener(
+        &mut self,
+        config: ResidentTcpListenerConfig,
+    ) -> Result<(), ResidentSupervisorError> {
+        let endpoint_id = self.session.register_endpoint(ResidentEndpoint::Tcp {
+            name: config.name.clone(),
+            bind: config.bind,
+        })?;
+
+        let driver = match TcpEndpointDriver::start(
+            &config,
+            self.session.owner().clone(),
+            endpoint_id,
+            &self.async_runtime,
+            self.sender.clone(),
+            self.accepting_events.clone(),
+        ) {
+            Ok(driver) => driver,
+            Err(error) => {
+                let _ = self.session.release_resource(endpoint_id);
+                return Err(error.into());
+            }
+        };
+        self.tcp_endpoints.insert(endpoint_id, driver);
+        Ok(())
+    }
+
+    fn tcp_accepted(
+        &mut self,
+        endpoint_id: ResourceId,
+        stream: TcpStream,
+        peer: SocketAddr,
+    ) -> Result<(), ResidentSupervisorError> {
+        let (stream_id, operations) = self
+            .session
+            .accept_stream(endpoint_id, Some(peer.to_string()))?;
+        let driver = TcpStreamDriver::start(
+            self.session.owner().clone(),
+            stream_id,
+            stream,
+            self.stream_read_bytes,
+            self.stream_command_capacity,
+            &self.async_runtime,
+            self.sender.clone(),
+            self.accepting_events.clone(),
+        );
+        self.streams.insert(stream_id, driver);
+        if let Err(error) = self.apply_operations(operations) {
+            self.discard_stream(stream_id);
+            return Err(error);
+        }
+        Ok(())
+    }
+
+    fn stream_data(
+        &mut self,
+        stream_id: ResourceId,
+        bytes: Vec<u8>,
+    ) -> Result<(), ResidentSupervisorError> {
+        if !self.streams.contains_key(&stream_id) {
+            return Ok(());
+        }
+        let result = self
+            .session
+            .stream_data(stream_id, bytes)
+            .map_err(ResidentSupervisorError::from)
+            .and_then(|operations| self.apply_operations(operations));
+        if result.is_err() {
+            self.discard_stream(stream_id);
+        }
+        result
+    }
+
+    fn stream_half_closed(&mut self, stream_id: ResourceId) -> Result<(), ResidentSupervisorError> {
+        if !self.streams.contains_key(&stream_id) {
+            return Ok(());
+        }
+        let result = self
+            .session
+            .stream_half_closed(stream_id)
+            .map_err(ResidentSupervisorError::from)
+            .and_then(|operations| self.apply_operations(operations));
+        if result.is_err() {
+            self.discard_stream(stream_id);
+        }
+        result
+    }
+
+    fn stream_closed(
+        &mut self,
+        stream_id: ResourceId,
+        reason: StreamCloseReason,
+    ) -> Result<(), ResidentSupervisorError> {
+        let Some(driver) = self.streams.remove(&stream_id) else {
+            // A task can race with Host-initiated termination. Resource IDs are
+            // never reused, so a late terminal fact is safe to ignore.
+            return Ok(());
+        };
+        drop(driver);
+        let operations = self.session.stream_closed(stream_id, reason)?;
+        self.apply_operations(operations)
+    }
+
+    fn discard_stream(&mut self, stream_id: ResourceId) {
+        self.streams.remove(&stream_id);
+        let _ = self.session.release_resource(stream_id);
+    }
+
+    fn timer_fired(
+        &mut self,
+        timer_id: ResourceId,
+        generation: u64,
+        scheduled_at_ns: u64,
+    ) -> Result<(), ResidentSupervisorError> {
+        let interval_ms = {
+            let timer = self
+                .timers
+                .get_mut(&timer_id)
+                .ok_or(ResidentSupervisorError::UnknownTimer(timer_id.get()))?;
+            if timer.generation != generation {
+                return Ok(());
+            }
+            timer.task = None;
+            timer.interval_ms
+        };
+        let operations = self.session.timer_fired(timer_id, scheduled_at_ns)?;
+        let schedule_replaced = operations.iter().any(|operation| {
+            matches!(
+                operation,
+                ResidentOperation::ArmTimer {
+                    timer_id: operation_timer,
+                    ..
+                } | ResidentOperation::CancelTimer {
+                    timer_id: operation_timer
+                } if *operation_timer == timer_id
+            )
+        });
+        self.apply_operations(operations)?;
+        if !schedule_replaced && let Some(interval_ms) = interval_ms {
+            self.arm_timer(timer_id, interval_ms, Some(interval_ms))?;
+        }
+        Ok(())
+    }
+
+    fn apply_operations(
+        &mut self,
+        operations: Vec<ResidentOperation>,
+    ) -> Result<(), ResidentSupervisorError> {
+        let mut pending = VecDeque::from(operations);
+        while let Some(operation) = pending.pop_front() {
+            match operation {
+                ResidentOperation::ArmTimer {
+                    timer_id,
+                    delay_ms,
+                    interval_ms,
+                } => self.arm_timer(timer_id, delay_ms, interval_ms)?,
+                ResidentOperation::CancelTimer { timer_id } => self.cancel_timer(timer_id)?,
+                ResidentOperation::WriteStream { stream_id, bytes } => {
+                    if !self.queue_stream_command(stream_id, TcpStreamCommand::Write(bytes))? {
+                        pending.extend(
+                            self.terminate_stream(stream_id, StreamCloseReason::TransportError)?,
+                        );
+                    }
+                }
+                ResidentOperation::CloseStream { stream_id } => {
+                    if !self.queue_stream_command(stream_id, TcpStreamCommand::Close)? {
+                        pending.extend(
+                            self.terminate_stream(stream_id, StreamCloseReason::HostClosed)?,
+                        );
+                    }
+                }
+                ResidentOperation::PauseStream { stream_id } => {
+                    if !self.queue_stream_command(stream_id, TcpStreamCommand::Pause)? {
+                        pending.extend(
+                            self.terminate_stream(stream_id, StreamCloseReason::TransportError)?,
+                        );
+                    }
+                }
+                ResidentOperation::ResumeStream { stream_id } => {
+                    if !self.queue_stream_command(stream_id, TcpStreamCommand::Resume)? {
+                        pending.extend(
+                            self.terminate_stream(stream_id, StreamCloseReason::TransportError)?,
+                        );
+                    }
+                }
+                ResidentOperation::SendDatagram { .. } => {
+                    return Err(ResidentSupervisorError::UnsupportedOperation(
+                        "send-datagram",
+                    ));
+                }
+                ResidentOperation::AcknowledgeMessage { .. } => {
+                    return Err(ResidentSupervisorError::UnsupportedOperation(
+                        "acknowledge-message",
+                    ));
+                }
+                ResidentOperation::SourceCommand { .. } => {
+                    return Err(ResidentSupervisorError::UnsupportedOperation(
+                        "source-command",
+                    ));
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// Queues one bounded operation without ever blocking the supervisor.
+    ///
+    /// A full per-stream queue means the peer is not consuming output quickly
+    /// enough. The caller terminates that stream instead of stalling unrelated
+    /// connections or allowing an unbounded Host buffer.
+    fn queue_stream_command(
+        &self,
+        stream_id: ResourceId,
+        command: TcpStreamCommand,
+    ) -> Result<bool, ResidentSupervisorError> {
+        let driver = self
+            .streams
+            .get(&stream_id)
+            .ok_or(ResidentSupervisorError::UnknownStream(stream_id.get()))?;
+        match driver.try_send(command) {
+            Ok(()) => Ok(true),
+            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
+                tracing::warn!(
+                    service = %self.session.owner(),
+                    resource_id = stream_id.get(),
+                    "resident TCP stream closed because its write/control queue is full"
+                );
+                Ok(false)
+            }
+            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => Ok(false),
+        }
+    }
+
+    fn terminate_stream(
+        &mut self,
+        stream_id: ResourceId,
+        reason: StreamCloseReason,
+    ) -> Result<Vec<ResidentOperation>, ResidentSupervisorError> {
+        let driver = self
+            .streams
+            .remove(&stream_id)
+            .ok_or(ResidentSupervisorError::UnknownStream(stream_id.get()))?;
+        drop(driver);
+        Ok(self.session.stream_closed(stream_id, reason)?)
+    }
+
+    fn arm_timer(
+        &mut self,
+        timer_id: ResourceId,
+        delay_ms: u64,
+        interval_ms: Option<u64>,
+    ) -> Result<(), ResidentSupervisorError> {
+        let timer = self
+            .timers
+            .get_mut(&timer_id)
+            .ok_or(ResidentSupervisorError::UnknownTimer(timer_id.get()))?;
+        timer.cancel_task();
+        timer.generation = next_generation(timer.generation);
+        timer.interval_ms = interval_ms;
+        let generation = timer.generation;
+        let delay = Duration::from_millis(delay_ms);
+        let scheduled_at_ns = u64::try_from(self.origin.elapsed().saturating_add(delay).as_nanos())
+            .unwrap_or(u64::MAX);
+        let sender = self.sender.clone();
+        let accepting_events = self.accepting_events.clone();
+        let owner = self.session.owner().clone();
+        timer.task = Some(self.async_runtime.spawn(async move {
+            tokio::time::sleep(delay).await;
+            if !accepting_events.load(Ordering::Acquire) {
+                return;
+            }
+            match sender.try_send(SupervisorCommand::TimerFired {
+                timer_id,
+                generation,
+                scheduled_at_ns,
+            }) {
+                Ok(()) => {}
+                Err(TrySendError::Full(_)) => {
+                    tracing::warn!(
+                        service = %owner,
+                        resource_id = timer_id.get(),
+                        "resident timer occurrence dropped because the supervisor queue is full"
+                    );
+                }
+                Err(TrySendError::Disconnected(_)) => {}
+            }
+        }));
+        Ok(())
+    }
+
+    fn cancel_timer(&mut self, timer_id: ResourceId) -> Result<(), ResidentSupervisorError> {
+        let timer = self
+            .timers
+            .get_mut(&timer_id)
+            .ok_or(ResidentSupervisorError::UnknownTimer(timer_id.get()))?;
+        timer.cancel_task();
+        timer.generation = next_generation(timer.generation);
+        timer.interval_ms = None;
+        Ok(())
+    }
+
+    fn shutdown(&mut self) -> Result<(), ResidentSupervisorError> {
+        self.accepting_events.store(false, Ordering::Release);
+        for endpoint in self.tcp_endpoints.values_mut() {
+            endpoint.abort();
+        }
+        for timer in self.timers.values_mut() {
+            timer.cancel_task();
+        }
+
+        let mut first_error = match self.session.shutdown() {
+            Ok(operations) => self.apply_operations(operations).err(),
+            Err(error) => Some(error.into()),
+        };
+
+        // A shutdown callback may return stream or timer operations. They have
+        // been validated, but no source may remain active beyond this edge.
+        let stream_ids = self.streams.keys().copied().collect::<Vec<_>>();
+        for stream_id in stream_ids {
+            self.streams.remove(&stream_id);
+            if let Err(error) = self.session.release_resource(stream_id)
+                && first_error.is_none()
+            {
+                first_error = Some(error.into());
+            }
+        }
+        self.streams.clear();
+
+        let endpoint_ids = self.tcp_endpoints.keys().copied().collect::<Vec<_>>();
+        for endpoint_id in endpoint_ids {
+            self.tcp_endpoints.remove(&endpoint_id);
+            if let Err(error) = self.session.release_resource(endpoint_id)
+                && first_error.is_none()
+            {
+                first_error = Some(error.into());
+            }
+        }
+        self.tcp_endpoints.clear();
+
+        for timer in self.timers.values_mut() {
+            timer.cancel_task();
+        }
+        let timer_ids = self.timers.keys().copied().collect::<Vec<_>>();
+        for timer_id in timer_ids {
+            if let Err(error) = self.session.release_resource(timer_id)
+                && first_error.is_none()
+            {
+                first_error = Some(error.into());
+            }
+        }
+        self.timers.clear();
+
+        match first_error {
+            Some(error) => Err(error),
+            None => Ok(()),
+        }
+    }
+}
+
+fn run_supervisor(
+    key: ServiceKey,
+    mut core: SupervisorCore,
+    receiver: Receiver<SupervisorCommand>,
+) {
+    while let Ok(command) = receiver.recv() {
+        match command {
+            SupervisorCommand::TcpAccepted {
+                endpoint_id,
+                stream,
+                peer,
+            } => {
+                if let Err(error) = core.tcp_accepted(endpoint_id, stream, peer) {
+                    tracing::warn!(
+                        service = %key,
+                        resource_id = endpoint_id.get(),
+                        %peer,
+                        %error,
+                        "resident TCP connection was rejected"
+                    );
+                }
+            }
+            SupervisorCommand::StreamData { stream_id, bytes } => {
+                if let Err(error) = core.stream_data(stream_id, bytes) {
+                    tracing::warn!(
+                        service = %key,
+                        resource_id = stream_id.get(),
+                        %error,
+                        "resident TCP stream data failed and the connection was closed"
+                    );
+                }
+            }
+            SupervisorCommand::StreamHalfClosed { stream_id } => {
+                if let Err(error) = core.stream_half_closed(stream_id) {
+                    tracing::warn!(
+                        service = %key,
+                        resource_id = stream_id.get(),
+                        %error,
+                        "resident TCP half-close event failed and the connection was closed"
+                    );
+                }
+            }
+            SupervisorCommand::StreamClosed { stream_id, reason } => {
+                if let Err(error) = core.stream_closed(stream_id, reason) {
+                    tracing::warn!(
+                        service = %key,
+                        resource_id = stream_id.get(),
+                        %error,
+                        "resident TCP terminal event failed"
+                    );
+                }
+            }
+            SupervisorCommand::TimerFired {
+                timer_id,
+                generation,
+                scheduled_at_ns,
+            } => {
+                if let Err(error) = core.timer_fired(timer_id, generation, scheduled_at_ns) {
+                    let _ = core.cancel_timer(timer_id);
+                    tracing::warn!(
+                        service = %key,
+                        resource_id = timer_id.get(),
+                        %error,
+                        "resident timer event failed and was cancelled"
+                    );
+                }
+            }
+            SupervisorCommand::Shutdown { response_sender } => {
+                let result = core.shutdown();
+                let _ = response_sender.send(result);
+                return;
+            }
+        }
+    }
+
+    if let Err(error) = core.shutdown() {
+        tracing::warn!(service = %key, %error, "resident supervisor channel closed during drain");
+    }
+}
+
+#[derive(Default)]
+struct TimerDriver {
+    generation: u64,
+    interval_ms: Option<u64>,
+    task: Option<TaskJoinHandle<()>>,
+}
+
+impl TimerDriver {
+    fn cancel_task(&mut self) {
+        if let Some(task) = self.task.take() {
+            task.abort();
+        }
+    }
+}
+
+impl Drop for TimerDriver {
+    fn drop(&mut self) {
+        self.cancel_task();
+    }
+}
+
+fn next_generation(current: u64) -> u64 {
+    let next = current.wrapping_add(1);
+    if next == 0 { 1 } else { next }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        ResidentServiceConfig, ResidentTcpListenerConfig, ResidentTimerConfig, next_generation,
+    };
+
+    #[test]
+    fn rejects_invalid_timer_configuration() {
+        for timer in [
+            ResidentTimerConfig {
+                name: String::new(),
+                initial_delay_ms: 1,
+                interval_ms: None,
+            },
+            ResidentTimerConfig {
+                name: "zero-delay".to_owned(),
+                initial_delay_ms: 0,
+                interval_ms: None,
+            },
+            ResidentTimerConfig {
+                name: "zero-interval".to_owned(),
+                initial_delay_ms: 1,
+                interval_ms: Some(0),
+            },
+        ] {
+            assert!(
+                ResidentServiceConfig {
+                    timers: vec![timer],
+                    ..ResidentServiceConfig::default()
+                }
+                .validate()
+                .is_err()
+            );
+        }
+    }
+
+    #[test]
+    fn rejects_duplicate_timer_names() {
+        let timer = ResidentTimerConfig {
+            name: "heartbeat".to_owned(),
+            initial_delay_ms: 1,
+            interval_ms: None,
+        };
+        assert!(
+            ResidentServiceConfig {
+                timers: vec![timer.clone(), timer],
+                ..ResidentServiceConfig::default()
+            }
+            .validate()
+            .is_err()
+        );
+    }
+
+    #[test]
+    fn rejects_names_shared_by_listener_and_timer() {
+        assert!(
+            ResidentServiceConfig {
+                tcp_listeners: vec![ResidentTcpListenerConfig {
+                    name: "events".to_owned(),
+                    bind: "127.0.0.1:9000".parse().unwrap(),
+                }],
+                timers: vec![ResidentTimerConfig {
+                    name: "events".to_owned(),
+                    initial_delay_ms: 1,
+                    interval_ms: None,
+                }],
+                ..ResidentServiceConfig::default()
+            }
+            .validate()
+            .is_err()
+        );
+    }
+
+    #[test]
+    fn rejects_tcp_listener_outside_default_policy() {
+        assert!(
+            ResidentServiceConfig {
+                tcp_listeners: vec![ResidentTcpListenerConfig {
+                    name: "denied".to_owned(),
+                    bind: "127.0.0.1:9000".parse().unwrap(),
+                }],
+                ..ResidentServiceConfig::default()
+            }
+            .validate()
+            .is_err()
+        );
+    }
+
+    #[test]
+    fn timer_generation_never_uses_zero() {
+        assert_eq!(next_generation(0), 1);
+        assert_eq!(next_generation(u64::MAX), 1);
+    }
+}

crates/wasmeld-console/src/resident_supervisor/tcp.rs

@@ -0,0 +1,402 @@
+//! Tokio TCP transport owned by one resident supervisor.
+//!
+//! This module owns operating-system socket handles and knows nothing about
+//! Wasm bindings or Component effects. It reports accepted connections and
+//! bounded stream facts through [`SupervisorCommand`], while the parent
+//! supervisor validates resource identity and translates Component operations
+//! back into [`TcpStreamCommand`] values.
+
+use std::{
+    io,
+    net::SocketAddr,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+        mpsc::{SyncSender, TrySendError},
+    },
+    time::Duration,
+};
+
+use thiserror::Error;
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::{TcpListener, TcpSocket, TcpStream},
+    runtime::Handle,
+    sync::mpsc,
+    task::JoinHandle,
+};
+use wasmeld_runtime::{ResourceId, ServiceKey, StreamCloseReason};
+
+use super::SupervisorCommand;
+
+/// One management-owned TCP listener for a resident Component.
+///
+/// The address must be numeric; DNS and socket handles are never exposed to
+/// the Component. Port zero is accepted for tests and private embedding, but a
+/// fixed configured port is required when external callers must discover it.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ResidentTcpListenerConfig {
+    /// Stable operator-facing name, unique among all resources for the service.
+    pub name: String,
+    /// Numeric address bound by the Host after policy validation.
+    pub bind: SocketAddr,
+}
+
+impl ResidentTcpListenerConfig {
+    pub(super) fn validate(&self) -> Result<(), String> {
+        if self.name.trim().is_empty() {
+            return Err("resident TCP listener name must not be empty".to_owned());
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, Error)]
+pub(crate) enum TcpDriverError {
+    #[error("failed to {action} resident TCP listener {name:?} at {bind}: {source}")]
+    Listener {
+        name: String,
+        bind: SocketAddr,
+        action: &'static str,
+        #[source]
+        source: io::Error,
+    },
+}
+
+pub(super) struct TcpEndpointDriver {
+    task: JoinHandle<()>,
+}
+
+impl TcpEndpointDriver {
+    pub(super) fn start(
+        config: &ResidentTcpListenerConfig,
+        owner: ServiceKey,
+        endpoint_id: ResourceId,
+        async_runtime: &Handle,
+        sender: SyncSender<SupervisorCommand>,
+        accepting_events: Arc<AtomicBool>,
+    ) -> Result<Self, TcpDriverError> {
+        let listener = bind_listener(config, async_runtime)?;
+        let name = config.name.clone();
+        let task = async_runtime.spawn(async move {
+            run_listener(owner, endpoint_id, name, listener, sender, accepting_events).await;
+        });
+        Ok(Self { task })
+    }
+
+    pub(super) fn abort(&mut self) {
+        self.task.abort();
+    }
+}
+
+impl Drop for TcpEndpointDriver {
+    fn drop(&mut self) {
+        self.abort();
+    }
+}
+
+pub(super) struct TcpStreamDriver {
+    commands: mpsc::Sender<TcpStreamCommand>,
+    task: JoinHandle<()>,
+}
+
+impl TcpStreamDriver {
+    #[allow(clippy::too_many_arguments)]
+    pub(super) fn start(
+        owner: ServiceKey,
+        stream_id: ResourceId,
+        stream: TcpStream,
+        read_bytes: usize,
+        command_capacity: usize,
+        async_runtime: &Handle,
+        sender: SyncSender<SupervisorCommand>,
+        accepting_events: Arc<AtomicBool>,
+    ) -> Self {
+        let (commands, command_receiver) = mpsc::channel(command_capacity);
+        let task = async_runtime.spawn(async move {
+            run_stream(
+                owner,
+                stream_id,
+                stream,
+                read_bytes,
+                command_receiver,
+                sender,
+                accepting_events,
+            )
+            .await;
+        });
+        Self { commands, task }
+    }
+
+    pub(super) fn try_send(
+        &self,
+        command: TcpStreamCommand,
+    ) -> Result<(), mpsc::error::TrySendError<TcpStreamCommand>> {
+        self.commands.try_send(command)
+    }
+}
+
+impl Drop for TcpStreamDriver {
+    fn drop(&mut self) {
+        // Aborting the task drops both owned socket halves. The session owner
+        // removes the resource identity separately and in deterministic order.
+        self.task.abort();
+    }
+}
+
+pub(super) enum TcpStreamCommand {
+    Write(Vec<u8>),
+    Close,
+    Pause,
+    Resume,
+}
+
+fn bind_listener(
+    config: &ResidentTcpListenerConfig,
+    async_runtime: &Handle,
+) -> Result<TcpListener, TcpDriverError> {
+    // TcpSocket gives the platform an explicit backlog and SO_REUSEADDR, which
+    // makes an immediate service restart able to reclaim its stable address
+    // after existing connections are closed.
+    let _runtime_context = async_runtime.enter();
+    let socket = if config.bind.is_ipv4() {
+        TcpSocket::new_v4()
+    } else {
+        TcpSocket::new_v6()
+    }
+    .map_err(|source| listener_error(config, "create socket for", source))?;
+    socket
+        .set_reuseaddr(true)
+        .map_err(|source| listener_error(config, "configure", source))?;
+    socket
+        .bind(config.bind)
+        .map_err(|source| listener_error(config, "bind", source))?;
+    socket
+        .listen(1024)
+        .map_err(|source| listener_error(config, "listen on", source))
+}
+
+async fn run_listener(
+    owner: ServiceKey,
+    endpoint_id: ResourceId,
+    name: String,
+    listener: TcpListener,
+    sender: SyncSender<SupervisorCommand>,
+    accepting_events: Arc<AtomicBool>,
+) {
+    while accepting_events.load(Ordering::Acquire) {
+        match listener.accept().await {
+            Ok((stream, peer)) => {
+                if let Err(error) = stream.set_nodelay(true) {
+                    tracing::warn!(
+                        service = %owner,
+                        endpoint = %name,
+                        %peer,
+                        %error,
+                        "resident TCP connection rejected because TCP_NODELAY could not be set"
+                    );
+                    continue;
+                }
+                if !send_stream_event(
+                    &sender,
+                    &accepting_events,
+                    SupervisorCommand::TcpAccepted {
+                        endpoint_id,
+                        stream,
+                        peer,
+                    },
+                )
+                .await
+                {
+                    return;
+                }
+            }
+            Err(error) => {
+                if !accepting_events.load(Ordering::Acquire) {
+                    return;
+                }
+                tracing::warn!(
+                    service = %owner,
+                    endpoint = %name,
+                    %error,
+                    "resident TCP accept failed; listener will retry"
+                );
+                // Persistent descriptor exhaustion must not turn into a hot
+                // retry loop. A healthy listener resumes after the backoff.
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            }
+        }
+    }
+}
+
+async fn run_stream(
+    owner: ServiceKey,
+    stream_id: ResourceId,
+    stream: TcpStream,
+    read_bytes: usize,
+    mut commands: mpsc::Receiver<TcpStreamCommand>,
+    sender: SyncSender<SupervisorCommand>,
+    accepting_events: Arc<AtomicBool>,
+) {
+    let (mut reader, mut writer) = stream.into_split();
+    let mut buffer = vec![0_u8; read_bytes];
+    let mut read_paused = false;
+    let mut peer_half_closed = false;
+
+    loop {
+        if !accepting_events.load(Ordering::Acquire) {
+            return;
+        }
+
+        if read_paused || peer_half_closed {
+            let Some(command) = commands.recv().await else {
+                return;
+            };
+            if let Some(reason) = apply_stream_command(command, &mut writer, &mut read_paused).await
+            {
+                let _ = send_stream_terminal(&sender, &accepting_events, stream_id, reason).await;
+                return;
+            }
+            continue;
+        }
+
+        tokio::select! {
+            biased;
+            command = commands.recv() => {
+                let Some(command) = command else {
+                    return;
+                };
+                if let Some(reason) =
+                    apply_stream_command(command, &mut writer, &mut read_paused).await
+                {
+                    let _ =
+                        send_stream_terminal(&sender, &accepting_events, stream_id, reason).await;
+                    return;
+                }
+            }
+            read = reader.read(&mut buffer) => {
+                match read {
+                    Ok(0) => {
+                        peer_half_closed = true;
+                        if !send_stream_event(
+                            &sender,
+                            &accepting_events,
+                            SupervisorCommand::StreamHalfClosed { stream_id },
+                        )
+                        .await
+                        {
+                            return;
+                        }
+                    }
+                    Ok(read) => {
+                        if !send_stream_event(
+                            &sender,
+                            &accepting_events,
+                            SupervisorCommand::StreamData {
+                                stream_id,
+                                bytes: buffer[..read].to_vec(),
+                            },
+                        )
+                        .await
+                        {
+                            return;
+                        }
+                    }
+                    Err(error) => {
+                        tracing::debug!(
+                            service = %owner,
+                            resource_id = stream_id.get(),
+                            %error,
+                            "resident TCP read failed"
+                        );
+                        let _ = send_stream_terminal(
+                            &sender,
+                            &accepting_events,
+                            stream_id,
+                            StreamCloseReason::TransportError,
+                        )
+                        .await;
+                        return;
+                    }
+                }
+            }
+        }
+    }
+}
+
+async fn apply_stream_command(
+    command: TcpStreamCommand,
+    writer: &mut tokio::net::tcp::OwnedWriteHalf,
+    read_paused: &mut bool,
+) -> Option<StreamCloseReason> {
+    match command {
+        TcpStreamCommand::Write(bytes) => writer
+            .write_all(&bytes)
+            .await
+            .err()
+            .map(|_| StreamCloseReason::TransportError),
+        TcpStreamCommand::Close => {
+            let _ = writer.shutdown().await;
+            Some(StreamCloseReason::HostClosed)
+        }
+        TcpStreamCommand::Pause => {
+            *read_paused = true;
+            None
+        }
+        TcpStreamCommand::Resume => {
+            *read_paused = false;
+            None
+        }
+    }
+}
+
+async fn send_stream_terminal(
+    sender: &SyncSender<SupervisorCommand>,
+    accepting_events: &AtomicBool,
+    stream_id: ResourceId,
+    reason: StreamCloseReason,
+) -> bool {
+    send_stream_event(
+        sender,
+        accepting_events,
+        SupervisorCommand::StreamClosed { stream_id, reason },
+    )
+    .await
+}
+
+/// Sends a non-lossy stream fact through the bounded supervisor queue.
+///
+/// Unlike timer ticks, TCP chunks cannot be dropped. Waiting here naturally
+/// stops the connection task from reading more bytes and propagates
+/// backpressure into the kernel receive buffer. Shutdown flips
+/// `accepting_events`, which cancels the retry without blocking Actor drain.
+async fn send_stream_event(
+    sender: &SyncSender<SupervisorCommand>,
+    accepting_events: &AtomicBool,
+    mut command: SupervisorCommand,
+) -> bool {
+    while accepting_events.load(Ordering::Acquire) {
+        match sender.try_send(command) {
+            Ok(()) => return true,
+            Err(TrySendError::Full(returned)) => {
+                command = returned;
+                tokio::time::sleep(Duration::from_millis(1)).await;
+            }
+            Err(TrySendError::Disconnected(_)) => return false,
+        }
+    }
+    false
+}
+
+fn listener_error(
+    config: &ResidentTcpListenerConfig,
+    action: &'static str,
+    source: io::Error,
+) -> TcpDriverError {
+    TcpDriverError::Listener {
+        name: config.name.clone(),
+        bind: config.bind,
+        action,
+        source,
+    }
+}

crates/wasmeld-console/tests/api.rs

@@ -1,9 +1,12 @@
 use std::{
+    collections::BTreeMap,
     fs,
     io::Cursor,
+    net::{SocketAddr, TcpListener as StdTcpListener},
     path::{Path, PathBuf},
     process::Command,
     sync::{Arc, OnceLock},
+    time::Duration,
 };
 
 use axum::{
@@ -14,8 +17,15 @@ use axum::{
 use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
 use serde_json::{Value, json};
 use tempfile::TempDir;
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::TcpStream,
+};
 use tower::ServiceExt;
-use wasmeld_console::{Console, ConsoleConfig, app, gateway_app};
+use wasmeld_console::{
+    Console, ConsoleConfig, NetworkScope, ResidentPolicy, ResidentServiceConfig,
+    ResidentTcpListenerConfig, ResidentTimerConfig, app, gateway_app,
+};
 use wasmeld_package::{
     module::{ModuleLock, sync_dependencies},
     wit_package::build_wit_package,
@@ -90,6 +100,192 @@ async fn manages_a_resident_component_over_http() {
     assert!(events["events"].as_array().unwrap().len() >= 5);
 }
 
+#[tokio::test]
+async fn drives_resident_timers_across_service_and_runtime_restarts() {
+    let artifact_dir = TempDir::new().expect("temporary artifact directory");
+    let mut resident_services = BTreeMap::new();
+    resident_services.insert(
+        "resident-probe".to_owned(),
+        ResidentServiceConfig {
+            timers: vec![ResidentTimerConfig {
+                name: "heartbeat".to_owned(),
+                initial_delay_ms: 5,
+                interval_ms: None,
+            }],
+            ..ResidentServiceConfig::default()
+        },
+    );
+    let console = Arc::new(
+        Console::new(ConsoleConfig {
+            artifact_dir: artifact_dir.path().to_path_buf(),
+            wit_registry_dir: artifact_dir.path().join("wit-packages"),
+            database_path: artifact_dir.path().join("console.db"),
+            resident_services,
+            ..ConsoleConfig::default()
+        })
+        .await
+        .expect("console should start"),
+    );
+    let application = app(console, Vec::new());
+    let component = fs::read(component_artifact("resident_probe_component.wasm"))
+        .expect("resident probe component should be readable");
+
+    let response = application
+        .clone()
+        .oneshot(package_request_with_world(
+            "resident-probe",
+            "0.1.0",
+            "component:resident-probe/resident-probe-component@0.1.0",
+            &component,
+        ))
+        .await
+        .expect("resident package registration should complete");
+    assert_eq!(response.status(), StatusCode::CREATED);
+
+    let response = application
+        .clone()
+        .oneshot(json_request(
+            "/api/v1/deployments/resident-probe/activate",
+            json!({ "revision": "0.1.0" }),
+        ))
+        .await
+        .expect("resident deployment activation should complete");
+    assert_eq!(response.status(), StatusCode::OK);
+    assert!(
+        wait_for_resident_count(&application).await > 0,
+        "configured timer should reach the resident Component"
+    );
+
+    let response = application
+        .clone()
+        .oneshot(empty_post("/api/v1/services/resident-probe/0.1.0/stop"))
+        .await
+        .expect("resident service stop should complete");
+    assert_eq!(response.status(), StatusCode::OK);
+
+    let response = application
+        .clone()
+        .oneshot(empty_post("/api/v1/services/resident-probe/0.1.0/restart"))
+        .await
+        .expect("resident service restart should complete");
+    assert_eq!(response.status(), StatusCode::OK);
+    assert!(
+        wait_for_resident_count(&application).await > 0,
+        "service restart should create a new timer session"
+    );
+
+    let response = application
+        .clone()
+        .oneshot(empty_post("/api/v1/runtime/restart"))
+        .await
+        .expect("Runtime restart should complete");
+    assert_eq!(response.status(), StatusCode::OK);
+    assert!(
+        wait_for_resident_count(&application).await > 0,
+        "deployment restore should recreate the timer supervisor"
+    );
+}
+
+#[tokio::test]
+async fn drives_tcp_streams_across_service_and_runtime_restarts() {
+    let artifact_dir = TempDir::new().expect("temporary artifact directory");
+    let occupied_listener =
+        StdTcpListener::bind("127.0.0.1:0").expect("occupied test listener should bind");
+    let address = occupied_listener
+        .local_addr()
+        .expect("occupied test listener should have an address");
+    let mut resident_services = BTreeMap::new();
+    resident_services.insert(
+        "resident-probe".to_owned(),
+        ResidentServiceConfig {
+            policy: ResidentPolicy {
+                tcp_listen: NetworkScope::Loopback,
+                ..ResidentPolicy::default()
+            },
+            tcp_listeners: vec![ResidentTcpListenerConfig {
+                name: "echo".to_owned(),
+                bind: address,
+            }],
+            ..ResidentServiceConfig::default()
+        },
+    );
+    let console = Arc::new(
+        Console::new(ConsoleConfig {
+            artifact_dir: artifact_dir.path().to_path_buf(),
+            wit_registry_dir: artifact_dir.path().join("wit-packages"),
+            database_path: artifact_dir.path().join("console.db"),
+            resident_services,
+            ..ConsoleConfig::default()
+        })
+        .await
+        .expect("console should start"),
+    );
+    let application = app(console, Vec::new());
+    let component = fs::read(component_artifact("resident_probe_component.wasm"))
+        .expect("resident probe component should be readable");
+
+    let response = application
+        .clone()
+        .oneshot(package_request_with_world(
+            "resident-probe",
+            "0.1.0",
+            "component:resident-probe/resident-probe-component@0.1.0",
+            &component,
+        ))
+        .await
+        .expect("resident package registration should complete");
+    assert_eq!(response.status(), StatusCode::CREATED);
+
+    let response = application
+        .clone()
+        .oneshot(json_request(
+            "/api/v1/deployments/resident-probe/activate",
+            json!({ "revision": "0.1.0" }),
+        ))
+        .await
+        .expect("conflicting resident deployment activation should complete");
+    assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
+
+    // Driver startup failure must stop the Actor and leave deployment state
+    // untouched. Once the external conflict is gone, the same revision can
+    // start normally without repairing Runtime state manually.
+    drop(occupied_listener);
+    let response = application
+        .clone()
+        .oneshot(json_request(
+            "/api/v1/deployments/resident-probe/activate",
+            json!({ "revision": "0.1.0" }),
+        ))
+        .await
+        .expect("resident deployment activation should complete");
+    assert_eq!(response.status(), StatusCode::OK);
+    let large_payload = vec![0xA5; 2 * 64 * 1024 + 17];
+    assert_tcp_echo(address, &large_payload).await;
+
+    let response = application
+        .clone()
+        .oneshot(empty_post("/api/v1/services/resident-probe/0.1.0/stop"))
+        .await
+        .expect("resident service stop should complete");
+    assert_eq!(response.status(), StatusCode::OK);
+
+    let response = application
+        .clone()
+        .oneshot(empty_post("/api/v1/services/resident-probe/0.1.0/restart"))
+        .await
+        .expect("resident service restart should complete");
+    assert_eq!(response.status(), StatusCode::OK);
+    assert_tcp_echo(address, b"service restart").await;
+
+    let response = application
+        .clone()
+        .oneshot(empty_post("/api/v1/runtime/restart"))
+        .await
+        .expect("Runtime restart should complete");
+    assert_eq!(response.status(), StatusCode::OK);
+    assert_tcp_echo(address, b"runtime restart").await;
+}
+
 #[tokio::test]
 async fn reports_exact_component_host_capabilities() {
     let artifact_dir = TempDir::new().expect("temporary artifact directory");
@@ -999,6 +1195,53 @@ async fn invoke_service(application: &Router, id: &str, revision: &str, input: &
         .unwrap()
 }
 
+async fn wait_for_resident_count(application: &Router) -> u64 {
+    for _ in 0..100 {
+        let output = invoke_service(application, "resident-probe", "0.1.0", &[]).await;
+        let count = u64::from_le_bytes(
+            output
+                .try_into()
+                .expect("resident probe response should be a u64"),
+        );
+        if count > 0 {
+            return count;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    0
+}
+
+async fn assert_tcp_echo(address: SocketAddr, payload: &[u8]) {
+    let mut stream = connect_tcp(address).await;
+    stream
+        .write_all(payload)
+        .await
+        .expect("TCP request should be written");
+    let mut echoed = vec![0_u8; payload.len()];
+    tokio::time::timeout(Duration::from_secs(2), stream.read_exact(&mut echoed))
+        .await
+        .expect("resident TCP echo should not time out")
+        .expect("resident TCP echo should be readable");
+    assert_eq!(echoed, payload);
+}
+
+async fn connect_tcp(address: SocketAddr) -> TcpStream {
+    let mut last_error = None;
+    for _ in 0..100 {
+        match TcpStream::connect(address).await {
+            Ok(stream) => return stream,
+            Err(error) => {
+                last_error = Some(error);
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        }
+    }
+    panic!(
+        "resident TCP listener {address} did not accept connections: {}",
+        last_error.expect("at least one connection should have been attempted")
+    );
+}
+
 fn echo_component() -> PathBuf {
     component_artifact("echo_component.wasm")
 }
@@ -1010,6 +1253,7 @@ fn component_artifact(name: &str) -> PathBuf {
             "components/counter/Cargo.toml",
             "components/clock-probe/Cargo.toml",
             "components/kv-probe/Cargo.toml",
+            "components/resident-probe/Cargo.toml",
         ] {
             let status = Command::new("rustup")
                 .args([