From 14baba41b74bd56434f16dc4260b238b01ad5e47 Mon Sep 17 00:00:00 2001 From: lex Date: Thu, 30 Jul 2026 10:22:55 +0800 Subject: [PATCH] fix(dag): harden workflow spec handoffs --- docs/harness-dag.md | 20 +- .../plugin/command/orchestration-policy.md | 21 +- packages/core/src/plugin/command/workflow.md | 76 ++- packages/core/test/plugin/command.test.ts | 9 +- packages/opencode/src/dag/admission.ts | 24 + .../opencode/src/dag/templates/resolve.ts | 4 +- packages/opencode/src/tool/workflow.ts | 129 ++++- .../opencode/test/dag/dag-templates.test.ts | 22 +- .../opencode/test/dag/workflow-tool.test.ts | 501 +++++++++++------- 9 files changed, 552 insertions(+), 254 deletions(-) diff --git a/docs/harness-dag.md b/docs/harness-dag.md index 6737a7d99..8e9e110bd 100644 --- a/docs/harness-dag.md +++ b/docs/harness-dag.md @@ -51,13 +51,19 @@ QA 覆盖六个维度：目标、范围、约束与假设、验收标准、证 } ``` -准入记录还包含 `protocol_version`、`brief_revision`、`qa_mode`、`verdict`、 -`state` 和 `fingerprint`。目标、范围、约束、假设或验收标准发生实质变化时， -应增加 Brief 修订号、生成新指纹，并把旧准入置为 `INVALIDATED` 后重新问答。 +YAML 准入输入只包含 `brief_revision`、`qa_mode`、`verdict`、`brief`，以及 +WAIVED 所需的审计字段。不要在输入中提供 `protocol_version`、`state` 或 +`fingerprint`：工作流边界会设置协议版本，从 verdict 初始化状态，规范化 Brief +后计算小写十六进制 SHA-256 指纹；只有成功启动后才把持久化状态转为 +`CONSUMED`。 -`action: start` 调用中，`mode` 和 `admission` 与 `config` 同级，而不是 -`config` 的子字段。指纹计算会裁剪目标和所有数组字符串、删除空项、对数组排序， -按 Brief 的固定字段顺序序列化为紧凑 JSON，再计算小写十六进制 SHA-256。 +启动、扩展和 replan 的图配置都先写入 `.yaml` 或 `.yml` 文件，工具调用只传 +`action`、`spec_path` 和该动作所需的少量标识字段。启动文件中，`mode`、 +`admission` 与 `config` 同级。校验失败时保留并修改同一文件后重试，不要重新 +生成整段 tool-call 参数。 + +目标、范围、约束、假设或验收标准发生实质变化时，应增加 Brief 修订号， +使旧指纹失效并重新问答；新指纹仍由工作流边界生成。 ## Verdict 与恢复路径 @@ -109,7 +115,7 @@ REJECT 准入和 review 元数据；若显式声明了不完整的 diff review 元数据，引擎只产生 非阻塞诊断。 -本能力扩展的是模型可调用的 `workflow` 工具配置。现有 HTTP DAG +本能力扩展的是模型可调用的 `workflow` 工具文件输入。现有 HTTP DAG 查询仍返回持久化工作流行和字符串化 `config`，HTTP 请求/响应 schema、 SDK 的 DAG summary 类型以及 TUI re-export 均未改变，因此不需要重新生成 JavaScript SDK。 diff --git a/packages/core/src/plugin/command/orchestration-policy.md b/packages/core/src/plugin/command/orchestration-policy.md index 48e73c527..58ba076a5 100644 --- a/packages/core/src/plugin/command/orchestration-policy.md +++ b/packages/core/src/plugin/command/orchestration-policy.md @@ -125,25 +125,26 @@ Maintain a versioned Requirement Brief with this structure: ``` Before start, show a concise brief summary and verdict: -`READY | NOT_READY | WAIVED`, plus QA mode, brief revision, fingerprint, and -remaining blockers. `READY` requires a non-empty goal, scope boundaries, +`READY | NOT_READY | WAIVED`, plus QA mode, brief revision, and remaining +blockers. `READY` requires a non-empty goal, scope boundaries, acceptance criteria, evidence obligations, review plan, and no blocking questions. For `NOT_READY`, remain in the parent conversation and offer: continue QA, reduce scope, use `standard`, or explicitly waive. A `WAIVED` start is informed only when both `waiver_reason` and `acknowledged_risks` are non-empty; preserve them for audit. -Compute `fingerprint` exactly as the workflow boundary does: trim `goal`; for -every array in the Brief, trim strings, remove blanks, and sort them; preserve -the documented key order; then SHA-256 hash the compact JSON object and encode -it as lowercase hexadecimal. Use normal local calculation tools rather than -inventing a digest. +Do not supply `protocol_version`, `state`, or `fingerprint` in the YAML +admission input. Those are durable audit fields owned by the workflow boundary: +it sets protocol version 1, initializes state from the verdict, normalizes the +Brief for fingerprint computation, and computes the lowercase hexadecimal +SHA-256 hash. A successful deep start alone transitions the durable record to +`CONSUMED`. Material changes to goal, scope, constraints, assumptions, or acceptance criteria create a new brief revision, invalidate the prior fingerprint, and -return admission to questioning. Only a successful deep workflow start consumes -a `READY` or `WAIVED` record. Do not replay QA from a consumed record after -recovery. +return admission to questioning. The workflow boundary generates the +replacement fingerprint from the revised Brief. Do not replay QA from a +consumed record after recovery. ## Role Resolution diff --git a/packages/core/src/plugin/command/workflow.md b/packages/core/src/plugin/command/workflow.md index 546ab4228..8c1f25b9a 100644 --- a/packages/core/src/plugin/command/workflow.md +++ b/packages/core/src/plugin/command/workflow.md @@ -34,12 +34,47 @@ as independent workstreams, cross-domain uncertainty, high blast radius, conflicting constraints, evidence gathering, or multiple verification perspectives. -Before a deep start, qualify the request interactively in the parent session -and pass `mode: deep` plus a versioned `READY` or informed `WAIVED` admission -record beside `config` in the `action: start` tool call. Do not put admission -QA inside the graph: its answers define the graph. Use the orchestration policy -below for QA modes, round budgets, verdict recovery, revision invalidation, and -waiver audit fields. +Before any graph-carrying action (`start`, `extend`, or `control(replan)`), +write its configuration to a `.yaml` or `.yml` file, then pass only +`spec_path` beside the shallow action fields. Never inline graph nodes, +admission, or replan fragments in the tool call. Keep the file after a +validation failure, edit only the reported problem, and retry the same path. + +Before a deep start, qualify the request interactively in the parent session. +The start YAML places `mode: deep`, a versioned `READY` or informed `WAIVED` +admission input, and `config` at the same level. The admission input contains +`brief_revision`, `qa_mode`, `verdict`, `brief`, and waiver audit fields when +applicable; the workflow boundary owns `protocol_version`, `state`, and +`fingerprint`. Do not put admission QA inside the graph: its answers define the +graph. Use the orchestration policy below for QA modes, round budgets, verdict +recovery, revision invalidation, and waiver audit fields. + +A deep start spec has this outer shape: + +```yaml +title: Deep review +mode: deep +admission: + brief_revision: 1 + qa_mode: STANDARD + verdict: READY + brief: + goal: Review the requested implementation + scope: + in: [requested modules] + out: [unrelated modules] + constraints: [read-only reviewers] + assumptions: [the working tree is the review target] + acceptance_criteria: [all material findings are evidence-backed] + evidence_required: [code references, relevant test results] + risks: [missed cross-module regressions] + review_plan: [parallel review, claim verification, final arbitration] + open_questions: [] + blocking_questions: [] +config: + name: deep-review + nodes: [] +``` ## Orchestration Lifecycle @@ -92,14 +127,16 @@ the workflow uncreated so the user can configure a model and retry. ## Collaboration Patterns -Four structural patterns cover the common cases. Real workflows often combine them. +Four structural patterns cover the common cases. Real workflows often combine +them. Every YAML block below is workflow spec file content. Save the selected +shape to a `.yaml` file, then call the tool with +`{ action: "start", spec_path: ".yaml" }`. ### 1. Staged Pipeline with Gate Sequential phases where each depends on the previous. Insert a gate node between phases to block downstream execution until quality is confirmed. ```yaml -action: start config: name: staged-gate nodes: @@ -148,7 +185,6 @@ the parent an actionable replan or stop decision. One preparatory node feeds N independent worker nodes, which fan back into a single assembler. ```yaml -action: start config: name: parallel-fan-out nodes: @@ -191,7 +227,6 @@ config: Multiple reviewer nodes with different perspectives examine the same artifact. A final arbiter synthesizes their verdicts. The arbiter must not be a silent terminal leaf: gate an in-graph continuation node on its verdict (shown below), or dispose of the reported verdict at the wake boundary per the Verdict Disposal Contract. ```yaml -action: start config: name: adversarial-review nodes: @@ -270,7 +305,6 @@ never silently terminalize the graph. Multiple independent generators produce candidate solutions; a converger selects and refines. ```yaml -action: start config: name: diverge-converge nodes: @@ -381,7 +415,9 @@ For ad-hoc prompts, use `prompt_template: { inline: "...", input: {...} }`. Static `prompt_template.input` supplies literal, local template values; it does not read upstream node output. Inline templates interpolate those static values and direct dependency variables. Use `input_mapping` when an upstream output -needs a stable variable name or field selection. +needs a stable variable name or field selection. When a mapped placeholder +contains an object or array, interpolation renders it as indented JSON; it must +never appear as `[object Object]`. ## Budget Declaration @@ -415,17 +451,19 @@ All nodes share the same workspace. Write conflicts are an orchestration concern ### Actions -**start** — Create a workflow from a YAML-declared graph. Pass the graph as -`{ action: "start", config: { name, nodes, ...defaultsAndBudgets } }`; `nodes` -at the tool-call top level is only for `extend`, not `start`. Returns the -workflow ID. Nodes declare `depends_on` (node IDs); layers and execution order -are computed automatically. +**start** — Create a workflow from a YAML spec with `config` and optional +`title`, `mode`, and admission input at the file root. Write the file first, +then call `{ action: "start", spec_path: ".opencode/workflows/name.yaml" }`. +Returns the workflow ID. Nodes declare `depends_on` (node IDs); layers and +execution order are computed automatically. **extend** — Add nodes to a running workflow. Existing nodes are unaffected; new nodes are immediately eligible for scheduling if their dependencies are met. It also accepts a genuinely additive wave after a reporting leaf checkpoint naturally completed the current graph; an early -`control(complete)` workflow remains terminal. +`control(complete)` workflow remains terminal. Put the new nodes under a +file-root `nodes` array, then call +`{ action: "extend", workflow_id: "dag_...", spec_path: "extend.yaml" }`. **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. @@ -433,7 +471,7 @@ checkpoint naturally completed the current graph; an early - `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. Valid while paused — the pause → compose fragment → replan → resume sequence is the safe path. +- `replan` — write a YAML file with a file-root `fragment` object containing the graph fields and node definitions, then pass its `spec_path`; running nodes can be `restart: true` or `cancel: true`; pending nodes absent from the fragment are cancelled. Valid while paused — the pause → compose file → 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 405c19792..85c344ae4 100644 --- a/packages/core/test/plugin/command.test.ts +++ b/packages/core/test/plugin/command.test.ts @@ -281,11 +281,12 @@ describe("CommandPlugin.Plugin", () => { expect(CommandPlugin.OrchestrationPolicyContent).toContain("invalidate the prior fingerprint") expect(CommandPlugin.OrchestrationPolicyContent).toContain("SHA-256 hash") expect(CommandPlugin.WorkflowFactsContent).toContain( - "pass `mode: deep` plus a versioned `READY` or informed `WAIVED` admission", + "The start YAML places `mode: deep`, a versioned `READY` or informed `WAIVED`", ) expect(CommandPlugin.WorkflowFactsContent).toContain( - "beside `config` in the `action: start` tool call", + "the workflow boundary owns `protocol_version`, `state`, and\n`fingerprint`", ) + expect(CommandPlugin.WorkflowFactsContent).toContain("pass only\n`spec_path` beside the shallow action fields") expect(CommandPlugin.WorkflowFactsContent).not.toContain("`config.mode`") }), ) @@ -316,7 +317,8 @@ describe("CommandPlugin.Plugin", () => { expect(graphExamples.length).toBeGreaterThan(0) for (const example of graphExamples) { - expect(example).toContain("action: start\nconfig:") + expect(example).toMatch(/^config:/m) + expect(example).not.toMatch(/^action:/m) expect(example).toMatch(/\n nodes:/) expect(example).not.toMatch(/^nodes:/m) expect(example.match(/^\s+- id:/gm)?.length).toBe(example.match(/^ {6}name:/gm)?.length) @@ -328,6 +330,7 @@ describe("CommandPlugin.Plugin", () => { expect(CommandPlugin.WorkflowFactsContent).not.toContain('input: { findings: "from explore" }') expect(CommandPlugin.WorkflowFactsContent).not.toContain("Gate failure cancels the workflow automatically") expect(CommandPlugin.WorkflowFactsContent).toContain("Static `prompt_template.input`") + expect(CommandPlugin.WorkflowFactsContent).toContain("it must\nnever appear as `[object Object]`") expect(CommandPlugin.WorkflowFactsContent).toContain( "Workflow definitions MUST NOT specify `node.model` or\n`config.node_defaults.model`", ) diff --git a/packages/opencode/src/dag/admission.ts b/packages/opencode/src/dag/admission.ts index 5a31568c9..fad431b19 100644 --- a/packages/opencode/src/dag/admission.ts +++ b/packages/opencode/src/dag/admission.ts @@ -62,6 +62,16 @@ export const AdmissionRecord = Schema.Struct({ }) export type AdmissionRecord = typeof AdmissionRecord.Type +export const AdmissionInput = Schema.Struct({ + brief_revision: Schema.Number, + qa_mode: Schema.Literals(Modes), + verdict: Schema.Literals(Verdicts), + brief: RequirementBrief, + waiver_reason: Schema.optional(Schema.String), + acknowledged_risks: Schema.optional(StringArray), +}) +export type AdmissionInput = typeof AdmissionInput.Type + const Transitions = { UNASSESSED: ["QUESTIONING", "WAIVED"], QUESTIONING: ["READY", "NOT_READY", "WAIVED"], @@ -104,6 +114,20 @@ export function fingerprintBrief(brief: RequirementBrief) { return new Bun.CryptoHasher("sha256").update(JSON.stringify(canonical)).digest("hex") } +export function createAdmissionRecord(input: AdmissionInput): AdmissionRecord { + return { + protocol_version: 1, + brief_revision: input.brief_revision, + qa_mode: input.qa_mode, + verdict: input.verdict, + state: input.verdict, + fingerprint: fingerprintBrief(input.brief), + brief: input.brief, + ...(input.waiver_reason ? { waiver_reason: input.waiver_reason } : {}), + ...(input.acknowledged_risks ? { acknowledged_risks: input.acknowledged_risks } : {}), + } +} + export function validateAdmission(record: AdmissionRecord) { const errors = [ ...(record.protocol_version === 1 ? [] : ["protocol_version must be 1"]), diff --git a/packages/opencode/src/dag/templates/resolve.ts b/packages/opencode/src/dag/templates/resolve.ts index a90166cf0..3bba9d257 100644 --- a/packages/opencode/src/dag/templates/resolve.ts +++ b/packages/opencode/src/dag/templates/resolve.ts @@ -97,7 +97,9 @@ function interpolate(template: string, input: Record) { const unresolvedPlaceholders: string[] = [] const text = template.replace(INTERPOLATION_RE, (match, key: string) => { const value = input[key] - if (value !== null && value !== undefined) return String(value) + if (value !== null && value !== undefined) { + return typeof value === "object" ? JSON.stringify(value, null, 2) : String(value) + } unresolvedPlaceholders.push(key) return match }) diff --git a/packages/opencode/src/tool/workflow.ts b/packages/opencode/src/tool/workflow.ts index ea0ffaff7..e43fa63a6 100644 --- a/packages/opencode/src/tool/workflow.ts +++ b/packages/opencode/src/tool/workflow.ts @@ -9,13 +9,17 @@ import { Question } from "@/question" import { Session } from "@/session/session" import { SessionID } from "@/session/schema" import type { NodeConfig, WorkflowConfig } from "@/dag/dag" -import { AdmissionRecord, ExecutionMode } from "@/dag/admission" +import { AdmissionInput, createAdmissionRecord, ExecutionMode } from "@/dag/admission" import { TerminalViolationError } from "@opencode-ai/core/dag/core/types" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { assertExternalDirectoryEffect } from "./external-directory" +import path from "node:path" const id = "workflow" +const MAX_WORKFLOW_SPEC_BYTES = 1_000_000 // ============================================================================ -// Schemas (flat Struct with action discriminator — matches codebase convention) +// File schemas stay rich; tool-call parameters below stay shallow. // ============================================================================ const NodeSchema = Schema.Struct({ @@ -78,18 +82,32 @@ const WorkflowGraphSchema = Schema.Struct({ nodes: Schema.Array(NodeSchema).annotate({ description: "Node declarations" }), }) +const StartSpec = Schema.Struct({ + title: Schema.optional(Schema.String), + mode: Schema.optional(ExecutionMode), + admission: Schema.optional(AdmissionInput), + config: WorkflowGraphSchema, +}) + +const ExtendSpec = Schema.Struct({ + nodes: Schema.Array(NodeSchema), +}) + +const ReplanSpec = Schema.Struct({ + fragment: WorkflowGraphSchema, +}) + +const decodeStartSpec = Schema.decodeUnknownEffect(StartSpec) +const decodeExtendSpec = Schema.decodeUnknownEffect(ExtendSpec) +const decodeReplanSpec = Schema.decodeUnknownEffect(ReplanSpec) + export const Parameters = Schema.Struct({ action: Schema.Literals(["start", "extend", "control", "status"]).annotate({ description: "start: create workflow; extend: add nodes; control: pause/resume/cancel/replan/step/complete; status: inspect durable workflow and node state" }), - config: Schema.optional(WorkflowGraphSchema).annotate({ description: "(start) Workflow graph definition" }), - mode: Schema.optional(ExecutionMode).annotate({ description: "(start) standard by default; deep requires completed parent-session admission QA" }), - admission: Schema.optional(AdmissionRecord).annotate({ description: "(start deep) Structured Requirement Brief and READY/WAIVED verdict" }), + spec_path: Schema.optional(Schema.String).annotate({ description: "(start/extend/control replan) Path to a YAML workflow spec. Relative paths resolve from the session directory" }), session_id: Schema.optional(Schema.String).annotate({ description: "(start) Parent session ID" }), project_id: Schema.optional(Schema.String).annotate({ description: "(start) Optional Project ID; must match the parent session project" }), - title: Schema.optional(Schema.String).annotate({ description: "(start) Workflow title" }), workflow_id: Schema.optional(Schema.String).annotate({ description: "(extend/control/status) Target workflow ID" }), - nodes: Schema.optional(Schema.Array(NodeSchema)).annotate({ description: "(extend) Nodes to add" }), operation: Schema.optional(Schema.Literals(["pause", "resume", "cancel", "replan", "step", "complete"])).annotate({ description: "(control) Operation to perform" }), - fragment: Schema.optional(WorkflowGraphSchema).annotate({ description: "(control replan) Replan fragment with node definitions" }), }) // ============================================================================ @@ -165,15 +183,19 @@ export const WorkflowTool = Tool.define< } } case "start": { - if (!params.config) return yield* Effect.die(new Error("start requires 'config'")) const sessionID = SessionID.make(params.session_id ?? ctx.sessionID) const session = yield* sessions.get(sessionID).pipe(Effect.orDie) if (params.project_id && params.project_id !== session.projectID) { return yield* Effect.die(new Error("project_id must match the parent session project")) } + const specFile = yield* readWorkflowSpec(params.spec_path, session.directory, ctx).pipe(Effect.orDie) + const spec = yield* decodeStartSpec(specFile.value).pipe( + Effect.mapError((error) => new Error(`Invalid workflow spec ${specFile.path}: ${String(error)}`)), + Effect.orDie, + ) const missingModels = yield* findNodesWithoutModel({ - nodes: params.config.nodes, - defaults: params.config.node_defaults, + nodes: spec.config.nodes, + defaults: spec.config.node_defaults, directory: session.directory, parent: session.model, agents, @@ -207,23 +229,29 @@ export const WorkflowTool = Tool.define< const dagID = yield* dag.create({ projectID: session.projectID, sessionID, - title: params.title ?? params.config.name, + title: spec.title ?? spec.config.name, config: { - ...params.config, - mode: params.mode ?? "standard", - ...(params.admission ? { admission: params.admission } : {}), + ...spec.config, + mode: spec.mode ?? "standard", + ...(spec.admission ? { admission: createAdmissionRecord(spec.admission) } : {}), } as WorkflowConfig, }).pipe(Effect.orDie) - const mode = params.mode ?? "standard" + const mode = spec.mode ?? "standard" return { - title: `Workflow started: ${params.config.name}`, - output: `\n${params.config.nodes.length} nodes registered.\nDo not poll this workflow. It runs asynchronously and will wake the parent session when attention is required.\n`, + title: `Workflow started: ${spec.config.name}`, + output: `\n${spec.config.nodes.length} nodes registered.\nDo not poll this workflow. It runs asynchronously and will wake the parent session when attention is required.\n`, metadata: { workflowId: dagID } as Metadata, } } case "extend": { - if (!params.workflow_id || !params.nodes) return yield* Effect.die(new Error("extend requires 'workflow_id' and 'nodes'")) - const r = yield* dag.extend(params.workflow_id, params.nodes as NodeConfig[]).pipe(Effect.orDie) + if (!params.workflow_id) return yield* Effect.die(new Error("extend requires 'workflow_id'")) + const session = yield* sessions.get(SessionID.make(ctx.sessionID)).pipe(Effect.orDie) + const specFile = yield* readWorkflowSpec(params.spec_path, session.directory, ctx).pipe(Effect.orDie) + const spec = yield* decodeExtendSpec(specFile.value).pipe( + Effect.mapError((error) => new Error(`Invalid workflow spec ${specFile.path}: ${String(error)}`)), + Effect.orDie, + ) + const r = yield* dag.extend(params.workflow_id, spec.nodes as NodeConfig[]).pipe(Effect.orDie) return { title: `Workflow extended: ${r.add.length} nodes added`, output: `\nAdded: ${r.add.join(", ")}\n`, @@ -233,14 +261,14 @@ export const WorkflowTool = Tool.define< case "control": { if (!params.workflow_id || !params.operation) { return yield* Effect.die(new Error( - `control requires 'workflow_id' and 'operation' (got workflow_id=${params.workflow_id ?? ""}, 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.`, + `control requires 'workflow_id' and 'operation' (got workflow_id=${params.workflow_id ?? ""}, operation=${params.operation ?? ""}). Example: { action: "control", workflow_id: "dag_...", operation: "pause" }. On a cancel/replan intent, issue pause FIRST — it needs no spec file 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. 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 } + 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 spec 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 } @@ -251,12 +279,13 @@ export const WorkflowTool = Tool.define< yield* dag.complete(wfId).pipe(Effect.orDie) return { title: "Workflow completed (early)", output: ``, metadata: { workflowId: wfId } as Metadata } case "replan": { - 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( + const session = yield* sessions.get(SessionID.make(ctx.sessionID)).pipe(Effect.orDie) + const specFile = yield* readWorkflowSpec(params.spec_path, session.directory, ctx).pipe(Effect.orDie) + const spec = yield* decodeReplanSpec(specFile.value).pipe( + Effect.mapError((error) => new Error(`Invalid workflow spec ${specFile.path}: ${String(error)}`)), + Effect.orDie, + ) + const r = yield* dag.replan(wfId, { nodes: spec.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. @@ -264,7 +293,7 @@ export const WorkflowTool = Tool.define< (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.`, + `${err.message}. The workflow reached a terminal status before the replan arrived — terminal workflows are immutable. Recover by writing a new start spec with the updated node definitions and passing its spec_path, or extend if a reporting leaf checkpoint naturally completed the graph. Next time issue control(pause) BEFORE composing the spec file.`, )), ), Effect.orDie, @@ -291,6 +320,48 @@ export const WorkflowTool = Tool.define< }), ) +function readWorkflowSpec(specPath: string | undefined, directory: string, ctx: Tool.Context) { + return Effect.gen(function* () { + if (!specPath) { + return yield* Effect.fail(new Error( + "Workflow configuration requires 'spec_path'. Write the YAML spec to a file, then retry with its path.", + )) + } + const filepath = path.isAbsolute(specPath) ? path.normalize(specPath) : path.resolve(directory, specPath) + if (![".yaml", ".yml"].includes(path.extname(filepath).toLowerCase())) { + return yield* Effect.fail(new Error(`Workflow spec must be a .yaml or .yml file: ${filepath}`)) + } + if (!FSUtil.contains(directory, filepath)) { + yield* assertExternalDirectoryEffect(ctx, filepath, { + bypass: Boolean(ctx.extra?.["bypassCwdCheck"]), + }) + } + + const file = Bun.file(filepath) + if (!(yield* Effect.promise(() => file.exists()))) { + return yield* Effect.fail(new Error(`Workflow spec not found: ${filepath}`)) + } + if (file.size > MAX_WORKFLOW_SPEC_BYTES) { + return yield* Effect.fail(new Error( + `Workflow spec is too large: ${file.size} bytes exceeds ${MAX_WORKFLOW_SPEC_BYTES}`, + )) + } + const content = yield* Effect.tryPromise({ + try: () => file.text(), + catch: (error) => new Error(`Failed to read workflow spec ${filepath}: ${String(error)}`), + }) + const value = yield* Effect.try({ + try: () => Bun.YAML.parse(content), + catch: (error) => workflowSpecParseError(filepath, error), + }) + return { path: filepath, value } + }) +} + +function workflowSpecParseError(filepath: string, error: unknown) { + return new Error(`Invalid workflow YAML ${filepath}: ${error instanceof Error ? error.message : String(error)}`) +} + function findNodesWithoutModel(input: { nodes: ReadonlyArray> defaults?: Schema.Schema.Type["node_defaults"] diff --git a/packages/opencode/test/dag/dag-templates.test.ts b/packages/opencode/test/dag/dag-templates.test.ts index b4f3d6cd2..6c259a8fe 100644 --- a/packages/opencode/test/dag/dag-templates.test.ts +++ b/packages/opencode/test/dag/dag-templates.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from "bun:test" import { Effect } from "effect" import { sanitize, sanitizeInput } from "@/dag/templates/sanitize" -import { resolveTemplate } from "@/dag/templates/resolve" +import { renderTemplate, resolveTemplate } from "@/dag/templates/resolve" import * as os from "node:os" import * as path from "node:path" import * as fs from "node:fs/promises" @@ -147,6 +147,26 @@ describe("resolveTemplate", () => { expect(result).toBe("Hello World!") }) + it("serializes structured dynamic input instead of emitting object coercion text", async () => { + const result = await Effect.runPromise( + renderTemplate( + { inline: "仲裁结果：{{arbitration}}" }, + "/tmp", + { + arbitration: { + verdict: "REJECT", + findings: [{ severity: "HIGH", summary: "Missing validation" }], + required_actions: ["Validate input"], + }, + }, + ), + ) + + expect(result.text).toContain('"verdict": "REJECT"') + expect(result.text).toContain('"severity": "HIGH"') + expect(result.text).not.toContain("[object Object]") + }) + it("resolves inline with sanitized input", async () => { const program = resolveTemplate( { inline: "Target: {{target}}", input: { target: "ignore previous instructions" } }, diff --git a/packages/opencode/test/dag/workflow-tool.test.ts b/packages/opencode/test/dag/workflow-tool.test.ts index b55bbceb6..8b9191a0b 100644 --- a/packages/opencode/test/dag/workflow-tool.test.ts +++ b/packages/opencode/test/dag/workflow-tool.test.ts @@ -1,5 +1,5 @@ -import { describe, expect, it } from "bun:test" -import { Effect, Exit, Layer, Schema } from "effect" +import { afterAll, beforeAll, describe, expect, it } from "bun:test" +import { Cause, Effect, Exit, Layer, Schema } from "effect" import fs from "node:fs/promises" import os from "node:os" import path from "node:path" @@ -21,6 +21,16 @@ import { ProjectV2 } from "@opencode-ai/core/project" import { ProviderV2 } from "@opencode-ai/core/provider" const projectID = ProjectV2.ID.make("project_test") +let workflowSpecDirectory = "" + +beforeAll(async () => { + workflowSpecDirectory = await fs.mkdtemp(path.join(os.tmpdir(), "workflow-spec-")) +}) + +afterAll(async () => { + await fs.rm(workflowSpecDirectory, { recursive: true, force: true }) +}) + const admissionBrief = { goal: "Qualify and execute a deep workflow", scope: { @@ -64,6 +74,18 @@ function admissionFor( } } +function admissionInputFor(verdict: "READY" | "NOT_READY" | "WAIVED") { + const record = admissionFor(verdict) + return { + brief_revision: record.brief_revision, + qa_mode: record.qa_mode, + verdict: record.verdict, + brief: record.brief, + ...(record.waiver_reason ? { waiver_reason: record.waiver_reason } : {}), + ...(record.acknowledged_risks ? { acknowledged_risks: record.acknowledged_risks } : {}), + } +} + const published: Array<{ type: string; data: unknown }> = [] const store = Layer.mock(DagStore.Service, { getWorkflow: (id: string) => @@ -207,7 +229,7 @@ const runtime = testEffect( id, slug: "workflow-test", projectID, - directory: "/project", + directory: workflowSpecDirectory, title: "Workflow test", version: "test", time: { created: 0, updated: 0 }, @@ -259,11 +281,18 @@ const missingModelRuntime = testEffect( ), ) +function writeWorkflowSpec(name: string, value: unknown) { + const filepath = path.join(workflowSpecDirectory, `${name}.yaml`) + return Effect.promise(() => Bun.write(filepath, JSON.stringify(value, null, 2))).pipe( + Effect.as(filepath), + ) +} + describe("workflow tool schema (negative tests)", () => { it("action field accepts start/extend/control/status", () => { const decode = Schema.decodeUnknownSync(Parameters) - expect(() => decode({ action: "start", config: { name: "test", nodes: [], max_concurrency: 3 } })).not.toThrow() - expect(() => decode({ action: "extend", workflow_id: "wf-1", nodes: [] })).not.toThrow() + expect(() => decode({ action: "start", spec_path: ".opencode/workflows/test.yaml" })).not.toThrow() + expect(() => decode({ action: "extend", workflow_id: "wf-1", spec_path: ".opencode/workflows/extend.yaml" })).not.toThrow() expect(() => decode({ action: "control", workflow_id: "wf-1", operation: "pause" })).not.toThrow() expect(() => decode({ action: "status", workflow_id: "wf-1" })).not.toThrow() }) @@ -298,55 +327,23 @@ describe("workflow tool schema (negative tests)", () => { expect(() => decode({ action: "control", workflow_id: "wf-1", operation: "start" })).toThrow() }) - it("leaves omitted node defaults unresolved and strips graph-level model fields", () => { - const decode = Schema.decodeUnknownSync(Parameters) - const input = decode({ - action: "start", - config: { - name: "required-default", - node_defaults: { - required: true, - report_to_parent: true, - worker_config: { timeout_ms: 1234 }, - model: { providerID: "configured", modelID: "configured/model" }, - }, - nodes: [ - { - id: "optional-node", - name: "Optional node", - worker_type: "build", - depends_on: [], - prompt_template: { inline: "work" }, - model: { providerID: "configured", modelID: "configured/model" }, - }, - ], - }, - }) - - expect(input.config?.nodes[0]?.required).toBeUndefined() - expect(input.config?.node_defaults).toEqual({ - required: true, - report_to_parent: true, - worker_config: { timeout_ms: 1234 }, - }) - expect(input.config?.nodes[0]).not.toHaveProperty("model") - }) - - it("decodes deep mode and structured admission", () => { + it("keeps workflow graph and admission internals out of tool-call parameters", () => { const decode = Schema.decodeUnknownSync(Parameters) expect(decode({ action: "start", + spec_path: ".opencode/workflows/deep.yaml", mode: "deep", - admission: admissionFor("READY"), + admission: admissionFor("READY", "CONSUMED"), config: { name: "deep-schema", nodes: [], }, - })).toEqual(expect.objectContaining({ - mode: "deep", - admission: expect.objectContaining({ verdict: "READY" }), - })) + })).toEqual({ + action: "start", + spec_path: ".opencode/workflows/deep.yaml", + }) }) + }) describe("workflow tool execution", () => { @@ -427,20 +424,155 @@ describe("workflow tool execution", () => { }), ) + runtime.effect("deep start repairs one YAML file and owns admission audit fields", () => + Effect.gen(function* () { + published.length = 0 + const specPath = path.join(workflowSpecDirectory, "deep.yaml") + yield* Effect.promise(() => + Bun.write( + specPath, + `title: Deep ready +mode: deep +admission: + brief_revision: 1 + qa_mode: STANDARD + verdict: READY +config: + name: deep-ready + nodes: [] +`, + ), + ) + + const info = yield* WorkflowTool + const workflow = yield* info.init() + const context = { + sessionID: SessionID.make("ses_workflow_parent"), + messageID: MessageID.ascending(), + agent: "build", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + } satisfies Tool.Context + const invalid = yield* workflow.execute( + { + action: "start", + spec_path: "deep.yaml", + }, + context, + ).pipe(Effect.exit) + + expect(Exit.isFailure(invalid)).toBe(true) + if (Exit.isFailure(invalid)) { + expect(Cause.pretty(invalid.cause)).toContain('["admission"]["brief"]') + } + expect(published).toHaveLength(0) + + yield* Effect.promise(() => + Bun.write( + specPath, + `title: Deep ready +mode: deep +admission: + protocol_version: 999 + brief_revision: 1 + qa_mode: STANDARD + verdict: READY + state: CONSUMED + fingerprint: ${"0".repeat(64)} + brief: + goal: Qualify and execute a deep workflow + scope: + in: [workflow start, review lifecycle] + out: [new admission UI] + constraints: [standard workflows stay compatible] + assumptions: [the parent session can ask questions] + acceptance_criteria: [deep start requires READY or WAIVED] + evidence_required: [unit tests, integration tests] + risks: [waiver misuse] + review_plan: [verify, review the implementation diff] + open_questions: [] + blocking_questions: [] +config: + name: deep-ready + nodes: [] +`, + ), + ) + + const result = yield* workflow.execute( + { + action: "start", + spec_path: "deep.yaml", + }, + context, + ) + + expect(result.output).toContain('mode="deep"') + const created = published.find((event) => event.type === DagEvent.WorkflowCreated.type)?.data as { + config?: string + } + expect(JSON.parse(created.config ?? "{}")).toEqual(expect.objectContaining({ + mode: "deep", + admission: expect.objectContaining({ + protocol_version: 1, + verdict: "READY", + state: "CONSUMED", + fingerprint: fingerprintBrief(admissionBrief), + }), + })) + }), + ) + + runtime.effect("invalid YAML reports its source file without workflow side effects", () => + Effect.gen(function* () { + published.length = 0 + const specPath = path.join(workflowSpecDirectory, "invalid.yaml") + yield* Effect.promise(() => Bun.write(specPath, "config:\n nodes: [\n")) + const info = yield* WorkflowTool + const workflow = yield* info.init() + const exit = yield* workflow.execute( + { + action: "start", + spec_path: specPath, + }, + { + sessionID: SessionID.make("ses_workflow_parent"), + messageID: MessageID.ascending(), + agent: "build", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + } satisfies Tool.Context, + ).pipe(Effect.exit) + + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) { + expect(Cause.pretty(exit.cause)).toContain(`Invalid workflow YAML ${specPath}:`) + } + expect(published).toHaveLength(0) + }), + ) + runtime.effect("start derives the project ID from the parent session", () => Effect.gen(function* () { published.length = 0 const parentID = SessionID.make("ses_workflow_parent") const info = yield* WorkflowTool const workflow = yield* info.init() + const specPath = yield* writeWorkflowSpec("project-id-regression", { + config: { + name: "project-id-regression", + nodes: [], + }, + }) const result = yield* workflow.execute( { action: "start", - config: { - name: "project-id-regression", - nodes: [], - }, + spec_path: specPath, }, { sessionID: parentID, @@ -481,22 +613,30 @@ describe("workflow tool execution", () => { '{ "model": {} }\n', ) ) + yield* Effect.promise(() => + Bun.write( + path.join(missingModelDirectory, "missing-model.yaml"), + JSON.stringify({ + config: { + name: "missing-model", + nodes: [{ + id: "worker", + name: "Worker", + worker_type: "build", + depends_on: [], + prompt_template: { inline: "work" }, + }], + }, + }), + ) + ) const info = yield* WorkflowTool const workflow = yield* info.init() const result = yield* workflow.execute( { action: "start", - config: { - name: "missing-model", - nodes: [{ - id: "worker", - name: "Worker", - worker_type: "build", - depends_on: [], - prompt_template: { inline: "work" }, - }], - }, + spec_path: "missing-model.yaml", }, { sessionID: SessionID.make("ses_workflow_parent"), @@ -517,61 +657,23 @@ describe("workflow tool execution", () => { }), ) - runtime.effect("deep start consumes and persists a READY admission", () => - Effect.gen(function* () { - published.length = 0 - const info = yield* WorkflowTool - const workflow = yield* info.init() - const result = yield* workflow.execute( - { - action: "start", - mode: "deep", - admission: admissionFor("READY"), - config: { - name: "deep-ready", - nodes: [], - }, - }, - { - sessionID: SessionID.make("ses_workflow_parent"), - messageID: MessageID.ascending(), - agent: "build", - abort: new AbortController().signal, - messages: [], - metadata: () => Effect.void, - ask: () => Effect.void, - } satisfies Tool.Context, - ) - - expect(result.output).toContain('mode="deep"') - const created = published.find((event) => event.type === DagEvent.WorkflowCreated.type)?.data as { - config?: string - } - expect(JSON.parse(created.config ?? "{}")).toEqual(expect.objectContaining({ - mode: "deep", - admission: expect.objectContaining({ - verdict: "READY", - state: "CONSUMED", - fingerprint: admissionFor("READY").fingerprint, - }), - })) - }), - ) - runtime.effect("deep start consumes and retains an informed WAIVED admission", () => Effect.gen(function* () { published.length = 0 const info = yield* WorkflowTool const workflow = yield* info.init() + const specPath = yield* writeWorkflowSpec("deep-waived", { + mode: "deep", + admission: admissionInputFor("WAIVED"), + config: { + name: "deep-waived", + nodes: [], + }, + }) yield* workflow.execute( { action: "start", - mode: "deep", - admission: admissionFor("WAIVED"), - config: { - name: "deep-waived", - nodes: [], - }, + spec_path: specPath, }, { sessionID: SessionID.make("ses_workflow_parent"), @@ -596,34 +698,56 @@ describe("workflow tool execution", () => { }), ) - runtime.effect("deep start blocks every non-admitted state without side effects", () => + runtime.effect("deep start blocks missing or non-ready admission without side effects", () => Effect.gen(function* () { const info = yield* WorkflowTool const workflow = yield* info.init() const cases = [ - { name: "missing", admission: undefined }, - { name: "not-ready", admission: admissionFor("NOT_READY") }, - { name: "invalidated", admission: admissionFor("NOT_READY", "INVALIDATED") }, { - name: "bad-fingerprint", - admission: { - ...admissionFor("READY"), - fingerprint: "0".repeat(64), + name: "missing", + value: { + mode: "deep", + config: { + name: "deep-missing", + nodes: [], + }, + }, + }, + { + name: "not-ready", + value: { + mode: "deep", + admission: admissionInputFor("NOT_READY"), + config: { + name: "deep-not-ready", + nodes: [], + }, + }, + }, + { + name: "waived-without-audit", + value: { + mode: "deep", + admission: { + ...admissionInputFor("WAIVED"), + waiver_reason: undefined, + acknowledged_risks: undefined, + }, + config: { + name: "deep-waived-without-audit", + nodes: [], + }, }, }, ] for (const item of cases) { published.length = 0 + const specPath = yield* writeWorkflowSpec(`blocked-${item.name}`, item.value) const exit = yield* workflow.execute( { action: "start", - mode: "deep", - admission: item.admission, - config: { - name: `deep-${item.name}`, - nodes: [], - }, + spec_path: specPath, }, { sessionID: SessionID.make("ses_workflow_parent"), @@ -647,22 +771,25 @@ describe("workflow tool execution", () => { published.length = 0 const info = yield* WorkflowTool const workflow = yield* info.init() + const specPath = yield* writeWorkflowSpec("required-default", { + config: { + name: "required-default", + nodes: [ + { + id: "optional-node", + name: "Optional node", + worker_type: "build", + depends_on: [], + prompt_template: { inline: "work" }, + }, + ], + }, + }) yield* workflow.execute( { action: "start", - config: { - name: "required-default", - nodes: [ - { - id: "optional-node", - name: "Optional node", - worker_type: "build", - depends_on: [], - prompt_template: { inline: "work" }, - }, - ], - }, + spec_path: specPath, }, { sessionID: SessionID.make("ses_workflow_parent"), @@ -686,37 +813,40 @@ describe("workflow tool execution", () => { published.length = 0 const info = yield* WorkflowTool const workflow = yield* info.init() + const specPath = yield* writeWorkflowSpec("configured-defaults", { + config: { + name: "configured-defaults", + node_defaults: { + required: true, + report_to_parent: true, + worker_config: { timeout_ms: 1234 }, + }, + nodes: [ + { + id: "inherits", + name: "Inherits defaults", + worker_type: "general", + depends_on: [], + prompt_template: { inline: "work" }, + }, + { + id: "overrides", + name: "Overrides defaults", + worker_type: "general", + depends_on: [], + required: false, + report_to_parent: false, + worker_config: { timeout_ms: 4321 }, + prompt_template: { inline: "work" }, + }, + ], + }, + }) yield* workflow.execute( { action: "start", - config: { - name: "configured-defaults", - node_defaults: { - required: true, - report_to_parent: true, - worker_config: { timeout_ms: 1234 }, - }, - nodes: [ - { - id: "inherits", - name: "Inherits defaults", - worker_type: "general", - depends_on: [], - prompt_template: { inline: "work" }, - }, - { - id: "overrides", - name: "Overrides defaults", - worker_type: "general", - depends_on: [], - required: false, - report_to_parent: false, - worker_config: { timeout_ms: 4321 }, - prompt_template: { inline: "work" }, - }, - ], - }, + spec_path: specPath, }, { sessionID: SessionID.make("ses_workflow_parent"), @@ -762,20 +892,23 @@ describe("workflow tool execution", () => { published.length = 0 const info = yield* WorkflowTool const workflow = yield* info.init() + const specPath = yield* writeWorkflowSpec("extend-defaults", { + nodes: [ + { + id: "added", + name: "Added node", + worker_type: "general", + depends_on: [], + prompt_template: { inline: "work" }, + }, + ], + }) yield* workflow.execute( { action: "extend", workflow_id: "dag_defaults", - nodes: [ - { - id: "added", - name: "Added node", - worker_type: "general", - depends_on: [], - prompt_template: { inline: "work" }, - }, - ], + spec_path: specPath, }, { sessionID: SessionID.make("ses_workflow_parent"), @@ -815,24 +948,27 @@ describe("workflow tool execution", () => { published.length = 0 const info = yield* WorkflowTool const workflow = yield* info.init() + const specPath = yield* writeWorkflowSpec("replan-defaults", { + fragment: { + name: "replan-fragment", + nodes: [ + { + id: "replanned", + name: "Replanned node", + worker_type: "general", + depends_on: [], + prompt_template: { inline: "work" }, + }, + ], + }, + }) yield* workflow.execute( { action: "control", workflow_id: "dag_defaults", operation: "replan", - fragment: { - name: "replan-fragment", - nodes: [ - { - id: "replanned", - name: "Replanned node", - worker_type: "general", - depends_on: [], - prompt_template: { inline: "work" }, - }, - ], - }, + spec_path: specPath, }, { sessionID: SessionID.make("ses_workflow_parent"), @@ -878,10 +1014,7 @@ describe("workflow tool execution", () => { { action: "start", project_id: "project_other", - config: { - name: "project-id-mismatch", - nodes: [], - }, + spec_path: "project-id-mismatch.yaml", }, { sessionID: parentID,