From bc7600625d45f5bb5f4d59fdff66be3d2a7a228b Mon Sep 17 00:00:00 2001 From: Blizzard Date: Mon, 29 Jun 2026 12:46:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(hitl):=20=E5=A2=9E=E9=87=8F2b=20=E2=80=94?= =?UTF-8?q?=E2=80=94=20=E5=AE=A1=E6=89=B9=E8=8A=82=E7=82=B9=E6=94=B9=20com?= =?UTF-8?q?pose.Interrupt=20+=20orchestrator=20=E8=AF=86=E5=88=AB=E4=B8=AD?= =?UTF-8?q?=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 接了 checkpoint 后端时，审批节点从「阻塞 goroutine 等 5min」改为持久化中断： 首次执行发待审 + 置 waiting + compose.Interrupt → compose 把整图状态（含 board） 落进 checkpoint store 并返回中断错误 → Handle 释放 goroutine、任务停在 waiting， 不收尾不评测不判 done。抗 dispatcher 重启。 - orchestrator: 新增 errInterrupted 哨兵 + checkpoints 字段 + SetCheckpoints setter（沿用 guard/usageSink 的 setter 注入，不动构造签名）；Handle 识别 errInterrupted → 释放 goroutine、保留 waiting、SSE 流不关。 - compose_compiler: 编译挂 WithCheckPointStore + WithGraphName("root")，Invoke 带 WithCheckPointID(task_id)；审批节点接 checkpoint 时改走专用 approvalInterruptLambda（须把中断错误作节点返回值上抛，泛型 lambda 会吞掉）； ExtractInterruptInfo 识别中断 → 上抛 errInterrupted。 - graph.go: 抽出 approvalSummary / applyApprovalDecision，阻塞式与中断式审批共用， 杜绝两路文案/语义漂移。 双路径并存：未接 checkpoint 后端（store=nil）维持阻塞模型，行为不变；main 暂不 接，生产保持阻塞，待增量3 resume 闭环补齐再打开（建在 flag 后）。 测试：中断半边端到端——errInterrupted + 置 waiting + checkpoint 落 KV(key=task_id) + 下游 agent 不执行。既有阻塞式审批/等价测试全绿。 Co-Authored-By: Claude Opus 4.8 --- .../internal/eino/compose_compiler.go | 82 +++++++++++++++++-- .../internal/eino/compose_compiler_test.go | 34 ++++++++ sundynix-dispatcher/internal/eino/graph.go | 53 ++++++++---- .../internal/eino/orchestrator.go | 39 ++++++--- 4 files changed, 176 insertions(+), 32 deletions(-) diff --git a/sundynix-dispatcher/internal/eino/compose_compiler.go b/sundynix-dispatcher/internal/eino/compose_compiler.go index 525ae44..bf28ea1 100644 --- a/sundynix-dispatcher/internal/eino/compose_compiler.go +++ b/sundynix-dispatcher/internal/eino/compose_compiler.go @@ -102,6 +102,13 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr func(context.Context, flowSignal) (flowSignal, error) { return flowSignal{}, nil })) continue } + // 接了 checkpoint 后端 → 审批走中断式（compose.Interrupt 落盘释放 goroutine、抗重启）， + // 用专用 lambda（须把中断错误作为节点返回值上抛，泛型 lambda 会吞掉它）。 + // 未接后端 → 落入下方泛型 lambda 经 execDSLNode 走阻塞式 approvalNode（行为不变）。 + if node.Kind == "approval" && o.checkpoints != nil { + _ = g.AddLambdaNode(key(node.ID), compose.InvokableLambda(o.approvalInterruptLambda(t, node, tr))) + continue + } _ = g.AddLambdaNode(key(node.ID), compose.InvokableLambda( func(c context.Context, _ flowSignal) (flowSignal, error) { perr := compose.ProcessState(c, func(sc context.Context, bd *board) error { @@ -165,14 +172,37 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr } } - // 4) 编译（DAG 模式：无依赖节点并行调度）。编译失败 → 降级回自研 graph.go（安全网）。 - r, cerr := g.Compile(ctx, compose.WithNodeTriggerMode(compose.AllPredecessor)) + // 4) 编译（DAG 模式：无依赖节点并行调度）。接了 checkpoint 后端则挂上，使审批中断可落盘恢复。 + // 编译失败 → 降级回自研 graph.go（安全网）。 + compileOpts := []compose.GraphCompileOption{ + compose.WithNodeTriggerMode(compose.AllPredecessor), + compose.WithGraphName("root"), + } + if o.checkpoints != nil { + compileOpts = append(compileOpts, compose.WithCheckPointStore(newCheckpointStore(o.checkpoints))) + } + r, cerr := g.Compile(ctx, compileOpts...) if cerr != nil { tr.info("task", "system", "compose 编译失败", "退回自研 graph.go："+cerr.Error()) return o.runGraph(ctx, t, tr) // 降级回权威实现（带 refs / 终态） } - if _, ierr := r.Invoke(ctx, flowSignal{}); ierr != nil { - tr.info("task", "system", "compose 执行告警", ierr.Error()) // 副作用已落 board；下方按终态收尾 + // checkpoint id = task id：审批中断时 compose 据此把整图状态（含 board）落进 store。 + var invokeOpts []compose.Option + if o.checkpoints != nil { + invokeOpts = append(invokeOpts, compose.WithCheckPointID(t.ID)) + } + if _, ierr := r.Invoke(ctx, flowSignal{}, invokeOpts...); ierr != nil { + // HITL 审批中断：checkpoint 已落、任务停在 waiting（审批 lambda 内已置）→ 上抛哨兵， + // Handle 据此释放 goroutine 而不收尾，等决定到达后 resume（增量3）。 + if info, ok := compose.ExtractInterruptInfo(ierr); ok { + id := "" + if len(info.InterruptContexts) > 0 { + id = info.InterruptContexts[0].ID + } + tr.info("task", "approval", "已中断等待审批", "checkpoint 已落，释放执行；interrupt="+id) + return b.answer, nil, errInterrupted + } + tr.info("task", "system", "compose 执行告警", ierr.Error()) // 非中断的执行告警：副作用已落 board，下方按终态收尾 } // 终态对齐 graph.go：审批拒绝 / 预算触顶 / 模型失败要显式上抛，否则任务误判 done-空。 @@ -227,8 +257,9 @@ func (o *Orchestrator) execDSLNode(ctx context.Context, t *contract.Task, n dsl. b.refs, b.toolOut = merged, nil tr.info("aggregate:"+n.ID, "system", labelOf(n, "汇聚"), "策略："+firstNonEmpty(cstr(n.Config, "strategy"), "拼接")) case "approval": - // HITL 人工审批中断：阻塞等批准/拒绝，拒绝/超时置 b.rejected（节点入口守卫据此中止下游）。 - // 返回的「放行下游」清单在 compose 由静态边 + rejected 守卫接管，这里丢弃。 + // 阻塞式审批（未接 checkpoint 后端时走这里）：阻塞等批准/拒绝，拒绝/超时置 b.rejected + // （节点入口守卫据此中止下游）。返回的放行清单由静态边 + rejected 守卫接管，丢弃。 + // 接了 checkpoint 后端时，审批已在建图阶段改走 approvalInterruptLambda，不会到这。 o.approvalNode(ctx, t.ID, n, b, tr, nil) case "render": o.renderNode(ctx, t.ID, n, b, tr) @@ -240,3 +271,42 @@ func (o *Orchestrator) execDSLNode(ctx context.Context, t *contract.Task, n dsl. tr.info(n.Kind+":"+n.ID, "system", labelOf(n, n.Kind), "未识别节点，跳过") } } + +// approvalInterruptLambda 是中断式 HITL 审批节点体（compose checkpoint）： +// - 首次执行 → 发待审事件 + 置 waiting + compose.Interrupt：compose 把整图状态（含 board， +// 经 *board 的自定义 JSON 序列化）落进 checkpoint store 并返回中断错误， +// runComposeGraph 据 ExtractInterruptInfo 上抛 errInterrupted、Handle 释放 goroutine； +// - resume 流 → GetResumeContext 取人工决定，落黑板（批准放行 / 拒绝置 rejected 中止下游）。 +// +// 上游已中止（rejected/fatalErr）则直接跳过，不发待审、不中断。 +func (o *Orchestrator) approvalInterruptLambda(t *contract.Task, n dsl.Node, tr *execTracer) func(context.Context, flowSignal) (flowSignal, error) { + return func(ctx context.Context, _ flowSignal) (flowSignal, error) { + // resume 流：应用人工决定。 + if isResume, hasData, dec := compose.GetResumeContext[*contract.ApprovalDecision](ctx); isResume { + approved := hasData && dec != nil && dec.Approved + perr := compose.ProcessState(ctx, func(_ context.Context, b *board) error { + o.applyApprovalDecision(t.ID, n, b, dec, approved, tr) + return nil + }) + return flowSignal{}, perr + } + // 首次执行：读待审摘要；上游已中止则跳过（不发待审、不中断）。 + var title, summary string + skip := false + _ = compose.ProcessState(ctx, func(_ context.Context, b *board) error { + if b.rejected || b.fatalErr != nil { + skip = true + return nil + } + title, summary = approvalSummary(n, b) + return nil + }) + if skip { + return flowSignal{}, nil + } + // 发待审 + 置 waiting，然后中断（compose 落 checkpoint、释放 goroutine）。 + tr.emit("approval:"+n.ID, "approval", "await", title, summary, 0) + o.setStatus(t.ID, contract.TaskWaiting, title) + return flowSignal{}, compose.Interrupt(ctx, title) + } +} diff --git a/sundynix-dispatcher/internal/eino/compose_compiler_test.go b/sundynix-dispatcher/internal/eino/compose_compiler_test.go index 0f93ef1..893b847 100644 --- a/sundynix-dispatcher/internal/eino/compose_compiler_test.go +++ b/sundynix-dispatcher/internal/eino/compose_compiler_test.go @@ -81,6 +81,40 @@ func TestComposeApprovalRejectStopsDownstream(t *testing.T) { } } +// TestComposeApprovalInterruptCheckpoints 钉死中断式 HITL（接了 checkpoint 后端）： +// 审批节点应 compose.Interrupt —— 返回 errInterrupted、置 waiting、checkpoint 落盘、下游 agent 不执行。 +// 这是「释放 goroutine + 抗重启」的中断半边；恢复半边（resume）由增量3 覆盖。 +func TestComposeApprovalInterruptCheckpoints(t *testing.T) { + graph := `{"version":"1","nodes":[ + {"id":"in","kind":"input","config":{"text":"hi"}}, + {"id":"ap","kind":"approval","config":{"title":"上线审批"}}, + {"id":"a","kind":"agent","config":{"system":"机密操作"}} + ],"edges":[ + {"source":"in","target":"ap"},{"source":"ap","target":"a"} + ]}` + kv := newMemKV() + st := &fakeStatus{} + fs := &fakeSink{} + o := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: fs, status: st} + o.SetCheckpoints(kv) // 接 checkpoint 后端 → 审批走中断模型 + + task := &contract.Task{ID: "t_intr", Graph: []byte(graph)} + _, _, err := o.runComposeGraph(context.Background(), task, &execTracer{}) + + if !errors.Is(err, errInterrupted) { + t.Fatalf("审批应中断并返回 errInterrupted，got %v", err) + } + if st.last() != contract.TaskWaiting { + t.Fatalf("中断应置 waiting，got %q", st.last()) + } + if _, ok, _ := kv.Get(context.Background(), task.ID); !ok { + t.Fatalf("compose 应把图状态落进 checkpoint（key=task_id），但 KV 未命中") + } + if strings.Contains(fs.text(), "ANS:") { + t.Fatalf("审批中断后下游 agent 不应执行，但 sink 含 agent 产出: %q", fs.text()) + } +} + // TestComposeReturnsRefs 钉死 compose 路径回传检索来源——曾被硬写成 nil，导致忠实度评测静默失效。 // 与 runGraph 同图对照：两路径都应回流含检索片段的 refs（喂 grounded judge）。 func TestComposeReturnsRefs(t *testing.T) { diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index 585c546..bc553c7 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -425,12 +425,7 @@ func (o *Orchestrator) branchNode(n dsl.Node, b *board, outs []dsl.Edge, byID ma // + 状态 waiting），阻塞等人工批准/拒绝（带超时，安全默认拒绝）。 // 批准 → 状态回 running 并放行下游；拒绝/超时 → 置 b.rejected 中止全图。返回应激活的下游（拒绝=空）。 func (o *Orchestrator) approvalNode(ctx context.Context, taskID string, n dsl.Node, b *board, tr *execTracer, outs []dsl.Edge) []string { - title := firstNonEmpty(cstr(n.Config, "title"), labelOf(n, "人工审批")) - prompt := firstNonEmpty(cstr(n.Config, "prompt"), "请审批是否继续执行后续步骤") - summary := prompt - if b.answer != "" { // 带上当前产出预览，便于审批人判断 - summary = prompt + "\n—— 当前产出预览 ——\n" + truncate(b.answer, 400) - } + title, summary := approvalSummary(n, b) // 未接审批通道（单测/降级）→ 自动放行，避免无人应答卡死。 if o.approval == nil { @@ -443,23 +438,51 @@ func (o *Orchestrator) approvalNode(ctx context.Context, taskID string, n dsl.No o.setStatus(taskID, contract.TaskWaiting, title) dec, err := o.approval.WaitApproval(ctx, taskID, approvalTimeout) - switch { - case err != nil: // 超时 / ctx 取消 → 安全默认拒绝 + if err != nil { // 超时 / ctx 取消 → 安全默认拒绝 b.rejected = true b.answer = "❌ 审批超时未决，已自动拒绝：" + title tr.emit("approval:"+n.ID, "approval", "error", title, "审批超时，自动拒绝", 0) return nil - case !dec.Approved: + } + // 批准/拒绝的落盘与终态事件与中断式审批共用，杜绝两路文案/语义漂移。 + o.applyApprovalDecision(taskID, n, b, dec, dec.Approved, tr) + if b.rejected { + return nil + } + return targetsOf(outs) +} + +// approvalSummary 据审批节点配置 + 当前成稿，算出待审标题与摘要（带产出预览，便于审批人判断）。 +func approvalSummary(n dsl.Node, b *board) (title, summary string) { + title = firstNonEmpty(cstr(n.Config, "title"), labelOf(n, "人工审批")) + prompt := firstNonEmpty(cstr(n.Config, "prompt"), "请审批是否继续执行后续步骤") + summary = prompt + if b.answer != "" { + summary = prompt + "\n—— 当前产出预览 ——\n" + truncate(b.answer, 400) + } + return title, summary +} + +// applyApprovalDecision 把一次人工决定落到黑板并发终态事件：拒绝 → 置 rejected + 拒绝语； +// 批准 → 状态回 running 放行。阻塞式(approvalNode)与中断式(approvalInterruptLambda)审批共用。 +func (o *Orchestrator) applyApprovalDecision(taskID string, n dsl.Node, b *board, dec *contract.ApprovalDecision, approved bool, tr *execTracer) { + title := firstNonEmpty(cstr(n.Config, "title"), labelOf(n, "人工审批")) + if !approved { + note := "审批人拒绝" + if dec != nil { + note = firstNonEmpty(dec.Note, note) + } b.rejected = true - note := firstNonEmpty(dec.Note, "审批人拒绝") b.answer = "❌ 已被拒绝：" + note tr.emit("approval:"+n.ID, "approval", "end", title, "拒绝："+note, 0) - return nil - default: // 批准 → 恢复执行，放行下游 - o.setStatus(taskID, contract.TaskRunning, "审批通过，继续执行") - tr.emit("approval:"+n.ID, "approval", "end", title, "批准："+firstNonEmpty(dec.Note, "放行"), 0) - return targetsOf(outs) + return } + note := "放行" + if dec != nil { + note = firstNonEmpty(dec.Note, note) + } + o.setStatus(taskID, contract.TaskRunning, "审批通过，继续执行") + tr.emit("approval:"+n.ID, "approval", "end", title, "批准："+note, 0) } // targetsOf 取一组边的目标节点 ID（保持顺序）。 diff --git a/sundynix-dispatcher/internal/eino/orchestrator.go b/sundynix-dispatcher/internal/eino/orchestrator.go index 76015f9..851f89a 100644 --- a/sundynix-dispatcher/internal/eino/orchestrator.go +++ b/sundynix-dispatcher/internal/eino/orchestrator.go @@ -65,6 +65,10 @@ var errBudget = errors.New("token 预算超限，已中止") // Handle 据此判 rejected 并优雅收尾（不计熔断失败）。 var errRejected = errors.New("approval rejected") +// errInterrupted 是 HITL 审批中断（compose checkpoint）时图执行返回的哨兵错误：任务停在 +// waiting、checkpoint 已落、goroutine 即可释放；非故障非终态，等人工决定到达后 resume 续跑。 +var errInterrupted = errors.New("approval interrupted (checkpointed)") + // LLM 是编排所需的语言模型能力（生产由 *llm.Pool 实现）。抽成接口便于测试注入假模型。 type LLM interface { Ready() bool @@ -89,17 +93,18 @@ const approvalTimeout = 5 * time.Minute // Orchestrator 把每个 DSL 任务动态编译为 Eino 图并执行（记忆召回 → 工具节点 → 注入 → 流式）。 type Orchestrator struct { - pool LLM - breaker *harness.CircuitBreaker - eval *harness.Evaluator - sink TokenSink - tools ToolCaller - exec ExecSink - status StatusSink // 任务生命周期状态回写（可为 nil） - approval ApprovalWaiter // HITL 审批等待（可为 nil → 审批节点自动放行） - evalSink EvalSink // 评测结果回写落库（可为 nil → 仅打日志） - guard *harness.Classifier // 输入护栏 Tier2：对网关标记的灰区任务做 LLM 裁决（可为 nil → 不做） - usageSink UsageSink // token 用量回写（可为 nil → 不回写） + pool LLM + breaker *harness.CircuitBreaker + eval *harness.Evaluator + sink TokenSink + tools ToolCaller + exec ExecSink + status StatusSink // 任务生命周期状态回写（可为 nil） + approval ApprovalWaiter // HITL 审批等待（可为 nil → 审批节点自动放行） + evalSink EvalSink // 评测结果回写落库（可为 nil → 仅打日志） + guard *harness.Classifier // 输入护栏 Tier2：对网关标记的灰区任务做 LLM 裁决（可为 nil → 不做） + usageSink UsageSink // token 用量回写（可为 nil → 不回写） + checkpoints CheckpointKV // HITL 持久化中断的 checkpoint 后端（可为 nil → 审批走阻塞模型） turnMu sync.Mutex // 保护 turns（攒批计数，多任务 goroutine 共享） turns map[string]int // sessionID → 累计轮次，用于每 N 轮触发 consolidate @@ -118,6 +123,10 @@ func (o *Orchestrator) SetGuardian(c *harness.Classifier) { o.guard = c } // SetUsageSink 注入 token 用量回写出口（可选；不注入则不上报用量）。 func (o *Orchestrator) SetUsageSink(s UsageSink) { o.usageSink = s } +// SetCheckpoints 注入 HITL 持久化中断的 checkpoint 后端（可选）。注入后审批节点改走 +// compose.Interrupt 中断模型（落 checkpoint、释放 goroutine、抗重启）；不注入则维持阻塞模型。 +func (o *Orchestrator) SetCheckpoints(kv CheckpointKV) { o.checkpoints = kv } + // taskBudget 取本任务的 token 预算上限：优先 Meta（网关按用户/套餐下发），否则 env TASK_TOKEN_BUDGET（默认 20 万）。 func (o *Orchestrator) taskBudget(t *contract.Task) int { switch n := t.Meta[contract.MetaTokenBudget].(type) { @@ -254,6 +263,14 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error { // 按 DSL 图执行：compose.Graph（EINO_COMPOSE=1）或自研 graph.go（默认）；agent 节点流式回流 token。 answer, refs, err := o.executeGraph(tctx, t, tr) + if errors.Is(err, errInterrupted) { + // HITL 持久化中断：checkpoint 已落、任务停在 waiting（状态在审批节点内已置）。 + // 释放 goroutine 但不收尾——不 CompleteStream、不评测、不判 done、不计熔断失败； + // SSE 流保持打开，UI 继续显示待审。人工决定到达后由 resume 续跑（增量3）。 + slog.InfoContext(ctx, "task interrupted for approval (checkpointed)", "task_id", t.ID) + o.breaker.Report(true) // 中断是正常暂停，非后端故障 + return nil + } if errors.Is(err, errRejected) { // HITL 拒绝：合法终态，非故障。收尾流 + 置 rejected，不计熔断、不重投。 slog.InfoContext(ctx, "task rejected by approval", "task_id", t.ID)