diff --git a/internal/runtime/history/document.go b/internal/runtime/history/document.go index a05f6c4..390c2eb 100644 --- a/internal/runtime/history/document.go +++ b/internal/runtime/history/document.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "encoding/json" + "errors" "fmt" "strings" ) @@ -15,6 +16,10 @@ import ( // 字符数、是否截断、抽取状态；**绝不存抽取正文**。G3（base64 原样落库 → 每页 // 历史背着重传）与 H1（每回合重算）两条教训说的都是同一件事：正文一旦进了 // 历史，每读一页都要为它付一次代价，而且它是提示词注入面（design/11 §10）。 +// +// 列的内容是**引用数组**（≤maxDocumentsPerMessage 条），因为一个回合可以挂 +// 多个附件（design/24 §6.1 的定稿上限 5）；S2/S3 早期写下的**单对象**行仍能 +// 读回（见 DecodeDocuments）。列名与 Message.Document 字段名保持不变。 // 文档在持久化与历史 DTO 里可能处于的抽取状态（design/24 §6.1），由装配期 // 与抽取器填入。 @@ -27,8 +32,27 @@ const ( DocumentStatusUnreadable = "unreadable" ) +// maxDocumentsPerMessage 是一条消息（一个回合）携带的文档引用数上限 +// （design/24 §6.1/§9.4-3 的定稿值 5）。它与网关的 maxChatDocuments **同值 +// 同义**：这里刻意不复用那个常量 —— 网关依赖本包，反向依赖会成环。两个包 +// 各自具名并互相交叉引用，任一侧改动都必须同时改另一侧（网关侧另有 +// TestMaxChatDocumentsMatchesDesign 把值钉在设计定稿上）。 +const maxDocumentsPerMessage = 5 + +// 区分 document 列读写失败的哨兵错误。调用方用 errors.Is 判类，并按「不静默」 +// 处理它：打一条带来源的 WARN 并省掉该条引用，**绝不**把它当成「这条消息没有 +// 附件」—— 一方以为附件传上去了、另一方一无所知，正是 F2 要消灭的误解。 +var ( + // ErrBadDocuments 报告 document 列的内容不是合法的引用数据（乱码、非法 + // JSON、无名引用等）。 + ErrBadDocuments = errors.New("history: malformed document column") + + // ErrTooManyDocuments 报告一次读写里的引用数超过 maxDocumentsPerMessage。 + ErrTooManyDocuments = errors.New("history: too many document references") +) + // DocumentRef 是一条消息携带的文档引用与抽取统计。它就是 chat_messages.document -// 列里的 JSON 小对象，**同时**也是历史 DTO 里 document 对象的形状：两边共用 +// 列里的 JSON 小对象，**同时**也是历史 DTO 里 documents 数组的元素形状：两边共用 // 同一个结构体，所以「不回传正文」是结构上保证的，而不是靠约定或代码审查。 type DocumentRef struct { // Name 是服务端存储名（design/24 §3.2：32 位十六进制 + 白名单扩展名）。 @@ -46,31 +70,92 @@ type DocumentRef struct { Status string `json:"status,omitempty"` } -// EncodeDocument 把文档引用编码成落库用的 JSON 小对象。空引用返回 ""，使 -// 「这条消息没有附件」与「有附件」在库里、在 DTO 里都有确定区别。 -func EncodeDocument(ref DocumentRef) (string, error) { - if ref == (DocumentRef{}) { +// EncodeDocuments 把一条消息携带的文档引用编码成落库用的 JSON 数组。空列表返回 +// ""，使「这条消息没有附件」与「有附件」在库里、在 DTO 里都有确定区别。 +// +// 超限与无名引用都**报错**而不是就地裁剪：悄悄砍掉几条、或把一条没有名字的引用 +// （它的 URL 只能从名字推出）塞进数组，都会让「客户端传上去的附件」与「库里记着 +// 的附件」对不上 —— 那正是本列要消灭的静默。 +func EncodeDocuments(refs []DocumentRef) (string, error) { + if len(refs) == 0 { return "", nil } - b, err := json.Marshal(ref) + if len(refs) > maxDocumentsPerMessage { + return "", fmt.Errorf("history: encode %d documents: %w", len(refs), ErrTooManyDocuments) + } + for _, ref := range refs { + if ref.Name == "" { + return "", fmt.Errorf("history: encode document ref without name: %w", ErrBadDocuments) + } + } + b, err := json.Marshal(refs) if err != nil { - return "", fmt.Errorf("history: encode document ref: %w", err) + return "", fmt.Errorf("history: encode document refs: %w", err) } return string(b), nil } -// DecodeDocument 解码落库的文档引用。空串（这条消息没有附件）返回 (nil, nil)； -// 非法 JSON 返回错误，由调用方决定如何**可见地**报告 —— 解析失败绝不静默当成 -// 「没有附件」，那正是 F2 要消灭的那种「一方以为传上去了，另一方一无所知」。 -func DecodeDocument(raw string) (*DocumentRef, error) { - if raw == "" { +// DecodeDocuments 解码落库的文档引用。空串（这条消息没有附件）返回 (nil, nil)； +// 坏内容返回包装了 ErrBadDocuments / ErrTooManyDocuments 的错误，由调用方决定 +// 如何**可见地**报告 —— 解析失败绝不静默当成「没有附件」，那正是 F2 要消灭的 +// 那种「一方以为传上去了，另一方一无所知」。 +// +// 它兼容 S2/S3 早期写下的**单对象形状**：内容的首字符是 '{' 就按「单元素数组」 +// 处理。那些行里带着一条真实引用，绝不能因为形状老就被当成坏数据丢掉。 +func DecodeDocuments(raw string) ([]DocumentRef, error) { + trimmed := strings.TrimSpace(raw) + if trimmed == "" { return nil, nil } - var ref DocumentRef - if err := json.Unmarshal([]byte(raw), &ref); err != nil { - return nil, fmt.Errorf("history: decode document ref: %w", err) + refs, err := decodeDocumentData(trimmed) + if err != nil { + return nil, err } - return &ref, nil + if len(refs) > maxDocumentsPerMessage { + return nil, fmt.Errorf("history: decode %d documents: %w", len(refs), ErrTooManyDocuments) + } + return refs, nil +} + +// decodeDocumentData 按内容形状解码：'{' 开头是旧单对象形状（当成单元素数组）， +// 其余按数组解析。空数组与字面量 null 都解出空结果，等价于「这条消息没有附件」。 +func decodeDocumentData(raw string) ([]DocumentRef, error) { + if raw[0] == '{' { + var ref DocumentRef + if err := json.Unmarshal([]byte(raw), &ref); err != nil { + return nil, fmt.Errorf("history: decode document ref: %w", ErrBadDocuments) + } + return []DocumentRef{ref}, nil + } + var refs []DocumentRef + if err := json.Unmarshal([]byte(raw), &refs); err != nil { + return nil, fmt.Errorf("history: decode document refs: %w", ErrBadDocuments) + } + return refs, nil +} + +// UpdateMessageDocument 只替换 id 所指消息的 document 列（引用数组），其余任何 +// 列都保持已存值。它是「回合结束后补抽取结论」的写入端（design/24 §6.1）。 +// +// 为什么需要它：用户消息在回合**开始前**就已落库，那时装配期的抽取还没发生， +// 因此列里只有引用、没有 status/chars/truncated。结论只能在回合结束后补进同一 +// 列 —— 两段式是刻意的，也是客户端能看到「昔涟读不到这个文件」的唯一路径。 +// +// 未知 id 返回 ErrMessageNotFound（与 UpdateMessage 同一口径），使调用方能区分 +// 「那一行不见了」与「写失败」，并据此打带来源的 WARN，而不是静默当作写完了。 +func (s *Store) UpdateMessageDocument(ctx context.Context, id, document string) error { + res, err := s.db.ExecContext(ctx, `UPDATE chat_messages SET document = ? WHERE id = ?`, document, id) + if err != nil { + return fmt.Errorf("history: update message document %s: %w", id, err) + } + n, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("history: update message document %s rows: %w", id, err) + } + if n == 0 { + return fmt.Errorf("history: update message document %s: %w", id, ErrMessageNotFound) + } + return nil } // ensureMessageColumns 把 chat_messages 上「建表之后才引入的列」前滚到既有表。 diff --git a/internal/runtime/history/document_test.go b/internal/runtime/history/document_test.go index 79689bb..26cfa80 100644 --- a/internal/runtime/history/document_test.go +++ b/internal/runtime/history/document_test.go @@ -2,32 +2,37 @@ package history import ( "context" + "errors" "strings" "testing" "git.yeij.top/AskaEth/Cyrene_v2/internal/db" ) -// 本文件是文档引用列（chat_messages.document，design/24 §2.4）的存储测试： -// 引用与统计往返一致、库里**任何一列都不含抽取正文**、坏 JSON 不静默当成 -// 「没有附件」，以及早于 0015 建出的旧表能被幂等前滚。 +// 本文件是文档引用列（chat_messages.document，design/24 §2.4/§6.1）的存储测试： +// 引用与统计往返一致、旧单对象形状兼容、库里**任何一列都不含抽取正文**、坏内容 +// 不静默当成「没有附件」、上限报错、按 id 回写抽取结论，以及早于 0015 建出的 +// 旧表能被幂等前滚。 -// TestDocumentRefRoundTrip 验证「引用 + 统计」落库再读出后逐字段一致。该列是 -// 不透明 JSON：存储不解释、不裁剪它。 +// TestDocumentRefRoundTrip 验证「引用 + 统计」落库再读出后逐字段一致、顺序不变。 +// 该列是不透明 JSON：存储不解释、不裁剪它。 func TestDocumentRefRoundTrip(t *testing.T) { t.Parallel() st := newStore(t) ctx := context.Background() - ref := DocumentRef{ - Name: "0123456789abcdef0123456789abcdef.txt", - MIME: "text/plain", - Size: 12345, - Chars: 8000, - Truncated: true, - Status: DocumentStatusExtracted, + refs := []DocumentRef{ + { + Name: "0123456789abcdef0123456789abcdef.txt", + MIME: "text/plain", + Size: 12345, + Chars: 8000, + Truncated: true, + Status: DocumentStatusExtracted, + }, + {Name: "fedcba9876543210fedcba9876543210.md", MIME: "text/markdown", Size: 42, Status: DocumentStatusListed}, } - raw, err := EncodeDocument(ref) + raw, err := EncodeDocuments(refs) if err != nil { t.Fatalf("encode: %v", err) } @@ -42,44 +47,104 @@ func TestDocumentRefRoundTrip(t *testing.T) { if len(msgs) != 1 { t.Fatalf("len = %d, want 1", len(msgs)) } - got, err := DecodeDocument(msgs[0].Document) + got, err := DecodeDocuments(msgs[0].Document) if err != nil { t.Fatalf("decode: %v", err) } - if got == nil { - t.Fatal("document ref lost in round trip") + if len(got) != len(refs) { + t.Fatalf("refs = %+v, want %d 条（顺序与写入一致）", got, len(refs)) } - if *got != ref { - t.Fatalf("ref = %+v, want %+v", *got, ref) + for i := range refs { + if got[i] != refs[i] { + t.Fatalf("refs[%d] = %+v, want %+v", i, got[i], refs[i]) + } + } +} + +// TestDecodeDocumentsAcceptsLegacySingleObject 钉住旧行兼容：S2/S3 早期把 +// document 列写成**单个**对象，那些行里带着一条真实引用，绝不能因为形状老就被 +// 当成坏数据丢掉（这正是缺口①的另一半）。 +func TestDecodeDocumentsAcceptsLegacySingleObject(t *testing.T) { + t.Parallel() + const legacy = `{"name":"aaaabbbbccccdddd.txt","mime":"text/plain","size":1234,"status":"extracted"}` + refs, err := DecodeDocuments(legacy) + if err != nil { + t.Fatalf("decode legacy: %v", err) + } + if len(refs) != 1 { + t.Fatalf("legacy refs = %+v, want 长度 1 的数组", refs) + } + if refs[0].Name != "aaaabbbbccccdddd.txt" || refs[0].Status != DocumentStatusExtracted || refs[0].Size != 1234 { + t.Fatalf("legacy refs[0] = %+v, want 旧行里的那条引用", refs[0]) } } // TestDocumentRefEmptyMeansNoAttachment 验证「没有附件」与「有附件」在库里、 -// 在 DTO 里都有确定区别：空引用编码成空串，空串解码回 nil。 +// 在 DTO 里都有确定区别：空列表编码成空串，空串解码回 nil。 func TestDocumentRefEmptyMeansNoAttachment(t *testing.T) { t.Parallel() - raw, err := EncodeDocument(DocumentRef{}) + raw, err := EncodeDocuments(nil) if err != nil { - t.Fatalf("encode zero ref: %v", err) + t.Fatalf("encode zero refs: %v", err) } if raw != "" { t.Fatalf("raw = %q, want empty column", raw) } - ref, err := DecodeDocument("") + refs, err := DecodeDocuments("") if err != nil { t.Fatalf("decode empty: %v", err) } - if ref != nil { - t.Fatalf("ref = %+v, want nil", ref) + if refs != nil { + t.Fatalf("refs = %+v, want nil", refs) } } -// TestDecodeDocumentRejectsMalformed 验证坏 JSON 返回错误而不是 (nil, nil)： -// 解析失败绝不能静默变成「这条消息没有附件」，那正是 F2 要消灭的误解。 -func TestDecodeDocumentRejectsMalformed(t *testing.T) { +// TestDecodeDocumentsRejectsMalformed 验证坏内容返回包装了 ErrBadDocuments 的错误 +// 而不是 (nil, nil)：解析失败绝不能静默变成「这条消息没有附件」，那正是 F2 要 +// 消灭的误解（调用方按 errors.Is 判类后打带来源的 WARN）。 +func TestDecodeDocumentsRejectsMalformed(t *testing.T) { t.Parallel() - if _, err := DecodeDocument("{not json"); err == nil { - t.Fatal("DecodeDocument(malformed) = nil, want an error") + for _, raw := range []string{"{not json", "not json at all", `{"name":42}`} { + if _, err := DecodeDocuments(raw); !errors.Is(err, ErrBadDocuments) { + t.Fatalf("DecodeDocuments(%q) = %v, want ErrBadDocuments", raw, err) + } + } +} + +// TestEncodeDocumentsRejectsUnnamedRef 验证无名引用报错而不是写进库：名字是 URL +// 与展示身份的唯一来源，存一条没有名字的引用等于存一条读不出来的附件。 +func TestEncodeDocumentsRejectsUnnamedRef(t *testing.T) { + t.Parallel() + if _, err := EncodeDocuments([]DocumentRef{{MIME: "text/plain", Size: 12}}); !errors.Is(err, ErrBadDocuments) { + t.Fatalf("EncodeDocuments(unnamed) = %v, want ErrBadDocuments", err) + } +} + +// TestDocumentCountCapIsFive 把上限的**值**单独钉在设计定稿上 +// （design/24 §9.4-3、§6.1）。为什么必须单独钉一次：下面的超限用例只证明「逻辑 +// 自洽」，上限被悄悄改宽时它们照样绿（S3-B 的测试质量教训）。 +func TestDocumentCountCapIsFive(t *testing.T) { + t.Parallel() + if maxDocumentsPerMessage != 5 { + t.Fatalf("单条消息文档引用上限 = %d, want 5（design/24 §6.1 的定稿值）", maxDocumentsPerMessage) + } +} + +// TestDocumentCountCapRejectsSix 覆盖防御性上限：超过 5 条引用时读写两端都**报错** +// 而不是悄悄裁掉几条 —— 悄悄裁剪就是静默丢失，而 6 条只可能来自坏数据或外部改动。 +// 六条是字面量而不是由常量算出，常量本身另由 TestDocumentCountCapIsFive 钉住。 +func TestDocumentCountCapRejectsSix(t *testing.T) { + t.Parallel() + six := []DocumentRef{ + {Name: "a.txt"}, {Name: "b.txt"}, {Name: "c.txt"}, + {Name: "d.txt"}, {Name: "e.txt"}, {Name: "f.txt"}, + } + if _, err := EncodeDocuments(six); !errors.Is(err, ErrTooManyDocuments) { + t.Fatalf("EncodeDocuments(6) = %v, want ErrTooManyDocuments", err) + } + const sixJSON = `[{"name":"a.txt"},{"name":"b.txt"},{"name":"c.txt"},{"name":"d.txt"},{"name":"e.txt"},{"name":"f.txt"}]` + if _, err := DecodeDocuments(sixJSON); !errors.Is(err, ErrTooManyDocuments) { + t.Fatalf("DecodeDocuments(6) = %v, want ErrTooManyDocuments", err) } } @@ -93,13 +158,13 @@ func TestPersistedDocumentHoldsNoBody(t *testing.T) { ctx := context.Background() const body = "抽取正文不可落库·SECRET-BODY-0123456789" - raw, err := EncodeDocument(DocumentRef{ + raw, err := EncodeDocuments([]DocumentRef{{ Name: "aaaabbbbccccdddd.txt", MIME: "text/plain", Size: int64(len(body)), Chars: len(body), Status: DocumentStatusExtracted, - }) + }}) if err != nil { t.Fatalf("encode: %v", err) } @@ -125,6 +190,65 @@ func TestPersistedDocumentHoldsNoBody(t *testing.T) { } } +// TestUpdateMessageDocumentPersistsConclusion 覆盖状态回写的写入端 +// （design/24 §6.1）：回合结束后按消息 id 把抽取结论补进 document 列，且**只** +// 改这一列。 +func TestUpdateMessageDocumentPersistsConclusion(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + before, err := EncodeDocuments([]DocumentRef{{Name: "aaaabbbbccccdddd.txt", MIME: "text/plain", Size: 1234}}) + if err != nil { + t.Fatalf("encode: %v", err) + } + full, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "读一下", Document: before}) + if err != nil { + t.Fatalf("append: %v", err) + } + + after, err := EncodeDocuments([]DocumentRef{{ + Name: "aaaabbbbccccdddd.txt", MIME: "text/plain", Size: 1234, + Chars: 900, Truncated: true, Status: DocumentStatusExtracted, + }}) + if err != nil { + t.Fatalf("encode: %v", err) + } + if err := st.UpdateMessageDocument(ctx, full.ID, after); err != nil { + t.Fatalf("UpdateMessageDocument: %v", err) + } + + msgs, err := st.List(ctx, MainSessionID, 0, 10, "") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 1 { + t.Fatalf("len = %d, want 1", len(msgs)) + } + refs, err := DecodeDocuments(msgs[0].Document) + if err != nil { + t.Fatalf("decode: %v", err) + } + if len(refs) != 1 || refs[0].Status != DocumentStatusExtracted || refs[0].Chars != 900 || !refs[0].Truncated { + t.Fatalf("refs = %+v, want extracted/chars=900/truncated", refs) + } + // 其余列保持已存值：回写只碰 document 一列。 + if msgs[0].Content != "读一下" || msgs[0].Role != RoleUser || msgs[0].ID != full.ID || msgs[0].CreatedAt != full.CreatedAt { + t.Fatalf("msg = %+v, want 除 document 外与落库时一致", msgs[0]) + } +} + +// TestUpdateMessageDocumentUnknownID 验证未知 id 返回 ErrMessageNotFound（与 +// UpdateMessage 同一口径），使调用方能区分「那一行不见了」与「写失败」。 +func TestUpdateMessageDocumentUnknownID(t *testing.T) { + t.Parallel() + st := newStore(t) + err := st.UpdateMessageDocument(context.Background(), "no-such-message", "[]") + if !errors.Is(err, ErrMessageNotFound) { + t.Fatalf("UpdateMessageDocument(unknown) = %v, want ErrMessageNotFound", err) + } +} + // TestNewStoreRollsForwardDocumentColumn 验证早于 0015 建出的 chat_messages 表 // 会被 NewStore 幂等前滚出 document 列（与 client_msg_id 同一手法）：全新库由 // 建表 DDL 一次建全，既有库靠补列，两条路都必须能写能读带引用的消息。 @@ -163,7 +287,7 @@ CREATE TABLE chat_messages ( if err != nil { t.Fatalf("new store: %v", err) } - raw, err := EncodeDocument(DocumentRef{Name: "aaaabbbbccccdddd.txt", MIME: "text/plain", Status: DocumentStatusListed}) + raw, err := EncodeDocuments([]DocumentRef{{Name: "aaaabbbbccccdddd.txt", MIME: "text/plain", Status: DocumentStatusListed}}) if err != nil { t.Fatalf("encode: %v", err) } diff --git a/internal/runtime/history/types.go b/internal/runtime/history/types.go index ed576d6..fc1d00d 100644 --- a/internal/runtime/history/types.go +++ b/internal/runtime/history/types.go @@ -16,11 +16,16 @@ type Message struct { Mode string Mood string Image string - // Document 是这条消息携带的文档引用与抽取统计（design/24 §2.4），即 - // DocumentRef 序列化出的小对象 {name,mime,size,chars,truncated,status}。 - // 它刻意只装引用与统计 —— **绝不装抽取正文**：正文一旦落库，每翻一页 - // 历史都要背着它重传一次（G3 的教训），而且它是提示词注入面 - // （design/11 §10）。空串表示这条消息没有附件。 + // Document 是这条消息携带的文档引用与抽取统计（design/24 §2.4、§6.1），即 + // DocumentRef 的 JSON **数组**（一个回合最多 5 个附件，见 document.go 的 + // maxDocumentsPerMessage；S2/S3 早期的单对象行仍能读回）。它刻意只装引用与 + // 统计 —— **绝不装抽取正文**：正文一旦落库，每翻一页历史都要背着它重传一次 + // （G3 的教训），而且它是提示词注入面（design/11 §10）。空串表示这条消息没有 + // 附件。 + // + // 列内容有两个写入时点：回合开始前只写引用，回合结束后由 + // (*Store).UpdateMessageDocument 把抽取结论（status/chars/truncated）补进同 + // 一列 —— 抽取发生在回合的装配期，那时消息早已落库。 Document string Thinking string // Platform 标注一条消息的来源平台（design/16 §4.4）。Web 主会话是跨平台 diff --git a/internal/services/gateway/chat_documents_test.go b/internal/services/gateway/chat_documents_test.go index 6147bd4..f8842b7 100644 --- a/internal/services/gateway/chat_documents_test.go +++ b/internal/services/gateway/chat_documents_test.go @@ -436,9 +436,9 @@ func TestHubStreamChatResolvesDocuments(t *testing.T) { func TestTranscriptMessageKeepsDocumentReference(t *testing.T) { t.Parallel() const name = "0123456789abcdef0123456789abcdef.txt" - raw, err := history.EncodeDocument(history.DocumentRef{ + raw, err := history.EncodeDocuments([]history.DocumentRef{{ Name: name, MIME: documentSniffOKMIME, Size: 12345, Status: history.DocumentStatusExtracted, - }) + }}) if err != nil { t.Fatalf("encode document: %v", err) } diff --git a/internal/services/gateway/history.go b/internal/services/gateway/history.go index 8704be0..cbbdcae 100644 --- a/internal/services/gateway/history.go +++ b/internal/services/gateway/history.go @@ -61,14 +61,19 @@ type historyMessage struct { Mode string `json:"mode,omitempty"` Mood string `json:"mood,omitempty"` Image string `json:"image,omitempty"` - // Document 是这条消息携带的文档引用与抽取统计（design/24 §2.4、§6.1）， - // 形状为 {name,mime,size,chars,truncated,status}。它**只带引用与统计， - // 绝不带抽取正文** —— 正文若从这里回传，每翻一页历史都要背着它（G3 的 - // 教训）。没有附件的消息省略该字段。 - Document *history.DocumentRef `json:"document,omitempty"` - Thinking string `json:"thinking,omitempty"` - Platform string `json:"platform,omitempty"` - DeviceID string `json:"deviceId,omitempty"` + // Documents 是这条消息携带的文档引用与抽取统计（design/24 §2.4、§6.1）， + // 形状为 [{name,mime,size,chars,truncated,status}, ...]：一个回合最多 5 条， + // 顺序即客户端发送顺序。它们**只带引用与统计，绝不带抽取正文** —— 正文若 + // 从这里回传，每翻一页历史都要背着它（G3 的教训）。没有附件（或附件列表为 + // 空）的消息**省略该字段**，与引入文档之前逐字相同。 + // + // ⚠️ 契约变更（S3.5）：此前这里是单对象 `document`。单数形状从未进 + // Docs/api-client.md（v0.59 待发、客户端未接），因此不留兼容期，直接换成 + // 数组 —— 客户端按 `documents`（可为空/缺失）读取。 + Documents []history.DocumentRef `json:"documents,omitempty"` + Thinking string `json:"thinking,omitempty"` + Platform string `json:"platform,omitempty"` + DeviceID string `json:"deviceId,omitempty"` // SenderName 是第三方消息（群聊/陌生人，design/16 §4.6） // 给人看的发送者显示名。对主会话中主人的私聊它为 // 空，因此空意味着「标记为主人」。 @@ -407,7 +412,7 @@ func toHistoryMessage(m history.Message) historyMessage { Mode: m.Mode, Mood: m.Mood, Image: m.Image, - Document: decodeHistoryDocument(m.Document), + Documents: decodeHistoryDocuments(m.Document), Thinking: m.Thinking, Platform: m.Platform, DeviceID: m.DeviceID, @@ -425,15 +430,16 @@ func toHistoryMessage(m history.Message) historyMessage { // historyDefaultLimit 是客户端省略 limit 查询参数时使用的页大小。 const historyDefaultLimit = 50 -// decodeHistoryDocument 解码已存消息的 document 列（JSON 小对象）。空列表示 -// 这条消息没有附件，返回 nil 以省略该字段。列被写坏时（本服务端只会写合法 -// JSON，因此那只能是外部改动或未来格式变更）**绝不静默当成没有附件** —— -// 「一方以为附件传上去了，另一方一无所知」正是 F2 要消灭的误解；这时回一个 -// 标着 status=unreadable 的最小对象，让客户端看得见这条引用读不出来。 -func decodeHistoryDocument(raw string) *history.DocumentRef { - ref, err := history.DecodeDocument(raw) +// decodeHistoryDocuments 解码已存消息的 document 列（引用数组，兼容 S2/S3 早期 +// 的单对象旧形状）。空列表示这条消息没有附件，返回 nil 以省略该字段。列被写坏时 +// （本服务端只会写合法 JSON，因此那只能是外部改动或未来格式变更）**绝不静默 +// 当成没有附件** —— 「一方以为附件传上去了，另一方一无所知」正是 F2 要消灭的 +// 误解；这时回一个标着 status=unreadable 的最小引用，让客户端看得见这条引用 +// 读不出来。 +func decodeHistoryDocuments(raw string) []history.DocumentRef { + refs, err := history.DecodeDocuments(raw) if err != nil { - return &history.DocumentRef{Status: history.DocumentStatusUnreadable} + return []history.DocumentRef{{Status: history.DocumentStatusUnreadable}} } - return ref + return refs } diff --git a/internal/services/gateway/history_document_test.go b/internal/services/gateway/history_document_test.go index fd9b5aa..f8ce6c7 100644 --- a/internal/services/gateway/history_document_test.go +++ b/internal/services/gateway/history_document_test.go @@ -14,19 +14,19 @@ import ( // 引用与统计往返、**不回传抽取正文**、坏列可见，以及 session_persist 的 // messageTextImageDocument 取值。 -// TestHistoryMessageDocumentCarriesReferenceOnly 验证 document 对象只带 +// TestHistoryMessageDocumentsCarryReferenceOnly 验证 documents 数组的元素只带 // 引用与统计：字段集被逐个钉住，将来谁往 DocumentRef 里加一个正文/摘要字段， // 这里就会红。 -func TestHistoryMessageDocumentCarriesReferenceOnly(t *testing.T) { +func TestHistoryMessageDocumentsCarryReferenceOnly(t *testing.T) { t.Parallel() - raw, err := history.EncodeDocument(history.DocumentRef{ + raw, err := history.EncodeDocuments([]history.DocumentRef{{ Name: "0123456789abcdef0123456789abcdef.txt", MIME: "text/plain", Size: 12345, Chars: 8000, Truncated: true, Status: history.DocumentStatusExtracted, - }) + }}) if err != nil { t.Fatalf("encode: %v", err) } @@ -40,9 +40,13 @@ func TestHistoryMessageDocumentCarriesReferenceOnly(t *testing.T) { if err := json.Unmarshal(b, &decoded); err != nil { t.Fatalf("unmarshal: %v", err) } - doc, ok := decoded["document"].(map[string]any) + docs, ok := decoded["documents"].([]any) + if !ok || len(docs) != 1 { + t.Fatalf("documents array missing in %s", b) + } + doc, ok := docs[0].(map[string]any) if !ok { - t.Fatalf("document object missing in %s", b) + t.Fatalf("documents[0] is not an object in %s", b) } keys := make([]string, 0, len(doc)) for k := range doc { @@ -50,38 +54,61 @@ func TestHistoryMessageDocumentCarriesReferenceOnly(t *testing.T) { } sort.Strings(keys) if got := strings.Join(keys, ","); got != "chars,mime,name,size,status,truncated" { - t.Fatalf("document keys = %q, want chars,mime,name,size,status,truncated", got) + t.Fatalf("documents[0] keys = %q, want chars,mime,name,size,status,truncated", got) } if doc["name"] != "0123456789abcdef0123456789abcdef.txt" || doc["status"] != history.DocumentStatusExtracted { - t.Fatalf("document = %v, want the persisted reference", doc) + t.Fatalf("documents[0] = %v, want the persisted reference", doc) } } -// TestHistoryMessageDocumentAbsentWhenNoColumn 验证没有附件的消息省略该字段： +// TestHistoryMessageDocumentsAbsentWhenNoColumn 验证没有附件的消息省略该字段： // 空列与「带附件」必须可区分（省略 = 没有附件）。 -func TestHistoryMessageDocumentAbsentWhenNoColumn(t *testing.T) { +func TestHistoryMessageDocumentsAbsentWhenNoColumn(t *testing.T) { t.Parallel() wire := toHistoryMessage(history.Message{ID: "m1", Role: "user", Content: "没有附件"}) b, err := json.Marshal(wire) if err != nil { t.Fatalf("marshal: %v", err) } - if strings.Contains(string(b), `"document"`) { - t.Fatalf("empty column should omit document, got %s", b) + if strings.Contains(string(b), `"documents"`) { + t.Fatalf("empty column should omit documents, got %s", b) } } -// TestHistoryMessageDocumentMalformedColumnIsVisible 验证列里是坏 JSON 时 -// 不静默当成「没有附件」：回一个标着 status=unreadable 的最小对象，让客户端 +// TestHistoryMessageDocumentsMultipleKeepOrder 验证一回合多个附件原样回传 +// （缺口①）：数组长度与顺序都一致，客户端因此不会只看到第一个附件。 +func TestHistoryMessageDocumentsMultipleKeepOrder(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, Status: history.DocumentStatusUnreadable}, + {Name: "ccc.csv", MIME: "text/csv", Size: 3, Chars: 9, Status: history.DocumentStatusExtracted}, + }) + if err != nil { + t.Fatalf("encode: %v", err) + } + wire := toHistoryMessage(history.Message{ID: "m1", Role: "user", Content: "三个附件", Document: raw}) + if len(wire.Documents) != 3 { + t.Fatalf("Documents = %+v, want 3 条", wire.Documents) + } + for i, want := range []string{"aaa.txt", "bbb.pdf", "ccc.csv"} { + if wire.Documents[i].Name != want { + t.Fatalf("Documents[%d].Name = %q, want %q（顺序必须与发送一致）", i, wire.Documents[i].Name, want) + } + } +} + +// TestHistoryMessageDocumentsMalformedColumnIsVisible 验证列里是坏内容时 +// 不静默当成「没有附件」：回一条标着 status=unreadable 的最小引用，让客户端 // 看得见这条引用读不出来（design/24 §4.4 的「失败必须可见」）。 -func TestHistoryMessageDocumentMalformedColumnIsVisible(t *testing.T) { +func TestHistoryMessageDocumentsMalformedColumnIsVisible(t *testing.T) { t.Parallel() wire := toHistoryMessage(history.Message{ID: "m1", Role: "user", Content: "带附件", Document: "{not json"}) - if wire.Document == nil { - t.Fatal("malformed document column was silently dropped") + if len(wire.Documents) != 1 { + t.Fatalf("Documents = %+v, want 一条 unreadable 的最小引用", wire.Documents) } - if wire.Document.Status != history.DocumentStatusUnreadable { - t.Fatalf("status = %q, want %q", wire.Document.Status, history.DocumentStatusUnreadable) + if wire.Documents[0].Status != history.DocumentStatusUnreadable { + t.Fatalf("status = %q, want %q", wire.Documents[0].Status, history.DocumentStatusUnreadable) } } @@ -103,17 +130,17 @@ func TestMessageTextImageDocument(t *testing.T) { if image != "data:image/png;base64,abc" { t.Fatalf("image = %q", image) } - ref, err := history.DecodeDocument(raw) + refs, err := history.DecodeDocuments(raw) if err != nil { - t.Fatalf("decode document: %v", err) + t.Fatalf("decode documents: %v", err) } - if ref == nil || ref.Name != doc.Name || ref.MIME != doc.MIME || ref.Size != doc.Size { - t.Fatalf("ref = %+v, want name/mime/size of %+v", ref, doc) + if len(refs) != 1 || refs[0].Name != doc.Name || refs[0].MIME != doc.MIME || refs[0].Size != doc.Size { + t.Fatalf("refs = %+v, want name/mime/size of %+v", refs, doc) } } // TestMessageTextImageDocumentWithoutDocument 验证没有文档部件时文档列为空 -// （而不是 "{}"）：拼写空对象会让读取侧把它当成一条读不出来的附件。 +// （而不是 "[]"/"{}"）：拼一个空形状会让读取侧把它当成一条读不出来的附件。 func TestMessageTextImageDocumentWithoutDocument(t *testing.T) { t.Parallel() _, _, raw, err := messageTextImageDocument(model.TextMessage(model.RoleUser, "只有文本")) diff --git a/internal/services/gateway/session_document_e2e_test.go b/internal/services/gateway/session_document_e2e_test.go new file mode 100644 index 0000000..1626eff --- /dev/null +++ b/internal/services/gateway/session_document_e2e_test.go @@ -0,0 +1,157 @@ +package gateway + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "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" +) + +// 本文件是 S3.5 在 **HTTP 层**的端到端回归（design/24 §6.1、§9.7）：一次 `chat` +// 带上 3 个附件 → 全部落库 → 回合结束后抽取结论回写 → 历史接口按 `documents` +// 数组把引用与结论一起交回客户端。它补齐单测覆盖不到的那一段接线：REST 入口 → +// 编排器 → 真实 history 存储 → 历史读端点。 + +// docStatusRunner 是端到端的假回合驱动器：它把这一回合用户消息里的**每一个** +// 文档部件都当成「已抽取」的结论回给网关（正是真实抽取层 reshape 后的形状）， +// 并记录它实际看到的文档部件数。 +type docStatusRunner struct { + mu sync.Mutex + docs []model.Content + calls int +} + +func (r *docStatusRunner) Run(_ context.Context, _ string, msgs []model.Message) (*loop.Result, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.calls++ + refs := make([]history.DocumentRef, 0, len(r.docs)) + for _, m := range msgs { + for _, c := range m.Content { + if c.Type != model.ContentTypeDocument { + continue + } + r.docs = append(r.docs, c) + refs = append(refs, history.DocumentRef{ + Name: c.DocumentName, MIME: c.DocumentMIME, Size: c.DocumentSize, + Chars: 12, Status: history.DocumentStatusExtracted, + }) + } + } + return &loop.Result{Text: "读完了", Documents: refs, Steps: 1, StopReason: "stop"}, nil +} + +// documentCount 返回该回合被驱动时看到的文档部件数。 +func (r *docStatusRunner) documentCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.docs) +} + +// TestChatDocumentsEndToEndCarriesEveryAttachmentAndConclusion 是 S3.5 两个缺口的 +// 端到端回归：3 个附件全部进入模型视图、全部落库，且落库的每一行都带回 +// status/chars/truncated（此前只有第一个附件落库，且结论永远到不了客户端）。 +func TestChatDocumentsEndToEndCarriesEveryAttachmentAndConclusion(t *testing.T) { + t.Parallel() + dataDir := t.TempDir() + jwt := auth.New(config.AuthConfig{JWTSecret: testSecret}) + cfg := &config.Config{Auth: config.AuthConfig{AdminUser: "admin", AdminPass: "secret", JWTSecret: testSecret}} + runner := &docStatusRunner{} + st := newGatewayHistoryStore(t) + srv, err := New(cfg, Deps{Auth: jwt, DataDir: dataDir, Loop: runner, History: st}) + if err != nil { + t.Fatalf("构造网关失败: %v", err) + } + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(ts.Close) + t.Cleanup(srv.Close) + tok, err := jwt.Sign("acct", auth.RoleOwner, time.Hour) + if err != nil { + t.Fatalf("签发 token 失败: %v", err) + } + + names := []string{ + storeGatewayDocument(t, dataDir, "第一份", ".txt"), + storeGatewayDocument(t, dataDir, "第二份", ".md"), + storeGatewayDocument(t, dataDir, "第三份", ".csv"), + } + refs := make([]chatDocumentRef, 0, len(names)) + for _, name := range names { + refs = append(refs, chatDocumentRef{Name: name}) + } + resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/chat", map[string]any{ + "sessionId": history.MainSessionID, "text": "这三份都看看", "documents": refs, + }, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("chat status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + if got := runner.documentCount(); got != 3 { + t.Fatalf("进入模型可见视图的文档部件 = %d 个, want 3（缺口①：附件不能丢）", got) + } + + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("messages status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var envelope struct { + Data []struct { + ID string `json:"id"` + Role string `json:"role"` + Content string `json:"content"` + Documents []history.DocumentRef `json:"documents"` + } `json:"data"` + } + if err := json.Unmarshal(body, &envelope); err != nil { + t.Fatalf("unmarshal: %v (body=%s)", err, body) + } + var userDocs []history.DocumentRef + for _, m := range envelope.Data { + if m.Role == history.RoleUser && m.Content == "这三份都看看" { + userDocs = m.Documents + } + } + if len(userDocs) != 3 { + t.Fatalf("历史接口返回的 documents = %+v, want 3 条（缺口①）", userDocs) + } + for i, name := range names { + if userDocs[i].Name != name { + t.Fatalf("documents[%d].Name = %q, want %q（顺序必须与发送一致）", i, userDocs[i].Name, name) + } + if userDocs[i].Status != history.DocumentStatusExtracted || userDocs[i].Chars != 12 { + t.Fatalf("documents[%d] = %+v, want extracted/chars=12（缺口②）", i, userDocs[i]) + } + } +} + +// storeGatewayDocument 按 S1 的内容寻址规则（sha256 前 32 位 + 扩展名）在 +// `/uploads/documents` 下落一份文档并返回其存储名（与 chat_documents_test +// 的 helper 同规则；这里在 HTTP 端到端用例里独立使用，避免依赖那边的 env）。 +func storeGatewayDocument(t *testing.T, dataDir, content, ext string) string { + t.Helper() + sum := sha256.Sum256([]byte(content)) + name := hex.EncodeToString(sum[:])[:documentHashHexLen] + ext + if !validDocumentName(name) { + t.Fatalf("构造的存储名 %q 不是合法形态", name) + } + dir := filepath.Join(dataDir, "uploads", documentSubDir) + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatalf("创建文档目录失败: %v", err) + } + if err := os.WriteFile(filepath.Join(dir, name), []byte(content), 0o600); err != nil { + t.Fatalf("写入文档失败: %v", err) + } + return name +} diff --git a/internal/services/gateway/session_document_status.go b/internal/services/gateway/session_document_status.go new file mode 100644 index 0000000..d91927b --- /dev/null +++ b/internal/services/gateway/session_document_status.go @@ -0,0 +1,190 @@ +package gateway + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" +) + +// 本文件归属**文档抽取结论的回写闭环**（design/24 §4.4/§6.1，S3.5 缺口②）。 +// +// 用户消息在回合**开始前**落库（session_persist.go 的 persistUserMessages）， +// 那时文档抽取还没发生，document 列里只有引用、没有结论。抽取结论要等回合跑 +// 起来（loop 的装配期解析）才存在，于是本文件负责在**回合结束后**把它补回同一 +// 列 —— 两段式是刻意的，也是客户端能看到「昔涟读不到这个文件」的唯一路径。 +// +// 三条纪律： +// - 只补结论（status/chars/truncated），身份与大小逐字保留； +// - 按**服务端存储名**匹配，匹配不到就原样留着（凭空给它一个结论就是编造）； +// - 找不到行、存储未实现该视图、写库失败一律带来源 WARN，绝不静默。 +// +// 它从 session_persist.go 拆出来，使那两个文件都留在「单文件 ≤500 行」的硬约束 +// 内（与 session_events.go / session_tail.go 同一理由）。 + +// documentStatusLogSource 是文档状态回写的日志来源标签，使这条冷路径上的 +// WARN 能被追溯到具体子功能（AGENTS §10 的「日志必须标来源」）。 +const documentStatusLogSource = "gateway/session-persist" + +// turnDocuments 配对一次回合同它已落库的用户消息与所属代次（gen）。代次是 +// 必须的：被打断或被新回合取代的陈旧回合在结算时不该去写别人的那一行。 +type turnDocuments struct { + gen uint64 + msgs []history.Message +} + +// documentStatusWriter 是 HistoryStore 的一个**可选**更窄视图（与 +// clientMsgIDReader / transcriptReader 同一做法）：按消息 id 更新 document 列。 +// 未实现该视图的存储替身或后端退回「不回写」并留下带来源的 WARN，不因此让 +// 回合失败 —— 状态回传是增强，绝不能反过来卡住会话。 +type documentStatusWriter interface { + UpdateMessageDocument(ctx context.Context, id, document string) error +} + +// rememberTurnDocuments 记下本回合已落库的用户消息（按代次配对），供回合结束 +// 后把抽取结论回写进它们的 document 列。空列表是空操作（这条回合没有落库任何 +// 用户消息，也就没有结论可补）。 +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} + } +} + +// attachPendingDocuments 把刚落库的用户消息挂到队列里 seq 所指的那条待处理回合 +// 上，使它的续跑回合结算时能回写结论（它在入队那一刻就落了库，而那时还没有 +// 任何结论）。 +// +// 队列里已经找不到那条（被中断清空、已被弹出或已被续跑消费）时什么都不做： +// 那一行的结论没有谁能再补，也没有必要为此刷一条 WARN —— 它不是缺陷，是中断 +// 的正常后果。 +func (o *Orchestrator) attachPendingDocuments(sessionID string, seq uint64, msgs []history.Message) { + if len(msgs) == 0 { + return + } + o.mu.Lock() + defer o.mu.Unlock() + st := o.states[sessionID] + if st == nil { + return + } + for i := range st.pending { + if st.pending[i].seq == seq { + st.pending[i].persisted = msgs + return + } + } +} + +// takeTurnDocuments 取走与 gen 配对的已落库用户消息并清空记录。第二个返回值 +// 报告这次代次是否仍是当代：非当代说明这是一次陈旧回合的迟到结算（中断/取代）， +// 调用方据此静默返回 —— 迟到的结算是正常竞态，不是缺陷。 +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 { + return nil, false + } + msgs := st.turnDocs.msgs + st.turnDocs = turnDocuments{} + return msgs, true +} + +// persistDocumentStatus 在回合结束后把装配期解析到的文档结论回写进该回合用户 +// 消息的 document 列（design/24 §4.4/§6.1）。它由 completeTurn 在助手回复落库 +// 之后、状态机迁移之前调用，因此结论与该回合的对话内容一起同批入账。 +// +// 找不到对应行（ErrMessageNotFound）、存储未接该视图、列被写坏、写库失败时都 +// **不静默**：打一条带来源的 WARN 说明「状态未回写」。日志只带会话、消息 id、 +// 条数与错误，**绝不带正文**，也不带列内容。 +// +// 写入使用全新的后台 context（与 persistAssistantReply 同一理由）：被取消回合 +// 的 ctx 绝不该中断这次补写。 +func (o *Orchestrator) persistDocumentStatus(sessionID string, gen uint64, res *loop.Result) { + if res == nil || len(res.Documents) == 0 { + return + } + log := logger.WithSource(o.log, documentStatusLogSource) + if o.history == nil { + log.Warn("document status not persisted", "reason", "no history store", "session", sessionID, "documents", len(res.Documents)) + return + } + writer, ok := o.history.(documentStatusWriter) + if !ok { + log.Warn("document status not persisted", "reason", "history store has no document update path", "session", sessionID, "documents", len(res.Documents)) + return + } + msgs, current := o.takeTurnDocuments(sessionID, gen) + if !current { + return + } + if len(msgs) == 0 { + log.Warn("document status not persisted", "reason", "no persisted user message for this turn", "session", sessionID, "documents", len(res.Documents)) + return + } + ctx := context.Background() + for _, m := range msgs { + if m.Document == "" { + continue + } + merged, changed, err := mergeDocumentStatus(m.Document, res.Documents) + if err != nil { + log.Warn("document status merge", "err", err, "session", sessionID, "message", m.ID) + continue + } + if !changed { + continue + } + if err := writer.UpdateMessageDocument(ctx, m.ID, merged); err != nil { + log.Warn("document status update", "err", err, "session", sessionID, "message", m.ID) + } + } +} + +// mergeDocumentStatus 把本回合解析到的结论按存储名并进已落库的引用数组：匹配到 +// 的补上 status/chars/truncated，没匹配到的原样保留（那意味着本回合没有解析到 +// 它，凭空给它一个结论就是编造）。第二个返回值报告是否真的补进了新结论 —— +// 没变化就不必写库、也不必刷一条「什么都没变」的日志。 +// +// 列被写坏时返回包装了 history 哨兵的错误，由调用方按「不静默」处理。 +func mergeDocumentStatus(raw string, resolved []history.DocumentRef) (string, bool, error) { + refs, err := history.DecodeDocuments(raw) + if err != nil { + return "", false, err + } + conclusion := make(map[string]history.DocumentRef, len(resolved)) + for _, ref := range resolved { + if ref.Name != "" { + conclusion[ref.Name] = ref + } + } + changed := false + for i := range refs { + got, ok := conclusion[refs[i].Name] + if !ok { + continue + } + if refs[i].Status == got.Status && refs[i].Chars == got.Chars && refs[i].Truncated == got.Truncated { + continue + } + refs[i].Status, refs[i].Chars, refs[i].Truncated = got.Status, got.Chars, got.Truncated + changed = true + } + if !changed { + return raw, false, nil + } + merged, err := history.EncodeDocuments(refs) + if err != nil { + return "", false, err + } + return merged, true, nil +} diff --git a/internal/services/gateway/session_document_status_test.go b/internal/services/gateway/session_document_status_test.go new file mode 100644 index 0000000..b10c450 --- /dev/null +++ b/internal/services/gateway/session_document_status_test.go @@ -0,0 +1,448 @@ +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" +) + +// 本文件是 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{transcriptMessage(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{ + transcriptMessage(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}, + }) + if err != nil { + t.Fatalf("encode: %v", err) + } + merged, changed, err := mergeDocumentStatus(raw, []history.DocumentRef{ + {Name: "aaa.txt", Status: history.DocumentStatusExtracted, Chars: 42, Truncated: true}, + }) + 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) != 2 { + t.Fatalf("refs = %+v, want 2 条（没匹配到的原样保留）", 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]) + } + if refs[1].Status != "" || refs[1].Name != "bbb.pdf" { + t.Fatalf("refs[1] = %+v, want 原样保留（本回合没解析到它）", refs[1]) + } + + if _, changed, err := mergeDocumentStatus(merged, []history.DocumentRef{{Name: "aaa.txt", Status: history.DocumentStatusExtracted, Chars: 42, Truncated: true}}); 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) + 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("坏列不该被写回") + } + }) +} + +// 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 +} diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index 45110f6..c4c3b38 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -142,6 +142,10 @@ type sessionState struct { // 队列是一条只在队尾追加、可能被整体弹出/清空的先进先出队列，因此「第几条」 // 不稳定；按 seq 认领已注入的消息，才不会被同一时刻的追加或清空误伤。 pendingSeq uint64 + // turnDocs 是本回合已落库的用户消息（id + 只有引用的 document 列），供回合 + // 结算时把抽取结论回写进那一列（design/24 §6.1）。它按代次配对，因此被中断/ + // 取代的陈旧回合读不到本回合的条目。见 session_document_status.go。 + turnDocs turnDocuments } // pendingMsg 是一条已排队用户回合。它携带自己的 env，使续跑 @@ -161,6 +165,10 @@ type pendingMsg struct { // 它只在状态锁下分配，是转向注入「注入＝已消费」认领的依据：队列会被追加、 // 被弹出、被中断清空，序号却能唯一定位这一条。 seq uint64 + // persisted 是这条待处理回合已落库的用户消息，供它的续跑回合结算时回写 + // 文档抽取结论（design/24 §6.1）：它在入队那一刻就落了库，而那时抽取还没 + // 发生。见 session_document_status.go。 + persisted []history.Message } // Orchestrator 拥有每会话的消息状态机。它对并发安全 @@ -263,11 +271,14 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes ctx = withClientMsgHint(ctx, a.hint) if st.mode != ModeIdle { st.pendingSeq++ - st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, mode: msgMode, at: time.Now(), seq: st.pendingSeq}) + seq := st.pendingSeq + st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, mode: msgMode, at: time.Now(), seq: seq}) depth := len(st.pending) mode := st.mode o.mu.Unlock() - o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID) + // 落库在锁外（慢操作）完成，随后把它挂回那条待处理回合：抽取结论要等 + // 这个回合真正跑起来才有，只能在它的续跑结算时回写（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 } @@ -279,16 +290,7 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes // 消息不会抢在被取消回合落定旧快照之前开跑（session_turn_fence.go）。 fence := newTurnFence(st.fence.done) st.fence = fence - // 新回合延续该会话的对话：持久化转录被播种在 - // 刚发送的消息之前（design/15 §2.2/§4，design/22 - // §7-5），且其助手话语保持说话时的角色。该读取 - // 在此发生，在下面 persistUserMessages 写入新消息之前，因此刚 - // 发送的消息绝不在上下文中重复。没有历史（或空会话）的 - // 存储让该回合只使用调用方给出的消息。它是在状态锁下 - // 的有界索引读取（transcriptContextCount 行），而状态锁 - // 从不会被运行中的回合持有，因此它无法把 Submit 堵在回合后面。 - turn := o.startTurnMessages(ctx, sessionID, messages) - st.tail = o.boundedTail(turn) + turn := o.seedTurnLocked(ctx, st, sessionID, messages) runCtx, cancel := context.WithCancel(ctx) st.cancel = cancel o.mu.Unlock() @@ -299,9 +301,10 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, gen)) o.emitState(sessionID, EventSessionBusy, map[string]any{"sessionId": sessionID, "mode": ModeThinking}) - // 只有此刻到达的消息会被持久化：已播种的转录 - // 已经持久，重新追加它会重复整个对话。 - o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID) + // 只有此刻到达的消息会被持久化（已播种的转录已持久，重追加会重复整个对话）， + // 且落库只写**引用** —— 抽取还发生在后面，结论由回合结算时补进同一列 + // （两段式见 session_document_status.go）。 + o.rememberTurnDocuments(sessionID, gen, o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID)) if sync { return o.runImmediateSync(runCtx, sessionID, gen, turn, env, fence, deviceID, msgMode, a.hint.MessageID) @@ -351,8 +354,13 @@ func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Mes // deviceID 是产出回合的客户端 id，mode 是其行为模式， // personaID 是它说话时的角色，都用于标注持久化的回复 // （design/16 §4.5、§5.1；design/22 §7-5）。 +// +// 它同时是文档抽取结论回写的落点（design/24 §6.1）：本回合用户消息的 +// document 列里此前只有引用，装配期解析出的 status/chars/truncated 在这里补上 +// —— 见 session_document_status.go 的两段式说明。 func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Result, runErr error, deviceID, mode, personaID string) error { o.persistAssistantReply(sessionID, res, deviceID, mode, personaID) + o.persistDocumentStatus(sessionID, gen, res) o.mu.Lock() st := o.states[sessionID] if st == nil || st.gen != gen { @@ -378,6 +386,9 @@ func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Resu // 的回合无需再等任何在途回合：围栏只负责放行它后面的那一个。 fence := newTurnFence(nil) st.fence = fence + // 续跑的这一轮同样有「回合前已落库的用户消息」，因此把它们的文档状态 + // 记到新代次上（design/24 §6.1）。 + st.turnDocs = turnDocuments{gen: gen, msgs: next.persisted} depth := len(st.pending) resume := append([]model.Message(nil), st.tail...) env := next.env @@ -471,20 +482,3 @@ func (o *Orchestrator) notifyPersisted(msg history.Message) { o.onPersisted(msg) } } - -// deviceIDContextKey 是携带消息客户端 deviceID 的 context 键 -// （design/16 §4.5）从 HTTP/WS 边界引入编排器的持久化。 -type deviceIDContextKey struct{} - -// withDeviceID 返回携带 deviceID 的 context，使编排器能标注 -// 持久化的用户消息。空 deviceID 是空操作（不写入标签）。 -func withDeviceID(ctx context.Context, deviceID string) context.Context { - return context.WithValue(ctx, deviceIDContextKey{}, deviceID) -} - -// deviceIDFrom 返回 ctx 携带的 deviceID，未设置时返回 ""。它 -// 只返回格式良好的非空值；其他一律视为缺失。 -func deviceIDFrom(ctx context.Context) string { - v, _ := ctx.Value(deviceIDContextKey{}).(string) - return v -} diff --git a/internal/services/gateway/session_persist.go b/internal/services/gateway/session_persist.go index a3414d3..f0027b1 100644 --- a/internal/services/gateway/session_persist.go +++ b/internal/services/gateway/session_persist.go @@ -60,26 +60,32 @@ func modeFrom(ctx context.Context) string { // ctx 上若带着客户端的幂等键（G4，clientMsgHintFrom），第一条用户消息会带上它 // 与编排器预分配的消息 id：写入因此可被 (session_id, client_msg_id) 的唯一索引 // 判重。省略幂等键时 ctx 上没有该值，写入与加入这一列之前逐字相同。 -func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string, messages []model.Message, deviceID, mode, personaID string) { +// +// 它返回刚落库的那些消息（含存储分配的 id 与**只有引用**的 document 列）， +// 使调用方能把它们挂到本回合上：抽取结论要等回合跑起来才有，只能在回合结束后 +// 回写进同一列（design/24 §6.1，见 session_document_status.go）。这就是那条 +// 「回合前落引用、回合后补状态」的两段式。 +func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string, messages []model.Message, deviceID, mode, personaID string) []history.Message { if o.history == nil { - return + return nil } hint := clientMsgHintFrom(ctx) + persisted := make([]history.Message, 0, len(messages)) for _, m := range messages { if m.Role != model.RoleUser { continue } - text, image, document, err := messageTextImageDocument(m) + text, image, documents, err := messageTextImageDocument(m) if err != nil { - // 文档引用编码失败（理论上不可达：引用只含字符串与整数）只丢那一列 - // 并留下带来源的日志，整条消息照常落库 —— 丢一条消息远比丢一个附件 - // 引用严重，而且这里绝不静默。 + // 文档引用编码失败（只可能是坏数据或超限）只丢那一列并留下带来源的 + // 日志，整条消息照常落库 —— 丢一条消息远比丢一个附件引用严重，而且 + // 这里绝不静默。 o.log.Error("orchestrator persist document", "err", err, "session", sessionID, "device_id", deviceID) } - if text == "" && image == "" && document == "" { + if text == "" && image == "" && documents == "" { continue } - msg := history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, Document: document, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID} + msg := history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, Document: documents, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID} // 幂等键只标注**第一条**用户消息：一个回合至多只有一条用户消息，而把 // 同一个键盖到多条上会让它们互相冲突（唯一索引把它们判成同一条）。 if hint.ClientMsgID != "" { @@ -96,6 +102,7 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string } continue } + persisted = append(persisted, full) if msg.ClientMsgID != "" && full.ID != msg.ID { // 存储侧唯一索引裁决出既有行（例如本进程重启后另一次写入已落库）： // 把权威 id 覆盖进认领表，别让后续重复请求拿到一个悬空的 messageId。 @@ -103,6 +110,7 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string } o.notifyPersisted(full) } + return persisted } // persistAssistantReply 把已结算回合的最终助手文本写入 @@ -128,15 +136,21 @@ func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result, o.notifyPersisted(full) } -// messageTextImageDocument 返回消息内容部分拼接后的文本、首个图片 URL 与首个 -// 文档引用（已编码为落库用的 JSON 小对象），供持久化进 chat_messages 各列。 +// messageTextImageDocument 返回消息内容部分拼接后的文本、首个图片 URL 与**全部** +// 文档引用（已编码为落库用的 JSON 数组），供持久化进 chat_messages 各列。 // // 文档那一半只落**引用与统计**（design/24 §2.4）：装配期已经把抽取所得的正文 // 替换成文本部件，因此这里绝不把正文写进 document 列 —— 正文一旦落库，每翻 // 一页历史都要背着它重传（G3 的教训）。抽取统计（chars/truncated/status）属于 -// 装配期与抽取器，模型层只携带引用本身；编码失败时返回错误，由调用方决定如何 -// 可见地报告，而不是在这里静默当成「没有附件」。 -func messageTextImageDocument(m model.Message) (text, image, document string, err error) { +// 装配期与抽取器，模型层只携带引用本身，它们由回合结束后的回写补进同一列 +// （见 session_document_status.go）。 +// +// 一个回合可以挂多个附件（design/24 §6.1 的上限 5），因此这里按**部件顺序**收集 +// 全部引用：此前「只留第一个」的写法会让其余几个从历史与后续回合的上下文里凭空 +// 消失（缺口①）。编码失败（含超限）时返回错误，由调用方决定如何可见地报告， +// 而不是在这里静默当成「没有附件」。 +func messageTextImageDocument(m model.Message) (text, image, documents string, err error) { + refs := make([]history.DocumentRef, 0, len(m.Content)) for _, c := range m.Content { switch c.Type { case model.ContentTypeText: @@ -146,17 +160,14 @@ func messageTextImageDocument(m model.Message) (text, image, document string, er image = c.ImageURL } case model.ContentTypeDocument: - if document != "" { - continue - } - encoded, err := history.EncodeDocument(documentRefOf(c)) - if err != nil { - return text, image, "", err - } - document = encoded + refs = append(refs, documentRefOf(c)) } } - return text, image, document, nil + encoded, err := history.EncodeDocuments(refs) + if err != nil { + return text, image, "", err + } + return text, image, encoded, nil } // documentRefOf 把文档内容片段收成落库用的引用。四个字段全部来自服务端 @@ -359,3 +370,20 @@ func (o *Orchestrator) releaseClientMsg(sessionID, clientMsgID, msgID string) { } } } + +// deviceIDContextKey 是携带消息客户端 deviceID 的 context 键 +// （design/16 §4.5）从 HTTP/WS 边界引入编排器的持久化。 +type deviceIDContextKey struct{} + +// withDeviceID 返回携带 deviceID 的 context，使编排器能标注 +// 持久化的用户消息。空 deviceID 是空操作（不写入标签）。 +func withDeviceID(ctx context.Context, deviceID string) context.Context { + return context.WithValue(ctx, deviceIDContextKey{}, deviceID) +} + +// deviceIDFrom 返回 ctx 携带的 deviceID，未设置时返回 ""。它 +// 只返回格式良好的非空值；其他一律视为缺失。 +func deviceIDFrom(ctx context.Context) string { + v, _ := ctx.Value(deviceIDContextKey{}).(string) + return v +} diff --git a/internal/services/gateway/session_transcript.go b/internal/services/gateway/session_transcript.go index 12b45ea..3887eed 100644 --- a/internal/services/gateway/session_transcript.go +++ b/internal/services/gateway/session_transcript.go @@ -58,6 +58,23 @@ func (o *Orchestrator) startTurnMessages(ctx context.Context, sessionID string, return concatMessages(seed, messages...) } +// seedTurnLocked 为新回合播种对话（持久化转录在刚发送的消息之前），并把播种后的 +// 尾部记进会话状态。**必须在 o.mu 下调用**（它读写 st.tail；与 admitLocked / +// claimClientMsgLocked 同一约定），调用方负责解锁 —— 因此它从 +// session_orchestrator.go 的 submitInternal 里拆出来，让那个函数留在「函数 ≤80 +// 行」的硬约束内。 +// +// 播种的时刻是刻意的：它在 persistUserMessages 写入新消息**之前**，因此刚发送的 +// 消息绝不在上下文中重复（design/15 §2.2/§4、design/22 §7-5）。没有历史（或空 +// 会话）的存储让该回合只使用调用方给出的消息。它是在状态锁下的有界索引读取 +// （transcriptContextCount 行），而状态锁从不会被运行中的回合持有，因此它无法把 +// Submit 堵在回合后面。 +func (o *Orchestrator) seedTurnLocked(ctx context.Context, st *sessionState, sessionID string, messages []model.Message) []model.Message { + turn := o.startTurnMessages(ctx, sessionID, messages) + st.tail = o.boundedTail(turn) + return turn +} + // sessionTranscript 读取该会话最近的持久化消息并把它们 // 映射为模型消息（最旧优先）。nil/不可读的存储或读错误 // 都得到 nil，因此聊天绝不被元数据查找阻塞；没有 @@ -88,7 +105,7 @@ func (o *Orchestrator) sessionTranscript(ctx context.Context, sessionID string) // persona_id）作为其说话者标签，因此即使会话被切换过， // 上下文也能把它归属到那个角色；非助手消息保持其 // 已存发送者标签（design/16 §4.6 的第三方显示名），绝不 -// 被重新标为说话角色。文档引用一并还原（design/24 §9.5）： +// 被重新标为说话角色。文档引用**全部**一并还原（design/24 §9.5）： // 少了它，附件会在发起它的那一回合之后的所有回合里消失。 func transcriptMessage(h history.Message) model.Message { role := historyRoleToModel(h.Role) @@ -108,9 +125,13 @@ func transcriptMessage(h history.Message) model.Message { return msg } -// transcriptDocumentParts 把已存 document 列还原为该回合的文档部件（design/24 -// §9.5）。部件只交给 RoleUser 回合（§2.3-③）：文档在任何别的角色上都会让编码器 -// fail-loud，而一行坏数据不该把整个会话卡死。 +// transcriptDocumentParts 把已存 document 列里的**全部**引用还原为该回合的文档 +// 部件（design/24 §9.5）。部件只交给 RoleUser 回合（§2.3-③）：文档在任何别的 +// 角色上都会让编码器 fail-loud，而一行坏数据不该把整个会话卡死。 +// +// 列里是引用数组（一个回合最多 5 个附件，缺口①），因此这里逐条还原、保持部件 +// 顺序 —— 少还原一条，那个附件就会在发起它的回合之后的所有回合里消失，与 +// 「清单恒在」的承诺自相矛盾。 // // 列被写坏（或附件落在非用户行上）时**不静默**：打一条带来源的 WARN 并省掉 // 该部件，回合照常继续 —— 附件凭空消失正是这条清单要修的缺陷，因此必须留下 @@ -123,22 +144,26 @@ func transcriptDocumentParts(role model.Role, h history.Message) []model.Content transcriptDocumentLog().Warn("transcript document on non-user row", "message", h.ID, "role", h.Role) return nil } - ref, err := history.DecodeDocument(h.Document) + refs, err := history.DecodeDocuments(h.Document) if err != nil { transcriptDocumentLog().Warn("transcript document decode", "message", h.ID, "err", err) return nil } - if ref == nil || ref.Name == "" { - transcriptDocumentLog().Warn("transcript document without name", "message", h.ID) - return nil + parts := make([]model.Content, 0, len(refs)) + for _, ref := range refs { + if ref.Name == "" { + transcriptDocumentLog().Warn("transcript document without name", "message", h.ID) + continue + } + doc := model.Document{ + Name: ref.Name, + MIME: ref.MIME, + URL: documentFilePrefix + ref.Name, + Size: ref.Size, + } + parts = append(parts, model.DocumentMessage(model.RoleUser, "", doc).Content...) } - doc := model.Document{ - Name: ref.Name, - MIME: ref.MIME, - URL: documentFilePrefix + ref.Name, - Size: ref.Size, - } - return model.DocumentMessage(model.RoleUser, "", doc).Content + return parts } // historyRoleToModel 把已存聊天角色映射为模型的角色词汇，