featherbit/graph/

engine.rs

//! Policy compilation and graph execution.
//!
//! Turns a [`PolicyConfig`] (nodes + `from`/`to` edges from `gateway.yaml`)
//! into a [`CompiledGraph`] with instantiated plugins, then walks that graph
//! per request: success edges on `Ok`, error edges (or the policy-level
//! catch-all handler) on failure, stopping at terminal `client` nodes.

use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use serde::Serialize;

use crate::config::{NodeConfig, PolicyConfig};
use crate::context::Context;
use crate::debug::{EdgeKind, StepOutcome, TraceRecorder};
use crate::plugins::resources::PluginResources;
use crate::plugins::{self, Plugin};

/// A compiled, ready-to-execute graph with instantiated plugins.
///
/// Built by [`compile_policy`] and shared read-only by the data-plane; one
/// instance serves all requests matching the route that references its policy.
pub struct CompiledGraph {
    nodes: HashMap<String, Box<dyn Plugin>>,
    /// node_id -> (port -> target node_id). Ports normalized (`out` -> `success`).
    edges: HashMap<String, HashMap<String, String>>,
    /// The node that starts the pipeline (first node after listener)
    entry_node_id: String,
    /// Node IDs that are terminal (client nodes) — execution stops when we reach one
    terminal_node_ids: HashSet<String>,
    /// Policy-level catch-all error handler node ID
    catch_all_handler: Option<String>,
    /// Policy name, used as a metrics label.
    policy_name: String,
    /// Process-wide services (metrics registry, shared clients); per-node
    /// metrics recording is disabled when `resources.metrics` is `None`.
    resources: Arc<PluginResources>,
    /// Ids of `upstream` nodes whose `success` path reaches `client` without
    /// passing any node that reads the response body. Read by
    /// `is_stream_capable`, consulted per node in `run`.
    stream_capable: HashSet<String>,
    /// Why each non-capable upstream must buffer, for operator-visible reporting.
    buffering_reasons: Vec<BufferingReason>,
    cache_pair_warnings: Vec<CachePairWarning>,
    /// Every `proxy-cache` half's backend, for invalidation by pair `id`.
    cache_targets: Vec<crate::traffic::CacheTarget>,
}

/// Records that one node on an upstream's success path forces buffering.
///
/// Serializes as `{"upstream": ..., "blocked_by": ..., "node_type": ...}` —
/// the shape the Admin API's policy-validate endpoint and the MCP
/// One half of a `proxy-cache` pair with no counterpart.
///
/// A lookup with no store caches nothing; a store with no lookup is never
/// read. Both are silent no-ops rather than errors -- and both are also what
/// a policy looks like halfway through being built -- so they are reported
/// the way a buffering reason is rather than refused.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct CachePairWarning {
    /// The shared `id` that links a pair.
    pub cache_id: String,
    /// The half that is present.
    pub present_node_id: String,
    /// The half that is missing: `"lookup"` or `"store"`.
    pub missing_role: String,
}

/// `validate_policy` tool both report to operators/agents.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct BufferingReason {
    #[serde(rename = "upstream")]
    pub upstream_node_id: String,
    #[serde(rename = "blocked_by")]
    pub blocked_by_node_id: String,
    pub node_type: String,
}

// Manual impl: `Box<dyn Plugin>` doesn't implement `Debug`, so `#[derive]`
// isn't available. This exists solely so `Result<CompiledGraph, String>` can
// be `.unwrap_err()`'d in tests; the node table is summarized by id only.
impl std::fmt::Debug for CompiledGraph {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CompiledGraph")
            .field("policy_name", &self.policy_name)
            .field("node_ids", &self.nodes.keys().collect::<Vec<_>>())
            .field("edges", &self.edges)
            .field("entry_node_id", &self.entry_node_id)
            .field("terminal_node_ids", &self.terminal_node_ids)
            .field("catch_all_handler", &self.catch_all_handler)
            .finish()
    }
}

impl CompiledGraph {
    /// Executes the graph with the given initial context and returns the
    /// final context (with `response` populated).
    ///
    /// Walks nodes starting at the entry node (the first node after the
    /// listener), following the success edge after each `Ok` result. On a
    /// plugin error the error is tagged with the failing node's ID, pushed
    /// onto `ctx.errors`, and execution jumps to the node's `error` edge if
    /// one exists, otherwise to the policy-level catch-all handler; with
    /// neither, execution stops and a generic JSON 500 response is written.
    /// Reaching a terminal `client` node (or a node with no success edge)
    /// ends the walk. This method never fails: every outcome is expressed
    /// through the returned context's response.
    pub async fn execute(&self, ctx: Context) -> Context {
        self.run(ctx, None).await.0
    }

    /// Same walk as [`execute`](Self::execute), additionally recording a
    /// [`Trace`] of every node: the context after each one, the outcome, and
    /// which edge the engine followed.
    ///
    /// Used by debug mode and the sandbox. Deliberately the *same* loop rather
    /// than a parallel implementation — a trace that could drift from the real
    /// execution path would be worse than no trace at all.
    pub async fn execute_traced(
        &self,
        ctx: Context,
        recorder: TraceRecorder,
    ) -> (Context, TraceRecorder) {
        let (ctx, rec) = self.run(ctx, Some(recorder)).await;
        (
            ctx,
            rec.expect("recorder is returned when one was supplied"),
        )
    }

    /// The single graph walk, optionally recording each step.
    ///
    /// When `recorder` is `None` the only added cost is one `Option`
    /// discriminant check per node — the same shape as the existing metrics
    /// branch, and perfectly branch-predicted.
    async fn run(
        &self,
        mut ctx: Context,
        mut recorder: Option<TraceRecorder>,
    ) -> (Context, Option<TraceRecorder>) {
        let mut current_node_id = self.entry_node_id.clone();

        loop {
            let node = match self.nodes.get(&current_node_id) {
                Some(n) => n,
                None => {
                    ctx.response.status_code = 500;
                    ctx.response.body = bytes::Bytes::from(format!(
                        "Graph error: node '{}' not found",
                        current_node_id
                    ));
                    // Same rule as the no-handler 500 fallback and
                    // `ErrorHandlerPlugin::execute`: this overwrites
                    // `response.body`, so a stream left by an earlier node
                    // (reachable here because `compile_policy` validates only
                    // `from_node`, never `to_node` — see
                    // `infer_stream_capability`'s doc below) must not survive
                    // alongside it.
                    ctx.response.stream = None;
                    if let Some(r) = recorder.as_mut() {
                        r.record_step(
                            &current_node_id,
                            "<missing>",
                            StepOutcome::Error {
                                code: "NODE_NOT_FOUND".to_string(),
                                message: format!("node '{}' not found", current_node_id),
                            },
                            std::time::Duration::ZERO,
                            EdgeKind::NodeNotFound,
                            None,
                            None,
                            &ctx,
                        );
                    }
                    break;
                }
            };

            let node_type = node.plugin_type().to_string();
            // Reserved `context.message` key, same mechanism as `__policy`,
            // `__route`, and `__request_start_ms` (set by the listener before
            // the graph runs; see `src/server/listener.rs`): tells a
            // stream-capable `upstream` node it may hand back a
            // `response.stream` instead of buffering.
            //
            // Written unconditionally (never merely inserted when `true`) on
            // every node, capable or not: `ctx`/`ctx.message` survives an
            // error-port hop, so on a failover shape (a capable `upstream`'s
            // `error` port routed to a non-capable one) a stale `true` left
            // by the failed node would otherwise still be sitting in
            // `ctx.message` when the non-capable node runs, telling it to
            // stream when it must not — silently bypassing whatever reads
            // the buffered body downstream of it (a `response-rewrite`, a
            // logger). Always overwriting with this node's own capability
            // closes that: a non-capable node always sees `false`.
            ctx.message.insert(
                "__may_stream".to_string(),
                serde_json::json!(self.is_stream_capable(&current_node_id)),
            );
            let started = std::time::Instant::now();
            let result = node.execute(ctx).await;
            let elapsed = started.elapsed();
            if let Some(ref m) = self.resources.metrics {
                m.node_execution_count
                    .with_label_values(&[&self.policy_name, &current_node_id, &node_type])
                    .inc();
                m.node_execution_duration
                    .with_label_values(&[&self.policy_name, &current_node_id])
                    .observe(elapsed.as_secs_f64());
            }

            match result {
                Ok(output) => {
                    ctx = output.context;
                    let port = output.port.unwrap_or("success");
                    // `error` is reachable only through `Err`, which records a
                    // GatewayError and follows the error/catch-all fallback
                    // chain. An `Ok` naming it would silently take an error
                    // edge with no error record attached.
                    debug_assert!(
                        port != "error",
                        "plugins must not emit on the error port via Ok"
                    );

                    // If this is a terminal node (client), we're done
                    let terminal = self.terminal_node_ids.contains(&current_node_id);
                    let next = if terminal {
                        None
                    } else {
                        self.edges.get(&current_node_id).and_then(|m| m.get(port))
                    };
                    if let Some(r) = recorder.as_mut() {
                        let edge = if terminal {
                            EdgeKind::Terminal
                        } else if next.is_none() {
                            EdgeKind::EndOfChain // defensive: validation makes this unreachable
                        } else if port == "success" {
                            EdgeKind::Success
                        } else {
                            EdgeKind::Outcome
                        };
                        r.record_step(
                            &current_node_id,
                            &node_type,
                            StepOutcome::Success,
                            elapsed,
                            edge,
                            (port != "success").then_some(port),
                            next.map(String::as_str),
                            &ctx,
                        );
                    }
                    match next {
                        Some(next_id) => current_node_id = next_id.clone(),
                        // Terminal, or no edge for this port — end of chain.
                        None => break,
                    }
                }
                Err(mut err) => {
                    if let Some(ref m) = self.resources.metrics {
                        m.node_errors
                            .with_label_values(&[
                                &self.policy_name,
                                &current_node_id,
                                &err.error.code,
                            ])
                            .inc();
                    }

                    // Tag the error with the node_id that produced it
                    err.error.node_id = current_node_id.clone();
                    // Operational visibility: an error-port exit is the only
                    // signal an operator gets without debug traces, so it is
                    // logged here regardless of which edge picks it up.
                    tracing::warn!(
                        "policy '{}': node '{}' exited on error port: {} ({})",
                        self.policy_name,
                        current_node_id,
                        err.error.code,
                        err.error.message
                    );
                    let outcome = StepOutcome::Error {
                        code: err.error.code.clone(),
                        message: err.error.message.clone(),
                    };
                    err.context.errors.push(err.error);
                    ctx = err.context;

                    // Try per-node error edge first, then the policy catch-all.
                    let next_error = self
                        .edges
                        .get(&current_node_id)
                        .and_then(|m| m.get("error"))
                        .map(|id| (id.clone(), EdgeKind::Error))
                        .or_else(|| {
                            self.catch_all_handler
                                .as_ref()
                                .map(|id| (id.clone(), EdgeKind::CatchAll))
                        });

                    if next_error.is_none() {
                        // No error handling — return 500
                        ctx.response.status_code = 500;
                        ctx.response.body = bytes::Bytes::from(
                            r#"{"error": "internal_error", "message": "Unhandled error in routing policy"}"#,
                        );
                        // This overwrites `response.body`; a stream set by an
                        // earlier node in the same chain (e.g. `upstream`)
                        // must not survive alongside it — see
                        // `ErrorHandlerPlugin::execute` for the same rule and
                        // why it matters (the listener treats a set stream as
                        // authoritative over `body`).
                        ctx.response.stream = None;
                        ctx.response.headers.insert(
                            "content-type".to_string(),
                            vec!["application/json".to_string()],
                        );
                    }

                    if let Some(r) = recorder.as_mut() {
                        let (edge, next_id) = match &next_error {
                            Some((id, kind)) => (*kind, Some(id.as_str())),
                            None => (EdgeKind::Unhandled, None),
                        };
                        r.record_step(
                            &current_node_id,
                            &node_type,
                            outcome,
                            elapsed,
                            edge,
                            None,
                            next_id,
                            &ctx,
                        );
                    }

                    match next_error {
                        Some((id, _)) => current_node_id = id,
                        None => break,
                    }
                }
            }
        }

        (ctx, recorder)
    }

