diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index 47792ac..e035c6c 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -412,6 +412,13 @@ func (o *Orchestrator) completeTurn(sessionID string, leg *turnLeg, res *loop.Re 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, runLeg.turnID)) + // 续跑路同样可能有**别的在途路**（并行的另一路还在飞）：它也照答自己的那条消息， + // 同样需要「另有一路正在进行」的结构化摘要（design/15 §2.5.6，主人 2026-10-05 + // 裁决：同一类串话、同一个修法）。快照与上面同一个临界区取（st.legs / st.pending）， + // 没有其它在途路时不挂 ⇒ 单路续跑零成本、行为逐字不变。 + if notice, ok := parallelNoticeLocked(st, runLeg.gen); ok { + runCtx = loop.WithParallelNotice(runCtx, notice) + } // 续跑的这一轮同样有「回合前已落库的用户消息」，因此把它们的文档状态 // 记到**这一路**上（design/24 §6.1）；语音回调同理（协同 C1：忙时排队 // 不许把音频丢掉）。 diff --git a/internal/services/gateway/session_parallel.go b/internal/services/gateway/session_parallel.go index 7ca9a06..9725e73 100644 --- a/internal/services/gateway/session_parallel.go +++ b/internal/services/gateway/session_parallel.go @@ -13,18 +13,21 @@ import ( // 2026-10-05 裁决 B）。它从 session_orchestrator.go 拆出，使编排器本体继续只留 // 状态机，并保持在「单文件 ≤500 行 / 函数 ≤80 行」的硬约束内。 // -// 它做五件事： +// 它做六件事： // - 解析上行的并行声明（ctx 通道，与 deviceID / mode / 幂等键同一种机制）； // - 判定并发额度（默认 2，硬上限 3，<2 = 关闭真并行 ⇒ 完全回落到 steer-only 现状）； // - 在状态锁下登记一条新路（自己的世代号 / turnId / 取消链 / checkpoint 归属键）； // - 忙态帧的在途路数（activeTurns）与排队原因（reason）； // - 下发新帧 `turn_begin`（`publishLegStart` / `emitTurnBegin`）：单路 // kind:"turn"、并行路 kind:"parallel"，两条路径都发。`steer_ack` 的帧壳 -// turnId 由 session_steer.go 的转向游标带出，与这里**取值同源**（leg.turnID）。 +// turnId 由 session_steer.go 的转向游标带出，与这里**取值同源**（leg.turnID）； +// - 把「本会话其它在途路」的**进行态摘要**（design/15 §2.5.6，见 +// parallelNoticeLocked）挂到新路的 runCtx 上，使循环每一步都能把一条独立的 +// system 提示注入本轮视图 —— 这正是「一题多答 / 跨端串话」的治法。摘要只含 +// 元数据，结构里**没有**任何能承载对方正文的字段（红线）。 // -// 它**不做**（留给后续切片）：第二路的进行态 system 摘要。写类工具的文件级互斥 -// 在**两个地方**落地：互斥**身份**由 beginLegLocked 对每一条在途路盖 -// （loop.WithToolLegID，取值 = 本路 turnID），登记 / 判忙 / 释放由工具侧 +// 写类工具的文件级互斥在**两个地方**落地：互斥**身份**由 beginLegLocked 对每一条 +// 在途路盖（loop.WithToolLegID，取值 = 本路 turnID），登记 / 判忙 / 释放由工具侧 // （internal/runtime/tool/fileclaims.go）在执行前后来做。 // 并发额度的取值口径（design/15 §2.5.7）。 @@ -144,6 +147,52 @@ func (o *Orchestrator) canStartParallelLocked(st *sessionState) bool { return len(st.legs) >= 1 && len(st.legs) < o.maxParallelTurns } +// parallelNoticeLocked 把该会话**其它在途路**的进行态收成一份结构化的 +// loop.ParallelNotice（design/15 §2.5.6），供新起的这一路挂到自己的 runCtx 上。 +// 第二个返回值报告「是否有其它在途路」：false 时调用方**不挂**提示，使单路 / +// 收尾只剩一路 / 未接线的路径零成本、行为逐字不变。 +// +// 摘要的每一项都取**结构化数据**，绝无例外： +// - OtherMessageID = 其它某一路已落库用户消息（leg.docs.msgs）**末条**的权威 id； +// 该路还没落库（落库在锁外、比登记稍晚）时留空，对应段不渲染，绝不凭空编 id； +// - Queued = st.pending 的条数 —— 仍在排队、尚未被任何一路接手的消息。 +// +// 它**没有**「对方正文」这个概念：ParallelNotice 结构里根本没有能承载正文的字段 +// （结构性红线，见 loop/steer.go）。半句话会诱导本路去补全或抢答，也无法诚实地 +// 标注 who said what（§2.5.6「不做的事」①）。 +// +// 多路同时在场（上限 3）时取**世代号最大**的那一条其它路，即最近起跑的那一路： +// 契约里只有单个 OtherMessageID，按 map 遍历序取值会让提示内容随调度抖动，因此 +// 这里刻意取一个确定的口径。 +// +// **Step / ToolNames 恒为 0 / nil**：当前**没有数据源** —— turnLeg 不带步数、也不带 +// 已调用工具名，loop 也不回传任何中途进度。这两个字段对应的事实行因此不渲染；进度源 +// 待另开切片，**不在这里猜或伪造数据**。 +// +// 必须在 o.mu 下调用（它读 st.legs / st.pending，与 activeTurns 同一临界区）。 +func parallelNoticeLocked(st *sessionState, selfGen uint64) (loop.ParallelNotice, bool) { + if st == nil { + return loop.ParallelNotice{}, false + } + var other *turnLeg + for gen, leg := range st.legs { + if gen == selfGen || leg == nil { + continue + } + if other == nil || leg.gen > other.gen { + other = leg + } + } + if other == nil { + return loop.ParallelNotice{}, false + } + notice := loop.ParallelNotice{Queued: len(st.pending)} + if msgs := other.docs.msgs; len(msgs) > 0 { + notice.OtherMessageID = msgs[len(msgs)-1].ID + } + return notice, true +} + // startParallelLocked 在状态锁下把一条忙时到达、且显式声明并行的用户消息升为 // **独立的一路**（design/15 §2.5.1 / §2.5.3），并在锁外起 goroutine、返回本次提交 // 的判定。它必须在 o.mu 下调用，且由调用方保证 canStartParallelLocked 已经成立 @@ -168,6 +217,12 @@ func (o *Orchestrator) startParallelLocked(ctx context.Context, st *sessionState // 这与续跑路用 Background 派生属于同一条口径：后台路的生死由编排器说了算。 leg, runCtx, cancel := o.beginLegLocked(context.WithoutCancel(ctx), st, newTurnID(), true, newTurnFence(nil)) activeTurns := len(st.legs) + // 进行态摘要（design/15 §2.5.6）在**解锁之前**取快照：它读 st.legs / st.pending， + // 与 activeTurns 同一条纪律 —— 解锁后再读数会随另一路的收尾漂移。没有其它在途路 + // 时 ok=false ⇒ 不挂，runCtx 与加入这条通道之前逐字相同（零成本）。 + if notice, ok := parallelNoticeLocked(st, leg.gen); ok { + runCtx = loop.WithParallelNotice(runCtx, notice) + } o.mu.Unlock() runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, leg.gen, leg.turnID)) diff --git a/internal/services/gateway/session_parallel_notice_test.go b/internal/services/gateway/session_parallel_notice_test.go new file mode 100644 index 0000000..6588368 --- /dev/null +++ b/internal/services/gateway/session_parallel_notice_test.go @@ -0,0 +1,204 @@ +package gateway + +import ( + "context" + "testing" + + "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.6「并行路进行态摘要」**接线**的回归套件（它单独成文件， +// 使 session_parallel_test.go 不因这一组用例越过「单文件 ≤500 行」的硬约束）。 +// +// 它钉住三件事： +// - 第二路起跑时它的 runCtx 上带着摘要，且 OtherMessageID 指向**另一路**的消息 id； +// - 摘要里的 Queued 取自该会话仍在 pending 的消息条数； +// - 没有其它在途路时（单路 / 串行续跑）**不挂**该提示 —— 未接线的路径零成本。 +// +// 断言一律**从 runCtx 上读**（loop.ParallelNoticeFrom），因此接线一旦漏掉或对方 +// 消息 id 一旦不填，这些用例立刻变红。 + +// --------------------------------------------------------------------------- +// 并行进行态摘要（design/15 §2.5.6）：起第二路时挂给**它自己** +// --------------------------------------------------------------------------- + +// firstUserMessageID 返回落库记录里第一条用户消息的权威 id（夹具由 fakeHistory +// 分配）。断言「提示指向另一路的那条消息」时用它，而不是硬编码一个 id。 +func firstUserMessageID(t *testing.T, f *parallelFixture) string { + t.Helper() + for _, m := range f.history.records() { + if m.Role == history.RoleUser && m.ID != "" { + return m.ID + } + } + t.Fatal("夹具没有给第一路的用户消息落库 id，这条断言无从谈起") + return "" +} + +// userMessageIDByText 返回落库记录里正文为 text 的那条用户消息的权威 id。 +func userMessageIDByText(t *testing.T, f *parallelFixture, text string) string { + t.Helper() + for _, m := range f.history.records() { + if m.Role == history.RoleUser && m.Content == text && m.ID != "" { + return m.ID + } + } + t.Fatalf("落库记录里找不到正文为 %q 的用户消息", text) + return "" +} + +// TestParallelNoticeAttachedToSecondLegOnly 钉住 §2.5.6 的**接线**：第二路起跑时它 +// 的 runCtx 上带着「另有一路在跑」的结构化摘要，且 OtherMessageID 指向**另一路** +// 的那条消息；而第一路（起跑时本会话没有别的在途路）**不挂**该提示 —— 单路零成本。 +// +// 它刻意**从 runCtx 上读**（而不是读编排器内部字段）：接线漏掉时这个断言立刻变红， +// 因为提示此刻根本不在 ctx 上。 +func TestParallelNoticeAttachedToSecondLegOnly(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) + } + firstCalls := f.run.waitStarted(1) + if notice, ok := loop.ParallelNoticeFrom(firstCalls[0].ctx); ok { + t.Fatalf("第一路的 runCtx 不该挂并行提示（起跑时没有其它在途路），得到 %+v", notice) + } + + second := f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("第二问")}, loop.OwnerEnv()) + if !second.Immediate || !second.Parallel { + t.Fatalf("第二路 = %+v, want Immediate 且 Parallel", second) + } + calls := f.run.waitStarted(2) + + notice, ok := loop.ParallelNoticeFrom(calls[1].ctx) + if !ok { + t.Fatal("第二路的 runCtx 上必须挂并行提示（§2.5.6 的接线点）") + } + wantID := firstUserMessageID(t, f) + if notice.OtherMessageID != wantID { + t.Fatalf("提示里的对方消息 id = %q, want %q（另一路已落库的那条）", notice.OtherMessageID, wantID) + } + // 摘要只含元数据：末条消息 id 必须是**另一路**的，绝不能等于本路自己的 + // （本路此刻还没落库，也绝不该拿自己的消息冒充对方）。 + if notice.OtherMessageID == "" { + t.Fatal("提示里的对方消息 id 不该为空") + } + // Step / ToolNames 当前没有数据源 ⇒ 恒为 0 / nil，对应事实行不渲染。 + if notice.Step != 0 || len(notice.ToolNames) != 0 { + t.Fatalf("Step/ToolNames 没有数据源，必须留空，得到 %d / %v", notice.Step, notice.ToolNames) + } + + f.run.release(0) + f.run.release(1) + if !waitIdle(f.o, "web") { + t.Fatal("会话没有回到空闲") + } +} + +// TestParallelNoticeReportsQueueDepth 钉住摘要里的排队条数取自 st.pending：那条 +// 已排队、尚未被任何一路接手的消息计入 Queued，而不是被漏掉或凭空推断。 +func TestParallelNoticeReportsQueueDepth(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(context.Background(), "web", []model.Message{userMsg("乙")}, loop.OwnerEnv()); !sr.Pending { + t.Fatalf("忙时未声明并行的提交 = %+v, want Pending", sr) + } + + if sr := f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("丙")}, loop.OwnerEnv()); !sr.Parallel { + t.Fatalf("第三问 = %+v, want 照准并行", sr) + } + calls := f.run.waitStarted(2) + notice, ok := loop.ParallelNoticeFrom(calls[1].ctx) + if !ok { + t.Fatal("第二路的 runCtx 上必须挂并行提示") + } + if notice.Queued != 1 { + t.Fatalf("提示里的排队条数 = %d, want 1（那条尚未被接手的「乙」）", notice.Queued) + } + + // 两路逐条收尾；排队的那条由结算中的一路续跑（第 3 条路），必须一并放行。 + f.run.release(0) + f.run.release(1) + f.run.waitStarted(3) + f.run.release(2) + if !waitIdle(f.o, "web") { + t.Fatal("会话没有回到空闲") + } +} + +// TestParallelNoticeAbsentWithoutOtherLegs 钉住「没有其它在途路就不挂提示」： +// 单路提交、以及它的**串行续跑路**（起跑时在途路已清空）都拿不到提示 —— 未接线的 +// 路径零成本，行为与加入这条 ctx 通道之前逐字不变。 +func TestParallelNoticeAbsentWithoutOtherLegs(t *testing.T) { + t.Parallel() + f := newParallelFixture(5, 2) + + f.o.Submit(context.Background(), "web", []model.Message{userMsg("甲")}, loop.OwnerEnv()) + f.run.waitStarted(1) + f.o.Submit(context.Background(), "web", []model.Message{userMsg("乙")}, loop.OwnerEnv()) + + // 放行第一路 ⇒ 弹队首续跑；此刻在途路已清空，续跑路不得挂任何提示。 + f.run.release(0) + calls := f.run.waitStarted(2) + for i, c := range calls { + if notice, ok := loop.ParallelNoticeFrom(c.ctx); ok { + t.Fatalf("第 %d 路的 runCtx 不该挂并行提示（本会话没有其它在途路），得到 %+v", i+1, notice) + } + } + + f.run.release(1) + if !waitIdle(f.o, "web") { + t.Fatal("会话没有回到空闲") + } +} + +// TestParallelNoticeAttachedToResumeLeg 钉住**续跑路**的接线（主人 2026-10-05 裁决： +// 同一类串话、同一个修法）：一路收尾、队列有货、而**另一并行路仍在飞**时弹出来的 +// 那条续跑路，同样拿到进行态摘要，且 OtherMessageID 指向**当时仍在飞的那一路**。 +// +// 「没有其它在途路的续跑路不挂提示」由 TestParallelNoticeAbsentWithoutOtherLegs 钉住， +// 两条合起来才是完整的接线口径（有则挂、无则零成本）。 +func TestParallelNoticeAttachedToResumeLeg(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, want 照准并行", sr) + } + live := f.run.waitStarted(2) + if notice, ok := loop.ParallelNoticeFrom(live[0].ctx); ok { + t.Fatalf("第一路不该挂并行提示（它起跑时本会话没有其它在途路），得到 %+v", notice) + } + // 忙时未声明并行的第三条 ⇒ 入队，等某一路收尾时被续跑。 + if sr := f.o.Submit(context.Background(), "web", []model.Message{userMsg("丙")}, loop.OwnerEnv()); !sr.Pending { + t.Fatalf("第三条 = %+v, want Pending", sr) + } + + // 放行**第一**路 ⇒ 它收尾后弹队首起续跑路；此刻第二路仍在飞。 + f.run.release(0) + calls := f.run.waitStarted(3) + notice, ok := loop.ParallelNoticeFrom(calls[2].ctx) + if !ok { + t.Fatal("续跑路的 runCtx 上必须挂并行提示（当时仍有另一路在飞）") + } + wantID := userMessageIDByText(t, f, "乙") + if notice.OtherMessageID != wantID { + t.Fatalf("续跑路提示里的对方消息 id = %q, want %q（当时仍在飞的那一路）", notice.OtherMessageID, wantID) + } + + // 两路逐条收尾（第二路释放在前、续跑路在后），会话回到空闲。 + f.run.release(1) + f.run.release(2) + if !waitIdle(f.o, "web") { + t.Fatal("会话没有回到空闲") + } +}