diff --git a/internal/runtime/loop/checkpoint.go b/internal/runtime/loop/checkpoint.go index 7c7278b..1e95b5b 100644 --- a/internal/runtime/loop/checkpoint.go +++ b/internal/runtime/loop/checkpoint.go @@ -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 } diff --git a/internal/runtime/loop/parallel_leg_test.go b/internal/runtime/loop/parallel_leg_test.go new file mode 100644 index 0000000..fef3dc7 --- /dev/null +++ b/internal/runtime/loop/parallel_leg_test.go @@ -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) + } +} diff --git a/internal/runtime/loop/session_ctx.go b/internal/runtime/loop/session_ctx.go index 83d72c8..7c93e65 100644 --- a/internal/runtime/loop/session_ctx.go +++ b/internal/runtime/loop/session_ctx.go @@ -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 +} diff --git a/internal/runtime/tool/legfile.go b/internal/runtime/tool/legfile.go new file mode 100644 index 0000000..04ceabd --- /dev/null +++ b/internal/runtime/tool/legfile.go @@ -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 +} diff --git a/internal/runtime/tool/legfile_test.go b/internal/runtime/tool/legfile_test.go new file mode 100644 index 0000000..f1bd28c --- /dev/null +++ b/internal/runtime/tool/legfile_test.go @@ -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) + } +} diff --git a/internal/services/gateway/chat_stream_dispatch.go b/internal/services/gateway/chat_stream_dispatch.go index 0d868ec..38481b9 100644 --- a/internal/services/gateway/chat_stream_dispatch.go +++ b/internal/services/gateway/chat_stream_dispatch.go @@ -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}) } diff --git a/internal/services/gateway/deps.go b/internal/services/gateway/deps.go index ca0dcf1..0c52f1e 100644 --- a/internal/services/gateway/deps.go +++ b/internal/services/gateway/deps.go @@ -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 } diff --git a/internal/services/gateway/rest.go b/internal/services/gateway/rest.go index 5e1e0d7..ceed3ba 100644 --- a/internal/services/gateway/rest.go +++ b/internal/services/gateway/rest.go @@ -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"` } diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index ab4a1d1..1733ca3 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -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, }) } diff --git a/internal/services/gateway/session_document_status.go b/internal/services/gateway/session_document_status.go index 6900f45..c97d97b 100644 --- a/internal/services/gateway/session_document_status.go +++ b/internal/services/gateway/session_document_status.go @@ -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 } diff --git a/internal/services/gateway/session_document_status_test.go b/internal/services/gateway/session_document_status_test.go index f905b7b..bfffe86 100644 --- a/internal/services/gateway/session_document_status_test.go +++ b/internal/services/gateway/session_document_status_test.go @@ -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` 数组回传引用与结论，且没有附件的消息**省略**该字段（省略 = // 没有附件，与引入文档之前逐字一致）。 diff --git a/internal/services/gateway/session_interrupt.go b/internal/services/gateway/session_interrupt.go index 277b0a0..1e1672e 100644 --- a/internal/services/gateway/session_interrupt.go +++ b/internal/services/gateway/session_interrupt.go @@ -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, + } +} diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index 812709c..112088a 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -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) - } -} diff --git a/internal/services/gateway/session_parallel.go b/internal/services/gateway/session_parallel.go new file mode 100644 index 0000000..e7d526b --- /dev/null +++ b/internal/services/gateway/session_parallel.go @@ -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 +} diff --git a/internal/services/gateway/session_parallel_protocol_test.go b/internal/services/gateway/session_parallel_protocol_test.go new file mode 100644 index 0000000..f9784ce --- /dev/null +++ b/internal/services/gateway/session_parallel_protocol_test.go @@ -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 不该被判为声明了并行") + } +} diff --git a/internal/services/gateway/session_parallel_test.go b/internal/services/gateway/session_parallel_test.go new file mode 100644 index 0000000..dbca4dd --- /dev/null +++ b/internal/services/gateway/session_parallel_test.go @@ -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 ¶llelFixture{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) + } + } +} diff --git a/internal/services/gateway/session_pending.go b/internal/services/gateway/session_pending.go new file mode 100644 index 0000000..186629a --- /dev/null +++ b/internal/services/gateway/session_pending.go @@ -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} +} diff --git a/internal/services/gateway/session_state.go b/internal/services/gateway/session_state.go new file mode 100644 index 0000000..ea39f29 --- /dev/null +++ b/internal/services/gateway/session_state.go @@ -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) + } +} diff --git a/internal/services/gateway/session_steer.go b/internal/services/gateway/session_steer.go index a706cab..b9d8342 100644 --- a/internal/services/gateway/session_steer.go +++ b/internal/services/gateway/session_steer.go @@ -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)) diff --git a/internal/services/gateway/session_turn_fence.go b/internal/services/gateway/session_turn_fence.go index 73796d4..5c803b1 100644 --- a/internal/services/gateway/session_turn_fence.go +++ b/internal/services/gateway/session_turn_fence.go @@ -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{} diff --git a/internal/services/gateway/session_turn_reply.go b/internal/services/gateway/session_turn_reply.go index 58c8064..68d0d6d 100644 --- a/internal/services/gateway/session_turn_reply.go +++ b/internal/services/gateway/session_turn_reply.go @@ -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 的收尾第一 diff --git a/internal/services/gateway/session_voice.go b/internal/services/gateway/session_voice.go index 79e6a37..54c36fd 100644 --- a/internal/services/gateway/session_voice.go +++ b/internal/services/gateway/session_voice.go @@ -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 } diff --git a/internal/services/gateway/ws_hub_turn.go b/internal/services/gateway/ws_hub_turn.go index f9ad279..f75b3f3 100644 --- a/internal/services/gateway/ws_hub_turn.go +++ b/internal/services/gateway/ws_hub_turn.go @@ -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）。一条都不带