From 3c09614662b3f0739f972ea0425b07c70fea12dc Mon Sep 17 00:00:00 2001 From: chumingjun <15951837502@163.com> Date: Sat, 29 Aug 2026 02:50:18 +0800 Subject: [PATCH 1/6] =?UTF-8?q?feat(orchestrator):=20=E7=BB=AD=E8=B7=91?= =?UTF-8?q?=E4=BA=A7=E7=89=A9=E7=89=A9=E5=8C=96=20+=20artifact=20=E8=B7=AF?= =?UTF-8?q?=E7=94=B1=20resumedFrom=20=E7=A5=96=E5=85=88=E5=9B=9E=E9=80=80?= =?UTF-8?q?=20+=20agent-progress=20=E5=AE=9E=E6=97=B6=E6=B5=81=E5=A2=9E?= =?UTF-8?q?=E5=BC=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 断点续跑的可复用节点不重新执行，其工作区文件物理上留在祖先运行的 runtime 目录，导致按当前 runId 定位的两处消费方全部失联： - persistRun 快照：snapshotRunArtifacts 在新 run 的节点工作区找不到文件， 可复用节点的产物整个进不了 artifactIndex（成果面板/导出缺文件） - /wf1/api/artifact 兜底：历史 resume 运行的过程产物 404 修法两层： - materializeResumedWorkspaces：持久化前把祖先运行中可复用节点的产物按 nodeStates.artifacts 物化拷贝进本次运行目录（COPYFILE_EXCL、路径越界 校验、失败仅告警不阻塞），让快照天然命中 - artifact 路由兜底沿 resumedFrom 祖先链回退解析（深度 10 防环）， 救已存在的历史 resume 运行数据 agent-progress 增强（文稿视图实时流消费）： - preview 截断 200→4096 字符：assistant 全文拼接，多轮生成期前端可看 文稿长大；带宽 = 4KB × 并发 agent ÷ 2s，量级安全 - 首个 turn/end 后延迟 2.5s 复查一次再确认结束：多工具轮 agent 常在 首轮文本后继续调用工具，立即退出会漏报后续轮次（单轮语义不变） 集成测试：seed 祖先运行工作区产物 → 续跑 → 断言新 run 目录物化拷贝 存在 + /artifact 路由 200 命中。 --- .../dsh-ccpg-orchestrator/lib/index.js | 74 ++++++++++++++++--- .../test/plugin-storage.integration.test.mjs | 38 ++++++++++ 2 files changed, 100 insertions(+), 12 deletions(-) diff --git a/dsh-plugins/dsh-ccpg-orchestrator/lib/index.js b/dsh-plugins/dsh-ccpg-orchestrator/lib/index.js index c101235..87a451c 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/lib/index.js +++ b/dsh-plugins/dsh-ccpg-orchestrator/lib/index.js @@ -15,7 +15,7 @@ import { randomUUID } from 'node:crypto'; import { AsyncLocalStorage } from 'node:async_hooks'; -import { readFileSync, writeFileSync, existsSync, mkdirSync, readdirSync, statSync, copyFileSync, cpSync, unlinkSync, renameSync, realpathSync, rmSync } from 'node:fs'; +import { readFileSync, writeFileSync, existsSync, mkdirSync, readdirSync, statSync, copyFileSync, cpSync, unlinkSync, renameSync, realpathSync, rmSync, constants as fsConstants } from 'node:fs'; import { join, dirname, extname, isAbsolute, relative, resolve, sep } from 'node:path'; import { fileURLToPath } from 'node:url'; import { homedir } from 'node:os'; @@ -849,6 +849,34 @@ export function apply(ctx, config) { ...(recoveredStatus === 'interrupted' ? { error: run.error || '运行进程异常终止' } : {}), }); }; + // 断点续跑的可复用节点不重新执行，其工作区文件仍留在祖先运行的 runtime 目录； + // 先物化拷贝到本次运行目录，快照与 /artifact 路由按 runId 定位才能命中。 + // 拷贝失败只记 issue 不阻塞持久化（祖先目录被清理时产物缺失属既成事实）。 + const materializeResumedWorkspaces = (run) => { + if (!run.resumedFrom || !run.nodeStates) return; + const scope = { workflowId: run.workflowId || 'draft', runId: run.runId }; + for (const [nodeId, state] of Object.entries(run.nodeStates)) { + if (state?.status !== 'success' || !Array.isArray(state.artifacts) || !state.artifacts.length) continue; + const sourceRoot = STORAGE.workspaceForNode({ workflowId: run.workflowId || 'draft', runId: run.resumedFrom, nodeId }); + const targetRoot = STORAGE.workspaceForNode({ ...scope, nodeId }); + for (const relativePath of state.artifacts) { + if (!relativePath || String(relativePath).endsWith('/')) continue; + try { + const source = resolveInside(sourceRoot, relativePath); + const target = resolveInside(targetRoot, relativePath); + if (!source || !target) continue; + if (existsSync(target)) continue; + if (!existsSync(source) || !statSync(source).isFile()) continue; + const realSource = realpathSync(source); + if (resolveInside(realpathSync(sourceRoot), realSource) !== realSource) continue; + mkdirSync(dirname(target), { recursive: true, mode: 0o700 }); + copyFileSync(realSource, target, fsConstants.COPYFILE_EXCL); + } catch (error) { + ctx.logger?.warn?.(`[wf1] 续跑产物物化失败（${run.runId}/${nodeId}/${relativePath}）：${error.message}`); + } + } + } + }; const persistRun = (run, graph, workflowName, workflowId) => { try { const light = { ...run, _resolved: true }; @@ -866,6 +894,7 @@ export function apply(ctx, config) { graph: graphSnapshot, }); const scope = { workflowId: base.workflowId || 'draft', runId: base.runId }; + materializeResumedWorkspaces(base); const snapshot = snapshotRunArtifacts(base, { workspaceForNode: ({ nodeId }) => STORAGE.workspaceForNode({ ...scope, nodeId }), artifactRunDir: STORAGE.artifactRunDir(scope), @@ -1238,7 +1267,9 @@ export function apply(ctx, config) { const { turns, preview, turnEnded } = scanEvents(); emit('agent-progress', { runId, nodeId: node.id, turns, - preview: outputConfig.mode === 'structured' ? '' : preview.slice(0, 200), + // 实时输出流（文稿视图消费）：assistant 全文拼接，4KB 截断——多工具轮 agent 生成期 + // 前端可看文稿长大；带宽 = 4KB × 并发 agent ÷ 2s，量级安全 + preview: outputConfig.mode === 'structured' ? '' : preview.slice(0, 4096), structured: outputConfig.mode === 'structured' || undefined, maxRounds: maxRounds || undefined, }); @@ -1246,6 +1277,18 @@ export function apply(ctx, config) { try { agent.cancel({ kind: 'user' }); } catch { /* noop */ } return watchDone(); } + // 首个 turn/end 后延迟复查一次：多工具轮 agent 常在首轮文本后继续调用工具， + // 立即退出会漏报后续轮次；复查仍无新 turn 才确认结束（单轮 agent 语义不变） + if (turnEnded && !watchState.rechecking) { + watchState.rechecking = true; + watchState.timer = setTimeout(() => { + if (watchState.stop) return; + const next = scanEvents(); + if (next.turns > turns) { watchState.rechecking = false; watchTick(); return; } + watchDone(); + }, 2500); + return; + } if (turnEnded) return watchDone(); watchState.timer = setTimeout(watchTick, 2000); }; @@ -2568,18 +2611,25 @@ export function apply(ctx, config) { file: resolved.file, filename: resolved.artifact.name, mediaType, preview, }); } - // 运行中/试运行：运行文档还没有快照，直接从节点工作区解析 - const ws = resolveInside(STORAGE.workspaceForNode({ - workflowId: run?.workflowId || 'draft', runId, nodeId: nodeParam, - }), file); - if (ws && existsSync(ws) && statSync(ws).isFile()) { + // 运行中/试运行：运行文档还没有快照，直接从节点工作区解析； + // 断点续跑的节点产物物理上在祖先运行目录，沿 resumedFrom 链回退（有限深度防环） + const ancestorRunIds = [runId]; + let cursor = readRun(runId); + for (let depth = 0; cursor?.resumedFrom && depth < 10; depth += 1) { + ancestorRunIds.push(cursor.resumedFrom); + cursor = readRun(cursor.resumedFrom); + } + for (const candidateRunId of ancestorRunIds) { + const ws = resolveInside(STORAGE.workspaceForNode({ + workflowId: run?.workflowId || 'draft', runId: candidateRunId, nodeId: nodeParam, + }), file); + if (!ws || !existsSync(ws) || !statSync(ws).isFile()) continue; const realWsParent = realpathSync(dirname(ws)); const realWs = realpathSync(ws); - if (resolveInside(realWsParent, realWs) === realWs) { - const mediaType = mediaTypeFor(file); - const preview = url.searchParams.get('preview') === '1' && isPreviewableMediaType(mediaType); - return streamArtifactResponse(req, res, { file: realWs, filename: file, mediaType, preview }); - } + if (resolveInside(realWsParent, realWs) !== realWs) continue; + const mediaType = mediaTypeFor(file); + const preview = url.searchParams.get('preview') === '1' && isPreviewableMediaType(mediaType); + return streamArtifactResponse(req, res, { file: realWs, filename: file, mediaType, preview }); } return json(res, 404, { error: '产物不存在' }); } diff --git a/dsh-plugins/dsh-ccpg-orchestrator/test/plugin-storage.integration.test.mjs b/dsh-plugins/dsh-ccpg-orchestrator/test/plugin-storage.integration.test.mjs index 95a535c..b0dc5bb 100644 --- a/dsh-plugins/dsh-ccpg-orchestrator/test/plugin-storage.integration.test.mjs +++ b/dsh-plugins/dsh-ccpg-orchestrator/test/plugin-storage.integration.test.mjs @@ -6,6 +6,7 @@ import { tmpdir } from 'node:os'; import { DatabaseSync } from 'node:sqlite'; import { apply } from '../lib/index.js'; import { graphFingerprint } from '../lib/run-scope.js'; +import { hashedKey } from '../lib/storage-paths.js'; function responseCapture() { const listeners = new Map(); @@ -373,6 +374,20 @@ try { nodeStates: { resume_input: { status: 'success' }, resume_output: { status: 'running' } }, outputs: { resume_input: 'hello' }, structuredOutputs: {}, nodeOrder: ['resume_input', 'resume_output'], }); + // 续跑物化前置：祖先运行的可复用节点在工作区留了文件，新 run 目录初始为空 + const resumeSeedWorkspace = join( + workspaceB, '.workflow-one', 'runtime', + hashedKey('wf_resume_named'), hashedKey('run_resume_named_seed'), + 'nodes', hashedKey('resume_input'), 'workspace', + ); + mkdirSync(resumeSeedWorkspace, { recursive: true }); + writeFileSync(join(resumeSeedWorkspace, '底稿.md'), '# 祖先产物\n'); + // 同步进 seed 的 nodeStates.artifacts（真实运行里 success 节点都会带清单） + { + const seedDoc = readStoredRun(workspaceB, 'run_resume_named_seed'); + seedDoc.nodeStates.resume_input.artifacts = ['底稿.md']; + seedStoredRun(workspaceB, seedDoc); + } const namedMismatch = responseCapture(); await route('/wf1/api/runs/resume')(request('POST', withSession('/wf1/api/runs/resume', 'session-b'), { runId: 'run_resume_named_seed', graph: resumeGraph, @@ -392,6 +407,29 @@ try { assert.equal(namedResume.status, 200); assert.equal(namedResume.json().resumedFrom, 'run_resume_named_seed'); + // 续跑物化：可复用节点的祖先工作区文件必须拷进新 run 目录， + // 且 /artifact 兜底沿 resumedFrom 链也能命中（祖先目录被清后仍可读） + const resumedRunId = namedResume.json().runId; + let resumedRunDoc; + for (let attempt = 0; attempt < 60; attempt += 1) { + resumedRunDoc = readStoredRun(workspaceB, resumedRunId); + if (resumedRunDoc && resumedRunDoc.status !== 'running') break; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + assert.equal(resumedRunDoc?.status, 'success', `续跑运行未完成：${resumedRunDoc?.status}`); + const resumedWorkspaceFile = join( + workspaceB, '.workflow-one', 'runtime', + hashedKey('wf_resume_named'), hashedKey(resumedRunId), + 'nodes', hashedKey('resume_input'), 'workspace', '底稿.md', + ); + assert.equal(readFileSync(resumedWorkspaceFile, 'utf8'), '# 祖先产物\n'); + const resumedArtifact = responseCapture(); + await route('/wf1/api/artifact')(request('GET', withSession( + `/wf1/api/artifact?run=${resumedRunId}&node=resume_input&file=${encodeURIComponent('底稿.md')}&preview=1`, 'session-b', + )), resumedArtifact); + assert.equal(resumedArtifact.status, 200); + assert.match(resumedArtifact.headers['Content-Type'] || '', /text\/markdown/); + const missingSession = responseCapture(); await route('/wf1/api/graph')(request('GET', '/wf1/api/graph'), missingSession); assert.equal(missingSession.status, 409); From 3116b15d09ff75333dbfd031d4b51b4b954342f4 Mon Sep 17 00:00:00 2001 From: chumingjun <15951837502@163.com> Date: Sat, 29 Aug 2026 02:50:39 +0800 Subject: [PATCH 2/6] =?UTF-8?q?feat(web):=20=E6=96=87=E7=A8=BF=E8=A7=86?= =?UTF-8?q?=E5=9B=BE=E2=80=94=E2=80=94=E8=BF=90=E8=A1=8C=E4=BA=A7=E7=89=A9?= =?UTF-8?q?=E4=B8=BB=E4=BB=8E=E5=A2=99=EF=BC=88=E8=8A=82=E7=82=B9=E6=9D=A1?= =?UTF-8?q?=E5=B8=A6=20+=20=E5=A4=A7=E5=8D=A1=20+=20=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E6=B5=81=E6=B8=B2=E6=9F=93=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 画布级第三视图（画布 | 文稿 | 工作流）：整块主区切换为「左侧节点列表 + 右侧大卡横向条带」的文稿墙，把一次运行的过程与成果铺成可读文档。 数据层 doc-wall-data.js（纯函数、node 直测）： - buildDocWallModel 把 run-results（磁盘事实投影）+ nodeStates（节点工作 区产物清单）+ progressByNode（SSE 实时态）整理成 主从模型：finals 成果 带 + nodes 按执行拓扑序分桶（doc/image/video 铺卡，data 折 chip） - 去重时带 downloadUrl 的行胜出：adaptRunResults 派生的 processFiles 只有 文件名，同键 stateArtifacts（scoped artifact URL）必须覆盖它 - 卡内正文 2000 字符截断；scopedArtifactUrl 工厂注入（apiUrl 带 sessionId） 视图 DocWallView.jsx： - DocCard：md 卡 MarkdownDocument 渲染；nodeStates 产物无内联正文，进入 视口后惰性 fetch（startedRef 防重入），失败渲染「正文暂不可读」占位 + 下载链接，不白屏；图片懒加载；视频不挂