238 lines
13 KiB
Go
238 lines
13 KiB
Go
package loop
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strings"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
)
|
||
|
||
// 本文件实现「回合内转向注入」（design/15 §2.0/§2.4，裁决 13.4.5 目标态）。
|
||
//
|
||
// 不变量按 design/15 §2.0 实现，语义不在循环里自创：
|
||
//
|
||
// - **I1（同会话在途回合数恒为 1）**：转向只在**同一回合的边界**注入，绝不取消
|
||
// 当前回合、也绝不新开第二个回合。循环对这一点的保证是结构性的——它没有任何
|
||
// 「起新回合」的分支，注入只是把消息并进当前回合下一次模型调用的视图。
|
||
// - **I2（消息恰好被处理一次）**：注入＝已消费。循环看不到消息 id
|
||
// （model.Message 没有 id 字段），因此「按 created_at + id 去重」与「从待处理
|
||
// 队列移除」都归 SteerSource 的实现方（网关会话编排器，它同时持有转录与
|
||
// pending 队列）。循环只保证同一个边界只问一次、按返回顺序追加。
|
||
// - **I3（转向不取消已发起的工具）**：注入点只在**回合内边界**——一次模型步结束
|
||
// 与它那一批工具调用结束**之后**、下一次模型调用**之前**。此刻没有任何工具在
|
||
// 飞，因此注入不可能取消任何已发起的工具；正在执行的工具永远跑完（与 Cline
|
||
// PR #13958「只打断在飞的模型请求、工具与整轮继续」同口径）。
|
||
//
|
||
// 为什么注入只进**本轮视图**而不写回运行窗口：写回会让注入内容进入 checkpoint
|
||
// 快照与后台记忆提取的输入面（两者都读该回合的 msgs），而它们本应只看到该回合真正
|
||
// 的输入与工具结果。注入在每一步重新并入视图，因此对模型而言它仍是**同一个窗口**
|
||
// （一致性＝同一窗口，design/15 §2.4），只是不落盘、不出现在记忆面。
|
||
|
||
// SteerRequest 是循环在回合内边界向 SteerSource 发起的一次「重读转录」请求。
|
||
type SteerRequest struct {
|
||
// SessionID 是本次 Run 驱动的会话。
|
||
SessionID string
|
||
// Step 是**即将开始**的模型步序号（从 1 开始）：被返回的消息会出现在该步的
|
||
// 模型可见视图里，因此它就是网关 `steer_ack` 帧上报的 `atStep`。
|
||
Step int
|
||
}
|
||
|
||
// SteerSource 由调用方（网关会话编排器）提供，在指定的回合内边界重读该会话转录，
|
||
// 返回「本回合开始后新到、且尚未被本回合吃进」的用户消息，供循环追加进本轮视图。
|
||
// 返回空切片表示本次边界没有可注入的消息，是正常情形而非错误。
|
||
//
|
||
// 实现方负责去重与「已消费」口径（见文件头 I2）：循环按返回顺序原样追加，不做也
|
||
// 无法做 id 去重。它必须是非阻塞的——它在 Run 路径上被调用；网络/存储读取有界，
|
||
// 读失败时返回空（该消息留在待处理队列，照旧作为自己的回合被回答，绝不静默丢失）。
|
||
type SteerSource interface {
|
||
Steer(ctx context.Context, req SteerRequest) []model.Message
|
||
}
|
||
|
||
// steerSourceKey 是回合内转向源的 context key。
|
||
type steerSourceKey struct{}
|
||
|
||
// WithSteerSource 把转向源挂到 ctx 上，使本次 Run 的每个回合内边界都能重读转录并
|
||
// 注入新到的用户消息（design/15 §2.4）。调用方（网关编排器）在**每个回合启动之前**
|
||
// 挂一份新的源：转向游标是按回合的状态，跨回合复用会让上一个回合已吃进的消息被
|
||
// 过滤掉或重复注入。
|
||
//
|
||
// nil 源是空操作，ctx 原样返回——未接线的调用方（子代理、后台思考、未接网关的循环）
|
||
// 行为逐字不变。
|
||
func WithSteerSource(ctx context.Context, src SteerSource) context.Context {
|
||
if ctx == nil || src == nil {
|
||
return ctx
|
||
}
|
||
return context.WithValue(ctx, steerSourceKey{}, src)
|
||
}
|
||
|
||
// SteerSourceFrom 返回 ctx 上挂的转向源，未挂载时为 nil。它是 WithSteerSource 的
|
||
// 读侧，供拿到 context 的调用方（网关测试、断言注入边界的实现）观察本次 Run 的
|
||
// 注入能力；nil 表示本次运行不做任何转向注入。
|
||
func SteerSourceFrom(ctx context.Context) SteerSource {
|
||
if ctx == nil {
|
||
return nil
|
||
}
|
||
src, _ := ctx.Value(steerSourceKey{}).(SteerSource)
|
||
return src
|
||
}
|
||
|
||
// steerAtBoundary 在一个回合内边界重读该会话转录，并把新到的用户消息合并进本回合
|
||
// 已注入列表。未接线转向源、ctx 已取消、或本次边界无新消息时，它原样返回 steered
|
||
// （无新消息是常态：工具批次之间通常没有用户插话）。
|
||
//
|
||
// 它绝不返回错误：转向是增益，任何一个边界的读取失败都不该让用户已经在跑的回合
|
||
// 失败——失败时返回空（见 SteerSource 契约），该消息留在待处理队列里等下一个回合。
|
||
func (l *Loop) steerAtBoundary(ctx context.Context, sessionID string, step int, steered []model.Message) []model.Message {
|
||
src := SteerSourceFrom(ctx)
|
||
if src == nil || ctx.Err() != nil {
|
||
return steered
|
||
}
|
||
msgs := src.Steer(ctx, SteerRequest{SessionID: sessionID, Step: step})
|
||
if len(msgs) == 0 {
|
||
return steered
|
||
}
|
||
// 注入的图片消息走与回合开始处同一条视觉改写（design/06 §0）：否则一条中途
|
||
// 注入的图片会绕过 resolver，直接落到不支持视觉的主模型面前。
|
||
msgs = l.resolveImages(ctx, msgs)
|
||
// 链路图如实标记「本轮输入被中途追加过」（design/15 §5）。子代理 Run 不拥有
|
||
// 自己的图（它的 span 并入派生它的会话），因此分支 Run 不打这个标记。
|
||
if _, isBranch := flowBranch(ctx); !isBranch {
|
||
l.flow.MarkInputSteered(sessionID)
|
||
}
|
||
l.log.Info("steer injected", "session", sessionID, "step", step, "messages", len(msgs))
|
||
return append(steered, msgs...)
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// 并行路进行态摘要（design/15 §2.5.6，主人 2026-10-05 裁决 B 的落地）
|
||
// ---------------------------------------------------------------------------
|
||
//
|
||
// 背景（真实事故）：主人连发三条消息（两条来自安卓语音助手设备 `voiceassist:cyrene:v1`、
|
||
// 一条微信图片），后端真并发跑了三路；三路因为**互相看不见**，各自把同一句
|
||
// 「写 2000 字关于海洋」当成自己的题作答（三篇作文），其中一路还把别路的作文答进了
|
||
// 另一个客户端（跨端串话）。§2.5.6 的「进行态摘要」就是为了让每一路**结构化地**
|
||
// 知道「本会话此刻另有一路在跑」，从而各答各的。
|
||
//
|
||
// 它只含**元数据**（对方消息 id / 步数 / 已调用工具名 / 排队条数），结构里没有任何承载
|
||
// 对方正文的字段 —— 半句话会诱导本路去「补全」或「抢答」，也无法诚实地标注
|
||
// who said what（§2.5.6「不做的事」①）。它作为**独立的 system 消息**注入本轮视图，
|
||
// **绝不拼进任何 user 正文**（项目红线：用户内容只进 user 角色消息）；这段文本也
|
||
// **不**承载任何身份 / 权限信息（身份识别靠 channel 层 actor 元数据 + 策略引擎）。
|
||
//
|
||
// 注入点与口径与 steer 注入**完全相同**（§2.5.6 明文要求）：在 driveTurn 每步的
|
||
// assembleContext 之后追加进**本轮视图**，不写回 msgs ⇒ 不进 checkpoint 快照、
|
||
// 不进后台记忆提取的输入面。
|
||
|
||
// design/15 §2.5.6 定死的措辞（逐字，含标点）。头尾两句在任何字段为空时也照发 ——
|
||
// 「另有一路在跑、你独立作答」本身就是这条提示要传递的核心事实，元数据只是增益。
|
||
const (
|
||
parallelNoticeHeader = "[并行路提示] 该会话当前另有一路正在进行中（不打断它）。"
|
||
parallelNoticeTail = "你这一路独立作答；不要复述或改写对方可能给出的内容。"
|
||
)
|
||
|
||
// ParallelNotice 是本会话「另有一路正在进行」的**结构化**进行态摘要（design/15 §2.5.6）。
|
||
//
|
||
// 它由**网关**（会话编排器）在起第二路时按该路上游的实时状态构造，经
|
||
// WithParallelNotice 挂到该路的 runCtx 上；循环在每一步把它渲染成一条独立的
|
||
// system 消息（appendParallelNotice）。**空字段对应的那一段不渲染**，因此哪怕一条
|
||
// 元数据都没有，挂载这个值本身仍然是有意义的信号（头尾两句照发）。
|
||
//
|
||
// 结构里**没有**任何能承载对方半成品正文的字段 —— 这条红线是**结构性**的：
|
||
// 想拼也拼不进来。
|
||
type ParallelNotice struct {
|
||
// OtherMessageID 是另一路（或上一条）用户消息的服务端 id，作为「上一条消息 id」
|
||
// 渲染。为空时该段不渲染。
|
||
OtherMessageID string
|
||
// Step 是另一路**已进行到**的步序号（1 基）。<= 0 时该段不渲染。
|
||
Step int
|
||
// ToolNames 是另一路已经调用过的工具名，按调用顺序、已去重。为空时该段不渲染。
|
||
ToolNames []string
|
||
// Queued 是该会话仍在 pending、尚未被任何一路接手的消息条数。<= 0 时该段不渲染。
|
||
Queued int
|
||
}
|
||
|
||
// parallelNoticeKey 是并行路进行态摘要的 context key。
|
||
type parallelNoticeKey struct{}
|
||
|
||
// WithParallelNotice 把进行态摘要挂到 ctx 上，使本次 Run 每一步的模型可见视图都带上
|
||
// 一条独立的 system 提示（design/15 §2.5.6）。调用方（网关会话编排器）在**起第二路
|
||
// 之前**挂一份：摘要是该路起点的一份**快照**，中途不再改写（I4：两路之间不做运行中
|
||
// 的实时可见）。nil ctx 是空操作，ctx 原样返回 —— 未挂该键的运行（子代理 / 后台思考 /
|
||
// 未接网关的循环 / 单路）**行为逐字不变**。
|
||
func WithParallelNotice(ctx context.Context, notice ParallelNotice) context.Context {
|
||
if ctx == nil {
|
||
return ctx
|
||
}
|
||
return context.WithValue(ctx, parallelNoticeKey{}, notice)
|
||
}
|
||
|
||
// ParallelNoticeFrom 返回 ctx 上挂的进行态摘要，ok 报告是否挂载。它是
|
||
// WithParallelNotice 的读侧，供注入点与测试观察本路的提示能力。
|
||
//
|
||
// 它刻意返回 ok 而不只返回值：零值摘要也是「另有一路在跑」的有效信号（头尾两句本身
|
||
// 就是核心事实），因此「未挂载」不能与「挂了但没元数据」混为一谈。
|
||
func ParallelNoticeFrom(ctx context.Context) (ParallelNotice, bool) {
|
||
if ctx == nil {
|
||
return ParallelNotice{}, false
|
||
}
|
||
notice, ok := ctx.Value(parallelNoticeKey{}).(ParallelNotice)
|
||
return notice, ok
|
||
}
|
||
|
||
// parallelNoticeText 把进行态摘要渲染成 design/15 §2.5.6 定死的那段文本。措辞（含
|
||
// 标点与全角分隔）逐字取自设计；只有**有数据的字段**才出现在事实行里，使空字段不
|
||
// 产生空段（如「已进行到第 0 步」这种无信息量的噪声）。
|
||
func parallelNoticeText(notice ParallelNotice) string {
|
||
var b strings.Builder
|
||
b.WriteString(parallelNoticeHeader)
|
||
if facts := parallelNoticeFacts(notice); facts != "" {
|
||
b.WriteString("\n")
|
||
b.WriteString(facts)
|
||
}
|
||
if notice.Queued > 0 {
|
||
fmt.Fprintf(&b, "\n另有 %d 条消息在排队（未开始）。", notice.Queued)
|
||
}
|
||
b.WriteString("\n")
|
||
b.WriteString(parallelNoticeTail)
|
||
return b.String()
|
||
}
|
||
|
||
// parallelNoticeFacts 渲染事实行（设计示例里的
|
||
// 「上一条消息 id：msg-…；已进行到第 3 步；已调用工具：search_documents、document-read。」）：
|
||
// 只列出有值的元数据段，以全角「；」分隔、以「。」收尾。一段都没有时返回空串，
|
||
// 该行整体不出现。
|
||
func parallelNoticeFacts(notice ParallelNotice) string {
|
||
parts := make([]string, 0, 3)
|
||
if notice.OtherMessageID != "" {
|
||
parts = append(parts, "上一条消息 id："+notice.OtherMessageID)
|
||
}
|
||
if notice.Step > 0 {
|
||
parts = append(parts, fmt.Sprintf("已进行到第 %d 步", notice.Step))
|
||
}
|
||
if len(notice.ToolNames) > 0 {
|
||
parts = append(parts, "已调用工具："+strings.Join(notice.ToolNames, "、"))
|
||
}
|
||
if len(parts) == 0 {
|
||
return ""
|
||
}
|
||
return strings.Join(parts, "；") + "。"
|
||
}
|
||
|
||
// appendParallelNotice 在**本步视图**末尾追加 §2.5.6 的进行态摘要，作为一条
|
||
// **独立的 system 消息**。它必须在 assembleContext **之后**调用（§2.5.6 的落点）。
|
||
//
|
||
// 它只改本轮视图、**不写回 msgs**：checkpoint 快照与后台记忆提取的输入面因此完全
|
||
// 看不到它（与 steer 注入同一口径，见本文件头）。ctx 上未挂摘要时它原样返回 view，
|
||
// 不做任何分配 ⇒ 单路 / 子代理 / 后台思考的路径**零成本、逐字不变**。
|
||
func (l *Loop) appendParallelNotice(ctx context.Context, sessionID string, step int, view []model.Message) []model.Message {
|
||
notice, ok := ParallelNoticeFrom(ctx)
|
||
if !ok {
|
||
return view
|
||
}
|
||
l.log.Debug("parallel notice injected", "session", sessionID, "step", step,
|
||
"other_message", notice.OtherMessageID, "other_step", notice.Step,
|
||
"tools", len(notice.ToolNames), "queued", notice.Queued)
|
||
return append(view, model.TextMessage(model.RoleSystem, parallelNoticeText(notice)))
|
||
}
|