diff --git a/crates/traverse-contracts/src/lib.rs b/crates/traverse-contracts/src/lib.rs index 3bfc84cf..003cf2ec 100644 --- a/crates/traverse-contracts/src/lib.rs +++ b/crates/traverse-contracts/src/lib.rs @@ -9,9 +9,12 @@ pub mod proposal; pub mod usage_telemetry; pub mod violations; pub use proposal::{ - CanonicalProposal, ManifestReference, MappingSource, ProposalEdge, ProposalLimits, - ProposalMapping, ProposalNode, ProposalValidationError, ProposalValidationErrorCode, - ProposalValidationFailure, SnapshotDigests, WorkflowProposal, canonicalize_proposal, + CanonicalProposal, DEFAULT_MAX_CONCURRENT_NODES, DEFAULT_MAX_FAN_OUT, DEFAULT_MAX_JOIN_WIDTH, + DEFAULT_MAX_QUEUE_DEPTH, ManifestReference, MappingSource, ParallelSchedule, + ParallelScheduleError, ParallelScheduleErrorCode, ParallelScheduleFailure, + ParallelScheduleLimits, ProposalEdge, ProposalLimits, ProposalMapping, ProposalNode, + ProposalValidationError, ProposalValidationErrorCode, ProposalValidationFailure, + SnapshotDigests, WorkflowProposal, canonicalize_proposal, compute_parallel_schedule, proposal_digest, proposal_snapshot_digest, }; pub use usage_telemetry::{NoOpUsageTelemetrySink, UsageEvent, UsageEventKind, UsageTelemetrySink}; diff --git a/crates/traverse-contracts/src/proposal.rs b/crates/traverse-contracts/src/proposal.rs index 4e4b9955..4ef8054f 100644 --- a/crates/traverse-contracts/src/proposal.rs +++ b/crates/traverse-contracts/src/proposal.rs @@ -431,6 +431,172 @@ fn topological_order( } } +// --------------------------------------------------------------------------- +// Bounded parallel scheduling (spec 110 P2, ADR-0042) +// --------------------------------------------------------------------------- + +/// Configured parallel-scheduling bounds a P2 execution schedule must fall +/// within (spec 110 FR-001, FR-005). Values are host configuration, never +/// caller-supplied. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ParallelScheduleLimits { + /// Max node ids eligible to become ready and run concurrently in any + /// single wave. + pub max_fan_out: usize, + /// Max direct predecessors (in-degree) converging into any single node. + pub max_join_width: usize, + /// Max total node ids across the whole schedule. + pub max_queue_depth: usize, + /// Max node executions the runtime may dispatch at once within a wave. + pub max_concurrent_nodes: usize, +} + +pub const DEFAULT_MAX_FAN_OUT: usize = 8; +pub const DEFAULT_MAX_JOIN_WIDTH: usize = 8; +pub const DEFAULT_MAX_QUEUE_DEPTH: usize = 16; +pub const DEFAULT_MAX_CONCURRENT_NODES: usize = 8; + +impl Default for ParallelScheduleLimits { + fn default() -> Self { + Self { + max_fan_out: DEFAULT_MAX_FAN_OUT, + max_join_width: DEFAULT_MAX_JOIN_WIDTH, + max_queue_depth: DEFAULT_MAX_QUEUE_DEPTH, + max_concurrent_nodes: DEFAULT_MAX_CONCURRENT_NODES, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct ParallelScheduleError { + pub code: ParallelScheduleErrorCode, + pub message: String, + pub path: String, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ParallelScheduleErrorCode { + FanOutExceeded, + JoinWidthExceeded, + QueueDepthExceeded, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ParallelScheduleFailure { + pub errors: Vec, +} + +/// A bounded schedule of concurrency waves over an already-canonicalized +/// proposal (spec 110 FR-001, FR-002). Each wave is the set of node ids +/// whose dependencies are fully satisfied by earlier waves, sorted +/// lexicographically for deterministic dispatch order. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ParallelSchedule { + pub waves: Vec>, +} + +/// Levelizes an already-canonicalized, acyclic proposal into concurrency +/// waves and checks it against configured fan-out/join-width/queue-depth +/// bounds (spec 110 FR-001, FR-005: reject before doing any work). Performs +/// no capability-contract lookups — the `pure_read`-only constraint +/// (FR-004a) requires resolved contracts and is enforced in +/// `traverse-runtime`. +/// +/// # Errors +/// +/// Returns [`ParallelScheduleFailure`] listing every exceeded bound. +pub fn compute_parallel_schedule( + canonical: &CanonicalProposal, + limits: &ParallelScheduleLimits, +) -> Result { + let mut errors = Vec::new(); + + let node_ids: BTreeSet = canonical + .proposal + .nodes + .iter() + .map(|node| node.node_id.clone()) + .collect(); + if node_ids.len() > limits.max_queue_depth { + errors.push(ParallelScheduleError { + code: ParallelScheduleErrorCode::QueueDepthExceeded, + message: format!( + "schedule has {} nodes, exceeding the configured queue depth of {}", + node_ids.len(), + limits.max_queue_depth + ), + path: "$.nodes".to_string(), + }); + } + + let mut adjacency: BTreeMap> = BTreeMap::new(); + let mut in_degree: BTreeMap = + node_ids.iter().map(|id| (id.clone(), 0)).collect(); + for edge in &canonical.proposal.edges { + adjacency + .entry(edge.from_node_id.clone()) + .or_default() + .insert(edge.to_node_id.clone()); + *in_degree.entry(edge.to_node_id.clone()).or_insert(0) += 1; + } + + for (node_id, degree) in &in_degree { + if *degree > limits.max_join_width { + errors.push(ParallelScheduleError { + code: ParallelScheduleErrorCode::JoinWidthExceeded, + message: format!( + "node '{node_id}' has {degree} direct predecessors, exceeding the configured join width of {}", + limits.max_join_width + ), + path: format!("$.nodes[?node_id={node_id}]"), + }); + } + } + + let mut remaining_in_degree = in_degree.clone(); + let mut ready: BTreeSet = node_ids + .iter() + .filter(|id| remaining_in_degree.get(*id).copied().unwrap_or(0) == 0) + .cloned() + .collect(); + let mut waves = Vec::new(); + while !ready.is_empty() { + let wave: Vec = ready.iter().cloned().collect(); + if wave.len() > limits.max_fan_out { + errors.push(ParallelScheduleError { + code: ParallelScheduleErrorCode::FanOutExceeded, + message: format!( + "{} nodes became ready concurrently, exceeding the configured fan-out of {}", + wave.len(), + limits.max_fan_out + ), + path: "$.nodes".to_string(), + }); + } + let mut next_ready = BTreeSet::new(); + for node_id in &wave { + if let Some(successors) = adjacency.get(node_id) { + for successor in successors { + let degree = remaining_in_degree.entry(successor.clone()).or_insert(0); + *degree -= 1; + if *degree == 0 { + next_ready.insert(successor.clone()); + } + } + } + } + waves.push(wave); + ready = next_ready; + } + + if errors.is_empty() { + Ok(ParallelSchedule { waves }) + } else { + Err(ParallelScheduleFailure { errors }) + } +} + /// Deterministic, independently-reproducible digest of a proposal's canonical /// JSON form (spec 109 FR-003, FR-007a). Uses recursively key-sorted JSON /// (not Rust `Debug` formatting) hashed with SHA-256, so an external proposer diff --git a/crates/traverse-contracts/tests/proposal.rs b/crates/traverse-contracts/tests/proposal.rs index 348b8724..4bf9ab6d 100644 --- a/crates/traverse-contracts/tests/proposal.rs +++ b/crates/traverse-contracts/tests/proposal.rs @@ -1,8 +1,8 @@ use traverse_contracts::{ - CanonicalProposal, ManifestReference, MappingSource, ProposalEdge, ProposalLimits, - ProposalMapping, ProposalNode, ProposalValidationErrorCode, ProposalValidationFailure, - SnapshotDigests, WorkflowProposal, canonicalize_proposal, proposal_digest, - proposal_snapshot_digest, + CanonicalProposal, ManifestReference, MappingSource, ParallelScheduleErrorCode, + ParallelScheduleLimits, ProposalEdge, ProposalLimits, ProposalMapping, ProposalNode, + ProposalValidationErrorCode, ProposalValidationFailure, SnapshotDigests, WorkflowProposal, + canonicalize_proposal, compute_parallel_schedule, proposal_digest, proposal_snapshot_digest, }; fn expect_failure( @@ -58,6 +58,36 @@ fn linear_proposal() -> WorkflowProposal { } } +fn diamond_proposal() -> WorkflowProposal { + let mut proposal = linear_proposal(); + proposal.nodes = vec![ + node("a", "content.comments.create-comment-draft"), + node("b", "content.comments.publish-comment"), + node("c", "content.comments.publish-comment"), + node("d", "content.comments.publish-comment"), + ]; + proposal.edges = vec![ + ProposalEdge { + from_node_id: "a".to_string(), + to_node_id: "b".to_string(), + }, + ProposalEdge { + from_node_id: "a".to_string(), + to_node_id: "c".to_string(), + }, + ProposalEdge { + from_node_id: "b".to_string(), + to_node_id: "d".to_string(), + }, + ProposalEdge { + from_node_id: "c".to_string(), + to_node_id: "d".to_string(), + }, + ]; + proposal.mappings = Vec::new(); + proposal +} + #[test] fn canonicalizes_a_valid_linear_proposal_in_dependency_order() -> Result<(), String> { let canonical = canonicalize_proposal(linear_proposal(), &ProposalLimits::default()) @@ -531,3 +561,147 @@ fn snapshot_digest_changes_when_any_pinned_snapshot_changes_even_if_proposal_is_ assert_ne!(base, with_changed_policy); } + +// -- Parallel scheduling (spec 110 P2) --------------------------------------- + +#[test] +fn compute_parallel_schedule_levelizes_a_linear_proposal_into_singleton_waves() -> Result<(), String> +{ + let canonical = canonicalize_proposal(linear_proposal(), &ProposalLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + assert_eq!( + schedule.waves, + vec![vec!["a".to_string()], vec!["b".to_string()]] + ); + Ok(()) +} + +#[test] +fn compute_parallel_schedule_levelizes_the_diamond_graph_into_a_concurrent_wave() +-> Result<(), String> { + let canonical = canonicalize_proposal(diamond_proposal(), &ProposalLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + assert_eq!( + schedule.waves, + vec![ + vec!["a".to_string()], + vec!["b".to_string(), "c".to_string()], + vec!["d".to_string()], + ] + ); + Ok(()) +} + +#[test] +fn compute_parallel_schedule_levelizes_the_wide_fan_out_graph() -> Result<(), String> { + let mut proposal = diamond_proposal(); + proposal + .nodes + .push(node("e", "content.comments.publish-comment")); + proposal.edges.push(ProposalEdge { + from_node_id: "a".to_string(), + to_node_id: "e".to_string(), + }); + proposal.edges.push(ProposalEdge { + from_node_id: "e".to_string(), + to_node_id: "d".to_string(), + }); + + let canonical = canonicalize_proposal(proposal, &ProposalLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + assert_eq!( + schedule.waves, + vec![ + vec!["a".to_string()], + vec!["b".to_string(), "c".to_string(), "e".to_string()], + vec!["d".to_string()], + ] + ); + Ok(()) +} + +#[test] +fn rejects_a_schedule_exceeding_the_configured_fan_out_limit() -> Result<(), String> { + let canonical = canonicalize_proposal(diamond_proposal(), &ProposalLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let limits = ParallelScheduleLimits { + max_fan_out: 1, + ..ParallelScheduleLimits::default() + }; + let Err(failure) = compute_parallel_schedule(&canonical, &limits) else { + return Err("a fan-out over the configured limit must be rejected".to_string()); + }; + assert!( + failure + .errors + .iter() + .any(|e| e.code == ParallelScheduleErrorCode::FanOutExceeded) + ); + Ok(()) +} + +#[test] +fn rejects_a_schedule_exceeding_the_configured_join_width_limit() -> Result<(), String> { + let canonical = canonicalize_proposal(diamond_proposal(), &ProposalLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let limits = ParallelScheduleLimits { + max_join_width: 1, + ..ParallelScheduleLimits::default() + }; + let Err(failure) = compute_parallel_schedule(&canonical, &limits) else { + return Err("a join width over the configured limit must be rejected".to_string()); + }; + assert!( + failure + .errors + .iter() + .any(|e| e.code == ParallelScheduleErrorCode::JoinWidthExceeded) + ); + Ok(()) +} + +#[test] +fn rejects_a_schedule_exceeding_the_configured_queue_depth_limit() -> Result<(), String> { + let canonical = canonicalize_proposal(diamond_proposal(), &ProposalLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let limits = ParallelScheduleLimits { + max_queue_depth: 1, + ..ParallelScheduleLimits::default() + }; + let Err(failure) = compute_parallel_schedule(&canonical, &limits) else { + return Err("a queue depth over the configured limit must be rejected".to_string()); + }; + assert!( + failure + .errors + .iter() + .any(|e| e.code == ParallelScheduleErrorCode::QueueDepthExceeded) + ); + Ok(()) +} + +#[test] +fn compute_parallel_schedule_reports_every_exceeded_bound_in_one_pass() -> Result<(), String> { + let canonical = canonicalize_proposal(diamond_proposal(), &ProposalLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let limits = ParallelScheduleLimits { + max_fan_out: 1, + max_join_width: 1, + max_queue_depth: 1, + max_concurrent_nodes: 1, + }; + let Err(failure) = compute_parallel_schedule(&canonical, &limits) else { + return Err("every configured bound is violated and must be rejected".to_string()); + }; + let codes: std::collections::BTreeSet<_> = failure.errors.iter().map(|e| e.code).collect(); + assert!(codes.contains(&ParallelScheduleErrorCode::FanOutExceeded)); + assert!(codes.contains(&ParallelScheduleErrorCode::JoinWidthExceeded)); + assert!(codes.contains(&ParallelScheduleErrorCode::QueueDepthExceeded)); + Ok(()) +} diff --git a/crates/traverse-mcp/src/tools/mod.rs b/crates/traverse-mcp/src/tools/mod.rs index 5d7cb2ca..4dd0a2df 100644 --- a/crates/traverse-mcp/src/tools/mod.rs +++ b/crates/traverse-mcp/src/tools/mod.rs @@ -4,5 +4,6 @@ pub mod capabilities; pub mod events; +pub mod parallel_proposals; pub mod proposals; pub mod traces; diff --git a/crates/traverse-mcp/src/tools/parallel_proposals.rs b/crates/traverse-mcp/src/tools/parallel_proposals.rs new file mode 100644 index 00000000..aeff41f0 --- /dev/null +++ b/crates/traverse-mcp/src/tools/parallel_proposals.rs @@ -0,0 +1,284 @@ +//! MCP tool surfaces for bounded parallel proposal scheduling and execution +//! (spec `110-bounded-parallel-workflow-scheduling`, P2, ADR-0042). +//! +//! Extends `tools::proposals`' P1 lifecycle with a schedule-aware variant: +//! the same `WorkflowProposal` wire format, cross-validated the same way, +//! but levelized into concurrency waves and authorized against the +//! `pure_read`-only constraint (FR-004a) before execution. Mirrors the same +//! plain, fully-tested-function pattern as `tools::proposals` and +//! `tools::capabilities` rather than wiring into `stdio_server.rs`. + +use ed25519_dalek::VerifyingKey; +use serde::Serialize; +use std::collections::HashMap; + +use traverse_contracts::{ + ParallelScheduleLimits, ProposalLimits, SnapshotDigests, canonicalize_proposal, + compute_parallel_schedule, proposal_snapshot_digest, +}; +use traverse_registry::{ApplicationBundleManifest, CapabilityRegistry}; +use traverse_runtime::parallel_proposal::{ + ParallelExecutionLimits, enforce_pure_read_only_parallelism, execute_parallel_proposal, +}; +use traverse_runtime::proposal::{ + ApprovalTokenStore, QuotaLimits, QuotaTracker, validate_proposal_against_host_state, +}; +use traverse_runtime::{LocalExecutor, Runtime}; + +use crate::McpError; +use crate::tools::proposals::{ + AuthorizeAndReserveQuotaRequest, ProposalDenial, ProposalExecutionResponse, cross_denial, + parse_and_digest, structural_denial, +}; + +#[derive(Debug, Clone, Serialize)] +pub struct ParallelScheduleResponse { + pub proposal_id: String, + pub proposal_digest: String, + pub valid: bool, + pub errors: Vec, + /// Concurrency waves in dependency order; empty unless `valid` is true. + pub waves: Vec>, + pub automatic_eligible: bool, +} + +/// Validates a proposal (spec 109, unchanged) and, if valid, levelizes it +/// into concurrency waves and checks the FR-004a `pure_read`-only +/// constraint. A structural, cross-validation, schedule-bound, or +/// concurrent-side-effect denial is a normal, structured response, never an +/// [`McpError`] — matching FR-010's stable denial-code contract. +/// +/// # Errors +/// +/// Returns [`McpError`] only when `proposal_json` is not valid JSON. +pub fn compute_schedule_for_proposal( + proposal_json: &str, + manifest: &ApplicationBundleManifest, + registry: &CapabilityRegistry, + proposal_limits: &ProposalLimits, + schedule_limits: &ParallelScheduleLimits, +) -> Result { + let (proposal, proposal_id, digest) = parse_and_digest(proposal_json)?; + + let canonical = match canonicalize_proposal(proposal, proposal_limits) { + Ok(canonical) => canonical, + Err(failure) => { + return Ok(ParallelScheduleResponse { + proposal_id, + proposal_digest: digest, + valid: false, + errors: failure.errors.into_iter().map(structural_denial).collect(), + waves: Vec::new(), + automatic_eligible: false, + }); + } + }; + + let resolved = match validate_proposal_against_host_state(&canonical, manifest, registry) { + Ok(resolved) => resolved, + Err(failure) => { + return Ok(ParallelScheduleResponse { + proposal_id, + proposal_digest: digest, + valid: false, + errors: failure.errors.into_iter().map(cross_denial).collect(), + waves: Vec::new(), + automatic_eligible: false, + }); + } + }; + + let schedule = match compute_parallel_schedule(&canonical, schedule_limits) { + Ok(schedule) => schedule, + Err(failure) => { + return Ok(ParallelScheduleResponse { + proposal_id, + proposal_digest: digest, + valid: false, + errors: failure.errors.into_iter().map(schedule_denial).collect(), + waves: Vec::new(), + automatic_eligible: false, + }); + } + }; + + if let Err(failure) = enforce_pure_read_only_parallelism(&schedule, &resolved) { + return Ok(ParallelScheduleResponse { + proposal_id, + proposal_digest: digest, + valid: false, + errors: failure + .errors + .into_iter() + .map(authorization_denial) + .collect(), + waves: Vec::new(), + automatic_eligible: false, + }); + } + + Ok(ParallelScheduleResponse { + proposal_id, + proposal_digest: digest, + valid: true, + errors: Vec::new(), + waves: schedule.waves, + automatic_eligible: traverse_runtime::proposal::proposal_is_automatic_eligible(&resolved), + }) +} + +/// Everything [`execute_parallel_proposal_via_mcp`] needs beyond the +/// caller's runtime and shared authorization/quota state. +pub struct ParallelProposalExecutionRequest<'a> { + pub proposal_json: &'a str, + pub manifest: &'a ApplicationBundleManifest, + pub registry: &'a CapabilityRegistry, + pub proposal_limits: &'a ProposalLimits, + pub schedule_limits: &'a ParallelScheduleLimits, + pub execution_limits: &'a ParallelExecutionLimits, + pub snapshots: &'a SnapshotDigests, + pub approval_token: Option<&'a str>, + pub expected_token_issuer: &'a str, + pub expected_token_audience: &'a str, + pub token_verifying_keys_by_key_id: &'a HashMap, + pub principal: &'a str, + pub app_id: &'a str, +} + +/// Authorizes and executes a proposal's bounded parallel schedule end to end +/// (spec 110 FR-001 through FR-005, reusing spec 109 FR-006 through FR-009 +/// for authorization, quotas, and tracing). A denial for any reason — +/// invalid proposal, schedule bound exceeded, a non-`pure_read` node in a +/// concurrent wave, missing/invalid approval token, exhausted quota — is a +/// normal structured response, never an [`McpError`]. +/// +/// # Errors +/// +/// Returns [`McpError`] only when `proposal_json` is not valid JSON. +pub fn execute_parallel_proposal_via_mcp( + runtime: &Runtime, + request: &ParallelProposalExecutionRequest<'_>, + token_store: &ApprovalTokenStore, + quota_tracker: &QuotaTracker, + quota_limits: &QuotaLimits, +) -> Result +where + Runtime: Sync, +{ + let (proposal, _proposal_id, digest) = parse_and_digest(request.proposal_json)?; + let workspace_id = proposal.workspace_id.clone(); + let snapshot_digest = proposal_snapshot_digest(&digest, request.snapshots); + + let canonical = match canonicalize_proposal(proposal, request.proposal_limits) { + Ok(canonical) => canonical, + Err(failure) => { + return Ok(ProposalExecutionResponse::Denied { + code: "invalid_proposal".to_string(), + message: format!("{} structural validation error(s)", failure.errors.len()), + }); + } + }; + + let resolved = match validate_proposal_against_host_state( + &canonical, + request.manifest, + request.registry, + ) { + Ok(resolved) => resolved, + Err(failure) => { + return Ok(ProposalExecutionResponse::Denied { + code: "invalid_proposal".to_string(), + message: format!("{} cross-validation error(s)", failure.errors.len()), + }); + } + }; + + let schedule = match compute_parallel_schedule(&canonical, request.schedule_limits) { + Ok(schedule) => schedule, + Err(failure) => { + return Ok(ProposalExecutionResponse::Denied { + code: "invalid_parallel_schedule".to_string(), + message: format!("{} schedule bound violation(s)", failure.errors.len()), + }); + } + }; + + if let Err(failure) = enforce_pure_read_only_parallelism(&schedule, &resolved) { + return Ok(ProposalExecutionResponse::Denied { + code: "concurrent_side_effect_denied".to_string(), + message: format!( + "{} node(s) denied concurrent execution", + failure.errors.len() + ), + }); + } + + let authorized = + crate::tools::proposals::authorize_and_reserve_quota(&AuthorizeAndReserveQuotaRequest { + resolved: &resolved, + digest: &digest, + snapshot_digest: &snapshot_digest, + workspace_id: &workspace_id, + approval_token: request.approval_token, + expected_token_issuer: request.expected_token_issuer, + expected_token_audience: request.expected_token_audience, + token_verifying_keys_by_key_id: request.token_verifying_keys_by_key_id, + token_store, + quota_tracker, + quota_limits, + principal: request.principal, + app_id: request.app_id, + }); + let authorized = match authorized { + Ok(authorized) => authorized, + Err(denial) => return Ok(*denial), + }; + + let trace = execute_parallel_proposal( + runtime, + &canonical, + &schedule, + authorized.summary, + &digest, + &snapshot_digest, + request.execution_limits, + ); + drop(authorized.reservation); + Ok(ProposalExecutionResponse::Trace(trace)) +} + +fn schedule_denial(error: traverse_contracts::ParallelScheduleError) -> ProposalDenial { + ProposalDenial { + code: debug_enum_to_snake_case(&format!("{:?}", error.code)), + path: error.path, + message: error.message, + } +} + +fn authorization_denial( + error: traverse_runtime::parallel_proposal::ParallelAuthorizationError, +) -> ProposalDenial { + ProposalDenial { + code: debug_enum_to_snake_case(&format!("{:?}", error.code)), + path: error.path, + message: error.message, + } +} + +/// Converts a Rust `Debug`-formatted `PascalCase` enum variant into the +/// stable `snake_case` string used for every machine-readable code this +/// module emits (spec 109 FR-010, carried over for spec 110). +fn debug_enum_to_snake_case(value: &str) -> String { + let mut output = String::with_capacity(value.len() + 4); + for (index, ch) in value.chars().enumerate() { + if ch.is_ascii_uppercase() { + if index > 0 { + output.push('_'); + } + output.push(ch.to_ascii_lowercase()); + } else { + output.push(ch); + } + } + output +} diff --git a/crates/traverse-mcp/src/tools/proposals.rs b/crates/traverse-mcp/src/tools/proposals.rs index 1ad8103e..fd5fdd7d 100644 --- a/crates/traverse-mcp/src/tools/proposals.rs +++ b/crates/traverse-mcp/src/tools/proposals.rs @@ -18,8 +18,8 @@ use traverse_registry::{ApplicationBundleManifest, CapabilityRegistry}; use traverse_runtime::proposal::{ ApprovalTokenStore, ApprovalTokenVerificationContext, AuthorizationDecision, AuthorizationSummary, ProposalCrossValidationError as CrossError, ProposalTrace, QuotaLimits, - QuotaTracker, execute_proposal, proposal_is_automatic_eligible, - validate_proposal_against_host_state, verify_approval_token, + QuotaReservation, QuotaTracker, ResolvedProposalNode, execute_proposal, + proposal_is_automatic_eligible, validate_proposal_against_host_state, verify_approval_token, }; use traverse_runtime::{LocalExecutor, Runtime}; @@ -261,57 +261,119 @@ pub fn execute_proposal_via_mcp( } }; - let authorization = if proposal_is_automatic_eligible(&resolved) { + let authorized = match authorize_and_reserve_quota(&AuthorizeAndReserveQuotaRequest { + resolved: &resolved, + digest: &digest, + snapshot_digest: &snapshot_digest, + workspace_id: &workspace_id, + approval_token: request.approval_token, + expected_token_issuer: request.expected_token_issuer, + expected_token_audience: request.expected_token_audience, + token_verifying_keys_by_key_id: request.token_verifying_keys_by_key_id, + token_store, + quota_tracker, + quota_limits, + principal: request.principal, + app_id: request.app_id, + }) { + Ok(authorized) => authorized, + Err(denial) => return Ok(*denial), + }; + + let trace = execute_proposal( + runtime, + &canonical, + &resolved, + authorized.summary, + &digest, + &snapshot_digest, + ); + drop(authorized.reservation); + Ok(ProposalExecutionResponse::Trace(trace)) +} + +/// Everything [`authorize_and_reserve_quota`] needs to decide automatic vs. +/// approval-token authorization and reserve a concurrency quota slot. +/// Shared by the P1 sequential and P2 parallel MCP execute paths. +pub(crate) struct AuthorizeAndReserveQuotaRequest<'a> { + pub resolved: &'a [ResolvedProposalNode], + pub digest: &'a str, + pub snapshot_digest: &'a str, + pub workspace_id: &'a str, + pub approval_token: Option<&'a str>, + pub expected_token_issuer: &'a str, + pub expected_token_audience: &'a str, + pub token_verifying_keys_by_key_id: &'a HashMap, + pub token_store: &'a ApprovalTokenStore, + pub quota_tracker: &'a QuotaTracker, + pub quota_limits: &'a QuotaLimits, + pub principal: &'a str, + pub app_id: &'a str, +} + +pub(crate) struct AuthorizedQuota<'a> { + pub summary: AuthorizationSummary, + pub reservation: QuotaReservation<'a>, +} + +/// Decides automatic-vs-approval-token authorization (spec 109 FR-006, +/// FR-006a) and reserves a per-principal/app/workspace concurrency quota +/// slot (FR-007b). On any denial, returns the exact +/// [`ProposalExecutionResponse::Denied`] the caller should return unchanged. +pub(crate) fn authorize_and_reserve_quota<'a>( + request: &AuthorizeAndReserveQuotaRequest<'a>, +) -> Result, Box> { + let authorization = if proposal_is_automatic_eligible(request.resolved) { AuthorizationDecision::Automatic } else { let Some(token) = request.approval_token else { - return Ok(ProposalExecutionResponse::Denied { + return Err(Box::new(ProposalExecutionResponse::Denied { code: "approval_token_required".to_string(), message: "this proposal requires a verified approval token".to_string(), - }); + })); }; let verification_context = ApprovalTokenVerificationContext { expected_issuer: request.expected_token_issuer, expected_audience: request.expected_token_audience, - expected_workspace_id: &workspace_id, - expected_proposal_digest: &digest, - expected_snapshot_digest: &snapshot_digest, + expected_workspace_id: request.workspace_id, + expected_proposal_digest: request.digest, + expected_snapshot_digest: request.snapshot_digest, verifying_keys_by_key_id: request.token_verifying_keys_by_key_id, }; let claims = match verify_approval_token(token, &verification_context) { Ok(claims) => claims, Err(error) => { - return Ok(ProposalExecutionResponse::Denied { + return Err(Box::new(ProposalExecutionResponse::Denied { code: token_error_code(&error.code), message: error.message, - }); + })); } }; - if let Err(error) = token_store.check_and_record_use(&claims) { - return Ok(ProposalExecutionResponse::Denied { + if let Err(error) = request.token_store.check_and_record_use(&claims) { + return Err(Box::new(ProposalExecutionResponse::Denied { code: token_error_code(&error.code), message: error.message, - }); + })); } AuthorizationDecision::Approved(Box::new(claims)) }; - let reservation = match quota_tracker.reserve( + let reservation = match request.quota_tracker.reserve( request.principal, request.app_id, - &workspace_id, - quota_limits, + request.workspace_id, + request.quota_limits, ) { Ok(reservation) => reservation, Err(denial) => { - return Ok(ProposalExecutionResponse::Denied { + return Err(Box::new(ProposalExecutionResponse::Denied { code: format!("quota_exhausted_{}", denial.scope), message: denial.message, - }); + })); } }; - let authorization_summary = match &authorization { + let summary = match &authorization { AuthorizationDecision::Automatic => AuthorizationSummary { automatic: true, approval_token_id: None, @@ -322,16 +384,10 @@ pub fn execute_proposal_via_mcp( }, }; - let trace = execute_proposal( - runtime, - &canonical, - &resolved, - authorization_summary, - &digest, - &snapshot_digest, - ); - drop(reservation); - Ok(ProposalExecutionResponse::Trace(trace)) + Ok(AuthorizedQuota { + summary, + reservation, + }) } /// Renders a completed execution's redacted trace for MCP observation (spec @@ -379,7 +435,9 @@ pub fn export_proposal( }) } -fn parse_and_digest(proposal_json: &str) -> Result<(WorkflowProposal, String, String), McpError> { +pub(crate) fn parse_and_digest( + proposal_json: &str, +) -> Result<(WorkflowProposal, String, String), McpError> { let proposal: WorkflowProposal = serde_json::from_str(proposal_json).map_err(|e| McpError { code: McpErrorCode::InvalidRequest, message: format!("proposal is not valid JSON: {e}"), @@ -389,7 +447,7 @@ fn parse_and_digest(proposal_json: &str) -> Result<(WorkflowProposal, String, St Ok((proposal, proposal_id, digest)) } -fn structural_denial(error: StructuralError) -> ProposalDenial { +pub(crate) fn structural_denial(error: StructuralError) -> ProposalDenial { ProposalDenial { code: debug_enum_to_snake_case(&format!("{:?}", error.code)), path: error.path, @@ -397,7 +455,7 @@ fn structural_denial(error: StructuralError) -> ProposalDenial { } } -fn cross_denial(error: CrossError) -> ProposalDenial { +pub(crate) fn cross_denial(error: CrossError) -> ProposalDenial { ProposalDenial { code: debug_enum_to_snake_case(&format!("{:?}", error.code)), path: error.path, @@ -405,7 +463,9 @@ fn cross_denial(error: CrossError) -> ProposalDenial { } } -fn token_error_code(code: &traverse_runtime::proposal::ApprovalTokenErrorCode) -> String { +pub(crate) fn token_error_code( + code: &traverse_runtime::proposal::ApprovalTokenErrorCode, +) -> String { debug_enum_to_snake_case(&format!("{code:?}")) } diff --git a/crates/traverse-mcp/tests/parallel_proposal_tests.rs b/crates/traverse-mcp/tests/parallel_proposal_tests.rs new file mode 100644 index 00000000..ba400482 --- /dev/null +++ b/crates/traverse-mcp/tests/parallel_proposal_tests.rs @@ -0,0 +1,1001 @@ +//! End-to-end MCP tests for bounded parallel proposal scheduling and +//! execution (spec `110-bounded-parallel-workflow-scheduling`, P2, +//! ADR-0042). +//! +//! Exercises `traverse_mcp::tools::parallel_proposals` — schedule +//! computation, FR-004a `pure_read`-only authorization, and full +//! execute-via-mcp — reusing the same wire format and manifest/registry +//! fixtures as spec 109's P1 surface. Authorization-token and quota +//! mechanics themselves are already covered end to end by +//! `proposal_tests.rs`; these tests focus on what P2 adds on top. + +use ed25519_dalek::{Signer, SigningKey}; +use serde_json::{Value, json}; +use std::collections::HashMap; + +use traverse_contracts::{ + BinaryFormat as ContractBinaryFormat, CapabilityContract, DataFlowPolicy, DeterminismClass, + EffectClass, Entrypoint, EntrypointKind, Execution, ExecutionConstraints, ExecutionTarget, + FilesystemAccess, HostApiAccess, Lifecycle, NetworkAccess, Owner, ParallelScheduleLimits, + ProposalLimits, ReliabilityMetadata, RiskMetadata, SchemaContainer, ServiceType, SideEffect, + SideEffectKind, +}; +use traverse_mcp::tools::parallel_proposals::{ + ParallelProposalExecutionRequest, compute_schedule_for_proposal, + execute_parallel_proposal_via_mcp, +}; +use traverse_mcp::tools::proposals::ProposalExecutionResponse; +use traverse_registry::{ + ApplicationBundleManifest, ApplicationComponent, ApplicationComponentRef, + ApplicationEffectiveConfig, ArtifactDigests, BinaryFormat as RegistryBinaryFormat, + BinaryReference, CapabilityArtifactRecord, CapabilityRegistration, CapabilityRegistry, + ComponentExecutionMode, ComposabilityMetadata, CompositionKind, CompositionPattern, + ImplementationKind, RegistryProvenance, RegistryScope, SourceKind, SourceReference, + WasmComponentManifest, +}; +use traverse_runtime::parallel_proposal::ParallelExecutionLimits; +use traverse_runtime::proposal::{ + ApprovalTokenStore, ProposalTerminalState, QuotaLimits, QuotaTracker, +}; +use traverse_runtime::security::RuntimeSecurityConfig; +use traverse_runtime::{ + LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutionOutput, LocalExecutor, Runtime, +}; + +fn risk(effect_class: EffectClass) -> RiskMetadata { + RiskMetadata { + effect_class, + determinism_class: DeterminismClass::Deterministic, + data_flow: DataFlowPolicy::default(), + reliability: ReliabilityMetadata { + idempotency_required: false, + retryable: true, + compensation_available: false, + }, + } +} + +fn contract(id: &str, risk: RiskMetadata) -> CapabilityContract { + let (namespace, name) = id.rsplit_once('.').unwrap_or(("test", id)); + CapabilityContract { + kind: "capability_contract".to_string(), + schema_version: "1.0.0".to_string(), + id: id.to_string(), + namespace: namespace.to_string(), + name: name.to_string(), + version: "1.0.0".to_string(), + lifecycle: Lifecycle::Active, + owner: Owner { + team: "traverse-core".to_string(), + contact: "enrico.piovesan10@gmail.com".to_string(), + }, + summary: "Test capability for parallel proposal MCP coverage.".to_string(), + description: "Portable test capability used to exercise the parallel proposal MCP surface." + .to_string(), + inputs: SchemaContainer { + schema: json!({"type": "object"}), + }, + outputs: SchemaContainer { + schema: json!({"type": "object"}), + }, + preconditions: Vec::new(), + postconditions: Vec::new(), + side_effects: vec![SideEffect { + kind: SideEffectKind::MemoryOnly, + description: "No durable side effect.".to_string(), + }], + emits: Vec::new(), + consumes: Vec::new(), + permissions: Vec::new(), + execution: Execution { + binary_format: ContractBinaryFormat::Wasm, + entrypoint: Entrypoint { + kind: EntrypointKind::WasiCommand, + command: "run".to_string(), + }, + preferred_targets: vec![ExecutionTarget::Local], + constraints: ExecutionConstraints { + host_api_access: HostApiAccess::None, + network_access: NetworkAccess::Forbidden, + filesystem_access: FilesystemAccess::None, + }, + }, + policies: Vec::new(), + dependencies: Vec::new(), + provenance: traverse_contracts::Provenance { + source: traverse_contracts::ProvenanceSource::Greenfield, + author: "test".to_string(), + created_at: "2026-08-23T00:00:00Z".to_string(), + spec_ref: None, + adr_refs: Vec::new(), + exception_refs: Vec::new(), + }, + evidence: Vec::new(), + service_type: ServiceType::Stateless, + permitted_targets: vec![ExecutionTarget::Local], + event_trigger: None, + connector_requirements: Vec::new(), + state_schema: None, + use_cases: Vec::new(), + risk, + } +} + +fn artifact(digest: &str) -> CapabilityArtifactRecord { + CapabilityArtifactRecord { + artifact_ref: format!("artifact:{digest}"), + implementation_kind: ImplementationKind::Executable, + source: SourceReference { + kind: SourceKind::Git, + location: "https://example.invalid/repo".to_string(), + }, + binary: Some(BinaryReference { + format: RegistryBinaryFormat::Wasm, + location: format!("artifacts/{digest}/capability.wasm"), + signature: None, + }), + workflow_ref: None, + digests: ArtifactDigests { + source_digest: format!("src-{digest}"), + binary_digest: Some(digest.to_string()), + }, + provenance: RegistryProvenance { + source: "test".to_string(), + author: "test".to_string(), + created_at: "2026-08-23T00:00:00Z".to_string(), + }, + } +} + +fn registry_with( + entries: Vec<(CapabilityContract, CapabilityArtifactRecord)>, +) -> CapabilityRegistry { + let mut registry = CapabilityRegistry::new(); + for (contract, artifact) in entries { + let outcome = registry.register(CapabilityRegistration { + scope: RegistryScope::Public, + contract, + contract_path: "registry/test/contract.json".to_string(), + artifact, + registered_at: "2026-08-23T00:00:00Z".to_string(), + tags: Vec::new(), + composability: ComposabilityMetadata { + kind: CompositionKind::Atomic, + patterns: vec![CompositionPattern::Sequential], + provides: Vec::new(), + requires: Vec::new(), + }, + governing_spec: "005-capability-registry".to_string(), + validator_version: "0.1.0".to_string(), + }); + assert!(outcome.is_ok(), "registration must succeed: {outcome:?}"); + } + registry +} + +fn manifest_declaring(components: &[(&str, EffectClass)]) -> ApplicationBundleManifest { + ApplicationBundleManifest { + app_id: "test-app".to_string(), + version: "1.0.0".to_string(), + schema_version: "1.0.0".to_string(), + workspace_defaults: json!({}), + components: components + .iter() + .map(|(capability_id, effect_class)| ApplicationComponent { + reference: ApplicationComponentRef { + component_id: (*capability_id).to_string(), + version: "1.0.0".to_string(), + digest: "sha256:component-digest".to_string(), + manifest_path: "component.manifest.json".to_string(), + }, + manifest_path: "component.manifest.json".into(), + manifest: WasmComponentManifest { + component_id: (*capability_id).to_string(), + version: "1.0.0".to_string(), + schema_version: "1.0.0".to_string(), + execution_mode: ComponentExecutionMode::Wasm, + capability_id: (*capability_id).to_string(), + capability_version: "1.0.0".to_string(), + contract_path: None, + registry_ref: None, + wasm_binary_path: None, + wasm_digest: None, + platforms: vec!["local".to_string()], + wrapper_path: None, + runtime_constraints: json!({}), + permitted_targets: vec![ExecutionTarget::Local], + dependencies: Vec::new(), + connector_requirements: Vec::new(), + validation_evidence: Vec::new(), + executable_pin: None, + }, + contract_path: "contract.json".into(), + contract: contract(capability_id, risk(*effect_class)), + wasm_binary_path: None, + verified_wasm_digest: None, + }) + .collect(), + workflows: Vec::new(), + connector_bindings: Vec::new(), + model_dependencies: Vec::new(), + config_schema: json!({}), + default_config: json!({}), + effective_config: ApplicationEffectiveConfig { + values: json!({}), + redacted_secret_keys: Vec::new(), + }, + placement_policy: json!({}), + public_surfaces: Vec::new(), + state_machine: None, + } +} + +fn diamond_registry(non_read_node: Option<&str>) -> CapabilityRegistry { + registry_with( + ["a", "b", "c", "d"] + .iter() + .map(|id| { + let effect_class = if Some(*id) == non_read_node { + EffectClass::ExternalEffect + } else { + EffectClass::PureRead + }; + ( + contract(&format!("test.{id}"), risk(effect_class)), + artifact(&format!("digest-{id}")), + ) + }) + .collect(), + ) +} + +fn diamond_manifest(non_read_node: Option<&str>) -> ApplicationBundleManifest { + let effect_for = |id: &str| -> EffectClass { + if Some(id) == non_read_node { + EffectClass::ExternalEffect + } else { + EffectClass::PureRead + } + }; + manifest_declaring(&[ + ("test.a", effect_for("a")), + ("test.b", effect_for("b")), + ("test.c", effect_for("c")), + ("test.d", effect_for("d")), + ]) +} + +/// a fans out to b and c, which both feed the join node d. +fn diamond_proposal_json(proposal_id: &str) -> String { + json!({ + "kind": "workflow_proposal", + "schema_version": "1.0.0", + "proposal_id": proposal_id, + "workspace_id": "workspace-001", + "app_manifest": { + "app_id": "test-app", + "app_version": "1.0.0", + "manifest_digest": "sha256:manifest-digest" + }, + "nodes": [ + {"node_id": "a", "capability_id": "test.a", "capability_version": "1.0.0", "artifact_digest": "digest-a"}, + {"node_id": "b", "capability_id": "test.b", "capability_version": "1.0.0", "artifact_digest": "digest-b"}, + {"node_id": "c", "capability_id": "test.c", "capability_version": "1.0.0", "artifact_digest": "digest-c"}, + {"node_id": "d", "capability_id": "test.d", "capability_version": "1.0.0", "artifact_digest": "digest-d"} + ], + "edges": [ + {"from_node_id": "a", "to_node_id": "b"}, + {"from_node_id": "a", "to_node_id": "c"}, + {"from_node_id": "b", "to_node_id": "d"}, + {"from_node_id": "c", "to_node_id": "d"} + ], + "mappings": [], + "initial_input": {} + }) + .to_string() +} + +fn structurally_invalid_proposal_json(proposal_id: &str) -> String { + json!({ + "kind": "not_a_workflow_proposal", + "schema_version": "1.0.0", + "proposal_id": proposal_id, + "workspace_id": "workspace-001", + "app_manifest": {"app_id": "test-app", "app_version": "1.0.0", "manifest_digest": "sha256:manifest-digest"}, + "nodes": [], + "edges": [], + "mappings": [], + "initial_input": {} + }) + .to_string() +} + +struct EchoExecutor; + +impl LocalExecutor for EchoExecutor { + fn execute( + &self, + _capability: &traverse_registry::ResolvedCapability, + _input: &Value, + ) -> Result { + Ok(LocalExecutionOutput { + value: json!({"status": "ok"}), + emitted_events: Vec::new(), + }) + } +} + +struct FailsOneCapabilityExecutor { + failing_capability_id: String, +} + +impl LocalExecutor for FailsOneCapabilityExecutor { + fn execute( + &self, + capability: &traverse_registry::ResolvedCapability, + _input: &Value, + ) -> Result { + if capability.contract.id == self.failing_capability_id { + Err(LocalExecutionFailure { + code: LocalExecutionFailureCode::ExecutionFailed, + message: "scripted failure".to_string(), + }) + } else { + Ok(LocalExecutionOutput { + value: json!({"status": "ok"}), + emitted_events: Vec::new(), + }) + } + } +} + +fn keys() -> HashMap { + HashMap::new() +} + +// -- compute_schedule_for_proposal ------------------------------------------- + +#[test] +fn compute_schedule_accepts_a_well_formed_diamond_proposal() -> Result<(), String> { + let manifest = diamond_manifest(None); + let registry = diamond_registry(None); + + let response = compute_schedule_for_proposal( + &diamond_proposal_json("proposal-schedule-accept"), + &manifest, + ®istry, + &ProposalLimits::default(), + &ParallelScheduleLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + assert!(response.valid, "errors: {:?}", response.errors); + assert!(response.automatic_eligible); + assert_eq!( + response.waves, + vec![ + vec!["a".to_string()], + vec!["b".to_string(), "c".to_string()], + vec!["d".to_string()], + ] + ); + Ok(()) +} + +#[test] +fn compute_schedule_rejects_a_structurally_invalid_proposal() -> Result<(), String> { + let manifest = diamond_manifest(None); + let registry = diamond_registry(None); + + let response = compute_schedule_for_proposal( + &structurally_invalid_proposal_json("proposal-schedule-bad-kind"), + &manifest, + ®istry, + &ProposalLimits::default(), + &ParallelScheduleLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + assert!(!response.valid); + assert!(response.waves.is_empty()); + Ok(()) +} + +#[test] +fn compute_schedule_rejects_a_cross_validation_invalid_proposal() -> Result<(), String> { + let mut manifest = diamond_manifest(None); + manifest + .components + .retain(|c| c.reference.component_id != "test.d"); // missing test.d + let registry = diamond_registry(None); + + let response = compute_schedule_for_proposal( + &diamond_proposal_json("proposal-schedule-undeclared"), + &manifest, + ®istry, + &ProposalLimits::default(), + &ParallelScheduleLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + assert!(!response.valid); + assert!( + response + .errors + .iter() + .any(|e| e.code == "undeclared_capability") + ); + Ok(()) +} + +#[test] +fn compute_schedule_rejects_a_fan_out_over_the_configured_limit() -> Result<(), String> { + let manifest = diamond_manifest(None); + let registry = diamond_registry(None); + let limits = ParallelScheduleLimits { + max_fan_out: 1, + ..ParallelScheduleLimits::default() + }; + + let response = compute_schedule_for_proposal( + &diamond_proposal_json("proposal-schedule-fan-out"), + &manifest, + ®istry, + &ProposalLimits::default(), + &limits, + ) + .map_err(|e| format!("{e:?}"))?; + + assert!(!response.valid); + assert!(response.errors.iter().any(|e| e.code == "fan_out_exceeded")); + Ok(()) +} + +#[test] +fn compute_schedule_denies_a_concurrent_wave_with_a_non_pure_read_node() -> Result<(), String> { + let manifest = diamond_manifest(Some("c")); + let registry = diamond_registry(Some("c")); + + let response = compute_schedule_for_proposal( + &diamond_proposal_json("proposal-schedule-side-effect"), + &manifest, + ®istry, + &ProposalLimits::default(), + &ParallelScheduleLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + assert!(!response.valid); + assert!( + response + .errors + .iter() + .any(|e| e.code == "concurrent_side_effect_denied") + ); + Ok(()) +} + +#[test] +fn compute_schedule_reports_invalid_json_as_an_mcp_error() { + let manifest = diamond_manifest(None); + let registry = diamond_registry(None); + + let result = compute_schedule_for_proposal( + "not json", + &manifest, + ®istry, + &ProposalLimits::default(), + &ParallelScheduleLimits::default(), + ); + assert!(result.is_err()); +} + +// -- execute_parallel_proposal_via_mcp --------------------------------------- + +/// A struct literal so temporary fields (`ProposalLimits::default()`, the +/// inline `SnapshotDigests`) get Rust's `let`-initializer lifetime +/// extension, matching `request`'s scope — a plain function call would drop +/// them at the end of the call expression instead. +macro_rules! execution_request { + ($proposal_json:expr, $manifest:expr, $registry:expr, $schedule_limits:expr, $execution_limits:expr, $approval_token:expr, $keys:expr $(,)?) => { + ParallelProposalExecutionRequest { + proposal_json: $proposal_json, + manifest: $manifest, + registry: $registry, + proposal_limits: &ProposalLimits::default(), + schedule_limits: $schedule_limits, + execution_limits: $execution_limits, + snapshots: &traverse_contracts::SnapshotDigests { + manifest_digest: "manifest-1".to_string(), + registry_digest: "registry-1".to_string(), + binding_digest: "binding-1".to_string(), + policy_digest: "policy-1".to_string(), + budget_digest: "budget-1".to_string(), + }, + approval_token: $approval_token, + expected_token_issuer: "traverse-approval-service", + expected_token_audience: "traverse-runtime", + token_verifying_keys_by_key_id: $keys, + principal: "principal-001", + app_id: "test-app", + } + }; +} + +#[test] +fn execute_parallel_proposal_via_mcp_succeeds_for_an_automatic_eligible_diamond() +-> Result<(), String> { + let manifest = diamond_manifest(None); + let registry = diamond_registry(None); + let runtime = Runtime::new(registry.clone(), EchoExecutor) + .with_security_config(RuntimeSecurityConfig::development()); + let token_store = ApprovalTokenStore::new(); + let quota_tracker = QuotaTracker::new(); + let keys = keys(); + + let proposal_json = diamond_proposal_json("proposal-exec-accept"); + let schedule_limits = ParallelScheduleLimits::default(); + let execution_limits = ParallelExecutionLimits::default(); + let request = execution_request!( + &proposal_json, + &manifest, + ®istry, + &schedule_limits, + &execution_limits, + None, + &keys, + ); + + let response = execute_parallel_proposal_via_mcp( + &runtime, + &request, + &token_store, + "a_tracker, + &QuotaLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + let ProposalExecutionResponse::Trace(trace) = response else { + return Err(format!("expected Trace, got {response:?}")); + }; + assert_eq!(trace.terminal_state, ProposalTerminalState::Succeeded); + assert_eq!(trace.node_outcomes.len(), 4); + Ok(()) +} + +#[test] +fn execute_parallel_proposal_via_mcp_denies_a_structurally_invalid_proposal() -> Result<(), String> +{ + let manifest = diamond_manifest(None); + let registry = diamond_registry(None); + let runtime = Runtime::new(registry.clone(), EchoExecutor) + .with_security_config(RuntimeSecurityConfig::development()); + let token_store = ApprovalTokenStore::new(); + let quota_tracker = QuotaTracker::new(); + let keys = keys(); + + let proposal_json = structurally_invalid_proposal_json("proposal-exec-bad-kind"); + let schedule_limits = ParallelScheduleLimits::default(); + let execution_limits = ParallelExecutionLimits::default(); + let request = execution_request!( + &proposal_json, + &manifest, + ®istry, + &schedule_limits, + &execution_limits, + None, + &keys, + ); + + let response = execute_parallel_proposal_via_mcp( + &runtime, + &request, + &token_store, + "a_tracker, + &QuotaLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + let ProposalExecutionResponse::Denied { code, .. } = response else { + return Err(format!("expected Denied, got {response:?}")); + }; + assert_eq!(code, "invalid_proposal"); + Ok(()) +} + +#[test] +fn execute_parallel_proposal_via_mcp_denies_a_cross_validation_invalid_proposal() +-> Result<(), String> { + let mut manifest = diamond_manifest(None); + manifest + .components + .retain(|c| c.reference.component_id != "test.d"); // missing test.d + let registry = diamond_registry(None); + let runtime = Runtime::new(registry.clone(), EchoExecutor) + .with_security_config(RuntimeSecurityConfig::development()); + let token_store = ApprovalTokenStore::new(); + let quota_tracker = QuotaTracker::new(); + let keys = keys(); + + let proposal_json = diamond_proposal_json("proposal-exec-undeclared"); + let schedule_limits = ParallelScheduleLimits::default(); + let execution_limits = ParallelExecutionLimits::default(); + let request = execution_request!( + &proposal_json, + &manifest, + ®istry, + &schedule_limits, + &execution_limits, + None, + &keys, + ); + + let response = execute_parallel_proposal_via_mcp( + &runtime, + &request, + &token_store, + "a_tracker, + &QuotaLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + let ProposalExecutionResponse::Denied { code, .. } = response else { + return Err(format!("expected Denied, got {response:?}")); + }; + assert_eq!(code, "invalid_proposal"); + Ok(()) +} + +#[test] +fn execute_parallel_proposal_via_mcp_denies_a_schedule_bound_violation() -> Result<(), String> { + let manifest = diamond_manifest(None); + let registry = diamond_registry(None); + let runtime = Runtime::new(registry.clone(), EchoExecutor) + .with_security_config(RuntimeSecurityConfig::development()); + let token_store = ApprovalTokenStore::new(); + let quota_tracker = QuotaTracker::new(); + let keys = keys(); + let schedule_limits = ParallelScheduleLimits { + max_fan_out: 1, + ..ParallelScheduleLimits::default() + }; + + let proposal_json = diamond_proposal_json("proposal-exec-fan-out"); + let execution_limits = ParallelExecutionLimits::default(); + let request = execution_request!( + &proposal_json, + &manifest, + ®istry, + &schedule_limits, + &execution_limits, + None, + &keys, + ); + + let response = execute_parallel_proposal_via_mcp( + &runtime, + &request, + &token_store, + "a_tracker, + &QuotaLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + let ProposalExecutionResponse::Denied { code, .. } = response else { + return Err(format!("expected Denied, got {response:?}")); + }; + assert_eq!(code, "invalid_parallel_schedule"); + Ok(()) +} + +#[test] +fn execute_parallel_proposal_via_mcp_denies_a_concurrent_side_effect() -> Result<(), String> { + let manifest = diamond_manifest(Some("b")); + let registry = diamond_registry(Some("b")); + let runtime = Runtime::new(registry.clone(), EchoExecutor) + .with_security_config(RuntimeSecurityConfig::development()); + let token_store = ApprovalTokenStore::new(); + let quota_tracker = QuotaTracker::new(); + let keys = keys(); + + let proposal_json = diamond_proposal_json("proposal-exec-side-effect"); + let schedule_limits = ParallelScheduleLimits::default(); + let execution_limits = ParallelExecutionLimits::default(); + let request = execution_request!( + &proposal_json, + &manifest, + ®istry, + &schedule_limits, + &execution_limits, + None, + &keys, + ); + + let response = execute_parallel_proposal_via_mcp( + &runtime, + &request, + &token_store, + "a_tracker, + &QuotaLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + let ProposalExecutionResponse::Denied { code, .. } = response else { + return Err(format!("expected Denied, got {response:?}")); + }; + assert_eq!(code, "concurrent_side_effect_denied"); + Ok(()) +} + +#[test] +fn execute_parallel_proposal_via_mcp_denies_execution_without_a_required_approval_token() +-> Result<(), String> { + // d has an external effect but sits alone in the final wave, so FR-004a + // does not deny it — it still requires a token via FR-006 (P1 carryover). + let manifest = diamond_manifest(Some("d")); + let registry = diamond_registry(Some("d")); + let runtime = Runtime::new(registry.clone(), EchoExecutor) + .with_security_config(RuntimeSecurityConfig::development()); + let token_store = ApprovalTokenStore::new(); + let quota_tracker = QuotaTracker::new(); + let keys = keys(); + + let proposal_json = diamond_proposal_json("proposal-exec-needs-token"); + let schedule_limits = ParallelScheduleLimits::default(); + let execution_limits = ParallelExecutionLimits::default(); + let request = execution_request!( + &proposal_json, + &manifest, + ®istry, + &schedule_limits, + &execution_limits, + None, + &keys, + ); + + let response = execute_parallel_proposal_via_mcp( + &runtime, + &request, + &token_store, + "a_tracker, + &QuotaLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + let ProposalExecutionResponse::Denied { code, .. } = response else { + return Err(format!("expected Denied, got {response:?}")); + }; + assert_eq!(code, "approval_token_required"); + Ok(()) +} + +fn base64url_encode(input: &[u8]) -> String { + const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"; + let mut out = String::new(); + let mut i = 0; + while i + 3 <= input.len() { + let n = + (u32::from(input[i]) << 16) | (u32::from(input[i + 1]) << 8) | u32::from(input[i + 2]); + out.push(ALPHABET[((n >> 18) & 63) as usize] as char); + out.push(ALPHABET[((n >> 12) & 63) as usize] as char); + out.push(ALPHABET[((n >> 6) & 63) as usize] as char); + out.push(ALPHABET[(n & 63) as usize] as char); + i += 3; + } + let remainder = input.len() - i; + if remainder == 1 { + let n = u32::from(input[i]) << 16; + out.push(ALPHABET[((n >> 18) & 63) as usize] as char); + out.push(ALPHABET[((n >> 12) & 63) as usize] as char); + } else if remainder == 2 { + let n = (u32::from(input[i]) << 16) | (u32::from(input[i + 1]) << 8); + out.push(ALPHABET[((n >> 18) & 63) as usize] as char); + out.push(ALPHABET[((n >> 12) & 63) as usize] as char); + out.push(ALPHABET[((n >> 6) & 63) as usize] as char); + } + out +} + +fn sign_token(payload: &Value, key: &SigningKey, key_id: &str) -> String { + let header = base64url_encode(format!(r#"{{"alg":"EdDSA","kid":"{key_id}"}}"#).as_bytes()); + let payload_b64 = base64url_encode(payload.to_string().as_bytes()); + let signing_input = format!("{header}.{payload_b64}"); + let signature = key.sign(signing_input.as_bytes()); + let signature_b64 = base64url_encode(&signature.to_bytes()); + format!("{header}.{payload_b64}.{signature_b64}") +} + +#[test] +fn execute_parallel_proposal_via_mcp_succeeds_with_a_valid_approval_token() -> Result<(), String> { + let manifest = diamond_manifest(Some("d")); + let registry = diamond_registry(Some("d")); + let runtime = Runtime::new(registry.clone(), EchoExecutor) + .with_security_config(RuntimeSecurityConfig::development()); + let token_store = ApprovalTokenStore::new(); + let quota_tracker = QuotaTracker::new(); + + let signing_key = SigningKey::from_bytes(&[9_u8; 32]); + let mut keys = HashMap::new(); + keys.insert("key-1".to_string(), signing_key.verifying_key()); + + let proposal_json = diamond_proposal_json("proposal-exec-with-token"); + let proposal_digest = traverse_contracts::proposal_digest( + &serde_json::from_str::(&proposal_json) + .map_err(|e| e.to_string())?, + ); + let snapshots = traverse_contracts::SnapshotDigests { + manifest_digest: "manifest-1".to_string(), + registry_digest: "registry-1".to_string(), + binding_digest: "binding-1".to_string(), + policy_digest: "policy-1".to_string(), + budget_digest: "budget-1".to_string(), + }; + let snapshot_digest = + traverse_contracts::proposal_snapshot_digest(&proposal_digest, &snapshots); + + let payload = json!({ + "jti": "token-001", + "iss": "traverse-approval-service", + "aud": "traverse-runtime", + "sub": "principal-001", + "workspace_id": "workspace-001", + "proposal_digest": proposal_digest, + "snapshot_digest": snapshot_digest, + "permitted_effects": ["external_effect"], + "permitted_connectors": [], + "max_use_count": 1, + "exp": 4_102_444_800_i64, + }); + let token = sign_token(&payload, &signing_key, "key-1"); + + let request = ParallelProposalExecutionRequest { + proposal_json: &proposal_json, + manifest: &manifest, + registry: ®istry, + proposal_limits: &ProposalLimits::default(), + schedule_limits: &ParallelScheduleLimits::default(), + execution_limits: &ParallelExecutionLimits::default(), + snapshots: &snapshots, + approval_token: Some(&token), + expected_token_issuer: "traverse-approval-service", + expected_token_audience: "traverse-runtime", + token_verifying_keys_by_key_id: &keys, + principal: "principal-001", + app_id: "test-app", + }; + + let response = execute_parallel_proposal_via_mcp( + &runtime, + &request, + &token_store, + "a_tracker, + &QuotaLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + let ProposalExecutionResponse::Trace(trace) = response else { + return Err(format!("expected Trace, got {response:?}")); + }; + assert_eq!(trace.terminal_state, ProposalTerminalState::Succeeded); + Ok(()) +} + +#[test] +fn execute_parallel_proposal_via_mcp_denies_when_quota_is_exhausted() -> Result<(), String> { + let manifest = diamond_manifest(None); + let registry = diamond_registry(None); + let runtime = Runtime::new(registry.clone(), EchoExecutor) + .with_security_config(RuntimeSecurityConfig::development()); + let token_store = ApprovalTokenStore::new(); + let quota_tracker = QuotaTracker::new(); + let keys = keys(); + let quota_limits = QuotaLimits { + max_concurrent_per_principal: 0, + max_concurrent_per_app: 10, + max_concurrent_per_workspace: 10, + }; + + let proposal_json = diamond_proposal_json("proposal-exec-quota"); + let schedule_limits = ParallelScheduleLimits::default(); + let execution_limits = ParallelExecutionLimits::default(); + let request = execution_request!( + &proposal_json, + &manifest, + ®istry, + &schedule_limits, + &execution_limits, + None, + &keys, + ); + + let response = execute_parallel_proposal_via_mcp( + &runtime, + &request, + &token_store, + "a_tracker, + "a_limits, + ) + .map_err(|e| format!("{e:?}"))?; + + let ProposalExecutionResponse::Denied { code, .. } = response else { + return Err(format!("expected Denied, got {response:?}")); + }; + assert_eq!(code, "quota_exhausted_principal"); + Ok(()) +} + +#[test] +fn execute_parallel_proposal_via_mcp_reports_a_failed_terminal_state_when_a_node_fails() +-> Result<(), String> { + let manifest = diamond_manifest(None); + let registry = diamond_registry(None); + let executor = FailsOneCapabilityExecutor { + failing_capability_id: "test.c".to_string(), + }; + let runtime = Runtime::new(registry.clone(), executor) + .with_security_config(RuntimeSecurityConfig::development()); + let token_store = ApprovalTokenStore::new(); + let quota_tracker = QuotaTracker::new(); + let keys = keys(); + + let proposal_json = diamond_proposal_json("proposal-exec-node-fails"); + let schedule_limits = ParallelScheduleLimits::default(); + let execution_limits = ParallelExecutionLimits::default(); + let request = execution_request!( + &proposal_json, + &manifest, + ®istry, + &schedule_limits, + &execution_limits, + None, + &keys, + ); + + let response = execute_parallel_proposal_via_mcp( + &runtime, + &request, + &token_store, + "a_tracker, + &QuotaLimits::default(), + ) + .map_err(|e| format!("{e:?}"))?; + + let ProposalExecutionResponse::Trace(trace) = response else { + return Err(format!("expected Trace, got {response:?}")); + }; + assert_eq!(trace.terminal_state, ProposalTerminalState::Failed); + Ok(()) +} + +#[test] +fn execute_parallel_proposal_via_mcp_reports_invalid_json_as_an_mcp_error() { + let manifest = diamond_manifest(None); + let registry = diamond_registry(None); + let runtime = Runtime::new(registry.clone(), EchoExecutor) + .with_security_config(RuntimeSecurityConfig::development()); + let token_store = ApprovalTokenStore::new(); + let quota_tracker = QuotaTracker::new(); + let keys = keys(); + + let proposal_json = "not json"; + let schedule_limits = ParallelScheduleLimits::default(); + let execution_limits = ParallelExecutionLimits::default(); + let request = execution_request!( + &proposal_json, + &manifest, + ®istry, + &schedule_limits, + &execution_limits, + None, + &keys, + ); + + let result = execute_parallel_proposal_via_mcp( + &runtime, + &request, + &token_store, + "a_tracker, + &QuotaLimits::default(), + ); + assert!(result.is_err()); +} diff --git a/crates/traverse-runtime/src/lib.rs b/crates/traverse-runtime/src/lib.rs index 93b49065..5bab5568 100644 --- a/crates/traverse-runtime/src/lib.rs +++ b/crates/traverse-runtime/src/lib.rs @@ -12,6 +12,7 @@ pub mod executor; /// wasm32 builds because its Ollama implementation requires TCP sockets. #[cfg(feature = "native-inference")] pub mod inference; +pub mod parallel_proposal; pub mod placement; pub mod proposal; pub mod router; diff --git a/crates/traverse-runtime/src/parallel_proposal.rs b/crates/traverse-runtime/src/parallel_proposal.rs new file mode 100644 index 00000000..940a2883 --- /dev/null +++ b/crates/traverse-runtime/src/parallel_proposal.rs @@ -0,0 +1,1119 @@ +//! Bounded deterministic parallel proposal execution (spec +//! `110-bounded-parallel-workflow-scheduling`, P2, ADR-0042). +//! +//! Extends the P1 sequential proposal executor in `crate::proposal` with a +//! wave-based concurrent dispatcher over the same `CanonicalProposal`/ +//! `ResolvedProposalNode` shapes. A [`traverse_contracts::ParallelSchedule`] +//! (computed by `traverse_contracts::compute_parallel_schedule`) levelizes +//! the already-validated acyclic graph into waves of node ids whose +//! dependencies are satisfied by earlier waves; this module authorizes and +//! executes that schedule. +//! +//! FR-004a (spec 110): the first P2 implementation permits a wave with more +//! than one member only when every member's declared `effect_class` is +//! `pure_read` — [`enforce_pure_read_only_parallelism`] checks this before +//! any dispatch. Once authorized, each wave runs on real OS threads via +//! [`std::thread::scope`], bounded to `max_concurrent_nodes` per batch; +//! outcomes are folded back into the trace in the wave's lexicographic +//! order (not completion order), so the observable trace is deterministic +//! regardless of real scheduling (FR-002). +//! +//! Wall-clock and payload-size bounds (FR-001, FR-005) are checked *between* +//! waves, before committing to the next one — Rust gives no safe way to +//! preemptively interrupt an in-flight OS thread without `unsafe`, and +//! FR-004a already restricts concurrent work to side-effect-free local reads, +//! so a wave that is already dispatched is always allowed to finish; a +//! budget that is already exhausted simply stops further waves from +//! starting, reported as [`traverse_contracts::CanonicalProposal`]'s trace +//! terminal state `cancelled`. + +use serde::Serialize; +use serde_json::Value; +use std::collections::HashMap; +use std::time::{Duration, Instant}; + +use traverse_contracts::{CanonicalProposal, EffectClass, ParallelSchedule, ProposalNode}; + +use crate::proposal::{ + AuthorizationSummary, ProposalNodeOutcome, ProposalNodeStatus, ProposalTerminalState, + ProposalTrace, ResolvedProposalNode, assemble_node_input, build_node_execution_request, +}; +use crate::{Runtime, RuntimeResultStatus}; + +// --------------------------------------------------------------------------- +// FR-004a: pure_read-only concurrency authorization +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct ParallelAuthorizationError { + pub code: ParallelAuthorizationErrorCode, + pub message: String, + pub path: String, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ParallelAuthorizationErrorCode { + ConcurrentSideEffectDenied, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ParallelAuthorizationFailure { + pub errors: Vec, +} + +/// Enforces spec 110 FR-004a: the first P2 implementation permits parallel +/// execution only for `pure_read` nodes. Any wave with more than one member +/// is denied outright unless every member in it is `pure_read`. +/// +/// # Errors +/// +/// Returns [`ParallelAuthorizationFailure`] listing every offending node. +pub fn enforce_pure_read_only_parallelism( + schedule: &ParallelSchedule, + resolved_nodes: &[ResolvedProposalNode], +) -> Result<(), ParallelAuthorizationFailure> { + let effect_class_by_node: HashMap<&str, EffectClass> = resolved_nodes + .iter() + .map(|node| (node.node_id.as_str(), node.contract.risk.effect_class)) + .collect(); + + let mut errors = Vec::new(); + for (wave_index, wave) in schedule.waves.iter().enumerate() { + if wave.len() <= 1 { + continue; + } + for node_id in wave { + if effect_class_by_node.get(node_id.as_str()) != Some(&EffectClass::PureRead) { + errors.push(ParallelAuthorizationError { + code: ParallelAuthorizationErrorCode::ConcurrentSideEffectDenied, + message: format!( + "node '{node_id}' does not have effect_class pure_read and cannot run \ + concurrently with other nodes in wave {wave_index}" + ), + path: format!("$.schedule.waves[{wave_index}]"), + }); + } + } + } + + if errors.is_empty() { + Ok(()) + } else { + Err(ParallelAuthorizationFailure { errors }) + } +} + +// --------------------------------------------------------------------------- +// Execution-time bounds (spec 110 FR-001, FR-005): wall time and payload size +// --------------------------------------------------------------------------- + +/// Execution-time bounds that structural schedule validation cannot express +/// (spec 110 FR-001: "time, memory" bounds). Checked between waves. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ParallelExecutionLimits { + /// Total wall-clock budget for the whole parallel execution. Checked + /// before starting each wave, not preemptively mid-wave. + pub max_wall_time: Duration, + /// Max total serialized JSON byte size of one wave's assembled node + /// inputs — a bounded, honest proxy for a per-wave memory budget. + pub max_wave_payload_bytes: usize, + /// Max node executions the runtime dispatches at once within a wave — + /// the execution-time enforcement of + /// `ParallelScheduleLimits::max_concurrent_nodes`. + pub max_concurrent_nodes: usize, +} + +pub const DEFAULT_MAX_WALL_TIME_MS: u64 = 30_000; +pub const DEFAULT_MAX_WAVE_PAYLOAD_BYTES: usize = 1_048_576; + +impl Default for ParallelExecutionLimits { + fn default() -> Self { + Self { + max_wall_time: Duration::from_millis(DEFAULT_MAX_WALL_TIME_MS), + max_wave_payload_bytes: DEFAULT_MAX_WAVE_PAYLOAD_BYTES, + max_concurrent_nodes: traverse_contracts::DEFAULT_MAX_CONCURRENT_NODES, + } + } +} + +// --------------------------------------------------------------------------- +// Execution +// --------------------------------------------------------------------------- + +/// Executes an authorized [`ParallelSchedule`] wave by wave, dispatching +/// each wave's nodes concurrently (bounded to +/// `limits.max_concurrent_nodes` batches) when the wave has more than one +/// member, and folding results back in lexicographic order regardless of +/// real completion order (spec 110 FR-002). +/// +/// Stops advancing to further waves — but never interrupts an +/// already-dispatched one — at the first node failure, exhausted wall-time +/// budget, or exceeded wave payload budget (spec 110 FR-005, FR-008 +/// carried over from P1). +#[must_use] +#[allow(clippy::too_many_lines)] +pub fn execute_parallel_proposal( + runtime: &Runtime, + canonical: &CanonicalProposal, + schedule: &ParallelSchedule, + authorization: AuthorizationSummary, + proposal_digest: &str, + snapshot_digest: &str, + limits: &ParallelExecutionLimits, +) -> ProposalTrace +where + Runtime: Sync, +{ + let nodes_by_id: HashMap<&str, &ProposalNode> = canonical + .proposal + .nodes + .iter() + .map(|node| (node.node_id.as_str(), node)) + .collect(); + + let mut outputs: HashMap = HashMap::new(); + let mut outcomes: Vec = Vec::with_capacity(canonical.proposal.nodes.len()); + let mut failed = false; + let mut cancelled = false; + let start = Instant::now(); + let batch_size = limits.max_concurrent_nodes.max(1); + + for wave in &schedule.waves { + if failed || cancelled { + push_skipped_wave(&mut outcomes, wave, &nodes_by_id); + continue; + } + if start.elapsed() > limits.max_wall_time { + cancelled = true; + push_skipped_wave(&mut outcomes, wave, &nodes_by_id); + continue; + } + + let wave_inputs: Vec<(String, Value)> = wave + .iter() + .map(|node_id| { + ( + node_id.clone(), + assemble_node_input(canonical, node_id, &outputs), + ) + }) + .collect(); + let wave_payload_bytes: usize = wave_inputs + .iter() + .map(|(_, input)| input.to_string().len()) + .sum(); + if wave_payload_bytes > limits.max_wave_payload_bytes { + cancelled = true; + push_skipped_wave(&mut outcomes, wave, &nodes_by_id); + continue; + } + + for batch in wave_inputs.chunks(batch_size) { + for (node_id, outcome, output) in + dispatch_batch(runtime, canonical, &nodes_by_id, batch) + { + if outcome.status == ProposalNodeStatus::Failed { + failed = true; + } + if let Some(output) = output { + outputs.insert(node_id, output); + } + outcomes.push(outcome); + } + } + } + + let terminal_state = if failed { + ProposalTerminalState::Failed + } else if cancelled { + ProposalTerminalState::Cancelled + } else { + ProposalTerminalState::Succeeded + }; + + ProposalTrace { + proposal_id: canonical.proposal.proposal_id.clone(), + proposal_digest: proposal_digest.to_string(), + snapshot_digest: snapshot_digest.to_string(), + authorization, + node_outcomes: outcomes, + mapping_paths: canonical + .proposal + .mappings + .iter() + .map(|m| (m.source_path.clone(), m.target_path.clone())) + .collect(), + terminal_state, + } +} + +/// Dispatches one bounded batch of a wave concurrently on real OS threads +/// and returns each entry's outcome (plus its output to store, on success) +/// in the batch's original lexicographic order, independent of actual +/// completion order. A node id absent from `nodes_by_id` (only reachable by +/// hand-constructing a [`ParallelSchedule`] that disagrees with `canonical` +/// — never produced by `traverse_contracts::compute_parallel_schedule`) is +/// silently skipped. A panicking executor is surfaced as a `Failed` outcome +/// rather than silently dropped, matching this crate's fail-closed +/// convention for host-side faults. +fn dispatch_batch<'a, E: crate::LocalExecutor>( + runtime: &Runtime, + canonical: &CanonicalProposal, + nodes_by_id: &HashMap<&'a str, &'a ProposalNode>, + batch: &[(String, Value)], +) -> Vec<(String, ProposalNodeOutcome, Option)> +where + Runtime: Sync, +{ + let dispatchable: Vec<(&str, &'a ProposalNode, &Value)> = batch + .iter() + .filter_map(|(node_id, input)| { + nodes_by_id + .get(node_id.as_str()) + .map(|node| (node_id.as_str(), *node, input)) + }) + .collect(); + + let joined: Vec<( + String, + &'a ProposalNode, + std::thread::Result, + )> = std::thread::scope(|scope| { + let handles: Vec<( + String, + &'a ProposalNode, + std::thread::ScopedJoinHandle<'_, crate::RuntimeExecutionOutcome>, + )> = dispatchable + .iter() + .map(|(node_id, node, input)| { + let owned_node_id = (*node_id).to_string(); + let spawn_node_id = owned_node_id.clone(); + let node = *node; + let input = (*input).clone(); + let handle = scope.spawn(move || { + let request = + build_node_execution_request(canonical, node, &spawn_node_id, input); + runtime.execute(request) + }); + (owned_node_id, node, handle) + }) + .collect(); + handles + .into_iter() + .map(|(node_id, node, handle)| (node_id, node, handle.join())) + .collect() + }); + + joined + .into_iter() + .map(|(node_id, node, joined_result)| match joined_result { + Ok(outcome) => match outcome.result.status { + RuntimeResultStatus::Completed => ( + node_id.clone(), + succeeded_outcome(&node_id, node), + outcome.result.output.clone(), + ), + RuntimeResultStatus::Error => ( + node_id.clone(), + failed_outcome(&node_id, node, &outcome), + None, + ), + }, + Err(_) => (node_id.clone(), panicked_outcome(&node_id, node), None), + }) + .collect() +} + +fn succeeded_outcome(node_id: &str, node: &ProposalNode) -> ProposalNodeOutcome { + ProposalNodeOutcome { + node_id: node_id.to_string(), + capability_id: node.capability_id.clone(), + capability_version: node.capability_version.clone(), + artifact_digest: node.artifact_digest.clone(), + status: ProposalNodeStatus::Succeeded, + error_code: None, + } +} + +fn failed_outcome( + node_id: &str, + node: &ProposalNode, + outcome: &crate::RuntimeExecutionOutcome, +) -> ProposalNodeOutcome { + ProposalNodeOutcome { + node_id: node_id.to_string(), + capability_id: node.capability_id.clone(), + capability_version: node.capability_version.clone(), + artifact_digest: node.artifact_digest.clone(), + status: ProposalNodeStatus::Failed, + error_code: outcome + .result + .error + .as_ref() + .map(|error| format!("{:?}", error.code)), + } +} + +/// A node execution thread panicked. Surfaced as a `Failed` outcome — never +/// silently dropped — matching this crate's fail-closed convention for +/// host-side faults (e.g. `events/broker.rs`'s poisoned-lock handling). +fn panicked_outcome(node_id: &str, node: &ProposalNode) -> ProposalNodeOutcome { + ProposalNodeOutcome { + node_id: node_id.to_string(), + capability_id: node.capability_id.clone(), + capability_version: node.capability_version.clone(), + artifact_digest: node.artifact_digest.clone(), + status: ProposalNodeStatus::Failed, + error_code: Some("executor_panicked".to_string()), + } +} + +fn push_skipped_wave( + outcomes: &mut Vec, + wave: &[String], + nodes_by_id: &HashMap<&str, &ProposalNode>, +) { + for node_id in wave { + let Some(node) = nodes_by_id.get(node_id.as_str()) else { + continue; + }; + outcomes.push(ProposalNodeOutcome { + node_id: node_id.clone(), + capability_id: node.capability_id.clone(), + capability_version: node.capability_version.clone(), + artifact_digest: node.artifact_digest.clone(), + status: ProposalNodeStatus::SkippedAfterEarlierFailure, + error_code: None, + }); + } +} + +#[cfg(test)] +#[allow(clippy::expect_used)] +#[allow(clippy::panic)] +mod tests { + use super::*; + use crate::security::RuntimeSecurityConfig; + use crate::{ + LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutionOutput, LocalExecutor, + }; + use serde_json::json; + use std::collections::{HashMap as StdHashMap, HashSet}; + use std::sync::{Arc, Mutex}; + use traverse_contracts::{ + BinaryFormat as ContractBinaryFormat, CapabilityContract, DataFlowPolicy, DeterminismClass, + Entrypoint, EntrypointKind, Execution, ExecutionConstraints, ExecutionTarget, + FilesystemAccess, HostApiAccess, Lifecycle, ManifestReference, NetworkAccess, Owner, + ParallelScheduleLimits, ProposalEdge, ProposalLimits, ReliabilityMetadata, RiskMetadata, + SchemaContainer, ServiceType, SideEffect, SideEffectKind, WorkflowProposal, + canonicalize_proposal, compute_parallel_schedule, + }; + use traverse_registry::{ + ArtifactDigests, BinaryFormat as RegistryBinaryFormat, BinaryReference, + CapabilityArtifactRecord, CapabilityRegistration, CapabilityRegistry, + ComposabilityMetadata, CompositionKind, CompositionPattern, ImplementationKind, + RegistryProvenance, RegistryScope, SourceKind, SourceReference, + }; + + fn risk(effect_class: EffectClass) -> RiskMetadata { + RiskMetadata { + effect_class, + determinism_class: DeterminismClass::Deterministic, + data_flow: DataFlowPolicy::default(), + reliability: ReliabilityMetadata { + idempotency_required: false, + retryable: true, + compensation_available: false, + }, + } + } + + fn contract(capability_id: &str, effect_class: EffectClass) -> CapabilityContract { + let (namespace, name) = capability_id + .rsplit_once('.') + .unwrap_or(("test", capability_id)); + CapabilityContract { + kind: "capability_contract".to_string(), + schema_version: "1.0.0".to_string(), + id: capability_id.to_string(), + namespace: namespace.to_string(), + name: name.to_string(), + version: "1.0.0".to_string(), + lifecycle: Lifecycle::Active, + owner: Owner { + team: "traverse-core".to_string(), + contact: "enrico.piovesan10@gmail.com".to_string(), + }, + summary: "Test capability for parallel proposal scheduling.".to_string(), + description: "Portable test capability used to validate parallel scheduling." + .to_string(), + inputs: SchemaContainer { schema: json!({}) }, + outputs: SchemaContainer { schema: json!({}) }, + preconditions: Vec::new(), + postconditions: Vec::new(), + side_effects: vec![SideEffect { + kind: SideEffectKind::MemoryOnly, + description: "No durable side effect.".to_string(), + }], + emits: Vec::new(), + consumes: Vec::new(), + permissions: Vec::new(), + execution: Execution { + binary_format: ContractBinaryFormat::Wasm, + entrypoint: Entrypoint { + kind: EntrypointKind::WasiCommand, + command: "run".to_string(), + }, + preferred_targets: vec![ExecutionTarget::Local], + constraints: ExecutionConstraints { + host_api_access: HostApiAccess::None, + network_access: NetworkAccess::Forbidden, + filesystem_access: FilesystemAccess::None, + }, + }, + policies: Vec::new(), + dependencies: Vec::new(), + provenance: traverse_contracts::Provenance { + source: traverse_contracts::ProvenanceSource::Greenfield, + author: "test".to_string(), + created_at: "2026-08-23T00:00:00Z".to_string(), + spec_ref: None, + adr_refs: Vec::new(), + exception_refs: Vec::new(), + }, + evidence: Vec::new(), + service_type: ServiceType::Stateless, + permitted_targets: vec![ExecutionTarget::Local], + event_trigger: None, + connector_requirements: Vec::new(), + state_schema: None, + use_cases: Vec::new(), + risk: risk(effect_class), + } + } + + fn artifact(digest: &str) -> CapabilityArtifactRecord { + CapabilityArtifactRecord { + artifact_ref: format!("artifact:{digest}"), + implementation_kind: ImplementationKind::Executable, + source: SourceReference { + kind: SourceKind::Git, + location: "https://example.invalid/repo".to_string(), + }, + binary: Some(BinaryReference { + format: RegistryBinaryFormat::Wasm, + location: format!("artifacts/{digest}/capability.wasm"), + signature: None, + }), + workflow_ref: None, + digests: ArtifactDigests { + source_digest: format!("src-{digest}"), + binary_digest: Some(digest.to_string()), + }, + provenance: RegistryProvenance { + source: "test".to_string(), + author: "test".to_string(), + created_at: "2026-08-23T00:00:00Z".to_string(), + }, + } + } + + fn registry_with( + entries: Vec<(CapabilityContract, CapabilityArtifactRecord)>, + ) -> CapabilityRegistry { + let mut registry = CapabilityRegistry::new(); + for (contract, artifact) in entries { + let outcome = registry.register(CapabilityRegistration { + scope: RegistryScope::Public, + contract, + contract_path: "registry/test/contract.json".to_string(), + artifact, + registered_at: "2026-08-23T00:00:00Z".to_string(), + tags: Vec::new(), + composability: ComposabilityMetadata { + kind: CompositionKind::Atomic, + patterns: vec![CompositionPattern::Sequential], + provides: Vec::new(), + requires: Vec::new(), + }, + governing_spec: "005-capability-registry".to_string(), + validator_version: "0.1.0".to_string(), + }); + assert!(outcome.is_ok(), "registration must succeed: {outcome:?}"); + } + registry + } + + fn node(node_id: &str, capability_id: &str) -> ProposalNode { + ProposalNode { + node_id: node_id.to_string(), + capability_id: capability_id.to_string(), + capability_version: "1.0.0".to_string(), + artifact_digest: format!("digest-{node_id}"), + } + } + + fn resolved(node_id: &str, effect_class: EffectClass) -> ResolvedProposalNode { + ResolvedProposalNode { + node_id: node_id.to_string(), + contract: contract(&format!("test.{node_id}"), effect_class), + } + } + + /// a fans out to b and c (both independently mapped from a's output), + /// which both feed the join node d. + fn diamond_workflow_proposal() -> WorkflowProposal { + WorkflowProposal { + kind: "workflow_proposal".to_string(), + schema_version: "1.0.0".to_string(), + proposal_id: "proposal-p2-001".to_string(), + workspace_id: "workspace-001".to_string(), + app_manifest: ManifestReference { + app_id: "test-app".to_string(), + app_version: "1.0.0".to_string(), + manifest_digest: "sha256:manifest-digest".to_string(), + }, + nodes: vec![ + node("a", "test.a"), + node("b", "test.b"), + node("c", "test.c"), + node("d", "test.d"), + ], + edges: vec![ + ProposalEdge { + from_node_id: "a".to_string(), + to_node_id: "b".to_string(), + }, + ProposalEdge { + from_node_id: "a".to_string(), + to_node_id: "c".to_string(), + }, + ProposalEdge { + from_node_id: "b".to_string(), + to_node_id: "d".to_string(), + }, + ProposalEdge { + from_node_id: "c".to_string(), + to_node_id: "d".to_string(), + }, + ], + mappings: Vec::new(), + initial_input: json!({}), + } + } + + fn diamond_canonical() -> CanonicalProposal { + canonicalize_proposal(diamond_workflow_proposal(), &ProposalLimits::default()) + .expect("diamond proposal must canonicalize") + } + + fn diamond_resolved_nodes(non_read_effect: Option<&str>) -> Vec { + ["a", "b", "c", "d"] + .iter() + .map(|id| { + let effect_class = if Some(*id) == non_read_effect { + EffectClass::ExternalEffect + } else { + EffectClass::PureRead + }; + resolved(id, effect_class) + }) + .collect() + } + + // -- FR-004a: pure_read-only concurrency authorization ------------------- + + #[test] + fn allows_a_diamond_schedule_when_every_concurrent_wave_is_pure_read() -> Result<(), String> { + let canonical = diamond_canonical(); + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + enforce_pure_read_only_parallelism(&schedule, &diamond_resolved_nodes(None)) + .map_err(|e| format!("{e:?}")) + } + + #[test] + fn denies_a_concurrent_wave_containing_a_non_pure_read_node() -> Result<(), String> { + let canonical = diamond_canonical(); + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let failure = + enforce_pure_read_only_parallelism(&schedule, &diamond_resolved_nodes(Some("c"))) + .expect_err("a non-pure_read node in a concurrent wave must be denied"); + assert!( + failure + .errors + .iter() + .any(|e| e.code == ParallelAuthorizationErrorCode::ConcurrentSideEffectDenied) + ); + Ok(()) + } + + #[test] + fn allows_a_non_pure_read_node_when_its_wave_has_no_sibling() -> Result<(), String> { + // b and c are pure_read and run concurrently in wave 1; a and d are + // singleton waves and may be any effect class. + let canonical = diamond_canonical(); + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + enforce_pure_read_only_parallelism(&schedule, &diamond_resolved_nodes(Some("d"))) + .map_err(|e| format!("{e:?}")) + } + + // -- Execution ------------------------------------------------------------- + + #[derive(Default)] + struct ConcurrencyTracker { + current: Mutex, + max_seen: Mutex, + } + + impl ConcurrencyTracker { + fn enter(&self) { + let mut current = self + .current + .lock() + .expect("tracker lock must not be poisoned"); + *current += 1; + let mut max_seen = self + .max_seen + .lock() + .expect("tracker lock must not be poisoned"); + if *current > *max_seen { + *max_seen = *current; + } + } + + fn exit(&self) { + let mut current = self + .current + .lock() + .expect("tracker lock must not be poisoned"); + *current -= 1; + } + + fn max_seen(&self) -> usize { + *self + .max_seen + .lock() + .expect("tracker lock must not be poisoned") + } + } + + #[derive(Default, Clone)] + struct ScriptedExecutor { + fail_capability_ids: HashSet, + panic_capability_ids: HashSet, + sleep_capability_ids: StdHashMap, + concurrency: Arc, + } + + impl LocalExecutor for ScriptedExecutor { + fn execute( + &self, + capability: &traverse_registry::ResolvedCapability, + input: &Value, + ) -> Result { + self.concurrency.enter(); + if self.panic_capability_ids.contains(&capability.contract.id) { + self.concurrency.exit(); + panic!("scripted executor panic for test"); + } + if let Some(duration) = self.sleep_capability_ids.get(&capability.contract.id) { + std::thread::sleep(*duration); + } + let result = if self.fail_capability_ids.contains(&capability.contract.id) { + Err(LocalExecutionFailure { + code: LocalExecutionFailureCode::ExecutionFailed, + message: "scripted failure".to_string(), + }) + } else { + Ok(LocalExecutionOutput { + value: json!({"node": capability.contract.id, "received": input.clone()}), + emitted_events: Vec::new(), + }) + }; + self.concurrency.exit(); + result + } + } + + fn diamond_registry() -> CapabilityRegistry { + registry_with(vec![ + ( + contract("test.a", EffectClass::PureRead), + artifact("digest-a"), + ), + ( + contract("test.b", EffectClass::PureRead), + artifact("digest-b"), + ), + ( + contract("test.c", EffectClass::PureRead), + artifact("digest-c"), + ), + ( + contract("test.d", EffectClass::PureRead), + artifact("digest-d"), + ), + ]) + } + + #[test] + fn executes_independent_branches_concurrently_with_a_deterministic_trace_order() + -> Result<(), String> { + let canonical = diamond_canonical(); + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + + let mut sleep_capability_ids = StdHashMap::new(); + sleep_capability_ids.insert("test.c".to_string(), std::time::Duration::from_millis(20)); + let executor = ScriptedExecutor { + sleep_capability_ids, + ..ScriptedExecutor::default() + }; + let runtime = Runtime::new(diamond_registry(), executor) + .with_security_config(RuntimeSecurityConfig::development()); + + let trace = execute_parallel_proposal( + &runtime, + &canonical, + &schedule, + AuthorizationSummary { + automatic: true, + approval_token_id: None, + }, + "digest", + "snapshot-digest", + &ParallelExecutionLimits::default(), + ); + + assert_eq!(trace.terminal_state, ProposalTerminalState::Succeeded); + // b finishes before c (c sleeps), but the trace order is always + // lexicographic (b, c), never completion order. + let node_order: Vec<&str> = trace + .node_outcomes + .iter() + .map(|o| o.node_id.as_str()) + .collect(); + assert_eq!(node_order, vec!["a", "b", "c", "d"]); + assert!( + trace + .node_outcomes + .iter() + .all(|o| o.status == ProposalNodeStatus::Succeeded) + ); + Ok(()) + } + + #[test] + fn bounds_real_concurrency_to_the_configured_max_concurrent_nodes() -> Result<(), String> { + let canonical = diamond_canonical(); + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + + let mut sleep_capability_ids = StdHashMap::new(); + sleep_capability_ids.insert("test.b".to_string(), std::time::Duration::from_millis(15)); + sleep_capability_ids.insert("test.c".to_string(), std::time::Duration::from_millis(15)); + let executor = ScriptedExecutor { + sleep_capability_ids, + ..ScriptedExecutor::default() + }; + let concurrency = Arc::clone(&executor.concurrency); + let runtime = Runtime::new(diamond_registry(), executor) + .with_security_config(RuntimeSecurityConfig::development()); + + let limits = ParallelExecutionLimits { + max_concurrent_nodes: 1, + ..ParallelExecutionLimits::default() + }; + let trace = execute_parallel_proposal( + &runtime, + &canonical, + &schedule, + AuthorizationSummary { + automatic: true, + approval_token_id: None, + }, + "digest", + "snapshot-digest", + &limits, + ); + + assert_eq!(trace.terminal_state, ProposalTerminalState::Succeeded); + assert_eq!(concurrency.max_seen(), 1); + Ok(()) + } + + #[test] + fn stops_advancing_after_a_node_failure_but_lets_the_dispatched_wave_finish() + -> Result<(), String> { + let canonical = diamond_canonical(); + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + + let mut fail_capability_ids = HashSet::new(); + fail_capability_ids.insert("test.c".to_string()); + let executor = ScriptedExecutor { + fail_capability_ids, + ..ScriptedExecutor::default() + }; + let runtime = Runtime::new(diamond_registry(), executor) + .with_security_config(RuntimeSecurityConfig::development()); + + let trace = execute_parallel_proposal( + &runtime, + &canonical, + &schedule, + AuthorizationSummary { + automatic: true, + approval_token_id: None, + }, + "digest", + "snapshot-digest", + &ParallelExecutionLimits::default(), + ); + + assert_eq!(trace.terminal_state, ProposalTerminalState::Failed); + let outcome_by_node: StdHashMap<&str, ProposalNodeStatus> = trace + .node_outcomes + .iter() + .map(|o| (o.node_id.as_str(), o.status.clone())) + .collect(); + assert_eq!(outcome_by_node["a"], ProposalNodeStatus::Succeeded); + assert_eq!(outcome_by_node["b"], ProposalNodeStatus::Succeeded); + assert_eq!(outcome_by_node["c"], ProposalNodeStatus::Failed); + assert_eq!( + outcome_by_node["d"], + ProposalNodeStatus::SkippedAfterEarlierFailure + ); + Ok(()) + } + + #[test] + fn surfaces_an_executor_panic_as_a_failed_outcome_instead_of_dropping_it() -> Result<(), String> + { + let canonical = diamond_canonical(); + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + + let mut panic_capability_ids = HashSet::new(); + panic_capability_ids.insert("test.c".to_string()); + let executor = ScriptedExecutor { + panic_capability_ids, + ..ScriptedExecutor::default() + }; + let runtime = Runtime::new(diamond_registry(), executor) + .with_security_config(RuntimeSecurityConfig::development()); + + let trace = execute_parallel_proposal( + &runtime, + &canonical, + &schedule, + AuthorizationSummary { + automatic: true, + approval_token_id: None, + }, + "digest", + "snapshot-digest", + &ParallelExecutionLimits::default(), + ); + + assert_eq!(trace.terminal_state, ProposalTerminalState::Failed); + let c_outcome = trace + .node_outcomes + .iter() + .find(|o| o.node_id == "c") + .expect("c must have an outcome, not be silently dropped"); + assert_eq!(c_outcome.status, ProposalNodeStatus::Failed); + assert_eq!(c_outcome.error_code, Some("executor_panicked".to_string())); + Ok(()) + } + + #[test] + fn cancels_further_waves_once_the_wall_time_budget_is_exhausted() -> Result<(), String> { + let canonical = diamond_canonical(); + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + + let mut sleep_capability_ids = StdHashMap::new(); + sleep_capability_ids.insert("test.a".to_string(), std::time::Duration::from_millis(40)); + let executor = ScriptedExecutor { + sleep_capability_ids, + ..ScriptedExecutor::default() + }; + let runtime = Runtime::new(diamond_registry(), executor) + .with_security_config(RuntimeSecurityConfig::development()); + + let limits = ParallelExecutionLimits { + max_wall_time: std::time::Duration::from_millis(10), + ..ParallelExecutionLimits::default() + }; + let trace = execute_parallel_proposal( + &runtime, + &canonical, + &schedule, + AuthorizationSummary { + automatic: true, + approval_token_id: None, + }, + "digest", + "snapshot-digest", + &limits, + ); + + assert_eq!(trace.terminal_state, ProposalTerminalState::Cancelled); + let outcome_by_node: StdHashMap<&str, ProposalNodeStatus> = trace + .node_outcomes + .iter() + .map(|o| (o.node_id.as_str(), o.status.clone())) + .collect(); + assert_eq!(outcome_by_node["a"], ProposalNodeStatus::Succeeded); + assert_eq!( + outcome_by_node["b"], + ProposalNodeStatus::SkippedAfterEarlierFailure + ); + Ok(()) + } + + #[test] + fn cancels_a_wave_whose_assembled_payload_exceeds_the_configured_byte_budget() + -> Result<(), String> { + let canonical = diamond_canonical(); + let schedule = compute_parallel_schedule(&canonical, &ParallelScheduleLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let executor = ScriptedExecutor::default(); + let runtime = Runtime::new(diamond_registry(), executor) + .with_security_config(RuntimeSecurityConfig::development()); + + let limits = ParallelExecutionLimits { + max_wave_payload_bytes: 0, + ..ParallelExecutionLimits::default() + }; + let trace = execute_parallel_proposal( + &runtime, + &canonical, + &schedule, + AuthorizationSummary { + automatic: true, + approval_token_id: None, + }, + "digest", + "snapshot-digest", + &limits, + ); + + assert_eq!(trace.terminal_state, ProposalTerminalState::Cancelled); + Ok(()) + } + + #[test] + fn execute_parallel_proposal_skips_an_unresolved_node_id_within_a_dispatched_wave() + -> Result<(), String> { + // `ParallelSchedule` has public fields, so a caller could hand-build + // one that disagrees with `canonical` — never produced by + // `compute_parallel_schedule` itself. This proves the executor + // degrades gracefully (silently skips the unresolved id) rather than + // panicking. + let mut proposal = diamond_workflow_proposal(); + proposal.nodes.retain(|n| n.node_id == "a"); + proposal.edges.clear(); + let canonical = canonicalize_proposal(proposal, &ProposalLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let schedule = ParallelSchedule { + waves: vec![vec!["a".to_string(), "ghost".to_string()]], + }; + + let executor = ScriptedExecutor::default(); + let runtime = Runtime::new( + registry_with(vec![( + contract("test.a", EffectClass::PureRead), + artifact("digest-a"), + )]), + executor, + ) + .with_security_config(RuntimeSecurityConfig::development()); + + let trace = execute_parallel_proposal( + &runtime, + &canonical, + &schedule, + AuthorizationSummary { + automatic: true, + approval_token_id: None, + }, + "digest", + "snapshot-digest", + &ParallelExecutionLimits::default(), + ); + + assert_eq!(trace.terminal_state, ProposalTerminalState::Succeeded); + assert_eq!(trace.node_outcomes.len(), 1); + assert_eq!(trace.node_outcomes[0].node_id, "a"); + Ok(()) + } + + #[test] + fn execute_parallel_proposal_skips_an_unresolved_node_id_in_a_skipped_wave() + -> Result<(), String> { + let mut proposal = diamond_workflow_proposal(); + proposal.nodes.retain(|n| n.node_id == "a"); + proposal.edges.clear(); + let canonical = canonicalize_proposal(proposal, &ProposalLimits::default()) + .map_err(|e| format!("{e:?}"))?; + let schedule = ParallelSchedule { + waves: vec![vec!["a".to_string()], vec!["ghost".to_string()]], + }; + + let mut fail_capability_ids = HashSet::new(); + fail_capability_ids.insert("test.a".to_string()); + let executor = ScriptedExecutor { + fail_capability_ids, + ..ScriptedExecutor::default() + }; + let runtime = Runtime::new( + registry_with(vec![( + contract("test.a", EffectClass::PureRead), + artifact("digest-a"), + )]), + executor, + ) + .with_security_config(RuntimeSecurityConfig::development()); + + let trace = execute_parallel_proposal( + &runtime, + &canonical, + &schedule, + AuthorizationSummary { + automatic: true, + approval_token_id: None, + }, + "digest", + "snapshot-digest", + &ParallelExecutionLimits::default(), + ); + + assert_eq!(trace.terminal_state, ProposalTerminalState::Failed); + assert_eq!(trace.node_outcomes.len(), 1); + assert_eq!(trace.node_outcomes[0].node_id, "a"); + Ok(()) + } + + #[test] + fn parallel_execution_limits_default_matches_the_documented_defaults() { + let limits = ParallelExecutionLimits::default(); + assert_eq!( + limits.max_wall_time, + std::time::Duration::from_millis(DEFAULT_MAX_WALL_TIME_MS) + ); + assert_eq!( + limits.max_wave_payload_bytes, + DEFAULT_MAX_WAVE_PAYLOAD_BYTES + ); + assert_eq!( + limits.max_concurrent_nodes, + traverse_contracts::DEFAULT_MAX_CONCURRENT_NODES + ); + } +} diff --git a/crates/traverse-runtime/src/proposal.rs b/crates/traverse-runtime/src/proposal.rs index e674ed41..a0831909 100644 --- a/crates/traverse-runtime/src/proposal.rs +++ b/crates/traverse-runtime/src/proposal.rs @@ -968,51 +968,8 @@ pub fn execute_proposal( continue; } - let mut input = Value::Object(serde_json::Map::new()); - for mapping in &canonical.proposal.mappings { - if mapping.target_node_id != *node_id { - continue; - } - let value = match &mapping.source { - MappingSource::InitialInput => { - pointer_get(&canonical.proposal.initial_input, &mapping.source_path) - } - MappingSource::Node { node_id: source_id } => outputs - .get(source_id) - .and_then(|output| pointer_get(output, &mapping.source_path)), - }; - if let Some(value) = value { - pointer_set(&mut input, &mapping.target_path, value.clone()); - } - } - - let request = RuntimeRequest { - kind: "runtime_request".to_string(), - schema_version: "1.0.0".to_string(), - request_id: format!("{}-{node_id}", canonical.proposal.proposal_id), - intent: RuntimeIntent { - capability_id: Some(node.capability_id.clone()), - capability_version: Some(node.capability_version.clone()), - version_range: None, - intent_key: None, - }, - input, - lookup: RuntimeLookup { - scope: RuntimeLookupScope::PreferPrivate, - allow_ambiguity: false, - }, - context: RuntimeContext { - requested_target: PlacementTarget::Local, - correlation_id: Some(canonical.proposal.proposal_id.clone()), - caller: Some("workflow_proposal".to_string()), - traceparent: None, - tracestate: None, - metadata: None, - identity: None, - }, - governing_spec: "006-runtime-request-execution".to_string(), - }; - + let input = assemble_node_input(canonical, node_id, &outputs); + let request = build_node_execution_request(canonical, node, node_id, input); let outcome = runtime.execute(request); match outcome.result.status { RuntimeResultStatus::Completed => { @@ -1069,11 +1026,75 @@ pub fn execute_proposal( } } -fn pointer_get<'a>(value: &'a Value, pointer: &str) -> Option<&'a Value> { +/// Resolves every mapping targeting `node_id` against `outputs` (and the +/// proposal's `initial_input`) into that node's assembled input object. +/// Shared by the sequential P1 executor and the P2 wave executor. +pub(crate) fn assemble_node_input( + canonical: &CanonicalProposal, + node_id: &str, + outputs: &HashMap, +) -> Value { + let mut input = Value::Object(serde_json::Map::new()); + for mapping in &canonical.proposal.mappings { + if mapping.target_node_id != *node_id { + continue; + } + let value = match &mapping.source { + MappingSource::InitialInput => { + pointer_get(&canonical.proposal.initial_input, &mapping.source_path) + } + MappingSource::Node { node_id: source_id } => outputs + .get(source_id) + .and_then(|output| pointer_get(output, &mapping.source_path)), + }; + if let Some(value) = value { + pointer_set(&mut input, &mapping.target_path, value.clone()); + } + } + input +} + +/// Builds the `006-runtime-request-execution` request for one proposal node. +/// Shared by the sequential P1 executor and the P2 wave executor. +pub(crate) fn build_node_execution_request( + canonical: &CanonicalProposal, + node: &ProposalNode, + node_id: &str, + input: Value, +) -> RuntimeRequest { + RuntimeRequest { + kind: "runtime_request".to_string(), + schema_version: "1.0.0".to_string(), + request_id: format!("{}-{node_id}", canonical.proposal.proposal_id), + intent: RuntimeIntent { + capability_id: Some(node.capability_id.clone()), + capability_version: Some(node.capability_version.clone()), + version_range: None, + intent_key: None, + }, + input, + lookup: RuntimeLookup { + scope: RuntimeLookupScope::PreferPrivate, + allow_ambiguity: false, + }, + context: RuntimeContext { + requested_target: PlacementTarget::Local, + correlation_id: Some(canonical.proposal.proposal_id.clone()), + caller: Some("workflow_proposal".to_string()), + traceparent: None, + tracestate: None, + metadata: None, + identity: None, + }, + governing_spec: "006-runtime-request-execution".to_string(), + } +} + +pub(crate) fn pointer_get<'a>(value: &'a Value, pointer: &str) -> Option<&'a Value> { value.pointer(pointer) } -fn pointer_set(target: &mut Value, pointer: &str, new_value: Value) { +pub(crate) fn pointer_set(target: &mut Value, pointer: &str, new_value: Value) { let segments: Vec<&str> = pointer.split('/').filter(|s| !s.is_empty()).collect(); *target = set_at_segments(std::mem::take(target), &segments, new_value); } diff --git a/docs/bounded-parallel-proposal-scheduling.md b/docs/bounded-parallel-proposal-scheduling.md new file mode 100644 index 00000000..f97aee4d --- /dev/null +++ b/docs/bounded-parallel-proposal-scheduling.md @@ -0,0 +1,136 @@ +# Bounded Parallel Proposal Scheduling (P2) + +Governed by spec [`110-bounded-parallel-workflow-scheduling`](../specs/110-bounded-parallel-workflow-scheduling/spec.md) +and [ADR-0042](adr/0042-phased-dynamic-orchestration-evolution.md). Tracks +issue `#1092`. + +P2 extends the [P1 sequential proposal lifecycle](workflow-proposal-lifecycle.md) +with deterministic, resource-bounded **concurrent** execution of independent +branches in the same proposal DAG. It reuses the exact same +`WorkflowProposal` wire format, canonicalization, and cross-validation as +P1 — P2 is a smarter, bounded *execution strategy* over an unchanged +document, not a new schema. + +## Where the code lives + +| Layer | Crate | What it owns | +|---|---|---| +| Wave levelization, fan-out/join-width/queue-depth bounds | `traverse-contracts::proposal` (`compute_parallel_schedule`) | Pure graph analysis over an already-canonicalized `CanonicalProposal` — no manifest/registry access, same portability guarantee as P1's structural validation. | +| `pure_read`-only authorization (FR-004a), concurrent execution, wall-time/payload bounds | `traverse-runtime::parallel_proposal` | Needs resolved capability contracts (for `effect_class`) and the `Runtime` execution engine, same split rationale as P1's `traverse-runtime::proposal`. | +| Public MCP tool surface | `traverse-mcp::tools::parallel_proposals` | Mirrors `tools::proposals`' plain-function pattern; reuses its authorization/quota helper (`authorize_and_reserve_quota`) so P1 and P2 share one FR-006/FR-006a/FR-007b implementation. | + +## Why the proposal document doesn't change + +P1's DAG model (`nodes`/`edges`/`mappings`) already supports arbitrary +fan-out and fan-in — `canonicalize_proposal`'s topological sort already +handles diamonds and wide fan-out deterministically. What P1's executor +does *not* do is run independent branches concurrently; it walks the total +order one node at a time. P2 adds a second analysis pass — +`compute_parallel_schedule` — that levelizes the same validated graph into +**waves**: each wave is the set of node ids whose dependencies are fully +satisfied by earlier waves, sorted lexicographically for determinism. A +linear proposal produces one node per wave (no observable behavior change); +a diamond (`a → {b, c} → d`) produces three waves, with `b` and `c` +eligible to run concurrently in the middle one. + +## Bounds (FR-001, FR-005: reject before doing any work) + +`ParallelScheduleLimits` (`traverse-contracts`), checked structurally +before any execution: + +| Bound | Meaning | +|---|---| +| `max_fan_out` | Max node ids ready to run concurrently in any single wave. | +| `max_join_width` | Max direct predecessors (in-degree) converging into any one node. | +| `max_queue_depth` | Max total node ids across the whole schedule. | +| `max_concurrent_nodes` | Max node executions the runtime dispatches at once within a wave (also the execution-time throttle — see below). | + +`ParallelExecutionLimits` (`traverse-runtime::parallel_proposal`), checked +during execution because they aren't expressible from graph shape alone: + +| Bound | Meaning | +|---|---| +| `max_wall_time` | Total wall-clock budget for the whole execution, checked **before starting each wave** — not preemptively mid-wave (see below). | +| `max_wave_payload_bytes` | Max total serialized JSON byte size of one wave's assembled node inputs — a bounded, honest proxy for a per-wave memory budget. | +| `max_concurrent_nodes` | Mirrors the schedule-level bound; batches a wave into chunks of this size before dispatch. | + +A structural bound violation is reported as `ParallelScheduleFailure` with a +stable `fan_out_exceeded` / `join_width_exceeded` / `queue_depth_exceeded` +code (FR-010 style, matching P1). An execution-time bound violation reports +`ProposalTerminalState::Cancelled`, with every un-dispatched node marked +`skipped_after_earlier_failure`. + +## `pure_read`-only concurrency (FR-004a) + +The first P2 implementation permits a wave with more than one member only +when **every** member's declared `effect_class` is `pure_read`. +`enforce_pure_read_only_parallelism` checks this after cross-validation +resolves each node's capability contract, denying with +`ConcurrentSideEffectDenied` otherwise. A non-`pure_read` node is still +allowed to execute — just never concurrently with a sibling; a singleton +wave (no concurrent sibling) is unaffected regardless of effect class. State +writes and external effects stay entirely sequential until a successor spec +proves declared independence, idempotency, cancellation, and budget +semantics (ADR-0042). + +## Execution and determinism (FR-002) + +`execute_parallel_proposal` dispatches each wave's nodes on real OS threads +via `std::thread::scope`, bounded to `max_concurrent_nodes` per batch. A +join may only consume declared, completed predecessor outputs — guaranteed +structurally, since a mapping's source can only be `initial_input` or a +node in the *same or earlier* wave by construction of the levelization +(nodes in the same wave have no edge between them, so no mapping between +them is structurally valid either). + +Real thread completion order is not deterministic, but the **observable +trace is**: outcomes are folded back in the wave's lexicographic order +(the same tie-break rule P1 uses for its sequential execution order), never +completion order. A test proposal where one branch sleeps and the other +doesn't demonstrates this: the trace always lists the fast branch before +the slow one, matching lexicographic order. + +## Cancellation semantics + +Rust has no safe way to preemptively interrupt an in-flight OS thread +without `unsafe`, and FR-004a already restricts concurrent work to +side-effect-free local reads. So a wave that is already dispatched is +always allowed to finish — including every node within it, even after a +sibling in the same batch fails. A wall-time or payload budget that is +already exhausted simply refuses to **start** the next wave; it never +attempts to interrupt a running one. This is checked and tested explicitly +(a slow first wave plus a tight time budget results in the first wave's +outcomes present and `Cancelled`, with the second wave's nodes marked +skipped). + +A panicking host executor is surfaced as a `Failed` outcome — never +silently dropped — matching this crate's fail-closed convention for +host-side faults (e.g. mutex-poisoning handling in `events/broker.rs`). + +## MCP tool surface + +`tools::parallel_proposals` exposes two functions, mirroring +`tools::proposals`' plain-function pattern (not wired into +`stdio_server.rs`, matching spec 015's precedent that pattern already set): + +- `compute_schedule_for_proposal` — validates a proposal (P1, unchanged) + and, if valid, computes and returns its concurrency waves plus automatic + eligibility. A structural, cross-validation, schedule-bound, or + concurrent-side-effect denial is a normal structured response. +- `execute_parallel_proposal_via_mcp` — the full P2 pipeline: validate, + compute schedule, authorize concurrency (FR-004a), decide or verify + authorization (FR-006/FR-006a, shared with P1 via + `authorize_and_reserve_quota`), reserve a quota slot (FR-007b), execute, + and return the trace. Every denial reason has its own stable code: + `invalid_proposal`, `invalid_parallel_schedule`, + `concurrent_side_effect_denied`, `approval_token_required`, a token error + code, or `quota_exhausted_`. + +`observe_proposal` from `tools::proposals` is reused as-is for P2 traces — +the trace shape is identical, so no new observation function exists. + +## Non-goals (spec 110's own "Out of scope") + +Dynamic expansion, loops, durable waits, retry/saga semantics, and implicit +parallelism inferred by a planner. These remain out of scope for every P2 +implementation, not just this first one.