node-signals.js
6.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
/**
* 节点信号的共用读法 —— 事件轨过滤 / 事件摘要 / 时长格式化。
*
* 三个消费方（NodeRunLog 的分组、NodeDetailPanel 的失败原因回落与「还活着」条、
* WorkGraphCanvas 的节点秒表）读的是同一份数据，过滤条件与格式化抽在这里一份，
* 免得三处各写一遍再各错一遍。
*
* ⚠️ **SSE 帧里没有 summary**：`graph-run-sse.emitTransition` 把人类可读的 summary
* 只写进 timeline.ndjson，广播给浏览器的 `data:` 只有 `ev.data`。所以前端要看懂一条
* 事件，只能自己从 data 拼句子 —— 这就是 describeEvent 存在的理由。
* （`d.summary` 那条分支留着：引擎哪天把 summary 并进 data，这里自动优先用它。）
*/
/** GraphRunPane 要订阅的事件名全集 —— 引擎 emitEvent / timelineNote 实际发出的那些。
* EventSource 对具名事件必须逐个 addEventListener，漏一个就是**整类事件收不到**：
* 在此之前 node.error（执行器异常的原因）/ node.triage（归因判词）/ node.boundary
* 三类从来没到过浏览器，失败节点因此在界面上「无原因」。 */
export const RUN_EVENT_KINDS = [
'node.status', 'run.status',
'agent.call', 'node.output', 'node.error',
'node.triage', 'node.contract', 'node.boundary', 'node.subgraph',
'loop.registered', 'loop.terminated', 'loop.cross_layer',
'budget.spent', 'budget.extended',
'plan.call', 'plan.done', 'plan.rejected',
];
/** 本节点的事件，保持到达顺序（= stream.log 顺序 = 事件 id 顺序） */
export function nodeEvents(events, node) {
if (!node) return [];
return (events || []).filter((e) => {
const d = e.data || {};
return (d.node_run_id || d.ref) === node.id;
});
}
/** 事件属于哪一次 attempt：事件自带的优先，没带就算在节点当前 attempt 上 */
export function attemptOf(e, node) {
const d = e.data || {};
return d.attempt != null ? d.attempt : ((node && node.attempt) || 0);
}
/** 倒着找第一个满足条件的（「最近一条」的通用写法） */
export function lastWhere(list, fn) {
for (let i = list.length - 1; i >= 0; i -= 1) if (fn(list[i])) return list[i];
return null;
}
function clip(s, n = 160) {
const str = String(s == null ? '' : s).replace(/\s+/g, ' ').trim();
return str.length <= n ? str : `${str.slice(0, n - 1)}…`;
}
/** 一条事件的人类可读摘要（前端自拼，理由见文件头） */
export function describeEvent(e) {
const d = (e && e.data) || {};
if (d.summary) return clip(d.summary);
switch (e && e.kind) {
case 'node.status':
return `状态 ${d.from || '*'} → ${d.to}`;
case 'run.status':
return `run 状态 ${d.from || '*'} → ${d.to}`;
case 'agent.call':
return `调用 ${d.system_agent_id || d.node_name || 'agent'}（attempt #${(d.attempt || 0) + 1}）`;
case 'node.output':
return `产出 ${d.status || '?'}${d.fail_reason ? `（${d.fail_reason}）` : ''}`
+ (d.output_excerpt ? ` · ${clip(d.output_excerpt, 90)}` : '');
case 'node.error':
return `执行器抛异常 ${d.error_type || 'Error'}：${clip(d.error, 110)}`;
case 'node.triage':
return d.degraded
? `归因降级（${d.degraded}）`
: `归因 verdict=${d.verdict || '(缺省)'} fail_reason=${d.fail_reason || 'null'}`;
case 'node.contract':
return `契约 ${d.dirty ? '变了' : '未变'}（evidence=${d.evidence || '?'}，级联重置 ${(d.reset || []).length}）`;
case 'node.boundary':
if (d.violation === 'boundary_escape') {
// **只报个数等于没报**：判罚就是按这份清单下的，而人要判断「这真是 agent 写的吗」
// 唯一的依据就是文件名（并行的开发会话改宿主仓库会被算到当时在跑的节点头上，
// 那是误判，看一眼路径就知道）。以前这里只吐 count，清单躺在 timeline.ndjson 里，
// 界面上翻不到 —— 于是「为什么判我越界」在控制台上无法回答。
// entries 是 porcelain 行（"M path" / "?? path"），引擎已截到 20 条，原样列出。
const list = (d.entries || []).join('、');
return `写入越界：宿主仓库留下 ${d.count || 0} 处改动${list ? ` —— ${clip(list, 240)}` : ''}`;
}
if (d.degraded) return `边界守卫降级（${d.degraded}）`;
return `项目根 ${clip(d.root_path, 90)}`;
case 'node.subgraph':
return `子 run ${d.child_run_id || '?'}（${d.scope === 'node' ? '只重跑被点名的节点' : '整张图'}）`;
case 'loop.registered':
return `回退备案 loop_count=${d.loop_count ?? '?'}`;
case 'loop.terminated':
return `熔断（${d.cap || 'run'} 级硬顶已撞）`;
case 'loop.cross_layer':
return '跨层定点回退';
case 'budget.spent':
return `预算扣减 +${(d.delta && d.delta.agent_calls) || 0} 次调用 / +${(d.delta && d.delta.tokens) || 0} tokens`;
case 'budget.extended':
return `运维追加预算：${Object.entries(d.applied || {}).map(([k, v]) => `${k} +${v}`).join('、') || '?'}`;
case 'plan.call':
return `调 ${d.agent || 'workgraph-planning'} 规划子图（第 ${d.round ?? '?'} 轮）`;
case 'plan.done':
return `子图规划完成：${d.work_graph_id || '?'}`
+ (d.nodes != null ? `（${d.nodes} 节点 / ${d.edges ?? '?'} 边）` : '');
case 'plan.rejected':
return `子图未过写入前校验（第 ${d.round ?? '?'} 轮）：`
+ clip((d.errors || []).map((x) => x.rule || x.detail).join('、') || '未给出规则', 110);
default:
return clip(JSON.stringify(stripNoise(d)), 140);
}
}
/** 每条事件都带的通道字段，摘要里没有信息量 */
export function stripNoise(d) {
const { run_id, root_run_id, node_run_id, ts, ...rest } = d || {};
return rest;
}
/** 毫秒 → `12s` / `3m 07s` / `1h 04m`；不可用时返回 null（由调用方决定显示成什么） */
export function fmtDuration(ms) {
if (!Number.isFinite(ms) || ms < 0) return null;
const s = Math.floor(ms / 1000);
if (s < 60) return `${s}s`;
const m = Math.floor(s / 60);
if (m < 60) return `${m}m ${String(s % 60).padStart(2, '0')}s`;
return `${Math.floor(m / 60)}h ${String(m % 60).padStart(2, '0')}m`;
}
/** ISO 时间戳 → 距 nowTs 过去了多久（同上，不可用返回 null） */
export function since(iso, nowTs) {
if (!iso) return null;
const t = typeof iso === 'number' ? iso : Date.parse(iso);
return Number.isNaN(t) ? null : fmtDuration(nowTs - t);
}