feat(infra): full-surface integration suite + docs sync + publish prep

Full-surface integration suite (tests/full_surface.rs, mcp feature): - one HttpAdapter over real TCP (ProtocolHandler::handle path) serving gateway endpoints, /openapi.json, /mcp, and the WS channels session - gateway: search/schema/call/subscribe/batch/publish presence, envelope shapes, error fidelity end-to-end - from_openapi import -> Internal-by-default invisible from the wire -> External facade composes it via env.invoke -> upstream HTTP API called end-to-end (ADR-015 composition model exercised) - to_openapi 6-path doc validated against openapiv3 over the wire - to_mcp: MCP client connects to /mcp on the served adapter, lists the 4 gateway tools, search returns ACL-filtered ops (Sub excluded) Production fix: the WS upgrade route was reserved but never wired into HttpAdapter's router (the ws-upgrade-session tests built their own router). Now wired with ws_bearer_auth (401 without a resolvable token) around ws_upgrade_handler. Docs sync: all 28 'Port notes' sections/blockquotes stripped from ported ADRs/specs; OQ-01/OQ-02 statuses corrected to resolved in overview.md, websocket.md, and the README table (open-questions.md was already current). Publish prep: cargo publish --dry-run --allow-dirty succeeds; cargo doc --no-deps warning-free (ADR link targets fixed); feature combinations (default / test-support / mcp / wss / all) compile warning-free under clippy -D warnings. Verified: cargo test (182 lib default), --all-features (227 lib + 29 integration), clippy -D warnings x3 feature sets, fmt, doc, publish --dry-run.
2026-08-28 16:07:56 +00:00
parent bc99ec7188
commit 4a825d33e7
41 changed files with 700 additions and 1035 deletions
@@ -0,0 +1,583 @@
+//! Full-surface integration suite (phase 4): one in-process `HttpAdapter`
+//! serving the gateway endpoints, the `/openapi.json` projection, and the
+//! WS channels session over real I/O; `from_openapi` imported against a
+//! local HTTP echo server; `to_openapi`/`to_mcp` projections consumed
+//! back. Exercises the composition the assembly layer performs.
+
+#![cfg(feature = "mcp")]
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use alkcall::core::auth::{AuthContext, Identity, IdentityProvider};
+use alkcall::core::types::{Capabilities, Connection};
+use alkcall::protocol::wire::{EventEnvelope, ResponseEnvelope};
+use alkcall::registry::discovery::{
+    services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
+};
+use alkcall::registry::registration::{
+    make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
+    OperationRegistry,
+};
+use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
+use alkhttp::adapters::FromOpenAPI;
+use alkhttp::client::HttpClientConfig;
+use alkhttp::server::HttpAdapter;
+use alkhttp::websocket::{frame_channel0_chunk, ChunkAssembler, FrameAssembler, WsClient};
+
+fn identity(id: &str, scopes: &[&str]) -> Identity {
+    Identity {
+        id: id.to_string(),
+        scopes: scopes.iter().map(|s| s.to_string()).collect(),
+        resources: HashMap::new(),
+    }
+}
+
+struct StaticTokens {
+    tokens: std::sync::Mutex<HashMap<String, Identity>>,
+}
+
+impl IdentityProvider for StaticTokens {
+    fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
+        None
+    }
+    fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
+        let s = String::from_utf8_lossy(&token.raw).to_string();
+        self.tokens
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .get(&s)
+            .cloned()
+    }
+}
+
+fn provider_with(tokens: Vec<(&str, Identity)>) -> Arc<dyn IdentityProvider> {
+    let map: HashMap<String, Identity> = tokens
+        .into_iter()
+        .map(|(t, i)| (t.to_string(), i))
+        .collect();
+    Arc::new(StaticTokens {
+        tokens: std::sync::Mutex::new(map),
+    })
+}
+
+/// The local operation registry the `HttpAdapter` serves: an echo op (open
+/// and echo-restricted variants), a streaming sub op, and the discovery
+/// ops the adapters need.
+fn local_registry() -> Arc<OperationRegistry> {
+    let mut inner = OperationRegistry::new();
+    inner
+        .register(HandlerRegistration::new(
+            OperationSpec::new(
+                "echo/run",
+                OperationType::Query,
+                Visibility::External,
+                serde_json::json!({}),
+                serde_json::json!({}),
+                vec![],
+                AccessControl::default(),
+                None,
+            ),
+            HandlerKind::Once(make_handler(|input, ctx| async move {
+                ResponseEnvelope::ok(ctx.request_id, input)
+            })),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ))
+        .unwrap();
+    inner
+        .register(HandlerRegistration::new(
+            OperationSpec::new(
+                "events/tick",
+                OperationType::Sub,
+                Visibility::External,
+                serde_json::json!({}),
+                serde_json::json!({}),
+                vec![],
+                AccessControl::default(),
+                None,
+            ),
+            HandlerKind::Stream(make_streaming_handler(|input, ctx| {
+                futures::stream::iter(vec![
+                    ResponseEnvelope::ok(
+                        ctx.request_id.clone(),
+                        serde_json::json!({ "n": 1, "input": input }),
+                    ),
+                    ResponseEnvelope::ok(
+                        ctx.request_id.clone(),
+                        serde_json::json!({ "n": 2, "input": input }),
+                    ),
+                ])
+            })),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ))
+        .unwrap();
+    let inner = Arc::new(inner);
+
+    let mut registry = OperationRegistry::new();
+    registry
+        .register(HandlerRegistration::new(
+            services_list_spec(),
+            HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ))
+        .unwrap();
+    registry
+        .register(HandlerRegistration::new(
+            services_schema_spec(),
+            HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ))
+        .unwrap();
+    for spec in inner.list_operations() {
+        let name = spec.name.clone();
+        let reg = inner.registration(&name).unwrap();
+        registry
+            .register(HandlerRegistration::new(
+                reg.spec.clone(),
+                reg.handler.clone(),
+                reg.provenance,
+                reg.composition_authority.clone(),
+                reg.scoped_env.clone(),
+                reg.capabilities.clone(),
+            ))
+            .unwrap();
+    }
+    Arc::new(registry)
+}
+
+/// Serve the full adapter surface over a real TCP listener. Each accepted
+/// TCP connection is wrapped as an alkcall `Connection` (single-stream,
+/// `http/1.1` ALPN) and handed to the adapter's `ProtocolHandler::handle`
+/// — the same path a production endpoint drives. Returns the base URL.
+async fn spawn_full_server(
+    registry: Arc<OperationRegistry>,
+    provider: Arc<dyn IdentityProvider>,
+) -> String {
+    let adapter = std::sync::Arc::new(HttpAdapter::new(Arc::clone(&provider), registry));
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        loop {
+            let Ok((sock, _)) = listener.accept().await else {
+                break;
+            };
+            let conn = Connection::from_bidi(sock, b"http/1.1".to_vec(), None);
+            // The bearer middleware resolves the identity per request from
+            // the Authorization header; the transport-level AuthContext
+            // carries no identity (TLS-identity binding is the endpoint's
+            // job, not this test's).
+            let auth = AuthContext::anonymous(b"http/1.1");
+            let a = std::sync::Arc::clone(&adapter);
+            tokio::spawn(async move {
+                let _ =
+                    alkcall::core::types::ProtocolHandler::handle(a.as_ref(), conn, &auth).await;
+            });
+        }
+    });
+    format!("http://{addr}")
+}
+
+#[tokio::test]
+async fn full_surface_gateway_over_http() {
+    // A minimal registry with the discovery ops; the gateway endpoints
+    // must serve search/schema/call/subscribe against it over HTTP.
+    let registry = local_registry();
+    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);
+    let base = spawn_full_server(Arc::clone(&registry), Arc::clone(&provider)).await;
+    let client = reqwest::Client::new();
+
+    // /healthz — no auth.
+    let resp = client.get(format!("{base}/healthz")).send().await.unwrap();
+    assert_eq!(resp.status(), 200);
+
+    // /search — ACL-filtered discovery.
+    let resp = client
+        .get(format!("{base}/search"))
+        .header("Authorization", "Bearer tok-1")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    let body: serde_json::Value = resp.json().await.unwrap();
+    let names: Vec<&str> = body["output"]["operations"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .filter_map(|o| o["name"].as_str())
+        .collect();
+    assert!(names.contains(&"echo/run"), "got {names:?}");
+    assert!(names.contains(&"events/tick"), "got {names:?}");
+
+    // /call — request/response round trip.
+    let resp = client
+        .post(format!("{base}/call"))
+        .header("Authorization", "Bearer tok-1")
+        .json(&serde_json::json!({ "operation": "echo/run", "input": { "v": 42 } }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    let body: serde_json::Value = resp.json().await.unwrap();
+    assert_eq!(body["output"]["v"], 42);
+
+    // /subscribe — SSE stream of the Sub op.
+    let resp = client
+        .post(format!("{base}/subscribe"))
+        .header("Authorization", "Bearer tok-1")
+        .json(&serde_json::json!({ "operation": "events/tick", "input": {} }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    let text = resp.text().await.unwrap();
+    assert!(text.contains("\"n\":1"), "first chunk in SSE: {text}");
+    assert!(text.contains("\"n\":2"), "second chunk in SSE: {text}");
+
+    // /schema — the full spec (GET with a name query param).
+    let resp = client
+        .get(format!("{base}/schema?name=echo%2Frun"))
+        .header("Authorization", "Bearer tok-1")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    let body: serde_json::Value = resp.json().await.unwrap();
+    assert_eq!(body["output"]["name"], "echo/run");
+
+    // /openapi.json — the 6-endpoint projection including /publish.
+    let resp = client
+        .get(format!("{base}/openapi.json"))
+        .header("Authorization", "Bearer tok-1")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    let body: serde_json::Value = resp.json().await.unwrap();
+    assert_eq!(body["info"]["version"], "1.1.0");
+    assert!(body["paths"].get("/publish").is_some());
+    assert!(body["paths"].get("/call").is_some());
+}
+
+#[tokio::test]
+async fn full_surface_ws_call_round_trip() {
+    let registry = local_registry();
+    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);
+    let base = spawn_full_server(registry, provider).await;
+    // WS endpoint rides the same TCP listener.
+    let ws_base = base.replacen("http://", "ws://", 1);
+
+    let mut ws = WsClient::connect_authorized(&format!("{ws_base}/alk/channels"), "tok-1")
+        .await
+        .unwrap();
+    let frame = EventEnvelope::requested(
+        "ws-full-1",
+        serde_json::json!({ "operationId": "echo/run", "input": { "v": 7 } }),
+    );
+    ws.send_binary(frame_channel0_chunk(&frame)).await;
+
+    let mut chunks = ChunkAssembler::new();
+    let mut frames = FrameAssembler::new();
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
+    loop {
+        assert!(tokio::time::Instant::now() < deadline, "timed out");
+        if let Some(env) = frames.next_frame() {
+            assert_eq!(env.r#type, "call.responded");
+            assert_eq!(env.id, "ws-full-1");
+            assert_eq!(env.payload["output"]["v"], 7);
+            break;
+        }
+        let bin = ws.next_binary(std::time::Duration::from_millis(500)).await;
+        match bin {
+            Some(bytes) => {
+                chunks.push(&bytes);
+                while let Some((channel_id, payload)) = chunks.next_chunk() {
+                    assert_eq!(channel_id, 0);
+                    frames.push(&payload);
+                }
+            }
+            None => panic!("ws closed unexpectedly"),
+        }
+    }
+    ws.close().await;
+}
+
+#[tokio::test]
+async fn from_openapi_import_then_gateway_call() {
+    // A local HTTP service the adapter imports; the gateway dispatch then
+    // reaches it through the imported forwarding handler.
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let upstream = format!("http://{addr}");
+    tokio::spawn(async move {
+        let app = axum::Router::new().route(
+            "/widgets",
+            axum::routing::get(|| async {
+                axum::Json(serde_json::json!({ "widgets": ["a", "b"] }))
+            }),
+        );
+        axum::serve(listener, app).await.unwrap();
+    });
+
+    let doc = r#"{"openapi":"3.0.0","info":{"title":"T","version":"1"},"paths":{"/widgets":{"get":{"operationId":"listWidgets","responses":{"200":{"content":{"application/json":{"schema":{}}}}}}}}}"#;
+    let spec = alkhttp::adapters::OpenAPISpec::from_json(doc).unwrap();
+    let config = alkhttp::adapters::HttpServiceConfig {
+        namespace: "upstream".to_string(),
+        base_url: upstream.clone(),
+        auth: None,
+        default_headers: HashMap::new(),
+    };
+    let http_client =
+        Arc::new(alkhttp::client::SharedHttpClient::new(HttpClientConfig::default()).unwrap());
+    let adapter = FromOpenAPI::new(spec, config, http_client);
+    let bundles = alkcall::client::OperationAdapter::import(&adapter)
+        .await
+        .expect("import succeeds");
+    assert_eq!(bundles.len(), 1);
+    assert_eq!(bundles[0].spec.name, "upstream/listWidgets");
+
+    // The imported bundles are Internal (ADR-015 — composition material);
+    // a wire call to them is NOT_FOUND (ADR-015 §2). The assembly layer
+    // composes them under an External facade. Verify Internal-not-callable
+    // through the gateway, then compose the External facade and call that.
+    let mut registry = OperationRegistry::new();
+    for b in bundles {
+        registry.register(b).unwrap();
+    }
+    registry
+        .register(HandlerRegistration::new(
+            OperationSpec::new(
+                "widgets/list",
+                OperationType::Query,
+                Visibility::External,
+                serde_json::json!({}),
+                serde_json::json!({}),
+                vec![],
+                AccessControl::default(),
+                None,
+            ),
+            HandlerKind::Once(make_handler(|_input, ctx| {
+                // The facade composes the imported leaf (env.invoke —
+                // composition-only per ADR-015). Its scoped_env declares
+                // the imported leaf as the reachable set.
+                async move {
+                    let response = ctx
+                        .env
+                        .invoke("upstream", "listWidgets", serde_json::json!({}), &ctx)
+                        .await;
+                    ResponseEnvelope {
+                        request_id: ctx.request_id,
+                        result: response.result,
+                    }
+                }
+            })),
+            OperationProvenance::Local,
+            None,
+            Some(alkcall::registry::context::ScopedPeerEnv::new([
+                "upstream/listWidgets",
+            ])),
+            Capabilities::new(),
+        ))
+        .unwrap();
+    let provider = provider_with(vec![("tok-1", identity("alice", &[]))]);
+    let base = spawn_full_server(Arc::new(registry), provider).await;
+
+    let client = reqwest::Client::new();
+
+    // Internal op from the wire → NOT_FOUND (does not leak existence).
+    let resp = client
+        .post(format!("{base}/call"))
+        .header("Authorization", "Bearer tok-1")
+        .json(&serde_json::json!({ "operation": "upstream/listWidgets", "input": {} }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(
+        resp.status(),
+        404,
+        "Internal imported op is invisible from the wire"
+    );
+
+    // The External facade composes it: external HTTP API → from_openapi
+    // forwarding handler → upstream HTTP API.
+    let resp = client
+        .post(format!("{base}/call"))
+        .header("Authorization", "Bearer tok-1")
+        .json(&serde_json::json!({ "operation": "widgets/list", "input": {} }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200, "facade composes the imported op");
+    let body: serde_json::Value = resp.json().await.unwrap();
+    assert_eq!(body["output"]["widgets"], serde_json::json!(["a", "b"]));
+}
+
+#[tokio::test]
+async fn to_openapi_and_to_mcp_projections_over_served_registry() {
+    use rmcp::model::{CallToolRequestParams, ClientCapabilities, ClientInfo, Implementation};
+    use rmcp::service::RoleClient;
+    use rmcp::transport::streamable_http_client::{
+        StreamableHttpClientTransport, StreamableHttpClientTransportConfig,
+    };
+    use rmcp::{Peer, ServiceExt};
+
+    let registry = local_registry();
+    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);
+
+    // /openapi.json served from the same registry: the to_openapi
+    // projection sees the local ops through services/list.
+    let base = spawn_full_server(Arc::clone(&registry), Arc::clone(&provider)).await;
+    let client = reqwest::Client::new();
+    let resp = client
+        .get(format!("{base}/openapi.json"))
+        .header("Authorization", "Bearer tok-1")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 200);
+    let doc: serde_json::Value = resp.json().await.unwrap();
+    assert_eq!(doc["info"]["title"], "alk gateway");
+    assert_eq!(doc["paths"].as_object().unwrap().len(), 6);
+    // Validates against openapiv3 (ADR-042 contract).
+    let text = serde_json::to_string(&doc).unwrap();
+    let _: openapiv3::OpenAPI = serde_json::from_str(&text).unwrap();
+
+    // /mcp served by the same adapter: an MCP client connects, lists the
+    // 4 gateway tools, calls search — the tool-gateway pattern (ADR-041).
+    let url = format!("{base}/mcp");
+    let transport = StreamableHttpClientTransport::from_config(
+        StreamableHttpClientTransportConfig::with_uri(url),
+    );
+    let client_info = ClientInfo::new(
+        ClientCapabilities::default(),
+        Implementation::new("integration-test", "0.1.0"),
+    );
+    let running = client_info.serve(transport).await.expect("initialize");
+    let peer: Peer<RoleClient> = running.peer().clone();
+
+    let tools = peer
+        .list_tools(Default::default())
+        .await
+        .expect("tools/list");
+    let names: Vec<String> = tools.tools.iter().map(|t| t.name.to_string()).collect();
+    assert_eq!(names.len(), 4);
+    assert!(names.contains(&"search".to_string()));
+    assert!(names.contains(&"schema".to_string()));
+    assert!(names.contains(&"call".to_string()));
+    assert!(names.contains(&"batch".to_string()));
+
+    let mut args = serde_json::Map::new();
+    args.insert("query".to_string(), serde_json::Value::Null);
+    let params = CallToolRequestParams::new("search".to_string()).with_arguments(args);
+    let result = peer.call_tool(params).await.expect("search call");
+    assert_eq!(result.is_error, Some(false));
+    let structured = result.structured_content.expect("structured present");
+    let ops = structured
+        .get("operations")
+        .and_then(serde_json::Value::as_array)
+        .expect("operations array");
+    let names: Vec<&str> = ops
+        .iter()
+        .filter_map(|o| o.get("name").and_then(|v| v.as_str()))
+        .collect();
+    assert!(names.contains(&"echo/run"), "got {names:?}");
+    assert!(
+        !names.contains(&"events/tick"),
+        "Sub ops excluded from search"
+    );
+}
+
+#[tokio::test]
+async fn gateway_error_fidelity_end_to_end() {
+    // Unknown op through /call → 404 NOT_FOUND; internal op → 404.
+    let registry = local_registry();
+    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);
+    let base = spawn_full_server(registry, provider).await;
+    let client = reqwest::Client::new();
+
+    let resp = client
+        .post(format!("{base}/call"))
+        .header("Authorization", "Bearer tok-1")
+        .json(&serde_json::json!({ "operation": "no/such", "input": {} }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), 404);
+    let body: serde_json::Value = resp.json().await.unwrap();
+    assert_eq!(body["code"], "NOT_FOUND");
+}
+
+use http::header::AUTHORIZATION;
+
+#[tokio::test]
+async fn gateway_endpoints_exist_with_bearer_enforcement() {
+    let registry = local_registry();
+    let provider = provider_with(vec![("tok-1", identity("alice", &["user"]))]);
+    let base = spawn_full_server(registry, provider).await;
+    let client = reqwest::Client::new();
+
+    // GET endpoints: /search, /schema (schema needs a known name).
+    let resp = client
+        .get(format!("{base}/search"))
+        .header(AUTHORIZATION, "Bearer tok-1")
+        .send()
+        .await
+        .unwrap();
+    assert_ne!(
+        resp.status(),
+        404,
+        "/search must exist on the gateway surface"
+    );
+    let resp = client
+        .get(format!("{base}/schema?name=echo/run"))
+        .header(AUTHORIZATION, "Bearer tok-1")
+        .send()
+        .await
+        .unwrap();
+    assert_ne!(
+        resp.status(),
+        404,
+        "/schema must exist on the gateway surface"
+    );
+    for path in ["/call", "/batch", "/subscribe", "/publish"] {
+        let resp = client
+            .post(format!("{base}{path}"))
+            .header(AUTHORIZATION, "Bearer tok-1")
+            .header("Content-Type", "application/json")
+            .body("{}")
+            .send()
+            .await
+            .unwrap();
+        assert_ne!(
+            resp.status(),
+            404,
+            "{path} must exist on the gateway surface"
+        );
+    }
+
+    // Unauthenticated call to an op with no restrictions: allowed
+    // (AccessControl::default() passes for any identity, including none).
+    let resp = client
+        .post(format!("{base}/call"))
+        .header("Content-Type", "application/json")
+        .body(r#"{"operation": "no/such", "input": {}}"#)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(
+        resp.status(),
+        404,
+        "unknown op is NOT_FOUND regardless of auth"
+    );
+}