camel_component_wasm/

source_host.rs

//! Source host state, linker functions, and host imports.
//!
//! This module provides:
//! - [`SourceHostState`]: host state for the source world (separate from `WasmHostState`)
//! - [`SourceChannels`]: bounded tokio channels bridging async guest imports to host tasks
//! - `HostWithStore` impl for async `accept-http`, `submit-exchange`; sync `Host` impl for
//!   `is-cancelled`
//! - `run_http_listener`: async axum task feeding HTTP requests into the channel
//! - [`run_pipeline_bridge`]: async task forwarding exchanges to the pipeline

use std::sync::Arc;

use bytes::Bytes;
use tokio::sync::{Notify, mpsc, oneshot};
use tokio_util::sync::CancellationToken;
use wasmtime::component::{Accessor, AccessorTask, HasSelf, Linker, Resource, ResourceTable};

use camel_api::security_policy::{AccessMode, RouteSecurityPlan};
use camel_api::{Body, CamelError, Exchange, ExchangePattern, Message, StreamBody, Value};
use camel_auth::AuthenticatedPrincipal;
use camel_auth::ProviderRegistry;
use camel_component_api::SecurityContext;
use camel_component_api::consumer::ConsumerContext;

use crate::return_stream::{
    DEFAULT_DRAIN_CHANNEL_BOUND, DrainCoord, DrainEvent, StreamReturnable, TerminalSlot,
    drain_guest_stream, receiver_to_body_stream, write_terminal,
};
use crate::source_auth_edge::{EdgeAuthOutcome, authenticate_edge};
use crate::source_bindings::camel::plugin::source_host::{HttpRequest, SubmitOutcome};
use crate::source_bindings::camel::plugin::types::{
    WasmBody, WasmExchange, WasmMessage, WasmPattern,
};

/// Kernel authentication state captured by [`WasmSourceConsumer`] when the
/// route controller delivers the route's [`SecurityContext`] before
/// `start()` (`wasm-source-auth-kernel`, Task 1.3).
///
/// Construction-order lifecycle: the compiled plan and the provider registry
/// arrive from the route's security context through
/// `Consumer::set_security_context` — before the listener binds and before
/// any request reaches the guest. A context lacking either piece leaves the
/// handshake unwired: the classification plan is still retained separately
/// by the consumer, so a non-Public classification without kernel state
/// fails closed per request (Task 2.2) rather than degrading to Public.
// Fields are read by the bind-gate snapshot (Task 1.4) and the host-edge
// handshake (Task 2.2 — plan via credential extraction + kernel mint,
// providers via `kernel_authenticate`).
pub(crate) struct WasmSourceKernelAuth {
    pub(crate) plan: RouteSecurityPlan,
    pub(crate) providers: Arc<ProviderRegistry>,
}

impl WasmSourceKernelAuth {
    /// Capture the kernel state from a route's security context.
    ///
    /// `None` unless both the compiled plan and the provider registry are
    /// present: a plan without providers can never mint a principal, and a
    /// registry without a plan has nothing to enforce.
    pub(crate) fn from_security_context(ctx: &SecurityContext) -> Option<Self> {
        Some(Self {
            plan: ctx.plan.clone()?,
            providers: ctx.providers.clone()?,
        })
    }
}

/// Concrete type for the http-listener resource in the ResourceTable.
/// Stateless — the handle is a marker that the guest holds while running.
/// Mapped via `with:` in source_bindings.rs to replace the empty enum
/// generated by bindgen for the `resource http-listener` declaration.
pub struct HttpListenerHandle;

/// Materialized HTTP request metadata (everything except the body).
///
/// Sent through the request channel alongside a per-request body receiver.
/// The body itself is streamed incrementally through the receiver — never
/// materialized as a single `Vec<u8>` — so large payloads are not capped.
pub struct HttpMeta {
    pub method: String,
    pub path: String,
    pub headers: Vec<(String, String)>,
    /// Principal minted at the host edge before the request entered the
    /// channel (Task 2.1; the edge handshake that mints it lands in Task
    /// 2.2). Private on purpose: the guest never observes it — the
    /// `accept_http` conversion maps only the public fields into the WIT
    /// `HttpRequest` — and `accept_http` stashes it into
    /// [`SourceHostState::pending_principal`] for the Exchange that
    /// `submit-exchange` assembles next.
    principal: Option<AuthenticatedPrincipal>,
}

