feat: server foundation (phase 1 core) — state, auth, healthz/decoy, gateway dispatch, HttpAdapter

Tasks completed: server-core-types, server-auth, server-healthz-decoy, gateway-dispatch, server-adapter (5 of 17). - src/server/state.rs: DecoyConfig + RouterState (alkcall type paths, 6-endpoint reserved-path docs) - src/server/auth.rs: bearer middleware + ResolvedIdentity extractor (10 tests: missing/malformed/basic/failed-resolution matrix) - src/server/healthz.rs + decoy.rs: raw healthz; nginx-style 404, static site (path-traversal guarded), redirect decoys - src/gateway/dispatch.rs: GatewayDispatch invoke/invoke_streaming (internal:false, forwarded_for:None, bounded deadline) + src/gateway/error.rs: CallError→HTTP status mapping (HTTP_<status> passthrough, retryable→Retry-After) - src/server/adapter.rs: HttpAdapter ProtocolHandler — accept_bi → BiStream → TokioIo → hyper auto builder (h2 CONNECT enabled); integration tests over DuplexStream (request/response cycle, healthz, decoy 404) Verified: cargo test (46 lib tests), clippy -D warnings, fmt, test --all-features.
2026-08-28 07:35:02 +00:00
parent a85500d3d9
commit d070e548ad
15 changed files with 1669 additions and 31 deletions
@@ -0,0 +1,289 @@
+//! Shared dispatch spine for the `to_openapi` / `to_mcp` gateway
+//! projections.
+//!
+//! Thin concrete struct (not a trait — the alknet research ruled out a
+//! trait with an associated output type). Holds `Arc<OperationRegistry>`
+//! and `Arc<dyn IdentityProvider>` and exposes a `resolve_bearer()` and
+//! `invoke()` method pair returning the neutral `ResponseEnvelope`. Each
+//! gateway maps the envelope to its own wire shape (`to_openapi` → HTTP
+//! `Response`, `to_mcp` → `CallToolResult`).
+//!
+//! # Security invariants
+//!
+//! - `internal: false` — ACL runs against the caller's `identity`, not a
+//!   handler's composition authority (alkcall ADR-017).
+//! - `forwarded_for: None` — wire-ingress only.
+//!
+//! The root `OperationContext` is constructed identically for both
+//! gateways, making them provably identical on the security axis (auth,
+//! authority, ACL); they diverge only on wire-framing.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use alkcall::core::auth::{AuthToken, Identity, IdentityProvider};
+use alkcall::core::types::Capabilities;
+use alkcall::protocol::wire::ResponseEnvelope;
+use alkcall::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
+use alkcall::registry::env::LocalOperationEnv;
+use alkcall::registry::registration::OperationRegistry;
+use futures::stream::BoxStream;
+use serde_json::Value;
+
+const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
+
+pub struct GatewayDispatch {
+    registry: Arc<OperationRegistry>,
+    identity_provider: Arc<dyn IdentityProvider>,
+}
+
+impl GatewayDispatch {
+    pub fn new(
+        registry: Arc<OperationRegistry>,
+        identity_provider: Arc<dyn IdentityProvider>,
+    ) -> Self {
+        Self {
+            registry,
+            identity_provider,
+        }
+    }
+
+    pub fn registry(&self) -> &Arc<OperationRegistry> {
+        &self.registry
+    }
+
+    pub fn identity_provider(&self) -> &Arc<dyn IdentityProvider> {
+        &self.identity_provider
+    }
+
+    pub fn resolve_bearer(&self, token: &AuthToken) -> Option<Identity> {
+        self.identity_provider.resolve_from_token(token)
+    }
+
+    pub async fn invoke(
+        &self,
+        identity: Option<Identity>,
+        op: &str,
+        input: Value,
+    ) -> ResponseEnvelope {
+        let operation_name = strip_leading_slash(op).to_string();
+        let request_id = uuid::Uuid::new_v4().to_string();
+        let context = self.build_root_context(&request_id, &operation_name, identity);
+        self.registry.invoke(&operation_name, input, context).await
+    }
+
+    pub fn invoke_streaming(
+        &self,
+        identity: Option<Identity>,
+        op: &str,
+        input: Value,
+    ) -> BoxStream<'static, ResponseEnvelope> {
+        let operation_name = strip_leading_slash(op).to_string();
+        let request_id = uuid::Uuid::new_v4().to_string();
+        let context = self.build_root_context_streaming(&request_id, &operation_name, identity);
+        self.registry
+            .invoke_streaming(&operation_name, input, context)
+    }
+
+    fn build_root_context(
+        &self,
+        request_id: &str,
+        operation_name: &str,
+        identity: Option<Identity>,
+    ) -> OperationContext {
+        self.build_root_context_inner(request_id, operation_name, identity, true)
+    }
+
+    fn build_root_context_streaming(
+        &self,
+        request_id: &str,
+        operation_name: &str,
+        identity: Option<Identity>,
+    ) -> OperationContext {
+        self.build_root_context_inner(request_id, operation_name, identity, false)
+    }
+
+    fn build_root_context_inner(
+        &self,
+        request_id: &str,
+        operation_name: &str,
+        identity: Option<Identity>,
+        bounded: bool,
+    ) -> OperationContext {
+        let registration = self.registry.registration(operation_name);
+        let (composition_authority, capabilities, scoped_env) = match registration {
+            Some(r) => (
+                r.composition_authority.clone(),
+                r.capabilities.clone(),
+                r.scoped_env.clone().unwrap_or_else(ScopedPeerEnv::empty),
+            ),
+            None => (None, Capabilities::new(), ScopedPeerEnv::empty()),
+        };
+
+        let env: Arc<dyn alkcall::registry::env::OperationEnv + Send + Sync> =
+            Arc::new(LocalOperationEnv::new(Arc::clone(&self.registry)));
+
+        OperationContext {
+            request_id: request_id.to_string(),
+            parent_request_id: None,
+            identity,
+            handler_identity: composition_authority,
+            forwarded_for: None,
+            capabilities,
+            metadata: HashMap::new(),
+            deadline: bounded.then(|| Instant::now() + DEFAULT_TIMEOUT),
+            scoped_env,
+            env,
+            abort_policy: AbortPolicy::default(),
+            internal: false,
+            ownership: None,
+        }
+    }
+}
+
+fn strip_leading_slash(operation_id: &str) -> &str {
+    operation_id.strip_prefix('/').unwrap_or(operation_id)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alkcall::registry::registration::{
+        make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+    };
+    use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
+    use futures::StreamExt;
+    use std::sync::Mutex as StdMutex;
+
+    struct StaticIdentityProvider {
+        tokens: StdMutex<HashMap<String, Identity>>,
+    }
+
+    impl StaticIdentityProvider {
+        fn new() -> Self {
+            Self {
+                tokens: StdMutex::new(HashMap::new()),
+            }
+        }
+    }
+
+    impl IdentityProvider for StaticIdentityProvider {
+        fn resolve_from_fingerprint(&self, _fp: &str) -> Option<Identity> {
+            None
+        }
+        fn resolve_from_token(&self, token: &AuthToken) -> Option<Identity> {
+            let token_str = String::from_utf8_lossy(&token.raw);
+            self.tokens.lock().unwrap().get(token_str.as_ref()).cloned()
+        }
+    }
+
+    fn spec(name: &str, visibility: Visibility, op_type: OperationType) -> OperationSpec {
+        OperationSpec::new(
+            name,
+            op_type,
+            visibility,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            AccessControl::default(),
+            None,
+        )
+    }
+
+    fn echo_registry() -> Arc<OperationRegistry> {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec("echo/run", Visibility::External, OperationType::Query),
+                HandlerKind::Once(make_handler(|input, ctx| async move {
+                    ResponseEnvelope::ok(ctx.request_id, input)
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        Arc::new(registry)
+    }
+
+    #[tokio::test]
+    async fn invoke_external_op_round_trips() {
+        let dispatch =
+            GatewayDispatch::new(echo_registry(), Arc::new(StaticIdentityProvider::new()));
+        let envelope = dispatch
+            .invoke(None, "/echo/run", serde_json::json!({ "x": 1 }))
+            .await;
+        assert!(envelope.result.is_ok(), "expected ok, got {envelope:?}");
+    }
+
+    #[tokio::test]
+    async fn invoke_unknown_op_returns_not_found() {
+        let dispatch =
+            GatewayDispatch::new(echo_registry(), Arc::new(StaticIdentityProvider::new()));
+        let envelope = dispatch
+            .invoke(None, "/missing/op", serde_json::json!({}))
+            .await;
+        match envelope.result {
+            Err(error) => assert_eq!(error.code, "NOT_FOUND"),
+            Ok(v) => panic!("expected error, got {v:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_internal_op_returns_not_found() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec("internal/op", Visibility::Internal, OperationType::Query),
+                HandlerKind::Once(make_handler(|_input, ctx| async move {
+                    ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let dispatch =
+            GatewayDispatch::new(Arc::new(registry), Arc::new(StaticIdentityProvider::new()));
+        let envelope = dispatch
+            .invoke(None, "/internal/op", serde_json::json!({}))
+            .await;
+        match envelope.result {
+            Err(error) => assert_eq!(error.code, "NOT_FOUND"),
+            Ok(v) => panic!("expected error, got {v:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn streaming_sub_op_streams_envelopes() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                spec("tick/stream", Visibility::External, OperationType::Sub),
+                HandlerKind::Stream(make_streaming_handler(|input, ctx| {
+                    let count = input.get("count").and_then(|v| v.as_u64()).unwrap_or(2);
+                    let request_id = ctx.request_id.clone();
+                    futures::stream::iter(0..count).map(move |i| {
+                        ResponseEnvelope::ok(request_id.clone(), serde_json::json!({ "tick": i }))
+                    })
+                })),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let dispatch =
+            GatewayDispatch::new(Arc::new(registry), Arc::new(StaticIdentityProvider::new()));
+        let mut stream =
+            dispatch.invoke_streaming(None, "/tick/stream", serde_json::json!({ "count": 3 }));
+        let mut ticks = Vec::new();
+        while let Some(envelope) = stream.next().await {
+            ticks.push(envelope);
+        }
+        assert_eq!(ticks.len(), 3);
+    }
+}