fix(hitl): 审批复盘轨迹补全 —— 中断不关 exec 流,resume 事件续录
现象:已完成的审批任务,复盘轨迹停在「已中断等待审批」、审批节点永远转圈,看着 像卡住(实际任务已 done、有输出)。 根因:任务中断时 Handle 的 defer tr.done() 给 exec 流发了 CompleteExec → 网关 exec 录制器关闭。resume 是另一次独立调用,其 exec 事件(审批通过/拒绝、续跑节点)发到 同一主题时录制器已关 → 没录进去。(token 流中断时特意没关,所以输出录到了;exec 流却被关了,不对称。) - exec.go: execTracer 加 suspended 标志,done() 挂起时不关流。 - orchestrator.go: 中断分支置 tr.suspended=true(与 token 流一致)。 - ExecTrace.tsx: 前端兜底——运行已完成(phase done)时把悬挂的 waiting/running 节点 收敛为 done,修旧任务(已 done、轨迹缺 resume 事件)的转圈假象。 live 验证:审批任务批准后复盘轨迹完整呈现 approval await→end「批准:放行」→ agent start→推理→end,审批节点不再转圈。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -43,7 +43,13 @@ function StatusDot({ status }: { status: NodeTrace["status"] }) {
|
|||||||
|
|
||||||
// ExecTrace 把执行事件流渲染为竖向轨道:每个节点一颗灯,实时点亮 + 耗时 + 入参/产出。
|
// ExecTrace 把执行事件流渲染为竖向轨道:每个节点一颗灯,实时点亮 + 耗时 + 入参/产出。
|
||||||
export function ExecTrace({ events, phase, compact }: { events: ExecEvent[]; phase?: RunPhase; compact?: boolean }) {
|
export function ExecTrace({ events, phase, compact }: { events: ExecEvent[]; phase?: RunPhase; compact?: boolean }) {
|
||||||
const nodes = deriveNodes(events);
|
const raw = deriveNodes(events);
|
||||||
|
// 运行整体已完成,但轨迹可能缺了中断后(resume)续录的事件(如审批通过、续跑节点),
|
||||||
|
// 导致审批节点停在 waiting 转圈。既然已完成,把悬挂的 waiting/running 收敛为 done,不再假装进行中。
|
||||||
|
const nodes =
|
||||||
|
phase === "done"
|
||||||
|
? raw.map((n) => (n.status === "waiting" || n.status === "running" ? { ...n, status: "done" as const } : n))
|
||||||
|
: raw;
|
||||||
if (nodes.length === 0) {
|
if (nodes.length === 0) {
|
||||||
return (
|
return (
|
||||||
<div className="flex h-full flex-col items-center justify-center gap-1 px-4 py-8 text-center text-xs text-slate-600">
|
<div className="flex h-full flex-col items-center justify-center gap-1 px-4 py-8 text-center text-xs text-slate-600">
|
||||||
|
|||||||
@@ -17,9 +17,10 @@ type ExecSink interface {
|
|||||||
|
|
||||||
// execTracer 为一个任务发结构化执行事件,自增 Seq 保序、span 自动计耗时。
|
// execTracer 为一个任务发结构化执行事件,自增 Seq 保序、span 自动计耗时。
|
||||||
type execTracer struct {
|
type execTracer struct {
|
||||||
sink ExecSink
|
sink ExecSink
|
||||||
task string
|
task string
|
||||||
seq int32
|
seq int32
|
||||||
|
suspended bool // HITL 中断挂起:done() 不关 exec 流,留给 resume 续录(对齐 token 流的中断处理)
|
||||||
}
|
}
|
||||||
|
|
||||||
// tracer 为某任务建一个事件发射器(sink 为空时所有方法变空操作)。
|
// tracer 为某任务建一个事件发射器(sink 为空时所有方法变空操作)。
|
||||||
@@ -60,8 +61,10 @@ func (e *execTracer) span(node, kind, label string) func(detail string, err erro
|
|||||||
}
|
}
|
||||||
|
|
||||||
// done 关闭该任务的执行事件流(让 SSE 客户端收尾)。
|
// done 关闭该任务的执行事件流(让 SSE 客户端收尾)。
|
||||||
|
// 中断挂起时不关:任务停在审批、resume 是另一次调用,过早关流会让录制器错过 resume 的轨迹事件
|
||||||
|
// (审批通过/拒绝、续跑的节点),导致复盘里审批节点永远停在「等待中」。
|
||||||
func (e *execTracer) done() {
|
func (e *execTracer) done() {
|
||||||
if e == nil || e.sink == nil {
|
if e == nil || e.sink == nil || e.suspended {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
_ = e.sink.CompleteExec(e.task)
|
_ = e.sink.CompleteExec(e.task)
|
||||||
|
|||||||
@@ -266,7 +266,8 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
|
|||||||
if errors.Is(err, errInterrupted) {
|
if errors.Is(err, errInterrupted) {
|
||||||
// HITL 持久化中断:checkpoint 已落、任务停在 waiting(状态在审批节点内已置)。
|
// HITL 持久化中断:checkpoint 已落、任务停在 waiting(状态在审批节点内已置)。
|
||||||
// 释放 goroutine 但不收尾——不 CompleteStream、不评测、不判 done、不计熔断失败;
|
// 释放 goroutine 但不收尾——不 CompleteStream、不评测、不判 done、不计熔断失败;
|
||||||
// SSE 流保持打开,UI 继续显示待审。人工决定到达后由 resume 续跑(增量3)。
|
// token 流与 exec 流都保持打开(suspended),resume 时续录,否则复盘里审批节点永停在等待中。
|
||||||
|
tr.suspended = true
|
||||||
slog.InfoContext(ctx, "task interrupted for approval (checkpointed)", "task_id", t.ID)
|
slog.InfoContext(ctx, "task interrupted for approval (checkpointed)", "task_id", t.ID)
|
||||||
o.breaker.Report(true) // 中断是正常暂停,非后端故障
|
o.breaker.Report(true) // 中断是正常暂停,非后端故障
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
Reference in New Issue
Block a user