    /// Whether the `upstream` node named `node_id` may stream its response
    /// body straight through to the client — nothing between it and `client`
    /// on the `success` path reads the buffered response body. Consulted by
    /// `run` to set the `__may_stream` reserved key before invoking the node.
    pub fn is_stream_capable(&self, node_id: &str) -> bool {
        self.stream_capable.contains(node_id)
    }

    /// Why each non-stream-capable `upstream` node must buffer, one entry per
    /// blocked upstream, for operator-visible reporting.
    /// `proxy-cache` halves with no counterpart. Informational, like
    /// [`CompiledGraph::buffering_reasons`].
    pub fn cache_pair_warnings(&self) -> &[CachePairWarning] {
        &self.cache_pair_warnings
    }

    /// Every `proxy-cache` half's backend, for invalidation by pair `id`.
    pub fn cache_targets(&self) -> &[crate::traffic::CacheTarget] {
        &self.cache_targets
    }

    pub fn buffering_reasons(&self) -> &[BufferingReason] {
        &self.buffering_reasons
    }
}

/// Compiles a [`PolicyConfig`] into a ready-to-execute [`CompiledGraph`].
///
/// Instantiates each node's plugin via `plugins::create_plugin`, records
/// `client` nodes as terminals, and indexes edges per node by source port
/// (`out` normalizes to `success`). The entry node is the target of the
/// listener's success/out edge. Fails if the policy has no `listener` node,
/// a plugin cannot be constructed, an edge names a port its node type does
/// not declare, two edges leave the same `node.port` (fan-out is not
/// supported), or a node's `success`/outcome port ([`PortKind::Success`] or
/// [`PortKind::Outcome`]) has no outgoing edge — `error` ports are exempt
/// (fallback chain: per-node error edge -> policy catch-all -> default 500).
///
/// [`PortKind::Success`]: crate::plugins::ports::PortKind::Success
/// [`PortKind::Outcome`]: crate::plugins::ports::PortKind::Outcome
///
/// Edge endpoints use the `node_id.port` form:
///
/// ```yaml
/// edges:
///   - from: listener.out
///     to: rewrite.in
///   - from: rewrite.success
///     to: client.in
///   - from: rewrite.error
///     to: error-handler.in
/// ```
pub fn compile_policy(
    policy: &PolicyConfig,
    resources: Arc<PluginResources>,
) -> Result<CompiledGraph, String> {
    // Before anything is constructed: this reads config only, and running it
    // first means a split pair is reported as a split pair rather than as
    // whatever its half happens to fail on downstream.
    let cache_pair_warnings = validate_cache_pairs(&policy.nodes)?;

    let mut nodes: HashMap<String, Box<dyn Plugin>> = HashMap::new();
    let mut listener_node_id = None;
    let mut terminal_node_ids = HashSet::new();

    // Instantiate all nodes
    for node_config in &policy.nodes {
        tracing::debug!(
            "Compiling node '{}' type='{}' config={:?}",
            node_config.id,
            node_config.node_type,
            node_config.config
        );
        // Interpolate `${ENV_VAR}` in the node config before instantiating the
        // plugin. This is the source-agnostic choke point for every config
        // source — gateway.yaml (loaded raw), the Web UI / Admin API, and etcd
        // all deliver placeholder-form config; only the compiled graph ever
        // holds resolved values, so the stored config the Admin API serves
        // never contains resolved secrets.
        let mut config = node_config.config.clone();
        for value in config.values_mut() {
            crate::config::interpolate_env_json(value);
        }
        let plugin = plugins::create_plugin(&node_config.node_type, &config, &resources)?;
        if node_config.node_type == "listener" {
            listener_node_id = Some(node_config.id.clone());
        }
        if node_config.node_type == "client" {
            terminal_node_ids.insert(node_config.id.clone());
        }
        nodes.insert(node_config.id.clone(), plugin);
    }

    let listener_node_id = listener_node_id.ok_or("Policy must have a listener node")?;

    let node_types: HashMap<String, String> = policy
        .nodes
        .iter()
        .map(|n| (n.id.clone(), n.node_type.clone()))
        .collect();

    // Parse edges, indexed by source port. `out` normalizes to `success`.
    let mut edges: HashMap<String, HashMap<String, String>> = HashMap::new();
    for edge in &policy.edges {
        let (from_node, from_port) = parse_edge_endpoint(&edge.from)?;
        let (to_node, _to_port) = parse_edge_endpoint(&edge.to)?;
        let from_port = if from_port == "out" {
            "success".to_string()
        } else {
            from_port
        };

        let node_type = node_types.get(&from_node).ok_or_else(|| {
            format!(
                "policy '{}': edge references unknown node '{}'",
                policy.name, from_node
            )
        })?;
        let spec = plugins::port_spec(node_type)
            .ok_or_else(|| format!("Unknown plugin type: {}", node_type))?;
        if !spec.outputs.iter().any(|p| p.name == from_port) {
            return Err(format!(
                "policy '{}': node '{}' (type '{}') has no output port '{}'",
                policy.name, from_node, node_type, from_port
            ));
        }
        if edges
            .entry(from_node.clone())
            .or_default()
            .insert(from_port.clone(), to_node)
            .is_some()
        {
            return Err(format!(
                "policy '{}': duplicate edge from '{}.{}' — fan-out is not supported",
                policy.name, from_node, from_port
            ));
        }
    }

    // Mandatory wiring: every success/outcome port of every node must have an edge.
    for node_config in &policy.nodes {
        let spec = plugins::port_spec(&node_config.node_type)
            .ok_or_else(|| format!("Unknown plugin type: {}", node_config.node_type))?;
        for p in spec.outputs {
            if matches!(p.kind, plugins::ports::PortKind::Error) {
                continue;
            }
            let wired = edges
                .get(&node_config.id)
                .is_some_and(|m| m.contains_key(p.name));
            if !wired {
                return Err(format!(
                    "policy '{}': output port '{}' of node '{}' (type '{}') must be wired — add an edge from '{}.{}'",
                    policy.name, p.name, node_config.id, node_config.node_type, node_config.id, p.name
                ));
            }
        }
    }

    // Reject cycles: the runtime walk (`run`) follows edges with no step
    // limit, so a loop anywhere in the graph — through any port, error edges
    // included — would never terminate. Checking each declared node for
    // self-reachability keeps the reported node independent of HashMap
    // iteration order.
    for node_config in &policy.nodes {
        let start = &node_config.id;
        let mut stack: Vec<&String> = edges
            .get(start)
            .map(|m| m.values().collect())
            .unwrap_or_default();
        let mut seen: HashSet<&String> = HashSet::new();
        while let Some(next) = stack.pop() {
            if next == start {
                return Err(format!(
                    "policy '{}': policy graph contains a cycle through node '{}' — policies must be acyclic",
                    policy.name, start
                ));
            }
            if seen.insert(next) {
                if let Some(m) = edges.get(next) {
                    stack.extend(m.values());
                }
            }
        }
    }

    // The entry node is the first node connected from the listener's success/out edge
    let entry_node_id = edges
        .get(&listener_node_id)
        .and_then(|m| m.get("success"))
        .cloned()
        .unwrap_or_else(|| listener_node_id.clone());

    // Compile-time streaming inference: see `infer_stream_capability` for the
    // walk itself; it is a free function (rather than inlined here) so tests
    // can drive it directly with hand-built node/edge maps, the same way
    // `failing_graph`/`outcome_graph` below already do for other engine
    // behaviour.
    let (stream_capable, buffering_reasons) =
        infer_stream_capability(&policy.nodes, &nodes, &edges);

    for reason in &buffering_reasons {
        tracing::info!(
            policy = %policy.name,
            upstream = %reason.upstream_node_id,
            blocked_by = %reason.blocked_by_node_id,
            node_type = %reason.node_type,
            "response buffering: upstream cannot stream because a downstream node reads the response body"
        );
    }

    let cache_targets: Vec<_> = nodes.values().filter_map(|n| n.cache_target()).collect();

    Ok(CompiledGraph {
        nodes,
        edges,
        entry_node_id,
        terminal_node_ids,
        catch_all_handler: policy.error_handler.clone(),
        policy_name: policy.name.clone(),
        resources,
        stream_capable,
        buffering_reasons,
        cache_pair_warnings,
        cache_targets,
    })
}

