diff --git a/internal/services/gateway/interrupt_scope_test.go b/internal/services/gateway/interrupt_scope_test.go new file mode 100644 index 0000000..79ed4f9 --- /dev/null +++ b/internal/services/gateway/interrupt_scope_test.go @@ -0,0 +1,341 @@ +package gateway + +import ( + "context" + "encoding/json" + "errors" + "net" + "slices" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" + + "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" +) + +// 本文件锁定协同项 A2 的 `interrupt {sessionId, scope}` 口径（上游清单： +// `Cyrene-For-Android/devdocs/02-前后端协同事项.md`）。客户端有「追加队列」与 +// 「打断」两种模式，而后端过去只有一种口径（取消在途回合 + 清空待处理队列）， +// 于是用户选「打断」时连发的多条消息会被静默丢弃，只回一个 dropped 数字。 +// +// 落地口径（省略 = 维持现状）： +// - scope 省略 / "all" ⇒ 现状语义：取消在途回合 + 清空待处理队列； +// - scope "current" ⇒ 只取消在途回合，**保留**待处理队列（它们按编排器 +// 的续跑规则继续执行），dropped 恒为 0； +// - 其余取值 ⇒ 只回发送者 {type:"error", data:{code:"bad_request"}}， +// 不广播、不打断任何会话（显式报错，绝不静默当默认执行）。 +// +// 连接、令牌、编排器与 WS 路由都是真的（httptest + 真 JWT + 真 orchestrator）， +// 只有模型适配器是脚本化的，因此「消息到底有没有被驱动」是被观测到的事实， +// 而不是从下行帧反推的推测。 +const ( + // a2ReadWait 界定一次预期会产出帧的读取。 + a2ReadWait = 3 * time.Second + // a2Settle 让一个「本不该发生」的副作用有足够时间暴露自己，再断言它不存在。 + a2Settle = 200 * time.Millisecond + // a2MainSession 是本组用例使用的会话 id。 + a2MainSession = "web" +) + +// a2Runner 是 WS 用例使用的假 Chatter：它记录每次 Run 的末条用户文本后阻塞， +// 直到 release 被关闭或自己的 ctx 被取消（打断路径）。会话因此稳定处于 busy， +// 测试能确定地断言忙时连发的消息最终有没有被驱动。 +type a2Runner struct { + mu sync.Mutex + calls []string + release chan struct{} +} + +func newA2Runner() *a2Runner { return &a2Runner{release: make(chan struct{})} } + +func (r *a2Runner) Run(ctx context.Context, _ string, msgs []model.Message) (*loop.Result, error) { + text := "" + if n := len(msgs); n > 0 { + text = messageText(msgs[n-1]) + } + r.mu.Lock() + r.calls = append(r.calls, text) + r.mu.Unlock() + select { + case <-r.release: + case <-ctx.Done(): + return nil, ctx.Err() + } + return &loop.Result{Text: "reply-" + text, Steps: 1, StopReason: "stop"}, nil +} + +// count 返回已被驱动的回合数。 +func (r *a2Runner) count() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.calls) +} + +// texts 返回已驱动回合的末条用户文本快照。 +func (r *a2Runner) texts() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.calls...) +} + +// a2Runner 与 a2Event/a2Sink 等夹具、以及单元级用例见 +// interrupt_scope_unit_test.go（拆开以守住单文件 ≤500 行）。 + +// interruptFrameWire 是解码后用于断言的 session_interrupted 帧（§3.2 形状： +// {sessionId, mode, dropped}）。 +type interruptFrameWire struct { + Type string `json:"type"` + SessionID string `json:"sessionId"` + Data struct { + SessionID string `json:"sessionId"` + Mode string `json:"mode"` + Dropped int `json:"dropped"` + } `json:"data"` +} + +// a2Env 为一个 A2 用例接线网关：脚本化的 runner + 真编排器 + 真 JWT + 真 WS。 +type a2Env struct { + srv *Server + url string + jwt *auth.JWTService + run *a2Runner + o *Orchestrator +} + +func newA2Env(t *testing.T, run *a2Runner) *a2Env { + t.Helper() + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: &fakeHistory{}}) + return &a2Env{srv: srv, url: ts.URL, jwt: jwt, run: run, o: srv.hub.orchestrator} +} + +// owner 拨一条 owner 连接的 WS。 +func (e *a2Env) owner(t *testing.T) *websocket.Conn { + t.Helper() + return dialWS(t, e.url, signAny(t, e.jwt, "acct", auth.RoleOwner)) +} + +// chatAndQueue 发出一条在飞消息与一条待处理消息，并等到两条都真的落地 +// （在飞 + 队列深度 1），使后续断言面对一个确定的忙态而不是靠 sleep 赌时序。 +func (e *a2Env) chatAndQueue(t *testing.T, conn *websocket.Conn, inFlight, queued string) { + t.Helper() + if err := conn.WriteJSON(wsInbound{Type: "chat", Data: a2ChatData(inFlight)}); err != nil { + t.Fatalf("write chat %s: %v", inFlight, err) + } + waitUntil(t, inFlight+" 回合在飞", func() bool { return e.run.count() == 1 }) + if err := conn.WriteJSON(wsInbound{Type: "chat", Data: a2ChatData(queued)}); err != nil { + t.Fatalf("write chat %s: %v", queued, err) + } + waitUntil(t, queued+" 进入 pending", func() bool { return e.o.QueueDepth(a2MainSession) == 1 }) +} + +// a2ChatData 构造 chat 上行帧的 data。 +func a2ChatData(text string) json.RawMessage { + b, _ := json.Marshal(map[string]string{"sessionId": a2MainSession, "text": text}) + return b +} + +// readFrameOfType 读帧直到出现 want 类型（握手回放、持久化增量等一并跳过）， +// 返回那条帧的原始字节；超期未出现即失败。 +func readFrameOfType(t *testing.T, conn *websocket.Conn, want, what string) []byte { + t.Helper() + deadline := time.Now().Add(a2ReadWait) + for time.Now().Before(deadline) { + _ = conn.SetReadDeadline(deadline) + _, msg, err := conn.ReadMessage() + if err != nil { + t.Fatalf("%s: 等待 %q 帧时读取失败: %v", what, want, err) + } + var shell struct { + Type string `json:"type"` + } + if err := json.Unmarshal(msg, &shell); err != nil { + t.Fatalf("%s: 解帧 %s: %v", what, msg, err) + } + if shell.Type == want { + return msg + } + } + t.Fatalf("%s: 截止时间前没有收到 %q 帧", what, want) + return nil +} + +// a2DrainTypes 读取帧直到 window 内再无帧到达，返回窗口内收到的帧类型。 +// 它给「本不该出现的帧」一个真实的时间窗口，并按类型断言（例如「不再有 +// session_interrupted」），比单个 assertNoFrame 更能表达意图。它是该连接的 +// 最后一次读取 —— 读截止时间到期后 gorilla 会留下一个已损坏的连接。 +func a2DrainTypes(t *testing.T, conn *websocket.Conn, window time.Duration) []string { + t.Helper() + var types []string + // 每次读取单独续期，但总窗口有上限，使迟到的帧仍能被看见而测试不会挂住。 + stop := time.Now().Add(4 * window) + for { + readBy := time.Now().Add(window) + if readBy.After(stop) { + readBy = stop + } + _ = conn.SetReadDeadline(readBy) + _, msg, err := conn.ReadMessage() + if err != nil { + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return types + } + t.Fatalf("读取帧失败: %v", err) + } + var shell struct { + Type string `json:"type"` + } + if err := json.Unmarshal(msg, &shell); err != nil { + t.Fatalf("解帧 %s: %v", msg, err) + } + types = append(types, shell.Type) + } +} + +// TestInterruptScopeOmittedAndAllDropPending 锁定 A2 的「维持现状」侧：scope +// 省略（老客户端只发 {sessionId}）与显式 "all" 必须逐字同解 —— 取消在途回合 + +// 清空待处理队列，session_interrupted 的 dropped 等于真的被丢弃的条数，且被 +// 丢弃的那条绝不被驱动。 +func TestInterruptScopeOmittedAndAllDropPending(t *testing.T) { + t.Parallel() + cases := []struct { + name string + data string + }{ + {"scope 省略", `{"sessionId":"web"}`}, + {"scope=all", `{"sessionId":"web","scope":"all"}`}, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + run := newA2Runner() + env := newA2Env(t, run) + conn := env.owner(t) + defer conn.Close() + waitConnCount(t, env.srv, 1) + env.chatAndQueue(t, conn, "A", "B") + + if err := conn.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(tc.data)}); err != nil { + t.Fatalf("write interrupt: %v", err) + } + raw := readFrameOfType(t, conn, "session_interrupted", tc.name) + var frame interruptFrameWire + if err := json.Unmarshal(raw, &frame); err != nil { + t.Fatalf("%s: 解 session_interrupted 帧 %s: %v", tc.name, raw, err) + } + if frame.SessionID != a2MainSession || frame.Data.SessionID != a2MainSession { + t.Fatalf("%s: 帧的会话作用域 = %q/%q, want %q", tc.name, frame.SessionID, frame.Data.SessionID, a2MainSession) + } + if frame.Data.Dropped != 1 { + t.Fatalf("%s: dropped = %d, want 1（队列里那一条真的被丢弃）frame=%s", tc.name, frame.Data.Dropped, raw) + } + if frame.Data.Mode != string(ModeIdle) { + t.Fatalf("%s: mode = %q, want %q（队列清空后会话真的空闲）", tc.name, frame.Data.Mode, ModeIdle) + } + waitUntil(t, "队列清空且会话空闲", func() bool { + return !env.o.Busy(a2MainSession) && env.o.QueueDepth(a2MainSession) == 0 + }) + + // 打断后 A 的收尾会发出一条 stream_end（被取消的回合如实收口）， + // 因此这里按类型断言窗口内容，而不是断言「再无任何帧」：既不允许 + // 第二条 session_interrupted，也不允许被丢弃的 B 被驱动。 + close(run.release) + types := a2DrainTypes(t, conn, a2Settle) + if slices.Contains(types, "session_interrupted") { + t.Fatalf("%s: 一次 interrupt 只应上报一次 session_interrupted，实际帧序列 %v", tc.name, types) + } + if got := run.texts(); len(got) != 1 || got[0] != "A" { + t.Fatalf("%s: driven = %v, want [A]（被丢弃的 B 绝不能执行）", tc.name, got) + } + }) + } +} + +// TestInterruptScopeCurrentOverWSResumesPending 是 A2 在真实 WS 路由上的端到端 +// 落地：scope:"current" 的下行帧必须如实报告 dropped=0 且会话未空闲，而下一条 +// 已被排队的消息必须真的被驱动（这就是「不再静默丢弃」的证据）。 +func TestInterruptScopeCurrentOverWSResumesPending(t *testing.T) { + t.Parallel() + run := newA2Runner() + env := newA2Env(t, run) + conn := env.owner(t) + defer conn.Close() + waitConnCount(t, env.srv, 1) + env.chatAndQueue(t, conn, "A", "B") + + if err := conn.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(`{"sessionId":"web","scope":"current"}`)}); err != nil { + t.Fatalf("write interrupt: %v", err) + } + raw := readFrameOfType(t, conn, "session_interrupted", "scope=current") + var frame interruptFrameWire + if err := json.Unmarshal(raw, &frame); err != nil { + t.Fatalf("解 session_interrupted 帧 %s: %v", raw, err) + } + if frame.Data.Dropped != 0 { + t.Fatalf("dropped = %d, want 0（current 口径不丢弃任何待处理消息）frame=%s", frame.Data.Dropped, raw) + } + if frame.Data.Mode == string(ModeIdle) { + t.Fatalf("mode = %q, want 非 idle（队列仍在续跑，会话没有被报成空闲）frame=%s", frame.Data.Mode, raw) + } + + waitUntil(t, "被保留的 B 续跑", func() bool { return run.count() == 2 }) + if got := run.texts(); len(got) != 2 || got[0] != "A" || got[1] != "B" { + t.Fatalf("driven = %v, want [A B]（pending 必须续跑）", got) + } + close(run.release) + waitUntil(t, "会话安定", func() bool { + return !env.o.Busy(a2MainSession) && env.o.QueueDepth(a2MainSession) == 0 + }) +} + +// TestInterruptScopeInvalidIsDirectedErrorOnly 锁定 A2 的报错侧：非法 scope +// 只回发送者一条 {code:"bad_request"}，不广播（第二条连接收不到）、不打断任何 +// 会话（在飞回合继续跑、待处理队列原样保留）。 +func TestInterruptScopeInvalidIsDirectedErrorOnly(t *testing.T) { + t.Parallel() + run := newA2Runner() + env := newA2Env(t, run) + sender := env.owner(t) + defer sender.Close() + observer := env.owner(t) + defer observer.Close() + waitConnCount(t, env.srv, 2) + env.chatAndQueue(t, sender, "A", "B") + if err := sender.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(`{"sessionId":"web","scope":"pending"}`)}); err != nil { + t.Fatalf("write interrupt: %v", err) + } + raw := readFrameOfType(t, sender, "error", "非法 scope") + var ef errorFrameWire + if err := json.Unmarshal(raw, &ef); err != nil { + t.Fatalf("解错误帧 %s: %v", raw, err) + } + if ef.Data.Code != wsBadRequestCode { + t.Fatalf("error code = %q, want %q（非法 scope 必须显式报错，不得静默当默认）frame=%s", ef.Data.Code, wsBadRequestCode, raw) + } + + // 只到发送者：观察者（它先收到了 chat 带来的状态帧与握手回放，这些被跳过） + // 在窗口里既收不到 error、也收不到 session_interrupted —— 拒绝不是广播。 + seen := a2DrainTypes(t, observer, a2Settle) + if slices.Contains(seen, "error") || slices.Contains(seen, "session_interrupted") { + t.Fatalf("非法 scope 被广播了：观察者收到 %v（必须只定向回发送者）", seen) + } + // 发送者自己也只多了那一条错误帧：没有 session_interrupted。 + sendSeen := a2DrainTypes(t, sender, a2Settle) + if slices.Contains(sendSeen, "session_interrupted") { + t.Fatalf("非法 scope 仍然打断了会话：发送者收到 %v", sendSeen) + } + + // 会话完全没有被触动：回合 A 仍在飞（a2Runner 没被取消），队列仍是 1。 + if got := run.texts(); len(got) != 1 || got[0] != "A" { + t.Fatalf("driven = %v, want [A]（非法 scope 不得打断或续跑任何东西）", got) + } + if env.o.QueueDepth(a2MainSession) != 1 || !env.o.Busy(a2MainSession) { + t.Fatalf("会话被非法 scope 触动了：busy=%v depth=%d", env.o.Busy(a2MainSession), env.o.QueueDepth(a2MainSession)) + } + close(run.release) +} diff --git a/internal/services/gateway/interrupt_scope_unit_test.go b/internal/services/gateway/interrupt_scope_unit_test.go new file mode 100644 index 0000000..8d4c55b --- /dev/null +++ b/internal/services/gateway/interrupt_scope_unit_test.go @@ -0,0 +1,188 @@ +package gateway + +import ( + "context" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// 本文件是协同项 A2 的单元级断言（编排器 + scope 取值域），与 +// interrupt_scope_test.go 的 WS 端到端用例配套。两处合起来钉住： +// `interrupt {sessionId, scope}` 的省略 / "all" / "current" / 非法值四种口径。 +// 拆成两个文件是为了守住「单文件 ≤500 行」的硬约束。 + +// a2HoldRunner 与 a2Runner 同形，但 ctx 被取消后**仍不返回**（真实 loop 会 +// 立刻返回，这里刻意留在 Run 里），使「pending 在打断那一刻真的还在」有一个 +// 不被续跑打断的确定性窗口。release 关闭后它才按其 ctx 状态返回。 +type a2HoldRunner struct { + mu sync.Mutex + calls []string + release chan struct{} +} + +func newA2HoldRunner() *a2HoldRunner { return &a2HoldRunner{release: make(chan struct{})} } + +func (r *a2HoldRunner) Run(ctx context.Context, _ string, msgs []model.Message, _ loop.SessionEnv) (*loop.Result, error) { + text := "" + if n := len(msgs); n > 0 { + text = messageText(msgs[n-1]) + } + r.mu.Lock() + r.calls = append(r.calls, text) + r.mu.Unlock() + <-r.release + if err := ctx.Err(); err != nil { + return nil, err + } + return &loop.Result{Text: "reply-" + text, Steps: 1, StopReason: "stop"}, nil +} + +func (r *a2HoldRunner) count() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.calls) +} + +func (r *a2HoldRunner) texts() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.calls...) +} + +// a2Event 是一次被捕获的状态转移。 +type a2Event struct { + typ string + data map[string]any +} + +// a2Sink 记录编排器发出的每一次状态转移（含 data），使「帧里报的 dropped / +// mode」能在单元级被断言，而不必先架一条 WS。它与 recordingSink 分开是为了 +// 让本组用例不依赖别的测试文件的数据结构。 +type a2Sink struct { + mu sync.Mutex + events []a2Event +} + +func (s *a2Sink) Emit(_, typ string, data map[string]any) { + s.mu.Lock() + defer s.mu.Unlock() + s.events = append(s.events, a2Event{typ: typ, data: data}) +} + +// lastInterrupt 返回最后一次 session_interrupted 的 mode 与 dropped。mode 在 +// 编排器里是 Mode 类型（不是 string），因此两种形状都接受。 +func (s *a2Sink) lastInterrupt() (string, int, bool) { + s.mu.Lock() + defer s.mu.Unlock() + for i := len(s.events) - 1; i >= 0; i-- { + if s.events[i].typ != EventSessionInterrupt { + continue + } + var mode string + switch v := s.events[i].data["mode"].(type) { + case Mode: + mode = string(v) + case string: + mode = v + } + dropped, _ := s.events[i].data["dropped"].(int) + return mode, dropped, true + } + return "", 0, false +} + +// TestInterruptScopeCurrentKeepsPending 是 A2 的单元级核心断言：scope=current +// 只取消当前回合，待处理队列一条不丢（确定性地断言队列仍在，因为 a2HoldRunner +// 被取消后仍留在 Run 里），状态帧如实报告 dropped=0 且 mode 不是 idle；随后被 +// 保留的消息真的按续跑规则执行。 +func TestInterruptScopeCurrentKeepsPending(t *testing.T) { + t.Parallel() + run := newA2HoldRunner() + sink := &a2Sink{} + o := NewOrchestrator(OrchestratorOptions{Run: run.Run, Emit: sink.Emit, MaxPending: 5, MaxTail: 20}) + + if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv()); !sr.Immediate { + t.Fatalf("A = %+v, want Immediate", sr) + } + waitUntil(t, "回合 A 在飞", func() bool { return run.count() == 1 }) + if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("B")}, loop.OwnerEnv()); !sr.Pending { + t.Fatalf("B = %+v, want Pending", sr) + } + if o.QueueDepth("s1") != 1 { + t.Fatalf("queue depth = %d, want 1（B 必须已经入队）", o.QueueDepth("s1")) + } + + if dropped := o.InterruptScope("s1", InterruptScopeCurrent); dropped != 0 { + t.Fatalf("dropped = %d, want 0（current 口径不丢弃任何待处理消息）", dropped) + } + // 确定性窗口：A 尚未结算，因此队列与忙态就是打断那一刻的真实状态。 + if got := o.QueueDepth("s1"); got != 1 { + t.Fatalf("queue depth = %d, want 1（scope=current 必须保留待处理队列）", got) + } + if !o.Busy("s1") { + t.Fatal("scope=current 只取消当前回合，会话不该被报成空闲（队列还在续跑）") + } + if got := run.texts(); len(got) != 1 || got[0] != "A" { + t.Fatalf("driven = %v, want [A]（此时不得有任何续跑回合）", got) + } + mode, dropped, ok := sink.lastInterrupt() + if !ok { + t.Fatal("没有发出 session_interrupted（打断必须如实上报发生了什么）") + } + if dropped != 0 { + t.Fatalf("session_interrupted.dropped = %d, want 0（真的没有消息被丢弃）", dropped) + } + if mode != string(ModeThinking) { + t.Fatalf("session_interrupted.mode = %q, want %q（会话仍在收尾并续跑队列）", mode, ModeThinking) + } + + // 放行 A 的收尾：被保留的 B 必须带着会话尾部上下文真的续跑。 + close(run.release) + waitUntil(t, "被保留的 B 续跑", func() bool { return run.count() == 2 }) + if got := run.texts(); len(got) != 2 || got[0] != "A" || got[1] != "B" { + t.Fatalf("driven = %v, want [A B]（pending 必须续跑，绝不能静默丢弃）", got) + } + if !waitIdle(o, "s1") { + t.Fatalf("会话没有安定：busy=%v depth=%d", o.Busy("s1"), o.QueueDepth("s1")) + } +} + +// TestParseInterruptScopeValueDomain 钉住 scope 的取值域：省略 / 空串 / "all" +// 落到现状语义，"current" 落到保留队列口径，其余一律报错 —— 未知取值被悄悄当成 +// 默认执行，正是 A2 要修掉的那类静默丢消息。 +func TestParseInterruptScopeValueDomain(t *testing.T) { + t.Parallel() + cases := []struct { + raw string + want InterruptScope + wantErr bool + }{ + {"", InterruptScopeAll, false}, + {"all", InterruptScopeAll, false}, + {"current", InterruptScopeCurrent, false}, + {" current ", InterruptScopeCurrent, false}, + {"Current", "", true}, + {"ALL", "", true}, + {"pending", "", true}, + {"both", "", true}, + {"*", "", true}, + } + for _, tc := range cases { + got, err := parseInterruptScope(tc.raw) + if tc.wantErr { + if err == nil { + t.Fatalf("scope=%q 被接受为 %q：未知取值必须报错，不得静默按默认执行", tc.raw, got) + } + continue + } + if err != nil { + t.Fatalf("scope=%q: %v", tc.raw, err) + } + if got != tc.want { + t.Fatalf("scope=%q = %q, want %q", tc.raw, got, tc.want) + } + } +} diff --git a/internal/services/gateway/session_interrupt.go b/internal/services/gateway/session_interrupt.go new file mode 100644 index 0000000..277b0a0 --- /dev/null +++ b/internal/services/gateway/session_interrupt.go @@ -0,0 +1,94 @@ +package gateway + +// 本文件归属协同项 A2 的打断口径：`interrupt {sessionId, scope}` 决定一次 +// 用户打断清到哪一层。它从 session_orchestrator.go 拆出，使编排器本体只留 +// 消息状态机（Submit/pending/续跑），打断的选择与语义在这里回答 —— 也避免 +// 编排器主文件越过「单文件 ≤500 行」的硬约束。 +// +// 上游清单：`Cyrene-For-Android/devdocs/02-前后端协同事项.md`（A2.scope）。 +// 客户端有「追加队列」与「打断」两种模式，若打断一律连队列一起清，用户选 +// 「打断」时连发的多条消息会被静默丢弃，只回一个 dropped 数字。 + +// InterruptScope 选择一次 interrupt 清到哪一层（协同项 A2）。 +type InterruptScope string + +const ( + // InterruptScopeAll 取消在飞回合并清空待处理队列。它是省略 scope 时的 + // 默认口径，与历史行为逐字一致。 + InterruptScopeAll InterruptScope = "all" + // InterruptScopeCurrent 只取消在飞回合，**保留**待处理队列：队列里的消息 + // 按编排器正常的续跑规则继续执行，一条都不丢。 + InterruptScopeCurrent InterruptScope = "current" +) + +// Interrupt 取消在飞回合、清空待处理队列并把会话回到空闲（design/15 §3， +// 「停下/算了/换个话题」路径）。它是忙碌会话不必排空队列就能复位的唯一方式； +// 运行中 goroutine 的完成变成陈旧代次，不续跑任何东西。它返回被丢弃的 +// 待处理回合数。 +// +// 它把会话立即报成空闲并立即接受下一条消息（design/15 §3 的「立即 +// 处理新消息」），但被取消的回合仍要收尾：下一个回合的启动被 session_turn_fence.go +// 的围栏挡在那次收尾之后，因此它绝不会读到被放弃回合那份仍可恢复的旧快照。 +func (o *Orchestrator) Interrupt(sessionID string) int { + return o.InterruptScope(sessionID, InterruptScopeAll) +} + +// InterruptScope 按 scope 中断一个会话（协同项 A2），返回真的被丢弃的 +// 待处理回合数： +// +// - InterruptScopeAll：取消在飞回合、清空待处理队列并把会话回到空闲 +// —— 省略 scope 时的历史语义，逐字不变； +// - InterruptScopeCurrent：只取消在飞回合，**保留**待处理队列。它不递增 +// 代次，因此被取消回合的结算（completeTurn）仍属当代，会按正常续跑 +// 规则弹出队列里的下一条消息、带上会话尾部上下文继续跑（design/15 +// §2.2），返回 0 —— 真的没有消息被丢弃。 +// +// 空字符串与省略同义（都是现状语义），使老客户端只发 {sessionId} 的 +// 行为完全不变。 +func (o *Orchestrator) InterruptScope(sessionID string, scope InterruptScope) int { + if scope == "" { + scope = InterruptScopeAll + } + sessionID = o.normalize(sessionID) + o.mu.Lock() + defer o.mu.Unlock() + return o.interruptLocked(sessionID, o.states[sessionID], scope) +} + +// InterruptAll 取消每一个在飞会话并清空所有待处理队列。它在网关关闭时使用， +// 使没有后台回合比 hub 活得更久。它返回被丢弃的待处理回合总数。 +func (o *Orchestrator) InterruptAll() int { + o.mu.Lock() + total := 0 + for id, st := range o.states { + total += o.interruptLocked(id, st, InterruptScopeAll) + } + o.mu.Unlock() + return total +} + +// interruptLocked 在状态锁下执行一次中断。scope 决定待处理队列的去留 +// （见 InterruptScope）：All 清队列、递增代次并把会话回到空闲；Current +// 只取消在飞回合，队列与代次原样留下，使被取消回合的结算仍会续跑队列。 +// 它返回真的被丢弃的待处理回合数（Current 恒为 0）。 +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() + } + if scope == InterruptScopeCurrent { + // 只取消在途回合：不碰 pending、不递增代次。mode 如实报告为 + // 当时的忙碌通道（回合仍在收尾、队列还在续跑），而不是 idle。 + o.emitState(sessionID, EventSessionInterrupt, map[string]any{"sessionId": sessionID, "mode": st.mode, "dropped": 0}) + return 0 + } + dropped := len(st.pending) + st.pending = nil + st.gen++ + st.mode = ModeIdle + st.cancel = nil + o.emitState(sessionID, EventSessionInterrupt, map[string]any{"sessionId": sessionID, "mode": ModeIdle, "dropped": dropped}) + return dropped +} diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index 2231908..3d3b484 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -125,6 +125,10 @@ type sessionState struct { cancel context.CancelFunc pending []pendingMsg tail []model.Message // 用于上下文续跑的有界对话尾部 + // fence 是当前在途回合的启动围栏（见 session_turn_fence.go）：下一个 + // 回合以它为先驱，等本回合真正返回后才开始 —— 因此中断后立即到达的 + // 消息绝不会在被取消回合落定其 checkpoint 之前读到那份旧快照。 + fence turnFence } // pendingMsg 是一条已排队用户回合。它携带自己的 env，使续跑 @@ -243,6 +247,10 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes st.gen++ gen := st.gen st.mode = ModeThinking + // 本回合的围栏以当前在途回合的收尾信号为先驱，使中断后来到的这条 + // 消息不会抢在被取消回合落定旧快照之前开跑（session_turn_fence.go）。 + fence := newTurnFence(st.fence.done) + st.fence = fence // 新回合延续该会话的对话：持久化转录被播种在 // 刚发送的消息之前（design/15 §2.2/§4，design/22 // §7-5），且其助手话语保持说话时的角色。该读取 @@ -263,7 +271,8 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID) if sync { - res, err := o.run(runCtx, sessionID, turn, env) + res, err := o.runTurnFenced(runCtx, sessionID, turn, env, fence) + fence.release() if err != nil { o.log.Error("orchestrator turn", "err", err, "session", sessionID) } @@ -271,7 +280,7 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes return SubmitResult{Accepted: true, Immediate: true, Err: err}, res } - go o.goTurn(sessionID, gen, turn, env, runCtx, cancel, deviceID, msgMode) + go o.goTurn(sessionID, gen, turn, env, runCtx, cancel, deviceID, msgMode, fence) return SubmitResult{Accepted: true, Immediate: true}, nil } @@ -280,10 +289,12 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes // deviceID 用发送该回合的客户端标注持久化助手回复 // （design/16 §4.5），mode 用该回合的行为模式（design/16 §5.1），而 env 是 // 该回合解析出的角色/隔离环境（design/22 §4.5），因此回复被 -// 归到产出它的角色名下。 -func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Message, env loop.SessionEnv, ctx context.Context, cancel context.CancelFunc, deviceID, mode string) { +// 归到产出它的角色名下。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) { defer cancel() - res, err := o.run(ctx, sessionID, messages, env) + defer fence.release() + res, err := o.runTurnFenced(ctx, sessionID, messages, env, fence) if err != nil && ctx.Err() == nil { o.log.Error("orchestrator turn", "err", err, "session", sessionID) } @@ -319,13 +330,17 @@ func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Resu gen = st.gen runCtx, cancel := context.WithCancel(context.Background()) st.cancel = cancel + // 本回合的这次 Run 已经返回（我们正处在它的结算里），因此被续跑 + // 的回合无需再等任何在途回合：围栏只负责放行它后面的那一个。 + fence := newTurnFence(nil) + st.fence = fence depth := len(st.pending) 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) + go o.goTurn(sessionID, gen, resume, env, runCtx, cancel, next.deviceID, next.mode, fence) return nil } @@ -337,50 +352,6 @@ func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Resu return nil } -// Interrupt 取消在飞回合、清空待处理队列并把会话 -// 回到空闲（design/15 §3，「停下/算了/换个话题」路径）。它是忙碌 -// 会话不必排空队列就能复位的唯一方式；运行中 goroutine 的 -// 完成变成陈旧代次，不续跑任何东西。它返回 -// 被丢弃的待处理回合数。 -func (o *Orchestrator) Interrupt(sessionID string) int { - return o.interrupt(o.normalize(sessionID)) -} - -// InterruptAll 取消每一个在飞会话并清空所有待处理队列。它 -// 在网关关闭时使用，使没有后台回合比 hub 活得更久。它返回 -// 被丢弃的待处理回合总数。 -func (o *Orchestrator) InterruptAll() int { - o.mu.Lock() - total := 0 - for id, st := range o.states { - total += o.interruptLocked(id, st) - } - o.mu.Unlock() - return total -} - -func (o *Orchestrator) interrupt(sessionID string) int { - o.mu.Lock() - defer o.mu.Unlock() - return o.interruptLocked(sessionID, o.states[sessionID]) -} - -func (o *Orchestrator) interruptLocked(sessionID string, st *sessionState) int { - if st == nil || st.mode == ModeIdle { - return 0 - } - if st.cancel != nil { - st.cancel() - } - dropped := len(st.pending) - st.pending = nil - st.gen++ - st.mode = ModeIdle - st.cancel = nil - o.emitState(sessionID, EventSessionInterrupt, map[string]any{"sessionId": sessionID, "mode": ModeIdle, "dropped": dropped}) - return dropped -} - // SetMode 提升某会话所报告的忙碌状态通道（thinking → // executing → subagent）。它是 loop/subagent 钩子可调用的注入点； // 任何非空闲模式都让会话保持忙碌，而 idle 解除忙碌。为 nil 的会话 diff --git a/internal/services/gateway/ws_hub_turn.go b/internal/services/gateway/ws_hub_turn.go index c0c7eda..37b6f83 100644 --- a/internal/services/gateway/ws_hub_turn.go +++ b/internal/services/gateway/ws_hub_turn.go @@ -5,6 +5,7 @@ import ( "crypto/rand" "encoding/hex" "encoding/json" + "fmt" "strings" "time" @@ -103,21 +104,66 @@ func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Messag return res, nil } -// handleInterrupt 把客户端的 "interrupt" 帧（design/15 §3）路由到 -// orchestrator，取消在途轮次并清空指定会话的待处理队列。未接 -// orchestrator 时是空操作。 -func (h *hub) handleInterrupt(payload json.RawMessage) { - if h.orchestrator == nil { +// wsBadRequestCode 是上行帧载荷非法时定向回发送者的机器可读错误码。它与 +// ws_route.go 的 wsForbiddenCode 同形（{type:"error", data:{code}}）， +// 因此客户端只看 data.code 就能把「你没有这个权限」与「这个帧写错了」 +// 区分开。 +const wsBadRequestCode = "bad_request" + +// interruptRequest 是解码后的 "interrupt" 上行载荷（design/15 §3）。 +type interruptRequest struct { + SessionID string `json:"sessionId"` + // Scope 选择这次打断清到哪一层（协同项 A2，取值见 InterruptScope）： + // 省略 / 空串 / "all" = 现状语义（取消在途回合并清空待处理队列）， + // "current" = 只取消在途回合、保留待处理队列让它们照常续跑。 + // 其余取值一律显式拒绝 —— 客户端以为队列保住了、实际被清空， + // 就是静默丢消息，正是 A2 要修掉的那类缺陷。 + Scope string `json:"scope,omitempty"` +} + +// parseInterruptScope 把上行 scope 解析为取值域内的口径。省略、空串与 +// "all" 都落到现状语义（省略 = 什么都不改，因此老客户端只发 +// {sessionId} 的行为逐字不变）；"current" 是新增的保留队列口径；其余 +// 取值返回错误，由调用方定向回 bad_request，绝不静默当默认执行。 +// +// 取值大小写敏感（与契约里的枚举 id 同口径），只容忍首尾空白。 +func parseInterruptScope(raw string) (InterruptScope, error) { + switch strings.TrimSpace(raw) { + case "", string(InterruptScopeAll): + return InterruptScopeAll, nil + case string(InterruptScopeCurrent): + return InterruptScopeCurrent, nil + default: + return "", fmt.Errorf("interrupt: 未知 scope %q（合法值：%q / %q）", raw, InterruptScopeAll, InterruptScopeCurrent) + } +} + +// handleInterrupt 处理客户端的 "interrupt" 帧（design/15 §3 + 协同项 A2）： +// 按 scope 让 orchestrator 取消该会话的在途回合，必要时保留待处理队列。 +// 未接 orchestrator 时是空操作。 +// +// 它是**连接级**的，因为非法 scope 必须定向回发送者一条 +// {type:"error", data:{code:"bad_request"}}：只到发送者、不广播、也不打断 +// 任何会话（显式报错，绝不静默按默认执行）。该帧只由 owner/admin 连接 +// 分发到此（design/11 §7.2 B）。 +// +// 缺 sessionId 时是空操作（既有行为，本次不改）。 +func (c *wsConn) handleInterrupt(payload json.RawMessage) { + if c.h == nil { return } - var req struct { - SessionID string `json:"sessionId"` - } + var req interruptRequest _ = json.Unmarshal(payload, &req) - if req.SessionID == "" { + scope, err := parseInterruptScope(req.Scope) + if err != nil { + c.h.log.Debug("ws interrupt rejected", "err", err, "scope", req.Scope) + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"code": wsBadRequestCode}}) return } - h.orchestrator.Interrupt(req.SessionID) + if c.h.orchestrator == nil || req.SessionID == "" { + return + } + c.h.orchestrator.InterruptScope(req.SessionID, scope) } // broadcastStream 把流式的 model.StreamEvent 映射为其下行 WS 事件类型，