From d1e502d367a20207f3c667212738b19276f10aaf Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Tue, 25 Aug 2026 07:52:51 +0800 Subject: [PATCH] fix(agent): self-input channel carries target output channel flag The selfInputCh previously treated ALL internal messages as memory consolidation tasks (hardcoded _consolidation_ output channel), which silently discarded child-agent completion notifications: - processConsolidation never appends to conversation context, so the parent agent could not see that its child had finished - it also discards the LLM response without emitting to any output channel, so nothing reached the user - net effect: notifications vanished; parent never called child_result Restore the intended design: each self-input message now carries a target output channel. Only consolidation tasks (_consolidation_) go through the no-memory path (no context write, no emit). Child notifications carry the parent's original output channel and are processed as normal input: appended to context, LLM sees them and can call child_result, and the response is emitted back to the user. Changes: - new selfInputMsg{text, channel} type + channelConsolidation const - selfInputCh: chan string -> chan selfInputMsg - injectSelf (consolidation) keeps _consolidation_; new injectSelfChannel for flagged messages - handleSelfInput routes on msg.channel instead of hardcoding - executeSpawnChild captures a.currentOutputChannel and passes it to runChildTask so the notification returns to the originating channel (falls back to "cli" when unset or consolidation) - executeChildResultTool: remove dead double-lock/re-check block Verified end-to-end with tmux PTY against llmsproxy: spawn_child -> child done -> notification processed via normal path (log shows 'input from system -> response, tools=[child_result]'), parent agent retrieved the child result successfully. --- internal/agent/core/agent.go | 28 ++++++++++++++++++++-------- internal/agent/core/eventloop.go | 28 ++++++++++++++++++++++------ internal/agent/core/spawn.go | 28 ++++++++++++++-------------- 3 files changed, 56 insertions(+), 28 deletions(-) diff --git a/internal/agent/core/agent.go b/internal/agent/core/agent.go index 76c6b72..c826dc0 100644 --- a/internal/agent/core/agent.go +++ b/internal/agent/core/agent.go @@ -76,8 +76,11 @@ type Agent struct { eventBus *events.Bus pluginHealth *pluginHealthTracker - // 自循环输入通道：核心内部任务（记忆消歧、系统维护），不经过 IO 层 - selfInputCh chan string + // 自循环输入通道：核心内部任务（记忆消歧、系统维护、子 Agent 通知）， + // 不经过 IO 层。每条消息携带目标输出通道： + // "_consolidation_" = 记忆整理（无记忆路径，不写入上下文、不 emit 响应） + // 其他 = 正常处理（写入上下文、emit 响应到该通道） + selfInputCh chan selfInputMsg // 子任务异步执行 childMu sync.Mutex @@ -218,7 +221,7 @@ func New(cfg AgentConfig) *Agent { maxContextSize: cfg.MaxContextSize, stageHost: cfg.StageHost, eventBus: cfg.EventBus, - selfInputCh: make(chan string, 64), + selfInputCh: make(chan selfInputMsg, 64), childResults: make(map[string]string), interceptCh: make(chan *agentIO.InputEvent, 64), pluginHealth: newPluginHealthTracker(), @@ -275,16 +278,25 @@ func (a *Agent) IsDuplicateInput(source, content string) bool { } // SelfInputChan 返回自循环输入通道（只读，供内部测试验证） -func (a *Agent) SelfInputChan() <-chan string { +func (a *Agent) SelfInputChan() <-chan selfInputMsg { return a.selfInputCh } -// injectSelf 向自循环通道发送内部任务（记忆消歧、系统维护） -// 线程安全，不阻塞发送者（通道缓冲 64） +// injectSelf 向自循环通道发送记忆整理类内部任务（无记忆路径）。 +// 线程安全，不阻塞发送者（通道缓冲 64）。 func (a *Agent) injectSelf(task string) { + a.injectSelfChannel(selfInputMsg{ + text: task, + channel: channelConsolidation, + }) +} + +// injectSelfChannel 向自循环通道发送一条带目标通道标志的消息。 +// channel == "_consolidation_" 走无记忆整理路径；其他值走正常处理路径。 +func (a *Agent) injectSelfChannel(msg selfInputMsg) { select { - case a.selfInputCh <- task: + case a.selfInputCh <- msg: default: - log.Printf("[agent] self input channel full, dropping task: %s", truncateStr(task, 80)) + log.Printf("[agent] self input channel full, dropping task: %s", truncateStr(msg.text, 80)) } } diff --git a/internal/agent/core/eventloop.go b/internal/agent/core/eventloop.go index d9b58eb..c538772 100644 --- a/internal/agent/core/eventloop.go +++ b/internal/agent/core/eventloop.go @@ -24,8 +24,8 @@ func (a *Agent) eventLoop() { select { case evt := <-a.io.InputChan(): a.handleInput(evt) - case task := <-a.selfInputCh: - a.handleSelfInput(task) + case msg := <-a.selfInputCh: + a.handleSelfInput(msg) case <-a.ctx.Done(): return } @@ -108,13 +108,29 @@ func (a *Agent) interceptLoop() { } } -func (a *Agent) handleSelfInput(task string) { +// channelConsolidation 标记记忆整理类自输入：无记忆路径处理， +// 不写入对话上下文、不向任何输出通道 emit 响应。 +const channelConsolidation = "_consolidation_" + +// selfInputMsg 自循环输入消息。channel 决定处理路径： +// - channelConsolidation：记忆整理，无记忆（不污染上下文/知识库） +// - 其他值（如 "cli"、"webui"）：正常输入路径，写入上下文并 emit 响应 +// （典型场景：子 Agent 完成通知，需让父 Agent 感知并可回复用户） +type selfInputMsg struct { + text string + channel string +} + +func (a *Agent) handleSelfInput(msg selfInputMsg) { + if msg.channel == "" { + msg.channel = channelConsolidation // 兼容空值：默认走整理路径 + } a.processTextInput(&agentIO.InputEvent{ Source: "system", Type: "text", - Payload: map[string]interface{}{"content": task}, - OutputChannel: "_consolidation_", - }, task) + Payload: map[string]interface{}{"content": msg.text}, + OutputChannel: msg.channel, + }, msg.text) } func (a *Agent) handleInput(evt *agentIO.InputEvent) { diff --git a/internal/agent/core/spawn.go b/internal/agent/core/spawn.go index 6b63c18..ca809b8 100644 --- a/internal/agent/core/spawn.go +++ b/internal/agent/core/spawn.go @@ -19,12 +19,19 @@ func (a *Agent) executeSpawnChild(tc agentAPI.ToolCall) string { taskID := fmt.Sprintf("child_%d", a.childNextID) a.childMu.Unlock() - go a.runChildTask(taskID, task) + // 捕获父 Agent 当前输出通道：子任务完成通知需回到发起对话的通道， + // 让父 Agent 正常感知并可回复用户（而非走无记忆整理路径丢失通知）。 + parentChannel := a.currentOutputChannel + if parentChannel == "" || parentChannel == channelConsolidation { + parentChannel = "cli" + } + + go a.runChildTask(taskID, task, parentChannel) return fmt.Sprintf("子任务已启动（ID: %s），完成后会自动通知你，届时请使用 child_result 工具查看输出", taskID) } -func (a *Agent) runChildTask(taskID, task string) { +func (a *Agent) runChildTask(taskID, task string, parentChannel string) { if a.provider == nil { log.Printf("[child] %s failed: no LLM provider configured", taskID) return @@ -105,11 +112,10 @@ func (a *Agent) runChildTask(taskID, task string) { log.Printf("[child] %s done: %s", taskID, truncateStr(finalResult, 100)) notification := fmt.Sprintf("子任务 %s 已完成，请调用 child_result 工具查看输出", taskID) - select { - case a.selfInputCh <- notification: - default: - log.Printf("[child] self input channel full, dropping notification for %s", taskID) - } + a.injectSelfChannel(selfInputMsg{ + text: notification, + channel: parentChannel, // 回到父对话通道，正常处理（写入上下文 + emit 响应） + }) } func (a *Agent) executeChildResultTool(tc agentAPI.ToolCall) string { @@ -122,13 +128,7 @@ func (a *Agent) executeChildResultTool(tc agentAPI.ToolCall) string { result, ok := a.childResults[taskID] if !ok { a.childMu.Unlock() - - a.childMu.Lock() - _, exists := a.childResults[taskID] - a.childMu.Unlock() - if !exists { - return fmt.Sprintf("子任务 %s 不存在或已过期", taskID) - } + return fmt.Sprintf("子任务 %s 不存在或已过期", taskID) } delete(a.childResults, taskID) a.childMu.Unlock()