diff --git a/src/adapters/forward.rs b/src/adapters/forward.rs new file mode 100644 index 0000000..93d2a4b --- /dev/null +++ b/src/adapters/forward.rs @@ -0,0 +1,442 @@ +//! Shared HTTP forwarding core for the HTTP-backed adapters +//! (`from_openapi`, `from_jsonschema`): request construction +//! (path templates, query split, body, credential injection), the +//! reqwest forwarding handlers (`forward` / `forward_stream`), and the +//! SSE frame parser for streaming (SSE → `ResponseEnvelope`) handlers +//! (ADR-049). +//! +//! The forwarding handler is the no-env-vars credential injection point +//! (ADR-014): it reads `OperationContext.capabilities`, never +//! `std::env::var`. Imported error codes are `HTTP_` to avoid +//! collision with the protocol-level codes (ADR-023). + +use std::collections::HashMap; +use std::sync::Arc; + +use alkcall::protocol::wire::{CallError, ResponseEnvelope}; +use alkcall::registry::context::OperationContext; +use alkcall::registry::registration::ResponseStream; +use alkcall::registry::spec::OperationType; +use futures::stream; +use futures::StreamExt; +use reqwest::header::{HeaderMap, HeaderName, HeaderValue, ACCEPT, AUTHORIZATION, CONTENT_TYPE}; +use reqwest::Method; +use serde_json::Value; +use url::Url; + +use crate::client::SharedHttpClient; + +#[derive(Clone)] +pub enum HttpAuthScheme { + Bearer, + ApiKey { header_name: String }, + Basic, +} + +pub struct HttpServiceConfig { + pub namespace: String, + pub base_url: String, + pub auth: Option, + pub default_headers: HashMap, +} + +#[allow(clippy::too_many_arguments)] +pub(crate) fn build_request( + base_url: &str, + path_template: &str, + method: &str, + auth_scheme: &Option, + default_headers: &HashMap, + namespace: &str, + input: &Value, + context: &OperationContext, +) -> Result<(Method, Url, Option, HeaderMap), CallError> { + let input_obj = input.as_object(); + + let mut url_path = path_template.to_string(); + let mut query_params: Vec<(String, String)> = Vec::new(); + let mut body: Option = None; + + if let Some(obj) = input_obj { + for (key, value) in obj { + let placeholder = format!("{{{key}}}"); + if url_path.contains(&placeholder) { + let rendered = value_to_path_segment(value); + url_path = url_path.replace(&placeholder, &rendered); + } else if key == "body" { + body = Some(value.clone()); + } else { + query_params.push((key.clone(), value_to_query(value))); + } + } + } + + let base = Url::parse(base_url) + .map_err(|e| CallError::internal(format!("invalid base_url `{base_url}`: {e}")))?; + let mut url = base + .join(url_path.trim_start_matches('/')) + .map_err(|e| CallError::internal(format!("invalid path `{url_path}`: {e}")))?; + if !query_params.is_empty() { + let mut pairs = url.query_pairs_mut(); + for (k, v) in &query_params { + pairs.append_pair(k, v); + } + } + + let mut headers = HeaderMap::new(); + for (k, v) in default_headers { + if let (Ok(name), Ok(value)) = ( + HeaderName::try_from(k.as_str()), + HeaderValue::try_from(v.as_str()), + ) { + headers.insert(name, value); + } + } + + if body.is_some() { + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + } + + if let Some(scheme) = auth_scheme { + if let Some(secret) = context.capabilities.get(namespace) { + let credential = secret.expose_secret().clone(); + match scheme { + HttpAuthScheme::Bearer => { + let header_value = format!("Bearer {credential}"); + if let Ok(value) = HeaderValue::try_from(header_value) { + headers.insert(AUTHORIZATION, value); + } + } + HttpAuthScheme::ApiKey { header_name } => { + if let (Ok(name), Ok(value)) = ( + HeaderName::try_from(header_name.as_str()), + HeaderValue::try_from(credential.as_str()), + ) { + headers.insert(name, value); + } + } + HttpAuthScheme::Basic => { + let header_value = format!("Basic {credential}"); + if let Ok(value) = HeaderValue::try_from(header_value) { + headers.insert(AUTHORIZATION, value); + } + } + } + } + } + + let http_method = Method::from_bytes(method.as_bytes()) + .map_err(|_| CallError::internal(format!("invalid HTTP method `{method}`")))?; + Ok((http_method, url, body, headers)) +} + +pub(crate) fn value_to_path_segment(value: &Value) -> String { + match value { + Value::String(s) => s.clone(), + Value::Number(n) => n.to_string(), + Value::Bool(b) => b.to_string(), + Value::Null => String::new(), + other => other.to_string(), + } +} + +pub(crate) fn value_to_query(value: &Value) -> String { + match value { + Value::String(s) => s.clone(), + Value::Number(n) => n.to_string(), + Value::Bool(b) => b.to_string(), + Value::Null => String::new(), + other => other.to_string(), + } +} + +#[allow(clippy::too_many_arguments)] +pub(crate) async fn forward( + http_client: &Arc, + base_url: &str, + path_template: &str, + method: &str, + auth_scheme: &Option, + default_headers: &HashMap, + namespace: &str, + error_status_codes: &[(u16, String)], + op_type: OperationType, + input: Value, + context: OperationContext, +) -> ResponseEnvelope { + let request_id = context.request_id.clone(); + + let (http_method, url, body, headers) = match build_request( + base_url, + path_template, + method, + auth_scheme, + default_headers, + namespace, + &input, + &context, + ) { + Ok(parts) => parts, + Err(err) => return ResponseEnvelope::error(request_id, err), + }; + + let http_client = http_client.client(); + + let request_builder = http_client + .request(http_method, url.as_str()) + .headers(headers); + + let request_builder = if op_type == OperationType::Sub { + request_builder.header(ACCEPT, "text/event-stream") + } else { + request_builder.header(ACCEPT, "*/*") + }; + + let request_builder = match body.as_ref() { + Some(b) => { + let serialized = serde_json::to_string(b).unwrap_or_else(|_| String::from("null")); + request_builder.body(serialized) + } + None => request_builder, + }; + + let response: reqwest::Response = match request_builder.send().await { + Ok(r) => r, + Err(err) => { + return ResponseEnvelope::error( + request_id, + CallError::internal(format!("HTTP request failed: {err}")), + ); + } + }; + + let status = response.status(); + + if !status.is_success() { + let code = error_status_codes + .iter() + .find(|(s, _)| *s == status.as_u16()) + .map(|(_, code)| code.clone()) + .unwrap_or_else(|| format!("HTTP_{}", status.as_u16())); + let message = format!( + "HTTP {}: {}", + status.as_u16(), + status.canonical_reason().unwrap_or("") + ); + return ResponseEnvelope::error(request_id, CallError::new(code, message, false)); + } + + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|v: &reqwest::header::HeaderValue| v.to_str().ok()) + .unwrap_or("") + .to_string(); + + if content_type.contains("application/json") { + match response.json::().await { + Ok(v) => ResponseEnvelope::ok(request_id, v), + Err(err) => ResponseEnvelope::error( + request_id, + CallError::internal(format!("failed to decode JSON body: {err}")), + ), + } + } else if content_type.starts_with("text/") { + match response.text().await { + Ok(t) => ResponseEnvelope::ok(request_id, Value::String(t)), + Err(err) => ResponseEnvelope::error( + request_id, + CallError::internal(format!("failed to decode text body: {err}")), + ), + } + } else { + match response.bytes().await { + Ok(b) => { + let arr: Vec = b.iter().map(|byte| Value::Number((*byte).into())).collect(); + ResponseEnvelope::ok(request_id, Value::Array(arr)) + } + Err(err) => ResponseEnvelope::error( + request_id, + CallError::internal(format!("failed to read body: {err}")), + ), + } + } +} + +#[allow(clippy::too_many_arguments)] +pub(crate) fn forward_stream( + http_client: &Arc, + base_url: &str, + path_template: &str, + method: &str, + auth_scheme: &Option, + default_headers: &HashMap, + namespace: &str, + error_status_codes: &[(u16, String)], + input: Value, + context: OperationContext, +) -> ResponseStream { + let request_id = context.request_id.clone(); + + let (http_method, url, body, headers) = match build_request( + base_url, + path_template, + method, + auth_scheme, + default_headers, + namespace, + &input, + &context, + ) { + Ok(parts) => parts, + Err(err) => { + return Box::pin(stream::once(async move { + ResponseEnvelope::error(request_id, err) + })); + } + }; + + let http_client = Arc::clone(http_client); + let error_status_codes = error_status_codes.to_vec(); + + let request_id_stream = request_id.clone(); + let error_status_codes_stream = error_status_codes.clone(); + + let init = async move { + let request_builder = http_client + .client() + .request(http_method, url.as_str()) + .headers(headers) + .header(ACCEPT, "text/event-stream"); + let request_builder = match body.as_ref() { + Some(b) => { + let serialized = serde_json::to_string(b).unwrap_or_else(|_| String::from("null")); + request_builder.body(serialized) + } + None => request_builder, + }; + request_builder.send().await + }; + + let sse = stream::once(init).flat_map(move |result| { + let request_id = request_id_stream.clone(); + let error_status_codes = error_status_codes_stream.clone(); + match result { + Err(err) => Box::pin(stream::once(async move { + ResponseEnvelope::error( + request_id, + CallError::internal(format!("HTTP request failed: {err}")), + ) + })) as ResponseStream, + Ok(response) => { + let status = response.status(); + if !status.is_success() { + let code = error_status_codes + .iter() + .find(|(s, _)| *s == status.as_u16()) + .map(|(_, c)| c.clone()) + .unwrap_or_else(|| format!("HTTP_{}", status.as_u16())); + let message = format!( + "HTTP {}: {}", + status.as_u16(), + status.canonical_reason().unwrap_or("") + ); + Box::pin(stream::once(async move { + ResponseEnvelope::error(request_id, CallError::new(code, message, false)) + })) as ResponseStream + } else { + let request_id_inner = request_id.clone(); + Box::pin( + stream::unfold( + (response.bytes_stream(), String::new()), + move |(mut bytes, mut buffer)| { + let request_id = request_id_inner.clone(); + async move { + match bytes.next().await { + Some(Ok(chunk)) => { + buffer.push_str(&String::from_utf8_lossy(&chunk)); + let (events, remaining) = parse_sse_frames(&buffer); + let envelopes: Vec = events + .into_iter() + .map(|e| { + let parsed = if e.data.trim().is_empty() { + Value::Null + } else { + serde_json::from_str(&e.data).unwrap_or( + Value::String(e.data.clone()), + ) + }; + ResponseEnvelope::ok(&request_id, parsed) + }) + .collect(); + Some((envelopes, (bytes, remaining))) + } + Some(Err(err)) => { + let error = CallError::internal(format!( + "SSE stream error: {err}" + )); + Some(( + vec![ResponseEnvelope::error(request_id, error)], + (bytes, buffer), + )) + } + None => None, + } + } + }, + ) + .flat_map(stream::iter), + ) as ResponseStream + } + } + } + }); + + Box::pin(sse) +} + +pub(crate) struct SseEvent { + pub(crate) data: String, +} + +pub(crate) fn parse_sse_frames(buffer: &str) -> (Vec, String) { + let mut events = Vec::new(); + let text = if let Some(stripped) = buffer.strip_prefix('\u{feff}') { + stripped + } else { + buffer + }; + let lines: Vec<&str> = text.split('\n').collect(); + let mut data_buffer: Vec = Vec::new(); + let mut remaining = String::new(); + + for (i, line) in lines.iter().enumerate() { + if i == lines.len() - 1 { + remaining = line.to_string(); + break; + } + let line = line.strip_suffix('\r').unwrap_or(line); + if line.is_empty() { + if !data_buffer.is_empty() { + events.push(SseEvent { + data: data_buffer.join("\n"), + }); + } + data_buffer.clear(); + continue; + } + if line.starts_with(':') { + continue; + } + if let Some((field, value)) = line.split_once(':') { + let value = value.strip_prefix(' ').unwrap_or(value); + if field == "data" { + data_buffer.push(value.to_string()); + } + } else if line == "data" { + data_buffer.push(String::new()); + } + } + + (events, remaining) +} diff --git a/src/adapters/from_jsonschema.rs b/src/adapters/from_jsonschema.rs new file mode 100644 index 0000000..70b785e --- /dev/null +++ b/src/adapters/from_jsonschema.rs @@ -0,0 +1,493 @@ +//! `from_jsonschema` adapter: register a single HTTP-backed operation from a +//! caller-supplied [`OperationSpec`], path template, and HTTP method +//! ([ADR-066]). +//! +//! One `HandlerRegistration` per call with a reqwest forwarding handler +//! (the shared [`super::forward`] code path with `from_openapi`). +//! Provenance is `FromJsonSchema` (leaf, `composition_authority: None`, +//! `scoped_env: None`, `Internal` by default — ADR-015/022). `Sub` op +//! type → `HandlerKind::Stream` expecting `text/event-stream` (ADR-049). +//! +//! [ADR-066]: crate::docs + +use std::sync::Arc; + +use alkcall::client::{AdapterError, OperationAdapter}; +use alkcall::core::types::Capabilities; +use alkcall::registry::context::OperationContext; +use alkcall::registry::registration::{ + make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance, +}; +use alkcall::registry::spec::{OperationSpec, OperationType}; +use async_trait::async_trait; +use serde_json::Value; + +use super::forward::{forward, forward_stream, HttpServiceConfig}; +use crate::client::SharedHttpClient; + +pub struct FromJsonSchema { + spec: OperationSpec, + config: HttpServiceConfig, + path_template: String, + method: String, + http_client: Arc, +} + +impl FromJsonSchema { + pub fn new( + spec: OperationSpec, + config: HttpServiceConfig, + path_template: String, + method: String, + http_client: Arc, + ) -> Self { + Self { + spec, + config, + path_template, + method, + http_client, + } + } +} + +#[async_trait] +impl OperationAdapter for FromJsonSchema { + async fn import(&self) -> Result, AdapterError> { + let path_template = self.path_template.clone(); + let method_upper = self.method.to_ascii_uppercase(); + let auth_scheme = self.config.auth.clone(); + let default_headers = self.config.default_headers.clone(); + let base_url = self.config.base_url.clone(); + let namespace = self.config.namespace.clone(); + let http_client = Arc::clone(&self.http_client); + let op_type = self.spec.op_type; + + let error_status_codes: Vec<(u16, String)> = self + .spec + .error_schemas + .iter() + .map(|e| (e.http_status.unwrap_or(0), e.code.clone())) + .collect(); + + let handler = if op_type == OperationType::Sub { + let stream_handler = + make_streaming_handler(move |input: Value, context: OperationContext| { + let path_template = path_template.clone(); + let method_upper = method_upper.clone(); + let auth_scheme = auth_scheme.clone(); + let default_headers = default_headers.clone(); + let base_url = base_url.clone(); + let namespace = namespace.clone(); + let http_client = Arc::clone(&http_client); + let error_status_codes = error_status_codes.clone(); + forward_stream( + &http_client, + &base_url, + &path_template, + &method_upper, + &auth_scheme, + &default_headers, + &namespace, + &error_status_codes, + input, + context, + ) + }); + HandlerKind::Stream(stream_handler) + } else { + let once_handler = make_handler(move |input: Value, context: OperationContext| { + let path_template = path_template.clone(); + let method_upper = method_upper.clone(); + let auth_scheme = auth_scheme.clone(); + let default_headers = default_headers.clone(); + let base_url = base_url.clone(); + let namespace = namespace.clone(); + let http_client = Arc::clone(&http_client); + let error_status_codes = error_status_codes.clone(); + let op_type = op_type; + async move { + forward( + &http_client, + &base_url, + &path_template, + &method_upper, + &auth_scheme, + &default_headers, + &namespace, + &error_status_codes, + op_type, + input, + context, + ) + .await + } + }); + HandlerKind::Once(once_handler) + }; + + let capabilities = Capabilities::new(); + Ok(vec![HandlerRegistration::new( + self.spec.clone(), + handler, + OperationProvenance::FromJsonSchema, + None, + None, + capabilities, + )]) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::adapters::forward::{build_request, HttpAuthScheme}; + use crate::client::HttpClientConfig; + use alkcall::protocol::wire::ResponseEnvelope; + use alkcall::registry::context::AbortPolicy; + use alkcall::registry::env::OperationEnv; + use alkcall::registry::spec::{AccessControl, ErrorDefinition, Visibility}; + use futures::StreamExt; + use reqwest::header::AUTHORIZATION; + use reqwest::Method; + use std::collections::HashMap; + use std::time::Duration; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + fn noop_context(request_id: &str, capabilities: Capabilities) -> OperationContext { + struct NoopEnv; + #[async_trait] + impl OperationEnv for NoopEnv { + async fn invoke_with_policy( + &self, + _ns: &str, + _op: &str, + _input: Value, + parent: &OperationContext, + _policy: AbortPolicy, + ) -> ResponseEnvelope { + ResponseEnvelope::ok(parent.request_id.clone(), Value::Null) + } + fn contains(&self, _name: &str) -> bool { + false + } + } + OperationContext { + request_id: request_id.to_string(), + parent_request_id: None, + identity: None, + handler_identity: None, + forwarded_for: None, + capabilities, + metadata: HashMap::new(), + scoped_env: alkcall::registry::context::ScopedPeerEnv::empty(), + env: Arc::new(NoopEnv), + abort_policy: AbortPolicy::default(), + deadline: Some(std::time::Instant::now() + Duration::from_secs(30)), + internal: true, + ownership: None, + } + } + + fn test_spec(name: &str, op_type: OperationType) -> OperationSpec { + OperationSpec::new( + name, + op_type, + Visibility::Internal, + serde_json::json!({"type":"object","properties":{"id":{"type":"string"}}}), + serde_json::json!({"type":"object"}), + vec![], + AccessControl::default(), + None, + ) + } + + fn test_spec_with_errors( + name: &str, + op_type: OperationType, + errors: Vec, + ) -> OperationSpec { + OperationSpec::new( + name, + op_type, + Visibility::Internal, + serde_json::json!({"type":"object"}), + serde_json::json!({"type":"object"}), + errors, + AccessControl::default(), + None, + ) + } + + fn test_config(namespace: &str, base_url: &str) -> HttpServiceConfig { + HttpServiceConfig { + namespace: namespace.to_string(), + base_url: base_url.to_string(), + auth: None, + default_headers: HashMap::new(), + } + } + + fn test_http_client() -> Arc { + Arc::new(SharedHttpClient::new(HttpClientConfig::default()).unwrap()) + } + + #[tokio::test] + async fn import_produces_one_handler_registration() { + let adapter = FromJsonSchema::new( + test_spec("svc/getWidget", OperationType::Query), + test_config("svc", "https://api.example.com"), + "/widgets".to_string(), + "GET".to_string(), + test_http_client(), + ); + let bundles = adapter.import().await.unwrap(); + assert_eq!(bundles.len(), 1); + assert_eq!(bundles[0].spec.name, "svc/getWidget"); + assert_eq!(bundles[0].provenance, OperationProvenance::FromJsonSchema); + assert!(bundles[0].composition_authority.is_none()); + assert!(bundles[0].scoped_env.is_none()); + } + + #[tokio::test] + async fn query_op_registration_is_handler_kind_once() { + let adapter = FromJsonSchema::new( + test_spec("svc/getWidget", OperationType::Query), + test_config("svc", "https://api.example.com"), + "/widgets".to_string(), + "GET".to_string(), + test_http_client(), + ); + let bundles = adapter.import().await.unwrap(); + assert!(matches!(bundles[0].handler, HandlerKind::Once(_))); + } + + #[tokio::test] + async fn sub_op_registration_is_handler_kind_stream() { + let adapter = FromJsonSchema::new( + test_spec("svc/stream", OperationType::Sub), + test_config("svc", "https://api.example.com"), + "/stream".to_string(), + "POST".to_string(), + test_http_client(), + ); + let bundles = adapter.import().await.unwrap(); + assert!(matches!(bundles[0].handler, HandlerKind::Stream(_))); + } + + #[tokio::test] + async fn mutation_op_registration_is_handler_kind_once() { + let adapter = FromJsonSchema::new( + test_spec("svc/createWidget", OperationType::Mutation), + test_config("svc", "https://api.example.com"), + "/widgets".to_string(), + "POST".to_string(), + test_http_client(), + ); + let bundles = adapter.import().await.unwrap(); + assert!(matches!(bundles[0].handler, HandlerKind::Once(_))); + } + + #[tokio::test] + async fn build_request_injects_bearer_from_capabilities() { + let caps = Capabilities::new().with_http_token("github", "tok-123".to_string()); + let ctx = noop_context("req-1", caps); + let (method, url, _body, headers) = build_request( + "https://api.github.com", + "/repos/{owner}/{repo}/issues", + "GET", + &Some(HttpAuthScheme::Bearer), + &HashMap::new(), + "github", + &serde_json::json!({"owner":"a","repo":"b"}), + &ctx, + ) + .unwrap(); + assert_eq!(method, Method::GET); + assert_eq!(url.path(), "/repos/a/b/issues"); + assert_eq!(url.host_str(), Some("api.github.com")); + let auth = headers.get(AUTHORIZATION).unwrap(); + assert_eq!(auth.to_str().unwrap(), "Bearer tok-123"); + } + + #[tokio::test] + async fn build_request_path_and_query_split() { + let ctx = noop_context("req-2", Capabilities::new()); + let (_, url, _, _) = build_request( + "https://api.example.com", + "/widgets/{id}", + "GET", + &None, + &HashMap::new(), + "svc", + &serde_json::json!({"id":42,"filter":"active"}), + &ctx, + ) + .unwrap(); + assert_eq!(url.path(), "/widgets/42"); + assert_eq!(url.query().unwrap(), "filter=active"); + } + + async fn spawn_echo_server( + status: u16, + body: &'static str, + content_type: &'static str, + ) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + loop { + let (mut sock, _) = match listener.accept().await { + Ok(pair) => pair, + Err(_) => break, + }; + let status_line = match status { + 200 => "200 OK", + 201 => "201 Created", + 404 => "404 Not Found", + 500 => "500 Internal Server Error", + _ => "200 OK", + }; + let body_bytes = body.as_bytes(); + let response = format!( + "HTTP/1.1 {status_line}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body_bytes.len(), + body + ); + let mut buf = [0u8; 4096]; + let _ = sock.read(&mut buf).await; + sock.write_all(response.as_bytes()).await.unwrap(); + sock.flush().await.unwrap(); + } + }); + format!("http://{addr}") + } + + #[tokio::test] + async fn integration_forwarding_handler_calls_external_endpoint() { + let base = spawn_echo_server(200, r#"{"ok":true}"#, "application/json").await; + let adapter = FromJsonSchema::new( + test_spec("svc/data", OperationType::Query), + test_config("svc", &base), + "/data".to_string(), + "GET".to_string(), + test_http_client(), + ); + let bundles = adapter.import().await.unwrap(); + let registration = &bundles[0]; + let ctx = noop_context("req-10", Capabilities::new()); + let response = match ®istration.handler { + HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await, + _ => panic!("expected Once handler"), + }; + assert_eq!(response.request_id, "req-10"); + match response.result { + Ok(v) => assert_eq!(v, serde_json::json!({"ok":true})), + Err(e) => panic!("expected Ok, got {e:?}"), + } + } + + #[tokio::test] + async fn integration_non_2xx_returns_declared_error() { + let base = spawn_echo_server(404, r#"{"error":"missing"}"#, "application/json").await; + let errors = vec![ErrorDefinition { + code: "HTTP_404".to_string(), + description: "Not found".to_string(), + schema: serde_json::json!({"type":"object"}), + http_status: Some(404), + }]; + let adapter = FromJsonSchema::new( + test_spec_with_errors("svc/missing", OperationType::Query, errors), + test_config("svc", &base), + "/missing".to_string(), + "GET".to_string(), + test_http_client(), + ); + let bundles = adapter.import().await.unwrap(); + let registration = &bundles[0]; + let ctx = noop_context("req-11", Capabilities::new()); + let response = match ®istration.handler { + HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await, + _ => panic!("expected Once handler"), + }; + match response.result { + Err(e) => { + assert_eq!(e.code, "HTTP_404"); + assert!(!e.retryable); + } + other => panic!("expected HTTP_404 error, got {other:?}"), + } + } + + #[tokio::test] + async fn integration_undeclared_error_status_returns_http_status_code() { + let base = spawn_echo_server(500, "boom", "text/plain").await; + let adapter = FromJsonSchema::new( + test_spec("svc/x", OperationType::Query), + test_config("svc", &base), + "/x".to_string(), + "GET".to_string(), + test_http_client(), + ); + let bundles = adapter.import().await.unwrap(); + let registration = &bundles[0]; + let ctx = noop_context("req-12", Capabilities::new()); + let response = match ®istration.handler { + HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await, + _ => panic!("expected Once handler"), + }; + match response.result { + Err(e) => assert_eq!(e.code, "HTTP_500"), + other => panic!("expected HTTP_500, got {other:?}"), + } + } + + #[tokio::test] + async fn integration_sse_subscription_streams_responded_events() { + let sse_body = "data: {\"n\":1}\n\ndata: {\"n\":2}\n\n"; + let base = spawn_echo_server(200, sse_body, "text/event-stream").await; + let adapter = FromJsonSchema::new( + test_spec("svc/stream", OperationType::Sub), + test_config("svc", &base), + "/stream".to_string(), + "POST".to_string(), + test_http_client(), + ); + let bundles = adapter.import().await.unwrap(); + let registration = &bundles[0]; + let ctx = noop_context("req-13", Capabilities::new()); + let stream = match ®istration.handler { + HandlerKind::Stream(h) => h(serde_json::json!({}), ctx), + _ => panic!("expected Stream handler"), + }; + let collected: Vec = stream.collect().await; + assert_eq!(collected.len(), 2); + assert_eq!(collected[0].result, Ok(serde_json::json!({"n":1}))); + assert_eq!(collected[1].result, Ok(serde_json::json!({"n":2}))); + assert_eq!(collected[0].request_id, "req-13"); + assert_eq!(collected[1].request_id, "req-13"); + } + + #[test] + fn no_env_vars_read_in_build_request() { + std::env::set_var("OPENAI_API_KEY", "should-not-be-used"); + let ctx = noop_context("req-14", Capabilities::new()); + let (_, _, _, headers) = build_request( + "https://api.openai.com", + "/v1/chat", + "POST", + &Some(HttpAuthScheme::Bearer), + &HashMap::new(), + "openai", + &serde_json::json!({"body":{"prompt":"hi"}}), + &ctx, + ) + .unwrap(); + assert!( + headers.get(AUTHORIZATION).is_none(), + "no auth header when capabilities absent" + ); + std::env::remove_var("OPENAI_API_KEY"); + } +} diff --git a/src/adapters/mod.rs b/src/adapters/mod.rs index 8b13789..ce64cb3 100644 --- a/src/adapters/mod.rs +++ b/src/adapters/mod.rs @@ -1 +1,10 @@ +//! HTTP-backed call-protocol adapters: `from_openapi` / `from_jsonschema` +//! (import external HTTP APIs as operations), `to_openapi` / `to_mcp` +//! (project local operations onto HTTP surfaces), `from_mcp`, and +//! `from_wss` (feature `wss`). +pub mod forward; +pub mod from_jsonschema; + +pub use forward::{HttpAuthScheme, HttpServiceConfig}; +pub use from_jsonschema::FromJsonSchema; diff --git a/tasks/adapters/from-jsonschema.md b/tasks/adapters/from-jsonschema.md index 6ad2b81..09c6ba8 100644 --- a/tasks/adapters/from-jsonschema.md +++ b/tasks/adapters/from-jsonschema.md @@ -1,7 +1,7 @@ --- id: adapter-from-jsonschema name: from_jsonschema single-endpoint adapter -status: pending +status: completed depends_on: [client-http-host] scope: narrow risk: low @@ -23,9 +23,9 @@ from_openapi. ## Acceptance Criteria -- [ ] Adapter ported; one registration per call, correct provenance -- [ ] Forwarding round-trip test against a local mock server -- [ ] `cargo test` passes +- [x] Adapter ported; one registration per call, correct provenance +- [x] Forwarding round-trip test against a local mock server +- [x] `cargo test` passes ## References @@ -39,4 +39,26 @@ from_openapi. ## Summary -> Agent fills on completion. \ No newline at end of file +Ported `src/adapters/from_jsonschema.rs` per ADR-066, plus the shared +forwarding core it depends on: + +- `src/adapters/forward.rs`: extracted from old from_openapi — + HttpServiceConfig/HttpAuthScheme, build_request (path templates, + query split, body key, credential injection from + OperationContext.capabilities), forward (2xx -> JSON/text/bytes + envelope; non-2xx -> declared code or HTTP_), forward_stream + (SSE -> ResponseEnvelope stream, ADR-049), parse_sse_frames. + Pre-positions the from_openapi port to reuse the same core. +- `src/adapters/from_jsonschema.rs`: FromJsonSchema OperationAdapter — + one HandlerRegistration per call (spec + path template + method), + FromJsonSchema provenance (leaf, composition_authority None, + scoped_env None, Internal default), Sub op type -> HandlerKind::Stream + expecting text/event-stream. + +Adaptations for alkcall 0.1.1: OperationType::Sub (not Subscription), +adapter imports via alkcall::client::{AdapterError, OperationAdapter}. + +11 tests: registration shape, handler kinds (Query/Sub/Mutation), +build_request bearer-from-capabilities + path/query split, forwarding +round-trip vs local mock server (200/404/500), SSE subscription stream, +no-env-vars invariant. 106 lib tests green. \ No newline at end of file