Documentation
¶
Overview ¶
Package run 提供 worker agent run 命令的 cmd.run lane handler。
Handler 实现了基于持久化 inbox 的异步分发，提供 at-least-once 的崩溃恢复语义。 消息经过校验、持久化后，在后台 goroutine 中分发给 RunCoordinator， 使得 NATS Ack 可以立即返回，不阻塞消息确认。
确认决策：
- 永久错误（payload、route、model 校验失败）→ Term
- inbox 写入失败 → NakWithDelay(5s)，等待重试
- 持久化成功并启动后台 goroutine → Ack
- 重启后通过 RecoverNonTerminal 恢复未完成的任务
Index ¶
- func RequestFromWorkerTask(task runTask) *agentrundomain.RunRequest
- type Config
- type Handler
- func (h *Handler) Close() error
- func (h *Handler) Drain(timeout time.Duration) bool
- func (h *Handler) HandleControlCommand(ctx context.Context, cmd messaging.WorkerCommand) error
- func (h *Handler) HandleRunCommand(ctx context.Context, cmd messaging.WorkerCommand, ...) error
- func (h *Handler) RecoverNonTerminal(ctx context.Context) error
- func (h *Handler) RunInbox() inbox.RunInbox
- func (h *Handler) RunSubject() string
- func (h *Handler) Status(ctx context.Context) messaging.WorkerStatusSnapshot
- func (h *Handler) StopAdmission()
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func RequestFromWorkerTask ¶
func RequestFromWorkerTask(task runTask) *agentrundomain.RunRequest
RequestFromWorkerTask converts the internal runTask into the agent runtime boundary.
Types ¶
type Config ¶
type Config struct {
OrgID uint
WorkerID uint
Env string
MaxConcurrency int
MaxInflight int
DebounceWindow time.Duration
// MaxInteractionWaits 最大并发交互等待数量。
MaxInteractionWaits int
// InteractionWaitTimeout 审批/问题等待默认硬超时。
InteractionWaitTimeout time.Duration
// MaxQueuedCommands limits non-terminal durable inbox records. It protects admission without blocking NATS callbacks.
MaxQueuedCommands int
QueueRetry time.Duration
QueueStartTimeout time.Duration
MaxRunDuration time.Duration
InboxDBPath string // required
}
Config controls a worker run handler.
type Handler ¶
type Handler struct {
// contains filtered or unexported fields
}
Handler receives run commands and dispatches them asynchronously to the RunCoordinator.
func New ¶
New creates a worker run handler backed by the agentrun.Service through a Coordinator. InboxDBPath is required — the handler must not operate without a durable inbox.
func (*Handler) Drain ¶ added in v0.1.19
Drain 等待所有正在执行的后台分发任务（包括恢复 feeder）完成。 返回 true 表示在超时前全部完成，false 表示超时。 必须在 StopAdmission 之后调用，确保没有新的任务再加入。
func (*Handler) HandleControlCommand ¶
HandleControlCommand 处理 control lane 的控制命令（如 cancel）。 控制命令不需要手动确认，handler 同步完成后由 dispatcher 的自动 Ack 机制确认。
func (*Handler) HandleRunCommand ¶
func (h *Handler) HandleRunCommand(ctx context.Context, cmd messaging.WorkerCommand, delivery eventbus.ManualDelivery) error
HandleRunCommand 处理 run 命令，使用 ManualDelivery 手动控制确认时机。
确认决策：
- 永久错误（payload、route、model 校验失败）→ Term，不再重试
- inbox 写入失败 → NakWithDelay(5s)，请求 NATS 延迟重投
- 持久化成功 + 注册 inflight + 启动后台 goroutine → Ack，异步执行
func (*Handler) RecoverNonTerminal ¶ added in v0.1.19
RecoverNonTerminal 加载当前 topic 下所有非终态的 inbox 记录， 将它们注册为 owned（防止重复投递），然后启动后台 feeder goroutine 通过 semaphore 逐个恢复执行。
必须在 NATS 订阅开始前调用，确保崩溃恢复的消息不会被实时消息插队。
func (*Handler) RunSubject ¶
RunSubject returns the NATS subject for this handler's cmd.run lane.
func (*Handler) Status ¶ added in v0.3.9
func (h *Handler) Status(ctx context.Context) messaging.WorkerStatusSnapshot
Status 返回 Worker 本地运行状态快照，供运维状态查询使用。
数据来源：
- running/waiting 任务清单来自 Coordinator 的调度状态；
- command_id、stream_seq、created_at、updated_at 来自持久化 inbox 记录， 通过 run_id 关联补齐。
- admission_waiting 为当前阻塞在准入 semaphore 上的 goroutine 数；
- accepted 为当前拥有的 stream_seq 数（inflight 映射大小）。
摘要只包含定位与生命周期字段，不携带 prompt、模型配置、环境变量或原始命令。
func (*Handler) StopAdmission ¶ added in v0.1.19
func (h *Handler) StopAdmission()
StopAdmission 停止接受新的消息准入。 调用后不会有新的 WaitGroup.Add 发生，用于优雅关闭的第一步。
Source Files
Directories