Files
Genarrative/rust/crates/agent-runtime-orchestration/src/lib.rs
T
kdletters 202279c6d9 新增独立 Agent Runtime Rust 工作区
新增 Core、Engine、Runtime、SQLite、Provider、MCP、Skill、Codex、CLI 与 DAG crate

补齐 OpenAI endpoint 配置、Provider 实例/协议路由和统一工具权限边界

加入持久化、lease、checkpoint、reconciliation、审批恢复与消息历史回归

加入独立 workspace CI、依赖边界、能力集和 Fake Agent 测试脚本

同步建设计划、TODO、架构、测试与验收文档
2026-09-06 17:44:54 +08:00

4341 lines
156 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! 单 Agent Core/Engine 之上的最小 DAG 扩展。
//!
//! 本 crate 做图的校验、ready/wave 计算、受限提案,以及 Coordinator 控制面
//! (并发配额、消息去重和节点隔离);不拥有 Provider、线程、Runtime run 或业务
//! 完成真相。它只定义可替换的快照持久化端口和轻量 JSON/内存实现,Host 可以把
//! 每个 Delegation 映射成独立 Runtime run,用 Join 汇总结果,并在新的 epoch 中
//! 安装经过校验的 GraphProposal。
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::fs::{self, OpenOptions};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use agent_runtime_core::AgentCatalog;
use serde::{Deserialize, Deserializer, Serialize};
use thiserror::Error;
const MAX_ID_CHARS: usize = 128;
/// 默认候选图允许的最大任务数。
pub const DEFAULT_GRAPH_MAX_TASKS: usize = 128;
/// 默认候选图允许的最大依赖边数。
pub const DEFAULT_GRAPH_MAX_EDGES: usize = 512;
/// 默认候选图允许的最大层深(根任务深度为 1)。
pub const DEFAULT_GRAPH_MAX_DEPTH: usize = 32;
/// 默认单个前置任务允许的最大直接下游数。
pub const DEFAULT_GRAPH_MAX_OUT_DEGREE: usize = 32;
/// Coordinator 快照的稳定版本标识。
///
/// 快照只覆盖内存协调控制面;真正的 Runtime/TaskGraph 持久化仍由宿主负责。
pub const COORDINATOR_SNAPSHOT_SCHEMA_VERSION: &str = "agent-runtime-coordinator.v1";
/// TaskGraph + Coordinator 的 durable 快照版本标识。
pub const ORCHESTRATION_SNAPSHOT_SCHEMA_VERSION: &str = "agent-runtime-orchestration.v1";
/// Coordinator 快照的最大持久化大小。
///
/// 控制面状态不应成为无界日志;文件适配器在读取前先拒绝超出上限的输入,
/// 具体任务结果仍由各个 Runtime 自己保存。
pub const MAX_COORDINATOR_SNAPSHOT_BYTES: usize = 8 * 1024 * 1024;
static NEXT_SNAPSHOT_TEMP_ID: AtomicU64 = AtomicU64::new(1);
#[derive(Debug, Error, Clone, PartialEq, Eq)]
pub enum OrchestrationError {
#[error("无效输入: {0}")]
InvalidInput(String),
#[error("任务重复: {0}")]
DuplicateTask(String),
#[error("依赖无效: {0}")]
InvalidDependency(String),
#[error("图包含环")]
Cycle,
#[error("任务依赖未满足: {0}")]
DependencyBlocked(String),
#[error("未知任务: {0}")]
UnknownTask(String),
#[error("提案违反限制: {0}")]
Limit(String),
#[error("提案为空")]
EmptyProposal,
#[error("并发配额不足: scope={scope}, limit={limit}, active={active}, requested={requested}")]
QuotaExceeded {
scope: String,
limit: usize,
active: usize,
requested: usize,
},
#[error("运行重复: {0}")]
DuplicateRun(String),
#[error("未知运行: {0}")]
UnknownRun(String),
#[error("节点已隔离: {0}")]
NodeIsolated(String),
#[error("消息 ID 冲突: {0}")]
MessageConflict(String),
#[error("编排快照 revision 冲突: expected={expected:?}, actual={actual:?}")]
RevisionConflict {
expected: Option<u64>,
actual: Option<u64>,
},
}
fn validate_id(value: &str, label: &str) -> Result<(), OrchestrationError> {
if value.trim().is_empty() || value.chars().count() > MAX_ID_CHARS {
return Err(OrchestrationError::InvalidInput(format!(
"{label} 必须是非空且不超过 {MAX_ID_CHARS} 字符"
)));
}
if value.chars().any(char::is_control) {
return Err(OrchestrationError::InvalidInput(format!(
"{label} 不能包含控制字符"
)));
}
Ok(())
}
fn validate_isolation_reason(reason: &str) -> Result<(), OrchestrationError> {
if reason.trim().is_empty() {
return Err(OrchestrationError::InvalidInput(
"节点隔离原因不能为空".to_owned(),
));
}
if reason.chars().count() > MAX_MESSAGE_CHARS {
return Err(OrchestrationError::InvalidInput(format!(
"节点隔离原因不能超过 {MAX_MESSAGE_CHARS} 字符"
)));
}
Ok(())
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum TaskStatus {
Pending,
Running,
Waiting,
Completed,
Failed,
Cancelled,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct TaskNode {
pub id: String,
pub agent_id: String,
#[serde(default = "default_task_status")]
pub status: TaskStatus,
#[serde(default)]
pub dependencies: Vec<String>,
}
fn default_task_status() -> TaskStatus {
TaskStatus::Pending
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct TaskNodeInput {
id: String,
agent_id: String,
#[serde(default = "default_task_status")]
status: TaskStatus,
#[serde(default)]
dependencies: Vec<String>,
}
impl<'de> Deserialize<'de> for TaskNode {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let input = TaskNodeInput::deserialize(deserializer)?;
let node = Self {
id: input.id,
agent_id: input.agent_id,
status: input.status,
dependencies: input.dependencies,
};
node.validate().map_err(serde::de::Error::custom)?;
Ok(node)
}
}
impl TaskNode {
pub fn try_new(
id: impl Into<String>,
agent_id: impl Into<String>,
dependencies: impl IntoIterator<Item = impl Into<String>>,
) -> Result<Self, OrchestrationError> {
let node = Self {
id: id.into(),
agent_id: agent_id.into(),
status: TaskStatus::Pending,
dependencies: dependencies.into_iter().map(Into::into).collect(),
};
node.validate()?;
Ok(node)
}
fn validate(&self) -> Result<(), OrchestrationError> {
validate_id(&self.id, "task id")?;
validate_id(&self.agent_id, "agent id")?;
let mut seen = BTreeSet::new();
for dependency in &self.dependencies {
validate_id(dependency, "task dependency")?;
if dependency == &self.id {
return Err(OrchestrationError::InvalidDependency(format!(
"task {} 不能依赖自身",
self.id
)));
}
if !seen.insert(dependency) {
return Err(OrchestrationError::InvalidDependency(format!(
"task {} 重复依赖 {dependency}",
self.id
)));
}
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct TaskGraph {
pub goal: String,
pub tasks: Vec<TaskNode>,
#[serde(skip)]
index: BTreeMap<String, usize>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct TaskGraphInput {
goal: String,
tasks: Vec<TaskNode>,
}
impl<'de> Deserialize<'de> for TaskGraph {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let input = TaskGraphInput::deserialize(deserializer)?;
Self::try_new(input.goal, input.tasks).map_err(serde::de::Error::custom)
}
}
impl TaskGraph {
pub fn try_new(
goal: impl Into<String>,
tasks: impl IntoIterator<Item = TaskNode>,
) -> Result<Self, OrchestrationError> {
let goal = goal.into();
if goal.trim().is_empty() {
return Err(OrchestrationError::InvalidInput("goal 不能为空".to_owned()));
}
let tasks = tasks.into_iter().collect::<Vec<_>>();
if tasks.is_empty() {
return Err(OrchestrationError::InvalidInput(
"图至少需要一个任务".to_owned(),
));
}
let mut index = BTreeMap::new();
for (position, task) in tasks.iter().enumerate() {
task.validate()?;
if index.insert(task.id.clone(), position).is_some() {
return Err(OrchestrationError::DuplicateTask(task.id.clone()));
}
}
for task in &tasks {
for dependency in &task.dependencies {
if !index.contains_key(dependency) {
return Err(OrchestrationError::InvalidDependency(format!(
"{} 引用了未知任务 {dependency}",
task.id
)));
}
}
}
let graph = Self { goal, tasks, index };
graph.validate_acyclic()?;
Ok(graph)
}
pub fn task(&self, id: &str) -> Option<&TaskNode> {
self.index
.get(id)
.and_then(|position| self.tasks.get(*position))
}
/// 返回当前图中的任务数量。
pub fn task_count(&self) -> usize {
self.tasks.len()
}
/// `node_count` 是面向图调用方的同义入口。
pub fn node_count(&self) -> usize {
self.task_count()
}
/// 返回当前图中的依赖边数量。
pub fn edge_count(&self) -> usize {
self.tasks.iter().map(|task| task.dependencies.len()).sum()
}
/// 返回图的最长依赖层数;无依赖的根任务深度为 1。
pub fn depth(&self) -> usize {
if self.tasks.is_empty() {
return 0;
}
let mut indegree = self
.tasks
.iter()
.map(|task| (task.id.clone(), task.dependencies.len()))
.collect::<BTreeMap<_, _>>();
let mut layers = self
.tasks
.iter()
.map(|task| (task.id.clone(), 1usize))
.collect::<BTreeMap<_, _>>();
let mut queue = VecDeque::from_iter(
indegree
.iter()
.filter(|(_, degree)| **degree == 0)
.map(|(id, _)| id.clone()),
);
let mut max_depth = 1;
while let Some(id) = queue.pop_front() {
let current_depth = layers.get(&id).copied().unwrap_or(1);
max_depth = max_depth.max(current_depth);
for task in &self.tasks {
if !task.dependencies.iter().any(|dependency| dependency == &id) {
continue;
}
let next_depth = current_depth.saturating_add(1);
let layer = layers
.get_mut(&task.id)
.expect("task layers are initialized from the graph");
*layer = (*layer).max(next_depth);
let degree = indegree
.get_mut(&task.id)
.expect("task indegree is initialized from the graph");
*degree -= 1;
if *degree == 0 {
queue.push_back(task.id.clone());
}
}
}
max_depth
}
/// 返回一个任务的直接下游数量。
pub fn fan_out(&self, task_id: &str) -> Option<usize> {
self.task(task_id)?;
Some(
self.tasks
.iter()
.filter(|task| task.dependencies.iter().any(|id| id == task_id))
.count(),
)
}
/// `out_degree` 是 `fan_out` 的图术语别名。
pub fn out_degree(&self, task_id: &str) -> Option<usize> {
self.fan_out(task_id)
}
/// 校验图中每个任务都绑定到已注册的 Agent。
pub fn validate_agents(&self, catalog: &AgentCatalog) -> Result<(), OrchestrationError> {
for task in &self.tasks {
if catalog.get(&task.agent_id).is_none() {
return Err(OrchestrationError::InvalidInput(format!(
"task {} 引用了未注册 Agent:{}",
task.id, task.agent_id
)));
}
}
Ok(())
}
/// 按给定目录和资源限制,以不可变方式应用一个增量图提案。
///
/// 结构化的 [`GraphProposal::apply`] 仍保留为兼容入口;需要接收不可信
/// planner 输入的宿主应使用本方法,确保完整候选图(而非仅 delta)通过
/// 任务数、边数、层深、扇出和 Agent 目录校验。
pub fn apply_proposal(
&self,
proposal: &GraphProposal,
catalog: &AgentCatalog,
limits: &GraphLimits,
) -> Result<Self, OrchestrationError> {
self.expand_with_proposal(proposal, catalog, limits)
.map(GraphExpansion::into_graph)
}
/// 返回完整候选图及本次提案的已校验增量。
pub fn expand_with_proposal(
&self,
proposal: &GraphProposal,
catalog: &AgentCatalog,
limits: &GraphLimits,
) -> Result<GraphExpansion, OrchestrationError> {
limits.validate()?;
proposal.validate()?;
self.validate_agents(catalog)?;
let resulting_task_count = self
.task_count()
.checked_add(proposal.nodes.len())
.ok_or_else(|| OrchestrationError::Limit("proposal task 数量计算溢出".to_owned()))?;
if self.task_count() > limits.max_tasks {
return Err(OrchestrationError::Limit(format!(
"现有 task 数量 {} 已超过 maxTasks {}",
self.task_count(),
limits.max_tasks
)));
}
if resulting_task_count > limits.max_tasks {
return Err(OrchestrationError::Limit(format!(
"扩图后 task 数量 {} 超过 maxTasks {}",
resulting_task_count, limits.max_tasks
)));
}
for node in &proposal.nodes {
if self.task(&node.id).is_some() {
return Err(OrchestrationError::DuplicateTask(node.id.clone()));
}
if catalog.get(&node.agent_id).is_none() {
return Err(OrchestrationError::InvalidInput(format!(
"proposal task {} 引用了未注册 Agent:{}",
node.id, node.agent_id
)));
}
}
let resulting_edge_count = self
.edge_count()
.checked_add(proposal.edges.len())
.ok_or_else(|| OrchestrationError::Limit("proposal edge 数量计算溢出".to_owned()))?;
if resulting_edge_count > limits.max_edges {
return Err(OrchestrationError::Limit(format!(
"扩图后 dependency edge 数量 {} 超过 maxEdges {}",
resulting_edge_count, limits.max_edges
)));
}
let candidate = proposal.apply(self)?;
if candidate.depth() > limits.max_depth {
return Err(OrchestrationError::Limit(format!(
"扩图后 graph depth {} 超过 maxDepth {}",
candidate.depth(),
limits.max_depth
)));
}
if let Some((task_id, degree)) = candidate
.tasks
.iter()
.filter_map(|task| candidate.fan_out(&task.id).map(|degree| (&task.id, degree)))
.find(|(_, degree)| *degree > limits.max_out_degree)
{
return Err(OrchestrationError::Limit(format!(
"task {task_id} 的 fan-out {degree} 超过 maxOutDegree {}",
limits.max_out_degree
)));
}
Ok(GraphExpansion {
graph: candidate,
added_task_ids: proposal.nodes.iter().map(|node| node.id.clone()).collect(),
added_edges: proposal.edges.clone(),
})
}
/// 参数顺序与部分宿主配置代码兼容的别名。
pub fn apply_proposal_with_limits(
&self,
proposal: &GraphProposal,
limits: &GraphLimits,
catalog: &AgentCatalog,
) -> Result<Self, OrchestrationError> {
self.apply_proposal(proposal, catalog, limits)
}
/// `expand` 是面向宿主的简短别名。
pub fn expand(
&self,
proposal: &GraphProposal,
catalog: &AgentCatalog,
limits: &GraphLimits,
) -> Result<Self, OrchestrationError> {
self.apply_proposal(proposal, catalog, limits)
}
/// 更新一个任务的状态。
///
/// 只有启动任务需要检查依赖;终态和取消状态由宿主根据自己的运行结果
/// 写入。这样 orchestration 不会复制单 Agent reducer 的完整状态机。
pub fn set_task_status(
&mut self,
id: impl AsRef<str>,
status: TaskStatus,
) -> Result<(), OrchestrationError> {
let id = id.as_ref();
let position = *self
.index
.get(id)
.ok_or_else(|| OrchestrationError::UnknownTask(id.to_owned()))?;
if status == TaskStatus::Running {
let task = self.tasks.get(position).expect("任务索引由 try_new 建立");
if let Some(dependency) = task.dependencies.iter().find(|dependency| {
self.task(dependency.as_str())
.is_none_or(|task| task.status != TaskStatus::Completed)
}) {
return Err(OrchestrationError::DependencyBlocked(format!(
"{id} 依赖 {dependency} 尚未完成"
)));
}
}
self.tasks[position].status = status;
Ok(())
}
/// 以不可变视角创建状态更新后的新 epoch,便于调用方在并发协调时避免
/// 共享可变图;原图不会被修改。
pub fn with_task_status(
&self,
id: impl AsRef<str>,
status: TaskStatus,
) -> Result<Self, OrchestrationError> {
let mut next = self.clone();
next.set_task_status(id, status)?;
Ok(next)
}
/// 将种子任务及其全部下游任务重置为 Pending,返回稳定顺序的受影响任务。
///
/// 这是显式修复入口,不会自动重跑任何 Runtime run;宿主仍需按返回值重新
/// 创建或排队对应运行。
pub fn repair_downstream(
&mut self,
seeds: &[impl AsRef<str>],
) -> Result<Vec<String>, OrchestrationError> {
let impacted = self.expand_downstream(seeds)?;
for id in &impacted {
let position = self
.index
.get(id)
.copied()
.expect("expand_downstream 已验证任务存在");
self.tasks[position].status = TaskStatus::Pending;
}
Ok(impacted)
}
/// 不修改原图地执行一次下游修复。
pub fn repaired_downstream(
&self,
seeds: &[impl AsRef<str>],
) -> Result<(Self, Vec<String>), OrchestrationError> {
let mut next = self.clone();
let impacted = next.repair_downstream(seeds)?;
Ok((next, impacted))
}
pub fn ready_task_ids(&self) -> Vec<&str> {
let completed = self
.tasks
.iter()
.filter(|task| task.status == TaskStatus::Completed)
.map(|task| task.id.as_str())
.collect::<BTreeSet<_>>();
self.tasks
.iter()
.filter(|task| {
task.status == TaskStatus::Pending
&& task
.dependencies
.iter()
.all(|dependency| completed.contains(dependency.as_str()))
})
.map(|task| task.id.as_str())
.collect()
}
/// 根据当前图生成依赖波次;同一波次内的任务互不依赖,可由 Host 并发运行。
pub fn dependency_waves(&self) -> Result<Vec<Vec<String>>, OrchestrationError> {
let mut done = self
.tasks
.iter()
.filter(|task| task.status == TaskStatus::Completed)
.map(|task| task.id.clone())
.collect::<BTreeSet<_>>();
let mut pending = self
.tasks
.iter()
.filter(|task| task.status == TaskStatus::Pending)
.map(|task| task.id.clone())
.collect::<BTreeSet<_>>();
let mut waves = Vec::new();
while !pending.is_empty() {
let wave = pending
.iter()
.filter(|id| {
self.task(id).is_some_and(|task| {
task.dependencies
.iter()
.all(|dependency| done.contains(dependency))
})
})
.cloned()
.collect::<Vec<_>>();
if wave.is_empty() {
return Err(OrchestrationError::DependencyBlocked(
pending.into_iter().collect::<Vec<_>>().join(", "),
));
}
for id in &wave {
pending.remove(id);
done.insert(id.clone());
}
waves.push(wave);
}
Ok(waves)
}
pub fn expand_downstream(
&self,
seeds: &[impl AsRef<str>],
) -> Result<Vec<String>, OrchestrationError> {
let mut impacted = BTreeSet::new();
for seed in seeds {
let id = seed.as_ref();
if self.task(id).is_none() {
return Err(OrchestrationError::UnknownTask(id.to_owned()));
}
impacted.insert(id.to_owned());
}
let mut changed = true;
while changed {
changed = false;
for task in &self.tasks {
if !impacted.contains(&task.id)
&& task
.dependencies
.iter()
.any(|dependency| impacted.contains(dependency))
{
impacted.insert(task.id.clone());
changed = true;
}
}
}
Ok(self
.tasks
.iter()
.filter(|task| impacted.contains(&task.id))
.map(|task| task.id.clone())
.collect())
}
fn validate_acyclic(&self) -> Result<(), OrchestrationError> {
let mut indegree = self
.tasks
.iter()
.map(|task| (task.id.clone(), task.dependencies.len()))
.collect::<BTreeMap<_, _>>();
let mut queue = VecDeque::from_iter(
indegree
.iter()
.filter(|(_, degree)| **degree == 0)
.map(|(id, _)| id.clone()),
);
let mut visited = 0;
while let Some(id) = queue.pop_front() {
visited += 1;
for task in &self.tasks {
if task.dependencies.iter().any(|dependency| dependency == &id) {
let degree = indegree.get_mut(&task.id).expect("task index validated");
*degree -= 1;
if *degree == 0 {
queue.push_back(task.id.clone());
}
}
}
}
if visited == self.tasks.len() {
Ok(())
} else {
Err(OrchestrationError::Cycle)
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct Delegation {
pub id: String,
pub parent_run_id: String,
pub child_agent_id: String,
pub task: String,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct DelegationInput {
id: String,
parent_run_id: String,
child_agent_id: String,
task: String,
}
impl<'de> Deserialize<'de> for Delegation {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let input = DelegationInput::deserialize(deserializer)?;
Self::try_new(
input.id,
input.parent_run_id,
input.child_agent_id,
input.task,
)
.map_err(serde::de::Error::custom)
}
}
impl Delegation {
pub fn try_new(
id: impl Into<String>,
parent_run_id: impl Into<String>,
child_agent_id: impl Into<String>,
task: impl Into<String>,
) -> Result<Self, OrchestrationError> {
let value = Self {
id: id.into(),
parent_run_id: parent_run_id.into(),
child_agent_id: child_agent_id.into(),
task: task.into(),
};
value.validate()?;
Ok(value)
}
fn validate(&self) -> Result<(), OrchestrationError> {
validate_id(&self.id, "delegation id")?;
validate_id(&self.parent_run_id, "parent run id")?;
validate_id(&self.child_agent_id, "child agent id")?;
if self.task.trim().is_empty() {
return Err(OrchestrationError::InvalidInput(
"delegation task 不能为空".to_owned(),
));
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct Join {
pub id: String,
pub delegation_ids: Vec<String>,
pub strategy: String,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct JoinInput {
id: String,
delegation_ids: Vec<String>,
strategy: String,
}
impl<'de> Deserialize<'de> for Join {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let input = JoinInput::deserialize(deserializer)?;
Self::try_new(input.id, input.delegation_ids, input.strategy)
.map_err(serde::de::Error::custom)
}
}
impl Join {
pub fn try_new(
id: impl Into<String>,
delegation_ids: impl IntoIterator<Item = impl Into<String>>,
strategy: impl Into<String>,
) -> Result<Self, OrchestrationError> {
let id = id.into();
let strategy = strategy.into();
validate_id(&id, "join id")?;
validate_id(&strategy, "join strategy")?;
let delegation_ids = delegation_ids
.into_iter()
.map(Into::into)
.collect::<Vec<_>>();
if delegation_ids.is_empty() {
return Err(OrchestrationError::InvalidInput(
"join 至少需要一个 delegation".to_owned(),
));
}
let join = Self {
id,
delegation_ids,
strategy,
};
join.validate()?;
Ok(join)
}
fn validate(&self) -> Result<(), OrchestrationError> {
validate_id(&self.id, "join id")?;
validate_id(&self.strategy, "join strategy")?;
if self.delegation_ids.is_empty() {
return Err(OrchestrationError::InvalidInput(
"join 至少需要一个 delegation".to_owned(),
));
}
let mut seen = BTreeSet::new();
for delegation_id in &self.delegation_ids {
validate_id(delegation_id, "join delegation id")?;
if !seen.insert(delegation_id) {
return Err(OrchestrationError::InvalidInput(format!(
"join 重复 delegation {delegation_id}"
)));
}
}
Ok(())
}
}
/// 候选任务图的资源边界。
///
/// 限制始终针对提案应用后的完整候选图计算;根任务的层深为 1,
/// `max_out_degree` 统计一个前置任务的直接下游数量。
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct GraphLimits {
#[serde(alias = "maxNodes")]
pub max_tasks: usize,
pub max_edges: usize,
pub max_depth: usize,
#[serde(alias = "maxFanOut")]
pub max_out_degree: usize,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct GraphLimitsInput {
#[serde(alias = "maxNodes")]
max_tasks: usize,
max_edges: usize,
max_depth: usize,
#[serde(alias = "maxFanOut")]
max_out_degree: usize,
}
impl<'de> Deserialize<'de> for GraphLimits {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let input = GraphLimitsInput::deserialize(deserializer)?;
Self::try_new(
input.max_tasks,
input.max_edges,
input.max_depth,
input.max_out_degree,
)
.map_err(serde::de::Error::custom)
}
}
impl Default for GraphLimits {
fn default() -> Self {
Self {
max_tasks: DEFAULT_GRAPH_MAX_TASKS,
max_edges: DEFAULT_GRAPH_MAX_EDGES,
max_depth: DEFAULT_GRAPH_MAX_DEPTH,
max_out_degree: DEFAULT_GRAPH_MAX_OUT_DEGREE,
}
}
}
impl GraphLimits {
pub const fn new(
max_tasks: usize,
max_edges: usize,
max_depth: usize,
max_out_degree: usize,
) -> Self {
Self {
max_tasks,
max_edges,
max_depth,
max_out_degree,
}
}
pub fn try_new(
max_tasks: usize,
max_edges: usize,
max_depth: usize,
max_out_degree: usize,
) -> Result<Self, OrchestrationError> {
let limits = Self::new(max_tasks, max_edges, max_depth, max_out_degree);
limits.validate().map(|()| limits)
}
pub fn with_max_tasks(mut self, value: usize) -> Self {
self.max_tasks = value;
self
}
pub fn with_max_nodes(self, value: usize) -> Self {
self.with_max_tasks(value)
}
pub fn with_max_edges(mut self, value: usize) -> Self {
self.max_edges = value;
self
}
pub fn with_max_depth(mut self, value: usize) -> Self {
self.max_depth = value;
self
}
pub fn with_max_out_degree(mut self, value: usize) -> Self {
self.max_out_degree = value;
self
}
pub fn with_max_fan_out(self, value: usize) -> Self {
self.with_max_out_degree(value)
}
pub fn max_nodes(&self) -> usize {
self.max_tasks
}
pub fn max_fan_out(&self) -> usize {
self.max_out_degree
}
pub fn validate(&self) -> Result<(), OrchestrationError> {
let invalid = [
(self.max_tasks, "maxTasks"),
(self.max_edges, "maxEdges"),
(self.max_depth, "maxDepth"),
(self.max_out_degree, "maxOutDegree"),
]
.into_iter()
.find(|(value, _)| *value == 0);
if let Some((_, field)) = invalid {
return Err(OrchestrationError::Limit(format!(
"graph limits 的 {field} 必须大于 0"
)));
}
Ok(())
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct GraphProposal {
pub nodes: Vec<TaskNodeProposal>,
pub edges: Vec<GraphEdge>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct GraphProposalInput {
nodes: Vec<TaskNodeProposal>,
edges: Vec<GraphEdge>,
}
impl<'de> Deserialize<'de> for GraphProposal {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let input = GraphProposalInput::deserialize(deserializer)?;
Self::try_new(input.nodes, input.edges).map_err(serde::de::Error::custom)
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct TaskNodeProposal {
pub id: String,
pub agent_id: String,
}
/// Planner-facing name for a proposed task.
///
/// `TaskNodeProposal` remains the historical public name used by this crate;
/// the alias keeps both names source-compatible while the proposal contract is
/// shared with hosts that call the value `TaskProposal`.
pub type TaskProposal = TaskNodeProposal;
impl TaskNodeProposal {
pub fn try_new(
id: impl Into<String>,
agent_id: impl Into<String>,
) -> Result<Self, OrchestrationError> {
let value = Self {
id: id.into(),
agent_id: agent_id.into(),
};
validate_id(&value.id, "proposal task id")?;
validate_id(&value.agent_id, "proposal task agent id")?;
Ok(value)
}
pub fn new(
id: impl Into<String>,
agent_id: impl Into<String>,
) -> Result<Self, OrchestrationError> {
Self::try_new(id, agent_id)
}
pub fn id(&self) -> &str {
&self.id
}
pub fn agent_id(&self) -> &str {
&self.agent_id
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct GraphEdge {
pub from: String,
pub to: String,
}
impl GraphEdge {
pub fn try_new(
from: impl Into<String>,
to: impl Into<String>,
) -> Result<Self, OrchestrationError> {
let value = Self {
from: from.into(),
to: to.into(),
};
validate_id(&value.from, "proposal edge from")?;
validate_id(&value.to, "proposal edge to")?;
if value.from == value.to {
return Err(OrchestrationError::InvalidDependency(
"proposal 自环".to_owned(),
));
}
Ok(value)
}
pub fn new(from: impl Into<String>, to: impl Into<String>) -> Result<Self, OrchestrationError> {
Self::try_new(from, to)
}
pub fn from(&self) -> &str {
&self.from
}
pub fn to(&self) -> &str {
&self.to
}
}
impl GraphProposal {
pub fn try_new(
nodes: impl IntoIterator<Item = TaskNodeProposal>,
edges: impl IntoIterator<Item = GraphEdge>,
) -> Result<Self, OrchestrationError> {
let proposal = Self {
nodes: nodes.into_iter().collect(),
edges: edges.into_iter().collect(),
};
proposal.validate()?;
Ok(proposal)
}
fn validate(&self) -> Result<(), OrchestrationError> {
if self.nodes.is_empty() && self.edges.is_empty() {
return Err(OrchestrationError::EmptyProposal);
}
let mut ids = BTreeSet::new();
for node in &self.nodes {
validate_id(&node.id, "proposal task id")?;
validate_id(&node.agent_id, "proposal agent id")?;
if !ids.insert(&node.id) {
return Err(OrchestrationError::DuplicateTask(node.id.clone()));
}
}
let mut edges = BTreeSet::new();
for edge in &self.edges {
validate_id(&edge.from, "proposal edge from")?;
validate_id(&edge.to, "proposal edge to")?;
if edge.from == edge.to {
return Err(OrchestrationError::InvalidDependency(
"proposal 自环".to_owned(),
));
}
if !edges.insert((&edge.from, &edge.to)) {
return Err(OrchestrationError::InvalidDependency(format!(
"proposal 重复边: {} -> {}",
edge.from, edge.to
)));
}
}
Ok(())
}
/// 将新增节点/边应用到旧图;现有节点不会被原地改写,调用方可把结果视为新 epoch。
pub fn apply(&self, base: &TaskGraph) -> Result<TaskGraph, OrchestrationError> {
self.validate()?;
let mut tasks = base.tasks.clone();
let mut by_id = tasks
.iter()
.map(|task| (task.id.clone(), task.dependencies.clone()))
.collect::<BTreeMap<_, _>>();
for node in &self.nodes {
if by_id.contains_key(&node.id) {
return Err(OrchestrationError::DuplicateTask(node.id.clone()));
}
tasks.push(TaskNode::try_new(
&node.id,
&node.agent_id,
std::iter::empty::<String>(),
)?);
by_id.insert(node.id.clone(), Vec::new());
}
for edge in &self.edges {
if !by_id.contains_key(&edge.from) || !by_id.contains_key(&edge.to) {
return Err(OrchestrationError::UnknownTask(format!(
"{} -> {}",
edge.from, edge.to
)));
}
if base.task(&edge.to).is_some() {
return Err(OrchestrationError::InvalidDependency(format!(
"不能在新 epoch 外原地修改既有任务依赖: {} -> {}",
edge.from, edge.to
)));
}
let dependencies = by_id.get_mut(&edge.to).expect("checked above");
if !dependencies.contains(&edge.from) {
dependencies.push(edge.from.clone());
}
}
for task in &mut tasks {
if let Some(dependencies) = by_id.get(&task.id) {
task.dependencies = dependencies.clone();
}
}
TaskGraph::try_new(base.goal.clone(), tasks)
}
/// 使用 Agent 目录和图资源限制应用提案。
pub fn apply_with_limits(
&self,
base: &TaskGraph,
catalog: &AgentCatalog,
limits: &GraphLimits,
) -> Result<TaskGraph, OrchestrationError> {
base.apply_proposal(self, catalog, limits)
}
/// 参数顺序与配置驱动宿主兼容的别名。
pub fn apply_with_graph_limits(
&self,
base: &TaskGraph,
limits: &GraphLimits,
catalog: &AgentCatalog,
) -> Result<TaskGraph, OrchestrationError> {
base.apply_proposal(self, catalog, limits)
}
}
/// 经过完整校验的候选图及其新增节点/边摘要。
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct GraphExpansion {
graph: TaskGraph,
added_task_ids: Vec<String>,
added_edges: Vec<GraphEdge>,
}
/// `AppliedGraphProposal` 是给宿主持久化层使用的描述性别名。
pub type AppliedGraphProposal = GraphExpansion;
impl GraphExpansion {
pub fn graph(&self) -> &TaskGraph {
&self.graph
}
pub fn into_graph(self) -> TaskGraph {
self.graph
}
pub fn added_task_ids(&self) -> &[String] {
&self.added_task_ids
}
pub fn added_edges(&self) -> &[GraphEdge] {
&self.added_edges
}
}
const MAX_MESSAGE_CHARS: usize = 16 * 1024;
/// 多运行协调器的有界并发配置。
///
/// 该配置只约束协调器持有的活动 run 数,不替代 Runtime reducer,也不决定
/// worker 如何执行。真正的线程/进程调度仍由 Host 或上层执行器负责。
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct CoordinatorQuota {
max_active_runs: usize,
max_active_runs_per_agent: usize,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct CoordinatorQuotaInput {
max_active_runs: usize,
max_active_runs_per_agent: usize,
}
impl<'de> Deserialize<'de> for CoordinatorQuota {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let input = CoordinatorQuotaInput::deserialize(deserializer)?;
Self::try_new(input.max_active_runs, input.max_active_runs_per_agent)
.map_err(serde::de::Error::custom)
}
}
impl CoordinatorQuota {
pub fn try_new(
max_active_runs: usize,
max_active_runs_per_agent: usize,
) -> Result<Self, OrchestrationError> {
if max_active_runs == 0 || max_active_runs_per_agent == 0 {
return Err(OrchestrationError::InvalidInput(
"coordinator 并发配额必须大于零".to_owned(),
));
}
Ok(Self {
max_active_runs,
max_active_runs_per_agent,
})
}
pub fn max_active_runs(self) -> usize {
self.max_active_runs
}
pub fn max_active_runs_per_agent(self) -> usize {
self.max_active_runs_per_agent
}
}
impl Default for CoordinatorQuota {
fn default() -> Self {
Self {
max_active_runs: 16,
max_active_runs_per_agent: 4,
}
}
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct CoordinatorRunRequest {
pub run_id: String,
pub task_id: String,
pub agent_id: String,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct CoordinatorRunRequestInput {
run_id: String,
task_id: String,
agent_id: String,
}
impl<'de> Deserialize<'de> for CoordinatorRunRequest {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let input = CoordinatorRunRequestInput::deserialize(deserializer)?;
Self::try_new(input.run_id, input.task_id, input.agent_id).map_err(serde::de::Error::custom)
}
}
impl CoordinatorRunRequest {
pub fn try_new(
run_id: impl Into<String>,
task_id: impl Into<String>,
agent_id: impl Into<String>,
) -> Result<Self, OrchestrationError> {
let request = Self {
run_id: run_id.into(),
task_id: task_id.into(),
agent_id: agent_id.into(),
};
request.validate()?;
Ok(request)
}
fn validate(&self) -> Result<(), OrchestrationError> {
validate_id(&self.run_id, "coordinator run id")?;
validate_id(&self.task_id, "coordinator task id")?;
validate_id(&self.agent_id, "coordinator agent id")?;
Ok(())
}
}
/// 一个由宿主消费的 ready-task 计划项。
///
/// 计划只携带任务与其声明的 Agent,不生成 Runtime run ID;run ID 仍由宿主
/// 负责分配。这样 orchestration 不需要引入 UUID、Runtime 或业务命名规则。
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ReadyTaskCandidate {
pub task_id: String,
pub agent_id: String,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct ReadyTaskCandidateInput {
task_id: String,
agent_id: String,
}
impl<'de> Deserialize<'de> for ReadyTaskCandidate {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let input = ReadyTaskCandidateInput::deserialize(deserializer)?;
Self::try_new(input.task_id, input.agent_id).map_err(serde::de::Error::custom)
}
}
impl ReadyTaskCandidate {
pub fn try_new(
task_id: impl Into<String>,
agent_id: impl Into<String>,
) -> Result<Self, OrchestrationError> {
let candidate = Self {
task_id: task_id.into(),
agent_id: agent_id.into(),
};
candidate.validate()?;
Ok(candidate)
}
fn validate(&self) -> Result<(), OrchestrationError> {
validate_id(&self.task_id, "ready task id")?;
validate_id(&self.agent_id, "ready agent id")?;
Ok(())
}
}
/// 基于某个编排 revision 生成的 ready-task 派发计划。
///
/// 计划本身不占用任务,也不写 Store。宿主拿到计划后调用
/// [`PersistentCoordinator::claim_ready_tasks`],该方法会再次检查 revision、
/// 任务状态、Agent 归属、隔离和配额,并通过现有 `dispatch_wave` 一次性提交。
/// `candidates` 可以为空,表示当前 revision 没有可派发任务;对空计划的 claim
/// 是无写入的幂等操作。
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ReadyTaskDispatchPlan {
pub expected_revision: u64,
pub candidates: Vec<ReadyTaskCandidate>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct ReadyTaskDispatchPlanInput {
expected_revision: u64,
candidates: Vec<ReadyTaskCandidate>,
}
impl<'de> Deserialize<'de> for ReadyTaskDispatchPlan {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let input = ReadyTaskDispatchPlanInput::deserialize(deserializer)?;
Self::try_new(input.expected_revision, input.candidates).map_err(serde::de::Error::custom)
}
}
impl ReadyTaskDispatchPlan {
pub fn try_new(
expected_revision: u64,
candidates: impl IntoIterator<Item = ReadyTaskCandidate>,
) -> Result<Self, OrchestrationError> {
let plan = Self {
expected_revision,
candidates: candidates.into_iter().collect(),
};
plan.validate()?;
Ok(plan)
}
fn validate(&self) -> Result<(), OrchestrationError> {
let mut seen_tasks = BTreeSet::new();
for candidate in &self.candidates {
candidate.validate()?;
if !seen_tasks.insert(candidate.task_id.as_str()) {
return Err(OrchestrationError::DuplicateTask(candidate.task_id.clone()));
}
}
Ok(())
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct CoordinatorRun {
pub run_id: String,
pub task_id: String,
pub agent_id: String,
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct CoordinatorMessage {
pub message_id: String,
pub from_node: String,
pub to_node: String,
pub payload: serde_json::Value,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
struct CoordinatorMessageInput {
message_id: String,
from_node: String,
to_node: String,
payload: serde_json::Value,
}
impl<'de> Deserialize<'de> for CoordinatorMessage {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let input = CoordinatorMessageInput::deserialize(deserializer)?;
Self::try_new(
input.message_id,
input.from_node,
input.to_node,
input.payload,
)
.map_err(serde::de::Error::custom)
}
}
impl CoordinatorMessage {
pub fn try_new(
message_id: impl Into<String>,
from_node: impl Into<String>,
to_node: impl Into<String>,
payload: impl Into<serde_json::Value>,
) -> Result<Self, OrchestrationError> {
let message = Self {
message_id: message_id.into(),
from_node: from_node.into(),
to_node: to_node.into(),
payload: payload.into(),
};
message.validate()?;
Ok(message)
}
/// 文本消息的便捷构造;结构化消息可直接传入 `serde_json::Value`。
pub fn try_text(
message_id: impl Into<String>,
from_node: impl Into<String>,
to_node: impl Into<String>,
payload: impl Into<String>,
) -> Result<Self, OrchestrationError> {
Self::try_new(
message_id,
from_node,
to_node,
serde_json::Value::String(payload.into()),
)
}
fn validate(&self) -> Result<(), OrchestrationError> {
validate_id(&self.message_id, "coordinator message id")?;
validate_id(&self.from_node, "coordinator message source")?;
validate_id(&self.to_node, "coordinator message target")?;
let encoded = serde_json::to_string(&self.payload).map_err(|error| {
OrchestrationError::InvalidInput(format!("coordinator 消息 payload 无法编码: {error}"))
})?;
if encoded.chars().count() > MAX_MESSAGE_CHARS {
return Err(OrchestrationError::InvalidInput(format!(
"coordinator 消息 payload 不能超过 {MAX_MESSAGE_CHARS} 字符"
)));
}
Ok(())
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum MessageDelivery {
Delivered,
Duplicate,
}
/// Coordinator 的可序列化状态边界。
///
/// `active_tasks`、`active_by_agent` 等索引是运行时派生数据,不写入快照;导入
/// 时会从 `active_runs` 重新建立。收件箱保留消息 ID 顺序,`messages` 则保留
/// 去重账本,因此已经消费过的消息仍能在恢复后继续被识别为重复投递。
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct CoordinatorSnapshot {
pub schema_version: String,
pub quota: CoordinatorQuota,
pub active_runs: Vec<CoordinatorRun>,
pub isolated_nodes: BTreeMap<String, String>,
pub messages: Vec<CoordinatorMessage>,
pub mailboxes: BTreeMap<String, Vec<String>>,
}
/// Coordinator 控制面的可替换持久化端口。
///
/// 该端口只承载经过校验的 [`CoordinatorSnapshot`],不把 SQLite、线程或具体
/// Runtime 生命周期带进编排 crate。宿主可以实现自己的数据库/对象存储适配器,
/// 或直接使用下面的内存、JSON 文件实现。
pub trait CoordinatorStore: Send + Sync {
/// 读取当前快照;没有持久化状态时返回 `Ok(None)`。
fn load_snapshot(&self) -> Result<Option<CoordinatorSnapshot>, OrchestrationError>;
/// 原子保存一个完整快照。实现方不得把半成品快照暴露给读取者。
fn save_snapshot(&self, snapshot: &CoordinatorSnapshot) -> Result<(), OrchestrationError>;
}
/// TaskGraph 与 Coordinator 控制面一起保存的 durable 快照。
///
/// 旧的 [`CoordinatorStore`] 只保存活动 run、消息账本和隔离标记,无法在
/// 进程重启后恢复任务图本身。这个合同把两者放进同一个有版本的文档,供
/// `PersistentCoordinator` 以一次原子写入更新。它仍然只保存编排控制面,
/// 不保存单 Agent Runtime 的消息/结果真相。
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct OrchestrationSnapshot {
pub schema_version: String,
/// 单调递增的控制面版本;首次创建使用 0。
pub revision: u64,
pub graph: TaskGraph,
pub coordinator: CoordinatorSnapshot,
}
/// `PersistentCoordinator` 使用的持久化端口。
///
/// `expected_revision` 是 CAS 保护:`None` 只允许创建新快照,`Some(n)` 只
/// 允许覆盖当前 revision 为 `n` 的快照。实现方必须保证一次写入要么看到
/// 完整新快照,要么继续看到旧快照,不能暴露半个 JSON 文档。
pub trait OrchestrationSnapshotStore: Send + Sync {
fn load_snapshot(&self) -> Result<Option<OrchestrationSnapshot>, OrchestrationError>;
fn save_snapshot(
&self,
expected_revision: Option<u64>,
snapshot: &OrchestrationSnapshot,
) -> Result<(), OrchestrationError>;
}
fn validate_orchestration_snapshot(
snapshot: &OrchestrationSnapshot,
) -> Result<(), OrchestrationError> {
if snapshot.schema_version != ORCHESTRATION_SNAPSHOT_SCHEMA_VERSION {
return Err(OrchestrationError::InvalidInput(format!(
"不支持的 orchestration 快照版本: {}",
snapshot.schema_version
)));
}
Coordinator::from_snapshot(snapshot.coordinator.clone())?;
// 活动 run 必须指向图中存在且处于 Running 的节点;否则恢复后配额索引
// 虽然看似有效,下一轮 ready 计算却会与活动 run 互相矛盾。反过来,图中
// 每个 Running 节点也必须有且仅有一个对应活动 run;隔离释放 run 后由
// PersistentCoordinator 将节点写成 Waiting,不能把不完整的 Running 状态
// 写进可恢复快照。
let active_runs_by_task = snapshot
.coordinator
.active_runs
.iter()
.map(|run| (run.task_id.as_str(), run))
.collect::<BTreeMap<_, _>>();
for run in &snapshot.coordinator.active_runs {
let Some(task) = snapshot.graph.task(&run.task_id) else {
return Err(OrchestrationError::InvalidInput(format!(
"活动 run {} 引用了未知任务 {}",
run.run_id, run.task_id
)));
};
if task.agent_id != run.agent_id {
return Err(OrchestrationError::InvalidInput(format!(
"活动 run {} 的 agent 与任务 {} 不一致",
run.run_id, run.task_id
)));
}
if task.status != TaskStatus::Running {
return Err(OrchestrationError::InvalidInput(format!(
"活动 run {} 的任务 {} 不是 running",
run.run_id, run.task_id
)));
}
}
for task in &snapshot.graph.tasks {
if task.status != TaskStatus::Running {
continue;
}
let Some(run) = active_runs_by_task.get(task.id.as_str()) else {
return Err(OrchestrationError::InvalidInput(format!(
"running 任务 {} 没有对应活动 run",
task.id
)));
};
if run.agent_id != task.agent_id {
return Err(OrchestrationError::InvalidInput(format!(
"running 任务 {} 的 agent 与活动 run {} 不一致",
task.id, run.run_id
)));
}
}
Ok(())
}
/// 进程内的完整编排快照存储,带 revision CAS。
#[derive(Clone, Debug, Default)]
pub struct InMemoryOrchestrationSnapshotStore {
snapshot: Arc<Mutex<Option<OrchestrationSnapshot>>>,
}
impl InMemoryOrchestrationSnapshotStore {
pub fn new() -> Self {
Self::default()
}
}
impl OrchestrationSnapshotStore for InMemoryOrchestrationSnapshotStore {
fn load_snapshot(&self) -> Result<Option<OrchestrationSnapshot>, OrchestrationError> {
self.snapshot
.lock()
.map(|snapshot| snapshot.clone())
.map_err(|_| OrchestrationError::InvalidInput("编排快照存储锁已损坏".to_owned()))
}
fn save_snapshot(
&self,
expected_revision: Option<u64>,
snapshot: &OrchestrationSnapshot,
) -> Result<(), OrchestrationError> {
validate_orchestration_snapshot(snapshot)?;
let mut current = self
.snapshot
.lock()
.map_err(|_| OrchestrationError::InvalidInput("编排快照存储锁已损坏".to_owned()))?;
let actual = current.as_ref().map(|value| value.revision);
if actual != expected_revision {
return Err(OrchestrationError::RevisionConflict {
expected: expected_revision,
actual,
});
}
*current = Some(snapshot.clone());
Ok(())
}
}
/// 使用临时文件 + rename 保存完整编排快照的轻量适配器。
///
/// 同一实例内写入串行化,另外使用与快照同目录的 sidecar advisory lock
/// 把“读取 revision → CAS → rename”收成跨进程临界区。读取仍只会看到旧文件
/// 或完整新文件,不会看到截断 JSON;锁文件本身不承载业务状态,丢失后可由
/// 下一次写入重新创建。
#[derive(Debug)]
pub struct JsonFileOrchestrationSnapshotStore {
path: PathBuf,
write_lock: Mutex<()>,
}
impl JsonFileOrchestrationSnapshotStore {
pub fn new(path: impl Into<PathBuf>) -> Result<Self, OrchestrationError> {
let path = path.into();
if path.as_os_str().is_empty() {
return Err(OrchestrationError::InvalidInput(
"orchestration 快照路径不能为空".to_owned(),
));
}
Ok(Self {
path,
write_lock: Mutex::new(()),
})
}
pub fn path(&self) -> &Path {
&self.path
}
}
impl OrchestrationSnapshotStore for JsonFileOrchestrationSnapshotStore {
fn load_snapshot(&self) -> Result<Option<OrchestrationSnapshot>, OrchestrationError> {
let file = match OpenOptions::new().read(true).open(&self.path) {
Ok(file) => file,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(file_io_error("读取 orchestration 快照", error)),
};
let metadata = file
.metadata()
.map_err(|error| file_io_error("读取 orchestration 快照元数据", error))?;
let size = usize::try_from(metadata.len()).unwrap_or(usize::MAX);
if size > MAX_COORDINATOR_SNAPSHOT_BYTES {
return Err(OrchestrationError::InvalidInput(format!(
"orchestration 快照超过 {} 字节上限",
MAX_COORDINATOR_SNAPSHOT_BYTES
)));
}
let mut encoded = Vec::with_capacity(size.min(MAX_COORDINATOR_SNAPSHOT_BYTES + 1));
file.take((MAX_COORDINATOR_SNAPSHOT_BYTES as u64) + 1)
.read_to_end(&mut encoded)
.map_err(|error| file_io_error("读取 orchestration 快照内容", error))?;
if encoded.len() > MAX_COORDINATOR_SNAPSHOT_BYTES {
return Err(OrchestrationError::InvalidInput(format!(
"orchestration 快照超过 {} 字节上限",
MAX_COORDINATOR_SNAPSHOT_BYTES
)));
}
let snapshot =
serde_json::from_slice::<OrchestrationSnapshot>(&encoded).map_err(|error| {
OrchestrationError::InvalidInput(format!("orchestration 快照无法解码: {error}"))
})?;
validate_orchestration_snapshot(&snapshot)?;
Ok(Some(snapshot))
}
fn save_snapshot(
&self,
expected_revision: Option<u64>,
snapshot: &OrchestrationSnapshot,
) -> Result<(), OrchestrationError> {
validate_orchestration_snapshot(snapshot)?;
let encoded = serde_json::to_vec(snapshot).map_err(|error| {
OrchestrationError::InvalidInput(format!("orchestration 快照无法编码: {error}"))
})?;
if encoded.len() > MAX_COORDINATOR_SNAPSHOT_BYTES {
return Err(OrchestrationError::InvalidInput(format!(
"orchestration 快照超过 {} 字节上限",
MAX_COORDINATOR_SNAPSHOT_BYTES
)));
}
let _guard = self.write_lock.lock().map_err(|_| {
OrchestrationError::InvalidInput("orchestration 文件写锁已损坏".to_owned())
})?;
// 先创建父目录,再打开 sidecar;否则首次写入嵌套目录时锁文件无法
// 创建,而目标快照本身尚未有机会落盘。
if let Some(parent) = self
.path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
{
fs::create_dir_all(parent)
.map_err(|error| file_io_error("创建 orchestration 快照目录", error))?;
}
let lock_path = orchestration_lock_path(&self.path);
let lock_file = OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(&lock_path)
.map_err(|error| file_io_error("创建 orchestration sidecar 锁", error))?;
lock_file
.lock()
.map_err(|error| file_io_error("获取 orchestration sidecar 锁", error))?;
// CAS 必须同时受进程内 Mutex 和 sidecar 文件锁保护;这样两个进程
// 不会都基于同一旧 revision 通过检查。lock_file 持有到 rename 完成,
// 离开作用域时自动释放 advisory lock。
let actual = self.load_snapshot()?.map(|value| value.revision);
if actual != expected_revision {
return Err(OrchestrationError::RevisionConflict {
expected: expected_revision,
actual,
});
}
let temp_path = temporary_orchestration_snapshot_path(&self.path);
let write_result = (|| {
let mut file = OpenOptions::new()
.create_new(true)
.write(true)
.open(&temp_path)
.map_err(|error| file_io_error("创建 orchestration 临时快照", error))?;
file.write_all(&encoded)
.map_err(|error| file_io_error("写入 orchestration 临时快照", error))?;
file.sync_all()
.map_err(|error| file_io_error("同步 orchestration 临时快照", error))?;
fs::rename(&temp_path, &self.path)
.map_err(|error| file_io_error("替换 orchestration 快照", error))
})();
if write_result.is_err() {
let _ = fs::remove_file(&temp_path);
}
write_result
}
}
fn orchestration_lock_path(path: &Path) -> PathBuf {
let file_name = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("orchestration.json");
path.with_file_name(format!(".{file_name}.lock"))
}
fn temporary_orchestration_snapshot_path(path: &Path) -> PathBuf {
let id = NEXT_SNAPSHOT_TEMP_ID.fetch_add(1, Ordering::Relaxed);
let file_name = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("orchestration.json");
path.with_file_name(format!(
".{file_name}.orchestration-tmp-{}-{id}",
std::process::id()
))
}
/// 进程内 Coordinator 快照存储,适合测试和短生命周期宿主。
#[derive(Clone, Debug, Default)]
pub struct InMemoryCoordinatorStore {
snapshot: Arc<Mutex<Option<CoordinatorSnapshot>>>,
}
impl InMemoryCoordinatorStore {
pub fn new() -> Self {
Self::default()
}
}
impl CoordinatorStore for InMemoryCoordinatorStore {
fn load_snapshot(&self) -> Result<Option<CoordinatorSnapshot>, OrchestrationError> {
self.snapshot
.lock()
.map(|snapshot| snapshot.clone())
.map_err(|_| OrchestrationError::InvalidInput("coordinator 存储锁已损坏".to_owned()))
}
fn save_snapshot(&self, snapshot: &CoordinatorSnapshot) -> Result<(), OrchestrationError> {
validate_snapshot(snapshot)?;
self.snapshot
.lock()
.map(|mut current| {
*current = Some(snapshot.clone());
})
.map_err(|_| OrchestrationError::InvalidInput("coordinator 存储锁已损坏".to_owned()))
}
}
/// 使用同目录临时文件 + rename 保存 Coordinator 快照的文件适配器。
///
/// 这是一个轻量宿主适配器,不是跨主机锁服务:同一实例内的写入会串行化,
/// 多进程同时写入时由最后一次成功 rename 胜出。读取始终看到旧文件或完整的
/// 新文件,不会看到截断 JSON。
#[derive(Debug)]
pub struct JsonFileCoordinatorStore {
path: PathBuf,
write_lock: Mutex<()>,
}
impl JsonFileCoordinatorStore {
pub fn new(path: impl Into<PathBuf>) -> Result<Self, OrchestrationError> {
let path = path.into();
if path.as_os_str().is_empty() {
return Err(OrchestrationError::InvalidInput(
"coordinator 快照路径不能为空".to_owned(),
));
}
Ok(Self {
path,
write_lock: Mutex::new(()),
})
}
pub fn path(&self) -> &Path {
&self.path
}
}
impl CoordinatorStore for JsonFileCoordinatorStore {
fn load_snapshot(&self) -> Result<Option<CoordinatorSnapshot>, OrchestrationError> {
let file = match OpenOptions::new().read(true).open(&self.path) {
Ok(file) => file,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(file_io_error("读取 coordinator 快照", error)),
};
let metadata = file
.metadata()
.map_err(|error| file_io_error("读取 coordinator 快照元数据", error))?;
let size = usize::try_from(metadata.len()).unwrap_or(usize::MAX);
if size > MAX_COORDINATOR_SNAPSHOT_BYTES {
return Err(OrchestrationError::InvalidInput(format!(
"coordinator 快照超过 {} 字节上限",
MAX_COORDINATOR_SNAPSHOT_BYTES
)));
}
// 元数据和实际读取之间可能发生替换/增长;即使文件在检查后变大,
// `take` 也把本次读取限制在上限加一个字节,避免把适配器变成无界内存入口。
let mut encoded = Vec::with_capacity(size.min(MAX_COORDINATOR_SNAPSHOT_BYTES + 1));
file.take((MAX_COORDINATOR_SNAPSHOT_BYTES as u64) + 1)
.read_to_end(&mut encoded)
.map_err(|error| file_io_error("读取 coordinator 快照内容", error))?;
if encoded.len() > MAX_COORDINATOR_SNAPSHOT_BYTES {
return Err(OrchestrationError::InvalidInput(format!(
"coordinator 快照超过 {} 字节上限",
MAX_COORDINATOR_SNAPSHOT_BYTES
)));
}
let snapshot =
serde_json::from_slice::<CoordinatorSnapshot>(&encoded).map_err(|error| {
OrchestrationError::InvalidInput(format!("coordinator 快照无法解码: {error}"))
})?;
validate_snapshot(&snapshot)?;
Ok(Some(snapshot))
}
fn save_snapshot(&self, snapshot: &CoordinatorSnapshot) -> Result<(), OrchestrationError> {
validate_snapshot(snapshot)?;
let encoded = serde_json::to_vec(snapshot).map_err(|error| {
OrchestrationError::InvalidInput(format!("coordinator 快照无法编码: {error}"))
})?;
if encoded.len() > MAX_COORDINATOR_SNAPSHOT_BYTES {
return Err(OrchestrationError::InvalidInput(format!(
"coordinator 快照超过 {} 字节上限",
MAX_COORDINATOR_SNAPSHOT_BYTES
)));
}
let _guard = self.write_lock.lock().map_err(|_| {
OrchestrationError::InvalidInput("coordinator 文件写锁已损坏".to_owned())
})?;
if let Some(parent) = self
.path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
{
fs::create_dir_all(parent)
.map_err(|error| file_io_error("创建 coordinator 快照目录", error))?;
}
let temp_path = temporary_snapshot_path(&self.path);
let write_result = (|| {
let mut file = OpenOptions::new()
.create_new(true)
.write(true)
.open(&temp_path)
.map_err(|error| file_io_error("创建 coordinator 临时快照", error))?;
file.write_all(&encoded)
.map_err(|error| file_io_error("写入 coordinator 临时快照", error))?;
file.sync_all()
.map_err(|error| file_io_error("同步 coordinator 临时快照", error))?;
fs::rename(&temp_path, &self.path)
.map_err(|error| file_io_error("替换 coordinator 快照", error))
})();
if write_result.is_err() {
// 只删除本次明确创建的临时文件;不触碰目录中的其它内容。
let _ = fs::remove_file(&temp_path);
}
write_result
}
}
fn validate_snapshot(snapshot: &CoordinatorSnapshot) -> Result<(), OrchestrationError> {
// 复用 Coordinator 的完整恢复校验,避免不同 Store 保存出无法导入的状态。
Coordinator::from_snapshot(snapshot.clone()).map(|_| ())
}
fn file_io_error(operation: &str, error: std::io::Error) -> OrchestrationError {
OrchestrationError::InvalidInput(format!("{operation}失败: {error}"))
}
fn temporary_snapshot_path(path: &Path) -> PathBuf {
let id = NEXT_SNAPSHOT_TEMP_ID.fetch_add(1, Ordering::Relaxed);
let file_name = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("coordinator.json");
path.with_file_name(format!(".{file_name}.tmp-{}-{id}", std::process::id()))
}
/// 多 Agent 的轻量协调控制面。
///
/// Coordinator 只保留活动 run、消息去重账本和节点隔离标记;它不启动线程、
/// 不执行单 Agent reducer,也不承担持久化。宿主可以把 `dispatch_wave` 返回的
/// run 交给任意执行器,并在完成/失败后调用相应的收口 API。
#[derive(Clone, Debug)]
pub struct Coordinator {
quota: CoordinatorQuota,
active_runs: BTreeMap<String, CoordinatorRun>,
active_tasks: BTreeMap<String, String>,
active_by_agent: BTreeMap<String, usize>,
isolated_nodes: BTreeMap<String, String>,
messages: BTreeMap<String, CoordinatorMessage>,
mailboxes: BTreeMap<String, VecDeque<String>>,
}
impl Coordinator {
pub fn new(quota: CoordinatorQuota) -> Self {
Self {
quota,
active_runs: BTreeMap::new(),
active_tasks: BTreeMap::new(),
active_by_agent: BTreeMap::new(),
isolated_nodes: BTreeMap::new(),
messages: BTreeMap::new(),
mailboxes: BTreeMap::new(),
}
}
pub fn try_new(quota: CoordinatorQuota) -> Result<Self, OrchestrationError> {
// `CoordinatorQuota` 通常来自 try_new;再次校验可避免未经校验的反序列化
// 配置把协调器置于永远无法接收 run 的状态。
CoordinatorQuota::try_new(quota.max_active_runs, quota.max_active_runs_per_agent)?;
Ok(Self::new(quota))
}
pub fn quota(&self) -> CoordinatorQuota {
self.quota
}
/// 导出当前协调控制面的内存状态。
///
/// 该快照不包含 TaskGraph,也不暗示任何自动恢复或重放行为;调用方可以在
/// 自己选择的存储中保存它,并通过 [`Self::from_snapshot`] 显式恢复。
pub fn snapshot(&self) -> CoordinatorSnapshot {
CoordinatorSnapshot {
schema_version: COORDINATOR_SNAPSHOT_SCHEMA_VERSION.to_owned(),
quota: self.quota,
active_runs: self.active_runs.values().cloned().collect(),
isolated_nodes: self.isolated_nodes.clone(),
messages: self.messages.values().cloned().collect(),
mailboxes: self
.mailboxes
.iter()
.map(|(node_id, message_ids)| {
(
node_id.clone(),
message_ids.iter().cloned().collect::<Vec<_>>(),
)
})
.collect(),
}
}
/// 从结构化快照显式恢复 Coordinator,并重建所有派生索引。
pub fn from_snapshot(snapshot: CoordinatorSnapshot) -> Result<Self, OrchestrationError> {
if snapshot.schema_version != COORDINATOR_SNAPSHOT_SCHEMA_VERSION {
return Err(OrchestrationError::InvalidInput(format!(
"不支持的 coordinator 快照版本: {}",
snapshot.schema_version
)));
}
let quota = CoordinatorQuota::try_new(
snapshot.quota.max_active_runs,
snapshot.quota.max_active_runs_per_agent,
)?;
// 先恢复隔离标记,避免导入一个同时占用隔离节点配额的无效状态。
let mut coordinator = Self {
quota,
active_runs: BTreeMap::new(),
active_tasks: BTreeMap::new(),
active_by_agent: BTreeMap::new(),
isolated_nodes: BTreeMap::new(),
messages: BTreeMap::new(),
mailboxes: BTreeMap::new(),
};
for (node_id, reason) in snapshot.isolated_nodes {
validate_id(&node_id, "coordinator node id")?;
validate_isolation_reason(&reason)?;
coordinator.isolated_nodes.insert(node_id, reason);
}
for run in snapshot.active_runs {
let request = CoordinatorRunRequest::try_new(
run.run_id.clone(),
run.task_id.clone(),
run.agent_id.clone(),
)?;
// `try_start_run` 同时检查重复键、隔离节点和两级配额,并在成功后
// 通过同一条内部路径重建 active_tasks/active_by_agent 索引。
coordinator.ensure_run_available(&request, 1)?;
coordinator.insert_active_run(run);
}
for message in snapshot.messages {
message.validate()?;
if coordinator
.messages
.insert(message.message_id.clone(), message)
.is_some()
{
return Err(OrchestrationError::InvalidInput(
"coordinator 快照包含重复消息 ID".to_owned(),
));
}
}
let mut queued_ids = BTreeSet::new();
for (node_id, message_ids) in snapshot.mailboxes {
validate_id(&node_id, "coordinator mailbox node id")?;
let mut mailbox = VecDeque::with_capacity(message_ids.len());
for message_id in message_ids {
validate_id(&message_id, "coordinator mailbox message id")?;
let message = coordinator.messages.get(&message_id).ok_or_else(|| {
OrchestrationError::InvalidInput(format!(
"coordinator mailbox 引用了未知消息: {message_id}"
))
})?;
if message.to_node != node_id {
return Err(OrchestrationError::InvalidInput(format!(
"coordinator mailbox 与消息目标不一致: {message_id}"
)));
}
if !queued_ids.insert(message_id.clone()) {
return Err(OrchestrationError::InvalidInput(format!(
"coordinator 消息重复出现在 mailbox: {message_id}"
)));
}
mailbox.push_back(message_id);
}
coordinator.mailboxes.insert(node_id, mailbox);
}
Ok(coordinator)
}
/// 将协调器快照编码为 JSON;编码结果只包含上述显式状态,不包含派生索引。
pub fn export_snapshot_json(&self) -> Result<String, OrchestrationError> {
serde_json::to_string(&self.snapshot()).map_err(|error| {
OrchestrationError::InvalidInput(format!("coordinator 快照无法编码: {error}"))
})
}
/// 从 JSON 显式导入协调器快照。
pub fn import_snapshot_json(encoded: &str) -> Result<Self, OrchestrationError> {
let snapshot = serde_json::from_str::<CoordinatorSnapshot>(encoded).map_err(|error| {
OrchestrationError::InvalidInput(format!("coordinator 快照无法解码: {error}"))
})?;
Self::from_snapshot(snapshot)
}
/// 把当前控制面快照交给调用方选择的持久化适配器。
pub fn save_to<S: CoordinatorStore + ?Sized>(
&self,
store: &S,
) -> Result<(), OrchestrationError> {
store.save_snapshot(&self.snapshot())
}
/// 从调用方选择的持久化适配器恢复 Coordinator。
pub fn load_from<S: CoordinatorStore + ?Sized>(
store: &S,
) -> Result<Option<Self>, OrchestrationError> {
store.load_snapshot()?.map(Self::from_snapshot).transpose()
}
/// 保留一个 run 配额,但不触碰 TaskGraph;适用于图外的独立运行。
pub fn try_start_run(
&mut self,
run_id: impl Into<String>,
task_id: impl Into<String>,
agent_id: impl Into<String>,
) -> Result<CoordinatorRun, OrchestrationError> {
let request = CoordinatorRunRequest::try_new(run_id, task_id, agent_id)?;
self.ensure_run_available(&request, 1)?;
let run = CoordinatorRun {
run_id: request.run_id,
task_id: request.task_id,
agent_id: request.agent_id,
};
self.insert_active_run(run.clone());
Ok(run)
}
/// 将同一 ready wave 的多个任务一次性占用为 Running。
///
/// 所有校验和配额检查都在图及协调器状态修改前完成,因此拒绝时不会留下
/// 半个波次或半个配额。worker 并发执行仍由调用方决定。
pub fn dispatch_wave<I>(
&mut self,
graph: &mut TaskGraph,
requests: I,
) -> Result<Vec<CoordinatorRun>, OrchestrationError>
where
I: IntoIterator<Item = CoordinatorRunRequest>,
{
let requests = requests.into_iter().collect::<Vec<_>>();
if requests.is_empty() {
return Err(OrchestrationError::InvalidInput(
"coordinator 波次不能为空".to_owned(),
));
}
let ready = graph.ready_task_ids().into_iter().collect::<BTreeSet<_>>();
let mut seen_tasks = BTreeSet::new();
let mut seen_runs = BTreeSet::new();
let mut requested_by_agent = BTreeMap::<String, usize>::new();
for request in &requests {
request.validate()?;
if !seen_runs.insert(request.run_id.as_str()) {
return Err(OrchestrationError::DuplicateRun(request.run_id.clone()));
}
if !seen_tasks.insert(request.task_id.as_str()) {
return Err(OrchestrationError::DuplicateTask(request.task_id.clone()));
}
if graph.task(&request.task_id).is_none() {
return Err(OrchestrationError::UnknownTask(request.task_id.clone()));
}
if graph
.task(&request.task_id)
.is_some_and(|task| task.agent_id != request.agent_id)
{
return Err(OrchestrationError::InvalidInput(format!(
"task {} 的 agent 与 run 请求不一致",
request.task_id
)));
}
if !ready.contains(request.task_id.as_str()) {
return Err(OrchestrationError::DependencyBlocked(format!(
"{} 尚未处于当前 ready wave",
request.task_id
)));
}
if self.is_node_isolated(&request.task_id) {
return Err(OrchestrationError::NodeIsolated(request.task_id.clone()));
}
if self.active_runs.contains_key(&request.run_id) {
return Err(OrchestrationError::DuplicateRun(request.run_id.clone()));
}
if self.active_tasks.contains_key(&request.task_id) {
return Err(OrchestrationError::DuplicateRun(format!(
"task {} 已有活动运行",
request.task_id
)));
}
*requested_by_agent
.entry(request.agent_id.clone())
.or_default() += 1;
}
let active = self.active_runs.len();
if active.saturating_add(requests.len()) > self.quota.max_active_runs {
return Err(OrchestrationError::QuotaExceeded {
scope: "all-runs".to_owned(),
limit: self.quota.max_active_runs,
active,
requested: requests.len(),
});
}
for (agent_id, requested) in &requested_by_agent {
let active_for_agent = self.active_by_agent.get(agent_id).copied().unwrap_or(0);
if active_for_agent.saturating_add(*requested) > self.quota.max_active_runs_per_agent {
return Err(OrchestrationError::QuotaExceeded {
scope: format!("agent:{agent_id}"),
limit: self.quota.max_active_runs_per_agent,
active: active_for_agent,
requested: *requested,
});
}
}
// 在副本上推进图,确保后续任何意外校验失败都不会污染调用方的 epoch。
let mut next_graph = graph.clone();
for request in &requests {
next_graph.set_task_status(&request.task_id, TaskStatus::Running)?;
}
*graph = next_graph;
let mut started = Vec::with_capacity(requests.len());
for request in requests {
let run = CoordinatorRun {
run_id: request.run_id,
task_id: request.task_id,
agent_id: request.agent_id,
};
self.insert_active_run(run.clone());
started.push(run);
}
Ok(started)
}
pub fn finish_run(&mut self, run_id: &str) -> Result<CoordinatorRun, OrchestrationError> {
self.remove_active_run(run_id)
.ok_or_else(|| OrchestrationError::UnknownRun(run_id.to_owned()))
}
/// 原子取消一个活动 run,并同步把它的图节点标记为 `Cancelled`。
///
/// 低层 Coordinator 同时持有图和活动索引时也应走这条入口,避免调用方
/// 先释放配额、再更新图状态而在中途失败。真正的 Runtime 取消信号仍由
/// 宿主负责;这里仅维护编排控制面的状态。
pub fn cancel_run(
&mut self,
graph: &mut TaskGraph,
run_id: &str,
) -> Result<CoordinatorRun, OrchestrationError> {
let run = self
.active_run(run_id)
.cloned()
.ok_or_else(|| OrchestrationError::UnknownRun(run_id.to_owned()))?;
// 先在副本中校验状态更新,再修改活动索引,确保错误不会留下半个取消。
let mut next_graph = graph.clone();
next_graph.set_task_status(&run.task_id, TaskStatus::Cancelled)?;
self.remove_active_run(run_id);
*graph = next_graph;
Ok(run)
}
pub fn active_run_count(&self) -> usize {
self.active_runs.len()
}
pub fn active_run_count_for_agent(&self, agent_id: &str) -> usize {
self.active_by_agent.get(agent_id).copied().unwrap_or(0)
}
pub fn active_run(&self, run_id: &str) -> Option<&CoordinatorRun> {
self.active_runs.get(run_id)
}
/// 隔离单个节点,并释放该节点仍占用的活动 run 配额。
///
/// 重复隔离是幂等的:第一次故障原因保留,返回空的释放列表。
pub fn isolate_node(
&mut self,
node_id: impl Into<String>,
reason: impl Into<String>,
) -> Result<Vec<String>, OrchestrationError> {
let node_id = node_id.into();
let reason = reason.into();
validate_id(&node_id, "coordinator node id")?;
validate_isolation_reason(&reason)?;
if self.isolated_nodes.contains_key(&node_id) {
return Ok(Vec::new());
}
let run_ids = self
.active_runs
.values()
.filter(|run| run.task_id == node_id)
.map(|run| run.run_id.clone())
.collect::<Vec<_>>();
for run_id in &run_ids {
self.remove_active_run(run_id);
}
self.isolated_nodes.insert(node_id, reason);
Ok(run_ids)
}
/// 同步标记图节点失败并隔离它;下游是否重跑必须另行调用 repair_downstream。
pub fn fail_node(
&mut self,
graph: &mut TaskGraph,
node_id: impl AsRef<str>,
reason: impl Into<String>,
) -> Result<Vec<String>, OrchestrationError> {
let node_id = node_id.as_ref();
let reason = reason.into();
validate_id(node_id, "coordinator node id")?;
validate_isolation_reason(&reason)?;
graph.set_task_status(node_id, TaskStatus::Failed)?;
self.isolate_node(node_id.to_owned(), reason)
}
pub fn is_node_isolated(&self, node_id: &str) -> bool {
self.isolated_nodes.contains_key(node_id)
}
pub fn isolation_reason(&self, node_id: &str) -> Option<&str> {
self.isolated_nodes.get(node_id).map(String::as_str)
}
/// 显式修复图及协调器状态;不会自动启动新的 run。
pub fn repair_downstream(
&mut self,
graph: &mut TaskGraph,
seeds: &[impl AsRef<str>],
) -> Result<Vec<String>, OrchestrationError> {
let impacted = graph.expand_downstream(seeds)?;
// 修复会把受影响节点重新置为 Pending,因此不能保留旧的活动 run 预约。
let to_release = self
.active_runs
.values()
.filter(|run| impacted.iter().any(|id| id == &run.task_id))
.map(|run| run.run_id.clone())
.collect::<Vec<_>>();
for run_id in &to_release {
self.remove_active_run(run_id);
}
let repaired = graph.repair_downstream(seeds)?;
for node_id in &repaired {
self.isolated_nodes.remove(node_id);
}
Ok(repaired)
}
/// 投递消息并按 message_id 去重;相同 ID 但内容不同会被拒绝。
pub fn deliver_message(
&mut self,
message: CoordinatorMessage,
) -> Result<MessageDelivery, OrchestrationError> {
message.validate()?;
if self.is_node_isolated(&message.from_node) {
return Err(OrchestrationError::NodeIsolated(message.from_node));
}
if self.is_node_isolated(&message.to_node) {
return Err(OrchestrationError::NodeIsolated(message.to_node));
}
if let Some(previous) = self.messages.get(&message.message_id) {
if previous == &message {
return Ok(MessageDelivery::Duplicate);
}
return Err(OrchestrationError::MessageConflict(message.message_id));
}
let message_id = message.message_id.clone();
let target = message.to_node.clone();
self.messages.insert(message_id.clone(), message);
self.mailboxes
.entry(target)
.or_default()
.push_back(message_id);
Ok(MessageDelivery::Delivered)
}
/// 取出节点当前收件箱中的消息;去重账本会保留到协调器生命周期结束。
pub fn receive_messages(
&mut self,
node_id: &str,
) -> Result<Vec<CoordinatorMessage>, OrchestrationError> {
validate_id(node_id, "coordinator node id")?;
if self.is_node_isolated(node_id) {
return Err(OrchestrationError::NodeIsolated(node_id.to_owned()));
}
let ids = self.mailboxes.remove(node_id).unwrap_or_default();
Ok(ids
.into_iter()
.filter_map(|message_id| self.messages.get(&message_id).cloned())
.collect())
}
pub fn pending_message_count(&self, node_id: &str) -> usize {
self.mailboxes.get(node_id).map_or(0, VecDeque::len)
}
fn ensure_run_available(
&self,
request: &CoordinatorRunRequest,
requested: usize,
) -> Result<(), OrchestrationError> {
if self.active_runs.contains_key(&request.run_id) {
return Err(OrchestrationError::DuplicateRun(request.run_id.clone()));
}
if self.active_tasks.contains_key(&request.task_id) {
return Err(OrchestrationError::DuplicateRun(format!(
"task {} 已有活动运行",
request.task_id
)));
}
if self.is_node_isolated(&request.task_id) {
return Err(OrchestrationError::NodeIsolated(request.task_id.clone()));
}
if self.active_runs.len().saturating_add(requested) > self.quota.max_active_runs {
return Err(OrchestrationError::QuotaExceeded {
scope: "all-runs".to_owned(),
limit: self.quota.max_active_runs,
active: self.active_runs.len(),
requested,
});
}
let active_for_agent = self
.active_by_agent
.get(&request.agent_id)
.copied()
.unwrap_or(0);
if active_for_agent.saturating_add(requested) > self.quota.max_active_runs_per_agent {
return Err(OrchestrationError::QuotaExceeded {
scope: format!("agent:{}", request.agent_id),
limit: self.quota.max_active_runs_per_agent,
active: active_for_agent,
requested,
});
}
Ok(())
}
fn insert_active_run(&mut self, run: CoordinatorRun) {
self.active_tasks
.insert(run.task_id.clone(), run.run_id.clone());
*self
.active_by_agent
.entry(run.agent_id.clone())
.or_default() += 1;
self.active_runs.insert(run.run_id.clone(), run);
}
fn remove_active_run(&mut self, run_id: &str) -> Option<CoordinatorRun> {
let run = self.active_runs.remove(run_id)?;
self.active_tasks.remove(&run.task_id);
if let Some(count) = self.active_by_agent.get_mut(&run.agent_id) {
if *count <= 1 {
self.active_by_agent.remove(&run.agent_id);
} else {
*count -= 1;
}
}
Some(run)
}
}
impl Default for Coordinator {
fn default() -> Self {
Self::new(CoordinatorQuota::default())
}
}
/// 把任务图和协调器状态作为一个原子控制面管理。
///
/// 这个类型是“持久化调度控制器”,不是线程池:它只在每次状态变更后提交
/// 快照,实际的 Agent Runtime 创建、worker 生命周期和 Provider 调用仍由宿主
/// 决定。这样多 Agent 扩展可以先拥有可靠的图/配额/消息恢复边界,再接入任意
/// 执行器,而不用复制单 Agent reducer。
#[derive(Debug)]
pub struct PersistentCoordinator<S> {
store: S,
graph: TaskGraph,
coordinator: Coordinator,
revision: u64,
}
impl<S> PersistentCoordinator<S>
where
S: OrchestrationSnapshotStore,
{
/// 创建并立即保存一个新的编排控制面。已有快照不会被覆盖。
pub fn new(
graph: TaskGraph,
quota: CoordinatorQuota,
store: S,
) -> Result<Self, OrchestrationError> {
let coordinator = Coordinator::try_new(quota)?;
let controller = Self {
store,
graph,
coordinator,
revision: 0,
};
controller.persist(None)?;
Ok(controller)
}
/// 从已有快照恢复;没有快照时返回 `None`,不会隐式创建空图。
pub fn load(store: S) -> Result<Option<Self>, OrchestrationError> {
let Some(snapshot) = store.load_snapshot()? else {
return Ok(None);
};
validate_orchestration_snapshot(&snapshot)?;
let coordinator = Coordinator::from_snapshot(snapshot.coordinator)?;
Ok(Some(Self {
store,
graph: snapshot.graph,
coordinator,
revision: snapshot.revision,
}))
}
/// 使用调用方已经取得的完整快照构造控制器,不写回存储。
pub fn from_snapshot(
snapshot: OrchestrationSnapshot,
store: S,
) -> Result<Self, OrchestrationError> {
validate_orchestration_snapshot(&snapshot)?;
let coordinator = Coordinator::from_snapshot(snapshot.coordinator)?;
Ok(Self {
store,
graph: snapshot.graph,
coordinator,
revision: snapshot.revision,
})
}
pub fn store(&self) -> &S {
&self.store
}
pub fn graph(&self) -> &TaskGraph {
&self.graph
}
pub fn coordinator(&self) -> &Coordinator {
&self.coordinator
}
pub fn revision(&self) -> u64 {
self.revision
}
/// 按图中稳定顺序生成一个受当前配额和隔离状态约束的 ready-task 计划。
///
/// 该方法只读,不领取任务;`max_tasks` 为本次计划的上限,必须大于零。
/// 计划中的 `expected_revision` 是宿主稍后 claim 时使用的乐观并发凭证。
/// run ID 不在这里生成,因为它属于宿主的 Runtime 命名/持久化策略。
pub fn plan_ready_tasks(
&self,
max_tasks: usize,
) -> Result<ReadyTaskDispatchPlan, OrchestrationError> {
if max_tasks == 0 {
return Err(OrchestrationError::InvalidInput(
"ready task 计划上限必须大于零".to_owned(),
));
}
let quota = self.coordinator.quota();
let global_room = quota
.max_active_runs()
.saturating_sub(self.coordinator.active_run_count());
let target_count = max_tasks.min(global_room);
let mut active_by_agent = self.coordinator.active_by_agent.clone();
let mut candidates = Vec::with_capacity(target_count);
// `ready_task_ids` 已经按图的稳定顺序返回;按此顺序选取可用任务,
// 让多个宿主在同一 revision 上得到相同的计划,冲突时只需 reload。
for task_id in self.graph.ready_task_ids() {
if candidates.len() >= target_count {
break;
}
if self.coordinator.is_node_isolated(task_id) {
continue;
}
let task = self
.graph
.task(task_id)
.expect("ready_task_ids 返回的任务必须存在");
let active_for_agent = active_by_agent.get(&task.agent_id).copied().unwrap_or(0);
if active_for_agent >= quota.max_active_runs_per_agent() {
continue;
}
candidates.push(ReadyTaskCandidate {
task_id: task.id.clone(),
agent_id: task.agent_id.clone(),
});
active_by_agent.insert(task.agent_id.clone(), active_for_agent + 1);
}
ReadyTaskDispatchPlan::try_new(self.revision, candidates)
}
/// 原子领取一个 ready-task 计划,并为每个候选任务绑定宿主生成的 run ID。
///
/// `expected_revision` 不匹配时不会读取或修改图状态;匹配后仍会复用
/// `Coordinator::dispatch_wave` 的全部校验和 Store CAS,因此计划过期、被
/// 隔离、重复领取或超出配额都会在持久化前失败且不留下半个 claim。
pub fn claim_ready_tasks<I, R>(
&mut self,
plan: &ReadyTaskDispatchPlan,
run_id_source: I,
) -> Result<Vec<CoordinatorRun>, OrchestrationError>
where
I: IntoIterator<Item = R>,
R: Into<String>,
{
plan.validate()?;
if plan.expected_revision != self.revision {
return Err(OrchestrationError::RevisionConflict {
expected: Some(plan.expected_revision),
actual: Some(self.revision),
});
}
// 最多读取候选数加一个 ID,既能检测多余输入,也不会让一个恶意无限
// iterator 把 claim 入口变成无界内存消耗。
let expected_count = plan.candidates.len();
let mut collected_run_ids = Vec::with_capacity(expected_count);
for run_id in run_id_source {
if collected_run_ids.len() >= expected_count {
return Err(OrchestrationError::InvalidInput(
"ready task 计划与 run ID 数量不一致".to_owned(),
));
}
collected_run_ids.push(run_id.into());
}
if collected_run_ids.len() != expected_count {
return Err(OrchestrationError::InvalidInput(
"ready task 计划与 run ID 数量不一致".to_owned(),
));
}
if expected_count == 0 {
return Ok(Vec::new());
}
let requests =
plan.candidates
.iter()
.zip(collected_run_ids)
.map(|(candidate, run_id)| {
let task = self.graph.task(&candidate.task_id).ok_or_else(|| {
OrchestrationError::UnknownTask(candidate.task_id.clone())
})?;
if task.agent_id != candidate.agent_id {
return Err(OrchestrationError::InvalidInput(format!(
"task {} 的 agent 与 ready 计划不一致",
candidate.task_id
)));
}
CoordinatorRunRequest::try_new(
run_id,
candidate.task_id.clone(),
candidate.agent_id.clone(),
)
})
.collect::<Result<Vec<_>, _>>()?;
// 统一复用已有的 ready/dependency/isolation/quota 校验和 CAS 提交路径。
self.dispatch_wave(requests)
}
pub fn snapshot(&self) -> OrchestrationSnapshot {
OrchestrationSnapshot {
schema_version: ORCHESTRATION_SNAPSHOT_SCHEMA_VERSION.to_owned(),
revision: self.revision,
graph: self.graph.clone(),
coordinator: self.coordinator.snapshot(),
}
}
/// 重新从 Store 读取最新状态,供 CAS 冲突后的调用方显式恢复。
pub fn reload(&mut self) -> Result<(), OrchestrationError> {
let snapshot = self
.store
.load_snapshot()?
.ok_or_else(|| OrchestrationError::InvalidInput("编排快照不存在".to_owned()))?;
validate_orchestration_snapshot(&snapshot)?;
self.graph = snapshot.graph;
self.coordinator = Coordinator::from_snapshot(snapshot.coordinator)?;
self.revision = snapshot.revision;
Ok(())
}
/// 显式重写当前 revision;一般只在恢复/迁移代码中使用。
pub fn save(&self) -> Result<(), OrchestrationError> {
self.persist(Some(self.revision))
}
/// 原子派发一个 ready wave;图状态和活动 run 配额要么一起更新,要么都
/// 保持旧值。Store 冲突时内存中的控制器也保持旧 epoch,调用方可 reload。
pub fn dispatch_wave<I>(
&mut self,
requests: I,
) -> Result<Vec<CoordinatorRun>, OrchestrationError>
where
I: IntoIterator<Item = CoordinatorRunRequest>,
{
self.mutate(|graph, coordinator| coordinator.dispatch_wave(graph, requests))
}
/// 结束一个活动 run 并将对应任务置为 Completed。
pub fn complete_run(&mut self, run_id: &str) -> Result<CoordinatorRun, OrchestrationError> {
self.mutate(|graph, coordinator| {
let run = coordinator
.active_run(run_id)
.cloned()
.ok_or_else(|| OrchestrationError::UnknownRun(run_id.to_owned()))?;
graph.set_task_status(&run.task_id, TaskStatus::Completed)?;
coordinator.finish_run(run_id)
})
}
/// 取消一个活动 run,并在同一 revision-CAS 中释放配额、持久化图状态。
///
/// Runtime 的 cooperative cancel/lease 收口由宿主负责;该方法只更新
/// orchestration 控制面,依赖它的下游任务仍需显式 repair。
pub fn cancel_run(&mut self, run_id: &str) -> Result<CoordinatorRun, OrchestrationError> {
self.mutate(|graph, coordinator| coordinator.cancel_run(graph, run_id))
}
/// 结束一个活动 run 并将对应任务置为 Failed;是否隔离节点由调用方另行
/// 选择,避免把失败策略硬编码进持久化控制器。
pub fn fail_run(&mut self, run_id: &str) -> Result<CoordinatorRun, OrchestrationError> {
self.mutate(|graph, coordinator| {
let run = coordinator
.active_run(run_id)
.cloned()
.ok_or_else(|| OrchestrationError::UnknownRun(run_id.to_owned()))?;
graph.set_task_status(&run.task_id, TaskStatus::Failed)?;
coordinator.finish_run(run_id)
})
}
/// 失败并隔离节点,同时释放该节点活动 run;下游修复仍须显式调用。
pub fn fail_node(
&mut self,
node_id: impl AsRef<str>,
reason: impl Into<String>,
) -> Result<Vec<String>, OrchestrationError> {
let node_id = node_id.as_ref().to_owned();
let reason = reason.into();
self.mutate(|graph, coordinator| coordinator.fail_node(graph, &node_id, reason))
}
pub fn isolate_node(
&mut self,
node_id: impl Into<String>,
reason: impl Into<String>,
) -> Result<Vec<String>, OrchestrationError> {
let node_id = node_id.into();
let reason = reason.into();
self.mutate(|graph, coordinator| {
let released = coordinator.isolate_node(&node_id, reason)?;
// 低层 Coordinator 只维护控制面索引;PersistentCoordinator 同时拥有
// TaskGraph,因此释放活动 run 后不能留下 Running 节点。Waiting 表示
// 可恢复的隔离态,后续必须显式调用 repair_downstream。
let should_wait = graph.task(&node_id).is_some_and(|task| {
matches!(task.status, TaskStatus::Pending | TaskStatus::Running)
});
if should_wait {
graph.set_task_status(&node_id, TaskStatus::Waiting)?;
}
Ok(released)
})
}
/// 显式修复种子节点及全部下游节点,并释放受影响的活动 run。
pub fn repair_downstream(
&mut self,
seeds: &[impl AsRef<str>],
) -> Result<Vec<String>, OrchestrationError> {
self.mutate(|graph, coordinator| coordinator.repair_downstream(graph, seeds))
}
/// 投递消息并持久化去重账本;重复投递是幂等的,不产生新 revision。
pub fn deliver_message(
&mut self,
message: CoordinatorMessage,
) -> Result<MessageDelivery, OrchestrationError> {
self.mutate(|_, coordinator| coordinator.deliver_message(message))
}
pub fn receive_messages(
&mut self,
node_id: &str,
) -> Result<Vec<CoordinatorMessage>, OrchestrationError> {
self.mutate(|_, coordinator| coordinator.receive_messages(node_id))
}
/// 把经过校验的 GraphProposal 安装为新图 epoch;现有活动 run 和消息账本
/// 原样保留,提案应用与快照写入在同一次控制面更新中完成。
pub fn apply_proposal(
&mut self,
proposal: &GraphProposal,
) -> Result<TaskGraph, OrchestrationError> {
let next_graph = proposal.apply(&self.graph)?;
let result = next_graph.clone();
self.mutate(|graph, _| {
*graph = next_graph;
Ok(result.clone())
})
}
/// 在同一 revision-CAS 更新中安装经过 Agent 目录和资源边界校验的提案。
///
/// 候选图先在当前 epoch 的不可变副本上构造;只有完整快照成功写入后,
/// 图、Coordinator 状态和 revision 才会一起替换,CAS 冲突不会泄漏半成品。
pub fn apply_proposal_with_limits(
&mut self,
proposal: &GraphProposal,
catalog: &AgentCatalog,
limits: &GraphLimits,
) -> Result<TaskGraph, OrchestrationError> {
let next_graph = self.graph.apply_proposal(proposal, catalog, limits)?;
let result = next_graph.clone();
self.mutate(|graph, _| {
*graph = next_graph;
Ok(result.clone())
})
}
fn mutate<T, F>(&mut self, operation: F) -> Result<T, OrchestrationError>
where
F: FnOnce(&mut TaskGraph, &mut Coordinator) -> Result<T, OrchestrationError>,
{
let mut next_graph = self.graph.clone();
let mut next_coordinator = self.coordinator.clone();
let result = operation(&mut next_graph, &mut next_coordinator)?;
// Idempotent operations such as delivering an already-known message do
// not consume a revision or perform an unnecessary disk write.
if next_graph == self.graph && next_coordinator.snapshot() == self.coordinator.snapshot() {
return Ok(result);
}
let next_revision = self
.revision
.checked_add(1)
.ok_or_else(|| OrchestrationError::InvalidInput("编排 revision 溢出".to_owned()))?;
let next_snapshot = OrchestrationSnapshot {
schema_version: ORCHESTRATION_SNAPSHOT_SCHEMA_VERSION.to_owned(),
revision: next_revision,
graph: next_graph.clone(),
coordinator: next_coordinator.snapshot(),
};
self.store
.save_snapshot(Some(self.revision), &next_snapshot)?;
self.graph = next_graph;
self.coordinator = next_coordinator;
self.revision = next_revision;
Ok(result)
}
fn persist(&self, expected_revision: Option<u64>) -> Result<(), OrchestrationError> {
let snapshot = self.snapshot();
self.store.save_snapshot(expected_revision, &snapshot)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{Arc, Barrier, Mutex};
use std::thread;
fn test_temp_path(prefix: &str) -> std::path::PathBuf {
// Respect an explicit TMPDIR, but keep direct test runs out of /tmp by
// defaulting to the repository-wide ~/data/tmp location.
let parent = std::env::var_os("TMPDIR")
.filter(|value| !value.is_empty())
.map(std::path::PathBuf::from)
.or_else(|| {
std::env::var_os("HOME")
.map(std::path::PathBuf::from)
.map(|home| home.join("data/tmp"))
})
.expect("TMPDIR 或 HOME 未设置,无法创建测试快照");
fs::create_dir_all(&parent).expect("创建测试临时目录");
let id = NEXT_SNAPSHOT_TEMP_ID.fetch_add(1, Ordering::Relaxed);
parent.join(format!("{prefix}-{}-{id}.json", std::process::id()))
}
fn graph() -> TaskGraph {
TaskGraph::try_new(
"goal",
[
TaskNode::try_new("a", "agent-a", std::iter::empty::<String>()).unwrap(),
TaskNode::try_new("b", "agent-b", ["a"]).unwrap(),
TaskNode::try_new("c", "agent-c", ["a"]).unwrap(),
],
)
.unwrap()
}
fn graph_catalog() -> AgentCatalog {
AgentCatalog::try_new([
agent_runtime_core::AgentDescriptor::try_new(
"agent-a",
"role-a",
std::iter::empty::<&str>(),
)
.unwrap(),
agent_runtime_core::AgentDescriptor::try_new(
"agent-b",
"role-b",
std::iter::empty::<&str>(),
)
.unwrap(),
agent_runtime_core::AgentDescriptor::try_new(
"agent-c",
"role-c",
std::iter::empty::<&str>(),
)
.unwrap(),
agent_runtime_core::AgentDescriptor::try_new(
"agent-d",
"role-d",
std::iter::empty::<&str>(),
)
.unwrap(),
agent_runtime_core::AgentDescriptor::try_new(
"agent-e",
"role-e",
std::iter::empty::<&str>(),
)
.unwrap(),
])
.unwrap()
}
#[test]
fn ready_and_waves_are_deterministic() {
let graph = graph();
assert_eq!(graph.ready_task_ids(), ["a"]);
let waves = graph.dependency_waves().unwrap();
assert_eq!(waves, vec![vec!["a"], vec!["b", "c"]]);
}
#[test]
fn status_updates_unlock_dependencies_and_repair_downstream() {
let mut current = graph();
assert!(matches!(
current.set_task_status("b", TaskStatus::Running),
Err(OrchestrationError::DependencyBlocked(_))
));
current.set_task_status("a", TaskStatus::Running).unwrap();
current.set_task_status("a", TaskStatus::Completed).unwrap();
assert_eq!(current.ready_task_ids(), ["b", "c"]);
current.set_task_status("b", TaskStatus::Running).unwrap();
current.set_task_status("b", TaskStatus::Completed).unwrap();
current.set_task_status("c", TaskStatus::Cancelled).unwrap();
assert!(current.dependency_waves().unwrap().is_empty());
// 上游失败时,下游仍是 Pending,但不能被误报为可执行波次。
let mut blocked = graph();
blocked.set_task_status("a", TaskStatus::Failed).unwrap();
assert!(matches!(
blocked.dependency_waves(),
Err(OrchestrationError::DependencyBlocked(_))
));
let repaired = current.repair_downstream(&["a"]).unwrap();
assert_eq!(repaired, ["a", "b", "c"]);
assert!(
current
.tasks
.iter()
.all(|task| task.status == TaskStatus::Pending)
);
assert_eq!(
current.dependency_waves().unwrap(),
vec![vec!["a"], vec!["b", "c"]]
);
}
#[test]
fn immutable_status_update_keeps_original_epoch() {
let graph = graph();
let next = graph.with_task_status("a", TaskStatus::Completed).unwrap();
assert_eq!(graph.task("a").unwrap().status, TaskStatus::Pending);
assert_eq!(next.task("a").unwrap().status, TaskStatus::Completed);
assert_eq!(next.ready_task_ids(), ["b", "c"]);
}
#[test]
fn independent_tasks_in_one_wave_can_run_concurrently() {
let graph = Arc::new(Mutex::new(graph()));
{
let mut graph = graph.lock().unwrap();
graph.set_task_status("a", TaskStatus::Completed).unwrap();
}
let wave = graph
.lock()
.unwrap()
.dependency_waves()
.unwrap()
.pop()
.unwrap();
let barrier = Arc::new(Barrier::new(wave.len()));
let handles = wave
.into_iter()
.map(|id| {
let graph = Arc::clone(&graph);
let barrier = Arc::clone(&barrier);
thread::spawn(move || {
assert_eq!(graph.lock().unwrap().task(&id).unwrap().dependencies, ["a"]);
// 两个 worker 都到达屏障后才继续,证明波次可以并行交给宿主。
barrier.wait();
let mut graph = graph.lock().unwrap();
graph.set_task_status(&id, TaskStatus::Running).unwrap();
graph.set_task_status(&id, TaskStatus::Completed).unwrap();
id
})
})
.collect::<Vec<_>>();
let mut completed = handles
.into_iter()
.map(|handle| handle.join().unwrap())
.collect::<Vec<_>>();
completed.sort();
assert_eq!(completed, ["b", "c"]);
assert!(
graph
.lock()
.unwrap()
.tasks
.iter()
.all(|task| task.status == TaskStatus::Completed)
);
}
#[test]
fn cycles_and_unknown_dependencies_fail() {
let a = TaskNode::try_new("a", "agent", ["b"]).unwrap();
let b = TaskNode::try_new("b", "agent", ["a"]).unwrap();
assert_eq!(
TaskGraph::try_new("goal", [a, b]).unwrap_err(),
OrchestrationError::Cycle
);
let unknown = TaskNode::try_new("a", "agent", ["missing"]).unwrap();
assert!(matches!(
TaskGraph::try_new("goal", [unknown]),
Err(OrchestrationError::InvalidDependency(_))
));
}
#[test]
fn proposal_creates_new_epoch_without_mutating_base() {
let base = graph();
let proposal = GraphProposal::try_new(
[TaskNodeProposal {
id: "d".into(),
agent_id: "agent-d".into(),
}],
[GraphEdge {
from: "b".into(),
to: "d".into(),
}],
)
.unwrap();
let next = proposal.apply(&base).unwrap();
assert!(base.task("d").is_none());
assert_eq!(next.task("d").unwrap().dependencies, ["b"]);
}
#[test]
fn graph_round_trip_rebuilds_lookup_index() {
let original = graph();
let encoded = serde_json::to_string(&original).unwrap();
let restored: TaskGraph = serde_json::from_str(&encoded).unwrap();
assert_eq!(restored.task("b").unwrap().agent_id, "agent-b");
assert_eq!(restored.ready_task_ids(), ["a"]);
}
#[test]
fn proposal_cannot_mutate_existing_task_dependencies_in_place() {
let base = graph();
let proposal = GraphProposal::try_new(
[TaskNodeProposal {
id: "d".into(),
agent_id: "agent-d".into(),
}],
[GraphEdge {
from: "a".into(),
to: "b".into(),
}],
)
.unwrap();
assert!(proposal.apply(&base).is_err());
}
#[test]
fn checked_proposal_enforces_catalog_and_graph_limits_atomically() {
let base = graph();
let catalog = graph_catalog();
assert_eq!(base.task_count(), 3);
assert_eq!(base.node_count(), 3);
assert_eq!(base.edge_count(), 2);
assert_eq!(base.depth(), 2);
assert_eq!(base.fan_out("a"), Some(2));
assert_eq!(base.fan_out("missing"), None);
let valid = GraphProposal::try_new(
[TaskProposal::try_new("d", "agent-d").unwrap()],
[GraphEdge::try_new("b", "d").unwrap()],
)
.unwrap();
let candidate = base
.apply_proposal(&valid, &catalog, &GraphLimits::default())
.unwrap();
assert_eq!(candidate.task_count(), 4);
assert_eq!(candidate.edge_count(), 3);
assert_eq!(candidate.depth(), 3);
assert_eq!(candidate.task("d").unwrap().status, TaskStatus::Pending);
assert_eq!(candidate.task("d").unwrap().dependencies, ["b"]);
assert_eq!(base.task("d"), None, "失败或成功都不能原地改写旧 epoch");
let unknown_agent = GraphProposal::try_new(
[TaskProposal::try_new("unknown", "agent-missing").unwrap()],
std::iter::empty::<GraphEdge>(),
)
.unwrap();
assert!(matches!(
base.apply_proposal(&unknown_agent, &catalog, &GraphLimits::default()),
Err(OrchestrationError::InvalidInput(message))
if message.contains("未注册 Agent")
));
let limits = GraphLimits::new(3, 3, 2, 2);
assert!(matches!(
base.apply_proposal(&valid, &catalog, &limits),
Err(OrchestrationError::Limit(message)) if message.contains("maxTasks")
));
assert!(matches!(
base.apply_proposal(
&valid,
&catalog,
&GraphLimits::new(8, 2, 8, 8)
),
Err(OrchestrationError::Limit(message)) if message.contains("maxEdges")
));
assert!(matches!(
base.apply_proposal(
&valid,
&catalog,
&GraphLimits::new(8, 8, 2, 8)
),
Err(OrchestrationError::Limit(message)) if message.contains("maxDepth")
));
let fan_out = GraphProposal::try_new(
[
TaskProposal::try_new("d", "agent-d").unwrap(),
TaskProposal::try_new("e", "agent-e").unwrap(),
],
[
GraphEdge::try_new("a", "d").unwrap(),
GraphEdge::try_new("a", "e").unwrap(),
],
)
.unwrap();
assert!(matches!(
base.apply_proposal(
&fan_out,
&catalog,
&GraphLimits::new(8, 8, 8, 2)
),
Err(OrchestrationError::Limit(message)) if message.contains("fan-out")
));
assert!(matches!(
GraphLimits::try_new(0, 1, 1, 1),
Err(OrchestrationError::Limit(message)) if message.contains("maxTasks")
));
let encoded = serde_json::to_value(GraphLimits::default()).unwrap();
assert_eq!(
encoded,
serde_json::json!({
"maxTasks": 128,
"maxEdges": 512,
"maxDepth": 32,
"maxOutDegree": 32
})
);
assert!(serde_json::from_value::<GraphLimits>(encoded).is_ok());
assert!(
serde_json::from_str::<GraphLimits>(
r#"{"maxTasks":1,"maxEdges":1,"maxDepth":1,"maxOutDegree":1,"extra":true}"#
)
.is_err()
);
}
#[test]
fn delegation_and_join_validate_boundaries() {
assert!(Delegation::try_new("d", "run", "agent", "do it").is_ok());
assert!(Join::try_new("j", ["d"], "all").is_ok());
assert!(Join::try_new("j", std::iter::empty::<String>(), "all").is_err());
assert!(matches!(
Join::try_new("j", ["d", "d"], "all"),
Err(OrchestrationError::InvalidInput(message))
if message.contains("重复 delegation")
));
assert!(Join::try_new("j", [""], "all").is_err());
}
#[test]
fn public_contract_deserialization_reapplies_constructor_validation() {
let invalid_node = serde_json::from_str::<TaskNode>(
r#"{"id":"a","agentId":"agent","dependencies":["a"]}"#,
);
assert!(invalid_node.is_err());
let invalid_delegation = serde_json::from_str::<Delegation>(
r#"{"id":"d","parentRunId":"run","childAgentId":"agent","task":""}"#,
);
assert!(invalid_delegation.is_err());
let invalid_join = serde_json::from_str::<Join>(
r#"{"id":"j","delegationIds":["d","d"],"strategy":"all"}"#,
);
assert!(invalid_join.is_err());
let invalid_proposal = serde_json::from_str::<GraphProposal>(
r#"{"nodes":[{"id":"n","agentId":"a"}],"edges":[{"from":"n","to":"n"}]}"#,
);
assert!(invalid_proposal.is_err());
let invalid_quota = serde_json::from_str::<CoordinatorQuota>(
r#"{"maxActiveRuns":0,"maxActiveRunsPerAgent":1}"#,
);
assert!(invalid_quota.is_err());
let invalid_run = serde_json::from_str::<CoordinatorRunRequest>(
r#"{"runId":"","taskId":"task","agentId":"agent"}"#,
);
assert!(invalid_run.is_err());
let invalid_message = serde_json::from_str::<CoordinatorMessage>(
r#"{"messageId":"m","fromNode":"","toNode":"node","payload":"x"}"#,
);
assert!(invalid_message.is_err());
}
#[test]
fn coordinator_rejects_runs_over_bounded_quota_without_partial_state() {
let quota = CoordinatorQuota::try_new(1, 1).unwrap();
let mut coordinator = Coordinator::new(quota);
coordinator
.try_start_run("run-a", "node-a", "agent-a")
.unwrap();
let error = coordinator
.try_start_run("run-b", "node-b", "agent-b")
.unwrap_err();
assert!(matches!(
error,
OrchestrationError::QuotaExceeded {
scope,
limit: 1,
active: 1,
requested: 1,
} if scope == "all-runs"
));
assert_eq!(coordinator.active_run_count(), 1);
assert!(coordinator.active_run("run-b").is_none());
}
#[test]
fn coordinator_dispatch_rejects_per_agent_quota_without_partial_state() {
let mut graph = TaskGraph::try_new(
"goal",
[
TaskNode::try_new("node-a", "agent-a", std::iter::empty::<String>()).unwrap(),
TaskNode::try_new("node-b", "agent-a", std::iter::empty::<String>()).unwrap(),
],
)
.unwrap();
let mut coordinator = Coordinator::new(CoordinatorQuota::try_new(2, 1).unwrap());
let error = coordinator
.dispatch_wave(
&mut graph,
[
CoordinatorRunRequest::try_new("run-a", "node-a", "agent-a").unwrap(),
CoordinatorRunRequest::try_new("run-b", "node-b", "agent-a").unwrap(),
],
)
.unwrap_err();
assert!(matches!(
error,
OrchestrationError::QuotaExceeded {
scope,
limit: 1,
active: 0,
requested: 2,
} if scope == "agent:agent-a"
));
assert_eq!(coordinator.active_run_count(), 0);
assert!(
graph
.tasks
.iter()
.all(|task| task.status == TaskStatus::Pending)
);
}
#[test]
fn coordinator_dispatches_independent_tasks_in_one_wave() {
let mut graph = graph();
graph.set_task_status("a", TaskStatus::Completed).unwrap();
let mut coordinator = Coordinator::new(CoordinatorQuota::try_new(2, 1).unwrap());
let started = coordinator
.dispatch_wave(
&mut graph,
[
CoordinatorRunRequest::try_new("run-b", "b", "agent-b").unwrap(),
CoordinatorRunRequest::try_new("run-c", "c", "agent-c").unwrap(),
],
)
.unwrap();
assert_eq!(
started
.iter()
.map(|run| run.run_id.as_str())
.collect::<Vec<_>>(),
["run-b", "run-c"]
);
assert_eq!(coordinator.active_run_count(), 2);
assert_eq!(graph.task("b").unwrap().status, TaskStatus::Running);
assert_eq!(graph.task("c").unwrap().status, TaskStatus::Running);
coordinator.finish_run("run-b").unwrap();
coordinator.finish_run("run-c").unwrap();
assert_eq!(coordinator.active_run_count(), 0);
}
#[test]
fn coordinator_cancel_run_updates_graph_and_releases_quota_atomically() {
let mut graph = graph();
let mut coordinator = Coordinator::new(CoordinatorQuota::try_new(1, 1).unwrap());
coordinator
.dispatch_wave(
&mut graph,
[CoordinatorRunRequest::try_new("run-a", "a", "agent-a").unwrap()],
)
.unwrap();
let cancelled = coordinator.cancel_run(&mut graph, "run-a").unwrap();
assert_eq!(cancelled.run_id, "run-a");
assert_eq!(graph.task("a").unwrap().status, TaskStatus::Cancelled);
assert_eq!(coordinator.active_run_count(), 0);
// 取消后配额已经释放,但依赖该节点的任务不会被误报为 ready。
assert!(
coordinator
.try_start_run("run-independent", "node-independent", "agent-a")
.is_ok()
);
assert!(graph.ready_task_ids().is_empty());
}
#[test]
fn coordinator_message_delivery_is_idempotent() {
let mut coordinator = Coordinator::default();
let message = CoordinatorMessage::try_text("m-1", "node-a", "node-b", "hello").unwrap();
assert_eq!(
coordinator.deliver_message(message.clone()).unwrap(),
MessageDelivery::Delivered
);
assert_eq!(coordinator.pending_message_count("node-b"), 1);
assert_eq!(
coordinator.deliver_message(message.clone()).unwrap(),
MessageDelivery::Duplicate
);
assert_eq!(coordinator.pending_message_count("node-b"), 1);
let received = coordinator.receive_messages("node-b").unwrap();
assert_eq!(received, std::slice::from_ref(&message));
assert!(coordinator.receive_messages("node-b").unwrap().is_empty());
assert_eq!(
coordinator.deliver_message(message).unwrap(),
MessageDelivery::Duplicate
);
let conflict = CoordinatorMessage::try_text("m-1", "node-a", "node-b", "changed").unwrap();
assert!(matches!(
coordinator.deliver_message(conflict),
Err(OrchestrationError::MessageConflict(id)) if id == "m-1"
));
}
#[test]
fn coordinator_rejects_messages_to_or_from_isolated_nodes() {
let mut coordinator = Coordinator::default();
coordinator
.isolate_node("node-a", "temporarily unavailable")
.unwrap();
let from_isolated =
CoordinatorMessage::try_text("m-from", "node-a", "node-b", "hello").unwrap();
assert!(matches!(
coordinator.deliver_message(from_isolated),
Err(OrchestrationError::NodeIsolated(id)) if id == "node-a"
));
let to_isolated =
CoordinatorMessage::try_text("m-to", "node-b", "node-a", "hello").unwrap();
assert!(matches!(
coordinator.deliver_message(to_isolated),
Err(OrchestrationError::NodeIsolated(id)) if id == "node-a"
));
assert!(matches!(
coordinator.receive_messages("node-a"),
Err(OrchestrationError::NodeIsolated(id)) if id == "node-a"
));
}
#[test]
fn coordinator_isolates_one_failed_node_and_keeps_siblings_running() {
let mut graph = TaskGraph::try_new(
"goal",
[
TaskNode::try_new("node-a", "agent-a", std::iter::empty::<String>()).unwrap(),
TaskNode::try_new("node-b", "agent-b", std::iter::empty::<String>()).unwrap(),
],
)
.unwrap();
let mut coordinator = Coordinator::new(CoordinatorQuota::try_new(2, 1).unwrap());
coordinator
.dispatch_wave(
&mut graph,
[
CoordinatorRunRequest::try_new("run-a", "node-a", "agent-a").unwrap(),
CoordinatorRunRequest::try_new("run-b", "node-b", "agent-b").unwrap(),
],
)
.unwrap();
let released = coordinator
.fail_node(&mut graph, "node-a", "provider failed")
.unwrap();
assert_eq!(released, ["run-a"]);
assert_eq!(graph.task("node-a").unwrap().status, TaskStatus::Failed);
assert!(coordinator.is_node_isolated("node-a"));
assert_eq!(
coordinator.isolation_reason("node-a"),
Some("provider failed")
);
assert!(coordinator.active_run("run-a").is_none());
assert!(coordinator.active_run("run-b").is_some());
assert!(matches!(
coordinator.try_start_run("run-a2", "node-a", "agent-a"),
Err(OrchestrationError::NodeIsolated(id)) if id == "node-a"
));
coordinator.finish_run("run-b").unwrap();
}
#[test]
fn coordinator_repair_is_explicit_and_releases_affected_runs() {
let mut graph = graph();
graph.set_task_status("a", TaskStatus::Completed).unwrap();
let mut coordinator = Coordinator::new(CoordinatorQuota::try_new(2, 1).unwrap());
coordinator
.dispatch_wave(
&mut graph,
[
CoordinatorRunRequest::try_new("run-b", "b", "agent-b").unwrap(),
CoordinatorRunRequest::try_new("run-c", "c", "agent-c").unwrap(),
],
)
.unwrap();
coordinator.isolate_node("a", "upstream failed").unwrap();
let repaired = coordinator.repair_downstream(&mut graph, &["a"]).unwrap();
assert_eq!(repaired, ["a", "b", "c"]);
assert!(!coordinator.is_node_isolated("a"));
assert_eq!(coordinator.active_run_count(), 0);
assert!(
graph
.tasks
.iter()
.all(|task| task.status == TaskStatus::Pending)
);
let restarted = coordinator
.dispatch_wave(
&mut graph,
[CoordinatorRunRequest::try_new("run-a2", "a", "agent-a").unwrap()],
)
.unwrap();
assert_eq!(restarted[0].task_id, "a");
assert_eq!(graph.task("a").unwrap().status, TaskStatus::Running);
}
#[test]
fn coordinator_repair_rejects_unknown_seed_without_partial_state() {
let mut graph = graph();
graph.set_task_status("a", TaskStatus::Completed).unwrap();
let mut coordinator = Coordinator::new(CoordinatorQuota::try_new(2, 1).unwrap());
coordinator
.dispatch_wave(
&mut graph,
[CoordinatorRunRequest::try_new("run-b", "b", "agent-b").unwrap()],
)
.unwrap();
let graph_before = graph.clone();
let snapshot_before = coordinator.snapshot();
assert!(matches!(
coordinator.repair_downstream(&mut graph, &["missing"]),
Err(OrchestrationError::UnknownTask(id)) if id == "missing"
));
assert_eq!(graph, graph_before);
assert_eq!(coordinator.snapshot(), snapshot_before);
}
#[test]
fn coordinator_snapshot_round_trip_rebuilds_indexes_and_message_ledger() {
let quota = CoordinatorQuota::try_new(2, 1).unwrap();
let mut coordinator = Coordinator::new(quota);
coordinator
.try_start_run("run-a", "node-a", "agent-a")
.unwrap();
coordinator
.isolate_node("node-x", "temporarily unavailable")
.unwrap();
let consumed =
CoordinatorMessage::try_text("m-consumed", "node-a", "node-b", "one").unwrap();
let queued = CoordinatorMessage::try_text("m-queued", "node-a", "node-c", "two").unwrap();
coordinator.deliver_message(consumed.clone()).unwrap();
coordinator.deliver_message(queued.clone()).unwrap();
assert_eq!(
coordinator.receive_messages("node-b").unwrap(),
std::slice::from_ref(&consumed)
);
let encoded = coordinator.export_snapshot_json().unwrap();
let restored = Coordinator::import_snapshot_json(&encoded).unwrap();
assert_eq!(restored.snapshot(), coordinator.snapshot());
assert_eq!(restored.active_run_count_for_agent("agent-a"), 1);
assert_eq!(restored.pending_message_count("node-c"), 1);
assert_eq!(
restored.isolation_reason("node-x"),
Some("temporarily unavailable")
);
// 派生索引和已消费消息的去重账本都在导入后继续生效。
let mut restored = restored;
assert!(matches!(
restored.try_start_run("run-a-duplicate", "node-a", "agent-a"),
Err(OrchestrationError::DuplicateRun(_))
));
assert_eq!(
restored.deliver_message(consumed).unwrap(),
MessageDelivery::Duplicate
);
assert_eq!(
restored.receive_messages("node-c").unwrap(),
std::slice::from_ref(&queued)
);
assert_eq!(
restored.deliver_message(queued).unwrap(),
MessageDelivery::Duplicate
);
}
#[test]
fn coordinator_snapshot_restore_preserves_quota_isolation_and_message_dedupe() {
let quota = CoordinatorQuota::try_new(2, 1).unwrap();
let mut coordinator = Coordinator::new(quota);
coordinator
.try_start_run("run-a", "node-a", "agent-a")
.unwrap();
coordinator
.isolate_node("node-x", "provider unavailable")
.unwrap();
let message = CoordinatorMessage::try_text("m-1", "node-a", "node-b", "hello").unwrap();
assert_eq!(
coordinator.deliver_message(message.clone()).unwrap(),
MessageDelivery::Delivered
);
let store = InMemoryCoordinatorStore::new();
coordinator.save_to(&store).unwrap();
let mut restored = Coordinator::load_from(&store).unwrap().unwrap();
assert_eq!(restored.pending_message_count("node-b"), 1);
assert_eq!(
restored.deliver_message(message).unwrap(),
MessageDelivery::Duplicate
);
assert!(matches!(
restored.try_start_run("run-b", "node-b", "agent-a"),
Err(OrchestrationError::QuotaExceeded { scope, .. })
if scope == "agent:agent-a"
));
assert!(matches!(
restored.try_start_run("run-x", "node-x", "agent-x"),
Err(OrchestrationError::NodeIsolated(id)) if id == "node-x"
));
assert!(matches!(
restored.deliver_message(
CoordinatorMessage::try_text("m-x", "node-x", "node-b", "blocked").unwrap()
),
Err(OrchestrationError::NodeIsolated(id)) if id == "node-x"
));
}
#[test]
fn coordinator_snapshot_rejects_unknown_version_and_mailbox_mismatch() {
let coordinator = Coordinator::default();
let mut snapshot = coordinator.snapshot();
snapshot.schema_version = "agent-runtime-coordinator.v0".to_owned();
assert!(matches!(
Coordinator::from_snapshot(snapshot),
Err(OrchestrationError::InvalidInput(message))
if message.contains("不支持的 coordinator 快照版本")
));
let mut snapshot = coordinator.snapshot();
let message = CoordinatorMessage::try_text("m-1", "node-a", "node-b", "hello").unwrap();
snapshot.messages.push(message);
snapshot
.mailboxes
.insert("node-c".to_owned(), vec!["m-1".to_owned()]);
assert!(matches!(
Coordinator::from_snapshot(snapshot),
Err(OrchestrationError::InvalidInput(message))
if message.contains("mailbox 与消息目标不一致")
));
}
#[test]
fn coordinator_store_round_trip_and_empty_load_are_explicit() {
let store = InMemoryCoordinatorStore::new();
assert!(Coordinator::load_from(&store).unwrap().is_none());
let mut coordinator = Coordinator::default();
coordinator
.try_start_run("run-a", "node-a", "agent-a")
.unwrap();
coordinator.save_to(&store).unwrap();
let restored = Coordinator::load_from(&store).unwrap().unwrap();
assert_eq!(restored.snapshot(), coordinator.snapshot());
}
fn assert_store_round_trip<S: CoordinatorStore + ?Sized>(store: &S) {
assert!(Coordinator::load_from(store).unwrap().is_none());
let mut coordinator = Coordinator::default();
coordinator
.try_start_run("run-contract", "node-contract", "agent-contract")
.unwrap();
let expected = coordinator.snapshot();
coordinator.save_to(store).unwrap();
let restored = Coordinator::load_from(store).unwrap().unwrap();
assert_eq!(restored.snapshot(), expected);
// Store 合同要求拒绝无法由 Coordinator 导入的快照,并保留之前的有效值。
let mut invalid = expected.clone();
invalid.schema_version = "invalid-version".to_owned();
assert!(store.save_snapshot(&invalid).is_err());
assert_eq!(
Coordinator::load_from(store).unwrap().unwrap().snapshot(),
expected
);
}
#[test]
fn coordinator_store_contract_is_shared_by_memory_and_json_file_adapters() {
let memory = InMemoryCoordinatorStore::new();
assert_store_round_trip(&memory);
let path = test_temp_path("agent-runtime-coordinator-contract");
let _ = std::fs::remove_file(&path);
let file = JsonFileCoordinatorStore::new(&path).unwrap();
assert_store_round_trip(&file);
let _ = std::fs::remove_file(path);
}
#[test]
fn json_file_store_replaces_atomically_and_rejects_corrupt_or_oversized_data() {
let path = test_temp_path("agent-runtime-coordinator-test");
let _ = std::fs::remove_file(&path);
let store = JsonFileCoordinatorStore::new(&path).unwrap();
let mut coordinator = Coordinator::default();
coordinator
.try_start_run("run-a", "node-a", "agent-a")
.unwrap();
coordinator.save_to(&store).unwrap();
let restored = Coordinator::load_from(&store).unwrap().unwrap();
assert_eq!(restored.snapshot(), coordinator.snapshot());
// 写入非 JSON 内容时,读取会失败而不会返回半个控制面状态。
std::fs::write(&path, b"{").unwrap();
assert!(matches!(
store.load_snapshot(),
Err(OrchestrationError::InvalidInput(message))
if message.contains("无法解码")
));
// 大文件在解析前被拒绝,避免把文件适配器变成无界内存入口。
std::fs::write(&path, vec![b'x'; MAX_COORDINATOR_SNAPSHOT_BYTES + 1]).unwrap();
assert!(matches!(
store.load_snapshot(),
Err(OrchestrationError::InvalidInput(message))
if message.contains("字节上限")
));
let _ = std::fs::remove_file(path);
}
#[test]
fn coordinator_store_does_not_save_invalid_snapshot() {
let store = InMemoryCoordinatorStore::new();
let mut snapshot = Coordinator::default().snapshot();
snapshot.schema_version = "unknown-version".to_owned();
assert!(store.save_snapshot(&snapshot).is_err());
assert!(store.load_snapshot().unwrap().is_none());
}
#[test]
fn persistent_coordinator_round_trips_graph_and_control_state() {
let store = InMemoryOrchestrationSnapshotStore::new();
let mut controller = PersistentCoordinator::new(
graph(),
CoordinatorQuota::try_new(2, 1).unwrap(),
store.clone(),
)
.unwrap();
assert_eq!(controller.revision(), 0);
controller
.dispatch_wave([CoordinatorRunRequest::try_new("run-a", "a", "agent-a").unwrap()])
.unwrap();
controller.complete_run("run-a").unwrap();
controller
.dispatch_wave([
CoordinatorRunRequest::try_new("run-b", "b", "agent-b").unwrap(),
CoordinatorRunRequest::try_new("run-c", "c", "agent-c").unwrap(),
])
.unwrap();
let message = CoordinatorMessage::try_text("m-1", "b", "c", "hello").unwrap();
assert_eq!(
controller.deliver_message(message.clone()).unwrap(),
MessageDelivery::Delivered
);
let restored = PersistentCoordinator::load(store)
.unwrap()
.expect("snapshot should exist");
assert_eq!(restored.revision(), controller.revision());
assert_eq!(restored.graph(), controller.graph());
assert_eq!(
restored.coordinator().active_run_count_for_agent("agent-b"),
1
);
assert_eq!(restored.coordinator().pending_message_count("c"), 1);
assert_eq!(
restored.graph().task("a").unwrap().status,
TaskStatus::Completed
);
let mut restored = restored;
assert_eq!(
restored.receive_messages("c").unwrap(),
std::slice::from_ref(&message)
);
assert_eq!(restored.coordinator().pending_message_count("c"), 0);
}
#[test]
fn persistent_cancel_run_commits_cancelled_state_and_releases_quota() {
let store = InMemoryOrchestrationSnapshotStore::new();
let mut controller = PersistentCoordinator::new(
graph(),
CoordinatorQuota::try_new(1, 1).unwrap(),
store.clone(),
)
.unwrap();
controller
.dispatch_wave([CoordinatorRunRequest::try_new("run-a", "a", "agent-a").unwrap()])
.unwrap();
let cancelled = controller.cancel_run("run-a").unwrap();
assert_eq!(cancelled.run_id, "run-a");
assert_eq!(controller.revision(), 2);
assert_eq!(
controller.graph().task("a").unwrap().status,
TaskStatus::Cancelled
);
assert_eq!(controller.coordinator().active_run_count(), 0);
assert!(controller.graph().ready_task_ids().is_empty());
let restored = PersistentCoordinator::load(store)
.unwrap()
.expect("cancelled snapshot should be recoverable");
assert_eq!(restored.revision(), 2);
assert_eq!(
restored.graph().task("a").unwrap().status,
TaskStatus::Cancelled
);
assert_eq!(restored.coordinator().active_run_count(), 0);
}
#[test]
fn persistent_coordinator_cas_conflict_does_not_leak_uncommitted_state() {
let store = InMemoryOrchestrationSnapshotStore::new();
let mut first = PersistentCoordinator::new(
graph(),
CoordinatorQuota::try_new(2, 1).unwrap(),
store.clone(),
)
.unwrap();
let mut stale = PersistentCoordinator::load(store.clone())
.unwrap()
.expect("stale controller");
first
.deliver_message(CoordinatorMessage::try_text("m-1", "a", "b", "first").unwrap())
.unwrap();
let error = stale
.deliver_message(CoordinatorMessage::try_text("m-2", "a", "b", "stale").unwrap())
.expect_err("stale write must fail CAS");
assert!(matches!(error, OrchestrationError::RevisionConflict { .. }));
assert_eq!(stale.revision(), 0);
assert_eq!(stale.coordinator().pending_message_count("b"), 0);
stale.reload().unwrap();
assert_eq!(stale.revision(), 1);
assert_eq!(stale.coordinator().pending_message_count("b"), 1);
}
#[test]
fn persistent_apply_proposal_is_atomic_and_revision_cas_protected() {
let store = InMemoryOrchestrationSnapshotStore::new();
let mut first = PersistentCoordinator::new(
graph(),
CoordinatorQuota::try_new(2, 1).unwrap(),
store.clone(),
)
.unwrap();
let mut stale = PersistentCoordinator::load(store.clone())
.unwrap()
.expect("stale controller");
// Keep an active run in the same snapshot so proposal installation must
// preserve Coordinator state while replacing only the graph epoch.
first
.dispatch_wave([CoordinatorRunRequest::try_new("run-a", "a", "agent-a").unwrap()])
.unwrap();
let proposal = GraphProposal::try_new(
[TaskProposal::try_new("d", "agent-d").unwrap()],
[GraphEdge::try_new("a", "d").unwrap()],
)
.unwrap();
let candidate = first.apply_proposal(&proposal).unwrap();
assert_eq!(candidate.task("d").unwrap().dependencies, ["a"]);
assert_eq!(first.revision(), 2);
assert_eq!(first.coordinator().active_run_count(), 1);
assert_eq!(first.graph().task("d").unwrap().status, TaskStatus::Pending);
let committed = store.load_snapshot().unwrap().expect("committed snapshot");
assert_eq!(committed.revision, 2);
assert!(committed.graph.task("d").is_some());
assert_eq!(committed.coordinator.active_runs.len(), 1);
// The stale controller computes a candidate locally, but its old
// revision cannot overwrite the committed graph. Its in-memory state
// and the durable snapshot both remain unchanged after the conflict.
let stale_before = stale.snapshot();
let stale_proposal = GraphProposal::try_new(
[TaskProposal::try_new("e", "agent-e").unwrap()],
[GraphEdge::try_new("a", "e").unwrap()],
)
.unwrap();
let error = stale
.apply_proposal(&stale_proposal)
.expect_err("stale proposal must fail revision CAS");
assert!(matches!(error, OrchestrationError::RevisionConflict { .. }));
assert_eq!(stale.snapshot(), stale_before);
assert_eq!(store.load_snapshot().unwrap().unwrap(), committed);
// A proposal rejected before mutate also leaves the active run, graph,
// revision and durable snapshot untouched.
let invalid = GraphProposal::try_new(
[TaskProposal::try_new("f", "agent-d").unwrap()],
[GraphEdge::try_new("a", "b").unwrap()],
)
.unwrap();
assert!(first.apply_proposal(&invalid).is_err());
assert_eq!(first.revision(), committed.revision);
assert_eq!(first.coordinator().active_run_count(), 1);
assert_eq!(store.load_snapshot().unwrap().unwrap(), committed);
}
#[test]
fn persistent_file_store_recovers_graph_after_reopen() {
let path = test_temp_path("agent-runtime-orchestration");
let _ = fs::remove_file(&path);
{
let store = JsonFileOrchestrationSnapshotStore::new(&path).unwrap();
let mut controller = PersistentCoordinator::new(
graph(),
CoordinatorQuota::try_new(2, 1).unwrap(),
store,
)
.unwrap();
controller
.dispatch_wave([CoordinatorRunRequest::try_new("run-a", "a", "agent-a").unwrap()])
.unwrap();
}
let reopened = JsonFileOrchestrationSnapshotStore::new(&path).unwrap();
let restored = PersistentCoordinator::load(reopened)
.unwrap()
.expect("reopened snapshot");
assert_eq!(restored.revision(), 1);
assert_eq!(
restored.graph().task("a").unwrap().status,
TaskStatus::Running
);
assert_eq!(restored.coordinator().active_run_count(), 1);
let lock_path = orchestration_lock_path(&path);
let _ = fs::remove_file(&path);
// The orchestration store creates a same-directory advisory lock on
// the first write. Remove that sidecar with the fixture so repeated
// test runs do not accumulate control-plane artifacts in temp.
let _ = fs::remove_file(lock_path);
}
#[test]
fn persistent_file_store_serializes_cross_instance_revision_cas() {
let path = test_temp_path("agent-runtime-orchestration-lock");
let lock_path = orchestration_lock_path(&path);
let _ = fs::remove_file(&path);
let _ = fs::remove_file(&lock_path);
// 先写入 revision 0,两个独立 store 随后都基于同一旧值竞争 revision 1。
// sidecar 锁让其中一个完成 rename 后,另一个在临界区内重新读取到
// revision 1 并返回 CAS 冲突,而不是双双声称写入成功。
let seed = JsonFileOrchestrationSnapshotStore::new(&path).unwrap();
let initial = PersistentCoordinator::new(
graph(),
CoordinatorQuota::try_new(2, 1).unwrap(),
InMemoryOrchestrationSnapshotStore::new(),
)
.unwrap();
let snapshot = initial.snapshot();
seed.save_snapshot(None, &snapshot).unwrap();
let next = OrchestrationSnapshot {
revision: 1,
..snapshot.clone()
};
let path_a = path.clone();
let path_b = path.clone();
let next_a = next.clone();
let next_b = next;
let (first, second) = std::thread::scope(|scope| {
let a = scope.spawn(|| {
JsonFileOrchestrationSnapshotStore::new(path_a)
.unwrap()
.save_snapshot(Some(0), &next_a)
});
let b = scope.spawn(|| {
JsonFileOrchestrationSnapshotStore::new(path_b)
.unwrap()
.save_snapshot(Some(0), &next_b)
});
(a.join().unwrap(), b.join().unwrap())
});
let successes = (if first.is_ok() { 1 } else { 0 }) + (if second.is_ok() { 1 } else { 0 });
assert_eq!(successes, 1, "exactly one CAS writer may win");
assert!(matches!(
(first, second),
(Err(OrchestrationError::RevisionConflict { .. }), Ok(()))
| (Ok(()), Err(OrchestrationError::RevisionConflict { .. }))
));
assert_eq!(
JsonFileOrchestrationSnapshotStore::new(&path)
.unwrap()
.load_snapshot()
.unwrap()
.unwrap()
.revision,
1
);
let _ = fs::remove_file(path);
let _ = fs::remove_file(lock_path);
}
#[test]
fn persistent_isolation_moves_active_task_to_waiting_and_survives_reload() {
let store = InMemoryOrchestrationSnapshotStore::new();
let mut controller = PersistentCoordinator::new(
graph(),
CoordinatorQuota::try_new(2, 1).unwrap(),
store.clone(),
)
.unwrap();
controller
.dispatch_wave([CoordinatorRunRequest::try_new("run-a", "a", "agent-a").unwrap()])
.unwrap();
let released = controller
.isolate_node("a", "provider temporarily unavailable")
.unwrap();
assert_eq!(released, ["run-a"]);
assert_eq!(
controller.graph().task("a").unwrap().status,
TaskStatus::Waiting
);
assert_eq!(controller.coordinator().active_run_count(), 0);
assert!(controller.graph().ready_task_ids().is_empty());
// The persisted epoch must not resurrect a Running task without a run.
let mut restored = PersistentCoordinator::load(store)
.unwrap()
.expect("isolated snapshot should be recoverable");
assert_eq!(
restored.graph().task("a").unwrap().status,
TaskStatus::Waiting
);
assert_eq!(restored.coordinator().active_run_count(), 0);
assert!(matches!(
restored.graph().dependency_waves(),
Err(OrchestrationError::DependencyBlocked(_))
));
// Repair is the explicit recovery boundary: it clears quarantine and
// makes the seed task dispatchable in the next epoch.
assert_eq!(restored.repair_downstream(&["a"]).unwrap(), ["a", "b", "c"]);
assert_eq!(restored.graph().ready_task_ids(), ["a"]);
let started = restored
.dispatch_wave([CoordinatorRunRequest::try_new("run-a2", "a", "agent-a").unwrap()])
.unwrap();
assert_eq!(started[0].run_id, "run-a2");
assert_eq!(
restored.graph().task("a").unwrap().status,
TaskStatus::Running
);
}
#[test]
fn persistent_snapshot_rejects_active_run_graph_mismatch() {
let store = InMemoryOrchestrationSnapshotStore::new();
let graph = graph();
let coordinator = {
let mut coordinator = Coordinator::default();
coordinator.try_start_run("run-a", "a", "agent-a").unwrap();
coordinator.snapshot()
};
let snapshot = OrchestrationSnapshot {
schema_version: ORCHESTRATION_SNAPSHOT_SCHEMA_VERSION.to_owned(),
revision: 0,
graph,
coordinator,
};
assert!(matches!(
store.save_snapshot(None, &snapshot),
Err(OrchestrationError::InvalidInput(message))
if message.contains("不是 running")
));
assert!(store.load_snapshot().unwrap().is_none());
}
#[test]
fn persistent_snapshot_rejects_running_task_without_active_run() {
let store = InMemoryOrchestrationSnapshotStore::new();
let mut graph = graph();
graph.set_task_status("a", TaskStatus::Running).unwrap();
let snapshot = OrchestrationSnapshot {
schema_version: ORCHESTRATION_SNAPSHOT_SCHEMA_VERSION.to_owned(),
revision: 0,
graph,
coordinator: Coordinator::default().snapshot(),
};
// 反向不变量同样是持久化合同的一部分,避免恢复后 Running 永远卡在
// 图中却没有可完成、可释放的活动 run。
assert!(matches!(
store.save_snapshot(None, &snapshot),
Err(OrchestrationError::InvalidInput(message))
if message.contains("running 任务 a 没有对应活动 run")
));
assert!(store.load_snapshot().unwrap().is_none());
assert!(matches!(
PersistentCoordinator::from_snapshot(snapshot, store),
Err(OrchestrationError::InvalidInput(message))
if message.contains("running 任务 a 没有对应活动 run")
));
}
#[test]
fn ready_plan_is_bounded_deterministic_and_claims_in_one_epoch() {
let store = InMemoryOrchestrationSnapshotStore::new();
let mut controller = PersistentCoordinator::new(
graph(),
CoordinatorQuota::try_new(2, 1).unwrap(),
store.clone(),
)
.unwrap();
let plan = controller.plan_ready_tasks(8).unwrap();
assert_eq!(plan.expected_revision, 0);
assert_eq!(
plan.candidates
.iter()
.map(|candidate| candidate.task_id.as_str())
.collect::<Vec<_>>(),
["a"]
);
let started = controller.claim_ready_tasks(&plan, ["run-a"]).unwrap();
assert_eq!(started[0].task_id, "a");
assert_eq!(controller.revision(), 1);
assert_eq!(
controller.graph().task("a").unwrap().status,
TaskStatus::Running
);
assert_eq!(
PersistentCoordinator::load(store)
.unwrap()
.unwrap()
.revision(),
1
);
// 同一计划不能被重复领取;revision CAS 在任何图/配额变更前拒绝它。
assert!(matches!(
controller.claim_ready_tasks(&plan, ["run-a-retry"]),
Err(OrchestrationError::RevisionConflict {
expected: Some(0),
actual: Some(1)
})
));
controller.complete_run("run-a").unwrap();
let next_plan = controller.plan_ready_tasks(2).unwrap();
assert_eq!(
next_plan
.candidates
.iter()
.map(|candidate| candidate.task_id.as_str())
.collect::<Vec<_>>(),
["b", "c"]
);
let started = controller
.claim_ready_tasks(&next_plan, ["run-b", "run-c"])
.unwrap();
assert_eq!(
started
.iter()
.map(|run| run.run_id.as_str())
.collect::<Vec<_>>(),
["run-b", "run-c"]
);
}
#[test]
fn ready_plan_honors_isolation_and_per_agent_quota() {
let graph = TaskGraph::try_new(
"goal",
[
TaskNode::try_new("a", "agent-a", std::iter::empty::<String>()).unwrap(),
TaskNode::try_new("b", "agent-a", std::iter::empty::<String>()).unwrap(),
TaskNode::try_new("c", "agent-b", std::iter::empty::<String>()).unwrap(),
],
)
.unwrap();
let store = InMemoryOrchestrationSnapshotStore::new();
let mut controller =
PersistentCoordinator::new(graph, CoordinatorQuota::try_new(3, 1).unwrap(), store)
.unwrap();
// 同一个 Agent 的第二个 ready 任务被跳过,但其它 Agent 仍可入计划。
let plan = controller.plan_ready_tasks(3).unwrap();
assert_eq!(
plan.candidates
.iter()
.map(|candidate| candidate.task_id.as_str())
.collect::<Vec<_>>(),
["a", "c"]
);
// 隔离 a 后释放其配额;隔离节点本身不会再次进入 ready 计划,b 可以
// 在下一个 revision 中使用 agent-a 的空闲配额。
controller
.isolate_node("a", "agent-a temporarily unavailable")
.unwrap();
let plan = controller.plan_ready_tasks(3).unwrap();
assert_eq!(
plan.candidates
.iter()
.map(|candidate| candidate.task_id.as_str())
.collect::<Vec<_>>(),
["b", "c"]
);
assert!(
plan.candidates
.iter()
.all(|candidate| candidate.task_id != "a")
);
}
#[test]
fn ready_plan_claim_rejects_stale_or_mismatched_run_ids_without_state_leak() {
let store = InMemoryOrchestrationSnapshotStore::new();
let mut controller =
PersistentCoordinator::new(graph(), CoordinatorQuota::try_new(2, 1).unwrap(), store)
.unwrap();
let plan = controller.plan_ready_tasks(1).unwrap();
controller
.deliver_message(CoordinatorMessage::try_text("m-1", "a", "b", "advance").unwrap())
.unwrap();
let before = controller.snapshot();
assert!(matches!(
controller.claim_ready_tasks(&plan, ["run-a"]),
Err(OrchestrationError::RevisionConflict {
expected: Some(0),
actual: Some(1)
})
));
assert_eq!(controller.snapshot(), before);
let fresh = controller.plan_ready_tasks(1).unwrap();
let before = controller.snapshot();
assert!(matches!(
controller.claim_ready_tasks(&fresh, std::iter::empty::<String>()),
Err(OrchestrationError::InvalidInput(message))
if message.contains("run ID 数量不一致")
));
assert_eq!(controller.snapshot(), before);
assert!(matches!(
controller.claim_ready_tasks(&fresh, ["run-a", "run-extra"]),
Err(OrchestrationError::InvalidInput(message))
if message.contains("run ID 数量不一致")
));
assert_eq!(controller.snapshot(), before);
}
#[test]
fn ready_plan_constructor_and_json_deserialization_reapply_validation() {
assert!(matches!(
ReadyTaskDispatchPlan::try_new(
0,
[
ReadyTaskCandidate::try_new("a", "agent").unwrap(),
ReadyTaskCandidate::try_new("a", "agent").unwrap(),
],
),
Err(OrchestrationError::DuplicateTask(id)) if id == "a"
));
assert!(
serde_json::from_str::<ReadyTaskDispatchPlan>(
r#"{"expectedRevision":0,"candidates":[{"taskId":"","agentId":"agent"}]}"#
)
.is_err()
);
assert!(serde_json::from_str::<ReadyTaskDispatchPlan>(
r#"{"expectedRevision":0,"candidates":[{"taskId":"a","agentId":"agent"},{"taskId":"a","agentId":"agent"}]}"#
)
.is_err());
assert!(matches!(
ReadyTaskDispatchPlan::try_new(0, std::iter::empty::<ReadyTaskCandidate>())
.unwrap()
.candidates,
candidates if candidates.is_empty()
));
}
}