use std::sync::Arc;
use serde_json::{json, Value};
use super::context::OperationContext;
use super::registration::{
Handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry,
};
use super::spec::{AccessControl, AccessResult, OperationSpec, OperationType, Visibility};
use crate::core::types::Capabilities;
use crate::protocol::wire::{CallError, ResponseEnvelope};
const NAME_SERVICES_LIST: &str = "services/list";
const NAME_SERVICES_LIST_PEERS: &str = "services/list-peers";
const NAME_SERVICES_SCHEMA: &str = "services/schema";
pub const BOOTSTRAP_DISCOVERY_OPS: [&str; 3] = [
NAME_SERVICES_LIST,
NAME_SERVICES_LIST_PEERS,
NAME_SERVICES_SCHEMA,
];
pub fn services_list_spec() -> OperationSpec {
OperationSpec::new(
NAME_SERVICES_LIST,
OperationType::Query,
Visibility::External,
json!({}),
json!({
"type": "object",
"properties": {
"operations": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": { "type": "string" },
"namespace": { "type": "string" },
"op_type": {
"type": "string",
"enum": ["query", "mutation", "sub", "pub"]
},
"description": {
"type": "string",
"description": "Human-readable op description (review 006 E-02). Absent when the producer declares none."
}
}
}
}
}
}),
vec![],
AccessControl::default(),
None,
)
}
pub fn services_schema_spec() -> OperationSpec {
OperationSpec::new(
NAME_SERVICES_SCHEMA,
OperationType::Query,
Visibility::External,
json!({
"type": "object",
"properties": { "name": { "type": "string" } },
"required": ["name"]
}),
operation_spec_schema(),
vec![],
AccessControl::default(),
None,
)
}
pub fn services_list_peers_spec() -> OperationSpec {
OperationSpec::new(
NAME_SERVICES_LIST_PEERS,
OperationType::Query,
Visibility::External,
json!({}),
json!({
"type": "object",
"properties": {
"peers": {
"type": "array",
"items": {
"type": "object",
"properties": {
"peer_id": { "type": "string" },
"operations": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": { "type": "string" },
"namespace": { "type": "string" },
"op_type": {
"type": "string",
"enum": ["query", "mutation", "sub", "pub"]
},
"description": {
"type": "string",
"description": "Human-readable op description. Absent when the op declares none."
}
}
}
}
}
}
}
}
}),
vec![],
AccessControl::default(),
None,
)
}
fn operation_spec_schema() -> Value {
json!({
"type": "object",
"properties": {
"name": { "type": "string" },
"namespace": { "type": "string" },
"op_type": {
"type": "string",
"enum": ["query", "mutation", "sub", "pub"]
},
"visibility": {
"type": "string",
"enum": ["external", "internal"]
},
"description": {
"description": "Human-readable op description (review 006 E-02). Absent when the op declares none; disclosed verbatim in `services/list` when set."
},
"input_schema": {},
"output_schema": {},
"error_schemas": {
"type": "array",
"items": {
"type": "object",
"properties": {
"code": { "type": "string" },
"description": { "type": "string" },
"schema": {},
"http_status": { "type": ["integer", "null"] }
}
}
},
"access_control": {
"type": "object",
"properties": {
"required_scopes": {
"type": "array",
"items": { "type": "string" }
},
"required_scopes_any": {
"type": ["array", "null"],
"items": { "type": "string" }
},
"resource_type": { "type": ["string", "null"] },
"resource_action": { "type": ["string", "null"] }
}
},
"channel_open": {
"type": ["boolean", "null"],
"description": "Marker (ADR-047): when true, the op's stream is binary and the channels layer allocates a data channel for it. Absent/null for JSON-stream ops."
},
"channel_open_alpn": {
"type": ["string", "null"],
"description": "Explicit data-plane ALPN for a marked op whose name is not the standard channels/<segment>/(sub|pub) shape (ADR-047 amendment — review 008 U-1, the flavor form). The boolean marker remains the dispatch hint; this string only carries the ALPN the name alone cannot disambiguate. Absent for standard-shape marked ops and for JSON-stream ops."
},
"publish_schema": {
"description": "Schema for each published chunk's `input` (Pub ops only, ADR-046 §4). Absent/null for Query/Mutation/Sub ops and for Pub ops with no per-chunk validation. When present, the dispatch path validates each `call.published` event's `payload.input` against this schema before yielding it to the SinkHandler."
}
},
"required": [
"name",
"namespace",
"op_type",
"visibility",
"input_schema",
"output_schema",
"error_schemas",
"access_control"
]
})
}
fn op_type_str(op_type: OperationType) -> &'static str {
match op_type {
OperationType::Query => "query",
OperationType::Mutation => "mutation",
OperationType::Sub => "sub",
OperationType::Pub => "pub",
}
}
fn visibility_str(visibility: Visibility) -> &'static str {
match visibility {
Visibility::External => "external",
Visibility::Internal => "internal",
}
}
fn access_control_to_json(acl: &AccessControl) -> Value {
json!({
"required_scopes": acl.required_scopes,
"required_scopes_any": acl.required_scopes_any,
"resource_type": acl.resource_type,
"resource_action": acl.resource_action,
})
}
fn error_definition_to_json(def: &super::spec::ErrorDefinition) -> Value {
json!({
"code": def.code,
"description": def.description,
"schema": def.schema,
"http_status": def.http_status,
})
}
pub(crate) fn spec_to_json(spec: &OperationSpec) -> Value {
spec_to_json_pub(spec)
}
pub(crate) fn op_name_is_standard_channel_open_shape(name: &str) -> bool {
let Some(rest) = name.strip_prefix("channels/") else {
return false;
};
let Some((segment, flavor)) = rest.rsplit_once('/') else {
return false;
};
!segment.is_empty() && (flavor == "sub" || flavor == "pub")
}
pub fn spec_to_json_pub(spec: &OperationSpec) -> Value {
let error_schemas: Vec<Value> = spec
.error_schemas
.iter()
.map(error_definition_to_json)
.collect();
let mut json = json!({
"name": spec.name,
"namespace": spec.namespace,
"op_type": op_type_str(spec.op_type),
"visibility": visibility_str(spec.visibility),
"input_schema": spec.input_schema,
"output_schema": spec.output_schema,
"error_schemas": error_schemas,
"access_control": access_control_to_json(&spec.access_control),
});
if let Some(resource_id_path) = &spec.resource_id_path {
json["resource_id_path"] = json!(resource_id_path);
}
if let Some(description) = &spec.description {
json["description"] = json!(description);
}
if let Some(channel_open) = &spec.channel_open {
json["channel_open"] = json!(true);
if !op_name_is_standard_channel_open_shape(&spec.name) {
json["channel_open_alpn"] = json!(channel_open.alpn.as_ref());
}
}
if let Some(publish_schema) = &spec.publish_schema {
json["publish_schema"] = publish_schema.clone();
}
json
}
fn normalize_name(name: &str) -> String {
if let Some(rest) = name.strip_prefix('/') {
rest.to_string()
} else {
name.to_string()
}
}
pub fn services_list_handler(registry: Arc<OperationRegistry>) -> Handler {
Arc::new(move |input: Value, ctx: OperationContext| {
let registry = Arc::clone(®istry);
Box::pin(async move {
let _ = input;
let calling_identity = ctx.identity.as_ref();
let ops: Vec<Value> = registry
.list_operations()
.into_iter()
.filter(|spec| {
spec.access_control
.check(calling_identity, None, None)
.is_allowed()
})
.map(|s| {
let mut listing = json!({
"name": s.name,
"namespace": s.namespace,
"op_type": op_type_str(s.op_type),
});
if let Some(description) = &s.description {
listing["description"] = json!(description);
}
listing
})
.collect();
ResponseEnvelope::ok(ctx.request_id, json!({ "operations": ops }))
})
})
}
pub fn install_bootstrap_discovery(registry: &Arc<OperationRegistry>) -> Result<(), String> {
registry.register(HandlerRegistration::new(
services_list_spec(),
HandlerKind::Once(services_list_handler(Arc::clone(registry))),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))?;
registry.register(HandlerRegistration::new(
services_list_peers_spec(),
HandlerKind::Once(services_list_peers_handler(Arc::clone(registry))),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))?;
registry.register(HandlerRegistration::new(
services_schema_spec(),
HandlerKind::Once(services_schema_handler(Arc::clone(registry))),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))?;
Ok(())
}
pub fn services_list_peers_handler(registry: Arc<OperationRegistry>) -> Handler {
Arc::new(move |input: Value, ctx: OperationContext| {
let registry = Arc::clone(®istry);
Box::pin(async move {
let _ = input;
let calling_identity = ctx.identity.as_ref();
let local_ops: Vec<Value> = registry
.list_operations()
.into_iter()
.filter(|spec| {
spec.access_control
.check(calling_identity, None, None)
.is_allowed()
})
.map(|s| {
let mut listing = json!({
"name": s.name,
"namespace": s.namespace,
"op_type": op_type_str(s.op_type),
});
if let Some(description) = &s.description {
listing["description"] = json!(description);
}
listing
})
.collect();
let mut peers: Vec<Value> = Vec::new();
if !local_ops.is_empty() {
peers.push(json!({ "peer_id": "local", "operations": local_ops }));
}
for peer_id in ctx.env.peer_ids() {
let peer_ops: Vec<Value> = ctx
.env
.peer_operations(&peer_id)
.into_iter()
.filter(|name| {
let spec = registry.registration(name);
match spec {
Some(reg) => reg
.spec
.access_control
.check(calling_identity, None, None)
.is_allowed(),
None => true,
}
})
.map(name_to_listing_json)
.collect();
if !peer_ops.is_empty() {
peers.push(json!({ "peer_id": peer_id, "operations": peer_ops }));
}
}
ResponseEnvelope::ok(ctx.request_id, json!({ "peers": peers }))
})
})
}
fn name_to_listing_json(name: String) -> Value {
let namespace = name
.split('/')
.next()
.filter(|s| !s.is_empty())
.unwrap_or("")
.to_string();
json!({
"name": name,
"namespace": namespace,
"op_type": "query",
})
}
pub fn services_schema_handler(registry: Arc<OperationRegistry>) -> Handler {
Arc::new(move |input: Value, ctx: OperationContext| {
let registry = Arc::clone(®istry);
Box::pin(async move {
let name = match input.get("name").and_then(|v| v.as_str()) {
Some(n) => normalize_name(n),
None => {
return ResponseEnvelope::error(
ctx.request_id,
CallError::invalid_input("missing required field: name"),
);
}
};
let registration = match registry.registration(&name) {
Some(reg) => reg,
None => return ResponseEnvelope::not_found(ctx.request_id, &name),
};
if registration.spec.visibility == Visibility::Internal && !ctx.internal {
return ResponseEnvelope::not_found(ctx.request_id, &name);
}
let identity = if ctx.internal {
ctx.handler_identity
.as_ref()
.and_then(|ca| ca.as_identity())
} else {
ctx.identity.clone()
};
if let AccessResult::Forbidden(_) = registration.spec.access_control.check(
identity.as_ref(),
None,
ctx.ownership.as_deref(),
) {
return ResponseEnvelope::not_found(ctx.request_id, &name);
}
let spec_json = spec_to_json(®istration.spec);
ResponseEnvelope::ok(ctx.request_id, spec_json)
})
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::core::types::Capabilities;
use crate::registry::context::{CompositionAuthority, ScopedPeerEnv};
use crate::registry::registration::{
make_handler, make_streaming_handler, HandlerKind, HandlerRegistration,
OperationProvenance, StreamingHandler,
};
use std::collections::HashMap;
use std::time::Duration;
fn external_spec(name: &str) -> OperationSpec {
OperationSpec::new(
name,
OperationType::Query,
Visibility::External,
json!({}),
json!({}),
vec![],
AccessControl::default(),
None,
)
}
fn internal_spec(name: &str) -> OperationSpec {
OperationSpec::new(
name,
OperationType::Mutation,
Visibility::Internal,
json!({}),
json!({}),
vec![],
AccessControl::default(),
None,
)
}
fn echo_handler() -> Handler {
make_handler(
|input, context| async move { ResponseEnvelope::ok(context.request_id, input) },
)
}
fn echo_streaming_handler() -> StreamingHandler {
make_streaming_handler(|input, context| {
futures::stream::iter(vec![ResponseEnvelope::ok(context.request_id, input)])
})
}
fn noop_env() -> Arc<dyn crate::registry::env::OperationEnv + Send + Sync> {
struct NoopEnv;
#[async_trait::async_trait]
impl crate::registry::env::OperationEnv for NoopEnv {
async fn invoke_with_policy(
&self,
_ns: &str,
_op: &str,
_input: Value,
_parent: &OperationContext,
_policy: crate::registry::context::AbortPolicy,
) -> ResponseEnvelope {
ResponseEnvelope::error("test", CallError::internal("noop env does not dispatch"))
}
fn contains(&self, _name: &str) -> bool {
false
}
}
Arc::new(NoopEnv)
}
fn root_context(request_id: &str) -> OperationContext {
OperationContext {
request_id: request_id.to_string(),
parent_request_id: None,
identity: None,
handler_identity: None,
forwarded_for: None,
capabilities: Capabilities::new(),
metadata: HashMap::new(),
scoped_env: ScopedPeerEnv::empty(),
env: noop_env(),
abort_policy: crate::registry::context::AbortPolicy::default(),
deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
internal: false,
ownership: None,
}
}
fn root_context_with_identity(
request_id: &str,
identity: Option<crate::core::auth::Identity>,
) -> 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: ScopedPeerEnv::empty(),
env: noop_env(),
abort_policy: crate::registry::context::AbortPolicy::default(),
deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
internal: false,
ownership: None,
}
}
fn identity_with_scopes(id: &str, scopes: &[&str]) -> crate::core::auth::Identity {
crate::core::auth::Identity {
id: id.to_string(),
scopes: scopes.iter().map(|s| s.to_string()).collect(),
resources: HashMap::new(),
}
}
fn external_spec_with_acl(name: &str, acl: AccessControl) -> OperationSpec {
OperationSpec::new(
name,
OperationType::Query,
Visibility::External,
json!({}),
json!({}),
vec![],
acl,
None,
)
}
fn registry_with_access_controlled_ops() -> Arc<OperationRegistry> {
let registry = OperationRegistry::new();
registry
.register(HandlerRegistration::new(
external_spec_with_acl("public/echo", AccessControl::default()),
HandlerKind::Once(echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
registry
.register(HandlerRegistration::new(
external_spec_with_acl(
"admin/secret",
AccessControl {
required_scopes: vec!["admin".to_string()],
..Default::default()
},
),
HandlerKind::Once(echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
registry
.register(HandlerRegistration::new(
internal_spec("internal/hidden"),
HandlerKind::Once(echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
Arc::new(registry)
}
fn op_names(response: ResponseEnvelope) -> Vec<String> {
let output = response.result.expect("ok response");
output
.get("operations")
.and_then(|v| v.as_array())
.expect("operations array")
.iter()
.filter_map(|o| o.get("name").and_then(|n| n.as_str()).map(String::from))
.collect()
}
fn registry_with_ops() -> Arc<OperationRegistry> {
let registry = OperationRegistry::new();
registry
.register(HandlerRegistration::new(
external_spec("fs/readFile"),
HandlerKind::Once(echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
registry
.register(HandlerRegistration::new(
internal_spec("secret/internal"),
HandlerKind::Once(echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
registry
.register(HandlerRegistration::new(
OperationSpec::new(
"events/subscribe",
OperationType::Sub,
Visibility::External,
json!({}),
json!({}),
vec![],
AccessControl::default(),
None,
),
HandlerKind::Stream(echo_streaming_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
registry
.register(HandlerRegistration::new(
OperationSpec::new(
"fs/readFileErr",
OperationType::Query,
Visibility::External,
json!({}),
json!({}),
vec![super::super::spec::ErrorDefinition {
code: "FILE_NOT_FOUND".to_string(),
description: "file not found".to_string(),
schema: json!({ "type": "object" }),
http_status: None,
}],
AccessControl::default(),
None,
),
HandlerKind::Once(echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
Arc::new(registry)
}
#[test]
fn services_list_spec_has_correct_fields() {
let spec = services_list_spec();
assert_eq!(spec.name, NAME_SERVICES_LIST);
assert_eq!(spec.namespace, "services");
assert_eq!(spec.op_type, OperationType::Query);
assert_eq!(spec.visibility, Visibility::External);
assert_eq!(spec.input_schema, json!({}));
assert!(spec.output_schema.get("properties").is_some());
assert!(spec.error_schemas.is_empty());
assert!(!spec.access_control.has_restrictions());
}
#[test]
fn services_schema_spec_has_correct_fields() {
let spec = services_schema_spec();
assert_eq!(spec.name, NAME_SERVICES_SCHEMA);
assert_eq!(spec.namespace, "services");
assert_eq!(spec.op_type, OperationType::Query);
assert_eq!(spec.visibility, Visibility::External);
assert!(spec.input_schema.get("required").is_some());
assert!(spec.output_schema.get("properties").is_some());
assert!(spec.error_schemas.is_empty());
assert!(!spec.access_control.has_restrictions());
}
#[tokio::test]
async fn services_list_returns_external_ops_only() {
let registry = registry_with_ops();
let handler = services_list_handler(Arc::clone(®istry));
let ctx = root_context("req-1");
let response = handler(serde_json::json!({}), ctx).await;
let output = response.result.expect("ok response");
let ops = output
.get("operations")
.and_then(|v| v.as_array())
.expect("operations array");
let names: Vec<&str> = ops
.iter()
.filter_map(|o| o.get("name").and_then(|n| n.as_str()))
.collect();
assert!(names.contains(&"fs/readFile"));
assert!(names.contains(&"events/subscribe"));
assert!(names.contains(&"fs/readFileErr"));
assert!(
!names.contains(&"secret/internal"),
"internal ops must not be listed"
);
}
#[tokio::test]
async fn services_list_output_format_matches_spec() {
let registry = registry_with_ops();
let handler = services_list_handler(Arc::clone(®istry));
let ctx = root_context("req-1");
let response = handler(serde_json::json!({}), ctx).await;
let output = response.result.expect("ok response");
let ops = output
.get("operations")
.and_then(|v| v.as_array())
.expect("operations array");
let fs_op = ops
.iter()
.find(|o| o.get("name").and_then(|n| n.as_str()) == Some("fs/readFile"))
.expect("fs/readFile present");
assert_eq!(fs_op.get("namespace"), Some(&json!("fs")));
assert_eq!(fs_op.get("op_type"), Some(&json!("query")));
}
#[tokio::test]
async fn services_schema_returns_spec_for_known_op() {
let registry = registry_with_ops();
let handler = services_schema_handler(Arc::clone(®istry));
let ctx = root_context("req-2");
let response = handler(serde_json::json!({ "name": "fs/readFileErr" }), ctx).await;
let spec = response.result.expect("ok response");
assert_eq!(spec.get("name"), Some(&json!("fs/readFileErr")));
assert_eq!(spec.get("namespace"), Some(&json!("fs")));
assert_eq!(spec.get("op_type"), Some(&json!("query")));
let error_schemas = spec
.get("error_schemas")
.and_then(|v| v.as_array())
.expect("error_schemas array");
assert_eq!(error_schemas.len(), 1);
assert_eq!(error_schemas[0].get("code"), Some(&json!("FILE_NOT_FOUND")));
}
#[tokio::test]
async fn services_schema_returns_not_found_for_unknown_op() {
let registry = registry_with_ops();
let handler = services_schema_handler(Arc::clone(®istry));
let ctx = root_context("req-3");
let response = handler(serde_json::json!({ "name": "no/such" }), ctx).await;
match response.result {
Err(e) => assert_eq!(e.code, "NOT_FOUND"),
other => panic!("expected NOT_FOUND, got {other:?}"),
}
}
#[tokio::test]
async fn services_schema_accepts_name_with_leading_slash() {
let registry = registry_with_ops();
let handler = services_schema_handler(Arc::clone(®istry));
let ctx = root_context("req-4");
let response = handler(serde_json::json!({ "name": "/fs/readFile" }), ctx).await;
let spec = response.result.expect("ok response");
assert_eq!(spec.get("name"), Some(&json!("fs/readFile")));
}
#[tokio::test]
async fn services_schema_rejects_missing_name() {
let registry = registry_with_ops();
let handler = services_schema_handler(Arc::clone(®istry));
let ctx = root_context("req-5");
let response = handler(serde_json::json!({}), ctx).await;
match response.result {
Err(e) => assert_eq!(e.code, "INVALID_INPUT"),
other => panic!("expected INVALID_INPUT, got {other:?}"),
}
}
#[tokio::test]
async fn services_schema_hides_internal_op_from_external_caller() {
let registry = registry_with_access_controlled_ops();
let handler = services_schema_handler(Arc::clone(®istry));
let ctx = root_context("req-cf4-1");
let response = handler(serde_json::json!({ "name": "internal/hidden" }), ctx).await;
match response.result {
Err(e) => assert_eq!(e.code, "NOT_FOUND"),
other => panic!("expected NOT_FOUND for internal op, got {other:?}"),
}
}
#[tokio::test]
async fn services_schema_shows_internal_op_to_internal_caller() {
let registry = registry_with_access_controlled_ops();
let handler = services_schema_handler(Arc::clone(®istry));
let mut ctx = root_context("req-cf4-2");
ctx.internal = true;
ctx.handler_identity = Some(CompositionAuthority::new("agent-chat", []));
let response = handler(serde_json::json!({ "name": "internal/hidden" }), ctx).await;
let spec = response.result.expect("internal caller sees internal op");
assert_eq!(spec.get("name"), Some(&json!("internal/hidden")));
}
#[tokio::test]
async fn services_schema_hides_acl_restricted_op_from_unauthorized_caller() {
let registry = registry_with_access_controlled_ops();
let handler = services_schema_handler(Arc::clone(®istry));
let ctx = root_context_with_identity(
"req-cf4-3",
Some(identity_with_scopes("regular-peer", &["user"])),
);
let response = handler(serde_json::json!({ "name": "admin/secret" }), ctx).await;
match response.result {
Err(e) => assert_eq!(e.code, "NOT_FOUND"),
other => panic!("expected NOT_FOUND for unauthorized ACL, got {other:?}"),
}
}
#[tokio::test]
async fn services_schema_shows_acl_restricted_op_to_authorized_caller() {
let registry = registry_with_access_controlled_ops();
let handler = services_schema_handler(Arc::clone(®istry));
let ctx = root_context_with_identity(
"req-cf4-4",
Some(identity_with_scopes("admin-peer", &["admin"])),
);
let response = handler(serde_json::json!({ "name": "admin/secret" }), ctx).await;
let spec = response.result.expect("authorized caller sees the spec");
assert_eq!(spec.get("name"), Some(&json!("admin/secret")));
assert_eq!(
spec.get("access_control")
.and_then(|a| a.get("required_scopes")),
Some(&json!(["admin"]))
);
}
#[tokio::test]
async fn services_schema_unrestricted_op_fetchable_without_identity() {
let registry = registry_with_access_controlled_ops();
let handler = services_schema_handler(Arc::clone(®istry));
let ctx = root_context("req-cf4-5");
let response = handler(serde_json::json!({ "name": "public/echo" }), ctx).await;
let spec = response
.result
.expect("default-ACL op fetchable unauthenticated");
assert_eq!(spec.get("name"), Some(&json!("public/echo")));
}
#[tokio::test]
async fn services_list_handler_registered_and_invocable_via_registry() {
let registry = registry_with_ops();
let list_handler = services_list_handler(Arc::clone(®istry));
let schema_handler = services_schema_handler(Arc::clone(®istry));
let discovery_registry = OperationRegistry::new();
discovery_registry
.register(HandlerRegistration::new(
services_list_spec(),
HandlerKind::Once(list_handler),
OperationProvenance::Local,
CompositionAuthority::none(),
ScopedPeerEnv::empty().into(),
Capabilities::new(),
))
.unwrap();
discovery_registry
.register(HandlerRegistration::new(
services_schema_spec(),
HandlerKind::Once(schema_handler),
OperationProvenance::Local,
CompositionAuthority::none(),
ScopedPeerEnv::empty().into(),
Capabilities::new(),
))
.unwrap();
let discovery = Arc::new(discovery_registry);
let ctx = root_context("req-6");
let response = discovery
.invoke(NAME_SERVICES_LIST, serde_json::json!({}), ctx)
.await;
let output = response.result.expect("list ok");
assert!(output.get("operations").is_some());
}
#[test]
fn normalize_name_strips_leading_slash() {
assert_eq!(normalize_name("/fs/readFile"), "fs/readFile");
assert_eq!(normalize_name("fs/readFile"), "fs/readFile");
}
#[test]
fn op_type_str_matches_wire_enum() {
assert_eq!(op_type_str(OperationType::Query), "query");
assert_eq!(op_type_str(OperationType::Mutation), "mutation");
assert_eq!(op_type_str(OperationType::Sub), "sub");
assert_eq!(op_type_str(OperationType::Pub), "pub");
}
#[test]
fn visibility_str_matches_wire_enum() {
assert_eq!(visibility_str(Visibility::External), "external");
assert_eq!(visibility_str(Visibility::Internal), "internal");
}
#[test]
fn spec_to_json_round_trips_error_schemas() {
let spec = OperationSpec::new(
"fs/readFile",
OperationType::Query,
Visibility::External,
json!({ "type": "object" }),
json!({ "type": "string" }),
vec![super::super::spec::ErrorDefinition {
code: "FILE_NOT_FOUND".to_string(),
description: "file not found".to_string(),
schema: json!({ "type": "object", "properties": { "path": { "type": "string" } } }),
http_status: Some(404),
}],
AccessControl {
required_scopes: vec!["fs:read".to_string()],
..Default::default()
},
None,
);
let json_val = spec_to_json(&spec);
let error_schemas = json_val
.get("error_schemas")
.and_then(|v| v.as_array())
.expect("error_schemas");
assert_eq!(error_schemas.len(), 1);
assert_eq!(error_schemas[0].get("code"), Some(&json!("FILE_NOT_FOUND")));
assert_eq!(error_schemas[0].get("http_status"), Some(&json!(404)));
let acl = json_val.get("access_control").expect("access_control");
assert_eq!(acl.get("required_scopes"), Some(&json!(["fs:read"])));
}
#[test]
fn spec_to_json_emits_channel_open_boolean_when_set() {
let spec = OperationSpec::new(
"channels/tty/sub",
OperationType::Sub,
Visibility::External,
json!({}),
json!({}),
vec![],
AccessControl::default(),
None,
)
.with_channel_open(super::super::spec::ChannelOpenSpec::new("alk/tty"));
let json_val = spec_to_json(&spec);
assert_eq!(json_val.get("channel_open"), Some(&json!(true)));
}
#[test]
fn spec_to_json_omits_channel_open_when_absent() {
let spec = OperationSpec::new(
"fs/readFile",
OperationType::Query,
Visibility::External,
json!({}),
json!({}),
vec![],
AccessControl::default(),
None,
);
let json_val = spec_to_json(&spec);
assert!(
json_val.get("channel_open").is_none(),
"channel_open must be absent for a plain op"
);
}
#[test]
fn spec_to_json_emits_publish_schema_when_set() {
let publish_schema = json!({
"type": "object",
"properties": { "bytes": { "type": "string" } },
"required": ["bytes"]
});
let spec = OperationSpec::new(
"fs/upload",
OperationType::Pub,
Visibility::External,
json!({}),
json!({}),
vec![],
AccessControl::default(),
None,
)
.with_publish_schema(publish_schema.clone());
let json_val = spec_to_json(&spec);
assert_eq!(json_val.get("publish_schema"), Some(&publish_schema));
}
#[test]
fn spec_to_json_omits_publish_schema_when_absent() {
let spec = OperationSpec::new(
"fs/readFile",
OperationType::Query,
Visibility::External,
json!({}),
json!({}),
vec![],
AccessControl::default(),
None,
);
let json_val = spec_to_json(&spec);
assert!(
json_val.get("publish_schema").is_none(),
"publish_schema must be absent when not set"
);
}
#[test]
fn operation_spec_schema_documents_publish_schema_property() {
let schema = operation_spec_schema();
let props = schema
.get("properties")
.and_then(|v| v.as_object())
.expect("properties object");
assert!(
props.contains_key("publish_schema"),
"operation_spec_schema must advertise publish_schema"
);
}
#[test]
fn spec_to_json_emits_channel_open_alpn_for_flavor_form() {
let spec = external_spec("channels/tunnel/direct")
.with_channel_open(super::super::spec::ChannelOpenSpec::new("alk/tunnel"));
let json_val = spec_to_json(&spec);
assert_eq!(json_val.get("channel_open"), Some(&json!(true)));
assert_eq!(
json_val.get("channel_open_alpn"),
Some(&json!("alk/tunnel")),
"flavor-form marked op emits the explicit ALPN"
);
}
#[test]
fn spec_to_json_omits_channel_open_alpn_for_standard_shape() {
let spec = external_spec("channels/tty/sub")
.with_channel_open(super::super::spec::ChannelOpenSpec::new("alk/tty"));
let json_val = spec_to_json(&spec);
assert_eq!(json_val.get("channel_open"), Some(&json!(true)));
assert!(
json_val.get("channel_open_alpn").is_none(),
"standard shape keeps the boolean-only payload (byte-stable)"
);
}
#[test]
fn spec_to_json_omits_channel_open_alpn_when_marker_absent() {
let spec = external_spec("fs/readFile");
let json_val = spec_to_json(&spec);
assert!(json_val.get("channel_open").is_none());
assert!(json_val.get("channel_open_alpn").is_none());
}
#[test]
fn operation_spec_schema_documents_channel_open_alpn_property() {
let schema = operation_spec_schema();
let props = schema
.get("properties")
.and_then(|v| v.as_object())
.expect("properties object");
let prop = props
.get("channel_open_alpn")
.expect("operation_spec_schema must advertise channel_open_alpn");
assert_eq!(
prop["type"],
json!(["string", "null"]),
"optional string property (schema-type widening, additive)"
);
}
#[test]
fn op_name_is_standard_channel_open_shape_covers_both_flavors() {
assert!(op_name_is_standard_channel_open_shape("channels/tty/sub"));
assert!(op_name_is_standard_channel_open_shape("channels/tty/pub"));
assert!(op_name_is_standard_channel_open_shape(
"channels/custom/proto/sub"
));
assert!(!op_name_is_standard_channel_open_shape(
"channels/tunnel/direct"
));
assert!(!op_name_is_standard_channel_open_shape(
"channels/tunnel/forwarded"
));
assert!(!op_name_is_standard_channel_open_shape("fs/readFile"));
assert!(!op_name_is_standard_channel_open_shape("channels/tty"));
}
#[test]
fn spec_to_json_emits_description_when_set() {
let spec = external_spec("fs/readFile").with_description("Read a file");
let json_val = spec_to_json(&spec);
assert_eq!(json_val.get("description"), Some(&json!("Read a file")));
}
#[test]
fn spec_to_json_omits_description_when_absent() {
let spec = external_spec("fs/readFile");
let json_val = spec_to_json(&spec);
assert!(
json_val.get("description").is_none(),
"description must be absent when not set (additive optional field)"
);
}
#[tokio::test]
async fn services_list_emits_description_when_set() {
let registry = Arc::new(OperationRegistry::new());
registry
.register(HandlerRegistration::new(
external_spec("channels/tty/sub").with_description("Interactive TTY sessions"),
HandlerKind::Once(echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
registry
.register(HandlerRegistration::new(
external_spec("fs/readFile"),
HandlerKind::Once(echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
let handler = services_list_handler(Arc::clone(®istry));
let response = handler(json!({}), root_context("req-e02-1")).await;
let output = response.result.expect("ok response");
let ops = output
.get("operations")
.and_then(|v| v.as_array())
.expect("operations array")
.iter()
.map(|o| {
(
o.get("name").and_then(|n| n.as_str()).unwrap_or(""),
o.get("description").and_then(|d| d.as_str()),
)
})
.collect::<Vec<_>>();
assert!(
ops.contains(&("channels/tty/sub", Some("Interactive TTY sessions"))),
"described op carries its description: {ops:?}"
);
assert!(
ops.contains(&("fs/readFile", None)),
"undescribed op omits the description key: {ops:?}"
);
}
#[tokio::test]
async fn services_schema_discloses_description() {
let registry = registry_with_ops();
let handler = services_schema_handler(Arc::clone(®istry));
let described = external_spec("fs/readFile").with_description("Read a file");
registry
.register(HandlerRegistration::new(
described,
HandlerKind::Once(echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
let response = handler(json!({ "name": "fs/readFile" }), root_context("req-e02-2")).await;
let spec = response.result.expect("ok response");
assert_eq!(spec.get("description"), Some(&json!("Read a file")));
}
#[test]
fn operation_spec_schema_documents_description_property() {
let schema = operation_spec_schema();
let props = schema
.get("properties")
.and_then(|v| v.as_object())
.expect("properties object");
assert!(
props.contains_key("description"),
"operation_spec_schema must advertise description"
);
}
#[tokio::test]
async fn services_list_filters_by_access_control_authorized_peer() {
let registry = registry_with_access_controlled_ops();
let handler = services_list_handler(Arc::clone(®istry));
let ctx = root_context_with_identity(
"req-acl-1",
Some(identity_with_scopes("admin-peer", &["admin"])),
);
let names = op_names(handler(serde_json::json!({}), ctx).await);
assert!(names.contains(&"public/echo".to_string()));
assert!(names.contains(&"admin/secret".to_string()));
assert!(!names.contains(&"internal/hidden".to_string()));
}
#[tokio::test]
async fn services_list_filters_by_access_control_unauthorized_peer() {
let registry = registry_with_access_controlled_ops();
let handler = services_list_handler(Arc::clone(®istry));
let ctx = root_context_with_identity(
"req-acl-2",
Some(identity_with_scopes("regular-peer", &["user"])),
);
let names = op_names(handler(serde_json::json!({}), ctx).await);
assert!(names.contains(&"public/echo".to_string()));
assert!(
!names.contains(&"admin/secret".to_string()),
"unauthorized peer must not see admin/secret"
);
assert!(!names.contains(&"internal/hidden".to_string()));
}
#[tokio::test]
async fn services_list_op_with_default_acl_listed_to_any_peer() {
let registry = registry_with_access_controlled_ops();
let handler = services_list_handler(Arc::clone(®istry));
let ctx = root_context_with_identity("req-acl-3", None);
let names = op_names(handler(serde_json::json!({}), ctx).await);
assert!(
names.contains(&"public/echo".to_string()),
"default AccessControl op must be listed to unauthenticated peer"
);
assert!(!names.contains(&"admin/secret".to_string()));
}
#[tokio::test]
async fn services_list_peers_attributes_ops_by_peer_id() {
struct PeerEnv {
peers: HashMap<String, Vec<String>>,
}
#[async_trait::async_trait]
impl crate::registry::env::OperationEnv for PeerEnv {
async fn invoke_with_policy(
&self,
_ns: &str,
_op: &str,
_input: Value,
parent: &OperationContext,
_policy: crate::registry::context::AbortPolicy,
) -> ResponseEnvelope {
ResponseEnvelope::ok(parent.request_id.clone(), json!({}))
}
fn contains(&self, _name: &str) -> bool {
false
}
fn peer_ids(&self) -> Vec<crate::registry::env::PeerId> {
self.peers.keys().cloned().collect()
}
fn peer_operations(&self, peer: &crate::registry::env::PeerId) -> Vec<String> {
self.peers.get(peer).cloned().unwrap_or_default()
}
}
let mut peers = HashMap::new();
peers.insert(
"worker-a".to_string(),
vec!["container/exec".to_string(), "container/logs".to_string()],
);
peers.insert("worker-b".to_string(), vec!["container/exec".to_string()]);
let env: Arc<dyn crate::registry::env::OperationEnv + Send + Sync> =
Arc::new(PeerEnv { peers });
let registry = registry_with_access_controlled_ops();
let handler = services_list_peers_handler(Arc::clone(®istry));
let ctx = OperationContext {
request_id: "req-peers-1".to_string(),
parent_request_id: None,
identity: None,
handler_identity: None,
forwarded_for: None,
capabilities: Capabilities::new(),
metadata: HashMap::new(),
scoped_env: ScopedPeerEnv::empty(),
env,
abort_policy: crate::registry::context::AbortPolicy::default(),
deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
internal: false,
ownership: None,
};
let response = handler(serde_json::json!({}), ctx).await;
let output = response.result.expect("ok response");
let peers_arr = output
.get("peers")
.and_then(|v| v.as_array())
.expect("peers array");
let peer_ids: Vec<&str> = peers_arr
.iter()
.filter_map(|p| p.get("peer_id").and_then(|v| v.as_str()))
.collect();
assert!(peer_ids.contains(&"local"));
assert!(peer_ids.contains(&"worker-a"));
assert!(peer_ids.contains(&"worker-b"));
let worker_a = peers_arr
.iter()
.find(|p| p.get("peer_id").and_then(|v| v.as_str()) == Some("worker-a"))
.expect("worker-a present");
let worker_a_ops = worker_a
.get("operations")
.and_then(|v| v.as_array())
.expect("worker-a operations");
let worker_a_names: Vec<&str> = worker_a_ops
.iter()
.filter_map(|o| o.get("name").and_then(|n| n.as_str()))
.collect();
assert!(worker_a_names.contains(&"container/exec"));
assert!(worker_a_names.contains(&"container/logs"));
}
#[test]
fn services_list_peers_spec_has_correct_fields() {
let spec = services_list_peers_spec();
assert_eq!(spec.name, NAME_SERVICES_LIST_PEERS);
assert_eq!(spec.namespace, "services");
assert_eq!(spec.op_type, OperationType::Query);
assert_eq!(spec.visibility, Visibility::External);
assert!(spec.error_schemas.is_empty());
assert!(!spec.access_control.has_restrictions());
}
#[tokio::test]
async fn services_list_peers_filters_by_access_control() {
struct PeerEnv;
#[async_trait::async_trait]
impl crate::registry::env::OperationEnv for PeerEnv {
async fn invoke_with_policy(
&self,
_ns: &str,
_op: &str,
_input: Value,
parent: &OperationContext,
_policy: crate::registry::context::AbortPolicy,
) -> ResponseEnvelope {
ResponseEnvelope::ok(parent.request_id.clone(), json!({}))
}
fn contains(&self, _name: &str) -> bool {
false
}
fn peer_ids(&self) -> Vec<crate::registry::env::PeerId> {
vec!["restricted-peer".to_string()]
}
fn peer_operations(&self, _peer: &crate::registry::env::PeerId) -> Vec<String> {
vec!["admin/secret".to_string(), "public/echo".to_string()]
}
}
let registry = registry_with_access_controlled_ops();
let handler = services_list_peers_handler(Arc::clone(®istry));
let env: Arc<dyn crate::registry::env::OperationEnv + Send + Sync> = Arc::new(PeerEnv);
let ctx = OperationContext {
request_id: "req-peers-2".to_string(),
parent_request_id: None,
identity: Some(identity_with_scopes("regular-peer", &["user"])),
handler_identity: None,
forwarded_for: None,
capabilities: Capabilities::new(),
metadata: HashMap::new(),
scoped_env: ScopedPeerEnv::empty(),
env,
abort_policy: crate::registry::context::AbortPolicy::default(),
deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
internal: false,
ownership: None,
};
let response = handler(serde_json::json!({}), ctx).await;
let output = response.result.expect("ok response");
let peers_arr = output
.get("peers")
.and_then(|v| v.as_array())
.expect("peers array");
let restricted = peers_arr
.iter()
.find(|p| p.get("peer_id").and_then(|v| v.as_str()) == Some("restricted-peer"))
.expect("restricted-peer present");
let ops = restricted
.get("operations")
.and_then(|v| v.as_array())
.expect("operations");
let names: Vec<&str> = ops
.iter()
.filter_map(|o| o.get("name").and_then(|n| n.as_str()))
.collect();
assert!(names.contains(&"public/echo"));
assert!(
!names.contains(&"admin/secret"),
"unauthorized peer must not see admin op in list-peers"
);
}
fn context_for(
request_id: &str,
identity: Option<crate::core::auth::Identity>,
) -> 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: ScopedPeerEnv::empty(),
env: Arc::new(crate::registry::env::LocalOperationEnv::new(Arc::new(
OperationRegistry::new(),
))),
abort_policy: crate::registry::context::AbortPolicy::default(),
deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
internal: false,
ownership: None,
}
}
fn identity_scopes(id: &str, scopes: &[&str]) -> crate::core::auth::Identity {
crate::core::auth::Identity {
id: id.to_string(),
scopes: scopes.iter().map(|s| s.to_string()).collect(),
resources: HashMap::new(),
}
}
#[tokio::test]
async fn bootstrap_discovery_on_fork_sees_per_session_ops() {
let base = OperationRegistry::new();
base.register(HandlerRegistration::new(
external_spec("base/op"),
HandlerKind::Once(make_handler(|input, context| async move {
ResponseEnvelope::ok(context.request_id, input)
})),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
let fork = Arc::new(base.fork());
fork.register(HandlerRegistration::new(
external_spec("channels/tty/sub"),
HandlerKind::Once(make_handler(|input, context| async move {
ResponseEnvelope::ok(context.request_id, input)
})),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
install_bootstrap_discovery(&fork).expect("bootstrap discovery install");
let handler = fork
.registration("services/list")
.map(|r| match r.handler {
HandlerKind::Once(h) => h,
_ => panic!("services/list must be Once"),
})
.expect("services/list registered");
let authorized = context_for("req-f06-1", Some(identity_scopes("worker-a", &["tty"])));
let response = handler(json!({}), authorized).await;
let names: Vec<String> = response
.result
.expect("ok")
.get("operations")
.and_then(|v| v.as_array())
.expect("operations array")
.iter()
.filter_map(|o| o.get("name").and_then(|n| n.as_str().map(String::from)))
.collect();
assert!(
names.contains(&"channels/tty/sub".to_string()),
"per-session openable discoverable on the fork: {names:?}"
);
assert!(names.contains(&"base/op".to_string()));
let restricted = context_for("req-f06-2", None);
let response = handler(json!({}), restricted).await;
assert!(response.result.is_ok(), "list itself is callable");
}
}