astrid-capsule 0.7.0

//! Shared state for wasmtime Component Model host functions.
//!
//! [`HostState`] is used as the `T` in `wasmtime::Store<T>` and is directly
//! accessible to all host trait implementations generated by `bindgen!`.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;

use tokio::sync::{Semaphore, mpsc, watch};
use tokio_util::sync::CancellationToken;

use crate::capsule::CapsuleId;
use astrid_core::uplink::{InboundMessage, MAX_UPLINKS_PER_CAPSULE, UplinkDescriptor};
use astrid_storage::ScopedKvStore;
use astrid_storage::secret::SecretStore;

/// An active network stream owned by a capsule.
///
/// Holds either an inbound Unix-socket connection (accepted from the
/// capsule's pre-provisioned listener) or an outbound TCP connection
/// (opened via `net.connect-tcp`). The read/write/close host fns
/// dispatch on the variant; both variants implement
/// [`tokio::io::AsyncRead`] / [`tokio::io::AsyncWrite`], so the inner
/// framing logic is shared.
#[derive(Debug, Clone)]
pub enum NetStream {
    /// Inbound Unix-domain socket accepted from the kernel's listener.
    Unix(Arc<tokio::sync::Mutex<tokio::net::UnixStream>>),
    /// Outbound TCP connection opened via `net.connect-tcp`.
    Tcp(TcpStreamSlot),
}

/// Per-stream state for an outbound TCP connection.
///
/// Holds the raw [`tokio::net::TcpStream`] (behind an `Arc<Mutex<…>>` so
/// host fns that all take `&mut HostState` can each lock it cooperatively)
/// plus the std-style configurable socket options that the WIT surface
/// exposes — read/write timeouts. `TCP_NODELAY` and `TTL` are not cached
/// here; they live on the underlying socket and are read back via
/// `TcpStream::nodelay()` / `ttl()` on demand.
#[derive(Debug, Clone)]
pub struct TcpStreamSlot {
    /// The connected TCP socket. Shared via `Arc<Mutex<…>>` so the
    /// existing `net-read` / `net-write` dispatchers can clone the
    /// outer `NetStream` and then lock the inner stream.
    pub stream: Arc<tokio::sync::Mutex<tokio::net::TcpStream>>,
    /// Read timeout applied to each `net-read-bytes` / `net-peek` call.
    /// `None` → use the host default (~50 ms poll for `Pending` parity
    /// with `net-read`).
    pub read_timeout: Option<std::time::Duration>,
    /// Write timeout applied to each `net-write-bytes` call.
    /// `None` → no extra timeout beyond the host cancellation token.
    pub write_timeout: Option<std::time::Duration>,
}

/// The lifecycle phase a capsule is currently executing in.
///
/// Set on [`HostState`] during `#[install]` or `#[upgrade]` dispatch.
/// The `astrid_elicit` host function checks this field and rejects calls
/// outside of a lifecycle phase.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LifecyclePhase {
    /// First-time installation.
    Install,
    /// Upgrading from a previous version.
    Upgrade,
}

/// Metadata for an interceptor binding declared in `Capsule.toml`.
///
/// Under the new per-domain WIT, interceptors are NOT capsule-side
/// `Resource<Subscription>` handles — the kernel matches incoming IPC
/// messages against interceptor patterns and dispatches them directly
/// to `astrid-hook-trigger`. This struct exists purely to:
///
/// 1. Let the guest enumerate what it's subscribed to via
///    `astrid:ipc/host.get-interceptor-bindings` (debugging /
///    tooling), and
/// 2. Anchor the registry-side routing table used by the run-loop
///    dispatcher.
///
/// `handle_id` is informational only — the guest cannot turn it
/// back into a `Resource<Subscription>` and the kernel does not key
/// any storage on it.
#[derive(Debug, Clone, serde::Serialize)]
pub struct InterceptorHandle {
    /// Enumeration index; informational only under the new ABI.
    pub handle_id: u64,
    /// The interceptor action name from the manifest.
    pub action: String,
    /// The event topic this interceptor subscribes to.
    pub topic: String,
}

