Cyrene_v2/internal/runtime/loop/memory.go

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 全权限环境保持既有的 Inject 行为（候选集已并入
// user/project 共享域）。受限环境（群聊 / 陌生人）只注入
// 非敏感的共享域记忆 —— 用户习惯（ScopeUser）与项目记忆
// （ScopeProject，人人可读）—— 并把该块标记为群聊隔离，
// 使调用方能区分隔离读取与主人的完整读取；主人私有的
// 角色日常记忆（ScopePersona）属敏感项，绝不注入
// （design/03 §5「不泄露主人敏感信息」+ 主人 2026-10-03
// 裁决「用户习惯记忆与不敏感数据：可查、不可写」）。
//
// 同样按敏感项排除的是 L4 意图记录（capacity.RecordIntent 的回写格式，
// 内容是主人的逐句原话）：它虽落在共享的用户习惯域以便 L4 闭环学习，
// 但群聊/陌生人只该看到「不敏感的习惯结论」，不该看到主人原话。
//
// 召回本身包在一个 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)
		block, err := l.retriever.Inject(rctx, personaID, query, 0)
		l.finishMemoryRecallSpan(span, "owner", 0, len(block), err)
		return block, err
	}
	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 := make([]memory.Result, 0, len(results))
	for _, r := range results {
		if !sharedScope(r.Item.Scope) {
			continue
		}
		// L4 意图记录（capacity.RecordIntent 的回写格式）是主人原话的
		// 机器可读形态，属主人敏感信息：共享域可查的是「不敏感的用户
		// 习惯」，主人的逐句原话绝不注入群聊/陌生人上下文
		// （design/03 §5「不泄露主人敏感信息」）。
		if capacity.IsIntentRecord(r.Item.Content) {
			continue
		}
		shared = append(shared, r)
	}
	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
}

// sharedScope 报告 scope 是否为群聊/陌生人可查的非敏感共享域：
// 用户习惯（user，全局共享）与项目记忆（project，人人可读）。
// 角色日常（persona）是主人私有的敏感项，不在其列
// （design/03 §3/§5 的敏感度划分即作用域本身）。
func sharedScope(s memory.Scope) bool {
	return s == memory.ScopeUser || s == memory.ScopeProject
}

// formatMemoryBlock 由检索结果构建前导系统块。非空
// marker 作为该块的前缀，使隔离的群聊读取可与
// 主人的完整读取区分开。
func formatMemoryBlock(results []memory.Result, marker string) string {
	var b strings.Builder
	if marker != "" {
		b.WriteString(marker)
		b.WriteString("：")
	}
	b.WriteString("相关记忆（按相关度 × 衰减排序）：")
	for _, res := range results {
		b.WriteByte('\n')
		b.WriteString("- [")
		b.WriteString(string(res.Item.Kind))
		b.WriteString("] ")
		b.WriteString(res.Item.Content)
	}
	return b.String()
}

// 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 或 store 时为空操作。提取失败与 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 || l.memory == nil {
		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()
}