From 368938ce893017a92efcc643cfdff42a97eb98b8 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Mon, 29 Jun 2026 12:57:07 +0800 Subject: [PATCH] =?UTF-8?q?feat(hitl):=20=E5=A2=9E=E9=87=8F3a=20=E2=80=94?= =?UTF-8?q?=E2=80=94=20resume=20=E9=87=8D=E5=85=A5=E9=97=AD=E7=8E=AF?= =?UTF-8?q?=EF=BC=88=E4=BB=8E=20checkpoint=20=E7=BB=AD=E8=B7=91=E5=AE=A1?= =?UTF-8?q?=E6=89=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 把 HITL 的恢复半边补上：人工决定到达后从 checkpoint 重入图、喂给审批节点续跑。 至此中断→恢复全闭环在 compose 路径打通（in-process 端到端测试钉死）。 - runComposeGraph 重构为统一入口 execComposeGraph(rc)：rc==nil 全新执行、rc!=nil 从 checkpoint resume，两路共用建图/编译，仅三处分叉——①记忆注入仅 fresh（resume 时黑板由 checkpoint 还原，重注入会覆盖已积累态）；②Invoke ctx（resume 经 ResumeWithData 注入决定）；③终态黑板来源。 - live 捕获：resume 时 compose 用 checkpoint 还原的实例作 local state（非闭包 b）， 故节点 ProcessState 内捕获 live 指针，Invoke 后据此读终态（fresh 仍读 b）。 - resume 记录：中断时把 {interruptID, Task} 落 KV(pending:task_id)，供决定在另一 goroutine/重启进程独立重建任务并续跑；终态清记录 + checkpoint（幂等）。 - ResumeApproval(t, dec) 入口：载记录定位中断点 → execComposeGraph resume。续跑 语义同 fresh：再遇审批→errInterrupted、批准跑完→稿+refs、拒绝→errRejected。 测试：批准（黑板无损还原 + 下游执行 + 状态拨回 running + 清记录/checkpoint）、 拒绝（errRejected + 拒绝语 + 下游不跑 + 清 checkpoint）。这条用例也透过 eino 真实 checkpoint 路径端到端验证了 board 序列化（2a）。全量 go test ./... 绿。 下一步 3b：审批决定改 JetStream 持久投递 + orchestrator 持久消费者触发 ResumeApproval + main 接 Bus.Checkpoints 打开开关。 Co-Authored-By: Claude Opus 4.8 --- .../internal/eino/compose_compiler.go | 178 +++++++++++++++--- .../internal/eino/compose_compiler_test.go | 76 ++++++++ 2 files changed, 225 insertions(+), 29 deletions(-) diff --git a/sundynix-dispatcher/internal/eino/compose_compiler.go b/sundynix-dispatcher/internal/eino/compose_compiler.go index bf28ea1..0613be4 100644 --- a/sundynix-dispatcher/internal/eino/compose_compiler.go +++ b/sundynix-dispatcher/internal/eino/compose_compiler.go @@ -2,6 +2,8 @@ package eino import ( "context" + "encoding/json" + "errors" "fmt" "sync" @@ -38,6 +40,20 @@ func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *e // 返回 (成稿, 检索来源, error)：审批拒绝→errRejected、预算触顶/模型失败→fatalErr， // 与 graph.go 终态严格对齐，否则任务会被误判 done-空。 func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) { + return o.execComposeGraph(ctx, t, tr, nil) +} + +// resumeCtx 标记一次 resume 续跑：interruptID 定位中断点，dec 是要喂给审批节点的人工决定。 +type resumeCtx struct { + interruptID string + dec *contract.ApprovalDecision +} + +// execComposeGraph 是 compose 编排的统一入口：rc==nil 全新执行；rc!=nil 从 checkpoint resume。 +// 两路共用建图/编译，仅在三处分叉：①记忆注入(仅 fresh，resume 时黑板由 checkpoint 还原)； +// ②Invoke 的 ctx(resume 注入人工决定)；③终态黑板来源(resume 时 compose 用 checkpoint 还原的 +// 实例、非本闭包 b → 经 live 捕获读终态)。 +func (o *Orchestrator) execComposeGraph(ctx context.Context, t *contract.Task, tr *execTracer, rc *resumeCtx) (string, []string, error) { registerFlowMerge() flow, ferr := dsl.Parse(t.Graph) plan := dsl.Compile(t.Graph) @@ -76,19 +92,29 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr } // 图里无 memory 节点 → 沿用默认：注入画像+历史（与 graph.go 对齐，避免回归）。 - hasMemory := false - for _, n := range flow.Nodes { - if n.Kind == "memory" { - hasMemory = true - break + // 仅全新执行注入；resume 时黑板由 checkpoint 还原，重注入会覆盖已积累的执行态。 + if rc == nil { + hasMemory := false + for _, n := range flow.Nodes { + if n.Kind == "memory" { + hasMemory = true + break + } + } + if !hasMemory { + b.profile = o.fetchMemory(ctx, b.uid, b.query) + b.history = o.fetchHistory(ctx, b.sid) } } - if !hasMemory { - b.profile = o.fetchMemory(ctx, b.uid, b.query) - b.history = o.fetchHistory(ctx, b.sid) - } - // 建 compose 图：黑板进本地状态（GenLocalState 闭包持有本任务的 b）。 + // 建 compose 图：黑板进本地状态。fresh 用本闭包 b；resume 时 compose 改用 checkpoint 还原的 + // 实例（非 b）——故节点执行时捕获 live 指针，Invoke 后据此读终态（见函数头注释③）。 + var live *board + capture := func(bd *board) { + if live == nil { + live = bd // ProcessState 持状态锁 → 串行无竞态；全程同一黑板指针，设一次即可 + } + } g := compose.NewGraph[flowSignal, flowSignal]( compose.WithGenLocalState(func(context.Context) *board { return b }), ) @@ -106,12 +132,13 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr // 用专用 lambda（须把中断错误作为节点返回值上抛，泛型 lambda 会吞掉它）。 // 未接后端 → 落入下方泛型 lambda 经 execDSLNode 走阻塞式 approvalNode（行为不变）。 if node.Kind == "approval" && o.checkpoints != nil { - _ = g.AddLambdaNode(key(node.ID), compose.InvokableLambda(o.approvalInterruptLambda(t, node, tr))) + _ = g.AddLambdaNode(key(node.ID), compose.InvokableLambda(o.approvalInterruptLambda(t, node, tr, capture))) 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 { + capture(bd) // 记下实际执行的黑板（resume 时为 checkpoint 还原的实例） // 上游审批拒绝 / 预算触顶 / 模型失败 → 跳过下游（对齐 graph.go 的 break 中止语义）。 if bd.rejected || bd.fatalErr != nil { return nil @@ -138,6 +165,7 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr cond := func(c context.Context, _ flowSignal) (map[string]bool, error) { chosen := map[string]bool{} _ = compose.ProcessState(c, func(sc context.Context, bd *board) error { + capture(bd) if bd.rejected || bd.fatalErr != nil { return nil // 已中止 → 不选任何下游，下方收口到 END } @@ -191,36 +219,47 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr 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）。 + // resume 续跑：把人工决定注入 ctx，喂给中断点的审批节点（fresh 则原样 invoke）。 + invokeCtx := ctx + if rc != nil { + invokeCtx = compose.ResumeWithData(ctx, rc.interruptID, rc.dec) + } + if _, ierr := r.Invoke(invokeCtx, flowSignal{}, invokeOpts...); ierr != nil { + // HITL 审批中断：checkpoint 已落、任务停在 waiting（审批 lambda 内已置）→ 落 resume 记录 + // 供决定到达时续跑，上抛哨兵让 Handle 释放 goroutine 不收尾。 if info, ok := compose.ExtractInterruptInfo(ierr); ok { - id := "" - if len(info.InterruptContexts) > 0 { - id = info.InterruptContexts[0].ID - } + id := firstInterruptID(info) + o.persistResume(ctx, t, id, tr) tr.info("task", "approval", "已中断等待审批", "checkpoint 已落，释放执行；interrupt="+id) return b.answer, nil, errInterrupted } - tr.info("task", "system", "compose 执行告警", ierr.Error()) // 非中断的执行告警：副作用已落 board，下方按终态收尾 + tr.info("task", "system", "compose 执行告警", ierr.Error()) // 非中断告警：副作用已落 board，按终态收尾 + } + + // 终态读「实际执行的黑板」：fresh=闭包 b；resume=compose 用 checkpoint 还原的实例(live)。 + fb := live + if fb == nil { + fb = b } // 终态对齐 graph.go：审批拒绝 / 预算触顶 / 模型失败要显式上抛，否则任务误判 done-空。 - if b.rejected { - return b.answer, nil, errRejected // 合法终态，Handle 据此判 rejected 并优雅收尾 + if fb.rejected { + o.clearResume(ctx, t.ID) // 拒绝是终态，清 resume 记录 + checkpoint + return fb.answer, nil, errRejected // 合法终态，Handle 据此判 rejected 并优雅收尾 } - if b.fatalErr != nil { - return b.answer, nil, b.fatalErr // 上抛 → Handle 判 failed（带原因） + if fb.fatalErr != nil { + return fb.answer, nil, fb.fatalErr // 上抛 → Handle 判 failed（带原因） } // 图里无 agent 节点（纯工具/检索图）也要出一段答复。 - if b.answer == "" { - o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent") - if b.fatalErr != nil { // 兜底对话也可能触预算顶 / 模型失败 - return b.answer, nil, b.fatalErr + if fb.answer == "" { + o.runComposeConversation(ctx, t.ID, fb, plan.System, tr, "agent") + if fb.fatalErr != nil { // 兜底对话也可能触预算顶 / 模型失败 + return fb.answer, nil, fb.fatalErr } } - return b.answer, refsOf(b), nil // 成功：带回检索来源供忠实度评测 + o.clearResume(ctx, t.ID) // 成功收尾：清 resume 记录 + checkpoint + return fb.answer, refsOf(fb), nil // 成功：带回检索来源供忠实度评测 } // execDSLNode 执行一个非 branch 的 DSL 节点（compose 编译器用；节点体与 graph.go 一致， @@ -279,12 +318,13 @@ func (o *Orchestrator) execDSLNode(ctx context.Context, t *contract.Task, n dsl. // - resume 流 → GetResumeContext 取人工决定，落黑板（批准放行 / 拒绝置 rejected 中止下游）。 // // 上游已中止（rejected/fatalErr）则直接跳过，不发待审、不中断。 -func (o *Orchestrator) approvalInterruptLambda(t *contract.Task, n dsl.Node, tr *execTracer) func(context.Context, flowSignal) (flowSignal, error) { +func (o *Orchestrator) approvalInterruptLambda(t *contract.Task, n dsl.Node, tr *execTracer, capture func(*board)) 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 { + capture(b) // resume 时这是 checkpoint 还原的黑板，须捕获以读终态 o.applyApprovalDecision(t.ID, n, b, dec, approved, tr) return nil }) @@ -294,6 +334,7 @@ func (o *Orchestrator) approvalInterruptLambda(t *contract.Task, n dsl.Node, tr var title, summary string skip := false _ = compose.ProcessState(ctx, func(_ context.Context, b *board) error { + capture(b) if b.rejected || b.fatalErr != nil { skip = true return nil @@ -310,3 +351,82 @@ func (o *Orchestrator) approvalInterruptLambda(t *contract.Task, n dsl.Node, tr return flowSignal{}, compose.Interrupt(ctx, title) } } + +// ---- HITL resume：中断记录的持久化与续跑 ---- + +// errNoPending 表示无 resume 记录（任务未中断 / 记录已过 TTL 被清）。 +var errNoPending = errors.New("no pending approval record") + +// pendingApproval 是审批中断的 resume 记录：决定到达时据此重建任务并从 checkpoint 续跑。 +// 与 compose checkpoint 同桶分键存（checkpoint 键=task_id，本记录键=pending:task_id）。 +// 存 Task 原文是为了让决定能在另一 goroutine / 重启后的新进程里独立重建任务再 resume。 +type pendingApproval struct { + InterruptID string `json:"interrupt_id"` + Task json.RawMessage `json:"task"` +} + +func pendingKey(taskID string) string { return "pending:" + taskID } + +// firstInterruptID 取本次中断的首个 interrupt id（审批是单点中断，取首个即可）。 +func firstInterruptID(info *compose.InterruptInfo) string { + if info != nil && len(info.InterruptContexts) > 0 { + return info.InterruptContexts[0].ID + } + return "" +} + +// persistResume 落 resume 记录（无 checkpoint 后端则跳过；失败仅告警不阻断——checkpoint 仍在， +// 大不了靠人工/兜底重投）。 +func (o *Orchestrator) persistResume(ctx context.Context, t *contract.Task, interruptID string, tr *execTracer) { + if o.checkpoints == nil { + return + } + tb, err := t.Marshal() + if err != nil { + tr.info("task", "system", "resume 记录序列化失败", err.Error()) + return + } + rec, _ := json.Marshal(pendingApproval{InterruptID: interruptID, Task: tb}) + if err := o.checkpoints.Put(ctx, pendingKey(t.ID), rec); err != nil { + tr.info("task", "system", "resume 记录落盘失败", err.Error()) + } +} + +// loadResume 取 resume 记录（决定消费者据 task_id 续跑用）。无记录返回 errNoPending。 +func (o *Orchestrator) loadResume(ctx context.Context, taskID string) (*pendingApproval, error) { + if o.checkpoints == nil { + return nil, errNoPending + } + data, ok, err := o.checkpoints.Get(ctx, pendingKey(taskID)) + if err != nil { + return nil, err + } + if !ok { + return nil, errNoPending + } + var p pendingApproval + if err := json.Unmarshal(data, &p); err != nil { + return nil, err + } + return &p, nil +} + +// clearResume 终态清理：删 resume 记录 + compose checkpoint（幂等；无后端则跳过）。 +func (o *Orchestrator) clearResume(ctx context.Context, taskID string) { + if o.checkpoints == nil { + return + } + _ = o.checkpoints.Delete(ctx, pendingKey(taskID)) + _ = o.checkpoints.Delete(ctx, taskID) +} + +// ResumeApproval 续跑一个等待审批的任务：据 task_id 取 resume 记录定位中断点，把人工决定喂给 +// 审批节点、从 checkpoint 重入图。供决定消费者调用（增量3b）。返回终态语义同 runComposeGraph： +// 续跑中再遇审批 → 仍返回 errInterrupted（重新落记录）；批准跑完 → 成稿+refs；拒绝 → errRejected。 +func (o *Orchestrator) ResumeApproval(ctx context.Context, t *contract.Task, dec *contract.ApprovalDecision, tr *execTracer) (string, []string, error) { + p, err := o.loadResume(ctx, t.ID) + if err != nil { + return "", nil, err + } + return o.execComposeGraph(ctx, t, tr, &resumeCtx{interruptID: p.InterruptID, dec: dec}) +} diff --git a/sundynix-dispatcher/internal/eino/compose_compiler_test.go b/sundynix-dispatcher/internal/eino/compose_compiler_test.go index 893b847..9a5035e 100644 --- a/sundynix-dispatcher/internal/eino/compose_compiler_test.go +++ b/sundynix-dispatcher/internal/eino/compose_compiler_test.go @@ -115,6 +115,82 @@ func TestComposeApprovalInterruptCheckpoints(t *testing.T) { } } +// approvalGraph 是 input→approval→agent 的最小 HITL 图，agent 用 echoLLM 回 "ANS:"， +// 借此判断审批放行后下游是否真的执行。 +const approvalGraph = `{"version":"1","nodes":[ + {"id":"in","kind":"input","config":{"text":"写上线报告"}}, + {"id":"ap","kind":"approval","config":{"title":"上线审批"}}, + {"id":"a","kind":"agent","config":{"system":"报告助手"}} +],"edges":[ + {"source":"in","target":"ap"},{"source":"ap","target":"a"} +]}` + +// interruptThenSetup 跑一次 fresh，断言中断并落了 resume 记录，返回可继续 resume 的编排器与任务。 +func interruptThenSetup(t *testing.T, kv *memKV, st *fakeStatus) (*Orchestrator, *contract.Task) { + t.Helper() + o := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}, status: st} + o.SetCheckpoints(kv) + task := &contract.Task{ID: "t_resume", Graph: []byte(approvalGraph)} + if _, _, err := o.runComposeGraph(context.Background(), task, &execTracer{}); !errors.Is(err, errInterrupted) { + t.Fatalf("fresh 跑应中断，got %v", err) + } + if _, ok, _ := kv.Get(context.Background(), pendingKey(task.ID)); !ok { + t.Fatalf("中断应落 resume 记录(pending:%s)", task.ID) + } + return o, task +} + +// TestComposeApprovalResumeApprove 钉死 HITL 恢复半边——批准：从 checkpoint 重入图，黑板无损还原， +// 下游 agent 执行出稿，终态清 resume 记录 + checkpoint。这条用例也透过 eino 真实 checkpoint 路径 +// 验证了 board 序列化（增量2a）端到端可用。 +func TestComposeApprovalResumeApprove(t *testing.T) { + kv, st := newMemKV(), &fakeStatus{} + o, task := interruptThenSetup(t, kv, st) + + dec := &contract.ApprovalDecision{TaskID: task.ID, Approved: true, Note: "放行"} + ans, _, err := o.ResumeApproval(context.Background(), task, dec, &execTracer{}) + if err != nil { + t.Fatalf("批准 resume 不应报错: %v", err) + } + if !strings.Contains(ans, "ANS:") { + t.Fatalf("批准后下游 agent 应执行并出稿，got %q", ans) + } + if st.last() != contract.TaskRunning { + t.Fatalf("批准应把状态从 waiting 拨回 running，got %q", st.last()) + } + if _, ok, _ := kv.Get(context.Background(), pendingKey(task.ID)); ok { + t.Fatalf("成功收尾应清 resume 记录") + } + if _, ok, _ := kv.Get(context.Background(), task.ID); ok { + t.Fatalf("成功收尾应清 compose checkpoint") + } +} + +// TestComposeApprovalResumeReject 钉死 HITL 恢复半边——拒绝：resume 置 rejected、返回 errRejected、 +// 出拒绝语、下游 agent 不执行、清 checkpoint。 +func TestComposeApprovalResumeReject(t *testing.T) { + kv, st := newMemKV(), &fakeStatus{} + o, task := interruptThenSetup(t, kv, st) + + dec := &contract.ApprovalDecision{TaskID: task.ID, Approved: false, Note: "不批"} + ans, refs, err := o.ResumeApproval(context.Background(), task, dec, &execTracer{}) + if !errors.Is(err, errRejected) { + t.Fatalf("拒绝 resume 应返回 errRejected，got %v", err) + } + if !strings.Contains(ans, "已被拒绝") { + t.Fatalf("应出拒绝语，got %q", ans) + } + if strings.Contains(ans, "ANS:") { + t.Fatalf("拒绝后下游 agent 不应执行，got %q", ans) + } + if refs != nil { + t.Fatalf("拒绝终态 refs 应为 nil，got %v", refs) + } + if _, ok, _ := kv.Get(context.Background(), task.ID); ok { + t.Fatalf("拒绝收尾应清 compose checkpoint") + } +} + // TestComposeReturnsRefs 钉死 compose 路径回传检索来源——曾被硬写成 nil，导致忠实度评测静默失效。 // 与 runGraph 同图对照：两路径都应回流含检索片段的 refs（喂 grounded judge）。 func TestComposeReturnsRefs(t *testing.T) {