alkcall 0.8.0

//! `op/register` — the wire mechanism by which a connected peer
//! announces the operations it serves (review 004 F-05, ADR-022
//! amendment). The envelope kind set stays closed at six; bootstrap ops
//! over channel 0 are the door.
//!
//! Shape: the peer sends `call.requested` for `op/register` with a
//! payload of serializable registration parts — the `OperationSpec` in
//! the `services/schema` wire shape (`spec_to_json`) plus a `replace`
//! flag. The serving-side handler rebuilds the spec, wraps a
//! call-forwarding handler that issues a nested `call.requested` back
//! over channel 0 to the announcing peer (the same shape `from_call`'s
//! imported bundles use), and writes the bundle into that connection's
//! overlay via `CallConnection::register_imported`.
//!
//! The overlay is the landing zone: `compose_root_env` attaches it
//! keyed by peer identity, so nested invocations from any composed
//! handler reach the peer-announced op, and `services/list-peers`
//! discovers it (`ctx.env.peer_operations()`).
//!
//! `AccessControl` gates the surface: the `op/register` op itself
//! carries an `AccessControl` (an unprivileged peer cannot reach the
//! handler at all — the registry's normal invoke path enforces it), and
//! an announced op that collides with an existing registration is
//! rejected unless `replace` is set (the reconnect path).
//!
//! The `Handler` closures cannot cross the wire — the announcing peer
//! keeps its handler locally; the registered bundle is a forwarding
//! stub. This is the same contract `from_call` produces for the
//! hub→consumer import direction, extended to the peer→hub direction.

use std::sync::Arc;

use serde_json::{json, Value};

use crate::core::types::Capabilities;
use crate::protocol::connection::CallConnection;
use crate::protocol::wire::{CallError, ResponseEnvelope};
use crate::registry::registration::{
    make_handler, Handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry,
};
use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};

pub const OP_REGISTER_NAME: &str = "op/register";

/// The wire DTO a peer sends as the `op/register` input. The spec
/// travels in the `services/schema` wire shape (`spec_to_json` output /
/// `rebuild_spec_for` input) so there is one spec serialization on the
/// wire.
#[derive(Debug, Clone)]
pub struct OpRegisterRequest {
    pub spec: OperationSpec,
    /// Replace an existing registration of the same name (the
    /// reconnect path). `false` (default) rejects a collision with
    /// `ALREADY_EXISTS`.
    pub replace: bool,
}

impl OpRegisterRequest {
    pub fn to_json(&self) -> Value {
        json!({
            "spec": crate::registry::discovery::spec_to_json_pub(&self.spec),
            "replace": self.replace,
        })
    }

    pub fn from_json(value: &Value) -> Result<Self, CallError> {
        let spec_json = value
            .get("spec")
            .ok_or_else(|| CallError::invalid_input("op/register payload missing `spec`"))?;
        let name = spec_json
            .get("name")
            .and_then(|v| v.as_str())
            .ok_or_else(|| CallError::invalid_input("op/register spec missing `name`"))?
            .to_string();
        let spec = crate::client::rebuild_spec_for(spec_json, &name, &None).map_err(|e| {
            CallError::invalid_input(format!("op/register spec rebuild failed: {e:?}"))
        })?;
        let replace = value
            .get("replace")
            .and_then(|v| v.as_bool())
            .unwrap_or(false);
        Ok(Self { spec, replace })
    }
}

/// The `op/register` `OperationSpec`. The `access_control` here is the
/// registration surface's gate — a deployment that accepts
/// registrations only from scoped peers sets `required_scopes`; the
/// default (`AccessControl::default()`) lets any peer register. The op
/// is `Mutation`-typed (it mutates the connection overlay).
pub fn op_register_spec(access_control: AccessControl) -> OperationSpec {
    OperationSpec::new(
        OP_REGISTER_NAME,
        OperationType::Mutation,
        Visibility::External,
        json!({
            "type": "object",
            "properties": {
                "spec": { "type": "object" },
                "replace": { "type": "boolean" }
            },
            "required": ["spec"]
        }),
        json!({
            "type": "object",
            "properties": {
                "name": { "type": "string" },
                "registered": { "type": "boolean" }
            },
            "required": ["name", "registered"]
        }),
        vec![],
        access_control,
        None,
    )
}

