diff --git a/web/src/components/WorkGraphCanvas.vue b/web/src/components/WorkGraphCanvas.vue index d62886a..9daea21 100644 --- a/web/src/components/WorkGraphCanvas.vue +++ b/web/src/components/WorkGraphCanvas.vue @@ -86,10 +86,21 @@ ⏱ {{ elapsedLabels.get(n.key) }} + + + + {{ joinWaitLabels.get(n.key).text }} + {{ joinWaitLabels.get(n.key).title }} + + ⊞ - NEXT ▶ + + NEXT ▶
@@ -102,6 +113,7 @@ import { computed, ref, watch, onBeforeUnmount } from 'vue'; import { fmtDuration } from './graph-console/node-signals'; import { edgeContractLabel } from './graph-console/io-model'; +import { resolveJoinWait, joinWaitLabel } from './graph-console/join-status'; /** * 图画布（T3）——从 devloop/UnitsDAG.vue 抽取并扩展。 @@ -121,11 +133,15 @@ const props = defineProps({ // graph_run_loop_records 原始行（snake_case），徽标数字取同一 edge_id 的最新一条 loopRecords: { type: Array, default: () => [] }, selectedId: { type: String, default: null }, // defNodeId - nextNodeId: { type: String, default: null }, // 下一步会拾取哪个（§四.C：并行 ready 时必须标出来） + // 下一批 advanceBatch 会拾取哪些（defNodeId 数组）——27-P4 T4 之后不再是单数， + // 并行分支下真实会被同一批认领的可能是好几个节点。 + nextNodeIds: { type: Array, default: () => [] }, hardCap: { type: Number, default: 3 }, // 单边回退硬顶（§四.D.2.1） }); defineEmits(['select']); +const nextNodeIdSet = computed(() => new Set((props.nextNodeIds || []).map(String))); + const uid = `wg${++instanceCount}`; /** @@ -159,6 +175,20 @@ watch( ); onBeforeUnmount(() => clearInterval(ticker)); +/** + * fan-in 等待态徽标（27-P4 T10）：defNodeId → {text, title} | undefined。 + * 与 elapsedLabels 同款理由单独拆 computed——它依赖全部节点+全部边而不是单个节点， + * 混进 laidOut 会让布局 computed 背上一份不相关的依赖。 + */ +const joinWaitLabels = computed(() => { + const m = new Map(); + for (const n of props.nodes) { + const label = joinWaitLabel(resolveJoinWait(n, props.nodes, props.edges)); + if (label) m.set(String(n.defNodeId), label); + } + return m; +}); + const NW = 158, NH = 54, GAP_X = 52, GAP_Y = 18, PAD = 20; const LANE_H = 34, BADGE_W = 34; @@ -377,6 +407,7 @@ let instanceCount = 0; .name-label { font-size:12.5px; font-weight:600; font-family: ui-monospace, monospace; pointer-events:none; } .status-label { font-size:10px; fill:#666; text-transform:uppercase; pointer-events:none; } .elapsed-label { font-size:10px; fill:#004085; font-weight:600; font-variant-numeric: tabular-nums; pointer-events:none; } +.join-wait-label { font-size:10px; fill:#8a4b00; font-weight:600; font-variant-numeric: tabular-nums; cursor:help; } .sub-badge { font-size:13px; fill:#5a67d8; pointer-events:none; } .next-badge { font-size:9px; fill:#0069d9; font-weight:700; pointer-events:none; } diff --git a/web/src/components/graph-console/GraphRunPane.vue b/web/src/components/graph-console/GraphRunPane.vue index 20c4dd6..97872b0 100644 --- a/web/src/components/graph-console/GraphRunPane.vue +++ b/web/src/components/graph-console/GraphRunPane.vue @@ -28,7 +28,9 @@ - 下一步将执行 {{ nextNode.name || nextNode.defNodeId }} + 下一步将执行 {{ nextNode.name || nextNode.defNodeId }} + 自动模式下一批最多并发 {{ nextNodes.length }} 个： + {{ n.name || n.defNodeId }}

{{ err }}

@@ -40,7 +42,7 @@ :edges="edges" :loop-records="loopRecords" :selected-id="selectedDefId" - :next-node-id="nextNode ? String(nextNode.defNodeId) : null" + :next-node-ids="nextNodes.map((n) => String(n.defNodeId))" @select="onSelect" > @@ -115,8 +117,19 @@ let refreshTimer = null; let pollTimer = null; const MAX_EVENTS = 400; -/** 引擎按 (created_at, id) 取第一个 ready 节点；getRunState 用同样的排序，取首个 ready 即「下一步会跑哪个」 */ -const nextNode = computed(() => nodes.value.find((n) => n.status === 'ready') || null); +// 27-P4 T4：手动模式的「下一步」（nextStep→advanceOne）仍然只推一个，但自动模式的 +// advanceBatch 一次最多认领 run.max_concurrency 个 ready 节点——「下一步会跑哪个」 +// 不再是单数。引擎按 (created_at, id) 取候选，getRunState 用同样的排序返回 nodes， +// 取前 N 个 ready 即「下一批大概率会认领谁」。 +// ponytail: 这是提示，不是承诺——真实拾取时若跟别的 advanceBatch 撞了车（§4.1 决策）， +// 实际认领的集合可能有出入；纯展示用途，不影响任何控制逻辑，不值得为了精确复刻 +// 后端排序再传一次 max_concurrency 的默认值同步逻辑。 +const DEFAULT_MAX_CONCURRENCY_HINT = 5; // 与 work-graph-db.DEFAULT_MAX_CONCURRENCY 对齐，纯展示兜底 +const nextNodes = computed(() => { + const cap = (run.value && run.value.max_concurrency) || DEFAULT_MAX_CONCURRENCY_HINT; + return nodes.value.filter((n) => n.status === 'ready').slice(0, cap); +}); +const nextNode = computed(() => nextNodes.value[0] || null); // 手动模式「下一步」提示仍用单数 const selectedNode = computed(() => nodes.value.find((n) => String(n.defNodeId) === selectedDefId.value) || null); const canRetry = computed(() => selectedNode.value && ['fail', 'blocked'].includes(selectedNode.value.status)); diff --git a/web/src/components/graph-console/join-status.check.js b/web/src/components/graph-console/join-status.check.js new file mode 100644 index 0000000..ea5228b --- /dev/null +++ b/web/src/components/graph-console/join-status.check.js @@ -0,0 +1,70 @@ +/** + * join-status 的可运行检查（`node web/src/components/graph-console/join-status.check.js`）。 + * 同款手法见 io-model.check.js 的头注——web/ 不在 vitest include 里，用 assert 自跑。 + */ +import assert from 'node:assert/strict'; +import { resolveJoinWait, joinWaitLabel } from './join-status.js'; + +const nodes = [ + { defNodeId: 'entry', name: 'entry', status: 'success' }, + { defNodeId: 'a', name: 'a', status: 'success' }, + { defNodeId: 'b', name: 'b', status: 'running' }, + { defNodeId: 'c', name: 'c', status: 'fail' }, + { defNodeId: 'join', name: 'join', status: 'pending' }, + { defNodeId: 'chain2', name: 'chain2', status: 'pending' }, +]; +const edges = [ + { from: 'entry', to: 'a', condition: 'success', edgeKind: 'forward' }, + { from: 'entry', to: 'b', condition: 'success', edgeKind: 'forward' }, + { from: 'entry', to: 'c', condition: 'success', edgeKind: 'forward' }, + { from: 'a', to: 'join', condition: 'success', edgeKind: 'forward' }, + { from: 'b', to: 'join', condition: 'success', edgeKind: 'forward' }, + { from: 'c', to: 'join', condition: 'success', edgeKind: 'forward' }, + { from: 'a', to: 'chain2', condition: 'success', edgeKind: 'forward' }, + // backward 边不参与等待判定，即使指向 join 也不该被数进 waiting + { from: 'join', to: 'a', condition: 'fail', edgeKind: 'backward' }, +]; + +// ① 真正的 fan-in：join 在等 b(running)/c(fail)，a 已经 success +{ + const wait = resolveJoinWait(nodes.find((n) => n.defNodeId === 'join'), nodes, edges); + assert.equal(wait.total, 3); + assert.equal(wait.done, 1); + assert.equal(wait.isFanIn, true); + assert.equal(wait.satisfied, false); + assert.deepEqual(wait.blockedBy.map((w) => w.defNodeId).sort(), ['b', 'c']); + const label = joinWaitLabel(wait); + assert.equal(label.text, '⏳1/3'); + assert.match(label.title, /a\(success\)/); + assert.match(label.title, /b\(running\)/); + assert.match(label.title, /c\(fail\)/); +} + +// ② 单一入边的普通顺序等待：不是 fan-in，不出徽标（避免给每一条链都加噪声） +{ + const wait = resolveJoinWait(nodes.find((n) => n.defNodeId === 'chain2'), nodes, edges); + assert.equal(wait.total, 1); + assert.equal(wait.isFanIn, false); + assert.equal(joinWaitLabel(wait), null); +} + +// ③ 非 pending 节点：不适用（已经跑过或正在跑，没有「在等谁」这回事） +{ + assert.equal(resolveJoinWait(nodes.find((n) => n.defNodeId === 'a'), nodes, edges), null); +} + +// ④ 已满足（引擎下一拍就会点亮，这一刻不该显示「在等」） +{ + const satisfiedNodes = nodes.map((n) => (n.defNodeId === 'b' || n.defNodeId === 'c' + ? { ...n, status: 'success' } : n)); + const wait = resolveJoinWait(satisfiedNodes.find((n) => n.defNodeId === 'join'), satisfiedNodes, edges); + assert.equal(wait.satisfied, true); + assert.equal(joinWaitLabel(wait), null); +} + +// ⑤ 没有 forward success 入边（图的入口节点）：不适用 +{ + assert.equal(resolveJoinWait({ defNodeId: 'entry', status: 'pending' }, nodes, edges), null); +} + +console.log('join-status.check.js: 全部通过'); diff --git a/web/src/components/graph-console/join-status.js b/web/src/components/graph-console/join-status.js new file mode 100644 index 0000000..eb13f0d --- /dev/null +++ b/web/src/components/graph-console/join-status.js @@ -0,0 +1,67 @@ +/** + * fan-in 等待态（27-P4 T10）：一个 pending 节点在等哪几个上游、各自是什么状态。 + * + * 并行（27-P4 T4/T6）落地之后一张图可能同时有好几个分支在跑，「这个节点为什么还没轮到」 + * 不再是一眼能看出来的事——以前只有一条链，pending 的下一个必然是紧邻上游那一个； + * 现在它可能在等三个并行分支里最慢的那一个。§4.3 决策要求「run 层面要能回答『我卡在等 + * 哪几个节点』」，这里就是那句话的数据源。 + * + * 判据与引擎 `graph-engine.activateSuccessTargets`/`allIncomingSatisfied` 同源： + * 只看 forward 的 success 入边——backward 边是异常流程，不参与「汇合点」这道闸， + * fail 边同理。跟 io-model.js 的 inboundEdges 用的是同一条筛选条件，别改出分叉。 + */ + +/** + * @param {object} node getRunState 的合并节点（要 status 与 defNodeId 两个键） + * @param {Array} nodes 同图全部节点（拿名字与上游状态用） + * @param {Array} edges 同图全部边 + * @returns {null | { + * waiting: Array<{defNodeId, name, status}>, + * total: number, done: number, satisfied: boolean, + * isFanIn: boolean, // 2+ 条 forward success 入边才算真正的 fan-in；单一入边只是普通顺序等待 + * blockedBy: Array<{defNodeId, name, status}>, + * }} + */ +export function resolveJoinWait(node, nodes = [], edges = []) { + if (!node || node.status !== 'pending') return null; + const incoming = (edges || []).filter((e) => ( + String(e.to) === String(node.defNodeId) + && (e.condition || 'success') === 'success' + && (e.edgeKind || 'forward') !== 'backward' + )); + if (!incoming.length) return null; // 没有 forward success 入边：不是在等 fan-in，是别的原因还没轮到 + + const byId = new Map((nodes || []).map((n) => [String(n.defNodeId), n])); + const waiting = incoming.map((e) => { + const src = byId.get(String(e.from)); + return { + defNodeId: e.from, + name: (src && (src.name || src.defNodeId)) || e.from, + status: (src && src.status) || 'pending', + }; + }); + + const done = waiting.filter((w) => w.status === 'success').length; + return { + waiting, + total: waiting.length, + done, + satisfied: done === waiting.length, + isFanIn: waiting.length > 1, + blockedBy: waiting.filter((w) => w.status !== 'success'), + }; +} + +/** + * 画布上的紧凑摘要——克制原则同 `edgeContractLabel`：只占一个小徽标的位置， + * 完整明细留给 title。`satisfied` 时不显示（该节点这一刻就会被点亮，不算「在等」）； + * 非 fan-in（只有一条入边）时也不显示——普通顺序等待，看紧邻上游的状态就够了， + * 加一个「1/1」徽标只会制造噪声。 + */ +export function joinWaitLabel(wait) { + if (!wait || !wait.isFanIn || wait.satisfied) return null; + return { + text: `⏳${wait.done}/${wait.total}`, + title: `在等：${wait.waiting.map((w) => `${w.name}(${w.status})`).join(' · ')}`, + }; +}