AskaEth
074ffc4ebf
主人 2026-10-03 拍板「注释一律中文」；gateway 至此整包注释中文化完成（本会话共 3 批： 已推库 16 个文件 + 本批剩余文件）。 校验：机械校验（非注释改动行）= 空；剥注释后代码文本逐行比对全等（不一致文件数 0）； 全树 go test ./... 58 包 ok / 0 FAIL；交叉编译绿。
813 lines
25 KiB
Go
813 lines
25 KiB
Go
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"
|
||
)
|
||
|
||
// callRecord 是假 runner 捕获的一次调用。
|
||
type callRecord struct {
|
||
sessionID string
|
||
messages []model.Message
|
||
env loop.SessionEnv
|
||
ctx context.Context
|
||
}
|
||
|
||
// gatedRunner 是一个可控 TurnRunner。每次 Run 都会阻塞，直到它
|
||
// 从 tickets 消费一张票，或直到 ctx 被取消（打断路径）。当它
|
||
// 继续时，它记录这次调用并返回确定性回复。err 为 nil
|
||
// 会产出一个把*传入*消息作为文本持有的回复，使测试能断言
|
||
// 续跑回合究竟携带了什么上下文。
|
||
//
|
||
// 与真实的 loop runner 一样，它绝不执行 context 已被取消的
|
||
// 回合，并统计那些被取消的返回，使测试能等待一个
|
||
// 被中断的回合真正停下（waitCanceled），然后再释放下一张
|
||
// 票。没有这一点，一个在下面 select 中驻留时被取消的回合
|
||
// 仍可能在同一调度步骤中赢下 ticket 分支（两个分支都
|
||
// 就绪）并被记录成仿佛已经运行 —— 这是测试夹具的产物，不是
|
||
// 编排器的顺序保证。
|
||
type gatedRunner struct {
|
||
mu sync.Mutex
|
||
calls []callRecord
|
||
canceled int
|
||
tickets chan struct{}
|
||
err error
|
||
}
|
||
|
||
func newGatedRunner(cap int) *gatedRunner {
|
||
return &gatedRunner{tickets: make(chan struct{}, cap)}
|
||
}
|
||
|
||
func (g *gatedRunner) Run(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) (*loop.Result, error) {
|
||
if err := ctx.Err(); err != nil {
|
||
g.recordCanceled()
|
||
return nil, err
|
||
}
|
||
select {
|
||
case <-g.tickets:
|
||
case <-ctx.Done():
|
||
g.recordCanceled()
|
||
return nil, ctx.Err()
|
||
}
|
||
g.mu.Lock()
|
||
n := len(g.calls) + 1
|
||
g.calls = append(g.calls, callRecord{sessionID: sessionID, messages: append([]model.Message(nil), messages...), env: env, ctx: ctx})
|
||
g.mu.Unlock()
|
||
if g.err != nil {
|
||
return nil, g.err
|
||
}
|
||
reply := n + len(messages)
|
||
return &loop.Result{Text: fmt.Sprintf("reply-%d", reply), Steps: 1, StopReason: "stop"}, nil
|
||
}
|
||
|
||
// waitCalls 阻塞直到至少记录 n 次调用，并返回前 n 次的
|
||
// 副本，使测试能确定性地推进受门控的链条。
|
||
func (g *gatedRunner) waitCalls(n int) []callRecord {
|
||
deadline := time.Now().Add(3 * time.Second)
|
||
for time.Now().Before(deadline) {
|
||
g.mu.Lock()
|
||
if len(g.calls) >= n {
|
||
out := append([]callRecord(nil), g.calls[:n]...)
|
||
g.mu.Unlock()
|
||
return out
|
||
}
|
||
g.mu.Unlock()
|
||
time.Sleep(time.Millisecond)
|
||
}
|
||
panic("timed out waiting for runner calls")
|
||
}
|
||
|
||
// count 返回已记录调用的总数。
|
||
func (g *gatedRunner) count() int {
|
||
g.mu.Lock()
|
||
defer g.mu.Unlock()
|
||
return len(g.calls)
|
||
}
|
||
|
||
// recordCanceled 记下一次因其 context 被取消而返回的调用，
|
||
// 即一个从未执行的回合。
|
||
func (g *gatedRunner) recordCanceled() {
|
||
g.mu.Lock()
|
||
g.canceled++
|
||
g.mu.Unlock()
|
||
}
|
||
|
||
// waitCanceled 阻塞直到至少 n 次调用因 context 被取消而
|
||
// 停止，未能在期限内发生时报 false。它是测试在 Interrupt
|
||
// 与释放下一张票之间使用的屏障：它一旦成立，就没有
|
||
// 被取消的回合还能消费一张票。
|
||
func (g *gatedRunner) waitCanceled(n int) bool {
|
||
deadline := time.Now().Add(3 * time.Second)
|
||
for time.Now().Before(deadline) {
|
||
g.mu.Lock()
|
||
ok := g.canceled >= n
|
||
g.mu.Unlock()
|
||
if ok {
|
||
return true
|
||
}
|
||
time.Sleep(time.Millisecond)
|
||
}
|
||
return false
|
||
}
|
||
|
||
// waitIdle 阻塞直到该会话报告空闲且待处理队列为空，
|
||
// 未能在期限内安定时报 false。它让测试能确定性地断言
|
||
// 安定的状态而不是睡眠等待，并把「被丢弃的回合竟仍被
|
||
// 续跑」变成测试失败，而不是静默通过。
|
||
func waitIdle(o *Orchestrator, sessionID string) bool {
|
||
deadline := time.Now().Add(3 * time.Second)
|
||
for time.Now().Before(deadline) {
|
||
if !o.Busy(sessionID) && o.QueueDepth(sessionID) == 0 {
|
||
return true
|
||
}
|
||
time.Sleep(time.Millisecond)
|
||
}
|
||
return false
|
||
}
|
||
|
||
// eventRec 是记录型汇聚点捕获的一次状态转移。
|
||
type eventRec struct {
|
||
sessionID, typ string
|
||
data map[string]any
|
||
}
|
||
|
||
// recordingSink 捕获每一次发出的状态转移。
|
||
type recordingSink struct {
|
||
mu sync.Mutex
|
||
events []eventRec
|
||
}
|
||
|
||
func (s *recordingSink) Emit(sessionID, typ string, data map[string]any) {
|
||
s.mu.Lock()
|
||
s.events = append(s.events, eventRec{sessionID: sessionID, typ: typ, data: data})
|
||
s.mu.Unlock()
|
||
}
|
||
|
||
func (s *recordingSink) types() []string {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
out := make([]string, 0, len(s.events))
|
||
for _, e := range s.events {
|
||
out = append(out, e.typ)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func (s *recordingSink) last() (string, bool) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
if len(s.events) == 0 {
|
||
return "", false
|
||
}
|
||
return s.events[len(s.events)-1].typ, true
|
||
}
|
||
|
||
// userMsg 构建一条纯文本用户消息。
|
||
func userMsg(text string) model.Message {
|
||
return model.TextMessage(model.RoleUser, text)
|
||
}
|
||
|
||
// messageText 返回消息文本部分拼接后的文本。
|
||
func messageText(m model.Message) string {
|
||
var out string
|
||
for _, c := range m.Content {
|
||
if c.Type == model.ContentTypeText {
|
||
out += c.Text
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// seedOrchestrator 构建一个带受门控 runner 与记录型汇聚点的
|
||
// 编排器，二者都返回，使测试能驱动链条并断言状态事件。
|
||
func seedOrchestrator(run *gatedRunner, maxPending int) (*Orchestrator, *recordingSink) {
|
||
sink := &recordingSink{}
|
||
o := NewOrchestrator(OrchestratorOptions{
|
||
Run: run.Run,
|
||
Emit: sink.Emit,
|
||
Log: slog.Default(),
|
||
MaxPending: maxPending,
|
||
MaxTail: 20,
|
||
})
|
||
return o, sink
|
||
}
|
||
|
||
// readOnlyEnv 返回群聊 / 陌生人的只读记忆通道。
|
||
func readOnlyEnv() loop.SessionEnv {
|
||
return loop.SessionEnv{MemWritable: false}
|
||
}
|
||
|
||
// fakeHistory 记录每一次 Append，使测试能断言编排器把用户
|
||
// 消息流与助手回复持久化进聊天历史存储。
|
||
type fakeHistory struct {
|
||
mu sync.Mutex
|
||
appends []history.Message
|
||
}
|
||
|
||
func (f *fakeHistory) Append(ctx context.Context, msg history.Message) error {
|
||
_, err := f.AppendFull(ctx, msg)
|
||
return err
|
||
}
|
||
|
||
func (f *fakeHistory) AppendFull(_ context.Context, msg history.Message) (history.Message, error) {
|
||
f.mu.Lock()
|
||
defer f.mu.Unlock()
|
||
if msg.ID == "" {
|
||
msg.ID = fmt.Sprintf("id-%d", len(f.appends)+1)
|
||
}
|
||
if msg.CreatedAt == 0 {
|
||
msg.CreatedAt = int64(len(f.appends) + 1)
|
||
}
|
||
f.appends = append(f.appends, msg)
|
||
return msg, nil
|
||
}
|
||
|
||
func (f *fakeHistory) List(context.Context, string, int64, int, string) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
|
||
func (f *fakeHistory) ListAfter(context.Context, string, int64, int, string) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
|
||
func (f *fakeHistory) Recent(context.Context, string, int) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
|
||
func (f *fakeHistory) ListPlatformLog(context.Context, string, int64, int) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
|
||
func (f *fakeHistory) ListAdapterLog(context.Context, string, int64, int) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
|
||
func (f *fakeHistory) ListAllLog(context.Context, int64, int) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
|
||
func (f *fakeHistory) DeleteSession(context.Context, string) error {
|
||
return nil
|
||
}
|
||
|
||
func (f *fakeHistory) Search(context.Context, string, string, int64, int) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
|
||
func (f *fakeHistory) UpdateMessage(_ context.Context, id, content string) (history.Message, error) {
|
||
return history.Message{ID: id, Content: content}, nil
|
||
}
|
||
|
||
func (f *fakeHistory) DeleteMessage(context.Context, string) error {
|
||
return nil
|
||
}
|
||
|
||
func (f *fakeHistory) RenameSession(context.Context, string, string) error {
|
||
return nil
|
||
}
|
||
|
||
func (f *fakeHistory) SetPinned(context.Context, string, bool) error {
|
||
return nil
|
||
}
|
||
|
||
func (f *fakeHistory) NewSubSession(context.Context, string) (string, error) {
|
||
return "", nil
|
||
}
|
||
|
||
func (f *fakeHistory) ListSessions(context.Context) ([]history.SessionMeta, error) {
|
||
return nil, nil
|
||
}
|
||
|
||
// MarkSessionRead 为读游标路径（D4）满足 HistoryStore 契约。
|
||
// 编排器绝不推进游标，因此该假实现只把请求的游标
|
||
// 原样报告回去，不予存储。
|
||
func (f *fakeHistory) MarkSessionRead(_ context.Context, _ string, before time.Time) (int64, error) {
|
||
return before.UnixMilli(), nil
|
||
}
|
||
|
||
func (f *fakeHistory) records() []history.Message {
|
||
f.mu.Lock()
|
||
defer f.mu.Unlock()
|
||
return append([]history.Message(nil), f.appends...)
|
||
}
|
||
|
||
func TestSubmitPersistsUserAndAssistant(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(1)
|
||
hist := &fakeHistory{}
|
||
o := NewOrchestrator(OrchestratorOptions{
|
||
Run: run.Run,
|
||
Log: slog.Default(),
|
||
History: hist,
|
||
})
|
||
run.tickets <- struct{}{}
|
||
sr, res := o.SubmitSync(context.Background(), "web", []model.Message{userMsg("你好")}, loop.OwnerEnv())
|
||
if !sr.Accepted || !sr.Immediate {
|
||
t.Fatalf("submit = %+v, want Accepted+Immediate", sr)
|
||
}
|
||
if res == nil {
|
||
t.Fatal("sync result is nil")
|
||
}
|
||
recs := hist.records()
|
||
if len(recs) != 2 {
|
||
t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs))
|
||
}
|
||
if recs[0].SessionID != "web" || recs[0].Role != history.RoleUser || recs[0].Content != "你好" {
|
||
t.Fatalf("user record = %+v", recs[0])
|
||
}
|
||
if recs[1].SessionID != "web" || recs[1].Role != history.RoleAssistant || recs[1].Content == "" {
|
||
t.Fatalf("assistant record = %+v", recs[1])
|
||
}
|
||
// ctx 上没有 mode → 二者都回退到 "chat"（design/16 §5.1）。
|
||
for i, rec := range recs {
|
||
if rec.Mode != "chat" {
|
||
t.Fatalf("record[%d] Mode = %q, want chat fallback", i, rec.Mode)
|
||
}
|
||
}
|
||
}
|
||
|
||
func TestSubmitPersistsMessageMode(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(1)
|
||
hist := &fakeHistory{}
|
||
o := NewOrchestrator(OrchestratorOptions{
|
||
Run: run.Run,
|
||
Log: slog.Default(),
|
||
History: hist,
|
||
})
|
||
run.tickets <- struct{}{}
|
||
// HTTP/WS 边界把客户端的行为模式经 ctx 串下来
|
||
// （design/01 §3、design/16 §5.1）；用户回合与助手回复
|
||
// 都必须以该模式被持久化。
|
||
ctx := withMode(context.Background(), "code")
|
||
sr, res := o.SubmitSync(ctx, "web", []model.Message{userMsg("改代码")}, loop.OwnerEnv())
|
||
if !sr.Accepted || !sr.Immediate {
|
||
t.Fatalf("submit = %+v, want Accepted+Immediate", sr)
|
||
}
|
||
if res == nil {
|
||
t.Fatal("sync result is nil")
|
||
}
|
||
recs := hist.records()
|
||
if len(recs) != 2 {
|
||
t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs))
|
||
}
|
||
if recs[0].Mode != "code" {
|
||
t.Fatalf("user Mode = %q, want code", recs[0].Mode)
|
||
}
|
||
if recs[1].Mode != "code" {
|
||
t.Fatalf("assistant Mode = %q, want code", recs[1].Mode)
|
||
}
|
||
}
|
||
|
||
func TestSubmitPersistsDeviceID(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(1)
|
||
hist := &fakeHistory{}
|
||
o := NewOrchestrator(OrchestratorOptions{
|
||
Run: run.Run,
|
||
Log: slog.Default(),
|
||
History: hist,
|
||
})
|
||
run.tickets <- struct{}{}
|
||
// HTTP/WS 边界把浏览器 deviceID 经 ctx 串下来（design/16 §4.5）；
|
||
// 用户回合与助手回复都必须被它标注。
|
||
ctx := withDeviceID(context.Background(), "web:chrome:a1b2c3")
|
||
sr, res := o.SubmitSync(ctx, "web", []model.Message{userMsg("你好")}, loop.OwnerEnv())
|
||
if !sr.Accepted || !sr.Immediate {
|
||
t.Fatalf("submit = %+v, want Accepted+Immediate", sr)
|
||
}
|
||
if res == nil {
|
||
t.Fatal("sync result is nil")
|
||
}
|
||
recs := hist.records()
|
||
if len(recs) != 2 {
|
||
t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs))
|
||
}
|
||
if recs[0].Role != history.RoleUser || recs[0].Content != "你好" || recs[0].DeviceID != "web:chrome:a1b2c3" {
|
||
t.Fatalf("user record = %+v, want DeviceID web:chrome:a1b2c3", recs[0])
|
||
}
|
||
if recs[1].Role != history.RoleAssistant || recs[1].Content == "" || recs[1].DeviceID != "web:chrome:a1b2c3" {
|
||
t.Fatalf("assistant record = %+v, want DeviceID web:chrome:a1b2c3", recs[1])
|
||
}
|
||
}
|
||
|
||
func TestSubmitPersistedBroadcast(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(1)
|
||
hist := &fakeHistory{}
|
||
var mu sync.Mutex
|
||
var persisted []history.Message
|
||
o := NewOrchestrator(OrchestratorOptions{
|
||
Run: run.Run,
|
||
Log: slog.Default(),
|
||
History: hist,
|
||
OnPersisted: func(m history.Message) {
|
||
mu.Lock()
|
||
persisted = append(persisted, m)
|
||
mu.Unlock()
|
||
},
|
||
})
|
||
run.tickets <- struct{}{}
|
||
sr, _ := o.SubmitSync(context.Background(), "web", []model.Message{userMsg("你好")}, loop.OwnerEnv())
|
||
if !sr.Accepted || !sr.Immediate {
|
||
t.Fatalf("submit = %+v, want Accepted+Immediate", sr)
|
||
}
|
||
mu.Lock()
|
||
got := append([]history.Message(nil), persisted...)
|
||
mu.Unlock()
|
||
if len(got) != 2 {
|
||
t.Fatalf("persisted = %d, want 2 (user + assistant)", len(got))
|
||
}
|
||
if got[0].Role != history.RoleUser || got[0].ID == "" || got[0].SessionID != "web" || got[0].Content != "你好" {
|
||
t.Fatalf("user persisted = %+v", got[0])
|
||
}
|
||
if got[1].Role != history.RoleAssistant || got[1].ID == "" || got[1].SessionID != "web" || got[1].Content == "" {
|
||
t.Fatalf("assistant persisted = %+v", got[1])
|
||
}
|
||
if got[0].ID == got[1].ID {
|
||
t.Fatalf("user and assistant share id %q", got[0].ID)
|
||
}
|
||
}
|
||
|
||
func TestSubmitDoesNotPersistRejectedTurn(t *testing.T) {
|
||
t.Parallel()
|
||
hist := &fakeHistory{}
|
||
o := NewOrchestrator(OrchestratorOptions{Run: newGatedRunner(0).Run, Log: slog.Default(), History: hist, MaxPending: 1})
|
||
// 空回合在任何持久化之前就被拒绝。
|
||
sr := o.Submit(context.Background(), "web", nil, loop.OwnerEnv())
|
||
if sr.Accepted || !sr.Empty {
|
||
t.Fatalf("empty submit = %+v, want Empty reject", sr)
|
||
}
|
||
if n := len(hist.records()); n != 0 {
|
||
t.Fatalf("appends = %d, want 0 for a rejected turn", n)
|
||
}
|
||
}
|
||
|
||
func TestSubmitIdleStartsImmediate(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(1)
|
||
o, _ := seedOrchestrator(run, 5)
|
||
|
||
sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("hi")}, loop.OwnerEnv())
|
||
if !sr.Accepted || !sr.Immediate {
|
||
t.Fatalf("submit = %+v, want Accepted+Immediate", sr)
|
||
}
|
||
run.tickets <- struct{}{}
|
||
got := run.waitCalls(1)
|
||
if len(got) != 1 {
|
||
t.Fatalf("calls = %d, want 1", len(got))
|
||
}
|
||
if o.Busy("s1") {
|
||
t.Fatal("session should be idle after a completed turn")
|
||
}
|
||
}
|
||
|
||
func TestSubmitBusyPendingNotDropped(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(4)
|
||
o, sink := seedOrchestrator(run, 5)
|
||
|
||
// 回合 A 启动并阻塞（还没有票），因此会话处于忙碌。
|
||
a := o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv())
|
||
if !a.Immediate {
|
||
t.Fatalf("A = %+v, want Immediate", a)
|
||
}
|
||
// B 与 C 在 A 在飞时到达：它们必须被排队，而不是被丢弃。
|
||
b := o.Submit(context.Background(), "s1", []model.Message{userMsg("B")}, readOnlyEnv())
|
||
c := o.Submit(context.Background(), "s1", []model.Message{userMsg("C")}, loop.OwnerEnv())
|
||
if !b.Accepted || !b.Pending || b.QueueDepth != 1 {
|
||
t.Fatalf("B = %+v, want Pending depth=1", b)
|
||
}
|
||
if !c.Accepted || !c.Pending || c.QueueDepth != 2 {
|
||
t.Fatalf("C = %+v, want Pending depth=2", c)
|
||
}
|
||
if o.QueueDepth("s1") != 2 {
|
||
t.Fatalf("queue depth = %d, want 2", o.QueueDepth("s1"))
|
||
}
|
||
|
||
// 释放链条：A，然后 B，然后 C。发出的 ack 应包含
|
||
// 待处理深度，且汇聚点应已看到 pending_ack 事件。
|
||
types := sink.types()
|
||
foundAck := false
|
||
for _, ty := range types {
|
||
if ty == EventPendingAck {
|
||
foundAck = true
|
||
}
|
||
}
|
||
if !foundAck {
|
||
t.Fatalf("no pending_ack in events: %v", types)
|
||
}
|
||
|
||
run.tickets <- struct{}{}
|
||
run.tickets <- struct{}{}
|
||
run.tickets <- struct{}{}
|
||
if n := len(run.waitCalls(3)); n != 3 {
|
||
t.Fatalf("total calls = %d, want 3 (A+B+C)", n)
|
||
}
|
||
if o.QueueDepth("s1") != 0 || o.Busy("s1") {
|
||
t.Fatalf("session should be idle and empty after drain: busy=%v depth=%d", o.Busy("s1"), o.QueueDepth("s1"))
|
||
}
|
||
}
|
||
|
||
func TestSubmitBusyResumeCarriesPreviousContext(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(3)
|
||
o, _ := seedOrchestrator(run, 5)
|
||
|
||
if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv()); !sr.Immediate {
|
||
t.Fatalf("A = %+v, want Immediate", sr)
|
||
}
|
||
if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("B")}, readOnlyEnv()); !sr.Pending {
|
||
t.Fatalf("B = %+v, want Pending", sr)
|
||
}
|
||
if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("C")}, loop.OwnerEnv()); !sr.Pending {
|
||
t.Fatalf("C = %+v, want Pending", sr)
|
||
}
|
||
|
||
run.tickets <- struct{}{} // A
|
||
run.tickets <- struct{}{} // B
|
||
run.tickets <- struct{}{} // C
|
||
got := run.waitCalls(3)
|
||
|
||
// A 从空白回合开始。
|
||
if len(got[0].messages) != 1 || messageText(got[0].messages[0]) != "A" {
|
||
t.Fatalf("A messages = %v, want [A]", messageTextList(got[0].messages))
|
||
}
|
||
// B 必须带着折进尾部的 A 助手回复续跑。
|
||
if !matchesText(got[1].messages, "A", "B") {
|
||
t.Fatalf("B messages = %v, want B to resume carrying the prior tail", messageTextList(got[1].messages))
|
||
}
|
||
// C 必须带着 A 与 B 的助手回复续跑。
|
||
if !matchesText(got[2].messages, "A", "C") {
|
||
t.Fatalf("C messages = %v, want C to resume carrying the prior tail", messageTextList(got[2].messages))
|
||
}
|
||
// B 的 env 必须是该消息到达时所用的那个（只读通道）。
|
||
if got[1].env.MemWritable {
|
||
t.Fatal("B resumed with MemWritable=true, want the caller's read-only env")
|
||
}
|
||
}
|
||
|
||
// matchesText 报告一个续跑回合是否以捕获的首个用户
|
||
// 回合开头（即先前上下文被带了过来）、以预期的新闻用户
|
||
// 回合结尾，且中间至少含有一条助手回复。
|
||
func matchesText(msgs []model.Message, first, last string) bool {
|
||
if len(msgs) < 3 {
|
||
return false
|
||
}
|
||
if messageText(msgs[0]) != first {
|
||
return false
|
||
}
|
||
if messageText(msgs[len(msgs)-1]) != last {
|
||
return false
|
||
}
|
||
for i := 1; i < len(msgs)-1; i++ {
|
||
if msgs[i].Role == model.RoleAssistant {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func messageTextList(msgs []model.Message) []string {
|
||
out := make([]string, 0, len(msgs))
|
||
for _, m := range msgs {
|
||
out = append(out, messageText(m))
|
||
}
|
||
return out
|
||
}
|
||
|
||
func TestSubmitPendingBoundedRejectsOverflow(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(8)
|
||
o, _ := seedOrchestrator(run, 3) // 紧凑队列
|
||
|
||
if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv()); !sr.Immediate {
|
||
t.Fatalf("A = %+v, want Immediate", sr)
|
||
}
|
||
for i := 0; i < 3; i++ {
|
||
sr := o.Submit(context.Background(), "s1", []model.Message{userMsg(fmt.Sprintf("p%d", i))}, loop.OwnerEnv())
|
||
if !sr.Pending || sr.QueueDepth != i+1 {
|
||
t.Fatalf("pending %d = %+v, want Pending depth=%d", i, sr, i+1)
|
||
}
|
||
}
|
||
// 队列已达容量（3），因此下一条消息被拒绝，而不是被丢弃。
|
||
over := o.Submit(context.Background(), "s1", []model.Message{userMsg("overflow")}, loop.OwnerEnv())
|
||
if over.Accepted || !over.PendingFull {
|
||
t.Fatalf("overflow = %+v, want PendingFull reject", over)
|
||
}
|
||
if o.QueueDepth("s1") != 3 {
|
||
t.Fatalf("queue depth = %d, want 3 (unchanged)", o.QueueDepth("s1"))
|
||
}
|
||
}
|
||
|
||
func TestInterruptCancelsTurnAndClearsPending(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(2)
|
||
o, sink := seedOrchestrator(run, 5)
|
||
|
||
if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv()); !sr.Immediate {
|
||
t.Fatalf("A = %+v, want Immediate", sr)
|
||
}
|
||
if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("B")}, loop.OwnerEnv()); !sr.Pending {
|
||
t.Fatalf("B = %+v, want Pending", sr)
|
||
}
|
||
|
||
// 不释放任何票：A 阻塞在飞。Interrupt 取消它并清除 B。
|
||
dropped := o.Interrupt("s1")
|
||
if dropped != 1 {
|
||
t.Fatalf("dropped = %d, want 1", dropped)
|
||
}
|
||
if o.Busy("s1") || o.QueueDepth("s1") != 0 {
|
||
t.Fatalf("session not idle after interrupt: busy=%v depth=%d", o.Busy("s1"), o.QueueDepth("s1"))
|
||
}
|
||
|
||
// 取消必须抵达在飞回合（design/15 §3：打断是优雅的
|
||
// context 取消），且它必须在释放任何后续回合之前停下：
|
||
// 被取消的回合绝不执行，因此它绝不可能消费本该属于
|
||
// 下一个回合的票。
|
||
if !run.waitCanceled(1) {
|
||
t.Fatal("interrupted turn never observed its context cancellation")
|
||
}
|
||
|
||
// 打断之后的新消息必须全新开始（Immediate），且
|
||
// 被取消的链条绝不续跑过 B。
|
||
if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("C")}, loop.OwnerEnv()); !sr.Immediate {
|
||
t.Fatalf("C = %+v, want Immediate after interrupt", sr)
|
||
}
|
||
run.tickets <- struct{}{}
|
||
got := run.waitCalls(1)
|
||
if len(got) != 1 {
|
||
t.Fatalf("calls = %d, want 1 (only C, not the canceled A or B)", len(got))
|
||
}
|
||
if messageText(got[0].messages[0]) != "C" {
|
||
t.Fatalf("only surviving call = %v, want the C turn", messageTextList(got[0].messages))
|
||
}
|
||
|
||
// C 的链条安定下来且无待运行项：若打断未能取消 A
|
||
// 或未能清除 B，存活的链条本会续跑一个回合，
|
||
// 而该会话就再也不会回到空闲。
|
||
if !waitIdle(o, "s1") {
|
||
t.Fatalf("session never settled idle: busy=%v depth=%d", o.Busy("s1"), o.QueueDepth("s1"))
|
||
}
|
||
if n := run.count(); n != 1 {
|
||
t.Fatalf("runner calls = %d, want 1 (neither the canceled A nor the dropped B may execute)", n)
|
||
}
|
||
// 被取消回合的完成对会话状态是空操作（其代次已
|
||
// 陈旧），因此它不得发出正在运行那个回合的 idle 事件：
|
||
// 本场景恰好有一个 session_idle（C 安定）。
|
||
idles := 0
|
||
for _, ty := range sink.types() {
|
||
if ty == EventSessionIdle {
|
||
idles++
|
||
}
|
||
}
|
||
if idles != 1 {
|
||
t.Fatalf("session_idle events = %d, want 1 (a canceled turn must not settle the fresh chain)", idles)
|
||
}
|
||
}
|
||
|
||
func TestSubmitConcurrentSameSessionIsSafeAndBounded(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(8)
|
||
o, _ := seedOrchestrator(run, 3)
|
||
|
||
const senders = 8
|
||
var wg sync.WaitGroup
|
||
accepted := make(chan SubmitResult, senders)
|
||
for i := 0; i < senders; i++ {
|
||
wg.Add(1)
|
||
go func(i int) {
|
||
defer wg.Done()
|
||
accepted <- o.Submit(context.Background(), "s1", []model.Message{userMsg(fmt.Sprintf("m%d", i))}, loop.OwnerEnv())
|
||
}(i)
|
||
}
|
||
wg.Wait()
|
||
close(accepted)
|
||
|
||
immediate, pending, full := 0, 0, 0
|
||
for sr := range accepted {
|
||
switch {
|
||
case sr.Immediate:
|
||
immediate++
|
||
case sr.Pending:
|
||
pending++
|
||
case sr.PendingFull:
|
||
full++
|
||
default:
|
||
t.Fatalf("unexpected submit result: %+v", sr)
|
||
}
|
||
}
|
||
// maxPending=3 时：恰好 1 个回合立即运行，最多 3 个排在其后；
|
||
// 其余发送者在容量处被拒绝，而不是被丢弃。
|
||
if immediate != 1 {
|
||
t.Fatalf("immediate = %d, want 1", immediate)
|
||
}
|
||
if pending != 3 {
|
||
t.Fatalf("pending = %d, want 3", pending)
|
||
}
|
||
if full != senders-4 {
|
||
t.Fatalf("pendingFull = %d, want %d", full, senders-4)
|
||
}
|
||
if o.QueueDepth("s1") != 3 {
|
||
t.Fatalf("queue depth = %d, want 3", o.QueueDepth("s1"))
|
||
}
|
||
|
||
// 排空链条：1 个在跑 + 3 个排队 = 4 个回合。无 panic / 无消息丢失。
|
||
for i := 0; i < 4; i++ {
|
||
run.tickets <- struct{}{}
|
||
}
|
||
if n := len(run.waitCalls(4)); n != 4 {
|
||
t.Fatalf("total calls = %d, want 4", n)
|
||
}
|
||
if o.Busy("s1") || o.QueueDepth("s1") != 0 {
|
||
t.Fatalf("session should be idle and empty after drain")
|
||
}
|
||
}
|
||
|
||
func TestSubmitEmptyRejected(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(1)
|
||
o, _ := seedOrchestrator(run, 5)
|
||
sr := o.Submit(context.Background(), "s1", nil, loop.OwnerEnv())
|
||
if sr.Accepted || !sr.Empty {
|
||
t.Fatalf("empty submit = %+v, want Empty reject", sr)
|
||
}
|
||
if run.count() != 0 {
|
||
t.Fatalf("runner invoked for an empty turn")
|
||
}
|
||
}
|
||
|
||
func TestSubmitSyncIdleReturnsResultBusyReturnsPending(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(2)
|
||
o, _ := seedOrchestrator(run, 5)
|
||
|
||
// 空闲 → 同步运行立即返回回复。推入一张票，使
|
||
// 受门控 runner 的阻塞 select 在同步等待期间得以解决。
|
||
run.tickets <- struct{}{}
|
||
sr, res := o.SubmitSync(context.Background(), "s1", []model.Message{userMsg("hi")}, loop.OwnerEnv())
|
||
if !sr.Accepted || !sr.Immediate {
|
||
t.Fatalf("sync idle = %+v, want Accepted+Immediate", sr)
|
||
}
|
||
if res == nil {
|
||
t.Fatal("sync idle should return a result")
|
||
}
|
||
|
||
// 经异步 Submit 握一个在飞的忙碌回合（阻塞，无票），随后
|
||
// 同步 submit 必须不阻塞、无结果地返回 Pending ack。
|
||
if sr2 := o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv()); !sr2.Immediate {
|
||
t.Fatalf("async A = %+v, want Immediate", sr2)
|
||
}
|
||
sr3, res3 := o.SubmitSync(context.Background(), "s1", []model.Message{userMsg("B")}, loop.OwnerEnv())
|
||
if !sr3.Accepted || !sr3.Pending {
|
||
t.Fatalf("sync busy = %+v, want Pending", sr3)
|
||
}
|
||
if res3 != nil {
|
||
t.Fatal("sync busy should not return a result (must not block)")
|
||
}
|
||
run.tickets <- struct{}{} // 让 A 结束，使编排器回到空闲
|
||
}
|
||
|
||
func TestInterruptAllClearsEverySession(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(2)
|
||
o, _ := seedOrchestrator(run, 5)
|
||
|
||
o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv())
|
||
o.Submit(context.Background(), "s1", []model.Message{userMsg("B")}, loop.OwnerEnv())
|
||
o.Submit(context.Background(), "s2", []model.Message{userMsg("X")}, loop.OwnerEnv())
|
||
|
||
if got := o.InterruptAll(); got != 1 {
|
||
t.Fatalf("InterruptAll dropped = %d, want 1 (B)", got)
|
||
}
|
||
if o.Busy("s1") || o.Busy("s2") {
|
||
t.Fatal("all sessions should be idle after InterruptAll")
|
||
}
|
||
}
|
||
|
||
func TestSetModeBusyAndIdle(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(1)
|
||
o, sink := seedOrchestrator(run, 5)
|
||
|
||
o.SetMode("s1", ModeExecuting)
|
||
if !o.Busy("s1") || o.ModeOf("s1") != ModeExecuting {
|
||
t.Fatalf("after executing: busy=%v mode=%q", o.Busy("s1"), o.ModeOf("s1"))
|
||
}
|
||
o.SetMode("s1", ModeIdle)
|
||
if o.Busy("s1") {
|
||
t.Fatal("session must be idle after SetMode(idle)")
|
||
}
|
||
if ty, ok := sink.last(); !ok || ty != EventSessionIdle {
|
||
t.Fatalf("last event = %q, want session_idle", ty)
|
||
}
|
||
}
|