/// Per-request body chunk receiver. The host's axum handler streams body
/// frames into the matching `mpsc::Sender`; the guest reads via the
/// `stream-body-handle` minted by `accept_http`.
type BodyChunkRx = mpsc::Receiver<Result<Bytes, CamelError>>;

/// Request channel payload: metadata + body streaming receiver.
type RequestChannelItem = (HttpMeta, BodyChunkRx);

// ─── Channel capacity constants ───────────────────────────────────────────

/// Capacity for the HTTP request channel (host listener → guest).
///
/// 1 propagates backpressure all the way to the HTTP client: the axum handler
/// `.send().await`s onto this channel, so once the guest is busy draining a
/// prior exchange into the (also capacity-1) pipeline, further inbound
/// requests park on the send until the guest calls `accept-http` again. A
/// larger buffer would let a burst of clients receive 202 responses for work
/// the pipeline has not yet accepted, silently decoupling client-visible
/// success from actual delivery — undesirable for a source that must not
/// acknowledge messages it may drop on shutdown.
pub const REQUEST_CHANNEL_CAPACITY: usize = 1;

/// Capacity for the exchange channel (guest → pipeline bridge).
/// 1 enforces strict backpressure — guest blocks until pipeline accepts.
pub const EXCHANGE_CHANNEL_CAPACITY: usize = 1;

// ─── SourceChannels ───────────────────────────────────────────────────────

/// Paired channel endpoints for bridging the guest to async host tasks.
///
/// Created once per source consumer instance. The `*_tx` halves go to the
/// HTTP listener and pipeline bridge tasks; `request_rx` lives in
/// [`SourceHostState`] inside the wasmtime `Store`, wrapped in an
/// `Arc<tokio::sync::Mutex<…>>` so the async host import (which receives only
/// `&Accessor`, not `&mut self`) can `lock().await` it to call `recv().await`.
///
/// The request channel carries `(HttpMeta, body_rx)` — metadata plus a
/// per-request bounded channel streaming body chunks. The body is never
/// materialized as a single `Vec<u8>`, removing the old 10 MiB cap.
pub struct SourceChannels {
    pub request_tx: mpsc::Sender<RequestChannelItem>,
    pub request_rx: Arc<tokio::sync::Mutex<mpsc::Receiver<RequestChannelItem>>>,
    pub exchange_tx: mpsc::Sender<(Exchange, oneshot::Sender<SubmitOutcome>)>,
    pub exchange_rx: mpsc::Receiver<(Exchange, oneshot::Sender<SubmitOutcome>)>,
}

impl SourceChannels {
    /// Create a new set of channels with the documented capacities.
    pub fn new() -> Self {
        let (request_tx, request_rx) = mpsc::channel(REQUEST_CHANNEL_CAPACITY);
        let (exchange_tx, exchange_rx) = mpsc::channel(EXCHANGE_CHANNEL_CAPACITY);
        Self {
            request_tx,
            request_rx: Arc::new(tokio::sync::Mutex::new(request_rx)),
            exchange_tx,
            exchange_rx,
        }
    }
}

impl Default for SourceChannels {
    fn default() -> Self {
        Self::new()
    }
}

// ─── SourceHostState ──────────────────────────────────────────────────────

/// Default maximum request body size in bytes (10 MiB).
///
/// Matches [`camel_api::body::DEFAULT_MATERIALIZE_LIMIT`] — the same cap the
/// old `to_bytes` path enforced. Restores the DoS backstop that was
/// inadvertently removed when the body switched from materialized to streamed.
pub const DEFAULT_MAX_REQUEST_BODY_BYTES: u64 = 10 * 1024 * 1024;

