alkcall 0.8.0

//! `from_call` adapter (ADR-017 §3): discovers the remote peer's `External`
//! operations via `services/list` + `services/schema` and registers them in
//! the connection's Layer 2 overlay as `FromCall`-provenance leaves with
//! forwarding handlers.
//!
//! The discovery mechanism (`services/list` + `services/schema`) is already
//! implemented in `registry/discovery.rs`; `from_call` is the client-side
//! consumer of that API.
//!
//! See `docs/architecture/` §from_call for
//! the spec and the v1 defaults (auto-on-reconnect, error-on-collision).

use std::collections::HashSet;
use std::pin::Pin;
use std::sync::Arc;

use futures::stream::Stream;
use serde_json::{json, Value};

use crate::client::AdapterError;
use crate::core::types::Capabilities;
use crate::protocol::connection::CallConnection;
use crate::protocol::wire::ResponseEnvelope;
use crate::registry::context::OperationContext;
use crate::registry::registration::{
    Handler, HandlerKind, HandlerRegistration, OperationProvenance, SinkHandler, StreamingHandler,
};
use crate::registry::spec::{
    AccessControl, ChannelOpenSpec, ErrorDefinition, OperationSpec, OperationType, Visibility,
};

/// Configuration for [`from_call`].
///
/// Under the peer-keyed overlay model (ADR-029 §5), cross-peer collision
/// dissolves — same name on different peers lives in separate sub-overlays.
/// Same-peer collision stays an error (`AdapterError::SamePeerCollision`):
/// a peer shouldn't expose two ops with the same name.
#[derive(Debug, Clone, Default)]
pub struct FromCallConfig {
    /// Optional namespace prefix applied to imported operation names. This is
    /// local-naming sugar for when the importing node wants to expose a peer's
    /// ops under a different name *locally* — not a disambiguation mechanism
    /// (cross-peer collision dissolves under the peer-keyed model, ADR-029
    /// §5). Defaults to `None`.
    pub namespace_prefix: Option<String>,
    /// Optional filter — import only operations whose names match. `None`
    /// imports all `External` ops discovered via `services/list`.
    pub operation_filter: Option<HashSet<String>>,
}

impl FromCallConfig {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn with_namespace_prefix(mut self, prefix: impl Into<String>) -> Self {
        self.namespace_prefix = Some(prefix.into());
        self
    }

    pub fn with_operation_filter(mut self, filter: HashSet<String>) -> Self {
        self.operation_filter = Some(filter);
        self
    }
}

/// Discover the remote peer's `External` ops via `services/list` +
/// `services/schema` and construct `HandlerRegistration` bundles with
/// `FromCall` provenance and forwarding handlers. The caller registers the
/// bundles in the connection's overlay via
/// `CallConnection::register_imported_all()` — this is the peer-keyed
/// registration model (ADR-029 §5): the connection's overlay is the peer's
/// sub-overlay, aggregated into `PeerCompositeEnv` by `PeerId`.
///
/// v1 defaults (two-way doors recorded in `client-and-adapters.md`):
/// - auto-on-reconnect: the overlay is per-connection (Layer 2, ADR-024), so
///   re-import on reconnect is naturally scoped; the assembly layer calls
///   `from_call` immediately after `AlknetClient::dial_*` + `spawn_dispatch`.
/// - same-peer collision = error: two ops with the same name from the same
///   peer (after applying the optional prefix) → `AdapterError::SamePeerCollision`.
///   Cross-peer collision dissolves (ADR-029 §5).
pub async fn from_call(
    connection: &CallConnection,
    config: FromCallConfig,
) -> Result<Vec<HandlerRegistration>, AdapterError> {
    let discovered = discover_operations(connection).await?;
    build_bundles(
        discovered,
        &config.namespace_prefix,
        &config.operation_filter,
    )
}

/// Pure bundle construction extracted from [`from_call`] for testability —
/// the discovery round-trip against a live `CallConnection` is exercised by
/// integration tests; the collision rule and the forwarded_for-populating
/// handler are unit-tested here. The `peer_id` parameter records which peer's
/// sub-overlay these bundles target (ADR-029 §5); it is metadata on the
/// bundles' forwarding handlers, not used for collision detection (collision
/// is same-peer only and is checked within this set).
fn build_bundles(
    discovered: Vec<OpSummary>,
    namespace_prefix: &Option<String>,
    operation_filter: &Option<HashSet<String>>,
) -> Result<Vec<HandlerRegistration>, AdapterError> {
    let mut bundles = Vec::with_capacity(discovered.len());
    let mut seen_names = HashSet::new();

    for op_summary in discovered {
        let remote_name = op_summary.name;
        if let Some(filter) = operation_filter {
            if !filter.contains(&remote_name) {
                continue;
            }
        }

        let spec = rebuild_spec_for(&op_summary.schema, &remote_name, namespace_prefix)?;

        if !seen_names.insert(spec.name.clone()) {
            return Err(AdapterError::SamePeerCollision {
                message: format!(
                    "same-peer collision on import: {} (peer exposes two ops with the same name after prefix)",
                    spec.name
                ),
            });
        }

        let kind = match spec.op_type {
            OperationType::Sub => HandlerKind::Stream(make_streaming_forwarding_handler(
                Arc::new(op_summary.connection.clone()),
                remote_name,
            )),
            OperationType::Pub => HandlerKind::Sink(make_sink_forwarding_handler(
                Arc::new(op_summary.connection.clone()),
                remote_name,
            )),
            OperationType::Query | OperationType::Mutation => HandlerKind::Once(
                make_forwarding_handler(Arc::new(op_summary.connection.clone()), remote_name),
            ),
        };
        bundles.push(HandlerRegistration::new(
            spec,
            kind,
            OperationProvenance::FromCall,
            None,
            None,
            Capabilities::new(),
        ));
    }

    Ok(bundles)
}

#[derive(Clone)]
struct OpSummary {
    name: String,
    schema: Value,
    connection: CallConnection,
}

async fn discover_operations(connection: &CallConnection) -> Result<Vec<OpSummary>, AdapterError> {
    let response = connection.call("services/list", json!({})).await;
    let output = response.result.map_err(|e| AdapterError::DiscoveryFailed {
        message: format!("services/list failed: {} ({})", e.code, e.message),
    })?;
    let ops = output
        .get("operations")
        .and_then(|v| v.as_array())
        .ok_or_else(|| AdapterError::SchemaParse {
            message: "services/list response missing 'operations' array".to_string(),
        })?;
    let mut summaries = Vec::with_capacity(ops.len());
    for op in ops {
        let name =
            op.get("name")
                .and_then(|v| v.as_str())
                .ok_or_else(|| AdapterError::SchemaParse {
                    message: "services/list entry missing 'name'".to_string(),
                })?;
        let schema = fetch_schema(connection, name).await?;
        summaries.push(OpSummary {
            name: name.to_string(),
            schema,
            connection: connection.clone(),
        });
    }
    Ok(summaries)
}

async fn fetch_schema(connection: &CallConnection, name: &str) -> Result<Value, AdapterError> {
    let response = connection
        .call("services/schema", json!({ "name": name }))
        .await;
    response.result.map_err(|e| AdapterError::DiscoveryFailed {
        message: format!(
            "services/schema for {name} failed: {} ({})",
            e.code, e.message
        ),
    })
}

