From 0378a770caaf02ef5b31a5be40460ac6bc943d93 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Mon, 29 Jun 2026 14:25:47 +0800 Subject: [PATCH] =?UTF-8?q?fix(hitl):=20resume=20=E8=AE=B0=E5=BD=95=20KV?= =?UTF-8?q?=20=E9=94=AE=E4=B8=8D=E5=8F=AF=E5=90=AB=E5=86=92=E5=8F=B7?= =?UTF-8?q?=EF=BC=88NATS:=20invalid=20key=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit live NATS 联调发现：resume 记录键用 "pending:"+taskID，冒号是 NATS JetStream KV 非法字符（仅允许 [-/_=.a-zA-Z0-9]）→ 中断时 persistResume 的 Put 静默失败、决定 到达时 loadResume 报 "nats: invalid key"，任务永卡 waiting、无法恢复。内存桩接受 任意键，故单测漏过——正是只有 live NATS 才暴露的那类。 - pendingKey: "pending:"+id → "pending_"+id（合法键）。 - persistResume: Put 失败改 log.Printf 大声告警（不止 exec 轨迹），关键失败可见。 - 回归测试 TestPendingKeyIsNATSValid：直接钉键形匹配 NATS KV 字符集，绕开内存桩盲区。 live 验证（devnats + 真链路）：提交 HITL 任务→waiting→杀 dispatcher→离线批准→ 重启→1s waiting→2s running→3s done，deepseek 真实出稿。证明 checkpoint 抗重启 + 决定经 JetStream 抗离线 + 断点恢复。 Co-Authored-By: Claude Opus 4.8 --- .../internal/eino/checkpoint_test.go | 13 +++++++++++++ .../internal/eino/compose_compiler.go | 6 +++++- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/sundynix-dispatcher/internal/eino/checkpoint_test.go b/sundynix-dispatcher/internal/eino/checkpoint_test.go index 2b5d4b2..1866f90 100644 --- a/sundynix-dispatcher/internal/eino/checkpoint_test.go +++ b/sundynix-dispatcher/internal/eino/checkpoint_test.go @@ -2,12 +2,25 @@ package eino import ( "context" + "regexp" "sync" "testing" "github.com/sundynix/sundynix-shared/bus" ) +// natsKVKey 是 NATS JetStream KV 允许的键字符集（首字符不可为 . 或 _）。 +// resume 记录键曾用冒号分隔 → 真 NATS 报 "invalid key"，内存桩测不到，故在此直接钉键形。 +var natsKVKey = regexp.MustCompile(`^[a-zA-Z0-9][-/_=.a-zA-Z0-9]*$`) + +func TestPendingKeyIsNATSValid(t *testing.T) { + for _, id := range []string{"task_4c47de5ce1a29e16", "abc123", "t_0"} { + if k := pendingKey(id); !natsKVKey.MatchString(k) { + t.Fatalf("pendingKey(%q)=%q 不是合法 NATS KV 键（冒号等字符会被拒）", id, k) + } + } +} + // memKV 是 CheckpointKV 的内存桩（并发安全），用于不依赖 NATS 的单测。 type memKV struct { mu sync.Mutex diff --git a/sundynix-dispatcher/internal/eino/compose_compiler.go b/sundynix-dispatcher/internal/eino/compose_compiler.go index e33ec53..86f2e50 100644 --- a/sundynix-dispatcher/internal/eino/compose_compiler.go +++ b/sundynix-dispatcher/internal/eino/compose_compiler.go @@ -366,7 +366,9 @@ type pendingApproval struct { Task json.RawMessage `json:"task"` } -func pendingKey(taskID string) string { return "pending:" + taskID } +// pendingKey 是 resume 记录的 KV 键。注意：NATS JetStream KV 键只允许 [-/_=.a-zA-Z0-9]， +// 冒号等字符会被拒（nats: invalid key）——故用下划线分隔，不可改回冒号。 +func pendingKey(taskID string) string { return "pending_" + taskID } // firstInterruptID 取本次中断的首个 interrupt id（审批是单点中断，取首个即可）。 func firstInterruptID(info *compose.InterruptInfo) string { @@ -389,6 +391,8 @@ func (o *Orchestrator) persistResume(ctx context.Context, t *contract.Task, inte } rec, _ := json.Marshal(pendingApproval{InterruptID: interruptID, Task: tb}) if err := o.checkpoints.Put(ctx, pendingKey(t.ID), rec); err != nil { + // 关键：落盘失败则决定到达时无从 resume，任务永卡 waiting → 大声 log（非仅 exec 轨迹）。 + log.Printf("[eino] resume 记录落盘失败 task=%s: %v（该任务将无法恢复）", t.ID, err) tr.info("task", "system", "resume 记录落盘失败", err.Error()) } }