use crate::engine::wasm::host::process::ProcessTracker;
use crate::security::CapsuleSecurityGate;

/// A principal-scoped filesystem mount: physical root, VFS, and capability handle.
///
/// The three are bound together — the [`DirHandle`](astrid_capabilities::DirHandle)
/// is confined to the specific [`Vfs`](astrid_vfs::Vfs) instance, and both are rooted
/// at `root`. They must always be installed and cleared as a unit, so callers
/// cannot accidentally pair an invocation-scoped VFS with a load-time handle
/// (which would break capability confinement).
pub struct PrincipalMount {
    /// Canonical physical directory this mount is rooted at.
    pub root: PathBuf,
    /// VFS wrapping `root`. Direct [`HostVfs`](astrid_vfs::HostVfs) —
    /// writes are permanent (no OverlayVfs CoW layer).
    pub vfs: Arc<dyn astrid_vfs::Vfs>,
    /// Capability handle that confines access to `root`.
    pub handle: astrid_capabilities::DirHandle,
}

impl std::fmt::Debug for PrincipalMount {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PrincipalMount")
            .field("root", &self.root)
            .field("handle", &self.handle)
            .finish_non_exhaustive()
    }
}

/// Shared state accessible to all host functions via `Store<HostState>`.
pub struct HostState {
    /// WASI context for Component Model WASI imports (clocks, random, etc.).
    pub wasi_ctx: wasmtime_wasi::WasiCtx,
    /// Resource table for WASI resource types (streams, descriptors, etc.).
    pub resource_table: wasmtime::component::ResourceTable,
    /// Memory limits for the WASM store (64 MB cap).
    pub store_limits: wasmtime::StoreLimits,
    /// The principal this capsule is running on behalf of.
    pub principal: astrid_core::principal::PrincipalId,
    /// The plugin this state belongs to.
    pub capsule_id: CapsuleId,
    /// Per-capsule log file at `home/{principal}/.local/log/{capsule}.log`.
    /// Capsule `astrid_log` output writes here instead of the system tracing
    /// subscriber. `None` if the log file couldn't be opened.
    pub capsule_log: Option<Arc<std::sync::Mutex<std::fs::File>>>,
    /// Context of the current caller (set per-invocation by the dispatcher).
    pub caller_context: Option<astrid_events::ipc::IpcMessage>,
    /// The unique session UUID for this plugin's execution state.
    pub capsule_uuid: uuid::Uuid,
    /// Workspace root directory (file operations are confined here).
    pub workspace_root: PathBuf,
    /// The Virtual File System (VFS) instance for this plugin.
    pub vfs: Arc<dyn astrid_vfs::Vfs>,
    /// The root capability handle for the VFS.
    pub vfs_root_handle: astrid_capabilities::DirHandle,
    /// Load-time principal home mount (`home://` paths). `None` if the
    /// principal's home directory does not exist on disk.
    pub home: Option<PrincipalMount>,
    /// Load-time principal tmp mount (`/tmp/` paths, backed by
    /// `~/.astrid/home/{principal}/.local/tmp/`). `None` if unavailable.
    pub tmp: Option<PrincipalMount>,
    /// Concrete reference to the [`OverlayVfs`](astrid_vfs::OverlayVfs) for
    /// commit/rollback operations. `None` for non-overlay VFS configurations
    /// (e.g., tests with a plain `HostVfs`).
    pub overlay_vfs: Option<Arc<astrid_vfs::OverlayVfs>>,
    /// Reference to the ephemeral upper directory to keep it alive for the session.
    pub upper_dir: Option<Arc<tempfile::TempDir>>,
    /// Plugin-scoped KV store (`plugin:{capsule_id}` namespace).
    pub kv: ScopedKvStore,
    /// Per-invocation KV store scoped to a different principal.
    ///
    /// Set by `WasmEngine::invoke_interceptor` when the IPC message's
    /// principal differs from `self.principal`. Host functions use
    /// `invocation_kv.as_ref().unwrap_or(&kv)` to transparently scope
    /// reads/writes to the calling principal.
    pub invocation_kv: Option<ScopedKvStore>,
    /// Per-invocation home mount for the calling principal.
    ///
    /// Populated by `WasmEngine::invoke_interceptor` when the IPC message's
    /// principal differs from `self.principal`. When set, overrides `home`
    /// for scheme resolution and security gate checks. Cleared on exit.
    pub invocation_home: Option<PrincipalMount>,
    /// Per-invocation tmp mount for the calling principal. Same lifecycle
    /// as `invocation_home`.
    pub invocation_tmp: Option<PrincipalMount>,
    /// Per-invocation secret store scoped to the calling principal.
    ///
    /// Set by `WasmEngine::invoke_interceptor` when the IPC message's
    /// principal differs from `self.principal`. When set, overrides
    /// `secret_store` so `has_secret` and future secret host fns scope
    /// reads/writes to the invoking principal. Cleared on exit.
    pub invocation_secret_store: Option<Arc<dyn SecretStore>>,
    /// Per-invocation capsule log file scoped to the calling principal.
    ///
    /// Same lifecycle as [`invocation_secret_store`](Self::invocation_secret_store).
    /// When set, `astrid_log` writes to the invoking principal's
    /// `~/.astrid/home/{principal}/.local/log/{capsule}/{date}.log` instead
    /// of the capsule owner's.
    pub invocation_capsule_log: Option<Arc<std::sync::Mutex<std::fs::File>>>,
    /// Per-invocation quota profile for the invoking principal.
    ///
    /// Set by [`WasmEngine::invoke_interceptor`](super::WasmEngine::invoke_interceptor)
    /// after resolving the invoking principal's `PrincipalProfile` through
    /// the shared [`PrincipalProfileCache`](crate::profile_cache::PrincipalProfileCache).
    /// Host functions that gate on per-principal sub-budgets (IPC
    /// throughput, background processes, HTTP streams) read this through
    /// [`effective_profile`](Self::effective_profile). Cleared on exit.
    pub invocation_profile: Option<Arc<astrid_core::profile::PrincipalProfile>>,
    /// System Event Bus for IPC publish/subscribe.
    pub event_bus: astrid_events::EventBus,
    /// Rate limiter for IPC message publishing.
    pub ipc_limiter: astrid_events::ipc::IpcRateLimiter,
    /// Plugin configuration from the manifest.
    ///
    /// Holds only **non-secret** env values (the `[env]` declarations
    /// in `Capsule.toml` whose `type` is not `"secret"` — base URLs,
    /// model names, log levels, etc.). Secret-typed keys are
    /// resolved live in [`crate::engine::wasm::host::sys`] through
    /// `resolve_secret` (file-per-secret store, see
    /// [`astrid_storage::FileSecretStore`]) instead of being preloaded
    /// here, so plaintext secret material never sits in
    /// `wasm_config`'s hot memory.
    pub config: HashMap<String, serde_json::Value>,
    /// Manifest-declared secret-typed env keys. Populated at load
    /// time from `manifest.env` entries with `type = "secret"`. Used
    /// by [`crate::engine::wasm::host::sys::Host::get_config`] to
    /// route reads through the file-per-secret store instead of
    /// `config`. The scope (per-agent vs host-wide) is an
    /// operator-side decision at `astrid secret set` time, not a
    /// manifest declaration, so the kernel always tries per-agent
    /// first and falls through to host-wide regardless of where the
    /// operator stored the value.
    pub secret_env: std::collections::HashSet<String>,
    /// IPC topic patterns this capsule is allowed to publish to.
    /// Empty means DENY ALL (fail-closed).
    pub ipc_publish_patterns: Vec<String>,
    /// IPC topic patterns this capsule is allowed to subscribe to.
    /// Empty means DENY ALL (fail-closed).
    pub ipc_subscribe_patterns: Vec<String>,
    /// Optional security gate for gated operations (HTTP, file I/O).
    pub security: Option<Arc<dyn CapsuleSecurityGate>>,
    /// Hook manager for executing user scripts synchronously via airlock.
    pub hook_manager: Option<Arc<dyn std::any::Any + Send + Sync>>,
    /// Shared capsule registry for `hooks::trigger` fan-out dispatch.
    ///
    /// When set, the `astrid_trigger_hook` host function can iterate the
    /// registry to find capsules with matching interceptors, invoke them,
    /// and collect responses. This is the kernel mechanism that WASM
    /// capsules use to dispatch hooks to other capsules.
    pub capsule_registry: Option<Arc<tokio::sync::RwLock<crate::registry::CapsuleRegistry>>>,
    /// Tokio runtime handle for bridging async operations in sync host functions.
    pub runtime_handle: tokio::runtime::Handle,
    /// Whether the plugin manifest declares `CapsuleCapability::Uplink`.
    ///
    /// Used to gate `astrid_register_uplink` — only uplink plugins
    /// are allowed to register uplinks.
    pub has_uplink_capability: bool,
    /// Sender for inbound messages from uplink plugins.
    ///
    /// Set during plugin loading when the manifest declares
    /// [`CapsuleCapability::Uplink`](crate::CapsuleCapability). Feeds into
    /// the gateway's inbound router.
    pub inbound_tx: Option<mpsc::Sender<InboundMessage>>,
    /// Uplinks registered by the WASM guest via `astrid_register_uplink`.
    pub registered_uplinks: Vec<UplinkDescriptor>,
    /// Optional natively bound unix listener.
    pub cli_socket_listener: Option<Arc<tokio::sync::Mutex<tokio::net::UnixListener>>>,
    /// Active lifecycle phase, if any. `None` during normal runtime.
    /// Set to `Some(Install)` or `Some(Upgrade)` during lifecycle dispatch.
    /// Gates the `astrid_elicit` host function.
    pub lifecycle_phase: Option<LifecyclePhase>,
    /// Secret store for capsule credentials (keychain with KV fallback).
    pub secret_store: Arc<dyn SecretStore>,
    /// Readiness signal sender for run-loop capsules.
    ///
    /// When the WASM guest calls `astrid_signal_ready`, the host sends `true`
    /// on this channel. The kernel waits on the corresponding receiver before
    /// loading dependent capsules.
    pub ready_tx: Option<watch::Sender<bool>>,
    /// Bounded concurrency semaphore for host function blocking calls.
    ///
    /// Limits the number of concurrent `block_in_place` / `block_on` operations
    /// across all capsules to prevent tokio thread-pool exhaustion.
    /// Created via [`default_host_semaphore`].
    pub host_semaphore: Arc<Semaphore>,
    /// Cooperative cancellation token for long-running host function calls.
    ///
    /// Triggered during capsule unload to unblock every blocking host
    /// fn that races it: `Subscription.recv`, `elicit`, the various
    /// listener `accept` paths, `connect-tcp`, `read`/`read-bytes`/
    /// `write-bytes` on `tcp-stream`, `read-chunk` on `http-stream`,
    /// `pollable.block`, `poll.poll`, `sleep-ns`, and the `wait` /
    /// `spawn` paths on `process`. Streams (`astrid:io/streams`)
    /// short-circuit with `Closed` when this fires; pollables
    /// short-circuit with `Cancelled`.
    pub cancel_token: CancellationToken,
    /// Session token for authenticating CLI socket connections. Only set for
    /// the CLI proxy capsule (which has `net_bind` capability).
    pub session_token: Option<std::sync::Arc<astrid_core::session_token::SessionToken>>,
    /// Interceptor binding metadata, populated during `WasmEngine::load()`
    /// from `[[interceptor]]` entries in `Capsule.toml`. The kernel
    /// matches incoming IPC messages against these patterns and calls
    /// `astrid-hook-trigger` directly — these handles are not
    /// `Resource<Subscription>` references and the guest cannot
    /// consume from them. See [`InterceptorHandle`] for the metadata
    /// shape and `astrid:ipc/host.get-interceptor-bindings` for the
    /// guest-facing accessor.
    pub interceptor_handles: Vec<InterceptorHandle>,
    /// Shared allowance store for capsule-level approval requests.
    ///
    /// When set, the `astrid_request_approval` host function can check
    /// existing allowances before prompting the user. Approvals with
    /// session/always scope create new allowances here.
    pub allowance_store: Option<std::sync::Arc<astrid_approval::AllowanceStore>>,
    /// Shared identity store for resolving platform users to `AstridUserId`.
    ///
    /// When `None`, identity host functions return an error.
    pub identity_store: Option<std::sync::Arc<dyn astrid_storage::IdentityStore>>,
    /// Active HTTP streaming responses, keyed by handle ID.
    ///
    /// Each entry holds a `reqwest::Response` whose body is being consumed
    /// incrementally by `astrid_http_stream_read` plus the principal that
    /// started the stream, so per-principal HTTP-stream caps can be
    /// enforced by creator without blocking other principals on the same
    /// capsule. Cleaned up by `astrid_http_stream_close` or when the
    /// capsule unloads.
    pub active_http_streams: HashMap<u64, crate::engine::wasm::host::http::ActiveHttpStream>,
    /// Monotonic counter for HTTP stream handle IDs.
    /// Starts at 1 (0 reserved as sentinel).
    pub next_http_stream_id: u64,
    /// Tracks active child process PIDs for cancellation.
    ///
    /// Shared with the cancel listener background task. The spawn host function
    /// registers/unregisters PIDs; the listener calls `cancel_all()` when a
    /// `tool.v1.request.cancel` event arrives.
    pub process_tracker: Arc<ProcessTracker>,
    /// Live count of `NetStream` entries currently in the resource table.
    /// Maintained alongside `ResourceTable` insertions / drops so the
    /// `MAX_ACTIVE_STREAMS` gate is O(1) instead of iterating every
    /// resource (the table may hold hundreds of pollables / errors /
    /// http handles unrelated to net). Single-threaded: wasmtime
    /// stores are owned by exactly one OS thread.
    pub net_stream_count: usize,
    /// Live count of `SubscriptionEntry` entries. Same rationale as
    /// `net_stream_count`.
    pub subscription_count: usize,
    /// Live count of `ManagedProcess` entries — overall.
    pub process_count_total: usize,
    /// Per-creator-principal count of `ManagedProcess` entries. The
    /// process-spawn gate needs to enforce per-principal sub-budgets
    /// without iterating the whole resource table.
    pub process_count_by_principal:
        std::collections::HashMap<astrid_core::principal::PrincipalId, usize>,
}