/// Checks that every `proxy-cache` pair agrees about where it caches.
///
/// The two halves of a pair are linked only by their shared `id`; nothing
/// else ties them together. If they disagree about `policy` or `store`, the
/// store half writes somewhere the lookup half never reads, and the route
/// compiles, serves traffic, and returns a permanent 100% miss with no error
/// anywhere -- it simply looks like a cache that is never warm. There is no
/// configuration for which that is correct, so it is refused.
///
/// A half with no counterpart is returned as a warning instead: equally
/// useless, but also what a policy looks like halfway through being built.
fn validate_cache_pairs(policy_nodes: &[NodeConfig]) -> Result<Vec<CachePairWarning>, String> {
    /// One half, as configured.
    struct Half<'a> {
        cache_id: &'a str,
        node_id: &'a str,
        role: &'a str,
        policy: &'a str,
        store: &'a str,
    }

    let mut halves: Vec<Half> = Vec::new();
    for node in policy_nodes {
        if node.node_type != "proxy-cache" {
            continue;
        }
        let get = |key: &str| -> Option<&str> { node.config.get(key).and_then(|v| v.as_str()) };
        let Some(cache_id) = get("id") else { continue };
        // `phase` is the documented key; `role` is accepted as an alias.
        let Some(role) = get("phase").or_else(|| get("role")) else {
            continue;
        };
        halves.push(Half {
            node_id: &node.id,
            role,
            policy: get("policy").unwrap_or("local"),
            store: get("store").unwrap_or(""),
            cache_id,
        });
    }

    let mut warnings = Vec::new();
    let ids: Vec<&str> = {
        let mut seen: Vec<&str> = Vec::new();
        for h in &halves {
            if !seen.contains(&h.cache_id) {
                seen.push(h.cache_id);
            }
        }
        seen
    };

    for cache_id in ids {
        let group: Vec<&Half> = halves.iter().filter(|h| h.cache_id == cache_id).collect();

        // Disagreement is a hard error: whichever half is wrong, the pair can
        // never see its own entries.
        for pair in group.windows(2) {
            let (a, b) = (pair[0], pair[1]);
            for (key, left, right) in [("policy", a.policy, b.policy), ("store", a.store, b.store)]
            {
                if left != right {
                    return Err(format!(
                        "proxy-cache pair '{}': node '{}' uses {} '{}' but node '{}' uses {} '{}'.                          Both halves must agree, or the store half writes where the lookup half                          never reads and the route serves a permanent 100% miss with no error",
                        cache_id, a.node_id, key, left, b.node_id, key, right
                    ));
                }
            }
        }

        // A half with no counterpart: report, do not refuse. A group is
        // complete only once it has both a lookup and a store -- a purge
        // alone (nothing to purge for) is a lone half too.
        let has_lookup = group.iter().any(|h| h.role == "lookup");
        let has_store = group.iter().any(|h| h.role == "store");
        if !(has_lookup && has_store) {
            // Whatever IS present (a lone lookup, a lone store, or a purge
            // with nothing to purge for) is reported; the first missing role
            // names what would complete it.
            let present = group.first().expect("groups are non-empty");
            warnings.push(CachePairWarning {
                cache_id: cache_id.to_string(),
                present_node_id: present.node_id.to_string(),
                missing_role: if !has_lookup { "lookup" } else { "store" }.to_string(),
            });
        }
    }

    Ok(warnings)
}

/// For each `upstream` node, walks only the `success`/outcome ports forward
/// to see whether anything before `client` reads the response body.
///
/// `error` ports are never walked, on the upstream itself or on any node
/// further along the chain: an error exit runs through the node's own error
/// edge (unwalked here for the same reason), the policy's `error_handler`
/// catch-all, or the engine's built-in fallbacks — none of which this walk
/// can see.
///
/// This is safe only because of two separate facts, not one general
/// guarantee:
///
/// - Every gateway-generated error body — `ErrorHandlerPlugin::execute`, the
///   engine's no-handler 500 fallback, and the `NODE_NOT_FOUND` fallback,
///   all in this crate — explicitly clears `response.stream` before writing
///   `response.body`, so none of *those* three specific sites can leave a
///   stale stream alongside a generated body.
/// - An error edge can in principle route to *any* node, including one that
///   writes `response.body` itself without going through those three sites.
///   Nothing here walks that edge to rule it out; it is safe today only
///   because no plugin that opts out of `reads_response_body()` (the set
///   that can run downstream of a stream-capable `upstream` without forcing
///   it to buffer: `client`, `opentelemetry`, `prometheus`, `proxy-rewrite`,
///   `request-id`, `response-rewrite`, `skywalking`, `traffic-label`,
///   `zipkin`, and `proxy-cache` in its `lookup` role only) ever returns
///   `Err` from `execute` — every `return Err` in the first nine plugins'
///   source is in `from_config` (construction-time validation), never in
///   `execute`; `proxy-cache`'s `lookup` role does call a fallible backend
///   (`ResponseCache::get`) from `execute`, but matches on the result and
///   degrades a failure to a miss rather than propagating `Err`. This is
///   exactly why `proxy-cache`'s `store` and `purge` roles do **not** opt
///   out despite neither reading the response body: `store` because it
///   reads it to cache it, and `purge` because it *can* return `Err` from
///   `execute` on a failed backend, which this invariant forbids for an
///   opt-out node. If a future change to any of these plugins starts
///   erroring from `execute` where it previously didn't, this walk would
///   not catch it, and nothing else currently enforces it either.
///
/// Iterates `policy_nodes` (a `Vec`), not the `nodes` map, and visits each
/// node's outgoing ports in sorted-name order: the same precedent as the
/// cycle check in [`compile_policy`], so the reported blocker for an
/// unchanged policy is stable across compiles instead of depending on
/// `HashMap` iteration order.
///
/// An edge whose target doesn't resolve to a real node in `nodes` is treated
/// as blocking rather than assumed harmless — `to_node`s aren't validated
/// against the node table the way `from_node`s are, so this walk cannot
/// assume every edge target exists.
fn infer_stream_capability(
    policy_nodes: &[NodeConfig],
    nodes: &HashMap<String, Box<dyn Plugin>>,
    edges: &HashMap<String, HashMap<String, String>>,
) -> (HashSet<String>, Vec<BufferingReason>) {
    let mut stream_capable = HashSet::new();
    let mut buffering_reasons = Vec::new();

    for node_config in policy_nodes {
        if node_config.node_type != "upstream" {
            continue;
        }
        let node_id = &node_config.id;
        let mut blocked_by: Option<(&String, &str)> = None;
        let mut seen: HashSet<&String> = HashSet::new();
        let mut queue: Vec<&String> = edges
            .get(node_id)
            .and_then(|ports| ports.get("success"))
            .into_iter()
            .collect();

        while let Some(current) = queue.pop() {
            if !seen.insert(current) {
                continue;
            }
            let Some(p) = nodes.get(current) else {
                // An edge whose target doesn't resolve to a real node must
                // not be assumed harmless — treat it as blocking rather than
                // silently marking the upstream stream-capable.
                blocked_by = Some((current, "unknown"));
                break;
            };
            if p.reads_response_body() {
                blocked_by = Some((current, p.plugin_type()));
                break;
            }
            if let Some(ports) = edges.get(current) {
                let mut names: Vec<&String> = ports
                    .keys()
                    .filter(|name| name.as_str() != "error")
                    .collect();
                names.sort();
                queue.extend(names.into_iter().filter_map(|name| ports.get(name)));
            }
        }

        match blocked_by {
            None => {
                stream_capable.insert(node_id.clone());
            }
            Some((blocker, node_type)) => buffering_reasons.push(BufferingReason {
                upstream_node_id: node_id.clone(),
                blocked_by_node_id: blocker.clone(),
                node_type: node_type.to_string(),
            }),
        }
    }

    (stream_capable, buffering_reasons)
}