/// Host state for a source consumer. Lives inside the `wasmtime::Store`.
///
/// Separate from [`crate::runtime::WasmHostState`] because the source world
/// has a fundamentally different lifecycle: the guest IS the source, owns a
/// run loop, and communicates via channels rather than direct call-process.
///
/// `request_rx` is wrapped in `Arc<tokio::sync::Mutex<…>>` because the async
/// host imports receive `&Accessor` (shared access), not `&mut self`; the
/// Mutex provides the interior mutability needed to `recv().await` the
/// receiver across import calls.
pub struct SourceHostState {
    pub table: ResourceTable,
    pub wasi: wasmtime_wasi::WasiCtx,
    pub request_rx: Arc<tokio::sync::Mutex<mpsc::Receiver<RequestChannelItem>>>,
    pub exchange_tx: mpsc::Sender<(Exchange, oneshot::Sender<SubmitOutcome>)>,
    pub cancel_token: CancellationToken,
    /// Maximum bytes the guest may read from a single request body before
    /// the stream producer terminates with an overflow error. Restores the
    /// DoS backstop that the old `to_bytes(MAX_BODY_BYTES)` path provided.
    pub max_request_body_bytes: u64,
    /// Principal minted at the host edge for the most recently accepted,
    /// not-yet-submitted request (Task 2.1). Stashed by `accept_http`,
    /// consumed by `submit_exchange` when the typed carrier is installed on
    /// the native Exchange. `None` while no request is in flight or when the
    /// edge handshake minted no principal.
    pub pending_principal: Option<AuthenticatedPrincipal>,
    /// One-outstanding-request invariant marker: an `accept-http` returned
    /// without an intervening `submit-exchange`. While set, a further
    /// `accept-http` fails closed instead of overwriting
    /// `pending_principal` — the WIT contract does not force accept/submit
    /// alternation, so the host enforces it for identity integrity.
    pub accept_outstanding: bool,
}

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

// ─── Pending-principal slot (Task 2.1) ─────────────────────────────────────

impl SourceHostState {
    /// Stash the principal of a just-accepted request and mark the
    /// accept/submit slot outstanding.
    ///
    /// One-outstanding-request invariant: the WIT contract does not force
    /// accept/submit alternation, and a buggy guest may accept twice — the
    /// host must not let a second accept overwrite the first request's
    /// identity. Returns `Err` (fail closed) when an accept is already
    /// outstanding; the caller surfaces it as the WIT error result for the
    /// second accept. `Ok` replaces the slot unconditionally (the previous
    /// request, if any, was already submitted and consumed it).
    fn stash_pending_principal(
        &mut self,
        principal: Option<AuthenticatedPrincipal>,
    ) -> Result<(), &'static str> {
        if self.accept_outstanding {
            return Err(
                "accept-http rejected: a previous request is outstanding; submit-exchange it before accepting again",
            );
        }
        self.accept_outstanding = true;
        self.pending_principal = principal;
        Ok(())
    }

    /// Take the stashed principal, install it as the submitted Exchange's
    /// typed carrier, and clear the outstanding marker — the slot is free
    /// for the next accept.
    ///
    /// Called by `submit-exchange` after the native Exchange is assembled
    /// and before it is sent down the pipeline channel.
    fn install_pending_carrier(&mut self, exchange: &mut Exchange) {
        if let Some(principal) = self.pending_principal.take() {
            camel_auth::install_carrier(exchange, &principal);
        }
        self.accept_outstanding = false;
    }

    /// Clear the pending-principal slot and the outstanding marker after a
    /// failed accept (body-assembly failure): the request never reaches the
    /// guest, so the slot must not poison a follow-up accept under the
    /// one-outstanding-request invariant.
    fn abort_pending_accept(&mut self) {
        self.pending_principal = None;
        self.accept_outstanding = false;
    }
}

// ─── Host trait implementation ────────────────────────────────────────────

type SourceWasmError = crate::source_bindings::camel::plugin::types::WasmError;

// The http-listener resource has no methods in WIT, but wasmtime bindgen
// generates a HostHttpListener trait that must be implemented.
impl crate::source_bindings::camel::plugin::source_host::HostHttpListener for SourceHostState {
    fn drop(
        &mut self,
        _resource: wasmtime::component::Resource<HttpListenerHandle>,
    ) -> wasmtime::Result<()> {
        // HttpListenerHandle is stateless; nothing to clean up.
        Ok(())
    }
}

impl crate::source_bindings::camel::plugin::source_host::Host for SourceHostState {
    fn is_cancelled(&mut self) -> bool {
        // Stays sync (a `with`-style peek that must not yield). Returns
        // immediately so tight guest loops can poll cancellation cheaply.
        self.cancel_token.is_cancelled()
    }
}

