package gateway import ( "context" "encoding/json" "errors" "log/slog" "net/http" "sync" "testing" "git.yeij.top/AskaEth/Cyrene_v2/internal/db" "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" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" ) // 本文件是 S3.5 两个契约缺口的回归测试(design/24 §6.1、§9.7): // - 缺口①:一个回合的**全部**附件都要落库、都要从历史还原(此前只剩第一个); // - 缺口②:抽取结论(status/chars/truncated)要在回合结束后回写进该回合用户 // 消息的 document 列,使历史接口与客户端看得到「读到了 / 读不到」。 // // 测试断言一律哨兵 + errors.Is(绝不按错误文本子串断言);日志断言看的是记录 // 本身的 message/属性,而不是拼接后的文本。 // TestMessageTextImageDocumentCollectsEveryDocument 是缺口①的直接回归:一回合 // 挂 3 个附件时,落库的引用数组必须恰好是 3 条且顺序与部件顺序一致(此前的实现 // 对第二个及以后的文档部件直接 continue,只剩第一个)。 func TestMessageTextImageDocumentCollectsEveryDocument(t *testing.T) { t.Parallel() m := model.Message{Role: model.RoleUser, Content: []model.Content{{Type: model.ContentTypeText, Text: "看这三个"}}} for _, name := range []string{"aaaabbbbccccdddd.txt", "eeff001122334455.md", "0011223344556677.csv"} { m.Content = append(m.Content, model.DocumentMessage(model.RoleUser, "", model.Document{Name: name, MIME: "text/plain", Size: 7}).Content...) } text, _, raw, err := messageTextImageDocument(m) if err != nil { t.Fatalf("messageTextImageDocument: %v", err) } if text != "看这三个" { t.Fatalf("text = %q", text) } refs, err := history.DecodeDocuments(raw) if err != nil { t.Fatalf("decode documents: %v", err) } if len(refs) != 3 { t.Fatalf("refs = %+v, want 3 条(全部附件都要落库)", refs) } for i, want := range []string{"aaaabbbbccccdddd.txt", "eeff001122334455.md", "0011223344556677.csv"} { if refs[i].Name != want { t.Fatalf("refs[%d].Name = %q, want %q(顺序必须与部件一致)", i, refs[i].Name, want) } } } // TestTranscriptRestoresAllDocumentParts 是缺口①的另一半,且走真实存储:一条带 // 3 个引用的用户消息落库、读回、还原为 3 个文档部件(顺序一致);附件绝不能 // 在发起它的回合之后凭空消失。 func TestTranscriptRestoresAllDocumentParts(t *testing.T) { t.Parallel() ctx := context.Background() st := newGatewayHistoryStore(t) names := []string{"aaaabbbbccccdddd.txt", "eeff001122334455.md", "0011223344556677.csv"} refs := make([]history.DocumentRef, 0, len(names)) for _, name := range names { refs = append(refs, history.DocumentRef{Name: name, MIME: "text/plain", Size: 11, Status: history.DocumentStatusExtracted}) } raw, err := history.EncodeDocuments(refs) if err != nil { t.Fatalf("encode: %v", err) } if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "读这三份", Document: raw}); err != nil { t.Fatalf("append: %v", err) } recent, err := st.Recent(ctx, history.MainSessionID, 10) if err != nil { t.Fatalf("recent: %v", err) } if len(recent) != 1 { t.Fatalf("recent = %+v, want 1 条", recent) } parts := chatDocumentsOf([]model.Message{transcript.Message(recent[0])}) if len(parts) != 3 { t.Fatalf("还原出的文档部件 = %d 个, want 3(全部附件都要回到上下文)", len(parts)) } for i, want := range names { if parts[i].DocumentName != want { t.Fatalf("parts[%d].DocumentName = %q, want %q", i, parts[i].DocumentName, want) } if parts[i].DocumentURL != documentFilePrefix+want { t.Fatalf("parts[%d].DocumentURL = %q", i, parts[i].DocumentURL) } } } // TestTranscriptRestoresLegacySingleObjectColumn 钉住旧行兼容:S2/S3 早期写下的 // 单对象列仍能还原出那一个部件,绝不因为形状老就被当成坏数据丢掉。 func TestTranscriptRestoresLegacySingleObjectColumn(t *testing.T) { t.Parallel() const legacy = `{"name":"aaaabbbbccccdddd.txt","mime":"text/plain","size":1234,"status":"extracted"}` parts := chatDocumentsOf([]model.Message{ transcript.Message(history.Message{ID: "m1", Role: history.RoleUser, Content: "旧行", Document: legacy}), }) if len(parts) != 1 || parts[0].DocumentName != "aaaabbbbccccdddd.txt" { t.Fatalf("旧单对象列还原出 = %+v, want 那一条引用", parts) } } // TestMergeDocumentStatus 覆盖合并规则:按存储名补结论(身份与大小不动)、没匹配 // 到的原样保留、无变化时报「不必写库」、坏列报哨兵错误。 func TestMergeDocumentStatus(t *testing.T) { t.Parallel() raw, err := history.EncodeDocuments([]history.DocumentRef{ {Name: "aaa.txt", MIME: "text/plain", Size: 1}, {Name: "bbb.pdf", MIME: "application/pdf", Size: 2}, {Name: "ccc.txt", MIME: "text/plain", Size: 3}, }) if err != nil { t.Fatalf("encode: %v", err) } const scanReason = "无法抽取正文:这是图片型/无文本层 PDF,需要 OCR 才能读(当前未做)" merged, changed, err := mergeDocumentStatus(raw, []history.DocumentRef{ {Name: "aaa.txt", Status: history.DocumentStatusExtracted, Chars: 42, Truncated: true}, {Name: "bbb.pdf", Status: history.DocumentStatusUnreadable, Reason: scanReason}, }) if err != nil { t.Fatalf("merge: %v", err) } if !changed { t.Fatal("changed = false, want 补进了新结论") } refs, err := history.DecodeDocuments(merged) if err != nil { t.Fatalf("decode merged: %v", err) } if len(refs) != 3 { t.Fatalf("refs = %+v, want 3 条(没匹配到的原样保留)", refs) } if refs[0].Status != history.DocumentStatusExtracted || refs[0].Chars != 42 || !refs[0].Truncated || refs[0].MIME != "text/plain" || refs[0].Size != 1 { t.Fatalf("refs[0] = %+v, want 结论补上且身份不变", refs[0]) } // v0.63:reason 与 status 同批回写,客户端因此不会看到「unreadable 但没有理由」。 if refs[1].Status != history.DocumentStatusUnreadable || refs[1].Reason != scanReason { t.Fatalf("refs[1] = %+v, want unreadable + 原因文案", refs[1]) } if refs[2].Status != "" || refs[2].Reason != "" || refs[2].Name != "ccc.txt" { t.Fatalf("refs[2] = %+v, want 原样保留(本回合没解析到它)", refs[2]) } // 结论与原因都补齐后,同一批再合一次必须报「无变化」——reason 也是变化判据的 // 一部分(否则每回合都会因为空 reason 而重写一次)。 if _, changed, err := mergeDocumentStatus(merged, []history.DocumentRef{ {Name: "aaa.txt", Status: history.DocumentStatusExtracted, Chars: 42, Truncated: true}, {Name: "bbb.pdf", Status: history.DocumentStatusUnreadable, Reason: scanReason}, }); err != nil || changed { t.Fatalf("changed = %v, err = %v, want 无变化且无错(不必写库)", changed, err) } if _, _, err := mergeDocumentStatus("{not json", nil); !errors.Is(err, history.ErrBadDocuments) { t.Fatalf("merge(坏列) = %v, want history.ErrBadDocuments", err) } } // TestOrchestratorWritesBackDocumentStatus 是缺口②的端到端回归(走真实存储): // 回合结束后,该回合用户消息的 document 列里 status/chars/truncated 由空变为 // 装配期解析出的结论,其它行一个字都不动。 func TestOrchestratorWritesBackDocumentStatus(t *testing.T) { t.Parallel() ctx := context.Background() st := newGatewayHistoryStore(t) const nameA = "aaaabbbbccccdddd.txt" const nameB = "eeff001122334455.pdf" resA := &loop.Result{Text: "读完了", Documents: []history.DocumentRef{{ Name: nameA, MIME: "text/plain", Size: 2048, Chars: 900, Truncated: true, Status: history.DocumentStatusExtracted, }}} resB := &loop.Result{Text: "读不到", Documents: []history.DocumentRef{{ Name: nameB, MIME: "application/pdf", Size: 4096, Status: history.DocumentStatusUnreadable, }}} answers := []*loop.Result{resA, resB} call := 0 o := NewOrchestrator(OrchestratorOptions{ Run: func(context.Context, string, []model.Message, loop.SessionEnv) (*loop.Result, error) { res := answers[call] call++ return res, nil }, Log: testLogger(), History: st, }) for _, tc := range []struct{ text, name string }{{"读一下 A", nameA}, {"读一下 B", nameB}} { docs := []model.Document{{Name: tc.name, MIME: "text/plain", URL: documentFilePrefix + tc.name, Size: 2048}} o.SubmitSync(ctx, history.MainSessionID, chatTurnMessages(tc.text, "", docs), loop.SessionEnv{}) } msgs, err := st.List(ctx, history.MainSessionID, 0, 10, "") if err != nil { t.Fatalf("list: %v", err) } got := make(map[string][]history.DocumentRef, len(msgs)) for _, m := range msgs { if m.Document == "" { continue } refs, err := history.DecodeDocuments(m.Document) if err != nil { t.Fatalf("decode %s: %v", m.ID, err) } got[m.Content] = refs } first, ok := got["读一下 A"] if !ok || len(first) != 1 { t.Fatalf("读一下 A 的引用 = %+v, want 1 条", first) } if first[0].Status != history.DocumentStatusExtracted || first[0].Chars != 900 || !first[0].Truncated { t.Fatalf("A 的结论 = %+v, want extracted/chars=900/truncated", first[0]) } second, ok := got["读一下 B"] if !ok || len(second) != 1 || second[0].Status != history.DocumentStatusUnreadable { t.Fatalf("B 的结论 = %+v, want unreadable(「昔涟读不到这个文件」必须回写到列里)", second) } if len(got) != 2 { t.Fatalf("带回结论的行 = %d 条, want 2(助手回复不该有 document 列)", len(got)) } } // TestPersistDocumentStatusIsNotSilent 覆盖「不静默」的四条冷路径:存储未接线、 // 存储没有回写视图、按 id 找不到行、列被写坏 —— 每一条都要留下带来源的 WARN, // 而回合本身绝不因此失败(不卡会话)。 func TestPersistDocumentStatusIsNotSilent(t *testing.T) { t.Parallel() ctx := context.Background() res := &loop.Result{Text: "ok", Documents: []history.DocumentRef{{Name: "aaa.txt", Status: history.DocumentStatusExtracted}}} msgs := chatTurnMessages("读一下", "", []model.Document{{Name: "aaa.txt", MIME: "text/plain", Size: 1}}) t.Run("未接线历史存储", func(t *testing.T) { cap := newCaptureHandler() o := NewOrchestrator(OrchestratorOptions{Run: fixedRunner(res), Log: slog.New(cap)}) o.SubmitSync(ctx, "web", msgs, loop.SessionEnv{}) if !cap.has(`document status not persisted`, "no history store") { t.Fatal("未接线时没有留下带来源的 WARN(静默丢结论)") } }) t.Run("存储没有回写视图", func(t *testing.T) { cap := newCaptureHandler() o := NewOrchestrator(OrchestratorOptions{Run: fixedRunner(res), Log: slog.New(cap), History: &fakeHistory{}}) o.SubmitSync(ctx, "web", msgs, loop.SessionEnv{}) if !cap.has(`document status not persisted`, "history store has no document update path") { t.Fatal("存储缺少回写视图时没有留下带来源的 WARN(静默丢结论)") } }) t.Run("找不到对应行", func(t *testing.T) { cap := newCaptureHandler() writer := &fakeDocumentWriter{fakeHistory: &fakeHistory{}, err: history.ErrMessageNotFound} o := NewOrchestrator(OrchestratorOptions{Run: fixedRunner(res), Log: slog.New(cap), History: writer}) o.SubmitSync(ctx, "web", msgs, loop.SessionEnv{}) if !cap.has(`document status update`, "") { t.Fatal("回写失败时没有留下带来源的 WARN(静默丢结论)") } if len(writer.snapshot()) != 0 { t.Fatal("失败的写入不该被记成功") } }) t.Run("列被写坏", func(t *testing.T) { cap := newCaptureHandler() 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`, "") { t.Fatal("坏列没有留下带来源的 WARN(静默丢结论)") } if len(writer.snapshot()) != 0 { t.Fatal("坏列不该被写回") } }) } // 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` 数组回传引用与结论,且没有附件的消息**省略**该字段(省略 = // 没有附件,与引入文档之前逐字一致)。 func TestHistoryEndpointReturnsDocumentStatusArray(t *testing.T) { t.Parallel() ctx := context.Background() _, ts, jwt, st := newHistoryEnv(t) raw, err := history.EncodeDocuments([]history.DocumentRef{ {Name: "aaa.txt", MIME: "text/plain", Size: 1, Status: history.DocumentStatusExtracted, Chars: 7}, {Name: "bbb.pdf", MIME: "application/pdf", Size: 2, Status: history.DocumentStatusUnreadable}, }) if err != nil { t.Fatalf("encode: %v", err) } if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "两个附件", Document: raw, CreatedAt: 1000}); err != nil { t.Fatalf("append: %v", err) } if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "没有附件", CreatedAt: 2000}); err != nil { t.Fatalf("append: %v", err) } tok := signAny(t, jwt, "acct", "owner") resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, tok) if resp.StatusCode != http.StatusOK { t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) } var envelope struct { Data []map[string]json.RawMessage `json:"data"` } if err := json.Unmarshal(body, &envelope); err != nil { t.Fatalf("unmarshal: %v (body=%s)", err, body) } if len(envelope.Data) != 2 { t.Fatalf("messages = %d 条, want 2 (body=%s)", len(envelope.Data), body) } // 最新优先:第一条是「没有附件」的那条 —— 它必须省略 documents。 if _, ok := envelope.Data[0]["documents"]; ok { t.Fatalf("没有附件的消息不该出现 documents 字段 (body=%s)", body) } var docs []history.DocumentRef if err := json.Unmarshal(envelope.Data[1]["documents"], &docs); err != nil { t.Fatalf("decode documents: %v (body=%s)", err, body) } if len(docs) != 2 { t.Fatalf("documents = %+v, want 2 条", docs) } if docs[0].Name != "aaa.txt" || docs[0].Status != history.DocumentStatusExtracted || docs[0].Chars != 7 { t.Fatalf("documents[0] = %+v", docs[0]) } if docs[1].Name != "bbb.pdf" || docs[1].Status != history.DocumentStatusUnreadable { t.Fatalf("documents[1] = %+v", docs[1]) } } // newGatewayHistoryStore 在隔离的内存 SQLite 上建一个真实 history 存储(测试用 // 临时库,绝不碰 data/cyrene.db)。 func newGatewayHistoryStore(t *testing.T) *history.Store { t.Helper() ctx := context.Background() conn, err := db.Open(ctx, ":memory:", testLogger()) if err != nil { t.Fatalf("open memory db: %v", err) } t.Cleanup(func() { _ = conn.Close() }) st, err := history.NewStore(ctx, conn, testLogger()) if err != nil { t.Fatalf("new history store: %v", err) } return st } // fixedRunner 返回一个总是产出同一结果的回合驱动器。 func fixedRunner(res *loop.Result) TurnRunner { return func(context.Context, string, []model.Message, loop.SessionEnv) (*loop.Result, error) { return res, nil } } // fakeDocumentWriter 在 fakeHistory 之上加按 id 更新 document 列的能力(回写 // 视图),记录每次成功调用,并可按需让写入失败。 type fakeDocumentWriter struct { *fakeHistory err error mu sync.Mutex updates []string } // UpdateMessageDocument 记录一次成功的回写(id 与列内容),err 非 nil 时失败。 func (f *fakeDocumentWriter) UpdateMessageDocument(_ context.Context, id, document string) error { f.mu.Lock() defer f.mu.Unlock() if f.err != nil { return f.err } f.updates = append(f.updates, id+"|"+document) return nil } // snapshot 返回已成功回写的记录副本。 func (f *fakeDocumentWriter) snapshot() []string { f.mu.Lock() defer f.mu.Unlock() return append([]string(nil), f.updates...) } // captureHandler 把日志记录收进内存,使「不静默」可以被断言:断言的是记录本身的 // message 与属性,而不是拼接后的文本(错误一律按类型/哨兵判定,绝不匹配文本)。 // // WithAttrs 必须把预置属性(尤其是 logger.WithSource 盖上的 source)带进记录, // 否则断言「带来源的 WARN」时就看不到 source —— 那正是这条纪律要盯的东西。 type captureHandler struct { state *captureState attrs []slog.Attr } type captureState struct { mu sync.Mutex recs []slog.Record } // newCaptureHandler 建一个空的记录处理器。 func newCaptureHandler() *captureHandler { return &captureHandler{state: &captureState{}} } func (h *captureHandler) Enabled(context.Context, slog.Level) bool { return true } func (h *captureHandler) Handle(_ context.Context, r slog.Record) error { if len(h.attrs) > 0 { r.AddAttrs(h.attrs...) } h.state.mu.Lock() defer h.state.mu.Unlock() h.state.recs = append(h.state.recs, r.Clone()) return nil } func (h *captureHandler) WithAttrs(as []slog.Attr) slog.Handler { if len(as) == 0 { return h } attrs := make([]slog.Attr, 0, len(h.attrs)+len(as)) attrs = append(attrs, h.attrs...) attrs = append(attrs, as...) return &captureHandler{state: h.state, attrs: attrs} } func (h *captureHandler) WithGroup(string) slog.Handler { return h } // has 报告是否存在一条 message 为 msg、reason 属性为 reason(空串表示不限定)且 // 带 source 属性的记录 —— 也就是说这条 WARN 是**有来源**的。 func (h *captureHandler) has(msg, reason string) bool { h.state.mu.Lock() defer h.state.mu.Unlock() for _, r := range h.state.recs { if r.Message != msg { continue } var gotReason, gotSource string r.Attrs(func(a slog.Attr) bool { switch a.Key { case "reason": gotReason = a.Value.String() case "source": gotSource = a.Value.String() } return true }) if gotSource == "" { continue } if reason == "" || gotReason == reason { return true } } return false }