docs(resident): document timer and TCP drivers

Describe ResidentSupervisor ownership, TCP policy and configuration, bounded stream backpressure, timer scheduling, restart behavior, and deterministic Host-resource shutdown order.

.zed/settings.json

@@ -0,0 +1,26 @@
+{
+  "lsp": {
+    "tailwindcss-language-server": {
+      "settings": {
+        "classFunctions": [
+          "cva",
+          "cx",
+          "cn"
+        ],
+        "experimental": {
+          "classRegex": [
+            "[cls|className]\\s\\:\\=\\s\"([^\"]*)"
+          ]
+        }
+      }
+    }
+  },
+  "languages": {
+    "CSS": {
+      "language_servers": [
+        "tailwindcss-intellisense-css",
+        "!vscode-css-language-server"
+      ]
+    }
+  }
+}

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"

README.md

@@ -29,6 +29,12 @@ Host driver <- validated operation <- raw effect <-
 线程外；协议级驱动以及文件监听、串口、系统信号等其它来源通过独立版本的 WIT 能力演进，
 不需要扩大基础 service world。
 
+Console 已接入 Timer 与 TCP Host Driver：平台配置的 Timer、Listener 和 Stream 在
+Tokio task 中等待，并通过有界队列交给专用 `ResidentSupervisor` 线程串行进入
+`ResidentSession`。TCP 地址执行默认拒绝的 Host policy，字节流支持有界读写、半关闭和
+pause/resume；服务停止、重启和 Runtime Deployment 恢复会按 Revision 创建或释放 Host
+资源。UDP 和 Unix Driver 仍待接入。
+
 ## 结构
 
 ```text

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/README.md

@@ -30,6 +30,71 @@ Host KV 使用 `(service_id, key)` 复合主键存储在该数据库中：同一
 编译后的本地机器码默认缓存在 `var/wasmeld/component-cache`，相同 Component 在
 Console 或 Runtime 重启后可由 Wasmtime 直接复用；缓存不包含 Actor 内存。
 
+## 常驻 Host Driver
+
+导出 `wasmeld:resident/actor@0.1.0` 的 Component 启动时，Console 会为它创建独立的
+`ResidentSupervisor`。Supervisor 在专用阻塞线程上独占 `ResidentSession`，避免同步
+Actor mailbox 阻塞 Tokio executor。Timer、TCP Listener 和每条 TCP Stream 由 Tokio
+task 等待外部事件，再通过容量受限的 channel 汇聚到同一个 Supervisor。
+
+Host 资源属于平台配置，不由 Component 创建，也不写入 `.wasmpkg`。当前可通过
+`ConsoleConfig::resident_services` 为稳定服务 ID 配置 TCP Listener 和 Timer：
+
+```rust
+use std::collections::BTreeMap;
+use wasmeld_console::{
+    ConsoleConfig, NetworkScope, ResidentPolicy, ResidentServiceConfig,
+    ResidentTcpListenerConfig, ResidentTimerConfig,
+};
+
+let mut resident_services = BTreeMap::new();
+resident_services.insert(
+    "scheduler".to_owned(),
+    ResidentServiceConfig {
+        policy: ResidentPolicy {
+            tcp_listen: NetworkScope::Loopback,
+            ..ResidentPolicy::default()
+        },
+        tcp_listeners: vec![ResidentTcpListenerConfig {
+            name: "internal-api".to_owned(),
+            bind: "127.0.0.1:9000".parse().unwrap(),
+        }],
+        timers: vec![ResidentTimerConfig {
+            name: "heartbeat".to_owned(),
+            initial_delay_ms: 1_000,
+            interval_ms: Some(30_000),
+        }],
+        ..ResidentServiceConfig::default()
+    },
+);
+let config = ConsoleConfig {
+    resident_services,
+    ..ConsoleConfig::default()
+};
+```
+
+`ResidentPolicy` 默认拒绝全部网络地址；配置 Listener 时还必须显式选择 `Loopback`
+或 `Any`。Listener 在 Actor 启动过程中完成策略校验和 bind，任何失败都会回滚刚启动的
+Actor，避免出现“Actor 正在运行但端口未监听”的半启动状态。Listener、Timer 和其它
+顶层资源名称在同一服务内必须唯一，每个 Revision 都会获得独立的 Host handle 和不复用
+资源 ID。
+
+TCP Driver 每次最多读取 `limits.resident.max_stream_chunk_bytes`，不会把 socket handle
+交给 Wasm。TCP 字节在 Supervisor 队列满时等待并把背压传递到内核接收缓冲区，不能像
+Timer occurrence 一样丢弃。Component 可通过 WIT 返回 `write-stream`、`close-stream`、
+`pause-stream` 和 `resume-stream`；每条连接的写入/控制队列也是有界的，慢客户端填满
+队列时只关闭该连接，不阻塞同一 Actor 的其它连接。
+
+Timer task 只负责等待，队列满时丢弃该次 occurrence，而不是创建无界积压。Component
+返回 `arm-timer` 会替换 Host 初始日程，`cancel-timer` 会取消它；generation 使取消前
+已经排队的迟到事件失效。
+
+停止顺序固定为：停止接收新 Driver 事件、停止 accept、投递 `shutdown`、关闭 Stream
+和 Timer task、先释放子 Stream 再释放 Listener、最后停止 Actor。服务重启和 Runtime
+Deployment 恢复都会重新 bind 并创建新 Session，不会复用旧 Revision 的 socket、Timer
+或 Component 内存。UDP 和 Unix Driver 尚未接入，后续继续复用相同的 Supervisor
+序列化边界。
+
 ## 启动
 
 `wasmeld-console` 需要 Rust 1.95 或更高版本；Wasm 组件继续使用项目约定的 Rust 1.90

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,9 +778,9 @@ impl Console {
             .write()
             .map_err(|_| ConsoleError::LockPoisoned)?;
 
-        if let Some(running) = runtime.take()
-            && let Err(error) = running.stop_all()
-        {
+        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(
@@ -750,11 +790,22 @@ impl Console {
                 )?;
                 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([

crates/wasmeld-console/web/src/host/app.tsx

@@ -403,7 +403,7 @@ export default function HostApp() {
     <div class="grid h-dvh min-w-0 grid-cols-[minmax(0,1fr)] grid-rows-[auto_56px_minmax(0,1fr)_30px] overflow-hidden bg-canvas lg:grid-cols-[232px_minmax(0,1fr)] lg:grid-rows-[64px_minmax(0,1fr)_30px]">
       <aside class="border-line min-w-0 bg-[#16221f] text-white lg:row-span-3 lg:flex lg:min-h-0 lg:flex-col lg:border-r">
         <div class="flex h-16 shrink-0 items-center gap-3 px-4 lg:px-5">
-          <span class="flex size-9 items-center justify-center rounded-md bg-[#e5f4ef] text-[#155a46]">
+          <span class="flex size-9 items-center justify-center rounded-md bg-[#e5f4ef] text-brand-strong">
             <Boxes size={19} />
           </span>
           <div>
@@ -533,7 +533,7 @@ export default function HostApp() {
 
       <Show when={toast()}>
         {(message) => (
-          <div class="fixed bottom-12 right-5 z-[70] max-w-sm rounded-md bg-[#16221f] px-4 py-3 text-sm text-white shadow-xl">
+          <div class="fixed bottom-12 right-5 z-70 max-w-sm rounded-md bg-[#16221f] px-4 py-3 text-sm text-white shadow-xl">
             {message()}
           </div>
         )}

crates/wasmeld-runtime/src/resident_host.rs

@@ -515,6 +515,15 @@ impl ResidentSession {
         self.actor.key()
     }
 
+    /// Returns the largest stream chunk that a driver may deliver.
+    ///
+    /// A stream driver should size its read buffer to this value or a smaller
+    /// protocol-specific limit. The Runtime applies the same limit to both
+    /// incoming `stream-data` events and Component `write-stream` effects.
+    pub fn max_stream_chunk_bytes(&self) -> usize {
+        self.actor.resident_stream_chunk_limit()
+    }
+
     /// Registers a configured listener or datagram endpoint after policy checks.
     ///
     /// Registration does not bind an operating-system socket. The driver binds
@@ -664,6 +673,7 @@ impl ResidentSession {
         reason: StreamCloseReason,
     ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
         self.ensure_kind(stream_id, ResidentResourceKind::Stream, "stream")?;
+        *self.stream_state_mut(stream_id)? = StreamState::Closing;
         let result = self.dispatch(ResidentEvent::StreamClosed { stream_id, reason });
         self.resources.remove(&stream_id);
         result
@@ -1056,7 +1066,13 @@ impl ResidentSession {
 }
 
 impl ResidentPolicy {
-    fn validate_endpoint(&self, endpoint: &ResidentEndpoint) -> Result<(), ResidentHostError> {
+    /// Validates one endpoint without registering or binding it.
+    ///
+    /// Management layers can call this while loading static configuration so
+    /// an address outside policy fails before any Actor is started. Drivers
+    /// must still register the endpoint immediately before bind because policy
+    /// is also a runtime ownership boundary.
+    pub fn validate_endpoint(&self, endpoint: &ResidentEndpoint) -> Result<(), ResidentHostError> {
         match endpoint {
             ResidentEndpoint::Tcp { bind, .. } if !self.tcp_listen.allows(*bind) => {
                 Err(ResidentHostError::EndpointDenied(format!(

crates/wasmeld-runtime/src/runtime.rs

@@ -615,10 +615,33 @@ impl ActorHandle {
         self.execution
     }
 
+    /// Returns whether two handles address the exact same Actor instance.
+    ///
+    /// Service keys are revision identities and therefore survive an Actor
+    /// fault/restart. Host resource supervisors use this stronger check to
+    /// retain themselves for an idempotent start but replace stale resources
+    /// when the Runtime created a fresh Store for the same revision.
+    pub fn is_same_instance(&self, other: &Self) -> bool {
+        Arc::ptr_eq(&self.status, &other.status)
+    }
+
+    /// Returns the immutable mailbox capacity configured for this Actor.
+    ///
+    /// Host supervisors use the same bound for their external event queue so a
+    /// restored revision retains the limits stored in its own manifest instead
+    /// of inheriting newer process defaults.
+    pub fn mailbox_capacity(&self) -> usize {
+        self.limits.mailbox_capacity
+    }
+
     pub(crate) fn resident_resource_limit(&self) -> usize {
         self.limits.resident.max_resources
     }
 
+    pub(crate) fn resident_stream_chunk_limit(&self) -> usize {
+        self.limits.resident.max_stream_chunk_bytes
+    }
+
     /// Sends an invocation and waits up to the remaining service deadline.
     ///
     /// Cloned handles compete for the same bounded mailbox. A Wasm trap is

crates/wasmeld-runtime/tests/runtime_components.rs

@@ -96,12 +96,16 @@ fn faulted_actor_can_be_started_again() {
     let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
     let (key, actor) = start_component(&runtime, "fault", "fault_component.wasm");
 
+    let already_running = runtime.start(&key, Vec::new()).unwrap();
+    assert!(actor.is_same_instance(&already_running));
+
     assert!(matches!(
         actor.invoke(b"fault".to_vec()),
         Err(RuntimeError::ActorFault { .. })
     ));
 
     let restarted = runtime.start(&key, Vec::new()).unwrap();
+    assert!(!actor.is_same_instance(&restarted));
     assert_eq!(restarted.invoke(b"ready".to_vec()).unwrap(), b"ready");
 
     runtime.stop(&key).unwrap();