fix(node): terminal 之后与取消之后不再转发流式工具的尾随 artifact

[wuchulonly]

问题

流式工具(如 katana)通过后台 worker 的 OnResult 回调持续 Emit artifact,这些 goroutine 与工具 Execute() 的返回并不同步。Execute() 返回后 tool node 已发出 terminal tool.result,而后台 worker 仍在为同一个 call 继续发 artifact。

由于 eventbus.Emit 是同步的、WS 写又是单条 FIFO 通道(sendCh),线上顺序 = 发射顺序,于是这些晚到的 artifact 排在 terminal 之后到达控制面。控制面一旦结算了该 call,就把后续每一条 artifact 判为 runner artifact arrived after terminal barrier 拒收 + 记一条 ERROR。

一次长扫描可产生数万条这样的拒收(实测单个 katana 调用一次爬取 34,012 条),足以打满控制面单核 CPU 并刷爆日志,而任务域状态看上去完全正常(工具结果已正确返回引擎)。

还有一条同样通向该拒收的路径:取消停不掉扫描器。katana 的 crawler.Crawl(u) 不接受 ctx(tools/katana/katana.go:204-208 只在 URL 之间检查一次),控制面 300s 超时发 CancelOperation 之后,爬虫会把整轮爬完并继续发 artifact;此时 call 在 node 侧仍然"活跃",而控制面早已 retire 掉它,于是剩下的每一条照样被拒。

修复

核心不变式:一个 call 的 terminal 是它在连线上的最后一条消息;控制面已经放弃的 call 不再往线上发 artifact。

pkg/node/proto_connection.go:

1. 新增 sealed map[string]time.Time(与 operations 共用 operationsMu),记录本连接已关闭 artifact 窗口的 call id;
2. 订阅者转发 artifact 前查 sealed,命中就丢,连 WS 都不发,从发射端掐断尾流;
3. handleAgentToolMessage 在发送 terminal 之前先 seal。配合 sendCh 的 FIFO:seal 之前发的 artifact 必排在 terminal 前(正常转发);
4. seal 排在 AgentRuntime.EmitEvent 之前——runtime 发布用的是订阅者读的同一条总线,agent 模式下 terminal 其实是从 EmitEvent 里上线的,seal 排在它后面等于没排;
5. handleAgentCoreMessage 处理 CancelOperation 时,在 cancel() 之前先 seal,盖住"取消停不掉扫描器"那条路径;
6. sealed 按 1 分钟保留期在每次 seal 时顺带清理,不随连接寿命无界增长。

为什么是墓碑集而不是活跃集:按"不在 operations 里就丢"会连从未被本连接派发过的 call 一并杀掉——agent 模式下节点自身 agent loop 的工具调用,CallID 是模型给的 tc.Id(agent/loop.go:538);standalone scan 用的是 scan-…-call(pkg/runner/runner.go:647)。两者都进不了 operations,它们的 artifact 会被静默丢弃,服务端 forwardAOPFrame → IngestArtifact 再也收不到资产/漏洞。改成只丢明确 seal 过的 id,未知 id 一律放行。

ToolResult 事件不受影响——门控只作用于 artifact 扩展事件;finishOperation 仍在 defer 里 cancel taskCtx 并从 operations 删除。

已知边界:检查与入队之间仍有一个窄窗口(订阅者查完 sealed 释放锁,到真正入 sendCh 之间),极少数情况下可能漏一条尾随记录排到 terminal 之后。把 check+enqueue 放进同一把锁会在 sendCh 满时持锁阻塞,连带卡住 CancelOperation 查表,不划算。堵住整条尾巴是目的,漏网的那条由控制面计数丢弃。

测试

• TestRunToolNodeWireInterop 改成真实形态:工具执行期间从 invocation context 取 CallID 发 artifact(等同 toolargs.Base 的真实链路),而不是在 call 结算之后再补发;末尾新增断言——terminal 之后为同一 call 发的被丢弃,未派发过的 call 的 artifact 仍然送达。
• 新增 TestCancelOperationSealsTheCallArtifactWindow。
• 反向验证:把门控置为恒不丢,新断言会挂在尾随 artifact 抢先到达。
• gofmt / go vet / golangci-lint v2.12.2(全仓,含 re2 tags)干净;go test ./pkg/node/ -count=3 绿。

附带

第二个 commit 修的是 master 就存在、与本 PR 无关的 archtest 失败:websocket_liveness_test.go 没有同名源文件,TestGoTestFilesFollowSourceFiles 一直红。这三个用例测的符号全部定义在 proto_connection.go,并入 proto_connection_test.go 即归位,用例本身一行未改。

控制面侧另有独立改动兜底(把 "after terminal barrier" 从逐条 ERROR + 无界 errors.Join 改为静默计数 + 每 call 汇总一次);本 PR 只治发射端源头。另建议控制面把 state.retired 与 state.terminalQueued 两条拒收路径拆成不同文案/计数——它们目前共用同一条 errArtifactAfterTerminal,分不清"terminal 之后"还是"call 被放弃之后"。

🤖 Generated with Claude Code