/// Build the `op/register` handler for a connection: announces land in
/// `connection`'s overlay via `register_imported`, wrapped as
/// call-forwarding stubs that issue a nested `call.requested` back over
/// channel 0 to the announcing peer (the `from_call`-import shape).
///
/// `Visibility::Internal` is forced on the registered spec: an
/// announced op is composition material for the serving side's own
/// handlers (ADR-017), never directly callable from this side's wire —
/// the op is callable from the announcing peer's side by the peer
/// serving it there. `services/list-peers` still discovers it (the
/// overlay is peer-keyed, provenance `FromCall`).
///
/// Collision policy (review 005 G-03): announced ops may collide with
/// other *announced* ops on the same connection (`replace` governs,
/// the reconnect path) but **never** with the serving side's own
/// registrations — a name on `serving_registry` rejects with
/// `ALREADY_EXISTS` regardless of `replace`. Without this gate the
/// connection overlay shadows the base registry in `PeerCompositeEnv`
/// (connections resolve before base), so a peer could silently
/// rewrite what a wire-dispatched handler's `ctx.env.invoke` resolves
/// for any name the deployment registered — composition authority
/// (ADR-018) belongs to the composing handler's deployer, not the
/// connected peer.
///
/// Replace semantics: a registration for the same name already on the
/// overlay is rejected with `ALREADY_EXISTS` unless `replace: true`
/// (the reconnect path re-announces).
pub fn op_register_handler(
    connection: Arc<CallConnection>,
    serving_registry: Arc<OperationRegistry>,
) -> Handler {
    make_handler(move |input, context| {
        let connection = Arc::clone(&connection);
        let serving_registry = Arc::clone(&serving_registry);
        async move {
            let request = match OpRegisterRequest::from_json(&input) {
                Ok(r) => r,
                Err(e) => return ResponseEnvelope::error(context.request_id, e),
            };

            if serving_registry.registration(&request.spec.name).is_some() {
                return ResponseEnvelope::error(
                    context.request_id,
                    CallError::already_exists(format!(
                        "op/register: `{}` is registered by this side's own serving \
                         registry; peer-announced ops may not shadow it",
                        request.spec.name
                    )),
                );
            }

            if connection.overlay_contains(&request.spec.name) && !request.replace {
                return ResponseEnvelope::error(
                    context.request_id,
                    CallError::already_exists(format!(
                        "op/register: `{}` is already registered on this connection; \
                         set `replace: true` to replace it",
                        request.spec.name
                    )),
                );
            }

            let mut spec = request.spec;
            spec.visibility = Visibility::Internal;
            let remote_name = spec.name.clone();

            let handler = forwarding_stub_for_announced_op(
                Arc::clone(&connection),
                remote_name,
                spec.op_type,
            );

            connection.register_imported(HandlerRegistration::new(
                spec.clone(),
                handler,
                OperationProvenance::FromCall,
                None,
                None,
                Capabilities::new(),
            ));

            ResponseEnvelope::ok(
                context.request_id,
                json!({ "name": spec.name, "registered": true }),
            )
        }
    })
}

/// The forwarding stub for a peer-announced op. Query/Mutation ops get
/// the `from_call` forwarding shape (nested `call.requested` back over
/// channel 0, `forwarded_for` populated per ADR-032 §3). Announced
/// Sub/Pub ops are registered as stubs that return
/// `INVALID_OPERATION_TYPE` on invocation — nested composition is
/// request/response-only (`OverlayOperationEnv`'s contract); the
/// streaming/sink forwarding shapes ride on the `from_call` import
/// path, which the announcing side can use in the other direction.
fn forwarding_stub_for_announced_op(
    connection: Arc<CallConnection>,
    remote_name: String,
    op_type: OperationType,
) -> HandlerKind {
    match op_type {
        OperationType::Query | OperationType::Mutation => HandlerKind::Once(
            crate::client::make_forwarding_handler(connection, remote_name),
        ),
        OperationType::Sub | OperationType::Pub => {
            HandlerKind::Once(make_handler(|_input, context| async move {
                ResponseEnvelope::error(
                    context.request_id,
                    CallError::invalid_operation_type(
                        "peer-announced Sub/Pub ops are not invocable over nested \
                         composition (request/response only)",
                    ),
                )
            }))
        }
    }
}

