fix(orchestrator): 运行历史抽屉按画布工作流隔离（#79）

dsh-plugins/dsh-ccpg-orchestrator/lib/index.js

@@ -927,7 +927,7 @@ export function apply(ctx, config) {
     const document = normalizeRunDocument(run);
     return currentDatabase().putRun(document);
   };
-  const recentRuns = (limit = 50) => currentDatabase().listRuns(limit).map((run) => (
+  const recentRuns = (limit = 50, workflowId) => currentDatabase().listRuns(limit, workflowId).map((run) => (
     run.status === 'running' && !pendingRunIds.has(run.runId) ? (readRun(run.runId) || run) : run
   ));
   const checkpointRun = (runId) => {
@@ -1786,10 +1786,13 @@ export function apply(ctx, config) {
   register({ kind: 'exact', path: '/wf1/api/runs', handler(req, res) {
     const url = new URL(req.url, 'http://x');
     const limit = Math.min(Number(url.searchParams.get('limit')) || 20, 100);
+    // 可选 workflowId 过滤：历史抽屉按画布打开的工作流隔离（与 workflow_runs 工具同语义）；
+    // 缺省维持工作区级全量（草稿画布 / RunSwitcher 等既有调用不受影响）
+    const workflowId = url.searchParams.get('workflowId') || undefined;
     const liveIds = new Set([...orch.runs.values()].filter((entry) => entry.run.workspaceRoot === currentStore().workspaceRoot).map((entry) => entry.run.runId));
     json(res, 200, {
       // 保持既有字段面（triggerInput/canvasId 等平铺），整形逻辑与 workflow_runs 工具同源
-      runs: recentRuns(limit).map((r) => {
+      runs: recentRuns(limit, workflowId).map((r) => {
         const { structuredOutputs, graph, ...summary } = r;
         const isLive = liveIds.has(r.runId);
         return {

dsh-plugins/dsh-ccpg-orchestrator/lib/sqlite-store.js

@@ -180,6 +180,7 @@ export class WorkflowSqliteStore {
       deleteWorkflow: this.db.prepare('DELETE FROM workflows WHERE id = ?'),
       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 ?'),
       putRun: this.db.prepare(`
         INSERT INTO runs (run_id, workflow_id, status, started_at, finished_at, updated_at, document_json)
         VALUES (?, ?, ?, ?, ?, ?, ?)
@@ -255,9 +256,12 @@ export class WorkflowSqliteStore {
     return row ? { document: parseDocument(row, normalizeRunDocument), updatedAt: row.updated_at } : null;
   }
 
-  listRuns(limit = 50) {
+  listRuns(limit = 50, workflowId) {
     const count = Math.max(0, Math.floor(Number(limit) || 0));
-    return this.statements.listRuns.all(count).map((row) => parseDocument(row, normalizeRunDocument));
+    // workflowId 过滤命中 runs_workflow_started_at 索引；缺省（含 null/草稿）保持全量行为
+    const statement = workflowId ? this.statements.listRunsForWorkflow : this.statements.listRuns;
+    const rows = workflowId ? statement.all(String(workflowId), count) : statement.all(count);
+    return rows.map((row) => parseDocument(row, normalizeRunDocument));
   }
 
   putRun(value) {

dsh-plugins/dsh-ccpg-orchestrator/test/schedule-api.integration.test.mjs

@@ -308,6 +308,54 @@ await test('并发回归：同图并发两个 run 同时 live、输出互不串
   }
 });
 
+await test('GET /runs：workflowId 过滤按画布工作流隔离；缺省维持全量（#79）', async () => {
+  const originalFetch = globalThis.fetch;
+  const gates = new Map();
+  try {
+    globalThis.fetch = (url) => new Promise((resolve) => {
+      gates.set(String(url), () => resolve(new Response(String(url), { status: 200 })));
+    });
+    const graph = {
+      nodes: [
+        { id: 'hist_input', type: 'input', position: { x: 0, y: 0 }, data: { label: '输入', text: 'seed' } },
+        { id: 'hist_output', type: 'output', position: { x: 200, y: 0 }, data: { label: '输出' } },
+      ],
+      edges: [{ source: 'hist_input', target: 'hist_output' }],
+    };
+    // 两个已保存工作流各跑一次 + 一条草稿运行，形成跨工作区混合记录。
+    // 命名工作流运行须带匹配的 graphFingerprint（画布与库一致性的运行时校验）。
+    const wfA = await call('POST', '/wf1/api/workflows', { id: 'wf_hist_a', name: '历史工作流A', graph });
+    const wfB = await call('POST', '/wf1/api/workflows', { id: 'wf_hist_b', name: '历史工作流B', graph });
+    assert.equal(wfA.status, 200);
+    assert.equal(wfB.status, 200);
+    const runA = await call('POST', '/wf1/api/run', { workflowId: 'wf_hist_a', graphFingerprint: wfA.body.graphFingerprint });
+    const runB = await call('POST', '/wf1/api/run', { workflowId: 'wf_hist_b', graphFingerprint: wfB.body.graphFingerprint });
+    const runDraft = await call('POST', '/wf1/api/run', { graph });
+    assert.equal(runA.status, 200);
+    assert.equal(runB.status, 200);
+    assert.equal(runDraft.status, 200);
+    for (const release of gates.values()) release?.();
+
+    const scoped = await call('GET', '/wf1/api/runs?workflowId=wf_hist_a');
+    assert.equal(scoped.status, 200);
+    const scopedIds = scoped.body.runs.map((r) => r.runId);
+    assert.ok(scopedIds.includes(runA.body.runId), '过滤结果含本工作流运行');
+    assert.ok(!scopedIds.includes(runB.body.runId) && !scopedIds.includes(runDraft.body.runId), '过滤结果不含别的工作流/草稿运行');
+    assert.ok(scoped.body.runs.every((r) => r.workflowId === 'wf_hist_a'), '全部条目 workflowId 对齐');
+
+    const missing = await call('GET', '/wf1/api/runs?workflowId=wf_missing');
+    assert.equal(missing.status, 200);
+    assert.deepEqual(missing.body.runs, [], '无匹配返回空列表');
+
+    const all = await call('GET', '/wf1/api/runs');
+    const allIds = all.body.runs.map((r) => r.runId);
+    for (const id of [runA.body.runId, runB.body.runId, runDraft.body.runId]) assert.ok(allIds.includes(id), `全量列表含 ${id}`);
+  } finally {
+    for (const release of gates.values()) release?.();
+    globalThis.fetch = originalFetch;
+  }
+});
+
 for (const d of disposers) await d?.();
 rmSync(workspacesRoot, { recursive: true, force: true });
 rmSync(dshHome, { recursive: true, force: true });

dsh-plugins/dsh-ccpg-orchestrator/test/sqlite-store.test.mjs

@@ -121,6 +121,34 @@ test('prunes runs by startedAt and keeps the newest rows', () => {
   }
 });
 
+test('filters runs by workflowId and keeps the unfiltered default', () => {
+  const root = mkdtempSync(join(tmpdir(), 'wf1-sqlite-filter-'));
+  try {
+    const store = new WorkflowSqliteStore({
+      databaseFile: join(root, 'workflow-one.sqlite'),
+      workflowsDir: join(root, 'workflows'),
+      runsDir: join(root, 'runs'),
+    });
+    store.putRun(run('run_a1', '2026-08-01T00:00:00.000Z', 'success', 'wf_a'));
+    store.putRun(run('run_b1', '2026-08-02T00:00:00.000Z', 'success', 'wf_b'));
+    store.putRun(run('run_draft', '2026-08-03T00:00:00.000Z', 'success', null));
+    store.putRun(run('run_a2', '2026-08-04T00:00:00.000Z', 'success', 'wf_a'));
+    // 只看 wf_a：时间倒序且不含别的工作流/草稿
+    assert.deepEqual(store.listRuns(10, 'wf_a').map((row) => row.runId), ['run_a2', 'run_a1']);
+    // limit 在过滤内生效
+    assert.deepEqual(store.listRuns(1, 'wf_a').map((row) => row.runId), ['run_a2']);
+    // 无匹配返回空数组（不是 undefined/null）
+    assert.deepEqual(store.listRuns(10, 'wf_missing'), []);
+    // 缺省（undefined/null/空串）维持全量：草稿画布与既有调用不受影响
+    assert.deepEqual(store.listRuns(10).map((row) => row.runId), ['run_a2', 'run_draft', 'run_b1', 'run_a1']);
+    assert.deepEqual(store.listRuns(10, null).map((row) => row.runId), ['run_a2', 'run_draft', 'run_b1', 'run_a1']);
+    assert.deepEqual(store.listRuns(10, '').map((row) => row.runId), ['run_a2', 'run_draft', 'run_b1', 'run_a1']);
+    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 {

web/src/App.jsx

@@ -1677,7 +1677,7 @@ export default function App() {
           )}
         </Modal>
       )}
-      {historyOpen && <RunHistory onClose={() => setHistoryOpen(false)}
+      {historyOpen && <RunHistory onClose={() => setHistoryOpen(false)} workflowId={currentWf?.id || null}
         onResume={(runId, resumedNodes, rerunNodes) => {
           activeRunIdRef.current = runId;
           runningRef.current = true;

web/src/RunHistory.jsx

@@ -11,14 +11,17 @@ function progressText(r) {
   return `${p.done}/${p.total}`;
 }
 
-export function RunHistory({ onClose, onSelect, onResume }) {
+export function RunHistory({ onClose, onSelect, onResume, workflowId }) {
   const [runs, setRuns] = useState([]);
   const [resuming, setResuming] = useState('');
 
+  // 已保存工作流的画布只看该工作流的运行（后端按 workflowId 过滤）；
+  // 草稿画布没有 workflowId，维持工作区全量。
   const load = () => {
-    fetch(apiUrl('/runs')).then((r) => r.json()).then((d) => setRuns(d.runs || [])).catch(() => {});
+    const query = workflowId ? `?workflowId=${encodeURIComponent(workflowId)}` : '';
+    fetch(apiUrl(`/runs${query}`)).then((r) => r.json()).then((d) => setRuns(d.runs || [])).catch(() => {});
   };
-  useEffect(load, []);
+  useEffect(load, [workflowId]);
 
   const resumeRun = async (runId) => {
     if (resuming) return;