package loop import ( "context" "strings" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/capacity" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) // memoryExtractTimeout 限制单次后台提取。它只管辖 // 脱离出来的提取 context,因此挂起的模型调用不会让一个 goroutine // 泄露到进程生命周期结束。 const memoryExtractTimeout = 60 * time.Second // asyncExtractor 是循环脱离到后台 goroutine 的提取器能力: // memory.LLMExtractor 经 ExtractAsync 实现它(design/03 §1 // LLM 异步自动提取),返回一个恰好投递一个结果的 channel。 // 确定性提取器(memory.ManualExtractor)不实现它,留在 // 同步路径上。 type asyncExtractor interface { ExtractAsync(ctx context.Context, conversation []model.Message) (<-chan memory.AsyncResult, error) } // injectMemory 检索该角色的相关长期记忆并把它们作为前导 // 系统消息前置,使模型能把回答建立在召回出的上下文之上 // (design §6 / §2)。它经 assembleContext 每轮重新运行,因此 // 检索反映最新对话而非首轮快照。未配置 // retriever/persona、检索失败、或没找到相关内容时, // 它原样返回输入消息。 // // ctx 上的会话环境为检索设门:owner 全权限环境保留完整 // 召回(行为不变;候选集已并入 user/project 共享域, // design/03 §4);受限环境(群聊 / 陌生人, // design/02 §3, design/11 §2, design/03 §5)是只读隔离 —— // 只注入非敏感的共享记忆(用户习惯 + 项目),主人私有的 // 角色日常记忆绝不泄露给群聊或陌生人。 func (l *Loop) injectMemory(ctx context.Context, messages []model.Message) []model.Message { block := l.memoryBlock(ctx, messages) if block == "" { return messages } out := make([]model.Message, 0, len(messages)+1) out = append(out, model.TextMessage(model.RoleSystem, block)) out = append(out, messages...) return out } // memoryBlock 渲染该轮的记忆块;无可注入内容时返回 ""(无 // retriever/persona、检索出错、或没有相关记忆)。它 // 从 injectMemory 中拆出,因为组装会在记忆块与对话之间 // 放置其他前导块(design/22 §7-5 的角色上下文), // 因此它需要块本身而不是已前置好的切片。 func (l *Loop) memoryBlock(ctx context.Context, messages []model.Message) string { personaID := l.personaFor(ctx) if l.retriever == nil || personaID == "" { return "" } env := sessionEnvFrom(ctx) block, err := l.retrieveMemoryBlock(ctx, env, personaID, queryFromMessages(messages)) if err != nil { l.log.Warn("memory inject", "error", err, "env", env.ActorEnv) return "" } return block } // retrieveMemoryBlock 返回 env 与该 turn 角色的记忆注入块。 // owner 全权限环境保持既有的完整召回(候选集已并入 // user/project 共享域)。受限环境(群聊 / 陌生人)只注入 // 非敏感的共享域记忆 —— 用户习惯(ScopeUser)与项目记忆 // (ScopeProject,人人可读)—— 并把该块标记为群聊隔离, // 使调用方能区分隔离读取与主人的完整读取;主人私有的 // 角色日常记忆(ScopePersona)属敏感项,绝不注入 // (design/03 §5「不泄露主人敏感信息」+ 主人 2026-10-03 // 裁决「用户习惯记忆与不敏感数据:可查、不可写」)。 // // 隔离判据单一实现于 FilterSharedVisible:本条注入路径与模型可达的 // knowledge 检索工具共用它,绝不各造一套过滤。 // // 召回本身包在一个 memory span 里(裁决 11.4.2,design/19 §1 // 旁路节点):span 只记元数据(作用域/命中条数/注入块字节数/耗时), // 绝不记召回了什么内容。 func (l *Loop) retrieveMemoryBlock(ctx context.Context, env SessionEnv, personaID, query string) (string, error) { if !env.Restricted() { rctx, span := l.span(ctx, spanNameMemoryRecall, trace.SpanKindMemory) results, err := l.retriever.Retrieve(rctx, personaID, query, 0) if err != nil { l.finishMemoryRecallSpan(span, "owner", 0, 0, err) return "", err } if len(results) == 0 { l.finishMemoryRecallSpan(span, "owner", 0, 0, nil) return "", nil } block := formatMemoryBlock(results, "") l.finishMemoryRecallSpan(span, "owner", len(results), len(block), nil) return block, nil } rctx, span := l.span(ctx, spanNameMemoryRecall, trace.SpanKindMemory) results, err := l.retriever.Retrieve(rctx, personaID, query, 0) if err != nil { l.finishMemoryRecallSpan(span, "restricted", 0, 0, err) return "", err } shared := FilterSharedVisible(results) if len(shared) == 0 { l.finishMemoryRecallSpan(span, "restricted", 0, 0, nil) return "", nil } block := formatMemoryBlock(shared, "群聊隔离") l.finishMemoryRecallSpan(span, "restricted", len(shared), len(block), nil) return block, nil } // assembleContext 为一轮构建模型可见的消息列表(design // §1.3 每轮 assemble,循环中与 OpenClaw 的 transformContext 等价之物):它 // 重新注入该角色的相关长期记忆 —— 检索每轮都从 // 当前对话重新运行,不从首轮缓存,因此 // 支持跨轮刷新 —— 在存在时并入每会话的工作记忆摘要, // 渲染对话的角色/发言人归属(design/22 // §7-5),最后应用可选的 TransformContext 钩子,使调用方能 // 投影/过滤模型实际所见。什么都没配置时 // 它原样返回输入。 // // 既有顺序被精确保留:前导块依次是每轮的 // 时间、工作记忆摘要与记忆块;角色 // 声明紧接其后(就在它所描述的对话之前),而发言人 // 边界说明被插入对话本身。没有发言人标签的对话 // 完全不渲染归属,因此每条单角色路径都保持多角色支持 // 之前的提示词。唯一的头部新增是「人格知识」块(本批 P0-10 // 收窄,见下),它排在最前 —— 即紧接该轮 agent 的指令消息之后, // 与它过去作为指令尾段时的相对位置一致。 // // 「人格知识」(主人身份说明书)**按会话**注入(Options.PersonaKnowledgeFor): // 解析器拿该轮的 SessionEnv 决定返回什么,群聊 / 陌生人会话返回空串即 // 完全不出现(design/11 §0 红线:不入主会话上下文、不泄露主人敏感信息)。 // 为什么不能像过去那样静态拼进 system 指令:指令在装配期定死,静态注入 // 会让它进入该角色的每一个会话(含群聊与陌生人私聊)并随请求发给模型 // 供应商。未接线该解析器时逐字节等于接线前。 func (l *Loop) assembleContext(ctx context.Context, messages []model.Message, round int) []model.Message { var leading []model.Message if l.personaKnowledgeFor != nil { if block := l.personaKnowledgeFor(sessionEnvFrom(ctx)); block != "" { leading = append(leading, model.TextMessage(model.RoleSystem, block)) } } // 每轮「当前时间」,使模型自然知道墙上时钟(而非首轮 // 快照)。以已配置的时区为条件,使测试与无时区的运行 // 保持其精确的前导消息数。它位于工作记忆 / // 记忆块之前,且仍可被 transformContext 投影。 if l.timeLocation != nil { leading = append(leading, l.currentTimeBlock()) } if l.working != nil { // 读**本会话**的窗口(回合落定时由 recordWorkingMemory 写入): // 进程内只有一个 WorkingMemory,按会话读才不至于把别的会话 // (尤其是群聊/陌生人)的工作记忆混进主人的上下文。该会话 // 尚无窗口(如首轮之前)时摘要为空,前导块自然不出现。 if s := l.working.SummarizeSession(loopSession(ctx)); s != "" { leading = append(leading, model.TextMessage(model.RoleSystem, "工作记忆: "+s)) } } if block := l.memoryBlock(ctx, messages); block != "" { leading = append(leading, model.TextMessage(model.RoleSystem, block)) } decl, ok, view := l.attributeConversation(ctx, messages) if ok { leading = append(leading, decl) } out := append(leading, view...) if l.transformContext != nil { out = l.transformContext(ctx, out, round) } return out } // compactIfNeeded 在估算上下文超出配置的 MaxContextTokens // 时压缩对话。压缩被关闭(MaxContextTokens 非正)、 // compactor 为 nil、估算在预算内、或 compactor 返回错误时 // 它是空操作 —— 压缩失败会被记录日志并保持对话 // 原样,而不是让 run 失败。 func (l *Loop) compactIfNeeded(ctx context.Context, messages []model.Message) []model.Message { if l.maxContextTokens <= 0 || l.compactor == nil { return messages } compacted, err := l.compactor.Compact(ctx, messages, l.maxContextTokens) if err != nil { l.log.Warn("context compact", "error", err, "tokens", l.maxContextTokens) return messages } if compacted == nil { return messages } return compacted } // extractMemory 在当前对话上运行提取钩子(design §6) // 并持久化导出的条目。错误会被记录日志,绝不致命,因此 // 一次糟糕的提取无法中止 run。 // // 在受限上下文(群聊 / 陌生人,!MemWritable)中提取被 // 完全跳过:该对话可以只读地读取共享记忆,但绝不 // 持久化任何可能泄露主人敏感信息的东西(design/02 // §3, design/11 §2)。 func (l *Loop) extractMemory(ctx context.Context, messages []model.Message, step int) { env := sessionEnvFrom(ctx) if env.Restricted() { l.log.Debug("memory extract skipped: read-only context", "step", step, "env", env.ActorEnv) return } if l.extractor == nil || l.memory == nil { return } items, err := l.extractor.Extract(ctx, messages) if err != nil { l.log.Warn("memory extract", "step", step, "error", err) return } l.persistMemories(ctx, items, step) } // extractMemoryAsync 在 turn 边界触发提取钩子而不阻塞 // 回复(design/03 §1: 每轮对话后后台异步抽取)。配置的 // 提取器能异步提取(LLM extractor)时,对话被交给 // 后台 goroutine,本调用立即返回,因此慢速 // 提取模型绝不延迟响应。确定性提取器 // (ManualExtractor)不实现 asyncExtractor,就地运行,这 // 使其廉价的 "记住 X" 路径保持可观测且不变。 // // 与 extractMemory 相同的门仍然适用:受限对话绝不 // 被提取(只读隔离,design/02 §3 / design/11 §2)。未配置 // extractor 时为空操作;**配置了 extractor 却没有 store** 时 // 提取出的条目会被丢弃,因此该情形带来源记一条告警(l.log 已由 // sourcedLog 标记 source=runtime/loop),绝不静默吞掉 // (审计 2026-10-05 §7 静默吞点)。提取失败与 store 失败同样 // 带循环的来源记录日志,绝不向上传播。 // // 它同时是 loop 侧**回合落定**的记忆钩子入口(run 路径每个已落定 // 的 turn 调用它一次):无论提取器是否配置,都在此先落下工作记忆 // 与 L4 判定回写(见 recordTurnMemory)。两者各自带门(工作记忆写 // 会话窗口、L4 只写 owner 会话),因此不依赖这里的 extractor 门。 func (l *Loop) extractMemoryAsync(ctx context.Context, messages []model.Message, step int) { l.recordTurnMemory(ctx, messages, step) env := sessionEnvFrom(ctx) if env.Restricted() { l.log.Debug("memory extract skipped: read-only context", "step", step, "env", env.ActorEnv) return } if l.extractor == nil { return } if l.memory == nil { l.log.Warn("memory extract skipped: no memory store configured; extracted memories would be dropped", "step", step) return } async, ok := l.extractor.(asyncExtractor) if !ok { l.extractMemory(ctx, messages, step) return } // 该 turn 的 context 随请求消亡,而提取刻意比它活得 // 更久,因此 goroutine 拿到一份脱离的 context 副本(值 —— // 其中包括会话环境 —— 被保留),并处于自己的超时之下。 dctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), memoryExtractTimeout) // 随着该 turn 继续,调用方会不断向 messages 追加,因此 // goroutine 读取一份私有快照,而不是在共享切片上竞争。 snapshot := append([]model.Message(nil), messages...) results, err := async.ExtractAsync(dctx, snapshot) if err != nil { cancel() l.log.Warn("memory extract dispatch", "step", step, "error", err) return } go func() { defer cancel() res, ok := <-results if !ok { return } if res.Err != nil { l.log.Warn("memory extract", "step", step, "error", res.Err) return } l.persistMemories(dctx, res.Items, step) }() } // persistMemories 把提取出的条目写入记忆 store,为每个条目 // 打上产出该 turn 的角色。该赋值刻意无条件: // extractor 刻意不感知角色(它不再持有一个在构造期 // 内嵌的人格,cmd/cyrene/extractor.go),否则多角色 // 进程会把每条记忆都归到一个角色下。角色为空(完全未配置 // persona)时条目保持 extractor 产出时的样子。 // // 溯源同样在此盖真实 id(裁决 04.4.3,design/03 §2): // source_conversation 由本持久化点从运行上下文取 Run 的 // session id 与 gateway 挂载的 turn id 结构化编码 // (memory.FormatSourceConversation),无条件覆盖 extractor // 产出值 —— 模型自由文本绝无机会落库为来源。store // 失败会被记录日志,绝不致命,因此一条被拒的条目无法中止一个 turn。 // // 整批写入包在一个 memory span 里(裁决 11.4.2,design/19 §1 旁路节点): // 只记写入/失败条数与耗时,绝不记写入了什么内容;没有条目时不产生节点 // (span 绝不宣告未运行的工作)。 func (l *Loop) persistMemories(ctx context.Context, items []memory.MemoryItem, step int) { if len(items) == 0 { return } personaID := l.personaFor(ctx) source := memory.FormatSourceConversation(loopSession(ctx), loopTurn(ctx)) wctx, span := l.span(ctx, spanNameMemoryWrite, trace.SpanKindMemory) written, failed := 0, 0 for _, it := range items { if personaID != "" { it.PersonaID = personaID } it.SourceConversation = source if _, err := l.memory.Add(wctx, it); err != nil { failed++ l.log.Warn("memory store", "step", step, "error", err) continue } written++ } l.finishMemoryWriteSpan(span, written, failed, nil) } // recordTurnMemory 是回合落定时 loop 拥有的记忆钩子:把该会话的当前 // 对话窗口写进工作记忆(三层记忆的层①),并把该轮的 L4 判定回写进 // 记忆(design/14 §2.2 L4 / §2.4 闭环学习)。它由 extractMemoryAsync // 在每个已落定的 turn 上调用一次 —— 落点是 run 路径既有的 turn 结束 // 钩子,因此 sub-agent 与主会话走同一条路径。 // // 两个钩子各自带门(见 recordWorkingMemory / recordIntent),因此 // 未配置 capacity、未配置检索器、或处于群聊/陌生人环境时它们 // 安静跳过,绝不因为「钩子跑了」而多写一行记忆。 func (l *Loop) recordTurnMemory(ctx context.Context, messages []model.Message, step int) { l.recordWorkingMemory(ctx, messages) l.recordIntent(ctx, messages, step) } // recordWorkingMemory 把该会话的对话窗口记进工作记忆(design/03 §0 // 层①「当前会话的对话窗口(滑动窗口 + 摘要)」),使 assembleContext // 的「工作记忆: …」前导块与 Summarize 真的有内容 —— 接线之前生产 // 代码从不调用 Add,窗口恒空、摘要恒 ""(审计 P0-4)。 // // 两道门: // - 只写**主人会话**(MemWritable):群聊/陌生人的对话不入工作记忆 // (design/02 §3、design/11 §2 的只读隔离)。 // - 只写带会话 id 的窗口:进程内只有一个 WorkingMemory 实例,无 // 会话键的写入会污染别的会话的读点。 // // 传入的是该回合的完整对话(含 checkpoint 还原出的历史),窗口因此 // 就是「最近 N 条」滑窗;会话恢复时同一条路径把历史回填进窗口, // 而不是从空白重启(design §1 durable execution)。 func (l *Loop) recordWorkingMemory(ctx context.Context, messages []model.Message) { if l.working == nil || len(messages) == 0 { return } if !sessionEnvFrom(ctx).MemWritable { return } sessionID := loopSession(ctx) if sessionID == "" { return } l.working.RecordSession(sessionID, messages) } // recordIntent 把该轮的 L4 判定写回记忆(design/14 §2.2 L4「写入记忆… // 让意图理解越用越准」、§2.4 闭环学习)。接线之前 capacity.RecordIntent // 在生产代码里零调用,判定层因此只读不写,「越用越准」永远不成立 // (审计 P0-17)。 // // 红线:**只写 owner 会话**。群聊/陌生人有独立的记忆域且只读不写 // (design/02 §3、design/11 §2、design/03 §5),子代理的内部回合也 // 不是用户本人的表达习惯(经 flowBranch 识别),两者都不写。 // 判定层禁用、无角色、缺 store 时为空操作;判定为非任务时 // capacity.RecordIntent 返回 "",跳过写入(不堆噪声记忆)。 // // 条目落进**用户习惯域**(ScopeUser,design/03 §3「用户习惯记忆全局 // 共享」):L4 学的正是主人的任务表达习惯,写入与读取因此在同一个 // 作用域里闭环(capacity L4 只采信 user/persona/project 域的召回)。 func (l *Loop) recordIntent(ctx context.Context, messages []model.Message, step int) { if l.capacity == nil || !l.capacity.Enabled() || l.memory == nil || l.retriever == nil { return } if !sessionEnvFrom(ctx).MemWritable { return } if _, isBranch := flowBranch(ctx); isBranch { return } personaID := l.personaFor(ctx) if personaID == "" { return } judgment := l.capacity.Judge(ctx, capacity.Input{ Messages: capacityTexts(messages), Awakening: l.awakeningFor(ctx), Recall: l.capacityRecall(ctx, messages), }) content := l.capacity.RecordIntent(judgment) if content == "" { return } item := memory.MemoryItem{ PersonaID: personaID, Scope: memory.ScopeUser, Kind: memory.KindRelationship, Content: content, Confidence: judgment.Confidence, SourceConversation: memory.FormatSourceConversation(loopSession(ctx), loopTurn(ctx)), } if _, err := l.memory.Add(ctx, item); err != nil { l.log.Warn("memory: L4 intent writeback failed", "step", step, "error", err) return } l.log.Debug("memory: L4 intent writeback stored", "step", step, "task_kind", judgment.TaskKind) } // queryFromMessages 从对话导出检索查询:最后一个用户 // turn 的文本,回退到最后一条消息的文本,再回退到 ""。 func queryFromMessages(messages []model.Message) string { for i := len(messages) - 1; i >= 0; i-- { if messages[i].Role == model.RoleUser { return messageText(messages[i]) } } if n := len(messages); n > 0 { return messageText(messages[n-1]) } return "" } // messageText 返回一条消息各文本部件的拼接纯文本载荷, // 这是用于构建检索查询的共享视图。 func messageText(m model.Message) string { var b strings.Builder for _, part := range m.Content { if part.Type != model.ContentTypeText { continue } if b.Len() > 0 { b.WriteByte(' ') } b.WriteString(part.Text) } return b.String() }