/// Announce an op to the connected peer over `connection`'s channel 0:
/// sends `call.requested` for `op/register` and awaits the response.
/// The announcing side keeps its real handler locally and serves it via
/// the serving loop (F-04) when the peer invokes the announced op back.
pub async fn announce_op(
    connection: &CallConnection,
    spec: OperationSpec,
    replace: bool,
) -> ResponseEnvelope {
    let request = OpRegisterRequest { spec, replace };
    connection
        .call_with_payload(serde_json::json!({
            "operationId": OP_REGISTER_NAME,
            "input": request.to_json(),
        }))
        .await
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::protocol::connection::CallConnection;
    use crate::registry::context::OperationContext;
    use crate::registry::discovery::install_bootstrap_discovery;
    use crate::registry::registration::OperationRegistry;
    use crate::registry::spec::Visibility;
    use std::collections::HashMap;

    fn announced_spec(name: &str) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Query,
            Visibility::External,
            json!({}),
            json!({}),
            vec![],
            AccessControl::default(),
            None,
        )
    }

    fn stub_connection() -> crate::core::types::Connection {
        crate::protocol::sink_empty_connection()
    }

    fn test_context(request_id: &str) -> OperationContext {
        OperationContext {
            request_id: request_id.to_string(),
            parent_request_id: None,
            identity: None,
            handler_identity: None,
            forwarded_for: None,
            capabilities: Capabilities::new(),
            metadata: HashMap::new(),
            scoped_env: crate::registry::context::ScopedPeerEnv::empty(),
            env: Arc::new(crate::registry::env::LocalOperationEnv::new(Arc::new(
                OperationRegistry::new(),
            ))),
            abort_policy: crate::registry::context::AbortPolicy::default(),
            deadline: None,
            internal: false,
            ownership: None,
        }
    }

    #[test]
    fn request_round_trips_through_json() {
        let request = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: true,
        };
        let json = request.to_json();
        let parsed = OpRegisterRequest::from_json(&json).expect("parse");
        assert_eq!(parsed.spec.name, "worker/exec");
        assert_eq!(parsed.spec.op_type, OperationType::Query);
        assert!(parsed.replace);
    }

    #[test]
    fn request_missing_spec_is_invalid_input() {
        let err = OpRegisterRequest::from_json(&json!({})).unwrap_err();
        assert_eq!(err.code, "INVALID_INPUT");
    }

    #[test]
    fn request_missing_name_is_invalid_input() {
        let err = OpRegisterRequest::from_json(&json!({ "spec": {} })).unwrap_err();
        assert_eq!(err.code, "INVALID_INPUT");
    }

    #[tokio::test]
    async fn handler_registers_announced_op_in_overlay() {
        let conn = Arc::new(CallConnection::new(stub_connection()));
        let handler = op_register_handler(Arc::clone(&conn), Arc::new(OperationRegistry::new()));

        let input = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: false,
        }
        .to_json();
        let response = handler(input, test_context("req-or-1")).await;
        assert!(
            response.result.is_ok(),
            "register succeeded, got {:?}",
            response.result
        );

        let registered = conn.overlay_env().contains("worker/exec");
        assert!(registered, "announced op landed in the connection overlay");
        assert!(conn.overlay_contains("worker/exec"));
    }

    #[tokio::test]
    async fn handler_rejects_collision_without_replace() {
        let conn = Arc::new(CallConnection::new(stub_connection()));
        let handler = op_register_handler(Arc::clone(&conn), Arc::new(OperationRegistry::new()));

        let input = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: false,
        }
        .to_json();
        let first = handler(input.clone(), test_context("req-or-2a")).await;
        assert!(first.result.is_ok());

        let second = handler(input, test_context("req-or-2b")).await;
        let err = second.result.expect_err("collision rejected");
        assert_eq!(err.code, "ALREADY_EXISTS");
    }

    #[tokio::test]
    async fn handler_replaces_with_replace_flag() {
        let conn = Arc::new(CallConnection::new(stub_connection()));
        let handler = op_register_handler(Arc::clone(&conn), Arc::new(OperationRegistry::new()));

        let original = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: false,
        }
        .to_json();
        let first = handler(original, test_context("req-or-3a")).await;
        assert!(first.result.is_ok());

        let replacement = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: true,
        }
        .to_json();
        let second = handler(replacement, test_context("req-or-3b")).await;
        assert!(
            second.result.is_ok(),
            "replace flag permits re-registration, got {:?}",
            second.result
        );
    }

    #[tokio::test]
    async fn registered_spec_forced_internal_with_fromcall_provenance() {
        let conn = Arc::new(CallConnection::new(stub_connection()));
        let handler = op_register_handler(Arc::clone(&conn), Arc::new(OperationRegistry::new()));

        let input = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: false,
        }
        .to_json();
        let response = handler(input, test_context("req-or-4")).await;
        assert!(response.result.is_ok());

        let registration = conn
            .overlay_registration("worker/exec")
            .expect("registered");
        assert_eq!(registration.spec.visibility, Visibility::Internal);
        assert_eq!(
            registration.provenance,
            crate::registry::registration::OperationProvenance::FromCall
        );
    }

    // --- review 005 Unit 2 acceptance gates (G-03 collision policy) -------

    /// G-03 gate: an announce colliding with a **serving-registry**
    /// name rejects with `ALREADY_EXISTS` even with `replace: true` —
    /// peer-announced ops never shadow the serving side's own
    /// registrations (composition authority stays with the deployer).
    #[tokio::test]
    async fn handler_rejects_base_registry_collision_even_with_replace() {
        let serving = Arc::new(OperationRegistry::new());
        serving
            .register(HandlerRegistration::new(
                crate::registry::spec::OperationSpec::new(
                    "fs/readFile",
                    OperationType::Query,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Once(make_handler(|input, ctx| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                crate::registry::registration::OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let conn = Arc::new(CallConnection::new(stub_connection()));
        let handler = op_register_handler(Arc::clone(&conn), Arc::clone(&serving));

        let input = OpRegisterRequest {
            spec: announced_spec("fs/readFile"),
            replace: true,
        }
        .to_json();
        let response = handler(input, test_context("req-or-5")).await;
        let err = response.result.expect_err("base collision rejected");
        assert_eq!(err.code, "ALREADY_EXISTS");
        assert!(
            !conn.overlay_contains("fs/readFile"),
            "the rejected announce never lands in the overlay"
        );
        // The serving side's registration is untouched.
        assert!(serving.registration("fs/readFile").is_some());
    }

    /// G-03 gate: an announce colliding with an **Internal** serving
    /// op is also rejected — the visibility of the shadowed op is
    /// irrelevant to composition shadowing (`OverlayOperationEnv`
    /// gates on `AccessControl`, not visibility; the composed child is
    /// `internal: true` by design).
    #[tokio::test]
    async fn handler_rejects_collision_with_internal_serving_op() {
        let serving = Arc::new(OperationRegistry::new());
        serving
            .register(HandlerRegistration::new(
                crate::registry::spec::OperationSpec::new(
                    "internal/vault",
                    OperationType::Query,
                    Visibility::Internal,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Once(make_handler(|input, ctx| async move {
                    ResponseEnvelope::ok(ctx.request_id, input)
                })),
                crate::registry::registration::OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let conn = Arc::new(CallConnection::new(stub_connection()));
        let handler = op_register_handler(Arc::clone(&conn), Arc::clone(&serving));

        let input = OpRegisterRequest {
            spec: announced_spec("internal/vault"),
            replace: false,
        }
        .to_json();
        let response = handler(input, test_context("req-or-6")).await;
        let err = response
            .result
            .expect_err("internal base collision rejected");
        assert_eq!(err.code, "ALREADY_EXISTS");
    }

    /// G-03 gate: an announced op colliding with another *announced*
    /// op still follows `replace` semantics — the base-registry gate
    /// must not widen into the overlay.
    #[tokio::test]
    async fn handler_overlay_collision_still_governed_by_replace() {
        let serving = Arc::new(OperationRegistry::new());
        let conn = Arc::new(CallConnection::new(stub_connection()));
        let handler = op_register_handler(Arc::clone(&conn), Arc::clone(&serving));

        let first = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: false,
        }
        .to_json();
        assert!(handler(first, test_context("req-or-7a"))
            .await
            .result
            .is_ok());

        let collision = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: false,
        }
        .to_json();
        let err = handler(collision, test_context("req-or-7b"))
            .await
            .result
            .expect_err("overlay collision without replace rejected");
        assert_eq!(err.code, "ALREADY_EXISTS");

        let replacement = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: true,
        }
        .to_json();
        assert!(
            handler(replacement, test_context("req-or-7c"))
                .await
                .result
                .is_ok(),
            "overlay replace still permitted; base gate is scoped to the serving registry"
        );
    }

    /// G-03 gate: after a successful announce of a distinct name,
    /// nested composition of a **base-registered** op still resolves
    /// the serving side's own op — the exact `compose_root_env` shape
    /// (`PeerCompositeEnv` with the connection overlay attached). The
    /// G-03 defect was that the collision gate was overlay-only; this
    /// pins the invariant that survived the fix: the base layer is
    /// reachable and correct whenever the name is not announced.
    #[tokio::test]
    async fn nested_composition_of_base_op_unaffected_by_unrelated_announce() {
        let serving = Arc::new(OperationRegistry::new());
        serving
            .register(HandlerRegistration::new(
                crate::registry::spec::OperationSpec::new(
                    "fs/readFile",
                    OperationType::Query,
                    Visibility::External,
                    json!({}),
                    json!({}),
                    vec![],
                    AccessControl::default(),
                    None,
                ),
                HandlerKind::Once(make_handler(|_input, ctx| async move {
                    ResponseEnvelope::ok(ctx.request_id, json!({ "from": "base" }))
                })),
                crate::registry::registration::OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();

        let conn = Arc::new(CallConnection::new(stub_connection()));
        let handler = op_register_handler(Arc::clone(&conn), Arc::clone(&serving));

        // Announce a *distinct* name; it lands in the connection overlay.
        let input = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: false,
        }
        .to_json();
        assert!(handler(input, test_context("req-or-8a"))
            .await
            .result
            .is_ok());
        assert!(conn.overlay_contains("worker/exec"));

        // Compose the base op through the same env shape
        // `compose_root_env` produces for a wire-dispatched handler.
        let base = Arc::new(crate::registry::env::LocalOperationEnv::new(Arc::clone(
            &serving,
        )));
        let mut composite = crate::registry::env::PeerCompositeEnv::new(base);
        composite.attach_peer("consumer-peer".to_string(), conn.overlay_env());
        let env: Arc<dyn crate::registry::env::OperationEnv + Send + Sync> = Arc::new(composite);

        let mut ctx = test_context("req-or-8b");
        ctx.env = env;
        ctx.scoped_env = crate::registry::context::ScopedPeerEnv::new(["fs/readFile"]);
        let response = ctx.env.invoke("fs", "readFile", json!({}), &ctx).await;
        let out = response.result.expect("base op composes");
        assert_eq!(
            out,
            json!({ "from": "base" }),
            "the serving side's own op resolves through composition, not a peer stub"
        );
    }

    /// UP-03 gate (alkhttp review 006): after a peer announces an op,
    /// `services/list-peers` over the real `compose_root_env` shape
    /// (`PeerCompositeEnv` + the connection overlay attached under the
    /// peer's id) must list the announced op under that peer's
    /// entry — not an empty operations array. The pre-fix failure:
    /// `PeerCompositeEnv` overrode `peer_ids` only, so
    /// `peer_operations` fell to the trait default (`Vec::new()`) and
    /// every peer listed with `operations: []`. ADR-030's
    /// `list_operation_names` override is what this exercises.
    #[tokio::test]
    async fn announced_op_is_discoverable_via_services_list_peers() {
        use crate::registry::env::PeerCompositeEnv;
        use crate::registry::{
            context::ScopedPeerEnv, discovery::services_list_peers_handler, env::LocalOperationEnv,
        };

        let serving = Arc::new(OperationRegistry::new());
        install_bootstrap_discovery(&serving).expect("bootstrap discovery install");

        let peer_identity = crate::core::auth::Identity {
            id: "consumer-peer".to_string(),
            scopes: vec![],
            resources: HashMap::new(),
        };
        let conn = Arc::new(CallConnection::new_overlay_only(peer_identity));
        let handler = op_register_handler(Arc::clone(&conn), Arc::clone(&serving));

        let input = OpRegisterRequest {
            spec: announced_spec("worker/exec"),
            replace: false,
        }
        .to_json();
        assert!(
            handler(input, test_context("req-or-9a"))
                .await
                .result
                .is_ok(),
            "announce lands in the connection overlay"
        );
        assert!(conn.overlay_contains("worker/exec"));

        // The exact env shape `compose_root_env` produces for calls
        // arriving on this connection: LocalOperationEnv base +
        // the connection's overlay attached under the peer's id.
        let base = Arc::new(LocalOperationEnv::new(Arc::clone(&serving)));
        let mut composite = PeerCompositeEnv::new(base);
        composite.attach_peer("consumer-peer".to_string(), conn.overlay_env());
        let env: Arc<dyn crate::registry::env::OperationEnv + Send + Sync> = Arc::new(composite);

        // Direct probe: the composite resolves the announced name from
        // the attached overlay, and peer_operations lists it.
        assert!(env.contains("worker/exec"));
        let ops = env.peer_operations(&"consumer-peer".to_string());
        assert_eq!(
            ops,
            vec!["worker/exec".to_string()],
            "PeerCompositeEnv::peer_operations must surface the peer overlay's announced ops"
        );
        assert!(
            env.peer_operations(&"unknown-peer".to_string()).is_empty(),
            "an unattached peer has no operations"
        );

        // Wire-level probe: services/list-peers over the same env
        // attributes the announced op to the peer.
        let peers_registry = Arc::new(OperationRegistry::new());
        install_bootstrap_discovery(&peers_registry).expect("bootstrap install");
        let list_handler = services_list_peers_handler(Arc::clone(&peers_registry));

        let mut ctx = test_context("req-or-9b");
        ctx.env = env;
        ctx.scoped_env = ScopedPeerEnv::empty();
        let response = list_handler(json!({}), ctx).await;
        let out = response.result.expect("list-peers ok");
        let peers_arr = out
            .get("peers")
            .and_then(|v| v.as_array())
            .expect("peers array");
        let consumer = peers_arr
            .iter()
            .find(|p| p.get("peer_id").and_then(|v| v.as_str()) == Some("consumer-peer"))
            .expect("consumer-peer present in list-peers output");
        let names: Vec<&str> = consumer
            .get("operations")
            .and_then(|v| v.as_array())
            .expect("consumer operations array")
            .iter()
            .filter_map(|o| o.get("name").and_then(|n| n.as_str()))
            .collect();
        assert!(
            names.contains(&"worker/exec"),
            "the announced op must be discoverable via services/list-peers (UP-03)"
        );

        // The local entry still lists the bootstrap ops (the serving
        // registry's own surface is unaffected).
        let local = peers_arr
            .iter()
            .find(|p| p.get("peer_id").and_then(|v| v.as_str()) == Some("local"))
            .expect("local peer present");
        let local_names: Vec<&str> = local
            .get("operations")
            .and_then(|v| v.as_array())
            .expect("consumer operations array")
            .iter()
            .filter_map(|o| o.get("name").and_then(|n| n.as_str()))
            .collect();
        assert!(local_names.contains(&"services/list-peers"));
    }

    /// ADR-047 amendment (review 008 U-1): a flavor-form marked op
    /// announced through `op/register` round-trips with the
    /// `channel_open` marker reconstructed — the explicit
    /// `channel_open_alpn` string is the carrier for the
    /// non-derivable name.
    #[test]
    fn request_round_trips_flavor_form_channel_open_marker() {
        use crate::registry::spec::ChannelOpenSpec;

        let request = OpRegisterRequest {
            spec: announced_spec("channels/tunnel/direct")
                .with_channel_open(ChannelOpenSpec::new("alk/tunnel")),
            replace: false,
        };
        let json = request.to_json();
        assert_eq!(json["spec"]["channel_open"], json!(true));
        assert_eq!(
            json["spec"]["channel_open_alpn"],
            json!("alk/tunnel"),
            "the flavor form rides the explicit ALPN string"
        );
        let parsed = OpRegisterRequest::from_json(&json).expect("parse");
        let marker = parsed
            .spec
            .channel_open
            .expect("marker survives the announce");
        assert_eq!(marker.alpn, "alk/tunnel");
    }
}