alkcall 0.8.0

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;

use crate::core::types::Capabilities;
use futures::stream::{self, Stream};
use parking_lot::RwLock;
use serde_json::Value;

use super::context::{CompositionAuthority, OperationContext, ScopedPeerEnv};
use super::spec::{AccessResult, OperationSpec, OperationType, Visibility};
use crate::protocol::wire::{CallError, ResponseEnvelope};

pub type Handler = Arc<
    dyn Fn(Value, OperationContext) -> Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>
        + Send
        + Sync,
>;

pub type StreamingHandler = Arc<
    dyn Fn(Value, OperationContext) -> Pin<Box<dyn Stream<Item = ResponseEnvelope> + Send>>
        + Send
        + Sync,
>;

/// Sink handler — `Pub` operations (ADR-046). Receives the initiator's
/// data stream (each `call.published` chunk's `input` as one `Value`
/// item) and returns a single `ResponseEnvelope` (the result of
/// consuming the stream). An `Err` item in the stream (an initiator-side
/// error) terminates the stream early; the handler may produce its
/// result from the partial input or propagate the error.
pub type SinkHandler = Arc<
    dyn Fn(
            Value,
            OperationContext,
            Pin<Box<dyn Stream<Item = Result<Value, CallError>> + Send>>,
        ) -> Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>
        + Send
        + Sync,
>;

pub type ResponseStream = Pin<Box<dyn Stream<Item = ResponseEnvelope> + Send>>;

/// The publish stream shape consumed by a `SinkHandler`. Each item is
/// one published chunk: `Ok(value)` for a `call.published` event,
/// `Err(call_error)` for an initiator-side `call.error`. Stream end
/// (the initiator's `call.completed` or write-half close) yields
/// `None` — the `SinkHandler` sees a terminated stream.
pub type PublishStream = Pin<Box<dyn Stream<Item = Result<Value, CallError>> + Send>>;

#[derive(Clone)]
pub enum HandlerKind {
    Once(Handler),
    Stream(StreamingHandler),
    Sink(SinkHandler),
}

impl std::fmt::Debug for HandlerKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Once(_) => f.write_str("Once(<Handler>)"),
            Self::Stream(_) => f.write_str("Stream(<StreamingHandler>)"),
            Self::Sink(_) => f.write_str("Sink(<SinkHandler>)"),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OperationProvenance {
    Local,
    FromOpenAPI,
    FromMCP,
    FromCall,
    FromJsonSchema,
    Session,
}

#[derive(Debug, Clone)]
pub struct HandlerRegistration {
    pub spec: OperationSpec,
    pub handler: HandlerKind,
    pub provenance: OperationProvenance,
    pub composition_authority: Option<CompositionAuthority>,
    pub scoped_env: Option<ScopedPeerEnv>,
    pub capabilities: Capabilities,
}

impl HandlerRegistration {
    pub fn new(
        spec: OperationSpec,
        handler: HandlerKind,
        provenance: OperationProvenance,
        composition_authority: Option<CompositionAuthority>,
        scoped_env: Option<ScopedPeerEnv>,
        capabilities: Capabilities,
    ) -> Self {
        Self {
            spec,
            handler,
            provenance,
            composition_authority,
            scoped_env,
            capabilities,
        }
    }
}

pub struct OperationRegistry {
    operations: RwLock<HashMap<String, HandlerRegistration>>,
    validators: RwLock<HashMap<String, Option<jsonschema::Validator>>>,
    input_validators: RwLock<HashMap<String, Option<jsonschema::Validator>>>,
}

impl OperationRegistry {
    pub fn new() -> Self {
        Self {
            operations: RwLock::new(HashMap::new()),
            validators: RwLock::new(HashMap::new()),
            input_validators: RwLock::new(HashMap::new()),
        }
    }

    /// Fork the registry: a deep copy of every registration (handlers
    /// and provenance included) plus the cached publish-schema and
    /// input-schema validators (review 004 F-02/F-03). The fork is an
    /// independent registry — later registrations on the original do
    /// not appear on the fork and vice versa. It is the per-session
    /// composition seam (ADR-047 §4 amendment): fork the base
    /// registry, register the per-session ops (generic channel ops,
    /// openables, bootstrap ops) against the fork, and dispatch the
    /// session over the fork.
    pub fn fork(&self) -> OperationRegistry {
        let operations = self.operations.read().clone();
        let validators = self.validators.read().clone();
        let input_validators = self.input_validators.read().clone();
        OperationRegistry {
            operations: RwLock::new(operations),
            validators: RwLock::new(validators),
            input_validators: RwLock::new(input_validators),
        }
    }

    /// Compile a spec's `publish_schema` (CF-003). Un-compilable schemas
    /// are rejected at registration time — registering one is an error,
    /// never a silent unvalidated ingest path. A compiled validator is
    /// cached so the dispatch path does zero per-request work; the cached
    /// value also distinguishes "no schema declared" (`None`) from a
    /// schema that failed to compile, though the latter can no longer
    /// reach this registry.
    fn validate_and_cache_publish_schema(
        validators: &mut HashMap<String, Option<jsonschema::Validator>>,
        spec: &OperationSpec,
    ) -> Result<(), String> {
        let compiled = match &spec.publish_schema {
            None => None,
            Some(schema) => match jsonschema::options().build(schema) {
                Ok(v) => Some(v),
                Err(e) => {
                    return Err(format!(
                        "publish_schema for operation {:?} failed to compile: {e}",
                        spec.name
                    ));
                }
            },
        };
        validators.insert(spec.name.clone(), compiled);
        Ok(())
    }

    /// Compile a spec's `input_schema` for call-time enforcement
    /// (ADR-016: `INVALID_INPUT` is the registry's schema-mismatch
    /// error code; the advertise-vs-enforce leg every protocol crate
    /// leans on — the channels open-op path passes the registry-checked
    /// `input` to its `OpenHandler` as the authoritative params).
    /// Un-compilable schemas are rejected at registration time — the
    /// same fail-closed rule as `publish_schema` (CF-003): registering
    /// an un-validatable input contract must never produce an operation
    /// that silently skips validation. A compiled validator is cached
    /// so the dispatch path does zero per-request work; `None` means
    /// "no schema declared" (the empty-object schema alkcall's own ops
    /// declare accepts anything, but it is still compiled so the
    /// dispatch path never re-parses it).
    fn validate_and_cache_input_schema(
        input_validators: &mut HashMap<String, Option<jsonschema::Validator>>,
        spec: &OperationSpec,
    ) -> Result<(), String> {
        let compiled = match jsonschema::options().build(&spec.input_schema) {
            Ok(v) => Some(v),
            Err(e) => {
                return Err(format!(
                    "input_schema for operation {:?} failed to compile: {e}",
                    spec.name
                ));
            }
        };
        input_validators.insert(spec.name.clone(), compiled);
        Ok(())
    }

    pub fn register(&self, registration: HandlerRegistration) -> Result<(), String> {
        let expected = match registration.spec.op_type {
            OperationType::Query | OperationType::Mutation => "Once",
            OperationType::Sub => "Stream",
            OperationType::Pub => "Sink",
        };
        let actual = match registration.handler {
            HandlerKind::Once(_) => "Once",
            HandlerKind::Stream(_) => "Stream",
            HandlerKind::Sink(_) => "Sink",
        };
        if expected != actual {
            return Err(format!(
                "handler kind mismatch: {:?} requires HandlerKind::{} (got HandlerKind::{})",
                registration.spec.op_type, expected, actual
            ));
        }
        Self::validate_and_cache_publish_schema(&mut self.validators.write(), &registration.spec)?;
        Self::validate_and_cache_input_schema(
            &mut self.input_validators.write(),
            &registration.spec,
        )?;
        self.operations
            .write()
            .insert(registration.spec.name.clone(), registration);
        Ok(())
    }

    pub fn registration(&self, name: &str) -> Option<HandlerRegistration> {
        self.operations.read().get(name).cloned()
    }

    /// The cached per-chunk validator for a Pub op's `publish_schema`
    /// (ADR-046 §4). `None` when the op declares no schema. Compiled at
    /// registration time — a schema that fails to compile never reaches
    /// this registry, so the dispatch path cannot fail open (CF-003).
    pub fn publish_validator(&self, name: &str) -> Option<jsonschema::Validator> {
        self.validators.read().get(name).cloned().flatten()
    }

    /// The compiled call-time validator for an op's `input_schema`
    /// (ADR-016: `INVALID_INPUT` is the registry's schema-mismatch
    /// error). Compiled at registration time (CF-003 fail-closed — an
    /// un-compilable schema is a registration error, never a silent
    /// skip). Every declared schema compiles to `Some`; the `Option`
    /// exists for map-miss on an unknown op name.
    pub(crate) fn input_validator(&self, name: &str) -> Option<jsonschema::Validator> {
        self.input_validators.read().get(name).cloned().flatten()
    }

    /// Validate `input` against the op's compiled `input_schema`
    /// (ADR-016). Returns the `INVALID_INPUT` envelope on violation.
    /// Called by all three dispatch entry points (`invoke`,
    /// `invoke_streaming`, `invoke_sink` via `resolve_sink_handler`)
    /// after the ACL check and before the handler — the registry-level
    /// gate every protocol crate's dispatch funnels through.
    fn check_input_schema(
        &self,
        registration: &HandlerRegistration,
        input: &Value,
        request_id: &str,
    ) -> Result<(), ResponseEnvelope> {
        let Some(validator) = self.input_validator(&registration.spec.name) else {
            return Ok(());
        };
        if validator.is_valid(input) {
            return Ok(());
        }
        Err(ResponseEnvelope::error(
            request_id.to_string(),
            CallError::invalid_input("input failed input_schema validation")
                .with_details(serde_json::json!({ "input": input })),
        ))
    }

