diff --git a/.gitignore b/.gitignore index 73082d3cb..6b47da089 100644 --- a/.gitignore +++ b/.gitignore @@ -28,6 +28,7 @@ target # Local dev files opencode-dev +packages/opencode/script/config.json UPCOMING_CHANGELOG.md logs/ *.bun-build @@ -46,4 +47,5 @@ tsconfig.tsbuildinfo .agents .claude .opencode/commands/ -.opencode/skills \ No newline at end of file +.opencode/skills +.qoder \ No newline at end of file diff --git a/docs/dag-bug-condition-skip-as-satisfied.md b/docs/dag-bug-condition-skip-as-satisfied.md deleted file mode 100644 index a4a301439..000000000 --- a/docs/dag-bug-condition-skip-as-satisfied.md +++ /dev/null @@ -1,507 +0,0 @@ -# DAG 模组缺陷与改进清单（P0–P2） - -> **来源**：DAG 模组全盘审计（性能 / 运行闭环 / 场景覆盖 / TUI）+ 压测工作流 `dag_067ef539cffe6fbuucCw7L5nko` 实证 -> **报告日期**：2026-07-25（P0-1 实证）/ 2026-07-27（全量审计整合） -> **分级标准**：P0 = 语义正确性破坏或核心链路死路；P1 = 静默行为偏差、性能热点、契约自相矛盾；P2 = 语义弱化、死代码、UX 缺口 - ---- - -## 总览 - -| 编号 | 级别 | 标题 | 类型 | -|------|------|------|------| -| [P0-1](#p0-1) | P0 | condition_false 被当作 satisfied，门禁拒绝被下游完全无视 | 正确性（已实证） | -| [P0-2](#p0-2) | P0 | max_concurrency 只限流 LLM 调用，子会话被急切创建，QUEUED 是死状态 | 正确性 + 性能（**已修复：PR #122 queued admission**） | -| [P0-3](#p0-3) | P0 | 连续单步（step）是状态机死路 | 正确性 | -| [P1-1](#p1-1) | P1 | create 不校验 depends_on 引用 / 重复节点 ID / max_total_nodes | 正确性 | -| [P1-2](#p1-2) | P1 | sanitizer 破坏性替换损坏 diff/代码产出物，与 diff-review 契约矛盾 | 正确性（**已修复：PR #121 字段级豁免**） | -| [P1-3](#p1-3) | P1 | spawnReady 每节点重复全量读，O(ready × nodes) 查询 | 性能（**已修复：PR #121 快照复用**） | -| [P1-4](#p1-4) | P1 | getWorkflowSummaries JS 聚合 + 事件热路径前置查询 | 性能（**已修复：PR #121 SQL 聚合**） | -| [P1-5](#p1-5) | P1 | 失败节点无法通过 replan restart 重跑（必须换新 ID），工具描述未告知 | 使用陷阱 | -| [P2-1](#p2-1) | P2 | required 节点失败 → 工作流终态是 cancelled 而非 failed | 语义 | -| [P2-2](#p2-2) | P2 | 崩溃恢复对 running 节点一律判失败，required 级联即整流程报废 | 语义（**已修复：recovery-pause**） | -| [P2-3](#p2-3) | P2 | orchestrator_unresponsive 兜底的触发时机依赖未钉死的 promptIfIdle 语义 | 风险（**已验证关闭+契约测试**） | -| [P2-4](#p2-4) | P2 | 死代码/死状态：ViolationTable、ARCHIVED、NodeStatus.PAUSED/ABORTED | 清理 | -| [P2-5](#p2-5) | P2 | inline 模板走临时文件写读删，spawn 热路径纯多余 I/O | 性能 | -| [P2-6](#p2-6) | P2 | condition DSL 表达力不足，且只能引用直接依赖 | 场景 | -| [P2-7](#p2-7) | P2 | HTTP API 不对称：无 start / extend | 场景（**已修复：PR #123**） | -| [P2-8](#p2-8) | P2 | TUI：Inspector 无滚动、列表截断与导航不一致、无 step 控制、配色两套 | UX（**已修复：PR #120 + #124 残余项**） | -| [P2-9](#p2-9) | P2 | summary 进度不计 skipped，工作流"跑完了但分子不满" | UX（**已修复：PR #122 计数 + #124 展示**） | -| [P2-10](#p2-10) | P2 | pause 语义（不停止在跑节点）无任何文档/UI 提示 | UX | - ---- - - -## P0-1：condition_false 被当作 satisfied，门禁拒绝被下游完全无视 - -> **状态**：已确认（经工作流 `dag_067ef539cffe6fbuucCw7L5nko` 终态验证） -> **严重度**：Critical — 语义正确性缺陷，门禁形同虚设 - -### 摘要 - -当一个节点的 `condition` 求值为 false 时，节点发布 `NodeSkipped(condition_false)`，但调度层把 **skipped 与 completed 等价处理为 `satisfied`**。由于调度器把 `satisfied` 集合视为"依赖已满足"，**所有依赖该节点的下游节点会照常调度执行**——即使该节点是一个质量门禁且刚刚拒绝了放行。 - -结果是：门禁返回 `REVISE`/`REJECT` → 实现节点被 condition 跳过 → 但实现节点的下游（集成、验证、审计）仍然全部执行，最终审计甚至返回 `ACCEPT`。**整个质量门控链路形同虚设。** - -### 复现 - -#### 工作流信息 - -| 字段 | 值 | -|------|-----| -| Workflow ID | `dag_067ef539cffe6fbuucCw7L5nko` | -| Title | engine-stress-review-pipeline | -| Parent Session | `ses_067f06f1bffeAmDFfH51NmNBl5` | -| 模式 | standard | -| 终态 | completed | - -#### 图结构（相关部分） - -``` -quality-gate (required, output_schema: verdict) - ├── implement-core (condition: quality-gate.output.verdict == "ACCEPT") - ├── implement-docs (condition: quality-gate.output.verdict == "ACCEPT") - │ - implement-cli (condition: quality-gate.output.verdict == "ACCEPT", depends_on: implement-core) - │ - integrate (depends_on: implement-cli, implement-docs) ← 无 condition - ├── verify-unit (depends_on: integrate) - └── verify-e2e (depends_on: integrate) - │ - diff-review (depends_on: verify-unit, verify-e2e, output_schema: verdict) - │ - final-audit (required, depends_on: diff-review, output_schema: verdict) - │ - tail-telemetry (depends_on: final-audit) -``` - -#### 预期行为 - -quality-gate 返回 `REVISE` → condition 求值 false → implement-* 被跳过 → **整条下游链不应执行**（无实现产物可集成/验证/审计）。 - -#### 实际行为（终态快照） - -| 节点 | 状态 | 说明 | -|------|------|------| -| quality-gate | completed, verdict=**REVISE** | 子 agent 合法分析仲裁结果后拒绝放行 | -| implement-core | **skipped** (condition_false) | 正确跳过 | -| implement-docs | **skipped** (condition_false) | 正确跳过 | -| implement-cli | **skipped** (condition_false) | 正确跳过 | -| integrate | **completed** | ❌ 依赖的两个节点都被 skip，却仍然执行 | -| verify-unit | **completed** | ❌ 不应该运行 | -| verify-e2e | **completed** | ❌ 不应该运行 | -| diff-review | **completed**, verdict=ACCEPT | ❌ 不应该运行 | -| final-audit | **completed**, verdict=**ACCEPT** | ❌ 门禁拒绝了，审计却通过了 | -| tail-telemetry | **completed** | ❌ 不应该运行 | - -**工作流整体 completed**——门禁拒绝被完全无视，整条链跑完并"通过"。 - -### 根因分析（已修正定位） - -> 早期版本将根因定位在 `loop.ts` condition 分支内联 `markSatisfied`——**不准确**。当前源码中 condition 分支只发布 NodeSkipped 事件；skip≡satisfied 的合流发生在下述 **三个独立位置**，修复必须同时覆盖，否则任一遗漏路径都会在重建/单步时复现该缺陷。 - -**位置 ①：NodeSkipped 事件处理器（活跃路径）** — `packages/opencode/src/dag/runtime/loop.ts:318-353` - -```ts -for (const def of [DagEvent.NodeCompleted, DagEvent.NodeSkipped]) { // ← skip 与 complete 共用一个处理器 - yield* events.subscribe(def).pipe( - ... - if (entry.runtime.isActive(evt.data.nodeID as string)) { - entry.runtime.markSatisfied(evt.data.nodeID as string) // ← skip 被当作 satisfied - if (!entry.runtime.isStepMode()) yield* spawnReady(dagID) // ← 随即放行下游 - } -``` - -**位置 ②：`toSchedulingNodes` 重建映射（恢复/replan 重建路径）** — `loop.ts:36-51` - -```ts -export const SUCCESS_TERMINAL = new Set(["completed", "skipped", "aborted"]) // ← skip 归入成功终态 - -export function toSchedulingNodes(nodes) { - return nodes.map((n) => ({ - ... - status: SUCCESS_TERMINAL.has(n.status) - ? ("satisfied" as const) // ← 从 DB 重建 WorkflowRuntime 时 skip 再次变 satisfied - : ... -``` - -**位置 ③：`Dag.step` 的同款内联映射** — `packages/opencode/src/dag/dag.ts:356-385`（单步计算 ready 集合时复制了同一映射） - -**调度器侧的放大机制** — `packages/core/src/dag/core/scheduling.ts:50-109` - -```ts -markSatisfied(nodeID: string): void { - this.satisfied.add(nodeID) // ← skip 的节点进了这里 - ... -} - -getReadyNodes(): string[] { - const ready = this.graph - .getExecutableNodes(new Set([ - ...this.satisfied, // ← 下游据此判定依赖"已满足" - ...[...this.unsatisfied].filter((id) => !this.required.has(id)), - ])) - ... -} -``` - -`satisfied` 集合同时承载"成功完成"与"被 condition 跳过"两种语义，调度器无法区分。**WorkflowRuntime 只有 satisfied/unsatisfied/running/pending 四态，没有 skipped**——节点 STATUS 层有 `SKIPPED`（`types.ts:37`）和 `SkipReason.CONDITION_FALSE`（`types.ts:49-56`），但调度态丢失了这一信息。 - -#### 语义断裂链 - -``` -condition == false - → NodeSkipped(condition_false) 发布 - → 事件处理器 markSatisfied（skip 当 success） [位置①] - → 重建时 SUCCESS_TERMINAL 映射为 satisfied [位置②③] - → 下游 getExecutableNodes 判定依赖已满足 - → resolveInputMapping 注入 "Dependency skipped: no output" 文本作为下游输入 - → 下游带着占位文本照常执行 - → 整条链跑完，门禁形同虚设 -``` - -注意最后一环：下游并非"拿不到输入而失败"，而是 `resolveInputMapping`（`loop.ts:112-125`）会把 skip 依赖降级为一段说明文本注入 prompt——这是为"可选分支降级"设计的机制，但在门禁场景下变成了"用占位文本继续跑完全链"。 - -#### 设计张力：为什么不能简单地把 skip 全部级联 - -| 场景 | 当前行为（skip≡satisfied） | 期望行为 | -|------|---------------------------|----------| -| 门禁拒绝 → 实现链跳过 | 下游照常执行 ❌ | 下游级联 skip | -| 可选分支跳过 → 主链继续（fan-in 有其他 satisfied 依赖） | 下游执行 ✅ | 下游执行 ✅（合法场景，必须保留） | -| condition 跳过 → fan-in 汇总 | fan-in 收到占位文本仍执行 | 至少让 fan-in 可区分 skip 与 success | - -关键区分点：**下游是否"纯依赖" skipped 节点**。有任一 satisfied 依赖的 fan-in 继续执行是合法降级；全部依赖都 skip 的子树继续执行则是缺陷。 - -### 修复建议 - -#### 方案 A：引入 skipped 调度态 + 纯依赖级联（推荐） - -1. `WorkflowRuntime` 增加 `private readonly skipped: Set` 与 `markSkipped(nodeID)` -2. `getReadyNodes` 判定依赖满足时，`{satisfied ∪ skipped}` 仍可解锁下游，**但**一个节点若其依赖全部 ∈ skipped（无任何 satisfied），则该节点自动级联 skip（发布 `NodeSkipped(orphan_cascade)` 复用现有 SkipReason） -3. `isComplete()` 将 skipped 计入终态集合（保持现有完成判定不回归） -4. **三处合流点同步修改**：位置① NodeSkipped 处理器改调 `markSkipped`；位置② `SUCCESS_TERMINAL` 拆分为 `{completed, aborted} → satisfied`、`{skipped} → skipped`；位置③ `Dag.step` 内联映射同步（建议顺手消除该重复，复用 `toSchedulingNodes`） -5. 可选：节点级 `skip_propagation: cascade | allow_downstream` 配置覆盖默认级联策略 - -#### 方案 B：condition_false → markUnsatisfied（最小改动，不推荐） - -`markUnsatisfied` 仅对 required 节点级联（`scheduling.ts:60-75`），非必需节点的下游会永远卡 pending 导致 `isComplete()` 永假、工作流悬挂；补非必需级联变体后实质上等于方案 A 的劣化版，且把 skip 混入 unsatisfied 又制造新的语义合流。 - -#### 方案 C：编排层声明传播策略 - -`condition_propagation: skip_subtree` 类声明。可作为方案 A 第 5 步的配置面，不建议单独作为修复（默认行为仍是坏的）。 - -**推荐方案 A**——级联只影响纯依赖 skipped 的子树，保留"可选分支跳过→主链继续"合法场景；且三处合流点一次收敛，重建/单步路径不留后门。 - -#### 回归测试清单 - -- 门禁拒绝 → 全下游级联 skip、工作流 completed（非 cancelled） -- fan-in 一个依赖 skip 一个 completed → fan-in 照常执行且输入含 skip 说明文本 -- 级联 skip 后进程重启（走 `toSchedulingNodes` 重建）→ 不复活下游 -- stepping 模式下 condition_false → 级联 skip 不触发 auto-advance - ---- - - -## P0-2：max_concurrency 只限流 LLM 调用，子会话被急切创建，QUEUED 是死状态 - -> **状态**：已修复（PR #122：durable `dag.node.queued` 事件 + 子会话/NodeStarted 移入 permit 内；deadline 仍从录取时刻起算；queued 崩溃重启按 pending 重新排队不计 ownershipLost） - -**位置**：`packages/opencode/src/dag/runtime/spawn.ts:97-156`、`loop.ts:79-230` - -`spawnReady` 对每个 ready 节点立即执行 `spawnNode`，而 `sessions.create`（L97）与 `NodeStarted` 事件发布（L119）都发生在 **semaphore 取许可之前**（L156 才 `take(1)`）。后果链： - -- 100 个 ready 节点 → 瞬间创建 100 个子会话、发布 100 条 `NodeStarted`、100 个节点在 DB 中全部变为 `running` -- `NodeStatus.QUEUED` 全代码库**无任何发布点**（`transitionToNodeEvent` 对 QUEUED 返回 null）——状态机死状态 -- UI（sidebar `▶N running`、Inspector spinner）显示全部"运行中"，用户无法分辨真实并发是 5 -- 排队等待计入 deadline 是有意设计（正确），但大扇出下队尾节点"未执行先超时"，且全程显示 running - -**修复**：会话创建与 `NodeStarted` 移入 permit 内；取 permit 前发布 queued 语义（新增 NodeQueued 事件或延迟 NodeStarted）。注意联动：`recovery.ts` 对 running 节点的收敛逻辑、deadline 计算起点（保持 spawn 时刻不变）都要复核。 - ---- - - -## P0-3：连续单步（step）是状态机死路 - -**位置**：`packages/core/src/dag/core/types.ts:187-188`、`dag.ts:356-385`、`loop.ts:446-463` - -`dag.step` 的守卫要求 `当前状态 → STEPPING` 是合法迁移，而 `getValidNextWorkflowStatuses(STEPPING) = [RUNNING, PAUSED, COMPLETED, FAILED, CANCELLED]`——**不含 STEPPING**。完整时序推演： - -1. `running` 下 step → 状态 `stepping`，运行 1 个节点 ✓ -2. 节点完成，stepMode 阻止 auto-advance（正确）✓ -3. 再次 step → `InvalidTransitionError(stepping → stepping)` ✗ -4. 唯一出路 `resume` → 但 `WorkflowResumed` 处理器清 stepMode 并 `spawnReady` **全量并发放行** ✗ - -即"逐节点调试"只能走一步；第二步要么报错要么失控全放。`dag-step-semantics.test.ts` 只测了第一步，无连续 step 用例，故未暴露。 - -**修复**：迁移表允许 `STEPPING → STEPPING`（幂等 re-step，projector 的 `WorkflowStepped` 投影 from 列表同步加 `"stepping"`），或 stepped 节点完成后自动回落 running。TUI 侧同步补 step 命令（见 P2-8）。 - ---- - - -## P1-1：create 不校验 depends_on 引用 / 重复节点 ID / max_total_nodes - -**位置**：`dag.ts:296-317`（create 校验段）、`scheduling.ts:12-21`（buildGraph） - -- **悬空依赖静默丢边**：`buildGraph` 对不存在的依赖 `if (graph.hasNode(dep))` 静默跳过 → **依赖 ID 打错字的节点变成根节点立即执行**。replan 路径有完整引用校验（`planReplan` 第 2 步），create 没有——不对称。 -- **重复节点 ID 静默合并**：`addNode` 去重、projector `onConflictDoUpdate` 覆盖；`ErrorCode.DUPLICATE_NODE_NAME` 定义了但无人使用。 -- **max_total_nodes 只在 replan 检查**（`dag.ts:462`）：初始 config 500 节点直接放行，与 P0-2 叠加 = 瞬间 500 个子会话。 - -**修复**：create 增加三项校验，全部拒绝式（fail fast），错误信息对齐 replan 的措辞。 - ---- - - -## P1-2：sanitizer 破坏性替换损坏 diff/代码产出物，与 diff-review 契约矛盾 - -> **状态**：已修复（PR #121：`reviewEvidenceKeys` 推导的实现证据字段原文保留，`` 定界包裹；其余字段防注入面不变） - -**位置**：`packages/opencode/src/dag/templates/sanitize.ts:16-28`、`loop.ts:130` - -`sanitize` 把所有 ``` 替换为 ``、行首 `system:` 替换为 `[REDACTED]:`、`you are now a` 替换为 `[REDACTED]`，并作用于**所有上游节点输出**（`resolvedMapping = sanitizeInput(...)`）。而 `review-lifecycle.ts` 的 diff-review 契约**要求**下游收到真实的 diff/patch 工件——任何含 code fence 的 diff、含 "system:" 的日志都会被静默改写，**审查者审的是被篡改的补丁**。防注入与工件保真当前不可兼得。 - -**修复方向**（按侵入度递增）： -1. 对 `review.phase == "diff"` 声明的 implementation 工件字段豁免 sanitize -2. 改破坏性替换为包裹式中和（如 `` 定界 + 转义），保留原文 -3. 按 worker_type / 字段级 sanitize 策略配置 - ---- - - -## P1-3：spawnReady 每节点重复全量读，O(ready × nodes) 查询 - -> **状态**：已修复（PR #121：每轮调度一次 getNodes 快照；ready 节点依赖已终态、本轮内输出不可变） - -**位置**：`loop.ts:89`（condition 分支）、`loop.ts:111`（input_mapping 分支） - -每个 ready 节点最多两次 `store.getNodes(dagID)` 全表读，一轮调度 N 个 ready 节点 = 最多 2N 次全量查询。**修复**：每轮 `spawnReady` 开头读一次 nodes 快照并复用（condition 求值与 mapping 解析都只需要终态依赖的 output，快照一致性足够）。 - ---- - - -## P1-4：getWorkflowSummaries JS 聚合 + 事件热路径前置查询 - -> **状态**：已修复（PR #121：GROUP BY 计数下沉 SQL；无 sessionID 事件的 getWorkflow 前置查询移入 dagID 级去抖动窗口之后） - -**位置**：`packages/core/src/dag/store.ts:226-257`、`summary-publisher.ts:104-120` - -- `getWorkflowSummaries` 拉取 session 下**所有**节点行到内存计数；每个 `dag.*` 事件后触发（50ms 去抖动缓解突发）。长会话累积多个 100 节点工作流后，每次事件突发 = 数千行读取。改 `GROUP BY workflow_id, status` 一行等价。 -- 节点事件不带 sessionID，publisher 在**去抖动之前**每事件做一次 `getWorkflow` 查询（L109-112）。100 节点事件突发 = 100 次前置查询。可加 dagID→sessionID 短 TTL 缓存，或把去抖动窗口提到查询之前。 - ---- - - -## P1-5：失败节点无法 replan restart 重跑，必须换新 ID——工具描述未告知 - -**位置**：`packages/core/src/dag/core/replan.ts:108-110`、`replan.ts:133`（终态 ignore） - -`planReplan` 规定 `restart` 仅对 **running** 节点合法；failed 等终态节点出现在 fragment 中直接进 ignore 桶。即**失败节点想重试必须换一个新节点 ID 重新添加**。这是符合"终态不可逆"铁律的有意设计，但 `workflow.ts` 工具 schema 的 restart 描述（"Re-spawn this running node"）没有把这个陷阱讲透——父 agent 大概率先试 restart 失败节点、收到 ignore 静默结果后困惑。 - -**修复**：restart 非 running 节点时返回**显式错误提示**（"failed 节点请以新 ID 添加替代节点"），而非静默 ignore；工具描述补一句陷阱说明。 - ---- - - -## P2-1：required 节点失败 → 工作流终态 cancelled 而非 failed - -**位置**：`loop.ts:238`（`checkCompletion`：`hasRequiredFailure() → dag.cancel`） - -必需节点失败的工作流终态是 `cancelled`，与用户主动取消不可区分；TUI 把 failed 标红、cancelled 置灰——**必需节点炸了显示为灰色**，归因被弱化。wake 消息同样只说 "Workflow cancelled"。建议改走 `dag.fail(dagID, "required node failed: ")`，同时更新依赖该语义的测试与工具描述（当前描述"If true and this node fails, the workflow is cancelled"需同步）。 - ---- - - -## P2-2：崩溃恢复对 running 节点一律判失败 - -> **状态**：已修复（recovery-pause，分支 `fix/dag-recovery-pause`） - -**位置**：`recovery.ts:93-99`（判罚保留）+ `loop.ts` `recoverWorkflow`（处置改变） - -**原病理**：ownership lost 的 running 节点一律 `nodeFailed("execution ownership lost")`，不重试——有意的保守设计。但叠加 required 级联（P2-1）后：**重启一次进程 = required 链上任何在跑节点失败 = 整个工作流终态报废**，且终态节点不可变、终态工作流不可 replan，父 agent 收到的是无可修复的死亡通知。 - -**否决的方案**（本文档旧版建议）：ownership lost 且 deadline 未到的节点回落 `pending` 由 spawnReady 自动重拾取。**自动重拾取就是未经显式控制的新 execution attempt**，正面违反模块自身契约（`loop.ts` 恢复注释："a new execution attempt must come from explicit workflow control"，与 `core/session/runner/llm.ts` 的恢复铁律同源；该铁律经 07-27 实证为有效——双代码锚点、AGENTS.md 更新后 47 提交零漂移；其解除路径是完成显式恢复设计而非删除条款），且有崩溃循环与子会话副作用重放风险。 - -**实施的修复（recovery-pause）**：崩溃判罚保留（发明性失败照旧 `failed`，铁律无损），但当恢复过程**发明**了失败（ownership lost / 无子会话 / deadline 离线到期）且工作流原为 running 时，转 `paused` 而非放任 spawnReady+checkCompletion 立即级联 skip 并焊死终态： - -- **关键洞察**：暂停发生在级联之前，下游节点保持 `pending`——pending 可被 replan 改线，failed/skipped 终态不可。旧行为下 n2 已 skipped、工作流已 failed，同一 replan 根本无法提交。 -- 父 agent 收到 durable NodeFailed wake（paused 工作流本就处于投递边界）+ actionable 指令，三选一：`replan`（新 id 替换节点 + 下游改线）→ `resume` 复活；直接 `resume`（接受失败语义，走 P2-1 归因终态）；`cancel`。 -- 再次崩溃时已 paused、无 running 节点，恢复幂等——无崩溃循环。 -- 配套修复一个被此设计暴露的既有缺陷：`WorkflowResumed` 处理器只调 `spawnReady` 不调 `checkCompletion`，全节点已终态的 paused 工作流 resume 后永久悬挂。 -- 回归测试：`dag-loop-recovery-integration.test.ts` 新增三组契约（pause-then-resume 终态归因 / 下游可 replan / 全终态 resume 不悬挂）。 - ---- - - -## P2-3：orchestrator_unresponsive 兜底的触发时机假设未钉死 - -> **状态**：已验证关闭（07-27 源码实证）+ 契约测试钉死 - -**位置**：`loop.ts`（`tryDeliverWake`） - -投递 wake 后循环立即重读批次，无未上报行且工作流停摆即 `dag.fail(dagID, "orchestrator_unresponsive")`。该逻辑隐含假设 `promptIfIdle` 等到父 turn 完整结束才 resolve。 - -**实证结果**（`prompt.ts` `promptIfIdle` 实现）：`state.startIfIdle` 返回 wait handle，末尾 `yield* wait.value` 阻塞至完整 `runLoop`（父 turn）完成——**误杀窗口不存在**，假设成立。 - -**回归防护**：该假设是 session 层契约而非 DAG 层可控行为，因此防护落在 `test/session/prompt.test.ts`（"idle-only prompt resolves only after the full provider turn completes"）：provider 流未完成时 promptIfIdle 必未 resolve，流完成后才 resolve Some。若未来有人把 promptIfIdle 改成 admission 即返回，此测试会先于线上误杀暴露。 - ---- - - -## P2-4：死代码 / 死状态清理 - -| 项 | 证据 | 处置建议 | -|---|---|---| -| `WorkflowViolationTable` | 只有读方法（listViolations/queryViolations/countBySeverity），全库无写入点，HTTP/TUI 不暴露 | 接上（sanitizer 命中、ceiling 命中、unresponsive 判罚天然是 violation）或删除 | -| `WorkflowStatus.ARCHIVED` | 迁移表允许终态→ARCHIVED，但无 archive 事件定义、无发布点 | 删除或补 archive API | -| `NodeStatus.PAUSED` / `ABORTED` | 无节点级暂停 API；aborted 无发布点 | 从迁移表移除或明确 roadmap | -| `NodeStatus.QUEUED` | 见 P0-2，随 P0-2 修复激活 | 激活 | - ---- - - -## P2-5：inline 模板走临时文件写读删 - -**位置**：`templates/resolve.ts:68-82`。注释自认"simulating the template-file read path"——spawn 热路径上 3 次纯多余磁盘 I/O + 无谓的失败面（tmpdir 权限/磁盘满）。直接用字符串。 - ---- - - -## P2-6：condition DSL 表达力不足，且只能引用直接依赖 - -**位置**：`runtime/eval.ts:32-54`、`loop.ts:89-104` - -- 仅支持单个二元比较（`a.output.x == v`），无 `&&`/`||`、无存在性判断、无 contains -- condition 的 outputs map 只装 **direct dependsOn**（`loop.ts:91-94`），引用间接上游会静默 undefined → 比较恒 false → 静默 skip（叠加 P0-1 后下游还照跑） -- 字符串与数字比较 `>` 会得到 NaN 比较恒 false，无告警 - -**建议**：至少补 `&&`/`||` 与 `exists()`；对引用了非直接依赖的 condition 在 create/replan 校验期报错（静默 false 是最坏的失败模式）。 - ---- - - -## P2-7：HTTP API 不对称——无 start / extend - -> **状态**：已修复（PR #123：`POST /dag` start + `control(extend)`，同时修正 control 响应 schema 剥掉 replan/extend 处置数组的既有缺陷；SDK regen + exercise 场景门禁） - -**位置**：`server/routes/instance/httpapi/handlers/dag.ts` - -control 支持 pause/resume/cancel/complete/step/replan，但**无 extend、无 start**——外部系统无法通过 HTTP 发起或追加工作流，只能由 agent 工具面发起。若 HTTP 面定位为完整控制面，需补齐并同步 SDK 再生成（`./packages/sdk/js/script/build.ts`）与 `test/server/httpapi-exercise` 场景。 - ---- - - -## P2-8：TUI Inspector / 面板缺陷合集 - -> **状态**：已修复（主体 PR #120 三界面对齐；残余 PR #124：restarted ×N 历史标注 / running·queued 节点 deadline 倒计时 / queued 专属 glyph 与计数） - -**位置**：`packages/tui/src/feature-plugins/system/dag-inspector.tsx`、`sidebar/dag-panel.tsx` - -| 问题 | 位置 | 说明 | -|---|---|---| -| 无滚动容器 | inspector L379-501 | 节点树 `` 平铺，40+ 节点溢出屏幕；键盘选中可移动到不可见区域（无 scroll-into-view） | -| 列表截断与导航不一致 | L341 `slice(0, 10)` vs L153-159 moveWorkflow 全量列表 | 第 11 个工作流可被选中但列表看不到 | -| 无 step 控制 | — | 后端有 stepping 状态、面板显示黄色 stepping，TUI 无法触发/继续单步（受 P0-3 制约） | -| replan/extend 不可见 | — | `replanAttempts`、节点取消/替换/重启历史、`WorkflowReplanned` 计数均不呈现 | -| 节点详情缺失 | — | 无输出预览、无耗时（started_at/completed_at 有数据不渲染）、无模型标注、无 deadline 倒计时 | -| footer 快捷键硬编码 | L505-517 | `p/r/x/↑↓/←→` 写死，命令实际走 `keybinds.gather("dag", ...)` 可重绑；close/enter 用了动态 `useCommandShortcut`，同文件内不一致 | -| 两套状态配色 | inspector L295-302 vs dag-panel L13-21 | inspector 缺 paused/stepping 分支（落默认色），panel 里是 warning 黄；running/pending/skipped/cancelled 在 inspector 全是 textMuted | - -**建议**：提取共享 status→color 映射；补 scrollbox + scroll-into-view；节点行加耗时/模型/输出摘要；footer 全部走 `useCommandShortcut`。 - ---- - - -## P2-9：summary 进度不计 skipped - -> **状态**：已修复（计数面 PR #122：summary 增加 skippedNodes/queuedNodes；展示面 PR #124：进度分子 = completed+skipped） - -**位置**：`store.ts:242-250`（只计 completed/running/failed） - -进度显示 `completed/total`，条件跳过的节点永远不进分子——"3/5 · completed" 的工作流看起来像没跑完。skipped 应并入完成侧或单独列出（`⊘N`），schema `DagWorkflowSummary` 加 `skippedNodes` 字段后需再生成 SDK。 - ---- - - -## P2-10：pause 语义无提示 - -pause 只停新 spawn，运行中的子会话继续跑（合理设计），但工具描述、HTTP 文档、TUI toast 均无一处说明"暂停 ≠ 停止正在跑的节点"。补一句话成本极低。 - ---- - -## 附带观察（非缺陷，交接留档） - -### 观察 1：final-audit 输出引用了错误的工作流 ID - -final-audit 的结构化输出为： -```json -{"verdict":"ACCEPT","summary":"...工作流 dag_0679ada90ffeAHIyYfUv8WlxE0 已完成执行，4 节点（discover → {migrate-a, migrate-b condition:false} → assemble fan-in）..."} -``` - -- 引用的 `dag_0679ada90ffeAHIyYfUv8WlxE0` **不是**本工作流（`dag_067ef539cffe6fbuucCw7L5nko`） -- 描述的节点（discover/migrate-a/migrate-b/assemble）**不存在于本图** -- 可能原因：(a) 子 agent 在测试场景下幻觉编造；(b) 跨工作流上下文污染 - -**建议**：若是 (b)，排查子会话上下文是否混入其他工作流数据；若是 (a)，压测应使用更具约束性的 prompt。 - -### 观察 2：此前"永久卡死"诊断已被推翻 - -工作流执行中途查询 status 时 7 个节点显示 pending，当时诊断为"永久卡死"，但工作流最终 completed——这些节点随后被正常调度。**中途 status 快照不能判定卡死**，需配合 `isComplete()` / `hasRunning()` 等运行时方法。（另见 P0-2：大扇出下 pending/running 的显示语义本身有失真。） - ---- - -## 修复优先级路线 - -``` -第一批（正确性，可并行）： - P0-1 skipped 调度态 + 级联 ← 三处合流点一次收敛 - P0-3 STEPPING→STEPPING 迁移 ← 一行迁移表 + projector from 列表 - P1-1 create 三项校验 ← 对齐 replan 已有逻辑 - -第二批（需要设计推演）： - P0-2 permit 内建会话 ← 联动 deadline/recovery/QUEUED 语义 - P1-2 sanitize 豁免策略 ← 联动 review 契约 - P1-5 restart 显式报错 - -第三批（性能 + 清理 + UX）： - P1-3 / P1-4 / P2-* - -已完成（全部 16 项清零）： - PR #119：P0-1 / P0-3 / P1-1 / P1-5 / P2-1 / P2-2（recovery-pause）/ P2-3（验证关闭+契约测试）/ P2-4 / P2-5 / P2-6 / P2-10 - PR #120：P2-8 主体（TUI 三界面对齐） - PR #121（B 批）：P1-2（sanitize 字段级豁免）/ P1-3（spawn 快照）/ P1-4（SQL 聚合 + 去抖动前置查询消除） - PR #122（A 批）：P0-2（durable queued 录取 + permit 内建会话）/ P2-9 计数面（skipped/queued 进 summary） - PR #123（C 批）：P2-7（HTTP start/extend + control 响应 schema 修正 + SDK regen） - PR #124（D 批）：P2-8 残余（restarted ×N 历史标注 / deadline 倒计时 / queued 呈现）+ P2-9 展示语义（completed+skipped 进度分母） - -合入顺序（stacked）：#119 → #121 → #122 → #123 → #124；#120 独立，于 #124 内已合流。 -``` - ---- - -## 相关文件 - -| 文件 | 关键行 | 内容 | -|------|--------|------| -| `packages/opencode/src/dag/runtime/loop.ts` | 36-51, 318-353 | SUCCESS_TERMINAL 映射 / NodeSkipped→markSatisfied（**P0-1 根因**） | -| `packages/opencode/src/dag/runtime/loop.ts` | 88-104, 112-130 | condition 求值分支 / resolveInputMapping + sanitize 注入点 | -| `packages/opencode/src/dag/runtime/spawn.ts` | 97-156 | 会话急切创建（**P0-2 根因**） | -| `packages/opencode/src/dag/dag.ts` | 296-317, 356-385 | create 校验段（P1-1）/ step 守卫与内联映射（P0-3） | -| `packages/core/src/dag/core/scheduling.ts` | 50-113 | markSatisfied / getReadyNodes / isComplete | -| `packages/core/src/dag/core/types.ts` | 37-56, 156-200 | NodeStatus.SKIPPED / SkipReason / 迁移表 | -| `packages/core/src/dag/core/replan.ts` | 108-133 | restart 仅限 running / 终态 ignore（P1-5） | -| `packages/opencode/src/dag/templates/sanitize.ts` | 16-28 | 破坏性替换（P1-2） | -| `packages/core/src/dag/store.ts` | 226-257 | JS 聚合（P1-4）/ skipped 不计数（P2-9） | -| `packages/opencode/src/dag/runtime/eval.ts` | 32-54, 92-107 | evaluateCondition / resolvePath（求值逻辑本身正确，DSL 弱见 P2-6） | -| `packages/opencode/src/dag/runtime/recovery.ts` | 93-99 | ownership lost 判罚（P2-2，处置已改 recovery-pause） | -| `packages/tui/src/feature-plugins/system/dag-inspector.tsx` | 295-302, 341, 505-517 | 配色 / 截断 / footer（P2-8） | - ---- - -## 关键会话 ID（交接用） - -| 角色 | Session ID | -|------|-----------| -| 工作流父会话 | `ses_067f06f1bffeAmDFfH51NmNBl5` | -| quality-gate 子会话（返回 REVISE） | `ses_067ee102bffeQVj8nS1Jti7eet` | -| arbitrate 子会话 | `ses_067eebfe6ffe7tpZPn5m2EiV2U` | -| final-audit 子会话（返回 ACCEPT，引用错误 workflow ID） | `ses_06799fb9fffer5U2WPoyRYo3Jx` | -| integrate 子会话（skip 后仍执行） | `ses_0679b57baffexlZR1vLYr7VOwa` | diff --git a/docs/dag-orchestration-constraints.md b/docs/dag-orchestration-constraints.md deleted file mode 100644 index bea997c1a..000000000 --- a/docs/dag-orchestration-constraints.md +++ /dev/null @@ -1,116 +0,0 @@ -# DAG 编排约束提示词（condition / skip / 恢复语义） - -> **用途**：交给编写 DAG 工作流的 agent（编排者）作为硬约束规则。 -> **背景**：`condition_false → markSatisfied` 缺陷（P0-1）已修复：skipped 现在是独立调度态，纯 skip 依赖会级联跳过下游。本文档描述**修复后**的引擎语义。历史缺陷分析见 `dag-bug-condition-skip-as-satisfied.md`。 - ---- - -## 规则 1：condition 门禁会级联——纯 skip 依赖的下游自动跳过 - -### 引擎语义（修复后） - -`condition` 求值为 false 时节点被标记 `skipped`（终态，`condition_false`）。下游处置取决于依赖构成： - -- **下游的依赖全部为 skipped** → 级联跳过（`orphan_cascade`），一波一波传播到整条子链。门禁拒绝会真正阻断纯门禁子图。 -- **下游是混合 fan-in**（至少一个依赖 satisfied 或可降级的 failed-optional）→ 照常执行，skip 的上游以占位文本注入（见规则 3）。 - -``` -gate (verdict=REVISE) - └─ implement-A (condition: gate.verdict == "ACCEPT") → skipped (condition_false) - └─ integrate (仅依赖 implement-A) → skipped (orphan_cascade) ✅ - └─ audit (仅依赖 integrate) → skipped (orphan_cascade) ✅ -``` - -### 约束 - -- 单点 condition 即可门控其**纯依赖**子链，不再需要给子图每个节点重复相同 condition。 -- 若下游是混合 fan-in 且你希望它也被门控，仍需给该 fan-in 自己加 condition——混合 fan-in 的"继续执行"是有意的降级语义，不是缺陷。 -- skip 是终态：级联一旦发生不可逆。若门禁拒绝后还想走修复路径，用 `report_to_parent: true` 唤醒父会话决策 replan，而不是依赖已 skip 的子链复活。 - ---- - -## 规则 2：condition 只能引用 depends_on 中的节点（现在 create 会直接拒绝） - -### 引擎语义（修复后） - -condition 求值时只收集 `node.depends_on` 直接依赖的输出。**`dag.create` 与 `replan` 现在校验 condition 引用**：可解析的 condition 若引用了不在 `depends_on` 中的节点，提交直接报错（fail-fast），不再静默得到 `condition_false`。 - -### 约束 - -- condition 中出现的每个 nodeID 都必须存在于该节点的 `depends_on` 数组中。 -- 不可解析的表达式仍留给运行时 fail-loud，不要依赖格式错误的 condition"恰好为 false"。 - -```yaml -# ❌ create 时报错：condition 引用了 gate，但 depends_on 里没有 gate -- id: implement-core - depends_on: [some-other-node] - condition: 'gate.output.verdict == "ACCEPT"' - -# ✅ 正确 -- id: implement-core - depends_on: [gate] - condition: 'gate.output.verdict == "ACCEPT"' -``` - ---- - -## 规则 3：混合 fan-in 节点必须容忍上游 skip - -### 引擎语义（修复后） - -纯 skip 依赖的 fan-in 会被级联跳过（规则 1），**不再需要防护**。仍会执行的是混合 fan-in：部分上游 skipped、至少一个上游有真实产出。skip 的上游以 `"Dependency X skipped: condition_false"` 占位文本注入 input_mapping / prompt interpolation。 - -### 要求 - -- 混合 fan-in 的 prompt 必须显式说明如何处理 skip 的上游（如"如果某条输入标注为 skipped，在汇总中注明并跳过该项"）。 -- 若混合 fan-in 在部分上游 skip 时不应执行，给它加 condition 显式门控。 - ---- - -## 规则 4：崩溃恢复会暂停工作流——父会话必须处置 - -### 引擎语义（recovery-pause） - -进程崩溃重启后，恢复流程对无法从子会话持久状态证明结果的 running 节点**判失败**（`execution ownership lost on recovery` 等），这一保守判罚不会重试（恢复永不隐式重跑 provider 工作）。判罚后工作流**不会直接终态报废**，而是转 `paused`： - -- 失败节点的下游保持 `pending`（可 replan 改线）。 -- 父会话会收到 NodeFailed wake 消息（含 `execution ownership lost on recovery` 等原因）。 - -### 处置（父会话三选一） - -1. **replan + resume（推荐）**：失败节点是终态不可变的，用**新 id** 提交替换节点，并把下游节点的 `depends_on` 改线到新 id，然后 `resume`。 -2. **直接 resume**：接受失败语义。required 节点失败会把工作流归因为 `failed`（原因含具体节点 id）；optional 失败则降级继续。 -3. **cancel**：放弃整个工作流。 - -### 约束 - -- **禁止**假设崩溃后工作流会自动继续或自动重跑丢失的节点。不会。 -- 收到 ownership-lost wake 后必须在当轮处置（replan / resume / cancel），不要留着 paused 工作流不管。 - ---- - -## 规则 5：不要用中途 status 快照判定"卡死" - -工作流执行中查询 status 时可能看到部分节点 pending——不一定是卡死，可能是调度器尚未拾取。判定卡死需要：`hasRunning()` 为 false、无 ready 节点、且 `!isComplete()`。 - -- **禁止**仅凭"有 pending 节点 + status=running"就判定卡死。 -- 引擎自带 `orchestrator_unresponsive` 兜底：wake 投递后父会话当轮未对停摆工作流采取动作，工作流会被判 fail。收到 wake 里的 actionable 指令必须当轮响应。 - ---- - -## 规则 6：测试场景下占位 prompt 会导致子 agent 产出不可靠 - -压测中使用极简占位 prompt（如"输出一句话后结束"）时，部分子 agent 会产出幻觉内容（引用不存在的工作流 ID / 节点名）。 - -- 压测图若要验证 fan-in / 结构化输出的正确性，prompt 必须包含足够约束（如"只基于上游输入汇总，不要编造工作流 ID 或节点名"）。 -- 否则子 agent 的幻觉输出会干扰对引擎行为的判断。 - ---- - -## 自检清单（编排者提交图之前过一遍） - -- [ ] 每个 condition 引用的 nodeID 都在对应节点的 `depends_on` 中？（规则 2，create 会拒绝但别浪费一轮） -- [ ] 门禁 condition 的下游是纯依赖链还是混合 fan-in？混合 fan-in 需要自己的 condition 或 skip 容忍 prompt？（规则 1/3） -- [ ] 父会话的编排 prompt 是否涵盖崩溃恢复处置（ownership-lost wake → replan/resume/cancel）？（规则 4） -- [ ] 是否避免了对中途 status 快照的过度解读？（规则 5） -- [ ] 压测 prompt 是否足够约束，避免子 agent 幻觉？（规则 6） diff --git a/packages/core/src/plugin/command.ts b/packages/core/src/plugin/command.ts index 669f1687f..0a696db12 100644 --- a/packages/core/src/plugin/command.ts +++ b/packages/core/src/plugin/command.ts @@ -10,11 +10,13 @@ import PROMPT_REVIEW from "./command/review.txt" import DAG_FLOW_PROMPT from "./command/dag-flow.txt" import workflowContent from "./command/workflow.md" with { type: "text" } import orchestrationPolicy from "./command/orchestration-policy.md" with { type: "text" } +import orchestrationDomains from "./command/orchestration-domains.md" with { type: "text" } export const DagFlowDescription = "Start a dependency-graph multi-agent workflow for the supplied task" export const WorkflowFactsContent = workflowContent export const OrchestrationPolicyContent = orchestrationPolicy -export const WorkflowContent = `${WorkflowFactsContent}\n\n${OrchestrationPolicyContent}` +export const OrchestrationDomainsContent = orchestrationDomains +export const WorkflowContent = `${WorkflowFactsContent}\n\n${OrchestrationPolicyContent}\n\n${OrchestrationDomainsContent}` export const DagFlowContent = `${DAG_FLOW_PROMPT}\n\n${WorkflowContent}` export const Plugin = define({ diff --git a/packages/core/src/plugin/command/orchestration-domains.md b/packages/core/src/plugin/command/orchestration-domains.md new file mode 100644 index 000000000..85315f303 --- /dev/null +++ b/packages/core/src/plugin/command/orchestration-domains.md @@ -0,0 +1,116 @@ +# Orchestration Domains + +Productized workflow playbooks for recurring heavy-task domains. Each playbook +composes the existing primitives — profiles, review lifecycle, actionable +checkpoints, bounded repair, and the pause-first replan protocol — into a +repeatable graph shape. Resolve every role below as a capability slot per Role +Resolution: prefer a configured agent whose contract matches (an explore-style +scout, a reasoner-style logic prober, a review-style verdict gate, a +verify-style test runner), fall back to `explore`, `build`, or `general`. + +## The Simulated Audit Loop + +Iteration in a DAG is NOT a cyclic edge and NOT a harness loop. It is a +verdict-driven replan wave: + +1. An audit node declares `output_schema` with a normalized `verdict` and + `report_to_parent: true`. +2. On `REJECT` or `REVISE`, the wake delivers findings to the parent. The + parent issues `control(pause)`, then `control(replan)` appending a + correction node and a NEW audit node under NEW ids (terminal nodes are + immutable), wires `depends_on` forward, then `control(resume)`. +3. Repeat until the audit returns `ACCEPT`. The loop is bounded by + `max_node_replan_attempts` and `max_total_nodes` — on ceiling breach stop + with `BLOCKED` and report the residual findings instead of retrying the + identical plan. + +Every playbook below that says "audit loop" means exactly this mechanism. + +## Playbook: Deep Review + +Multi-role adversarial review of whether a code structure or design is sound. + +- Fan out 3+ reviewers with genuinely conflicting mandates: a prosecutor + (argues the structure is wrong — coupling, hidden invariants, failure + modes), a defender (argues the current shape is justified — constraints, + history, cost of change), and dimension specialists (architecture, + correctness, testability) as scope demands. +- Fan in to one arbiter that must resolve prosecutor/defender conflicts + finding-by-finding, not merely concatenate them, and emit the actionable + checkpoint shape (`verdict`, `findings`, `required_actions`, `next_action`). +- Pre-implementation structure reviews are `design` phase. Reviewing an actual + change requires the diff-phase hard contract: + `implementation → verification(PASS) → diff review` with fingerprint echo. +- On `REVISE`/`REJECT`, drive corrections through the audit loop. + +## Playbook: Deep Speculation + +Prophesy a whole design document — stress-test it end to end and emit an +automated verdict with zero human gates in the middle. + +- Internalized grill method, run as graph roles instead of user Q&A: parallel + nodes over the same document — a logic simulator (walk the described system, + surface contradictions and boundary gaps), an adversarial interrogator + (produce the hardest material questions: hidden assumptions, falsifiers, + failure modes, evidence quality), and an alternatives prober (steelman one + competing shape). +- A responder node answers the interrogation strictly from the document plus + codebase evidence, marking each question ANSWERED / GAP / CONTRADICTION. +- An arbiter synthesizes everything into a structured prophecy: verdict, + ranked risks, unresolved gaps, and a concrete revision list — then the audit + loop applies revisions and re-speculates until ACCEPT. +- Fully automated: no admission QA rounds with the user mid-flight. Reserve + interactive `GRILL` admission for before the workflow starts. + +## Playbook: Large Engineering + +Turn an execution document (todo list, work ledger, or spec) into audited, +parallel-safe delivery. + +1. **Deep analysis** — scout nodes map the affected surface; an analyst node + decomposes the document into work packages with explicit dependency edges + and disjoint write sets (the tickets: each package states its blocking + edges, not a bare list). +2. **Orchestrate** — compile the packages into a graph: independent packages + fan out in parallel, dependent ones serialize, propose-then-assemble where + write sets may overlap. +3. **Audit the plan** — a plan-audit node checks the decomposition itself: + missing edges, false parallelism, unstated assumptions, acceptance criteria + per package. `REJECT` re-orchestrates via the audit loop until the plan + passes. +4. **Execute** — run the audited graph with the develop-profile phases each + package still needs; verification consumes each implementation before any + diff review. +5. **Final adversarial review** — the Deep Review playbook over the assembled + result, with its own audit loop. +6. **Deliverable** — a final assembler emits the outcome report: shipped + packages, evidence, residual risks. + +## Playbook: Solution Bake-off + +N competing approaches implemented or prototyped in parallel against the same +acceptance criteria; a verify-style node exercises each candidate; one arbiter +picks the winner on evidence and records why the losers lost. + +## Playbook: Root-Cause Diagnosis + +Fan out one node per plausible hypothesis, each tasked to falsify its own +hypothesis with concrete evidence; an arbiter eliminates, ranks survivors, and +either declares the root cause or replans a deeper probe wave on the leading +survivor. + +## Playbook: Audit Sweeps + +The same fan-out/arbiter/audit-loop shape covers recurring sweep domains: +security surface audit (per-surface reviewers: input handling, authz, secrets, +dependencies), regression matrix fan-out (one verify node per axis cell), and +docs-code drift audit (per-document checkers comparing claims against the +code, with fix waves through the audit loop). + +## Choosing and Combining + +Playbooks compose: Large Engineering embeds Deep Review at its gate; Deep +Speculation can front-load any of them. Selection still obeys Execution Mode +Selection — a playbook is justified only when the task shows both a scenario +and a structural signal, and explicit user constraints always override the +playbook shape. diff --git a/packages/core/src/plugin/command/orchestration-policy.md b/packages/core/src/plugin/command/orchestration-policy.md index 193bdacde..10e7bd038 100644 --- a/packages/core/src/plugin/command/orchestration-policy.md +++ b/packages/core/src/plugin/command/orchestration-policy.md @@ -190,3 +190,14 @@ Do not poll `status` merely to wait. Atomic wake reports actionable checkpoints Implement review-and-repair with finite `extend` or `control(replan)` operations. Target only the nodes and findings that require repair. You MUST NOT create cyclic `depends_on`, predeclare unbounded speculative repair waves, or start an unrelated replacement workflow. Declare a finite `max_node_replan_attempts`. When the ceiling is exhausted, stop with `BLOCKED`, report the remaining findings, and do not retry the identical plan. + +## Replan Protocol (pause-first) + +A replan fragment takes real time to compose — template rendering, model reasoning, node rewiring. While you compose it, the workflow keeps scheduling and can reach a terminal status, after which replan is rejected (terminal workflows are immutable). Freeze first, then think: + +1. On any user cancel/replan/model-change intent, IMMEDIATELY issue `control(pause)` in the same turn. Pause needs no fragment, applies in milliseconds, and stops new node spawns. +2. Pause does not interrupt nodes that are already running. Decide their disposition inside the fragment: `restart: true` re-spawns a running node with the new definition (its in-flight child session is hard-aborted at re-spawn), `cancel: true` terminates it, absence keeps it running to completion. +3. Compose the fragment, then issue `control(replan)` — replan is valid while paused. +4. Issue `control(resume)` to restore scheduling. + +If the workflow terminalized before you paused, do not force the replan: start a new workflow carrying the updated definitions, and state which prior results are superseded. diff --git a/packages/core/src/plugin/command/workflow.md b/packages/core/src/plugin/command/workflow.md index b324b3cd4..b2f46dc56 100644 --- a/packages/core/src/plugin/command/workflow.md +++ b/packages/core/src/plugin/command/workflow.md @@ -474,7 +474,7 @@ Workflows are not static. After creating a workflow, use `extend` and `control(r - **Scale up**: a node reports the work is larger than expected → `extend` with additional parallel nodes to split the load. - **Cut short**: a node proves the remaining work is unnecessary → `control(complete)` to early-complete and skip pending nodes. -- **Redirect**: a gate or review reveals a wrong direction → `control(replan)` with `restart: true` on the affected nodes and `cancel: true` on their downstream dependents. +- **Redirect**: a gate or review reveals a wrong direction → `control(pause)` first to freeze scheduling, then `control(replan)` with `restart: true` on the affected nodes and `cancel: true` on their downstream dependents, then `control(resume)`. Only nodes with `report_to_parent: true` produce intermediate parent checkpoints, and those reports are delivered at the next actionable wake @@ -583,10 +583,10 @@ checkpoint naturally completed the current graph; an early **status** — Read the durable state of one workflow and all of its nodes. Pass `workflow_id`. Use it when the user explicitly asks for current state or once before a decision that requires fresh state, such as replan/control. Do not poll a running workflow merely to wait: node reports and terminal outcomes wake the parent session automatically. **control** — Control a running workflow: -- `pause` — let running nodes finish, don't spawn new ones (pause does NOT stop nodes that are already running) +- `pause` — let running nodes finish, don't spawn new ones (pause does NOT stop nodes that are already running). On a cancel/replan intent, always pause FIRST: it needs no fragment and freezes scheduling while you compose the replan, so the graph cannot terminalize under you. - `resume` — resume scheduling - `cancel` — cancel the entire workflow -- `replan` — submit a YAML fragment; running nodes can be `restart: true` or `cancel: true`; pending nodes absent from the fragment are cancelled +- `replan` — submit a YAML fragment; running nodes can be `restart: true` or `cancel: true`; pending nodes absent from the fragment are cancelled. Valid while paused — the pause → compose fragment → replan → resume sequence is the safe path. - `complete` — early-complete: remaining pending nodes are skipped (non-violation) - `step` — advance exactly one ready node (the first by node ID lexicographic order), then wait. Use for controlled debugging or staged verification of a critical path. Unlike `pause`, which freezes all scheduling, `step` advances one node and re-waits. A second `step` while the stepped node is still running is rejected. Use `resume` to return to full-speed scheduling. Nodes are selected in lexicographic ID order for determinism. diff --git a/packages/core/test/plugin/command.test.ts b/packages/core/test/plugin/command.test.ts index 01fce2845..695d46138 100644 --- a/packages/core/test/plugin/command.test.ts +++ b/packages/core/test/plugin/command.test.ts @@ -132,6 +132,39 @@ describe("CommandPlugin.Plugin", () => { }), ) + it.effect("defines the pause-first replan protocol", () => + Effect.sync(() => { + expect(CommandPlugin.OrchestrationPolicyContent).toContain("## Replan Protocol (pause-first)") + expect(CommandPlugin.OrchestrationPolicyContent).toContain("IMMEDIATELY issue `control(pause)`") + expect(CommandPlugin.OrchestrationPolicyContent).toContain("replan is valid while paused") + expect(CommandPlugin.OrchestrationPolicyContent).toContain("Pause does not interrupt nodes that are already running") + expect(CommandPlugin.WorkflowFactsContent).toContain("always pause FIRST") + }), + ) + + it.effect("defines productized orchestration domain playbooks", () => + Effect.sync(() => { + expect(CommandPlugin.WorkflowContent).toContain("# Orchestration Domains") + expect(CommandPlugin.OrchestrationDomainsContent).toContain("## The Simulated Audit Loop") + expect(CommandPlugin.OrchestrationDomainsContent).toContain("NOT a cyclic edge and NOT a harness loop") + expect(CommandPlugin.OrchestrationDomainsContent).toContain("NEW ids (terminal nodes are") + for (const playbook of [ + "## Playbook: Deep Review", + "## Playbook: Deep Speculation", + "## Playbook: Large Engineering", + "## Playbook: Solution Bake-off", + "## Playbook: Root-Cause Diagnosis", + "## Playbook: Audit Sweeps", + ]) { + expect(CommandPlugin.OrchestrationDomainsContent).toContain(playbook) + } + expect(CommandPlugin.OrchestrationDomainsContent).toContain("prosecutor") + expect(CommandPlugin.OrchestrationDomainsContent).toContain("zero human gates in the middle") + expect(CommandPlugin.OrchestrationDomainsContent).toContain("max_node_replan_attempts") + expect(CommandPlugin.OrchestrationDomainsContent).toContain("capability slot per Role") + }), + ) + it.effect("defines parent-session admission fixtures for standard deep and GRILL requests", () => Effect.sync(() => { const fixtures = [ diff --git a/packages/opencode/src/dag/config.ts b/packages/opencode/src/dag/config.ts new file mode 100644 index 000000000..fb21a14a8 --- /dev/null +++ b/packages/opencode/src/dag/config.ts @@ -0,0 +1,126 @@ +/** + * DAG defaults config — `dag.jsonc` beside the opencode config. + * + * Two-tier default working models plus a thinking-depth variant for DAG child + * sessions. Everything else inherits the main opencode configuration. + * + * Lookup order (first existing file wins, project overrides global): + * - project: `.opencode/dag.jsonc` / `.opencode/dag.json` + * - global: `/dag.jsonc` / `/dag.json` + * + * When neither exists, a commented default `dag.jsonc` is seeded into the + * global opencode config directory (same generation pattern as the main + * config's loadGlobal). Files are read lazily at spawn-scheduling time — like + * templates/resolve.ts, nothing is loaded at startup, and edits take effect on + * the next scheduling round. + */ + +export * as DagConfig from "./config" + +import { Effect, Option, Schema } from "effect" +import * as path from "node:path" +import * as fs from "node:fs/promises" +import { parse as parseJsonc } from "jsonc-parser" +import { Global } from "@opencode-ai/core/global" +import { Flag } from "@opencode-ai/core/flag/flag" +import { isReviewWorker } from "./review-lifecycle" + +export const Info = Schema.Struct({ + model: Schema.optional( + Schema.Struct({ + advanced: Schema.optional(Schema.String), + standard: Schema.optional(Schema.String), + }), + ), + thinking_depth: Schema.optional(Schema.Literals(["low", "medium", "high", "max"])), +}) +export type Info = typeof Info.Type + +const DEFAULT_CONTENT = `{ + // DAG workflow defaults — applies to every DAG child session. + // Model resolution per node: node.model → config.node_defaults.model + // → worker agent model → this file's tier → parent session model. + // Format: "provider/model", e.g. "anthropic/claude-sonnet-4-5". + "model": { + // Advanced tier — critical nodes: required: true and review/arbiter workers. + // "advanced": "", + // Standard tier — every other worker node. With only one tier configured, + // it serves as the unified default for all nodes. + // "standard": "" + }, + // Reasoning variant for DAG child sessions: "low" | "medium" | "high" | "max". + // Applied only when the resolved model defines a variant with this name; + // otherwise it is a no-op. + // "thinking_depth": "medium" +} +` + +export function load(projectDir: string): Effect.Effect { + return Effect.gen(function* () { + const found = yield* Effect.promise(() => readFirst(candidates(projectDir))) + if (!found) { + yield* Effect.promise(() => seedGlobalDefault()).pipe(Effect.ignore) + return {} + } + const decoded = Schema.decodeUnknownOption(Info)(parseJsonc(found.text, [], { allowTrailingComma: true }) ?? {}) + if (Option.isNone(decoded)) { + yield* Effect.logWarning("dag config is invalid — ignoring", { path: found.path }) + return {} + } + return decoded.value + }) +} + +/** + * Resolve the configured default model for a node. Critical nodes + * (required: true or review workers) take the advanced tier; everything else + * takes standard. A single configured tier acts as the unified default. + */ +export function tierModel(info: Info, node: { required: boolean; workerType: string }) { + const critical = node.required || isReviewWorker(node.workerType) + const ref = critical + ? info.model?.advanced ?? info.model?.standard + : info.model?.standard ?? info.model?.advanced + return parseModelRef(ref) +} + +function candidates(projectDir: string) { + const globalDir = globalConfigDir() + return [ + path.join(projectDir, ".opencode", "dag.jsonc"), + path.join(projectDir, ".opencode", "dag.json"), + path.join(globalDir, "dag.jsonc"), + path.join(globalDir, "dag.json"), + ] +} + +// Same redirect the Global service applies in make(), so tests and managed +// setups that point OPENCODE_CONFIG_DIR elsewhere are honored here too. +function globalConfigDir() { + return Flag.OPENCODE_CONFIG_DIR ?? Global.Path.config +} + +async function readFirst(paths: string[]) { + for (const file of paths) { + try { + return { path: file, text: await fs.readFile(file, "utf-8") } + } catch { + continue + } + } + return undefined +} + +async function seedGlobalDefault() { + const file = path.join(globalConfigDir(), "dag.jsonc") + await fs.mkdir(path.dirname(file), { recursive: true }) + // wx: never clobber a file that appeared between the read and the seed. + await fs.writeFile(file, DEFAULT_CONTENT, { flag: "wx" }) +} + +function parseModelRef(ref: string | undefined) { + if (!ref) return undefined + const slash = ref.indexOf("/") + if (slash <= 0 || slash === ref.length - 1) return undefined + return { providerID: ref.slice(0, slash), modelID: ref.slice(slash + 1) } +} diff --git a/packages/opencode/src/dag/review-lifecycle.ts b/packages/opencode/src/dag/review-lifecycle.ts index 3428e970f..a8a5eae4b 100644 --- a/packages/opencode/src/dag/review-lifecycle.ts +++ b/packages/opencode/src/dag/review-lifecycle.ts @@ -255,6 +255,6 @@ function dependsTransitively(config: WorkflowConfig, startID: string, targetID: return visit(startID) } -function isReviewWorker(workerType: string) { +export function isReviewWorker(workerType: string) { return workerType === "review" || workerType.startsWith("review-") } diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index 1e1a59b17..9f66f6c84 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -24,6 +24,7 @@ import { SessionID } from "@/session/schema" import { SessionStatus } from "@/session/status" import { renderTemplate } from "../templates/resolve" import { sanitizeInput } from "../templates/sanitize" +import { DagConfig } from "../config" import { spawnNode } from "./spawn" import { evaluateCondition, resolveInputMapping } from "./eval" import { reconcileWorkflow, makeSessionStatusChecker } from "./recovery" @@ -83,6 +84,9 @@ export const layer = Layer.effect( // their outputs cannot change during this loop — per-node re-reads // were O(ready × nodes) pure overhead. const nodesSnapshot = ready.length > 0 ? yield* store.getNodes(dagID) : [] + // dag.jsonc defaults are read once per scheduling round (lazy, like + // templates) so edits apply to the next round without a restart. + const dagConfig: DagConfig.Info = ready.length > 0 ? yield* DagConfig.load(ctx.directory) : {} for (const nodeID of ready) { const node = nodesSnapshot.find((n) => n.id === nodeID) if (!node) continue @@ -218,6 +222,8 @@ export const layer = Layer.effect( reviewImplementationFingerprint: nodeConfig ? reviewImplementationFingerprint(nodeConfig, resolvedMapping) : undefined, + fallbackModel: DagConfig.tierModel(dagConfig, { required: node.required, workerType: node.workerType }), + variant: dagConfig.thinking_depth, }).pipe( Effect.tap((result) => Effect.sync(() => entry.fibers.set(nodeID, result.fiber))), Effect.provideService(Dag.Service, dag), diff --git a/packages/opencode/src/dag/runtime/spawn.ts b/packages/opencode/src/dag/runtime/spawn.ts index c7e14da95..db0183d22 100644 --- a/packages/opencode/src/dag/runtime/spawn.ts +++ b/packages/opencode/src/dag/runtime/spawn.ts @@ -47,6 +47,10 @@ export interface NodeSpawnInput { timeoutMs?: number reportToParent?: boolean reviewImplementationFingerprint?: string + /** dag.jsonc tier default — slots between the agent model and the parent-session model. */ + fallbackModel?: { modelID: string; providerID: string } + /** dag.jsonc thinking_depth — forwarded as the prompt variant (no-op unless the model defines it). */ + variant?: string } export interface NodeSpawnResult { @@ -88,6 +92,7 @@ export function spawnNode( : undefined const model = nodeModel ?? (agent.model ? { modelID: agent.model.modelID, providerID: agent.model.providerID } : undefined) + ?? (input.fallbackModel ? { modelID: input.fallbackModel.modelID as never, providerID: input.fallbackModel.providerID as never } : undefined) ?? (parent.model ? { modelID: parent.model.id, providerID: parent.model.providerID } : undefined) if (!model) { yield* dag.nodeFailed(input.dagID, input.nodeID, `no model configured for agent: ${agent.name}`, "exec_failed") @@ -200,6 +205,7 @@ export function spawnNode( sessionID: childSession.id, model, agent: agent.name, + ...(input.variant ? { variant: input.variant } : {}), parts: input.promptParts, }).pipe(Effect.timeoutOption(remainingMs)) if (Option.isNone(resultOpt)) { diff --git a/packages/opencode/src/tool/workflow.ts b/packages/opencode/src/tool/workflow.ts index 740a7c404..1179b85b4 100644 --- a/packages/opencode/src/tool/workflow.ts +++ b/packages/opencode/src/tool/workflow.ts @@ -6,6 +6,7 @@ import { Session } from "@/session/session" import { SessionID } from "@/session/schema" import type { NodeConfig, WorkflowConfig } from "@/dag/dag" import { AdmissionRecord, ExecutionMode } from "@/dag/admission" +import { TerminalViolationError } from "@opencode-ai/core/dag/core/types" const id = "workflow" @@ -196,12 +197,16 @@ export const WorkflowTool = Tool.define"}, operation=${params.operation ?? ""}). Example: { action: "control", workflow_id: "dag_...", operation: "pause" }. On a cancel/replan intent, issue pause FIRST — it needs no fragment and freezes scheduling instantly while you compose the replan.`, + )) + } const wfId = params.workflow_id switch (params.operation) { case "pause": yield* dag.pause(wfId).pipe(Effect.orDie) - return { title: "Workflow paused", output: `\nNote: pause stops new node spawns only — nodes already running continue to completion.`, metadata: { workflowId: wfId } as Metadata } + return { title: "Workflow paused", output: `\nNote: pause stops new node spawns only — nodes already running continue to completion. To stop a running node, submit a replan fragment marking it restart: true or cancel: true (replan is valid while paused).`, metadata: { workflowId: wfId } as Metadata } case "resume": yield* dag.resume(wfId).pipe(Effect.orDie) return { title: "Workflow resumed", output: ``, metadata: { workflowId: wfId } as Metadata } @@ -212,8 +217,24 @@ export const WorkflowTool = Tool.define`, metadata: { workflowId: wfId } as Metadata } case "replan": { - if (!params.fragment) return yield* Effect.die(new Error("replan operation requires 'fragment'")) - const r = yield* dag.replan(wfId, { nodes: params.fragment.nodes as NodeConfig[] }).pipe(Effect.orDie) + if (!params.fragment) { + return yield* Effect.die(new Error( + "replan operation requires 'fragment' (a WorkflowGraphSchema with the node definitions to apply). If the fragment is not composed yet, issue control(pause) first so the graph cannot terminalize while you write it.", + )) + } + const r = yield* dag.replan(wfId, { nodes: params.fragment.nodes as NodeConfig[] }).pipe( + // The graph raced to terminal while the fragment was being + // composed (the pause-first protocol was skipped). Surface + // the recovery options instead of a bare iron-law rejection. + Effect.catchIf( + (err): err is TerminalViolationError => err instanceof TerminalViolationError, + (err) => + Effect.die(new Error( + `${err.message}. The workflow reached a terminal status before the replan arrived — terminal workflows are immutable. Recover by starting a new workflow with the updated node definitions (action: start), or extend if a reporting leaf checkpoint naturally completed the graph. Next time issue control(pause) BEFORE composing the fragment.`, + )), + ), + Effect.orDie, + ) const ignored = r.ignore.length > 0 ? `\nIgnored (terminal, immutable — add replacements under new ids to retry): ${r.ignore.join(", ")}` : "" return { title: `Workflow replanned: +${r.add.length} -${r.cancel.length} ↻${r.restart.length}`, diff --git a/packages/opencode/test/dag/dag-config.test.ts b/packages/opencode/test/dag/dag-config.test.ts new file mode 100644 index 000000000..de9b3fbba --- /dev/null +++ b/packages/opencode/test/dag/dag-config.test.ts @@ -0,0 +1,127 @@ +import { afterEach, beforeEach, describe, expect, it } from "bun:test" +import { Effect } from "effect" +import { DagConfig } from "@/dag/config" +import * as os from "node:os" +import * as path from "node:path" +import * as fs from "node:fs/promises" + +let dir: string +let projectDir: string +let globalDir: string +const originalConfigDir = process.env.OPENCODE_CONFIG_DIR + +beforeEach(async () => { + dir = await fs.mkdtemp(path.join(os.tmpdir(), "dag-config-")) + projectDir = path.join(dir, "project") + globalDir = path.join(dir, "global") + await fs.mkdir(projectDir, { recursive: true }) + // Redirect the global config dir (Flag.OPENCODE_CONFIG_DIR reads the env) + // so seeding never touches the real one. + process.env.OPENCODE_CONFIG_DIR = globalDir +}) + +afterEach(async () => { + if (originalConfigDir === undefined) delete process.env.OPENCODE_CONFIG_DIR + else process.env.OPENCODE_CONFIG_DIR = originalConfigDir + await fs.rm(dir, { recursive: true, force: true }) +}) + +describe("DagConfig.load", () => { + it("seeds a commented default dag.jsonc into the global config dir when none exists", async () => { + const info = await Effect.runPromise(DagConfig.load(projectDir)) + expect(info).toEqual({}) + const seeded = await fs.readFile(path.join(globalDir, "dag.jsonc"), "utf-8") + expect(seeded).toContain("thinking_depth") + expect(seeded).toContain("advanced") + // The seeded file keeps every value commented out — a subsequent load + // yields an empty model block and no thinking depth. + const reloaded = await Effect.runPromise(DagConfig.load(projectDir)) + expect(reloaded).toEqual({ model: {} }) + expect(reloaded.thinking_depth).toBeUndefined() + expect(DagConfig.tierModel(reloaded, { required: true, workerType: "build" })).toBeUndefined() + }) + + it("reads the global dag.jsonc with comments and trailing commas", async () => { + await fs.mkdir(globalDir, { recursive: true }) + await fs.writeFile( + path.join(globalDir, "dag.jsonc"), + `{ + // tiered defaults + "model": { "advanced": "prov/big", "standard": "prov/small", }, + "thinking_depth": "high", + }`, + ) + const info = await Effect.runPromise(DagConfig.load(projectDir)) + expect(info.model?.advanced).toBe("prov/big") + expect(info.model?.standard).toBe("prov/small") + expect(info.thinking_depth).toBe("high") + }) + + it("prefers the project .opencode/dag.jsonc over the global file", async () => { + await fs.mkdir(globalDir, { recursive: true }) + await fs.writeFile(path.join(globalDir, "dag.jsonc"), `{ "thinking_depth": "low" }`) + await fs.mkdir(path.join(projectDir, ".opencode"), { recursive: true }) + await fs.writeFile(path.join(projectDir, ".opencode", "dag.jsonc"), `{ "thinking_depth": "max" }`) + const info = await Effect.runPromise(DagConfig.load(projectDir)) + expect(info.thinking_depth).toBe("max") + }) + + it("ignores a config with an invalid shape instead of failing the round", async () => { + await fs.mkdir(globalDir, { recursive: true }) + await fs.writeFile(path.join(globalDir, "dag.jsonc"), `{ "thinking_depth": "ultra" }`) + const info = await Effect.runPromise(DagConfig.load(projectDir)) + expect(info).toEqual({}) + }) +}) + +describe("DagConfig.tierModel", () => { + const info: DagConfig.Info = { model: { advanced: "prov/big", standard: "prov/small" } } + + it("routes required nodes to the advanced tier", () => { + expect(DagConfig.tierModel(info, { required: true, workerType: "build" })).toEqual({ + providerID: "prov", + modelID: "big", + }) + }) + + it("routes review workers to the advanced tier", () => { + expect(DagConfig.tierModel(info, { required: false, workerType: "review-arch" })).toEqual({ + providerID: "prov", + modelID: "big", + }) + }) + + it("routes ordinary workers to the standard tier", () => { + expect(DagConfig.tierModel(info, { required: false, workerType: "explore" })).toEqual({ + providerID: "prov", + modelID: "small", + }) + }) + + it("uses a single configured tier as the unified default", () => { + const only = { model: { standard: "prov/small" } } + expect(DagConfig.tierModel(only, { required: true, workerType: "build" })).toEqual({ + providerID: "prov", + modelID: "small", + }) + const advancedOnly = { model: { advanced: "prov/big" } } + expect(DagConfig.tierModel(advancedOnly, { required: false, workerType: "explore" })).toEqual({ + providerID: "prov", + modelID: "big", + }) + }) + + it("keeps only the first slash as the provider separator", () => { + const nested = { model: { standard: "local-proxy-compatible/qwen3.8-max-preview" } } + expect(DagConfig.tierModel(nested, { required: false, workerType: "explore" })).toEqual({ + providerID: "local-proxy-compatible", + modelID: "qwen3.8-max-preview", + }) + }) + + it("returns undefined for missing or malformed refs", () => { + expect(DagConfig.tierModel({}, { required: true, workerType: "build" })).toBeUndefined() + expect(DagConfig.tierModel({ model: { standard: "no-slash" } }, { required: false, workerType: "x" })).toBeUndefined() + expect(DagConfig.tierModel({ model: { standard: "prov/" } }, { required: false, workerType: "x" })).toBeUndefined() + }) +})