fix(hitl): resume 记录 KV 键不可含冒号(NATS: invalid key)
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 <noreply@anthropic.com>
This commit is contained in:
@@ -2,12 +2,25 @@ package eino
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"regexp"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/sundynix/sundynix-shared/bus"
|
"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 的单测。
|
// memKV 是 CheckpointKV 的内存桩(并发安全),用于不依赖 NATS 的单测。
|
||||||
type memKV struct {
|
type memKV struct {
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
|
|||||||
@@ -366,7 +366,9 @@ type pendingApproval struct {
|
|||||||
Task json.RawMessage `json:"task"`
|
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(审批是单点中断,取首个即可)。
|
// firstInterruptID 取本次中断的首个 interrupt id(审批是单点中断,取首个即可)。
|
||||||
func firstInterruptID(info *compose.InterruptInfo) string {
|
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})
|
rec, _ := json.Marshal(pendingApproval{InterruptID: interruptID, Task: tb})
|
||||||
if err := o.checkpoints.Put(ctx, pendingKey(t.ID), rec); err != nil {
|
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())
|
tr.info("task", "system", "resume 记录落盘失败", err.Error())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user