    pub fn list_operations(&self) -> Vec<OperationSpec> {
        self.operations
            .read()
            .values()
            .filter(|r| r.spec.visibility == Visibility::External)
            .map(|r| r.spec.clone())
            .collect()
    }

    pub async fn invoke(
        &self,
        name: &str,
        input: Value,
        context: OperationContext,
    ) -> ResponseEnvelope {
        let request_id = context.request_id.clone();
        let registration = match self.registration(name) {
            Some(r) => r,
            None => return ResponseEnvelope::not_found(request_id, name),
        };

        if registration.spec.visibility == Visibility::Internal && !context.internal {
            return ResponseEnvelope::not_found(request_id, name);
        }

        let acl = &registration.spec.access_control;
        let identity = if context.internal {
            context
                .handler_identity
                .as_ref()
                .and_then(|ca| ca.as_identity())
        } else {
            context.identity.clone()
        };

        let resource_id = registration
            .spec
            .resource_id_path
            .as_ref()
            .and_then(|path| extract_json_pointer(&input, path));

        if let AccessResult::Forbidden(message) = acl.check(
            identity.as_ref(),
            resource_id.as_deref(),
            context.ownership.as_deref(),
        ) {
            return ResponseEnvelope::forbidden(request_id, message);
        }

        if let Err(envelope) = self.check_input_schema(&registration, &input, &request_id) {
            return envelope;
        }

        match &registration.handler {
            HandlerKind::Once(handler) => {
                let handler = Arc::clone(handler);
                (handler)(input, context).await
            }
            HandlerKind::Stream(_) => ResponseEnvelope::error(
                request_id,
                CallError::invalid_operation_type(
                    "invoke() called on a Sub op; use invoke_streaming()",
                ),
            ),
            HandlerKind::Sink(_) => ResponseEnvelope::error(
                request_id,
                CallError::invalid_operation_type("invoke() called on a Pub op; use invoke_sink()"),
            ),
        }
    }

    pub fn invoke_streaming(
        &self,
        name: &str,
        input: Value,
        context: OperationContext,
    ) -> ResponseStream {
        let request_id = context.request_id.clone();
        let name_owned = name.to_string();

        let registration = match self.registration(name) {
            Some(r) => r,
            None => {
                return Box::pin(stream::once(async move {
                    ResponseEnvelope::not_found(request_id, &name_owned)
                }));
            }
        };

        if registration.spec.visibility == Visibility::Internal && !context.internal {
            return Box::pin(stream::once(async move {
                ResponseEnvelope::not_found(request_id, &name_owned)
            }));
        }

        let acl = &registration.spec.access_control;
        let identity = if context.internal {
            context
                .handler_identity
                .as_ref()
                .and_then(|ca| ca.as_identity())
        } else {
            context.identity.clone()
        };

        let resource_id = registration
            .spec
            .resource_id_path
            .as_ref()
            .and_then(|path| extract_json_pointer(&input, path));

        if let AccessResult::Forbidden(message) = acl.check(
            identity.as_ref(),
            resource_id.as_deref(),
            context.ownership.as_deref(),
        ) {
            return Box::pin(stream::once(async move {
                ResponseEnvelope::forbidden(request_id, message)
            }));
        }

        if let Err(envelope) = self.check_input_schema(&registration, &input, &request_id) {
            return Box::pin(stream::once(async move { envelope }));
        }

        let streaming_handler = match &registration.handler {
            HandlerKind::Stream(h) => Arc::clone(h),
            HandlerKind::Once(_) => {
                return Box::pin(stream::once(async move {
                    ResponseEnvelope::error(
                        request_id,
                        CallError::invalid_operation_type(
                            "invoke_streaming() called on a Query/Mutation op; use invoke()",
                        ),
                    )
                }));
            }
            HandlerKind::Sink(_) => {
                return Box::pin(stream::once(async move {
                    ResponseEnvelope::error(
                        request_id,
                        CallError::invalid_operation_type(
                            "invoke_streaming() called on a Pub op; use invoke_sink()",
                        ),
                    )
                }));
            }
        };

        streaming_handler(input, context)
    }

    /// Resolve the `SinkHandler` for a `Pub` operation, performing the
    /// not-found / visibility / ACL / handler-kind checks that gate sink
    /// dispatch (ADR-046 §6). Returns the resolved `SinkHandler` on
    /// success, or a single error `ResponseEnvelope` on any pre-handler
    /// failure.
    ///
    /// This is the single source of truth for the sink dispatch checks —
    /// both `invoke_sink` (the registry-level dispatch) and the wire
    /// dispatch path (`Dispatcher::dispatch` Pub branch) call it, so the
    /// two paths cannot diverge (P-08: the `invoke_sink` path IS the
    /// wire path).
    #[allow(clippy::result_large_err)]
    pub(crate) fn resolve_sink_handler(
        &self,
        name: &str,
        input: &Value,
        context: &OperationContext,
    ) -> Result<SinkHandler, ResponseEnvelope> {
        let request_id = context.request_id.clone();
        let registration = match self.registration(name) {
            Some(r) => r,
            None => return Err(ResponseEnvelope::not_found(request_id, name)),
        };

        if registration.spec.visibility == Visibility::Internal && !context.internal {
            return Err(ResponseEnvelope::not_found(request_id, name));
        }

        let acl = &registration.spec.access_control;
        let identity = if context.internal {
            context
                .handler_identity
                .as_ref()
                .and_then(|ca| ca.as_identity())
        } else {
            context.identity.clone()
        };

        let resource_id = registration
            .spec
            .resource_id_path
            .as_ref()
            .and_then(|path| extract_json_pointer(input, path));

        if let AccessResult::Forbidden(message) = acl.check(
            identity.as_ref(),
            resource_id.as_deref(),
            context.ownership.as_deref(),
        ) {
            return Err(ResponseEnvelope::forbidden(request_id, message));
        }

        self.check_input_schema(&registration, input, &request_id)?;

        match &registration.handler {
            HandlerKind::Sink(h) => Ok(Arc::clone(h)),
            HandlerKind::Once(_) => Err(ResponseEnvelope::error(
                request_id,
                CallError::invalid_operation_type(
                    "invoke_sink() called on a Query/Mutation op; use invoke()",
                ),
            )),
            HandlerKind::Stream(_) => Err(ResponseEnvelope::error(
                request_id,
                CallError::invalid_operation_type(
                    "invoke_sink() called on a Sub op; use invoke_streaming()",
                ),
            )),
        }
    }

    /// Dispatch a `Pub` operation (ADR-046). The `publish_stream` is the
    /// initiator's data stream — each `call.published` chunk's `input` as
    /// one `Ok(Value)` item, or an initiator-side error as
    /// `Err(CallError)`. Pre-handler errors (not-found, forbidden,
    /// `INVALID_OPERATION_TYPE` for a non-Pub op) return a single error
    /// `ResponseEnvelope`.
    pub async fn invoke_sink(
        &self,
        name: &str,
        input: Value,
        publish_stream: PublishStream,
        context: OperationContext,
    ) -> ResponseEnvelope {
        match self.resolve_sink_handler(name, &input, &context) {
            Err(envelope) => envelope,
            Ok(sink_handler) => sink_handler(input, context, publish_stream).await,
        }
    }
}

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

pub struct OperationRegistryBuilder {
    operations: HashMap<String, HandlerRegistration>,
    validators: HashMap<String, Option<jsonschema::Validator>>,
    input_validators: HashMap<String, Option<jsonschema::Validator>>,
}

impl OperationRegistryBuilder {
    pub fn new() -> Self {
        Self {
            operations: HashMap::new(),
            validators: HashMap::new(),
            input_validators: HashMap::new(),
        }
    }

    /// Seed the builder from an existing registry's registrations — the
    /// fork surface expressed through the builder (review 004 F-02/F-03).
    /// Registration order is not preserved (HashMap iteration); use
    /// [`OperationRegistry::fork`] when order-independent deep copy is
    /// not a concern and the fork should carry validators as compiled.
    pub fn from_registry(registry: &OperationRegistry) -> Self {
        Self {
            operations: registry.operations.read().clone(),
            validators: registry.validators.read().clone(),
            input_validators: registry.input_validators.read().clone(),
        }
    }

    fn store(mut self, registration: HandlerRegistration) -> Result<Self, String> {
        OperationRegistry::validate_and_cache_publish_schema(
            &mut self.validators,
            &registration.spec,
        )?;
        OperationRegistry::validate_and_cache_input_schema(
            &mut self.input_validators,
            &registration.spec,
        )?;
        let name = registration.spec.name.clone();
        self.operations.insert(name, registration);
        Ok(self)
    }

    pub fn build(self) -> OperationRegistry {
        OperationRegistry {
            operations: RwLock::new(self.operations),
            validators: RwLock::new(self.validators),
            input_validators: RwLock::new(self.input_validators),
        }
    }

    fn wrap_once(spec: &OperationSpec, handler: Handler) -> Result<HandlerKind, String> {
        match spec.op_type {
            OperationType::Query | OperationType::Mutation => Ok(HandlerKind::Once(handler)),
            OperationType::Sub => Err(format!(
                "handler kind mismatch: {:?} requires HandlerKind::Stream (got Handler)",
                spec.op_type
            )),
            OperationType::Pub => Err(format!(
                "handler kind mismatch: {:?} requires HandlerKind::Sink (got Handler)",
                spec.op_type
            )),
        }
    }