impl wasmtime_wasi::WasiView for HostState {
    fn ctx(&mut self) -> wasmtime_wasi::WasiCtxView<'_> {
        wasmtime_wasi::WasiCtxView {
            ctx: &mut self.wasi_ctx,
            table: &mut self.resource_table,
        }
    }
}

impl HostState {
    /// Register a uplink descriptor (called from the host function).
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The per-capsule uplink limit ([`MAX_UPLINKS_PER_CAPSULE`]) has been reached.
    /// - A uplink with the same name and platform already exists.
    pub fn register_uplink(&mut self, descriptor: UplinkDescriptor) -> Result<(), &'static str> {
        if self.registered_uplinks.len() >= MAX_UPLINKS_PER_CAPSULE {
            return Err("uplink registration limit reached");
        }
        // Reject duplicate name+platform combinations
        let duplicate = self
            .registered_uplinks
            .iter()
            .any(|c| c.name == descriptor.name && c.platform == descriptor.platform);
        if duplicate {
            return Err("duplicate uplink name and platform");
        }
        self.registered_uplinks.push(descriptor);
        Ok(())
    }

    /// Return the registered uplinks.
    #[must_use]
    pub fn uplinks(&self) -> &[UplinkDescriptor] {
        &self.registered_uplinks
    }

    /// Create the default host semaphore for bounding concurrent blocking calls.
    ///
    /// Reserves 2 threads for the tokio scheduler and event dispatch, with a
    /// minimum of 2 permits so capsules can always make progress.
    #[must_use]
    pub fn default_host_semaphore() -> Arc<Semaphore> {
        Arc::new(Semaphore::new(
            std::thread::available_parallelism()
                .map(|n| n.get().saturating_sub(2).max(2))
                .unwrap_or(2),
        ))
    }

    /// Set the inbound message sender.
    pub fn set_inbound_tx(&mut self, tx: mpsc::Sender<InboundMessage>) {
        self.inbound_tx = Some(tx);
    }

    /// Return the KV namespace for this capsule scoped to its principal.
    ///
    /// Format: `{principal}:capsule:{capsule_id}`. This is the same namespace
    /// used when the `ScopedKvStore` was created, but exposed here for cases
    /// where host functions need to construct the namespace dynamically.
    #[must_use]
    pub fn principal_kv_namespace(&self) -> String {
        format!("{}:capsule:{}", self.principal, self.capsule_id)
    }

    /// Return the effective KV store for the current invocation.
    ///
    /// Uses `invocation_kv` if set (different principal), falls back to
    /// the capsule's default `kv` store.
    #[must_use]
    pub fn effective_kv(&self) -> &ScopedKvStore {
        #[cfg(debug_assertions)]
        self.debug_assert_invocation_field_set(self.invocation_kv.is_some(), "invocation_kv");
        self.invocation_kv.as_ref().unwrap_or(&self.kv)
    }

    /// Debug-only consistency check: when the caller's principal differs from
    /// the capsule owner's, the corresponding `invocation_*` field **must** be
    /// populated. Otherwise the accessor silently returns the owner's resource
    /// and the invoking principal's reads/writes leak to the owner's scope.
    ///
    /// In practice the setup in [`WasmEngine::invoke_interceptor`] guarantees
    /// `invocation_kv` and `invocation_secret_store` are populated whenever
    /// the principal mismatches (the only failure path is
    /// `ScopedKvStore::with_namespace` rejecting an empty/null-byte namespace,
    /// which our format string never produces). This assertion catches any
    /// regression that breaks that invariant in debug builds.
    ///
    /// Not applied to `invocation_home` / `invocation_tmp` / `invocation_capsule_log`:
    /// those legitimately stay `None` for unregistered principals (the VFS
    /// bundle registration gate and log-open registration gate).
    #[cfg(debug_assertions)]
    fn debug_assert_invocation_field_set(&self, is_set: bool, field_name: &str) {
        let principal_mismatches = self
            .caller_context
            .as_ref()
            .and_then(|m| m.principal.as_deref())
            .and_then(|p| astrid_core::PrincipalId::new(p).ok())
            .is_some_and(|p| p != self.principal);
        if principal_mismatches && !is_set {
            debug_assert!(
                false,
                "invocation principal differs from capsule owner ({owner}) but {field_name} is None — \
                 effective_* accessor would fall back to the owner's resource, leaking reads/writes",
                owner = self.principal,
            );
        }
    }

    /// Return the effective home mount for the current invocation.
    ///
    /// Prefers `invocation_home` (set when serving a different principal)
    /// over `home` (set at capsule load for the owning principal).
    #[must_use]
    pub fn effective_home(&self) -> Option<&PrincipalMount> {
        self.invocation_home.as_ref().or(self.home.as_ref())
    }

    /// Return the effective tmp mount for the current invocation. Same
    /// precedence as [`effective_home`](Self::effective_home).
    #[must_use]
    pub fn effective_tmp(&self) -> Option<&PrincipalMount> {
        self.invocation_tmp.as_ref().or(self.tmp.as_ref())
    }

    /// Owned copy of the effective home root path.
    ///
    /// Convenience for host fs functions that need to pass the principal
    /// home into a security-gate check running inside an `async move` block.
    #[must_use]
    pub fn effective_home_root_buf(&self) -> Option<PathBuf> {
        self.effective_home().map(|m| m.root.clone())
    }

    /// Return the effective secret store for the current invocation.
    ///
    /// Prefers `invocation_secret_store` (set when serving a different
    /// principal) over the load-time `secret_store`.
    #[must_use]
    pub fn effective_secret_store(&self) -> &Arc<dyn SecretStore> {
        #[cfg(debug_assertions)]
        self.debug_assert_invocation_field_set(
            self.invocation_secret_store.is_some(),
            "invocation_secret_store",
        );
        self.invocation_secret_store
            .as_ref()
            .unwrap_or(&self.secret_store)
    }

    /// Return the effective capsule log file for the current invocation.
    ///
    /// Same precedence as [`effective_secret_store`](Self::effective_secret_store).
    /// Returns `None` if neither the invocation nor load-time log is open.
    #[must_use]
    pub fn effective_capsule_log(&self) -> Option<&Arc<std::sync::Mutex<std::fs::File>>> {
        self.invocation_capsule_log
            .as_ref()
            .or(self.capsule_log.as_ref())
    }

    /// Return the principal whose budget should be charged for host-fn
    /// side-effects in the current invocation.
    ///
    /// Prefers the invoking principal from [`caller_context`](Self::caller_context)
    /// (set per-invocation by [`WasmEngine::invoke_interceptor`](super::WasmEngine::invoke_interceptor))
    /// and falls back to the capsule owner's [`principal`](Self::principal) when
    /// no caller is in scope — load-time host calls, tests, and daemons'
    /// self-triggered paths run on the owner's budget, matching the VFS/KV
    /// `effective_*` accessors.
    #[must_use]
    pub fn effective_principal(&self) -> astrid_core::principal::PrincipalId {
        self.caller_context
            .as_ref()
            .and_then(|m| m.principal.as_deref())
            .and_then(|p| astrid_core::principal::PrincipalId::new(p).ok())
            .unwrap_or_else(|| self.principal.clone())
    }

    /// Return the effective quota profile for the current invocation.
    ///
    /// Prefers `invocation_profile` (set by
    /// [`WasmEngine::invoke_interceptor`](super::WasmEngine::invoke_interceptor)
    /// for the calling principal) and falls back to the process-global
    /// [`PrincipalProfile::default_ref`](astrid_core::profile::PrincipalProfile::default_ref)
    /// when no invocation profile is in scope — load-time host calls, tests,
    /// and single-tenant deployments all legitimately run without one.
    ///
    /// The fallback path intentionally does **not** substitute the capsule
    /// owner's profile: that would leak the owner's quotas to every
    /// unauthenticated call path. Using `Default` preserves single-tenant
    /// parity while keeping the security invariant honest.
    #[must_use]
    pub fn effective_profile(&self) -> &astrid_core::profile::PrincipalProfile {
        match self.invocation_profile.as_deref() {
            Some(p) => p,
            None => astrid_core::profile::PrincipalProfile::default_ref(),
        }
    }

    /// Install per-invocation context from an inbound IPC message picked
    /// up via [`ipc::Host::ipc_recv`](crate::engine::wasm::host::ipc) /
    /// [`ipc::Host::ipc_poll`].
    ///
    /// Mirrors the principal-isolation setup done in
    /// [`WasmEngine::invoke_interceptor`](super::WasmEngine::invoke_interceptor)
    /// for the dispatcher path, but driven by `recv`/`poll` so that
    /// `run + ipc::recv` capsules (prompt-builder, registry,
    /// context-engine) also stamp publishes with the publisher's
    /// principal and route reads/writes to the invoking principal's
    /// namespaces. Without this hook these capsules silently fall back
    /// to the owner principal (`default` for the standard distro),
    /// breaking chat for any non-default agent: the publish goes out
    /// stamped `default`, downstream interceptors load the wrong KV
    /// namespace, the turn-state phase doesn't match, and the chain
    /// stalls.
    ///
    /// Sets up the subset relevant to publish stamping and per-principal
    /// KV / log routing:
    /// - [`caller_context`](Self::caller_context) — drives both
    ///   [`effective_principal`](Self::effective_principal) and the
    ///   `principal_str` chosen by `publish_inner`.
    /// - [`invocation_kv`](Self::invocation_kv) — per-principal KV
    ///   namespace; falls back to load-time `kv` on failure.
    /// - [`invocation_capsule_log`](Self::invocation_capsule_log) —
    ///   per-principal log file; falls back to load-time `capsule_log`
    ///   when the principal has no home directory yet.
    ///
    /// Skipped vs the interceptor path (each is independently
    /// recoverable; documenting the gaps so the omissions are
    /// auditable):
    /// - `invocation_home` / `invocation_tmp` / `invocation_secret_store` —
    ///   none of the current run+recv capsules touch home/tmp paths
    ///   or secrets from the recv loop. Add when one starts to.
    /// - `invocation_profile` / `store_limits` — quotas remain the
    ///   capsule owner's. Acceptable for the immediate fix because
    ///   the run+recv capsules are shared singletons whose per-call
    ///   work is bounded by the message size limits already in the
    ///   bus. Move to a real lookup when per-principal quota
    ///   enforcement is needed.
    pub(crate) fn install_recv_invocation_context(&mut self, msg: &astrid_events::ipc::IpcMessage) {
        // Fast path: if the new message's principal matches whatever
        // we already have installed, keep the existing
        // `invocation_kv` / `invocation_capsule_log` rather than
        // re-opening the namespace and log file. The chat-stack run
        // loop calls this on every recv tick — re-init each time
        // burns I/O and allocations for no behavioural change.
        let new_principal = msg.principal.clone();
        let existing_principal = self
            .caller_context
            .as_ref()
            .and_then(|c| c.principal.clone());
        if new_principal == existing_principal {
            // Refresh the caller context so e.g. topic name / payload
            // tracking stays current, but skip the expensive resets.
            self.caller_context = Some(msg.clone());
            return;
        }

        self.caller_context = Some(msg.clone());

        let invocation_principal: Option<astrid_core::PrincipalId> = msg
            .principal
            .as_deref()
            .and_then(|p| astrid_core::PrincipalId::new(p).ok())
            .filter(|p| *p != self.principal);

        let Some(p) = invocation_principal else {
            self.invocation_kv = None;
            self.invocation_capsule_log = None;
            return;
        };

        let ns = format!("{}:capsule:{}", p, self.capsule_id);
        self.invocation_kv = match self.kv.with_namespace(&ns) {
            Ok(kv) => Some(kv),
            Err(e) => {
                tracing::warn!(
                    principal = %p,
                    error = %e,
                    "Failed to create invocation KV scope on ipc::recv path"
                );
                None
            },
        };

        self.invocation_capsule_log = super::open_capsule_log(&p, self.capsule_id.as_str(), false);
    }

    /// Tear down per-invocation context installed by
    /// [`install_recv_invocation_context`](Self::install_recv_invocation_context).
    ///
    /// Called on empty `ipc_poll` / `ipc_recv` returns so a previous
    /// publisher's context doesn't leak into subsequent guest host
    /// calls (KV reads, publishes) that happen before the next recv
    /// pulls fresh state.
    pub(crate) fn clear_recv_invocation_context(&mut self) {
        self.caller_context = None;
        self.invocation_kv = None;
        self.invocation_capsule_log = None;
    }
}

impl std::fmt::Debug for HostState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("HostState")
            .field("capsule_id", &self.capsule_id)
            .field("workspace_root", &self.workspace_root)
            .field("vfs_root_handle", &self.vfs_root_handle)
            .field("has_home", &self.home.is_some())
            .field("has_tmp", &self.tmp.is_some())
            .field("has_security", &self.security.is_some())
            .field("has_uplink_capability", &self.has_uplink_capability)
            .field("has_inbound_tx", &self.inbound_tx.is_some())
            .field("registered_uplinks", &self.registered_uplinks.len())
            .field(
                "host_semaphore_permits",
                &self.host_semaphore.available_permits(),
            )
            .field("cancel_token_cancelled", &self.cancel_token.is_cancelled())
            .field("has_identity_store", &self.identity_store.is_some())
            .field("active_http_streams", &self.active_http_streams.len())
            .field("process_tracker", &self.process_tracker)
            .finish_non_exhaustive()
    }
}

#[cfg(test)]
#[path = "host_state_tests.rs"]
mod tests;