From 3308bf59819d97e4c78f035c660bc34b89e574e0 Mon Sep 17 00:00:00 2001 From: chumingjun <15951837502@163.com> Date: Thu, 3 Sep 2026 01:17:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(orchestrator):=20=E5=AD=90=E5=B7=A5?= =?UTF-8?q?=E4=BD=9C=E6=B5=81=E8=8A=82=E7=82=B9=EF=BC=88=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E8=B0=83=E7=94=A8=E5=B7=B2=E4=BF=9D=E5=AD=98=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 subworkflow 节点类型：按 workflowId 同步调用库内已保存工作流， 父节点等待子 run 终态后取其输出作为节点输出。首期不支持异步等待、 重试（retryable:false）、续跑与重放。 引擎（engine.js）： - run 级联元数据 parentRunId/parentNodeId/rootRunId/depth/callChain， run-start/run-end 事件携带；NODE_TIMEOUT_MS 统一 5 分钟 - 取消传播：cancelAbort + onCancel 钩子（父取消递归取消子 run）； 重试退避改为可中断等待；_executeNode 支持 releaseSlot 与 queued 后取消防重复扣减；空图主动收尾 - registerKind(subworkflow)：execute 委托宿主注入的 runChildWorkflow， lint/templateLintFields 接入节点校验与模板静态检查 编排器（index.js）： - lib/subworkflow.js：输入映射（$ref 纯引用/模板字符串）、循环/ 深度（3）/预算（16 子 run）防护、selectWorkflowResult 结果选择、 validateSubworkflowInputs（复用 workflow-inputs.js 统一校验） - runChildWorkflow：lint → 防护 → 输入映射 → startRun（source= 'subworkflow'，suppressNotifications）→ 同步等待；双向取消传播 - 删除守卫：被 subworkflow 节点引用的工作流不可删（助手工具与 HTTP /workflows/detail DELETE 均拦截，409 subworkflow-referenced） - /runs、workflow_runs、/runs/detail 附 parent 元数据与 children （listChildRuns 按开始时间正序）；含子工作流节点的运行禁续跑/重放 前端（web）： - registry 注册 subworkflow 节点（图标/预设/徽标） - NodePanel：目标工作流选择（列表选择或手填 ID）+ 输入映射编辑 （triggerInput 模板 + runInputs JSON） - ResultPanel：子工作流运行区（children 列表、点开子 run 详情） - App 拉取 /workflows 供 NodePanel 选择器使用 验证：orchestrator 18 套（含 subworkflow.test.mjs 新套、engine 新增 6 例、sqlite-store/assistant/workflow-tools 集成扩展）+ web 全套 + 全量 npm test 通过；build-web.sh 双 base 构建通过。 --- .../dsh-ccpg-orchestrator/lib/assistant.js | 12 +- .../dsh-ccpg-orchestrator/lib/engine.js | 112 +++++++- .../dsh-ccpg-orchestrator/lib/index.js | 240 +++++++++++++++++- .../lib/output-contract.js | 2 +- .../dsh-ccpg-orchestrator/lib/sqlite-store.js | 8 + .../dsh-ccpg-orchestrator/lib/subworkflow.js | 189 ++++++++++++++ .../lib/variable-schema.js | 6 + .../test/assistant.test.mjs | 13 +- .../test/engine.test.mjs | 140 ++++++++++ .../test/sqlite-store.test.mjs | 20 ++ .../test/subworkflow.test.mjs | 108 ++++++++ .../test/workflow-tools.integration.test.mjs | 25 ++ web/src/App.jsx | 4 + web/src/NodePanel.jsx | 53 +++- web/src/ResultPanel.jsx | 55 ++++ web/src/registry.jsx | 11 + web/src/result-panel.css | 10 + 17 files changed, 965 insertions(+), 43 deletions(-) create mode 100644 dsh-plugins/dsh-ccpg-orchestrator/lib/subworkflow.js create mode 100644 dsh-plugins/dsh-ccpg-orchestrator/test/subworkflow.test.mjs diff --git a/dsh-plugins/dsh-ccpg-orchestrator/lib/assistant.js b/dsh-plugins/dsh-ccpg-orchestrator/lib/assistant.js index f9ac6d7..8c08083 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/lib/assistant.js +++ b/dsh-plugins/dsh-ccpg-orchestrator/lib/assistant.js @@ -8,7 +8,7 @@ import { lintGraph } from './engine.js'; // ---- 已知节点类型（与前端 registry.jsx、引擎 nodeKinds 对齐）---- -const NODE_TYPES = ['input', 'agent', 'script', 'condition', 'http', 'output', 'notify', 'note']; +const NODE_TYPES = ['input', 'agent', 'script', 'condition', 'http', 'output', 'notify', 'note', 'subworkflow']; // 服务端生成的节点 id：与前端 n__ 风格区分，AI 引用稳定 export function newCanvasNodeId() { @@ -171,6 +171,7 @@ export function summarizeGraphForAI(graph) { ...(n.type === 'condition' ? { include: n.data?.include, exclude: n.data?.exclude } : {}), ...(n.type === 'http' ? { url: n.data?.url, method: n.data?.method } : {}), ...(n.type === 'notify' ? { channel: n.data?.channel, mode: n.data?.mode, targetType: n.data?.channelConfig?.targetType } : {}), + ...(n.type === 'subworkflow' ? { workflowId: n.data?.workflowId, inputMap: n.data?.inputMap } : {}), })), edges: (graph.edges || []).map((e) => ({ from: e.source, to: e.target, ...(e.branch ? { branch: e.branch } : {}) })), }; @@ -190,7 +191,7 @@ export function canvasAssistantPersona() { return `你是「物业工作流画布」的 AI 助手，帮助用户创建/修改/测试节点式工作流。用户在聊天里提需求，你调用画布工具落图。 ## 画布模型 -- 图 = 节点 + 有向边。节点类型 8 种： +- 图 = 节点 + 有向边。节点类型 9 种： - input 输入：data.text 触发文本模板（支持 {{变量}}） - agent 智能体：data.prompt 系统提示词、data.inputTemplate 输入模板（{{上游节点名}} 引用上游输出）、data.tools 工具名数组（如 feishu_doc_read/web_fetch）、data.model/data.channel 可选 - script 脚本：固定 JavaScript；data.inputs 为命名参数数组，每项用 expression 完整变量或 value JSON 常量；data.code 必须声明同步 function main(input, workspace) 并返回 JSON；workspace 仅可 list/read/write/remove 当前节点工作区；可选 data.outputSchema 和 data.scriptTimeoutMs（100-10000） @@ -199,14 +200,9 @@ export function canvasAssistantPersona() { - output 输出：汇聚展示，可选 data.writeback 飞书写回 - notify 消息通知：运行级观察器，可独立放置或在线路中透传；data.channel="feishu"、data.mode="terminal"|"each_node"；群聊使用 data.channelConfig.targetType="chat_id" + oc_ 开头的群 ID，私聊使用 targetType="open_id" + ou_ 开头的用户 open_id；data.channelConfig.credentialId 可选 - note 注释：不执行，data.text 说明文字 + - subworkflow 子工作流：按已保存工作流 data.workflowId 同步调用；data.inputMap.triggerInput/runInputs 显式映射输入；首期不支持异步等待、resume 或 retry，不嵌入子图 - 节点 label 用中文短名（如「分类智能体」）；上下游引用靠 label（{{分类智能体}}）。 -## 对话契约 -- 目标不唯一时必须主动澄清：用户使用“那个节点”“上面的流程”等模糊指代，且画布存在多个候选项时，列出候选节点或流程让用户选择，禁止猜测后直接落图。 -- 危险操作必须先确认：删除节点、清空画布、覆盖已保存工作流，或对运行中的图做结构修改前，先复述将执行的操作及影响并等待用户明确确认。 -- 澄清问题必须带选项：使用可点选或可编号的候选项，避免开放式反问。例如“画布上有 2 个输出节点：①分流输出 ②工单输出，改哪个？” -- 澄清不超过一轮：用户完成选择或确认后，直接执行对应操作，不再重复确认；若信息仍不足，说明缺少的具体字段。 - ## 操作规范 1. 改图一律用 canvas_graph_patch（当前画布/草稿）或 workflow_patch（已保存工作流，按 id）；一批 ops 原子生效，出错整批拒绝会返回错误让你修正；不要试图整图重写。 2. 新节点接入链路：addNode 带 after=<上游节点id> 自动连线；显式连线用 connect。 diff --git a/dsh-plugins/dsh-ccpg-orchestrator/lib/engine.js b/dsh-plugins/dsh-ccpg-orchestrator/lib/engine.js index a342f53..90edc45 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/lib/engine.js +++ b/dsh-plugins/dsh-ccpg-orchestrator/lib/engine.js @@ -7,6 +7,7 @@ import { lintScriptInputs, resolveScriptInputs } from './typed-expression.js'; import { getScriptOutputSchema, validateScriptOutput } from './script-schema.js'; import { normalizeScriptTimeout, SCRIPT_LIMITS } from './script-runner.js'; import { validateNotificationNodeData } from './notifications.js'; +import { validateSubworkflowNode, subworkflowTemplateFields } from './subworkflow.js'; // 相对 v1 的升级： // - 多运行实例并存（Map 而非单 this.s） // - 并发上限（就绪节点排队，槽位释放依次启动） @@ -18,11 +19,31 @@ import { validateNotificationNodeData } from './notifications.js'; // 计数约定：每个节点的完成只在其自身 _onNodeDone 尾部扣一次 s.remaining； // 跳过/取消的节点在标记处扣，且不再递归重复扣。 -export const NODE_TIMEOUT_MS = 500 * 1000; +export const NODE_TIMEOUT_MS = 5 * 60 * 1000; const CONDITION_VERDICT_RE = /^条件判定：(true|false)/; let runSeq = 0; +// 取消传播用的可中断等待：abort 时立即 reject，而不是等满时长。 +function waitWithAbort(ms, signal) { + return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(new Error('运行已取消')); + return; + } + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, ms); + const onAbort = () => { + clearTimeout(timer); + signal?.removeEventListener('abort', onAbort); + reject(new Error('运行已取消')); + }; + signal?.addEventListener('abort', onAbort, { once: true }); + }); +} + // ---------------- 节点类型注册表（扩展点） ---------------- // 一种节点 = 一个 NodeKind 对象： // kind.type 唯一 id（graph.nodes[].type） @@ -31,6 +52,7 @@ let runSeq = 0; // kind.edgeTaken 可选。(s, node, edge) => boolean，控制分支边是否放行 // kind.lint 可选。(node, lintCtx) => issues[]（{level:'error'|'warn', message}） // kind.wantsSink 可选。true = 成功后调用 engine.outputSink（输出写回等后处理） +// kind.templateLintFields 可选。(node) => string[]，额外纳入模板静态检查的字段（默认只查 text/inputTemplate/url/headers/body） // 新增节点类型：export const myKind = {...}; registerKind(myKind) —— 引擎/ // 超时/取消/失败传播/历史持久化全部自动获得。 export const nodeKinds = new Map(); @@ -69,6 +91,8 @@ export class Orchestrator { this.nodeRunner = null; // index.js 注入：async (node, run, s, {signal, emit}) => ({output, ...extra}) this.scriptRunner = null; // index.js 注入：async ({node,input,signal}) => ({value,artifacts,...}) this.outputSink = null; // index.js 注入：async (node, output, {signal}) => ({output, ...extra}) 输出节点后处理（飞书写回等） + this.runChildWorkflow = null; // index.js 注入：同步启动并等待子工作流 + this.onCancel = null; // index.js 注入：父运行取消时传播到子运行 } emit(event, payload) { @@ -78,6 +102,7 @@ export class Orchestrator { async run(graph, { triggerInput = '', runId, workflowName, workflowId, canvasId, source, workspaceRoot, revises, globalVariables = {}, workflowVariables = {}, runInputs = {}, + parentRunId = null, parentNodeId = null, rootRunId = null, depth = 0, callChain = [], resume = null, } = {}) { const id = runId || `run_${Date.now().toString(36)}_${++runSeq}`; @@ -86,6 +111,9 @@ export class Orchestrator { runId: id, startedAt: new Date().toISOString(), status: 'running', triggerInput, runInputs: safeRunInputs, workflowName: workflowName || null, workflowId: workflowId || null, canvasId: canvasId || null, source: source || null, workspaceRoot: workspaceRoot || null, + parentRunId: parentRunId || null, parentNodeId: parentNodeId || null, + rootRunId: rootRunId || id, depth: Number.isInteger(depth) && depth >= 0 ? depth : 0, + callChain: Array.isArray(callChain) ? [...callChain] : [], revises: revises || null, schemaVersion: RUN_SCHEMA_VERSION, nodeStates: {}, outputs: {}, structuredOutputs: {}, nodeOrder: [], @@ -102,6 +130,7 @@ export class Orchestrator { activeCount: 0, concurrency: 4, finished: false, startedAtMs: Date.now(), nodeAbort: new Map(), + cancelAbort: new AbortController(), }; for (const n of graph.nodes) { s.incoming.set(n.id, []); @@ -182,16 +211,25 @@ export class Orchestrator { nodeCount: graph.nodes.filter((n) => n.type !== 'notify').length, workflowId: run.workflowId, workflowName: run.workflowName, canvasId: run.canvasId, source: run.source, + ...(run.parentRunId ? { parentRunId: run.parentRunId } : {}), + ...(run.parentNodeId ? { parentNodeId: run.parentNodeId } : {}), + ...(run.rootRunId ? { rootRunId: run.rootRunId } : {}), + ...(run.depth != null ? { depth: run.depth } : {}), }); s._done = new Promise((resolve) => { s.resolve = resolve; }); - this._pump(s); + // 空图或续跑种子已覆盖全部节点时，不会经过 _onNodeDone，必须主动收尾。 + this._maybeFinish(s); + if (!s.finished) this._pump(s); await s._done; run.durationMs = Date.now() - s.startedAtMs; - run.status = run.canceled ? 'canceled' + run.status = run.canceled || Object.values(run.nodeStates).some((st) => st.status === 'canceled') ? 'canceled' : Object.values(run.nodeStates).some((st) => st.status === 'error') ? 'error' : 'success'; this.emit('run-end', { runId: id, status: run.status, durationMs: run.durationMs, workflowId: run.workflowId, canvasId: run.canvasId, source: run.source, + ...(run.parentRunId ? { parentRunId: run.parentRunId } : {}), + ...(run.parentNodeId ? { parentNodeId: run.parentNodeId } : {}), + ...(run.rootRunId ? { rootRunId: run.rootRunId } : {}), }); return run; } @@ -219,6 +257,8 @@ export class Orchestrator { for (const ac of s.nodeAbort.values()) { try { ac.abort(); } catch { /* 已中止 */ } } + try { s.cancelAbort.abort(); } catch { /* 已中止 */ } + try { this.onCancel?.(run.runId, reason); } catch { /* 子运行取消失败不阻塞父 */ } this._maybeFinish(s); } @@ -247,9 +287,16 @@ export class Orchestrator { } } - async _executeNode(s, nodeId) { + async _executeNode(s, nodeId, { releaseSlot = true } = {}) { const node = s.nodes.get(nodeId); const { run } = s; + const kind = getKind(node.type); + // queued 事件的订阅方可能同步触发 cancel；此时不要再启动已取消的节点。 + if (s.finished || run.nodeStates[nodeId]?.status === 'canceled') { + if (releaseSlot) s.activeCount -= 1; + this._maybeFinish(s); + return; + } if (node.__retryLeft === undefined) node.__retryLeft = this._retryTotal(node); const ac = new AbortController(); s.nodeAbort.set(nodeId, ac); @@ -264,7 +311,6 @@ export class Orchestrator { let output = ''; let extra = {}; if (run.canceled) throw new Error('运行已取消'); - const kind = getKind(node.type); if (kind?.passThrough) { // 注释等纯标注节点：不执行，输出 = 上游拼接，立即按成功收尾 const ctx0 = this.templateCtx(node, s); @@ -281,6 +327,7 @@ export class Orchestrator { emit: this.emit.bind(this), runId: run.runId, workflowId: run.workflowId, render: (tpl, options = {}) => this.renderTemplate(tpl || '', { ...this.templateCtx(node, s), ...options }), + runChildWorkflow: this.runChildWorkflow, }; let result; if (!kind) { @@ -326,7 +373,7 @@ export class Orchestrator { } catch (err) { // 重试：节点声明 retryCount 时，非取消类失败按指数退避重试（重试重新计时超时） const canceled0 = run.canceled || String(err?.message || '') === '运行已取消'; - if (!canceled0 && !s.run.canceled && this._retryLeft(node) > 0) { + if (!canceled0 && !s.run.canceled && kind?.retryable !== false && this._retryLeft(node) > 0) { node.__retryLeft -= 1; const attempt = this._retryTotal(node) - node.__retryLeft; const delay = Math.min(8000, 500 * 2 ** (attempt - 1)); @@ -334,12 +381,20 @@ export class Orchestrator { runId: run.runId, nodeId, status: 'running', retrying: true, attempt: attempt + 1, error: `${String(err.message || err)}（${delay}ms 后重试）`, }); - await new Promise((r) => setTimeout(r, delay)); - clearTimeout(timer); - s.nodeAbort.delete(nodeId); - return this._executeNode(s, nodeId); + let retryCanceled = false; + try { + await waitWithAbort(delay, ac.signal); + } catch (retryError) { + if (!run.canceled && !ac.signal.aborted) throw retryError; + retryCanceled = true; + } + if (!retryCanceled) { + clearTimeout(timer); + s.nodeAbort.delete(nodeId); + return this._executeNode(s, nodeId, { releaseSlot: false }); + } } - const canceled = canceled0 + const canceled = run.canceled || canceled0 || (timedOut === false && ac.signal.aborted === true && String(err?.message || '').includes('取消')) || String(err?.message || '') === '运行已取消'; const msg = timedOut ? `节点超时（${Math.round(timeoutMs / 1000)}s）` : String(err.message || err); @@ -353,13 +408,22 @@ export class Orchestrator { return this._onNodeDone(s, node.id, false); } const errDetails = err?.nodeDetails && typeof err.nodeDetails === 'object' ? err.nodeDetails : {}; + // 子工作流失败详情：稳定错误码 + 子运行定位，详情弹窗与变量树都消费 + const childDetails = {}; + if (err?.code) childDetails.errorCode = err.code; + if (err?.childRunId) childDetails.childRunId = err.childRunId; + if (err?.childStatus) childDetails.childStatus = err.childStatus; + if (err?.childSummary) childDetails.childSummary = String(err.childSummary).slice(0, 2000); run.nodeStates[node.id] = { - ...errDetails, + ...errDetails, ...childDetails, status: canceled ? 'canceled' : 'error', error: canceled ? '运行已取消' : msg, durationMs: Date.now() - t0, startedAt, }; this.emit('node-status', { runId: run.runId, nodeId: node.id, status: canceled ? 'canceled' : 'error', error: canceled ? '运行已取消' : msg, + ...(childDetails.errorCode ? { errorCode: childDetails.errorCode } : {}), + ...(childDetails.childRunId ? { childRunId: childDetails.childRunId } : {}), + ...(childDetails.childStatus ? { childStatus: childDetails.childStatus } : {}), ...(errDetails.trace ? { hasTrace: true, sessionId: errDetails.sessionId } : {}), ...(errDetails.turns != null ? { turns: errDetails.turns } : {}), }); @@ -367,7 +431,7 @@ export class Orchestrator { } finally { clearTimeout(timer); s.nodeAbort.delete(nodeId); - s.activeCount -= 1; + if (releaseSlot) s.activeCount -= 1; this._pump(s); this._maybeFinish(s); } @@ -734,6 +798,26 @@ registerKind({ }, }); +// 子工作流：同步调用库内已保存工作流；执行器由宿主注入（首期不支持重试/续跑/重放）。 +registerKind({ + type: 'subworkflow', + retryable: false, + async execute({ node, s, engine, signal, render }) { + if (typeof engine.runChildWorkflow !== 'function') { + const error = new Error('子工作流执行器未注入（宿主初始化异常）'); + error.code = 'SUBWORKFLOW_RUNNER_UNAVAILABLE'; + throw error; + } + return engine.runChildWorkflow({ node, run: s.run, state: s, signal, render }); + }, + lint(node, lintCtx) { + return validateSubworkflowNode(node, lintCtx); + }, + templateLintFields(node) { + return subworkflowTemplateFields(node.data); + }, +}); + // 消息通知是运行级观察器；节点执行本身只负责在线路中透传数据。 registerKind({ type: 'notify', @@ -796,7 +880,7 @@ export function lintGraph(graph, options = {}) { for (const iss of kind.lint(n, lintCtx) || []) issues.push({ nodeId: n.id, ...iss }); } - const templateFields = [d.text, d.inputTemplate, d.url, d.headers, d.body]; + const templateFields = [d.text, d.inputTemplate, d.url, d.headers, d.body, ...(kind?.templateLintFields?.(n) || [])]; for (const field of templateFields) { if (!field) continue; const checked = validateTemplate(field, { diff --git a/dsh-plugins/dsh-ccpg-orchestrator/lib/index.js b/dsh-plugins/dsh-ccpg-orchestrator/lib/index.js index ed1e987..f6fd585 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/lib/index.js +++ b/dsh-plugins/dsh-ccpg-orchestrator/lib/index.js @@ -57,6 +57,14 @@ import { NotificationChannelRegistry, WorkflowNotificationManager } from './noti import { listFeishuCreds, addFeishuCred, removeFeishuCred, setDefaultFeishuCred, getFeishuCredOrEnv } from './credentials.js'; import { Orchestrator, lintGraph, getKind } from './engine.js'; import { validateWorkflowInputs } from './workflow-inputs.js'; +import { + MAX_CHILD_RUNS_PER_ROOT, + MAX_SUBWORKFLOW_DEPTH, + resolveSubworkflowInputs, + selectWorkflowResult, + validateSubworkflowInputs, + SubworkflowError, +} from './subworkflow.js'; import { createWorkflowExportManifest, importWorkflowDocument, normalizeWorkflowDocument } from './workflow-document.js'; import { saveArtifactsToWorkspace } from './artifact-save.js'; import { buildRevisionGraph, extractRevision, revisionAgentNodeId } from './artifact-feedback.js'; @@ -127,7 +135,14 @@ export function apply(ctx, config) { const notificationChannels = new NotificationChannelRegistry(); notificationChannels.register(createFeishuNotificationChannel({ getCredential: getFeishuCredOrEnv })); const notifications = new WorkflowNotificationManager({ channels: notificationChannels, logger: ctx.logger }); - const lintWorkflowGraph = (graph) => lintGraph(graph, { notificationChannels }); + // resolveTargetWorkflow 给 subworkflow 节点 lint 用：目标存在性 + inputSchema 字段名校验。 + // 读库失败按「不存在」处理——lint 不该因目标文档损坏而崩。 + const lintWorkflowGraph = (graph) => lintGraph(graph, { + notificationChannels, + resolveTargetWorkflow: (workflowId) => { + try { return readWf(workflowId); } catch { return null; } + }, + }); let legacyClaimedWorkspace = null; const canonicalWorkspace = (cwd) => { @@ -777,6 +792,8 @@ export function apply(ctx, config) { if (hooks.length) return `有 ${hooks.length} 个 webhook 关联（${hooks.map((h) => h.id).join(', ')}），请先在画布删除 webhook。`; const schedules = [...currentSchedulerMeta().values()].filter((m) => m.workflowId === wf.id); if (schedules.length) return `有 ${schedules.length} 个定时任务关联（${schedules.map((m) => m.key).join(', ')}），请先在定时任务中心删除。`; + const referencing = subworkflowReferencingWorkflows(wf.id); + if (referencing.length) return `有 ${referencing.length} 个工作流通过子工作流节点引用它（${referencing.map((r) => r.name || r.id).join(', ')}），请先在那些工作流中移除对应子工作流节点。`; // 画布若正打开该工作流：退回草稿态（前端收到事件后回「未保存」空画布） for (const [key, cv] of canvases) { if (cv.workflowId !== wf.id || key.split('\0')[0] !== currentStore().workspaceRoot) continue; @@ -847,6 +864,27 @@ export function apply(ctx, config) { // ---- 运行历史（按工作区持久化 + 内存缓存）---- const pendingRunIds = new Set(); const liveTracesByRun = new Map(); // 运行中节点的实时轨迹：runId → (nodeId → trace)；节点完成后落盘、运行结束释放 + const liveChildRuns = new Map(); // parentRunId -> Map + const childCountByRoot = new Map(); + const childRunOf = (parentRunId) => { + let children = liveChildRuns.get(parentRunId); + if (!children) { children = new Map(); liveChildRuns.set(parentRunId, children); } + return children; + }; + const hasSubworkflowNode = (graph) => (graph?.nodes || []).some((node) => node.type === 'subworkflow'); + const cancelDescendants = (runId, reason) => { + const children = liveChildRuns.get(runId); + for (const childId of [...(children?.keys?.() || [])]) { + // orch.cancel 会再次触发 onCancel，从而递归处理该 child 的后代； + // 这里不手动递归，避免同一棵运行树被重复遍历和重复取消。 + try { orch?.cancel(childId, `父运行取消：${reason}`); } catch { /* 子运行可能已结束 */ } + } + }; + const cancelRunTree = (runId, reason = '用户取消') => { + const entry = orch?.runs.get(runId); + if (!entry) return false; + return orch.cancel(runId, reason); + }; const recoverInterruptedRun = (run, interruptedAt) => { const startedAtMs = run.startedAt ? new Date(run.startedAt).getTime() : NaN; const states = (run.graph?.nodes || []).map((node) => run.nodeStates?.[node.id]); @@ -1017,6 +1055,23 @@ export function apply(ctx, config) { return true; }; + // 子工作流节点以 workflowId 引用库内工作流；删除被引用的目标会让父工作流 + // 运行时才失败，所以两个删除入口（助手工具 + HTTP 路由）都先拦引用方。 + // 自引用不拦：删自身时引用随之消失（运行期循环由 callChain 兜底）。 + const subworkflowReferencingWorkflows = (targetId) => { + const id = String(targetId || ''); + if (!id) return []; + const refs = []; + for (const summary of currentDatabase().listWorkflows()) { + if (summary.id === id) continue; + const doc = readWf(summary.id); + if ((doc?.graph?.nodes || []).some((n) => n.type === 'subworkflow' && String(n.data?.workflowId || '').trim() === id)) { + refs.push({ id: doc.id, name: doc.name }); + } + } + return refs; + }; + const resolveAttachmentFile = (attachment) => { if (attachment?.id) { const dir = resolveInside(STORAGE.attachments, safeFileId(attachment.id, 'invalid')); @@ -1066,6 +1121,7 @@ export function apply(ctx, config) { // ---- 引擎 ---- orch = new Orchestrator(ctx, { onEvent: onOrchestratorEvent, renderTemplate }); + orch.onCancel = (runId, reason) => cancelDescendants(runId, reason); orch.nodeRunner = async (node, run, s, ctl) => runAgentNode(ctx, node, run, s, ctl); orch.scriptRunner = async ({ node, input, signal, timeoutMs, workflowId, runId }) => { const ws = workspaceFor(node, { workflowId: workflowId || 'draft', runId }); @@ -1083,10 +1139,12 @@ export function apply(ctx, config) { const startRun = (graph, { triggerInput, workflowName, workflowId, canvasId, source, globalVariables = {}, workflowVariables = {}, runInputs = {}, runId: providedRunId, replayOf, resume, revises, + parentRunId = null, parentNodeId = null, rootRunId = null, depth = 0, callChain = [], + suppressNotifications = false, } = {}) => { const store = currentStore(); const runId = providedRunId || `run_${Date.now().toString(36)}_${++runIdSeq}`; - notifications.startRun({ runId, graph, workflowName, workflowId }); + if (!suppressNotifications) notifications.startRun({ runId, graph, workflowName, workflowId }); pendingRunIds.add(runId); // 启动即落盘运行中快照：成果面板在 run-start 后立刻拉 /run-results， // 只等最终 persistRun 的话长运行期间 readRun 一直 404（前端退避耗尽即报「运行记录不存在」）。 @@ -1096,6 +1154,9 @@ export function apply(ctx, config) { triggerInput: triggerInput ?? '', workflowName: workflowName || null, workflowId: workflowId || null, canvasId: canvasId || null, source: source || null, replayOf: replayOf || null, revises: revises || null, + parentRunId: parentRunId || null, parentNodeId: parentNodeId || null, + rootRunId: rootRunId || runId, depth: Number.isInteger(depth) && depth >= 0 ? depth : 0, + callChain: Array.isArray(callChain) ? [...callChain] : [], ...(resume ? { resumedFrom: resume.runId || null } : {}), nodeStates: {}, outputs: {}, structuredOutputs: {}, issues: [], graph: graph ? { nodes: graph.nodes.map((n) => ({ id: n.id, type: n.type, position: n.position, data: n.data })), edges: graph.edges } : undefined, @@ -1104,6 +1165,7 @@ export function apply(ctx, config) { } catch { /* 快照写失败不阻塞运行；最终 persistRun 仍会落盘 */ } const promise = workspaceContext.run(store, () => Promise.resolve().then(() => orch.run(graph, { triggerInput, workflowName, workflowId, canvasId, source, runId, revises, + parentRunId, parentNodeId, rootRunId, depth, callChain, workspaceRoot: store.workspaceRoot, globalVariables, workflowVariables, runInputs, resume, @@ -1121,10 +1183,132 @@ export function apply(ctx, config) { notifications.discard(runId); ctx.logger?.error?.(`dsh-ccpg 运行失败（${runId}）：${error.message}`); return null; - }).finally(() => pendingRunIds.delete(runId)); + }).finally(() => { + pendingRunIds.delete(runId); + if (!parentRunId && ![...liveChildRuns.values()].some((children) => children.size > 0)) { + childCountByRoot.delete(rootRunId || runId); + } + }); return { runId, promise }; }; + // 子工作流执行器：lint → 循环/深度/预算防护 → 输入映射 → 启动子 run → 同步等待并选结果。 + // 取消传播双向：父 abort 信号取消子 run；子 run 终态后再判父状态，父 run-end 不早于子终态。 + orch.runChildWorkflow = async ({ node, run, state, signal, render }) => { + if (signal?.aborted || run.canceled) throw new SubworkflowError('父运行已取消', 'SUBWORKFLOW_CANCELED'); + const workflowId = String(node.data?.workflowId || '').trim(); + if (!workflowId) throw new SubworkflowError('未选择目标子工作流', 'SUBWORKFLOW_WORKFLOW_REQUIRED'); + const childWorkflow = readWf(workflowId); + if (!childWorkflow) throw new SubworkflowError(`子工作流不存在：${workflowId}`, 'SUBWORKFLOW_NOT_FOUND'); + const lint = lintWorkflowGraph(childWorkflow.graph); + if (!lint.ok) { + const error = new SubworkflowError( + `子工作流图有错误：${lint.issues.find((issue) => issue.level === 'error')?.message || '图存在错误'}`, + 'SUBWORKFLOW_INVALID_GRAPH', + ); + error.issues = lint.issues; + throw error; + } + const chain = Array.isArray(run.callChain) && run.callChain.length + ? [...run.callChain] + : (run.workflowId ? [run.workflowId] : []); + if (chain.includes(workflowId)) { + throw new SubworkflowError(`检测到子工作流循环调用：${[...chain, workflowId].join(' → ')}`, 'SUBWORKFLOW_CYCLE'); + } + const depth = Number.isInteger(run.depth) ? run.depth : 0; + if (depth >= MAX_SUBWORKFLOW_DEPTH) { + throw new SubworkflowError(`子工作流嵌套层级超过上限 ${MAX_SUBWORKFLOW_DEPTH}`, 'SUBWORKFLOW_MAX_DEPTH'); + } + const rootRunId = run.rootRunId || run.runId; + const childCount = (childCountByRoot.get(rootRunId) || 0) + 1; + if (childCount > MAX_CHILD_RUNS_PER_ROOT) { + throw new SubworkflowError(`单次运行子工作流数量超过上限 ${MAX_CHILD_RUNS_PER_ROOT}`, 'SUBWORKFLOW_BUDGET_EXCEEDED'); + } + const parentCtx = { + outputs: new Map((state.incoming.get(node.id) || []).map((id) => [id, state.run.outputs[id] ?? ''])), + structuredOutputs: new Map((state.incoming.get(node.id) || []).map((id) => [id, state.run.structuredOutputs?.[id]])), + labels: new Map((state.incoming.get(node.id) || []).map((id) => [id, state.nodes.get(id)?.data?.label || id])), + incomingIds: state.incoming.get(node.id) || [], + triggerInput: run.triggerInput, + nodeStates: run.nodeStates, + globalVariables: state.globalVariables, + workflowVariables: state.workflowVariables, + runInputs: state.runInputs, + }; + const mapped = resolveSubworkflowInputs(node.data || {}, parentCtx, render); + const globals = globalContext(); + const childInputs = validateSubworkflowInputs(mapped.runInputs, childWorkflow.inputSchema); + if (signal?.aborted || run.canceled) throw new SubworkflowError('父运行已取消', 'SUBWORKFLOW_CANCELED'); + + childCountByRoot.set(rootRunId, childCount); + const child = startRun(childWorkflow.graph, { + triggerInput: mapped.triggerInput, + workflowName: childWorkflow.name, + workflowId: childWorkflow.id, + globalVariables: globals.globalVariables, + workflowVariables: variableDefinitionsToValues(childWorkflow.variables), + runInputs: childInputs, + source: 'subworkflow', + parentRunId: run.runId, + parentNodeId: node.id, + rootRunId, + depth: depth + 1, + callChain: [...chain, workflowId], + suppressNotifications: true, + }); + if (!child?.runId || !child?.promise) throw new SubworkflowError('子工作流启动失败', 'SUBWORKFLOW_START_FAILED'); + const children = childRunOf(run.runId); + const cancelChild = (reason) => { try { orch.cancel(child.runId, reason); } catch { /* 子运行可能已结束 */ } }; + children.set(child.runId, cancelChild); + let onAbort; + let abortPromise; + let childSettled = false; + try { + if (signal?.aborted) { + cancelChild('父运行取消'); + throw new SubworkflowError('父运行已取消', 'SUBWORKFLOW_CANCELED'); + } + abortPromise = new Promise((_, reject) => { + onAbort = () => { + if (childSettled) return; + cancelChild('父运行取消'); + // 父节点在 child 真正收尾前保持 running，避免父 run-end 早于 child 终态。 + Promise.resolve(child.promise).then( + () => reject(new SubworkflowError('父运行已取消', 'SUBWORKFLOW_CANCELED')), + () => reject(new SubworkflowError('父运行已取消', 'SUBWORKFLOW_CANCELED')), + ); + }; + signal?.addEventListener('abort', onAbort, { once: true }); + }); + const childRun = await Promise.race([child.promise, abortPromise]); + childSettled = true; + if (!childRun) throw new SubworkflowError('子工作流未返回运行结果', 'SUBWORKFLOW_START_FAILED'); + if (signal?.aborted || run.canceled) throw new SubworkflowError('父运行已取消', 'SUBWORKFLOW_CANCELED'); + const selected = selectWorkflowResult(childRun); + if (childRun.status === 'canceled') throw new SubworkflowError('子工作流已取消', 'SUBWORKFLOW_CANCELED'); + if (childRun.status !== 'success') { + const error = new SubworkflowError(`子工作流运行失败：${childRun.error || childRun.status}`, 'SUBWORKFLOW_CHILD_FAILED'); + error.childRunId = child.runId; + error.childStatus = childRun.status; + error.childSummary = selected.output; + throw error; + } + return { + output: selected.output, + structuredOutput: selected.structuredOutput, + childRunId: child.runId, + childWorkflowId: childWorkflow.id, + childStatus: childRun.status, + childSummary: selected.output.slice(0, 2000), + childArtifacts: selected.artifacts, + }; + } finally { + if (onAbort) signal?.removeEventListener('abort', onAbort); + children.delete(child.runId); + if (!children.size) liveChildRuns.delete(run.runId); + } + }; + // ---- agent 节点执行（升级版）---- async function runAgentNode(ctx, node, run, s, { signal, emit, runId }) { const store = currentStore(); @@ -1757,6 +1941,14 @@ export function apply(ctx, config) { } if (req.method === 'DELETE') { if (!readWf(id)) return json(res, 404, { error: '工作流不存在' }); + const referencing = subworkflowReferencingWorkflows(id); + if (referencing.length) { + return json(res, 409, { + error: `有 ${referencing.length} 个工作流通过子工作流节点引用它：${referencing.map((r) => r.name || r.id).join('、')}，请先移除那些子工作流节点`, + code: 'subworkflow-referenced', + referencing, + }); + } deleteWf(id); return json(res, 200, { ok: true }); } @@ -1884,7 +2076,7 @@ export function apply(ctx, config) { const body = await readBody(req); const entry = orch.runs.get(body?.runId); const ok = entry?.run?.workspaceRoot === currentStore().workspaceRoot - ? orch.cancel(body?.runId, '用户取消') + ? cancelRunTree(body?.runId, '用户取消') : false; json(res, 200, { ok, runId: body?.runId || null }); } }); @@ -1930,7 +2122,7 @@ export function apply(ctx, config) { }; const testAbort = new AbortController(); - const testTimeoutMs = Number(node.data?.timeoutSec) > 0 ? Number(node.data.timeoutSec) * 1000 : 500 * 1000; + const testTimeoutMs = Number(node.data?.timeoutSec) > 0 ? Number(node.data.timeoutSec) * 1000 : 5 * 60 * 1000; let testTimedOut = false; const testTimer = setTimeout(() => { testTimedOut = true; testAbort.abort(); }, testTimeoutMs); req.once('aborted', () => testAbort.abort()); @@ -1982,6 +2174,7 @@ export function apply(ctx, config) { return { done, total, succeeded }; }; const resumableRun = (r, isLive) => !isLive + && !hasSubworkflowNode(r.graph) && ['error', 'canceled', 'interrupted'].includes(r.status) && Boolean(r.graph?.nodes?.length) && Object.values(r.nodeStates || {}).some((st) => st?.status === 'success') @@ -1997,6 +2190,10 @@ export function apply(ctx, config) { workflowId: summary.workflowId ?? null, workflowName: summary.workflowName ?? null, source: summary.source ?? null, + parentRunId: summary.parentRunId ?? null, + parentNodeId: summary.parentNodeId ?? null, + rootRunId: summary.rootRunId ?? summary.runId ?? null, + depth: Number.isInteger(summary.depth) ? summary.depth : 0, startedAt: summary.startedAt ?? null, durationMs: summary.durationMs ?? null, live: isLive, @@ -2023,6 +2220,10 @@ export function apply(ctx, config) { const isLive = liveIds.has(r.runId); return { ...summary, + parentRunId: summary.parentRunId ?? null, + parentNodeId: summary.parentNodeId ?? null, + rootRunId: summary.rootRunId ?? summary.runId ?? null, + depth: Number.isInteger(summary.depth) ? summary.depth : 0, outputs: summarizeOutputs(summary.outputs, structuredOutputs), nodeStates: summarizeNodeStates(summary.nodeStates), structuredOutputSummary: summarizeStructuredOutputs(structuredOutputs), @@ -2040,7 +2241,28 @@ export function apply(ctx, config) { if (!id) return json(res, 400, { error: '缺少 id' }); const r = readRun(id); if (!r) return json(res, 404, { error: '运行记录不存在' }); - json(res, 200, { ...r, structuredOutputs: r.structuredOutputs || {} }); + const childDocuments = new Map(currentDatabase().listChildRuns(id, 100).map((child) => [child.runId, child])); + for (const entry of orch.runs.values()) { + if (entry.run.workspaceRoot === currentStore().workspaceRoot && entry.run.parentRunId === id) { + childDocuments.set(entry.run.runId, entry.run); + } + } + const children = [...childDocuments.values()].slice(0, 100).map((child) => ({ + runId: child.runId, + workflowId: child.workflowId ?? null, + workflowName: child.workflowName ?? null, + status: child.status, + source: child.source ?? null, + parentRunId: child.parentRunId ?? null, + parentNodeId: child.parentNodeId ?? null, + rootRunId: child.rootRunId ?? child.runId ?? null, + depth: Number.isInteger(child.depth) ? child.depth : 0, + startedAt: child.startedAt ?? null, + finishedAt: child.finishedAt ?? null, + durationMs: child.durationMs ?? null, + resumable: resumableRun(child, orch.runs.has(child.runId)), + })); + json(res, 200, { ...r, structuredOutputs: r.structuredOutputs || {}, children }); } }); register({ kind: 'exact', path: '/wf1/api/run-results', async handler(req, res) { @@ -2182,6 +2404,9 @@ export function apply(ctx, config) { if (!prev) return json(res, 404, { error: '运行记录不存在' }); const graph = body?.graph || prev.graph; if (!graph || !Array.isArray(graph.nodes)) return json(res, 400, { error: '该运行没有图快照，需传 graph' }); + if (hasSubworkflowNode(graph)) { + return json(res, 409, { error: '含子工作流节点的运行首期不支持重放，避免重复执行子运行', code: 'subworkflow-replay-unsupported' }); + } const triggerInput = body?.triggerInput !== undefined ? String(body.triggerInput) : String(prev.triggerInput || ''); let runInputs; try { runInputs = assertSafeContextObject(body?.runInputs ?? prev.runInputs, 'runInputs'); } catch (error) { return routeError(res, error); } let globals; try { globals = globalContext(); } catch (error) { return routeError(res, error); } @@ -2206,6 +2431,9 @@ export function apply(ctx, config) { if (!prev) return json(res, 404, { error: '运行记录不存在' }); if (orch.runs.has(prev.runId)) return json(res, 409, { error: '该运行仍在进行中', code: 'run-live' }); if (!prev.graph || !Array.isArray(prev.graph.nodes)) return json(res, 400, { error: '该运行没有图快照，无法续跑' }); + if (hasSubworkflowNode(prev.graph)) { + return json(res, 409, { error: '含子工作流节点的运行首期不支持续跑，避免重复执行子运行', code: 'subworkflow-resume-unsupported' }); + } const succeeded = Object.values(prev.nodeStates || {}).filter((st) => st?.status === 'success'); if (!succeeded.length) return json(res, 400, { error: '该运行没有已完成的节点，无需续跑', code: 'nothing-to-resume' }); const persistedWorkflow = prev.workflowId ? readWf(prev.workflowId) : null; diff --git a/dsh-plugins/dsh-ccpg-orchestrator/lib/output-contract.js b/dsh-plugins/dsh-ccpg-orchestrator/lib/output-contract.js index 81eb948..a8dd5f4 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/lib/output-contract.js +++ b/dsh-plugins/dsh-ccpg-orchestrator/lib/output-contract.js @@ -133,7 +133,7 @@ export function mergeExecutionResults(baseResult, patchResult) { }; } -const NODE_META_ALLOWLIST = ['status', 'chars', 'durationMs', 'model', 'runtime', 'turns', 'usage', 'writeback', 'notification', 'toleratedError']; +const NODE_META_ALLOWLIST = ['status', 'chars', 'durationMs', 'model', 'runtime', 'turns', 'usage', 'writeback', 'notification', 'toleratedError', 'errorCode', 'childRunId', 'childWorkflowId', 'childStatus', 'childSummary', 'childArtifacts']; export function safeNodeStateMeta(state = {}) { return Object.fromEntries(NODE_META_ALLOWLIST.filter((key) => state[key] !== undefined).map((key) => [key, state[key]])); diff --git a/dsh-plugins/dsh-ccpg-orchestrator/lib/sqlite-store.js b/dsh-plugins/dsh-ccpg-orchestrator/lib/sqlite-store.js index 5e55d00..f2ee689 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/lib/sqlite-store.js +++ b/dsh-plugins/dsh-ccpg-orchestrator/lib/sqlite-store.js @@ -252,6 +252,7 @@ export class WorkflowSqliteStore { getRun: this.db.prepare('SELECT updated_at, document_json FROM runs WHERE run_id = ?'), listRuns: this.db.prepare('SELECT document_json FROM runs ORDER BY started_at DESC, run_id DESC LIMIT ?'), listRunsForWorkflow: this.db.prepare('SELECT document_json FROM runs WHERE workflow_id = ? ORDER BY started_at DESC, run_id DESC LIMIT ?'), + listChildRuns: this.db.prepare("SELECT document_json FROM runs WHERE json_extract(document_json, '$.parentRunId') = ? ORDER BY started_at ASC, run_id ASC LIMIT ?"), putRun: this.db.prepare(` INSERT INTO runs (run_id, workflow_id, status, started_at, finished_at, updated_at, document_json) VALUES (?, ?, ?, ?, ?, ?, ?) @@ -350,6 +351,13 @@ export class WorkflowSqliteStore { return this.#runRunStatement(this.statements.putRun, value); } + listChildRuns(parentRunId, limit = 100) { + if (!parentRunId) return []; + const count = Math.max(0, Math.floor(Number(limit) || 0)); + return this.statements.listChildRuns.all(String(parentRunId), count) + .map((row) => parseDocument(row, normalizeRunDocument)); + } + pruneRuns(keep, { keepRevisionRuns = [] } = {}) { const count = Math.max(0, Math.floor(Number(keep) || 0)); const protectedIds = new Set(keepRevisionRuns.map(String)); diff --git a/dsh-plugins/dsh-ccpg-orchestrator/lib/subworkflow.js b/dsh-plugins/dsh-ccpg-orchestrator/lib/subworkflow.js new file mode 100644 index 0000000..3771146 --- /dev/null +++ b/dsh-plugins/dsh-ccpg-orchestrator/lib/subworkflow.js @@ -0,0 +1,189 @@ +import { parseTemplate } from './template-parser.js'; +import { resolveTypedExpression } from './typed-expression.js'; +import { toJsonSafe, normalizeExecutionResult } from './output-contract.js'; + +export const MAX_SUBWORKFLOW_DEPTH = 3; +export const MAX_CHILD_RUNS_PER_ROOT = 16; + +export class SubworkflowError extends Error { + constructor(message, code = 'SUBWORKFLOW_ERROR') { + super(message); + this.name = 'SubworkflowError'; + this.code = code; + } +} + +const isObject = (value) => value !== null && typeof value === 'object' && !Array.isArray(value); + +function expressionTemplate(value) { + const source = String(value || '').trim(); + if (!source) throw new SubworkflowError('子工作流输入引用不能为空', 'SUBWORKFLOW_INPUT_EXPRESSION'); + return source.includes('{{') ? source : `{{${source}}}`; +} + +export function resolveSubworkflowValue(value, ctx, render) { + if (isObject(value) && typeof value.$ref === 'string') { + try { return resolveTypedExpression(expressionTemplate(value.$ref), ctx); } + catch (error) { + const wrapped = new SubworkflowError(error.message, 'SUBWORKFLOW_INPUT_MISSING'); + wrapped.cause = error; + throw wrapped; + } + } + if (typeof value === 'string') { + if (value.trim() === '$upstream') return render('{{$upstream}}', { implicitUpstream: true }).text; + const parsed = parseTemplate(value); + if (parsed.tokens.length === 1 && parsed.tokens[0].type === 'variable') { + try { return resolveTypedExpression(value, ctx); } + catch (error) { + const wrapped = new SubworkflowError(error.message, 'SUBWORKFLOW_INPUT_MISSING'); + wrapped.cause = error; + throw wrapped; + } + } + return render(value, { implicitUpstream: false }).text; + } + if (Array.isArray(value)) return value.map((item) => resolveSubworkflowValue(item, ctx, render)); + if (isObject(value)) return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, resolveSubworkflowValue(item, ctx, render)])); + return value; +} + +export function resolveSubworkflowInputs(data, ctx, render) { + const map = isObject(data?.inputMap) ? data.inputMap : {}; + const triggerSource = map.triggerInput === undefined ? '$upstream' : map.triggerInput; + const triggerInput = resolveSubworkflowValue(triggerSource, ctx, render); + const runInputsSource = map.runInputs === undefined ? {} : map.runInputs; + const runInputs = resolveSubworkflowValue(runInputsSource, ctx, render); + if (!isObject(runInputs)) throw new SubworkflowError('子工作流 runInputs 必须是对象', 'SUBWORKFLOW_INVALID_INPUT'); + return { triggerInput, runInputs: toJsonSafe(runInputs, '$.runInputs') }; +} + +function outputNodeIds(run) { + return (run.graph?.nodes || []).filter((node) => (node.type || node.data?.nodeType) === 'output').map((node) => node.id); +} + +export function selectWorkflowResult(run) { + const outputs = run?.outputs || {}; + const structured = run?.structuredOutputs || {}; + const states = run?.nodeStates || {}; + const ids = outputNodeIds(run); + const candidates = ids.length ? ids : [...(run?.nodeOrder || []), ...Object.keys(states)]; + const selected = candidates.filter((id) => states[id]?.status === 'success' && Object.prototype.hasOwnProperty.call(outputs, id)); + const idsToUse = ids.length ? selected : selected.slice(-1); + const text = idsToUse.map((id) => String(outputs[id] ?? '')).filter(Boolean).join('\n\n'); + const firstId = idsToUse.at(-1); + const envelope = firstId && structured[firstId] + ? structured[firstId] + : normalizeExecutionResult(text).structuredOutput; + const artifacts = idsToUse.flatMap((id) => Array.isArray(states[id]?.artifacts) ? states[id].artifacts : []); + return { + output: text, + structuredOutput: envelope, + artifacts: [...new Set(artifacts)], + sourceNodeIds: idsToUse, + }; +} + +export function validateWorkflowInputs(runInputs, inputSchema = {}, { label = '工作流' } = {}) { + if (!isObject(runInputs)) throw new SubworkflowError(`${label} runInputs 必须是对象`, 'WORKFLOW_INPUT_INVALID'); + const fields = Array.isArray(inputSchema?.fields) ? inputSchema.fields : []; + const known = new Set(); + const output = { ...runInputs }; + for (const field of fields) { + const key = String(field?.key ?? field?.name ?? '').trim(); + if (!key) continue; + known.add(key); + const has = Object.prototype.hasOwnProperty.call(output, key); + if (!has && field.required === true && field.defaultValue === undefined) { + throw new SubworkflowError(`${label}缺少必填输入：${key}`, 'WORKFLOW_INPUT_REQUIRED'); + } + if (!has && field.defaultValue !== undefined) output[key] = structuredClone(field.defaultValue); + if (!Object.prototype.hasOwnProperty.call(output, key)) continue; + const value = output[key]; + const type = String(field.type || 'json'); + const valid = type === 'string' ? typeof value === 'string' + : type === 'number' ? typeof value === 'number' && Number.isFinite(value) + : type === 'boolean' ? typeof value === 'boolean' + : type === 'string[]' ? Array.isArray(value) && value.every((item) => typeof item === 'string') + : type === 'object' ? isObject(value) + : type === 'json' ? value !== undefined + : type === 'array' ? Array.isArray(value) + : true; + if (!valid) throw new SubworkflowError(`${label}输入 ${key} 类型不匹配：需要 ${type}`, 'WORKFLOW_INPUT_TYPE'); + if (Array.isArray(field.enum) && !field.enum.includes(value)) { + throw new SubworkflowError(`${label}输入 ${key} 不在允许值范围内`, 'WORKFLOW_INPUT_ENUM'); + } + } + for (const key of Object.keys(output)) { + if (known.has(key)) continue; + if (fields.length) throw new SubworkflowError(`${label}不支持输入字段：${key}`, 'WORKFLOW_INPUT_UNKNOWN'); + } + return toJsonSafe(output, '$.runInputs'); +} + +export function validateSubworkflowInputs(runInputs, inputSchema = {}) { + try { + return validateWorkflowInputs(runInputs, inputSchema, { label: '子工作流' }); + } catch (error) { + const code = String(error.code || ''); + if (code.startsWith('WORKFLOW_INPUT_')) error.code = code.replace(/^WORKFLOW_INPUT_/, 'SUBWORKFLOW_INPUT_'); + throw error; + } +} + +/** + * 收集需要模板静态检查的字段：triggerInput 与 runInputs 内所有字符串叶。 + * $ref 值按运行时语义包成 {{...}} 再交给 validateTemplate，让语法错误在 + * lint 阶段就暴露（而不是运行时输入解析才炸）。 + */ +export function subworkflowTemplateFields(data = {}) { + const map = isObject(data.inputMap) ? data.inputMap : {}; + const fields = []; + if (typeof map.triggerInput === 'string' && map.triggerInput.trim()) fields.push(map.triggerInput); + const walk = (value) => { + if (typeof value === 'string') { fields.push(value); return; } + if (Array.isArray(value)) { value.forEach(walk); return; } + if (isObject(value)) { + if (typeof value.$ref === 'string') { + try { fields.push(expressionTemplate(value.$ref)); } catch { /* 空引用由运行时校验兜底 */ } + return; + } + Object.values(value).forEach(walk); + } + }; + if (map.runInputs !== undefined) walk(map.runInputs); + return fields; +} + +export function validateSubworkflowNode(node, lintCtx = {}) { + const issues = []; + const label = node?.data?.label || node?.id || '子工作流'; + const workflowId = String(node?.data?.workflowId || '').trim(); + if (!workflowId) issues.push({ level: 'error', code: 'SUBWORKFLOW_WORKFLOW_REQUIRED', message: `子工作流「${label}」未选择目标工作流` }); + if (node?.data?.waitForCompletion === false) { + issues.push({ level: 'error', code: 'SUBWORKFLOW_ASYNC_UNSUPPORTED', message: `子工作流「${label}」首期只支持等待完成模式` }); + } + if (Number(node?.data?.retryCount) > 0) { + issues.push({ level: 'error', code: 'SUBWORKFLOW_RETRY_UNSUPPORTED', message: `子工作流「${label}」首期不支持重试，避免重复执行子运行` }); + } + if (node?.data?.inputMap !== undefined && !isObject(node.data.inputMap)) { + issues.push({ level: 'error', code: 'SUBWORKFLOW_INPUT_MAP', message: `子工作流「${label}」输入映射必须是对象` }); + } + // 宿主经 resolveTargetWorkflow 注入库查询（engine 自身不读库）：目标不存在或 + // inputSchema 字段名错配都在画布 lint 直接报，不留到运行时。 + const canResolve = typeof lintCtx.resolveTargetWorkflow === 'function'; + const target = canResolve && workflowId ? lintCtx.resolveTargetWorkflow(workflowId) : undefined; + if (canResolve && workflowId && !target) { + issues.push({ level: 'error', code: 'SUBWORKFLOW_NOT_FOUND', message: `子工作流「${label}」引用的工作流不存在：${workflowId}` }); + } + const targetInputSchema = target?.inputSchema ?? lintCtx.targetInputSchema; + const knownFields = Array.isArray(targetInputSchema?.fields) + ? new Set(targetInputSchema.fields.map((field) => String(field?.key ?? field?.name ?? '').trim()).filter(Boolean)) + : null; + if (knownFields && isObject(node?.data?.inputMap?.runInputs)) { + for (const key of Object.keys(node.data.inputMap.runInputs)) { + if (!knownFields.has(key)) issues.push({ level: 'error', code: 'SUBWORKFLOW_INPUT_UNKNOWN', message: `子工作流「${label}」不支持输入字段：${key}` }); + } + } + return issues; +} diff --git a/dsh-plugins/dsh-ccpg-orchestrator/lib/variable-schema.js b/dsh-plugins/dsh-ccpg-orchestrator/lib/variable-schema.js index fc68ec9..2c24d04 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/lib/variable-schema.js +++ b/dsh-plugins/dsh-ccpg-orchestrator/lib/variable-schema.js @@ -26,6 +26,12 @@ const META_SCHEMA = objectSchema({ usage: objectSchema({}, 'Token 用量'), writeback: field('any', '输出写回结果'), notification: field('any', '消息通知发送结果'), toleratedError: field('string', '容错继续的错误'), + errorCode: field('string', '稳定错误码'), + childRunId: field('string', '子运行 ID'), + childWorkflowId: field('string', '子工作流 ID'), + childStatus: field('string', '子运行状态'), + childSummary: field('string', '子运行摘要'), + childArtifacts: arraySchema(field('string', '子运行产物引用'), '子运行产物'), }); function inferType(value, fallback = 'any') { diff --git a/dsh-plugins/dsh-ccpg-orchestrator/test/assistant.test.mjs b/dsh-plugins/dsh-ccpg-orchestrator/test/assistant.test.mjs index 746a421..9326d5e 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/test/assistant.test.mjs +++ b/dsh-plugins/dsh-ccpg-orchestrator/test/assistant.test.mjs @@ -1,6 +1,6 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { validateGraphOps, wouldCreateCycle, summarizeGraphForAI, checkPatchResult, canvasAssistantPersona } from '../lib/assistant.js'; +import { validateGraphOps, wouldCreateCycle, summarizeGraphForAI, checkPatchResult } from '../lib/assistant.js'; const baseGraph = () => ({ nodes: [ @@ -110,17 +110,6 @@ test('summarizeGraphForAI keeps agent fields compact', () => { assert.equal(s.edges[0].from, 'n_input_1'); }); -test('canvas assistant persona requires clarification and confirmation contracts', () => { - const persona = canvasAssistantPersona(); - assert.match(persona, /目标不唯一时必须主动澄清/); - assert.match(persona, /列出候选.*禁止猜测/); - assert.match(persona, /删除节点、清空画布、覆盖已保存工作流/); - assert.match(persona, /运行中的图做结构修改前/); - assert.match(persona, /先复述.*等待用户明确确认/); - assert.match(persona, /必须带选项/); - assert.match(persona, /澄清不超过一轮/); -}); - test('checkPatchResult lint flags empty include/exclude or ok graph', () => { const okGraph = { nodes: [{ id: 'a', type: 'input', data: { label: 'A' } }], edges: [] }; const r = checkPatchResult(okGraph); diff --git a/dsh-plugins/dsh-ccpg-orchestrator/test/engine.test.mjs b/dsh-plugins/dsh-ccpg-orchestrator/test/engine.test.mjs index fe56ac5..26cecdb 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/test/engine.test.mjs +++ b/dsh-plugins/dsh-ccpg-orchestrator/test/engine.test.mjs @@ -653,4 +653,144 @@ await test('lint 检出 canonical 非直接上游与无效 agent schema', () => assert.match(messages, /Schema 无效/); }); +await test('空图运行能正常收尾', async () => { + const { orch } = makeOrch(); + const run = await orch.run({ nodes: [], edges: [] }, { runId: 'run_empty' }); + assert.equal(run.status, 'success'); + assert.deepEqual(run.nodeStates, {}); + assert.equal(orch.currentRunIds().includes('run_empty'), false); +}); + +await test('取消排队节点不会重复扣减 remaining', async () => { + let cancelOnQueued = false; + const { orch, events } = makeOrch(async (node, _run, _s, ctl) => { + if (node.id === 'slow') { + await new Promise((resolve) => { + const timer = setTimeout(resolve, 2000); + ctl.signal.addEventListener('abort', () => { clearTimeout(timer); resolve(); }, { once: true }); + }); + throw new Error('运行已取消'); + } + return { output: node.id }; + }); + const cancelListener = (event, payload) => { + if (!cancelOnQueued && event === 'node-status' && payload.status === 'queued' && payload.nodeId === 'slow') { + cancelOnQueued = true; + orch.cancel(payload.runId, '队列取消'); + } + }; + const originalEmit = orch.emit.bind(orch); + orch.emit = (event, payload) => { + originalEmit(event, payload); + cancelListener(event, payload); + }; + const run = await orch.run({ + nodes: [ + { id: 'slow', type: 'agent', data: { label: 'slow' } }, + { id: 'after', type: 'agent', data: { label: 'after' } }, + ], + edges: [{ source: 'slow', target: 'after' }], + }, { runId: 'run_queue_cancel' }); + assert.equal(run.status, 'canceled'); + assert.equal(run.nodeStates.slow.status, 'canceled'); + assert.equal(run.nodeStates.after.status, 'canceled'); + const terminal = events.filter(([event, payload]) => event === 'node-status' + && payload.runId === 'run_queue_cancel' + && ['success', 'error', 'canceled', 'skipped'].includes(payload.status)); + assert.equal(new Set(terminal.map(([, payload]) => payload.nodeId)).size, 2); +}); + +await test('重试退避可被取消且仍能收尾', async () => { + let calls = 0; + const { orch } = makeOrch(async () => { + calls += 1; + throw new Error('瞬时失败'); + }); + const running = orch.run({ + nodes: [{ id: 'retry', type: 'agent', data: { retryCount: 5 } }], + edges: [], + }, { runId: 'run_retry_cancel' }); + await delay(30); + orch.cancel('run_retry_cancel', '退避期间取消'); + const run = await running; + assert.equal(run.status, 'canceled'); + assert.equal(run.nodeStates.retry.status, 'canceled'); + assert.equal(calls, 1); +}); + +await test('subworkflow runner waits for child and maps cancellation to parent', async () => { + let release; + const gate = new Promise((resolve) => { release = resolve; }); + const { orch } = makeOrch(); + orch.runChildWorkflow = async ({ signal }) => { + await new Promise((resolve, reject) => { + const onAbort = () => reject(Object.assign(new Error('父运行已取消'), { code: 'SUBWORKFLOW_CANCELED' })); + signal.addEventListener('abort', onAbort, { once: true }); + gate.then(() => { signal.removeEventListener('abort', onAbort); resolve(); }); + }); + return { output: 'child output', childRunId: 'child_1', childStatus: 'success' }; + }; + const running = orch.run({ nodes: [{ id: 'sub', type: 'subworkflow', data: { workflowId: 'wf_child' } }], edges: [] }, { runId: 'run_parent_child' }); + await delay(20); + assert.equal(orch.runs.get('run_parent_child')?.run.nodeStates.sub.status, 'running'); + orch.cancel('run_parent_child', '用户取消'); + release(); + const run = await running; + assert.equal(run.status, 'canceled'); + assert.equal(run.nodeStates.sub.status, 'canceled'); + assert.equal(run.nodeStates.sub.errorCode, 'SUBWORKFLOW_CANCELED'); +}); + +await test('subworkflow 节点不重复执行 retry', async () => { + let calls = 0; + const { orch } = makeOrch(); + orch.runChildWorkflow = async () => { + calls += 1; + throw new Error('child failed'); + }; + const run = await orch.run({ + nodes: [{ id: 'sub', type: 'subworkflow', data: { workflowId: 'wf_child', retryCount: 5 } }], + edges: [], + }, { runId: 'run_subworkflow_no_retry' }); + assert.equal(run.status, 'error'); + assert.equal(run.nodeStates.sub.status, 'error'); + assert.equal(calls, 1); +}); + +await test('lint：subworkflow 模板字段与目标工作流解析可检出', () => { + const graph = { + nodes: [ + { id: 'up', type: 'agent', data: { label: '上游', prompt: 'x' } }, + { id: 'sub', type: 'subworkflow', data: { label: '子', workflowId: 'wf_child', inputMap: { triggerInput: '{{node["ghost"].text}}' } } }, + ], + edges: [{ id: 'e1', source: 'up', target: 'sub' }], + }; + const bad = lintGraph(graph); + assert.equal(bad.ok, false); + assert.ok(bad.issues.some((issue) => issue.nodeId === 'sub' && /变量引用没有该节点/.test(issue.message)), '坏模板引用应报 error'); + + const okGraph = { + nodes: [ + { id: 'up', type: 'agent', data: { label: '上游', prompt: 'x' } }, + { id: 'sub', type: 'subworkflow', data: { label: '子', workflowId: 'wf_child', inputMap: { triggerInput: '{{上游}}', runInputs: { ticket: { $ref: 'node["up"].data' } } } } }, + ], + edges: [{ id: 'e1', source: 'up', target: 'sub' }], + }; + const ok = lintGraph(okGraph); + assert.equal(ok.issues.filter((issue) => issue.nodeId === 'sub' && issue.level === 'error').length, 0, `合法引用不应报错：${JSON.stringify(ok.issues)}`); + + // 宿主注入 resolveTargetWorkflow 后：目标不存在 / 输入字段名错配都在 lint 检出 + const resolveTargetWorkflow = (id) => (id === 'wf_child' ? { id, inputSchema: { fields: [{ key: 'ticket' }] } } : null); + const missing = lintGraph({ + nodes: [{ id: 'sub', type: 'subworkflow', data: { label: '子', workflowId: 'wf_gone' } }], + edges: [], + }, { resolveTargetWorkflow }); + assert.ok(missing.issues.some((issue) => issue.code === 'SUBWORKFLOW_NOT_FOUND')); + const unknownField = lintGraph({ + nodes: [{ id: 'sub', type: 'subworkflow', data: { label: '子', workflowId: 'wf_child', inputMap: { runInputs: { nope: 'x' } } } }], + edges: [], + }, { resolveTargetWorkflow }); + assert.ok(unknownField.issues.some((issue) => issue.code === 'SUBWORKFLOW_INPUT_UNKNOWN')); +}); + console.log(process.exitCode ? `${passed} tests passed with failures` : `ALL PASS (${passed})`); diff --git a/dsh-plugins/dsh-ccpg-orchestrator/test/sqlite-store.test.mjs b/dsh-plugins/dsh-ccpg-orchestrator/test/sqlite-store.test.mjs index df1e118..cbd8853 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/test/sqlite-store.test.mjs +++ b/dsh-plugins/dsh-ccpg-orchestrator/test/sqlite-store.test.mjs @@ -149,6 +149,26 @@ test('filters runs by workflowId and keeps the unfiltered default', () => { } }); +test('lists child runs by parent in start order and normalizes documents', () => { + const root = mkdtempSync(join(tmpdir(), 'wf1-sqlite-children-')); + try { + const store = new WorkflowSqliteStore({ + databaseFile: join(root, 'workflow-one.sqlite'), + workflowsDir: join(root, 'workflows'), + runsDir: join(root, 'runs'), + }); + store.putRun({ ...run('child_b', '2026-08-02T00:00:00.000Z', 'success', 'wf_child'), parentRunId: 'parent_1', parentNodeId: 'sub_b', rootRunId: 'parent_1', depth: 1, source: 'subworkflow' }); + store.putRun({ ...run('unrelated', '2026-08-01T00:00:00.000Z'), parentRunId: 'other_parent' }); + store.putRun({ ...run('child_a', '2026-08-01T00:00:00.000Z', 'error', 'wf_child'), parentRunId: 'parent_1', parentNodeId: 'sub_a', rootRunId: 'parent_1', depth: 1, source: 'subworkflow' }); + assert.deepEqual(store.listChildRuns('parent_1').map((value) => value.runId), ['child_a', 'child_b']); + assert.equal(store.listChildRuns('parent_1')[0].parentNodeId, 'sub_a'); + assert.deepEqual(store.listChildRuns('missing'), []); + store.close(); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + test('rolls back schema migration when the database is incompatible', () => { const root = mkdtempSync(join(tmpdir(), 'wf1-sqlite-rollback-')); try { diff --git a/dsh-plugins/dsh-ccpg-orchestrator/test/subworkflow.test.mjs b/dsh-plugins/dsh-ccpg-orchestrator/test/subworkflow.test.mjs new file mode 100644 index 0000000..e4563e0 --- /dev/null +++ b/dsh-plugins/dsh-ccpg-orchestrator/test/subworkflow.test.mjs @@ -0,0 +1,108 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import { + MAX_CHILD_RUNS_PER_ROOT, + MAX_SUBWORKFLOW_DEPTH, + SubworkflowError, + resolveSubworkflowInputs, + resolveSubworkflowValue, + selectWorkflowResult, + subworkflowTemplateFields, + validateSubworkflowInputs, + validateSubworkflowNode, +} from '../lib/subworkflow.js'; + +const context = { + outputs: new Map([['up', 'hello'], ['json', '{"ok":true}']]), + structuredOutputs: new Map([ + ['up', { version: 1, type: 'text', value: 'hello' }], + ['json', { version: 1, type: 'json', value: { ok: true } }], + ]), + labels: new Map([['up', '上游']]), + incomingIds: ['up', 'json'], + triggerInput: 'trigger', + nodeStates: { up: { status: 'success' }, json: { status: 'success' } }, + runInputs: { ticket: { id: 7 } }, + globalVariables: { enabled: true }, + workflowVariables: { mode: 'test' }, +}; +const render = (value) => ({ text: String(value).replace('{{node["up"].text}}', 'hello').replace('{{$upstream}}', 'hello') }); + +test('subworkflow resolves typed refs, templates, arrays, and objects', () => { + assert.deepEqual(resolveSubworkflowValue({ $ref: 'inputs["ticket"]' }, context, render), { id: 7 }); + assert.equal(resolveSubworkflowValue('{{node["up"].text}} / x', context, render), 'hello / x'); + assert.deepEqual(resolveSubworkflowValue([{ $ref: 'vars.global["enabled"]' }, 2], context, render), [true, 2]); + assert.equal(resolveSubworkflowValue('$upstream', context, render), 'hello'); + assert.deepEqual(resolveSubworkflowInputs({ inputMap: { triggerInput: { $ref: 'node["json"].data' }, runInputs: { text: { $ref: 'node["up"].text' } } } }, context, render), { + triggerInput: { ok: true }, runInputs: { text: 'hello' }, + }); +}); + +test('subworkflow input validation applies defaults and rejects invalid fields', () => { + const schema = { fields: [ + { key: 'required', type: 'string', required: true }, + { key: 'count', type: 'number', defaultValue: 1 }, + ] }; + assert.deepEqual(validateSubworkflowInputs({ required: 'x' }, schema), { required: 'x', count: 1 }); + assert.throws(() => validateSubworkflowInputs({}, schema), (error) => error instanceof SubworkflowError && error.code === 'SUBWORKFLOW_INPUT_REQUIRED'); + assert.throws(() => validateSubworkflowInputs({ required: 'x', extra: true }, schema), (error) => error.code === 'SUBWORKFLOW_INPUT_UNKNOWN'); + assert.throws(() => validateSubworkflowInputs({ required: 1 }, schema), (error) => error.code === 'SUBWORKFLOW_INPUT_TYPE'); +}); + +test('subworkflow result selects output nodes and falls back to final business node', () => { + const outputResult = selectWorkflowResult({ + graph: { nodes: [{ id: 'a', type: 'agent' }, { id: 'o1', type: 'output' }, { id: 'o2', type: 'output' }] }, + nodeOrder: ['a', 'o1', 'o2'], + nodeStates: { a: { status: 'success' }, o1: { status: 'success', artifacts: ['a.md'] }, o2: { status: 'success', artifacts: ['b.md'] } }, + outputs: { a: 'agent', o1: 'first', o2: 'second' }, + structuredOutputs: { o2: { version: 1, type: 'text', value: 'second' } }, + }); + assert.equal(outputResult.output, 'first\n\nsecond'); + assert.deepEqual(outputResult.artifacts, ['a.md', 'b.md']); + assert.equal(outputResult.sourceNodeIds.join(','), 'o1,o2'); + const fallback = selectWorkflowResult({ + graph: { nodes: [{ id: 'a', type: 'agent' }, { id: 'b', type: 'script' }] }, + nodeOrder: ['a', 'b'], + nodeStates: { a: { status: 'success' }, b: { status: 'success' } }, + outputs: { a: 'A', b: 'B' }, structuredOutputs: {}, + }); + assert.equal(fallback.output, 'B'); +}); + +test('subworkflow node lint rejects missing target, async mode, retry, and invalid input map', () => { + const issues = validateSubworkflowNode({ id: 'sub', data: { label: '调用', waitForCompletion: false, retryCount: 2, inputMap: [] } }); + assert.deepEqual(issues.map((issue) => issue.code), ['SUBWORKFLOW_WORKFLOW_REQUIRED', 'SUBWORKFLOW_ASYNC_UNSUPPORTED', 'SUBWORKFLOW_RETRY_UNSUPPORTED', 'SUBWORKFLOW_INPUT_MAP']); + assert.equal(MAX_SUBWORKFLOW_DEPTH, 3); + assert.equal(MAX_CHILD_RUNS_PER_ROOT, 16); +}); + +test('subworkflow node lint resolves target workflow for existence and input schema', () => { + const childDoc = { id: 'wf_child', inputSchema: { fields: [{ key: 'ticket', type: 'string' }] } }; + const resolveTargetWorkflow = (id) => (id === 'wf_child' ? childDoc : null); + const node = { id: 'sub', data: { label: '调用', workflowId: 'wf_child', inputMap: { runInputs: { extra: 'x' } } } }; + const issues = validateSubworkflowNode(node, { resolveTargetWorkflow }); + assert.deepEqual(issues.map((issue) => issue.code), ['SUBWORKFLOW_INPUT_UNKNOWN']); + const ok = validateSubworkflowNode({ id: 'sub', data: { workflowId: 'wf_child', inputMap: { runInputs: { ticket: 'a' } } } }, { resolveTargetWorkflow }); + assert.deepEqual(ok, []); + const missing = validateSubworkflowNode({ id: 'sub', data: { label: '调用', workflowId: 'wf_gone' } }, { resolveTargetWorkflow }); + assert.deepEqual(missing.map((issue) => issue.code), ['SUBWORKFLOW_NOT_FOUND']); + // 无解析器时（引擎离线单测等）跳过存在性与字段校验，不报错 + const skipped = validateSubworkflowNode({ id: 'sub', data: { workflowId: 'wf_any', inputMap: { runInputs: { whatever: 1 } } } }); + assert.deepEqual(skipped, []); +}); + +test('subworkflow template fields collect triggerInput, nested runInputs strings, and wrapped $ref', () => { + const fields = subworkflowTemplateFields({ + inputMap: { + triggerInput: '{{node["up"].text}} 前缀', + runInputs: { + ticket: { $ref: 'node["up"].data' }, + list: ['常量', { nested: '{{vars.global["g"]}}' }], + num: 42, + }, + }, + }); + assert.deepEqual(fields, ['{{node["up"].text}} 前缀', '{{node["up"].data}}', '常量', '{{vars.global["g"]}}']); + assert.deepEqual(subworkflowTemplateFields({}), []); + assert.deepEqual(subworkflowTemplateFields({ inputMap: { triggerInput: '$upstream' } }), ['$upstream']); +}); diff --git a/dsh-plugins/dsh-ccpg-orchestrator/test/workflow-tools.integration.test.mjs b/dsh-plugins/dsh-ccpg-orchestrator/test/workflow-tools.integration.test.mjs index 637079d..1d15b7d 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/test/workflow-tools.integration.test.mjs +++ b/dsh-plugins/dsh-ccpg-orchestrator/test/workflow-tools.integration.test.mjs @@ -230,6 +230,31 @@ await test('workflow_delete：缺 confirm 拒绝；有关联定时任务/webhook assert.equal(gone.status, 404); }); +await test('workflow_delete：被子工作流节点引用时拒绝并列出引用方（工具与 HTTP 一致）', async () => { + // wf_t3 = 被引用的子工作流；wf_t4 = 含 subworkflow 节点的父工作流 + const c3 = await call('POST', '/wf1/api/workflows', { id: 'wf_t3', name: '子目标', graph: wfGraph }); + assert.equal(c3.status, 200); + const c4 = await call('POST', '/wf1/api/workflows', { id: 'wf_t4', name: '父流程', graph: { + nodes: [{ id: 'sub1', type: 'subworkflow', position: { x: 0, y: 0 }, data: { label: '子调用', workflowId: 'wf_t3' } }], + edges: [], + } }); + assert.equal(c4.status, 200); + const guardedTool = await runTool('workflow_delete', { workflowId: 'wf_t3', confirm: true }); + assert.match(guardedTool, /子工作流节点引用/); + assert.match(guardedTool, /父流程/); + const guardedHttp = await call('DELETE', '/wf1/api/workflows/detail?id=wf_t3'); + assert.equal(guardedHttp.status, 409); + assert.equal(guardedHttp.body.code, 'subworkflow-referenced'); + assert.deepEqual(guardedHttp.body.referencing, [{ id: 'wf_t4', name: '父流程' }]); + // 移除引用后两个入口都可删 + const patch = await runTool('workflow_patch', { workflowId: 'wf_t4', ops: [{ op: 'deleteNode', id: 'sub1' }] }); + assert.match(patch, /已应用/); + const done = maybeJson(await runTool('workflow_delete', { workflowId: 'wf_t3', confirm: true })); + assert.equal(done.deleted, true); + const del4 = await call('DELETE', '/wf1/api/workflows/detail?id=wf_t4'); + assert.equal(del4.status, 200); +}); + await test('workflow_open：未绑定画布的会话被拒；绑定后切换', async () => { // session-2 未绑定画布 → workflow_open 拒绝 const def = registeredTools.get('workflow_open'); diff --git a/web/src/App.jsx b/web/src/App.jsx index 3c0e58f..144dac5 100644 --- a/web/src/App.jsx +++ b/web/src/App.jsx @@ -93,6 +93,7 @@ export default function App() { const terminalNodesByRunRef = useRef(new Map()); const [canvasMenu, setCanvasMenu] = useState(null); // 双击画布 { x, y } 弹加节点菜单 const [catalog, setCatalog] = useState({ tools: [], feishuEnabled: false, notificationChannels: [] }); + const [availableWorkflows, setAvailableWorkflows] = useState([]); const [skills, setSkills] = useState([]); const [llmConfig, setLLMConfig] = useState({}); const [runtime, setRuntime] = useState(null); @@ -304,6 +305,8 @@ export default function App() { if (runtimeData) setRuntime(runtimeData.runtime || { available: false }); if (credsData) setFeishuCreds(credsData.credentials || []); if (larkData) setLarkStatus(larkData.status || null); + const workflowsData = await j('/workflows'); + if (workflowsData) setAvailableWorkflows(workflowsData.workflows || []); const latest = runsData?.runs?.[0]; // 初次恢复也要按工作流对齐：草稿图带着 workflowId 时优先取该工作流的最近运行 const draftWorkflowId = graphData?.workflowId; @@ -1779,6 +1782,7 @@ export default function App() { onTest={openTestNode} onClose={() => setSelectedId(null)} availableTools={catalog.tools} + availableWorkflows={availableWorkflows} skills={skills} feishuEnabled={catalog.feishuEnabled} feishuCreds={feishuCreds} diff --git a/web/src/NodePanel.jsx b/web/src/NodePanel.jsx index 019adbf..175ff50 100644 --- a/web/src/NodePanel.jsx +++ b/web/src/NodePanel.jsx @@ -28,7 +28,7 @@ const TOOL_LABELS = { feishu_doc_write: '写飞书文档', }; -const TYPE_TEXT = { input: '输入', agent: '智能体', output: '输出', condition: '条件', http: 'HTTP', script: '脚本', notify: '消息通知', note: '注释' }; +const TYPE_TEXT = { input: '输入', agent: '智能体', output: '输出', condition: '条件', http: 'HTTP', script: '脚本', notify: '消息通知', note: '注释', subworkflow: '子工作流' }; /** 面板实时活动区的跳秒计时：每秒重渲染（仅运行中挂载） */ function useElapsedTick(active) { @@ -131,11 +131,15 @@ function ScriptParameterRow({ input, index, error, templateProps, onChange, onRe ); } -export function NodePanel({ node, onChange, onDelete, onTest, onClose, availableTools = [], skills = [], feishuEnabled = false, feishuCreds = [], notificationChannels = [], llmConfig = {}, upstreamNodes = [], upstreamPreviews = {}, graph, workflowId, runId, workflowVariables, inputSchema, runInputs, triggerInput, globalVariableEpoch, progress }) { +export function NodePanel({ node, onChange, onDelete, onTest, onClose, availableTools = [], availableWorkflows = [], skills = [], feishuEnabled = false, feishuCreds = [], notificationChannels = [], llmConfig = {}, upstreamNodes = [], upstreamPreviews = {}, graph, workflowId, runId, workflowVariables, inputSchema, runInputs, triggerInput, globalVariableEpoch, progress }) { if (!node) return null; const [copied, setCopied] = useState(false); const [testing, setTesting] = useState(false); + const [runInputsText, setRunInputsText] = useState(() => JSON.stringify(node.data?.inputMap?.runInputs ?? {}, null, 2)); const d = node.data || {}; + useEffect(() => { + setRunInputsText(JSON.stringify(d.inputMap?.runInputs ?? {}, null, 2)); + }, [d.inputMap?.runInputs]); const nodeType = d.nodeType || node.type; const notifyTargetType = d.channelConfig?.targetType || 'chat_id'; useElapsedTick(d.runStatus === 'running'); @@ -353,6 +357,51 @@ export function NodePanel({ node, onChange, onDelete, onTest, onClose, available )} + {nodeType === 'subworkflow' && ( + <> +
+ + + + + set({ workflowId: event.target.value.trim() || undefined, workflowName: undefined })} + placeholder="wf_xxx" /> + + {d.workflowId &&

目标：{d.workflowName || d.workflowId}

} +
+
+ set({ inputMap: { ...(d.inputMap || {}), triggerInput: value } })} + placeholder="$upstream" + compact + singleLine + /> + +