    fn wrap_stream(spec: &OperationSpec, handler: StreamingHandler) -> Result<HandlerKind, String> {
        match spec.op_type {
            OperationType::Sub => Ok(HandlerKind::Stream(handler)),
            OperationType::Query | OperationType::Mutation => Err(format!(
                "handler kind mismatch: {:?} requires HandlerKind::Once (got StreamingHandler)",
                spec.op_type
            )),
            OperationType::Pub => Err(format!(
                "handler kind mismatch: {:?} requires HandlerKind::Sink (got StreamingHandler)",
                spec.op_type
            )),
        }
    }

    fn wrap_sink(spec: &OperationSpec, handler: SinkHandler) -> Result<HandlerKind, String> {
        match spec.op_type {
            OperationType::Pub => Ok(HandlerKind::Sink(handler)),
            OperationType::Query | OperationType::Mutation => Err(format!(
                "handler kind mismatch: {:?} requires HandlerKind::Once (got SinkHandler)",
                spec.op_type
            )),
            OperationType::Sub => Err(format!(
                "handler kind mismatch: {:?} requires HandlerKind::Stream (got SinkHandler)",
                spec.op_type
            )),
        }
    }

    pub fn with_local(
        self,
        spec: OperationSpec,
        handler: Handler,
        composition_authority: Option<CompositionAuthority>,
        scoped_env: Option<ScopedPeerEnv>,
        capabilities: Capabilities,
    ) -> Result<Self, String> {
        let kind = Self::wrap_once(&spec, handler)?;
        let registration = HandlerRegistration::new(
            spec,
            kind,
            OperationProvenance::Local,
            composition_authority,
            scoped_env,
            capabilities,
        );
        self.store(registration)
    }

    pub fn with_local_streaming(
        self,
        spec: OperationSpec,
        handler: StreamingHandler,
        composition_authority: Option<CompositionAuthority>,
        scoped_env: Option<ScopedPeerEnv>,
        capabilities: Capabilities,
    ) -> Result<Self, String> {
        let kind = Self::wrap_stream(&spec, handler)?;
        let registration = HandlerRegistration::new(
            spec,
            kind,
            OperationProvenance::Local,
            composition_authority,
            scoped_env,
            capabilities,
        );
        self.store(registration)
    }

    pub fn with_leaf(
        self,
        spec: OperationSpec,
        handler: Handler,
        capabilities: Capabilities,
    ) -> Result<Self, String> {
        self.with_leaf_provenance(
            spec,
            handler,
            OperationProvenance::FromOpenAPI,
            capabilities,
        )
    }

    pub fn with_leaf_provenance(
        self,
        spec: OperationSpec,
        handler: Handler,
        provenance: OperationProvenance,
        capabilities: Capabilities,
    ) -> Result<Self, String> {
        let kind = Self::wrap_once(&spec, handler)?;
        let registration =
            HandlerRegistration::new(spec, kind, provenance, None, None, capabilities);
        self.store(registration)
    }

    pub fn with_leaf_streaming(
        self,
        spec: OperationSpec,
        handler: StreamingHandler,
        capabilities: Capabilities,
    ) -> Result<Self, String> {
        self.with_leaf_streaming_provenance(
            spec,
            handler,
            OperationProvenance::FromOpenAPI,
            capabilities,
        )
    }

    pub fn with_leaf_streaming_provenance(
        self,
        spec: OperationSpec,
        handler: StreamingHandler,
        provenance: OperationProvenance,
        capabilities: Capabilities,
    ) -> Result<Self, String> {
        let kind = Self::wrap_stream(&spec, handler)?;
        let registration =
            HandlerRegistration::new(spec, kind, provenance, None, None, capabilities);
        self.store(registration)
    }

    pub fn with_local_sink(
        self,
        spec: OperationSpec,
        handler: SinkHandler,
        composition_authority: Option<CompositionAuthority>,
        scoped_env: Option<ScopedPeerEnv>,
        capabilities: Capabilities,
    ) -> Result<Self, String> {
        let kind = Self::wrap_sink(&spec, handler)?;
        let registration = HandlerRegistration::new(
            spec,
            kind,
            OperationProvenance::Local,
            composition_authority,
            scoped_env,
            capabilities,
        );
        self.store(registration)
    }

    pub fn with_leaf_sink(
        self,
        spec: OperationSpec,
        handler: SinkHandler,
        capabilities: Capabilities,
    ) -> Result<Self, String> {
        self.with_leaf_sink_provenance(
            spec,
            handler,
            OperationProvenance::FromOpenAPI,
            capabilities,
        )
    }

    pub fn with_leaf_sink_provenance(
        self,
        spec: OperationSpec,
        handler: SinkHandler,
        provenance: OperationProvenance,
        capabilities: Capabilities,
    ) -> Result<Self, String> {
        let kind = Self::wrap_sink(&spec, handler)?;
        let registration =
            HandlerRegistration::new(spec, kind, provenance, None, None, capabilities);
        self.store(registration)
    }

    pub fn with(self, registration: HandlerRegistration) -> Result<Self, String> {
        self.store(registration)
    }
}

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

pub fn make_handler<F, Fut>(f: F) -> Handler
where
    F: Fn(Value, OperationContext) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = ResponseEnvelope> + Send + 'static,
{
    Arc::new(move |input, context| Box::pin(f(input, context)))
}

pub fn make_streaming_handler<S, St>(f: S) -> StreamingHandler
where
    S: Fn(Value, OperationContext) -> St + Send + Sync + 'static,
    St: Stream<Item = ResponseEnvelope> + Send + 'static,
{
    Arc::new(move |input, context| Box::pin(f(input, context)))
}

pub fn make_sink_handler<F, Fut>(f: F) -> SinkHandler
where
    F: Fn(Value, OperationContext, PublishStream) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = ResponseEnvelope> + Send + 'static,
{
    Arc::new(move |input, context, stream| Box::pin(f(input, context, stream)))
}

