Cyrene_v2/internal/runtime/loop/run.go

package loop

import (
	"context"
	"fmt"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/capacity"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)

// Run 驱动 agent 循环：每轮组装一份全新的模型可见上下文
// （从当前会话重新注入相关记忆，上下文超出配置预算时
// 可选地压缩更早的轮次），然后运行至多
// MaxRounds 个 step。每个 step 运行模型（配置了 OnStream 时
// 流式），所有被请求的工具调用在每次调用的超时下执行，
// 并以 RoleTool 消息回灌，使模型能继续。
// 每一步都持久化一个 checkpoint，每一步都作为
// state_transition span 追踪（每次调用带一个 llm_call 子 span 与
// tool_call span），且当 turn 落定时记忆 Extractor 在该
// turn 的对话上运行（design/03 §1：LLM extractor 跑在后台
// goroutine 上，因此提取绝不延迟回复），错误记录日志
// 而非致命。
//
// 回合边界发出 turn.start / turn.end 流式帧（design/04 §1.1）：
// start 在任何工作之前，end 经 defer 覆盖 Run 的每个出口
// （最终答案 / 安全网终止 / 预算耗尽 / 步骤错误）。
//
// 当 turn 带着 Result 落定时，OnTurnEnd 钩子对该
// turn 调用一次（绝不按步调用），其情绪标签记录在 Result.Mood 上，
// 使循环外围的分层能推进人格状态并标记持久化的回复。
//
// 它返回最终答案；预算耗尽而无最终答案时，返回带
// StopReasonMaxRounds 的部分 Result。step 错误用 %w 包装，
// 使调用方能以 errors.Is 分类它们。
func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Message) (*Result, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	if l.agent == nil && len(l.agents) == 0 {
		return nil, fmt.Errorf("loop: run: %w", ErrNilAgent)
	}

	// 以 quiet 播种该 Run 的 capacity 累加器，使状态能跨多步
	// 工具循环演进（quiet→awake→full）；nil/被禁用的判定器
	// 使其不被使用。它作用域限于本次 Run，因此并发 run 绝不共享。
	// 会话 id 搭同一个 context，使 capacity_awaken 的发出能
	// 定向到产出它的会话（design/14 §5）。
	ctx = withCapState(ctx, &capStateAccum{state: capacity.StateQuiet})
	// 同时播种该 Run 的「本轮工具允许集」持有者（design/14 §4，裁决
	// 14.4.2）：每轮由 restrictSchemas 发布模型可见的工具面，
	// Dispatcher.Invoke 据此做前置软判。作用域同样限于本次 Run，
	// 子代理 Run 各自播种一份，并发 run 绝不共享。
	ctx = tool.WithRoundAllowSet(ctx)
	ctx = withLoopSession(ctx, sessionID)

	l.emitTurnBoundary(ctx, model.StreamKindTurnStart)
	defer l.emitTurnBoundary(ctx, model.StreamKindTurnEnd)

	// 恢复被中断的 run（design §1 durable execution）：该会话有
	// 可恢复的 checkpoint 时，重建其上下文并从下一步继续，
	// 而不是从全新对话重启。不同的
	// sessionID —— 或已完成 run 留下的 checkpoint —— 落到
	// 全新开始（LoadLatest 未命中，或 stop reason 不可恢复）。
	msgs, startStep, _ := l.loadCheckpoint(ctx, sessionID, messages)
	msgs = l.resolveImages(ctx, msgs)
	res := &Result{}
	// 分支 Run（子代理）把其 span 并入父会话的实时
	// 图；只有所属的 Run 开始与结束轮次。没有它，子代理
	// 会重置前端唯一的 flowGraph，并让父级自身的节点出现断档。
	if _, isBranch := flowBranch(ctx); l.flow != nil && !isBranch {
		l.flow.StartRound(sessionID, lastUserText(msgs))
		defer func() { l.flow.EndRound(sessionID, res.Text) }()
	}
	return l.driveTurn(ctx, sessionID, msgs, startStep, res)
}