/// Parses `"node_id.port"` into `(node_id, port)`, defaulting the port to
/// `"out"` when no dot is present. Splits on the last dot, so node IDs may
/// themselves contain dots.
fn parse_edge_endpoint(endpoint: &str) -> Result<(String, String), String> {
    if let Some(dot_pos) = endpoint.rfind('.') {
        let node_id = endpoint[..dot_pos].to_string();
        let port = endpoint[dot_pos + 1..].to_string();
        Ok((node_id, port))
    } else {
        Ok((endpoint.to_string(), "out".to_string()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::{EdgeConfig, NodeConfig, PolicyConfig};
    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
    use bytes::Bytes;

    #[test]
    fn test_parse_edge_endpoint() {
        let (node, port) = parse_edge_endpoint("listener.out").unwrap();
        assert_eq!(node, "listener");
        assert_eq!(port, "out");

        let (node, port) = parse_edge_endpoint("upstream.error").unwrap();
        assert_eq!(node, "upstream");
        assert_eq!(port, "error");

        let (node, port) = parse_edge_endpoint("rewrite.success").unwrap();
        assert_eq!(node, "rewrite");
        assert_eq!(port, "success");
    }

    fn test_context(path: &str) -> Context {
        Context {
            request: GatewayRequest {
                method: "GET".to_string(),
                path: path.to_string(),
                host: "localhost".to_string(),
                scheme: "http".to_string(),
                headers: HashMap::new(),
                query_params: HashMap::new(),
                body: Bytes::new(),
                remote_addr: "127.0.0.1:12345".to_string(),
                protocol: Protocol::Http1,
            },
            response: GatewayResponse {
                status_code: 0,
                headers: HashMap::new(),
                body: Bytes::new(),
                stream: None,
            },
            message: HashMap::new(),
            errors: Vec::new(),
        }
    }

    #[tokio::test]
    async fn test_graph_proxy_rewrite_pipeline() {
        let mut rewrite_config = HashMap::new();
        rewrite_config.insert(
            "strip_path_prefix".to_string(),
            serde_json::Value::String("/api/v1".to_string()),
        );
        rewrite_config.insert(
            "phase".to_string(),
            serde_json::Value::String("request".to_string()),
        );

        let policy = PolicyConfig {
            name: "test".to_string(),
            error_handler: None,
            nodes: vec![
                NodeConfig {
                    id: "listener".to_string(),
                    node_type: "listener".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "rewrite".to_string(),
                    node_type: "proxy-rewrite".to_string(),
                    config: rewrite_config,
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "client".to_string(),
                    node_type: "client".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
            ],
            edges: vec![
                EdgeConfig {
                    from: "listener.out".to_string(),
                    to: "rewrite.in".to_string(),
                },
                EdgeConfig {
                    from: "rewrite.success".to_string(),
                    to: "client.in".to_string(),
                },
            ],
        };

        let graph = compile_policy(&policy, PluginResources::empty()).unwrap();
        let ctx = test_context("/api/v1/users");
        let result = graph.execute(ctx).await;

        assert_eq!(result.request.path, "/users");
    }

    #[tokio::test]
    async fn test_graph_records_node_metrics() {
        let policy = PolicyConfig {
            name: "metrics-test".to_string(),
            error_handler: None,
            nodes: vec![
                NodeConfig {
                    id: "listener".to_string(),
                    node_type: "listener".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "client".to_string(),
                    node_type: "client".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
            ],
            edges: vec![EdgeConfig {
                from: "listener.out".to_string(),
                to: "client.in".to_string(),
            }],
        };

        let metrics = Arc::new(crate::metrics::GatewayMetrics::new());
        let graph = compile_policy(&policy, PluginResources::new(Some(metrics.clone()))).unwrap();
        graph.execute(test_context("/test")).await;

        assert_eq!(
            metrics
                .node_execution_count
                .with_label_values(&["metrics-test", "client", "client"])
                .get(),
            1
        );
    }

    /// The policy-level catch-all: a node whose own `error` port is unwired
    /// falls through to the named handler, which then renders the error.
    ///
    /// The failing node is a real one (an `upstream` pointed at a closed port),
    /// not a bare `listener -> error-handler` chain: since the port migration,
    /// `error-handler` passes an **errorless** context straight through, so a
    /// handler reached without a preceding failure has nothing to render.
    #[tokio::test]
    async fn test_graph_error_handler_catch_all() {
        let policy = PolicyConfig {
            name: "test".to_string(),
            error_handler: Some("error-handler".to_string()),
            nodes: vec![
                NodeConfig {
                    id: "listener".to_string(),
                    node_type: "listener".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "backend".to_string(),
                    node_type: "upstream".to_string(),
                    config: {
                        let mut c = HashMap::new();
                        // Nothing listens on port 1: connect fails instantly.
                        c.insert(
                            "targets".to_string(),
                            serde_json::json!([{ "host": "127.0.0.1", "port": 1 }]),
                        );
                        c.insert("timeout_ms".to_string(), serde_json::json!(500));
                        c
                    },
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "error-handler".to_string(),
                    node_type: "error-handler".to_string(),
                    config: {
                        let mut c = HashMap::new();
                        c.insert("status_code".to_string(), serde_json::json!(503));
                        c.insert(
                            "body_template".to_string(),
                            serde_json::Value::String(r#"{"error": "{{error.code}}"}"#.to_string()),
                        );
                        c
                    },
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "client".to_string(),
                    node_type: "client".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
            ],
            edges: vec![
                EdgeConfig {
                    from: "listener.out".to_string(),
                    to: "backend.in".to_string(),
                },
                // backend.error is deliberately unwired -> catch-all.
                EdgeConfig {
                    from: "backend.success".to_string(),
                    to: "client.in".to_string(),
                },
                EdgeConfig {
                    from: "error-handler.success".to_string(),
                    to: "client.in".to_string(),
                },
            ],
        };

        let graph = compile_policy(&policy, PluginResources::empty()).unwrap();
        let ctx = test_context("/test");
        let result = graph.execute(ctx).await;

        assert_eq!(result.response.status_code, 503);
        let body = String::from_utf8(result.response.body.to_vec()).unwrap();
        assert!(!body.contains("{{"), "template must be rendered: {body}");
    }

    /// The complement of the test above: an `error-handler` reached with an
    /// **empty** `context.errors` — what every outcome exit looks like — must
    /// leave the already-prepared response alone instead of clobbering it with
    /// its own status and an unrendered template.
    #[tokio::test]
    async fn test_error_handler_passes_errorless_context_through() {
        let policy = PolicyConfig {
            name: "test".to_string(),
            error_handler: None,
            nodes: vec![
                NodeConfig {
                    id: "listener".to_string(),
                    node_type: "listener".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "error-handler".to_string(),
                    node_type: "error-handler".to_string(),
                    config: {
                        let mut c = HashMap::new();
                        c.insert("status_code".to_string(), serde_json::json!(503));
                        c.insert(
                            "body_template".to_string(),
                            serde_json::Value::String(r#"{"error": "{{error.code}}"}"#.to_string()),
                        );
                        c
                    },
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "client".to_string(),
                    node_type: "client".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
            ],
            edges: vec![
                EdgeConfig {
                    from: "listener.out".to_string(),
                    to: "error-handler.in".to_string(),
                },
                EdgeConfig {
                    from: "error-handler.success".to_string(),
                    to: "client.in".to_string(),
                },
            ],
        };

        let graph = compile_policy(&policy, PluginResources::empty()).unwrap();
        let mut ctx = test_context("/test");
        // Stand in for an outcome exit: a fully prepared 401, no error record.
        ctx.response.status_code = 401;
        ctx.response.body = Bytes::from(r#"{"error":"unauthorized"}"#);
        let result = graph.execute(ctx).await;

        assert_eq!(result.response.status_code, 401);
        assert_eq!(
            String::from_utf8(result.response.body.to_vec()).unwrap(),
            r#"{"error":"unauthorized"}"#
        );
    }

    // ---- debug tracing ---------------------------------------------------

    use crate::debug::{CaptureOptions, EdgeKind, StepOutcome, TraceRecorder, TraceSource};
    use std::time::Duration;

    fn recorder(ctx: &Context) -> TraceRecorder {
        TraceRecorder::new(ctx, CaptureOptions::default(), 100)
    }

    fn finish(rec: TraceRecorder, ctx: &Context) -> crate::debug::Trace {
        rec.finish(
            "t".to_string(),
            0,
            TraceSource::Request,
            Some("r".to_string()),
            "p".to_string(),
            ctx,
            Duration::from_millis(1),
        )
    }

    /// A node that always fails, so error-edge routing can be exercised without
    /// depending on a real plugin's failure mode.
    struct AlwaysFails;

    #[async_trait::async_trait]
    impl Plugin for AlwaysFails {
        fn plugin_type(&self) -> &str {
            "always-fails"
        }
        async fn execute(&self, ctx: Context) -> crate::plugins::PluginResult {
            Err(crate::plugins::PluginExecutionError {
                context: ctx,
                error: crate::context::GatewayError {
                    node_id: String::new(),
                    code: "BOOM".to_string(),
                    message: "exploded".to_string(),
                    metadata: HashMap::new(),
                },
            })
        }
    }

    /// Builds a graph by hand so a failing node can be injected.
    fn failing_graph(
        error_edges: HashMap<String, String>,
        catch_all: Option<String>,
    ) -> CompiledGraph {
        let mut nodes: HashMap<String, Box<dyn Plugin>> = HashMap::new();
        nodes.insert("boom".to_string(), Box::new(AlwaysFails));
        nodes.insert(
            "client".to_string(),
            plugins::create_plugin("client", &HashMap::new(), &PluginResources::empty()).unwrap(),
        );
        let mut edges: HashMap<String, HashMap<String, String>> = HashMap::new();
        if let Some(target) = error_edges.get("boom") {
            edges
                .entry("boom".to_string())
                .or_default()
                .insert("error".to_string(), target.clone());
        }
        CompiledGraph {
            nodes,
            edges,
            entry_node_id: "boom".to_string(),
            terminal_node_ids: HashSet::from(["client".to_string()]),
            catch_all_handler: catch_all,
            policy_name: "p".to_string(),
            resources: PluginResources::empty(),
            stream_capable: HashSet::new(),
            buffering_reasons: Vec::new(),
            cache_pair_warnings: Vec::new(),
            cache_targets: Vec::new(),
        }
    }

    fn rewrite_policy() -> PolicyConfig {
        let mut cfg = HashMap::new();
        cfg.insert(
            "strip_path_prefix".to_string(),
            serde_json::Value::String("/api/v1".to_string()),
        );
        cfg.insert(
            "phase".to_string(),
            serde_json::Value::String("request".to_string()),
        );
        PolicyConfig {
            name: "traced".to_string(),
            error_handler: None,
            nodes: vec![
                NodeConfig {
                    id: "listener".to_string(),
                    node_type: "listener".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "rewrite".to_string(),
                    node_type: "proxy-rewrite".to_string(),
                    config: cfg,
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "client".to_string(),
                    node_type: "client".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
            ],
            edges: vec![
                EdgeConfig {
                    from: "listener.out".to_string(),
                    to: "rewrite.in".to_string(),
                },
                EdgeConfig {
                    from: "rewrite.success".to_string(),
                    to: "client.in".to_string(),
                },
            ],
        }
    }

    #[tokio::test]
    async fn test_trace_records_each_node_in_order() {
        let graph = compile_policy(&rewrite_policy(), PluginResources::empty()).unwrap();
        let ctx = test_context("/api/v1/users");
        let rec = recorder(&ctx);
        let (out, rec) = graph.execute_traced(ctx, rec).await;
        let trace = finish(rec, &out);

        let ids: Vec<&str> = trace.steps.iter().map(|s| s.node_id.as_str()).collect();
        assert_eq!(ids, vec!["rewrite", "client"]);
        assert_eq!(trace.steps[0].node_type, "proxy-rewrite");
        assert_eq!(trace.steps[0].edge, EdgeKind::Success);
        assert_eq!(trace.steps[0].next_node_id.as_deref(), Some("client"));
        // The terminal node ends the walk.
        assert_eq!(trace.steps[1].edge, EdgeKind::Terminal);
        assert_eq!(trace.steps[1].next_node_id, None);
    }

    /// The trace must show the rewrite: initial path in, rewritten path out.
    #[tokio::test]
    async fn test_trace_captures_what_the_plugin_changed() {
        let graph = compile_policy(&rewrite_policy(), PluginResources::empty()).unwrap();
        let ctx = test_context("/api/v1/users");
        let rec = recorder(&ctx);
        let (out, rec) = graph.execute_traced(ctx, rec).await;
        let trace = finish(rec, &out);

        assert_eq!(trace.initial.request.path, "/api/v1/users");
        assert_eq!(trace.steps[0].after.request.path, "/users");

        let changes = crate::debug::diff::diff(&trace.initial, &trace.steps[0].after);
        let c = changes
            .iter()
            .find(|c| c.path == "request.path")
            .expect("path change");
        assert_eq!(c.before.as_deref(), Some("/api/v1/users"));
        assert_eq!(c.after.as_deref(), Some("/users"));
    }

    /// Tracing must never change what the gateway does. If this ever fails,
    /// the debug feature has become a heisenbug generator.
    #[tokio::test]
    async fn test_tracing_does_not_alter_behaviour() {
        let policy = rewrite_policy();
        let plain = compile_policy(&policy, PluginResources::empty()).unwrap();
        let traced = compile_policy(&policy, PluginResources::empty()).unwrap();

        let untraced_out = plain.execute(test_context("/api/v1/users")).await;
        let ctx = test_context("/api/v1/users");
        let rec = recorder(&ctx);
        let (traced_out, _) = traced.execute_traced(ctx, rec).await;

        assert_eq!(untraced_out.request.path, traced_out.request.path);
        assert_eq!(
            untraced_out.response.status_code,
            traced_out.response.status_code
        );
        assert_eq!(untraced_out.response.body, traced_out.response.body);
        assert_eq!(untraced_out.message, traced_out.message);
        assert_eq!(untraced_out.errors, traced_out.errors);
    }

    #[tokio::test]
    async fn test_trace_records_error_edge() {
        let graph = failing_graph(
            HashMap::from([("boom".to_string(), "client".to_string())]),
            None,
        );
        let ctx = test_context("/x");
        let rec = recorder(&ctx);
        let (out, rec) = graph.execute_traced(ctx, rec).await;
        let trace = finish(rec, &out);

        assert_eq!(trace.steps[0].node_id, "boom");
        assert_eq!(
            trace.steps[0].outcome,
            StepOutcome::Error {
                code: "BOOM".to_string(),
                message: "exploded".to_string()
            }
        );
        assert_eq!(trace.steps[0].edge, EdgeKind::Error);
        assert_eq!(trace.steps[0].next_node_id.as_deref(), Some("client"));
        // The error was recorded on the context the step captured.
        assert_eq!(trace.steps[0].after.errors.len(), 1);
        assert_eq!(trace.steps[0].after.errors[0].node_id, "boom");
    }

    /// A node leaving through its error port is an operational event, not
    /// just a response body: it must be logged at WARN with the node id and
    /// the error code/message so an operator can see why a request failed
    /// without enabling debug traces.
    #[tokio::test]
    async fn test_error_port_exit_is_logged_at_warn() {
        let graph = failing_graph(
            HashMap::from([("boom".to_string(), "client".to_string())]),
            None,
        );
        let (_guard, logs) = crate::test_log::capture_warnings();
        let _ = graph.execute(test_context("/x")).await;

        let out = logs.contents();
        assert!(out.contains("WARN"), "expected a WARN line, got: {out:?}");
        for needle in ["policy 'p'", "node 'boom'", "BOOM", "exploded"] {
            assert!(out.contains(needle), "missing {needle:?} in: {out:?}");
        }
    }

    #[tokio::test]
    async fn test_trace_records_catch_all_edge() {
        let graph = failing_graph(HashMap::new(), Some("client".to_string()));
        let ctx = test_context("/x");
        let rec = recorder(&ctx);
        let (out, rec) = graph.execute_traced(ctx, rec).await;
        let trace = finish(rec, &out);
        assert_eq!(trace.steps[0].edge, EdgeKind::CatchAll);
    }

    /// The unwired-error-port case: this is the footgun the sandbox's
    /// `on_error: "stop"` mode deliberately exposes.
    #[tokio::test]
    async fn test_trace_records_unhandled_error() {
        let graph = failing_graph(HashMap::new(), None);
        let ctx = test_context("/x");
        let rec = recorder(&ctx);
        let (out, rec) = graph.execute_traced(ctx, rec).await;
        let trace = finish(rec, &out);

        assert_eq!(trace.steps.len(), 1);
        assert_eq!(trace.steps[0].edge, EdgeKind::Unhandled);
        assert_eq!(trace.steps[0].next_node_id, None);
        // ...and the engine wrote its generic 500.
        assert_eq!(out.response.status_code, 500);
        assert_eq!(trace.steps[0].after.response.status_code, 500);
    }

    /// The `NODE_NOT_FOUND` path (an edge pointing at a node id absent from
    /// the node table — reachable because `compile_policy` validates only
    /// `from_node`, never `to_node`) also overwrites `response.body` with a
    /// generated message. Same hazard as the other two gateway-generated
    /// error bodies: a stale `response.stream` must not survive it.
    #[tokio::test]
    async fn test_node_not_found_clears_stale_stream() {
        use crate::context::stream::ResponseStream;
        use http_body_util::{BodyExt, Full};

        let graph = CompiledGraph {
            nodes: HashMap::new(),
            edges: HashMap::new(),
            entry_node_id: "missing".to_string(),
            terminal_node_ids: HashSet::new(),
            catch_all_handler: None,
            policy_name: "p".to_string(),
            resources: PluginResources::empty(),
            stream_capable: HashSet::new(),
            buffering_reasons: Vec::new(),
            cache_pair_warnings: Vec::new(),
            cache_targets: Vec::new(),
        };
        let mut ctx = test_context("/x");
        let boxed = Full::new(Bytes::from_static(b"partial-stream-bytes"))
            .map_err(|never| match never {})
            .boxed();
        ctx.response.stream = Some(ResponseStream::new(boxed));

        let out = graph.execute(ctx).await;

        assert_eq!(out.response.status_code, 500);
        assert!(
            out.response.stream.is_none(),
            "the generated node-not-found body must not coexist with a stale stream"
        );
        assert!(String::from_utf8(out.response.body.to_vec())
            .unwrap()
            .contains("node 'missing' not found"));
    }

    /// The engine's no-handler 500 fallback overwrites `response.body`. If a
    /// stream was set on the context that arrived here (e.g. an `upstream`
    /// node had started relaying a response, then a later node in the same
    /// chain errored with neither an error edge nor a catch-all), the stale
    /// stream must be cleared too — otherwise the listener would see both
    /// `body` and `stream` set and, per the documented invariant, send the
    /// half-finished stream instead of this generated 500.
    #[tokio::test]
    async fn test_unhandled_error_clears_stale_stream() {
        use crate::context::stream::ResponseStream;
        use http_body_util::{BodyExt, Full};

        let graph = failing_graph(HashMap::new(), None);
        let mut ctx = test_context("/x");
        let boxed = Full::new(Bytes::from_static(b"partial-stream-bytes"))
            .map_err(|never| match never {})
            .boxed();
        ctx.response.stream = Some(ResponseStream::new(boxed));

        let out = graph.execute(ctx).await;

        assert_eq!(out.response.status_code, 500);
        assert!(
            out.response.stream.is_none(),
            "the generated 500 body must not coexist with a stale stream"
        );
        assert!(String::from_utf8(out.response.body.to_vec())
            .unwrap()
            .contains("Unhandled error in routing policy"));
    }

    // ---- per-port outcome routing -----------------------------------------

    /// A test-only plugin that emits on the "denied" outcome port when the
    /// request carries `x-deny`, else the normal success port. Stands in for
    /// a real outcome-port plugin, which doesn't exist until Task 8.
    struct DenyOnHeader;

    #[async_trait::async_trait]
    impl Plugin for DenyOnHeader {
        fn plugin_type(&self) -> &str {
            "deny-on-header"
        }
        async fn execute(&self, ctx: Context) -> crate::plugins::PluginResult {
            if ctx.request.headers.contains_key("x-deny") {
                Ok(plugins::PluginOutput::on_port(ctx, "denied"))
            } else {
                Ok(plugins::PluginOutput::success(ctx))
            }
        }
    }

    /// Builds a graph by hand with a per-port edge map, so outcome-port
    /// routing can be exercised without a real plugin type declaring one.
    fn outcome_graph() -> CompiledGraph {
        let mut nodes: HashMap<String, Box<dyn Plugin>> = HashMap::new();
        nodes.insert("n".to_string(), Box::new(DenyOnHeader));
        nodes.insert(
            "client".to_string(),
            plugins::create_plugin("client", &HashMap::new(), &PluginResources::empty()).unwrap(),
        );
        nodes.insert(
            "deny-client".to_string(),
            plugins::create_plugin("client", &HashMap::new(), &PluginResources::empty()).unwrap(),
        );
        let mut n_edges = HashMap::new();
        n_edges.insert("denied".to_string(), "deny-client".to_string());
        n_edges.insert("success".to_string(), "client".to_string());
        let mut edges = HashMap::new();
        edges.insert("n".to_string(), n_edges);
        CompiledGraph {
            nodes,
            edges,
            entry_node_id: "n".to_string(),
            terminal_node_ids: HashSet::from(["client".to_string(), "deny-client".to_string()]),
            catch_all_handler: None,
            policy_name: "p".to_string(),
            resources: PluginResources::empty(),
            stream_capable: HashSet::new(),
            buffering_reasons: Vec::new(),
            cache_pair_warnings: Vec::new(),
            cache_targets: Vec::new(),
        }
    }

    /// A named outcome port routes to its wired target, not to success.
    #[tokio::test]
    async fn test_outcome_port_routes_to_its_edge() {
        let graph = outcome_graph();

        let mut ctx = test_context("/x");
        ctx.request
            .headers
            .insert("x-deny".to_string(), vec!["1".to_string()]);
        let rec = recorder(&ctx);
        let (out, rec) = graph.execute_traced(ctx, rec).await;
        let trace = finish(rec, &out);

        assert_eq!(trace.steps[0].node_id, "n");
        assert_eq!(trace.steps[0].edge, EdgeKind::Outcome);
        assert_eq!(trace.steps[0].port.as_deref(), Some("denied"));
        assert_eq!(trace.steps[0].next_node_id.as_deref(), Some("deny-client"));
        // The walk actually ended at deny-client, not client.
        assert_eq!(trace.steps.len(), 2);
        assert_eq!(trace.steps[1].node_id, "deny-client");
    }

    /// Without the header, the same node exits through `success` as usual.
    #[tokio::test]
    async fn test_outcome_port_falls_back_to_success() {
        let graph = outcome_graph();
        let ctx = test_context("/x");
        let rec = recorder(&ctx);
        let (out, rec) = graph.execute_traced(ctx, rec).await;
        let trace = finish(rec, &out);

        assert_eq!(trace.steps[0].edge, EdgeKind::Success);
        assert!(trace.steps[0].port.is_none());
        assert_eq!(trace.steps[0].next_node_id.as_deref(), Some("client"));
    }

    /// A real `cors` node, compiled through `compile_policy` (not a hand-built
    /// graph), with its `preflight` port wired to `client`. An OPTIONS
    /// preflight from an allowed origin must exit on the `preflight` outcome
    /// port — not `success` — with the prepared 204 reaching the final
    /// context, exercising the whole compile+execute+trace path end-to-end
    /// for an outcome-port node.
    fn cors_policy() -> PolicyConfig {
        let mut cfg = HashMap::new();
        cfg.insert(
            "allowed_origins".to_string(),
            serde_json::json!(["http://sub.domain.com"]),
        );
        PolicyConfig {
            name: "cors-preflight".to_string(),
            error_handler: None,
            nodes: vec![
                NodeConfig {
                    id: "listener".to_string(),
                    node_type: "listener".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "cors".to_string(),
                    node_type: "cors".to_string(),
                    config: cfg,
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "client".to_string(),
                    node_type: "client".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
            ],
            edges: vec![
                EdgeConfig {
                    from: "listener.out".to_string(),
                    to: "cors.in".to_string(),
                },
                EdgeConfig {
                    from: "cors.success".to_string(),
                    to: "client.in".to_string(),
                },
                EdgeConfig {
                    from: "cors.preflight".to_string(),
                    to: "client.in".to_string(),
                },
            ],
        }
    }

    #[tokio::test]
    async fn test_cors_preflight_exits_via_preflight_edge_end_to_end() {
        let graph = compile_policy(&cors_policy(), PluginResources::empty()).unwrap();

        let mut ctx = test_context("/x");
        ctx.request.method = "OPTIONS".to_string();
        ctx.request.headers.insert(
            "origin".to_string(),
            vec!["http://sub.domain.com".to_string()],
        );
        let rec = recorder(&ctx);
        let (out, rec) = graph.execute_traced(ctx, rec).await;
        let trace = finish(rec, &out);

        assert_eq!(trace.steps[0].node_id, "cors");
        assert_eq!(trace.steps[0].edge, EdgeKind::Outcome);
        assert_eq!(trace.steps[0].port.as_deref(), Some("preflight"));
        assert_eq!(trace.steps[0].next_node_id.as_deref(), Some("client"));
        assert_eq!(out.response.status_code, 204);
        assert_eq!(
            out.response.headers.get("access-control-allow-origin"),
            Some(&vec!["http://sub.domain.com".to_string()])
        );
    }

    // ---- compile-time mandatory-wiring validation -------------------------

    fn policy_missing_success_edge() -> PolicyConfig {
        PolicyConfig {
            name: "p".to_string(),
            error_handler: None,
            nodes: vec![
                NodeConfig {
                    id: "listener".to_string(),
                    node_type: "listener".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "rw".to_string(),
                    node_type: "proxy-rewrite".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "client".to_string(),
                    node_type: "client".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
            ],
            edges: vec![EdgeConfig {
                from: "listener.out".to_string(),
                to: "rw.in".to_string(),
            }],
        }
    }

    /// Unwired mandatory port fails compilation with the exact message.
    #[test]
    fn test_compile_rejects_unwired_mandatory_port() {
        let policy = policy_missing_success_edge();
        let err = compile_policy(&policy, PluginResources::empty()).unwrap_err();
        assert_eq!(
            err,
            "policy 'p': output port 'success' of node 'rw' (type 'proxy-rewrite') must be wired — add an edge from 'rw.success'"
        );
    }

    /// Duplicate (node, port) edge is an explicit error, not a silent overwrite.
    #[test]
    fn test_compile_rejects_fanout() {
        let mut policy = policy_missing_success_edge();
        policy.edges.push(EdgeConfig {
            from: "rw.success".to_string(),
            to: "client.in".to_string(),
        });
        policy.edges.push(EdgeConfig {
            from: "rw.success".to_string(),
            to: "client.in".to_string(),
        });
        let err = compile_policy(&policy, PluginResources::empty()).unwrap_err();
        assert_eq!(
            err,
            "policy 'p': duplicate edge from 'rw.success' — fan-out is not supported"
        );
    }

    /// Edge from a port the type does not declare.
    #[test]
    fn test_compile_rejects_undeclared_port() {
        let mut policy = policy_missing_success_edge();
        policy.edges.push(EdgeConfig {
            from: "rw.banana".to_string(),
            to: "client.in".to_string(),
        });
        let err = compile_policy(&policy, PluginResources::empty()).unwrap_err();
        assert_eq!(
            err,
            "policy 'p': node 'rw' (type 'proxy-rewrite') has no output port 'banana'"
        );
    }

    /// A fully-wired graph whose edges loop back on themselves must not
    /// compile: the runtime walk has no step limit, so a cycle would spin
    /// forever on the first matching request.
    #[test]
    fn test_compile_rejects_cycle() {
        let mut policy = policy_missing_success_edge();
        policy.nodes.push(NodeConfig {
            id: "rw2".to_string(),
            node_type: "proxy-rewrite".to_string(),
            config: HashMap::new(),
            config_ref: None,
            position: None,
        });
        policy.edges.push(EdgeConfig {
            from: "rw.success".to_string(),
            to: "rw2.in".to_string(),
        });
        policy.edges.push(EdgeConfig {
            from: "rw2.success".to_string(),
            to: "rw.in".to_string(),
        });
        let err = compile_policy(&policy, PluginResources::empty()).unwrap_err();
        assert_eq!(
            err,
            "policy 'p': policy graph contains a cycle through node 'rw' — policies must be acyclic"
        );
    }

    /// An error edge is still an edge the walk can follow, so it closes a
    /// cycle like any other: rw errors back into itself via rw2.
    #[test]
    fn test_compile_rejects_cycle_through_error_edge() {
        let mut policy = policy_missing_success_edge();
        policy.nodes.push(NodeConfig {
            id: "rw2".to_string(),
            node_type: "proxy-rewrite".to_string(),
            config: HashMap::new(),
            config_ref: None,
            position: None,
        });
        policy.edges.push(EdgeConfig {
            from: "rw.success".to_string(),
            to: "client.in".to_string(),
        });
        policy.edges.push(EdgeConfig {
            from: "rw.error".to_string(),
            to: "rw2.in".to_string(),
        });
        policy.edges.push(EdgeConfig {
            from: "rw2.success".to_string(),
            to: "rw.in".to_string(),
        });
        let err = compile_policy(&policy, PluginResources::empty()).unwrap_err();
        assert_eq!(
            err,
            "policy 'p': policy graph contains a cycle through node 'rw' — policies must be acyclic"
        );
    }

    /// Fan-in stays legal: two nodes converging on the same target is not a
    /// cycle and must compile.
    #[test]
    fn test_compile_allows_fan_in() {
        let mut policy = policy_missing_success_edge();
        policy.nodes.push(NodeConfig {
            id: "rw2".to_string(),
            node_type: "proxy-rewrite".to_string(),
            config: HashMap::new(),
            config_ref: None,
            position: None,
        });
        policy.edges.push(EdgeConfig {
            from: "rw.success".to_string(),
            to: "rw2.in".to_string(),
        });
        policy.edges.push(EdgeConfig {
            from: "rw.error".to_string(),
            to: "client.in".to_string(),
        });
        policy.edges.push(EdgeConfig {
            from: "rw2.success".to_string(),
            to: "client.in".to_string(),
        });
        assert!(compile_policy(&policy, PluginResources::empty()).is_ok());
    }

    /// error port stays optional: policy with no error edges still compiles.
    #[test]
    fn test_error_port_wiring_is_optional() {
        let mut policy = policy_missing_success_edge();
        policy.edges.push(EdgeConfig {
            from: "rw.success".to_string(),
            to: "client.in".to_string(),
        });
        assert!(compile_policy(&policy, PluginResources::empty()).is_ok());
    }

    // ---- streaming inference -----------------------------------------

    /// Turns a JSON policy (nodes + edges, `name` optional) into a compiled
    /// graph, panicking on any deserialize/compile failure so tests read as
    /// plain assertions.
    fn compile_test_policy(json: serde_json::Value) -> CompiledGraph {
        let mut value = json;
        if let serde_json::Value::Object(ref mut map) = value {
            map.entry("name")
                .or_insert_with(|| serde_json::Value::String("test".to_string()));
        }
        let policy: PolicyConfig =
            serde_json::from_value(value).expect("test policy JSON must deserialize");
        compile_policy(&policy, PluginResources::empty()).expect("test policy must compile")
    }

    /// The same, for a policy that must be REJECTED: returns the compile
    /// error so a test can assert on what it says.
    fn compile_test_policy_err(json: serde_json::Value) -> String {
        let mut value = json;
        if let serde_json::Value::Object(ref mut map) = value {
            map.entry("name")
                .or_insert_with(|| serde_json::Value::String("test".to_string()));
        }
        let policy: PolicyConfig =
            serde_json::from_value(value).expect("test policy JSON must deserialize");
        compile_policy(&policy, PluginResources::empty()).expect_err("this policy must not compile")
    }

    /// An upstream whose success path reaches `client` through header-only
    /// nodes can stream.
    #[test]
    fn test_upstream_is_stream_capable_with_header_only_tail() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "hdr", "type": "response-rewrite",
                  "config": { "headers": { "set": { "x-a": "b" } } } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "hdr.in" },
                { "from": "hdr.success", "to": "client.in" }
            ]
        }));

        assert!(graph.is_stream_capable("up"));
        assert!(graph.buffering_reasons().is_empty());
    }

    /// A body-rewriting node on the success path forces buffering, and the
    /// compiler must name it — silence here is the failure mode this feature
    /// exists to avoid.
    #[test]
    fn test_filters_force_buffering_and_are_reported() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "rw", "type": "response-rewrite",
                  "config": { "filters": [{ "regex": "a", "replace": "b" }] } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "rw.in" },
                { "from": "rw.success", "to": "client.in" }
            ]
        }));

        assert!(!graph.is_stream_capable("up"));
        let reasons = graph.buffering_reasons();
        assert_eq!(reasons.len(), 1);
        assert_eq!(reasons[0].upstream_node_id, "up");
        assert_eq!(reasons[0].blocked_by_node_id, "rw");
    }

    /// The error path must not influence the decision: it is taken only when
    /// the upstream produced no body at all.
    #[test]
    fn test_error_path_does_not_force_buffering() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "errs", "type": "error-handler",
                  "config": { "status_code": 502, "body_template": "{}" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "client.in" },
                { "from": "up.error", "to": "errs.in" },
                { "from": "errs.success", "to": "client.in" }
            ]
        }));

        assert!(
            graph.is_stream_capable("up"),
            "an error-handler on the error path must not block streaming"
        );
        assert!(graph.buffering_reasons().is_empty());
    }

    /// A `script` node on the success path forces buffering. This is not
    /// just another case: Task 1 left comments in `lua_runtime.rs` and
    /// `multi_auth.rs` asserting a discarded stream is safe there *because*
    /// those plugins never opt out of `reads_response_body`. Without this
    /// test that safety argument is only prose in a ledger; this makes the
    /// suite enforce it.
    #[test]
    fn test_script_node_forces_buffering() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "s", "type": "script",
                  "config": { "inline": "function execute(ctx) return ctx end" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "s.in" },
                { "from": "s.success", "to": "client.in" },
                { "from": "s.respond", "to": "client.in" }
            ]
        }));

        assert!(!graph.is_stream_capable("up"));
        let reasons = graph.buffering_reasons();
        assert_eq!(reasons.len(), 1);
        assert_eq!(reasons[0].upstream_node_id, "up");
        assert_eq!(reasons[0].blocked_by_node_id, "s");
        assert_eq!(reasons[0].node_type, "script");
    }

    /// A depth-1-only walk (checks only the upstream's direct successor)
    /// would already pass every test above, since all of them use a single
    /// hop. This one has a three-hop, all-header-only tail: only a real
    /// forward walk finds its way to `client` without blocking.
    #[test]
    fn test_multi_hop_header_only_chain_is_stream_capable() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "hdr1", "type": "response-rewrite",
                  "config": { "headers": { "set": { "x-a": "1" } } } },
                { "id": "hdr2", "type": "response-rewrite",
                  "config": { "headers": { "set": { "x-b": "2" } } } },
                { "id": "hdr3", "type": "response-rewrite",
                  "config": { "headers": { "set": { "x-c": "3" } } } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "hdr1.in" },
                { "from": "hdr1.success", "to": "hdr2.in" },
                { "from": "hdr2.success", "to": "hdr3.in" },
                { "from": "hdr3.success", "to": "client.in" }
            ]
        }));

        assert!(graph.is_stream_capable("up"));
        assert!(graph.buffering_reasons().is_empty());
    }

    /// The same three-hop chain, but the last node in the chain reads the
    /// body. A depth-1 implementation would report `hdr1` (or nothing at
    /// all) as the blocker, or miss the block entirely; a real forward walk
    /// must reach `hdr3` and name it.
    #[test]
    fn test_multi_hop_chain_blocks_on_body_reader_at_end() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "hdr1", "type": "response-rewrite",
                  "config": { "headers": { "set": { "x-a": "1" } } } },
                { "id": "hdr2", "type": "response-rewrite",
                  "config": { "headers": { "set": { "x-b": "2" } } } },
                { "id": "hdr3", "type": "response-rewrite",
                  "config": { "filters": [{ "regex": "a", "replace": "b" }] } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "hdr1.in" },
                { "from": "hdr1.success", "to": "hdr2.in" },
                { "from": "hdr2.success", "to": "hdr3.in" },
                { "from": "hdr3.success", "to": "client.in" }
            ]
        }));

        assert!(!graph.is_stream_capable("up"));
        let reasons = graph.buffering_reasons();
        assert_eq!(reasons.len(), 1);
        assert_eq!(reasons[0].upstream_node_id, "up");
        assert_eq!(reasons[0].blocked_by_node_id, "hdr3");
        assert_eq!(reasons[0].node_type, "response-rewrite");
    }

    /// A `proxy-cache` purge node on the success path must force buffering,
    /// even though `Role::Purge` reads nothing from the response it passes
    /// through: it *can* return `Err` from `execute` on a failed backend, and
    /// `infer_stream_capability`'s doc comment explains why that disqualifies
    /// it from opting out — an error edge from it could route to a node that
    /// writes `response.body` directly while a stream from the upstream is
    /// still live. Both `success` and `hit` are mandatory wiring on every
    /// `proxy-cache` node regardless of phase (the same `PortSpec` as
    /// lookup/store), even though a purge never emits `hit`, so both are
    /// wired to `client.in` here.
    #[test]
    fn test_proxy_cache_purge_node_forces_buffering_because_it_can_fail() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "purge", "type": "proxy-cache",
                  "config": { "phase": "purge", "id": "products", "policy": "local" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "purge.in" },
                { "from": "purge.success", "to": "client.in" },
                { "from": "purge.hit", "to": "client.in" }
            ]
        }));

        assert!(!graph.is_stream_capable("up"));
        let reasons = graph.buffering_reasons();
        assert_eq!(reasons.len(), 1);
        assert_eq!(reasons[0].upstream_node_id, "up");
        assert_eq!(reasons[0].blocked_by_node_id, "purge");
        assert_eq!(reasons[0].node_type, "proxy-cache");
    }

    /// A `proxy-cache` `lookup` node is normally placed *before* `upstream`,
    /// but nothing stops it from being wired after one too (unusual, but
    /// legal). Placed there, it must NOT force buffering: this pins the
    /// `Lookup`-only opt-out from `test_proxy_cache_purge_node_forces_
    /// buffering_because_it_can_fail`'s sibling case, proving it is the role
    /// -- not just the node type -- that decides.
    #[test]
    fn test_proxy_cache_lookup_node_after_upstream_does_not_force_buffering() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "look", "type": "proxy-cache",
                  "config": { "phase": "lookup", "id": "products", "policy": "local" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "look.in" },
                { "from": "look.success", "to": "client.in" },
                { "from": "look.hit", "to": "client.in" }
            ]
        }));

        assert!(graph.is_stream_capable("up"));
        assert!(graph.buffering_reasons().is_empty());
    }

    /// Two independent policies worth of `upstream` in one graph (a
    /// `condition` node routes to one or the other), one capable and one
    /// blocked — proving each upstream is judged on its own success path,
    /// not on the graph as a whole.
    #[test]
    fn test_two_upstreams_judged_independently() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "cond", "type": "condition",
                  "config": { "conditions": [["uri", "==", "/x"]] } },
                { "id": "up1", "type": "upstream",
                  "config": { "targets": [{ "host": "h1", "port": 80 }] } },
                { "id": "up2", "type": "upstream",
                  "config": { "targets": [{ "host": "h2", "port": 80 }] } },
                { "id": "s", "type": "script",
                  "config": { "inline": "function execute(ctx) return ctx end" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "cond.in" },
                { "from": "cond.true", "to": "up1.in" },
                { "from": "cond.false", "to": "up2.in" },
                { "from": "up1.success", "to": "client.in" },
                { "from": "up2.success", "to": "s.in" },
                { "from": "s.success", "to": "client.in" },
                { "from": "s.respond", "to": "client.in" }
            ]
        }));

        assert!(graph.is_stream_capable("up1"));
        assert!(!graph.is_stream_capable("up2"));
        let reasons = graph.buffering_reasons();
        assert_eq!(reasons.len(), 1);
        assert_eq!(reasons[0].upstream_node_id, "up2");
        assert_eq!(reasons[0].blocked_by_node_id, "s");
    }

    /// A no-op test plugin whose `reads_response_body` is set at
    /// construction, standing in for a real plugin so the diamond test below
    /// can shape a branching graph (two outcome ports on one node) that no
    /// currently-registered plugin type has.
    struct StubPlugin {
        reads_body: bool,
    }

    #[async_trait::async_trait]
    impl Plugin for StubPlugin {
        fn plugin_type(&self) -> &str {
            "stub"
        }
        fn reads_response_body(&self) -> bool {
            self.reads_body
        }
        async fn execute(&self, ctx: Context) -> crate::plugins::PluginResult {
            Ok(plugins::PluginOutput::success(ctx))
        }
    }

    /// `up.success -> a`, `a` branches to `b` and `c`, both rejoin at
    /// `client`; only `c` reads the response body. Exercises the part a
    /// straight-line (single-path) walk can't: multiple live branches off
    /// one node, only one of which blocks.
    ///
    /// Built by hand and driven through `infer_stream_capability` directly
    /// (rather than `compile_test_policy`) because no registered plugin type
    /// declares two non-error outcome ports without also defaulting to
    /// `reads_response_body() == true` itself, which would block the walk at
    /// `a` and never exercise the branch/rejoin logic this test is for.
    #[test]
    fn test_diamond_blocks_when_one_branch_reads_body() {
        let mut nodes: HashMap<String, Box<dyn Plugin>> = HashMap::new();
        nodes.insert("a".to_string(), Box::new(StubPlugin { reads_body: false }));
        nodes.insert("b".to_string(), Box::new(StubPlugin { reads_body: false }));
        nodes.insert("c".to_string(), Box::new(StubPlugin { reads_body: true }));
        nodes.insert(
            "client".to_string(),
            plugins::create_plugin("client", &HashMap::new(), &PluginResources::empty()).unwrap(),
        );

        let mut edges: HashMap<String, HashMap<String, String>> = HashMap::new();
        edges.insert(
            "up".to_string(),
            HashMap::from([("success".to_string(), "a".to_string())]),
        );
        edges.insert(
            "a".to_string(),
            HashMap::from([
                ("true".to_string(), "b".to_string()),
                ("false".to_string(), "c".to_string()),
            ]),
        );
        edges.insert(
            "b".to_string(),
            HashMap::from([("success".to_string(), "client".to_string())]),
        );
        edges.insert(
            "c".to_string(),
            HashMap::from([("success".to_string(), "client".to_string())]),
        );

        let policy_nodes = vec![NodeConfig {
            id: "up".to_string(),
            node_type: "upstream".to_string(),
            config: HashMap::new(),
            config_ref: None,
            position: None,
        }];

        let (stream_capable, reasons) = infer_stream_capability(&policy_nodes, &nodes, &edges);

        assert!(!stream_capable.contains("up"));
        assert_eq!(reasons.len(), 1);
        assert_eq!(reasons[0].upstream_node_id, "up");
        assert_eq!(reasons[0].blocked_by_node_id, "c");
        assert_eq!(reasons[0].node_type, "stub");
    }

    // ---- __may_stream engine signal -----------------------------------

    /// CRITICAL regression: `__may_stream` must not leak across an
    /// error-port hop. Shape: `up1` (stream-capable: its success path goes
    /// straight to `client`) fails and routes via its `error` port to `up2`
    /// (NOT capable: its success path is blocked by `rw`, a
    /// `response-rewrite` filter that reads the body) — an ordinary
    /// primary/failover-with-a-response-transform policy. `ctx`/`ctx.message`
    /// survives the error-port hop, so if the engine only ever *inserted*
    /// `true` for a capable node and never overwrote it for a non-capable
    /// one, `up2` would inherit the stale `true` `up1` left behind, stream
    /// instead of buffering, and silently bypass `rw`.
    #[tokio::test]
    async fn test_failover_to_non_capable_upstream_does_not_leak_may_stream() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        // up2's real backend: replies with a body the `rw` filter can
        // visibly transform, so a pass proves `rw` actually ran on a
        // buffered body — not just that `response.stream` happened to be
        // unset for some unrelated reason.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        tokio::spawn(async move {
            if let Ok((mut stream, _)) = listener.accept().await {
                let mut buf = [0u8; 4096];
                let _ = stream.read(&mut buf).await;
                let _ = stream
                    .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 3\r\n\r\naaa")
                    .await;
                let _ = stream.shutdown().await;
            }
        });

        let policy = PolicyConfig {
            name: "test".to_string(),
            error_handler: None,
            nodes: vec![
                NodeConfig {
                    id: "listener".to_string(),
                    node_type: "listener".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "up1".to_string(),
                    node_type: "upstream".to_string(),
                    config: {
                        let mut c = HashMap::new();
                        // Nothing listens on port 1: connect fails instantly.
                        c.insert(
                            "targets".to_string(),
                            serde_json::json!([{ "host": "127.0.0.1", "port": 1 }]),
                        );
                        c.insert("timeout_ms".to_string(), serde_json::json!(500));
                        c
                    },
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "up2".to_string(),
                    node_type: "upstream".to_string(),
                    config: {
                        let mut c = HashMap::new();
                        c.insert(
                            "targets".to_string(),
                            serde_json::json!([{ "host": "127.0.0.1", "port": port }]),
                        );
                        c
                    },
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "rw".to_string(),
                    node_type: "response-rewrite".to_string(),
                    config: {
                        let mut c = HashMap::new();
                        c.insert(
                            "filters".to_string(),
                            serde_json::json!([{ "regex": "a", "replace": "b" }]),
                        );
                        c
                    },
                    config_ref: None,
                    position: None,
                },
                NodeConfig {
                    id: "client".to_string(),
                    node_type: "client".to_string(),
                    config: HashMap::new(),
                    config_ref: None,
                    position: None,
                },
            ],
            edges: vec![
                EdgeConfig {
                    from: "listener.out".to_string(),
                    to: "up1.in".to_string(),
                },
                EdgeConfig {
                    from: "up1.success".to_string(),
                    to: "client.in".to_string(),
                },
                EdgeConfig {
                    from: "up1.error".to_string(),
                    to: "up2.in".to_string(),
                },
                EdgeConfig {
                    from: "up2.success".to_string(),
                    to: "rw.in".to_string(),
                },
                EdgeConfig {
                    from: "rw.success".to_string(),
                    to: "client.in".to_string(),
                },
            ],
        };

        let graph = compile_policy(&policy, PluginResources::empty()).unwrap();
        // Sanity-check the shape actually exercises the bug: up1 capable,
        // up2 not (confirms the compile-time inference agrees before we
        // even get to the runtime assertion below).
        assert!(
            graph.is_stream_capable("up1"),
            "up1's direct success path to client is header-only"
        );
        assert!(
            !graph.is_stream_capable("up2"),
            "up2's success path is blocked by rw's body filter"
        );

        let result = graph.execute(test_context("/test")).await;

        assert!(
            result.response.stream.is_none(),
            "up2 must have buffered: a stale __may_stream=true left by up1's \
             failed attempt must not leak across the error-port hop"
        );
        assert_eq!(
            result.response.body.as_ref(),
            b"baa",
            "rw's filter must have actually run on a real buffered body, \
             proving up2 did not silently bypass it by streaming"
        );
    }

    /// The payoff for the logger opt-out: `upstream -> http-logger -> client`
    /// is an ordinary policy, and every logger used to force it to buffer even
    /// when its format never mentioned the body.
    // `http-logger` starts a batch processor when it is constructed, so
    // building this policy needs a reactor even though the inference itself
    // is synchronous.
    #[tokio::test]
    async fn test_logger_with_a_body_free_format_is_stream_capable() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "log", "type": "http-logger",
                  "config": { "uri": "http://localhost:9/log",
                              "log_format": { "path": "{{request.path}}",
                                              "status": "{{response.status}}" } } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "log.in" },
                { "from": "log.success", "to": "client.in" }
            ]
        }));

        assert!(graph.is_stream_capable("up"));
        assert!(graph.buffering_reasons().is_empty());
    }

    /// A logger whose format does read the body still blocks, and the compiler
    /// must name it rather than silently logging an empty body.
    // `http-logger` starts a batch processor when it is constructed, so
    // building this policy needs a reactor even though the inference itself
    // is synchronous.
    #[tokio::test]
    async fn test_logger_reading_the_body_still_forces_buffering_and_is_reported() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "log", "type": "http-logger",
                  "config": { "uri": "http://localhost:9/log",
                              "log_format": { "body": "{{response.body}}" } } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "log.in" },
                { "from": "log.success", "to": "client.in" }
            ]
        }));

        assert!(!graph.is_stream_capable("up"));
        let reasons = graph.buffering_reasons();
        assert_eq!(reasons.len(), 1);
        assert_eq!(reasons[0].blocked_by_node_id, "log");
    }

    /// A logger with no `log_format` keeps buffering: the default entry
    /// records the body length, which streaming would reduce to 0.
    // `http-logger` starts a batch processor when it is constructed, so
    // building this policy needs a reactor even though the inference itself
    // is synchronous.
    #[tokio::test]
    async fn test_logger_without_a_format_still_forces_buffering() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "log", "type": "http-logger",
                  "config": { "uri": "http://localhost:9/log" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "log.in" },
                { "from": "log.success", "to": "client.in" }
            ]
        }));

        assert!(!graph.is_stream_capable("up"));
    }

    /// A cache pair links its two halves by `id`. If they disagree about the
    /// backend, the store half writes where the lookup half never reads: the
    /// route compiles, serves traffic, and returns a permanent 100% miss with
    /// no error anywhere. There is no configuration for which that is correct,
    /// so it is rejected rather than reported.
    #[tokio::test]
    async fn test_a_cache_pair_split_across_backends_is_rejected() {
        let err = compile_test_policy_err(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "look", "type": "proxy-cache",
                  "config": { "phase": "lookup", "id": "products", "policy": "local" } },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "keep", "type": "proxy-cache",
                  "config": { "phase": "store", "id": "products", "policy": "redis", "store": "s" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "look.in" },
                { "from": "look.success", "to": "up.in" },
                { "from": "look.hit", "to": "client.in" },
                { "from": "up.success", "to": "keep.in" },
                { "from": "keep.success", "to": "client.in" }
            ]
        }));

        assert!(
            err.contains("products"),
            "the error must name the pair: {err}"
        );
        assert!(
            err.contains("look") && err.contains("keep"),
            "and both halves: {err}"
        );
    }

    /// The same failure arrives through a different key: same backend kind,
    /// different store, so the two halves address different redis instances.
    #[tokio::test]
    async fn test_a_cache_pair_split_across_stores_is_rejected() {
        let err = compile_test_policy_err(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "look", "type": "proxy-cache",
                  "config": { "phase": "lookup", "id": "products", "policy": "redis", "store": "a" } },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "keep", "type": "proxy-cache",
                  "config": { "phase": "store", "id": "products", "policy": "redis", "store": "b" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "look.in" },
                { "from": "look.success", "to": "up.in" },
                { "from": "look.hit", "to": "client.in" },
                { "from": "up.success", "to": "keep.in" },
                { "from": "keep.success", "to": "client.in" }
            ]
        }));

        assert!(
            err.contains("store"),
            "the error must say which key disagrees: {err}"
        );
    }

    /// Two independent pairs must not be compared with each other -- only
    /// halves sharing an `id` form a pair, and a false positive here would
    /// reject a perfectly ordinary two-cache policy.
    #[tokio::test]
    async fn test_two_independent_cache_pairs_do_not_collide() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "look-a", "type": "proxy-cache",
                  "config": { "phase": "lookup", "id": "alpha", "policy": "local" } },
                { "id": "look-b", "type": "proxy-cache",
                  "config": { "phase": "lookup", "id": "beta", "policy": "local" } },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "keep-a", "type": "proxy-cache",
                  "config": { "phase": "store", "id": "alpha", "policy": "local" } },
                { "id": "keep-b", "type": "proxy-cache",
                  "config": { "phase": "store", "id": "beta", "policy": "local" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "look-a.in" },
                { "from": "look-a.success", "to": "look-b.in" },
                { "from": "look-a.hit", "to": "client.in" },
                { "from": "look-b.success", "to": "up.in" },
                { "from": "look-b.hit", "to": "client.in" },
                { "from": "up.success", "to": "keep-a.in" },
                { "from": "keep-a.success", "to": "keep-b.in" },
                { "from": "keep-b.success", "to": "client.in" },
                // The port spec is per TYPE, not per role, so a store-role
                // node declares `hit` too and the compiler requires it wired.
                { "from": "keep-a.hit", "to": "client.in" },
                { "from": "keep-b.hit", "to": "client.in" }
            ]
        }));

        assert!(
            graph.cache_pair_warnings().is_empty(),
            "both pairs are complete and agree"
        );
    }

    /// A lookup with no store caches nothing; a store with no lookup is never
    /// read. Both are silent no-ops -- but they are also what a policy looks
    /// like halfway through being built, so they are reported rather than
    /// rejected, the same way a buffering reason is.
    #[tokio::test]
    async fn test_a_lone_cache_half_is_reported_not_rejected() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "look", "type": "proxy-cache",
                  "config": { "phase": "lookup", "id": "orphan", "policy": "local" } },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "look.in" },
                { "from": "look.success", "to": "up.in" },
                { "from": "look.hit", "to": "client.in" },
                { "from": "up.success", "to": "client.in" }
            ]
        }));

        let warnings = graph.cache_pair_warnings();
        assert_eq!(warnings.len(), 1, "one incomplete pair: {warnings:?}");
        assert_eq!(warnings[0].cache_id, "orphan");
        assert_eq!(warnings[0].present_node_id, "look");
        assert_eq!(warnings[0].missing_role, "store");
    }

    /// The breaking change, kept visible: PortSpec is per node type, so a
    /// script node with no `respond` edge no longer compiles. The message
    /// names the port so the fix is obvious.
    #[tokio::test]
    async fn test_a_script_node_must_wire_respond() {
        let err = compile_test_policy_err(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "s", "type": "script",
                  "config": { "runtime": "lua", "inline": "function execute(ctx) return ctx end" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "s.in" },
                { "from": "s.success", "to": "client.in" }
            ]
        }));
        assert!(err.contains("respond") && err.contains("'s'"), "{err}");
    }

    /// End to end through the graph: the script's own 403 reaches the
    /// client because the node left on `respond`, skipping the upstream
    /// that would otherwise have replaced it. A browser UA takes success and
    /// gets the upstream's body -- the control that proves the branch.
    #[tokio::test]
    async fn test_a_script_taking_respond_short_circuits_the_upstream() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "block", "type": "script", "config": { "runtime": "lua", "inline":
                    "function execute(ctx)\n  local ua = (ctx.request.headers[\"user-agent\"] or {})[1] or \"\"\n  if string.find(string.lower(ua), \"scrapy\") then\n    ctx.response.status_code = 403\n    ctx.response.body = \"blocked\"\n    return ctx, \"respond\"\n  end\n  return ctx\nend" } },
                { "id": "up", "type": "mocking", "config": { "response_status": 200, "response_example": "proxied" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "block.in" },
                { "from": "block.respond", "to": "client.in" },
                { "from": "block.success", "to": "up.in" },
                { "from": "up.success", "to": "client.in" }
            ]
        }));

        let mut bot = test_context("/x");
        bot.request
            .headers
            .insert("user-agent".to_string(), vec!["scrapy/2.0".to_string()]);
        let out = graph.execute(bot).await;
        assert_eq!(out.response.status_code, 403);
        assert_eq!(out.response.body, bytes::Bytes::from_static(b"blocked"));

        let mut browser = test_context("/x");
        browser
            .request
            .headers
            .insert("user-agent".to_string(), vec!["Mozilla/5.0".to_string()]);
        let out = graph.execute(browser).await;
        assert_eq!(out.response.status_code, 200);
        assert_eq!(out.response.body, bytes::Bytes::from_static(b"proxied"));
    }

    /// A purge half is held to the same agreement rule as the other two: a
    /// purge pointed at a different backend than its pair clears nothing.
    #[tokio::test]
    async fn test_a_purge_half_split_from_its_pair_is_rejected() {
        let err = compile_test_policy_err(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "look", "type": "proxy-cache",
                  "config": { "phase": "lookup", "id": "products", "policy": "local" } },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "drop", "type": "proxy-cache",
                  "config": { "phase": "purge", "id": "products", "policy": "redis", "store": "s" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "look.in" },
                { "from": "look.success", "to": "up.in" },
                { "from": "look.hit", "to": "client.in" },
                { "from": "up.success", "to": "drop.in" },
                { "from": "drop.success", "to": "client.in" },
                { "from": "drop.hit", "to": "client.in" }
            ]
        }));
        assert!(err.contains("products") && err.contains("drop"), "{err}");
    }

    /// A purge with nothing to purge for is useless but not wrong -- reported
    /// like any other lone half.
    #[tokio::test]
    async fn test_a_lone_purge_half_is_reported() {
        let graph = compile_test_policy(serde_json::json!({
            "nodes": [
                { "id": "listener", "type": "listener", "config": {} },
                { "id": "up", "type": "upstream",
                  "config": { "targets": [{ "host": "h", "port": 80 }] } },
                { "id": "drop", "type": "proxy-cache",
                  "config": { "phase": "purge", "id": "orphan", "policy": "local" } },
                { "id": "client", "type": "client", "config": {} }
            ],
            "edges": [
                { "from": "listener.out", "to": "up.in" },
                { "from": "up.success", "to": "drop.in" },
                { "from": "drop.success", "to": "client.in" },
                { "from": "drop.hit", "to": "client.in" }
            ]
        }));
        let w = graph.cache_pair_warnings();
        assert_eq!(w.len(), 1);
        assert_eq!(w[0].present_node_id, "drop");
    }
}