/// Rebuild an `OperationSpec` from the `services/schema` JSON, applying the
/// optional namespace prefix. The spec JSON shape matches `spec_to_json` in
/// `registry/discovery.rs`.
///
/// `pub(crate)` so the `op/register` bootstrap path (review 004 F-05) can
/// rebuild peer-announced specs from the same wire shape — one parser, two
/// consumers.
pub(crate) fn rebuild_spec_for(
    schema_json: &Value,
    remote_name: &str,
    namespace_prefix: &Option<String>,
) -> Result<OperationSpec, AdapterError> {
    let op_type = parse_op_type(
        schema_json
            .get("op_type")
            .and_then(|v| v.as_str())
            .ok_or_else(|| AdapterError::SchemaParse {
                message: format!("schema for {remote_name} missing op_type"),
            })?,
    )?;
    let visibility = parse_visibility(
        schema_json
            .get("visibility")
            .and_then(|v| v.as_str())
            .unwrap_or("external"),
    );
    let input_schema = schema_json
        .get("input_schema")
        .cloned()
        .unwrap_or(Value::Null);
    let output_schema = schema_json
        .get("output_schema")
        .cloned()
        .unwrap_or(Value::Null);
    let error_schemas = schema_json
        .get("error_schemas")
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().filter_map(parse_error_definition).collect())
        .unwrap_or_default();
    let access_control = schema_json
        .get("access_control")
        .map(parse_access_control)
        .unwrap_or_default();

    let name = match namespace_prefix {
        Some(prefix) if !prefix.is_empty() => format!("{prefix}/{remote_name}"),
        _ => remote_name.to_string(),
    };

    let mut spec = OperationSpec::new(
        name,
        op_type,
        visibility,
        input_schema,
        output_schema,
        error_schemas,
        access_control,
        schema_json
            .get("resource_id_path")
            .and_then(|v| v.as_str())
            .map(String::from),
    );

    // ADR-047 §2: the `channel_open` marker survives discovery
    // serialization as a boolean (plus the explicit `channel_open_alpn`
    // string for non-derivable op names — ADR-047 amendment, review 008
    // U-1). The consumer (e.g. the hub) branches on the marker to wrap
    // marked ops with relay machinery (ADR-047 §1, Gap C) instead of the
    // plain forwarding stub. The marker is on the spec so the consumer
    // can see it without re-fetching `services/schema`.
    let explicit_alpn = schema_json
        .get("channel_open_alpn")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty());
    if schema_json
        .get("channel_open")
        .and_then(|v| v.as_bool())
        .unwrap_or(false)
    {
        let alpn = explicit_alpn
            .map(String::from)
            .or_else(|| derive_alpn_from_op_name(remote_name));
        if let Some(alpn) = alpn {
            spec = spec.with_channel_open(ChannelOpenSpec::new(alpn));
        }
    } else if let Some(_alpn) = explicit_alpn {
        // A `channel_open_alpn` without the boolean marker is not a
        // channel-open op — the boolean is the dispatch hint; the
        // string only disambiguates the ALPN for a marked op.
        tracing::debug!(
            op = remote_name,
            "rebuild_spec_for: channel_open_alpn present without the channel_open \
             boolean marker — ignored (the boolean is the dispatch hint)"
        );
    }

    if let Some(publish_schema) = schema_json.get("publish_schema") {
        if !publish_schema.is_null() {
            spec = spec.with_publish_schema(publish_schema.clone());
        }
    }

    if let Some(description) = schema_json.get("description").and_then(|v| v.as_str()) {
        spec = spec.with_description(description);
    }

    Ok(spec)
}

/// Derive the data-plane ALPN from an open-op name
/// (`channels/<alpn>/sub` → `alk/<alpn>`). Generalized (ADR-047
/// amendment — review 008 U-1): the derivation strips the LAST path
/// segment, which covers the standard `…/sub` and `…/pub` shapes
/// (byte-identical behavior) AND the flavor form
/// (`channels/tunnel/direct` → `alk/tunnel`) — the op-type suffix is
/// not special-cased; the boolean marker is the gate (consulted only
/// for marked ops in `rebuild_spec_for`). Returns `None` for op names
/// without a `channels/` prefix or a trailing segment — the op is not
/// a derivable channel-open op, and the marker (if present) without an
/// explicit `channel_open_alpn` is ignored. ADR-047 §"Negative": the
/// path segment is the ALPN with the `alk/` prefix stripped; ALPNs
/// without that prefix use their full ALPN string as the path segment
/// (rare case). Multi-segment non-`alk/*` ALPNs (e.g. `custom/proto`)
/// survive because the last segment is stripped from `rest` rather
/// than taking only the first path segment.
fn derive_alpn_from_op_name(op_name: &str) -> Option<String> {
    let rest = op_name.strip_prefix("channels/")?;
    let (segment, _flavor) = rest.rsplit_once('/')?;
    if segment.is_empty() {
        return None;
    }
    if segment.starts_with("alk/") || segment == "alk" || segment.contains('/') {
        Some(segment.to_string())
    } else {
        Some(format!("alk/{segment}"))
    }
}

fn parse_op_type(s: &str) -> Result<OperationType, AdapterError> {
    match s {
        "query" => Ok(OperationType::Query),
        "mutation" => Ok(OperationType::Mutation),
        "sub" => Ok(OperationType::Sub),
        "pub" => Ok(OperationType::Pub),
        other => Err(AdapterError::SchemaParse {
            message: format!("unknown op_type: {other}"),
        }),
    }
}

fn parse_visibility(s: &str) -> Visibility {
    match s {
        "internal" => Visibility::Internal,
        _ => Visibility::External,
    }
}

fn parse_error_definition(v: &Value) -> Option<ErrorDefinition> {
    Some(ErrorDefinition {
        code: v.get("code")?.as_str()?.to_string(),
        description: v
            .get("description")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string(),
        schema: v.get("schema").cloned().unwrap_or(Value::Null),
        http_status: v
            .get("http_status")
            .and_then(|v| v.as_u64())
            .map(|n| n as u16),
    })
}

fn parse_access_control(v: &Value) -> AccessControl {
    AccessControl {
        required_scopes: v
            .get("required_scopes")
            .and_then(|v| v.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|s| s.as_str().map(String::from))
                    .collect()
            })
            .unwrap_or_default(),
        required_scopes_any: v
            .get("required_scopes_any")
            .and_then(|v| v.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|s| s.as_str().map(String::from))
                    .collect()
            }),
        resource_type: v
            .get("resource_type")
            .and_then(|v| v.as_str())
            .map(String::from),
        resource_action: v
            .get("resource_action")
            .and_then(|v| v.as_str())
            .map(String::from),
    }
}

/// Construct a forwarding handler for a `FromCall` `Query`/`Mutation` leaf:
/// on invocation, calls the remote op via the `CallConnection` and returns
/// its `ResponseEnvelope` (single `call_with_payload()`, `HandlerKind::Once`).
/// `Subscription` ops use [`make_streaming_forwarding_handler`] instead.
///
/// Per ADR-032 §3, the handler populates `forwarded_for` on the
/// `call.requested` payload from the hub's `OperationContext.identity` (the
/// end user the hub authenticated). The hub authenticates as itself when
/// forwarding. The spoke authorizes the hub (its direct caller);
/// `forwarded_for` is metadata, never read by `AccessControl::check`.
///
/// If `context.identity` is `None` (the hub chose not to disclose, or has not
/// authenticated an originator), `forwarded_for` is omitted — the spoke
/// receives only the hub's identity.
pub(crate) fn make_forwarding_handler(
    connection: Arc<CallConnection>,
    remote_name: String,
) -> Handler {
    use crate::registry::registration::make_handler;
    make_handler(move |input, context| {
        let connection = Arc::clone(&connection);
        let remote_name = remote_name.clone();
        async move {
            let payload = build_forwarded_payload(&remote_name, input, &context);
            // The forwarding handler invokes the remote op via the
            // CallConnection. The parent_request_id participates in the abort
            // cascade (ADR-016 §6): if the parent is aborted, the cascade
            // reaches this handler, which sends call.aborted to the remote
            // node; the remote node cascades to its own descendants.
            // Cross-node abort is transparent.
            let response = connection.call_with_payload(payload).await;
            ResponseEnvelope {
                request_id: context.request_id,
                result: response.result,
            }
        }
    })
}

/// Construct a streaming forwarding handler for a `FromCall` `Subscription`
/// leaf: on invocation, calls `CallConnection::subscribe_with_payload()` and
/// forwards the remote stream end-to-end. Each `call.responded` from the
/// remote becomes a stream item, `call.completed` ends the stream, and
/// `call.aborted` drops it (ADR-021 §8). No truncation, no first-value
/// fallback.
///
/// `forwarded_for` is populated from `context.identity` (ADR-032 §3), exactly
/// as the request/response forwarding handler does — both via
/// `build_forwarded_payload` (no new payload-construction code). The
/// `subscribe_with_payload` path registers the request in
/// `PendingRequestMap`, so the abort cascade (ADR-016 §6) is already wired:
/// a parent abort drops the `SubscriptionStream`, which sends `call.aborted`
/// to the remote node.
pub(crate) fn make_streaming_forwarding_handler(
    connection: Arc<CallConnection>,
    remote_name: String,
) -> StreamingHandler {
    use crate::registry::registration::make_streaming_handler;
    use futures::stream::{once, StreamExt};
    make_streaming_handler(move |input, context| {
        let connection = Arc::clone(&connection);
        let remote_name = remote_name.clone();
        once(async move {
            let payload = build_forwarded_payload(&remote_name, input, &context);
            connection.subscribe_with_payload(payload).await
        })
        .flatten()
    })
}

