feat(runtime): implement P2 bounded deterministic parallel proposal scheduling

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};

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<ParallelScheduleError>,
+}
+
+/// 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<Vec<String>>,
+}
+
+/// 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<ParallelSchedule, ParallelScheduleFailure> {
+    let mut errors = Vec::new();
+
+    let node_ids: BTreeSet<String> = 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<String, BTreeSet<String>> = BTreeMap::new();
+    let mut in_degree: BTreeMap<String, usize> =
+        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<String> = 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<String> = 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

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(())
+}

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;