/// Async host imports. The guest's `run` (async) awaits these; each receives
/// `&Accessor<SourceHostState, HasSelf<SourceHostState>>` (shared access into
/// the store, NOT `&mut self`). We snapshot owned values (clones) out of a
/// `with` call, then `.await` OUTSIDE it — `Accessor::with` is sync and must
/// not be held across an await. Cancellation races every await against the
/// shared `CancellationToken` so `stop()` unblocks a parked import promptly.
impl crate::source_bindings::camel::plugin::source_host::HostWithStore<SourceHostState>
    for wasmtime::component::HasSelf<SourceHostState>
{
    async fn accept_http(
        accessor: &Accessor<SourceHostState, wasmtime::component::HasSelf<SourceHostState>>,
        _listener: Resource<HttpListenerHandle>,
    ) -> Result<Option<HttpRequest>, SourceWasmError> {
        // Snapshot clones via `with`, then await outside. request_rx is an
        // Arc<Mutex<Receiver>>; cloning the Arc is cheap and lets us lock it
        // across the await without borrowing the accessor.
        // Single `with` snapshot — one TLS lookup for all three fields.
        let (request_rx, cancel_token, max_body) = accessor.with(|mut view| {
            let state = view.get();
            (
                state.request_rx.clone(),
                state.cancel_token.clone(),
                state.max_request_body_bytes,
            )
        });

        // Uncontended lock: the guest runs a single sequential run loop, so
        // only one accept_http is in flight at a time.
        let mut guard = request_rx.lock().await;
        let meta_and_body = tokio::select! {
            r = guard.recv() => r,
            _ = cancel_token.cancelled() => {
                return Ok(None);
            }
        };
        // Drop the lock before minting the StreamReader (needs accessor.with).
        drop(guard);

        // None => channel closed (listener exited); treat as cancellation.
        let Some((mut meta, mut body_rx)) = meta_and_body else {
            return Ok(None);
        };

        // One-outstanding-request invariant (Task 2.1): stash the
        // edge-minted principal for the Exchange `submit-exchange` assembles
        // next. A second accept while one is outstanding fails closed — the
        // earlier request's identity is never overwritten.
        if let Err(msg) =
            accessor.with(|mut view| view.get().stash_pending_principal(meta.principal.take()))
        {
            // log-policy: handler-owned — the violation belongs to the guest.
            tracing::warn!("source: accept-http invariant: {msg}");
            return Err(SourceWasmError::ProcessorError(msg.to_string()));
        }

        // Build a BoxStream from the per-request body channel via poll_fn.
        // NOT receiver_to_body_stream — that is DrainEvent-specific (the
        // submit-exchange direction). Here the channel carries
        // Result<Bytes, CamelError> directly.
        let box_stream: futures::stream::BoxStream<'static, Result<Bytes, CamelError>> =
            Box::pin(futures::stream::poll_fn(move |cx| body_rx.poll_recv(cx)));

        // Mint the StreamReader + terminal FutureReader inside accessor.with.
        // assemble_stream_body_source takes &Accessor<S, U> and internally
        // calls accessor.with for each reader creation.
        let cancel = cancel_token.clone();
        let handle = crate::stream_bridge::assemble_stream_body_source(
            accessor, box_stream, cancel, max_body,
        );
        let request = match handle {
            Ok(body) => Some(HttpRequest {
                method: meta.method,
                path: meta.path,
                headers: meta.headers,
                body,
            }),
            Err(e) => {
                // log-policy: system-broken
                tracing::error!("failed to assemble stream body for accept-http: {e}");
                // The request is dropped on this failure return; clear the
                // stashed principal slot and the outstanding marker so a
                // follow-up accept is not poisoned by the one-outstanding
                // invariant.
                accessor.with(|mut view| view.get().abort_pending_accept());
                None
            }
        };
        Ok(request)
    }

    async fn submit_exchange(
        accessor: &Accessor<SourceHostState, wasmtime::component::HasSelf<SourceHostState>>,
        mut exchange: WasmExchange,
    ) -> Result<SubmitOutcome, SourceWasmError> {
        // Snapshot owned values (clones) + extract the stream (if any) inside a
        // single `with`. The closure must return before `spawn` is called:
        // `Accessor::spawn` internally calls `with`, which panics if nested
        // (concurrent.rs:454). `take_stream` replaces a `WasmBody::Stream` with
        // `Empty`, moving the live `StreamReader`/`FutureReader` out so the
        // metadata conversion below does not trip over the non-cloneable handle.
        let (exchange_tx, cancel_token, stream_parts) = accessor.with(|mut view| {
            let state = view.get();
            let exchange_tx = state.exchange_tx.clone();
            let cancel_token = state.cancel_token.clone();
            let stream_parts = exchange.take_stream();
            (exchange_tx, cancel_token, stream_parts)
        });
        // `with` guard dropped here — safe to call `accessor.spawn` below.

        // Build the native Exchange from the (now stream-stripped) metadata.
        // The body is `Empty` for the stream case; it is overwritten with
        // `Body::Stream` after the drain is spawned.
        let mut native = source_exchange_to_native(exchange);

        // Install the typed carrier from the principal stashed by
        // accept-http (Task 2.1) BEFORE the exchange reaches the pipeline.
        accessor.with(|mut view| view.get().install_pending_carrier(&mut native));

        let (reply_tx, reply_rx) = oneshot::channel();

        if let Some((stream_reader, terminal, stream_metadata)) = stream_parts {
            // Streaming submit-exchange: fire-and-return.
            //
            // The guest emits a `stream<u8>` body. Reading a wasmtime stream
            // requires the concurrent execution context (an `&Accessor`), which
            // is only live while the event loop driving `run` polls. We CANNOT
            // `tokio::spawn` a moved store + `run_concurrent` (the source
            // world's `run` is already inside `run_concurrent`;
            // `check_recursive_run` panics). Instead, `Accessor::spawn` queues
            // the drain onto the SAME event loop, where it progresses
            // concurrently with the guest fiber.
            let (chunk_tx, chunk_rx) = mpsc::channel::<DrainEvent>(DEFAULT_DRAIN_CHANNEL_BOUND);
            // Non-blocking terminal slot: written BEFORE sender drop so the
            // consumer observes the terminal result after buffered chunks drain.
            // Created ONCE and shared between the drain task and the body stream.
            let source_terminal: TerminalSlot = Arc::new(std::sync::Mutex::new(None));
            // progress/receiver_gone unused under source (no watchdog);
            // required by shared drain_guest_stream signature.
            let coord = DrainCoord {
                cancel: cancel_token.clone(),
                progress: Arc::new(Notify::new()),
                receiver_gone: Arc::new(Notify::new()),
                terminal_slot: source_terminal.clone(),
            };

            // Register the drain BEFORE sending the exchange. Ordering is
            // load-bearing: the chunk-channel receiver only gets a consumer
            // (the downstream `Body::Stream` reader) once the exchange is
            // delivered. If we awaited drain completion before sending, the
            // bounded chunk channel would fill, backpressure-stall the drain,
            // and self-deadlock.
            // wasmtime 48: Accessor::spawn is fallible (returns Err when the
            // concurrent event loop is already gone). On failure the drain
            // never runs, so the downstream body stream would hang forever —
            // fail fast by writing the terminal error the drain would have
            // written; chunk_tx drops inside spawn, closing the channel.
            // write_terminal is poison-safe: the error always surfaces.
            if let Err(spawn_err) = accessor.spawn(SubmitExchangeDrain {
                stream_reader,
                terminal,
                chunk_tx,
                coord,
            }) {
                write_terminal(
                    &source_terminal,
                    Err(CamelError::ProcessorError(format!(
                        "stream drain spawn failed: {spawn_err}"
                    ))),
                );
            }

            // Attach the lazy body. Downstream reads from this receiver; the
            // spawned drain feeds it from the guest's stream.
            native.input.body = Body::Stream(StreamBody {
                stream: Arc::new(tokio::sync::Mutex::new(Some(Box::pin(
                    receiver_to_body_stream(crate::return_stream::DrainReceiver {
                        rx: chunk_rx,
                        terminal: source_terminal,
                    }),
                )))),
                metadata: stream_metadata,
            });
        }

        // Send the (native) exchange and await the pipeline's acceptance reply.
        // submit-exchange returns once the pipeline ACCEPTS the envelope —
        // before full body drain (fire-and-return). Race against cancellation
        // so `stop()` unblocks a parked submit promptly.
        let outcome = tokio::select! {
            res = async {
                if exchange_tx.send((native, reply_tx)).await.is_err() {
                    // Bridge gone (clean shutdown) — report Stopped.
                    return SubmitOutcome::Stopped;
                }
                // Pipeline bridge replies with the delivery outcome; a dropped
                // reply (bridge exited after accept) is also a clean stop.
                reply_rx.await.unwrap_or(SubmitOutcome::Stopped)
            } => res,
            _ = cancel_token.cancelled() => SubmitOutcome::Stopped,
        };
        Ok(outcome)
    }
}