/// Construct a sink forwarding handler for a `FromCall` `Pub` leaf
/// (ADR-046): on invocation, calls `CallConnection::publish_with_payload()`
/// and forwards the local `PublishStream` to the remote as
/// `call.published` events. The remote's single `call.responded` becomes
/// the handler's `ResponseEnvelope`. No truncation — the full stream is
/// forwarded end-to-end (streamed, not buffered: the local stream's
/// items are pumped to the remote as they arrive).
///
/// An `Err` item in the local `PublishStream` (an initiator-side error)
/// terminates the forwarding — the stream is cut at the error, matching
/// the `SinkHandler` contract (`Err` terminates the stream). The
/// remote's `call.completed` is sent after the last `Ok` item; the
/// remote handler sees a truncated stream.
///
/// `forwarded_for` is populated from `context.identity` (ADR-032 §3),
/// exactly as the request/response and streaming forwarding handlers
/// do — both via `build_forwarded_payload`.
pub(crate) fn make_sink_forwarding_handler(
    connection: Arc<CallConnection>,
    remote_name: String,
) -> SinkHandler {
    use crate::registry::registration::make_sink_handler;
    use futures::stream::StreamExt;
    make_sink_handler(move |input, context, publish_stream| {
        let connection = Arc::clone(&connection);
        let remote_name = remote_name.clone();
        async move {
            let payload = build_forwarded_payload(&remote_name, input, &context);
            let value_stream: Pin<Box<dyn Stream<Item = Value> + Send>> = Box::pin(
                publish_stream
                    .take_while(|item| futures::future::ready(item.is_ok()))
                    .filter_map(|item| futures::future::ready(item.ok())),
            );
            connection.publish_with_payload(payload, value_stream).await
        }
    })
}

