feat(hitl): 增量3a —— resume 重入闭环（从 checkpoint 续跑审批）

把 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 <noreply@anthropic.com>

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
+	// 仅全新执行注入；resume 时黑板由 checkpoint 还原，重注入会覆盖已积累的执行态。
-	for _, n := range flow.Nodes {
+	if rc == nil {
-		if n.Kind == "memory" {
+		hasMemory := false
-			hasMemory = true
+		for _, n := range flow.Nodes {
-			break
+			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 {
+	// resume 续跑：把人工决定注入 ctx，喂给中断点的审批节点（fresh 则原样 invoke）。
-		// HITL 审批中断：checkpoint 已落、任务停在 waiting（审批 lambda 内已置）→ 上抛哨兵，
+	invokeCtx := ctx
-		// Handle 据此释放 goroutine 而不收尾，等决定到达后 resume（增量3）。
+	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 := ""
+			id := firstInterruptID(info)
-			if len(info.InterruptContexts) > 0 {
+			o.persistResume(ctx, t, id, tr)
-				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，下方按终态收尾
+		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 {
+	if fb.rejected {
-		return b.answer, nil, errRejected // 合法终态，Handle 据此判 rejected 并优雅收尾
+		o.clearResume(ctx, t.ID)           // 拒绝是终态，清 resume 记录 + checkpoint
+		return fb.answer, nil, errRejected // 合法终态，Handle 据此判 rejected 并优雅收尾
 	}
-	if b.fatalErr != nil {
+	if fb.fatalErr != nil {
-		return b.answer, nil, b.fatalErr // 上抛 → Handle 判 failed（带原因）
+		return fb.answer, nil, fb.fatalErr // 上抛 → Handle 判 failed（带原因）
 	}
 
 	// 图里无 agent 节点（纯工具/检索图）也要出一段答复。
-	if b.answer == "" {
+	if fb.answer == "" {
-		o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent")
+		o.runComposeConversation(ctx, t.ID, fb, plan.System, tr, "agent")
-		if b.fatalErr != nil { // 兜底对话也可能触预算顶 / 模型失败
+		if fb.fatalErr != nil { // 兜底对话也可能触预算顶 / 模型失败
-			return b.answer, nil, b.fatalErr
+			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})
+}

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:<query>"，
+// 借此判断审批放行后下游是否真的执行。
+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) {