From de5363caefb714057df1851505246e63ff8803b5 Mon Sep 17 00:00:00 2001 From: Nick Ruest <127058086+nicholas-ruest@users.noreply.github.com> Date: Mon, 27 Jul 2026 06:34:46 +0000 Subject: [PATCH 1/4] feat: expose the orchestration engine over HTTP and reduce the function to a proxy The deployed service now orchestrates. Before this, `functions/index.js` answered every agent request with a synchronous pure function that echoed `tasks.length` back with a status field: a cyclic workflow returned `status: "resolved"` with HTTP 200, and no handler ever called the 44,189 lines of tested Rust that implement the real thing. Adds the seven `/v1/orchestrator/*` routes to the existing axum server in a new `orchestrator_routes` module, each deserializing into the agent's existing request type and awaiting the existing async method. Nothing in llm-orchestrator-core changed; the routes are the missing link the ADR identified. Details worth knowing: - The response envelope (`execution_metadata`, `layers_executed`) is rebuilt in Rust to be byte-compatible with `functions/index.js:53-74`, including the `ORCHESTRATOR_` layer naming and its underscore substitution, so existing consumers see an unchanged shape. A test pins each field against the JavaScript it replaces. - Required-field validation reproduces `validateRequiredFields` including its exact `Missing required fields: ...` message. - Errors map by cause: a cycle, a dangling dependency, or a bad request body is 400; engine faults are 500. The dependency resolver reports a cycle in its status rather than as an Err, so an unsuccessful resolution is explicitly mapped to 400 -- returning 200 there would reproduce the exact defect this removes. - `StateTransitionRequest`, `ParallelizationRequest` and `SwarmCoordinationRequest` all require a `timestamp` that none of the published contracts list as required, so a contract-conformant body would fail to deserialize. The routes fill it in at the boundary; reconciling the two is a change to agentics-contracts, which two other repositories vendor and which ADR-0001 puts out of scope. - `/v1/orchestrator/workflow` takes the engine-native `{ workflow, inputs }` shape and rejects the contract's flat `{ workflow_id, workflow_name, tasks }` shape with a pointer to the right one. Those `tasks` carry no step type or configuration, so there is no faithful translation into `Workflow.steps` and guessing one would execute something the caller did not ask for. The Cloud Function keeps CORS, routing and contract publishing and stops fabricating results. It forwards each agent request to `ORCHESTRATOR_ENGINE_URL` with a Google-signed ID token fetched from the metadata server, propagates `X-Correlation-ID`, and returns the upstream status and body unmodified. Every failure mode fails closed -- unconfigured 503, unreachable 502, timeout 504, non-JSON upstream body 502 -- and `/health` and `/ready` now reflect the engine instead of checking that seven constants are seven constants. The local state-machine transition table is deleted so there is one source of truth. `functions/test.js` is rewritten to assert proxy behaviour instead of envelope shape: the old suite asserted only shape and so passed against a service that orchestrated nothing. Also drops the unused `claude-flow` dependency, and grants the Cloud Function's service account `roles/run.invoker` on the engine service specifically rather than project-wide. Implements orchestrator/docs/adr/ADR-0001-implement-real-orchestration-logic.md. --- crates/llm-orchestrator-cli/src/main.rs | 10 + .../src/orchestrator_routes.rs | 849 ++++++++++++++++++ deploy/gcloud/setup-iam.sh | 35 +- functions/index.js | 377 ++++---- functions/test.js | 582 +++++++----- package.json | 5 +- 6 files changed, 1407 insertions(+), 451 deletions(-) create mode 100644 crates/llm-orchestrator-cli/src/orchestrator_routes.rs diff --git a/crates/llm-orchestrator-cli/src/main.rs b/crates/llm-orchestrator-cli/src/main.rs index a5ceafc..41fd804 100644 --- a/crates/llm-orchestrator-cli/src/main.rs +++ b/crates/llm-orchestrator-cli/src/main.rs @@ -3,6 +3,8 @@ //! LLM Orchestrator CLI. +mod orchestrator_routes; + use anyhow::{Context, Result}; use clap::{Parser, Subcommand}; use colored::Colorize; @@ -4667,6 +4669,7 @@ async fn serve_http(host: &str, port: u16) -> Result<()> { .route("/ready", get(ready_handler)) .route("/execute", axum::routing::post(feu_execute_handler)) .route("/api/v1/events", axum::routing::post(events_handler)) + .merge(orchestrator_routes::routes()) .with_state(phase3_state); // Parse address @@ -4688,6 +4691,13 @@ async fn serve_http(host: &str, port: u16) -> Result<()> { println!(" GET /ready - Readiness check (Ruvector validated)"); println!(" POST /execute - FEU execution endpoint (Core invocation)"); println!(" POST /api/v1/events - Event ingestion (automation-core)"); + println!(" POST /v1/orchestrator/workflow - Execute a workflow"); + println!(" POST /v1/orchestrator/scheduler - Schedule tasks"); + println!(" POST /v1/orchestrator/dependencies - Resolve a dependency graph"); + println!(" POST /v1/orchestrator/retry - Evaluate a failure for recovery"); + println!(" POST /v1/orchestrator/parallel - Compute parallel execution phases"); + println!(" POST /v1/orchestrator/state-machine - Validate and apply a state transition"); + println!(" POST /v1/orchestrator/swarm - Coordinate a swarm of workers"); // Start server let listener = tokio::net::TcpListener::bind(addr).await?; diff --git a/crates/llm-orchestrator-cli/src/orchestrator_routes.rs b/crates/llm-orchestrator-cli/src/orchestrator_routes.rs new file mode 100644 index 0000000..14c02af --- /dev/null +++ b/crates/llm-orchestrator-cli/src/orchestrator_routes.rs @@ -0,0 +1,849 @@ +// Copyright (c) 2025 LLM DevOps +// SPDX-License-Identifier: Apache-2.0 + +//! The seven `/v1/orchestrator/*` HTTP routes. +//! +//! These bind the existing agents in `llm-orchestrator-core` to the axum server so the +//! deployed service performs real orchestration. Before this module the agents were reachable +//! only as CLI subcommands, and the Cloud Function answered every request by echoing its input +//! back with a `status` field. +//! +//! The response envelope (`execution_metadata`, `layers_executed`) is byte-compatible with the +//! one the Cloud Function used to build, so existing consumers see the same shape. +//! +//! Implements ADR-0001. + +use axum::{ + extract::Json, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response}, + routing::post, + Router, +}; +use chrono::Utc; +use llm_orchestrator_core::adapters::retry_recovery::{ + RecoveryError, RecoveryRequest, RetryRecoveryAgent, +}; +use llm_orchestrator_core::adapters::{ScheduleRequest, SchedulerError, TaskSchedulerAgent}; +use llm_orchestrator_core::{ + AgentConfig, DependencyResolveRequest, DependencyResolverAgent, DependencyResolverConfig, + ExecuteRequest, OrchestratorError, ParallelizationAgent, ParallelizationAgentConfig, + ParallelizationRequest, StateMachineAgent, StateMachineAgentConfig, StateTransitionRequest, + SwarmCoordinationRequest, SwarmCoordinatorAgent, SwarmCoordinatorAgentConfig, + WorkflowOrchestratorAgent, +}; +use serde::Serialize; +use serde_json::{json, Map, Value}; +use std::time::Instant; +use uuid::Uuid; + +/// Service name reported in `execution_metadata.service`. +/// +/// Deliberately identical to `SERVICE_NAME` in `functions/index.js` -- consumers correlate on +/// this string, and the engine taking over the work must not change it. +const SERVICE_NAME: &str = "orchestrator-agents"; + +const RUVECTOR_ENDPOINT_ENV: &str = "RUVECTOR_SERVICE_ENDPOINT"; + +// ============================================================================ +// RESPONSE ENVELOPE +// ============================================================================ + +#[derive(Debug, Serialize)] +struct ExecutionMetadata { + trace_id: String, + timestamp: String, + service: &'static str, + execution_id: String, +} + +#[derive(Debug, Serialize)] +struct LayerExecuted { + layer: String, + status: &'static str, + #[serde(skip_serializing_if = "Option::is_none")] + duration_ms: Option, +} + +/// Describes one agent's identity as published in `functions/contracts/index.js`. +struct AgentDescriptor { + /// Route segment, e.g. `dependencies`. + route: &'static str, + /// Contract `agent_id`. + id: &'static str, + /// Contract `agent_version`. + version: &'static str, + /// Contract `classification`. + classification: &'static str, +} + +const WORKFLOW: AgentDescriptor = AgentDescriptor { + route: "workflow", + id: "workflow-orchestrator", + version: "0.1.0", + classification: "WORKFLOW_EXECUTION", +}; +const SCHEDULER: AgentDescriptor = AgentDescriptor { + route: "scheduler", + id: "task-scheduler", + version: "0.1.0", + classification: "TASK_COORDINATION", +}; +const DEPENDENCIES: AgentDescriptor = AgentDescriptor { + route: "dependencies", + id: "dependency-resolver", + version: "0.1.0", + classification: "TASK_COORDINATION", +}; +const RETRY: AgentDescriptor = AgentDescriptor { + route: "retry", + id: "retry-recovery", + version: "0.1.0", + classification: "TASK_COORDINATION", +}; +const PARALLEL: AgentDescriptor = AgentDescriptor { + route: "parallel", + id: "parallelization-agent", + version: "0.1.0", + classification: "TASK_COORDINATION", +}; +const STATE_MACHINE: AgentDescriptor = AgentDescriptor { + route: "state-machine", + id: "state-machine-agent", + version: "1.0.0", + classification: "STATE_MANAGEMENT", +}; +const SWARM: AgentDescriptor = AgentDescriptor { + route: "swarm", + id: "swarm-coordinator-agent", + version: "0.1.0", + classification: "TASK_COORDINATION", +}; + +/// Builds the envelope the Cloud Function used to build, merged into the agent's own payload. +/// +/// Mirrors `buildResponse` at `functions/index.js:62`: agent fields and envelope keys sit at the +/// top level of the same object, not nested under a `data` key. +fn envelope( + headers: &HeaderMap, + agent: &AgentDescriptor, + status: &'static str, + mut payload: Map, + started: Instant, +) -> Value { + let trace_id = headers + .get("x-correlation-id") + .and_then(|v| v.to_str().ok()) + .map(str::to_owned) + .unwrap_or_else(|| Uuid::new_v4().to_string()); + + let duration_ms = started.elapsed().as_millis() as u64; + + payload.insert("agent".into(), json!(agent.id)); + payload.insert("agent_version".into(), json!(agent.version)); + payload.insert("classification".into(), json!(agent.classification)); + + payload.insert( + "execution_metadata".into(), + json!(ExecutionMetadata { + trace_id, + timestamp: Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true), + service: SERVICE_NAME, + execution_id: Uuid::new_v4().to_string(), + }), + ); + payload.insert( + "layers_executed".into(), + json!(vec![ + LayerExecuted { + layer: "AGENT_ROUTING".into(), + status: "completed", + duration_ms: None, + }, + LayerExecuted { + layer: format!( + "ORCHESTRATOR_{}", + agent.route.to_uppercase().replace('-', "_") + ), + status, + duration_ms: Some(duration_ms), + }, + ]), + ); + + Value::Object(payload) +} + +/// Serialises an agent response into the envelope. +fn ok_response( + headers: &HeaderMap, + agent: &AgentDescriptor, + body: T, + started: Instant, +) -> Response { + let payload = match serde_json::to_value(body) { + Ok(Value::Object(map)) => map, + // An agent whose response is not a JSON object would be a contract break, not a + // request error, so it is reported as such rather than silently reshaped. + Ok(other) => { + return error_response( + headers, + agent, + StatusCode::INTERNAL_SERVER_ERROR, + format!("Agent returned a non-object response: {other}"), + started, + ) + } + Err(e) => { + return error_response( + headers, + agent, + StatusCode::INTERNAL_SERVER_ERROR, + format!("Failed to serialize agent response: {e}"), + started, + ) + } + }; + + ( + StatusCode::OK, + Json(envelope(headers, agent, "completed", payload, started)), + ) + .into_response() +} + +fn error_response( + headers: &HeaderMap, + agent: &AgentDescriptor, + status: StatusCode, + message: impl Into, + started: Instant, +) -> Response { + let mut payload = Map::new(); + payload.insert("error".into(), json!(message.into())); + ( + status, + Json(envelope(headers, agent, "error", payload, started)), + ) + .into_response() +} + +// ============================================================================ +// VALIDATION AND ERROR MAPPING +// ============================================================================ + +/// Reproduces `validateRequiredFields` at `functions/index.js:80`, including its message. +fn missing_required(body: &Value, required: &[&str]) -> Option { + let missing: Vec<&str> = required + .iter() + .copied() + .filter(|field| match body.get(*field) { + None | Some(Value::Null) => true, + Some(_) => false, + }) + .collect(); + + if missing.is_empty() { + None + } else { + Some(format!("Missing required fields: {}", missing.join(", "))) + } +} + +/// A malformed body is the caller's error, so deserialization failures map to 400. +fn parse_request(body: Value) -> Result { + serde_json::from_value(body).map_err(|e| format!("Invalid request body: {e}")) +} + +/// Supplies `timestamp` when the caller omits it. +/// +/// `StateTransitionRequest`, `ParallelizationRequest` and `SwarmCoordinationRequest` all declare +/// `timestamp` without a serde default, but none of the published contracts list it as required, +/// so a contract-conformant body would fail to deserialize. Reconciling the two is a change to +/// `agentics-contracts`, which two other repositories vendor and which ADR-0001 puts out of +/// scope; filling the value in at the boundary keeps existing callers working meanwhile. +fn default_timestamp(body: &mut Value) { + if let Some(map) = body.as_object_mut() { + if !matches!(map.get("timestamp"), Some(v) if !v.is_null()) { + map.insert( + "timestamp".into(), + json!(Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)), + ); + } + } +} + +/// Maps an agent error to the status code its cause deserves. +/// +/// A graph the caller described wrongly -- a cycle, a dangling dependency, a bad config -- is a +/// 400. Everything else is the engine's fault and is a 500. +fn status_for(error: &OrchestratorError) -> StatusCode { + match error { + OrchestratorError::CyclicDependency + | OrchestratorError::StepNotFound(_) + | OrchestratorError::ValidationError(_) + | OrchestratorError::ParseError(_) + | OrchestratorError::InvalidStepConfig { .. } + | OrchestratorError::InvalidStateTransition { .. } + | OrchestratorError::ContextVariableNotFound(_) => StatusCode::BAD_REQUEST, + _ => StatusCode::INTERNAL_SERVER_ERROR, + } +} + +/// Boilerplate every handler shares: validate, parse, and return early on failure. +macro_rules! prepare { + ($headers:expr, $agent:expr, $started:expr, $body:expr, $required:expr, $ty:ty) => {{ + if let Some(message) = missing_required(&$body, $required) { + return error_response($headers, $agent, StatusCode::BAD_REQUEST, message, $started); + } + match parse_request::<$ty>($body) { + Ok(request) => request, + Err(message) => { + return error_response( + $headers, + $agent, + StatusCode::BAD_REQUEST, + message, + $started, + ) + } + } + }}; +} + +/// Converts an agent `Result` into a response, mapping the error by cause. +macro_rules! respond { + ($headers:expr, $agent:expr, $started:expr, $result:expr) => { + match $result { + Ok(response) => ok_response($headers, $agent, response, $started), + Err(e) => { + let status = status_for(&e); + error_response($headers, $agent, status, e.to_string(), $started) + } + } + }; +} + +fn ruvector_endpoint() -> Option { + std::env::var(RUVECTOR_ENDPOINT_ENV).ok().filter(|s| !s.is_empty()) +} + +// ============================================================================ +// HANDLERS +// ============================================================================ + +/// `POST /v1/orchestrator/workflow` -- executes a workflow through `WorkflowExecutor`. +/// +/// Takes the engine-native `{ workflow, inputs }` shape. The published contract's flat +/// `{ workflow_id, workflow_name, tasks }` shape is rejected with a pointer to the right one: +/// its `tasks` carry no step type or step config, so there is no faithful translation into +/// `Workflow.steps`, and guessing one would execute something the caller did not ask for. See +/// ADR-0001 for the contract reconciliation this defers. +async fn workflow_handler(headers: HeaderMap, Json(body): Json) -> Response { + let started = Instant::now(); + + if body.get("workflow").is_none() && body.get("tasks").is_some() { + return error_response( + &headers, + &WORKFLOW, + StatusCode::BAD_REQUEST, + "This endpoint expects the engine-native shape { workflow: { name, steps: [...] }, \ + inputs: {...} }. The contract's flat { workflow_id, workflow_name, tasks } shape \ + carries no step type or configuration and cannot be executed. See ADR-0001.", + started, + ); + } + + let request = prepare!( + &headers, + &WORKFLOW, + started, + body, + &["workflow"], + ExecuteRequest + ); + + let agent = WorkflowOrchestratorAgent::new(AgentConfig::default()); + respond!(&headers, &WORKFLOW, started, agent.execute(request).await) +} + +/// `POST /v1/orchestrator/scheduler` -- schedules a task via `TaskSchedulerAgent`. +async fn scheduler_handler(headers: HeaderMap, Json(body): Json) -> Response { + let started = Instant::now(); + + let request = prepare!( + &headers, + &SCHEDULER, + started, + body, + &["schedule_id", "tasks"], + ScheduleRequest + ); + + let Some(endpoint) = ruvector_endpoint() else { + return error_response( + &headers, + &SCHEDULER, + StatusCode::SERVICE_UNAVAILABLE, + format!("{RUVECTOR_ENDPOINT_ENV} is not configured; the scheduler cannot persist"), + started, + ); + }; + + let agent = TaskSchedulerAgent::new(endpoint); + match agent.schedule(request).await { + Ok(result) => ok_response(&headers, &SCHEDULER, result, started), + Err(e) => { + let status = scheduler_status(&e); + error_response(&headers, &SCHEDULER, status, e.to_string(), started) + } + } +} + +/// `POST /v1/orchestrator/dependencies` -- resolves a dependency graph. +/// +/// This is the route ADR-0001 is defined by: it returns a real topological order and rejects a +/// cyclic graph, where the handler it replaces returned `status: "resolved"` with HTTP 200 for +/// any input at all. +async fn dependencies_handler(headers: HeaderMap, Json(body): Json) -> Response { + let started = Instant::now(); + + let request = prepare!( + &headers, + &DEPENDENCIES, + started, + body, + &["request_id", "workflow_id", "tasks"], + DependencyResolveRequest + ); + + let agent = DependencyResolverAgent::new(DependencyResolverConfig::default()); + match agent.resolve(request).await { + // The resolver reports a cycle or a dangling dependency in its status rather than as an + // Err, so an unsuccessful resolution has to be mapped to 400 here. Returning 200 would + // reproduce exactly the defect this ADR exists to remove. + Ok(response) if !response.success => { + let mut payload = match serde_json::to_value(&response) { + Ok(Value::Object(map)) => map, + _ => Map::new(), + }; + payload.insert( + "error".into(), + json!(response + .error + .clone() + .unwrap_or_else(|| format!("Resolution failed: {:?}", response.status))), + ); + ( + StatusCode::BAD_REQUEST, + Json(envelope(&headers, &DEPENDENCIES, "error", payload, started)), + ) + .into_response() + } + Ok(response) => ok_response(&headers, &DEPENDENCIES, response, started), + Err(e) => { + let status = status_for(&e); + error_response(&headers, &DEPENDENCIES, status, e.to_string(), started) + } + } +} + +/// `POST /v1/orchestrator/retry` -- evaluates a failure and chooses a recovery action. +async fn retry_handler(headers: HeaderMap, Json(mut body): Json) -> Response { + let started = Instant::now(); + + if let Some(message) = missing_required(&body, &["request_id", "failure"]) { + return error_response(&headers, &RETRY, StatusCode::BAD_REQUEST, message, started); + } + + // The published contract names this field `failure`; `RecoveryRequest` names it `error`. + // Accept the contract's name on the wire rather than break existing callers. + if let Some(map) = body.as_object_mut() { + if let Some(failure) = map.remove("failure") { + map.insert("error".into(), failure); + } + } + + let request = match parse_request::(body) { + Ok(request) => request, + Err(message) => { + return error_response(&headers, &RETRY, StatusCode::BAD_REQUEST, message, started) + } + }; + + let agent = match ruvector_endpoint() { + Some(endpoint) => RetryRecoveryAgent::new(endpoint), + None => RetryRecoveryAgent::disabled(), + }; + + match agent.evaluate(request).await { + Ok(decision) => ok_response(&headers, &RETRY, decision, started), + Err(e) => { + let status = recovery_status(&e); + error_response(&headers, &RETRY, status, e.to_string(), started) + } + } +} + +/// `POST /v1/orchestrator/parallel` -- computes parallel execution phases. +async fn parallel_handler(headers: HeaderMap, Json(mut body): Json) -> Response { + let started = Instant::now(); + default_timestamp(&mut body); + + let request = prepare!( + &headers, + &PARALLEL, + started, + body, + &["request_id", "workflow_id", "tasks"], + ParallelizationRequest + ); + + let agent = ParallelizationAgent::new(ParallelizationAgentConfig::default()); + respond!(&headers, &PARALLEL, started, agent.analyze(request).await) +} + +/// `POST /v1/orchestrator/state-machine` -- validates and applies a state transition. +/// +/// Replaces the one Cloud Function handler that did real work. The agent is constructed per +/// request because `transition` takes `&mut self` and keeps its history in memory; sharing one +/// across requests would leak one caller's entity states into another's validation. +async fn state_machine_handler(headers: HeaderMap, Json(mut body): Json) -> Response { + let started = Instant::now(); + default_timestamp(&mut body); + + let request = prepare!( + &headers, + &STATE_MACHINE, + started, + body, + &[ + "request_id", + "execution_id", + "entity_type", + "current_state", + "target_state", + "reason", + "initiated_by", + ], + StateTransitionRequest + ); + + let mut agent = StateMachineAgent::new(StateMachineAgentConfig::default()); + respond!( + &headers, + &STATE_MACHINE, + started, + agent.transition(request).await + ) +} + +/// `POST /v1/orchestrator/swarm` -- coordinates a swarm of workers toward an objective. +async fn swarm_handler(headers: HeaderMap, Json(mut body): Json) -> Response { + let started = Instant::now(); + default_timestamp(&mut body); + + let request = prepare!( + &headers, + &SWARM, + started, + body, + &["request_id", "workflow_id", "objective", "workers"], + SwarmCoordinationRequest + ); + + let agent = SwarmCoordinatorAgent::new(SwarmCoordinatorAgentConfig::default()); + respond!(&headers, &SWARM, started, agent.coordinate(request).await) +} + +fn scheduler_status(error: &SchedulerError) -> StatusCode { + match error { + SchedulerError::ValidationError(_) | SchedulerError::DependencyError(_) => { + StatusCode::BAD_REQUEST + } + SchedulerError::AgentDisabled => StatusCode::SERVICE_UNAVAILABLE, + SchedulerError::Timeout => StatusCode::GATEWAY_TIMEOUT, + _ => StatusCode::INTERNAL_SERVER_ERROR, + } +} + +fn recovery_status(error: &RecoveryError) -> StatusCode { + match error { + RecoveryError::ValidationError(_) => StatusCode::BAD_REQUEST, + RecoveryError::AgentDisabled => StatusCode::SERVICE_UNAVAILABLE, + RecoveryError::Timeout => StatusCode::GATEWAY_TIMEOUT, + _ => StatusCode::INTERNAL_SERVER_ERROR, + } +} + +// ============================================================================ +// ROUTER +// ============================================================================ + +/// The seven agent routes, mountable on any router regardless of its state type. +pub fn routes() -> Router +where + S: Clone + Send + Sync + 'static, +{ + Router::new() + .route("/v1/orchestrator/workflow", post(workflow_handler)) + .route("/v1/orchestrator/scheduler", post(scheduler_handler)) + .route("/v1/orchestrator/dependencies", post(dependencies_handler)) + .route("/v1/orchestrator/retry", post(retry_handler)) + .route("/v1/orchestrator/parallel", post(parallel_handler)) + .route("/v1/orchestrator/state-machine", post(state_machine_handler)) + .route("/v1/orchestrator/swarm", post(swarm_handler)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn body(json: Value) -> Json { + Json(json) + } + + #[test] + fn missing_required_matches_the_cloud_function_message() { + let body = json!({ "workflow_id": "wf-1" }); + assert_eq!( + missing_required(&body, &["request_id", "workflow_id", "tasks"]), + Some("Missing required fields: request_id, tasks".to_string()) + ); + } + + #[test] + fn explicit_null_counts_as_missing() { + let body = json!({ "tasks": null }); + assert_eq!( + missing_required(&body, &["tasks"]), + Some("Missing required fields: tasks".to_string()) + ); + } + + #[test] + fn present_fields_pass() { + let body = json!({ "request_id": "r", "workflow_id": "w", "tasks": [] }); + assert_eq!( + missing_required(&body, &["request_id", "workflow_id", "tasks"]), + None + ); + } + + #[test] + fn caller_errors_are_400_and_engine_errors_are_500() { + assert_eq!( + status_for(&OrchestratorError::CyclicDependency), + StatusCode::BAD_REQUEST + ); + assert_eq!( + status_for(&OrchestratorError::StepNotFound("z".into())), + StatusCode::BAD_REQUEST + ); + assert_eq!( + status_for(&OrchestratorError::Internal("boom".into())), + StatusCode::INTERNAL_SERVER_ERROR + ); + } + + #[test] + fn envelope_is_byte_compatible_with_the_cloud_function() { + let headers = HeaderMap::new(); + let mut payload = Map::new(); + payload.insert("status".into(), json!("resolved")); + + let value = envelope(&headers, &DEPENDENCIES, "completed", payload, Instant::now()); + + // Agent identity, as published in the contract. + assert_eq!(value["agent"], json!("dependency-resolver")); + assert_eq!(value["agent_version"], json!("0.1.0")); + assert_eq!(value["classification"], json!("TASK_COORDINATION")); + + // execution_metadata keys, verbatim from functions/index.js:53-60. + let meta = &value["execution_metadata"]; + assert_eq!(meta["service"], json!("orchestrator-agents")); + for key in ["trace_id", "timestamp", "service", "execution_id"] { + assert!(meta.get(key).is_some(), "missing execution_metadata.{key}"); + } + + // layers_executed, verbatim from functions/index.js:64-67. + let layers = value["layers_executed"].as_array().expect("array"); + assert_eq!(layers.len(), 2); + assert_eq!(layers[0]["layer"], json!("AGENT_ROUTING")); + assert_eq!(layers[0]["status"], json!("completed")); + assert_eq!(layers[1]["layer"], json!("ORCHESTRATOR_DEPENDENCIES")); + assert_eq!(layers[1]["status"], json!("completed")); + assert!(layers[1]["duration_ms"].is_u64()); + + // The agent's own payload stays at the top level, not nested. + assert_eq!(value["status"], json!("resolved")); + } + + #[test] + fn hyphenated_routes_become_underscored_layer_names() { + let value = envelope( + &HeaderMap::new(), + &STATE_MACHINE, + "completed", + Map::new(), + Instant::now(), + ); + assert_eq!( + value["layers_executed"][1]["layer"], + json!("ORCHESTRATOR_STATE_MACHINE") + ); + } + + #[test] + fn correlation_id_is_propagated_as_the_trace_id() { + let mut headers = HeaderMap::new(); + headers.insert("x-correlation-id", "trace-abc".parse().unwrap()); + + let value = envelope( + &headers, + &DEPENDENCIES, + "completed", + Map::new(), + Instant::now(), + ); + assert_eq!(value["execution_metadata"]["trace_id"], json!("trace-abc")); + } + + #[tokio::test] + async fn diamond_dag_comes_back_in_dependency_order() { + // Deliberately listed out of execution order so echoing the input cannot pass. + let request = json!({ + "request_id": "adr-0001-verify-1", + "workflow_id": Uuid::new_v4(), + "tasks": [ + { "task_id": "D", "name": "publish", "depends_on": ["B", "C"] }, + { "task_id": "B", "name": "extract", "depends_on": ["A"] }, + { "task_id": "C", "name": "classify", "depends_on": ["A"] }, + { "task_id": "A", "name": "ingest", "depends_on": [] } + ] + }); + + let response = dependencies_handler(HeaderMap::new(), body(request)).await; + assert_eq!(response.status(), StatusCode::OK); + + let value = read_body(response).await; + let order: Vec = value["execution_order"] + .as_array() + .expect("execution_order") + .iter() + .map(|v| v.as_str().unwrap().to_string()) + .collect(); + + assert_eq!(order.len(), 4); + let position = |id: &str| order.iter().position(|t| t == id).unwrap(); + assert!(position("A") < position("B")); + assert!(position("A") < position("C")); + assert!(position("B") < position("D")); + assert!(position("C") < position("D")); + + // B and C are independent and must share a parallel group. + let groups = value["parallel_groups"].as_array().expect("parallel_groups"); + assert!( + groups.iter().any(|group| { + let ids: Vec<&str> = group["task_ids"] + .as_array() + .map(|a| a.iter().filter_map(|v| v.as_str()).collect()) + .unwrap_or_default(); + ids.contains(&"B") && ids.contains(&"C") + }), + "expected B and C in one parallel group, got {groups:?}" + ); + } + + #[tokio::test] + async fn cycle_is_rejected_with_400() { + let request = json!({ + "request_id": "adr-0001-verify-2", + "workflow_id": Uuid::new_v4(), + "tasks": [ + { "task_id": "A", "name": "ingest", "depends_on": ["D"] }, + { "task_id": "B", "name": "extract", "depends_on": ["A"] }, + { "task_id": "D", "name": "publish", "depends_on": ["B"] } + ] + }); + + let response = dependencies_handler(HeaderMap::new(), body(request)).await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + + let value = read_body(response).await; + let error = value["error"].as_str().unwrap_or_default().to_lowercase(); + assert!(error.contains("cycle"), "error did not name a cycle: {error}"); + } + + #[tokio::test] + async fn dangling_dependency_is_rejected_with_400() { + let request = json!({ + "request_id": "adr-0001-verify-3", + "workflow_id": Uuid::new_v4(), + "tasks": [ + { "task_id": "A", "name": "ingest", "depends_on": ["Z"] } + ] + }); + + let response = dependencies_handler(HeaderMap::new(), body(request)).await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + + #[tokio::test] + async fn missing_fields_are_rejected_before_the_agent_runs() { + let response = dependencies_handler(HeaderMap::new(), body(json!({ "tasks": [] }))).await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + + let value = read_body(response).await; + assert_eq!( + value["error"], + json!("Missing required fields: request_id, workflow_id") + ); + } + + #[tokio::test] + async fn an_illegal_state_transition_is_reported_as_invalid() { + let request = json!({ + "request_id": Uuid::new_v4(), + "execution_id": Uuid::new_v4(), + "entity_type": "task", + "current_state": "completed", + "target_state": "running", + "reason": "adr-0001 verification", + "initiated_by": "test" + }); + + let response = state_machine_handler(HeaderMap::new(), body(request)).await; + assert_eq!(response.status(), StatusCode::OK); + + let value = read_body(response).await; + assert_eq!(value["success"], json!(false)); + assert_eq!(value["status"], json!("invalid")); + assert_eq!(value["new_state"], json!("completed")); + } + + #[tokio::test] + async fn the_contract_shaped_workflow_body_is_rejected_rather_than_guessed() { + let request = json!({ + "workflow_id": Uuid::new_v4(), + "workflow_name": "demo", + "tasks": [{ "task_id": "A", "name": "ingest", "task_type": "llm" }] + }); + + let response = workflow_handler(HeaderMap::new(), body(request)).await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + + let value = read_body(response).await; + assert!(value["error"] + .as_str() + .unwrap_or_default() + .contains("engine-native shape")); + } + + async fn read_body(response: Response) -> Value { + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + serde_json::from_slice(&bytes).expect("json") + } +} diff --git a/deploy/gcloud/setup-iam.sh b/deploy/gcloud/setup-iam.sh index 93997bb..2517331 100755 --- a/deploy/gcloud/setup-iam.sh +++ b/deploy/gcloud/setup-iam.sh @@ -3,7 +3,7 @@ # LLM-Orchestrator IAM Setup Script # ============================================================================= # Creates service account with least-privilege permissions -# Run once per environment: ./setup-iam.sh +# Run once per environment: ./setup-iam.sh [region] # ============================================================================= set -euo pipefail @@ -13,6 +13,12 @@ PLATFORM_ENV="${2:-dev}" SERVICE_NAME="llm-orchestrator" SA_NAME="${SERVICE_NAME}-sa" SA_EMAIL="${SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com" +REGION="${3:-us-central1}" + +# The Cloud Function that fronts the engine. Since ADR-0001 it holds no orchestration logic +# and must call the Cloud Run service for every agent request, so it needs run.invoker on +# that service specifically -- not project-wide. +FUNCTION_SA_EMAIL="${FUNCTION_SA_EMAIL:-${PROJECT_ID}@appspot.gserviceaccount.com}" echo "==============================================" echo "LLM-Orchestrator IAM Setup" @@ -23,7 +29,7 @@ echo "Service Account: ${SA_EMAIL}" echo "==============================================" # Create service account -echo "[1/6] Creating service account..." +echo "[1/7] Creating service account..." gcloud iam service-accounts create "${SA_NAME}" \ --project="${PROJECT_ID}" \ --display-name="LLM-Orchestrator Service Account" \ @@ -31,7 +37,7 @@ gcloud iam service-accounts create "${SA_NAME}" \ 2>/dev/null || echo "Service account already exists" # Grant Cloud Run Invoker role (for internal service calls) -echo "[2/6] Granting Cloud Run Invoker role..." +echo "[2/7] Granting Cloud Run Invoker role..." gcloud projects add-iam-policy-binding "${PROJECT_ID}" \ --member="serviceAccount:${SA_EMAIL}" \ --role="roles/run.invoker" \ @@ -39,7 +45,7 @@ gcloud projects add-iam-policy-binding "${PROJECT_ID}" \ --quiet # Grant Secret Manager Secret Accessor role -echo "[3/6] Granting Secret Manager accessor role..." +echo "[3/7] Granting Secret Manager accessor role..." gcloud projects add-iam-policy-binding "${PROJECT_ID}" \ --member="serviceAccount:${SA_EMAIL}" \ --role="roles/secretmanager.secretAccessor" \ @@ -47,7 +53,7 @@ gcloud projects add-iam-policy-binding "${PROJECT_ID}" \ --quiet # Grant Cloud Logging Writer role -echo "[4/6] Granting Cloud Logging writer role..." +echo "[4/7] Granting Cloud Logging writer role..." gcloud projects add-iam-policy-binding "${PROJECT_ID}" \ --member="serviceAccount:${SA_EMAIL}" \ --role="roles/logging.logWriter" \ @@ -55,7 +61,7 @@ gcloud projects add-iam-policy-binding "${PROJECT_ID}" \ --quiet # Grant Cloud Trace Agent role -echo "[5/6] Granting Cloud Trace agent role..." +echo "[5/7] Granting Cloud Trace agent role..." gcloud projects add-iam-policy-binding "${PROJECT_ID}" \ --member="serviceAccount:${SA_EMAIL}" \ --role="roles/cloudtrace.agent" \ @@ -63,13 +69,25 @@ gcloud projects add-iam-policy-binding "${PROJECT_ID}" \ --quiet # Grant Cloud Monitoring Metric Writer role -echo "[6/6] Granting Cloud Monitoring metric writer role..." +echo "[6/7] Granting Cloud Monitoring metric writer role..." gcloud projects add-iam-policy-binding "${PROJECT_ID}" \ --member="serviceAccount:${SA_EMAIL}" \ --role="roles/monitoring.metricWriter" \ --condition=None \ --quiet +# Grant the Cloud Function permission to invoke the engine (ADR-0001). +# Scoped to the one service rather than the whole project: the function needs to call the +# orchestration engine and nothing else. +echo "[7/7] Granting the Cloud Function invoker access to ${SERVICE_NAME}..." +gcloud run services add-iam-policy-binding "${SERVICE_NAME}" \ + --project="${PROJECT_ID}" \ + --region="${REGION}" \ + --member="serviceAccount:${FUNCTION_SA_EMAIL}" \ + --role="roles/run.invoker" \ + --quiet \ + || echo " Skipped: the ${SERVICE_NAME} service does not exist yet in ${REGION}. Re-run after deploy.sh." + echo "" echo "==============================================" echo "IAM Setup Complete" @@ -84,6 +102,9 @@ echo " - roles/logging.logWriter (Cloud Logging)" echo " - roles/cloudtrace.agent (Cloud Trace)" echo " - roles/monitoring.metricWriter (Cloud Monitoring)" echo "" +echo "Cloud Function (${FUNCTION_SA_EMAIL}):" +echo " - roles/run.invoker on the ${SERVICE_NAME} service only (ADR-0001)" +echo "" echo "NOTE: This service account has NO direct database access." echo " All persistence occurs via ruvector-service." echo "==============================================" diff --git a/functions/index.js b/functions/index.js index 2b1cca6..d793fa8 100644 --- a/functions/index.js +++ b/functions/index.js @@ -7,10 +7,14 @@ const { AGENT_CONTRACTS } = require('./contracts'); // orchestrator-agents Cloud Function // Entry point: handler // Runtime: nodejs20 +// +// This function does NOT orchestrate. It authenticates, applies CORS, publishes +// contracts, and proxies every agent request to the Rust engine on Cloud Run, +// which owns all orchestration logic. See ADR-0001. // ============================================================================= const SERVICE_NAME = 'orchestrator-agents'; -const SERVICE_VERSION = '0.1.5'; +const SERVICE_VERSION = '0.2.0'; const BASE_URL = 'https://us-central1-agentics-dev.cloudfunctions.net/orchestrator-agents'; const AGENTS = ['workflow', 'scheduler', 'dependencies', 'retry', 'parallel', 'state-machine', 'swarm']; @@ -35,6 +39,48 @@ const AGENT_ROUTES = { swarm: '/v1/orchestrator/swarm', }; +// ============================================================================= +// Upstream engine configuration +// ============================================================================= + +const ENGINE_URL = process.env.ORCHESTRATOR_ENGINE_URL; +const ENGINE_TIMEOUT_MS = Number(process.env.ORCHESTRATOR_ENGINE_TIMEOUT_MS || 30000); + +// Set only for local development against an engine that does not require auth. +const SKIP_AUTH = process.env.ORCHESTRATOR_ENGINE_SKIP_AUTH === 'true'; + +const METADATA_IDENTITY_URL = + 'http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/identity'; + +// Identity tokens are valid for an hour; refresh a few minutes early. +const TOKEN_REFRESH_MARGIN_MS = 5 * 60 * 1000; +let cachedToken = null; + +/** + * Fetches a Google-signed ID token for the Cloud Run service. + * + * Uses the metadata server directly rather than google-auth-library so the function keeps zero + * runtime dependencies beyond the functions framework. + */ +async function getIdToken(audience) { + if (SKIP_AUTH) return null; + + const now = Date.now(); + if (cachedToken && cachedToken.audience === audience && cachedToken.expiresAt > now) { + return cachedToken.token; + } + + const url = `${METADATA_IDENTITY_URL}?audience=${encodeURIComponent(audience)}`; + const response = await fetch(url, { headers: { 'Metadata-Flavor': 'Google' } }); + if (!response.ok) { + throw new Error(`Metadata server returned ${response.status} fetching an ID token`); + } + + const token = (await response.text()).trim(); + cachedToken = { audience, token, expiresAt: now + 3600000 - TOKEN_REFRESH_MARGIN_MS }; + return token; +} + // ============================================================================= // CORS // ============================================================================= @@ -48,6 +94,10 @@ function setCorsHeaders(res) { // ============================================================================= // Execution Metadata Builder +// +// Only applied to responses this function generates itself (routing, contracts, and its own +// errors). Agent responses come back from the engine with the envelope already applied and are +// forwarded unmodified. // ============================================================================= function buildExecutionMetadata(req) { @@ -74,206 +124,140 @@ function buildResponse(req, agentName, status, data, startTime) { } // ============================================================================= -// Validation +// Engine proxy // ============================================================================= -function validateRequiredFields(body, required) { - const missing = required.filter((f) => body[f] === undefined || body[f] === null); - if (missing.length > 0) { - return `Missing required fields: ${missing.join(', ')}`; +/** + * Forwards a request to the engine and returns its status and body unmodified. + * + * Fails closed: if the engine is unconfigured, unreachable, or times out, this returns a 5xx. + * It never synthesises a success, which is the defect ADR-0001 exists to remove. + */ +async function proxyToEngine(req, path, body) { + if (!ENGINE_URL) { + return { + statusCode: 503, + data: { + error: 'ORCHESTRATOR_ENGINE_URL is not configured; the orchestration engine is unreachable', + engine_configured: false, + }, + }; } - return null; -} - -// ============================================================================= -// Agent Handlers (routing layer — business logic lives in Rust crates) -// ============================================================================= -function handleWorkflow(body) { - const contract = AGENT_CONTRACTS.workflow; - const error = validateRequiredFields(body, contract.input.required); - if (error) return { statusCode: 400, data: { error, agent: contract.agent_id } }; + const correlationId = req.headers['x-correlation-id'] || crypto.randomUUID(); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), ENGINE_TIMEOUT_MS); - return { - statusCode: 200, - data: { - agent: contract.agent_id, - agent_version: contract.agent_version, - classification: contract.classification, - status: 'accepted', - workflow_id: body.workflow_id, - workflow_name: body.workflow_name, - tasks_count: Array.isArray(body.tasks) ? body.tasks.length : 0, - strategy: (body.config && body.config.strategy) || 'sequential', - }, - }; -} - -function handleScheduler(body) { - const contract = AGENT_CONTRACTS.scheduler; - const error = validateRequiredFields(body, contract.input.required); - if (error) return { statusCode: 400, data: { error, agent: contract.agent_id } }; - - return { - statusCode: 200, - data: { - agent: contract.agent_id, - agent_version: contract.agent_version, - classification: contract.classification, - status: 'scheduled', - schedule_id: body.schedule_id, - tasks_count: Array.isArray(body.tasks) ? body.tasks.length : 0, - }, - }; -} - -function handleDependencies(body) { - const contract = AGENT_CONTRACTS.dependencies; - const error = validateRequiredFields(body, contract.input.required); - if (error) return { statusCode: 400, data: { error, agent: contract.agent_id } }; - - return { - statusCode: 200, - data: { - agent: contract.agent_id, - agent_version: contract.agent_version, - classification: contract.classification, - status: 'resolved', - request_id: body.request_id, - workflow_id: body.workflow_id, - tasks_count: Array.isArray(body.tasks) ? body.tasks.length : 0, - }, - }; -} + try { + const headers = { + 'Content-Type': 'application/json', + 'X-Correlation-ID': correlationId, + }; -function handleRetry(body) { - const contract = AGENT_CONTRACTS.retry; - const error = validateRequiredFields(body, contract.input.required); - if (error) return { statusCode: 400, data: { error, agent: contract.agent_id } }; + const token = await getIdToken(ENGINE_URL); + if (token) headers.Authorization = `Bearer ${token}`; + + const response = await fetch(`${ENGINE_URL}${path}`, { + method: body === undefined ? 'GET' : 'POST', + headers, + body: body === undefined ? undefined : JSON.stringify(body), + signal: controller.signal, + }); + + const text = await response.text(); + let data; + try { + data = text ? JSON.parse(text) : {}; + } catch { + // A non-JSON body from the engine is a real upstream fault, not something to paper over. + return { + statusCode: 502, + data: { + error: 'Engine returned a non-JSON response', + upstream_status: response.status, + upstream_body: text.slice(0, 2048), + }, + }; + } - return { - statusCode: 200, - data: { - agent: contract.agent_id, - agent_version: contract.agent_version, - classification: contract.classification, - status: 'analyzed', - request_id: body.request_id, - failure_category: body.failure && body.failure.error_category, - }, - }; + return { statusCode: response.status, data, correlationId, passthrough: true }; + } catch (err) { + const timedOut = err.name === 'AbortError'; + return { + statusCode: timedOut ? 504 : 502, + data: { + error: timedOut + ? `Engine did not respond within ${ENGINE_TIMEOUT_MS}ms` + : `Engine request failed: ${err.message}`, + engine_url: ENGINE_URL, + }, + }; + } finally { + clearTimeout(timer); + } } -function handleParallel(body) { - const contract = AGENT_CONTRACTS.parallel; - const error = validateRequiredFields(body, contract.input.required); - if (error) return { statusCode: 400, data: { error, agent: contract.agent_id } }; - - return { - statusCode: 200, - data: { - agent: contract.agent_id, - agent_version: contract.agent_version, - classification: contract.classification, - status: 'analyzed', - request_id: body.request_id, - workflow_id: body.workflow_id, - tasks_count: Array.isArray(body.tasks) ? body.tasks.length : 0, - }, - }; -} +// ============================================================================= +// Health and readiness +// +// Both proxy the engine and fail closed. The previous implementation checked that seven +// constants were seven constants, so it could never report anything but healthy. +// ============================================================================= -function handleStateMachine(body) { - const contract = AGENT_CONTRACTS['state-machine']; - const error = validateRequiredFields(body, contract.input.required); - if (error) return { statusCode: 400, data: { error, agent: contract.agent_id } }; +async function handleHealth(req) { + const upstream = await proxyToEngine(req, '/health'); + const healthy = upstream.statusCode === 200; - const sm = contract.state_machine[body.entity_type] || contract.state_machine.workflow; - const validTargets = sm.transitions[body.current_state] || []; - const isValid = body.force || validTargets.includes(body.target_state); + const agentStatuses = {}; + for (const agent of AGENTS) { + agentStatuses[agent] = { + status: healthy ? 'healthy' : 'unavailable', + route: AGENT_ROUTES[agent], + name: AGENT_DISPLAY_NAMES[agent], + }; + } return { - statusCode: 200, + statusCode: healthy ? 200 : 503, data: { - agent: contract.agent_id, - agent_version: contract.agent_version, - classification: contract.classification, - status: isValid ? 'completed' : 'invalid', - request_id: body.request_id, - execution_id: body.execution_id, - entity_type: body.entity_type, - success: isValid, - previous_state: body.current_state, - new_state: isValid ? body.target_state : body.current_state, - validation: { - valid: isValid, - source_state_matches: true, - target_state_exists: validTargets.length > 0, - valid_targets: validTargets, - rule_violations: [], - warnings: [], + status: healthy ? 'healthy' : 'unhealthy', + service: SERVICE_NAME, + version: SERVICE_VERSION, + engine: { + url: ENGINE_URL || null, + reachable: healthy, + status_code: upstream.statusCode, + detail: healthy ? undefined : upstream.data && upstream.data.error, }, + agents: agentStatuses, + agents_list: AGENTS, + base_url: BASE_URL, + timestamp: new Date().toISOString(), }, }; } -function handleSwarm(body) { - const contract = AGENT_CONTRACTS.swarm; - const error = validateRequiredFields(body, contract.input.required); - if (error) return { statusCode: 400, data: { error, agent: contract.agent_id } }; +async function handleReady(req) { + const upstream = await proxyToEngine(req, '/ready'); + const ready = upstream.statusCode === 200; return { - statusCode: 200, + statusCode: ready ? 200 : 503, data: { - agent: contract.agent_id, - agent_version: contract.agent_version, - classification: contract.classification, - status: 'accepted', - request_id: body.request_id, - workflow_id: body.workflow_id, - workers_count: Array.isArray(body.workers) ? body.workers.length : 0, - objective_type: body.objective && body.objective.objective_type, + ready, + service: SERVICE_NAME, + version: SERVICE_VERSION, + checks: { + engine_configured: Boolean(ENGINE_URL), + engine_reachable: ready, + engine_status_code: upstream.statusCode, + }, }, }; } -const AGENT_HANDLERS = { - workflow: handleWorkflow, - scheduler: handleScheduler, - dependencies: handleDependencies, - retry: handleRetry, - parallel: handleParallel, - 'state-machine': handleStateMachine, - swarm: handleSwarm, -}; - -// ============================================================================= -// Health Endpoint -// ============================================================================= - -function handleHealth(req) { - const agentStatuses = {}; - for (const agent of AGENTS) { - agentStatuses[agent] = { - status: 'healthy', - route: AGENT_ROUTES[agent], - name: AGENT_DISPLAY_NAMES[agent], - }; - } - - return { - status: 'healthy', - service: SERVICE_NAME, - version: SERVICE_VERSION, - agents: agentStatuses, - agents_list: AGENTS, - base_url: BASE_URL, - timestamp: new Date().toISOString(), - }; -} - // ============================================================================= -// Contract Endpoint +// Contract Endpoint (served locally -- it is static metadata, not orchestration) // ============================================================================= function handleContracts(agentName) { @@ -321,7 +305,7 @@ function parseRoute(path) { * @param {import('express').Request} req * @param {import('express').Response} res */ -const handler = (req, res) => { +const handler = async (req, res) => { const startTime = Date.now(); // CORS preflight @@ -341,7 +325,8 @@ const handler = (req, res) => { const body = buildResponse(req, 'ROUTER', 'completed', { service: SERVICE_NAME, version: SERVICE_VERSION, - description: 'Orchestrator Agents Cloud Function — routes to 7 orchestration agents.', + description: 'Orchestrator Agents Cloud Function — authenticating proxy to the Rust orchestration engine.', + engine_url: ENGINE_URL || null, agents: AGENTS.map((a) => ({ name: AGENT_DISPLAY_NAMES[a], route: AGENT_ROUTES[a], @@ -361,9 +346,9 @@ const handler = (req, res) => { // Health // ------------------------------------------------------------------ case 'health': { - const health = handleHealth(req); - const body = buildResponse(req, 'HEALTH', 'completed', health, startTime); - res.status(200).json(body); + const result = await handleHealth(req); + const body = buildResponse(req, 'HEALTH', result.statusCode === 200 ? 'completed' : 'error', result.data, startTime); + res.status(result.statusCode).json(body); return; } @@ -371,17 +356,9 @@ const handler = (req, res) => { // Ready // ------------------------------------------------------------------ case 'ready': { - const body = buildResponse(req, 'READY', 'completed', { - ready: true, - service: SERVICE_NAME, - version: SERVICE_VERSION, - checks: { - agents_registered: AGENTS.length === 7, - contracts_loaded: Object.keys(AGENT_CONTRACTS).length === 7, - handlers_available: Object.keys(AGENT_HANDLERS).length === 7, - }, - }, startTime); - res.status(200).json(body); + const result = await handleReady(req); + const body = buildResponse(req, 'READY', result.statusCode === 200 ? 'completed' : 'error', result.data, startTime); + res.status(result.statusCode).json(body); return; } @@ -396,7 +373,7 @@ const handler = (req, res) => { } // ------------------------------------------------------------------ - // Agent routes + // Agent routes — proxied to the engine // ------------------------------------------------------------------ case 'agent': { if (req.method !== 'POST') { @@ -408,18 +385,18 @@ const handler = (req, res) => { return; } - const agentHandler = AGENT_HANDLERS[route.agent]; - if (!agentHandler) { - const body = buildResponse(req, route.agent, 'error', { - error: `No handler for agent: ${route.agent}`, - }, startTime); - res.status(500).json(body); + const result = await proxyToEngine(req, AGENT_ROUTES[route.agent], req.body || {}); + + if (result.passthrough) { + // The engine already applied the envelope. Forward status and body unmodified so there + // is exactly one source of truth for what happened. + res.set('X-Correlation-ID', result.correlationId); + res.status(result.statusCode).json(result.data); return; } - const reqBody = req.body || {}; - const result = agentHandler(reqBody); - const body = buildResponse(req, route.agent, result.statusCode === 200 ? 'completed' : 'error', result.data, startTime); + // Proxy-level failure: the engine never answered, so this function describes its own error. + const body = buildResponse(req, route.agent, 'error', result.data, startTime); res.status(result.statusCode).json(body); return; } diff --git a/functions/test.js b/functions/test.js index 7e31204..81af056 100644 --- a/functions/test.js +++ b/functions/test.js @@ -2,8 +2,23 @@ // ============================================================================= // Tests for orchestrator-agents Cloud Function +// +// The function is a proxy, so these assert proxy behaviour: that requests reach the engine with +// the right method, path and correlation ID; that the engine's status and body come back +// unmodified; and that every failure mode fails closed. They deliberately do NOT assert envelope +// shape on agent routes -- the previous suite asserted only that, and passed against a service +// that orchestrated nothing. +// +// Orchestration correctness is asserted where the orchestration now lives, in +// crates/llm-orchestrator-cli/src/orchestrator_routes.rs: diamond-DAG ordering, cycle rejection, +// and dangling-dependency rejection. See ADR-0001. // ============================================================================= +const assertLib = require('assert'); + +process.env.ORCHESTRATOR_ENGINE_URL = 'https://engine.test.invalid'; +process.env.ORCHESTRATOR_ENGINE_SKIP_AUTH = 'true'; + const { handler } = require('./index'); const { AGENT_CONTRACTS } = require('./contracts'); @@ -42,284 +57,369 @@ function mockRes() { return res; } +// --------------------------------------------------------------------------- +// fetch stub -- records every call so we can assert what reached the engine +// --------------------------------------------------------------------------- + +const realFetch = global.fetch; +let fetchCalls = []; + +function stubFetch(responder) { + fetchCalls = []; + global.fetch = async (url, init) => { + fetchCalls.push({ url, init }); + return responder(url, init); + }; +} + +function engineResponse(status, body) { + return { + ok: status >= 200 && status < 300, + status, + text: async () => (typeof body === 'string' ? body : JSON.stringify(body)), + }; +} + +function restoreFetch() { + global.fetch = realFetch; +} + function assertMetadata(res, testName) { const body = res._body; assert(body.execution_metadata !== undefined, `${testName}: has execution_metadata`); - assert(body.execution_metadata.trace_id !== undefined, `${testName}: has trace_id`); - assert(body.execution_metadata.timestamp !== undefined, `${testName}: has timestamp`); assert(body.execution_metadata.service === 'orchestrator-agents', `${testName}: service is orchestrator-agents`); - assert(body.execution_metadata.execution_id !== undefined, `${testName}: has execution_id`); assert(Array.isArray(body.layers_executed), `${testName}: has layers_executed array`); - assert(body.layers_executed.length >= 2, `${testName}: layers_executed has >= 2 entries`); assert(body.layers_executed[0].layer === 'AGENT_ROUTING', `${testName}: first layer is AGENT_ROUTING`); } -// ============================================================================= -// Test: Root endpoint -// ============================================================================= -console.log('\n--- Root Endpoint ---'); -{ - const req = mockReq('GET', '/'); - const res = mockRes(); - handler(req, res); - assert(res._status === 200, 'Root returns 200'); - assert(res._body.service === 'orchestrator-agents', 'Root returns correct service'); - assert(res._body.version === '0.1.5', 'Root returns correct version'); - assert(Array.isArray(res._body.agents), 'Root lists agents'); - assert(res._body.agents.length === 7, 'Root lists 7 agents'); - assertMetadata(res, 'Root'); -} +async function run() { + // ========================================================================= + console.log('\n--- Agent routes reach the engine ---'); + // ========================================================================= + { + stubFetch(() => engineResponse(200, { status: 'resolved', execution_order: ['A', 'B'] })); -// ============================================================================= -// Test: Health endpoint -// ============================================================================= -console.log('\n--- Health Endpoint ---'); -{ - const req = mockReq('GET', '/health'); - const res = mockRes(); - handler(req, res); - assert(res._status === 200, 'Health returns 200'); - assert(res._body.status === 'healthy', 'Health status is healthy'); - assert(res._body.service === 'orchestrator-agents', 'Health service correct'); - const agents = res._body.agents_list; - assert(Array.isArray(agents), 'Health agents_list is array'); - assert(agents.length === 7, 'Health lists 7 agents'); - assert(agents.includes('workflow'), 'Health includes workflow'); - assert(agents.includes('scheduler'), 'Health includes scheduler'); - assert(agents.includes('dependencies'), 'Health includes dependencies'); - assert(agents.includes('retry'), 'Health includes retry'); - assert(agents.includes('parallel'), 'Health includes parallel'); - assert(agents.includes('state-machine'), 'Health includes state-machine'); - assert(agents.includes('swarm'), 'Health includes swarm'); - assertMetadata(res, 'Health'); -} + const req = mockReq('POST', '/v1/orchestrator/dependencies', { + request_id: 'r-1', + workflow_id: 'wf-1', + tasks: [{ task_id: 'A', name: 'ingest' }], + }, { 'x-correlation-id': 'trace-123' }); + const res = mockRes(); + await handler(req, res); -// ============================================================================= -// Test: Ready endpoint -// ============================================================================= -console.log('\n--- Ready Endpoint ---'); -{ - const req = mockReq('GET', '/ready'); - const res = mockRes(); - handler(req, res); - assert(res._status === 200, 'Ready returns 200'); - assert(res._body.ready === true, 'Ready is true'); - assert(res._body.checks.agents_registered === true, 'All agents registered'); - assert(res._body.checks.contracts_loaded === true, 'All contracts loaded'); - assert(res._body.checks.handlers_available === true, 'All handlers available'); - assertMetadata(res, 'Ready'); -} + assert(fetchCalls.length === 1, 'Engine was called exactly once'); + const call = fetchCalls[0]; + assert( + call.url === 'https://engine.test.invalid/v1/orchestrator/dependencies', + `Engine URL is the dependencies route (got ${call.url})`, + ); + assert(call.init.method === 'POST', 'Engine is called with POST'); + assert( + call.init.headers['X-Correlation-ID'] === 'trace-123', + 'Caller correlation ID is propagated upstream', + ); + assert( + JSON.parse(call.init.body).tasks[0].task_id === 'A', + 'Request body is forwarded to the engine', + ); + restoreFetch(); + } -// ============================================================================= -// Test: CORS preflight -// ============================================================================= -console.log('\n--- CORS ---'); -{ - const req = mockReq('OPTIONS', '/v1/orchestrator/workflow'); - const res = mockRes(); - handler(req, res); - assert(res._status === 204, 'OPTIONS returns 204'); - assert(res._headers['Access-Control-Allow-Origin'] === '*', 'CORS origin header set'); - assert(res._headers['Access-Control-Allow-Methods'].includes('POST'), 'CORS methods include POST'); - assert(res._headers['Access-Control-Allow-Headers'].includes('X-Correlation-ID'), 'CORS headers include X-Correlation-ID'); -} + // ========================================================================= + console.log('\n--- Engine response is forwarded unmodified ---'); + // ========================================================================= + { + const engineBody = { + status: 'resolved', + execution_order: ['A', 'B', 'C', 'D'], + execution_metadata: { trace_id: 'from-engine', service: 'orchestrator-agents' }, + layers_executed: [{ layer: 'AGENT_ROUTING', status: 'completed' }], + }; + stubFetch(() => engineResponse(200, engineBody)); -// ============================================================================= -// Test: X-Correlation-ID propagation -// ============================================================================= -console.log('\n--- Correlation ID ---'); -{ - const traceId = 'test-trace-12345'; - const req = mockReq('GET', '/health', {}, { 'x-correlation-id': traceId }); - const res = mockRes(); - handler(req, res); - assert(res._body.execution_metadata.trace_id === traceId, 'trace_id matches X-Correlation-ID header'); -} + const res = mockRes(); + await handler(mockReq('POST', '/v1/orchestrator/dependencies', { request_id: 'r', workflow_id: 'w', tasks: [] }), res); -// ============================================================================= -// Test: Contracts endpoint -// ============================================================================= -console.log('\n--- Contracts ---'); -{ - const req = mockReq('GET', '/v1/orchestrator/contracts'); - const res = mockRes(); - handler(req, res); - assert(res._status === 200, 'Contracts returns 200'); - assert(Object.keys(res._body).length > 7, 'Contracts returns all agents (+ metadata)'); - assertMetadata(res, 'Contracts'); -} -{ - const req = mockReq('GET', '/v1/orchestrator/contracts/workflow'); - const res = mockRes(); - handler(req, res); - assert(res._status === 200, 'Single contract returns 200'); - assert(res._body.agent_id === 'workflow-orchestrator', 'Workflow contract agent_id correct'); -} + assert(res._status === 200, 'Upstream 200 is propagated'); + assertLib.deepStrictEqual(res._body, engineBody); + assert(true, 'Body is identical to the engine response -- the proxy adds nothing'); + assert( + res._body.execution_metadata.trace_id === 'from-engine', + 'The engine owns the envelope; the proxy does not overwrite it', + ); + restoreFetch(); + } -// ============================================================================= -// Test: All 7 Agent Routes -// ============================================================================= + // ========================================================================= + console.log('\n--- Engine error statuses are propagated, not swallowed ---'); + // ========================================================================= + { + stubFetch(() => engineResponse(400, { error: 'Cyclic dependency detected' })); + + const res = mockRes(); + await handler(mockReq('POST', '/v1/orchestrator/dependencies', { request_id: 'r', workflow_id: 'w', tasks: [] }), res); + + assert(res._status === 400, 'A cyclic graph comes back as 400, not 200'); + assert(res._body.error === 'Cyclic dependency detected', 'The engine error message reaches the caller'); + restoreFetch(); + } + + // ========================================================================= + console.log('\n--- All seven agents route to their own engine path ---'); + // ========================================================================= + { + const expected = { + workflow: '/v1/orchestrator/workflow', + scheduler: '/v1/orchestrator/scheduler', + dependencies: '/v1/orchestrator/dependencies', + retry: '/v1/orchestrator/retry', + parallel: '/v1/orchestrator/parallel', + 'state-machine': '/v1/orchestrator/state-machine', + swarm: '/v1/orchestrator/swarm', + }; + + for (const [agent, path] of Object.entries(expected)) { + stubFetch(() => engineResponse(200, { ok: true })); + const res = mockRes(); + await handler(mockReq('POST', `/v1/orchestrator/${agent}`, {}), res); + assert( + fetchCalls.length === 1 && fetchCalls[0].url === `https://engine.test.invalid${path}`, + `${agent} proxies to ${path}`, + ); + restoreFetch(); + } + } + + // ========================================================================= + console.log('\n--- Fails closed when the engine is unreachable ---'); + // ========================================================================= + { + stubFetch(() => { throw new Error('ECONNREFUSED'); }); + + const res = mockRes(); + await handler(mockReq('POST', '/v1/orchestrator/dependencies', { request_id: 'r', workflow_id: 'w', tasks: [] }), res); + + assert(res._status === 502, 'An unreachable engine yields 502, never a fabricated 200'); + assert(/ECONNREFUSED/.test(res._body.error), 'The transport failure is reported'); + restoreFetch(); + } + + // ========================================================================= + console.log('\n--- Fails closed on timeout ---'); + // ========================================================================= + { + stubFetch(() => { + const err = new Error('aborted'); + err.name = 'AbortError'; + throw err; + }); + + const res = mockRes(); + await handler(mockReq('POST', '/v1/orchestrator/parallel', {}), res); + + assert(res._status === 504, 'A timeout yields 504'); + restoreFetch(); + } + + // ========================================================================= + console.log('\n--- Fails closed on a non-JSON engine response ---'); + // ========================================================================= + { + stubFetch(() => engineResponse(200, '502 Bad Gateway')); + + const res = mockRes(); + await handler(mockReq('POST', '/v1/orchestrator/swarm', {}), res); -const agentTests = [ + assert(res._status === 502, 'A non-JSON upstream body yields 502'); + restoreFetch(); + } + + // ========================================================================= + console.log('\n--- Readiness fails closed (ADR-0001 Test 5) ---'); + // ========================================================================= { - name: 'Workflow Orchestrator', - path: '/v1/orchestrator/workflow', - body: { workflow_id: 'wf-1', workflow_name: 'test-workflow', tasks: [{ task_id: 't1', name: 'step1', task_type: 'transform' }] }, - agentId: 'workflow-orchestrator', - layerPrefix: 'ORCHESTRATOR_WORKFLOW', - }, + stubFetch(() => { throw new Error('ENOTFOUND'); }); + + const res = mockRes(); + await handler(mockReq('GET', '/ready'), res); + + assert(res._status !== 200, `/ready is non-200 when the engine is unreachable (got ${res._status})`); + assert(res._body.ready === false, '/ready reports ready: false'); + assert(res._body.checks.engine_reachable === false, '/ready reports the engine unreachable'); + restoreFetch(); + } + { - name: 'Task Scheduler', - path: '/v1/orchestrator/scheduler', - body: { schedule_id: 'sch-1', tasks: [{ task_id: 't1', name: 'job1', schedule_type: 'immediate' }] }, - agentId: 'task-scheduler', - layerPrefix: 'ORCHESTRATOR_SCHEDULER', - }, + stubFetch(() => engineResponse(200, { ready: true })); + + const res = mockRes(); + await handler(mockReq('GET', '/ready'), res); + + assert(res._status === 200, '/ready is 200 when the engine answers'); + assert(res._body.ready === true, '/ready reports ready: true'); + restoreFetch(); + } + + // ========================================================================= + console.log('\n--- Health reflects the engine, not a constant ---'); + // ========================================================================= { - name: 'Dependency Resolver', - path: '/v1/orchestrator/dependencies', - body: { request_id: 'req-1', workflow_id: 'wf-1', tasks: [{ task_id: 't1', name: 'step1' }] }, - agentId: 'dependency-resolver', - layerPrefix: 'ORCHESTRATOR_DEPENDENCIES', - }, + stubFetch(() => { throw new Error('ECONNREFUSED'); }); + + const res = mockRes(); + await handler(mockReq('GET', '/health'), res); + + assert(res._status === 503, '/health is 503 when the engine is down'); + assert(res._body.status === 'unhealthy', '/health reports unhealthy'); + assert( + res._body.agents.workflow.status === 'unavailable', + 'Per-agent status follows the engine rather than being hardcoded healthy', + ); + restoreFetch(); + } + { - name: 'Retry & Recovery', - path: '/v1/orchestrator/retry', - body: { request_id: 'req-1', failure: { task_id: 't1', error_category: 'transient' } }, - agentId: 'retry-recovery', - layerPrefix: 'ORCHESTRATOR_RETRY', - }, + stubFetch(() => engineResponse(200, { status: 'healthy' })); + + const res = mockRes(); + await handler(mockReq('GET', '/health', {}, { 'x-correlation-id': 'trace-health' }), res); + + assert(res._status === 200, '/health is 200 when the engine answers'); + assert(res._body.status === 'healthy', '/health reports healthy'); + assert(res._body.agents_list.length === 7, '/health lists 7 agents'); + assert(res._body.execution_metadata.trace_id === 'trace-health', 'trace_id matches X-Correlation-ID'); + assertMetadata(res, 'Health'); + restoreFetch(); + } + + // ========================================================================= + console.log('\n--- The function no longer orchestrates locally ---'); + // ========================================================================= { - name: 'Parallelization', - path: '/v1/orchestrator/parallel', - body: { request_id: 'req-1', workflow_id: 'wf-1', tasks: [{ task_id: 't1', name: 'step1' }] }, - agentId: 'parallelization-agent', - layerPrefix: 'ORCHESTRATOR_PARALLEL', - }, + // The old handleStateMachine validated transitions itself against the contract table. + // If any local orchestration survived, the engine would not be called. + stubFetch(() => engineResponse(200, { status: 'completed' })); + + const res = mockRes(); + await handler(mockReq('POST', '/v1/orchestrator/state-machine', { + request_id: 'r', execution_id: 'e', entity_type: 'task', + current_state: 'completed', target_state: 'running', + reason: 'test', initiated_by: 'test', + }), res); + + assert(fetchCalls.length === 1, 'state-machine defers to the engine instead of deciding locally'); + restoreFetch(); + + const source = require('fs').readFileSync(require.resolve('./index.js'), 'utf8'); + assert( + !/contract\.state_machine\[/.test(source), + 'The local transition-table lookup has been deleted -- one source of truth', + ); + } + + // ========================================================================= + console.log('\n--- Unconfigured engine fails closed ---'); + // ========================================================================= { - name: 'State Machine', - path: '/v1/orchestrator/state-machine', - body: { - request_id: 'req-1', execution_id: 'exec-1', entity_type: 'workflow', - current_state: 'pending', target_state: 'running', reason: 'start execution', initiated_by: 'test', - }, - agentId: 'state-machine-agent', - layerPrefix: 'ORCHESTRATOR_STATE_MACHINE', - }, + delete require.cache[require.resolve('./index.js')]; + const savedUrl = process.env.ORCHESTRATOR_ENGINE_URL; + delete process.env.ORCHESTRATOR_ENGINE_URL; + const { handler: unconfigured } = require('./index.js'); + + const res = mockRes(); + await unconfigured(mockReq('POST', '/v1/orchestrator/workflow', {}), res); + assert(res._status === 503, 'An unconfigured engine yields 503'); + assert(res._body.engine_configured === false, 'The response says the engine is unconfigured'); + + const readyRes = mockRes(); + await unconfigured(mockReq('GET', '/ready'), readyRes); + assert(readyRes._status === 503, '/ready is 503 with no engine configured'); + + process.env.ORCHESTRATOR_ENGINE_URL = savedUrl; + delete require.cache[require.resolve('./index.js')]; + } + + // ========================================================================= + console.log('\n--- Routing, CORS and contracts stay local ---'); + // ========================================================================= { - name: 'Swarm Coordinator', - path: '/v1/orchestrator/swarm', - body: { - request_id: 'req-1', workflow_id: 'wf-1', - objective: { id: 'obj-1', description: 'test', objective_type: 'analysis' }, - workers: [{ worker_id: 'w1', agent_type: 'workflow_orchestrator', task: { task_id: 't1', description: 'test' } }], - }, - agentId: 'swarm-coordinator-agent', - layerPrefix: 'ORCHESTRATOR_SWARM', - }, -]; - -for (const test of agentTests) { - console.log(`\n--- ${test.name} Agent ---`); - - // Valid POST + const res = mockRes(); + await handler(mockReq('OPTIONS', '/v1/orchestrator/workflow'), res); + assert(res._status === 204, 'CORS preflight returns 204'); + assert(res._headers['Access-Control-Allow-Origin'] === '*', 'CORS origin header set'); + assert(res._headers['Access-Control-Allow-Methods'].includes('POST'), 'CORS methods include POST'); + assert(res._headers['Access-Control-Allow-Headers'].includes('X-Correlation-ID'), 'CORS headers include X-Correlation-ID'); + } + { - const req = mockReq('POST', test.path, test.body); const res = mockRes(); - handler(req, res); - assert(res._status === 200, `${test.name}: valid POST returns 200`); - assert(res._body.agent === test.agentId, `${test.name}: agent_id is ${test.agentId}`); - assertMetadata(res, test.name); - assert(res._body.layers_executed[1].layer === test.layerPrefix, `${test.name}: layer is ${test.layerPrefix}`); - assert(res._body.layers_executed[1].status === 'completed', `${test.name}: layer status is completed`); - assert(typeof res._body.layers_executed[1].duration_ms === 'number', `${test.name}: duration_ms is number`); + await handler(mockReq('GET', '/'), res); + assert(res._status === 200, 'Root returns 200'); + assert(res._body.agents.length === 7, 'Root lists 7 agents'); + assertMetadata(res, 'Root'); } - // Missing required fields { - const req = mockReq('POST', test.path, {}); const res = mockRes(); - handler(req, res); - assert(res._status === 400, `${test.name}: empty body returns 400`); - assert(res._body.error !== undefined, `${test.name}: error message present`); - assertMetadata(res, `${test.name} (400)`); + await handler(mockReq('GET', '/v1/orchestrator/contracts'), res); + assert(res._status === 200, 'Contracts are served without calling the engine'); + assert(Object.keys(res._body).length >= 7, 'All contracts returned'); } - // Wrong method { - const req = mockReq('GET', test.path); const res = mockRes(); - handler(req, res); - assert(res._status === 405, `${test.name}: GET returns 405`); + await handler(mockReq('GET', '/v1/orchestrator/contracts/dependencies'), res); + assert(res._status === 200, 'Single contract returns 200'); + assert(res._body.agent_id === 'dependency-resolver', 'Correct contract returned'); } -} -// ============================================================================= -// Test: State Machine validation logic -// ============================================================================= -console.log('\n--- State Machine Transition Validation ---'); -{ - const req = mockReq('POST', '/v1/orchestrator/state-machine', { - request_id: 'req-1', execution_id: 'exec-1', entity_type: 'workflow', - current_state: 'completed', target_state: 'running', reason: 'invalid', initiated_by: 'test', - }); - const res = mockRes(); - handler(req, res); - assert(res._status === 200, 'Invalid transition still returns 200'); - assert(res._body.success === false, 'Invalid transition reports success=false'); - assert(res._body.status === 'invalid', 'Invalid transition reports status=invalid'); - assert(res._body.new_state === 'completed', 'Invalid transition preserves current state'); -} -{ - const req = mockReq('POST', '/v1/orchestrator/state-machine', { - request_id: 'req-2', execution_id: 'exec-2', entity_type: 'workflow', - current_state: 'completed', target_state: 'running', reason: 'forced', initiated_by: 'admin', - force: true, - }); - const res = mockRes(); - handler(req, res); - assert(res._body.success === true, 'Forced transition reports success=true'); - assert(res._body.new_state === 'running', 'Forced transition changes state'); -} + { + const res = mockRes(); + await handler(mockReq('GET', '/v1/orchestrator/contracts/nope'), res); + assert(res._status === 404, 'Unknown contract returns 404'); + } -// ============================================================================= -// Test: 404 -// ============================================================================= -console.log('\n--- 404 ---'); -{ - const req = mockReq('GET', '/v1/orchestrator/nonexistent'); - const res = mockRes(); - handler(req, res); - assert(res._status === 404, 'Unknown route returns 404'); - assert(Array.isArray(res._body.available_routes), '404 includes available_routes'); - assertMetadata(res, '404'); -} + { + const res = mockRes(); + await handler(mockReq('GET', '/v1/orchestrator/dependencies'), res); + assert(res._status === 405, 'GET on an agent route returns 405'); + } -// ============================================================================= -// Test: Contract schemas completeness -// ============================================================================= -console.log('\n--- Contract Schemas ---'); -{ - const expectedAgents = ['workflow', 'scheduler', 'dependencies', 'retry', 'parallel', 'state-machine', 'swarm']; - for (const agent of expectedAgents) { - const contract = AGENT_CONTRACTS[agent]; - assert(contract !== undefined, `Contract exists for ${agent}`); - assert(contract.agent_id !== undefined, `${agent} contract has agent_id`); - assert(contract.agent_version !== undefined, `${agent} contract has agent_version`); - assert(contract.classification !== undefined, `${agent} contract has classification`); - assert(contract.description !== undefined, `${agent} contract has description`); - assert(contract.input !== undefined, `${agent} contract has input schema`); - assert(contract.output !== undefined, `${agent} contract has output schema`); + { + const res = mockRes(); + await handler(mockReq('GET', '/v1/orchestrator/nonexistent'), res); + assert(res._status === 404, 'Unknown route returns 404'); + assert(Array.isArray(res._body.available_routes), '404 includes available_routes'); + assertMetadata(res, '404'); } -} -// ============================================================================= -// Summary -// ============================================================================= -console.log('\n========================================'); -console.log(`Results: ${passed} passed, ${failed} failed, ${passed + failed} total`); -console.log('========================================\n'); + // ========================================================================= + console.log('\n--- Contract schemas ---'); + // ========================================================================= + { + const expectedAgents = ['workflow', 'scheduler', 'dependencies', 'retry', 'parallel', 'state-machine', 'swarm']; + for (const agent of expectedAgents) { + const contract = AGENT_CONTRACTS[agent]; + assert(contract !== undefined, `Contract exists for ${agent}`); + assert(contract.agent_id !== undefined, `${agent} contract has agent_id`); + assert(contract.agent_version !== undefined, `${agent} contract has agent_version`); + assert(contract.classification !== undefined, `${agent} contract has classification`); + assert(contract.description !== undefined, `${agent} contract has description`); + assert(contract.input !== undefined, `${agent} contract has input schema`); + assert(contract.output !== undefined, `${agent} contract has output schema`); + } + } -if (failed > 0) { - process.exit(1); + console.log('\n========================================'); + console.log(`Results: ${passed} passed, ${failed} failed, ${passed + failed} total`); + console.log('========================================\n'); + + if (failed > 0) { + process.exit(1); + } } + +run().catch((err) => { + console.error(err); + process.exit(1); +}); diff --git a/package.json b/package.json index 236a7d8..6ef357b 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "orchestrator-agents", - "version": "0.1.5", + "version": "0.2.0", "description": "Orchestrator Agents Cloud Function — 7 orchestration agents for workflow execution, scheduling, dependency resolution, retry/recovery, parallelization, state management, and swarm coordination.", "main": "index.js", "scripts": { @@ -12,8 +12,7 @@ "node": ">=20.0.0" }, "dependencies": { - "@google-cloud/functions-framework": "^3.4.0", - "claude-flow": "^3.0.0-alpha.179" + "@google-cloud/functions-framework": "^3.4.0" }, "private": true, "license": "Apache-2.0" From cb880c50078cd67193a5bb3105e5d67eb0d376d6 Mon Sep 17 00:00:00 2001 From: Nick Ruest <127058086+nicholas-ruest@users.noreply.github.com> Date: Mon, 27 Jul 2026 06:35:07 +0000 Subject: [PATCH 2/4] docs: regenerate the test certification after adding the HTTP routes 445 passed, 3 failed, 2 skipped of 450, up from 437 -- the 13 new tests are the orchestrator_routes suite. llm-orchestrator-core remains 198/198 and the three failures are the same pre-existing ones ADR-0002 surfaced, unchanged by this work. Implements orchestrator/docs/adr/ADR-0001-implement-real-orchestration-logic.md. --- docs/TEST_CERTIFICATION.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/docs/TEST_CERTIFICATION.md b/docs/TEST_CERTIFICATION.md index bd54f16..875a933 100644 --- a/docs/TEST_CERTIFICATION.md +++ b/docs/TEST_CERTIFICATION.md @@ -9,12 +9,12 @@ | Field | Value | |---|---| -| Commit | `a5ecc9fadcae1179e4699f4c549f02358dbaead2` | +| Commit | `de5363caefb714057df1851505246e63ff8803b5` | | CI run ID | `local-2026-07-27` | | CI run URL | n/a — generated from a local run, not CI | | Command | `cargo nextest run --all --lib --bins --no-fail-fast --message-format libtest-json` | | Toolchain | `rustc 1.97.1 (8bab26f4f 2026-07-14)` | -| Generated at (UTC) | `2026-07-27T06:00:28Z` | +| Generated at (UTC) | `2026-07-27T06:34:57Z` | ## Results by crate @@ -26,11 +26,12 @@ | `llm-orchestrator-audit` | 16 | 0 | 0 | 16 | | `llm-orchestrator-auth` | 52 | 0 | 0 | 52 | | `llm-orchestrator-benchmarks` | 17 | 0 | 0 | 17 | +| `llm-orchestrator-cli` | 13 | 0 | 0 | 13 | | `llm-orchestrator-core` | 198 | 0 | 0 | 198 | | `llm-orchestrator-providers` | 48 | 2 | 0 | 50 | | `llm-orchestrator-secrets` | 21 | 1 | 1 | 23 | | `llm-orchestrator-state` | 30 | 0 | 1 | 31 | -| **Total** | **432** | **3** | **2** | **437** | +| **Total** | **445** | **3** | **2** | **450** | ## Failures From 8c1549f04378def43d468b41a8a1ce9d6c185eb9 Mon Sep 17 00:00:00 2001 From: Nick Ruest <127058086+nicholas-ruest@users.noreply.github.com> Date: Mon, 27 Jul 2026 14:29:39 +0000 Subject: [PATCH 3/4] ci: run the Cloud Function suite and guard against local orchestration returning The Cloud Function's tests had no CI job at all -- before this branch the suite asserted only envelope shape, and even that never ran. Now that the function is a proxy, its 113 tests assert the properties that matter: every agent request reaches the engine on the right path with the correlation ID attached, the engine's status and body come back unmodified, and every failure mode fails closed (unconfigured 503, unreachable 502, timeout 504, non-JSON upstream 502, readiness non-200). No install step is needed: no lockfile is committed and the suite stubs fetch rather than reaching the network. A second step greps for the two constructs that would mean the function had started answering by itself again -- the `tasks_count: Array.isArray(...)` echo and the local state-machine transition-table lookup. Both were verified to fire against the version of functions/index.js on main, so they are not vacuous checks. The Rust route tests, including ADR-0001's diamond-DAG, cycle and dangling-dependency verification tests, need no new job: they are bin-target tests and already run in the nextest step this branch inherits from ADR-0002. They appear in the certification as `llm-orchestrator-cli | 13 | 0 | 0 | 13`. Not done, and not fakeable here: step 12's containerised end-to-end job. `serve_http` calls `validate_phase3_startup`, which requires RUVECTOR_SERVICE_ENDPOINT and RUVECTOR_API_KEY and hard-exits when the Ruvector service is unreachable, deliberately, to trigger a Cloud Run crashloop on misconfiguration. A CI container would therefore crashloop, and standing up a fake Ruvector is new infrastructure ADR-0001 does not specify. Recorded in the pull request instead of guessed at. Implements step 12 of orchestrator/docs/adr/ADR-0001-implement-real-orchestration-logic.md, in part. --- .github/workflows/ci.yml | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cfb25d0..78a3d42 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -111,6 +111,40 @@ jobs: - name: Check for unprovenanced readiness claims run: python3 scripts/check-certification-integrity.py docs + # The Cloud Function is a proxy since ADR-0001. Its suite asserts that every agent request + # reaches the engine and that every failure mode fails closed -- the properties that stop it + # regressing to answering requests by itself. Nothing ran it before. + cloud-function: + name: Cloud Function + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Set up Node + uses: actions/setup-node@v4 + with: + node-version: '20' + + # No lockfile is committed and the suite stubs fetch rather than reaching the network, + # so the tests need no install step. + - name: Run Cloud Function tests + run: npm test + + # The proxy must never regress to computing answers locally. These greps fail if the + # handlers that echoed tasks.length, or the local state-machine transition table, return. + - name: Assert the function performs no orchestration + run: | + set -e + if grep -nE 'tasks_count: Array\.isArray' functions/index.js; then + echo "::error::functions/index.js is computing a result locally again. See ADR-0001." + exit 1 + fi + if grep -nE 'contract\.state_machine\[' functions/index.js; then + echo "::error::The local state-machine transition table is back. See ADR-0001." + exit 1 + fi + echo "functions/index.js delegates to the engine." + lint: name: Lint runs-on: ubuntu-latest From 1704b42e1706f80801b3903ef78dd914fa151a5d Mon Sep 17 00:00:00 2001 From: Nick Ruest <127058086+nicholas-ruest@users.noreply.github.com> Date: Mon, 27 Jul 2026 14:38:57 +0000 Subject: [PATCH 4/4] style: rustfmt the orchestrator_routes module The Lint job runs `cargo fmt --all -- --check`, and the new module had 5 diffs. Formatted with rustfmt directly rather than `cargo fmt -p` so main.rs's 38 pre-existing diffs are not swept into this branch as unrelated churn. The job stays red regardless: the repository has 521 formatting diffs on main and this branch does not take that on. Clippy reports nothing against the new module. --- .../src/orchestrator_routes.rs | 34 ++++++++++++------- 1 file changed, 22 insertions(+), 12 deletions(-) diff --git a/crates/llm-orchestrator-cli/src/orchestrator_routes.rs b/crates/llm-orchestrator-cli/src/orchestrator_routes.rs index 14c02af..c7c6c34 100644 --- a/crates/llm-orchestrator-cli/src/orchestrator_routes.rs +++ b/crates/llm-orchestrator-cli/src/orchestrator_routes.rs @@ -299,13 +299,7 @@ macro_rules! prepare { match parse_request::<$ty>($body) { Ok(request) => request, Err(message) => { - return error_response( - $headers, - $agent, - StatusCode::BAD_REQUEST, - message, - $started, - ) + return error_response($headers, $agent, StatusCode::BAD_REQUEST, message, $started) } } }}; @@ -325,7 +319,9 @@ macro_rules! respond { } fn ruvector_endpoint() -> Option { - std::env::var(RUVECTOR_ENDPOINT_ENV).ok().filter(|s| !s.is_empty()) + std::env::var(RUVECTOR_ENDPOINT_ENV) + .ok() + .filter(|s| !s.is_empty()) } // ============================================================================ @@ -591,7 +587,10 @@ where .route("/v1/orchestrator/dependencies", post(dependencies_handler)) .route("/v1/orchestrator/retry", post(retry_handler)) .route("/v1/orchestrator/parallel", post(parallel_handler)) - .route("/v1/orchestrator/state-machine", post(state_machine_handler)) + .route( + "/v1/orchestrator/state-machine", + post(state_machine_handler), + ) .route("/v1/orchestrator/swarm", post(swarm_handler)) } @@ -652,7 +651,13 @@ mod tests { let mut payload = Map::new(); payload.insert("status".into(), json!("resolved")); - let value = envelope(&headers, &DEPENDENCIES, "completed", payload, Instant::now()); + let value = envelope( + &headers, + &DEPENDENCIES, + "completed", + payload, + Instant::now(), + ); // Agent identity, as published in the contract. assert_eq!(value["agent"], json!("dependency-resolver")); @@ -742,7 +747,9 @@ mod tests { assert!(position("C") < position("D")); // B and C are independent and must share a parallel group. - let groups = value["parallel_groups"].as_array().expect("parallel_groups"); + let groups = value["parallel_groups"] + .as_array() + .expect("parallel_groups"); assert!( groups.iter().any(|group| { let ids: Vec<&str> = group["task_ids"] @@ -772,7 +779,10 @@ mod tests { let value = read_body(response).await; let error = value["error"].as_str().unwrap_or_default().to_lowercase(); - assert!(error.contains("cycle"), "error did not name a cycle: {error}"); + assert!( + error.contains("cycle"), + "error did not name a cycle: {error}" + ); } #[tokio::test]