// ─── submit-exchange background drain (AccessorTask) ──────────────────────

/// `AccessorTask` driving the guest→host body drain for a streaming
/// `submit-exchange`.
///
/// Unlike the plugin/bean return path (`spawn_return_drain`), the source world
/// CANNOT `tokio::spawn` a moved store and open a fresh `run_concurrent` — the
/// source `run` is already inside `run_concurrent` and `check_recursive_run`
/// panics on a nested call. `Accessor::spawn` instead registers this task on
/// the SAME event loop already driving `run`, so the drain progresses
/// concurrently with the guest fiber (and the guest's `spawn_local` writer).
struct SubmitExchangeDrain {
    stream_reader: wasmtime::component::StreamReader<u8>,
    terminal: wasmtime::component::FutureReader<Result<(), SourceWasmError>>,
    chunk_tx: mpsc::Sender<DrainEvent>,
    coord: DrainCoord,
}

impl AccessorTask<SourceHostState, HasSelf<SourceHostState>> for SubmitExchangeDrain {
    async fn run(
        self,
        accessor: &Accessor<SourceHostState, HasSelf<SourceHostState>>,
    ) -> wasmtime::Result<()> {
        drain_guest_stream::<SourceWasmError, SourceHostState>(
            accessor,
            self.stream_reader,
            self.terminal,
            self.chunk_tx,
            self.coord,
        )
        .await;
        // Terminal errors are surfaced through the chunk channel
        // (`DrainEvent::Error`), not the task return — the downstream
        // `Body::Stream` reader observes them. Returning `Ok` keeps the
        // event loop's task accounting clean.
        Ok(())
    }
}

