mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-02 23:33:58 +00:00
refactor(core)!: N0 无状态化 —— 删除 Agent.currentOutputChannel,通道只跟输入事件/帧走
驻留式子 agent 设计(docs/zh/resident-subagent-design.md)的里程碑 N0。 ## 问题 `a.currentOutputChannel` 是 **agent 级可变字段**,只在 prepare 段写入,而被打断任务 恢复时**不重新 prepare**(resumeTask 只 rebase 前缀)。于是中断任务 prepare 时把它 覆盖成自己的通道,被恢复的任务再把回复发到**中断任务的通道**上——两个任务串台。 后果不只是标签错:工具提示词里那句"当前输入来源通道是 X,对应输出门工具是 output_send__X"会诱导模型**把回复主动发到错误的通道**。 ## 两处一起改(用户指出的两件事) 1. **内核不应持有"当前通道"**:通道是随输入事件带进来的,路由发生在**进内核之前**, 输出是 agent 的**主动调用**。删除该字段,改为一律从输入事件推导 (`outputChannelOf(evt)`)或读本任务的帧(`f.OutputChannel`)。 2. **提示词不应预设 outputch**:删掉"当前输入来源通道是 X → 用 output_send__X"那两行, 改为"不要假设当前通道是固定值;先看消息本身与上下文的来源信息,不确定时先调 output_list_channels"。 ## 改动面(把通道一路显式传下去,而不是读共享状态) - `agent.go`:删字段 - `task.go`:新增 `outputChannelOf` / `isCriticalChannel`;帧记录通道; 安全点与 setCritical 用帧/事件推导;步骤内事件标签改用 `f.OutputChannel`; `executeToolCall(f.CurTool, f.OutputChannel)`;`callLLMWithFallback(..., f.OutputChannel)` - `process.go`:`chatStreamWithFallback` / `accumulateStream` 增加 channel 参数 (增量事件的 channel 标签由此而来) - `stage.go`:`runStage` 从 `ctx.Extra["output_channel"]` 读(发起方写入) - `eventloop.go`:`emitResponse` 用 `outputChannelOf(evt)`;stageCtx 带上通道 - `spawn.go` / `toolcall.go`:`executeSpawnChild` 的 parentChannel 由调用方(帧)传入 (子任务完成通知要回到**发起这次 spawn 的那个任务**的通道) - `distill.go`:删掉 consolidation 路径里的赋值 - `tooldefs.go`:删掉提示词里的通道预设 ## 验收 - `scheduler_channel_routing_test.go`(N0 守卫):中断任务跑过之后,被恢复任务的 输出通道仍是它自己的(改前实测为 cli,期望 qq) - `TestCriticalSection_ConsolidationMarked`:补上推导链 「输入事件 → 通道 → isCriticalChannel → scheduler.critical」的集成断言 - 全仓 `go test ./...` 37 包 ok / 0 FAIL;`-race ./internal/agent/...` 干净 - 残留 `currentOutputChannel` 引用为 0(只剩描述历史的注释)
This commit is contained in:
@ -109,14 +109,14 @@ func dropContinuationPlaceholders(msgs []agentAPI.Message) []agentAPI.Message {
|
||||
//
|
||||
// 超时收益:首包 ~1-3s 到达即建立活性,后续只要 token 在流动就不会触发
|
||||
// 空闲超时;总生成时长不再受限於 180s 整体超时。
|
||||
func chatStreamWithFallback(ctx context.Context, p agentAPI.Provider, req *agentAPI.CompletionRequest, a *Agent) (*agentAPI.CompletionResponse, error) {
|
||||
func chatStreamWithFallback(ctx context.Context, p agentAPI.Provider, req *agentAPI.CompletionRequest, a *Agent, channel string) (*agentAPI.CompletionResponse, error) {
|
||||
ch, err := p.ChatStream(ctx, req)
|
||||
if err != nil {
|
||||
log.Printf("[agent] stream connect failed (%v), falling back to non-stream chat", err)
|
||||
return p.Chat(ctx, req)
|
||||
}
|
||||
|
||||
resp, accErr := accumulateStream(ctx, ch, a)
|
||||
resp, accErr := accumulateStream(ctx, ch, a, channel)
|
||||
|
||||
// 中断/超时取消必须保持取消语义传给调用方(与原 Chat() 行为一致:
|
||||
// 被 cancel 时丢弃已收内容返回 err),让 process() 的 continue 分支
|
||||
@ -127,7 +127,7 @@ func chatStreamWithFallback(ctx context.Context, p agentAPI.Provider, req *agent
|
||||
if a != nil {
|
||||
a.publishEvent(events.EventContentDelta, map[string]interface{}{
|
||||
"content": "",
|
||||
"channel": a.currentOutputChannel,
|
||||
"channel": channel,
|
||||
"reset": true,
|
||||
})
|
||||
}
|
||||
@ -157,7 +157,7 @@ type toolCallAcc struct {
|
||||
|
||||
// accumulateStream 消费 chunk channel,累积为完整 CompletionResponse,
|
||||
// 同时发布增量事件。返回的 response 与非流式 Chat() 的返回等价。
|
||||
func accumulateStream(ctx context.Context, ch <-chan agentAPI.StreamChunk, a *Agent) (*agentAPI.CompletionResponse, error) {
|
||||
func accumulateStream(ctx context.Context, ch <-chan agentAPI.StreamChunk, a *Agent, channel string) (*agentAPI.CompletionResponse, error) {
|
||||
resp := &agentAPI.CompletionResponse{
|
||||
ToolCalls: make([]agentAPI.ToolCall, 0),
|
||||
}
|
||||
@ -209,7 +209,7 @@ func accumulateStream(ctx context.Context, ch <-chan agentAPI.StreamChunk, a *Ag
|
||||
if a != nil {
|
||||
a.publishEvent(events.EventReasoningDelta, map[string]interface{}{
|
||||
"content": ck.ReasoningContent,
|
||||
"channel": a.currentOutputChannel,
|
||||
"channel": channel,
|
||||
})
|
||||
}
|
||||
}
|
||||
@ -218,7 +218,7 @@ func accumulateStream(ctx context.Context, ch <-chan agentAPI.StreamChunk, a *Ag
|
||||
if a != nil {
|
||||
a.publishEvent(events.EventContentDelta, map[string]interface{}{
|
||||
"content": ck.Content,
|
||||
"channel": a.currentOutputChannel,
|
||||
"channel": channel,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user