feat(gateway,loop): 忙时并行回合基础管线(I1′ 有条件放宽)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- turnId 由编排器生成、loop Run 复用(硬前提:否则 turn_begin 与流帧各铸一个,客户端两级分桶立即失配)
- StateSink 增带 turnId 的变体(未接线时退化为现签名,不丢状态帧)
- 上行 chat.parallel(WS chat 帧 + REST 三入口同形;省略 = 逐字不变)
- 单会话额度默认 2 / 硬上限 3 / <2 关闭;额度满或未声明 ⇒ 落回既有排队语义
- I1′:checkpoint 归属、历史配对、忙态计数三处从会话级单例升为路级
- session_idle 收紧为「所有在途路收尾且 pending 空」(仅并行时生效;单路逐字不变)
- 预留接口(形状定死、无实现无接线):tool/legfile.go 的 FileClaimer / ErrFileBusy / WithLegID
- 顺手修一个真 bug:并行路 ctx 派生自提交请求作用域,REST 同步入口一写响应就把它连坐杀死(改 context.WithoutCancel 派生)
- 23 个新用例(含 -race -count=3);变异 6 条(turnId 复用 / idle 时点 / 并行升级 / checkpoint 键 / 编排器交 id / pending_ack reason)
- 契约变更点 10 条待监工升 v0.65(本人未改 Docs/**)
This commit is contained in:
AskaEth committed 2026-10-05 09:18:50 +08:00
1 parent 7d2ff74cfc
commit 89e13cf53d
23 files changed
+1921 -275

No files matched your search

+24 -4
View File
@@ -33,12 +33,29 @@ type checkpointState struct {
StopReason string `json:"stop_reason,omitempty"`
}
// checkpointAgentID 返回本回合 checkpoint 的归属键(design/15 §2.5.9)。
//
// 单路(ctx 上未挂并发路标识)时它逐字等于 sessionID —— 与加入并发路之前
// 完全一致,归属语义不变;真并行路(ctx 上挂了 legID)时升为
// `{sessionID}#{legID}`,使 LoadLatest 按路隔离:并行的两路各写各的窗口,
// 绝不会把对方的快照当成自己的续接(并发写同一个 key 会互相覆盖)。
//
// 归属键同时用作 checkpoint 行 id 的前缀,因此两路的行在 `checkpoints` 表里
// 天然不冲突。
func checkpointAgentID(ctx context.Context, sessionID string) string {
if leg := ParallelLegFrom(ctx); leg != "" {
return sessionID + "#" + leg
}
return sessionID
}
// saveCheckpoint 在 step 处持久化一个快照。它绝不让 run 失败:
// store 为 nil 时为空操作,marshal/save 出错则记录日志。
func (l *Loop) saveCheckpoint(ctx context.Context, sessionID string, step int, messages []model.Message, stopReason string) {
if l.checkpoint == nil {
return
}
owner := checkpointAgentID(ctx, sessionID)
state := checkpointState{
Step: step,
Messages: messages,
@@ -50,8 +67,8 @@ func (l *Loop) saveCheckpoint(ctx context.Context, sessionID string, step int, m
return
}
cp := &persist.Checkpoint{
ID: fmt.Sprintf("%s:%d", sessionID, step),
AgentID: sessionID,
ID: fmt.Sprintf("%s:%d", owner, step),
AgentID: owner,
Step: step,
State: buf,
}
@@ -79,10 +96,13 @@ func (l *Loop) loadCheckpoint(ctx context.Context, sessionID string, messages []
if l.checkpoint == nil {
return messages, 1, false
}
cp, err := l.checkpoint.LoadLatest(ctx, sessionID)
// 归属键按路取(design/15 §2.5.9):单路 = sessionID(逐字不变),真并行路
// = {sessionID}#{legID},因此两路的崩溃快照互不可见,绝不会互相顶掉。
owner := checkpointAgentID(ctx, sessionID)
cp, err := l.checkpoint.LoadLatest(ctx, owner)
if err != nil {
if !errors.Is(err, persist.ErrNotFound) {
l.log.Warn("checkpoint load", "session", sessionID, "error", err)
l.log.Warn("checkpoint load", "session", sessionID, "agent", owner, "error", err)
}
return messages, 1, false
}
+169
View File
@@ -0,0 +1,169 @@
package loop
import (
"context"
"errors"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist"
)
// 本文件是 design/15 §2.5.9「checkpoint 归属键由会话升为**会话 + 路**」的回归:
// 真并行时两路各写各的归属键,LoadLatest 因此天然按路隔离 —— 否则两路合写同一个
// key,后写的那份会把前一路的窗口整段顶掉(LangGraph `Send` 扇出的同族教训)。
//
// 单路口径是**逐字不变**的:ctx 上不挂路标识时归属键仍等于 sessionID。
// keyedCheckpointStore 是按 AgentID 分桶的内存 checkpoint 存储,用来证明归属键
// 真的按路隔离(判据是**键**,而不是「循环怎么调用 store」)。
type keyedCheckpointStore struct {
saved []*persist.Checkpoint
latest map[string]*persist.Checkpoint
}
func newKeyedCheckpointStore() *keyedCheckpointStore {
return &keyedCheckpointStore{latest: make(map[string]*persist.Checkpoint)}
}
func (s *keyedCheckpointStore) Save(_ context.Context, cp *persist.Checkpoint) error {
s.saved = append(s.saved, cp)
s.latest[cp.AgentID] = cp
return nil
}
func (s *keyedCheckpointStore) LoadLatest(_ context.Context, agentID string) (*persist.Checkpoint, error) {
cp, ok := s.latest[agentID]
if !ok {
return nil, persist.ErrNotFound
}
return cp, nil
}
// TestCheckpointOwnerSingleLegUnchanged 钉住单路(未挂路标识)的归属键逐字等于
// sessionID:行 id 与 AgentID 都与引入并发路之前完全一致。
func TestCheckpointOwnerSingleLegUnchanged(t *testing.T) {
t.Parallel()
store := newKeyedCheckpointStore()
l := New(Options{Checkpoint: store})
l.saveCheckpoint(context.Background(), "sess", 2, []model.Message{model.TextMessage(model.RoleUser, "hi")}, agent.StopReasonToolCalls)
if len(store.saved) != 1 {
t.Fatalf("saved = %d, want 1", len(store.saved))
}
if got := store.saved[0].AgentID; got != "sess" {
t.Fatalf("单路 AgentID = %q, want %q(逐字不变)", got, "sess")
}
if got := store.saved[0].ID; got != "sess:2" {
t.Fatalf("单路 checkpoint id = %q, want %q", got, "sess:2")
}
}
// TestCheckpointOwnerIsolatedPerLeg 钉住归属键按路隔离:同一会话的两条路各写各的
// 键,且各自 LoadLatest 只命中自己的那份 —— 任何一路都绝不会把对方的崩溃快照
// 当成自己的续接。
func TestCheckpointOwnerIsolatedPerLeg(t *testing.T) {
t.Parallel()
store := newKeyedCheckpointStore()
l := New(Options{Checkpoint: store})
ctxA := WithParallelLeg(context.Background(), "turn-a")
ctxB := WithParallelLeg(context.Background(), "turn-b")
windowA := []model.Message{
model.TextMessage(model.RoleUser, "leg-a 的问题"),
model.TextMessage(model.RoleTool, "tool x result: a"),
}
l.saveCheckpoint(ctxA, "sess", 2, windowA, agent.StopReasonToolCalls)
if got := store.saved[0].AgentID; got != "sess#turn-a" {
t.Fatalf("并行路 AgentID = %q, want %q", got, "sess#turn-a")
}
if got := store.saved[0].ID; got != "sess#turn-a:2" {
t.Fatalf("并行路 checkpoint id = %q, want %q", got, "sess#turn-a:2")
}
// 同会话的另一条路:它自己的键上没有快照 ⇒ 全新开始(绝不续接 A 的窗口)。
msgs := []model.Message{model.TextMessage(model.RoleUser, "leg-b 的问题")}
gotB, stepB, resumedB := l.loadCheckpoint(ctxB, "sess", msgs)
if resumedB || stepB != 1 {
t.Fatalf("另一条路 resumed=%v step=%d, want false/1(绝不读对方的快照)", resumedB, stepB)
}
if len(gotB) != len(msgs) || messageText(gotB[0]) != "leg-b 的问题" {
t.Fatalf("另一条路的窗口被对方的快照顶掉了: %+v", gotB)
}
// 而 A 自己仍然续接得到(归属键保持有效)。
gotA, stepA, resumedA := l.loadCheckpoint(ctxA, "sess", windowA)
if !resumedA || stepA != 3 {
t.Fatalf("A resumed=%v step=%d, want true/3", resumedA, stepA)
}
if len(gotA) != len(windowA) {
t.Fatalf("A 的窗口 = %d 条, want %d", len(gotA), len(windowA))
}
}
// TestCheckpointOwnerSingleLegDoesNotSeeParallelSnapshot 钉住另一个方向:单路
// 回合(归属键 = sessionID)读不到并行路写的快照 —— 两边的键是两个空间。
func TestCheckpointOwnerSingleLegDoesNotSeeParallelSnapshot(t *testing.T) {
t.Parallel()
store := newKeyedCheckpointStore()
l := New(Options{Checkpoint: store})
l.saveCheckpoint(WithParallelLeg(context.Background(), "turn-a"), "sess", 2, []model.Message{
model.TextMessage(model.RoleUser, "leg-a 的问题"),
}, agent.StopReasonToolCalls)
// 单路那次 LoadLatest 请求的键是 "sess";存储里只有 "sess#turn-a"。
reader := &recordingLoadStore{inner: store}
l.checkpoint = reader
msgs := []model.Message{model.TextMessage(model.RoleUser, "单路的问题")}
if _, step, resumed := l.loadCheckpoint(context.Background(), "sess", msgs); resumed || step != 1 {
t.Fatalf("单路 resumed=%v step=%d, want false/1", resumed, step)
}
if len(reader.calls) != 1 || reader.calls[0] != "sess" {
t.Fatalf("LoadLatest 键 = %v, want [sess]", reader.calls)
}
}
// recordingLoadStore 记录 LoadLatest 的入参键,同时把读请求转发给内存存储。
type recordingLoadStore struct {
inner *keyedCheckpointStore
calls []string
}
func (r *recordingLoadStore) Save(ctx context.Context, cp *persist.Checkpoint) error {
return r.inner.Save(ctx, cp)
}
func (r *recordingLoadStore) LoadLatest(ctx context.Context, agentID string) (*persist.Checkpoint, error) {
r.calls = append(r.calls, agentID)
cp, err := r.inner.LoadLatest(ctx, agentID)
if errors.Is(err, persist.ErrNotFound) {
return nil, err
}
return cp, err
}
// TestParallelLegContextRoundTrip 钉住 ctx 通道本身:空 legID 是空操作
// (单路 / 子代理 / 后台思考不挂载 ⇒ 归属键不变),nil ctx 不 panic。
func TestParallelLegContextRoundTrip(t *testing.T) {
t.Parallel()
ctx := WithParallelLeg(context.Background(), "turn-a")
if got := ParallelLegFrom(ctx); got != "turn-a" {
t.Fatalf("ParallelLegFrom = %q, want turn-a", got)
}
if got := ParallelLegFrom(WithParallelLeg(context.Background(), "")); got != "" {
t.Fatalf("空 legID 应是不挂载的 ctx 副本,得到 %q", got)
}
if got := ParallelLegFrom(nil); got != "" {
t.Fatalf("nil ctx 应返回空串,得到 %q", got)
}
// TurnIDFrom 是 WithTurnID 的读侧:编排器先生成、loop 复用,两者必须同值。
turnCtx := WithTurnID(context.Background(), "turn-xyz")
if got := TurnIDFrom(turnCtx); got != "turn-xyz" {
t.Fatalf("TurnIDFrom = %q, want turn-xyz", got)
}
if got := TurnIDFrom(context.Background()); got != "" {
t.Fatalf("未挂载 TurnIDFrom = %q, want 空串", got)
}
}
+40
View File
@@ -39,8 +39,48 @@ func WithTurnID(ctx context.Context, turnID string) context.Context {
return context.WithValue(ctx, loopTurnIDKey{}, turnID)
}
// TurnIDFrom 返回 ctx 上的 turn id;未挂载时返回 ""。它是 WithTurnID 的读侧,
// 供网关在 Run **之前**复用自己已经生成的那个 id(design/15 §2.5.9 的 ⚠️ 必改项):
// 客户端按 `turnId` 两级分桶,而 `turn_begin` 由编排器在调用 Run 之前发出 ——
// 若 Run 内部再自己生成一个,两处 id 不同,分桶会立即失配。
func TurnIDFrom(ctx context.Context) string {
if ctx == nil {
return ""
}
return loopTurn(ctx)
}
// loopTurn 返回 ctx 上的 turn id;未挂载时返回 ""。
func loopTurn(ctx context.Context) string {
s, _ := ctx.Value(loopTurnIDKey{}).(string)
return s
}
// parallelLegKey 是「本 Run 驱动的是某条并发路(leg)」的 context key。
// 它只影响 checkpoint 归属(design/15 §2.5.9),不参与身份判定,也不改变
// 循环的任何行为。
type parallelLegKey struct{}
// WithParallelLeg 把本回合所属**并发路**的标识挂到 ctx 上(design/15 §2.5.9):
// 同一会话真并行时,两路各写各的 checkpoint 归属键 `{sessionID}#{legID}`,
// 使 LoadLatest 天然按路隔离 —— 否则两路会互相把对方的窗口当自己的续接
// (LangGraph `Send` 扇出的同族教训:并行分支必须有独立归属)。
//
// 空 legID 是空操作:单路、子代理、后台思考等未声明并行的运行不挂该键,
// 归属键逐字等于 sessionID,行为与加入本特性之前完全一致。
func WithParallelLeg(ctx context.Context, legID string) context.Context {
if ctx == nil || legID == "" {
return ctx
}
return context.WithValue(ctx, parallelLegKey{}, legID)
}
// ParallelLegFrom 返回 ctx 上挂的并发路标识;未挂载时返回 ""。它是
// WithParallelLeg 的读侧,供 checkpoint 归属键与网关侧断言使用。
func ParallelLegFrom(ctx context.Context) string {
if ctx == nil {
return ""
}
s, _ := ctx.Value(parallelLegKey{}).(string)
return s
}
+80
View File
@@ -0,0 +1,80 @@
package tool
import (
"context"
"errors"
"fmt"
)
// 本文件是**接口预留**(design/15 §2.5.7 与主人 2026-10-05 裁决 3),本切片
// **不实现也不接线**:忙时并行(design/15 §2.5)落地后,同一会话可以有多条在途
// 回合,而写类工具(执行命令 / 修改文件 / 对外发消息 / 改配置,即 design/11 §4
// 的敏感清单)在两路之间会真正争抢同一个副作用目标。
//
// 主人口径(裁决 3)是**文件级**互斥,而不是会话级串行:
// - 工具先检测目标文件此刻是否正被**别的在途路**操作;
// - **同一文件**正在被操作 ⇒ 把「该文件正在被操作」**显式**作为工具结果返回给
// LLM,并提示稍后再试(**不静默、不排队死等**);
// - **不是同一个文件** ⇒ 允许并行操作。
//
// 为什么只留接口:写类工具的「目标文件」需要逐工具声明(命令、补丁、配置写入
// 各自的目标语义不同),且占用登记要挂在**路**的生命周期上(路一收尾就必须释放,
// 否则一个中途报错的路会把那个文件永久锁住)。这两件都属于后续切片,本切片只把
// 通道与判定口径钉死,使实现方不必再改工具接口。
//
// 身份与权限**不走这条通道**:它只回答「此刻有没有另一路正在动这个文件」,与
// actor 元数据 / 策略引擎无关(项目红线)。
// ErrFileBusy 报告目标文件此刻正被另一条在途路操作。它不是执行失败:调用方
// (写类工具)应把它**显式**返回给 LLM,让模型稍后重试,而不是静默吞掉或排队
// 死等(主人裁决 3)。错误文案面向模型,因此写清「同一文件被另一路占用」与
// 「请稍后再试」,绝不泄露另一路的内容。
var ErrFileBusy = errors.New("tool: 目标文件正在被另一条并行回合操作,请稍后再试")
// FileBusyError 把 ErrFileBusy 与具体目标绑成一条可直接交给 LLM 的错误,
// 同时保持 errors.Is(err, ErrFileBusy) 成立(%w 包装)。target 是**服务端
// 规范化后**的文件标识,调用方不得把用户原始输入当作它(那会把路径穿越或
// 隐私内容带进模型上下文)。
func FileBusyError(target string) error {
return fmt.Errorf("%w(目标:%s)", ErrFileBusy, target)
}
// FileClaimer 是按**文件级**粒度登记在途操作的接口(主人裁决 3)。实现方
// (后续切片)负责:
// - 以规范化后的目标为键,记录「哪一条路正在操作它」;
// - Claim 在目标已被**另一条路**占用时返回 ErrFileBusy(同一路重复 Claim
// 同一目标必须是幂等的,否则一条路自己的多步工具调用会互相挡);
// - Release 无条件释放该路在该目标上的占用(路收尾时的兜底),legID 不是
// 占用者时是空操作。
//
// 实现必须是**有界的**:占用表随路收尾清空,绝不随会话长度增长。
type FileClaimer interface {
Claim(ctx context.Context, legID, target string) error
Release(ctx context.Context, legID, target string)
}
// legIDKey 是「本回合属于哪条并发路」的 context key。
type legIDKey struct{}
// WithLegID 把本回合所属的并发路标识挂到 ctx 上。循环在 Run 顶部按 ctx 上挂的
// 路标识播种一次(单路 / 子代理 / 后台思考为空串 ⇒ 不挂载),工具经
// Dispatcher.Invoke 的 ctx 读到它,从而知道「我是哪一路」——这是文件级占用
// 登记的唯一来源,绝不从模型输出推断。
//
// 空 legID 是空操作:单路时未挂载,读取侧返回空串,与加入本特性之前逐字相同。
func WithLegID(ctx context.Context, legID string) context.Context {
if ctx == nil || legID == "" {
return ctx
}
return context.WithValue(ctx, legIDKey{}, legID)
}
// LegIDFrom 返回 ctx 上挂的并发路标识;未挂载(单路、子代理、未接线)时返回
// 空串。空串表示「本回合不与其他路并行」,此时文件级互斥**无需生效**。
func LegIDFrom(ctx context.Context) string {
if ctx == nil {
return ""
}
s, _ := ctx.Value(legIDKey{}).(string)
return s
}
+97
View File
@@ -0,0 +1,97 @@
package tool
import (
"context"
"errors"
"testing"
)
// 本文件钉住 design/15 §2.5.7(主人 2026-10-05 裁决 3)为**后续切片**预留的
// 文件级互斥通道:本切片不实现、不接线,但通道形状与判定口径必须现在定死,
// 否则实现方以后还得改工具接口。
//
// 两件事缺一不可:① 「我是哪一路」只能来自 ctx(绝不从模型输出推断);
// ② 「该文件正被另一条路操作」必须是一条**能被 errors.Is 识别**的哨兵错误,
// 让写类工具把它当成正常返回值显式交给 LLM,而不是静默吞掉或排队死等。
// TestLegIDContextRoundTrip 钉住路标识的 ctx 通道:空 legID 是空操作(单路 /
// 子代理 / 未接线),nil ctx 不 panic。
func TestLegIDContextRoundTrip(t *testing.T) {
t.Parallel()
ctx := WithLegID(context.Background(), "turn-a")
if got := LegIDFrom(ctx); got != "turn-a" {
t.Fatalf("LegIDFrom = %q, want turn-a", got)
}
if got := LegIDFrom(WithLegID(context.Background(), "")); got != "" {
t.Fatalf("空 legID 应是不挂载的 ctx 副本,得到 %q", got)
}
if got := LegIDFrom(nil); got != "" {
t.Fatalf("nil ctx 应返回空串(单路 ⇒ 文件级互斥无需生效),得到 %q", got)
}
}
// TestFileBusyErrorWrapsSentinel 钉住「同一文件正在被别的路操作」的错误形状:
// 它必须能被 errors.Is 识别(调用方按它决定「让模型稍后再试」而不是「执行失败」),
// 且文案里带规范化后的目标,便于模型理解为何重试。
func TestFileBusyErrorWrapsSentinel(t *testing.T) {
t.Parallel()
err := FileBusyError("reports/2026.csv")
if !errors.Is(err, ErrFileBusy) {
t.Fatalf("errors.Is(err, ErrFileBusy) = false,错误形状已断: %v", err)
}
// 它绝不是执行失败:调用方不得把它当成「工具坏了」上报。
if errors.Is(err, ErrUnknownTool) {
t.Fatal("文件占用被误判成了未知工具")
}
if got := err.Error(); got == "" {
t.Fatal("面向模型的错误文案不能为空")
}
}
// fakeFileClaimer 是接口预留的形状检验:它证明 FileClaimer 只要两个方法
// (Claim / Release)就能表达「文件级、按路登记、被占用即明确拒绝」的口径,
// 不需要工具侧再改签名 —— 本切片到此为止,不接线任何实现。
type fakeFileClaimer struct {
owner map[string]string
}
func (f *fakeFileClaimer) Claim(_ context.Context, legID, target string) error {
if holder, ok := f.owner[target]; ok && holder != legID {
return FileBusyError(target)
}
f.owner[target] = legID
return nil
}
func (f *fakeFileClaimer) Release(_ context.Context, legID, target string) {
if f.owner[target] == legID {
delete(f.owner, target)
}
}
// TestFileClaimerShapeFileLevel 钉住接口预留的**粒度**语义(主人裁决 3):
// 同一目标被别的路占住 ⇒ 明确拒绝;**不同目标** ⇒ 允许并行;同一路重复 Claim
// 自己的目标 ⇒ 幂等;Release 之后别的路可以接手。
func TestFileClaimerShapeFileLevel(t *testing.T) {
t.Parallel()
var claimer FileClaimer = &fakeFileClaimer{owner: map[string]string{}}
ctx := context.Background()
if err := claimer.Claim(ctx, "turn-a", "a.txt"); err != nil {
t.Fatalf("首次占用: %v", err)
}
if err := claimer.Claim(ctx, "turn-a", "a.txt"); err != nil {
t.Fatalf("同一路重复占用同一目标必须是幂等的: %v", err)
}
if err := claimer.Claim(ctx, "turn-b", "b.txt"); err != nil {
t.Fatalf("不同文件必须允许并行操作: %v", err)
}
err := claimer.Claim(ctx, "turn-b", "a.txt")
if !errors.Is(err, ErrFileBusy) {
t.Fatalf("同一文件被另一路占用时必须明确拒绝,得到 %v", err)
}
claimer.Release(ctx, "turn-a", "a.txt")
if err := claimer.Claim(ctx, "turn-b", "a.txt"); err != nil {
t.Fatalf("占用释放后应可接手: %v", err)
}
}
@@ -77,6 +77,9 @@ func (s *Server) submitChatStream(ctx context.Context, sessionID string, req cha
"mode": req.Mode,
"clientMsgId": req.ClientMsgID,
"documents": req.Documents,
// parallel 与 WS chat 帧**同形**(design/15 §2.5.2):三条入口
//(REST 同步 / REST stream / WS chat)不能对同一条消息有不同口径。
"parallel": req.Parallel,
}
if s.hub == nil || s.hub.orchestrator == nil {
// 未接编排器:回退到既有的总线触发路径(未接线的 hub 保持完全相同的
@@ -123,8 +126,9 @@ func (s *Server) submitChatTurn(w http.ResponseWriter, r *http.Request, req chat
// 来自会话的绑定(design/22 §4.5)。
msgs := chatTurnMessages(req.Text, req.Image, docs)
env := s.envResolver.owner(r.Context(), req.SessionID)
// 客户端的 deviceID、行为模式与可选幂等键(协同 G4)都经 ctx 串进编排器。
ctx := withClientMsgHint(withMode(withDeviceID(r.Context(), req.DeviceID), req.Mode), clientMsgHint{ClientMsgID: req.ClientMsgID})
// 客户端的 deviceID、行为模式与可选幂等键(协同 G4)都经 ctx 串进编排器;
// 可选的并行声明同理(design/15 §2.5)。
ctx := withParallelTurn(withClientMsgHint(withMode(withDeviceID(r.Context(), req.DeviceID), req.Mode), clientMsgHint{ClientMsgID: req.ClientMsgID}), req.Parallel)
if s.hub.orchestrator != nil {
sr, res := s.hub.orchestrator.SubmitSync(ctx, req.SessionID, msgs, env)
// 重复到达(同一个 clientMsgId):明确告知既有消息 id,而不是再跑一遍
@@ -147,6 +151,13 @@ func (s *Server) submitChatTurn(w http.ResponseWriter, r *http.Request, req chat
writeJSON(w, http.StatusAccepted, map[string]any{"queued": true, "queueDepth": sr.QueueDepth})
return
}
// 升为独立一路(design/15 §2.5):第二路在后台跑,本次请求拿不到同步
// 回复,因此同样回 202 + 明确的判定 —— 绝不回一个 res 为 nil 的 200
//(那会让客户端以为拿到了空回复)。
if sr.Parallel {
writeJSON(w, http.StatusAccepted, map[string]any{"parallel": true, "turnId": sr.TurnID})
return
}
if sr.Err != nil {
s.log.Error("chat run", "err", sr.Err, "session", req.SessionID)
writeError(w, apperr.Wrap(apperr.ErrInternal, "chat_failed", "chat failed"))
@@ -179,11 +190,13 @@ func detachedTurnContext(parent context.Context) context.Context {
// writeChatStreamAck 把 stream:true 的提交判定写成响应(协同 G2 顺带项 + G4)。
// 此前它无条件回 200 {"stream":true},于是「排队满」与「重复到达」都只能靠客户端
// 猜;现在四种判定各有明确的状态码与载荷:
// 猜;现在五种判定各有明确的状态码与载荷:
// - 重复到达:200 {duplicate:true, messageId}(没有新回合、没有新帧,客户端用
// messageId 把本地乐观气泡对齐到既有消息)
// - 待处理队列满:409 {error:{code:"chat_busy"}}(与同步路径同一口径)
// - 会话正忙、已续接排队:202 {stream:true, queued:true, queueDepth}
// - 已升为独立一路:200 {stream:true, parallel:true, turnId}(design/15 §2.5;
// 该路的帧随后经 WS 到达,turnId 就是它的权威标识)
// - 已开始:200 {stream:true}
func (s *Server) writeChatStreamAck(w http.ResponseWriter, sr SubmitResult, sessionID string) {
switch {
@@ -193,6 +206,8 @@ func (s *Server) writeChatStreamAck(w http.ResponseWriter, sr SubmitResult, sess
writeError(w, apperr.Wrap(apperr.ErrConflict, "chat_busy", "会话正忙,待处理消息已满"))
case sr.Pending:
writeJSON(w, http.StatusAccepted, map[string]any{"stream": true, "queued": true, "queueDepth": sr.QueueDepth})
case sr.Parallel:
writeJSON(w, http.StatusOK, map[string]any{"stream": true, "parallel": true, "turnId": sr.TurnID})
default:
writeJSON(w, http.StatusOK, map[string]any{"stream": true})
}
+6
View File
@@ -142,4 +142,10 @@ type Deps struct {
// GET /api/v1/admin/llm/usage 诚实作答 500 usage_unavailable,
// 绝不回一份全 0 的假汇总。
Usage LlmUsageStore
// MaxParallelTurns 是单会话在途并行路数的上限(design/15 §2.5.7)。
// 零值/负数 ⇒ 编排器用默认 2;1 ⇒ 关闭真并行(恒 1 路,完全回落到
// queue + 回合内 steer 的现状);超过硬上限 3 ⇒ 钳到 3。
// 它刻意是编排器的构造期旋钮而不是一个全局:真并行只由显式声明
// (上行 `chat.parallel`)触发,因此调大它不会改变任何未声明并行的客户端行为。
MaxParallelTurns int
}
+4
View File
@@ -115,6 +115,9 @@ type loginResponse struct {
// 同一个非空值只承认第一条 —— 重复到达返回既有消息 id(`{duplicate:true,
// messageId}`)而不再落库、不再起回合,客户端据此把本地乐观气泡对齐到服务端
// 真相(替代「靠文本相等」的脆弱匹配)。省略它时行为与加入该字段之前逐字相同。
// Parallel 是「把本条作为**独立一路**处理」的显式声明(design/15 §2.5):它可选,
// 只有忙时且并发额度未满才照准(见 orchestrator.Submit 的并行分支)。省略它时
// 行为与加入该字段之前逐字相同 —— 服务端**不做**任何启发式自动升级。
type chatRequest struct {
SessionID string `json:"sessionId"`
Text string `json:"text"`
@@ -126,6 +129,7 @@ type chatRequest struct {
// 条,每条只给**服务端存储名**(绝不接受任意 URL)。解析与校验见
// chat_documents.go。
Documents []chatDocumentRef `json:"documents,omitempty"`
Parallel bool `json:"parallel,omitempty"`
Stream bool `json:"stream"`
}
+46 -35
View File
@@ -93,6 +93,9 @@ type Server struct {
// vector 是组合根报告的 pgvector 增强运行时状态(见
// Deps.Vector / VectorState)。零值 = 未配置。
vector VectorState
// maxParallelTurns 是单会话在途并行路数的上限(见 Deps.MaxParallelTurns)。
// 零值 = 未接线 ⇒ 编排器用默认 2。
maxParallelTurns int
// envResolver 解析会话每轮的 loop 环境(其角色)。见
// persona_env.go / design/22 §4.5。
envResolver turnEnvResolver
@@ -111,38 +114,39 @@ func New(cfg *config.Config, deps Deps) (*Server, error) {
return nil, fmt.Errorf("gateway: auth dependency required")
}
s := &Server{
cfg: cfg,
auth: deps.Auth,
bus: deps.Bus,
caps: deps.Registry,
runner: deps.Loop,
sessions: deps.Session,
personas: deps.Personas,
memories: deps.Memories,
history: deps.History,
iot: deps.IoT,
auditor: deps.Scheduler,
taskOutputs: deps.Tasks,
automations: deps.Automations,
users: deps.Users,
refresh: deps.Refresh,
backthink: deps.Backthink,
rag: deps.RAG,
obs: newObservability(deps),
streaming: deps.Streaming,
frontend: deps.Frontend,
llm: deps.LLM,
envPath: deps.EnvPath,
channels: deps.Channels,
plugins: deps.Plugins,
ownerRegistrar: deps.OwnerRegistrar,
webhooks: deps.Webhooks,
usage: deps.Usage,
dbPath: deps.DatabasePath,
dataDir: deps.DataDir,
db: deps.Database,
vector: deps.Vector,
log: sourcedLog(deps.Log),
cfg: cfg,
auth: deps.Auth,
bus: deps.Bus,
caps: deps.Registry,
runner: deps.Loop,
sessions: deps.Session,
personas: deps.Personas,
memories: deps.Memories,
history: deps.History,
iot: deps.IoT,
auditor: deps.Scheduler,
taskOutputs: deps.Tasks,
automations: deps.Automations,
users: deps.Users,
refresh: deps.Refresh,
backthink: deps.Backthink,
rag: deps.RAG,
obs: newObservability(deps),
streaming: deps.Streaming,
frontend: deps.Frontend,
llm: deps.LLM,
envPath: deps.EnvPath,
channels: deps.Channels,
plugins: deps.Plugins,
ownerRegistrar: deps.OwnerRegistrar,
webhooks: deps.Webhooks,
usage: deps.Usage,
dbPath: deps.DatabasePath,
dataDir: deps.DataDir,
db: deps.Database,
vector: deps.Vector,
log: sourcedLog(deps.Log),
maxParallelTurns: deps.MaxParallelTurns,
}
// 一个角色解析器由 Server 各处理器与 hub 共用,使每个面向 owner
// 的入口(REST、WS、语音)以同样方式解析会话角色(design/22 §4.5),
@@ -208,9 +212,16 @@ func newOrchestrator(s *Server) *Orchestrator {
Emit: func(sessionID, typ string, data map[string]any) {
s.hub.broadcast(wsEvent{Type: typ, SessionID: sessionID, Data: data})
},
OnPersisted: s.BroadcastPersisted,
Log: s.log,
History: s.history,
// EmitTurn 是带 turnId 的变体(design/15 §2.5.9):帧壳上的 turnId 让客户端
// 把「这条状态帧属于哪一路」与流帧对齐 —— 真并行下同一会话有两条交错的路,
// 只按 sessionId 分桶会把它们糊在一起。
EmitTurn: func(sessionID, turnID, typ string, data map[string]any) {
s.hub.broadcast(wsEvent{Type: typ, SessionID: sessionID, TurnID: turnID, Data: data})
},
OnPersisted: s.BroadcastPersisted,
Log: s.log,
History: s.history,
MaxParallelTurns: s.maxParallelTurns,
})
}
@@ -42,17 +42,22 @@ type documentStatusWriter interface {
UpdateMessageDocument(ctx context.Context, id, document string) error
}
// rememberTurnDocuments 记下本回合已落库的用户消息(按代次配对),供回合结束
// rememberTurnDocuments 记下本路已落库的用户消息(按世代号配对),供回合结束
// 后把抽取结论回写进它们的 document 列。空列表是空操作(这条回合没有落库任何
// 用户消息,也就没有结论可补)。
//
// 它按**路**配对(design/15 §2.5.9「历史配对」升为路级):并行时两路各自持有
// 自己那几条已落库的用户消息,任何一路结算都只回写自己那几行,绝不会把结论
// 写进另一路的行。路已失效(被打断清空)时什么都不做 —— 这次迟到结算不属于
// 当前状态。
func (o *Orchestrator) rememberTurnDocuments(sessionID string, gen uint64, msgs []history.Message) {
if len(msgs) == 0 {
return
}
o.mu.Lock()
defer o.mu.Unlock()
if st := o.states[sessionID]; st != nil {
st.turnDocs = turnDocuments{gen: gen, msgs: msgs}
if leg := o.legLocked(o.states[sessionID], gen); leg != nil {
leg.docs = turnDocuments{gen: gen, msgs: msgs}
}
}
@@ -81,21 +86,18 @@ func (o *Orchestrator) attachPendingDocuments(sessionID string, seq uint64, msgs
}
}
// takeTurnDocuments 取走与 gen 配对的已落库用户消息并清空记录。第二个返回值
// 报告这次代次是否仍是当代:非当代说明这是一次陈旧回合的迟到结算(中断/取代),
// 调用方据此静默返回 —— 迟到的结算是正常竞态,不是缺陷。
// takeTurnDocuments 取走**本路**(世代号配对)已落库的用户消息并清空记录。
// 第二个返回值报告这条路是否仍然在途:不在途说明这是一次失效路的迟到结算
// (被中断清空 / 被取代),调用方据此静默返回 —— 迟到的结算是正常竞态,不是缺陷。
func (o *Orchestrator) takeTurnDocuments(sessionID string, gen uint64) ([]history.Message, bool) {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[sessionID]
if st == nil {
return nil, true
}
if st.turnDocs.gen != gen {
leg := o.legLocked(o.states[sessionID], gen)
if leg == nil {
return nil, false
}
msgs := st.turnDocs.msgs
st.turnDocs = turnDocuments{}
msgs := leg.docs.msgs
leg.docs = turnDocuments{}
return msgs, true
}
@@ -269,6 +269,10 @@ func TestPersistDocumentStatusIsNotSilent(t *testing.T) {
writer := &fakeDocumentWriter{fakeHistory: &fakeHistory{}}
o := NewOrchestrator(OrchestratorOptions{Run: fixedRunner(res), Log: slog.New(cap), History: writer})
o.SetMode("web", ModeIdle)
// 文档结论回写现在是**按路配对**的(design/15 §2.5.9:并行时两路各写自己
// 那几行),因此白盒构造必须把那条路也登记进去 —— 真实路径上
// rememberTurnDocuments 紧跟 beginLegLocked 调用,恒有路在。
registerTestLeg(o, "web", 1)
o.rememberTurnDocuments("web", 1, []history.Message{{ID: "m1", SessionID: "web", Role: history.RoleUser, Content: "坏列", Document: "{not json"}})
o.persistDocumentStatus("web", 1, res)
if !cap.has(`document status merge`, "") {
@@ -280,6 +284,23 @@ func TestPersistDocumentStatusIsNotSilent(t *testing.T) {
})
}
// registerTestLeg 直接把一条在途路登记进会话状态(不存在时顺带建会话状态),
// 使白盒用例能在不驱动真实回合的前提下断言**按路配对**的收尾路径(文档结论
// 回写、语音回调)。它只服务测试:生产路径上的路一律由 beginLegLocked 登记。
func registerTestLeg(o *Orchestrator, sessionID string, gen uint64) {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[sessionID]
if st == nil {
st = &sessionState{mode: ModeIdle}
o.states[sessionID] = st
}
if st.legs == nil {
st.legs = make(map[uint64]*turnLeg, 1)
}
st.legs[gen] = &turnLeg{gen: gen, leg: 1, fence: newTurnFence(nil)}
}
// TestHistoryEndpointReturnsDocumentStatusArray 钉住线上契约(§4.5):历史接口
// 按 `documents` 数组回传引用与结论,且没有附件的消息**省略**该字段(省略 =
// 没有附件,与引入文档之前逐字一致)。
+37 -20
View File
@@ -13,17 +13,17 @@ package gateway
type InterruptScope string
const (
// InterruptScopeAll 取消在飞回合并清空待处理队列。它是省略 scope 时的
// InterruptScopeAll 取消全部在途路并清空待处理队列。它是省略 scope 时的
// 默认口径,与历史行为逐字一致。
InterruptScopeAll InterruptScope = "all"
// InterruptScopeCurrent 只取消在飞回合,**保留**待处理队列:队列里的消息
// 按编排器正常的续跑规则继续执行,一条都不丢。
// InterruptScopeCurrent 只取消在途路(多路时为**每一路**,见下),**保留**
// 待处理队列:队列里的消息按编排器正常的续跑规则继续执行,一条都不丢。
InterruptScopeCurrent InterruptScope = "current"
)
// Interrupt 取消在飞回合、清空待处理队列并把会话回到空闲(design/15 §3,
// Interrupt 取消**全部**在途路、清空待处理队列并把会话回到空闲(design/15 §3,
// 「停下/算了/换个话题」路径)。它是忙碌会话不必排空队列就能复位的唯一方式;
// 运行中 goroutine 的完成变成陈旧代次,不续跑任何东西。它返回被丢弃的
// 运行中 goroutine 的完成变成失效路,不续跑任何东西。它返回被丢弃的
// 待处理回合数。
//
// 它把会话立即报成空闲并立即接受下一条消息(design/15 §3 的「立即
@@ -36,11 +36,12 @@ func (o *Orchestrator) Interrupt(sessionID string) int {
// InterruptScope 按 scope 中断一个会话(协同项 A2),返回真的被丢弃的
// 待处理回合数:
//
// - InterruptScopeAll:取消在飞回合、清空待处理队列并把会话回到空闲
// - InterruptScopeAll:取消全部在途路、清空待处理队列并把会话回到空闲
// —— 省略 scope 时的历史语义,逐字不变;
// - InterruptScopeCurrent:只取消在飞回合,**保留**待处理队列。它不递增
// 代次,因此被取消回合的结算(completeTurn)仍属当代,会按正常续跑
// 规则弹出队列里的下一条消息、带上会话尾部上下文继续跑(design/15
// - InterruptScopeCurrent:只取消在途路(多路时每一路都取消,因为省略
// turnId 的 scope 就是「作用于每一路」——design/15 §2.5.4),**保留**待处理
// 队列。它不解除路的登记,因此被取消回合的结算(completeTurn)仍会按正常
// 续跑规则弹出队列里的下一条消息、带上会话尾部上下文继续跑(design/15
// §2.2),返回 0 —— 真的没有消息被丢弃。
//
// 空字符串与省略同义(都是现状语义),使老客户端只发 {sessionId} 的
@@ -68,27 +69,43 @@ func (o *Orchestrator) InterruptAll() int {
}
// interruptLocked 在状态锁下执行一次中断。scope 决定待处理队列的去留
// (见 InterruptScope):All 清队列、递增代次并把会话回到空闲;Current
// 只取消在飞回合,队列与代次原样留下,使被取消回合的结算仍会续跑队列。
// (见 InterruptScope):All 清队列、使全部在途路失效并把会话回到空闲;
// Current 只取消在途路,队列与路登记原样留下,使被取消回合的结算仍会续跑队列。
// 它返回真的被丢弃的待处理回合数(Current 恒为 0)。
//
// 多路口径(design/15 §2.5.4):无论哪个 scope,被取消的都是**当前全部在途路**
// —— 省略 turnId 的 scope 就是「作用于每一路」的意思(精确停某一路是后续切片的
// `interrupt.turnId`,本切片不做)。解除在途登记的方式与旧口径的 `st.gen++` 等价:
// 把路从 st.legs 里摘掉,被取消路的迟到结算因此找不到自己的路、成为空操作。
func (o *Orchestrator) interruptLocked(sessionID string, st *sessionState, scope InterruptScope) int {
if st == nil || st.mode == ModeIdle {
return 0
}
if st.cancel != nil {
st.cancel()
}
canceled := o.cancelLegsLocked(st)
if scope == InterruptScopeCurrent {
// 只取消在途回合:不碰 pending、不递增代次。mode 如实报告为
// 当时的忙碌通道(回合仍在收尾、队列还在续跑),而不是 idle。
o.emitState(sessionID, EventSessionInterrupt, map[string]any{"sessionId": sessionID, "mode": st.mode, "dropped": 0})
// 只取消在途路:不碰 pending、不解除路的登记(它们仍在收尾)。mode 如实
// 报告为当时的忙碌通道(回合仍在收尾、队列还在续跑),而不是 idle。
o.emitState(sessionID, EventSessionInterrupt, interruptPayload(sessionID, st.mode, 0, canceled, len(st.legs)))
return 0
}
dropped := len(st.pending)
st.pending = nil
st.gen++
st.legs = nil
st.mode = ModeIdle
st.cancel = nil
o.emitState(sessionID, EventSessionInterrupt, map[string]any{"sessionId": sessionID, "mode": ModeIdle, "dropped": dropped})
o.emitState(sessionID, EventSessionInterrupt, interruptPayload(sessionID, ModeIdle, dropped, canceled, 0))
return dropped
}
// interruptPayload 组装中断帧的载荷(design/15 §2.6):既有键(sessionId / mode /
// dropped)一个不改,新增两个**计数**键 —— cancelledTurns(本次真的取消了几路)
// 与 activeTurns(清点之后还在途的路,scope=all 恒为 0)。两者都不含内容,
// 因此该帧仍在「状态机信号」白名单内。
func interruptPayload(sessionID string, mode Mode, dropped, cancelledTurns, activeTurns int) map[string]any {
return map[string]any{
"sessionId": sessionID,
"mode": mode,
"dropped": dropped,
"cancelledTurns": cancelledTurns,
"activeTurns": activeTurns,
}
}
+108 -165
View File
@@ -83,6 +83,14 @@ type SubmitResult struct {
// (带幂等键时是编排器预分配的那一个),重复路径是既有那条。省略
// clientMsgId 的老客户端为 ""。
MessageID string
// Parallel 报告本次提交被**升为一条独立的并行路**(design/15 §2.5:显式声明
// parallel 且并发额度未满)。它总与 Immediate 同时为真 —— 第二路是「立即起」,
// 不排到队首;额度已满或未声明时它恒为假,本次落回既有排队语义。
Parallel bool
// TurnID 是本次**新起的那一路**的权威回合标识(design/15 §2.5.9):编排器生成、
// loop 复用,并与该路全部流帧的 turnId 同值。入队(Pending)或被拒时为空串,
// 因为那时还没有路。
TurnID string
// Err 在同步路径上于运行出错时设置;REST 调用方
// 用它返回 5xx,而不是编造回复。
Err error
@@ -93,19 +101,23 @@ type SubmitResult struct {
// 实际如何驱动保持不可知(它绝不重新实现 loop 引擎)。
type TurnRunner func(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) (*loop.Result, error)
// StateSink 把每会话状态转移(busy/pending/idle/interrupt)报告给
// 传输层。网关把它接到 WS hub 广播上。
type StateSink func(sessionID, typ string, data map[string]any)
// StateSink 与它的带 turnId 变体(TurnStateSink)住在 session_state.go,
// 与「谁在读状态 / 状态帧怎么出去」放在一起。
//
// OrchestratorOptions 配置 Orchestrator。Run 与 Log 必需
// (Log 可以是丢弃型 logger);Emit 可选(为 nil 时状态事件被丢弃)。
// MaxPending/MaxTail 非正时回退到默认值。
type OrchestratorOptions struct {
Run TurnRunner
Emit StateSink
EmitTurn TurnStateSink
Log *slog.Logger
MaxPending int
MaxTail int
// MaxParallelTurns 是单会话在途路数的上限(design/15 §2.5.7):
// 非正 ⇒ 默认 2;1 ⇒ 关闭真并行(恒 1 路,完全回落到 steer-only 现状);
// 超过硬上限 3 ⇒ 钳到 3。它由 resolveMaxParallelTurns 解析。
MaxParallelTurns int
// History 是可选的持久化聊天历史存储(design/16 §3.3)。
// 为 nil 时每次 Append 都被跳过,因此编排器无需存储也能运行
// (测试 / 未接 history 存储构建的网关)。
@@ -121,14 +133,24 @@ type OrchestratorOptions struct {
// 都在 Orchestrator 互斥锁下发生;实际的 loop 运行在锁外
// 完成,因此慢的模型调用绝不阻塞 Submit。
type sessionState struct {
mode Mode
gen uint64 // generation:在打断时以及新的顶层回合时递增
cancel context.CancelFunc
mode Mode
// gen 是会话内**世代号的分配源**:每条新路取一个(单调、不复用)。
// 旧口径里它同时兼任「当前回合」的身份与陈旧判据;并行落地后这两件事
// 都归 st.legs —— 「这一路还算数吗」=「它还在 st.legs 里吗」,
// 因此启动第二路绝不会让第一路变成陈旧(design/15 §2.5.9)。
gen uint64
// legs 是该会话**当前在途**的路(design/15 §2.5 的「路」),键为路的世代号。
// 单路时它至多一条,且其行为与引入多路之前逐字相同;len(legs) 就是
// 忙态帧里的 activeTurns,也是「所有路都收尾了吗」的判据。
legs map[uint64]*turnLeg
// legSeq 是路序号(1 基)的分配源,供后续切片的 `turn_begin.data.leg` 使用。
legSeq int
pending []pendingMsg
tail []model.Message // 用于上下文续跑的有界对话尾部
// fence 是当前在途回合的启动围栏(见 session_turn_fence.go):下一个
// 回合以它为先驱,等本回合真正返回后才开始 —— 因此中断后立即到达的
// 消息绝不会在被取消回合落定其 checkpoint 之前读到那份旧快照。
// fence 是当前在途**串行**路的启动围栏(见 session_turn_fence.go):下一个
// 串行回合以它为先驱,等本回合真正返回后才开始 —— 因此中断后立即到达的
// 消息绝不会在被取消回合落定其 checkpoint 之前读到那份旧快照。并行路不接入
// 这条链(它的归属键与谁都不同,无需等待)。
fence turnFence
// clientMsgs 记录本会话已认领的 clientMsgId → 该消息的 id(G4 幂等)。
// 它只在状态锁下读写,且只是**快路径 + 同进程并发仲裁**:跨重启的判重由
@@ -142,13 +164,6 @@ type sessionState struct {
// 队列是一条只在队尾追加、可能被整体弹出/清空的先进先出队列,因此「第几条」
// 不稳定;按 seq 认领已注入的消息,才不会被同一时刻的追加或清空误伤。
pendingSeq uint64
// turnDocs 是本回合已落库的用户消息(id + 只有引用的 document 列),供回合
// 结算时把抽取结论回写进那一列(design/24 §6.1)。它按代次配对,因此被中断/
// 取代的陈旧回合读不到本回合的条目。见 session_document_status.go。
turnDocs turnDocuments
// voiceTurn 是本回合的语音收尾回调(协同 C1),按代次配对;纯文本回合记为
// 零值。见 session_voice.go。
voiceTurn voiceTurn
}
// pendingMsg 是一条已排队用户回合。它携带自己的 env,使续跑
@@ -186,11 +201,15 @@ type Orchestrator struct {
states map[string]*sessionState
run TurnRunner
emit StateSink
emitTurn TurnStateSink
log *slog.Logger
history HistoryStore
onPersisted func(history.Message)
maxPending int
maxTail int
// maxParallelTurns 是单会话在途路数的生效上限(已解析,见
// resolveMaxParallelTurns):1 = 关闭真并行(恒 1 路)。
maxParallelTurns int
}
// NewOrchestrator 从 opts 构建编排器。Run 必需:它是编排器
@@ -210,14 +229,16 @@ func NewOrchestrator(opts OrchestratorOptions) *Orchestrator {
opts.Log = slog.New(slog.NewTextHandler(io.Discard, nil))
}
return &Orchestrator{
states: make(map[string]*sessionState),
run: opts.Run,
emit: opts.Emit,
log: opts.Log,
history: opts.History,
onPersisted: opts.OnPersisted,
maxPending: maxPending,
maxTail: maxTail,
states: make(map[string]*sessionState),
run: opts.Run,
emit: opts.Emit,
emitTurn: opts.EmitTurn,
log: opts.Log,
history: opts.History,
onPersisted: opts.OnPersisted,
maxPending: maxPending,
maxTail: maxTail,
maxParallelTurns: resolveMaxParallelTurns(opts.MaxParallelTurns),
}
}
@@ -277,107 +298,105 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes
// 加入该字段之前逐字相同。
ctx = withClientMsgHint(ctx, a.hint)
if st.mode != ModeIdle {
st.pendingSeq++
seq := st.pendingSeq
st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, mode: msgMode, at: time.Now(), seq: seq, voice: voiceReplySinkFrom(ctx)})
depth := len(st.pending)
mode := st.mode
o.mu.Unlock()
// 落库在锁外(慢操作)完成,随后把它挂回那条待处理回合:抽取结论要等
// 这个回合真正跑起来才有,只能在它的续跑结算时回写(design/24 §6.1)。
o.attachPendingDocuments(sessionID, seq, o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID))
o.emitState(sessionID, EventPendingAck, map[string]any{"sessionId": sessionID, "mode": mode, "queueDepth": depth})
return SubmitResult{Accepted: true, Pending: true, QueueDepth: depth, MessageID: a.hint.MessageID}, nil
// 忙时:**显式**声明并行且并发额度未满 ⇒ 升为第二路(design/15 §2.5.1
// 的 I1′)。额度判定与登记在同一个临界区(startParallelLocked 在锁内完成
// 登记),因此并发到达的两条并行消息不可能双双升级、突破上限。
// 未声明 / 额度已满 / 队列已满一律落回既有排队语义:不报错、不打断。
if parallelRequested(ctx) && o.canStartParallelLocked(st) {
return o.startParallelLocked(ctx, st, sessionID, messages, env, deviceID, msgMode, a.hint), nil
}
return o.enqueueLocked(ctx, st, sessionID, messages, env, deviceID, msgMode, a.hint), nil
}
st.gen++
gen := st.gen
st.mode = ModeThinking
// 本回合的围栏以当前在途回合的收尾信号为先驱,使中断后来到的这条
// 本回合的围栏以当前在途串行回合的收尾信号为先驱,使中断后来到的这条
// 消息不会抢在被取消回合落定旧快照之前开跑(session_turn_fence.go)。
fence := newTurnFence(st.fence.done)
st.fence = fence
turn := o.seedTurnLocked(ctx, st, sessionID, messages)
runCtx, cancel := context.WithCancel(ctx)
st.cancel = cancel
leg, runCtx, cancel := o.beginLegLocked(ctx, st, newTurnID(), false, newTurnFence(st.fence.done))
st.fence = leg.fence
o.mu.Unlock()
// 本回合的转向源(design/15 §2.4):loop 在**回合内边界**回调它重读转录,把忙时
// 新到、已入 pending 的用户消息并入**同一回合**的下一次模型调用(I1:绝不新开
// 第二回合)。它挂在 runCtx 上,因此只作用于本回合;回合结算或被打断后随 ctx 消亡。
runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, gen))
runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, leg.gen))
o.emitState(sessionID, EventSessionBusy, map[string]any{"sessionId": sessionID, "mode": ModeThinking})
o.emitState(sessionID, EventSessionBusy, busyPayload(sessionID, ModeThinking, 1))
// 只有此刻到达的消息会被持久化(已播种的转录已持久,重追加会重复整个对话),
// 且落库只写**引用** —— 抽取还发生在后面,结论由回合结算时补进同一列
// (两段式见 session_document_status.go)。
o.rememberTurnDocuments(sessionID, gen, o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID))
// 语音收尾回调同样按代次配对(协同 C1,见 session_voice.go)。
o.rememberVoiceTurn(sessionID, gen, voiceReplySinkFrom(ctx))
o.rememberTurnDocuments(sessionID, leg.gen, o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID))
// 语音收尾回调同样按**路**配对(协同 C1,见 session_voice.go)。
o.rememberVoiceTurn(sessionID, leg.gen, voiceReplySinkFrom(ctx))
if sync {
return o.runImmediateSync(runCtx, sessionID, gen, turn, env, fence, deviceID, msgMode, a.hint.MessageID)
return o.runImmediateSync(runCtx, sessionID, leg, turn, env, deviceID, msgMode, a.hint.MessageID)
}
go o.goTurn(sessionID, gen, turn, env, runCtx, cancel, deviceID, msgMode, fence)
return SubmitResult{Accepted: true, Immediate: true, MessageID: a.hint.MessageID}, nil
go o.goTurn(sessionID, leg, turn, env, runCtx, cancel, deviceID, msgMode)
return SubmitResult{Accepted: true, Immediate: true, MessageID: a.hint.MessageID, TurnID: leg.turnID}, nil
}
// runImmediateSync 在**调用方** goroutine 上跑完刚启动的那个回合(REST 同步
// 路径),结算它并把回合结果一并交回调用方。它与 goTurn 是同一轮的两个驱动
// 方式(前台等待 / 后台续跑),共享 runTurnFenced + completeTurn 这一对,因此
// 谁驱动都产生同样的持久化与状态迁移。
func (o *Orchestrator) runImmediateSync(ctx context.Context, sessionID string, gen uint64, turn []model.Message, env loop.SessionEnv, fence turnFence, deviceID, mode, msgID string) (SubmitResult, *loop.Result) {
res, err := o.runTurnFenced(ctx, sessionID, turn, env, fence)
fence.release()
func (o *Orchestrator) runImmediateSync(ctx context.Context, sessionID string, leg *turnLeg, turn []model.Message, env loop.SessionEnv, deviceID, mode, msgID string) (SubmitResult, *loop.Result) {
res, err := o.runTurnFenced(ctx, sessionID, turn, env, leg.fence)
leg.fence.release()
if err != nil {
o.log.Error("orchestrator turn", "err", err, "session", sessionID)
}
_ = o.completeTurn(sessionID, gen, res, err, deviceID, mode, env.PersonaID)
return SubmitResult{Accepted: true, Immediate: true, MessageID: msgID, Err: err}, res
_ = o.completeTurn(sessionID, leg, res, err, deviceID, mode, env.PersonaID)
return SubmitResult{Accepted: true, Immediate: true, MessageID: msgID, TurnID: leg.turnID, Err: err}, res
}
// goTurn 在后台 goroutine 上运行单个回合,并在它结束时
// goTurn 在后台 goroutine 上运行单条**路**,并在它结束时
// 把结果交给 completeTurn,使编排器能续跑队列。
// deviceID 用发送该回合的客户端标注持久化助手回复
// (design/16 §4.5),mode 用该回合的行为模式(design/16 §5.1),而 env 是
// 该回合解析出的角色/隔离环境(design/22 §4.5),因此回复被
// 归到产出它的角色名下。fence 是本回合的启动围栏:它先等上一回合
// 收尾,并在本回合返回时放行下一回合(session_turn_fence.go)。
func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Message, env loop.SessionEnv, ctx context.Context, cancel context.CancelFunc, deviceID, mode string, fence turnFence) {
// 归到产出它的角色名下。leg.fence 是本路的启动围栏:它先等上一**串行**回合
// 收尾,并在本路返回时放行下一路(session_turn_fence.go)。
func (o *Orchestrator) goTurn(sessionID string, leg *turnLeg, messages []model.Message, env loop.SessionEnv, ctx context.Context, cancel context.CancelFunc, deviceID, mode string) {
defer cancel()
defer fence.release()
res, err := o.runTurnFenced(ctx, sessionID, messages, env, fence)
defer leg.fence.release()
res, err := o.runTurnFenced(ctx, sessionID, messages, env, leg.fence)
if err != nil && ctx.Err() == nil {
o.log.Error("orchestrator turn", "err", err, "session", sessionID)
}
_ = o.completeTurn(sessionID, gen, res, err, deviceID, mode, env.PersonaID)
_ = o.completeTurn(sessionID, leg, res, err, deviceID, mode, env.PersonaID)
}
// completeTurn 在一个回合结算后运行(它完成、出错或被
// completeTurn 在一条路结算后运行(它完成、出错或被
// 取消)。它把助手回复折进会话的续跑尾部,
// 然后弹出下一条待处理回合并带着该尾部续跑它 —— 因此下一条
// 消息「记得」先前上下文,而不是从空白开始
// (design/15 §2.2:增量,而不是全量重放)。陈旧的代次(一次
// 打断或一个更新的顶层回合取代了本次)是空操作。
// (design/15 §2.2:增量,而不是全量重放)。**已经失效的路**(被
// 打断、被取代:不在 st.legs 里了)是空操作。
// deviceID 是产出回合的客户端 id,mode 是其行为模式,
// personaID 是它说话时的角色,都用于标注持久化的回复
// (design/16 §4.5、§5.1;design/22 §7-5)。
//
// 它同时是文档抽取结论回写的落点(design/24 §6.1):本回合用户消息的
// 它同时是文档抽取结论回写的落点(design/24 §6.1):本路用户消息的
// document 列里此前只有引用,装配期解析出的 status/chars/truncated 在这里补上
// —— 见 session_document_status.go;语音回合的音频回复同样在此收尾(协同 C1,
// 见 session_voice.go):只有这里知道这个回合真正产出了什么。
func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Result, runErr error, deviceID, mode, personaID string) error {
//
// 多路口径(design/15 §2.5.3):本路先把自己从在途路集合摘掉,**再**决定续跑
// 还是报空闲。续跑仍由结算中的这一路负责(弹队首、起一条新的**串行**路,占一个
// 额度);只有在途路全空且队列也空时才报空闲 —— 单路时该条件恒等价于现状。
func (o *Orchestrator) completeTurn(sessionID string, leg *turnLeg, res *loop.Result, runErr error, deviceID, mode, personaID string) error {
gen := leg.gen
res = o.persistTurnReply(sessionID, gen, res, runErr, deviceID, mode, personaID)
o.persistDocumentStatus(sessionID, gen, res)
o.dispatchVoiceReply(sessionID, gen, res)
o.mu.Lock()
st := o.states[sessionID]
if st == nil || st.gen != gen {
if o.legLocked(st, gen) == nil {
o.mu.Unlock()
return nil
}
delete(st.legs, gen)
if res != nil && res.Text != "" {
st.tail = o.boundedTail(concatMessages(st.tail, assistantTailMessage(res.Text, personaID)))
}
@@ -385,114 +404,38 @@ func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Resu
next := st.pending[0]
st.pending = st.pending[1:]
st.tail = o.boundedTail(concatMessages(st.tail, next.messages...))
st.mode = ModeThinking
st.gen++
gen = st.gen
runCtx, cancel := context.WithCancel(context.Background())
st.cancel = cancel
// 续跑回合同样带自己的转向源(design/15 §2.4):严格 FIFO 弹出下一条之后,
// 期间**仍在 pending** 的后到消息会在本回合的下一个边界被注入,而不是各起一回合。
runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, gen))
// 本回合的这次 Run 已经返回(我们正处在它的结算里),因此被续跑
// 的回合无需再等任何在途回合:围栏只负责放行它后面的那一个。
fence := newTurnFence(nil)
st.fence = fence
// 本路的 Run 已经返回(我们正处在它的结算里),因此被续跑的回合无需再等
// 任何在途路:围栏只负责放行它后面的那一个。
runLeg, runCtx, cancel := o.beginLegLocked(context.Background(), st, newTurnID(), false, newTurnFence(nil))
st.fence = runLeg.fence
runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, runLeg.gen))
// 续跑的这一轮同样有「回合前已落库的用户消息」,因此把它们的文档状态
// 记到新代次上(design/24 §6.1)。
st.turnDocs = turnDocuments{gen: gen, msgs: next.persisted}
// 续跑的这条若源自语音请求,它的音频回复照旧定向回发起它的那根连接
// (协同 C1:忙时排队不许把音频丢掉)。
st.voiceTurn = voiceTurn{gen: gen, sinks: voiceSinksOf(next.voice)}
// 记到**这一路**上(design/24 §6.1);语音回调同理(协同 C1:忙时排队
// 不许把音频丢掉)。
runLeg.docs = turnDocuments{gen: runLeg.gen, msgs: next.persisted}
runLeg.voice = voiceTurn{gen: runLeg.gen, sinks: voiceSinksOf(next.voice)}
depth := len(st.pending)
activeTurns := len(st.legs)
resume := append([]model.Message(nil), st.tail...)
env := next.env
o.mu.Unlock()
o.emitState(sessionID, EventPendingDrain, map[string]any{"sessionId": sessionID, "mode": ModeThinking, "queueDepth": depth})
go o.goTurn(sessionID, gen, resume, env, runCtx, cancel, next.deviceID, next.mode, fence)
o.emitState(sessionID, EventPendingDrain, map[string]any{"sessionId": sessionID, "mode": ModeThinking, "queueDepth": depth, "activeTurns": activeTurns})
go o.goTurn(sessionID, runLeg, resume, env, runCtx, cancel, next.deviceID, next.mode)
return nil
}
// 还有别的路在飞 ⇒ **不报空闲**(design/15 §2.5.3 的时点收紧,仅在并行时
// 与现状不同:单路时此刻 len(st.legs) 必为 0,因此这一分支不可达)。
if len(st.legs) > 0 {
o.mu.Unlock()
return nil
}
st.mode = ModeIdle
st.cancel = nil
o.mu.Unlock()
o.emitState(sessionID, EventSessionIdle, map[string]any{"sessionId": sessionID, "mode": ModeIdle})
o.emitState(sessionID, EventSessionIdle, busyPayload(sessionID, ModeIdle, 0))
return nil
}
// SetMode 提升某会话所报告的忙碌状态通道(thinking →
// executing → subagent)。它是 loop/subagent 钩子可调用的注入点;
// 任何非空闲模式都让会话保持忙碌,而 idle 解除忙碌。为 nil 的会话
// 被惰性创建,因此钩子可以在第一条消息之前设模式。
func (o *Orchestrator) SetMode(sessionID string, mode Mode) {
sessionID = o.normalize(sessionID)
o.mu.Lock()
st := o.states[sessionID]
if st == nil {
st = &sessionState{mode: mode}
o.states[sessionID] = st
} else {
st.mode = mode
}
o.mu.Unlock()
if mode == ModeIdle {
o.emitState(sessionID, EventSessionIdle, map[string]any{"sessionId": sessionID, "mode": mode})
return
}
o.emitState(sessionID, EventSessionBusy, map[string]any{"sessionId": sessionID, "mode": mode})
}
// Busy 报告该会话当前是否有在飞回合。
func (o *Orchestrator) Busy(sessionID string) bool {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[o.normalize(sessionID)]
return st != nil && st.mode != ModeIdle
}
// QueueDepth 报告某会话的待处理回合数。
func (o *Orchestrator) QueueDepth(sessionID string) int {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[o.normalize(sessionID)]
if st == nil {
return 0
}
return len(st.pending)
}
// ModeOf 报告某会话当前的忙碌状态通道(未知时为 idle)。
func (o *Orchestrator) ModeOf(sessionID string) Mode {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[o.normalize(sessionID)]
if st == nil {
return ModeIdle
}
return st.mode
}
// normalize 把空会话 id 映射为默认值,使状态 map 对省略它的
// 调用方保持稳定(Web 主会话)。
func (o *Orchestrator) normalize(sessionID string) string {
if sessionID == "" {
return defaultSessionID
}
return sessionID
}
// emitState 在接线了传输层时把状态转移转发给它。
func (o *Orchestrator) emitState(sessionID, typ string, data map[string]any) {
if o.emit != nil {
o.emit(sessionID, typ, data)
}
}
// notifyPersisted 在接线了传输钩子时把刚持久化的消息交给它,
// 使网关能广播 `message_persisted` 以进行跨端同步。
func (o *Orchestrator) notifyPersisted(msg history.Message) {
if o.onPersisted != nil {
o.onPersisted(msg)
}
}
@@ -0,0 +1,242 @@
package gateway
import (
"context"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// 本文件归属 design/15 §2.5「忙时并行回合(多路)」的**基础管线**(主人
// 2026-10-05 裁决 B)。它从 session_orchestrator.go 拆出,使编排器本体继续只留
// 状态机,并保持在「单文件 ≤500 行 / 函数 ≤80 行」的硬约束内。
//
// 它做四件事:
// - 解析上行的并行声明(ctx 通道,与 deviceID / mode / 幂等键同一种机制);
// - 判定并发额度(默认 2,硬上限 3,<2 = 关闭真并行 ⇒ 完全回落到 steer-only 现状);
// - 在状态锁下登记一条新路(自己的世代号 / turnId / 取消链 / checkpoint 归属键);
// - 忙态帧的在途路数(activeTurns)与排队原因(reason)。
//
// 它**不做**(留给后续切片):`turn_begin` 帧的发射、`steer_ack` 的帧壳 turnId、
// 写类工具的文件级互斥、第二路的进行态 system 摘要。预留点见 turnLeg 的字段。
// 并发额度的取值口径(design/15 §2.5.7)。
const (
// defaultMaxParallelTurns 是单会话在途路数的默认上限。
defaultMaxParallelTurns = 2
// maxParallelTurnsHardCap 是硬上限:配置再多也钳到它。上限存在的意义是
// 防失控(对齐 OpenAI Swarm / CrewAI 的「有上限的并行 worker」口径),
// 而不是成本旋钮 —— 成本按 design/15 §2.5.7 由既有 429/降级链兜。
maxParallelTurnsHardCap = 3
// pendingAckReasonParallelLimit 是 pending_ack 的 reason 取值(design/15 §2.6):
// 「声明了并行,但并发额度已满,本条已落回排队」。它不含任何内容,因此该帧
// 仍在「状态机信号」白名单内。
pendingAckReasonParallelLimit = "parallel_limit"
// parallelLogSource 是并发路相关日志的子功能来源标签。
parallelLogSource = "session-parallel"
)
// turnLeg 是会话内一条在途回合(design/15 §2.5 的「路」)。
//
// 单路时它恒只有一条,且其全部行为与引入本结构之前逐字相同;真并行时同一会话
// 有多条,每条各有自己的取消链(I3′)、自己的围栏、自己的历史配对(turnDocs /
// voiceTurn)与自己的 checkpoint 归属键。
//
// 它只在 Orchestrator.mu 下被读写,除了 cancel(取消是并发安全的)与该路自己的
// goroutine 读取 turnID/gen(出生后不可变)。
type turnLeg struct {
// turnID 是本路的权威回合标识(design/15 §2.5.9):编排器生成,挂进 runCtx
// 的 loop.WithTurnID,并由 loop 盖在该路全部流帧上。后续切片的 `turn_begin`
// 帧与 `steer_ack` 的帧壳 turnId 直接取它 —— 两处必须同值,否则客户端按
// turnId 两级分桶会立刻失配。
turnID string
// gen 是本路的世代号:会话内单调、永不复用。它取代了旧的「会话当前代次」
// 概念 —— 陈旧判据从「st.gen == gen」变成「本路还在 st.legs 里」,因此启动
// 第二路绝不会让第一路变成陈旧。
gen uint64
// leg 是本路的会话内序号(1 基)。它预留给后续切片的 `turn_begin.data.leg`
//(「并行路 2」的展示口径),本切片只在日志里用它。
leg int
// parallel 报告本路是否为**显式声明的并行路**:它决定 checkpoint 归属键
// (单路逐字等于 sessionID,并行路为 {sessionID}#{turnID})。
parallel bool
// cancel 派生自提交时的 ctx,是**本路独占**的取消链:任一路的注入 / 取消
// 都不影响另一路已发起的工具(I3′)。
cancel context.CancelFunc
// fence 是本路的启动围栏(session_turn_fence.go)。串行路接入会话级围栏链,
// 并行路恒自带一道无须等待的围栏 —— 它的归属键与任何在途路都不同,
// 因此不可能读到对方尚未落定的快照。
fence turnFence
// docs 是本路已落库的用户消息(design/24 §6.1 的文档结论回写),
// voice 是本路的语音收尾回调(协同 C1)。二者都按**路**配对:并行时两路
// 各自回写自己那几行,陈旧的路读不到别人的条目。
docs turnDocuments
voice voiceTurn
}
// parallelTurnKey 是携带「本次提交显式声明并行」的 context 键。它从 WS 帧 /
// REST body 边界引入编排器,与 deviceID / mode / 幂等键走同一种机制 ——
// Submit 的签名因此保持不变,所有既有调用方零改动。
type parallelTurnKey struct{}
// withParallelTurn 返回携带并行声明的 context。false 是空操作:省略
// `parallel` 的请求在 ctx 上不留任何痕迹,行为与加入该字段之前逐字相同。
func withParallelTurn(ctx context.Context, parallel bool) context.Context {
if !parallel {
return ctx
}
return context.WithValue(ctx, parallelTurnKey{}, true)
}
// parallelRequested 报告该提交是否显式声明了并行。服务端**不做启发式自动升级**
// (design/15 §2.5.1):触发权在客户端,否则「同一条消息被如何作答」会随负载
// 漂移,「恰好一次」的直觉被破坏。
func parallelRequested(ctx context.Context) bool {
if ctx == nil {
return false
}
v, _ := ctx.Value(parallelTurnKey{}).(bool)
return v
}
// resolveMaxParallelTurns 把配置值解析为生效的并发上限:非正 ⇒ 默认 2;
// 1 ⇒ 关闭真并行(恒 1 路,完全回落到 steer-only 现状);超过硬上限 ⇒ 钳到 3。
//
// 它是纯函数,因此「默认 / 关闭 / 钳位」三档边界可以被直接钉住(见
// session_parallel_test.go),而不必去构造一个完整的网关。
func resolveMaxParallelTurns(configured int) int {
if configured <= 0 {
return defaultMaxParallelTurns
}
if configured > maxParallelTurnsHardCap {
return maxParallelTurnsHardCap
}
return configured
}
// canStartParallelLocked 报告此刻能否把一条声明了并行的忙时消息升为第二路
// (design/15 §2.5.1 的 I1′)。三个条件缺一不可:
//
// - 真并行没有被关掉(上限 ≥ 2);
// - 会话已有**至少一条在途路**(没有在飞的路,就没有「并行」可言:那种情况
// 下 mode 非空闲只可能来自外部 SetMode 钩子,语义上应照旧排队);
// - 在途路数**未用满**额度。
//
// 必须在 o.mu 下调用。
func (o *Orchestrator) canStartParallelLocked(st *sessionState) bool {
if st == nil || o.maxParallelTurns < defaultMaxParallelTurns {
return false
}
return len(st.legs) >= 1 && len(st.legs) < o.maxParallelTurns
}
// startParallelLocked 在状态锁下把一条忙时到达、且显式声明并行的用户消息升为
// **独立的一路**(design/15 §2.5.1 / §2.5.3),并在锁外起 goroutine、返回本次提交
// 的判定。它必须在 o.mu 下调用,且由调用方保证 canStartParallelLocked 已经成立
// (额度判定与登记必须在同一个临界区,否则两条同时到达的并行消息会双双升级、
// 突破上限)。
//
// 三件事一次做完:
// - 本路的窗口取**该会话转录快照 + 本条消息**(I4:每路自己的窗口,绝不在运行中
// 看见对方的半成品)。它刻意**不**改写 st.tail —— 尾部是会话级增量游标、属于
// 串行续跑链,并行路不该把自己的窗口塞进别人要续跑的上下文;
// - 本路的 checkpoint 归属键升为 {sessionID}#{turnID}(§2.5.9 —— LangGraph
// `Send` 扇出的同族要求:并行分支必须有独立归属),因此它**不需要等任何在途路
// 收尾**:归属键不同就不可能读到对方的快照;
// - 本路的用户消息立即落库(与单路同一条两段式),使转录与跨端同步立刻看得到它。
func (o *Orchestrator) startParallelLocked(ctx context.Context, st *sessionState, sessionID string, messages []model.Message, env loop.SessionEnv, deviceID, mode string, hint clientMsgHint) SubmitResult {
turn := o.startTurnMessages(ctx, sessionID, messages)
// 第二路的 ctx **脱离提交请求的作用域**:它是一条后台路,提交方拿到的只是
// 「已受理」的 ack(WS 帧早已返回、REST 回 202),因此请求 ctx 随后被取消
// 绝不该把它连坐杀掉 —— 否则 REST 同步入口的并行路会在写响应的那一刻就死。
// WithoutCancel 保留 ctx 上的值(deviceID / 幂等键 / trace),只丢取消信号;
// 该路仍可被编排器自己的取消链(Interrupt / InterruptAll)取消。
// 这与续跑路用 Background 派生属于同一条口径:后台路的生死由编排器说了算。
leg, runCtx, cancel := o.beginLegLocked(context.WithoutCancel(ctx), st, newTurnID(), true, newTurnFence(nil))
activeTurns := len(st.legs)
o.mu.Unlock()
runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, leg.gen))
o.emitState(sessionID, EventSessionBusy, busyPayload(sessionID, ModeThinking, activeTurns))
logger.Scoped(o.log, source, parallelLogSource).Info("parallel leg started",
"session", sessionID, "turn", leg.turnID, "leg", leg.leg, "active_turns", activeTurns)
o.rememberTurnDocuments(sessionID, leg.gen, o.persistUserMessages(ctx, sessionID, messages, deviceID, mode, env.PersonaID))
o.rememberVoiceTurn(sessionID, leg.gen, voiceReplySinkFrom(ctx))
go o.goTurn(sessionID, leg, turn, env, runCtx, cancel, deviceID, mode)
return SubmitResult{Accepted: true, Immediate: true, Parallel: true, MessageID: hint.MessageID, TurnID: leg.turnID}
}
// beginLegLocked 在状态锁下登记一条新路(design/15 §2.5.9)。它分配该路自己的
// 世代号、派生**独立的**取消链(I3′),把 turnID 与(并行时的)路标识盖进 runCtx,
// 并把 runCtx 交回调用方 —— 调用方解锁后再起 goroutine。
//
// turnID 由**编排器**生成并由 loop 复用(design/15 §2.5.9 的 ⚠️ 必改项):
// `turn_begin` 帧由编排器在调用 Run 之前发出,若 Run 内部再自己生成一个,
// 客户端按 turnId 两级分桶会立即失配。单路时该 id 仍是每回合全新的随机值,
// 因此对客户端**无可见差异**。
//
// 它同时把会话标成忙(ModeThinking),使 mode 在任一路在途时恒非空闲。
func (o *Orchestrator) beginLegLocked(ctx context.Context, st *sessionState, turnID string, parallel bool, fence turnFence) (*turnLeg, context.Context, context.CancelFunc) {
st.gen++
st.legSeq++
runCtx, cancel := context.WithCancel(ctx)
runCtx = loop.WithTurnID(runCtx, turnID)
leg := &turnLeg{turnID: turnID, gen: st.gen, leg: st.legSeq, parallel: parallel, cancel: cancel, fence: fence}
if parallel {
runCtx = loop.WithParallelLeg(runCtx, turnID)
}
if st.legs == nil {
st.legs = make(map[uint64]*turnLeg, o.maxParallelTurns)
}
st.legs[leg.gen] = leg
st.mode = ModeThinking
return leg, runCtx, cancel
}
// legLocked 按世代号取该会话的在途路;不存在时返回 nil,表示这次收尾属于一条
// **已经失效**的路(被打断、被取代,或会话从未登记过它)。它取代了旧的
// `st.gen != gen` 判据:真并行下启动第二路不再让第一路失效。
//
// 必须在 o.mu 下调用。
func (o *Orchestrator) legLocked(st *sessionState, gen uint64) *turnLeg {
if st == nil {
return nil
}
return st.legs[gen]
}
// cancelLegsLocked 取消该会话当前**全部**在途路(design/15 §2.5.4:省略 turnId 的
// scope 作用于每一路),返回真的取消了几路。每路各有自己的取消链,因此取消一路
// 绝不影响另一路已发起的工具(I3′)。
//
// 必须在 o.mu 下调用。
func (o *Orchestrator) cancelLegsLocked(st *sessionState) int {
if st == nil {
return 0
}
for _, leg := range st.legs {
if leg.cancel != nil {
leg.cancel()
}
}
return len(st.legs)
}
// busyPayload 组装忙态 / 空闲帧的载荷(design/15 §2.6):既有键(sessionId / mode)
// 一个不改,只新增在途路数 activeTurns(整数,含正在收尾的路)。它不含任何内容,
// 因此这些帧仍在「状态机信号」白名单内、所有已认证连接都收得到。
func busyPayload(sessionID string, mode Mode, activeTurns int) map[string]any {
return map[string]any{"sessionId": sessionID, "mode": mode, "activeTurns": activeTurns}
}
// pendingAckPayload 组装排队应答的载荷(design/15 §2.6)。既有三个键
// (sessionId / mode / queueDepth)一个不改;只有「显式声明了并行却落回排队」
// 才附 reason,其它情况该键**缺席**,绝不伪造。
func pendingAckPayload(sessionID string, mode Mode, depth int, parallelLimit bool) map[string]any {
data := map[string]any{"sessionId": sessionID, "mode": mode, "queueDepth": depth}
if parallelLimit {
data["reason"] = pendingAckReasonParallelLimit
}
return data
}
@@ -0,0 +1,233 @@
package gateway
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// 本文件是 design/15 §2.5 基础管线的**协议面**回归:turnId 一致性
// (编排器生成 → loop 复用 → 流帧帧壳同值)、带 turnId 的状态通道变体、
// 上行 `parallel` 的解码,以及 REST 三条入口的应答形状。夹具留在
// session_parallel_test.go(同包),两半共用。
// ---------------------------------------------------------------------------
// ⑥ turnId 一致性(编排器生成 → loop 复用 → 流帧帧壳同值)
// ---------------------------------------------------------------------------
// TestRunTurnReusesOrchestratorTurnID 钉住①的硬前提:hub 的 runTurn 必须**复用**
// 编排器已经生成并交给 Run 的那个 turnId(否则 `turn_begin` 与流帧各生成一个,
// 客户端两级分桶立即失配);未挂载时才新铸一个,因此单路路径无可见差异。
func TestRunTurnReusesOrchestratorTurnID(t *testing.T) {
t.Parallel()
h := newTestHub(&stubRunner{res: &loop.Result{Text: "ok", StopReason: "stop"}}, nil)
conn := newOwnerConn(h, 16)
h.register(conn)
const want = "turn-orchestrator-made"
ctx := loop.WithTurnID(context.Background(), want)
if _, err := h.runTurn(ctx, "web", []model.Message{userMsg("你好")}, loop.OwnerEnv()); err != nil {
t.Fatalf("runTurn: %v", err)
}
// 一回合发两帧(stream_chunk + stream_end),两帧的帧壳都必须带上同一个 id。
for _, typ := range []string{"stream_chunk", "stream_end"} {
if ev := waitFrame(t, conn, typ); ev.TurnID != want {
t.Fatalf("%s 帧壳 turnId = %q, want %q(编排器生成的那个)", typ, ev.TurnID, want)
}
}
// 未挂载 ⇒ 每回合一个全新的随机 id(老契约不变)。
if _, err := h.runTurn(context.Background(), "web", []model.Message{userMsg("再来")}, loop.OwnerEnv()); err != nil {
t.Fatalf("runTurn: %v", err)
}
waitFrame(t, conn, "stream_chunk")
ev := waitFrame(t, conn, "stream_end")
if ev.TurnID == "" || ev.TurnID == want {
t.Fatalf("未挂载时应新铸一个 turnId,得到 %q", ev.TurnID)
}
}
// TestEmitTurnVariantCarriesTurnID 钉住②:带 turnId 的状态通道变体把 turnId 盖到
// 帧壳上;未接线该变体时退化为普通状态帧(绝不因此丢掉一条状态迁移)。
func TestEmitTurnVariantCarriesTurnID(t *testing.T) {
t.Parallel()
sink := &turnSink{}
o := NewOrchestrator(OrchestratorOptions{Run: fixedRunner(nil), Emit: sink.Emit, EmitTurn: sink.EmitTurn})
o.emitTurnState("web", "turn-1", EventSessionBusy, busyPayload("web", ModeThinking, 2))
got := sink.snapshot()
if len(got) != 1 || got[0].turnID != "turn-1" || got[0].typ != EventSessionBusy {
t.Fatalf("带 turnId 的变体 = %+v, want turnId=turn-1 的 session_busy", got)
}
if stateInt(got[0], "activeTurns") != 2 {
t.Fatalf("activeTurns = %v, want 2", got[0].data["activeTurns"])
}
plain := &turnSink{}
o2 := NewOrchestrator(OrchestratorOptions{Run: fixedRunner(nil), Emit: plain.Emit})
o2.emitTurnState("web", "turn-1", EventSessionBusy, busyPayload("web", ModeThinking, 1))
got = plain.snapshot()
if len(got) != 1 || got[0].turnID != "" {
t.Fatalf("未接线变体时应退化为普通帧(无 turnId),得到 %+v", got)
}
}
// TestChatStreamRequestDecodesParallel 钉住 WS 上行帧的解码:`parallel:true` 被
// 读懂,省略时为零值(因此省略该字段的老客户端行为逐字不变)。
func TestChatStreamRequestDecodesParallel(t *testing.T) {
t.Parallel()
var on chatStreamRequest
if err := json.Unmarshal([]byte(`{"sessionId":"web","text":"hi","parallel":true}`), &on); err != nil {
t.Fatalf("解码: %v", err)
}
if !on.Parallel {
t.Fatal("parallel:true 没有被解码到")
}
var off chatStreamRequest
if err := json.Unmarshal([]byte(`{"sessionId":"web","text":"hi"}`), &off); err != nil {
t.Fatalf("解码: %v", err)
}
if off.Parallel {
t.Fatal("省略 parallel 时必须为零值")
}
}
// TestRESTParallelAckShapes 钉住 REST 两条入口的应答形状:stream 路径在照准并行时
// 明确回报 parallel+turnId(而不是一个什么都没说的 200),同步路径回 202(第二路在
// 后台跑,拿不到同步回复)。
func TestRESTParallelAckShapes(t *testing.T) {
t.Parallel()
s := &Server{}
rec := httptest.NewRecorder()
s.writeChatStreamAck(rec, SubmitResult{Accepted: true, Immediate: true, Parallel: true, TurnID: "turn-9"}, "web")
if rec.Code != http.StatusOK {
t.Fatalf("stream 并行 ack 状态码 = %d, want 200", rec.Code)
}
var body struct {
Data map[string]any `json:"data"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
t.Fatalf("解码 ack: %v", err)
}
if body.Data["stream"] != true || body.Data["parallel"] != true || body.Data["turnId"] != "turn-9" {
t.Fatalf("stream 并行 ack = %v, want {stream:true, parallel:true, turnId:turn-9}", body.Data)
}
// 现状不动:排队仍是 202 + queued。
rec = httptest.NewRecorder()
s.writeChatStreamAck(rec, SubmitResult{Accepted: true, Pending: true, QueueDepth: 3}, "web")
if rec.Code != http.StatusAccepted {
t.Fatalf("排队 ack 状态码 = %d, want 202", rec.Code)
}
}
// blockingChatter 是**永不返回**的旧式 Chatter(`Deps.Loop`):它把会话钉在
// 「忙」上,使黑盒用例能在真实 HTTP 层断言并行分支的应答形状 —— 而不必让第一条
// 请求同步跑完。它尊重 ctx:网关关闭 / 中断会让它立刻返回,不留泄漏的 goroutine。
type blockingChatter struct {
started chan struct{}
release chan struct{}
stopped sync.Once
}
func newBlockingChatter() *blockingChatter {
return &blockingChatter{started: make(chan struct{}, 8), release: make(chan struct{})}
}
func (b *blockingChatter) Run(ctx context.Context, _ string, _ []model.Message) (*loop.Result, error) {
b.started <- struct{}{}
select {
case <-b.release:
return &loop.Result{Text: "ok", StopReason: "stop"}, nil
case <-ctx.Done():
return nil, ctx.Err()
}
}
func (b *blockingChatter) stop() { b.stopped.Do(func() { close(b.release) }) }
// TestRESTParallelEntriesShareOneShape 是 design/15 §2.5.2 的黑盒回归:REST 的
// 三条入口**同形** ——
// - 第一条经 `stream:true` 进入(提交即返回,回合在后台被钉住,会话因此忙碌);
// - 第二条走默认(同步)入口 + `parallel:true` ⇒ 202 + `parallel:true` + turnId
// (第二路在后台跑,同步入口拿不到回复,绝不回一个 res 为 nil 的 200);
// - 第三条省掉 `parallel` ⇒ 仍是既有的 202 `{queued:true, queueDepth}`。
//
// 它同时覆盖 Deps.MaxParallelTurns → 编排器的接线(第二路真的起来了,用 turnId
// 佐证:只有真的开了一路才会有它)。
func TestRESTParallelEntriesShareOneShape(t *testing.T) {
t.Parallel()
ch := newBlockingChatter()
srv, ts, jwt := buildEnv(t, nil, Deps{Loop: ch, MaxParallelTurns: 2})
t.Cleanup(func() { ch.stop(); srv.Close() })
tok := signAny(t, jwt, "acct", auth.RoleOwner)
resp, _ := doReq(t, http.MethodPost, ts.URL+"/api/v1/chat", map[string]any{
"sessionId": "web", "text": "第一问", "stream": true,
}, tok)
if resp.StatusCode != http.StatusOK {
t.Fatalf("stream 入口状态码 = %d, want 200", resp.StatusCode)
}
select {
case <-ch.started:
case <-time.After(3 * time.Second):
t.Fatal("后台回合没有开跑(会话应当是忙碌的)")
}
if !srv.hub.orchestrator.Busy("web") {
t.Fatal("会话没有进入忙碌态,后面的并行断言不成立")
}
resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/chat", map[string]any{
"sessionId": "web", "text": "第二问", "parallel": true,
}, tok)
if resp.StatusCode != http.StatusAccepted {
t.Fatalf("同步入口 + parallel 状态码 = %d, want 202(第二路在后台跑)", resp.StatusCode)
}
var accepted map[string]any
if err := json.Unmarshal(decodeBody(t, body)["data"], &accepted); err != nil {
t.Fatalf("解码并行 ack: %v (body=%s)", err, body)
}
if accepted["parallel"] != true {
t.Fatalf("同步入口并行 ack = %v, want parallel:true", accepted)
}
if turnID, _ := accepted["turnId"].(string); turnID == "" {
t.Fatalf("并行 ack 必须给出该路的 turnId,得到 %v", accepted)
}
waitActiveTurns(t, srv.hub.orchestrator, "web", 2)
resp, body = doReq(t, http.MethodPost, ts.URL+"/api/v1/chat", map[string]any{
"sessionId": "web", "text": "第三问",
}, tok)
if resp.StatusCode != http.StatusAccepted {
t.Fatalf("未声明并行的忙时提交状态码 = %d, want 202", resp.StatusCode)
}
var queued map[string]any
if err := json.Unmarshal(decodeBody(t, body)["data"], &queued); err != nil {
t.Fatalf("解码排队 ack: %v (body=%s)", err, body)
}
if queued["queued"] != true || queued["parallel"] != nil {
t.Fatalf("未声明并行的忙时 ack = %v, want {queued:true}(形状逐字不变)", queued)
}
}
// TestParallelTurnContextIsAdditive 钉住并行声明的 ctx 通道本身:false 是空操作
// (ctx 上不留痕迹),nil ctx 不 panic。
func TestParallelTurnContextIsAdditive(t *testing.T) {
t.Parallel()
if parallelRequested(withParallelTurn(context.Background(), false)) {
t.Fatal("false 必须是空操作:老客户端的 ctx 上不该留任何痕迹")
}
if !parallelRequested(withParallelTurn(context.Background(), true)) {
t.Fatal("true 必须可被读取侧看到")
}
if parallelRequested(nil) {
t.Fatal("nil ctx 不该被判为声明了并行")
}
}
@@ -0,0 +1,531 @@
package gateway
import (
"context"
"fmt"
"log/slog"
"sync"
"testing"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// 本文件是 design/15 §2.5「忙时并行回合(多路)」**基础管线**的回归套件
//(主人 2026-10-05 裁决 B)。它钉住六件事:
//
// ① 省略 `parallel` ⇒ 行为逐字不变(不并行、不新增帧键、报空闲时点不变);
// ② 显式 `parallel` 且额度未满 ⇒ 真的起第二路(两条不同的 turnId、各自的路标识);
// ③ 额度满 ⇒ 落回排队(不报错、不丢消息),并如实给出 reason;
// ④ 单路时 `session_idle` 的时点不变;
// ⑤ 并行时 `session_idle` 等到**所有路**收尾;
// ⑥ turnId 一致性(编排器生成的那个 = 交给 Run、并由流帧盖上壳的那个)。
//
// 夹具刻意用**可逐路放行**的 runner(放行顺序即收尾顺序),因此「谁先收尾」是
// 确定性的,断言的是时序而不是睡眠。
// gatedLegRunner 是可控的 TurnRunner:每条路的 Run 都阻塞在自己的放行闸上,
// 测试按启动顺序逐路放行,因此能精确断言「一路收尾而另一路仍在飞」的那一刻。
type gatedLegRunner struct {
mu sync.Mutex
calls []callRecord
gates []chan struct{}
started chan struct{}
}
func newGatedLegRunner() *gatedLegRunner {
return &gatedLegRunner{started: make(chan struct{}, 8)}
}
func (g *gatedLegRunner) Run(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) (*loop.Result, error) {
gate := make(chan struct{})
g.mu.Lock()
idx := len(g.calls)
g.calls = append(g.calls, callRecord{sessionID: sessionID, messages: append([]model.Message(nil), messages...), env: env, ctx: ctx})
g.gates = append(g.gates, gate)
g.mu.Unlock()
g.started <- struct{}{}
select {
case <-gate:
case <-ctx.Done():
return nil, ctx.Err()
}
return &loop.Result{Text: fmt.Sprintf("reply-%d", idx+1), Steps: 1, StopReason: "stop"}, nil
}
// waitStarted 等到至少 n 条路真的进了 runner(而不是只被登记)。
func (g *gatedLegRunner) waitStarted(n int) []callRecord {
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
g.mu.Lock()
got := len(g.calls)
out := append([]callRecord(nil), g.calls...)
g.mu.Unlock()
if got >= n {
return out
}
time.Sleep(time.Millisecond)
}
panic("timed out waiting for leg starts")
}
// release 放行第 i 条启动的路(i 从 0 起)。它幂等(重复关闭会 panic,
// 因此用 select 保护),使测试的放行顺序就是收尾顺序。
func (g *gatedLegRunner) release(i int) {
g.mu.Lock()
gate := g.gates[i]
g.mu.Unlock()
select {
case <-gate:
default:
close(gate)
}
}
// turnStateRec 是一次状态帧(含帧壳上的 turnId,可能为空)。
type turnStateRec struct {
sessionID string
turnID string
typ string
data map[string]any
}
// turnSink 同时实现 StateSink 与 TurnStateSink,把两类帧记在同一个序列里,
// 使「带 turnId 的变体真的把 turnId 盖到帧壳上」可被直接断言。
type turnSink struct {
mu sync.Mutex
events []turnStateRec
}
func (s *turnSink) Emit(sessionID, typ string, data map[string]any) {
s.record(turnStateRec{sessionID: sessionID, typ: typ, data: data})
}
func (s *turnSink) EmitTurn(sessionID, turnID, typ string, data map[string]any) {
s.record(turnStateRec{sessionID: sessionID, turnID: turnID, typ: typ, data: data})
}
func (s *turnSink) record(rec turnStateRec) {
s.mu.Lock()
s.events = append(s.events, rec)
s.mu.Unlock()
}
func (s *turnSink) snapshot() []turnStateRec {
s.mu.Lock()
defer s.mu.Unlock()
return append([]turnStateRec(nil), s.events...)
}
// ofType 返回某类状态帧(按发生顺序)。
func (s *turnSink) ofType(typ string) []turnStateRec {
var out []turnStateRec
for _, e := range s.snapshot() {
if e.typ == typ {
out = append(out, e)
}
}
return out
}
// parallelFixture 是并行管线用例的公共夹具。
type parallelFixture struct {
o *Orchestrator
run *gatedLegRunner
sink *turnSink
history *fakeHistory
}
// newParallelFixture 按给定的队列容量与并行上限构建编排器(Log 走丢弃型,
// 避免并行用例互相污染日志断言)。
func newParallelFixture(maxPending, maxParallel int) *parallelFixture {
run := newGatedLegRunner()
sink := &turnSink{}
hist := &fakeHistory{}
o := NewOrchestrator(OrchestratorOptions{
Run: run.Run,
Emit: sink.Emit,
EmitTurn: sink.EmitTurn,
Log: slog.New(slog.NewTextHandler(discardWriter{}, nil)),
MaxPending: maxPending,
MaxTail: 20,
History: hist,
MaxParallelTurns: maxParallel,
})
return &parallelFixture{o: o, run: run, sink: sink, history: hist}
}
// discardWriter 接收被丢弃的日志字节(用例只关心状态帧与历史行)。
type discardWriter struct{}
func (discardWriter) Write(p []byte) (int, error) { return len(p), nil }
// waitActiveTurns 等到某会话的在途路数等于 want(超时即失败),使断言不靠睡眠。
func waitActiveTurns(t *testing.T, o *Orchestrator, sessionID string, want int) {
t.Helper()
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
if o.ActiveTurns(sessionID) == want {
return
}
time.Sleep(time.Millisecond)
}
t.Fatalf("在途路数 = %d, want %d", o.ActiveTurns(sessionID), want)
}
// stateInt 从状态帧的 data 里取一个整数键(缺失或类型不符时返回 -1)。
func stateInt(rec turnStateRec, key string) int {
v, ok := rec.data[key].(int)
if !ok {
return -1
}
return v
}
// stateKeyPresent 报告状态帧的 data 里某个键是否**在场**(用于「该键缺席」的断言:
// 缺席与空串对客户端含义不同,绝不伪造)。
func stateKeyPresent(rec turnStateRec, key string) bool {
_, ok := rec.data[key]
return ok
}
// assistantContents 返回已落库记录里的助手回复正文(按顺序)。
func assistantContents(msgs []history.Message) []string {
out := make([]string, 0, len(msgs))
for _, m := range msgs {
if m.Role == history.RoleAssistant {
out = append(out, m.Content)
}
}
return out
}
// windowText 返回一路播种到的模型消息正文(按顺序),供断言 I4「每路一份窗口」。
func windowText(msgs []model.Message) []string {
out := make([]string, 0, len(msgs))
for _, m := range msgs {
out = append(out, messageText(m))
}
return out
}
// parallelCtx 返回一条**显式声明并行**的提交上下文(与 WS/REST 边界同源的 ctx 通道)。
func parallelCtx() context.Context { return withParallelTurn(context.Background(), true) }
// ---------------------------------------------------------------------------
// ① 省略 parallel ⇒ 逐字不变
// ---------------------------------------------------------------------------
// TestParallelOmittedKeepsQueueingBehaviour 钉住「老客户端零影响」:不声明 parallel
// 的忙时消息仍然入队,帧形状不变(**没有** reason 键),且始终只有一条在途路。
func TestParallelOmittedKeepsQueueingBehaviour(t *testing.T) {
t.Parallel()
f := newParallelFixture(5, 2)
if sr := f.o.Submit(context.Background(), "web", []model.Message{userMsg("第一问")}, loop.OwnerEnv()); !sr.Immediate || sr.Parallel {
t.Fatalf("空闲提交 = %+v, want Immediate 且非 Parallel", sr)
}
f.run.waitStarted(1)
if got := f.o.ActiveTurns("web"); got != 1 {
t.Fatalf("在途路数 = %d, want 1", got)
}
// 忙时第二条(未声明并行)⇒ 排队,且帧上没有任何新键。
sr := f.o.Submit(context.Background(), "web", []model.Message{userMsg("第二问")}, loop.OwnerEnv())
if !sr.Pending || sr.Parallel || sr.QueueDepth != 1 {
t.Fatalf("忙时未声明并行的提交 = %+v, want Pending(1) 且非 Parallel", sr)
}
if got := f.o.ActiveTurns("web"); got != 1 {
t.Fatalf("排队时在途路数 = %d, want 1(绝不新开路)", got)
}
acks := f.sink.ofType(EventPendingAck)
if len(acks) != 1 {
t.Fatalf("pending_ack 帧数 = %d, want 1", len(acks))
}
if stateKeyPresent(acks[0], "reason") {
t.Fatalf("未声明并行的排队不该带 reason,得到 %v", acks[0].data)
}
if stateInt(acks[0], "queueDepth") != 1 {
t.Fatalf("pending_ack queueDepth = %v, want 1", acks[0].data["queueDepth"])
}
if stateInt(acks[0], "activeTurns") != -1 {
t.Fatalf("pending_ack 不该带 activeTurns(契约里只有 session_busy/idle/drain/interrupted 带它):%v", acks[0].data)
}
// 放行第一路 ⇒ 弹队首续跑(仍是串行,一条在途路)。
f.run.release(0)
calls := f.run.waitStarted(2)
if got := f.o.ActiveTurns("web"); got != 1 {
t.Fatalf("续跑时在途路数 = %d, want 1", got)
}
if got := loop.ParallelLegFrom(calls[1].ctx); got != "" {
t.Fatalf("续跑路的路标识 = %q, want 空串(串行路不挂并发路标识)", got)
}
if loop.TurnIDFrom(calls[1].ctx) == loop.TurnIDFrom(calls[0].ctx) {
t.Fatal("续跑路必须是一条**新路**,turnId 不能与上一路相同")
}
drains := f.sink.ofType(EventPendingDrain)
if len(drains) != 1 || stateInt(drains[0], "activeTurns") != 1 {
t.Fatalf("pending_drain = %v, want 一帧且 activeTurns=1", drains)
}
f.run.release(1)
if !waitIdle(f.o, "web") {
t.Fatal("会话没有回到空闲")
}
idle := f.sink.ofType(EventSessionIdle)
if len(idle) != 1 || stateInt(idle[0], "activeTurns") != 0 {
t.Fatalf("session_idle = %v, want 一帧且 activeTurns=0", idle)
}
busy := f.sink.ofType(EventSessionBusy)
if len(busy) != 1 || stateInt(busy[0], "activeTurns") != 1 {
t.Fatalf("session_busy = %v, want 一帧且 activeTurns=1", busy)
}
if got := len(assistantContents(f.history.records())); got != 2 {
t.Fatalf("助手回复行数 = %d, want 2(每条消息各答一次)", got)
}
}
// TestSessionIdleTimingSingleLegUnchanged 钉住④:单路(含排队续跑)的
// `session_idle` 时点与现状一致 —— 队列空且本路收尾就报,绝不多等、绝不少报。
func TestSessionIdleTimingSingleLegUnchanged(t *testing.T) {
t.Parallel()
f := newParallelFixture(5, 2)
if sr := f.o.Submit(context.Background(), "web", []model.Message{userMsg("甲")}, loop.OwnerEnv()); !sr.Immediate {
t.Fatalf("提交 = %+v, want Immediate", sr)
}
f.run.waitStarted(1)
if got := f.sink.ofType(EventSessionIdle); len(got) != 0 {
t.Fatalf("回合在飞时不该报空闲,得到 %v", got)
}
if got := f.o.ActiveTurns("web"); got != 1 {
t.Fatalf("在途路数 = %d, want 1", got)
}
f.run.release(0)
if !waitIdle(f.o, "web") {
t.Fatal("会话没有回到空闲")
}
if got := f.sink.ofType(EventSessionIdle); len(got) != 1 {
t.Fatalf("session_idle 帧数 = %d, want 1", len(got))
}
if got := f.o.ActiveTurns("web"); got != 0 {
t.Fatalf("收尾后在途路数 = %d, want 0", got)
}
}
// ---------------------------------------------------------------------------
// ② 显式 parallel 且额度未满 ⇒ 真的起第二路
// ---------------------------------------------------------------------------
// TestParallelDeclaredStartsSecondLeg 钉住②与⑥:第二条显式声明并行的消息**立即**
// 起一条独立的路(两路 turnId 不同、并行路带自己的路标识),两路同时生长,各自
// 落一行助手回复,且 `session_idle` 只在两路都收尾后出现一次。
func TestParallelDeclaredStartsSecondLeg(t *testing.T) {
t.Parallel()
f := newParallelFixture(5, 2)
first := f.o.Submit(context.Background(), "web", []model.Message{userMsg("第一问")}, loop.OwnerEnv())
if !first.Immediate || first.TurnID == "" {
t.Fatalf("第一路 = %+v, want Immediate 且带 turnId", first)
}
f.run.waitStarted(1)
second := f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("第二问")}, loop.OwnerEnv())
if !second.Immediate || !second.Parallel {
t.Fatalf("显式并行提交 = %+v, want Immediate 且 Parallel", second)
}
if second.TurnID == "" || second.TurnID == first.TurnID {
t.Fatalf("第二路 turnId = %q,第一路 = %q:必须各自不同", second.TurnID, first.TurnID)
}
if f.o.QueueDepth("web") != 0 {
t.Fatalf("并行路不该入队,队列深度 = %d", f.o.QueueDepth("web"))
}
calls := f.run.waitStarted(2)
waitActiveTurns(t, f.o, "web", 2)
// I4:每一路都有**自己的一份窗口** —— 各自带来自己那条用户消息,且看不到
// 对方在跑的东西(对方的答复尚未落库,因此不在任何一路的转录里)。
if got := windowText(calls[0].messages); len(got) != 1 || got[0] != "第一问" {
t.Fatalf("第一路的窗口 = %v, want [第一问]", got)
}
if got := windowText(calls[1].messages); len(got) != 1 || got[0] != "第二问" {
t.Fatalf("第二路的窗口 = %v, want [第二问]", got)
}
// ⑥ 两路各自的 turnId:第一路不带并发路标识(checkpoint 归属键仍是 sessionID),
// 第二路带自己的(归属键升为 {sessionID}#{turnID}),且都与提交时交回的 turnId 同值。
if got := loop.TurnIDFrom(calls[0].ctx); got != first.TurnID {
t.Fatalf("第一路 ctx turnId = %q,提交回报 %q:两处必须同值", got, first.TurnID)
}
if got := loop.ParallelLegFrom(calls[0].ctx); got != "" {
t.Fatalf("第一路不该挂并发路标识,得到 %q", got)
}
if got := loop.TurnIDFrom(calls[1].ctx); got != second.TurnID {
t.Fatalf("第二路 ctx turnId = %q,提交回报 %q:两处必须同值", got, second.TurnID)
}
if got := loop.ParallelLegFrom(calls[1].ctx); got != second.TurnID {
t.Fatalf("第二路路标识 = %q, want %q(checkpoint 归属键按路隔离)", got, second.TurnID)
}
// 忙态帧如实报告两路在途。
busy := f.sink.ofType(EventSessionBusy)
if len(busy) != 2 || stateInt(busy[1], "activeTurns") != 2 {
t.Fatalf("session_busy 帧 = %v, want 两帧且第二帧 activeTurns=2", busy)
}
// ⑤ 先收尾第二路:会话仍不空闲(第一路还在飞)。
f.run.release(1)
waitActiveTurns(t, f.o, "web", 1)
if got := f.sink.ofType(EventSessionIdle); len(got) != 0 {
t.Fatalf("还有一路在飞时不该报空闲,得到 %v", got)
}
f.run.release(0)
if !waitIdle(f.o, "web") {
t.Fatal("所有路收尾后会话没有回到空闲")
}
idle := f.sink.ofType(EventSessionIdle)
if len(idle) != 1 || stateInt(idle[0], "activeTurns") != 0 {
t.Fatalf("session_idle = %v, want 恰好一帧且 activeTurns=0", idle)
}
// 两路各落一行助手回复(I5:绝不合并成一行、不覆盖对方),且每条用户消息只被
// 作答一次(两条 user 行、两条 assistant 行)。
rows := f.history.records()
if got := len(assistantContents(rows)); got != 2 {
t.Fatalf("助手回复行数 = %d, want 2", got)
}
users := 0
for _, m := range rows {
if m.Role == history.RoleUser {
users++
}
}
if users != 2 {
t.Fatalf("用户消息行数 = %d, want 2(恰好一次)", users)
}
}
// TestSessionIdleWaitsForAllLegs 是⑤的**最小时序**回归:并行两路,任意顺序收尾,
// 只有第二路也收尾时才出现唯一的 session_idle。
func TestSessionIdleWaitsForAllLegs(t *testing.T) {
t.Parallel()
f := newParallelFixture(5, 2)
f.o.Submit(context.Background(), "web", []model.Message{userMsg("A")}, loop.OwnerEnv())
f.run.waitStarted(1)
f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("B")}, loop.OwnerEnv())
f.run.waitStarted(2)
waitActiveTurns(t, f.o, "web", 2)
// 先收尾**第一**路(与上一个用例相反的收尾顺序,证明判据不是「最后启动的那一路」)。
f.run.release(0)
waitActiveTurns(t, f.o, "web", 1)
if got := f.sink.ofType(EventSessionIdle); len(got) != 0 {
t.Fatalf("一路收尾、另一路仍在飞 ⇒ 不该报空闲,得到 %v", got)
}
f.run.release(1)
if !waitIdle(f.o, "web") {
t.Fatal("所有路收尾后会话没有回到空闲")
}
if got := f.sink.ofType(EventSessionIdle); len(got) != 1 {
t.Fatalf("session_idle 帧数 = %d, want 1", len(got))
}
}
// ---------------------------------------------------------------------------
// ③ 额度满 ⇒ 落回排队
// ---------------------------------------------------------------------------
// TestParallelQuotaFullFallsBackToQueue 钉住③:额度(默认 2)用满后,再来的显式
// 并行消息**落回既有排队语义** —— 不报错、不打断、不丢,并如实给出
// `pending_ack{reason:"parallel_limit"}`;被排队的消息最终仍被作答一次。
func TestParallelQuotaFullFallsBackToQueue(t *testing.T) {
t.Parallel()
f := newParallelFixture(5, 2)
f.o.Submit(context.Background(), "web", []model.Message{userMsg("第一问")}, loop.OwnerEnv())
f.run.waitStarted(1)
if sr := f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("第二问")}, loop.OwnerEnv()); !sr.Parallel {
t.Fatalf("额度未满时应当照准并行,得到 %+v", sr)
}
f.run.waitStarted(2)
waitActiveTurns(t, f.o, "web", 2)
// 第三条声明并行的消息:额度已满 ⇒ 排队。
third := f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("第三问")}, loop.OwnerEnv())
if !third.Accepted || !third.Pending || third.Parallel {
t.Fatalf("额度满的并行提交 = %+v, want Accepted+Pending 且非 Parallel", third)
}
if third.QueueDepth != 1 {
t.Fatalf("队列深度 = %d, want 1", third.QueueDepth)
}
acks := f.sink.ofType(EventPendingAck)
if len(acks) != 1 || acks[0].data["reason"] != pendingAckReasonParallelLimit {
t.Fatalf("pending_ack = %v, want reason=%q", acks, pendingAckReasonParallelLimit)
}
if got := f.o.ActiveTurns("web"); got != 2 {
t.Fatalf("落回排队不该改变在途路数,得到 %d", got)
}
// 逐路收尾:排队的那条由结算中的一路续跑(起一条**串行**新路),最终被作答。
f.run.release(0)
f.run.waitStarted(3)
f.run.release(1)
f.run.release(2)
if !waitIdle(f.o, "web") {
t.Fatal("会话没有回到空闲")
}
if got := len(assistantContents(f.history.records())); got != 3 {
t.Fatalf("助手回复行数 = %d, want 3(排队的第三条也必须被作答一次)", got)
}
}
// TestParallelDisabledByLimitBelowTwo 钉住「<2 = 关闭真并行 ⇒ 完全回落到现状」:
// 上限为 1 时显式并行也一律排队,在途路数恒不越过 1。
func TestParallelDisabledByLimitBelowTwo(t *testing.T) {
t.Parallel()
f := newParallelFixture(5, 1)
f.o.Submit(context.Background(), "web", []model.Message{userMsg("第一问")}, loop.OwnerEnv())
f.run.waitStarted(1)
sr := f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("第二问")}, loop.OwnerEnv())
if !sr.Pending || sr.Parallel {
t.Fatalf("真并行被关闭时 = %+v, want Pending 且非 Parallel", sr)
}
if got := f.o.ActiveTurns("web"); got != 1 {
t.Fatalf("在途路数 = %d, want 1(关闭真并行)", got)
}
acks := f.sink.ofType(EventPendingAck)
if len(acks) != 1 || acks[0].data["reason"] != pendingAckReasonParallelLimit {
t.Fatalf("pending_ack = %v, want 仍如实给出 reason", acks)
}
f.run.release(0)
f.run.waitStarted(2)
f.run.release(1)
if !waitIdle(f.o, "web") {
t.Fatal("会话没有回到空闲")
}
}
// TestResolveMaxParallelTurns 钉住额度解析的三档边界(默认 / 关闭 / 硬上限)。
func TestResolveMaxParallelTurns(t *testing.T) {
t.Parallel()
cases := []struct {
in int
want int
}{
{-5, defaultMaxParallelTurns},
{0, defaultMaxParallelTurns},
{1, 1},
{2, 2},
{3, 3},
{99, maxParallelTurnsHardCap},
}
for _, c := range cases {
if got := resolveMaxParallelTurns(c.in); got != c.want {
t.Fatalf("resolveMaxParallelTurns(%d) = %d, want %d", c.in, got, c.want)
}
}
}
@@ -0,0 +1,46 @@
package gateway
import (
"context"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// 本文件归属会话编排器的**待处理队列**(design/15 §2.1/§2.2):忙时到达、且
// 未声明并行(或并行额度已满)的消息怎么入队、怎么落库、怎么回 ack。它从
// session_orchestrator.go 拆出,使那个文件继续只留状态机骨架(并守住
// 「函数 ≤80 行」的硬约束)。
// enqueueLocked 把一条忙时到达的用户消息放进有界待处理队列,并返回本次提交的
// 判定。它**必须在 o.mu 下调用**,并在返回前解锁:落库是慢操作,绝不能在状态
// 锁下做,因此它自己解锁后再做落库与状态帧广播(与 startParallelLocked 同一
// 约定,调用方直接 return 它的返回值)。
//
// 入队时携带该消息到达时的设备 / 模式 / 语音回调,使它的续跑回合与「它就是当前
// 回合」时逐项一致(design/16 §4.5、§5.1;协同 C1)。落库两段式的第一段在这里:
// 用户消息立刻进历史(抽取结论等回合跑起来才有,由续跑结算回写,见
// session_document_status.go)。
func (o *Orchestrator) enqueueLocked(ctx context.Context, st *sessionState, sessionID string, messages []model.Message, env loop.SessionEnv, deviceID, msgMode string, hint clientMsgHint) SubmitResult {
st.pendingSeq++
seq := st.pendingSeq
st.pending = append(st.pending, pendingMsg{
messages: messages,
env: env,
deviceID: deviceID,
mode: msgMode,
at: time.Now(),
seq: seq,
voice: voiceReplySinkFrom(ctx),
})
depth := len(st.pending)
mode := st.mode
o.mu.Unlock()
o.attachPendingDocuments(sessionID, seq, o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID))
// 声明了并行却落回排队时如实给出 reason,让客户端把气泡渲染成「排队中」而不是
// 「被拒」(design/15 §2.6);未声明时该键缺席,帧与今天逐字相同。
o.emitState(sessionID, EventPendingAck, pendingAckPayload(sessionID, mode, depth, parallelRequested(ctx)))
return SubmitResult{Accepted: true, Pending: true, QueueDepth: depth, MessageID: hint.MessageID}
}
+133
View File
@@ -0,0 +1,133 @@
package gateway
import "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
// 本文件归属会话编排器的**状态查询与状态帧出口**:谁在读「这个会话忙不忙」,
// 以及状态迁移怎么交给传输层。它从 session_orchestrator.go 拆出,使那个文件只留
// 消息状态机(Submit / pending / 续跑 / 结算),并保持在「单文件 ≤500 行」的
// 硬约束内(与 session_events.go / session_tail.go 同一理由)。
// StateSink 把每会话状态转移(busy/pending/idle/interrupt)报告给
// 传输层。网关把它接到 WS hub 广播上。
type StateSink func(sessionID, typ string, data map[string]any)
// TurnStateSink 是 StateSink 的**带 turnId 变体**(design/15 §2.5.9):帧壳上的
// `turnId` 是客户端按回合两级分桶的依据,而少数帧(后续切片的 `turn_begin`、
// 以及 `steer_ack`)必须让客户端知道**是哪一路**,只有 data 不足以表达。
//
// 它刻意是**另一个**回调而不是改 StateSink 的签名:既有接线点(server.go、
// 大量测试)的签名因此零改动,且空 turnID 时明确退化为现签名(帧壳上没有
// turnId,与今天逐字相同)。未接线该变体时,带 turnId 的状态帧退化为普通帧
// ——绝不因此丢一条状态迁移(状态是增益,不是回合能否跑完的前提)。
type TurnStateSink func(sessionID, turnID, typ string, data map[string]any)
// SetMode 提升某会话所报告的忙碌状态通道(thinking →
// executing → subagent)。它是 loop/subagent 钩子可调用的注入点;
// 任何非空闲模式都让会话保持忙碌,而 idle 解除忙碌。为 nil 的会话
// 被惰性创建,因此钩子可以在第一条消息之前设模式。
//
// 注意它与「在途路」的关系:真正的忙碌由 st.legs 承载(有路在飞就是忙),
// mode 只是给客户端看的通道标签。因此 SetMode 不会被并行的第二路顶掉,
// 也不会顶掉第二路的忙态。
//
// **已知偏离(诚实登记)**:design/15 §2.5.4 要求多路下 mode「取最重的一路」
// (subagent > executing > thinking)。那需要一个**按路**的模式源,而今天
// 提升模式下井的唯一入口是本函数(会话级、且生产路径上暂无调用方)—— 因此本
// 切片如实保留「会话级 mode:新路起于 thinking、全部收尾才回 idle」的既有语义,
// 不伪造一个按路的轻/重判定。等模式提升改成按路注入时,这里再实现加权。
func (o *Orchestrator) SetMode(sessionID string, mode Mode) {
sessionID = o.normalize(sessionID)
o.mu.Lock()
st := o.states[sessionID]
if st == nil {
st = &sessionState{mode: mode}
o.states[sessionID] = st
} else {
st.mode = mode
}
activeTurns := len(st.legs)
o.mu.Unlock()
if mode == ModeIdle {
o.emitState(sessionID, EventSessionIdle, busyPayload(sessionID, mode, activeTurns))
return
}
o.emitState(sessionID, EventSessionBusy, busyPayload(sessionID, mode, activeTurns))
}
// Busy 报告该会话当前是否有在飞回合。
func (o *Orchestrator) Busy(sessionID string) bool {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[o.normalize(sessionID)]
return st != nil && st.mode != ModeIdle
}
// ActiveTurns 报告该会话当前的在途路数(design/15 §2.6 的 `activeTurns`)。
// 单路时它非 0 即 1,与「忙不忙」同值;并行时它才是真正的路数 —— 恰好是所有路
// 都收尾后才报 0 的那个判据(design/15 §2.5.3)。
func (o *Orchestrator) ActiveTurns(sessionID string) int {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[o.normalize(sessionID)]
if st == nil {
return 0
}
return len(st.legs)
}
// QueueDepth 报告某会话的待处理回合数。
func (o *Orchestrator) QueueDepth(sessionID string) int {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[o.normalize(sessionID)]
if st == nil {
return 0
}
return len(st.pending)
}
// ModeOf 报告某会话当前的忙碌状态通道(未知时为 idle)。
func (o *Orchestrator) ModeOf(sessionID string) Mode {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[o.normalize(sessionID)]
if st == nil {
return ModeIdle
}
return st.mode
}
// normalize 把空会话 id 映射为默认值,使状态 map 对省略它的
// 调用方保持稳定(Web 主会话)。
func (o *Orchestrator) normalize(sessionID string) string {
if sessionID == "" {
return defaultSessionID
}
return sessionID
}
// emitState 在接线了传输层时把状态转移转发给它。
func (o *Orchestrator) emitState(sessionID, typ string, data map[string]any) {
if o.emit != nil {
o.emit(sessionID, typ, data)
}
}
// emitTurnState 发出一个**帧壳带 turnId** 的状态转移(design/15 §2.5.9)。
// 未接线带 turnId 的变体时退化为普通状态帧(空 turnID 与现签名等价),
// 因此未接线的构建与既有测试绝不因此丢掉一条状态迁移。
func (o *Orchestrator) emitTurnState(sessionID, turnID, typ string, data map[string]any) {
if o.emitTurn != nil {
o.emitTurn(sessionID, turnID, typ, data)
return
}
o.emitState(sessionID, typ, data)
}
// notifyPersisted 在接线了传输钩子时把刚持久化的消息交给它,
// 使网关能广播 `message_persisted` 以进行跨端同步。
func (o *Orchestrator) notifyPersisted(msg history.Message) {
if o.onPersisted != nil {
o.onPersisted(msg)
}
}
+11 -6
View File
@@ -186,13 +186,13 @@ func candidateSeqs(cs []steerCandidate) map[uint64]bool {
}
// pendingSnapshot 在状态锁下取该会话待处理队列的一份副本(世代不符时返回 nil:
// 那份队列已不属于本回合,例如被中断清空或被更新的顶层回合取代)。取副本而不是
// 那条路已失效,例如被打断、被取代或会话已清空)。取副本而不是
// 持锁做转录读取,是为了不让有界存储读取堵住 Submit。
func (o *Orchestrator) pendingSnapshot(sessionID string, gen uint64) []pendingMsg {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[sessionID]
if st == nil || st.gen != gen || len(st.pending) == 0 {
if o.legLocked(st, gen) == nil || len(st.pending) == 0 {
return nil
}
return append([]pendingMsg(nil), st.pending...)
@@ -202,9 +202,14 @@ func (o *Orchestrator) pendingSnapshot(sessionID string, gen uint64) []pendingMs
// 被第二个回合再答一遍)。它按 pendingMsg.seq 认领,因此即便队列在同一时刻被追加
// (只会加在队尾)或被中断清空,也绝不会误删别的消息。
//
// 返回 false 表示这次认领**没有生效**(队列里已无这些序号,或世代已被中断/更新的
// 顶层回合取代)——调用方此时必须放弃注入:否则这条消息会被本回合与那条新路径各答
// 一次。它必须在把消息交给循环**之前**调用。
// 返回 false 表示这次认领**没有生效**(队列里已无这些序号,或本路已失效
// ——被打断、被取代)。调用方此时必须放弃注入:否则这条消息会被本回合与那条
// 新路径各答一次。它必须在把消息交给循环**之前**调用。
//
// 多路口径(design/15 §2.5.2 I2):认领仍只走这一把锁与这一套序号,因此
// 「新起的第 2 路」与「在飞路的下一个 steer 边界」抢同一条消息时也只有一个赢家
// ——升级路径本就不把消息放进 pending(它直接登记为一条新路),而 steer 只能
// 认领 pending 里的条目,两者不可能同时拿走同一条。
func (o *Orchestrator) consumePending(sessionID string, gen uint64, seqs map[uint64]bool) bool {
if len(seqs) == 0 {
return false
@@ -212,7 +217,7 @@ func (o *Orchestrator) consumePending(sessionID string, gen uint64, seqs map[uin
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[sessionID]
if st == nil || st.gen != gen {
if o.legLocked(st, gen) == nil {
return false
}
kept := make([]pendingMsg, 0, len(st.pending))
@@ -7,8 +7,10 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// 本文件把「同一会话最多一个在途回合」从一句约定变成可等待的进程内围栏
// (design/15 §3 与裁决 13.4.5 的组合要求)。
// 本文件把「同一会话的**串行续跑链**上一次只有一个回合」从一句约定变成可等待的
// 进程内围栏(design/15 §3 与裁决 13.4.5 的组合要求;design/15 §2.5 落地真并行后,
// 需要这条护栏的仍只是串行链 —— 显式声明的并行路用**自己的 checkpoint 归属键**
// 消除同一竞态,因此不接入这条链)。
//
// 中断必须**立即**把会话交回空闲并接受新消息(design/15 §3),但被取消的
// 回合仍在收尾:loop 直到它自己的 Run 返回之前,才把被放弃回合的快照落定
@@ -21,9 +23,9 @@ import (
// 围栏关掉这条竞态窗口,且不阻塞调用方:新消息照样被立即接受、Submit 立即
// 返回(非阻塞是硬约束),只是它的回合要等上一回合真正返回之后才开始。
// turnFence 是一次回合的启动围栏。wait 是上一回合的收尾信号(nil 表示没有
// 在途的上一回合,无需等待);done 是本回合的收尾信号,在本回合返回时关闭,
// 从而放行等它的下一回合。
// turnFence 是**一条路**的启动围栏。wait 是上一串行回合的收尾信号(nil 表示没有
// 在途的上一回合,无需等待 —— 并行路与续跑路都用 nil);done 是本路的收尾信号,
// 在本路返回时关闭,从而放行等它的下一路。
type turnFence struct {
wait <-chan struct{}
done chan struct{}
@@ -28,8 +28,8 @@ import (
// 三类**不补**的情形(边界,与既有语义不冲突):
// - 被 interrupt 取消的回合:用户意图就是「停下」,此刻补一条「我出问题了」
// 反而违背那条指令;
// - 被更新的回合取代的陈旧代次(st.gen 已前进):那次收尾不属于当前会话状态,
// 补消息会落到别人的回合里;
// - 已经失效的路(被 interrupt 取消后又被清空、或会话从未登记过它):那次收尾
// 不属于当前会话状态,补消息会落到别人的回合里;
// - 静默内部流程(backthink、子代理):它们各自驱动 loop,从不经过本编排器,
// 因而天然不受这条保证影响(它们的产出只进内部状态,不进聊天历史)。
@@ -37,13 +37,17 @@ import (
// internal/runtime/turnreply 统一给出(IM 渠道桥走另一条编排路径,两边必须是
// 同一份诚实说明;各写一套迟早漂移)。本文件只负责判定**这次收尾该不该落库**。
// turnIsCurrent 报告该代次是否仍是该会话的当前代次。陈旧代次(已被 interrupt
// 或更新的顶层回合取代)意味着这次收尾属于**上一个**回合,不该再往历史里写。
// turnIsCurrent 报告该路是否仍是该会话的**在途路**。已失效的路(被打断、被
// interrupt-all 清空、或会话从未登记过它)意味着这次收尾属于**一条已经不存在的
// 回合**,不该再往历史里写。
//
// 判据从旧的「世代号等于会话当前代次」改成「这条路的世代号还在 st.legs 里」:
// 真并行下启动第二路会给它分配一个新世代号,但**绝不会**让第一路失效
// (design/15 §2.5.9 —— 这正是旧的会话级单例代次做不到的那件事)。
func (o *Orchestrator) turnIsCurrent(sessionID string, gen uint64) bool {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[sessionID]
return st != nil && st.gen == gen
return o.legLocked(o.states[sessionID], gen) != nil
}
// persistTurnReply 是本回合助手回复的**唯一落库入口**(completeTurn 的收尾第一
+22 -15
View File
@@ -40,8 +40,8 @@ func voiceReplySinkFrom(ctx context.Context) VoiceReplyFunc {
return sink
}
// voiceTurn 配对一次语音回合与它的收尾回调(按代次配对,与 turnDocuments
// 同理):被打断或被新回合取代的陈旧回合在结算时不该替别人回音频。
// voiceTurn 配对一条**路**与它的语音收尾回调(按世代号配对,与 turnDocuments
// 同理):被中断 / 已失效的路在结算时不该替别人回音频。
//
// sinks 是切片,因为一个回合可能同时有**多条**语音请求在等音频:忙时入 pending
// 的语音消息可能被转向注入(design/15 §2.4,见 session_steer.go 的 consumePending)
@@ -60,37 +60,44 @@ func voiceSinksOf(sink VoiceReplyFunc) []VoiceReplyFunc {
return []VoiceReplyFunc{sink}
}
// rememberVoiceTurn 记下本回合的语音收尾回调。**每个新回合都必须调用它一次**
// rememberVoiceTurn 记下**本路**的语音收尾回调。**每条新路都必须调用它一次**
// (纯文本回合记 nil):否则一个陈旧的语音回调会被沿用到后面的文本回合上,
// 让一次文本问答无端合成一段音频。
//
// 它按路配对(design/15 §2.5.9):并行时两路各自持有自己的回调,一路结算
// 绝不会替另一路回音频。路已失效(被打断清空)时什么都不做。
func (o *Orchestrator) rememberVoiceTurn(sessionID string, gen uint64, sink VoiceReplyFunc) {
o.mu.Lock()
defer o.mu.Unlock()
if st := o.states[sessionID]; st != nil {
st.voiceTurn = voiceTurn{gen: gen, sinks: voiceSinksOf(sink)}
if leg := o.legLocked(o.states[sessionID], gen); leg != nil {
leg.voice = voiceTurn{gen: gen, sinks: voiceSinksOf(sink)}
}
}
// addVoiceSinkLocked 把一条**被转向注入**的语音请求的收尾回调追加到当前回合上。
// 必须在 o.mu 下调用;代次不符(那条请求不属于本回合)时什么都不做。
// addVoiceSinkLocked 把一条**被转向注入**的语音请求的收尾回调追加到该路(gen
// 所指的那一条)上。必须在 o.mu 下调用;该路已失效时什么都不做。
func (o *Orchestrator) addVoiceSinkLocked(st *sessionState, gen uint64, sink VoiceReplyFunc) {
if st == nil || sink == nil || st.voiceTurn.gen != gen {
if sink == nil {
return
}
st.voiceTurn.sinks = append(st.voiceTurn.sinks, sink)
leg := o.legLocked(st, gen)
if leg == nil {
return
}
leg.voice.sinks = append(leg.voice.sinks, sink)
}
// takeVoiceTurn 取走与 gen 配对的所有语音收尾回调并清空记录。第二个返回值为
// false 表示代次已陈旧(中断/取代):迟到的结算不该回音频。
// takeVoiceTurn 取走**本路**的全部语音收尾回调并清空记录。第二个返回值为 false
// 表示这条路已失效(被中断清空 / 被取代):迟到的结算不该回音频。
func (o *Orchestrator) takeVoiceTurn(sessionID string, gen uint64) ([]VoiceReplyFunc, bool) {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[sessionID]
if st == nil || st.voiceTurn.gen != gen {
leg := o.legLocked(o.states[sessionID], gen)
if leg == nil {
return nil, false
}
sinks := st.voiceTurn.sinks
st.voiceTurn = voiceTurn{}
sinks := leg.voice.sinks
leg.voice = voiceTurn{}
return sinks, true
}
+21 -3
View File
@@ -83,6 +83,9 @@ func (h *hub) streamChat(ctx context.Context, payload any) SubmitResult {
// clientMsgId 被判重(design/16 §4.5、§5.1;协同 G4)。空 clientMsgId 是
// 空操作,因此老客户端的行为逐字不变。
ctx = withClientMsgHint(withMode(withDeviceID(ctx, req.DeviceID), req.Mode), clientMsgHint{ClientMsgID: req.ClientMsgID})
// 可选的并行声明走同一条 ctx 通道(design/15 §2.5):false 是空操作,
// 省略 `parallel` 的老客户端与加入该字段之前逐字相同。
ctx = withParallelTurn(ctx, req.Parallel)
env := h.envResolver.owner(ctx, req.SessionID)
if h.orchestrator != nil {
return h.orchestrator.Submit(ctx, req.SessionID, msgs, env)
@@ -98,11 +101,20 @@ func (h *hub) streamChat(ctx context.Context, payload any) SubmitResult {
// message.*/thinking.*/tool_call.* 事件扇出,因此只有非流式路径才发出
// 一整段文本 chunk 以维持流契约。
func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) {
turnID := newTurnID()
// 该路的 turnId 优先取 ctx 上已有的那个(design/15 §2.5.9 的 ⚠️ 必改项):
// 编排器在**启动该路之前**就生成了它,并把同一份 id 交给 runTurn —— 因为它
// 发出(后续切片的)`turn_begin` 帧时必须报出这个 id,而客户端按 turnId
// 两级分桶。若这里再自己生成一个,两处 id 不同,分桶会立即失配。
//
// 未挂载(子代理 / 后台思考 / 未接编排器的旧回退路径)时才新铸一个,因此
// 单路路径的帧上仍是一个**每回合全新的随机 id** ⇒ 对客户端无可见差异。
turnID := loop.TurnIDFrom(ctx)
if turnID == "" {
turnID = newTurnID()
}
// 把轮次身份串进运行上下文,使 loop 把它盖在它发出的每个流式
// 事件上(message.*/thinking.*/tool_call.*);hub 随后把它放到
// 下行帧上。每一轮都取全新的 id —— 一个会话最多一个在途轮次
// —— 因此客户端能把一串增量关联到同一个助手轮次
// 下行帧上,客户端因此能把一串增量关联到同一个助手轮次
// (design/16 §4.2)。
ctx = loop.WithTurnID(ctx, turnID)
if h.runner == nil || len(msgs) == 0 {
@@ -222,6 +234,12 @@ type chatStreamRequest struct {
// 上行帧仍有 4096 字节硬上限 ⇒ 文档实际上走 REST;这里保持同形解析,
// 使两条入口不会分叉(解析见 resolveDocuments)。
Documents []chatDocumentRef `json:"documents,omitempty"`
// Parallel 是「把本条作为**独立一路**处理」的显式声明(design/15 §2.5,
// 主人 2026-10-05 裁决 B):它可选,只有会话正忙且并发额度未满时才被照准
// (额度满 ⇒ 落回排队并回 `pending_ack{reason:"parallel_limit"}`)。
// 省略它时行为与加入该字段之前逐字相同 —— 服务端**不做**启发式自动升级,
// 触发权在客户端。
Parallel bool `json:"parallel,omitempty"`
}
// resolveDocuments 解析本回合携带的文档引用(design/24 §4.4/§6.1)。一条都不带