From 1104215b2af07b1c7ac48615b09d51c25f9ade02 Mon Sep 17 00:00:00 2001 From: kdletters Date: Wed, 26 Aug 2026 17:50:42 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E6=8A=BD=E5=8F=96=E9=80=9A=E7=94=A8?= =?UTF-8?q?=E5=A4=9A=20Agent=20DAG=20=E7=BC=96=E6=8E=92=20crate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增独立 agent-runtime-orchestration crate，统一任务图校验、ready、依赖波次和返工闭包。 接入 platform-agent 与 AGC Runtime catalog，保留游戏领域任务语义和现有持久化格式。 补充非游戏 conformance 测试、根检查脚本及架构文档。 --- .../src-tauri/Cargo.lock | 10 + .../src/agent/generation/pass_artifacts.rs | 3 +- .../src-tauri/src/agent/runtime_adapter.rs | 9 +- .../planning_strategy/repair_strategy.rs | 2 +- .../shared-memory/decision-log.md | 8 + .../shared-memory/project-overview.md | 1 + ...案】AI游戏创作Agent Runtime V1.1-2026-07-12.md | 30 ++ package.json | 3 +- server-rs/Cargo.toml | 1 + .../agent-runtime-orchestration/.gitignore | 2 + .../agent-runtime-orchestration/Cargo.toml | 13 + .../agent-runtime-orchestration/src/error.rs | 46 ++ .../agent-runtime-orchestration/src/graph.rs | 419 ++++++++++++++++++ .../agent-runtime-orchestration/src/lib.rs | 13 + .../agent-runtime-orchestration/src/plan.rs | 64 +++ .../tests/non_game_orchestration.rs | 154 +++++++ server-rs/crates/platform-agent/Cargo.toml | 2 + .../platform-agent/src/game_creation.rs | 303 +++++++------ server-rs/crates/platform-agent/src/lib.rs | 1 + 19 files changed, 926 insertions(+), 158 deletions(-) create mode 100644 server-rs/crates/agent-runtime-orchestration/.gitignore create mode 100644 server-rs/crates/agent-runtime-orchestration/Cargo.toml create mode 100644 server-rs/crates/agent-runtime-orchestration/src/error.rs create mode 100644 server-rs/crates/agent-runtime-orchestration/src/graph.rs create mode 100644 server-rs/crates/agent-runtime-orchestration/src/lib.rs create mode 100644 server-rs/crates/agent-runtime-orchestration/src/plan.rs create mode 100644 server-rs/crates/agent-runtime-orchestration/tests/non_game_orchestration.rs diff --git a/apps/ai-game-creator-shell/src-tauri/Cargo.lock b/apps/ai-game-creator-shell/src-tauri/Cargo.lock index 52f92e2e5..032383d0f 100644 --- a/apps/ai-game-creator-shell/src-tauri/Cargo.lock +++ b/apps/ai-game-creator-shell/src-tauri/Cargo.lock @@ -16,6 +16,14 @@ dependencies = [ "serde_json", ] +[[package]] +name = "agent-runtime-orchestration" +version = "0.1.0" +dependencies = [ + "agent-runtime-core", + "serde", +] + [[package]] name = "ahash" version = "0.8.12" @@ -3772,6 +3780,8 @@ checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" name = "platform-agent" version = "0.1.0" dependencies = [ + "agent-runtime-core", + "agent-runtime-orchestration", "platform-llm", "serde", "serde_json", diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/generation/pass_artifacts.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/generation/pass_artifacts.rs index bdc1046b5..979ee4ddb 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/generation/pass_artifacts.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/generation/pass_artifacts.rs @@ -7,7 +7,8 @@ pub(crate) fn write_agent_pass_agenda( ) -> Result { let graph = build_game_creation_seed_task_graph("AI 游戏创作") .map_err(|error| format!("构建 Agent 编排任务图失败：{error}"))?; - let pass_plan = plan_game_creation_agent_pass(&graph, pass, findings_markdown); + let pass_plan = plan_game_creation_agent_pass(&graph, pass, findings_markdown) + .map_err(|error| format!("规划 Agent 编排任务图失败：{error}"))?; let repair_routes = pass_plan .repair_routes .iter() diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_adapter.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_adapter.rs index b7360c185..0c9d72039 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_adapter.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_adapter.rs @@ -52,8 +52,13 @@ fn build_game_creator_runtime_agent_catalog() -> Result { ); } } - AgentCatalog::try_new(agents) - .map_err(|error| format!("AI 游戏创作 Agent catalog 无效：{error}")) + let catalog = AgentCatalog::try_new(agents) + .map_err(|error| format!("AI 游戏创作 Agent catalog 无效：{error}"))?; + let task_graph = build_game_creation_seed_task_graph("AI 游戏创作 Agent catalog 验证") + .map_err(|error| format!("AI 游戏创作任务图无效：{error}"))?; + platform_agent::validate_game_creation_task_agents(&task_graph, &catalog) + .map_err(|error| format!("AI 游戏创作 Agent catalog 与任务图不一致：{error}"))?; + Ok(catalog) } pub(crate) fn game_creator_runtime_agent_catalog() -> Result<&'static AgentCatalog, String> { diff --git a/apps/ai-game-creator-shell/src-tauri/src/tests/runtime_actions/planning_strategy/repair_strategy.rs b/apps/ai-game-creator-shell/src-tauri/src/tests/runtime_actions/planning_strategy/repair_strategy.rs index 25525d409..eeaeb7bc2 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/tests/runtime_actions/planning_strategy/repair_strategy.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/tests/runtime_actions/planning_strategy/repair_strategy.rs @@ -515,7 +515,7 @@ fn evaluator_findings_include_structured_repair_routes() { assert!(findings.contains("\"code-prototype\"")); let graph = build_game_creation_seed_task_graph("像素厨房弹幕").expect("task graph"); - let plan = plan_game_creation_agent_pass(&graph, 2, &findings); + let plan = plan_game_creation_agent_pass(&graph, 2, &findings).expect("repair plan"); assert_eq!(plan.mode, "repair"); assert!(plan.active_task_ids.contains(&"code-prototype".to_string())); assert!(plan diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index 4ffd04577..df5655f04 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -16,6 +16,14 @@ --- +## 2026-08-26 通用多 Agent DAG 编排与执行内核分层 + +- 背景：`agent-runtime-core` 已承接 catalog、run/action 生命周期、lane、宿主 ToolHost、spawn/all-join 和 Provider 契约，但动态任务图的 ready 选择、依赖波次与返工下游闭包仍混在 `platform-agent::game_creation`，其它产品无法复用且非法环会被合并成伪 wave。 +- 决策：新增纯 Rust `agent-runtime-orchestration`，依赖方向固定为 `agent-runtime-orchestration -> agent-runtime-core`。公共层只持有任务 ID、Agent ID、通用状态和依赖边，统一负责构图校验、ready、active/satisfied 波次、下游闭包和全量/返工选择；动态构图仍必须是 DAG，跨轮循环通过新的 pass / epoch 表达。 +- 产品边界：16 个游戏任务、六组角色、产物/验收条件、Evaluator Markdown 和中文语义路由继续留在 `platform-agent`；AGC 组合根使用公共层校验任务图与 `AgentCatalog`。Runtime store、Runner、Provider、权限、ToolHost、委派 journal、isolated write scope 和 `.agent/runtime/**` 不迁移、不双写。 +- 验证方式：非游戏 conformance 覆盖并行分支、汇合、repair closure、AgentCatalog 和非法图失败关闭；`platform-agent` 锁定种子 DAG 与现役波次/返工顺序，并验证环拒绝和 catalog 注入。根检查脚本必须执行新 crate 测试。 +- 关联文档：`docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md` V1.54。 + ## 2026-08-24 AGC Direct 媒体能力只通过客户端语义工具开放 - 背景：资源页已经补齐视频、角色动画、音效和背景音乐的 create/derive 能力，但 Direct Codex 只能准备标准美术包，无法查询已登记源资源或表达新增媒体意图。直接开放 Tauri invoke 会把项目路径、revision、operation、幂等键、登录态和事务权力交给模型。 diff --git a/docs/project-memory/shared-memory/project-overview.md b/docs/project-memory/shared-memory/project-overview.md index 58bcece5f..0ca84ed62 100644 --- a/docs/project-memory/shared-memory/project-overview.md +++ b/docs/project-memory/shared-memory/project-overview.md @@ -51,6 +51,7 @@ SpacetimeDB crate、SDK、CLI / standalone 与生成 bindings 按 `2.7.0` 对齐 ## AGC DirectProject 与 UI workflow +- 通用 Agent Rust 分层为 `agent-runtime-core`（catalog、执行生命周期、ToolHost/spawn/all-join/Provider 契约）、`agent-runtime-orchestration`（动态无环任务图、ready、依赖波次和返工下游闭包）与 `platform-agent` 游戏适配器；循环返工通过新 pass / epoch 表达，不在单张依赖图中建立回边。 - DirectProject 只连接客户端内置的 `agc_tools` STDIO MCP。它负责审核引用读取、标准美术准备、已登记资源有界查询、视频 / 角色动画 / 音效 / BGM 的 create-or-derive、已登记图片去背景、desktop/mobile 浏览器试玩和受控 `agc_web_search`；付费资源调用由客户端绑定回合、幂等账本、请求上限和投影权威。 - DirectProject 的 Codex 原生文件、搜索、命令、图片查看和 Skill 仅在真实 `game/` cwd 与 `workspaceWrite(writableRoots=[game])` 内可用；原生命令网络保持关闭。多 Agent、Apps、插件、hooks、图片生成、Goals、Workspace Dependencies、Tool Suggestion 和原生浏览器/电脑控制保持关闭。app-server 使用隔离 `CODEX_HOME`，provider 凭据只由 AGC 客户端代理持有，不能进入模型上下文或 shell 环境。 - `ui-prototype`（设计图片）与 UI 编辑器 `UI` JSON 是不同资源。白名单 `ui.workflow.run` 按页面执行 `prepare → recognize → status → finalize`，由 provider-backed 识别、合并和组件绑定持久化 State/revision，并把 `reference-ready → structure-ready → merge-ready → binding-ready → application-ready → completed` 投影到 manifest。Provider 缺失、请求失败、工具缺失、结果不匹配或仍有待审节点时保留真实阶段并返回 blocker，不得用 deterministic seed 伪造完成。 diff --git a/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md b/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md index 4616fd53a..38ac61005 100644 --- a/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md +++ b/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md @@ -1649,6 +1649,36 @@ V1.53 把根 Project Supervisor 的 same-run steer 从“收到消息立即中 - LLM 判定、解析或持久化失败时写入关联的非终态 fallback 回复，保持当前任务运行，并在下一安全边界应用已排队 steer；失败不能退回“默认中断”。判定与回复按 `agentId / runId / steerId` 幂等，冲突终态失败关闭。 - 验收必须覆盖判定 LLM 的 `true / false` 协议、公开回复幂等、入队本身不中断、`runtime.steer` 有活动 Provider 时仍不中断、缺失 decision 拒绝条件中断、`false` decision 不中断、`true` decision 只中断旧 cursor，以及判定失败后同一 run 继续。 +## V1.54 通用多 Agent DAG 编排 crate + +V1.48-V1.50 已把 catalog、执行生命周期与 Provider 契约收进 `agent-runtime-core`，但动态任务图的 ready 选择、依赖波次和返工下游闭包仍编译在 `platform-agent::game_creation`。V1.54 新增独立 `server-rs/crates/agent-runtime-orchestration`（package name `agent-runtime-orchestration`），作为 `agent-runtime-core` 上方的纯 Rust 编排层；它不复制 Runtime store、Runner、Provider、ToolHost、delegation journal、权限或产品持久文件。 + +### 图模型与无环合同 + +- 公共层只保存任务 ID、执行 Agent ID、通用状态和依赖边，不保存游戏组别、角色文案、产物路径、Evaluator Markdown 或中文关键词。任务在运行时注册，因此“动态 DAG”表示宿主可以按目标或 pass 动态构图与重规划，不表示依赖边可以形成环。 +- 构图一次性失败关闭：拒绝空或非法 ID、重复任务、重复依赖、未知依赖、自依赖和任意有向环，并保持任务注册顺序作为所有确定性输出的稳定顺序。环内没有可证明的首个 ready 节点，也无法给 completion、重放和下游失效定义单调顺序；需要迭代时由宿主建立新的 pass / epoch，并显式携带上一轮结果，不在同一依赖图里回边。 +- 编排层提供 pending ready 选择、active/satisfied 依赖波次、指定任务的下游影响闭包和全量/返工选择计划。active 任务依赖的非 active 节点必须显式位于 satisfied 集合；缺失前置不能按“图外即完成”静默放行。 +- 每个任务携带 `agentId`，并可对 V1.48 `AgentCatalog` 做引用校验；未知 Agent 在创建任何 run 或调用宿主前失败关闭。crate 依赖方向固定为 `agent-runtime-orchestration -> agent-runtime-core`，依赖闭包只使用标准库与 `serde / serde_json`，不得反向依赖 AGC、Tauri、`platform-agent` 或 `platform-llm`。 + +### AGC 生产适配与兼容 + +- `platform-agent` 继续拥有 16 个游戏任务、六组枚举、标题/角色/产物/验收条件、Evaluator Markdown 解析和游戏关键词路由；它把现有 `GameCreationTaskGraph` 映射为公共任务图，并由公共层计算 ready、dependency waves 和 repair downstream closure。 +- `GameCreationTaskGraph`、`GameCreationAgentPassPlan` 的 serde 字段、种子任务、有效 DAG 的顺序与现役 `.agent/passes/pass-N/task-graph.json` 输出保持不变。构图或计划函数改为显式返回错误，非法环、未知依赖或不完整 partition 不再合并成一个伪 wave。 +- AGC 的现役 Agent catalog 构造同时用公共编排层校验 16 个任务的 Agent 引用，证明新 crate 已进入生产组合根。isolated child 的项目路径、write scope、证据和深度限制仍是 AGC 工具/策略合同；通用 spawn/all-join 继续由 `agent-runtime-core` 执行，本轮不再造第二套委派协议。 + +### V1.54 验收 + +- 新 crate 的非游戏 conformance fixture 在运行时构造并行分支与汇合节点，覆盖稳定 ready 集合、依赖波次、返工下游闭包、AgentCatalog 引用，以及重复/未知/自依赖/有环/缺失 satisfied 的失败关闭；fixture 不得出现游戏任务名、AGC 路径或 Tauri 类型。 +- `platform-agent` 回归锁定 16 任务种子图、首轮全量波次、结构化返工和下游扩展的现有顺序，并增加非法游戏图不会进入计划的负向用例；AGC adapter 回归锁定 catalog 与任务图一致。 +- 根脚本增加 `agent-runtime-orchestration:check`，并纳入 `ai-game-creator-shell:check`。完成后至少运行新 crate、`platform-agent`、AGC adapter/生成编排定向测试、Tauri `cargo check --tests`、依赖树、`npm run check:encoding` 和 `git diff --check`；独立 crate 产生的本地 `Cargo.lock/target` 不进入提交。 + +### V1.54 本轮验证记录（2026-08-26） + +- 已通过：`agent-runtime-core` 20 项、`agent-runtime-orchestration` 5 项、`platform-agent` 19 项；新 crate 依赖树仅引入 `agent-runtime-core` 与 `serde`（测试专用 `serde_json`），`cargo fmt --check`、`npm run check:encoding` 和 `git diff --check` 均通过。 +- 已通过：AGC 任务图与注入 `AgentCatalog` 的一致性测试、非法环失败关闭、反序列化重复依赖失败关闭；测试产生的独立 crate `Cargo.lock/target` 已清理。 +- 未完成：Tauri `cargo check --tests` 已编译到 AGC 自定义 `build.rs`，随后因仓库四个候选路径均缺少内置 Codex CLI vendor 资源而退出（`build.rs:77`）；本轮未执行 `npm ci`，因此不能将 Tauri 组合根或完整 `ai-game-creator-shell:check` 记为通过。 +- 未执行：当前 Rust 1.96 工具链未安装 `clippy` component；没有把该静态检查结果用其它门禁结果替代。 + ## 验收命令 - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml structured_plan_ -- --nocapture` diff --git a/package.json b/package.json index a3bec46ba..f9844e1cd 100644 --- a/package.json +++ b/package.json @@ -187,8 +187,9 @@ "ai-game-creator-shell:agent-runtime:steer-real-e2e": "npm --prefix apps/ai-game-creator-shell run agent-runtime:steer-real-e2e --", "ai-game-creator-shell:agent-runtime:steer-runner-kill-real-e2e": "npm --prefix apps/ai-game-creator-shell run agent-runtime:steer-runner-kill-real-e2e --", "agent-runtime-core:check": "cargo test --manifest-path server-rs/crates/agent-runtime-core/Cargo.toml", + "agent-runtime-orchestration:check": "cargo test --manifest-path server-rs/crates/agent-runtime-orchestration/Cargo.toml", "ai-game-creator-shell:typecheck": "npm --prefix apps/ai-game-creator-shell run typecheck", - "ai-game-creator-shell:check": "npm run ai-game-creator-shell:typecheck && npm run test -- apps/ai-game-creator-shell/tests && npm run agent-runtime-core:check && cargo test --locked -p platform-llm --manifest-path server-rs/Cargo.toml && cargo test --locked -p shared-contracts --manifest-path server-rs/Cargo.toml game_creation_app && cargo test --locked --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml -- --test-threads=1 && npm run ai-game-creator-shell:agent-run:smoke", + "ai-game-creator-shell:check": "npm run ai-game-creator-shell:typecheck && npm run test -- apps/ai-game-creator-shell/tests && npm run agent-runtime-core:check && npm run agent-runtime-orchestration:check && cargo test --locked -p platform-llm --manifest-path server-rs/Cargo.toml && cargo test --locked -p shared-contracts --manifest-path server-rs/Cargo.toml game_creation_app && cargo test --locked --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml -- --test-threads=1 && npm run ai-game-creator-shell:agent-run:smoke", "check:native-shells": "node scripts/check-native-shells.mjs" }, "dependencies": { diff --git a/server-rs/Cargo.toml b/server-rs/Cargo.toml index 6ef1cca89..00a37db27 100644 --- a/server-rs/Cargo.toml +++ b/server-rs/Cargo.toml @@ -8,6 +8,7 @@ default-members = [ ] exclude = [ "crates/agent-runtime-core", + "crates/agent-runtime-orchestration", "crates/module-bark-battle", "crates/module-big-fish", "crates/module-combat", diff --git a/server-rs/crates/agent-runtime-orchestration/.gitignore b/server-rs/crates/agent-runtime-orchestration/.gitignore new file mode 100644 index 000000000..042776aad --- /dev/null +++ b/server-rs/crates/agent-runtime-orchestration/.gitignore @@ -0,0 +1,2 @@ +/Cargo.lock +/target/ diff --git a/server-rs/crates/agent-runtime-orchestration/Cargo.toml b/server-rs/crates/agent-runtime-orchestration/Cargo.toml new file mode 100644 index 000000000..9a35ff007 --- /dev/null +++ b/server-rs/crates/agent-runtime-orchestration/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "agent-runtime-orchestration" +edition = "2024" +version = "0.1.0" +license = "UNLICENSED" +publish = false + +[dependencies] +agent-runtime-core = { path = "../agent-runtime-core" } +serde = { version = "1", features = ["derive"] } + +[dev-dependencies] +serde_json = "1" diff --git a/server-rs/crates/agent-runtime-orchestration/src/error.rs b/server-rs/crates/agent-runtime-orchestration/src/error.rs new file mode 100644 index 000000000..25af0518f --- /dev/null +++ b/server-rs/crates/agent-runtime-orchestration/src/error.rs @@ -0,0 +1,46 @@ +use std::fmt; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum OrchestrationErrorKind { + InvalidInput, + DuplicateTask, + DuplicateDependency, + UnknownDependency, + SelfDependency, + Cycle, + UnknownAgent, + UnknownTask, + ConflictingTaskSet, + UnsatisfiedDependency, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct OrchestrationError { + kind: OrchestrationErrorKind, + detail: String, +} + +impl OrchestrationError { + pub(crate) fn new(kind: OrchestrationErrorKind, detail: impl Into) -> Self { + Self { + kind, + detail: detail.into(), + } + } + + pub fn kind(&self) -> OrchestrationErrorKind { + self.kind + } + + pub fn detail(&self) -> &str { + &self.detail + } +} + +impl fmt::Display for OrchestrationError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.detail) + } +} + +impl std::error::Error for OrchestrationError {} diff --git a/server-rs/crates/agent-runtime-orchestration/src/graph.rs b/server-rs/crates/agent-runtime-orchestration/src/graph.rs new file mode 100644 index 000000000..512610462 --- /dev/null +++ b/server-rs/crates/agent-runtime-orchestration/src/graph.rs @@ -0,0 +1,419 @@ +use std::collections::{BTreeMap, BTreeSet, HashSet, VecDeque}; + +use agent_runtime_core::AgentCatalog; +use serde::{Deserialize, Serialize}; + +use crate::{OrchestrationError, OrchestrationErrorKind}; + +const IDENTIFIER_MAX_CHARS: usize = 128; +const GOAL_MAX_CHARS: usize = 4_000; + +#[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 { + id: String, + agent_id: String, + status: TaskStatus, + dependencies: Vec, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct TaskNodeInput { + id: String, + agent_id: String, + status: TaskStatus, + dependencies: Vec, +} + +impl<'de> Deserialize<'de> for TaskNode { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let input = TaskNodeInput::deserialize(deserializer)?; + Self::try_new(input.id, input.agent_id, input.status, input.dependencies) + .map_err(serde::de::Error::custom) + } +} + +impl TaskNode { + pub fn try_new( + id: impl Into, + agent_id: impl Into, + status: TaskStatus, + dependencies: impl IntoIterator>, + ) -> Result { + let id = id.into(); + let agent_id = agent_id.into(); + validate_identifier(&id, "task id")?; + validate_identifier(&agent_id, "task agent id")?; + + let mut seen = BTreeSet::new(); + let mut dependencies_output = Vec::new(); + for dependency in dependencies { + let dependency = dependency.into(); + validate_identifier(&dependency, "task dependency")?; + if dependency == id { + return Err(OrchestrationError::new( + OrchestrationErrorKind::SelfDependency, + format!("task {id} 不能依赖自身"), + )); + } + if !seen.insert(dependency.clone()) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DuplicateDependency, + format!("task {id} 重复依赖：{dependency}"), + )); + } + dependencies_output.push(dependency); + } + + Ok(Self { + id, + agent_id, + status, + dependencies: dependencies_output, + }) + } + + pub fn id(&self) -> &str { + &self.id + } + + pub fn agent_id(&self) -> &str { + &self.agent_id + } + + pub fn status(&self) -> TaskStatus { + self.status + } + + pub fn dependencies(&self) -> &[String] { + &self.dependencies + } +} + +#[derive(Clone, Debug)] +pub struct TaskGraph { + goal: String, + tasks: Vec, + by_id: BTreeMap, +} + +impl TaskGraph { + pub fn try_new( + goal: impl Into, + tasks: impl IntoIterator, + ) -> Result { + let goal = goal.into(); + validate_goal(&goal)?; + let tasks = tasks.into_iter().collect::>(); + if tasks.is_empty() { + return Err(OrchestrationError::new( + OrchestrationErrorKind::InvalidInput, + "task graph 至少需要一个任务", + )); + } + + let mut by_id = BTreeMap::new(); + for (index, task) in tasks.iter().enumerate() { + if by_id.insert(task.id.clone(), index).is_some() { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DuplicateTask, + format!("task id 重复：{}", task.id), + )); + } + } + for task in &tasks { + for dependency in &task.dependencies { + if !by_id.contains_key(dependency) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::UnknownDependency, + format!("task {} 引用了未知依赖：{dependency}", task.id), + )); + } + } + } + + validate_acyclic(&tasks, &by_id)?; + Ok(Self { goal, tasks, by_id }) + } + + pub fn goal(&self) -> &str { + &self.goal + } + + pub fn tasks(&self) -> &[TaskNode] { + &self.tasks + } + + pub fn get(&self, task_id: &str) -> Option<&TaskNode> { + self.by_id + .get(task_id) + .and_then(|index| self.tasks.get(*index)) + } + + 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::new( + OrchestrationErrorKind::UnknownAgent, + format!("task {} 引用了未注册 Agent：{}", task.id, task.agent_id), + )); + } + } + Ok(()) + } + + 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::>(); + + 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() + } + + pub fn expand_downstream>( + &self, + task_ids: &[T], + ) -> Result, OrchestrationError> { + let seeds = self.collect_known_task_ids(task_ids, "downstream seeds")?; + if seeds.is_empty() { + return Ok(Vec::new()); + } + let mut impacted = seeds; + let mut changed = true; + while changed { + changed = false; + for task in &self.tasks { + if impacted.contains(&task.id) { + continue; + } + if 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()) + } + + pub fn dependency_waves, S: AsRef>( + &self, + active_task_ids: &[A], + satisfied_task_ids: &[S], + ) -> Result>, OrchestrationError> { + let active = self.collect_known_task_ids(active_task_ids, "active tasks")?; + let satisfied = self.collect_known_task_ids(satisfied_task_ids, "satisfied tasks")?; + if let Some(task_id) = active.iter().find(|task_id| satisfied.contains(*task_id)) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::ConflictingTaskSet, + format!("task 同时位于 active 与 satisfied：{task_id}"), + )); + } + + for task in self.tasks.iter().filter(|task| active.contains(&task.id)) { + for dependency in &task.dependencies { + if !active.contains(dependency) && !satisfied.contains(dependency) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::UnsatisfiedDependency, + format!( + "active task {} 的依赖既未 active 也未 satisfied：{dependency}", + task.id + ), + )); + } + } + } + + let mut remaining = self + .tasks + .iter() + .filter(|task| active.contains(&task.id)) + .map(|task| task.id.clone()) + .collect::>(); + let mut completed = satisfied; + let mut waves = Vec::new(); + while !remaining.is_empty() { + let wave = remaining + .iter() + .filter(|task_id| { + self.get(task_id).is_some_and(|task| { + task.dependencies + .iter() + .all(|dependency| completed.contains(dependency)) + }) + }) + .cloned() + .collect::>(); + if wave.is_empty() { + return Err(OrchestrationError::new( + OrchestrationErrorKind::Cycle, + format!( + "active task graph 无法生成下一依赖波次：{}", + remaining.join(", ") + ), + )); + } + for task_id in &wave { + completed.insert(task_id.clone()); + } + remaining.retain(|task_id| !completed.contains(task_id)); + waves.push(wave); + } + Ok(waves) + } + + pub(crate) fn all_task_ids(&self) -> Vec { + self.tasks.iter().map(|task| task.id.clone()).collect() + } + + fn collect_known_task_ids>( + &self, + task_ids: &[T], + label: &str, + ) -> Result, OrchestrationError> { + let mut output = BTreeSet::new(); + for task_id in task_ids { + let task_id = task_id.as_ref(); + if self.get(task_id).is_none() { + return Err(OrchestrationError::new( + OrchestrationErrorKind::UnknownTask, + format!("{label} 包含未知 task：{task_id}"), + )); + } + if !output.insert(task_id.to_string()) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DuplicateTask, + format!("{label} 包含重复 task：{task_id}"), + )); + } + } + Ok(output) + } +} + +fn validate_identifier(value: &str, field: &str) -> Result<(), OrchestrationError> { + if value != value.trim() { + return Err(OrchestrationError::new( + OrchestrationErrorKind::InvalidInput, + format!("{field} 不得包含首尾空白"), + )); + } + let mut chars = value.chars(); + let first = chars.next().ok_or_else(|| { + OrchestrationError::new( + OrchestrationErrorKind::InvalidInput, + format!("{field} 不能为空"), + ) + })?; + if value.chars().count() > IDENTIFIER_MAX_CHARS + || !first.is_ascii_alphanumeric() + || !chars.all(|character| { + character.is_ascii_alphanumeric() || matches!(character, '.' | '_' | '-' | ':') + }) + { + return Err(OrchestrationError::new( + OrchestrationErrorKind::InvalidInput, + format!("{field} 不是合法稳定标识：{value}"), + )); + } + Ok(()) +} + +fn validate_goal(goal: &str) -> Result<(), OrchestrationError> { + if goal != goal.trim() || goal.is_empty() || goal.chars().count() > GOAL_MAX_CHARS { + return Err(OrchestrationError::new( + OrchestrationErrorKind::InvalidInput, + format!("task graph goal 必须为 1..={GOAL_MAX_CHARS} 个无首尾空白字符"), + )); + } + if goal.chars().any(char::is_control) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::InvalidInput, + "task graph goal 不能包含控制字符", + )); + } + Ok(()) +} + +fn validate_acyclic( + tasks: &[TaskNode], + by_id: &BTreeMap, +) -> Result<(), OrchestrationError> { + let mut indegrees = tasks + .iter() + .map(|task| task.dependencies.len()) + .collect::>(); + let mut dependents = vec![Vec::::new(); tasks.len()]; + for (task_index, task) in tasks.iter().enumerate() { + for dependency in &task.dependencies { + let dependency_index = by_id[dependency]; + dependents[dependency_index].push(task_index); + } + } + + let mut ready = indegrees + .iter() + .enumerate() + .filter_map(|(index, indegree)| (*indegree == 0).then_some(index)) + .collect::>(); + let mut visited = 0; + while let Some(index) = ready.pop_front() { + visited += 1; + for dependent in &dependents[index] { + indegrees[*dependent] -= 1; + if indegrees[*dependent] == 0 { + ready.push_back(*dependent); + } + } + } + if visited == tasks.len() { + return Ok(()); + } + + let cyclic = tasks + .iter() + .zip(indegrees) + .filter_map(|(task, indegree)| (indegree > 0).then_some(task.id.as_str())) + .collect::>(); + Err(OrchestrationError::new( + OrchestrationErrorKind::Cycle, + format!("task graph 包含依赖环：{}", cyclic.join(", ")), + )) +} diff --git a/server-rs/crates/agent-runtime-orchestration/src/lib.rs b/server-rs/crates/agent-runtime-orchestration/src/lib.rs new file mode 100644 index 000000000..ecf418b75 --- /dev/null +++ b/server-rs/crates/agent-runtime-orchestration/src/lib.rs @@ -0,0 +1,13 @@ +//! Deterministic task-graph orchestration layered over `agent-runtime-core`. +//! +//! Hosts register task graphs at runtime. This crate validates the graph and +//! computes ready tasks, dependency waves and downstream repair impact without +//! owning persistence, threads, providers, tools or product-specific policy. + +mod error; +mod graph; +mod plan; + +pub use error::{OrchestrationError, OrchestrationErrorKind}; +pub use graph::{TaskGraph, TaskNode, TaskStatus}; +pub use plan::{OrchestrationPlan, PlanSelection}; diff --git a/server-rs/crates/agent-runtime-orchestration/src/plan.rs b/server-rs/crates/agent-runtime-orchestration/src/plan.rs new file mode 100644 index 000000000..a7ba32ff5 --- /dev/null +++ b/server-rs/crates/agent-runtime-orchestration/src/plan.rs @@ -0,0 +1,64 @@ +use serde::{Deserialize, Serialize}; + +use crate::{OrchestrationError, OrchestrationErrorKind, TaskGraph}; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum PlanSelection { + All, + Repair { task_ids: Vec }, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct OrchestrationPlan { + active_task_ids: Vec, + carried_task_ids: Vec, + dependency_waves: Vec>, +} + +impl OrchestrationPlan { + pub fn active_task_ids(&self) -> &[String] { + &self.active_task_ids + } + + pub fn carried_task_ids(&self) -> &[String] { + &self.carried_task_ids + } + + pub fn dependency_waves(&self) -> &[Vec] { + &self.dependency_waves + } +} + +impl TaskGraph { + pub fn plan(&self, selection: PlanSelection) -> Result { + let all_task_ids = self.all_task_ids(); + let active_task_ids = match selection { + PlanSelection::All => all_task_ids.clone(), + PlanSelection::Repair { task_ids } => { + if task_ids.is_empty() { + return Err(OrchestrationError::new( + OrchestrationErrorKind::InvalidInput, + "repair selection 至少需要一个 task", + )); + } + self.expand_downstream(&task_ids)? + } + }; + let active = active_task_ids + .iter() + .map(String::as_str) + .collect::>(); + let carried_task_ids = all_task_ids + .into_iter() + .filter(|task_id| !active.contains(task_id.as_str())) + .collect::>(); + let dependency_waves = self.dependency_waves(&active_task_ids, &carried_task_ids)?; + + Ok(OrchestrationPlan { + active_task_ids, + carried_task_ids, + dependency_waves, + }) + } +} diff --git a/server-rs/crates/agent-runtime-orchestration/tests/non_game_orchestration.rs b/server-rs/crates/agent-runtime-orchestration/tests/non_game_orchestration.rs new file mode 100644 index 000000000..53470dafa --- /dev/null +++ b/server-rs/crates/agent-runtime-orchestration/tests/non_game_orchestration.rs @@ -0,0 +1,154 @@ +use agent_runtime_core::{AgentCatalog, AgentDescriptor}; +use agent_runtime_orchestration::{ + OrchestrationErrorKind, PlanSelection, TaskGraph, TaskNode, TaskStatus, +}; + +fn task(id: &str, agent_id: &str, status: TaskStatus, dependencies: &[&str]) -> TaskNode { + TaskNode::try_new(id, agent_id, status, dependencies.iter().copied()).expect("valid task") +} + +fn document_review_graph() -> TaskGraph { + TaskGraph::try_new( + "Review a document collection", + [ + task("collect", "researcher", TaskStatus::Pending, &[]), + task("inspect", "reviewer", TaskStatus::Pending, &[]), + task("draft", "writer", TaskStatus::Pending, &["collect"]), + task("verify", "reviewer", TaskStatus::Pending, &["inspect"]), + task( + "publish", + "writer", + TaskStatus::Pending, + &["draft", "verify"], + ), + ], + ) + .expect("valid dynamic task graph") +} + +#[test] +fn dynamic_non_game_dag_produces_stable_ready_waves_and_repair_closure() { + let graph = document_review_graph(); + + assert_eq!(graph.ready_task_ids(), vec!["collect", "inspect"]); + let full = graph.plan(PlanSelection::All).expect("full plan"); + assert_eq!( + full.dependency_waves(), + &[ + vec!["collect".to_string(), "inspect".to_string()], + vec!["draft".to_string(), "verify".to_string()], + vec!["publish".to_string()], + ] + ); + + let repair = graph + .plan(PlanSelection::Repair { + task_ids: vec!["draft".to_string()], + }) + .expect("repair plan"); + assert_eq!(repair.active_task_ids(), ["draft", "publish"]); + assert_eq!(repair.carried_task_ids(), ["collect", "inspect", "verify"]); + assert_eq!( + repair.dependency_waves(), + &[vec!["draft".to_string()], vec!["publish".to_string()]] + ); +} + +#[test] +fn orchestration_graph_validates_agent_catalog_before_dispatch() { + let graph = document_review_graph(); + let catalog = AgentCatalog::try_new([ + AgentDescriptor::try_new("researcher", "research", std::iter::empty::<&str>()) + .expect("researcher"), + AgentDescriptor::try_new("reviewer", "review", std::iter::empty::<&str>()) + .expect("reviewer"), + AgentDescriptor::try_new("writer", "writing", std::iter::empty::<&str>()).expect("writer"), + ]) + .expect("catalog"); + graph.validate_agents(&catalog).expect("known agents"); + + let incomplete = AgentCatalog::try_new([ + AgentDescriptor::try_new("researcher", "research", std::iter::empty::<&str>()) + .expect("researcher"), + AgentDescriptor::try_new("reviewer", "review", std::iter::empty::<&str>()) + .expect("reviewer"), + ]) + .expect("incomplete catalog"); + let error = graph + .validate_agents(&incomplete) + .expect_err("writer must be registered"); + assert_eq!(error.kind(), OrchestrationErrorKind::UnknownAgent); +} + +#[test] +fn invalid_dependencies_and_cycles_fail_closed() { + let duplicate_dependency = TaskNode::try_new( + "draft", + "writer", + TaskStatus::Pending, + ["collect", "collect"], + ) + .expect_err("duplicate dependency"); + assert_eq!( + duplicate_dependency.kind(), + OrchestrationErrorKind::DuplicateDependency + ); + + let duplicate_task = TaskGraph::try_new( + "duplicate task", + [ + task("collect", "researcher", TaskStatus::Pending, &[]), + task("collect", "reviewer", TaskStatus::Pending, &[]), + ], + ) + .expect_err("duplicate task"); + assert_eq!(duplicate_task.kind(), OrchestrationErrorKind::DuplicateTask); + + let unknown = TaskGraph::try_new( + "unknown dependency", + [task("publish", "writer", TaskStatus::Pending, &["missing"])], + ) + .expect_err("unknown dependency"); + assert_eq!(unknown.kind(), OrchestrationErrorKind::UnknownDependency); + + let self_dependency = + TaskNode::try_new("inspect", "reviewer", TaskStatus::Pending, ["inspect"]) + .expect_err("self dependency"); + assert_eq!( + self_dependency.kind(), + OrchestrationErrorKind::SelfDependency + ); + + let cycle = TaskGraph::try_new( + "cycle", + [ + task("left", "researcher", TaskStatus::Pending, &["right"]), + task("right", "reviewer", TaskStatus::Pending, &["left"]), + ], + ) + .expect_err("cycle"); + assert_eq!(cycle.kind(), OrchestrationErrorKind::Cycle); +} + +#[test] +fn active_partition_requires_explicitly_satisfied_dependencies() { + let graph = document_review_graph(); + let error = graph + .dependency_waves(&["publish"], &[] as &[&str]) + .expect_err("publish prerequisites are neither active nor satisfied"); + assert_eq!(error.kind(), OrchestrationErrorKind::UnsatisfiedDependency); +} + +#[test] +fn deserialization_revalidates_task_node_contract() { + let error = serde_json::from_str::( + r#"{ + "id": "draft", + "agentId": "writer", + "status": "pending", + "dependencies": ["collect", "collect"] + }"#, + ) + .expect_err("duplicate dependency must not bypass the constructor"); + assert!(error.to_string().contains("重复依赖")); +} diff --git a/server-rs/crates/platform-agent/Cargo.toml b/server-rs/crates/platform-agent/Cargo.toml index c22ea18c3..62a291332 100644 --- a/server-rs/crates/platform-agent/Cargo.toml +++ b/server-rs/crates/platform-agent/Cargo.toml @@ -9,6 +9,8 @@ default = [] legacy-creative-agent = ["dep:async-trait", "dep:langchainrust", "dep:tokio"] [dependencies] +agent-runtime-core = { path = "../agent-runtime-core" } +agent-runtime-orchestration = { path = "../agent-runtime-orchestration" } async-trait = { version = "0.1", optional = true } langchainrust = { version = "0.2.20", optional = true } platform-llm = { path = "../platform-llm", default-features = false } diff --git a/server-rs/crates/platform-agent/src/game_creation.rs b/server-rs/crates/platform-agent/src/game_creation.rs index d41321bfc..2f403e922 100644 --- a/server-rs/crates/platform-agent/src/game_creation.rs +++ b/server-rs/crates/platform-agent/src/game_creation.rs @@ -1,3 +1,7 @@ +use agent_runtime_core::AgentCatalog; +use agent_runtime_orchestration::{ + OrchestrationError, PlanSelection, TaskGraph, TaskNode, TaskStatus, +}; use serde::{Deserialize, Serialize}; use std::collections::HashSet; @@ -825,7 +829,7 @@ pub fn build_game_creation_seed_task_graph( )); } - Ok(GameCreationTaskGraph { + let graph = GameCreationTaskGraph { goal: goal.to_string(), tasks: vec![ task( @@ -984,29 +988,55 @@ pub fn build_game_creation_seed_task_graph( ["标题、简介、标签、封面需求和导出检查已完成"], ), ], - }) + }; + compile_game_creation_task_graph(&graph)?; + Ok(graph) } -pub fn select_ready_game_creation_tasks(graph: &GameCreationTaskGraph) -> Vec { - let completed = graph +fn compile_game_creation_task_graph( + graph: &GameCreationTaskGraph, +) -> Result { + let tasks = graph .tasks .iter() - .filter(|task| task.status == GameCreationTaskStatus::Completed) - .map(|task| task.id.as_str()) - .collect::>(); - - graph - .tasks - .iter() - .filter(|task| { - task.status == GameCreationTaskStatus::Pending - && task - .dependencies - .iter() - .all(|dependency| completed.contains(dependency.as_str())) + .map(|task| { + TaskNode::try_new( + &task.id, + &task.id, + match task.status { + GameCreationTaskStatus::Pending => TaskStatus::Pending, + GameCreationTaskStatus::Running => TaskStatus::Running, + GameCreationTaskStatus::WaitingForConfirmation => TaskStatus::Waiting, + GameCreationTaskStatus::Completed => TaskStatus::Completed, + GameCreationTaskStatus::Failed => TaskStatus::Failed, + }, + task.dependencies.iter().cloned(), + ) }) + .collect::, _>>() + .map_err(invalid_orchestration)?; + TaskGraph::try_new(&graph.goal, tasks).map_err(invalid_orchestration) +} + +pub fn validate_game_creation_task_agents( + graph: &GameCreationTaskGraph, + catalog: &AgentCatalog, +) -> Result<(), PlatformAgentError> { + compile_game_creation_task_graph(graph)? + .validate_agents(catalog) + .map_err(invalid_orchestration) +} + +pub fn select_ready_game_creation_tasks( + graph: &GameCreationTaskGraph, +) -> Result, PlatformAgentError> { + let orchestration_graph = compile_game_creation_task_graph(graph)?; + Ok(orchestration_graph + .ready_task_ids() + .into_iter() + .filter_map(|task_id| graph.tasks.iter().find(|task| task.id == task_id)) .cloned() - .collect() + .collect()) } #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] @@ -1034,7 +1064,8 @@ pub fn plan_game_creation_agent_pass( graph: &GameCreationTaskGraph, pass: u8, findings_markdown: &str, -) -> GameCreationAgentPassPlan { +) -> Result { + let orchestration_graph = compile_game_creation_task_graph(graph)?; let structured_repair_routes = extract_game_creation_evaluator_repair_routes(graph, findings_markdown); let mut repair_focus = extract_game_creation_evaluator_issues(findings_markdown); @@ -1044,11 +1075,6 @@ pub fn plan_game_creation_agent_pass( .map(|route| route.issue.clone()) .collect(); } - let all_task_ids = graph - .tasks - .iter() - .map(|task| task.id.clone()) - .collect::>(); let repair_routes = if pass <= 1 || repair_focus.is_empty() { Vec::new() } else if !structured_repair_routes.is_empty() { @@ -1056,20 +1082,27 @@ pub fn plan_game_creation_agent_pass( } else { route_game_creation_repair_issues(graph, &repair_focus) }; - let repair_routes = expand_game_creation_repair_route_impacts(graph, repair_routes); - let active_task_ids = - select_agent_pass_active_tasks(pass, &repair_focus, &repair_routes, graph); - let active = active_task_ids - .iter() - .map(String::as_str) - .collect::>(); - let carried_task_ids = all_task_ids - .iter() - .filter(|task_id| !active.contains(task_id.as_str())) - .cloned() - .collect::>(); - let dependency_waves = - build_game_creation_dependency_waves(graph, &active_task_ids, &carried_task_ids); + let repair_routes = + expand_game_creation_repair_route_impacts(&orchestration_graph, repair_routes)?; + let mut selected_task_ids = Vec::new(); + for route in &repair_routes { + for task_id in &route.task_ids { + push_unique(&mut selected_task_ids, task_id); + } + } + let selection = if pass <= 1 || repair_focus.is_empty() || selected_task_ids.is_empty() { + PlanSelection::All + } else { + PlanSelection::Repair { + task_ids: selected_task_ids, + } + }; + let orchestration_plan = orchestration_graph + .plan(selection) + .map_err(invalid_orchestration)?; + let active_task_ids = orchestration_plan.active_task_ids().to_vec(); + let carried_task_ids = orchestration_plan.carried_task_ids().to_vec(); + let dependency_waves = orchestration_plan.dependency_waves().to_vec(); let mode = if pass <= 1 || repair_focus.is_empty() { "initial" } else { @@ -1093,7 +1126,7 @@ pub fn plan_game_creation_agent_pass( ) }; - GameCreationAgentPassPlan { + Ok(GameCreationAgentPassPlan { pass, mode: mode.to_string(), active_task_ids, @@ -1102,7 +1135,7 @@ pub fn plan_game_creation_agent_pass( repair_focus, repair_routes, summary, - } + }) } pub fn extract_game_creation_evaluator_issues(findings_markdown: &str) -> Vec { @@ -1152,30 +1185,6 @@ pub fn extract_game_creation_evaluator_repair_routes( sanitize_repair_routes(graph, routes) } -fn select_agent_pass_active_tasks( - pass: u8, - repair_focus: &[String], - repair_routes: &[GameCreationAgentRepairRoute], - graph: &GameCreationTaskGraph, -) -> Vec { - if pass <= 1 || repair_focus.is_empty() { - return graph.tasks.iter().map(|task| task.id.clone()).collect(); - } - - let mut task_ids = Vec::new(); - for route in repair_routes { - for task_id in &route.task_ids { - push_unique(&mut task_ids, task_id); - } - } - - if task_ids.is_empty() { - graph.tasks.iter().map(|task| task.id.clone()).collect() - } else { - task_ids - } -} - pub fn route_game_creation_repair_issues( graph: &GameCreationTaskGraph, issues: &[String], @@ -1273,13 +1282,15 @@ fn sanitize_repair_routes( } fn expand_game_creation_repair_route_impacts( - graph: &GameCreationTaskGraph, + graph: &TaskGraph, routes: Vec, -) -> Vec { +) -> Result, PlatformAgentError> { routes .into_iter() .map(|route| { - let expanded_task_ids = expand_task_ids_with_downstream_impacts(graph, &route.task_ids); + let expanded_task_ids = graph + .expand_downstream(&route.task_ids) + .map_err(invalid_orchestration)?; let reason = if expanded_task_ids.len() > route.task_ids.len() && !route.reason.contains("dependency-impact") { @@ -1288,46 +1299,15 @@ fn expand_game_creation_repair_route_impacts( route.reason }; - GameCreationAgentRepairRoute { + Ok(GameCreationAgentRepairRoute { issue: route.issue, task_ids: expanded_task_ids, reason, - } + }) }) .collect() } -fn expand_task_ids_with_downstream_impacts( - graph: &GameCreationTaskGraph, - task_ids: &[String], -) -> Vec { - let mut impacted = task_ids.iter().cloned().collect::>(); - let mut changed = true; - while changed { - changed = false; - for task in &graph.tasks { - if impacted.contains(&task.id) { - continue; - } - if task - .dependencies - .iter() - .any(|dependency| impacted.contains(dependency)) - { - impacted.insert(task.id.clone()); - changed = true; - } - } - } - - graph - .tasks - .iter() - .filter(|task| impacted.contains(&task.id)) - .map(|task| task.id.clone()) - .collect() -} - fn route_game_creation_repair_issue( graph: &GameCreationTaskGraph, issue: &str, @@ -1452,54 +1432,6 @@ fn push_unique(values: &mut Vec, value: &str) { } } -fn build_game_creation_dependency_waves( - graph: &GameCreationTaskGraph, - active_task_ids: &[String], - carried_task_ids: &[String], -) -> Vec> { - let active = active_task_ids.iter().cloned().collect::>(); - let known = graph - .tasks - .iter() - .map(|task| task.id.clone()) - .collect::>(); - let mut remaining = active_task_ids.to_vec(); - let mut completed = carried_task_ids.iter().cloned().collect::>(); - let mut waves = Vec::new(); - - while !remaining.is_empty() { - let wave = remaining - .iter() - .filter(|task_id| { - graph - .tasks - .iter() - .find(|task| task.id == **task_id) - .is_some_and(|task| { - task.dependencies.iter().all(|dependency| { - !active.contains(dependency) - || completed.contains(dependency) - || !known.contains(dependency) - }) - }) - }) - .cloned() - .collect::>(); - if wave.is_empty() { - waves.push(remaining); - break; - } - - for task_id in &wave { - completed.insert(task_id.clone()); - } - remaining.retain(|task_id| !wave.contains(task_id)); - waves.push(wave); - } - - waves -} - fn contains_any(value: &str, needles: &[&str]) -> bool { needles.iter().any(|needle| value.contains(needle)) } @@ -1525,6 +1457,10 @@ fn task( } } +fn invalid_orchestration(error: OrchestrationError) -> PlatformAgentError { + PlatformAgentError::InvalidInput(format!("多 Agent 编排任务图无效：{error}")) +} + #[cfg(test)] mod tests { use std::collections::HashSet; @@ -1886,6 +1822,28 @@ mod tests { ); } + #[test] + fn seed_task_graph_validates_against_an_injected_agent_catalog() { + use agent_runtime_core::AgentDescriptor; + + let graph = build_game_creation_seed_task_graph("做一个像素风横版动作原型").unwrap(); + let catalog = AgentCatalog::try_new(graph.tasks.iter().map(|task| { + AgentDescriptor::try_new(&task.id, &task.role, std::iter::empty::<&str>()) + .expect("agent descriptor") + })) + .expect("agent catalog"); + validate_game_creation_task_agents(&graph, &catalog).expect("known task agents"); + + let incomplete = AgentCatalog::try_new(graph.tasks.iter().skip(1).map(|task| { + AgentDescriptor::try_new(&task.id, &task.role, std::iter::empty::<&str>()) + .expect("agent descriptor") + })) + .expect("incomplete catalog"); + let error = validate_game_creation_task_agents(&graph, &incomplete) + .expect_err("missing task agent must fail closed"); + assert!(error.to_string().contains("未注册 Agent")); + } + #[test] fn code_director_waits_for_design_assets_audio_and_balance() { let graph = build_game_creation_seed_task_graph("做一个像素风横版动作原型").unwrap(); @@ -1912,6 +1870,7 @@ mod tests { assert_eq!( select_ready_game_creation_tasks(&graph) + .expect("ready tasks") .iter() .map(|task| task.id.as_str()) .collect::>(), @@ -1927,6 +1886,7 @@ mod tests { assert_eq!( select_ready_game_creation_tasks(&graph) + .expect("ready tasks") .iter() .map(|task| task.id.as_str()) .collect::>(), @@ -1942,6 +1902,7 @@ mod tests { assert_eq!( select_ready_game_creation_tasks(&graph) + .expect("ready tasks") .iter() .map(|task| task.id.as_str()) .collect::>(), @@ -1968,7 +1929,8 @@ mod tests { &graph, 1, "# Evaluator Findings\n\n- pass: 0\n- status: needs-revision\n\n- 暂无上一轮问题，Generator 可开始首轮实现。\n", - ); + ) + .expect("initial pass plan"); assert_eq!(plan.mode, "initial"); assert_eq!(plan.active_task_ids.len(), 16); @@ -1984,7 +1946,8 @@ mod tests { &graph, 2, "# Evaluator Findings\n\n- pass: 1\n- status: needs-revision\n\n- gameHtml 缺少 canvas、requestAnimationFrame 和输入监听。\n", - ); + ) + .expect("repair pass plan"); assert_eq!(plan.mode, "repair"); assert_eq!( @@ -2033,7 +1996,8 @@ mod tests { ] ``` "#, - ); + ) + .expect("structured repair pass plan"); assert_eq!(plan.mode, "repair"); assert_eq!( @@ -2088,7 +2052,8 @@ mod tests { ] ``` "#, - ); + ) + .expect("asset repair pass plan"); assert_eq!( plan.active_task_ids, @@ -2128,11 +2093,43 @@ mod tests { &graph, 2, "# Evaluator Findings\n\n- pass: 1\n- status: needs-revision\n\n- handoffs 缺少 publishing 专业组交接。\n", - ); + ) + .expect("cross-group pass plan"); assert_eq!(plan.mode, "repair"); assert_eq!(plan.active_task_ids.len(), 16); assert!(plan.carried_task_ids.is_empty()); assert_eq!(plan.repair_routes[0].reason, "cross-group-handoff"); } + + #[test] + fn pass_plan_rejects_a_cyclic_game_task_graph() { + let graph = GameCreationTaskGraph { + goal: "验证非法环".to_string(), + tasks: vec![ + task( + "left", + "左节点", + GameCreationAgentGroup::Design, + "Left", + ["right"], + [], + ["左节点完成"], + ), + task( + "right", + "右节点", + GameCreationAgentGroup::Code, + "Right", + ["left"], + [], + ["右节点完成"], + ), + ], + }; + + let error = plan_game_creation_agent_pass(&graph, 1, "") + .expect_err("cyclic graph must fail closed"); + assert!(error.to_string().contains("依赖环")); + } } diff --git a/server-rs/crates/platform-agent/src/lib.rs b/server-rs/crates/platform-agent/src/lib.rs index f7c609c1d..63d7403a8 100644 --- a/server-rs/crates/platform-agent/src/lib.rs +++ b/server-rs/crates/platform-agent/src/lib.rs @@ -32,6 +32,7 @@ pub use game_creation::{ build_game_creation_seed_task_graph, extract_game_creation_evaluator_issues, extract_game_creation_evaluator_repair_routes, plan_game_creation_agent_pass, route_game_creation_repair_issues, select_ready_game_creation_tasks, + validate_game_creation_task_agents, }; #[cfg(feature = "legacy-creative-agent")] pub use langchain_adapter::LangChainRustAdapter; -- 2.52.0 From 92d53ea6cc4b4188c70f3eaac0600a8edf1e50c4 Mon Sep 17 00:00:00 2001 From: kdletters Date: Wed, 26 Aug 2026 20:24:06 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E4=B8=AD=E8=87=AA=E4=B8=BB=E6=89=A9=E5=9B=BE=E6=8F=90=E6=A1=88?= =?UTF-8?q?=E8=83=BD=E5=8A=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 GraphProposal、GraphLimits 与 TaskGraph 原子扩图校验。 补充严格 JSON、预算、环和失败原子性测试。 同步 Agent Runtime 技术方案与项目记忆。 --- .../shared-memory/decision-log.md | 7 + .../shared-memory/project-overview.md | 2 +- ...案】AI游戏创作Agent Runtime V1.1-2026-07-12.md | 22 + ...案】AI游戏创作智能体App实施计划-2026-06-24.md | 6 + .../agent-runtime-orchestration/src/error.rs | 18 + .../agent-runtime-orchestration/src/graph.rs | 107 ++- .../agent-runtime-orchestration/src/lib.rs | 6 + .../src/proposal.rs | 744 ++++++++++++++++++ .../tests/dynamic_proposal.rs | 315 ++++++++ 9 files changed, 1223 insertions(+), 4 deletions(-) create mode 100644 server-rs/crates/agent-runtime-orchestration/src/proposal.rs create mode 100644 server-rs/crates/agent-runtime-orchestration/tests/dynamic_proposal.rs diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index df5655f04..00c219856 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -16,6 +16,13 @@ --- +## 2026-08-26 运行中自主扩图提案留在编排层 + +- 背景：`agent-runtime-orchestration` 已能构造和调度动态 DAG，但 LLM 在执行中发现缺少步骤时没有通用的安全扩图合同。 +- 决策：新增严格 serde 的 `GraphProposal`（`TaskProposal` + `GraphEdge`）和 `GraphLimits`，由 `TaskGraph::apply_proposal` / `expand_with_proposal` 在内存中构造不可变候选图；新节点默认 `Pending`，边方向为前置 `from` → 依赖方 `to`。 +- 安全与一致性：所有 Agent、端点、重复引用、环、节点/边/深度/扇出预算在候选返回前一次校验；边只能指向新节点，禁止给已运行任务原地追加依赖。任一失败保留旧图。成功后的 epoch、基图版本、proposal 幂等和持久化由宿主负责，crate 不调用 LLM/Provider/ToolHost/Runner，也不写 `.agent/runtime/**`。 +- 验证：非游戏 conformance 覆盖有效扩图、ready/wave 重算、未知 Agent/端点、重复边、已有任务修改、环、预算、严格 JSON 和原子失败；关联文档为 `docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md` V1.55。 + ## 2026-08-26 通用多 Agent DAG 编排与执行内核分层 - 背景：`agent-runtime-core` 已承接 catalog、run/action 生命周期、lane、宿主 ToolHost、spawn/all-join 和 Provider 契约，但动态任务图的 ready 选择、依赖波次与返工下游闭包仍混在 `platform-agent::game_creation`，其它产品无法复用且非法环会被合并成伪 wave。 diff --git a/docs/project-memory/shared-memory/project-overview.md b/docs/project-memory/shared-memory/project-overview.md index 0ca84ed62..c7a2bccc2 100644 --- a/docs/project-memory/shared-memory/project-overview.md +++ b/docs/project-memory/shared-memory/project-overview.md @@ -51,7 +51,7 @@ SpacetimeDB crate、SDK、CLI / standalone 与生成 bindings 按 `2.7.0` 对齐 ## AGC DirectProject 与 UI workflow -- 通用 Agent Rust 分层为 `agent-runtime-core`（catalog、执行生命周期、ToolHost/spawn/all-join/Provider 契约）、`agent-runtime-orchestration`（动态无环任务图、ready、依赖波次和返工下游闭包）与 `platform-agent` 游戏适配器；循环返工通过新 pass / epoch 表达，不在单张依赖图中建立回边。 +- 通用 Agent Rust 分层为 `agent-runtime-core`（catalog、执行生命周期、ToolHost/spawn/all-join/Provider 契约）、`agent-runtime-orchestration`（动态无环任务图、ready、依赖波次、返工下游闭包和受限自主扩图提案）与 `platform-agent` 游戏适配器；循环返工通过新 pass / epoch 表达，不在单张依赖图中建立回边。LLM 可经宿主结构化 function call 提出新增节点/边，编排层只生成经校验的新候选图，epoch 与持久化仍由宿主掌控。 - DirectProject 只连接客户端内置的 `agc_tools` STDIO MCP。它负责审核引用读取、标准美术准备、已登记资源有界查询、视频 / 角色动画 / 音效 / BGM 的 create-or-derive、已登记图片去背景、desktop/mobile 浏览器试玩和受控 `agc_web_search`；付费资源调用由客户端绑定回合、幂等账本、请求上限和投影权威。 - DirectProject 的 Codex 原生文件、搜索、命令、图片查看和 Skill 仅在真实 `game/` cwd 与 `workspaceWrite(writableRoots=[game])` 内可用；原生命令网络保持关闭。多 Agent、Apps、插件、hooks、图片生成、Goals、Workspace Dependencies、Tool Suggestion 和原生浏览器/电脑控制保持关闭。app-server 使用隔离 `CODEX_HOME`，provider 凭据只由 AGC 客户端代理持有，不能进入模型上下文或 shell 环境。 - `ui-prototype`（设计图片）与 UI 编辑器 `UI` JSON 是不同资源。白名单 `ui.workflow.run` 按页面执行 `prepare → recognize → status → finalize`，由 provider-backed 识别、合并和组件绑定持久化 State/revision，并把 `reference-ready → structure-ready → merge-ready → binding-ready → application-ready → completed` 投影到 manifest。Provider 缺失、请求失败、工具缺失、结果不匹配或仍有待审节点时保留真实阶段并返回 blocker，不得用 deterministic seed 伪造完成。 diff --git a/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md b/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md index 38ac61005..a93160c97 100644 --- a/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md +++ b/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md @@ -1679,6 +1679,28 @@ V1.48-V1.50 已把 catalog、执行生命周期与 Provider 契约收进 `agent- - 未完成：Tauri `cargo check --tests` 已编译到 AGC 自定义 `build.rs`，随后因仓库四个候选路径均缺少内置 Codex CLI vendor 资源而退出（`build.rs:77`）；本轮未执行 `npm ci`，因此不能将 Tauri 组合根或完整 `ai-game-creator-shell:check` 记为通过。 - 未执行：当前 Rust 1.96 工具链未安装 `clippy` component；没有把该静态检查结果用其它门禁结果替代。 +## V1.55 运行中自主扩图提案 + +V1.54 的公共编排层可以在运行前构造动态 DAG，但 LLM 在执行过程中发现缺少步骤时还没有一个通用、受限的扩图入口。V1.55 在同一 `agent-runtime-orchestration` crate 增加结构化 `GraphProposal`，让宿主能够把 LLM 的 function-call arguments 解析为候选节点和边，并在不改变 `agent-runtime-core` 执行职责的前提下生成下一张图。 + +### 提案 DTO 与宿主边界 + +- `GraphProposal` 只包含 `nodes` 与 `edges`。节点使用 `{ id, agentId }`，边使用 `{ from, to }`；`from` 是前置任务，`to` 是依赖它的任务。DTO 使用 `camelCase` 且拒绝未知字段，节点和边的稳定标识沿用公共图模型约束。 +- `TaskGraph::apply_proposal` / `expand_with_proposal` 是纯校验与候选构造 API：不调用 LLM、Provider、ToolHost、Runner 或持久化。宿主负责声明 function tool、把 arguments 反序列化为 `GraphProposal`，并在成功后把返回的候选图写入自己的新 epoch。 +- 新节点统一以 `Pending` 加入，并保留现有任务状态和注册顺序。提案边必须指向本次新增节点；不允许在运行中的旧任务上原地追加前置依赖。若业务确实要改旧边，宿主应构造完整候选图并按自己的 CAS/epoch 合同一次替换。 + +### 原子校验与预算 + +- 候选图只有在所有检查通过后才返回；未知 Agent、未知端点、重复节点/边、自依赖、有向环和已有节点依赖修改都会失败，原图保持不变。 +- `GraphLimits` 同时限制完整候选图的 `maxTasks`、`maxEdges`、`maxDepth`（根层计 1）和 `maxOutDegree`（一个前置任务的直接下游数）。默认值为 `128 / 512 / 32 / 32`；超限不截断、不部分提交。 +- 成功扩图后，宿主必须把它视为新的 graph/epoch，重新计算 ready task 与 dependency waves，并在自己的持久层记录提案身份、基图版本和幂等结果。crate 不把 epoch、proposal ID 或执行事实写入图，也不自动重放 Provider。 + +### V1.55 验收 + +- 非游戏 conformance 覆盖有效新增节点/边、全部新节点 `Pending`、ready/wave 重算、未知 Agent、未知端点、重复边、已有任务修改、环、节点/边/深度/扇出预算和失败原子性。 +- 覆盖 `GraphProposal`、`GraphEdge`、`GraphLimits` 与 `TaskGraph` 的严格 JSON round-trip；未知字段、非法标识和零预算均失败关闭。 +- 真实 LLM 接入仍由宿主后续提供；本切片证明了宿主可在一个结构化 function call 回合中安全生成候选新图，但不把 provider 请求或持久化当作 crate 的事实源。 + ## 验收命令 - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml structured_plan_ -- --nocapture` diff --git a/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md b/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md index da1e25334..b8a6883a2 100644 --- a/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md +++ b/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md @@ -1,5 +1,11 @@ # AI 游戏创作智能体 App 实施计划 +## 2026-08-26 运行中自主扩图提案 + +- `agent-runtime-orchestration` 提供严格 serde 的 `GraphProposal`、`TaskProposal`、`GraphEdge` 与 `GraphLimits`。宿主可把 LLM function-call arguments 解析后交给 `TaskGraph::apply_proposal`，在内存中得到新的、完整校验过的候选 DAG。 +- 新节点默认 `Pending`；边使用 `from`（前置）→ `to`（依赖方），且新增边只能指向本次新增节点。未知 Agent/端点、重复节点/边、自依赖、有向环和节点/边/深度/扇出预算超限整次失败，旧图不变。 +- crate 不调用 Provider、Runner、ToolHost 或持久化；宿主负责 function tool 暴露、基图/epoch CAS、proposal 幂等、落盘和重新调度 ready/dependency waves。修改既有任务依赖时必须由宿主构造完整候选图并切换新 epoch。 + ## 2026-08-25 账户 / 项目画布 / 本地素材导入 - 素材读取区分三类来源：`asset.list` / `agc_list_registered_assets` 是当前项目本地 manifest，`agc_list_project_files` / `file.list` 只发现项目目录中实际存在但可能未登记的文件，`asset.library.list` 是当前登录账号素材库，项目画布资源读取是当前网页项目/画布的完整图片清单；账户素材库不能替代项目画布清单。 diff --git a/server-rs/crates/agent-runtime-orchestration/src/error.rs b/server-rs/crates/agent-runtime-orchestration/src/error.rs index 25af0518f..3cdd4be03 100644 --- a/server-rs/crates/agent-runtime-orchestration/src/error.rs +++ b/server-rs/crates/agent-runtime-orchestration/src/error.rs @@ -3,17 +3,35 @@ use std::fmt; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum OrchestrationErrorKind { InvalidInput, + InvalidLimits, + EmptyProposal, DuplicateTask, DuplicateDependency, + DuplicateEdge, UnknownDependency, SelfDependency, Cycle, UnknownAgent, UnknownTask, + ExistingTaskMutation, + NodeBudgetExceeded, + EdgeBudgetExceeded, + DepthBudgetExceeded, + FanOutBudgetExceeded, ConflictingTaskSet, UnsatisfiedDependency, } +#[allow(non_upper_case_globals)] +impl OrchestrationErrorKind { + /// Compatibility alias for callers that describe the node budget as a + /// task budget. + pub const TaskBudgetExceeded: Self = Self::NodeBudgetExceeded; + + /// Compatibility alias for callers that use the shorter fan-out spelling. + pub const FanoutBudgetExceeded: Self = Self::FanOutBudgetExceeded; +} + #[derive(Clone, Debug, Eq, PartialEq)] pub struct OrchestrationError { kind: OrchestrationErrorKind, diff --git a/server-rs/crates/agent-runtime-orchestration/src/graph.rs b/server-rs/crates/agent-runtime-orchestration/src/graph.rs index 512610462..69ff57b2b 100644 --- a/server-rs/crates/agent-runtime-orchestration/src/graph.rs +++ b/server-rs/crates/agent-runtime-orchestration/src/graph.rs @@ -1,7 +1,7 @@ use std::collections::{BTreeMap, BTreeSet, HashSet, VecDeque}; use agent_runtime_core::AgentCatalog; -use serde::{Deserialize, Serialize}; +use serde::{Deserialize, Deserializer, Serialize}; use crate::{OrchestrationError, OrchestrationErrorKind}; @@ -105,13 +105,32 @@ impl TaskNode { } } -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] pub struct TaskGraph { goal: String, tasks: Vec, + #[serde(skip)] by_id: BTreeMap, } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct TaskGraphInput { + goal: String, + tasks: Vec, +} + +impl<'de> Deserialize<'de> for TaskGraph { + fn deserialize(deserializer: D) -> Result + where + D: 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, @@ -165,6 +184,44 @@ impl TaskGraph { .and_then(|index| self.tasks.get(*index)) } + /// Number of tasks in this graph. + pub fn task_count(&self) -> usize { + self.tasks.len() + } + + /// Alias for [`TaskGraph::task_count`] using graph terminology. + pub fn node_count(&self) -> usize { + self.task_count() + } + + /// Number of prerequisite edges in this graph. + pub fn edge_count(&self) -> usize { + self.tasks.iter().map(|task| task.dependencies.len()).sum() + } + + /// Longest dependency path measured in task layers. A root task has + /// depth 1. Graph construction rejects cycles, so this calculation is + /// total for every `TaskGraph` value. + pub fn depth(&self) -> usize { + graph_depth(&self.tasks, &self.by_id) + } + + /// Number of direct dependents of a prerequisite task. + pub fn fan_out(&self, task_id: &str) -> Option { + self.get(task_id)?; + Some( + self.tasks + .iter() + .filter(|task| task.dependencies.iter().any(|id| id == task_id)) + .count(), + ) + } + + /// Alias for [`TaskGraph::fan_out`]. + pub fn out_degree(&self, task_id: &str) -> Option { + self.fan_out(task_id) + } + pub fn validate_agents(&self, catalog: &AgentCatalog) -> Result<(), OrchestrationError> { for task in &self.tasks { if catalog.get(&task.agent_id).is_none() { @@ -328,7 +385,7 @@ impl TaskGraph { } } -fn validate_identifier(value: &str, field: &str) -> Result<(), OrchestrationError> { +pub(crate) fn validate_identifier(value: &str, field: &str) -> Result<(), OrchestrationError> { if value != value.trim() { return Err(OrchestrationError::new( OrchestrationErrorKind::InvalidInput, @@ -417,3 +474,47 @@ fn validate_acyclic( format!("task graph 包含依赖环：{}", cyclic.join(", ")), )) } + +fn graph_depth(tasks: &[TaskNode], by_id: &BTreeMap) -> usize { + if tasks.is_empty() { + return 0; + } + + let mut indegrees = tasks + .iter() + .map(|task| task.dependencies.len()) + .collect::>(); + let mut dependents = vec![Vec::::new(); tasks.len()]; + for (task_index, task) in tasks.iter().enumerate() { + for dependency in &task.dependencies { + // `TaskGraph::try_new` proves this lookup exists. Keeping the + // defensive branch makes this helper total if it is ever reused + // during a future internal refactor. + let Some(&dependency_index) = by_id.get(dependency) else { + return 0; + }; + dependents[dependency_index].push(task_index); + } + } + + let mut depths = vec![1usize; tasks.len()]; + let mut ready = indegrees + .iter() + .enumerate() + .filter_map(|(index, indegree)| (*indegree == 0).then_some(index)) + .collect::>(); + let mut visited = 0; + let mut maximum = 1; + while let Some(index) = ready.pop_front() { + visited += 1; + maximum = maximum.max(depths[index]); + for dependent in &dependents[index] { + depths[*dependent] = depths[*dependent].max(depths[index].saturating_add(1)); + indegrees[*dependent] -= 1; + if indegrees[*dependent] == 0 { + ready.push_back(*dependent); + } + } + } + if visited == tasks.len() { maximum } else { 0 } +} diff --git a/server-rs/crates/agent-runtime-orchestration/src/lib.rs b/server-rs/crates/agent-runtime-orchestration/src/lib.rs index ecf418b75..fe85e9199 100644 --- a/server-rs/crates/agent-runtime-orchestration/src/lib.rs +++ b/server-rs/crates/agent-runtime-orchestration/src/lib.rs @@ -7,7 +7,13 @@ mod error; mod graph; mod plan; +mod proposal; pub use error::{OrchestrationError, OrchestrationErrorKind}; pub use graph::{TaskGraph, TaskNode, TaskStatus}; pub use plan::{OrchestrationPlan, PlanSelection}; +pub use proposal::{ + AppliedGraphProposal, DEFAULT_GRAPH_MAX_DEPTH, DEFAULT_GRAPH_MAX_EDGES, + DEFAULT_GRAPH_MAX_OUT_DEGREE, DEFAULT_GRAPH_MAX_TASKS, GraphEdge, GraphExpansion, GraphLimits, + GraphProposal, TaskProposal, +}; diff --git a/server-rs/crates/agent-runtime-orchestration/src/proposal.rs b/server-rs/crates/agent-runtime-orchestration/src/proposal.rs new file mode 100644 index 000000000..df53cd180 --- /dev/null +++ b/server-rs/crates/agent-runtime-orchestration/src/proposal.rs @@ -0,0 +1,744 @@ +//! Structured, host-agnostic graph expansion proposed by an LLM or another +//! planner. +//! +//! This module deliberately stops at validation and candidate construction. +//! It does not call a provider, execute an agent, or persist an epoch. A host +//! can deserialize a provider function-call argument into [`GraphProposal`], +//! pass it to [`TaskGraph::expand_with_proposal`], and persist the returned +//! graph as the next epoch if the result is accepted. + +use std::collections::{BTreeMap, BTreeSet}; + +use agent_runtime_core::AgentCatalog; +use serde::{Deserialize, Deserializer, Serialize}; + +use crate::{ + OrchestrationError, OrchestrationErrorKind, TaskGraph, TaskNode, TaskStatus, + graph::validate_identifier, +}; + +/// Default maximum number of tasks in a candidate graph. +pub const DEFAULT_GRAPH_MAX_TASKS: usize = 128; +/// Default maximum number of dependency edges in a candidate graph. +pub const DEFAULT_GRAPH_MAX_EDGES: usize = 512; +/// Default maximum number of task layers in a candidate graph. +pub const DEFAULT_GRAPH_MAX_DEPTH: usize = 32; +/// Default maximum number of direct dependents of one task. +pub const DEFAULT_GRAPH_MAX_OUT_DEGREE: usize = 32; + +/// A task that a planner proposes to add to a graph. +/// +/// New tasks are always inserted with [`TaskStatus::Pending`]. Product +/// metadata such as a title, artifact path, or acceptance text belongs in the +/// host adapter and is intentionally not part of this generic DTO. +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct TaskProposal { + id: String, + agent_id: String, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct TaskProposalInput { + id: String, + agent_id: String, +} + +impl<'de> Deserialize<'de> for TaskProposal { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let input = TaskProposalInput::deserialize(deserializer)?; + Self::try_new(input.id, input.agent_id).map_err(serde::de::Error::custom) + } +} + +impl TaskProposal { + /// Creates a validated task proposal. + pub fn try_new( + id: impl Into, + agent_id: impl Into, + ) -> Result { + let id = id.into(); + let agent_id = agent_id.into(); + validate_identifier(&id, "proposal task id")?; + validate_identifier(&agent_id, "proposal task agent id")?; + Ok(Self { id, agent_id }) + } + + /// Alias for [`TaskProposal::try_new`] for hosts that use `new` for DTO + /// construction while still handling validation errors. + pub fn new( + id: impl Into, + agent_id: impl Into, + ) -> Result { + Self::try_new(id, agent_id) + } + + pub fn id(&self) -> &str { + &self.id + } + + pub fn agent_id(&self) -> &str { + &self.agent_id + } +} + +/// A directed dependency edge in a proposal. +/// +/// `from` is the prerequisite and `to` is the task that depends on it. The +/// edge therefore corresponds to adding `from` to `to.dependencies` in the +/// resulting graph. +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct GraphEdge { + from: String, + to: String, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct GraphEdgeInput { + #[serde(alias = "source")] + from: String, + #[serde(alias = "target")] + to: String, +} + +impl<'de> Deserialize<'de> for GraphEdge { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let input = GraphEdgeInput::deserialize(deserializer)?; + Self::try_new(input.from, input.to).map_err(serde::de::Error::custom) + } +} + +impl GraphEdge { + /// Creates a validated prerequisite-to-dependent edge. + pub fn try_new( + from: impl Into, + to: impl Into, + ) -> Result { + let from = from.into(); + let to = to.into(); + validate_identifier(&from, "proposal edge from")?; + validate_identifier(&to, "proposal edge to")?; + if from == to { + return Err(OrchestrationError::new( + OrchestrationErrorKind::SelfDependency, + format!("proposal edge 不能连接任务自身：{from}"), + )); + } + Ok(Self { from, to }) + } + + /// Alias for [`GraphEdge::try_new`]. + pub fn new(from: impl Into, to: impl Into) -> Result { + Self::try_new(from, to) + } + + /// Convenience constructor whose names make the dependency direction + /// explicit at call sites. + pub fn dependency( + prerequisite: impl Into, + dependent: impl Into, + ) -> Result { + Self::try_new(prerequisite, dependent) + } + + pub fn from(&self) -> &str { + &self.from + } + + pub fn to(&self) -> &str { + &self.to + } + + /// Alias for [`GraphEdge::from`], useful when a host calls the fields + /// source/target in its own graph model. + pub fn source(&self) -> &str { + &self.from + } + + /// Alias for [`GraphEdge::to`]. + pub fn target(&self) -> &str { + &self.to + } +} + +/// A structured graph change returned by a planner. +/// +/// The proposal contains only additions. Edges whose target is an existing +/// task are rejected so a running task never acquires a new prerequisite in +/// place. To replace existing dependencies, a host must build a complete +/// candidate graph and install it as a new epoch with its own persistence/CAS +/// contract. +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct GraphProposal { + nodes: Vec, + edges: Vec, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct GraphProposalInput { + nodes: Vec, + edges: Vec, +} + +impl<'de> Deserialize<'de> for GraphProposal { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let input = GraphProposalInput::deserialize(deserializer)?; + Self::try_new(input.nodes, input.edges).map_err(serde::de::Error::custom) + } +} + +impl GraphProposal { + pub fn try_new( + nodes: impl IntoIterator, + edges: impl IntoIterator, + ) -> Result { + let nodes = nodes.into_iter().collect::>(); + let edges = edges.into_iter().collect::>(); + if nodes.is_empty() && edges.is_empty() { + return Err(OrchestrationError::new( + OrchestrationErrorKind::EmptyProposal, + "graph proposal 至少需要一个新节点或一条新边", + )); + } + + let mut node_ids = BTreeSet::new(); + for node in &nodes { + if !node_ids.insert(node.id.clone()) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DuplicateTask, + format!("proposal task id 重复：{}", node.id), + )); + } + } + + let mut edge_ids = BTreeSet::new(); + for edge in &edges { + if !edge_ids.insert((edge.from.clone(), edge.to.clone())) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DuplicateEdge, + format!("proposal edge 重复：{} -> {}", edge.from, edge.to), + )); + } + } + + Ok(Self { nodes, edges }) + } + + pub fn new( + nodes: impl IntoIterator, + edges: impl IntoIterator, + ) -> Result { + Self::try_new(nodes, edges) + } + + pub fn nodes(&self) -> &[TaskProposal] { + &self.nodes + } + + /// Alias for [`GraphProposal::nodes`] for callers that use task-oriented + /// terminology. + pub fn tasks(&self) -> &[TaskProposal] { + &self.nodes + } + + pub fn edges(&self) -> &[GraphEdge] { + &self.edges + } + + pub fn is_empty(&self) -> bool { + self.nodes.is_empty() && self.edges.is_empty() + } + + pub fn validate(&self) -> Result<(), OrchestrationError> { + // The fields are private and constructors/deserialization already + // enforce these invariants. Re-running the cheap checks keeps this + // method useful as an explicit boundary for host adapters. + if self.is_empty() { + return Err(OrchestrationError::new( + OrchestrationErrorKind::EmptyProposal, + "graph proposal 至少需要一个新节点或一条新边", + )); + } + let mut node_ids = BTreeSet::new(); + for node in &self.nodes { + validate_identifier(&node.id, "proposal task id")?; + validate_identifier(&node.agent_id, "proposal task agent id")?; + if !node_ids.insert(node.id.as_str()) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DuplicateTask, + format!("proposal task id 重复：{}", node.id), + )); + } + } + let mut edge_ids = BTreeSet::new(); + for edge in &self.edges { + validate_identifier(&edge.from, "proposal edge from")?; + validate_identifier(&edge.to, "proposal edge to")?; + if edge.from == edge.to { + return Err(OrchestrationError::new( + OrchestrationErrorKind::SelfDependency, + format!("proposal edge 不能连接任务自身：{}", edge.from), + )); + } + if !edge_ids.insert((edge.from.as_str(), edge.to.as_str())) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DuplicateEdge, + format!("proposal edge 重复：{} -> {}", edge.from, edge.to), + )); + } + } + Ok(()) + } +} + +/// Resource limits applied to the candidate graph produced by a proposal. +/// +/// Limits are checked against the complete resulting graph, not just the +/// proposed delta. `max_depth` counts graph layers: a root task has depth 1. +/// `max_out_degree` counts dependents for one prerequisite (`from -> to`). +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct GraphLimits { + pub max_tasks: usize, + pub max_edges: usize, + pub max_depth: usize, + 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(deserializer: D) -> Result + 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 { + 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::new( + OrchestrationErrorKind::InvalidLimits, + format!("graph limits 的 {field} 必须大于 0"), + )); + } + Ok(()) + } +} + +/// A validated candidate graph plus the delta that produced it. +/// +/// The host may use this report to persist an epoch/change journal without +/// re-parsing the untrusted provider payload. The graph itself remains the +/// authoritative candidate. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct GraphExpansion { + graph: TaskGraph, + added_task_ids: Vec, + added_edges: Vec, +} + +/// Descriptive alias for [`GraphExpansion`]. +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 + } +} + +impl TaskGraph { + /// Applies an additive proposal and returns a new validated graph. + /// + /// This method is intentionally immutable: a successful return is a + /// candidate for a new host-managed epoch, while every error leaves the + /// current graph untouched. New edges must target a newly proposed task; + /// this prevents changing the prerequisites of a task that may already be + /// running or completed. + pub fn apply_proposal( + &self, + proposal: &GraphProposal, + catalog: &AgentCatalog, + limits: &GraphLimits, + ) -> Result { + self.expand_with_proposal(proposal, catalog, limits) + .map(GraphExpansion::into_graph) + } + + /// Returns the candidate graph together with its validated additive delta. + pub fn expand_with_proposal( + &self, + proposal: &GraphProposal, + catalog: &AgentCatalog, + limits: &GraphLimits, + ) -> Result { + limits.validate()?; + proposal.validate()?; + self.validate_agents(catalog)?; + + let existing_task_count = self.tasks().len(); + if existing_task_count > limits.max_tasks { + return Err(OrchestrationError::new( + OrchestrationErrorKind::NodeBudgetExceeded, + format!( + "现有 task 数量 {} 已超过 maxTasks {}", + existing_task_count, limits.max_tasks + ), + )); + } + let resulting_task_count = existing_task_count + .checked_add(proposal.nodes.len()) + .ok_or_else(|| { + OrchestrationError::new( + OrchestrationErrorKind::NodeBudgetExceeded, + "proposal task 数量计算溢出", + ) + })?; + if resulting_task_count > limits.max_tasks { + return Err(OrchestrationError::new( + OrchestrationErrorKind::NodeBudgetExceeded, + format!( + "扩图后 task 数量 {} 超过 maxTasks {}", + resulting_task_count, limits.max_tasks + ), + )); + } + + let existing_ids = self + .tasks() + .iter() + .map(|task| task.id().to_string()) + .collect::>(); + let proposed_ids = proposal + .nodes + .iter() + .map(|node| node.id.clone()) + .collect::>(); + for node in &proposal.nodes { + if existing_ids.contains(&node.id) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DuplicateTask, + format!("proposal task 已存在于当前 graph：{}", node.id), + )); + } + if catalog.get(&node.agent_id).is_none() { + return Err(OrchestrationError::new( + OrchestrationErrorKind::UnknownAgent, + format!( + "proposal task {} 引用了未注册 Agent：{}", + node.id, node.agent_id + ), + )); + } + } + + let all_ids = existing_ids + .iter() + .chain(proposed_ids.iter()) + .cloned() + .collect::>(); + let existing_edges = dependency_edges(self); + let mut proposed_edges = BTreeSet::new(); + let mut dependencies_by_target = BTreeMap::>::new(); + for edge in &proposal.edges { + if !all_ids.contains(edge.from()) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::UnknownDependency, + format!( + "proposal edge {} -> {} 引用了未知依赖：{}", + edge.from(), + edge.to(), + edge.from() + ), + )); + } + if !all_ids.contains(edge.to()) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::UnknownTask, + format!( + "proposal edge {} -> {} 引用了未知目标 task：{}", + edge.from(), + edge.to(), + edge.to() + ), + )); + } + if !proposed_ids.contains(edge.to()) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::ExistingTaskMutation, + format!( + "proposal edge {} -> {} 不能修改已有 task 的依赖", + edge.from(), + edge.to() + ), + )); + } + let edge_key = (edge.from().to_string(), edge.to().to_string()); + if !proposed_edges.insert(edge_key.clone()) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DuplicateEdge, + format!("proposal edge 重复：{} -> {}", edge.from(), edge.to()), + )); + } + if existing_edges.contains(&edge_key) { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DuplicateEdge, + format!("proposal edge 已存在：{} -> {}", edge.from(), edge.to()), + )); + } + dependencies_by_target + .entry(edge.to().to_string()) + .or_default() + .push(edge.from().to_string()); + } + + let resulting_edge_count = self + .edge_count() + .checked_add(proposal.edges.len()) + .ok_or_else(|| { + OrchestrationError::new( + OrchestrationErrorKind::EdgeBudgetExceeded, + "proposal edge 数量计算溢出", + ) + })?; + if resulting_edge_count > limits.max_edges { + return Err(OrchestrationError::new( + OrchestrationErrorKind::EdgeBudgetExceeded, + format!( + "扩图后 dependency edge 数量 {} 超过 maxEdges {}", + resulting_edge_count, limits.max_edges + ), + )); + } + + validate_out_degree(self, &proposal.edges, limits.max_out_degree)?; + + let mut tasks = self.tasks().to_vec(); + let added_task_ids = proposal + .nodes + .iter() + .map(|node| node.id.clone()) + .collect::>(); + for node in &proposal.nodes { + let dependencies = dependencies_by_target.remove(&node.id).unwrap_or_default(); + tasks.push(TaskNode::try_new( + node.id.clone(), + node.agent_id.clone(), + TaskStatus::Pending, + dependencies, + )?); + } + + // TaskGraph::try_new performs the final unknown-dependency and cycle + // checks over the complete candidate, so no partially built graph can + // escape this method. + let candidate = Self::try_new(self.goal().to_string(), tasks)?; + if candidate.depth() > limits.max_depth { + return Err(OrchestrationError::new( + OrchestrationErrorKind::DepthBudgetExceeded, + format!( + "扩图后 graph depth {} 超过 maxDepth {}", + candidate.depth(), + limits.max_depth + ), + )); + } + + Ok(GraphExpansion { + graph: candidate, + added_task_ids, + added_edges: proposal.edges.clone(), + }) + } + + /// Parameter-order variant for hosts that keep limits before the catalog. + pub fn apply_proposal_with_limits( + &self, + proposal: &GraphProposal, + limits: &GraphLimits, + catalog: &AgentCatalog, + ) -> Result { + self.apply_proposal(proposal, catalog, limits) + } + + /// Short alias for [`TaskGraph::apply_proposal`]. + pub fn expand( + &self, + proposal: &GraphProposal, + catalog: &AgentCatalog, + limits: &GraphLimits, + ) -> Result { + self.apply_proposal(proposal, catalog, limits) + } +} + +fn dependency_edges(graph: &TaskGraph) -> BTreeSet<(String, String)> { + graph + .tasks() + .iter() + .flat_map(|task| { + task.dependencies() + .iter() + .map(|dependency| (dependency.clone(), task.id().to_string())) + }) + .collect() +} + +fn validate_out_degree( + graph: &TaskGraph, + proposed_edges: &[GraphEdge], + max_out_degree: usize, +) -> Result<(), OrchestrationError> { + let mut out_degree = BTreeMap::::new(); + for (from, _) in dependency_edges(graph) { + let count = out_degree.entry(from.clone()).or_default(); + *count = count.checked_add(1).ok_or_else(|| { + OrchestrationError::new( + OrchestrationErrorKind::FanOutBudgetExceeded, + format!("task {from} 的 fan-out 数量计算溢出"), + ) + })?; + } + for edge in proposed_edges { + let count = out_degree.entry(edge.from().to_string()).or_default(); + *count = count.checked_add(1).ok_or_else(|| { + OrchestrationError::new( + OrchestrationErrorKind::FanOutBudgetExceeded, + format!("task {} 的 fan-out 数量计算溢出", edge.from()), + ) + })?; + } + if let Some((task_id, count)) = out_degree + .iter() + .find(|(_, count)| **count > max_out_degree) + { + return Err(OrchestrationError::new( + OrchestrationErrorKind::FanOutBudgetExceeded, + format!("task {task_id} 的 fan-out {count} 超过 maxOutDegree {max_out_degree}"), + )); + } + Ok(()) +} diff --git a/server-rs/crates/agent-runtime-orchestration/tests/dynamic_proposal.rs b/server-rs/crates/agent-runtime-orchestration/tests/dynamic_proposal.rs new file mode 100644 index 000000000..e12e6f60d --- /dev/null +++ b/server-rs/crates/agent-runtime-orchestration/tests/dynamic_proposal.rs @@ -0,0 +1,315 @@ +use agent_runtime_core::{AgentCatalog, AgentDescriptor}; +use agent_runtime_orchestration::{ + GraphEdge, GraphLimits, GraphProposal, OrchestrationErrorKind, PlanSelection, TaskGraph, + TaskNode, TaskProposal, TaskStatus, +}; + +fn task(id: &str, agent_id: &str, status: TaskStatus, dependencies: &[&str]) -> TaskNode { + TaskNode::try_new(id, agent_id, status, dependencies.iter().copied()).expect("valid task") +} + +fn catalog() -> AgentCatalog { + AgentCatalog::try_new([ + AgentDescriptor::try_new("researcher", "research", std::iter::empty::<&str>()) + .expect("researcher"), + AgentDescriptor::try_new("reviewer", "review", std::iter::empty::<&str>()) + .expect("reviewer"), + AgentDescriptor::try_new("writer", "writing", std::iter::empty::<&str>()).expect("writer"), + ]) + .expect("catalog") +} + +fn base_graph() -> TaskGraph { + TaskGraph::try_new( + "Review a document collection", + [ + task("collect", "researcher", TaskStatus::Completed, &[]), + task("draft", "writer", TaskStatus::Pending, &["collect"]), + ], + ) + .expect("base graph") +} + +fn proposal(nodes: &[(&str, &str)], edges: &[(&str, &str)]) -> GraphProposal { + GraphProposal::try_new( + nodes + .iter() + .map(|(id, agent)| TaskProposal::try_new(*id, *agent).expect("valid proposal node")), + edges + .iter() + .map(|(from, to)| GraphEdge::try_new(*from, *to).expect("valid proposal edge")), + ) + .expect("valid proposal") +} + +#[test] +fn llm_proposal_creates_a_new_pending_subgraph_without_mutating_the_old_graph() { + let graph = base_graph(); + let candidate = graph + .apply_proposal( + &proposal( + &[("review", "reviewer"), ("publish", "writer")], + &[ + ("collect", "review"), + ("draft", "publish"), + ("review", "publish"), + ], + ), + &catalog(), + &GraphLimits::default(), + ) + .expect("proposal should be accepted"); + + assert_eq!(graph.task_count(), 2); + assert_eq!(graph.edge_count(), 1); + assert_eq!(candidate.task_count(), 4); + assert_eq!(candidate.edge_count(), 4); + assert_eq!( + candidate.get("review").expect("review task").status(), + TaskStatus::Pending + ); + assert_eq!( + candidate.get("review").expect("review task").dependencies(), + &["collect".to_string()] + ); + assert_eq!( + candidate + .get("publish") + .expect("publish task") + .dependencies(), + &["draft".to_string(), "review".to_string()] + ); + assert_eq!(candidate.ready_task_ids(), vec!["draft", "review"]); + let plan = candidate + .plan(PlanSelection::All) + .expect("expanded graph should produce dependency waves"); + assert_eq!( + plan.dependency_waves(), + &[ + vec!["collect".to_string()], + vec!["draft".to_string(), "review".to_string()], + vec!["publish".to_string()], + ] + ); +} + +#[test] +fn expansion_report_contains_only_the_validated_delta() { + let graph = base_graph(); + let change = proposal(&[("review", "reviewer")], &[("collect", "review")]); + let expansion = graph + .expand_with_proposal(&change, &catalog(), &GraphLimits::default()) + .expect("proposal should be accepted"); + + assert_eq!(expansion.added_task_ids(), ["review"]); + assert_eq!(expansion.added_edges(), change.edges()); + assert_eq!( + expansion.graph().get("review").map(TaskNode::id), + Some("review") + ); +} + +#[test] +fn proposal_rejects_unknown_agents_and_keeps_the_current_graph_intact() { + let graph = base_graph(); + let error = graph + .apply_proposal( + &proposal(&[("review", "unknown-agent")], &[]), + &catalog(), + &GraphLimits::default(), + ) + .expect_err("unknown agent"); + assert_eq!(error.kind(), OrchestrationErrorKind::UnknownAgent); + assert_eq!(graph.task_count(), 2); + assert!(graph.get("review").is_none()); +} + +#[test] +fn proposal_cycle_is_rejected_atomically() { + let graph = base_graph(); + let error = graph + .apply_proposal( + &proposal( + &[("left", "researcher"), ("right", "reviewer")], + &[("left", "right"), ("right", "left")], + ), + &catalog(), + &GraphLimits::default(), + ) + .expect_err("cycle"); + assert_eq!(error.kind(), OrchestrationErrorKind::Cycle); + assert_eq!(graph.task_count(), 2); + assert!(graph.get("left").is_none()); +} + +#[test] +fn proposal_cannot_add_a_prerequisite_to_an_existing_task() { + let graph = base_graph(); + let error = graph + .apply_proposal( + &proposal(&[("review", "reviewer")], &[("review", "draft")]), + &catalog(), + &GraphLimits::default(), + ) + .expect_err("existing task mutation"); + assert_eq!(error.kind(), OrchestrationErrorKind::ExistingTaskMutation); + assert_eq!( + graph.get("draft").expect("draft").dependencies(), + &["collect".to_string()] + ); +} + +#[test] +fn proposal_limits_cover_nodes_edges_depth_and_fan_out() { + let graph = base_graph(); + let limits = GraphLimits::new(8, 8, 8, 1); + let fan_out = graph + .apply_proposal( + &proposal( + &[("review", "reviewer"), ("verify", "reviewer")], + &[("collect", "review"), ("collect", "verify")], + ), + &catalog(), + &limits, + ) + .expect_err("fan-out budget"); + assert_eq!(fan_out.kind(), OrchestrationErrorKind::FanOutBudgetExceeded); + + let node_budget = graph + .apply_proposal( + &proposal(&[("review", "reviewer"), ("verify", "reviewer")], &[]), + &catalog(), + &GraphLimits::new(3, 8, 8, 8), + ) + .expect_err("node budget"); + assert_eq!( + node_budget.kind(), + OrchestrationErrorKind::NodeBudgetExceeded + ); + + let edge_budget = graph + .apply_proposal( + &proposal( + &[("review", "reviewer"), ("verify", "reviewer")], + &[("collect", "review"), ("collect", "verify")], + ), + &catalog(), + &GraphLimits::new(8, 2, 8, 8), + ) + .expect_err("edge budget"); + assert_eq!( + edge_budget.kind(), + OrchestrationErrorKind::EdgeBudgetExceeded + ); +} + +#[test] +fn strict_json_round_trips_proposals_and_rejects_unknown_fields() { + let change = proposal(&[("review", "reviewer")], &[("collect", "review")]); + let json = serde_json::to_value(&change).expect("serialize proposal"); + assert_eq!( + json, + serde_json::json!({ + "nodes": [{"id": "review", "agentId": "reviewer"}], + "edges": [{"from": "collect", "to": "review"}] + }) + ); + let decoded: GraphProposal = serde_json::from_value(json).expect("decode proposal"); + assert_eq!(decoded, change); + + let unknown = serde_json::from_str::( + r#"{"nodes":[{"id":"review","agentId":"reviewer","title":"not allowed"}],"edges":[]}"#, + ) + .expect_err("unknown proposal field"); + assert!(unknown.to_string().contains("unknown field")); + + let graph = base_graph(); + let graph_json = serde_json::to_value(&graph).expect("serialize graph"); + let restored: TaskGraph = serde_json::from_value(graph_json).expect("decode graph"); + assert_eq!(restored, graph); + + let limits = GraphLimits::default(); + let limits_json = serde_json::to_value(limits).expect("serialize limits"); + assert_eq!( + limits_json, + serde_json::json!({ + "maxTasks": 128, + "maxEdges": 512, + "maxDepth": 32, + "maxOutDegree": 32 + }) + ); + let restored_limits: GraphLimits = serde_json::from_value(limits_json).expect("decode limits"); + assert_eq!(restored_limits, limits); + + let unknown_limits = serde_json::from_str::( + r#"{"maxTasks":1,"maxEdges":1,"maxDepth":1,"maxOutDegree":1,"extra":true}"#, + ) + .expect_err("unknown limits field"); + assert!(unknown_limits.to_string().contains("unknown field")); +} + +#[test] +fn proposal_rejects_duplicate_edges_unknown_endpoints_and_invalid_limits() { + let duplicate = GraphProposal::try_new( + [TaskProposal::try_new("review", "reviewer").expect("node")], + [ + GraphEdge::try_new("collect", "review").expect("edge"), + GraphEdge::try_new("collect", "review").expect("edge"), + ], + ) + .expect_err("duplicate edge"); + assert_eq!(duplicate.kind(), OrchestrationErrorKind::DuplicateEdge); + + let unknown = base_graph() + .apply_proposal( + &proposal(&[("review", "reviewer")], &[("missing", "review")]), + &catalog(), + &GraphLimits::default(), + ) + .expect_err("unknown edge source"); + assert_eq!(unknown.kind(), OrchestrationErrorKind::UnknownDependency); + + let invalid_limits = GraphLimits::try_new(0, 1, 1, 1).expect_err("zero limit"); + assert_eq!(invalid_limits.kind(), OrchestrationErrorKind::InvalidLimits); +} + +#[test] +fn graph_reports_layer_depth_and_fan_out() { + let graph = TaskGraph::try_new( + "depth", + [ + task("root", "researcher", TaskStatus::Pending, &[]), + task("middle", "reviewer", TaskStatus::Pending, &["root"]), + task("leaf", "writer", TaskStatus::Pending, &["middle"]), + ], + ) + .expect("graph"); + assert_eq!(graph.depth(), 3); + assert_eq!(graph.fan_out("root"), Some(1)); + assert_eq!(graph.node_count(), 3); + assert_eq!(graph.fan_out("missing"), None); +} + +#[test] +fn proposal_rejects_empty_payload_and_depth_overflow() { + let empty = GraphProposal::try_new( + std::iter::empty::(), + std::iter::empty::(), + ) + .expect_err("empty proposal"); + assert_eq!(empty.kind(), OrchestrationErrorKind::EmptyProposal); + + let graph = base_graph(); + let error = graph + .apply_proposal( + &proposal( + &[("review", "reviewer"), ("publish", "writer")], + &[("collect", "review"), ("review", "publish")], + ), + &catalog(), + &GraphLimits::new(8, 8, 2, 8), + ) + .expect_err("candidate depth should exceed the limit"); + assert_eq!(error.kind(), OrchestrationErrorKind::DepthBudgetExceeded); +} -- 2.52.0