From c69a1f11af8cb542d12dcbd523ea0881fe6e26a9 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sun, 13 Sep 2026 00:25:13 +0800 Subject: [PATCH] =?UTF-8?q?feat(scheduler):=20M3a+M3b=20=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F=E9=87=8D=E6=9E=84=20+=20?= =?UTF-8?q?=E5=9B=9B=E7=BA=A7=E4=BC=98=E5=85=88=E7=BA=A7=E6=8A=A2=E5=8D=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 设计依据 docs/zh/input-scheduler-design.md §3–§8、§14。 M3a（行为等价的所有权重构）： - processInput 拆为 prepareInputTask / runTaskSteps / finishInputTask， 帧覆盖 prepare→step…→finish；提交与回执只在 finish 段发生一次， 为安全点挂起做准备（挂起不重复提交） - process() 不再持 a.mu（挂起不能持锁），a.mu 字段随之移除 - TaskFrame 增加任务层现场（Evt/CleanInput/IsInterrupt/StartedAt/Terminal/ Level/SeedMsgs）与 taskTerminal / outcomeSuspended - 新增 task_lifecycle_test.go 5 项：正常恰好一次终态、去重 skipped、 on_input 短路、错误终态、consolidation 路由 M3b（优先级与抢占）： - interceptLoop 重写：只做「收中断 → 定级 → requestPreempt → 必要时取消 LLM」，绝不触碰帧（不变量 I2）；三条降级路径与 interceptCh 兜底退场， 改为统一的 pendingInterrupts - scheduler：pendingInterrupts / suspendPool / 让位信号，nextRef 在三集合上 按统一排序键取值；深度上限 4（canSuspend 在安全点拦下） - 抢占判据 incoming.level > running.level；相等与更低只入队 - 安全点只在 step 之间；执行中的 step（工具 RPC/ONNX/CAS）天然不可抢占； _consolidation_ 整任务视为临界区 - 恢复走 resumeTask：从 frame.Step 继续，不重跑 prepare - D1=A：suspend 把被打断任务的只读前缀交给抢占比它的中断任务（SeedMsgs） - 新增 scheduler_preempt_test.go 5 项：抢占-挂起-恢复（含 R1/R5）、同级更低 不抢占、深度上限、空闲中断不丢、seed 路径不污染标志位 验收：agent 全量 + -race 通过；全仓 build/vet 通过 --- internal/agent/core/agent.go | 16 +- internal/agent/core/eventloop.go | 219 ++---------- internal/agent/core/mediaref.go | 4 +- internal/agent/core/scheduler.go | 316 ++++++++++++++++-- internal/agent/core/scheduler_preempt_test.go | 297 ++++++++++++++++ internal/agent/core/task.go | 289 +++++++++++++++- internal/agent/core/task_lifecycle_test.go | 204 +++++++++++ internal/agent/core/task_test.go | 6 + 8 files changed, 1122 insertions(+), 229 deletions(-) create mode 100644 internal/agent/core/scheduler_preempt_test.go create mode 100644 internal/agent/core/task_lifecycle_test.go diff --git a/internal/agent/core/agent.go b/internal/agent/core/agent.go index 288d6d9..fca8d90 100644 --- a/internal/agent/core/agent.go +++ b/internal/agent/core/agent.go @@ -26,8 +26,12 @@ import ( // ContextEvent 和 RelevanceContext 定义在 context.go // Agent — 单 agent，不区分会话/实例 +// +// 并发现状（M3a 起）：所有任务状态只由 **schedulerLoop goroutine** 独占读写， +// 因此不再有保护整轮执行的互斥量——挂起不能持锁（见 docs/zh/input-scheduler-design.md §8.1 I3）。 +// 仍需跨 goroutine 保护的是：childMu/llmMu/lastInputMu/noMergeMu 与各子系统自己的锁； +// interceptLoop 只允许触碰 preemptionRequest 与 cancelLLM（经 llmMu）。 type Agent struct { - mu sync.Mutex id types.AgentID provider agentAPI.Provider providerManager *agentAPI.ProviderManager @@ -113,6 +117,9 @@ type Agent struct { // M2 起取代 eventLoop 的隐式 channel 排队。 sched *scheduler + // 优先级策略表（可空）；见 AgentConfig.PriorityLookup。 + priorityLookup func(source, channel string) Level + // 进行中的 LLM 请求取消函数，interceptLoop 可调用以在请求中打断 cancelLLM context.CancelFunc llmMu sync.Mutex @@ -132,7 +139,7 @@ type Agent struct { // // 需要缓存而不是当场挂到事件上：媒体在 process() 执行期间被捕获， // 而承载它的 ContextEvent 要等 process() 返回后才 Append——此刻还没有 owner_id。 - // 与 pendingMedia 同受 a.mu 保护。 + // 由 schedulerLoop goroutine 独占读写。 pendingMediaDigests []string // 当前输入是否为工具提醒/中断（以 system 角色注入，避免被当成用户消息） @@ -219,6 +226,10 @@ type AgentConfig struct { SkillIndexProvider SkillIndexProvider InputProcessing types.InputProcessingConfig // 非文本输入处理配置 + + // PriorityLookup 是内核的优先级策略表（设计文档 §3.2）。 + // 返回 L1..L4；返回 0 或越界值表示“无策略”，由 Agent 的通道名兜底决定。 + PriorityLookup func(source, channel string) Level } func New(cfg AgentConfig) *Agent { @@ -303,6 +314,7 @@ func New(cfg AgentConfig) *Agent { childTasks: make(map[string]*childTaskState), interceptCh: make(chan *agentIO.InputEvent, 64), sched: newScheduler(256), + priorityLookup: cfg.PriorityLookup, pluginHealth: newPluginHealthTracker(), thinkingEnabled: cfg.ThinkingEnabled, inputCfg: cfg.InputProcessing, diff --git a/internal/agent/core/eventloop.go b/internal/agent/core/eventloop.go index fc84196..082c5a7 100644 --- a/internal/agent/core/eventloop.go +++ b/internal/agent/core/eventloop.go @@ -50,43 +50,12 @@ func (a *Agent) interceptLoop() { clone.Payload["interrupt_source"] = evt.Source clone.Payload["interrupt_channel"] = evt.OutputChannel - a.llmMu.Lock() - hasActiveLLM := a.cancelLLM != nil - if hasActiveLLM { - a.cancelLLM() - log.Printf("[agent] LLM request cancelled by interrupt") - } - a.llmMu.Unlock() - - if hasActiveLLM { - if a.currentOutputChannel == "_consolidation_" { - log.Printf("[agent] consolidation interrupted, re-injecting input for %s/%s", evt.Source, evt.OutputChannel) - a.io.InjectInputTo(evt.Source, evt.OutputChannel, "text", map[string]interface{}{ - "content": text, - "interrupt": true, - "interrupt_source": evt.Source, - "interrupt_channel": evt.OutputChannel, - }) - } else { - select { - case a.interceptCh <- clone: - default: - log.Printf("[agent] intercept channel full, queuing input for %s", evt.Source) - a.io.InjectInputTo(evt.Source, evt.OutputChannel, "text", map[string]interface{}{ - "content": text, - "interrupt": true, - "interrupt_source": evt.Source, - "interrupt_channel": evt.OutputChannel, - }) - } - } - } else { - a.io.InjectInputTo(evt.Source, evt.OutputChannel, "text", map[string]interface{}{ - "content": text, - "interrupt": true, - "interrupt_source": evt.Source, - "interrupt_channel": evt.OutputChannel, - }) + // 决策交给调度器：requestPreempt 总是登记中断（进 pendingInterrupts， + // 因而不会丢），仅当它会真抢占时才告诉我“该取消可取消的步骤”。 + // 本 goroutine 不碰任何帧——只写 pendingInterrupts 与让位信号。 + level := a.taskLevel(evt.Source, evt.OutputChannel) + if a.sched.requestPreempt(clone, level) { + a.cancelCurrentLLM() } case <-a.ctx.Done(): @@ -95,6 +64,20 @@ func (a *Agent) interceptLoop() { } } +// cancelCurrentLLM 取消正在进行的 LLM 请求（若有）。 +// +// 只有 LLM 流式步骤是可取消的；工具 RPC / ONNX / CAS 在 v1 是临界区， +// 取消对它们无效——让位信号会等它们自然结束后的安全点（设计文档 D2）。 +func (a *Agent) cancelCurrentLLM() { + a.llmMu.Lock() + cancel := a.cancelLLM + a.llmMu.Unlock() + if cancel != nil { + cancel() + log.Printf("[agent] LLM request cancelled by preemption") + } +} + // channelConsolidation 标记记忆整理类自输入：无记忆路径处理， // 不写入对话上下文、不向任何输出通道 emit 响应。 const channelConsolidation = "_consolidation_" @@ -108,22 +91,27 @@ type selfInputMsg struct { channel string } -func (a *Agent) handleSelfInput(msg selfInputMsg) { +// selfEvent 把内核自循环消息归一成输入事件。 +func selfEvent(msg selfInputMsg) *agentIO.InputEvent { if msg.channel == "" { msg.channel = channelConsolidation // 兼容空值：默认走整理路径 } - a.processInput(&agentIO.InputEvent{ + return &agentIO.InputEvent{ Source: "system", Type: "text", Payload: map[string]interface{}{"content": msg.text}, OutputChannel: msg.channel, - }) + } +} + +func (a *Agent) handleSelfInput(msg selfInputMsg) { + _, _ = a.runInputTask(selfEvent(msg), nil) } func (a *Agent) handleInput(evt *agentIO.InputEvent) { switch evt.Type { case "text", "image", "audio": - a.processInput(evt) + _, _ = a.runInputTask(evt, nil) case "event": log.Printf("[agent] event from %s: %v", evt.Source, evt.Payload) @@ -283,155 +271,6 @@ func (a *Agent) mediaToBlocks(payload map[string]interface{}, mediaType string, return blocks, alt } -// processInput 是全部模态输入的唯一主干。 -// -// 文本、用户上传的图/音频、插件注入的多模态块走同一条路径，因此去重、 -// no_memory、通道 Cleaner、中断语义、EventRawInput、媒体入 CAS、媒体记忆绑定 -// 对所有模态一致——不会再出现「文本路径加了功能、媒体路径没跟上」。 -func (a *Agent) processInput(evt *agentIO.InputEvent) { - start := time.Now() - - in, ok := a.resolveInput(evt) - if !ok { - return - } - - // 去重按文本做：webui/GUI 断线重连会重放未确认消息。 - // 带媒体时跳过——媒体输入的 alt 文案（"[从 qq 收到了 image]"）对不同图片 - // 是同一句，拿它去重会把连发的两张图误判成重复。 - if len(in.blocks) == 0 && a.isDuplicateInput(evt.Source, in.text) { - log.Printf("[agent] dropped duplicate input from %s: %s", evt.Source, truncateStr(in.text, 60)) - return - } - - a.currentOutputChannel = evt.OutputChannel - if a.currentOutputChannel == "" { - a.currentOutputChannel = evt.Source - } - - if evt.OutputChannel == "_consolidation_" { - a.processConsolidation(evt, in.text) - return - } - - // pendingMedia 让 describe_image / transcribe_audio / ocr_image 拿到本轮媒体的 - // 原始 data/url，也是这三个工具是否出现在工具表里的开关。仅对用户直接上传成立 - //（payload 里才有 data/url）；插件注入的是成品 block，取不到原始数据。 - if evt.Type == "image" || evt.Type == "audio" { - a.pendingMedia = evt.Payload - defer func() { a.pendingMedia = nil }() - } - - // 媒体先落进 CAS。不存的后果是 ContextEvent.Input 只剩一句 alt 文本， - // base64 随 message 数组发给模型后就丢了。 - if len(in.blocks) > 0 { - a.stageMediaDigests(a.captureBlockMedia(in.blocks, in.captureTool)...) - } - - noMemory := false - if v, ok := evt.Payload["no_memory"].(bool); ok { - noMemory = v - } - if !noMemory && a.io != nil { - if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok && chDef.NoMemory { - noMemory = true - } - } - - // 工具提醒/中断（terminal_watch、timer 等）不是用户发言： - // 以 system 角色注入 LLM，且不写入用户对话履历。 - isInterrupt, _ := evt.Payload["interrupt"].(bool) - a.mu.Lock() - a.interruptInput = isInterrupt - a.mu.Unlock() - if isInterrupt { - noMemory = true - } - - stageCtx := a.stageCtxFromInput(in.text, evt.Source, "") - stageCtx.Extra["input_source"] = evt.Source - stageCtx.Extra["output_channel"] = evt.OutputChannel - if len(in.blocks) > 0 { - stageCtx.Extra["media_blocks"] = in.blocks - stageCtx.Extra["media_type"] = in.mediaType - } - if noMemory { - stageCtx.NoMemory = true - } - a.injectSourceContext(stageCtx, evt) - - if a.runStage(sdk.StageOnInput, stageCtx) { - a.emitResponse(evt, *stageCtx.Response) - return - } - - input := stageCtx.RawMessage - - // 计算层用的清洗文本（不改原文）：通道 Cleaner 提取语义内容后用于向量化/提关键词 - cleanInput := input - if a.io != nil { - if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok && chDef.Cleaner != nil { - cleanInput = chDef.Cleaner(input) - } - } - - // upload_* 字段一并转发：webui 的 EventRawInput 订阅方靠它们还原附件卡片。 - // 媒体路径此前把整个 payload 塞进 content（一个 map），订阅方按 string 断言 - // 直接失败 → 用户发的图从不出现在聊天记录里。 - rawPayload := map[string]interface{}{"content": input, "source": evt.Source} - for _, k := range []string{"upload_url", "upload_type", "upload_size", "upload_name"} { - if v, ok := evt.Payload[k]; ok { - rawPayload[k] = v - } - } - a.publishEvent(events.EventRawInput, rawPayload) - - archived := a.pruneOnInput(evt, cleanInput) - if archived > 0 { - log.Printf("[agent] pruned %d low-relevance events to document memory", archived) - } - - if !isInterrupt { - a.context.Append(ContextEvent{ - Timestamp: start, - Source: evt.Source, - Input: input, - }) - } - - response, toolsUsed, toolResults, err := a.process(input, stageCtx) - if err != nil { - log.Printf("[agent] process %s error: %v", evt.Type, err) - resp := fmt.Sprintf("处理错误: %v", err) - a.emitResponse(evt, resp) - a.context.Append(ContextEvent{Timestamp: time.Now(), Source: "agent", Input: input, Response: resp}) - return - } - - elapsed := time.Since(start) - log.Printf("[agent] %s from %s → response (%dms, tools=%v)", evt.Type, evt.Source, elapsed.Milliseconds(), toolsUsed) - - // 本轮捕获的媒体一起挂到这条事件上：用户上传的、插件注入的，以及模型调 - // multimodal_see_picture / see_video 时经 SetToolBlocks 注入的（后者在 - // process() 里被捕获，纯文本输入也会有）。 - turnEvt := ContextEvent{ - Timestamp: time.Now(), - Source: "agent", - Input: cleanInput, - Response: response, - ToolsUsed: toolsUsed, - ToolResults: toolResults, - } - a.bindEventMedia(&turnEvt, a.drainMediaDigests()) - a.context.Append(turnEvt) - - a.emitResponse(evt, response) - - if !stageCtx.NoMemory { - a.emitMemoryCandidate(evt.Source, cleanInput, response, toolResults, toolsUsed) - } -} - func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) { stageCtx := &sdk.StageContext{ FinalText: response, diff --git a/internal/agent/core/mediaref.go b/internal/agent/core/mediaref.go index 74b2ad4..50ec317 100644 --- a/internal/agent/core/mediaref.go +++ b/internal/agent/core/mediaref.go @@ -138,7 +138,7 @@ func (a *Agent) embedMediaOnIngest(digest, mime string, data []byte) { // // 为何要缓存而不是当场建块：媒体在 process() 执行期间被捕获，而承载它的 // ContextEvent 要等 process() 返回后才 Append——此刻还没有 owner_id。 -// 与既有的 a.pendingMedia 同一手法（都在 a.mu 保护下）。 +// 与既有的 a.pendingMedia 同一手法（均由 schedulerLoop goroutine 独占读写）。 func (a *Agent) stageMediaDigests(digests ...string) { if len(digests) == 0 { return @@ -197,7 +197,7 @@ func mediaLabel(it *media.Item) string { // // 沿用 document.Store 的 doc_ 手法（同一份代码库里保持一致， // 也避免为此引入 uuid 依赖）。纳秒精度足够：同一 Agent 的事件由 -// a.mu 串行化 Append，不存在同纳秒两条。 +// schedulerLoop 单 goroutine 串行 Append，不存在同纳秒两条。 func newEventID() string { return fmt.Sprintf("evt_%d", time.Now().UnixNano()) } diff --git a/internal/agent/core/scheduler.go b/internal/agent/core/scheduler.go index 1d0db20..e5d768c 100644 --- a/internal/agent/core/scheduler.go +++ b/internal/agent/core/scheduler.go @@ -23,6 +23,7 @@ import ( "sync" "time" + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" ) @@ -84,6 +85,10 @@ type Task struct { Event *agentIO.InputEvent // Kind == TaskKindInput Self selfInputMsg // Kind == TaskKindSelf + + // SeedMsgs 是抢占式中断任务的只读前缀（D1=A）：由被打断的任务在挂起时 + // 附上，使中断任务看得见「进行到哪一步」，但其产出不合并回原任务。 + SeedMsgs []agentAPI.Message } // SchedulerStats 是调度器的累计计数（可观测性，设计文档 §11 O2）。 @@ -97,9 +102,11 @@ type SchedulerStats struct { // SchedulerSnapshot 是调度器的原子快照。 type SchedulerSnapshot struct { - Running *Task - Queue []*Task - Stats SchedulerStats + Running *Task + Queue []*Task + PendingInterrupts []*Task + SuspendPool []*suspendedTask + Stats SchedulerStats } type scheduler struct { @@ -109,13 +116,41 @@ type scheduler struct { seq uint64 stats SchedulerStats maxQueue int + + // pendingInterrupts：因优先级不足（或运行任务在临界区）而未立即抢占的中断请求。 + // 与 readyQueue 分离：取出时以中断语义启动（设计文档 D3）。 + pendingInterrupts []*Task + // suspendPool：被抢占后保存了现场、等待恢复的任务（**不是栈**，按优先级取）。 + suspendPool []*suspendedTask + // preemptArmed/preemptLevel：运行任务的“让位信号”。 + // interruptLoop 只写这两个字段与 pendingInterrupts；帧永远只由调度器读写。 + preemptArmed bool + preemptLevel Level + // maxSuspendDepth：suspendPool 深度上限（设计文档 §6.3，默认 4）。 + maxSuspendDepth int } +// suspendedTask 是一个被抢占任务的现场。 +type suspendedTask struct { + Task *Task + Frame *TaskFrame +} + +// nextSelection 标识 nextRef 从哪个集合取出任务。 +type nextSelection int + +const ( + nextNone nextSelection = iota + nextReady + nextPending + nextSuspended +) + func newScheduler(maxQueue int) *scheduler { if maxQueue <= 0 { maxQueue = 256 } - return &scheduler{maxQueue: maxQueue} + return &scheduler{maxQueue: maxQueue, maxSuspendDepth: 4} } // hasRoom 报告就绪队列是否还能接收任务。泵入侧据此节流： @@ -145,17 +180,164 @@ func (s *scheduler) enqueue(t *Task) bool { } // next 取出下一个要执行的任务；队列空返回 nil。 +// +// 保留该签名供已有测试使用；调度器自用 nextRef（需要区分是否携带现场）。 func (s *scheduler) next() *Task { + t, _, _ := s.nextRef() + return t +} + +// nextRef 从三个集合中按统一排序键取出下一个任务。 +// +// 设计文档 §4.1：高有效级先；同级先到先服务。挂起任务保留其**原始**入队时刻， +// 因此同级时天然倾向“先把旧任务做完”，抑制饥饿。 +func (s *scheduler) nextRef() (*Task, *TaskFrame, nextSelection) { s.mu.Lock() defer s.mu.Unlock() - if len(s.queue) == 0 { - return nil + + var bestTask *Task + var bestFrame *TaskFrame + bestKind := nextNone + consider := func(t *Task, k nextSelection, fr *TaskFrame) { + if bestTask == nil || taskBefore(t, bestTask) { + bestTask, bestKind, bestFrame = t, k, fr + } } - i := pickTaskIndex(s.queue) - t := s.queue[i] - s.queue = append(s.queue[:i], s.queue[i+1:]...) - s.running = t - return t + for _, t := range s.queue { + consider(t, nextReady, nil) + } + for _, t := range s.pendingInterrupts { + consider(t, nextPending, nil) + } + for _, st := range s.suspendPool { + consider(st.Task, nextSuspended, st.Frame) + } + if bestTask == nil { + return nil, nil, nextNone + } + + switch bestKind { + case nextReady: + s.queue = removeTask(s.queue, bestTask) + case nextPending: + s.pendingInterrupts = removeTask(s.pendingInterrupts, bestTask) + case nextSuspended: + for i, st := range s.suspendPool { + if st.Task == bestTask { + s.suspendPool = append(s.suspendPool[:i], s.suspendPool[i+1:]...) + break + } + } + } + s.running = bestTask + return bestTask, bestFrame, bestKind +} + +func removeTask(list []*Task, target *Task) []*Task { + for i, t := range list { + if t == target { + return append(list[:i], list[i+1:]...) + } + } + return list +} + +// enqueueInterrupt 把一个未立即抢占的中断请求放进 pendingInterrupts。 +// +// 有界：满了丢**最老**的一条并计数（中断是提示性输入，宁可丢旧保新）。 +func (s *scheduler) enqueueInterrupt(t *Task) { + s.mu.Lock() + defer s.mu.Unlock() + s.seq++ + t.ID = s.seq + if t.EnqueuedAt.IsZero() { + t.EnqueuedAt = time.Now() + } + if len(s.pendingInterrupts) >= s.maxQueue { + s.pendingInterrupts = s.pendingInterrupts[1:] + s.stats.Rejected++ + } + s.pendingInterrupts = append(s.pendingInterrupts, t) +} + +// requestPreempt 登记一次中断请求。 +// +// 返回 true 表示“应该尝试取消运行任务正在进行的可取消步骤（LLM 流式）”。 +// +// 无论能否抢占，中断请求都进 pendingInterrupts——这样即使运行任务在抢占生效前 +// 就正常结束，中断也不会丢（它会被 nextRef 按优先级选出）。 +func (s *scheduler) requestPreempt(evt *agentIO.InputEvent, level Level) bool { + s.mu.Lock() + running := s.running + s.mu.Unlock() + + s.enqueueInterrupt(newInterruptTask(evt, level)) + + if running == nil || level <= running.Level { + return false + } + s.mu.Lock() + s.preemptArmed = true + s.preemptLevel = level + s.mu.Unlock() + return true +} + +// preemptGrantedFor 报告级别为 level 的运行任务是否应在当前安全点让位。 +func (s *scheduler) preemptGrantedFor(level Level) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.preemptArmed && s.preemptLevel > level +} + +func (s *scheduler) clearPreempt() { + s.mu.Lock() + s.preemptArmed = false + s.preemptLevel = 0 + s.mu.Unlock() +} + +// suspend 保存现场。 +// +// 深度上限（设计文档 §6.3）：安全点上的 canSuspend 已提前拦下超限情况， +// 此处仅在竞态下兜底计数——绝不丢弃帧（帧丢了会丢副作用记录）。 +func (s *scheduler) suspend(t *Task, f *TaskFrame) { + s.mu.Lock() + defer s.mu.Unlock() + if len(s.suspendPool) >= s.maxSuspendDepth { + s.stats.Rejected++ + } + s.suspendPool = append(s.suspendPool, &suspendedTask{Task: t, Frame: f}) + if s.running == t { + s.running = nil + } + + // D1=A：把被抢占任务的只读前缀交给造成本次抢占的中断任务。 + // 选最高优先级的待处理中断；若它已有前缀（嵌套抢占）则不覆盖。 + if s.preemptArmed { + var victim *Task + for _, it := range s.pendingInterrupts { + if it.Level < s.preemptLevel { + continue + } + if victim == nil || taskBefore(victim, it) { + victim = it + } + } + if victim != nil && len(victim.SeedMsgs) == 0 { + victim.SeedMsgs = append([]agentAPI.Message(nil), f.Msgs...) + } + } + + s.preemptArmed = false + s.preemptLevel = 0 +} + +// canSuspend 报告还有下潜余量（安全点用它决定是否真的让位）。 +func (s *scheduler) canSuspend() bool { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.suspendPool) < s.maxSuspendDepth } // done 标记任务执行结束。 @@ -165,9 +347,59 @@ func (s *scheduler) done(t *Task) { if s.running == t { s.running = nil } + // 任务正常结束：让位信号不再有意义（中断已在 pendingInterrupts 里）。 + s.preemptArmed = false + s.preemptLevel = 0 s.stats.Executed++ } +// currentLevel 返回当前正在执行任务的级别；无 running 时为默认级。 +// +// 用于在 prepare 段把级别写进帧（抢占比较的基准）。 +func (s *scheduler) currentLevel() Level { + s.mu.Lock() + defer s.mu.Unlock() + if s.running != nil { + return s.running.Level + } + return DefaultLevel +} + +// taskLevel 是内核的优先级策略：四级的来源（设计文档 §3.2）。 +// +// 优先走注入的查找函数（配置表）；未命中则用通道名兜底： +// cli/webui/http 为人机交互（L3），system/_consolidation_ 为后台（L1）， +// 其余一律默认级（L1）。显式才是特权：没有策略就不给抢占权。 +func (a *Agent) taskLevel(source, channel string) Level { + if a.priorityLookup != nil { + if l := a.priorityLookup(source, channel); l >= LevelBackground && l <= LevelCritical { + return l + } + } + switch channel { + case "cli", "webui", "http": + return LevelInteractive + case channelConsolidation, "system": + return LevelBackground + } + switch source { + case "cli", "webui": + return LevelInteractive + case "system": + return LevelBackground + } + return DefaultLevel +} + +// inCriticalSection 报告运行任务是否处于不可抢占区。 +// +// M3b 只处理「整个任务不可抢占」的情形（记忆整理）。工具执行、ONNX、 +// CAS 落盘属于**单步**临界区——它们由「只在 step 之间检查让位」天然保护， +// 不需要在这里列（M4 会把清单显式化）。 +func (a *Agent) inCriticalSection() bool { + return a.currentOutputChannel == channelConsolidation +} + // pickTaskIndex 返回下一个要执行的任务下标（设计文档 §4.1 的选择函数）。 // // 排序键：优先级降序 → 入队时刻升序 → ID 升序。 @@ -197,6 +429,11 @@ func newInputTask(evt *agentIO.InputEvent) *Task { return &Task{Kind: TaskKindInput, Level: DefaultLevel, Event: evt, EnqueuedAt: time.Now()} } +// newInterruptTask 把一个中断请求包装成任务。 +func newInterruptTask(evt *agentIO.InputEvent, level Level) *Task { + return &Task{Kind: TaskKindInput, Level: level, Event: evt, EnqueuedAt: time.Now()} +} + func newSelfTask(msg selfInputMsg) *Task { return &Task{Kind: TaskKindSelf, Level: DefaultLevel, Self: msg, EnqueuedAt: time.Now()} } @@ -210,6 +447,8 @@ func (a *Agent) DumpScheduler() SchedulerSnapshot { defer a.sched.mu.Unlock() snap := SchedulerSnapshot{Running: a.sched.running, Stats: a.sched.stats} snap.Queue = append(snap.Queue, a.sched.queue...) + snap.PendingInterrupts = append(snap.PendingInterrupts, a.sched.pendingInterrupts...) + snap.SuspendPool = append(snap.SuspendPool, a.sched.suspendPool...) return snap } @@ -226,9 +465,9 @@ func (a *Agent) schedulerLoop() { for { a.pumpInbox() - t := a.sched.next() - if t == nil { - // 就绪队列空：阻塞等新输入或退出。 + t, f, kind := a.sched.nextRef() + if kind == nextNone { + // 无待办：阻塞等新输入或退出。 select { case evt := <-a.io.InputChan(): a.sched.enqueue(newInputTask(evt)) @@ -239,7 +478,12 @@ func (a *Agent) schedulerLoop() { } continue } - a.executeTask(t) + + if kind == nextSuspended { + a.resumeTask(t, f) + continue + } + a.executeNewTask(t) } } @@ -264,11 +508,19 @@ func (a *Agent) pumpInbox() { } } -// executeTask 执行一个任务，并做**任务级 panic 隔离**（不变量 I6）。 +// executeTask 执行一个任务（测试与旧调用方的入口）；见 executeNewTask。 +func (a *Agent) executeTask(t *Task) { + a.executeNewTask(t) +} + +// executeNewTask 执行一个**新建**任务，并做任务级 panic 隔离（不变量 I6）。 // // 与改造前的差异（有意）：原 eventLoop 在 panic 后重启整个循环， // 现在一个任务的 panic 只丢弃该任务，调度器与其它任务不受影响。 -func (a *Agent) executeTask(t *Task) { +func (a *Agent) executeNewTask(t *Task) { + var f *TaskFrame + var out stepOutcome = outcomeDone + func() { defer func() { if r := recover(); r != nil { @@ -278,10 +530,36 @@ func (a *Agent) executeTask(t *Task) { }() switch t.Kind { case TaskKindInput: - a.handleInput(t.Event) + f, out = a.runInputTask(t.Event, t.SeedMsgs) case TaskKindSelf: - a.handleSelfInput(t.Self) + f, out = a.runInputTask(selfEvent(t.Self), nil) } }() + + if out == outcomeSuspended && f != nil { + a.sched.suspend(t, f) + return + } + a.sched.done(t) +} + +// resumeTask 从保存的现场继续一个被抢占的任务。 +// +// 关键：不重建帧、不重跑 prepare 段——否则会重复提交上下文与事件。 +func (a *Agent) resumeTask(t *Task, f *TaskFrame) { + defer func() { + if r := recover(); r != nil { + log.Printf("[agent] resume task#%d panic recovered: %v\n%s", + t.ID, r, debug.Stack()) + a.sched.done(t) + } + }() + + out := a.runTaskSteps(f) + if out == outcomeSuspended { + a.sched.suspend(t, f) + return + } + a.finishInputTask(f, out) a.sched.done(t) } diff --git a/internal/agent/core/scheduler_preempt_test.go b/internal/agent/core/scheduler_preempt_test.go new file mode 100644 index 0000000..0a514b0 --- /dev/null +++ b/internal/agent/core/scheduler_preempt_test.go @@ -0,0 +1,297 @@ +package core + +// M3b 验收测试：四级优先级 + 严格大于抢占 + 现场保存/恢复 + suspendPool。 +// +// 设计依据 docs/zh/input-scheduler-design.md §11（P1–P4、R1、R5、D1T、Q3）。 +// +// 测试手法：用一个「第一次调用阻塞到 ctx 取消、之后按脚本返回」的 provider， +// 让测试可以确定性地把运行任务停在 S_LLM 上，再注入中断观察让位与恢复。 + +import ( + "context" + "sync" + "testing" + "time" + + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" +) + +// preemptProvider 第 1 次 Chat 阻塞直到 ctx 取消；第 2 次起返回脚本。 +type preemptProvider struct { + mu sync.Mutex + calls int + entered chan struct{} + enteredOn sync.Once + responses []string +} + +func newPreemptProvider(responses ...string) *preemptProvider { + return &preemptProvider{entered: make(chan struct{}), responses: responses} +} + +func (p *preemptProvider) Name() string { return "preempt" } + +func (p *preemptProvider) Chat(ctx context.Context, req *agentAPI.CompletionRequest) (*agentAPI.CompletionResponse, error) { + p.mu.Lock() + p.calls++ + n := p.calls + p.mu.Unlock() + + if n == 1 { + p.enteredOn.Do(func() { close(p.entered) }) + <-ctx.Done() + return nil, ctx.Err() + } + p.mu.Lock() + defer p.mu.Unlock() + i := n - 2 + if i < len(p.responses) { + return &agentAPI.CompletionResponse{Content: p.responses[i]}, nil + } + return &agentAPI.CompletionResponse{Content: "done"}, nil +} + +func (p *preemptProvider) ChatStream(ctx context.Context, req *agentAPI.CompletionRequest) (<-chan agentAPI.StreamChunk, error) { + return nil, context.Canceled +} +func (p *preemptProvider) MaxContextTokens() int { return 8192 } + +func (p *preemptProvider) callCount() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.calls +} + +func newPreemptAgent(t *testing.T, sp agentAPI.Provider) *Agent { + t.Helper() + return New(AgentConfig{ + ID: "preempt", + Provider: sp, + ProviderManager: agentAPI.NewProviderManager(), + IO: agentIO.NewIOManager(), + StageHost: NewStageHost(), + }) +} + +func enqueueTask(t *testing.T, a *Agent, level Level, source, content string) (*Task, *agentIO.InputEvent) { + t.Helper() + evt, _ := textEvent(source, content) + task := &Task{Kind: TaskKindInput, Level: level, Event: evt, EnqueuedAt: time.Now()} + if !a.sched.enqueue(task) { + t.Fatal("入队失败") + } + return task, evt +} + +// P1 + R1 + R5 + D1=A：高优先级抢占 → 挂起在 S_LLM → 中断任务带只读前缀 → +// 恢复后从 S_LLM 重发，且原任务的 msgs 未被改动。 +func TestPreempt_HigherPreemptsAndResumes(t *testing.T) { + sp := newPreemptProvider("intr-done", "low-done") + a := newPreemptAgent(t, sp) + + lowTask, _ := enqueueTask(t, a, LevelBackground, "qq", "低优先级任务") + lt, _, kind := a.sched.nextRef() + if kind != nextReady || lt != lowTask { + t.Fatalf("应取到低优先级任务，kind=%v", kind) + } + + done := make(chan struct{}) + go func() { a.executeNewTask(lt); close(done) }() + + select { + case <-sp.entered: + case <-time.After(3 * time.Second): + t.Fatal("provider 第 1 次调用未发生") + } + + // 注入 L4 中断（cli） + intrEvt, _ := textEvent("cli", "紧急打断") + intrEvt.Payload["interrupt"] = true + if !a.sched.requestPreempt(intrEvt, LevelCritical) { + t.Fatal("L4 应请求抢占并返回 true（应取消 LLM）") + } + a.cancelCurrentLLM() + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("低优先级任务未在取消后挂起") + } + + snap := a.DumpScheduler() + if snap.Running != nil { + t.Fatal("挂起后不应还有 running") + } + if len(snap.SuspendPool) != 1 { + t.Fatalf("suspendPool=%d，期望 1", len(snap.SuspendPool)) + } + if len(snap.PendingInterrupts) != 1 { + t.Fatalf("pendingInterrupts=%d，期望 1", len(snap.PendingInterrupts)) + } + sf := snap.SuspendPool[0].Frame + if sf.Terminal != terminalSuspended { + t.Fatalf("挂起任务终态=%v，期望 terminalSuspended", sf.Terminal) + } + if sf.Step != StepLLM { + t.Fatalf("应在 StepLLM 安全点挂起，实际 step=%v", sf.Step) + } + msgsBefore := len(sf.Msgs) + + // R5：第一次 LLM 调用被丢弃，未产生新消息。 + if sp.callCount() != 1 { + t.Fatalf("挂起前 LLM 调用=%d，期望 1（不完整请求被丢弃）", sp.callCount()) + } + + // Q3：三集合统一比较 → 下一轮取中断（L4 > L1）。 + it, _, k := a.sched.nextRef() + if k != nextPending || it.Level != LevelCritical { + t.Fatalf("应取到 pending 中断，kind=%v level=%v", k, it.Level) + } + // D1=A：抢占式中断任务继承被打断任务的只读前缀。 + if len(it.SeedMsgs) == 0 { + t.Fatal("抢占式中断任务必须继承只读前缀（D1=A）") + } + a.executeNewTask(it) + if len(a.DumpScheduler().PendingInterrupts) != 0 { + t.Fatal("中断任务执行后 pendingInterrupts 应清空") + } + + // R1：恢复被抢占任务；msgs 与被抢占前逐字节一致（长度不变），并从 S_LLM 重发。 + rt, rf, k2 := a.sched.nextRef() + if k2 != nextSuspended || rt != lowTask { + t.Fatalf("应恢复被抢占任务，kind=%v", k2) + } + if rf.Step != StepLLM { + t.Fatalf("恢复游标=%v，期望 StepLLM", rf.Step) + } + if len(rf.Msgs) != msgsBefore { + t.Fatalf("恢复后 msgs 长度=%d，期望 %d（不得被中断污染）", len(rf.Msgs), msgsBefore) + } + a.resumeTask(rt, rf) + + if sp.callCount() != 3 { + t.Fatalf("LLM 总调用=%d，期望 3（丢弃 1 + 中断 1 + 恢复 1）", sp.callCount()) + } + snap = a.DumpScheduler() + if len(snap.SuspendPool) != 0 || snap.Running != nil { + t.Fatalf("全部结束后应无挂起与运行任务：%+v", snap) + } + if snap.Stats.Executed != 2 { + t.Fatalf("Executed=%d，期望 2（中断任务 + 被抢占任务）", snap.Stats.Executed) + } +} + +// P2/P3：同级与更低级都不得抢占，请求进 pendingInterrupts。 +func TestPreempt_LowerOrEqualDoesNotPreempt(t *testing.T) { + sp := newPreemptProvider("low-done", "intr-done") + a := newPreemptAgent(t, sp) + + lowTask, _ := enqueueTask(t, a, LevelInteractive, "cli", "运行中的 L3") + if _, _, kind := a.sched.nextRef(); kind != nextReady { + t.Fatal("应取到运行任务") + } + + done := make(chan struct{}) + go func() { a.executeNewTask(lowTask); close(done) }() + select { + case <-sp.entered: + case <-time.After(3 * time.Second): + t.Fatal("provider 未被调用") + } + + // 同级 L3 + e1, _ := textEvent("webui", "同级打断") + if a.sched.requestPreempt(e1, LevelInteractive) { + t.Fatal("同级不得抢占") + } + // 更低级 L1 + e2, _ := textEvent("system", "低优先级打断") + if a.sched.requestPreempt(e2, LevelBackground) { + t.Fatal("更低级不得抢占") + } + if a.sched.preemptGrantedFor(LevelInteractive) { + t.Fatal("未 arm 让位信号，preemptGrantedFor 应为 false") + } + + // 运行任务没有被取消，仍在等它自己的 ctx；用取消让它收尾，便于清理。 + a.cancelCurrentLLM() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("运行任务未结束") + } + if len(a.DumpScheduler().PendingInterrupts) != 2 { + t.Fatalf("两条未抢占中断都应保留在 pendingInterrupts，实际 %d", + len(a.DumpScheduler().PendingInterrupts)) + } +} + +// D1T：suspendPool 满时不再下潜（canSuspend=false），让位信号也不会 arm。 +func TestPreempt_DepthCapBlocksSuspension(t *testing.T) { + a := newPreemptAgent(t, newPreemptProvider()) + + if a.sched.maxSuspendDepth != 4 { + t.Fatalf("默认深度上限=%d，期望 4", a.sched.maxSuspendDepth) + } + frame := func() *TaskFrame { return a.newTaskFrame("x", a.stageCtxFromInput("x", "", "")) } + for i := 0; i < a.sched.maxSuspendDepth; i++ { + a.sched.suspend(&Task{ID: uint64(i + 1), Level: LevelBackground}, frame()) + } + if a.sched.canSuspend() { + t.Fatal("深度已达上限，canSuspend 应为 false") + } + // 超限兜底：仍保留帧（不丢副作用记录），但计数 Rejected。 + before := a.DumpScheduler().Stats.Rejected + a.sched.suspend(&Task{ID: 99, Level: LevelBackground}, frame()) + if a.DumpScheduler().Stats.Rejected != before+1 { + t.Fatal("超限挂起必须计数 Rejected") + } + if len(a.DumpScheduler().SuspendPool) != a.sched.maxSuspendDepth+1 { + t.Fatal("兜底路径必须保留帧而不是丢弃") + } +} + +// 中断不丢：空闲时请求抢占 → 不 arm 信号，但请求进 pendingInterrupts 并被选出。 +func TestPreempt_IdleInterruptIsQueuedNotLost(t *testing.T) { + a := newPreemptAgent(t, &scriptProvider{script: []*agentAPI.CompletionResponse{{Content: "已处理中断"}}}) + + evt, respCh := textEvent("qq", "空闲时的中断") + if a.sched.requestPreempt(evt, LevelMessage) { + t.Fatal("空闲时不应请求取消 LLM（没有运行任务）") + } + if len(a.DumpScheduler().PendingInterrupts) != 1 { + t.Fatal("空闲时的中断必须进 pendingInterrupts") + } + task, _, kind := a.sched.nextRef() + if kind != nextPending || task.Level != LevelMessage { + t.Fatalf("应取到待处理中断，kind=%v", kind) + } + a.executeNewTask(task) + if len(respCh) != 1 { + t.Fatal("中断任务应完成并回执") + } +} + +// 带媒体/中断标记的输入走新任务路径时不得污染下一个任务的尾部消息。 +func TestPreempt_SeedPathDoesNotLeakInterruptFlag(t *testing.T) { + sp := &scriptProvider{script: []*agentAPI.CompletionResponse{{Content: "ok"}}} + a := newPreemptAgent(t, sp) + + f := a.newTaskFrame("打断文本", a.stageCtxFromInput("打断文本", "", "")) + f.SeedMsgs = []agentAPI.Message{{Role: "system", Content: "S"}, {Role: "user", Content: "U"}} + a.interruptInput = true + if out := a.stepPrepare(f); out != outcomeContinue { + t.Fatalf("seed 路径应继续，实际 %v", out) + } + if a.interruptInput { + t.Fatal("seed 路径必须消费 interruptInput，否则下一个任务尾部会被误改") + } + last := f.Msgs[len(f.Msgs)-1] + if last.Role != "user" || last.Content != "打断文本" { + t.Fatalf("seed 路径尾部应为中断输入本身，实际 %+v", last) + } + if f.Step != StepLLM { + t.Fatalf("seed 路径应直接进入 StepLLM，实际 %v", f.Step) + } +} diff --git a/internal/agent/core/task.go b/internal/agent/core/task.go index ec4c40d..4ae32a8 100644 --- a/internal/agent/core/task.go +++ b/internal/agent/core/task.go @@ -28,6 +28,7 @@ import ( "time" agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" "gitcode.com/JianFeeeee/HomeAgent/internal/events" sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk" @@ -61,6 +62,28 @@ const ( outcomeDone // outcomeFailed 任务失败结束，错误在 frame.Err。 outcomeFailed + // outcomeSuspended 任务在安全点被抢占挂起，帧已保存（M3b 起使用）。 + outcomeSuspended +) + +// taskTerminal 是任务的终态种类（设计文档 §7：每个任务恰有一个终态）。 +type taskTerminal int + +const ( + // terminalNone 任务尚未结束，需进入 run 段。 + terminalNone taskTerminal = iota + // terminalOK 正常完成（已提交上下文并回执）。 + terminalOK + // terminalError 执行出错（已提交错误响应）。 + terminalError + // terminalStageShortCircuit 被 on_input 阶段短路（响应已发出）。 + terminalStageShortCircuit + // terminalSkipped 未进入执行：解析失败或被去重。 + terminalSkipped + // terminalConsolidation 走记忆整理专用路径，已处理完毕。 + terminalConsolidation + // terminalSuspended 被抢占挂起，等待恢复（M3b 起使用）。 + terminalSuspended ) // TaskFrame 承载一个任务在安全点之间必须存活的所有状态。 @@ -92,37 +115,259 @@ type TaskFrame struct { Step Step Response string Err error + + // ---- 任务层现场（原 processInput 的局部变量）---- + // + // 这些字段让帧覆盖 prepare → step… → finish 全生命周期：挂起发生在 run 段的 + // 安全点，恢复后由 finish 段统一提交（context.Append + emitResponse + + // emitMemoryCandidate），因此挂起不会重复提交。 + Evt *agentIO.InputEvent + CleanInput string + IsInterrupt bool + StartedAt time.Time + Terminal taskTerminal + Level Level + PreemptCount int + + // SeedMsgs 非空时，stepPrepare 不重建 system prompt / 记忆上下文， + // 而是以它为前缀继续（D1=A：抢占式中断任务继承被打断任务的**只读前缀**）。 + SeedMsgs []agentAPI.Message } func (a *Agent) newTaskFrame(input string, stageCtx *sdk.StageContext) *TaskFrame { return &TaskFrame{Input: input, StageCtx: stageCtx, Step: StepPrepare} } -// process 驱动状态机直到任务结束，返回与原实现完全相同的四元组。 +// runTaskSteps 驱动状态机直到任务结束或被抢占挂起。 // -// 保留该签名是为了让 M1 成为纯内部重构：所有调用方（processInput / -// processConsolidation / 测试）无需改动。 -func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response string, toolsUsed []string, toolResults []ToolResultItem, err error) { - a.mu.Lock() - defer a.mu.Unlock() - - if a.provider == nil { - return "", nil, nil, fmt.Errorf("agent: no LLM provider configured") - } - - f := a.newTaskFrame(input, stageCtx) +// 这是 M1 的驱动循环，M3a 从 process() 抽出来，使调用方可以拿到 +// outcomeSuspended 并把帧留给调度器保存。 +func (a *Agent) runTaskSteps(f *TaskFrame) stepOutcome { // 步数上限只是防"转移缺失导致死循环"的护栏；正常任务远达不到。 const maxSteps = 1 << 20 for i := 0; i < maxSteps; i++ { + // 安全点：只在 step 之间检查让位。临界区（StepToolExec）不在此列， + // 因为让位信号由 interruptLoop 置位、而本循环是唯一读帧者。 + if !a.inCriticalSection() && a.sched.preemptGrantedFor(f.Level) && a.sched.canSuspend() { + return outcomeSuspended + } switch a.step(f) { case outcomeDone: - return f.Response, f.ToolsUsed, f.ToolResults, nil + return outcomeDone case outcomeFailed: - return "", f.ToolsUsed, f.ToolResults, f.Err + return outcomeFailed + case outcomeSuspended: + return outcomeSuspended } } - return "", f.ToolsUsed, f.ToolResults, - fmt.Errorf("agent: task step budget exhausted（状态机未收敛，疑似转移缺失）") + f.Err = fmt.Errorf("agent: task step budget exhausted（状态机未收敛，疑似转移缺失）") + return outcomeFailed +} + +// process 是保留给 processConsolidation 与测试的薄壳，返回与原实现相同的四元组。 +// +// 注意：M3a 起**不再持 a.mu**——调度器是唯一执行者，而挂起不能持锁。 +func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response string, toolsUsed []string, toolResults []ToolResultItem, err error) { + if a.provider == nil { + return "", nil, nil, fmt.Errorf("agent: no LLM provider configured") + } + f := a.newTaskFrame(input, stageCtx) + switch a.runTaskSteps(f) { + case outcomeDone: + return f.Response, f.ToolsUsed, f.ToolResults, nil + case outcomeFailed: + return "", f.ToolsUsed, f.ToolResults, f.Err + default: + // 不该发生：process() 不参与挂起（只有 runInputTask 会）。 + return "", f.ToolsUsed, f.ToolResults, + fmt.Errorf("agent: task suspended outside scheduler") + } +} + +// runInputTask 是一个输入任务的完整生命周期：prepare → run → finish。 +// +// 它是原 processInput 的全部职责，被拆成三段而不是一个大函数，目的只有一个： +// 让帧可以跨安全点被挂起——挂起后由调度器保存，恢复时接着 run 段继续， +// 而 finish 段（上下文提交与回执）只在任务真正结束时执行一次。 +// +// M3a 还没有抢占，因此 outcomeSuspended 只会由 M3b 的抢占检查产生。 +func (a *Agent) runInputTask(evt *agentIO.InputEvent, seed []agentAPI.Message) (*TaskFrame, stepOutcome) { + f, term := a.prepareInputTask(evt) + switch term { + case terminalSkipped, terminalStageShortCircuit, terminalConsolidation: + return nil, outcomeDone + } + f.SeedMsgs = seed + + out := a.runTaskSteps(f) + if out == outcomeSuspended { + f.Terminal = terminalSuspended + return f, outcomeSuspended + } + a.finishInputTask(f, out) + return f, out +} + +// prepareInputTask 执行 processInput 的前半段（去重、通道解析、阶段、裁剪、 +// 输入事件落上下文）。返回终态不为 terminalNone 时调用方不得进入 run 段。 +func (a *Agent) prepareInputTask(evt *agentIO.InputEvent) (*TaskFrame, taskTerminal) { + start := time.Now() + + in, ok := a.resolveInput(evt) + if !ok { + return nil, terminalSkipped + } + + // 去重按文本做：webui/GUI 断线重连会重放未确认消息。 + // 带媒体时跳过——媒体输入的 alt 文案（"[从 qq 收到了 image]"）对不同图片 + // 是同一句，拿它去重会把连发的两张图误判成重复。 + if len(in.blocks) == 0 && a.isDuplicateInput(evt.Source, in.text) { + log.Printf("[agent] dropped duplicate input from %s: %s", evt.Source, truncateStr(in.text, 60)) + return nil, terminalSkipped + } + + a.currentOutputChannel = evt.OutputChannel + if a.currentOutputChannel == "" { + a.currentOutputChannel = evt.Source + } + + if evt.OutputChannel == channelConsolidation { + a.processConsolidation(evt, in.text) + return nil, terminalConsolidation + } + + // pendingMedia 让 describe_image / transcribe_audio / ocr_image 拿到本轮媒体的 + // 原始 data/url，也是这三个工具是否出现在工具表里的开关。仅对用户直接上传成立 + //（payload 里才有 data/url）；插件注入的是成品 block，取不到原始数据。 + if evt.Type == "image" || evt.Type == "audio" { + a.pendingMedia = evt.Payload + } + + // 媒体先落进 CAS。不存的后果是 ContextEvent.Input 只剩一句 alt 文本， + // base64 随 message 数组发给模型后就丢了。 + if len(in.blocks) > 0 { + a.stageMediaDigests(a.captureBlockMedia(in.blocks, in.captureTool)...) + } + + noMemory := false + if v, ok := evt.Payload["no_memory"].(bool); ok { + noMemory = v + } + if !noMemory && a.io != nil { + if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok && chDef.NoMemory { + noMemory = true + } + } + + // 工具提醒/中断（terminal_watch、timer 等）不是用户发言： + // 以 system 角色注入 LLM，且不写入用户对话履历。 + isInterrupt, _ := evt.Payload["interrupt"].(bool) + a.interruptInput = isInterrupt + if isInterrupt { + noMemory = true + } + + stageCtx := a.stageCtxFromInput(in.text, evt.Source, "") + stageCtx.Extra["input_source"] = evt.Source + stageCtx.Extra["output_channel"] = evt.OutputChannel + if len(in.blocks) > 0 { + stageCtx.Extra["media_blocks"] = in.blocks + stageCtx.Extra["media_type"] = in.mediaType + } + if noMemory { + stageCtx.NoMemory = true + } + a.injectSourceContext(stageCtx, evt) + + if a.runStage(sdk.StageOnInput, stageCtx) { + a.emitResponse(evt, *stageCtx.Response) + return nil, terminalStageShortCircuit + } + + input := stageCtx.RawMessage + + // 计算层用的清洗文本（不改原文）：通道 Cleaner 提取语义内容后用于向量化/提关键词 + cleanInput := input + if a.io != nil { + if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok && chDef.Cleaner != nil { + cleanInput = chDef.Cleaner(input) + } + } + + // upload_* 字段一并转发：webui 的 EventRawInput 订阅方靠它们还原附件卡片。 + rawPayload := map[string]interface{}{"content": input, "source": evt.Source} + for _, k := range []string{"upload_url", "upload_type", "upload_size", "upload_name"} { + if v, ok := evt.Payload[k]; ok { + rawPayload[k] = v + } + } + a.publishEvent(events.EventRawInput, rawPayload) + + archived := a.pruneOnInput(evt, cleanInput) + if archived > 0 { + log.Printf("[agent] pruned %d low-relevance events to document memory", archived) + } + + if !isInterrupt { + a.context.Append(ContextEvent{ + Timestamp: start, + Source: evt.Source, + Input: input, + }) + } + + f := a.newTaskFrame(input, stageCtx) + f.Evt = evt + f.CleanInput = cleanInput + f.IsInterrupt = isInterrupt + f.StartedAt = start + f.Level = a.sched.currentLevel() + return f, terminalNone +} + +// finishInputTask 执行 processInput 的后半段（日志、上下文提交、回执、记忆候选）。 +// +// 只在任务真正结束时调用一次——这正是不变量 I5（每任务恰一次终态）的落点。 +func (a *Agent) finishInputTask(f *TaskFrame, out stepOutcome) { + evt := f.Evt + + // pendingMedia 是「本轮」语义：任务结束即清（挂起时保留，见 runInputTask）。 + if evt != nil && (evt.Type == "image" || evt.Type == "audio") { + a.pendingMedia = nil + } + + if out == outcomeFailed { + log.Printf("[agent] process %s error: %v", evt.Type, f.Err) + resp := fmt.Sprintf("处理错误: %v", f.Err) + a.emitResponse(evt, resp) + a.context.Append(ContextEvent{Timestamp: time.Now(), Source: "agent", Input: f.Input, Response: resp}) + f.Terminal = terminalError + return + } + + elapsed := time.Since(f.StartedAt) + log.Printf("[agent] %s from %s → response (%dms, tools=%v)", + evt.Type, evt.Source, elapsed.Milliseconds(), f.ToolsUsed) + + // 本轮捕获的媒体一起挂到这条事件上：用户上传的、插件注入的，以及模型调 + // multimodal_see_picture / see_video 时经 SetToolBlocks 注入的。 + turnEvt := ContextEvent{ + Timestamp: time.Now(), + Source: "agent", + Input: f.CleanInput, + Response: f.Response, + ToolsUsed: f.ToolsUsed, + ToolResults: f.ToolResults, + } + a.bindEventMedia(&turnEvt, a.drainMediaDigests()) + a.context.Append(turnEvt) + + a.emitResponse(evt, f.Response) + + if !f.StageCtx.NoMemory { + a.emitMemoryCandidate(evt.Source, f.CleanInput, f.Response, f.ToolResults, f.ToolsUsed) + } + f.Terminal = terminalOK } // step 执行恰好一个 step。 @@ -148,6 +393,18 @@ func (a *Agent) step(f *TaskFrame) stepOutcome { // stepPrepare 构建本轮任务的初始帧。 func (a *Agent) stepPrepare(f *TaskFrame) stepOutcome { + // 抢占式中断任务：继承被打断任务的只读前缀（D1=A），不重建上下文。 + if len(f.SeedMsgs) > 0 { + f.Tools = a.buildToolDefs() + f.Msgs = append([]agentAPI.Message(nil), f.SeedMsgs...) + f.Msgs = append(f.Msgs, agentAPI.Message{Role: "user", Content: f.Input}) + // 前缀路径不重写尾部消息（那是「同行注入」的旧形态）； + // 必须消费掉标志位，否则下一个普通任务的尾部会被误改。 + a.interruptInput = false + f.Step = StepLLM + return outcomeContinue + } + budget := ComputeTokenBudget(a.provider, a.systemPrompt) memContext := a.buildMemoryContext(f.Input, budget.MemoryTokens) diff --git a/internal/agent/core/task_lifecycle_test.go b/internal/agent/core/task_lifecycle_test.go new file mode 100644 index 0000000..747f605 --- /dev/null +++ b/internal/agent/core/task_lifecycle_test.go @@ -0,0 +1,204 @@ +package core + +// M3a 验收测试：任务生命周期（prepare → run → finish）与「每任务恰一次终态」。 +// +// 设计依据 docs/zh/input-scheduler-design.md §11.3（X2/X3/X4 的 M3a 形态）： +// 帧覆盖全生命周期后，提交（context.Append）与回执（emitResponse）只能在 +// finish 段发生一次——挂起不会重复提交。 + +import ( + "strings" + "testing" + + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" + "gitcode.com/JianFeeeee/HomeAgent/internal/events" + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +func newLifecycleAgent(t *testing.T, sp agentAPI.Provider, bus *events.Bus, sh *StageHost) *Agent { + t.Helper() + return New(AgentConfig{ + ID: "lifecycle", + Provider: sp, + ProviderManager: agentAPI.NewProviderManager(), + IO: agentIO.NewIOManager(), + StageHost: sh, + EventBus: bus, + }) +} + +func textEvent(source, content string) (*agentIO.InputEvent, chan *agentIO.OutputEvent) { + ch := make(chan *agentIO.OutputEvent, 1) + return &agentIO.InputEvent{ + RequestID: "req-1", + Source: source, + Type: "text", + Payload: map[string]interface{}{"content": content}, + OutputChannel: source, + ResponseCh: ch, + }, ch +} + +// X3（M3a 形态）：正常任务在 finish 段**恰好**提交一次并回执一次。 +func TestLifecycle_NormalCommitsOnceAndReplies(t *testing.T) { + bus := events.NewBus() + var outputs, rawInputs int + bus.Subscribe(events.EventAgentOutput, func(*events.Event) { outputs++ }) + bus.Subscribe(events.EventRawInput, func(*events.Event) { rawInputs++ }) + + sp := &scriptProvider{script: []*agentAPI.CompletionResponse{{Content: "答复"}}} + a := newLifecycleAgent(t, sp, bus, NewStageHost()) + + evt, respCh := textEvent("cli", "你好") + if _, out := a.runInputTask(evt, nil); out != outcomeDone { + t.Fatalf("runInputTask=%v，期望 outcomeDone", out) + } + + select { + case r := <-respCh: + if got, _ := r.Payload["content"].(string); got != "答复" { + t.Fatalf("回执内容=%q，期望 答复", got) + } + if !r.Done { + t.Fatal("回执必须带 Done=true") + } + default: + t.Fatal("同步回执缺失：finish 段必须写 ResponseCh") + } + + if outputs != 1 { + t.Fatalf("agent_output 事件=%d，期望恰好 1（每任务一次终态）", outputs) + } + if rawInputs != 1 { + t.Fatalf("raw_input 事件=%d，期望 1", rawInputs) + } + if a.context.Len() != 2 { + t.Fatalf("上下文事件=%d，期望 2（输入事件 + 本轮事件）", a.context.Len()) + } +} + +// X2：被去重的输入以 skipped 终态结束——不提交、不回执、不发输出事件。 +func TestLifecycle_DuplicateSkippedHasTerminal(t *testing.T) { + bus := events.NewBus() + outputs := 0 + bus.Subscribe(events.EventAgentOutput, func(*events.Event) { outputs++ }) + + sp := &scriptProvider{script: []*agentAPI.CompletionResponse{ + {Content: "第一次"}, {Content: "第二次"}, + }} + a := newLifecycleAgent(t, sp, bus, NewStageHost()) + + e1, _ := textEvent("webui", "同样的消息") + if _, out := a.runInputTask(e1, nil); out != outcomeDone { + t.Fatalf("首次输入=%v，期望 outcomeDone", out) + } + after1, outputs1 := a.context.Len(), outputs + + e2, ch2 := textEvent("webui", "同样的消息") + if _, out := a.runInputTask(e2, nil); out != outcomeDone { + t.Fatalf("去重输入应正常返回（不挂起），实际 %v", out) + } + if a.context.Len() != after1 { + t.Fatalf("去重命中不得提交上下文：%d → %d", after1, a.context.Len()) + } + if len(ch2) != 0 { + t.Fatal("去重命中不得回执（原实现静默 return）") + } + if outputs != outputs1 { + t.Fatalf("去重命中不得发输出事件：%d → %d", outputs1, outputs) + } +} + +// 被 on_input 阶段短路：回执阶段给的响应，且不提交上下文（与原实现一致）。 +func TestLifecycle_OnInputShortCircuit(t *testing.T) { + sh := NewStageHost() + reply := "被插件短路" + sh.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error { + ctx.Response = &reply + return nil + }) + + sp := &scriptProvider{} // 不应被调用到 + a := newLifecycleAgent(t, sp, events.NewBus(), sh) + + evt, respCh := textEvent("cli", "任意") + if _, out := a.runInputTask(evt, nil); out != outcomeDone { + t.Fatalf("短路任务=%v，期望 outcomeDone", out) + } + select { + case r := <-respCh: + if got, _ := r.Payload["content"].(string); got != reply { + t.Fatalf("短路响应=%q，期望 %q", got, reply) + } + default: + t.Fatal("短路路径必须回执") + } + if a.context.Len() != 0 { + t.Fatalf("短路路径不得提交上下文，实际 %d 条", a.context.Len()) + } + if len(sp.reqs) != 0 { + t.Fatal("短路路径不得调用 LLM") + } +} + +// 错误路径：以 error 终态结束，回执错误文本，且提交的是**错误事件**（无 turn 事件）。 +func TestLifecycle_ErrorPathTerminal(t *testing.T) { + bus := events.NewBus() + outputs := 0 + bus.Subscribe(events.EventAgentOutput, func(*events.Event) { outputs++ }) + + sp := &scriptProvider{err: &agentAPI.ProviderError{StatusCode: 401, Message: "bad key"}} + a := newLifecycleAgent(t, sp, bus, NewStageHost()) + + evt, respCh := textEvent("cli", "会失败") + if _, out := a.runInputTask(evt, nil); out != outcomeFailed { + t.Fatalf("runInputTask=%v，期望 outcomeFailed", out) + } + select { + case r := <-respCh: + got, _ := r.Payload["content"].(string) + if !strings.HasPrefix(got, "处理错误:") { + t.Fatalf("错误回执=%q，期望以 处理错误: 开头", got) + } + default: + t.Fatal("错误路径必须回执（否则同步调用方永久挂起）") + } + if outputs != 1 { + t.Fatalf("错误路径的 agent_output 事件=%d，期望 1", outputs) + } + // 输入事件 + 错误事件 = 2；不得出现带 ToolsUsed 的 turn 事件。 + if a.context.Len() != 2 { + t.Fatalf("错误路径上下文事件=%d，期望 2", a.context.Len()) + } + recent := a.context.Recent(10) + last := recent[len(recent)-1] + if last.Response == "" { + t.Fatal("错误事件必须带 Response") + } +} + +// _consolidation_ 走记忆整理专用路径：不回执、不提交上下文。 +func TestLifecycle_ConsolidationRouted(t *testing.T) { + bus := events.NewBus() + outputs := 0 + bus.Subscribe(events.EventAgentOutput, func(*events.Event) { outputs++ }) + + sp := &scriptProvider{script: []*agentAPI.CompletionResponse{{Content: "整理完毕"}}} + a := newLifecycleAgent(t, sp, bus, NewStageHost()) + + evt, respCh := textEvent("system", "整理任务") + evt.OutputChannel = channelConsolidation + if _, out := a.runInputTask(evt, nil); out != outcomeDone { + t.Fatalf("consolidation=%v，期望 outcomeDone", out) + } + if len(respCh) != 0 { + t.Fatal("consolidation 路径不得回执") + } + if outputs != 0 { + t.Fatalf("consolidation 路径不得发输出事件，实际 %d", outputs) + } + if a.context.Len() != 0 { + t.Fatalf("consolidation 路径不得写用户上下文，实际 %d", a.context.Len()) + } +} diff --git a/internal/agent/core/task_test.go b/internal/agent/core/task_test.go index b36870b..24c8711 100644 --- a/internal/agent/core/task_test.go +++ b/internal/agent/core/task_test.go @@ -26,11 +26,17 @@ type scriptProvider struct { script []*agentAPI.CompletionResponse idx int reqs []*agentAPI.CompletionRequest + // err 非空时 Chat 直接返回它（用于错误路径测试）。 + // 配合 ProviderError(401) 可跳过 2s 瞬时重试，让测试保持快速。 + err error } func (s *scriptProvider) Name() string { return "script" } func (s *scriptProvider) Chat(ctx context.Context, req *agentAPI.CompletionRequest) (*agentAPI.CompletionResponse, error) { s.reqs = append(s.reqs, req) + if s.err != nil { + return nil, s.err + } if s.idx >= len(s.script) { return &agentAPI.CompletionResponse{Content: ""}, nil }