Files
HomeAgent/internal/agent/core/resident.go
JianFeeeee 8acd3ce1a8 fix(offload): 内核说明不能被再转投(自我循环)+ offload_owned 未接线
★ 线上实测两个缺陷:

1. **自我循环**:转投会在队列留一条 [系统] 说明(source=kernel),
   而转投条件把这条说明也算进「积压够了」⇒ 每次转投都产生下一轮要转投的东西。
   实测 5 秒内连续触发两次,分诊助手不断收到「N 条积压已转投」这类噪音。
   修法:takeQueuedInputs 排除 isKernelNotice(source=kernel)。

2. **offload_owned 永远为 false**:我加了 ResidentInfo 字段、加了状态面映射,
   却漏了在 rc.info() 里赋值 ⇒ 线上转投子明明存在,读出来是 null。
   这类「加了字段但没接线」不会报错,只会让父的判断悄悄失效
   (父据此决定回收策略,读到 false 就会把临时助手当成正式子)。

测试 +3:说明不转投 / 循环必须终止 / offload_owned 会被上报。
前两条已实测「禁用守卫会失败、恢复后通过」,是真回归测试。
2026-09-19 17:47:44 +08:00

683 lines
26 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package core
// 驻留式子 agent 的**生命周期与控制面**(设计 docs/zh/resident-subagent-design.md §7/§9/§10
//
// 父 ──创建/发送消息/查看/压缩/回收/销毁──► 驻留子
// 子 ──主动消息(L3) / contextfull(L4)──► 父
//
// 层级关系:
// - 父持**登记表**residents它是查看·发送·压缩·回收·销毁的寻址依据
// - 父 `Stop()` ⇒ 销毁全部子(**不留孤儿**
// - 子的 `KernelSource` = 父;父则把子的 contextfull 当**内核级事件**raiseKernelInterrupt上报给自己
// - 子持有自己的 **inputch 处理表**(父 pull不打断子
import (
"encoding/json"
"fmt"
"log"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"time"
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory"
"gitcode.com/JianFeeeee/HomeAgent/pkg/types"
)
// InputchRecord 是 inputch 处理表的一条记录(子持有,父 pull
type InputchRecord struct {
InputCh string `json:"inputch"`
At time.Time `json:"at"`
Proactive bool `json:"proactive"` // true = 子主动写入false = 系统自动写
Text string `json:"text"`
}
// ResidentOptions 是创建一个驻留子的参数(父的"创建"动作)。
type ResidentOptions struct {
// ID 是子 agent 的 id同时是登记表的键、跨 agent 寻址的依据)。
ID string
// TaskPrompt 是在固定提示词之上注入的**任务提示词**。
TaskPrompt string
// InputChs 是**划入**给这个子的 inputch单位 = inputch可来自同一插件的多个
InputChs []string
// AllowedOutputs 是授权给它的输出通道集合nil/空 = 完整授权)。
AllowedOutputs []string
// Capacity 是划入 inputch 的队列容量0 = 内核默认)。
Capacity int
// TempPath 是它 temp 图记忆的存储路径(必填;与子同生共死)。
TempPath string
// OffloadOwned 标记这是**内核为转投而拉起**的驻留子(见 offload.go
//
// 为什么要区分:自动转投会复用、也会按上限限制自己拉起的子,
// 但**绝不能**把人(父的模型)建的子算进去或拿去复用 ——
// 那会把人工安排的工作负载搬到一个本来在做别的事的 agent 上。
OffloadOwned bool
}
// ResidentInfo 是父对某个驻留子的可查询状态(登记表条目 + 状态面摘要)。
type ResidentInfo struct {
ID string `json:"id"`
State string `json:"state"`
InputChs []string `json:"inputchs"`
AllowedOutputs []string `json:"allowed_outputs"`
Rounds int `json:"rounds"`
ContextFull bool `json:"context_full"`
CreatedAt time.Time `json:"created_at"`
TableSize int `json:"table_size"`
// OffloadOwned 标记这是内核为承接积压而拉起的**临时分诊助手**(见 offload.go
OffloadOwned bool `json:"offload_owned,omitempty"`
Table []InputchRecord `json:"table,omitempty"`
// Sched* 是这个驻留子**自己的**输入调度器积压摘要(排队 / 待处理中断 /
// 中断栈 / 四级中断队列)。
//
// 为什么必须单列:每个驻留子是独立 agent跑自己的事件循环与调度器
// (见 agent.go 的 newScheduler。KernelStatus.Scheduler 只是**根**那份,
// 拿它代表全部子会让界面把「某个子堵死」显示成「一切正常」。
SchedReady int `json:"sched_ready"`
SchedPending int `json:"sched_pending"`
SchedStack int `json:"sched_stack"`
SchedQueues [5]int `json:"sched_queues"`
}
type residentChild struct {
id string
agent *Agent
light *memory.LightMemory
mainRO *memory.GraphDB
tempPath string
dir string
inputChs []string
allowed []string
createdAt time.Time
// offloadOwned 标记这是内核为转投拉起的子(见 offload.go
offloadOwned bool
mu sync.Mutex
state string // running | contextfull | stopped
}
// mainGraph 返回父自己的完整图记忆库(驻留子的受限句柄由它派生)。
func (a *Agent) mainGraph() *memory.GraphDB { return a.memory }
// SpawnResident 创建一个驻留子(父的"创建"动作)。
func (a *Agent) SpawnResident(opts ResidentOptions) (ResidentInfo, error) {
if strings.TrimSpace(opts.ID) == "" {
return ResidentInfo{}, fmt.Errorf("驻留子必须有 id")
}
if strings.TrimSpace(opts.TempPath) == "" {
return ResidentInfo{}, fmt.Errorf("驻留子必须给出 temp 图记忆路径")
}
main := a.mainGraph()
if main == nil {
return ResidentInfo{}, fmt.Errorf("父没有图记忆,无法为驻留子提供主库只读视图")
}
a.residentMu.Lock()
if a.residents == nil {
a.residents = map[string]*residentChild{}
}
if _, dup := a.residents[opts.ID]; dup {
a.residentMu.Unlock()
return ResidentInfo{}, fmt.Errorf("驻留子 %s 已存在", opts.ID)
}
a.residentMu.Unlock()
// ① 轻量内核的记忆装配:主库**受限句柄**(结构上写不进)+ 自己的 temp 实例。
// temp 目录由内核创建(调用方只给路径)——与子同生共死,销毁时整目录丢弃。
if dir := filepath.Dir(opts.TempPath); dir != "" && dir != "." {
if err := os.MkdirAll(dir, 0o700); err != nil {
return ResidentInfo{}, fmt.Errorf("创建 temp 目录: %w", err)
}
}
mainRO, err := memory.OpenGraphDBReadOnly(main.Path())
if err != nil {
return ResidentInfo{}, fmt.Errorf("打开主库受限句柄: %w", err)
}
light, err := memory.NewLightMemory(mainRO, opts.TempPath, true)
if err != nil {
_ = mainRO.Close()
return ResidentInfo{}, err
}
// ② 划入 inputch登记表里记归属一个插件的多个 inputch 可分别划给不同子)。
if reg := a.io.ChannelRegistry(); reg != nil {
for _, ch := range opts.InputChs {
if err := reg.Assign(ch, opts.ID, opts.Capacity); err != nil {
_ = light.Close()
_ = mainRO.Close()
return ResidentInfo{}, fmt.Errorf("划入 inputch %s: %w", ch, err)
}
}
}
// ③ 子的 io**独立**的 IOManager自己的输入通道入口但共享通道登记表
// 并把父的 io 挂成"上级"——**输出通道io 里的 Device由插件登记在父的 io 上**
// 子若不继承这张视图,`output_send__<通道>` 一律被判"通道不存在或不可用"、
// `output_list_channels` 为空、连 `output_send__*` 工具都不会生成
// (现场联调:父侧通道装载完整、子侧 childIO 空壳)。
// 回退是实时的(设备随资源生灭),授权仍由 opts.AllowedOutputs 白名单把关。
childIO := agentIO.NewIOManager()
if reg := a.io.ChannelRegistry(); reg != nil {
childIO.SetChannelRegistry(reg)
}
childIO.SetParentIO(a.io)
parentID := string(a.id)
child := New(AgentConfig{
ID: types.AgentID(opts.ID),
// SystemPrompt 必须**继承父的**
//
// 拿不到它时子只能用 buildSystemPrompt 里的兜底句(一句人格描述)。
// 而父的提示词里写着**回复投递规则** ——「面向 qq、wechat、a2a、acp 等异步
// 通道时,必须显式调用 output_send__{通道名};只返回纯文本会被直接丢弃,
// 用户永远收不到,而你会误以为已经回复过了」,以及事实性约束、命令与文件
// 操作策略等。缺了这些,子于异步通道(如 QQ 积压)处理完却发不出去,
// 而且它自己不会意识到(提示词没告诉它)。
//
// 实测2026-09-19webui 这类**同步**通道能回,是因为走 ResponseCh
// 而 QQ 是异步通道、必须子主动 output_send —— 这正是本字段必须接上的理由。
SystemPrompt: a.systemPrompt,
Provider: a.provider,
ProviderManager: a.providerManager,
IO: childIO,
StageHost: a.stageHost,
LightMemory: light, // 轻量内核:只有图记忆共同面
AllowedOutputs: opts.AllowedOutputs,
KernelSource: parentID, // 子的 L4 只属于父
ParentID: parentID,
TaskPrompt: opts.TaskPrompt,
})
rc := &residentChild{
id: opts.ID, agent: child, light: light, mainRO: mainRO,
tempPath: opts.TempPath, dir: filepath.Dir(opts.TempPath),
inputChs: append([]string(nil), opts.InputChs...),
allowed: append([]string(nil), opts.AllowedOutputs...),
createdAt: time.Now(), state: "running",
offloadOwned: opts.OffloadOwned,
}
// ④ 子 → 父的主动消息(**L3 中断**,带子标识):投进父的 inputch。
parentInCh := a.residentInboundChannel(opts.ID)
// 登记了入站 inputch 就必须有回滚路径:注册点与 residents 登记之间
// 当前无可失败步骤,但一旦将来插入(或 child.Start 变更),早退会把这条
// 登记留在共享表里——就是 child/<id> 残留那只 bug 的另一条入口。
// defer 保证“只要没走到成功返回,就注销这条登记”。
registered := false
defer func() {
if registered {
return
}
if reg := a.io.ChannelRegistry(); reg != nil {
reg.Unregister(inboundChannelName(opts.ID))
}
}()
child.notifyParent = func(text string) {
a.io.InjectInterruptTextOpts(opts.ID, parentInCh, text,
agentIO.InjectOptions{Priority: "L3"})
}
// ⑤ 子的 contextfull → 父侧的**内核级事件**L4带子标识
child.onContextFull = func() { a.handleChildContextFull(rc) }
a.residentMu.Lock()
a.residents[opts.ID] = rc
a.residentMu.Unlock()
child.Start()
// ⑥ create 即开工:把任务提示词作为**第一条排队输入**投给子。
//
// 为什么必须在这里投TaskPrompt 只进子的系统提示词("你是谁、要做什么"
// 而**不会**让子跑起来 —— 实测现象是子启动后 rounds=0、永远待机
// (日志 `[agent] r1 started, waiting for IO interrupts` 之后无事发生)。
// 走排队输入(非中断):创建是"安排工作",不是"打断它正在做的事"。
if strings.TrimSpace(opts.TaskPrompt) != "" {
child.io.InjectInputTo(a.residentParentSource(), parentInCh, "text",
map[string]interface{}{"content": opts.TaskPrompt})
}
registered = true
return rc.info(), nil
}
// residentParentSource 是"父给子投递"的输入来源名(子的视角里能看出是谁发的)。
func (a *Agent) residentParentSource() string { return "parent/" + string(a.id) }
// residentInboundChannel 是"父接收某个子的消息"的 inputch 名(登记进登记表可见)。
//
// inboundChannelName 只拼名字,不产生副作用(注销路径要用它算出同一个名字,
// 不能再去调 residentInboundChannel——那会顺手把刚摘掉的登记又写回去
func inboundChannelName(childID string) string { return "child/" + childID }
func (a *Agent) residentInboundChannel(childID string) string {
ch := inboundChannelName(childID)
if reg := a.io.ChannelRegistry(); reg != nil {
// 归属父自己:它是父的入站 inputch。
_ = reg.Register(agentIO.InputChannel{Name: ch, Plugin: "resident", Owner: string(a.id)})
}
return ch
}
// DestroyResident 立刻销毁一个驻留子并从登记表移除(父的"销毁"动作;不收割)。
//
// 销毁是父**随时**可做的;父退出时由 StopResidents 对全部子执行。
func (a *Agent) DestroyResident(id string) error {
a.residentMu.Lock()
rc, ok := a.residents[id]
if ok {
delete(a.residents, id)
}
a.residentMu.Unlock()
if !ok {
return fmt.Errorf("驻留子 %s 不存在", id)
}
a.teardownResident(rc)
return nil
}
// ResidualPolicy 是父对**残余任务**的显式决定。
//
// 为什么需要它(用户 2026-09-19 明确要求):回收/销毁驻留子时,
// 它手头可能还有**尚未处理的消息**。这些消息的处置不能由内核悄悄决定:
// - 直接丢弃 → 用户消息无声消失异步通道更是零反馈qq 无 ResponseCh
// 用户不知道发生了什么,系统里也没任何痕迹);
// - 无条件转回父 → 父本来就很忙,把一堆活重新塞回去可能反而加剧积压。
//
// 因此与其它控制面动作一致:**原语在内核,决定在父的模型**(设计 §7
// 内核负责把残余任务**列清楚**(来源、通道、内容),父选 keep转回自己/ drop。
//
// 无论选哪个,内核都会**逐条留日志**:丢弃必须可追溯。
// 默认(不传 policy取 keep宁可多做一件不可默默丢一条。
type ResidualPolicy string
const (
// ResidualKeep 把残余任务转回父自己的队列,父稍后处理。
ResidualKeep ResidualPolicy = "keep"
// ResidualDrop 明确丢弃残余任务(父已看过清单并确认)。
ResidualDrop ResidualPolicy = "drop"
)
// TakeResidual 取走一个驻留子手头**尚未处理**的残余任务(不处置)。
//
// 调它**会**从子的队列里移除这些任务,所以父必须先看返回值再决定:
// 一旦取走,不再调 ApplyResidual 就等于把它们丢了。控制面工具走 ApplyResidual
// (取+处置一步完成)以避免这个误用。
func (a *Agent) TakeResidual(id string) ([]*agentIO.InputEvent, error) {
a.residentMu.Lock()
rc := a.residents[id]
a.residentMu.Unlock()
if rc == nil || rc.agent == nil {
return nil, fmt.Errorf("驻留子 %s 不存在", id)
}
return rc.agent.sched.takeAllPendingEvents(), nil
}
// ApplyResidual 按父给出的策略处置一个驻留子的残余任务。
//
// keep逐条转回父自己的队列保留原事件含 ResponseCh 与来源/通道);
// drop逐条记录日志后丢弃不可追溯的丢弃是不允许的
//
// 无论哪种,带 ResponseCh 的都要给终态,否则 cli/a2a 这类无超时同步调用方
// 会永久挂起(设计 §7 I5
func (a *Agent) ApplyResidual(id string, policy ResidualPolicy) (int, string, error) {
events, err := a.TakeResidual(id)
if err != nil {
return 0, "", err
}
if len(events) == 0 {
return 0, "无残余任务", nil
}
if policy == ResidualDrop {
for _, evt := range events {
// 丢弃必须留痕异步通道qq/wechat没有 ResponseCh
// 不记日志的话“这条消息为什么没人回”永远查不出来。
log.Printf("[resident] %s 残余任务按父的决定丢弃source=%s channel=%s request=%s content=%s",
id, evt.Source, evt.OutputChannel, evt.RequestID, truncateStr(inputTextOf(evt), 80))
a.emitSkippedReply(evt, "residual_dropped_by_parent")
}
return len(events), fmt.Sprintf("已按父的决定丢弃 %d 条残余任务(逐条已记日志)", len(events)), nil
}
// 默认 keep转回父自己
for _, evt := range events {
if evt == nil {
continue
}
a.sched.enqueue(newInputTask(evt))
}
a.sched.signalWake()
return len(events), fmt.Sprintf("已把 %d 条残余任务转回主 agent 队列", len(events)), nil
}
// inputTextOf 取一条输入事件的正文(供日志描述残余任务用)。
func inputTextOf(evt *agentIO.InputEvent) string {
if evt == nil || evt.Payload == nil {
return ""
}
s, _ := evt.Payload["content"].(string)
return s
}
// teardownResident 停内核、放通道、丢 temp销毁与回收共用
func (a *Agent) teardownResident(rc *residentChild) {
rc.mu.Lock()
rc.state = "stopped"
rc.mu.Unlock()
rc.agent.Stop() // 停子的调度器(取消其运行中的任务)
if rc.light != nil {
_ = rc.light.Close() // 关掉 temp 实例
}
if rc.mainRO != nil {
_ = rc.mainRO.Close()
}
// temp 与子同生共死:连同目录一起丢弃(回收/销毁都不保留)。
if rc.dir != "" && strings.Contains(rc.tempPath, rc.id) {
_ = os.RemoveAll(rc.dir)
}
// 归还划入的 inputch归属清空 ⇒ 回到"未分配",可再分配)。
if reg := a.io.ChannelRegistry(); reg != nil {
for _, ch := range rc.inputChs {
_ = reg.Assign(ch, "", 0)
}
// 注销"父接收该子消息"的入站 inputchchild/<id>)。
//
// 它由 residentInboundChannel 在 create 时登记Owner=父),销毁时必须
// 一并摘掉:登记表是共享的、按 name 全局唯一,残留会随 create/destroy
// 次数单调累积脏数据。实测destroy 后 child/<id> 仍挂在根 agent 名下,
// 而外部没有任何工具能单独注销 inputch只能重启 homed 清。
// 注意用纯函数算名字,不要再走 residentInboundChannel会重新登记
reg.Unregister(inboundChannelName(rc.id))
}
}
// StopResidents 销毁全部驻留子(父退出时必须;不留孤儿)。
func (a *Agent) StopResidents() int {
a.residentMu.Lock()
all := make([]*residentChild, 0, len(a.residents))
for _, rc := range a.residents {
all = append(all, rc)
}
a.residents = map[string]*residentChild{}
a.residentMu.Unlock()
for _, rc := range all {
a.teardownResident(rc)
}
return len(all)
}
// Residents 返回登记表的快照(按 id 排序,便于断言与展示稳定)。
func (a *Agent) Residents() []ResidentInfo {
a.residentMu.Lock()
all := make([]*residentChild, 0, len(a.residents))
for _, rc := range a.residents {
all = append(all, rc)
}
a.residentMu.Unlock()
out := make([]ResidentInfo, 0, len(all))
for _, rc := range all {
out = append(out, rc.info())
}
sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
return out
}
// ResidentTable 是父**查看**子的 inputch 处理表pull不打断子
func (a *Agent) ResidentTable(id string) ([]InputchRecord, error) {
a.residentMu.Lock()
rc, ok := a.residents[id]
a.residentMu.Unlock()
if !ok {
return nil, fmt.Errorf("驻留子 %s 不存在", id)
}
return rc.agent.inputchTableSnapshot(), nil
}
// SendToResident 是父的"发送消息":经输出通道寻址到该子的 inputch
// 对子而言是 **L4 中断**(取消当前状态 + 插入新消息)。
func (a *Agent) SendToResident(id, text string) error {
a.residentMu.Lock()
rc, ok := a.residents[id]
a.residentMu.Unlock()
if !ok {
return fmt.Errorf("驻留子 %s 不存在", id)
}
ch := "sub/" + id
if len(rc.inputChs) > 0 {
ch = rc.inputChs[0]
}
// 来源 = 父;子的 KernelSource 是父 ⇒ 在子的阶梯上这是合法的 L4。
rc.agent.io.InjectInterruptTextOpts(string(a.id), ch, text,
agentIO.InjectOptions{Priority: "L4"})
return nil
}
// CompressResident 是父的"压缩"**保留语义**):压上下文 + **清理处理表**,子继续存在。
func (a *Agent) CompressResident(id string) (int, error) {
a.residentMu.Lock()
rc, ok := a.residents[id]
a.residentMu.Unlock()
if !ok {
return 0, fmt.Errorf("驻留子 %s 不存在", id)
}
dropped := rc.agent.context.TrimKeepRecent(residentKeepRecent)
rc.agent.clearInputchTable() // 处理表记的是被压掉那段窗口的逐轮处理 ⇒ 必须清
rc.agent.resetContextFull()
rc.mu.Lock()
rc.state = "running"
rc.mu.Unlock()
return dropped, nil
}
// ReclaimResident 是父的"回收"**取消语义**):父读子的 temp → 选记录 → 合入 main
// → 丢弃 temp → **取消**该驻留子。
//
// keep 由父决定"哪些纳入记忆";为 nil 时表示全部合入。
func (a *Agent) ReclaimResident(id string, keep func([]InputchRecord, []memory.Triple) []memory.Triple) (ResidentInfo, error) {
a.residentMu.Lock()
rc, ok := a.residents[id]
a.residentMu.Unlock()
if !ok {
return ResidentInfo{}, fmt.Errorf("驻留子 %s 不存在", id)
}
info := rc.info()
var promoted int
// 收割:读 temp 的全部活跃三元组(比通过图记录选择更直接)。
if rc.light != nil && rc.light.Temp() != nil && a.mainGraph() != nil {
exported, err := rc.light.Temp().ExportTriples(0)
if err != nil {
return info, fmt.Errorf("读取子 temp 失败: %w", err)
}
selected := exported
if keep != nil {
selected = keep(info.Table, exported)
}
if len(selected) > 0 {
if _, _, err := a.mainGraph().Commit(selected, "reclaim/"+id, 0); err != nil {
return info, fmt.Errorf("合入主记忆失败: %w", err)
}
promoted = len(selected)
}
}
// 收割完成 ⇒ 取消该驻留子(回收是取消语义,≠ 压缩)。
if err := a.DestroyResident(id); err != nil {
return info, err
}
info.State = fmt.Sprintf("reclaimed(promoted=%d)", promoted)
return info, nil
}
// handleChildContextFull 把子的 contextfull 当**内核级事件**上报给父自己:
// 父侧 L4 中断(带子标识)—— 只推信号,细节靠"查看"拉状态面。
func (a *Agent) handleChildContextFull(rc *residentChild) {
rc.mu.Lock()
rc.state = "contextfull"
rc.mu.Unlock()
a.raiseKernelInterrupt("child/"+rc.id, "kernel",
fmt.Sprintf("内核事件:驻留子 %s 上下文已满contextfull。请【查看】其状态面后决定【压缩】/【回收】/【销毁】。", rc.id))
}
// checkContextFull 判断本 agent 的**积累上下文是否已经装不下窗口**,是则触发一次 contextfull。
//
// ❗判据为什么不能写成"估算拼好的 f.Msgs"`buildMessages` 拿到的 `budget.ContextTokens`
// 本身就是按 `targetUsage = 0.8 × 窗口` 算出来的,时间线**在拼进消息之前就被预算裁过**了。
// 于是 `f.Msgs` 的规模结构上封顶在 ~80% 窗口 —— 对 90% 阈值而言是**永远不成立**的判据
// (写测试时我用一个比系统提示词还小的窗口才勉强越过线,那等于什么都没测)。
//
// 正确的事是"**要被裁了**":拿**未裁剪**的积累上下文(`a.context` 的全部事件)估算,
// 它超过窗口阈值就说明下一轮必须丢事件 ⇒ 这就是 contextfull。
//
// 只对**驻留子**生效(只有它们设了 onContextFull
func (a *Agent) checkContextFull(f *TaskFrame) {
if a.onContextFull == nil || a.ctxFullSignaled {
return
}
max := 0
if a.provider != nil {
max = a.provider.MaxContextTokens()
}
if max <= 0 {
max = defaultMaxContextTokens
}
if a.context == nil {
return
}
// 未裁剪的积累上下文规模。
acc := 0
for _, e := range a.context.Recent(0) {
acc += EstimateTokens(e.Input) + EstimateTokens(e.Response)
}
if float64(acc) < float64(max)*contextFullRatio {
return
}
a.ctxFullSignaled = true
a.onContextFull()
}
func (a *Agent) resetContextFull() { a.ctxFullSignaled = false }
// ---- 子侧inputch 处理表(子持有,父 pull ----
// recordInputchNote 是子**主动写入**本轮 inputch 的处理信息(工具 inputch_note
func (a *Agent) recordInputchNote(text string) {
a.tableMu.Lock()
defer a.tableMu.Unlock()
a.inputchPending = &InputchRecord{
InputCh: a.currentInputch, At: time.Now(), Proactive: true, Text: text,
}
}
// autoRecordInputch 是**系统自动写**兜底:本轮未主动写时,把该轮 inputch 的处理信息写入。
// 保证每一轮必有记录,父不会看到空洞。
func (a *Agent) autoRecordInputch(f *TaskFrame) {
if f == nil {
return
}
a.tableMu.Lock()
defer a.tableMu.Unlock()
if a.inputchPending != nil {
rec := *a.inputchPending
a.inputchPending = nil
rec.InputCh = firstNonEmpty(rec.InputCh, a.currentInputch)
a.inputchTable = append(a.inputchTable, rec)
return
}
text := fmt.Sprintf("轮次完成:输入=%s", truncateStr(f.Input, 80))
if f.Response != "" {
text += ";产出=" + truncateStr(f.Response, 120)
}
a.inputchTable = append(a.inputchTable, InputchRecord{
InputCh: a.currentInputch, At: time.Now(), Proactive: false, Text: text,
})
}
func (a *Agent) inputchTableSnapshot() []InputchRecord {
a.tableMu.Lock()
defer a.tableMu.Unlock()
return append([]InputchRecord(nil), a.inputchTable...)
}
func (a *Agent) clearInputchTable() {
a.tableMu.Lock()
defer a.tableMu.Unlock()
a.inputchTable = nil
a.inputchPending = nil
}
// ---- 子侧主动向父发消息L3 ----
func (a *Agent) notifyParentFrom(text string) string {
if a.notifyParent == nil {
return "本 agent 没有上级,无法发送消息"
}
if strings.TrimSpace(text) == "" {
return "消息内容不能为空"
}
a.notifyParent(text)
return "已发送给主 agent"
}
func firstNonEmpty(a, b string) string {
if a != "" {
return a
}
return b
}
// residentKeepRecent 是压缩时保留的最近事件条数。
const residentKeepRecent = 20
// contextFullRatio 是触发 contextfull 的占比:积累上下文超过窗口的这个比例就报。
// 取 0.9 而不是 1.0:留一点余量,让父 agent 在"下一次必须丢事件"之前就能决策
// (压缩 / 回收 / 销毁),而不是等已经丢了再报。
const contextFullRatio = 0.9
// defaultMaxContextTokens 是 provider 未报告窗口时的兜底(与 ComputeTokenBudget 一致)。
const defaultMaxContextTokens = 32768
func (rc *residentChild) info() ResidentInfo {
rc.mu.Lock()
state, full := rc.state, rc.state == "contextfull"
rc.mu.Unlock()
table := rc.agent.inputchTableSnapshot()
// 子自己的调度器积压per-agent 负载展示用。schedulerStatus 只读快照,
// 每次状态轮询为每个子算一次,成本可忽略(驻留子数量是个位数)。
sc := rc.agent.schedulerStatus()
info := ResidentInfo{
ID: rc.id, State: state, InputChs: append([]string(nil), rc.inputChs...),
AllowedOutputs: append([]string(nil), rc.allowed...),
ContextFull: full, CreatedAt: rc.createdAt, TableSize: len(table),
SchedReady: sc.ReadyQueueDepth,
SchedPending: sc.PendingInterrupts,
SchedStack: sc.SuspendStack,
SchedQueues: sc.InterruptQueues,
// Rounds = 子**已执行的轮次数**(调度器的执行计数,单调不减)。
//
// 此前这里根本没填这个字段 ⇒ 父看到的永远是 `轮次=0`,与"处理表已有 N 条"
// 自相矛盾(现场:子明明处理了两轮,父读到 rounds=0误判成"子没干活")。
// 注意它**不等于** len(table):处理表记的是"当前上下文窗口内"的轮次,
// 压缩会清空§8.3),所以窗口内的条数会被重置,而轮次总数不会。
Rounds: rc.agent.roundsExecuted(),
// OffloadOwned让父分清"我建的子"与"内核临时拉的分诊助手"。
// 父据此决定回收策略(临时的可以空闲时回收,正式的按需保留)。
OffloadOwned: rc.offloadOwned,
}
if len(table) > 0 {
info.Table = table
}
return info
}
// MarshalResidentInfo 便于工具输出(单工具多视图用 JSON 视图)。
func MarshalResidentInfo(info ResidentInfo) string {
b, err := json.Marshal(info)
if err != nil {
return fmt.Sprintf("%+v", info)
}
return string(b)
}