mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-28 05:13:27 +00:00
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.
This commit is contained in:
@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
@ -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) {
|
||||
|
||||
@ -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()
|
||||
|
||||
Reference in New Issue
Block a user