/// Build the `call.requested` payload for a forwarded call, populating
/// `forwarded_for` from the hub's `OperationContext.identity` (ADR-032 §3).
/// `forwarded_for` is omitted when `context.identity` is `None` (the hub
/// chooses not to disclose the originator).
pub(crate) fn build_forwarded_payload(
    operation_id: &str,
    input: Value,
    context: &OperationContext,
) -> Value {
    let mut payload = serde_json::Map::new();
    payload.insert(
        "operationId".to_string(),
        Value::String(operation_id.to_string()),
    );
    payload.insert("input".to_string(), input);
    if let Some(originator) = &context.identity {
        if let Ok(value) = serde_json::to_value(originator) {
            payload.insert("forwarded_for".to_string(), value);
        }
    }
    Value::Object(payload)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::auth::Identity;
    use crate::core::types::Capabilities;
    use crate::protocol::connection::CallConnection;
    use crate::registry::discovery::spec_to_json;
    use crate::registry::registration::{make_handler, make_streaming_handler, PublishStream};
    use crate::registry::spec::OperationType;
    use futures::StreamExt;
    use std::collections::HashMap;
    use std::sync::Mutex as StdMutex;

    use crate::protocol::sink_empty_connection as stub_connection;

    fn sample_schema_json(name: &str, op_type: &str) -> Value {
        json!({
            "name": name,
            "namespace": name.split('/').next().unwrap_or(""),
            "op_type": op_type,
            "visibility": "external",
            "input_schema": {"type": "object"},
            "output_schema": {"type": "string"},
            "error_schemas": [],
            "access_control": {"required_scopes": []},
        })
    }

    #[test]
    fn rebuild_spec_no_prefix_preserves_name() {
        let schema = sample_schema_json("fs/readFile", "query");
        let spec = rebuild_spec_for(&schema, "fs/readFile", &None).expect("rebuild");
        assert_eq!(spec.name, "fs/readFile");
        assert_eq!(spec.op_type, OperationType::Query);
        assert_eq!(spec.visibility, Visibility::External);
    }

    #[test]
    fn rebuild_spec_with_prefix_applies_prefix() {
        let schema = sample_schema_json("fs/readFile", "query");
        let spec =
            rebuild_spec_for(&schema, "fs/readFile", &Some("worker".to_string())).expect("rebuild");
        assert_eq!(spec.name, "worker/fs/readFile");
    }

    #[test]
    fn rebuild_spec_unknown_op_type_returns_schema_parse() {
        let schema = sample_schema_json("fs/readFile", "weird");
        match rebuild_spec_for(&schema, "fs/readFile", &None) {
            Err(AdapterError::SchemaParse { .. }) => {}
            other => panic!("expected SchemaParse, got {other:?}"),
        }
    }

    #[test]
    fn rebuild_spec_missing_op_type_returns_schema_parse() {
        let schema = json!({"name": "fs/readFile"});
        match rebuild_spec_for(&schema, "fs/readFile", &None) {
            Err(AdapterError::SchemaParse { .. }) => {}
            other => panic!("expected SchemaParse, got {other:?}"),
        }
    }

    #[test]
    fn rebuild_spec_parses_error_schemas_and_acl() {
        let schema = json!({
            "name": "fs/readFileErr",
            "namespace": "fs",
            "op_type": "query",
            "visibility": "external",
            "input_schema": {},
            "output_schema": {},
            "error_schemas": [{
                "code": "FILE_NOT_FOUND",
                "description": "file not found",
                "schema": {"type": "object"},
                "http_status": 404,
            }],
            "access_control": {
                "required_scopes": ["fs:read"],
                "required_scopes_any": null,
                "resource_type": "fs",
                "resource_action": "read",
            },
        });
        let spec = rebuild_spec_for(&schema, "fs/readFileErr", &None).expect("rebuild");
        assert_eq!(spec.error_schemas.len(), 1);
        assert_eq!(spec.error_schemas[0].code, "FILE_NOT_FOUND");
        assert_eq!(spec.error_schemas[0].http_status, Some(404));
        assert_eq!(
            spec.access_control.required_scopes,
            vec!["fs:read".to_string()]
        );
        assert_eq!(spec.access_control.resource_type.as_deref(), Some("fs"));
    }

    // --- review 005 Unit 3 gate (G-04 spec wire round-trip) ----------------

    /// G-04 gate: `resource_id_path` survives the spec wire round-trip —
    /// `spec_to_json_pub` serializes it, `rebuild_spec_for` parses it.
    /// The review found the field silently dropped: an announced (or
    /// `from_call`-imported) op with ownership-scoped resource
    /// extraction rebuilt with `resource_id: None`, so the rebuilt
    /// spec's ACL checks ran without the resource ID.
    #[test]
    fn spec_round_trips_resource_id_path() {
        use crate::registry::discovery::spec_to_json_pub;

        let spec = OperationSpec::new(
            "fs/readFile",
            OperationType::Query,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            crate::registry::spec::AccessControl::default(),
            Some("/path".to_string()),
        );
        let wire = spec_to_json_pub(&spec);
        assert_eq!(
            wire.get("resource_id_path").and_then(|v| v.as_str()),
            Some("/path"),
            "resource_id_path serialized"
        );

        let rebuilt = rebuild_spec_for(&wire, "fs/readFile", &None).expect("rebuild");
        assert_eq!(
            rebuilt.resource_id_path.as_deref(),
            Some("/path"),
            "resource_id_path survives the round-trip"
        );
    }

    /// G-04 companion: a spec without `resource_id_path` serializes no
    /// `resource_id_path` key and rebuilds with `None` (additive
    /// optional field — absent stays absent).
    #[test]
    fn spec_without_resource_id_path_stays_absent_through_round_trip() {
        use crate::registry::discovery::spec_to_json_pub;

        let spec = OperationSpec::new(
            "fs/readFile",
            OperationType::Query,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            crate::registry::spec::AccessControl::default(),
            None,
        );
        let wire = spec_to_json_pub(&spec);
        assert!(wire.get("resource_id_path").is_none());

        let rebuilt = rebuild_spec_for(&wire, "fs/readFile", &None).expect("rebuild");
        assert_eq!(rebuilt.resource_id_path, None);
    }

    #[test]
    fn rebuild_spec_channel_open_marker_set_for_channels_alpn_op() {
        let mut schema = sample_schema_json("channels/tty/sub", "sub");
        schema["channel_open"] = json!(true);
        let spec = rebuild_spec_for(&schema, "channels/tty/sub", &None).expect("rebuild");
        let marker = spec.channel_open.expect("channel_open marker parsed");
        assert_eq!(marker.alpn, "alk/tty");
    }

    #[test]
    fn rebuild_spec_channel_open_marker_absent_for_plain_op() {
        let schema = sample_schema_json("fs/readFile", "query");
        let spec = rebuild_spec_for(&schema, "fs/readFile", &None).expect("rebuild");
        assert!(
            spec.channel_open.is_none(),
            "plain op must not get a channel_open marker"
        );
    }

    #[test]
    fn rebuild_spec_channel_open_marker_false_is_absent() {
        let mut schema = sample_schema_json("channels/tty/sub", "sub");
        schema["channel_open"] = json!(false);
        let spec = rebuild_spec_for(&schema, "channels/tty/sub", &None).expect("rebuild");
        assert!(
            spec.channel_open.is_none(),
            "channel_open: false must be treated as absent"
        );
    }

    #[test]
    fn rebuild_spec_channel_open_marker_ignored_for_non_channels_op_name() {
        // A spec that claims channel_open=true but isn't a channels/<alpn>/*
        // op: the marker is ignored (no ALPN derivable). The op is treated
        // as a plain op. This is a defensive check — a well-behaved
        // producer shouldn't set the marker on a non-channels op.
        let mut schema = sample_schema_json("fs/readFile", "query");
        schema["channel_open"] = json!(true);
        let spec = rebuild_spec_for(&schema, "fs/readFile", &None).expect("rebuild");
        assert!(
            spec.channel_open.is_none(),
            "marker on non-channels op name is ignored"
        );
    }

    // --- ADR-047 amendment (review 008 U-1): flavor form + explicit ALPN --

    /// Review 008 U-1 gate 1: a flavor-form marked op reconstructs the
    /// spec WITH the marker — via the explicit `channel_open_alpn`
    /// string when the producer supplied it, via the generalized
    /// derivation otherwise (the deployment-skew case: an old producer
    /// serving the boolean only).
    #[test]
    fn rebuild_spec_flavor_form_reconstructs_marker_via_explicit_alpn() {
        use crate::registry::discovery::spec_to_json_pub;
        use crate::registry::spec::ChannelOpenSpec;

        let spec = OperationSpec::new(
            "channels/tunnel/direct",
            OperationType::Sub,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            crate::registry::spec::AccessControl::default(),
            None,
        )
        .with_channel_open(ChannelOpenSpec::new("alk/tunnel"));

        // New producer: the wire payload carries the explicit string.
        let wire = spec_to_json_pub(&spec);
        assert_eq!(wire["channel_open"], json!(true));
        assert_eq!(
            wire["channel_open_alpn"],
            json!("alk/tunnel"),
            "non-standard shape emits the explicit ALPN"
        );
        let rebuilt = rebuild_spec_for(&wire, "channels/tunnel/direct", &None).expect("rebuild");
        let marker = rebuilt.channel_open.expect("marker reconstructed");
        assert_eq!(marker.alpn, "alk/tunnel");

        // Deployment skew — an OLD alkcall producer serves the boolean
        // only (no `channel_open_alpn`); the new consumer still
        // reconstructs the marker via the generalized derivation.
        let mut old_wire = wire.clone();
        old_wire
            .as_object_mut()
            .expect("object")
            .remove("channel_open_alpn");
        let rebuilt =
            rebuild_spec_for(&old_wire, "channels/tunnel/direct", &None).expect("rebuild");
        let marker = rebuilt
            .channel_open
            .expect("marker reconstructed from the boolean alone (skew case)");
        assert_eq!(marker.alpn, "alk/tunnel");
    }

    #[test]
    fn rebuild_spec_explicit_alpn_overrides_derivation() {
        // The explicit string wins over the name-derived ALPN — the
        // disambiguator for the residual ambiguity (an ALPN segment
        // that collides with a flavor name).
        let mut schema = sample_schema_json("channels/x/direct", "sub");
        schema["channel_open"] = json!(true);
        schema["channel_open_alpn"] = json!("alk/x/direct");
        let spec = rebuild_spec_for(&schema, "channels/x/direct", &None).expect("rebuild");
        let marker = spec.channel_open.expect("marker");
        assert_eq!(
            marker.alpn, "alk/x/direct",
            "the explicit string wins over the strip-last derivation (alk/x)"
        );
    }

    #[test]
    fn rebuild_spec_explicit_alpn_without_boolean_is_ignored() {
        // The boolean marker is the dispatch hint; the string only
        // disambiguates the ALPN for a marked op. A string alone must
        // NOT mark a plain op.
        let mut schema = sample_schema_json("fs/readFile", "query");
        schema["channel_open_alpn"] = json!("alk/tty");
        let spec = rebuild_spec_for(&schema, "fs/readFile", &None).expect("rebuild");
        assert!(
            spec.channel_open.is_none(),
            "channel_open_alpn without the boolean never marks an op"
        );
    }

    /// An empty/whitespace explicit `channel_open_alpn` must not
    /// override a sane name derivation (review-008 post-landing audit
    /// F-3): a single misconfigured producer would otherwise poison the
    /// marker for all consumers with an empty ALPN.
    #[test]
    fn rebuild_spec_empty_explicit_alpn_falls_back_to_derivation() {
        for bad in ["", "   "] {
            let mut schema = sample_schema_json("channels/tunnel/direct", "sub");
            schema["channel_open"] = json!(true);
            schema["channel_open_alpn"] = json!(bad);
            let spec = rebuild_spec_for(&schema, "channels/tunnel/direct", &None).expect("rebuild");
            let marker = spec
                .channel_open
                .expect("marker reconstructed via the derivation fallback");
            assert_eq!(
                marker.alpn, "alk/tunnel",
                "empty explicit ALPN `{bad}` must not override the derivation"
            );
        }
    }

    /// A marked op named `channels//sub` (empty segment) must NOT
    /// serialize boolean-only and then reconstruct unmarked — the
    /// standard-shape check applies the same empty-segment guard the
    /// derivation does, so the explicit string rides the wire
    /// (review-008 post-landing audit N-4).
    #[test]
    fn spec_empty_segment_name_is_not_treated_as_standard_shape() {
        use crate::registry::discovery::{
            op_name_is_standard_channel_open_shape, spec_to_json_pub,
        };
        use crate::registry::spec::ChannelOpenSpec;

        assert!(!op_name_is_standard_channel_open_shape("channels//sub"));

        let spec = OperationSpec::new(
            "channels//sub",
            OperationType::Sub,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            crate::registry::spec::AccessControl::default(),
            None,
        )
        .with_channel_open(ChannelOpenSpec::new("alk/tty"));
        let wire = spec_to_json_pub(&spec);
        assert_eq!(
            wire["channel_open_alpn"],
            json!("alk/tty"),
            "the empty-segment name is non-derivable — the explicit string must ride"
        );
        let rebuilt = rebuild_spec_for(&wire, "channels//sub", &None).expect("rebuild");
        let marker = rebuilt.channel_open.expect("marker reconstructs");
        assert_eq!(marker.alpn, "alk/tty");
    }

    /// Review 008 U-1 gate 3: standard-shape ops round-trip unchanged —
    /// boolean only, no `channel_open_alpn` key, byte-stable.
    #[test]
    fn spec_standard_shape_channel_open_stays_boolean_only() {
        use crate::registry::discovery::spec_to_json_pub;
        use crate::registry::spec::ChannelOpenSpec;

        for name in ["channels/tty/sub", "channels/custom/proto/sub"] {
            let spec = OperationSpec::new(
                name,
                OperationType::Sub,
                Visibility::External,
                json!({}),
                json!({}),
                vec![],
                crate::registry::spec::AccessControl::default(),
                None,
            )
            .with_channel_open(ChannelOpenSpec::new(if name == "channels/tty/sub" {
                "alk/tty"
            } else {
                "custom/proto"
            }));
            let wire = spec_to_json_pub(&spec);
            assert_eq!(wire["channel_open"], json!(true), "{name}");
            assert!(
                wire.get("channel_open_alpn").is_none(),
                "{name}: standard shape must not emit the explicit key (byte-stable)"
            );
            let rebuilt = rebuild_spec_for(&wire, name, &None).expect("rebuild");
            let marker = rebuilt.channel_open.expect("marker");
            assert_eq!(
                marker.alpn.as_ref(),
                wire["channel_open_alpn"]
                    .as_str()
                    .unwrap_or(if name == "channels/tty/sub" {
                        "alk/tty"
                    } else {
                        "custom/proto"
                    }),
                "{name} round-trips"
            );
        }
    }

    /// Review 009 C-5: the standard-shape wire payload is
    /// golden-pinned as a full object against a literal (the entire
    /// key set, not just the `channel_open_alpn` absence) — the shape
    /// ADR-047 amendment 3 advertises as byte-stable.
    #[test]
    fn spec_standard_shape_wire_payload_golden_pin() {
        use crate::registry::discovery::spec_to_json_pub;
        use crate::registry::spec::ChannelOpenSpec;

        let spec = OperationSpec::new(
            "channels/tty/sub",
            OperationType::Sub,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            crate::registry::spec::AccessControl::default(),
            None,
        )
        .with_channel_open(ChannelOpenSpec::new("alk/tty"));
        let wire = spec_to_json_pub(&spec);
        assert_eq!(
            wire,
            json!({
                "name": "channels/tty/sub",
                "namespace": "channels",
                "op_type": "sub",
                "visibility": "external",
                "input_schema": {},
                "output_schema": {},
                "error_schemas": [],
                "access_control": {
                    "required_scopes": [],
                    "required_scopes_any": null,
                    "resource_type": null,
                    "resource_action": null,
                },
                "channel_open": true,
            }),
            "the standard-shape payload is exactly this literal — no \
             channel_open_alpn key, nothing else added"
        );
    }

    /// The full flavor-form round trip through the `op/register`
    /// announced-spec path (the same parser as `from_call`).
    #[test]
    fn spec_round_trips_flavor_form_marker() {
        use crate::registry::discovery::spec_to_json_pub;
        use crate::registry::spec::ChannelOpenSpec;

        let spec = OperationSpec::new(
            "channels/tunnel/direct",
            OperationType::Sub,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            crate::registry::spec::AccessControl::default(),
            None,
        )
        .with_channel_open(ChannelOpenSpec::new("alk/tunnel"))
        .with_description("dynamic-target egress (alktunnels ADR-007)");
        let wire = spec_to_json_pub(&spec);
        let rebuilt = rebuild_spec_for(&wire, "channels/tunnel/direct", &None).expect("rebuild");
        let marker = rebuilt
            .channel_open
            .expect("marker survives the round trip");
        assert_eq!(marker.alpn, "alk/tunnel");
        assert_eq!(
            rebuilt.description.as_deref(),
            Some("dynamic-target egress (alktunnels ADR-007)")
        );
    }

    #[test]
    fn rebuild_spec_publish_schema_set_when_present() {
        let publish_schema = json!({
            "type": "object",
            "properties": { "bytes": { "type": "string" } },
            "required": ["bytes"]
        });
        let mut schema = sample_schema_json("fs/upload", "pub");
        schema["publish_schema"] = publish_schema.clone();
        let spec = rebuild_spec_for(&schema, "fs/upload", &None).expect("rebuild");
        assert_eq!(spec.publish_schema.as_ref(), Some(&publish_schema));
    }

    #[test]
    fn rebuild_spec_publish_schema_absent_when_omitted() {
        let schema = sample_schema_json("fs/upload", "pub");
        let spec = rebuild_spec_for(&schema, "fs/upload", &None).expect("rebuild");
        assert!(
            spec.publish_schema.is_none(),
            "publish_schema must be absent when the discovered schema omits it"
        );
    }

    #[test]
    fn rebuild_spec_publish_schema_absent_when_null() {
        let mut schema = sample_schema_json("fs/upload", "pub");
        schema["publish_schema"] = Value::Null;
        let spec = rebuild_spec_for(&schema, "fs/upload", &None).expect("rebuild");
        assert!(
            spec.publish_schema.is_none(),
            "publish_schema: null must be treated as absent"
        );
    }

    #[test]
    fn rebuild_spec_publish_schema_round_trips_with_spec_to_json() {
        let publish_schema = json!({
            "type": "object",
            "properties": { "n": { "type": "integer" } },
            "required": ["n"]
        });
        let spec = OperationSpec::new(
            "fs/upload",
            OperationType::Pub,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            AccessControl::default(),
            None,
        )
        .with_publish_schema(publish_schema.clone());
        let serialized = spec_to_json(&spec);
        let rebuilt = rebuild_spec_for(&serialized, "fs/upload", &None).expect("rebuild");
        assert_eq!(rebuilt.publish_schema.as_ref(), Some(&publish_schema));
    }

    /// E-02 gate (review 006): `description` survives the spec wire
    /// round-trip — `spec_to_json_pub` serializes it when set,
    /// `rebuild_spec_for` parses it back (the `op/register` announced-spec
    /// path and the `from_call` import path both parse through here).
    #[test]
    fn spec_round_trips_description() {
        use crate::registry::discovery::spec_to_json_pub;

        let spec = OperationSpec::new(
            "channels/tty/sub",
            OperationType::Sub,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            crate::registry::spec::AccessControl::default(),
            None,
        )
        .with_description("Interactive TTY sessions");
        let wire = spec_to_json_pub(&spec);
        assert_eq!(
            wire.get("description").and_then(|v| v.as_str()),
            Some("Interactive TTY sessions"),
            "description serialized"
        );

        let rebuilt = rebuild_spec_for(&wire, "channels/tty/sub", &None).expect("rebuild");
        assert_eq!(
            rebuilt.description.as_deref(),
            Some("Interactive TTY sessions"),
            "description survives the round-trip"
        );
    }

    /// E-02 companion: a spec without `description` serializes no
    /// `description` key and rebuilds with `None` (additive optional
    /// field — absent stays absent, old producers stay parseable).
    #[test]
    fn spec_without_description_stays_absent_through_round_trip() {
        use crate::registry::discovery::spec_to_json_pub;

        let spec = OperationSpec::new(
            "fs/readFile",
            OperationType::Query,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            crate::registry::spec::AccessControl::default(),
            None,
        );
        let wire = spec_to_json_pub(&spec);
        assert!(wire.get("description").is_none());

        let rebuilt = rebuild_spec_for(&wire, "fs/readFile", &None).expect("rebuild");
        assert_eq!(rebuilt.description, None);
    }

    #[test]
    fn derive_alpn_from_op_name_strips_channels_prefix() {
        assert_eq!(
            derive_alpn_from_op_name("channels/tty/sub"),
            Some("alk/tty".to_string())
        );
        assert_eq!(
            derive_alpn_from_op_name("channels/tunnel/pub"),
            Some("alk/tunnel".to_string())
        );
    }

    #[test]
    fn derive_alpn_from_op_name_returns_none_for_non_channels_op() {
        assert_eq!(derive_alpn_from_op_name("fs/readFile"), None);
        assert_eq!(derive_alpn_from_op_name("channel/open"), None);
        assert_eq!(derive_alpn_from_op_name("channels/"), None);
    }

    #[test]
    fn derive_alpn_from_op_name_multi_segment_non_alknet_alpn_survives() {
        assert_eq!(
            derive_alpn_from_op_name("channels/custom/proto/sub"),
            Some("custom/proto".to_string()),
            "multi-segment non-alk/* ALPN uses full ALPN as the path segment"
        );
        assert_eq!(
            derive_alpn_from_op_name("channels/vendor/service/run/pub"),
            Some("vendor/service/run".to_string())
        );
    }

    #[test]
    fn derive_alpn_from_op_name_explicit_alknet_prefix_returned_as_is() {
        assert_eq!(
            derive_alpn_from_op_name("channels/alk/tty/sub"),
            Some("alk/tty".to_string())
        );
    }

    #[test]
    fn derive_alpn_from_op_name_strips_pub_suffix() {
        assert_eq!(
            derive_alpn_from_op_name("channels/tty/pub"),
            Some("alk/tty".to_string())
        );
    }

    #[test]
    fn derive_alpn_from_op_name_flavor_form() {
        // ADR-047 amendment (review 008 U-1): the flavor form — the
        // gate is the boolean marker at the rebuild level, the
        // derivation just strips the last segment.
        assert_eq!(
            derive_alpn_from_op_name("channels/tunnel/direct"),
            Some("alk/tunnel".to_string()),
            "alktunnels ADR-007's direct op (gate 1)"
        );
        assert_eq!(
            derive_alpn_from_op_name("channels/tunnel/forwarded"),
            Some("alk/tunnel".to_string()),
            "alktunnels ADR-008's forwarded op"
        );
    }

    #[test]
    fn derive_alpn_from_op_name_non_op_type_suffix_is_not_special_cased() {
        // The generalized derivation strips the LAST segment
        // unconditionally — `sub`/`pub` are not special-cased. The
        // op-type shape check lives at the rebuild level (the boolean
        // marker is the gate), so a plain op named `channels/tty/query`
        // derives here but never reaches the marker path
        // (`rebuild_spec_channel_open_marker_*` prove the gate).
        assert_eq!(
            derive_alpn_from_op_name("channels/tty/query"),
            Some("alk/tty".to_string()),
            "the derivation is shape-blind; the marker is the gate"
        );
        assert_eq!(derive_alpn_from_op_name("channels/tty"), None);
        assert_eq!(derive_alpn_from_op_name("channels/"), None);
    }

    /// Review 009 C-6: the derivation's edge shapes. The empty-segment
    /// guard (`channels//sub` → `None`) is the N-4 audit fix's
    /// foundation; the bare no-slash name (`channels` → `None`) has no
    /// trailing segment to strip. The 4-segment name is the deliberate
    /// strict-superset behavior (the LAST segment is stripped whatever
    /// it is) — a behavior change vs the pre-amendment derivation,
    /// which stripped only `/sub`//`/pub` and returned `None` here
    /// (review 008 U-1 generalized it).
    #[test]
    fn derive_alpn_from_op_name_edge_shapes() {
        assert_eq!(
            derive_alpn_from_op_name("channels//sub"),
            None,
            "the empty segment is not a derivable ALPN"
        );
        assert_eq!(
            derive_alpn_from_op_name("channels//direct"),
            None,
            "the guard is flavor-blind"
        );
        assert_eq!(
            derive_alpn_from_op_name("channels"),
            None,
            "no path segment to strip"
        );
        assert_eq!(
            derive_alpn_from_op_name("channels/x/sub/extra"),
            Some("x/sub".to_string()),
            "the last segment is stripped unconditionally — `x/sub` is \
             treated as a full (multi-segment) ALPN, the same rule the \
             5-segment case pins; pre-amendment this returned None \
             (only /sub//pub were stripped)"
        );
        assert_eq!(
            derive_alpn_from_op_name("channels/alk/tty/sub"),
            Some("alk/tty".to_string()),
            "the verbatim alk/* segment rides as-is"
        );
    }

    #[test]
    fn from_call_config_builder_methods() {
        let config = FromCallConfig::new()
            .with_namespace_prefix("worker")
            .with_operation_filter(HashSet::from(["fs/readFile".to_string()]));
        assert_eq!(config.namespace_prefix.as_deref(), Some("worker"));
        assert!(config.operation_filter.unwrap().contains("fs/readFile"));
    }

    /// `from_call` against a stub `CallConnection` (no real transport) returns
    /// a `DiscoveryFailed` because `services/list` can't dispatch on a mock
    /// connection. This verifies the error path rather than the happy path
    /// (the happy path is covered by the integration test in a later task).
    #[tokio::test]
    async fn from_call_against_mock_connection_returns_discovery_failed() {
        let conn = CallConnection::new(stub_connection());
        let result = from_call(&conn, FromCallConfig::new()).await;
        match result {
            Err(AdapterError::DiscoveryFailed { .. }) => {}
            Err(other) => panic!("expected DiscoveryFailed, got another error variant: {other}"),
            Ok(_) => panic!("expected DiscoveryFailed on mock connection, got Ok"),
        }
    }

    #[test]
    fn from_call_provenance_is_from_call_and_leaf_fields() {
        // Verify the registration shape produced by from_call: provenance
        // FromCall, no composition authority, no scoped_env, empty caps.
        // Uses a synthetic spec to avoid the transport round-trip.
        let spec = OperationSpec::new(
            "worker/echo",
            OperationType::Query,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            AccessControl::default(),
            None,
        );
        let handler = make_forwarding_handler(
            Arc::new(CallConnection::new(stub_connection())),
            "worker/echo".to_string(),
        );
        let reg = HandlerRegistration::new(
            spec,
            HandlerKind::Once(handler),
            OperationProvenance::FromCall,
            None,
            None,
            Capabilities::new(),
        );
        assert_eq!(reg.provenance, OperationProvenance::FromCall);
        assert!(reg.composition_authority.is_none());
        assert!(reg.scoped_env.is_none());
    }

    // --- ADR-032: forwarded_for population --------------------------------

    struct NoopEnv;
    #[async_trait::async_trait]
    impl crate::registry::env::OperationEnv for NoopEnv {
        async fn invoke_with_policy(
            &self,
            _ns: &str,
            _op: &str,
            _input: Value,
            parent: &OperationContext,
            _policy: crate::registry::context::AbortPolicy,
        ) -> ResponseEnvelope {
            ResponseEnvelope::ok(parent.request_id.clone(), Value::Null)
        }
        fn contains(&self, _name: &str) -> bool {
            false
        }
    }

    fn test_context(identity: Option<Identity>) -> OperationContext {
        use crate::registry::context::{AbortPolicy, ScopedPeerEnv};
        use std::collections::HashMap;
        use std::time::{Duration, Instant};
        OperationContext {
            request_id: "req-test".to_string(),
            parent_request_id: None,
            identity,
            handler_identity: None,
            forwarded_for: None,
            capabilities: Capabilities::new(),
            metadata: HashMap::new(),
            scoped_env: ScopedPeerEnv::empty(),
            env: Arc::new(NoopEnv),
            abort_policy: AbortPolicy::default(),
            deadline: Some(Instant::now() + Duration::from_secs(30)),
            internal: false,
            ownership: None,
        }
    }

    fn alice_identity() -> Identity {
        Identity {
            id: "alice".to_string(),
            scopes: vec!["fs:read".to_string()],
            resources: HashMap::new(),
        }
    }

    #[test]
    fn build_forwarded_payload_populates_forwarded_for_from_context_identity() {
        let ctx = test_context(Some(alice_identity()));
        let payload = build_forwarded_payload("fs/readFile", json!({"p": 1}), &ctx);
        assert_eq!(payload["operationId"], "fs/readFile");
        assert_eq!(payload["input"], json!({"p": 1}));
        let forwarded_for = payload.get("forwarded_for").expect("forwarded_for present");
        assert_eq!(forwarded_for["id"], "alice");
        assert_eq!(forwarded_for["scopes"][0], "fs:read");
    }

    #[test]
    fn build_forwarded_payload_omits_forwarded_for_when_context_identity_is_none() {
        let ctx = test_context(None);
        let payload = build_forwarded_payload("fs/readFile", json!({}), &ctx);
        assert!(payload.get("forwarded_for").is_none());
        assert_eq!(payload["operationId"], "fs/readFile");
    }

    /// Verify the forwarding handler actually populates `forwarded_for` on
    /// the wire payload it sends. We intercept the payload by using a handler
    /// that records the payload passed to `call_with_payload`. Since
    /// `call_with_payload` on a mock connection returns an error envelope
    /// (no transport), we instead test the payload-construction function
    /// directly (above) and rely on the handler wiring to call
    /// `call_with_payload(payload)`. The handler's contract is: read
    /// `context.identity`, build the payload, call. The payload-construction
    /// function is the unit under test.
    #[tokio::test]
    async fn forwarding_handler_populates_forwarded_for_from_context_identity() {
        let conn = Arc::new(CallConnection::new(stub_connection()));
        let captured_payload = Arc::new(StdMutex::new(None::<Value>));
        let captured = Arc::clone(&captured_payload);

        let handler: Handler = {
            let conn = Arc::clone(&conn);
            make_handler(move |input, context| {
                let conn = Arc::clone(&conn);
                let captured = Arc::clone(&captured);
                let remote_name = "fs/readFile".to_string();
                async move {
                    let payload = build_forwarded_payload(&remote_name, input, &context);
                    *captured.lock().unwrap() = Some(payload.clone());
                    let response = conn.call_with_payload(payload).await;
                    ResponseEnvelope {
                        request_id: context.request_id,
                        result: response.result,
                    }
                }
            })
        };

        let ctx = test_context(Some(alice_identity()));
        let _ = handler(json!({}), ctx).await;
        let payload = captured_payload.lock().unwrap().clone().expect("captured");
        assert_eq!(payload["forwarded_for"]["id"], "alice");
        assert_eq!(payload["operationId"], "fs/readFile");
    }

    #[tokio::test]
    async fn forwarding_handler_omits_forwarded_for_when_context_identity_is_none() {
        let conn = Arc::new(CallConnection::new(stub_connection()));
        let captured_payload = Arc::new(StdMutex::new(None::<Value>));
        let captured = Arc::clone(&captured_payload);

        let handler: Handler = {
            let conn = Arc::clone(&conn);
            make_handler(move |input, context| {
                let conn = Arc::clone(&conn);
                let captured = Arc::clone(&captured);
                let remote_name = "fs/readFile".to_string();
                async move {
                    let payload = build_forwarded_payload(&remote_name, input, &context);
                    *captured.lock().unwrap() = Some(payload.clone());
                    let response = conn.call_with_payload(payload).await;
                    ResponseEnvelope {
                        request_id: context.request_id,
                        result: response.result,
                    }
                }
            })
        };

        let ctx = test_context(None);
        let _ = handler(json!({}), ctx).await;
        let payload = captured_payload.lock().unwrap().clone().expect("captured");
        assert!(
            payload.get("forwarded_for").is_none(),
            "forwarded_for must be omitted when context.identity is None"
        );
    }

    // --- ADR-029 §5: collision rule ---------------------------------------

    fn op_summary(name: &str, conn: &CallConnection) -> OpSummary {
        OpSummary {
            name: name.to_string(),
            schema: sample_schema_json(name, "query"),
            connection: conn.clone(),
        }
    }

    fn op_summary_typed(name: &str, op_type: &str, conn: &CallConnection) -> OpSummary {
        OpSummary {
            name: name.to_string(),
            schema: sample_schema_json(name, op_type),
            connection: conn.clone(),
        }
    }

    #[test]
    fn build_bundles_same_peer_collision_returns_same_peer_collision_error() {
        let conn = CallConnection::new(stub_connection());
        // Same peer exposing two ops that resolve to the same name after the
        // (empty) prefix → SamePeerCollision.
        let discovered = vec![
            op_summary("worker/exec", &conn),
            op_summary("worker/exec", &conn),
        ];
        match build_bundles(discovered, &None, &None) {
            Err(AdapterError::SamePeerCollision { message }) => {
                assert!(message.contains("worker/exec"));
            }
            Err(other) => panic!("expected SamePeerCollision, got another error: {other}"),
            Ok(_) => panic!("expected SamePeerCollision, got Ok"),
        }
    }

    #[test]
    fn build_bundles_same_peer_collision_after_prefix_returns_error() {
        let conn = CallConnection::new(stub_connection());
        // Two ops with different remote names that collide after the prefix is
        // applied (prefix drops, then same name) → SamePeerCollision. Here we
        // use the same remote name twice, which is the canonical same-peer
        // collision.
        let discovered = vec![
            op_summary("fs/readFile", &conn),
            op_summary("fs/readFile", &conn),
        ];
        match build_bundles(discovered, &Some("worker".to_string()), &None) {
            Err(AdapterError::SamePeerCollision { message }) => {
                assert!(message.contains("worker/fs/readFile"));
            }
            Err(other) => panic!("expected SamePeerCollision, got another error: {other}"),
            Ok(_) => panic!("expected SamePeerCollision, got Ok"),
        }
    }

    #[test]
    fn build_bundles_cross_peer_same_name_does_not_collide() {
        // Cross-peer collision dissolves (ADR-029 §5): the same name on
        // different peers lives in separate sub-overlays. `from_call` runs
        // per-connection (per-peer), so the `build_bundles` collision check is
        // same-peer only. This test verifies that a single `build_bundles`
        // call with distinct names succeeds — the cross-peer case is
        // structurally separate `from_call` invocations on different
        // connections, each producing its own bundle set with no collision.
        let conn_a = CallConnection::new(stub_connection());
        let conn_b = CallConnection::new(stub_connection());

        let bundles_a = build_bundles(vec![op_summary("container/exec", &conn_a)], &None, &None)
            .expect("peer a bundles");
        let bundles_b = build_bundles(vec![op_summary("container/exec", &conn_b)], &None, &None)
            .expect("peer b bundles");

        assert_eq!(bundles_a.len(), 1);
        assert_eq!(bundles_b.len(), 1);
        assert_eq!(bundles_a[0].spec.name, "container/exec");
        assert_eq!(bundles_b[0].spec.name, "container/exec");
        // Same name, different peer sub-overlays — no collision.
    }

    #[test]
    fn build_bundles_distinct_names_in_same_peer_do_not_collide() {
        let conn = CallConnection::new(stub_connection());
        let discovered = vec![
            op_summary("worker/exec", &conn),
            op_summary("worker/status", &conn),
            op_summary("fs/readFile", &conn),
        ];
        let bundles = build_bundles(discovered, &None, &None).expect("distinct names ok");
        assert_eq!(bundles.len(), 3);
        for b in &bundles {
            assert_eq!(b.provenance, OperationProvenance::FromCall);
        }
    }

    #[test]
    fn build_bundles_applies_namespace_prefix_without_collision() {
        let conn = CallConnection::new(stub_connection());
        let discovered = vec![op_summary("exec", &conn), op_summary("status", &conn)];
        let bundles =
            build_bundles(discovered, &Some("worker".to_string()), &None).expect("prefixed ok");
        assert_eq!(bundles[0].spec.name, "worker/exec");
        assert_eq!(bundles[1].spec.name, "worker/status");
    }

    #[test]
    fn build_bundles_respects_operation_filter() {
        let conn = CallConnection::new(stub_connection());
        let discovered = vec![
            op_summary("worker/exec", &conn),
            op_summary("worker/status", &conn),
            op_summary("fs/readFile", &conn),
        ];
        let filter: HashSet<String> = HashSet::from(["worker/exec".to_string()]);
        let bundles = build_bundles(discovered, &None, &Some(filter)).expect("filtered ok");
        assert_eq!(bundles.len(), 1);
        assert_eq!(bundles[0].spec.name, "worker/exec");
    }

    // --- ADR-021 §8: streaming forwarding for Sub ops ---------------------

    #[test]
    fn build_bundles_subscription_op_produces_stream_kind() {
        let conn = CallConnection::new(stub_connection());
        let discovered = vec![op_summary_typed("events/stream", "sub", &conn)];
        let bundles = build_bundles(discovered, &None, &None).expect("bundles");
        assert_eq!(bundles.len(), 1);
        assert_eq!(bundles[0].spec.op_type, OperationType::Sub);
        assert!(
            matches!(bundles[0].handler, HandlerKind::Stream(_)),
            "Sub op must register HandlerKind::Stream"
        );
        assert_eq!(bundles[0].provenance, OperationProvenance::FromCall);
        assert!(bundles[0].composition_authority.is_none());
        assert!(bundles[0].scoped_env.is_none());
    }

    #[test]
    fn build_bundles_query_op_produces_once_kind() {
        let conn = CallConnection::new(stub_connection());
        let discovered = vec![op_summary_typed("fs/readFile", "query", &conn)];
        let bundles = build_bundles(discovered, &None, &None).expect("bundles");
        assert_eq!(bundles.len(), 1);
        assert_eq!(bundles[0].spec.op_type, OperationType::Query);
        assert!(
            matches!(bundles[0].handler, HandlerKind::Once(_)),
            "Query op must register HandlerKind::Once"
        );
    }

    #[test]
    fn build_bundles_mutation_op_produces_once_kind() {
        let conn = CallConnection::new(stub_connection());
        let discovered = vec![op_summary_typed("fs/writeFile", "mutation", &conn)];
        let bundles = build_bundles(discovered, &None, &None).expect("bundles");
        assert_eq!(bundles.len(), 1);
        assert_eq!(bundles[0].spec.op_type, OperationType::Mutation);
        assert!(
            matches!(bundles[0].handler, HandlerKind::Once(_)),
            "Mutation op must register HandlerKind::Once"
        );
    }

    #[test]
    fn build_bundles_mixed_op_types_route_to_correct_kind() {
        let conn = CallConnection::new(stub_connection());
        let discovered = vec![
            op_summary_typed("fs/readFile", "query", &conn),
            op_summary_typed("fs/writeFile", "mutation", &conn),
            op_summary_typed("events/stream", "sub", &conn),
        ];
        let bundles = build_bundles(discovered, &None, &None).expect("bundles");
        assert_eq!(bundles.len(), 3);
        let by_name: std::collections::HashMap<&str, &HandlerKind> = bundles
            .iter()
            .map(|b| (b.spec.name.as_str(), &b.handler))
            .collect();
        assert!(matches!(by_name["fs/readFile"], HandlerKind::Once(_)));
        assert!(matches!(by_name["fs/writeFile"], HandlerKind::Once(_)));
        assert!(matches!(by_name["events/stream"], HandlerKind::Stream(_)));
    }

    /// Verify `make_streaming_forwarding_handler` produces a `StreamingHandler`
    /// that builds the forwarded payload with `forwarded_for` populated from
    /// `context.identity` (ADR-032) and calls `subscribe_with_payload`. Since
    /// `subscribe_with_payload` on a mock connection returns a closed stream
    /// (no transport), we capture the payload by intercepting the build step:
    /// the handler's contract is "build payload via `build_forwarded_payload`,
    /// then call `subscribe_with_payload(payload)`". We mirror the existing
    /// `forwarding_handler_populates_forwarded_for` test by constructing the
    /// handler and exercising the payload-construction path it relies on, plus
    /// asserting the produced stream terminates (the mock-connection path
    /// yields one error envelope then ends — no truncation, no hang).
    #[tokio::test]
    async fn streaming_forwarding_handler_populates_forwarded_for_and_streams() {
        use futures::stream::StreamExt;

        let conn = Arc::new(CallConnection::new(stub_connection()));
        let captured_payload = Arc::new(StdMutex::new(None::<Value>));
        let captured = Arc::clone(&captured_payload);

        let handler: StreamingHandler = {
            let conn = Arc::clone(&conn);
            make_streaming_handler(move |input, context| {
                let conn = Arc::clone(&conn);
                let captured = Arc::clone(&captured);
                let remote_name = "events/stream".to_string();
                use futures::stream::{once, StreamExt};
                once(async move {
                    let payload = build_forwarded_payload(&remote_name, input, &context);
                    *captured.lock().unwrap() = Some(payload.clone());
                    conn.subscribe_with_payload(payload).await
                })
                .flatten()
            })
        };

        let ctx = test_context(Some(alice_identity()));
        let mut stream = handler(json!({}), ctx);
        let first = stream.next().await;
        assert!(
            first.is_some(),
            "streaming forwarding handler must produce at least one envelope"
        );
        if let Some(env) = first {
            assert!(
                env.result.is_err(),
                "mock connection has no transport, so the stream yields an error envelope"
            );
        }
        let second = stream.next().await;
        assert!(
            second.is_none(),
            "stream must terminate after the error (no truncation, no hang)"
        );

        let payload = captured_payload.lock().unwrap().clone().expect("captured");
        assert_eq!(payload["operationId"], "events/stream");
        assert_eq!(payload["forwarded_for"]["id"], "alice");
    }

    /// The streaming forwarding handler omits `forwarded_for` when
    /// `context.identity` is `None`, mirroring the request/response handler.
    #[tokio::test]
    async fn streaming_forwarding_handler_omits_forwarded_for_when_identity_none() {
        use futures::stream::StreamExt;

        let conn = Arc::new(CallConnection::new(stub_connection()));
        let captured_payload = Arc::new(StdMutex::new(None::<Value>));
        let captured = Arc::clone(&captured_payload);

        let handler: StreamingHandler = {
            let conn = Arc::clone(&conn);
            make_streaming_handler(move |input, context| {
                let conn = Arc::clone(&conn);
                let captured = Arc::clone(&captured);
                let remote_name = "events/stream".to_string();
                use futures::stream::{once, StreamExt};
                once(async move {
                    let payload = build_forwarded_payload(&remote_name, input, &context);
                    *captured.lock().unwrap() = Some(payload.clone());
                    conn.subscribe_with_payload(payload).await
                })
                .flatten()
            })
        };

        let ctx = test_context(None);
        let mut stream = handler(json!({}), ctx);
        let _ = stream.next().await;
        let payload = captured_payload.lock().unwrap().clone().expect("captured");
        assert!(
            payload.get("forwarded_for").is_none(),
            "forwarded_for must be omitted when context.identity is None"
        );
        assert_eq!(payload["operationId"], "events/stream");
    }

    /// `make_streaming_forwarding_handler` produces a `StreamingHandler` (not a
    /// `Handler`) — verifies the helper returns the right type and that
    /// `build_bundles` wires it into `HandlerKind::Stream`.
    #[test]
    fn make_streaming_forwarding_handler_returns_streaming_handler() {
        let handler = make_streaming_forwarding_handler(
            Arc::new(CallConnection::new(stub_connection())),
            "events/stream".to_string(),
        );
        let reg = HandlerRegistration::new(
            OperationSpec::new(
                "events/stream",
                OperationType::Sub,
                Visibility::External,
                json!({}),
                json!({}),
                vec![],
                AccessControl::default(),
                None,
            ),
            HandlerKind::Stream(handler),
            OperationProvenance::FromCall,
            None,
            None,
            Capabilities::new(),
        );
        assert!(matches!(reg.handler, HandlerKind::Stream(_)));
        assert_eq!(reg.provenance, OperationProvenance::FromCall);
        assert!(reg.composition_authority.is_none());
        assert!(reg.scoped_env.is_none());
    }

    // --- ADR-046: sink forwarding for Pub ops ----------------------------

    #[test]
    fn build_bundles_pub_op_produces_sink_kind() {
        let conn = CallConnection::new(stub_connection());
        let discovered = vec![op_summary_typed("fs/upload", "pub", &conn)];
        let bundles = build_bundles(discovered, &None, &None).expect("bundles");
        assert_eq!(bundles.len(), 1);
        assert_eq!(bundles[0].spec.op_type, OperationType::Pub);
        assert!(
            matches!(bundles[0].handler, HandlerKind::Sink(_)),
            "Pub op must register HandlerKind::Sink"
        );
        assert_eq!(bundles[0].provenance, OperationProvenance::FromCall);
        assert!(bundles[0].composition_authority.is_none());
        assert!(bundles[0].scoped_env.is_none());
    }

    #[test]
    fn build_bundles_mixed_with_pub_routes_to_correct_kind() {
        let conn = CallConnection::new(stub_connection());
        let discovered = vec![
            op_summary_typed("fs/readFile", "query", &conn),
            op_summary_typed("fs/upload", "pub", &conn),
            op_summary_typed("events/stream", "sub", &conn),
        ];
        let bundles = build_bundles(discovered, &None, &None).expect("bundles");
        assert_eq!(bundles.len(), 3);
        let by_name: std::collections::HashMap<&str, &HandlerKind> = bundles
            .iter()
            .map(|b| (b.spec.name.as_str(), &b.handler))
            .collect();
        assert!(matches!(by_name["fs/readFile"], HandlerKind::Once(_)));
        assert!(matches!(by_name["fs/upload"], HandlerKind::Sink(_)));
        assert!(matches!(by_name["events/stream"], HandlerKind::Stream(_)));
    }

    /// `make_sink_forwarding_handler` produces a `SinkHandler` that builds the
    /// forwarded payload with `forwarded_for` populated from `context.identity`
    /// (ADR-032) and calls `publish_with_payload`. Since `publish_with_payload`
    /// on a mock connection returns an error envelope (no transport), we capture
    /// the payload by intercepting the build step: the handler's contract is
    /// "build payload via `build_forwarded_payload`, then call
    /// `publish_with_payload(payload, stream)".
    #[tokio::test]
    async fn sink_forwarding_handler_populates_forwarded_for() {
        use crate::registry::registration::make_sink_handler;
        use futures::stream;

        let conn = Arc::new(CallConnection::new(stub_connection()));
        let captured_payload = Arc::new(StdMutex::new(None::<Value>));
        let captured = Arc::clone(&captured_payload);

        let handler: SinkHandler = {
            let conn = Arc::clone(&conn);
            make_sink_handler(move |input, context, publish_stream| {
                let conn = Arc::clone(&conn);
                let captured = Arc::clone(&captured);
                let remote_name = "fs/upload".to_string();
                async move {
                    let payload = build_forwarded_payload(&remote_name, input, &context);
                    *captured.lock().unwrap() = Some(payload.clone());
                    let value_stream: Pin<Box<dyn Stream<Item = Value> + Send>> = Box::pin(
                        publish_stream
                            .take_while(|item| futures::future::ready(item.is_ok()))
                            .filter_map(|item| futures::future::ready(item.ok())),
                    );
                    conn.publish_with_payload(payload, value_stream).await
                }
            })
        };

        let ctx = test_context(Some(alice_identity()));
        let chunks: Vec<Result<Value, crate::protocol::wire::CallError>> =
            vec![Ok(json!({"chunk": 1})), Ok(json!({"chunk": 2}))];
        let publish_stream: PublishStream = Box::pin(stream::iter(chunks));
        let response = handler(json!({}), ctx, publish_stream).await;
        assert!(
            response.result.is_err(),
            "mock connection has no transport, so the handler yields an error envelope"
        );

        let payload = captured_payload.lock().unwrap().clone().expect("captured");
        assert_eq!(payload["operationId"], "fs/upload");
        assert_eq!(payload["forwarded_for"]["id"], "alice");
    }

    /// The sink forwarding handler omits `forwarded_for` when
    /// `context.identity` is `None`, mirroring the request/response handler.
    #[tokio::test]
    async fn sink_forwarding_handler_omits_forwarded_for_when_identity_none() {
        use crate::registry::registration::make_sink_handler;
        use futures::stream;

        let conn = Arc::new(CallConnection::new(stub_connection()));
        let captured_payload = Arc::new(StdMutex::new(None::<Value>));
        let captured = Arc::clone(&captured_payload);

        let handler: SinkHandler = {
            let conn = Arc::clone(&conn);
            make_sink_handler(move |input, context, publish_stream| {
                let conn = Arc::clone(&conn);
                let captured = Arc::clone(&captured);
                let remote_name = "fs/upload".to_string();
                async move {
                    let payload = build_forwarded_payload(&remote_name, input, &context);
                    *captured.lock().unwrap() = Some(payload.clone());
                    let value_stream: Pin<Box<dyn Stream<Item = Value> + Send>> = Box::pin(
                        publish_stream
                            .take_while(|item| futures::future::ready(item.is_ok()))
                            .filter_map(|item| futures::future::ready(item.ok())),
                    );
                    conn.publish_with_payload(payload, value_stream).await
                }
            })
        };

        let ctx = test_context(None);
        let chunks: Vec<Result<Value, crate::protocol::wire::CallError>> =
            vec![Ok(json!({"c": 1}))];
        let publish_stream: PublishStream = Box::pin(stream::iter(chunks));
        let _ = handler(json!({}), ctx, publish_stream).await;

        let payload = captured_payload.lock().unwrap().clone().expect("captured");
        assert!(
            payload.get("forwarded_for").is_none(),
            "forwarded_for must be omitted when context.identity is None"
        );
        assert_eq!(payload["operationId"], "fs/upload");
    }

    /// `make_sink_forwarding_handler` produces a `SinkHandler` (not a
    /// `Handler`) — verifies the helper returns the right type and that
    /// `build_bundles` wires it into `HandlerKind::Sink`.
    #[test]
    fn make_sink_forwarding_handler_returns_sink_handler() {
        let handler = make_sink_forwarding_handler(
            Arc::new(CallConnection::new(stub_connection())),
            "fs/upload".to_string(),
        );
        let reg = HandlerRegistration::new(
            OperationSpec::new(
                "fs/upload",
                OperationType::Pub,
                Visibility::External,
                json!({}),
                json!({}),
                vec![],
                AccessControl::default(),
                None,
            ),
            HandlerKind::Sink(handler),
            OperationProvenance::FromCall,
            None,
            None,
            Capabilities::new(),
        );
        assert!(matches!(reg.handler, HandlerKind::Sink(_)));
        assert_eq!(reg.provenance, OperationProvenance::FromCall);
        assert!(reg.composition_authority.is_none());
        assert!(reg.scoped_env.is_none());
    }
}