// driveTurn 运行至多 maxRounds 个 step 的主循环：每步组装上下文、
// 调用模型、执行被请求的工具并回灌。它在任何一种落定路径上返回
// 已 settle 的 Result；step 或工具批次的错误以 %w 包装返回。
func (l *Loop) driveTurn(ctx context.Context, sessionID string, msgs []model.Message, startStep int, res *Result) (*Result, error) {
	toolState := &toolLoopState{prevFP: make(map[string]bool)}
	step := startStep
	// 被用户中断（design/15 §3）的回合不留下「可恢复的在途态」：design/15
	// 要求中断后**立即**按最新上下文处理下一条消息，而一个被放弃的回合若把
	// 可恢复 checkpoint 留在该会话上，下一条消息会被它顶掉——继承旧窗口、
	// 丢掉刚发出的那条，上下文既不最新也不完整（裁决 13.4.5）。因此取消后
	// 把该会话的快照落定为不可恢复态（快照本身保留，仍可查询）。
	//
	// 只在回合**未落定**时标记（落定路径已写入终态快照，其 stop reason 本就
	// 不可恢复），因此这里不会改写一个已交付答案的回合。
	defer func() {
		if ctx.Err() != nil && res.StopReason == "" {
			l.abandonCanceledCheckpoint(ctx, sessionID, step, msgs)
		}
	}()
	for ; step <= l.maxRounds; step++ {
		if err := ctx.Err(); err != nil {
			return nil, fmt.Errorf("loop: run: step %d: %w", step, err)
		}

		// 会话超出上下文预算时压缩更早的轮次，
		// 然后为本轮组装模型可见上下文（全新记忆
		// 注入 + transform 钩子），使每一步都看到最新上下文。
		msgs = l.compactIfNeeded(ctx, msgs)
		view := l.assembleContext(ctx, msgs, step)

		rc, roundSpan := l.span(ctx, "loop.step", trace.SpanKindStateTransition)
		outcome, err := l.step(rc, sessionID, view)
		if err != nil {
			l.spanProp(roundSpan, "error", err.Error())
			l.finishSpan(roundSpan, trace.SpanStatusError)
			return nil, fmt.Errorf("loop: run: step %d: %w", step, err)
		}
		l.spanProp(roundSpan, "step", step)
		l.spanProp(roundSpan, "stop_reason", outcome.stopReason)
		l.finishSpan(roundSpan, trace.SpanStatusOK)

		res.Steps = step
		if outcome.final {
			res.Text = outcome.text
			res.StopReason = outcome.stopReason
			return l.settleAndReturn(ctx, sessionID, msgs, step, outcome.stopReason, res)
		}

		batch, err := l.runToolCalls(rc, msgs, outcome.toolCalls, toolState)
		if err != nil {
			return nil, fmt.Errorf("loop: run: step %d: tools: %w", step, err)
		}
		msgs = batch.messages
		res.ToolCallsExecuted += batch.executed
		l.saveCheckpoint(ctx, sessionID, step, msgs, agent.StopReasonToolCalls)

		// 关键工具循环安全网（或终止钩子）不等真正的最终答案
		// 就停止本轮。
		if batch.terminate {
			res.Text = criticalToolLoopTerminalReason
			res.StopReason = StopReasonToolLoopTerminated
			return l.settleAndReturn(ctx, sessionID, msgs, step, StopReasonToolLoopTerminated, res)
		}
	}

	// 预算耗尽而无最终答案。
	res.Steps = l.maxRounds
	res.StopReason = StopReasonMaxRounds
	return l.settleAndReturn(ctx, sessionID, msgs, l.maxRounds, StopReasonMaxRounds, res)
}

// settleAndReturn 落定一个已得出 Result 的回合：以该路径的
// stop reason 持久化终态 checkpoint、触发后台记忆提取、
// 上报 turn 钩子（情绪标签写回 Result.Mood），然后把结果交还调用方。
func (l *Loop) settleAndReturn(ctx context.Context, sessionID string, msgs []model.Message, step int, stopReason string, res *Result) (*Result, error) {
	l.saveCheckpoint(ctx, sessionID, step, msgs, stopReason)
	l.extractMemoryAsync(ctx, msgs, step)
	l.settleTurn(ctx, sessionID, msgs, res)
	return res, nil
}

// StopReasonCanceled 标记一个因用户中断（design/15 §3「停下/算了/换个话题」）
// 而放弃的回合。它不是可恢复的在途态：design/15 要求中断后立即按**最新**
// 上下文处理下一条消息，因此这种快照绝不能被下一条消息续接
// （见 driveTurn 与 abandonCanceledCheckpoint）。
const StopReasonCanceled = "canceled"

// abandonCanceledCheckpoint 把一个被取消回合的 checkpoint 落定为不可恢复态
// （stop reason = StopReasonCanceled），使该会话的下一条消息从它自己的最新
// 上下文开始，而不是被放弃回合的旧窗口顶掉（裁决 13.4.5：上下文一致性 /
// 完整性 / 最新性）。快照本身保留在 store 里，仍可查询。
//
// 写入必须脱离已取消的 ctx（否则 ExecContext 必然立即失败），因此这里用
// WithoutCancel 派生背景上下文，保留原 ctx 的值而丢弃取消信号。失败只记日志，
// 绝不让这个收尾动作把中断路径变成错误。
func (l *Loop) abandonCanceledCheckpoint(ctx context.Context, sessionID string, step int, msgs []model.Message) {
	l.log.Info("checkpoint abandoned", "session", sessionID, "step", step, "stop_reason", StopReasonCanceled)
	l.saveCheckpoint(context.WithoutCancel(ctx), sessionID, step, msgs, StopReasonCanceled)
}