/// Extract a string value from `input` at a JSON-pointer-ish path described
/// by `$.field` or `$.field/sub` (a leading `$` followed by a slash-separated
/// pointer, the same shape `serde_json::Value::pointer` expects after the
/// leading `/` is restored). Also tolerates the dotted form `$.a.b` for
/// shallow nested lookups (translated to `/a/b`).
///
/// Returns `None` if the path is malformed, the field is missing, or the
/// value at the path is not a string. Graceful — never panics (ADR-011 §2a):
/// a missing field simply yields `resource_id: None`, and the caller's
/// `AccessControl::check` decides whether to deny based on whether the spec
/// requires a specific resource ID.
pub(crate) fn extract_json_pointer(input: &Value, path: &str) -> Option<String> {
    let trimmed = path.strip_prefix('$')?;
    if trimmed.is_empty() {
        return None;
    }
    let pointer = if let Some(rest) = trimmed.strip_prefix('/') {
        format!("/{rest}")
    } else {
        let dotted = trimmed.replace('.', "/");
        if let Some(rest) = dotted.strip_prefix('/') {
            format!("/{rest}")
        } else {
            format!("/{dotted}")
        }
    };
    input
        .pointer(pointer.as_str())
        .and_then(|v| v.as_str())
        .map(|s| s.to_string())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::auth::Identity;
    use crate::core::ownership::OwnershipProvider;
    use crate::protocol::wire::CallError;
    use crate::registry::context::AbortPolicy;
    use crate::registry::env::OperationEnv;
    use crate::registry::spec::{AccessControl, OperationType};
    use std::collections::HashMap;
    use std::time::Duration;

    struct NoopEnv;

    #[async_trait::async_trait]
    impl OperationEnv for NoopEnv {
        async fn invoke_with_policy(
            &self,
            _namespace: &str,
            _operation: &str,
            _input: Value,
            _parent: &OperationContext,
            _policy: AbortPolicy,
        ) -> ResponseEnvelope {
            ResponseEnvelope::error("test", CallError::internal("noop env does not dispatch"))
        }

        fn contains(&self, _name: &str) -> bool {
            false
        }
    }

    fn root_context(
        request_id: &str,
        identity: Option<Identity>,
        handler_identity: Option<CompositionAuthority>,
        internal: bool,
        scoped_env: ScopedPeerEnv,
    ) -> OperationContext {
        OperationContext {
            request_id: request_id.to_string(),
            parent_request_id: None,
            identity,
            handler_identity,
            forwarded_for: None,
            capabilities: Capabilities::new(),
            metadata: HashMap::new(),
            scoped_env,
            env: Arc::new(NoopEnv),
            abort_policy: AbortPolicy::default(),
            deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
            internal,
            ownership: None,
        }
    }

    fn root_context_with_ownership(
        request_id: &str,
        identity: Option<Identity>,
        ownership: Option<Arc<dyn OwnershipProvider>>,
        scoped_env: ScopedPeerEnv,
    ) -> OperationContext {
        OperationContext {
            request_id: request_id.to_string(),
            parent_request_id: None,
            identity,
            handler_identity: None,
            forwarded_for: None,
            capabilities: Capabilities::new(),
            metadata: HashMap::new(),
            scoped_env,
            env: Arc::new(NoopEnv),
            abort_policy: AbortPolicy::default(),
            deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
            internal: false,
            ownership,
        }
    }

    fn echo_handler() -> Handler {
        make_handler(
            |input, context| async move { ResponseEnvelope::ok(context.request_id, input) },
        )
    }

    fn error_handler() -> Handler {
        make_handler(|_input, context| async move {
            ResponseEnvelope::error(context.request_id, CallError::internal("handler failure"))
        })
    }

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

    fn internal_spec(name: &str, acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Query,
            Visibility::Internal,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            acl,
            None,
        )
    }

    fn identity_with_scopes(id: &str, scopes: &[&str]) -> Identity {
        Identity {
            id: id.to_string(),
            scopes: scopes.iter().map(|s| s.to_string()).collect(),
            resources: HashMap::new(),
        }
    }

    #[tokio::test]
    async fn register_and_invoke_simple_operation() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("echo", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-1", None, None, false, ScopedPeerEnv::empty());
        let response = registry
            .invoke("echo", serde_json::json!({"hi": 1}), ctx)
            .await;
        assert_eq!(response.request_id, "req-1");
        assert_eq!(response.result, Ok(serde_json::json!({"hi": 1})));
    }

    #[tokio::test]
    async fn internal_op_from_external_call_returns_not_found() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                internal_spec("secret", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-2", None, None, false, ScopedPeerEnv::empty());
        let response = registry.invoke("secret", serde_json::json!({}), ctx).await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "NOT_FOUND");
                assert!(e.message.contains("secret"));
            }
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn internal_op_from_internal_call_invokes_handler() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                internal_spec("secret", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-3", None, None, true, ScopedPeerEnv::empty());
        let response = registry
            .invoke("secret", serde_json::json!({"x": 2}), ctx)
            .await;
        assert_eq!(response.request_id, "req-3");
        assert_eq!(response.result, Ok(serde_json::json!({"x": 2})));
    }

    #[tokio::test]
    async fn unknown_op_returns_not_found() {
        let registry = OperationRegistry::new();
        let ctx = root_context("req-4", None, None, false, ScopedPeerEnv::empty());
        let response = registry.invoke("missing", serde_json::json!({}), ctx).await;
        match response.result {
            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn acl_sufficient_scopes_allowed() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec(
                    "admin",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context(
            "req-5",
            Some(identity_with_scopes("caller", &["admin"])),
            None,
            false,
            ScopedPeerEnv::empty(),
        );
        let response = registry.invoke("admin", serde_json::json!({}), ctx).await;
        assert!(response.result.is_ok());
    }

    #[tokio::test]
    async fn acl_insufficient_scopes_forbidden() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec(
                    "admin",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context(
            "req-6",
            Some(identity_with_scopes("caller", &["user"])),
            None,
            false,
            ScopedPeerEnv::empty(),
        );
        let response = registry.invoke("admin", serde_json::json!({}), ctx).await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "FORBIDDEN");
                assert!(e.message.contains("admin"));
            }
            other => panic!("expected FORBIDDEN, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn acl_restricted_op_no_identity_forbidden() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec(
                    "admin",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-7", None, None, false, ScopedPeerEnv::empty());
        let response = registry.invoke("admin", serde_json::json!({}), ctx).await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "FORBIDDEN");
                assert_eq!(e.message, "authentication required");
            }
            other => panic!("expected FORBIDDEN, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn internal_call_acl_uses_handler_identity() {
        let registry = OperationRegistry::new();
        let composing_authority = CompositionAuthority::new("agent-chat", ["admin".to_string()]);
        registry
            .register(HandlerRegistration::new(
                internal_spec(
                    "secret",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context(
            "req-8",
            Some(identity_with_scopes("user", &["user"])),
            Some(composing_authority),
            true,
            ScopedPeerEnv::empty(),
        );
        let response = registry.invoke("secret", serde_json::json!({}), ctx).await;
        assert!(
            response.result.is_ok(),
            "internal call should use handler_identity (admin), not caller (user)"
        );
    }

    #[tokio::test]
    async fn internal_call_acl_insufficient_handler_identity_forbidden() {
        let registry = OperationRegistry::new();
        let weak_authority = CompositionAuthority::new("weak", ["user".to_string()]);
        registry
            .register(HandlerRegistration::new(
                internal_spec(
                    "secret",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context(
            "req-9",
            Some(identity_with_scopes("user", &["admin"])),
            Some(weak_authority),
            true,
            ScopedPeerEnv::empty(),
        );
        let response = registry.invoke("secret", serde_json::json!({}), ctx).await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "FORBIDDEN");
                assert!(e.message.contains("admin"));
            }
            other => panic!("expected FORBIDDEN, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn external_call_acl_uses_caller_identity_not_handler_identity() {
        let registry = OperationRegistry::new();
        let handler_authority = CompositionAuthority::new("agent", ["admin".to_string()]);
        registry
            .register(HandlerRegistration::new(
                external_spec(
                    "gate",
                    AccessControl {
                        required_scopes: vec!["admin".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                Some(handler_authority),
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context(
            "req-10",
            Some(identity_with_scopes("user", &["user"])),
            Some(CompositionAuthority::new("agent", ["admin".to_string()])),
            false,
            ScopedPeerEnv::empty(),
        );
        let response = registry.invoke("gate", serde_json::json!({}), ctx).await;
        match response.result {
            Err(e) => assert_eq!(e.code, "FORBIDDEN"),
            other => panic!("expected FORBIDDEN, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn list_operations_returns_external_only() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("echo", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        registry
            .register(HandlerRegistration::new(
                internal_spec("secret", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ops = registry.list_operations();
        assert_eq!(ops.len(), 1);
        assert_eq!(ops[0].name, "echo");
    }

    #[tokio::test]
    async fn handler_returned_error_passes_through() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("boom", AccessControl::default()),
                HandlerKind::Once(error_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-11", None, None, false, ScopedPeerEnv::empty());
        let response = registry.invoke("boom", serde_json::json!({}), ctx).await;
        match response.result {
            Err(e) => assert_eq!(e.code, "INTERNAL"),
            other => panic!("expected INTERNAL error, got {other:?}"),
        }
    }

    #[test]
    fn builder_with_local_sets_provenance_local() {
        let registry = OperationRegistryBuilder::new()
            .with_local(
                external_spec("echo", AccessControl::default()),
                echo_handler(),
                CompositionAuthority::none(),
                ScopedPeerEnv::empty().into(),
                Capabilities::new(),
            )
            .unwrap()
            .build();
        let reg = registry.registration("echo").expect("registered");
        assert_eq!(reg.provenance, OperationProvenance::Local);
        assert!(reg.composition_authority.is_none());
        assert!(reg.scoped_env.is_some());
    }

    #[test]
    fn builder_with_local_carries_authority_and_scoped_env() {
        let registry = OperationRegistryBuilder::new()
            .with_local(
                external_spec("agent", AccessControl::default()),
                echo_handler(),
                Some(CompositionAuthority::new("agent", ["fs:read".to_string()])),
                Some(ScopedPeerEnv::new(["fs/readFile"])),
                Capabilities::new(),
            )
            .unwrap()
            .build();
        let reg = registry.registration("agent").expect("registered");
        assert_eq!(reg.provenance, OperationProvenance::Local);
        let authority = reg.composition_authority.as_ref().expect("authority set");
        assert_eq!(authority.label, "agent");
        assert_eq!(authority.scopes, vec!["fs:read".to_string()]);
        assert!(reg.scoped_env.is_some());
        assert!(reg.scoped_env.as_ref().unwrap().allows("fs/readFile"));
    }

    #[test]
    fn builder_with_leaf_sets_provenance_and_no_authority() {
        let registry = OperationRegistryBuilder::new()
            .with_leaf(
                external_spec("vastai", AccessControl::default()),
                echo_handler(),
                Capabilities::new(),
            )
            .unwrap()
            .build();
        let reg = registry.registration("vastai").expect("registered");
        assert_eq!(reg.provenance, OperationProvenance::FromOpenAPI);
        assert!(reg.composition_authority.is_none());
        assert!(reg.scoped_env.is_none());
    }

    #[test]
    fn builder_with_leaf_provenance_overrides_provenance() {
        let registry = OperationRegistryBuilder::new()
            .with_leaf_provenance(
                external_spec("remote", AccessControl::default()),
                echo_handler(),
                OperationProvenance::FromCall,
                Capabilities::new(),
            )
            .unwrap()
            .build();
        let reg = registry.registration("remote").expect("registered");
        assert_eq!(reg.provenance, OperationProvenance::FromCall);
        assert!(reg.composition_authority.is_none());
        assert!(reg.scoped_env.is_none());
    }

    #[test]
    fn builder_with_takes_full_bundle() {
        let registration = HandlerRegistration::new(
            external_spec("agent", AccessControl::default()),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Session,
            Some(CompositionAuthority::new("sandbox", [])),
            Some(ScopedPeerEnv::new(["fs/readFile"])),
            Capabilities::new(),
        );
        let registry = OperationRegistryBuilder::new()
            .with(registration)
            .unwrap()
            .build();
        let reg = registry.registration("agent").expect("registered");
        assert_eq!(reg.provenance, OperationProvenance::Session);
        assert!(reg.composition_authority.is_some());
        assert!(reg.scoped_env.is_some());
    }

    #[test]
    fn builder_default_is_new() {
        let builder = OperationRegistryBuilder::default();
        let registry = builder.build();
        assert!(registry.list_operations().is_empty());
    }

    #[test]
    fn registry_default_is_new() {
        let registry = OperationRegistry::default();
        assert!(registry.list_operations().is_empty());
        assert!(registry.registration("anything").is_none());
    }

    #[test]
    fn registration_lookup_returns_bundle_fields() {
        let registry = OperationRegistry::new();
        let authority = CompositionAuthority::new("agent", ["fs:read".to_string()]);
        let scoped = ScopedPeerEnv::new(["fs/readFile"]);
        let caps = Capabilities::new().with_api_key("google", "k".to_string());
        registry
            .register(HandlerRegistration::new(
                external_spec("agent", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                Some(authority.clone()),
                Some(scoped.clone()),
                caps.clone(),
            ))
            .unwrap();
        let reg = registry.registration("agent").expect("found");
        assert_eq!(reg.spec.name, "agent");
        assert_eq!(reg.provenance, OperationProvenance::Local);
        assert_eq!(reg.composition_authority.as_ref().unwrap().label, "agent");
        assert!(reg.scoped_env.as_ref().unwrap().allows("fs/readFile"));
    }

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

    fn echo_streaming_handler() -> StreamingHandler {
        make_streaming_handler(|input, context| {
            futures::stream::iter(vec![ResponseEnvelope::ok(context.request_id, input)])
        })
    }

    #[tokio::test]
    async fn invoke_on_stream_kind_returns_invalid_operation_type() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                subscription_spec("events/stream"),
                HandlerKind::Stream(echo_streaming_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-iot", None, None, false, ScopedPeerEnv::empty());
        let response = registry
            .invoke("events/stream", serde_json::json!({}), ctx)
            .await;
        match response.result {
            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_on_once_kind_dispatches_normally() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("echo", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-once", None, None, false, ScopedPeerEnv::empty());
        let response = registry
            .invoke("echo", serde_json::json!({"hi": 1}), ctx)
            .await;
        assert_eq!(response.result, Ok(serde_json::json!({"hi": 1})));
    }

    #[test]
    fn register_rejects_once_for_subscription_spec() {
        let registry = OperationRegistry::new();
        let result = registry.register(HandlerRegistration::new(
            subscription_spec("events/stream"),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ));
        match result {
            Err(msg) => assert!(
                msg.contains("Sub")
                    && msg.contains("HandlerKind::Stream")
                    && msg.contains("HandlerKind::Once"),
                "unexpected message: {msg}"
            ),
            other => panic!("expected Err, got {other:?}"),
        }
    }

    #[test]
    fn register_rejects_stream_for_query_spec() {
        let registry = OperationRegistry::new();
        let result = registry.register(HandlerRegistration::new(
            external_spec("echo", AccessControl::default()),
            HandlerKind::Stream(echo_streaming_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ));
        match result {
            Err(msg) => assert!(
                (msg.contains("Query") || msg.contains("Mutation"))
                    && msg.contains("HandlerKind::Once")
                    && msg.contains("HandlerKind::Stream"),
                "unexpected message: {msg}"
            ),
            other => panic!("expected Err, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn make_streaming_handler_produces_working_stream() {
        use futures::stream::StreamExt;
        let handler = echo_streaming_handler();
        let ctx = root_context("req-st", None, None, false, ScopedPeerEnv::empty());
        let mut stream = handler(serde_json::json!({"v": 1}), ctx);
        let first = stream.next().await.expect("one envelope");
        assert_eq!(first.result, Ok(serde_json::json!({"v": 1})));
        let second = stream.next().await;
        assert!(second.is_none(), "stream ends after one value");
    }

    #[test]
    fn call_error_invalid_operation_type_is_not_retryable() {
        let err = CallError::invalid_operation_type("bad path");
        assert_eq!(err.code, "INVALID_OPERATION_TYPE");
        assert!(!err.retryable);
        assert!(err.details.is_none());
    }

    async fn collect_stream(mut s: ResponseStream) -> Vec<ResponseEnvelope> {
        use futures::stream::StreamExt;
        let mut out = Vec::new();
        while let Some(env) = s.next().await {
            out.push(env);
        }
        out
    }

    #[tokio::test]
    async fn invoke_streaming_on_subscription_dispatches_handler_stream() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                subscription_spec("events/stream"),
                HandlerKind::Stream(echo_streaming_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-is-1", None, None, false, ScopedPeerEnv::empty());
        let stream = registry.invoke_streaming("events/stream", serde_json::json!({"v": 7}), ctx);
        let items = collect_stream(stream).await;
        assert_eq!(items.len(), 1);
        assert_eq!(items[0].request_id, "req-is-1");
        assert_eq!(items[0].result, Ok(serde_json::json!({"v": 7})));
    }

    #[tokio::test]
    async fn invoke_streaming_on_unknown_op_yields_single_not_found() {
        let registry = OperationRegistry::new();
        let ctx = root_context("req-is-2", None, None, false, ScopedPeerEnv::empty());
        let stream = registry.invoke_streaming("missing", serde_json::json!({}), ctx);
        let items = collect_stream(stream).await;
        assert_eq!(items.len(), 1);
        match &items[0].result {
            Err(e) => {
                assert_eq!(e.code, "NOT_FOUND");
                assert!(e.message.contains("missing"));
            }
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_streaming_on_query_op_yields_invalid_operation_type() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("echo", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-is-3", None, None, false, ScopedPeerEnv::empty());
        let stream = registry.invoke_streaming("echo", serde_json::json!({}), ctx);
        let items = collect_stream(stream).await;
        assert_eq!(items.len(), 1);
        match &items[0].result {
            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_streaming_internal_op_from_external_yields_not_found() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                internal_subscription_spec(AccessControl::default()),
                HandlerKind::Stream(echo_streaming_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-is-4", None, None, false, ScopedPeerEnv::empty());
        let stream = registry.invoke_streaming("events/stream", serde_json::json!({}), ctx);
        let items = collect_stream(stream).await;
        assert_eq!(items.len(), 1);
        match &items[0].result {
            Err(e) => {
                assert_eq!(e.code, "NOT_FOUND");
                assert!(e.message.contains("events/stream"));
            }
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_streaming_acl_denied_yields_forbidden() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                subscription_spec_with_acl(AccessControl {
                    required_scopes: vec!["admin".to_string()],
                    ..Default::default()
                }),
                HandlerKind::Stream(echo_streaming_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context(
            "req-is-5",
            Some(identity_with_scopes("caller", &["user"])),
            None,
            false,
            ScopedPeerEnv::empty(),
        );
        let stream = registry.invoke_streaming("events/stream", serde_json::json!({}), ctx);
        let items = collect_stream(stream).await;
        assert_eq!(items.len(), 1);
        match &items[0].result {
            Err(e) => {
                assert_eq!(e.code, "FORBIDDEN");
                assert!(e.message.contains("admin"));
            }
            other => panic!("expected FORBIDDEN, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_streaming_internal_call_uses_handler_identity_for_acl() {
        let registry = OperationRegistry::new();
        let composing_authority = CompositionAuthority::new("agent-chat", ["admin".to_string()]);
        registry
            .register(HandlerRegistration::new(
                internal_subscription_spec(AccessControl {
                    required_scopes: vec!["admin".to_string()],
                    ..Default::default()
                }),
                HandlerKind::Stream(echo_streaming_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context(
            "req-is-6",
            Some(identity_with_scopes("user", &["user"])),
            Some(composing_authority),
            true,
            ScopedPeerEnv::empty(),
        );
        let stream = registry.invoke_streaming("events/stream", serde_json::json!({"ok": 1}), ctx);
        let items = collect_stream(stream).await;
        assert_eq!(items.len(), 1);
        assert_eq!(items[0].request_id, "req-is-6");
        assert_eq!(items[0].result, Ok(serde_json::json!({"ok": 1})));
    }

    fn subscription_spec_with_acl(acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            "events/stream",
            OperationType::Sub,
            Visibility::External,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            acl,
            None,
        )
    }

    fn internal_subscription_spec(acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            "events/stream",
            OperationType::Sub,
            Visibility::Internal,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            acl,
            None,
        )
    }

    struct MockOwnership {
        owned: Vec<(String, String)>,
    }

    impl OwnershipProvider for MockOwnership {
        fn owns(
            &self,
            _identity: &Identity,
            resource_type: &str,
            resource_id: &str,
            _action: &str,
        ) -> bool {
            self.owned
                .iter()
                .any(|(rt, rid)| rt == resource_type && rid == resource_id)
        }

        fn owned_resources(&self, _identity: &Identity, resource_type: &str) -> Vec<String> {
            self.owned
                .iter()
                .filter(|(rt, _)| rt == resource_type)
                .map(|(_, rid)| rid.clone())
                .collect()
        }

        fn owns_any(&self, _identity: &Identity, resource_type: &str) -> bool {
            self.owned.iter().any(|(rt, _)| rt == resource_type)
        }
    }

    fn empty_identity(id: &str) -> Identity {
        Identity {
            id: id.to_string(),
            scopes: vec![],
            resources: HashMap::new(),
        }
    }

    fn resource_spec(
        name: &str,
        acl: AccessControl,
        resource_id_path: Option<&str>,
    ) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Query,
            Visibility::External,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            acl,
            resource_id_path.map(|s| s.to_string()),
        )
    }

    #[test]
    fn extract_json_pointer_resolves_single_field() {
        let input = serde_json::json!({"containerId": "abc123"});
        assert_eq!(
            extract_json_pointer(&input, "$.containerId"),
            Some("abc123".to_string())
        );
    }

    #[test]
    fn extract_json_pointer_missing_field_returns_none() {
        let input = serde_json::json!({"other": 1});
        assert_eq!(extract_json_pointer(&input, "$.containerId"), None);
    }

    #[test]
    fn extract_json_pointer_non_string_value_returns_none() {
        let input = serde_json::json!({"containerId": 42});
        assert_eq!(extract_json_pointer(&input, "$.containerId"), None);
    }

    #[test]
    fn extract_json_pointer_no_leading_dollar_returns_none() {
        let input = serde_json::json!({"containerId": "abc"});
        assert_eq!(extract_json_pointer(&input, "containerId"), None);
    }

    #[test]
    fn extract_json_pointer_empty_after_dollar_returns_none() {
        let input = serde_json::json!({"a": "b"});
        assert_eq!(extract_json_pointer(&input, "$"), None);
    }

    #[test]
    fn extract_json_pointer_nested_slash_path() {
        let input = serde_json::json!({"data": {"id": "xyz"}});
        assert_eq!(
            extract_json_pointer(&input, "$.data/id"),
            Some("xyz".to_string())
        );
    }

    #[test]
    fn extract_json_pointer_nested_dotted_path() {
        let input = serde_json::json!({"data": {"id": "xyz"}});
        assert_eq!(
            extract_json_pointer(&input, "$.data.id"),
            Some("xyz".to_string())
        );
    }

    #[tokio::test]
    async fn invoke_with_ownership_provider_allows_owned_resource() {
        let registry = OperationRegistry::new();
        let acl = AccessControl {
            resource_type: Some("container".to_string()),
            resource_action: Some("exec".to_string()),
            ..Default::default()
        };
        registry
            .register(HandlerRegistration::new(
                resource_spec("container/exec", acl, Some("$.containerId")),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let provider: Arc<dyn OwnershipProvider> = Arc::new(MockOwnership {
            owned: vec![("container".to_string(), "c1".to_string())],
        });
        let ctx = root_context_with_ownership(
            "req-owns",
            Some(empty_identity("alice")),
            Some(provider),
            ScopedPeerEnv::empty(),
        );
        let response = registry
            .invoke(
                "container/exec",
                serde_json::json!({"containerId": "c1"}),
                ctx,
            )
            .await;
        assert!(response.result.is_ok(), "owned resource should be allowed");
    }

    #[tokio::test]
    async fn invoke_with_ownership_provider_forbids_unowned_resource() {
        let registry = OperationRegistry::new();
        let acl = AccessControl {
            resource_type: Some("container".to_string()),
            resource_action: Some("exec".to_string()),
            ..Default::default()
        };
        registry
            .register(HandlerRegistration::new(
                resource_spec("container/exec", acl, Some("$.containerId")),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let provider: Arc<dyn OwnershipProvider> = Arc::new(MockOwnership {
            owned: vec![("container".to_string(), "c1".to_string())],
        });
        let ctx = root_context_with_ownership(
            "req-not-owns",
            Some(empty_identity("alice")),
            Some(provider),
            ScopedPeerEnv::empty(),
        );
        let response = registry
            .invoke(
                "container/exec",
                serde_json::json!({"containerId": "c2"}),
                ctx,
            )
            .await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "FORBIDDEN");
                assert!(e.message.contains("container/c2"));
            }
            other => panic!("expected FORBIDDEN, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_with_ownership_provider_missing_field_falls_back_to_owns_any() {
        let registry = OperationRegistry::new();
        let acl = AccessControl {
            resource_type: Some("container".to_string()),
            resource_action: Some("exec".to_string()),
            ..Default::default()
        };
        registry
            .register(HandlerRegistration::new(
                resource_spec("container/exec", acl, Some("$.containerId")),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let provider: Arc<dyn OwnershipProvider> = Arc::new(MockOwnership {
            owned: vec![("container".to_string(), "c1".to_string())],
        });
        let ctx = root_context_with_ownership(
            "req-missing",
            Some(empty_identity("alice")),
            Some(provider),
            ScopedPeerEnv::empty(),
        );
        let response = registry
            .invoke("container/exec", serde_json::json!({}), ctx)
            .await;
        assert!(
            response.result.is_ok(),
            "missing field → resource_id None → owns_any path allowed"
        );
    }

    #[tokio::test]
    async fn invoke_with_ownership_provider_missing_field_forbids_when_not_owns_any() {
        let registry = OperationRegistry::new();
        let acl = AccessControl {
            resource_type: Some("container".to_string()),
            resource_action: Some("exec".to_string()),
            ..Default::default()
        };
        registry
            .register(HandlerRegistration::new(
                resource_spec("container/exec", acl, Some("$.containerId")),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let provider: Arc<dyn OwnershipProvider> = Arc::new(MockOwnership {
            owned: vec![("volume".to_string(), "v1".to_string())],
        });
        let ctx = root_context_with_ownership(
            "req-missing-denied",
            Some(empty_identity("alice")),
            Some(provider),
            ScopedPeerEnv::empty(),
        );
        let response = registry
            .invoke("container/exec", serde_json::json!({}), ctx)
            .await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "FORBIDDEN");
                assert!(e.message.contains("container"));
            }
            other => panic!("expected FORBIDDEN, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_without_ownership_provider_falls_back_to_static_resources() {
        let registry = OperationRegistry::new();
        let acl = AccessControl {
            resource_type: Some("container".to_string()),
            resource_action: Some("exec".to_string()),
            ..Default::default()
        };
        registry
            .register(HandlerRegistration::new(
                resource_spec("container/exec", acl, Some("$.containerId")),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let mut resources = HashMap::new();
        resources.insert("container".to_string(), vec!["exec".to_string()]);
        let identity = Identity {
            id: "alice".to_string(),
            scopes: vec![],
            resources,
        };
        let ctx =
            root_context_with_ownership("req-static", Some(identity), None, ScopedPeerEnv::empty());
        let response = registry
            .invoke(
                "container/exec",
                serde_json::json!({"containerId": "c1"}),
                ctx,
            )
            .await;
        assert!(
            response.result.is_ok(),
            "no provider wired → static Identity.resources fallback allows"
        );
    }

    // --- ADR-046: Pub/SinkHandler/invoke_sink tests ------------------------

    fn pub_spec(name: &str, acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Pub,
            Visibility::External,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            acl,
            None,
        )
    }

    // --- CF-003: publish_schema validated at registration time -------------

    fn make_sink_noop() -> SinkHandler {
        make_sink_handler(|_input, ctx, _stream| async move {
            ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
        })
    }

    fn registering_pub_with_schema(schema: Value) -> Result<(), String> {
        let registry = OperationRegistry::new();
        registry.register(HandlerRegistration::new(
            pub_spec("fs/upload", AccessControl::default()).with_publish_schema(schema),
            HandlerKind::Sink(make_sink_noop()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ))
    }

    #[test]
    fn register_rejects_uncompilable_publish_schema() {
        // `required` must be an array — jsonschema rejects this at compile.
        let err = registering_pub_with_schema(serde_json::json!({
            "type": "object",
            "required": "not-an-array"
        }))
        .expect_err("un-compilable schema must be rejected at registration");
        assert!(
            err.contains("failed to compile"),
            "error names the failure: {err}"
        );
        assert!(err.contains("fs/upload"), "error names the op: {err}");
    }

    #[test]
    fn register_rejects_invalid_type_and_pattern_schema() {
        assert!(registering_pub_with_schema(serde_json::json!({
            "type": 123
        }))
        .is_err());

        assert!(registering_pub_with_schema(serde_json::json!({
            "type": "object",
            "properties": { "x": { "pattern": "(unclosed" } }
        }))
        .is_err());
    }

    #[test]
    fn register_accepts_compilable_publish_schema_and_caches_validator() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                pub_spec("fs/upload", AccessControl::default()).with_publish_schema(
                    serde_json::json!({
                        "type": "object",
                        "properties": { "bytes": { "type": "string" } },
                        "required": ["bytes"]
                    }),
                ),
                HandlerKind::Sink(make_sink_noop()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .expect("compilable schema registers");
        assert!(
            registry.publish_validator("fs/upload").is_some(),
            "validator cached at registration"
        );
    }

    #[test]
    fn publish_validator_none_when_no_schema_declared() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                pub_spec("fs/upload", AccessControl::default()),
                HandlerKind::Sink(make_sink_noop()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .expect("no-schema pub registers");
        assert!(registry.publish_validator("fs/upload").is_none());
        assert!(registry.publish_validator("no/such").is_none());
    }

    #[test]
    fn registry_builder_rejects_uncompilable_publish_schema() {
        let result = OperationRegistryBuilder::new().with_local_sink(
            pub_spec("fs/upload", AccessControl::default())
                .with_publish_schema(serde_json::json!({ "type": "object", "required": "bad" })),
            make_sink_noop(),
            None,
            None,
            Capabilities::new(),
        );
        assert!(result.is_err(), "builder path validates too");
    }

    #[test]
    fn failed_validation_leaves_no_validator_cache_entry() {
        let registry = OperationRegistry::new();
        let result = registry.register(HandlerRegistration::new(
            pub_spec("fs/upload", AccessControl::default())
                .with_publish_schema(serde_json::json!({ "type": "object", "required": "bad" })),
            HandlerKind::Sink(make_sink_noop()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ));
        assert!(result.is_err());
        assert!(
            registry.registration("fs/upload").is_none(),
            "op not registered on schema failure"
        );
    }

    fn internal_pub_spec(acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            "fs/upload",
            OperationType::Pub,
            Visibility::Internal,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            acl,
            None,
        )
    }

    fn collect_sink_handler() -> SinkHandler {
        use futures::stream::StreamExt;
        make_sink_handler(|_input, ctx, mut stream| async move {
            let mut count = 0u32;
            let mut last = Value::Null;
            while let Some(item) = stream.next().await {
                match item {
                    Ok(v) => {
                        count += 1;
                        last = v;
                    }
                    Err(e) => {
                        return ResponseEnvelope::error(
                            ctx.request_id,
                            CallError::internal(format!("publish error: {e:?}")),
                        );
                    }
                }
            }
            ResponseEnvelope::ok(
                ctx.request_id,
                serde_json::json!({ "count": count, "last": last }),
            )
        })
    }

    fn make_publish_stream(items: Vec<Value>) -> PublishStream {
        Box::pin(futures::stream::iter(
            items.into_iter().map(Ok::<_, CallError>),
        ))
    }

    #[tokio::test]
    async fn invoke_sink_pub_op_collects_stream_and_returns_result() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                pub_spec("fs/upload", AccessControl::default()),
                HandlerKind::Sink(collect_sink_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-pub-1", None, None, false, ScopedPeerEnv::empty());
        let stream = make_publish_stream(vec![
            serde_json::json!({"chunk": 1}),
            serde_json::json!({"chunk": 2}),
            serde_json::json!({"chunk": 3}),
        ]);
        let response = registry
            .invoke_sink("fs/upload", serde_json::json!({"path": "/x"}), stream, ctx)
            .await;
        let out = response.result.expect("ok");
        assert_eq!(out["count"], serde_json::json!(3));
        assert_eq!(out["last"], serde_json::json!({"chunk": 3}));
    }

    #[tokio::test]
    async fn invoke_sink_empty_stream_returns_zero_count() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                pub_spec("fs/upload", AccessControl::default()),
                HandlerKind::Sink(collect_sink_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-pub-2", None, None, false, ScopedPeerEnv::empty());
        let stream = make_publish_stream(vec![]);
        let response = registry
            .invoke_sink("fs/upload", serde_json::json!({}), stream, ctx)
            .await;
        let out = response.result.expect("ok");
        assert_eq!(out["count"], serde_json::json!(0));
    }

    #[tokio::test]
    async fn invoke_sink_unknown_op_returns_not_found() {
        let registry = OperationRegistry::new();
        let ctx = root_context("req-pub-3", None, None, false, ScopedPeerEnv::empty());
        let stream = make_publish_stream(vec![]);
        let response = registry
            .invoke_sink("missing", serde_json::json!({}), stream, ctx)
            .await;
        match response.result {
            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_sink_internal_op_from_external_returns_not_found() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                internal_pub_spec(AccessControl::default()),
                HandlerKind::Sink(collect_sink_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-pub-4", None, None, false, ScopedPeerEnv::empty());
        let stream = make_publish_stream(vec![]);
        let response = registry
            .invoke_sink("fs/upload", serde_json::json!({}), stream, ctx)
            .await;
        match response.result {
            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
            other => panic!("expected NOT_FOUND, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_sink_acl_denied_returns_forbidden() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                pub_spec(
                    "fs/upload",
                    AccessControl {
                        required_scopes: vec!["fs:write".to_string()],
                        ..Default::default()
                    },
                ),
                HandlerKind::Sink(collect_sink_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context(
            "req-pub-5",
            Some(identity_with_scopes("user", &["read"])),
            None,
            false,
            ScopedPeerEnv::empty(),
        );
        let stream = make_publish_stream(vec![]);
        let response = registry
            .invoke_sink("fs/upload", serde_json::json!({}), stream, ctx)
            .await;
        match response.result {
            Err(e) => {
                assert_eq!(e.code, "FORBIDDEN");
                assert!(e.message.contains("fs:write"));
            }
            other => panic!("expected FORBIDDEN, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_on_pub_op_returns_invalid_operation_type() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                pub_spec("fs/upload", AccessControl::default()),
                HandlerKind::Sink(collect_sink_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-pub-6", None, None, false, ScopedPeerEnv::empty());
        let response = registry
            .invoke("fs/upload", serde_json::json!({}), ctx)
            .await;
        match response.result {
            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_streaming_on_pub_op_returns_invalid_operation_type() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                pub_spec("fs/upload", AccessControl::default()),
                HandlerKind::Sink(collect_sink_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-pub-7", None, None, false, ScopedPeerEnv::empty());
        let stream = registry.invoke_streaming("fs/upload", serde_json::json!({}), ctx);
        let items = collect_stream(stream).await;
        assert_eq!(items.len(), 1);
        match &items[0].result {
            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_sink_on_query_op_returns_invalid_operation_type() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("echo", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-pub-8", None, None, false, ScopedPeerEnv::empty());
        let stream = make_publish_stream(vec![]);
        let response = registry
            .invoke_sink("echo", serde_json::json!({}), stream, ctx)
            .await;
        match response.result {
            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_sink_on_sub_op_returns_invalid_operation_type() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                subscription_spec("events/stream"),
                HandlerKind::Stream(echo_streaming_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context("req-pub-9", None, None, false, ScopedPeerEnv::empty());
        let stream = make_publish_stream(vec![]);
        let response = registry
            .invoke_sink("events/stream", serde_json::json!({}), stream, ctx)
            .await;
        match response.result {
            Err(e) => assert_eq!(e.code, "INVALID_OPERATION_TYPE"),
            other => panic!("expected INVALID_OPERATION_TYPE, got {other:?}"),
        }
    }

    #[test]
    fn register_rejects_once_for_pub_spec() {
        let registry = OperationRegistry::new();
        let result = registry.register(HandlerRegistration::new(
            pub_spec("fs/upload", AccessControl::default()),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ));
        match result {
            Err(msg) => assert!(
                msg.contains("Pub")
                    && msg.contains("HandlerKind::Sink")
                    && msg.contains("HandlerKind::Once"),
                "unexpected message: {msg}"
            ),
            other => panic!("expected Err, got {other:?}"),
        }
    }

    #[test]
    fn register_rejects_stream_for_pub_spec() {
        let registry = OperationRegistry::new();
        let result = registry.register(HandlerRegistration::new(
            pub_spec("fs/upload", AccessControl::default()),
            HandlerKind::Stream(echo_streaming_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ));
        match result {
            Err(msg) => assert!(
                msg.contains("Pub")
                    && msg.contains("HandlerKind::Sink")
                    && msg.contains("HandlerKind::Stream"),
                "unexpected message: {msg}"
            ),
            other => panic!("expected Err, got {other:?}"),
        }
    }

    #[test]
    fn register_rejects_sink_for_query_spec() {
        let registry = OperationRegistry::new();
        let result = registry.register(HandlerRegistration::new(
            external_spec("echo", AccessControl::default()),
            HandlerKind::Sink(collect_sink_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ));
        match result {
            Err(msg) => assert!(
                msg.contains("Query") || msg.contains("Mutation"),
                "unexpected message: {msg}"
            ),
            other => panic!("expected Err, got {other:?}"),
        }
    }

    #[test]
    fn register_rejects_sink_for_sub_spec() {
        let registry = OperationRegistry::new();
        let result = registry.register(HandlerRegistration::new(
            subscription_spec("events/stream"),
            HandlerKind::Sink(collect_sink_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ));
        match result {
            Err(msg) => assert!(
                msg.contains("Sub")
                    && msg.contains("HandlerKind::Stream")
                    && msg.contains("HandlerKind::Sink"),
                "unexpected message: {msg}"
            ),
            other => panic!("expected Err, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn invoke_sink_internal_call_uses_handler_identity_for_acl() {
        let registry = OperationRegistry::new();
        let composing_authority = CompositionAuthority::new("fs-writer", ["fs:write".to_string()]);
        registry
            .register(HandlerRegistration::new(
                internal_pub_spec(AccessControl {
                    required_scopes: vec!["fs:write".to_string()],
                    ..Default::default()
                }),
                HandlerKind::Sink(collect_sink_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let ctx = root_context(
            "req-pub-10",
            Some(identity_with_scopes("user", &["read"])),
            Some(composing_authority),
            true,
            ScopedPeerEnv::empty(),
        );
        let stream = make_publish_stream(vec![serde_json::json!({"chunk": 1})]);
        let response = registry
            .invoke_sink("fs/upload", serde_json::json!({}), stream, ctx)
            .await;
        let out = response.result.expect("ok — handler identity has fs:write");
        assert_eq!(out["count"], serde_json::json!(1));
    }

    #[test]
    fn builder_with_local_sink_sets_provenance_local() {
        let registry = OperationRegistryBuilder::new()
            .with_local_sink(
                pub_spec("fs/upload", AccessControl::default()),
                collect_sink_handler(),
                CompositionAuthority::none(),
                ScopedPeerEnv::empty().into(),
                Capabilities::new(),
            )
            .unwrap()
            .build();
        let reg = registry.registration("fs/upload").expect("registered");
        assert_eq!(reg.provenance, OperationProvenance::Local);
        assert!(matches!(reg.handler, HandlerKind::Sink(_)));
        assert!(reg.composition_authority.is_none());
        assert!(reg.scoped_env.is_some());
    }

    #[test]
    fn builder_with_leaf_sink_sets_provenance_and_no_authority() {
        let registry = OperationRegistryBuilder::new()
            .with_leaf_sink(
                pub_spec("fs/upload", AccessControl::default()),
                collect_sink_handler(),
                Capabilities::new(),
            )
            .unwrap()
            .build();
        let reg = registry.registration("fs/upload").expect("registered");
        assert_eq!(reg.provenance, OperationProvenance::FromOpenAPI);
        assert!(reg.composition_authority.is_none());
        assert!(reg.scoped_env.is_none());
        assert!(matches!(reg.handler, HandlerKind::Sink(_)));
    }

    #[test]
    fn builder_with_leaf_sink_provenance_overrides_provenance() {
        let registry = OperationRegistryBuilder::new()
            .with_leaf_sink_provenance(
                pub_spec("fs/upload", AccessControl::default()),
                collect_sink_handler(),
                OperationProvenance::FromCall,
                Capabilities::new(),
            )
            .unwrap()
            .build();
        let reg = registry.registration("fs/upload").expect("registered");
        assert_eq!(reg.provenance, OperationProvenance::FromCall);
        assert!(matches!(reg.handler, HandlerKind::Sink(_)));
    }

    // --- review 004 F-02/F-03: fork surface --------------------------------

    #[tokio::test]
    async fn fork_carries_handlers_validators_and_specs() {
        let registry = OperationRegistryBuilder::new()
            .with_local(
                external_spec("echo/run", AccessControl::default()),
                echo_handler(),
                None,
                None,
                Capabilities::new(),
            )
            .unwrap()
            .with_local_sink(
                pub_spec("fs/upload", AccessControl::default()),
                collect_sink_handler(),
                CompositionAuthority::none(),
                None,
                Capabilities::new(),
            )
            .unwrap()
            .build();

        let fork = registry.fork();
        assert!(fork.registration("echo/run").is_some());
        assert!(fork.registration("fs/upload").is_some());
        assert!(
            fork.publish_validator("fs/upload").is_none(),
            "op has no publish_schema — validator cache entry is None either way"
        );
        assert_eq!(fork.list_operations().len(), 2);
        assert!(
            fork.list_operations().iter().any(|s| s.name == "echo/run"),
            "External op appears in list_operations"
        );

        let ctx = root_context("req-fork-1", None, None, false, ScopedPeerEnv::empty());
        let response = fork
            .invoke("echo/run", serde_json::json!({"v": 1}), ctx)
            .await;
        assert_eq!(response.result.unwrap(), serde_json::json!({"v": 1}));
    }

    #[tokio::test]
    async fn fork_publish_schema_validator_is_carried_and_enforced() {
        let schema = serde_json::json!({
            "type": "object",
            "properties": { "n": { "type": "integer" } },
            "required": ["n"]
        });
        let spec = pub_spec("fs/upload", AccessControl::default()).with_publish_schema(schema);
        let registry = OperationRegistryBuilder::new()
            .with_local_sink(
                spec,
                collect_sink_handler(),
                CompositionAuthority::none(),
                None,
                Capabilities::new(),
            )
            .unwrap()
            .build();

        let fork = registry.fork();
        assert!(
            fork.publish_validator("fs/upload").is_some(),
            "F-03: validators map must be carried by the fork"
        );
    }

    #[tokio::test]
    async fn fork_is_independent_both_directions() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                external_spec("base/op", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();

        let fork = registry.fork();
        fork.register(HandlerRegistration::new(
            external_spec("fork/op", AccessControl::default()),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ))
        .unwrap();

        assert!(
            registry.registration("fork/op").is_none(),
            "fork registrations do not leak to the base"
        );
        assert!(
            fork.registration("base/op").is_some(),
            "base ops are on the fork"
        );
    }

    #[tokio::test]
    async fn register_through_shared_arc_after_dispatch_started() {
        let registry = Arc::new(OperationRegistry::new());
        registry
            .register(HandlerRegistration::new(
                external_spec("early/op", AccessControl::default()),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();

        let late = Arc::clone(&registry);
        let late_registration = HandlerRegistration::new(
            external_spec("late/op", AccessControl::default()),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        );
        late.register(late_registration).unwrap();

        let ctx = root_context("req-arc-1", None, None, false, ScopedPeerEnv::empty());
        let response = registry.invoke("late/op", serde_json::json!({}), ctx).await;
        assert!(response.result.is_ok());
    }

    // --- input_schema call-time enforcement (ADR-016 INVALID_INPUT leg) ----

    use futures::stream::StreamExt;

    fn spec_with_input_schema(name: &str, op_type: OperationType, schema: Value) -> OperationSpec {
        OperationSpec::new(
            name,
            op_type,
            Visibility::External,
            schema,
            serde_json::json!({}),
            vec![],
            AccessControl::default(),
            None,
        )
    }

    #[tokio::test]
    async fn invoke_rejects_input_failing_input_schema() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                spec_with_input_schema(
                    "echo",
                    OperationType::Query,
                    serde_json::json!({
                        "type": "object",
                        "properties": { "n": { "type": "integer" } },
                        "required": ["n"]
                    }),
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();

        let ctx = root_context("req-schema-1", None, None, false, ScopedPeerEnv::empty());
        let response = registry.invoke("echo", serde_json::json!({}), ctx).await;
        let err = response
            .result
            .expect_err("missing required field must be rejected");
        assert_eq!(err.code, "INVALID_INPUT");

        let ctx = root_context("req-schema-2", None, None, false, ScopedPeerEnv::empty());
        let response = registry
            .invoke("echo", serde_json::json!({"n": "not-an-int"}), ctx)
            .await;
        assert!(response.result.is_err(), "type mismatch must be rejected");

        let ctx = root_context("req-schema-3", None, None, false, ScopedPeerEnv::empty());
        let response = registry
            .invoke("echo", serde_json::json!({"n": 1}), ctx)
            .await;
        assert_eq!(response.result.unwrap(), serde_json::json!({"n": 1}));
    }

    #[tokio::test]
    async fn invoke_streaming_rejects_input_failing_input_schema() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                spec_with_input_schema(
                    "sub",
                    OperationType::Sub,
                    serde_json::json!({
                        "type": "object",
                        "properties": { "channel": { "type": "string" } },
                        "required": ["channel"]
                    }),
                ),
                HandlerKind::Stream(make_streaming_handler(|_input, ctx| {
                    Box::pin(futures::stream::once(async move {
                        ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
                    })) as Pin<Box<dyn Stream<Item = ResponseEnvelope> + Send>>
                })),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();

        let ctx = root_context("req-sub-1", None, None, false, ScopedPeerEnv::empty());
        let mut responses: Vec<_> = registry
            .invoke_streaming("sub", serde_json::json!({}), ctx)
            .collect()
            .await;
        assert_eq!(responses.len(), 1);
        let response = responses.pop().expect("one envelope");
        let err = response
            .result
            .expect_err("schema violation must be rejected");
        assert_eq!(err.code, "INVALID_INPUT");

        let ctx = root_context("req-sub-2", None, None, false, ScopedPeerEnv::empty());
        let responses: Vec<_> = registry
            .invoke_streaming("sub", serde_json::json!({"channel": "c"}), ctx)
            .collect()
            .await;
        assert_eq!(responses.len(), 1);
        assert!(responses[0].result.is_ok());
    }

    #[tokio::test]
    async fn invoke_sink_rejects_input_failing_input_schema() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                spec_with_input_schema(
                    "pub",
                    OperationType::Pub,
                    serde_json::json!({
                        "type": "object",
                        "properties": { "n": { "type": "integer" } },
                        "required": ["n"]
                    }),
                ),
                HandlerKind::Sink(collect_sink_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();

        let (chunk_tx, chunk_rx) = futures::channel::mpsc::channel::<Result<Value, CallError>>(8);
        drop(chunk_tx);
        let publish_stream: PublishStream = Box::pin(chunk_rx);
        let ctx = root_context("req-pub-1", None, None, false, ScopedPeerEnv::empty());
        let response = registry
            .invoke_sink("pub", serde_json::json!({}), publish_stream, ctx)
            .await;
        let err = response
            .result
            .expect_err("schema violation must be rejected");
        assert_eq!(err.code, "INVALID_INPUT");
    }

    #[test]
    fn uncompilable_input_schema_is_rejected_at_registration() {
        let registry = OperationRegistry::new();
        let result = registry.register(HandlerRegistration::new(
            spec_with_input_schema(
                "broken",
                OperationType::Query,
                serde_json::json!({
                    "required": "n"
                }),
            ),
            HandlerKind::Once(echo_handler()),
            OperationProvenance::Local,
            None,
            None,
            Capabilities::new(),
        ));
        assert!(
            result.is_err(),
            "un-compilable input_schema must fail registration"
        );
        assert!(
            result.unwrap_err().contains("input_schema"),
            "error must name the failing schema"
        );
    }

    #[test]
    fn empty_object_input_schema_compiles_and_accepts_anything() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                spec_with_input_schema("echo", OperationType::Query, serde_json::json!({})),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        assert!(
            registry.input_validator("echo").is_some(),
            "a declared schema compiles even when permissive"
        );
    }

    #[tokio::test]
    async fn fork_carries_input_validator_and_enforces() {
        let registry = OperationRegistry::new();
        registry
            .register(HandlerRegistration::new(
                spec_with_input_schema(
                    "echo",
                    OperationType::Query,
                    serde_json::json!({
                        "type": "object",
                        "properties": { "n": { "type": "integer" } },
                        "required": ["n"]
                    }),
                ),
                HandlerKind::Once(echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();

        let fork = registry.fork();
        let ctx = root_context("req-fork-schema", None, None, false, ScopedPeerEnv::empty());
        let response = fork.invoke("echo", serde_json::json!({}), ctx).await;
        let err = response
            .result
            .expect_err("the fork must carry the compiled input validator");
        assert_eq!(err.code, "INVALID_INPUT");
    }
}