feat(hitl): 增量3b —— 持久决定投递 + 生产激活(HITL 中断/恢复上线)
把中断/恢复模型在 dispatcher 接线启用,并让审批决定持久化抗离线。至此 HITL 从「阻塞 goroutine 等 5min、core NATS 非持久、不抗重启」升级为「持久化中断 + 决定持久投递 + 从 checkpoint 恢复续跑」。 - contract: 新增审批决定流 StreamApprovals(SUNDYNIX_APPROVALS)/通配 SubjectApprovalAll/消费者 ConsumerApprovals + checkpoint 桶 BucketCheckpoints。 - bus: EnsureApprovalStream(JetStream 流持久捕获 sundynix.approval.>,MaxAge 24h; gateway 现有 nc.Publish 的决定被本流自动捕获,无需改 gateway)+ ConsumeApprovals (持久消费者,队列组多副本安全)。 - orchestrator: HandleApprovalDecision(据 task_id 取 resume 记录续跑;无记录则忽略, 兼容阻塞态任务的决定 + 决定重投幂等)+ finishResumed(收尾对齐 Handle 尾段: 中断/拒绝/预算/失败/成功+评测落历史)。 - main: 开 checkpoint 存储 + 审批流 + 起决定消费者 → SetCheckpoints 启用中断模型; 任一步失败优雅降级回阻塞模型;停机 drain 在途 resume。 决定经 JetStream 持久:dispatcher 在决定发出时离线,重连后仍消费到并续跑;任一 dispatcher 副本都能据共享 KV 的 checkpoint+记录恢复(HA)。 测试:决定驱动恢复(续跑下游+判 done+清记录)、无记录忽略(幂等/兼容)。 全模块 go build + go test 绿。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -408,6 +408,61 @@ func (b *Bus) PublishApproval(dec *contract.ApprovalDecision) error {
|
||||
return b.nc.Publish(contract.ApprovalSubject(dec.TaskID), data)
|
||||
}
|
||||
|
||||
// EnsureApprovalStream 幂等地创建/更新审批决定流,持久捕获 sundynix.approval.>。
|
||||
// 决定带 MaxAge 过期(审批不会拖超一天,避免流无限增长);让中断/恢复模型即便 dispatcher
|
||||
// 在决定发出时离线,重连后仍能消费到决定续跑(core NATS 的 PublishApproval 也被本流捕获)。
|
||||
func (b *Bus) EnsureApprovalStream(ctx context.Context) error {
|
||||
_, err := b.js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
|
||||
Name: contract.StreamApprovals,
|
||||
Subjects: []string{contract.SubjectApprovalAll},
|
||||
Storage: jetstream.FileStorage,
|
||||
MaxAge: 24 * time.Hour,
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
// ConsumeApprovals 在持久消费者上消费审批决定,队列组内负载均衡(多 dispatcher 副本下每条决定
|
||||
// 只一个副本处理 resume)。每条决定派生独立 goroutine 处理(resume 续跑可能秒级),返回 stop 优雅停。
|
||||
func (b *Bus) ConsumeApprovals(ctx context.Context, h func(context.Context, *contract.ApprovalDecision)) (drain func(context.Context), err error) {
|
||||
cons, err := b.js.CreateOrUpdateConsumer(ctx, contract.StreamApprovals, jetstream.ConsumerConfig{
|
||||
Durable: contract.ConsumerApprovals,
|
||||
AckPolicy: jetstream.AckExplicitPolicy,
|
||||
FilterSubject: contract.SubjectApprovalAll,
|
||||
// resume 续跑可能再遇审批 / 出稿(LLM 秒级),AckWait 给足,未 ack 由重投兜底。
|
||||
AckWait: 15 * time.Minute,
|
||||
MaxAckPending: taskConcurrency(),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create approval consumer: %w", err)
|
||||
}
|
||||
var wg sync.WaitGroup
|
||||
cc, err := cons.Consume(func(msg jetstream.Msg) {
|
||||
var dec contract.ApprovalDecision
|
||||
if json.Unmarshal(msg.Data(), &dec) != nil {
|
||||
_ = msg.Term() // 脏数据,丢弃不重投
|
||||
return
|
||||
}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer func() {
|
||||
wg.Done()
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("[bus] approval %s handler panic: %v", dec.TaskID, r)
|
||||
_ = msg.Term()
|
||||
}
|
||||
}()
|
||||
// ctx 派生自 Background:决定处理独立于触发它的请求/信号生命周期(同 ConsumeTasks)。
|
||||
mctx := extractTrace(context.Background(), nats.Header(msg.Headers()))
|
||||
h(mctx, &dec)
|
||||
_ = msg.Ack()
|
||||
}()
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("consume approvals: %w", err)
|
||||
}
|
||||
return func(dctx context.Context) { cc.Stop(); drainWait(&wg, dctx) }, nil
|
||||
}
|
||||
|
||||
// WaitApproval 阻塞等待某任务的人工审批决定,直到收到、ctx 取消或超时。
|
||||
// dispatcher 在审批节点调用:先订阅再等待(订阅早于决定到达,避免错过)。
|
||||
// 超时返回 (nil, error) —— 调用方据安全默认(拒绝)处理。
|
||||
|
||||
@@ -39,8 +39,16 @@ const (
|
||||
SubjectTaskStatus = "sundynix.status.task"
|
||||
|
||||
// 人工审批(HITL)决定回传前缀:实际 sundynix.approval.<task_id>。
|
||||
// core NATS pub-sub:UI 点批准/拒绝 → 网关发到此 → dispatcher 解除审批节点的阻塞。
|
||||
SubjectApproval = "sundynix.approval"
|
||||
// UI 点批准/拒绝 → 网关 nc.Publish 到此。阻塞模型经 core NATS WaitApproval 收;
|
||||
// 中断/恢复模型经下方 JetStream 流持久捕获、由 dispatcher 持久消费者驱动 resume(抗离线)。
|
||||
SubjectApproval = "sundynix.approval"
|
||||
SubjectApprovalAll = "sundynix.approval.>" // 审批决定流捕获的通配
|
||||
StreamApprovals = "SUNDYNIX_APPROVALS" // 审批决定 JetStream 流(持久,决定不因 dispatcher 离线而丢)
|
||||
ConsumerApprovals = "approval-resumers" // 审批决定持久消费者(队列组:多副本下每条决定只一个副本处理 resume)
|
||||
|
||||
// BucketCheckpoints 是 HITL 持久化中断的 JetStream KV 桶名:存 compose 图 checkpoint
|
||||
// (键=task_id)与 resume 记录(键=pending:task_id),dispatcher 重启后可据此恢复在途审批。
|
||||
BucketCheckpoints = "SUNDYNIX_CHECKPOINTS"
|
||||
|
||||
// 自动化评测结果回写:dispatcher 评完经此广播,网关订阅落 PG 并供 UI 查询。core NATS pub-sub。
|
||||
SubjectEval = "sundynix.eval.task"
|
||||
|
||||
Reference in New Issue
Block a user