// ─── source WasmExchange → native Exchange conversion ──────────────────────

/// Convert a (stream-stripped) source-world `WasmExchange` into a native
/// [`Exchange`]. Inlines the `serde_bridge` field-mapping because the source
/// bindings generate distinct types from the plugin bindings (TODO WIT-001).
///
/// A `WasmBody::Stream` never reaches here: `submit_exchange` calls
/// `take_stream` (replacing it with `Empty`) before conversion and attaches a
/// `Body::Stream` separately.
fn source_exchange_to_native(wasm: WasmExchange) -> Exchange {
    let input = source_message_to_native(wasm.input);
    let mut exchange = Exchange::new(input);
    // Preserve the guest's correlation_id (Exchange::new generates a random one).
    if !wasm.correlation_id.is_empty() {
        exchange.correlation_id = wasm.correlation_id;
    }
    if let Some(out) = wasm.output {
        exchange.output = Some(source_message_to_native(out));
    }
    for (key, raw) in wasm.properties {
        let value = serde_json::from_str::<Value>(&raw).unwrap_or(Value::String(raw));
        exchange.properties.insert(key, value);
    }
    exchange.pattern = match wasm.pattern {
        WasmPattern::InOnly => ExchangePattern::InOnly,
        WasmPattern::InOut => ExchangePattern::InOut,
    };
    exchange
}

fn source_message_to_native(msg: WasmMessage) -> Message {
    let headers = msg
        .headers
        .into_iter()
        .map(|(k, raw)| {
            let value = serde_json::from_str::<Value>(&raw).unwrap_or(Value::String(raw));
            (k, value)
        })
        .collect();
    Message {
        headers,
        body: source_body_to_native(msg.body),
    }
}

fn source_body_to_native(body: WasmBody) -> Body {
    match body {
        WasmBody::Empty => Body::Empty,
        WasmBody::Text(s) => Body::Text(s),
        WasmBody::Bytes(v) => Body::Bytes(Bytes::from(v)),
        WasmBody::Json(s) => serde_json::from_str::<Value>(&s)
            .map(Body::Json)
            .unwrap_or(Body::Text(s)),
        WasmBody::Xml(s) => Body::Xml(s),
        // Unreachable: `submit_exchange` extracts the stream via `take_stream`
        // before conversion. Defensive guard (drop + warn) mirrors
        // `serde_bridge::wasm_to_body`.
        WasmBody::Stream(_) => {
            tracing::warn!("source: undrained stream body reached conversion — dropped");
            Body::Empty
        }
    }
}

// ─── Linker setup ─────────────────────────────────────────────────────────

/// Register the source-host interface and WASI p2 into a linker for [`SourceHostState`].
///
/// Uses the ASYNCHRONOUS WASI linker: the guest's `run` export is async and is
/// driven by `Store::run_concurrent` on a tokio task (see
/// [`crate::source_consumer`]). Its async host imports (`accept-http`,
/// `submit-exchange`) receive an `&Accessor` and use async channel ops.
pub fn add_to_linker(linker: &mut Linker<SourceHostState>) -> Result<(), wasmtime::Error> {
    // WASI p2 preview — register the command-adapter surface that the
    // `wasm32-wasip2` source fixture imports (hardened ctx backs it with no
    // resources). Filesystem and sockets stay unregistered (ADR-0050).
    crate::wasi_surface::register_command_adapter_wasi(linker)?;

    // Source-host interface (accept-http, submit-exchange, is-cancelled).
    crate::source_bindings::camel::plugin::source_host::add_to_linker::<_, HasSelf<_>>(
        linker,
        |state| state,
    )?;

    Ok(())
}

// ─── HTTP listener task (async) ───────────────────────────────────────────

/// Run an axum HTTP server that feeds incoming requests into `request_tx`.
///
/// The caller must have already bound the TCP listener (see
/// [`crate::source_consumer::WasmSourceConsumer::start`]) so that a bind
/// failure surfaces synchronously as a `start()` error rather than a
/// background warning that leaves the route appearing healthy with no
/// listener accepting requests.
///
/// Each request's body is streamed incrementally into a per-request bounded
/// channel — the old 10 MiB `to_bytes` materialization cap is removed. The
/// guest reads the body via the `stream-body-handle` returned by
/// `accept-http`.
///
/// # Response timing (fire-and-forget semantic)
///
/// Returns **202 Accepted** once the request metadata is handed to the guest
/// via the request channel. The body is streamed asynchronously *after* the
/// response is sent — the handler does NOT wait for the full body before
/// responding. This is inherent to the streaming shape: you cannot stream
/// and wait for full receipt simultaneously. Mid-body connection drops
/// surface as stream errors to the guest (the body channel receives an
/// `Err` frame), not as HTTP-level failures to the client. The source
/// accepts the request for processing; it does not guarantee full body
/// receipt.
///
/// # Host-edge authentication (Task 2.2)
///
/// For non-Public classifications the handler runs the kernel handshake
/// (see [`crate::source_auth_edge::authenticate_edge`]) BEFORE the request
/// channel is touched: a denial renders 401 (`unauthenticated`) and returns
/// without `tx.send` and without reading the body — the guest never wakes.
///
/// Shuts down gracefully when `cancel` is triggered.
pub(crate) async fn run_http_listener(
    listener: tokio::net::TcpListener,
    path_filter: Option<String>,
    request_tx: mpsc::Sender<RequestChannelItem>,
    cancel: CancellationToken,
    kernel: Option<Arc<WasmSourceKernelAuth>>,
    plan_access: Option<AccessMode>,
) -> Result<(), CamelError> {
    use axum::Router;
    use axum::extract::State;
    use axum::http::Request;
    use axum::response::Response;
    use axum::routing::any;
    use http_body_util::BodyExt;

    struct ListenerState {
        tx: mpsc::Sender<RequestChannelItem>,
        cancel: CancellationToken,
        kernel: Option<Arc<WasmSourceKernelAuth>>,
        plan_access: Option<AccessMode>,
    }

    async fn handler(
        State(state): State<Arc<ListenerState>>,
        req: Request<axum::body::Body>,
    ) -> Response {
        let (parts, body) = req.into_parts();

        // Host-edge kernel handshake (Task 2.2) BEFORE the request channel
        // is touched. Driven by the classification (`plan_access`), never by
        // kernel presence alone: absent/Public classifications pass through,
        // non-Public ones must mint or deny (fail-closed without wiring).
        // A denial returns without `tx.send` and without reading the body.
        let principal = match authenticate_edge(
            state.plan_access.as_ref(),
            state.kernel.as_deref(),
            &parts.headers,
            &parts.uri,
        )
        .await
        {
            Ok(EdgeAuthOutcome::PassThrough) => None,
            Ok(EdgeAuthOutcome::Authenticated(principal)) => Some(*principal),
            Err(status) => {
                return Response::builder()
                    .status(status)
                    .body(axum::body::Body::from("unauthenticated"))
                    .unwrap(); // allow-unwrap
            }
        };

        // Per-request bounded channel for streaming the body to the guest.
        // Capacity matches the drain-channel bound used elsewhere for
        // guest→host streaming (return_stream.rs).
        let (body_tx, body_rx) = mpsc::channel::<Result<Bytes, CamelError>>(
            crate::return_stream::DEFAULT_DRAIN_CHANNEL_BOUND,
        );

        let http_meta = HttpMeta {
            method: parts.method.to_string(),
            path: parts.uri.path().to_string(),
            headers: parts
                .headers
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
                .collect(),
            // Principal minted at the host edge above (Task 2.2); `None`
            // for pass-through classifications and unauthenticated-free
            // flows.
            principal,
        };

        if state.tx.send((http_meta, body_rx)).await.is_err() {
            return Response::builder()
                .status(503)
                .body(axum::body::Body::from("service unavailable"))
                .unwrap(); // allow-unwrap
        }

        // Stream the axum body into body_tx incrementally. The spawned task
        // owns the body and the sender; when the body is fully consumed (or
        // errors), body_tx drops = EOF on the guest's StreamReader.
        // Racing against cancel ensures the drain exits promptly on source
        // shutdown rather than relying solely on connection-close propagation.
        let drain_cancel = state.cancel.clone();
        tokio::spawn(async move {
            let mut body = body;
            loop {
                tokio::select! {
                    frame = body.frame() => {
                        match frame {
                            Some(Ok(frame)) => {
                                if let Ok(data) = frame.into_data()
                                    && body_tx.send(Ok(data)).await.is_err()
                                {
                                    // Guest stopped reading — abandon the stream.
                                    break;
                                }
                                // Trailers frame: silently ignored (HTTP metadata, not body bytes).
                            }
                            Some(Err(e)) => {
                                let _ = body_tx
                                    .send(Err(CamelError::Io(format!("body read: {e}"))))
                                    .await;
                                break;
                            }
                            // None = clean EOF from hyper — body fully consumed.
                            None => break,
                        }
                    }
                    _ = drain_cancel.cancelled() => break,
                }
            }
            // body_tx drops here → guest observes clean EOF.
        });

        Response::builder()
            .status(202)
            .body(axum::body::Body::from("accepted"))
            .unwrap() // allow-unwrap
    }

    let state = Arc::new(ListenerState {
        tx: request_tx,
        cancel: cancel.clone(),
        kernel,
        plan_access,
    });

    let route_path = path_filter
        .filter(|path| !path.is_empty())
        .map(|path| {
            if path.starts_with('/') {
                path
            } else {
                format!("/{path}")
            }
        })
        .unwrap_or_else(|| "/{*path}".to_string());

    let app = Router::new()
        .route(&route_path, any(handler))
        .with_state(state);

    // Listener is already bound by the caller; capture the address for logging.
    let local = listener.local_addr().ok();

    if let Some(addr) = &local {
        tracing::info!(%addr, "source HTTP listener started");
    }

    axum::serve(listener, app)
        .with_graceful_shutdown(async move { cancel.cancelled().await })
        .await
        .map_err(|e| CamelError::Io(format!("HTTP listener error: {e}")))?;

    if let Some(addr) = &local {
        tracing::info!(%addr, "source HTTP listener stopped");
    }
    Ok(())
}

// ─── Pipeline bridge task (async) ─────────────────────────────────────────

/// Receive native exchanges from the guest's `submit-exchange` import and
/// forward them to the pipeline via [`ConsumerContext`].
///
/// The WasmExchange→native conversion (and streaming-body drain setup) happens
/// inside the import now (see `source_exchange_to_native`), so the bridge is
/// reduced to a pure forwarder. Sends [`SubmitOutcome::Accepted`] /
/// [`SubmitOutcome::Stopped`] back to the guest via the oneshot reply channel.
pub async fn run_pipeline_bridge(
    mut exchange_rx: mpsc::Receiver<(Exchange, oneshot::Sender<SubmitOutcome>)>,
    ctx: ConsumerContext,
) -> Result<(), CamelError> {
    while let Some((exchange, reply_tx)) = exchange_rx.recv().await {
        let outcome = match ctx.send(exchange).await {
            Ok(()) => SubmitOutcome::Accepted,
            Err(_) => SubmitOutcome::Stopped,
        };
        // If the guest has already abandoned the exchange (unlikely), this
        // send fails silently.
        let _ = reply_tx.send(outcome);
    }
    Ok(())
}

// ─── Tests ────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests;