From 1420e8aaa67b783ca9f8d6816fc97a1627b6b44b Mon Sep 17 00:00:00 2001 From: AskaEth Date: Tue, 6 Oct 2026 14:12:27 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway):=20D3=20=E5=AE=A1=E6=9F=A5?= =?UTF-8?q?=E8=AF=BB=E4=BE=A7/revert=20=E7=AB=AF=E7=82=B9=20+=20=E8=AF=AD?= =?UTF-8?q?=E9=9F=B3=E7=94=A8=E6=88=B7=E5=8F=AF=E8=A7=81=E9=9D=A2=EF=BC=88?= =?UTF-8?q?L45=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit (a) 记忆入库前审查的两条 admin 端点（主人判「有必要就做」）： GET /api/v1/admin/memories/reviews?limit= （留痕读侧，不含记忆正文） POST /api/v1/admin/memories/{id}/revert （回滚软合并，落审计 memory.review.revert） 两条 writeAuth（裁定 DECISIONS #10：留痕属私有面，不给 viewer） (b) 语音用户可见面：DTO voice（含服务端派生 url）+ GET /api/v1/media/{name} （200/206/304/416/HEAD、Accept-Ranges、非法名/穿越统一 404 media_not_found） + documents.Store.SaveAudio（内容寻址、$DATA_DIR 0600）+ history voice 列读写与引用计数差动 缺口（如实）：message_persisted/握手回放仍不带 voice（文件面外，需给 history.Message 加字段）； AMR 需一行 VoiceTranscoder 注入（DECISIONS #11 已排）。4 个变异自检（含一条自认的等价变异）。 --- internal/runtime/documents/store_audio.go | 83 +++++ .../runtime/documents/store_audio_test.go | 170 ++++++++++ internal/runtime/history/voice_column.go | 203 ++++++++++++ internal/runtime/history/voice_column_test.go | 213 ++++++++++++ internal/runtime/voice/media.go | 118 +++++++ internal/runtime/voice/media_test.go | 126 ++++++++ .../services/gateway/admin_memory_review.go | 212 ++++++++++++ .../gateway/admin_memory_review_test.go | 250 +++++++++++++++ internal/services/gateway/history.go | 11 +- internal/services/gateway/history_voice.go | 134 ++++++++ .../services/gateway/history_voice_test.go | 303 ++++++++++++++++++ internal/services/gateway/history_write.go | 13 +- internal/services/gateway/media.go | 108 +++++++ internal/services/gateway/media_test.go | 243 ++++++++++++++ internal/services/gateway/routes.go | 20 ++ 15 files changed, 2202 insertions(+), 5 deletions(-) create mode 100644 internal/runtime/documents/store_audio.go create mode 100644 internal/runtime/documents/store_audio_test.go create mode 100644 internal/runtime/history/voice_column.go create mode 100644 internal/runtime/history/voice_column_test.go create mode 100644 internal/runtime/voice/media.go create mode 100644 internal/runtime/voice/media_test.go create mode 100644 internal/services/gateway/admin_memory_review.go create mode 100644 internal/services/gateway/admin_memory_review_test.go create mode 100644 internal/services/gateway/history_voice.go create mode 100644 internal/services/gateway/history_voice_test.go create mode 100644 internal/services/gateway/media.go create mode 100644 internal/services/gateway/media_test.go diff --git a/internal/runtime/documents/store_audio.go b/internal/runtime/documents/store_audio.go new file mode 100644 index 0000000..2b716f6 --- /dev/null +++ b/internal/runtime/documents/store_audio.go @@ -0,0 +1,83 @@ +package documents + +// 本文件归属**音频副本**的落盘分支（design/24 §10.9-2 的 `SaveAudio`；切片 CODE-VOICE +// 留下的跨片接缝，本片补齐）。语音副本**复用**文档那一整套存储（同一 +// `/uploads/documents` 目录、同一份内容寻址、同一张 `documents` 元数据表、 +// 同一套引用计数与 GC），只有**类型裁决**不同： +// +// - 文档：字节嗅探是唯一裁决者，且扩展名必须与嗅探**成对**（`.pdf` 必须嗅探成 +// `application/pdf`，文本类必须 `text/plain`）； +// - 音频：格式在**上游**已经由格式名确定（`voice.Prepare` 决定「原样落」还是 +// 「转码后落」），字节嗅探对它**无意义**（`http.DetectContentType` 对绝大多数 +// 音频容器返回 `application/octet-stream`，用它必误拒）⇒ 这里只认**扩展名**， +// 且限定在 10 项音频白名单内（`voice.MediaExtAllowed`，与回放端点的白名单同一份）。 +// +// 因此两者的关系是「同一存储核心 + 两套类型裁决」，而不是「两套存储」：目录、去重、 +// 路径双检、临时文件 + rename 的原子性、0600 文件权限全部走 `commit`，一行都没有分叉。 +// +// 为什么不进文档的抽取 / 注入链路（design/24 §10.9-14）：音频的「正文」是那次转写， +// 已经在消息的 `content`/`voice` 列里；把音频当文档抽取只会得到一堆噪声字节的摘要。 +// 因此音频只占存储与引用计数，不占注入预算、不进 `documents` DTO 数组。 + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + + voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" +) + +// SaveAudio 把一份语音副本落进共享存储，返回服务端签发的引用（形状与 Save 完全一致： +// `Name`/`MIME`/`Size`/`SHA256`/`Deduped`）。判定顺序刻意与 Save 同族，只是把 +// 「扩展名白名单 + 嗅探成对裁决」换成「**音频扩展名白名单**」： +// +// 1. **扩展名白名单**（deny-by-default）：`voice.MediaExtAllowed` 之外的扩展名一律 +// 拒绝（`ErrExtNotAllowed`）—— 调用方给的是 `voice.`，格式名来自 +// `voice.Prepare`，两者同一份归一； +// 2. **空内容**拒绝（`ErrEmpty`）：空副本播放不了，也没有落盘意义； +// 3. **字节上限**拒绝（`ErrTooLarge`，上限见 Limit）：与文档共用同一个生效上限 +// （`CHAT_DOCUMENT_MAX_BYTES`），因为两者落在**同一个目录、同一张表**，给它一个 +// 独立上限只会让「这份文件为什么被拒」取决于它是从哪条路来的； +// 4. 内容寻址落盘（`sha256(bytes)[:32] + ext`，同字节同扩展名天然去重；文件权限继承 +// 临时文件的 0600、目录 0700）+ 路径双检，随后（可选）写一行元数据。 +// +// 元数据的 uploader 与文档同源（由构造 Store 的一方给出：网关是请求 claims 的 sub， +// IM 是渠道来源标签）—— 语音副本在 `documents` 表里同样记着「是谁发来的」，身份 +// 绝不来自文件名或内容。 +// +// 返回错误时的契约与 Save 一致：元数据写入失败的错误同时满足 +// `errors.Is(err, ErrMetadata)`，且**已落盘的引用照旧返回**。 +func (s *Store) SaveAudio(ctx context.Context, origName string, data []byte) (Ref, error) { + ext := ExtOf(origName) + if !voiceref.MediaExtAllowed(ext) { + return Ref{}, fmt.Errorf("documents: save audio: %w", ErrExtNotAllowed) + } + if len(data) == 0 { + return Ref{}, fmt.Errorf("documents: save audio: %w", ErrEmpty) + } + if limit := s.Limit(); len(data) > limit { + return Ref{}, fmt.Errorf("documents: save audio: 上限 %d 字节，本次 %d 字节: %w", limit, len(data), ErrTooLarge) + } + mime := voiceref.MediaMIME(ext) + if mime == "" { + // 白名单已判过一次；这里是纵深防御 —— 没有下发类型就绝不落盘（落一份读不出 + // `Content-Type` 的副本，客户端只能靠猜）。 + return Ref{}, fmt.Errorf("documents: save audio: 扩展名 %q 没有下发类型: %w", ext, ErrExtNotAllowed) + } + sum := sha256.Sum256(data) + ref := Ref{ + Name: hashName(sum[:], ext), + MIME: mime, + Size: int64(len(data)), + SHA256: hex.EncodeToString(sum[:]), + } + committed, err := s.commit(ref, data) + if err != nil { + return Ref{}, err + } + if err := s.record(ctx, committed); err != nil { + return committed, err + } + return committed, nil +} diff --git a/internal/runtime/documents/store_audio_test.go b/internal/runtime/documents/store_audio_test.go new file mode 100644 index 0000000..b32201b --- /dev/null +++ b/internal/runtime/documents/store_audio_test.go @@ -0,0 +1,170 @@ +package documents + +// 本文件锁定**音频副本落盘分支**（design/24 §10.9-2 的 `SaveAudio`）的四条口径： +// ① 内容寻址 + 去重 + 0600 落盘（与文档共用同一套写入核心）； +// ② 类型裁决走**音频白名单**（10 项），与文档的 23 项白名单**互不越界**； +// ③ 空内容 / 超上限 / 未知扩展名都如实拒绝（不落半截文件、不写元数据行）； +// ④ 元数据行照写（uploader 只来自构造方，绝不从文件名或内容猜身份）。 + +import ( + "bytes" + "context" + "errors" + "os" + "path/filepath" + "runtime" + "testing" + + voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" +) + +// audioMetaRecorder 记录交给元数据面的行。它按 MetadataWriter 的契约模拟 +// `ON CONFLICT(name) DO NOTHING`（首次的 uploader/created_at 是权威），因此 +// 「重复落盘只留下一行」在本用例里是**契约级**断言，而不是替身的记账细节。 +type audioMetaRecorder struct { + seen map[string]bool + recs []Record +} + +func (m *audioMetaRecorder) WriteDocument(_ context.Context, rec Record) error { + if m.seen == nil { + m.seen = make(map[string]bool) + } + if m.seen[rec.Name] { + return nil + } + m.seen[rec.Name] = true + m.recs = append(m.recs, rec) + return nil +} + +// audioSample 是一份「像那么回事」的可播音频字节（RIFF/WAVE 头 + 载荷）。 +func audioSample() []byte { + return append([]byte("RIFF\x24\x00\x00\x00WAVEfmt "), []byte("voice-copy-payload")...) +} + +func TestSaveAudioStoresContentAddressedCopy(t *testing.T) { + t.Parallel() + dir := filepath.Join(t.TempDir(), uploadsSubDir, documentSubDir) + meta := &audioMetaRecorder{} + store := newTestStore(t, StoreConfig{UploadsDir: dir, Uploader: "im:qq-obv11:10001", Meta: meta}) + ctx := context.Background() + data := audioSample() + + ref, err := store.SaveAudio(ctx, "voice.wav", data) + if err != nil { + t.Fatalf("SaveAudio: %v", err) + } + wantName, wantSHA := expectedName(t, ".wav", data) + if ref.Name != wantName { + t.Fatalf("Name = %q, want %q（内容寻址）", ref.Name, wantName) + } + if ref.SHA256 != wantSHA { + t.Fatalf("SHA256 = %q, want %q", ref.SHA256, wantSHA) + } + if ref.MIME != "audio/wav" { + t.Fatalf("MIME = %q, want audio/wav（按扩展名裁决，与回放端点同一份表）", ref.MIME) + } + if ref.Size != int64(len(data)) || ref.Deduped { + t.Fatalf("Size/Deduped = %d/%v, want %d/false", ref.Size, ref.Deduped, len(data)) + } + if ref.Path == "" { + t.Fatal("Path 为空：落盘路径没有回给调用方") + } + // 写进共享目录（与文档同一个目录），绝不是另一个地方。 + if filepath.Dir(ref.Path) != filepath.Clean(dir) { + t.Fatalf("落盘目录 = %q, want %q", filepath.Dir(ref.Path), filepath.Clean(dir)) + } + got, err := os.ReadFile(ref.Path) + if err != nil { + t.Fatalf("读回副本: %v", err) + } + if !bytes.Equal(got, data) { + t.Fatal("落盘字节与输入不一致") + } + if runtime.GOOS != "windows" { + info, err := os.Stat(ref.Path) + if err != nil { + t.Fatalf("Stat 副本: %v", err) + } + if perm := info.Mode().Perm(); perm != 0o600 { + t.Fatalf("副本权限 = %o, want 600（继承临时文件）", perm) + } + } + if len(meta.recs) != 1 { + t.Fatalf("元数据行数 = %d, want 1", len(meta.recs)) + } + if rec := meta.recs[0]; rec.Uploader != "im:qq-obv11:10001" || rec.MIME != "audio/wav" || rec.Size != int64(len(data)) { + t.Fatalf("元数据行 = %+v", rec) + } + + // 同字节同扩展名第二次落盘 ⇒ 去重（不新建文件、不写第二行元数据）。 + again, err := store.SaveAudio(ctx, "voice.wav", data) + if err != nil { + t.Fatalf("重复 SaveAudio: %v", err) + } + if !again.Deduped || again.Name != ref.Name { + t.Fatalf("重复落盘 = {name:%q deduped:%v}, want {name:%q deduped:true}", again.Name, again.Deduped, ref.Name) + } + if len(meta.recs) != 1 { + t.Fatalf("去重后元数据行数 = %d, want 1", len(meta.recs)) + } + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("读目录: %v", err) + } + if len(entries) != 1 { + t.Fatalf("目录里文件数 = %d, want 1（盘上只留一份）", len(entries)) + } +} + +func TestSaveAudioRejectsBadInputs(t *testing.T) { + t.Parallel() + ctx := context.Background() + data := audioSample() + + store := newTestStore(t, StoreConfig{UploadsDir: filepath.Join(t.TempDir(), "docs")}) + for _, name := range []string{"voice.txt", "voice.pdf", "voice", "voice.silk", "voice.mp4"} { + if _, err := store.SaveAudio(ctx, name, data); !errors.Is(err, ErrExtNotAllowed) { + t.Fatalf("SaveAudio(%q) err = %v, want ErrExtNotAllowed", name, err) + } + } + if _, err := store.SaveAudio(ctx, "voice.wav", nil); !errors.Is(err, ErrEmpty) { + t.Fatalf("空内容 err = %v, want ErrEmpty", err) + } + small := newTestStore(t, StoreConfig{UploadsDir: filepath.Join(t.TempDir(), "docs"), MaxBytes: 4}) + if _, err := small.SaveAudio(ctx, "voice.wav", data); !errors.Is(err, ErrTooLarge) { + t.Fatalf("超上限 err = %v, want ErrTooLarge", err) + } + // 未配置目录 ⇒ 明确失败（绝不猜一个目录出来）。 + noDir := NewStore(StoreConfig{Log: nil}) + if _, err := noDir.SaveAudio(ctx, "voice.wav", data); !errors.Is(err, ErrNoDir) { + t.Fatalf("未配置目录 err = %v, want ErrNoDir", err) + } +} + +// TestSaveAudioUsesAudioWhitelistNotDocumentWhitelist 钉住「两套类型裁决」：音频白名单 +// 与文档白名单互不越界 —— 文档拒绝的音频扩展名在音频分支放行，音频分支拒绝的文档 +// 扩展名在文档白名单里放行。变异：把 SaveAudio 改走 ExtAllowed ⇒ 本用例红。 +func TestSaveAudioUsesAudioWhitelistNotDocumentWhitelist(t *testing.T) { + t.Parallel() + ctx := context.Background() + data := audioSample() + store := newTestStore(t, StoreConfig{UploadsDir: filepath.Join(t.TempDir(), "docs")}) + + if ExtAllowed(".wav") { + t.Fatal("文档白名单不该包含 .wav") + } + if voiceref.MediaExtAllowed(".txt") { + t.Fatal("音频白名单不该包含 .txt") + } + if _, err := store.Save(ctx, "doc.wav", data); !errors.Is(err, ErrExtNotAllowed) { + t.Fatalf("文档分支收 .wav 的 err = %v, want ErrExtNotAllowed", err) + } + if _, err := store.SaveAudio(ctx, "doc.txt", []byte("昔涟的文本\n")); !errors.Is(err, ErrExtNotAllowed) { + t.Fatalf("音频分支收 .txt 的 err = %v, want ErrExtNotAllowed", err) + } + if _, err := store.SaveAudio(ctx, "voice.ogg", data); err != nil { + t.Fatalf("音频分支拒了白名单内的 .ogg: %v", err) + } +} diff --git a/internal/runtime/history/voice_column.go b/internal/runtime/history/voice_column.go new file mode 100644 index 0000000..eb5efec --- /dev/null +++ b/internal/runtime/history/voice_column.go @@ -0,0 +1,203 @@ +package history + +// 本文件归属聊天消息的**语音事实**列（design/24 §10.4 / §10.9-4；迁移 0020）： +// 一条消息来自一条语音吗、那次转写成不成、盘上有没有一份可回放的副本。 +// +// 列的内容是**单个 JSON 对象**（形状见 `internal/runtime/voice.Ref`；空串 = 这条消息 +// 不是语音消息）。它与 document 列的演进史不同：自诞生起就是单对象，因此没有需要 +// 兼容的旧形状。 +// +// 写入是**两段式**（与 document 列同一手法）：入站语音的消息行先落库，副本落盘拿到 +// 内容寻址名之后再由渠道桥回填同一列 —— 因为「可播放副本」只有在转码/落盘之后才 +// 存在，而那时消息行早已写好。回填入口 = 本文件的 UpdateMessageVoice（渠道桥按 +// 窄接口 `voiceColumnWriter` 发现它，见 channelbridge/inbound_voice_copy.go）。 +// +// 音频副本与文档共用 `documents` 表与同一套引用计数（design/24 §10.5）：副本一旦被 +// 引用就要 +1，否则 GC 会把一份仍被播放的语音回收掉。因此本文件的每一次列变更都按 +// **新旧名字集合的差**在同一条事务里增减计数（与 UpdateMessageDocument 逐字同族）。 + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" + + voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" +) + +// voiceColumn 是语音事实的列名（迁移 0020 加的列）。 +const voiceColumn = "voice" + +// UpdateMessageVoice 只替换 id 所指消息的 voice 列，其余任何列保持已存值。它是 +// 「入站语音落库后回填来源事实与副本引用」的写入端（design/24 §10.2 第 5 步）。 +// +// 四条纪律： +// - **未知 id 返回 ErrMessageNotFound**（与 UpdateMessage / UpdateMessageDocument +// 同一口径），调用方能区分「那一行不见了」与「写失败」，并据此如实记 WARN； +// - **坏形状拒绝写**：非空值必须解码成自洽的语音事实（来源/状态在闭集内、副本三 +// 字段同生共死）。写一份读不回来的 JSON 等于让客户端永远看不到这条语音，所以这里 +// fail loud，绝不静默落一份半截数据。空串是**合法**值（这条消息不是语音）； +// - **引用计数与消息列同一个事务**：按新旧名字集合的差增减 `documents.ref_count` +// （切片 S7 的既有通道），任何一边失败就整体回滚 —— 绝不留下「消息认了这份音频、 +// 计数没加」这种让 GC 误删仍被引用文件的窗口； +// - **列不存在时前滚**：迁移 0020 之前建出的表（含本包 DDL 造的测试库）在这里幂等 +// 补列，与 ensureDocumentColumn / ensureTurnIDColumn 同一手法。 +func (s *Store) UpdateMessageVoice(ctx context.Context, id, raw string) error { + if _, err := voiceref.Decode(raw); err != nil { + return fmt.Errorf("history: update message voice %s: %w", id, err) + } + if err := ensureVoiceColumn(ctx, s.db); err != nil { + return err + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("history: update message voice %s: %w", id, err) + } + defer func() { _ = tx.Rollback() }() + + var previous string + err = tx.QueryRowContext(ctx, `SELECT `+voiceColumn+` FROM chat_messages WHERE id = ?`, id).Scan(&previous) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("history: update message voice %s: %w", id, ErrMessageNotFound) + } + return fmt.Errorf("history: read message voice %s: %w", id, err) + } + res, err := tx.ExecContext(ctx, `UPDATE chat_messages SET `+voiceColumn+` = ? WHERE id = ?`, raw, id) + if err != nil { + return fmt.Errorf("history: update message voice %s: %w", id, err) + } + n, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("history: update message voice %s rows: %w", id, err) + } + if n == 0 { + return fmt.Errorf("history: update message voice %s: %w", id, ErrMessageNotFound) + } + if err := s.syncVoiceRefs(ctx, tx, previous, raw); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("history: update message voice %s: %w", id, err) + } + return nil +} + +// VoiceColumns 按 id 批量取回各行的 voice 列原文（**一次** `IN (...)`，禁 N+1）。 +// 返回「消息 id → 列内容」；查询不到的 id（已删除、id 拼错）**不出现在映射里**， +// 与「取到空串」（这条消息不是语音）因此可以区分。 +// +// 它只读、不迁移、不建连接池（用本 Store 的活连接）。列不存在（迁移 0020 之前建出的 +// 表且从未写过语音）时返回错误，由调用方决定怎么如实报告 —— 这里**不**静默回空映射， +// 「没有这一面」与「这些消息都不是语音」是两件事。 +func (s *Store) VoiceColumns(ctx context.Context, ids []string) (map[string]string, error) { + unique := uniqueMessageIDs(ids) + if len(unique) == 0 { + return map[string]string{}, nil + } + placeholders := strings.TrimSuffix(strings.Repeat("?, ", len(unique)), ", ") + args := make([]any, 0, len(unique)) + for _, id := range unique { + args = append(args, id) + } + query := `SELECT id, ` + voiceColumn + ` FROM chat_messages WHERE id IN (` + placeholders + `)` + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("history: read voice columns for %d messages: %w", len(unique), err) + } + defer func() { _ = rows.Close() }() + + out := make(map[string]string, len(unique)) + for rows.Next() { + var id, raw string + if err := rows.Scan(&id, &raw); err != nil { + return nil, fmt.Errorf("history: scan voice column: %w", err) + } + out[id] = raw + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate voice columns: %w", err) + } + return out, nil +} + +// uniqueMessageIDs 去重并丢掉空 id（保持首次出现顺序）。 +func uniqueMessageIDs(ids []string) []string { + unique := make([]string, 0, len(ids)) + seen := make(map[string]bool, len(ids)) + for _, id := range ids { + if id == "" || seen[id] { + continue + } + seen[id] = true + unique = append(unique, id) + } + return unique +} + +// syncVoiceRefs 把一行消息的 voice 列从 oldRaw 换成 newRaw 时按**差**增减引用计数。 +// 两次内容相同时（含都是空）直接返回。 +// +// 与文档侧同一取舍：列被外部改坏（解不出名字）时**不让写入失败**（消息行才是用户的 +// 事实），只记 WARN 并跳过本次增减 —— 保守方向是「计数偏低 ⇒ 只会晚回收，绝不早删」。 +func (s *Store) syncVoiceRefs(ctx context.Context, ex SQLExecer, oldRaw, newRaw string) error { + if s.refs == nil || oldRaw == newRaw { + return nil + } + oldNames, err := voiceNames(oldRaw) + if err != nil { + s.log.Warn("voice refs sync skipped: malformed old column", "err", err) + return nil + } + newNames, err := voiceNames(newRaw) + if err != nil { + s.log.Warn("voice refs sync skipped: malformed new column", "err", err) + return nil + } + add, remove := diffDocumentNames(oldNames, newNames) + if len(add) > 0 { + if err := s.refs.Retain(ctx, ex, add); err != nil { + return fmt.Errorf("history: retain voice refs: %w", err) + } + } + if len(remove) > 0 { + if err := s.refs.Release(ctx, ex, remove); err != nil { + return fmt.Errorf("history: release voice refs: %w", err) + } + } + return nil +} + +// voiceNames 解出 voice 列涉及的音频副本名（有副本时恰好一个，没有副本/不是语音时为空）。 +// 与本包 documentNames 同形，因此复用同一套差集计算。 +func voiceNames(raw string) ([]string, error) { + ref, err := voiceref.Decode(raw) + if err != nil { + return nil, err + } + if !ref.HasCopy() { + return nil, nil + } + return []string{ref.Name}, nil +} + +// ensureVoiceColumn 在 chat_messages 早于本列（0020 迁移）建表时把它前滚：缺列则 +// 补加，已有则忽略重复列错误。它是幂等的（与 ensureDocumentColumn / ensureTurnIDColumn +// 同一手法），因此本包的写入路径可以在全新 schema 或既有 schema 上直接调用。 +// +// 只在**写入路径**调用：读路径不迁移（只读纪律），而一条消息要带语音事实必然先经过 +// 写入路径，因此列在读到它之前一定已经补上。 +func ensureVoiceColumn(ctx context.Context, db *sql.DB) error { + _, err := db.ExecContext(ctx, `ALTER TABLE chat_messages + ADD COLUMN voice TEXT NOT NULL DEFAULT ''`) + if err == nil { + return nil + } + // 列已存在时（由 0020 迁移补过，或上一次写入补过）SQLite 报 + // "duplicate column name: voice"。 + if strings.Contains(err.Error(), "duplicate column name") { + return nil + } + return fmt.Errorf("history: roll forward voice column: %w", err) +} diff --git a/internal/runtime/history/voice_column_test.go b/internal/runtime/history/voice_column_test.go new file mode 100644 index 0000000..92f5ba0 --- /dev/null +++ b/internal/runtime/history/voice_column_test.go @@ -0,0 +1,213 @@ +package history + +// 本文件锁定 chat_messages.voice 列的**读写视图**（design/24 §10.4 / §10.9-4）： +// ① 写入后能读回同样的 JSON 单对象，且列不存在时先幂等前滚（迁移 0020 之前建出的表）； +// ② 坏形状**拒绝写入**（绝不落下读不回来的列）； +// ③ 未知消息 id → ErrMessageNotFound（不假装写成功）； +// ④ 音频副本的名字并入**同一套**引用计数（按新旧名字集合的差增减，同一事务）； +// ⑤ 批量读是**按 id 集合一次查**，取不到的 id 不出现在映射里（与「取到空串」可区分）。 + +import ( + "context" + "errors" + "testing" + + voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" +) + +// seedVoiceMessage 经真实存储追加一条消息并返回已存行（含生成的 id）。 +func seedVoiceMessage(t *testing.T, st *Store, role, content string) Message { + t.Helper() + got, err := st.AppendFull(context.Background(), Message{SessionID: MainSessionID, Role: role, Content: content}) + if err != nil { + t.Fatalf("append %q: %v", content, err) + } + return got +} + +// voiceRawWithCopy 编码一条「有副本」的语音事实（ASR 转写成功 + 副本三字段齐备）。 +func voiceRawWithCopy(t *testing.T, name string) string { + t.Helper() + raw, err := voiceref.Encode(voiceref.Ref{ + Source: voiceref.SourceASR, + Status: voiceref.StatusTranscribed, + Name: name, + MIME: "audio/ogg", + Size: 1234, + DurationMs: 3200, + }) + if err != nil { + t.Fatalf("编码语音事实: %v", err) + } + return raw +} + +// voiceRawWithoutCopy 编码一条「没有副本」的语音事实（微信按既有裁决不取字节）。 +func voiceRawWithoutCopy(t *testing.T) string { + t.Helper() + raw, err := voiceref.Encode(voiceref.Ref{ + Source: voiceref.SourcePlatform, + Status: voiceref.StatusTranscribed, + }) + if err != nil { + t.Fatalf("编码语音事实: %v", err) + } + return raw +} + +func TestUpdateMessageVoiceRoundTrip(t *testing.T) { + t.Parallel() + refs := &recordingRefs{} + st := newRefStore(t, refs) + ctx := context.Background() + msg := seedVoiceMessage(t, st, RoleUser, "（语音）") + name := "0123456789abcdef0123456789abcdef.ogg" + raw := voiceRawWithCopy(t, name) + + // 全新库由本包 DDL 建表（**不含** voice 列）⇒ 这一次写入顺带把列幂等前滚。 + if err := st.UpdateMessageVoice(ctx, msg.ID, raw); err != nil { + t.Fatalf("UpdateMessageVoice: %v", err) + } + cols, err := st.VoiceColumns(ctx, []string{msg.ID, "does-not-exist"}) + if err != nil { + t.Fatalf("VoiceColumns: %v", err) + } + if got := cols[msg.ID]; got != raw { + t.Fatalf("读回 = %q, want %q", got, raw) + } + if _, ok := cols["does-not-exist"]; ok { + t.Fatal("取不到的 id 不该出现在映射里（与「取到空串」必须可区分）") + } + if len(refs.retained) != 1 || refs.retained[0] != name { + t.Fatalf("引用计数 retain = %v, want [%s]", refs.retained, name) + } + if len(refs.released) != 0 { + t.Fatalf("首次写入不该 release：%v", refs.released) + } + + // 清空（= 撤销语音事实）是合法值，且副本的名字被释放一次。 + if err := st.UpdateMessageVoice(ctx, msg.ID, ""); err != nil { + t.Fatalf("清空 voice: %v", err) + } + if len(refs.released) != 1 || refs.released[0] != name { + t.Fatalf("清空后 release = %v, want [%s]", refs.released, name) + } + cols, err = st.VoiceColumns(ctx, []string{msg.ID}) + if err != nil { + t.Fatalf("VoiceColumns(清空后): %v", err) + } + if got, ok := cols[msg.ID]; !ok || got != "" { + t.Fatalf("清空后列 = (%q, %v), want (\"\", true)", got, ok) + } +} + +func TestUpdateMessageVoiceRejectsMalformedAndUnknown(t *testing.T) { + t.Parallel() + refs := &recordingRefs{} + st := newRefStore(t, refs) + ctx := context.Background() + msg := seedVoiceMessage(t, st, RoleUser, "（语音）") + good := voiceRawWithCopy(t, "0123456789abcdef0123456789abcdef.wav") + if err := st.UpdateMessageVoice(ctx, msg.ID, good); err != nil { + t.Fatalf("先写一条合法值: %v", err) + } + + // 坏形状全部拒绝，且**列保持原值**（绝不落下读不回来的内容）。 + bad := []string{ + "{not json", + `{"source":"asr"}`, // 状态缺席（闭集外） + `{"source":"telepathy","status":"transcribed"}`, // 来源不在闭集 + `{"source":"asr","status":"transcribed","name":"x.ogg"}`, // 有 name 没有 mime/size（同生共死被破） + `{"source":"asr","status":"transcribed","mime":"audio/x"}`, // 有 mime 没有 name + `{"source":"asr","status":"transcribed","durationMs":-1}`, // 负时长 + } + for _, raw := range bad { + err := st.UpdateMessageVoice(ctx, msg.ID, raw) + if !errors.Is(err, voiceref.ErrBadRef) { + t.Fatalf("UpdateMessageVoice(%q) err = %v, want wrap of ErrBadRef", raw, err) + } + } + cols, err := st.VoiceColumns(ctx, []string{msg.ID}) + if err != nil { + t.Fatalf("VoiceColumns: %v", err) + } + if cols[msg.ID] != good { + t.Fatalf("坏写入改动了列：%q, want %q", cols[msg.ID], good) + } + + // 未知 id ⇒ ErrMessageNotFound（绝不假装写成功）。 + if err := st.UpdateMessageVoice(ctx, "no-such-message", good); !errors.Is(err, ErrMessageNotFound) { + t.Fatalf("未知 id err = %v, want ErrMessageNotFound", err) + } +} + +func TestUpdateMessageVoiceSyncsRefsByDifference(t *testing.T) { + t.Parallel() + refs := &recordingRefs{} + st := newRefStore(t, refs) + ctx := context.Background() + msg := seedVoiceMessage(t, st, RoleUser, "（语音）") + first := voiceRawWithCopy(t, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.ogg") + second := voiceRawWithCopy(t, "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.mp3") + + // ① 没有 → 有副本：retain 一次。 + if err := st.UpdateMessageVoice(ctx, msg.ID, first); err != nil { + t.Fatalf("第一次写入: %v", err) + } + // ② 内容逐字不变：一次都不增减（重复计数会让音频永远回收不掉）。 + if err := st.UpdateMessageVoice(ctx, msg.ID, first); err != nil { + t.Fatalf("同值重写: %v", err) + } + if len(refs.retained) != 1 || len(refs.released) != 0 { + t.Fatalf("同值重写后 retain/release = %v/%v, want 1/0", refs.retained, refs.released) + } + // ③ 换成另一份副本：旧的释放、新的保留（各一次）。 + if err := st.UpdateMessageVoice(ctx, msg.ID, second); err != nil { + t.Fatalf("换副本: %v", err) + } + if len(refs.retained) != 2 || refs.retained[1] != "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.mp3" { + t.Fatalf("换副本后 retain = %v", refs.retained) + } + if len(refs.released) != 1 || refs.released[0] != "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.ogg" { + t.Fatalf("换副本后 release = %v", refs.released) + } + // ④ 有副本 → 无副本（微信那条路）：release 一次，不新增 retain。 + if err := st.UpdateMessageVoice(ctx, msg.ID, voiceRawWithoutCopy(t)); err != nil { + t.Fatalf("写成无副本: %v", err) + } + if len(refs.retained) != 2 { + t.Fatalf("无副本写法不该 retain：%v", refs.retained) + } + if len(refs.released) != 2 || refs.released[1] != "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.mp3" { + t.Fatalf("无副本写法 release = %v", refs.released) + } +} + +func TestVoiceColumnsBatchesAndSkipsBlanks(t *testing.T) { + t.Parallel() + st := newRefStore(t, &recordingRefs{}) + ctx := context.Background() + withVoice := seedVoiceMessage(t, st, RoleUser, "语音") + without := seedVoiceMessage(t, st, RoleAssistant, "文本") + raw := voiceRawWithCopy(t, "cccccccccccccccccccccccccccccccc.webm") + if err := st.UpdateMessageVoice(ctx, withVoice.ID, raw); err != nil { + t.Fatalf("UpdateMessageVoice: %v", err) + } + + cols, err := st.VoiceColumns(ctx, []string{withVoice.ID, without.ID, "", withVoice.ID}) + if err != nil { + t.Fatalf("VoiceColumns: %v", err) + } + if len(cols) != 2 { + t.Fatalf("映射大小 = %d, want 2", len(cols)) + } + if cols[withVoice.ID] != raw { + t.Fatalf("语音行 = %q, want %q", cols[withVoice.ID], raw) + } + if got, ok := cols[without.ID]; !ok || got != "" { + t.Fatalf("文本行 = (%q, %v), want (\"\", true)", got, ok) + } + if empty, err := st.VoiceColumns(ctx, nil); err != nil || len(empty) != 0 { + t.Fatalf("空 id 列表 ⇒ 空映射且无错误，得到 (%v, %v)", empty, err) + } +} diff --git a/internal/runtime/voice/media.go b/internal/runtime/voice/media.go new file mode 100644 index 0000000..06ab28b --- /dev/null +++ b/internal/runtime/voice/media.go @@ -0,0 +1,118 @@ +package voice + +// 本文件承载**回放端点**（`GET /api/v1/media/{name}`）两侧的格式事实 +// （design/24 §10.11 的端点表；契约 v0.67 已登记形状）：**名字形态** +// （32 位小写十六进制主名 + 音频白名单扩展名）与**下发 MIME**。 +// +// 两条口径与 design/24 §10.11 逐字一致： +// - 下发 `Content-Type` **只看存储名的扩展名**（`.wav → audio/wav` …），**不查库、 +// 不重新嗅探** —— 「扩展名 ↔ 字节」在**保存那一刻**已成对裁决过（`documents.SaveAudio`）； +// - 名字形态是 deny-by-default：主名必须是 32 位小写十六进制、扩展名必须在 10 项 +// 音频白名单内，其余（含路径穿越的一切形态）一律拒绝，由调用方收敛成同一个 404。 + +import ( + "path/filepath" + "sort" + "strings" +) + +const ( + // MediaPathPrefix 是语音副本对外的**服务端相对路径**前缀（契约 v0.67 §4.4）： + // DTO 里的 `voice.url` 由服务端**一处**拼出（`MediaPath`），客户端绝不自己拼。 + MediaPathPrefix = "/api/v1/media/" + + // StoredNameHexLen 是内容寻址落盘名的十六进制主名长度（32 位 = 128 位）。 + // + // 它与 `documents.HashHexLen` 是**同一个数**（同一套内容寻址）。这里刻意独立具名： + // `documents` 已经 import 本包（`documents → history → voice` 的方向），本包一旦 + // import `documents` 就会成环。两处必须同值，由网关侧的交叉断言钉住 + // （`media_test.go` 的 TestMediaStoredNameHexLenMatchesDocuments）。 + StoredNameHexLen = 32 +) + +// mediaMIMEByExt 是音频白名单**唯一**的一份表：扩展名（小写、含点）→ 下发 MIME。 +// 白名单成员由它派生（`MediaExtAllowed`），因此「能下发」与「下发的类型」不可能分叉。 +// +// 10 项 = design/24 §10.11 的端点表定稿值：`.wav` `.mp3` `.aac` `.ogg` `.oga` `.opus` +// `.m4a` `.webm` `.flac` `.amr`。取值与 design/24 §10.11 的映射表逐字对应 +// （`.ogg`/`.oga`/`.opus` 一律 `audio/ogg` —— 三者是同一容器的不同封装名）。 +var mediaMIMEByExt = map[string]string{ + ".wav": "audio/wav", + ".mp3": "audio/mpeg", + ".aac": "audio/aac", + ".ogg": "audio/ogg", + ".oga": "audio/ogg", + ".opus": "audio/ogg", + ".m4a": "audio/mp4", + ".webm": "audio/webm", + ".flac": "audio/flac", + ".amr": "audio/amr", +} + +// MediaExtAllowed 报告扩展名（含点，大小写不敏感）是否在音频白名单内。 +// deny-by-default：不认识的扩展名一律 false。 +func MediaExtAllowed(ext string) bool { + return mediaMIMEByExt[normalizeExt(ext)] != "" +} + +// MediaMIME 给出该扩展名要下发的 MIME。扩展名不在白名单内时返回 ""（调用方据此 +// 拒绝，绝不回一个「看起来像音频」的兜底类型）。 +func MediaMIME(ext string) string { + return mediaMIMEByExt[normalizeExt(ext)] +} + +// MediaPath 拼出存储名对应的**服务端相对路径**（`/api/v1/media/`）。它是 +// DTO `voice.url` 的唯一派生点：客户端绝不自己拼路径（契约 v0.67 §4.4）。 +func MediaPath(name string) string { + return MediaPathPrefix + name +} + +// ValidMediaName 报告 name 是否恰好是「32 位小写十六进制主名 + 音频白名单扩展名」。 +// 它与网关文档读侧的 `validDocumentName` 同一手法（双检）：分隔符、点、双点及任何 +// 其他形态都在触碰文件系统之前被拒绝 —— 调用方因此可以把「名字非法」与「文件不存在」 +// 收敛成同一个 404，不泄露任何「存在与否」以外的信息。 +func ValidMediaName(name string) bool { + if name == "" || name != filepath.Base(name) || strings.ContainsAny(name, "/\\") { + return false + } + stem, ext, ok := strings.Cut(name, ".") + if !ok || len(stem) != StoredNameHexLen || ext == "" { + return false + } + for _, c := range stem { + if !isLowerHex(c) { + return false + } + } + return MediaExtAllowed("." + strings.ToLower(ext)) +} + +// MediaExtensions 返回音频白名单的扩展名（含点，已排序）。它是**发现**用出口 +// （面板/日志/测试能列出「当前认定哪些容器可回放」），调用方不得依赖它在 +// 业务判定上的顺序 —— 判定一律走 MediaExtAllowed。 +func MediaExtensions() []string { + out := make([]string, 0, len(mediaMIMEByExt)) + for ext := range mediaMIMEByExt { + out = append(out, ext) + } + sort.Strings(out) + return out +} + +// normalizeExt 把扩展名归一成「小写、带前导点」。归一只有这一处，使 +// `.OGG` / `OGG` / `ogg` 三种写法落到同一个判定上（读侧与写侧同源）。 +func normalizeExt(ext string) string { + e := strings.ToLower(strings.TrimSpace(ext)) + if e == "" { + return "" + } + if !strings.HasPrefix(e, ".") { + e = "." + e + } + return e +} + +// isLowerHex 报告 c 是否是 `0-9a-f`。主名形态据此拒绝大写十六进制与其他一切字符。 +func isLowerHex(c rune) bool { + return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') +} diff --git a/internal/runtime/voice/media_test.go b/internal/runtime/voice/media_test.go new file mode 100644 index 0000000..a29a662 --- /dev/null +++ b/internal/runtime/voice/media_test.go @@ -0,0 +1,126 @@ +package voice + +// 本文件锁定**回放端点**那一侧的格式事实（design/24 §10.11；契约 v0.67 的 +// `GET /api/v1/media/{name}` 表）：白名单**字面**成员、扩展名归一、名字形态的 +// deny-by-default，以及唯一派生字段 `url` 的拼法。 +// +// 白名单用**字面量**钉住（而不是从被测表反推）：表改成员时本用例必须红 —— +// 与「拿被测常量当期望」的教训同一条纪律。 + +import ( + "reflect" + "testing" +) + +// TestMediaExtensionsAreTheDeclaredTen 钉住白名单成员与项数（10 项，逐字）。 +func TestMediaExtensionsAreTheDeclaredTen(t *testing.T) { + t.Parallel() + want := []string{ + ".aac", ".amr", ".flac", ".m4a", ".mp3", + ".oga", ".ogg", ".opus", ".wav", ".webm", + } + if got := MediaExtensions(); !reflect.DeepEqual(got, want) { + t.Fatalf("MediaExtensions() = %v, want %v", got, want) + } +} + +// TestMediaExtAllowedNormalizesAndDeniesByDefault 覆盖归一与默认拒。 +func TestMediaExtAllowedNormalizesAndDeniesByDefault(t *testing.T) { + t.Parallel() + allowed := []string{".ogg", "ogg", ".OGG", " .Opus ", ".wav", ".amr", ".webm"} + for _, ext := range allowed { + if !MediaExtAllowed(ext) { + t.Fatalf("MediaExtAllowed(%q) = false, want true", ext) + } + } + denied := []string{"", ".", ".txt", ".mp4", ".pdf", ".ogg2", "ogs", ".OGX"} + for _, ext := range denied { + if MediaExtAllowed(ext) { + t.Fatalf("MediaExtAllowed(%q) = true, want false（deny-by-default）", ext) + } + } +} + +// TestMediaMIME 逐项钉住下发类型（契约 v0.67 §4.1 的映射表）。 +func TestMediaMIME(t *testing.T) { + t.Parallel() + cases := map[string]string{ + ".wav": "audio/wav", + ".mp3": "audio/mpeg", + ".aac": "audio/aac", + ".ogg": "audio/ogg", + ".oga": "audio/ogg", + ".opus": "audio/ogg", + ".m4a": "audio/mp4", + ".webm": "audio/webm", + ".flac": "audio/flac", + ".amr": "audio/amr", + } + for ext, want := range cases { + if got := MediaMIME(ext); got != want { + t.Fatalf("MediaMIME(%q) = %q, want %q", ext, got, want) + } + if got := MediaMIME("." + ext[1:]); got != want { + t.Fatalf("MediaMIME(无点 %q) = %q, want %q", ext, got, want) + } + } + // 未知扩展名绝不回一个「看起来像音频」的兜底类型。 + if got := MediaMIME(".txt"); got != "" { + t.Fatalf("MediaMIME(.txt) = %q, want 空串", got) + } +} + +// TestMediaPath 钉住唯一派生字段的拼法（服务端一处拼出，客户端绝不自己拼）。 +func TestMediaPath(t *testing.T) { + t.Parallel() + name := "0123456789abcdef0123456789abcdef.ogg" + if got, want := MediaPath(name), "/api/v1/media/"+name; got != want { + t.Fatalf("MediaPath = %q, want %q", got, want) + } + if MediaPathPrefix != "/api/v1/media/" { + t.Fatalf("MediaPathPrefix = %q, want /api/v1/media/", MediaPathPrefix) + } +} + +// TestStoredNameHexLen 钉住主名宽度（与 documents 的同名常量同值由网关侧交叉断言）。 +func TestStoredNameHexLen(t *testing.T) { + t.Parallel() + if StoredNameHexLen != 32 { + t.Fatalf("StoredNameHexLen = %d, want 32", StoredNameHexLen) + } +} + +// TestValidMediaName 覆盖名字形态的放行与拒绝（含路径穿越的一切常见形态）。 +func TestValidMediaName(t *testing.T) { + t.Parallel() + valid := []string{ + "0123456789abcdef0123456789abcdef.ogg", + "ffffffffffffffffffffffffffffffff.amr", + "0123456789abcdef0123456789abcdef.webm", + } + for _, name := range valid { + if !ValidMediaName(name) { + t.Fatalf("ValidMediaName(%q) = false, want true", name) + } + } + invalid := []string{ + "", + "0123456789ABCDEF0123456789abcdef.ogg", // 大写十六进制 + "0123456789abcdef0123456789abcde.ogg", // 主名不足 32 位 + "0123456789abcdef0123456789abcdefg.ogg", // 主名超 32 位 + "0123456789abcdef0123456789abcdef", // 没有扩展名 + "0123456789abcdef0123456789abcdef.xml", // 扩展名不在白名单 + "0123456789abcdef0123456789abcdef.exe", // 同上 + "0123456789abcdef0123456789abcdef.o.ogg", // 多一个点 + "../0123456789abcdef0123456789abcdef.ogg", + "..\\0123456789abcdef0123456789abcdef.ogg", + "/etc/passwd", + "sub/0123456789abcdef0123456789abcdef.ogg", + "voice.ogg", + } + for _, name := range invalid { + if ValidMediaName(name) { + t.Fatalf("ValidMediaName(%q) = true, want false", name) + } + } +} diff --git a/internal/services/gateway/admin_memory_review.go b/internal/services/gateway/admin_memory_review.go new file mode 100644 index 0000000..3be3ff6 --- /dev/null +++ b/internal/services/gateway/admin_memory_review.go @@ -0,0 +1,212 @@ +package gateway + +// 本文件承载**入库前审查（D3）的管理面**（design/23 §15 D3-8「面板可查」+ §8「回滚」； +// 切片 MT-B 已把能力做进 `internal/runtime/memory`，本片补 HTTP 入口）。 +// +// 两条端点，分别对应 D3 留痕的**看**与**撤销**： +// +// - `GET /api/v1/admin/memories/reviews?limit=` —— 最近的审查留痕（谁被审、四态 +// 判定、依据、合并去向、是否降级、时间）。留痕表 `memory_ingest_reviews` **不含 +// 记忆正文**（列清单见 preinsert_schema.go），命中的既有 id 足以让面板回表取原文； +// - `POST /api/v1/admin/memories/{id}/revert` —— 撤销一次**软合并**（把被吸收的行 +// 复原为 active 并清空 merged_into，同时把它的 id 从存活行的 merged_from 里摘掉）。 +// 行永不物理删除，因此这是**无损可逆**操作（design/23 §8）。 +// +// 门与审计：两条都走 `writeAuth`（owner|admin），与同族的 `GET /admin/memories` 同档 —— +// 留痕带 itemId / personaId / projectId / sessionId，是**私有记忆面**的一部分， +// design/23 §11 对同族读侧的既有取向是「含隐私正文的读侧不给 viewer，类比 +// GET /admin/memories」；`viewer` 的定位是只读监控投影（design/11 §7），不是记忆面。 +// 撤销是**写**，因此落审计（动作名 `memory.review.revert`，who/result 与既有口径一致）。 +// +// ⚠️ 与 design/23 §11 表格的一处口径差异：那张表把同族的 `candidates/{id}/revert` +// 标为 owner-only。本片按任务书「两条都 writeAuth」实现 —— 若以 §11 为准，把 +// routes.go 里那两行的 `writeAuth` 换成 `ownerAuth` 即可（一行一处，本文件不用改）。 + +import ( + "context" + "errors" + "net/http" + "strconv" + "time" + + apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" +) + +// ingestReviewStore 是入库前审查留痕的读写视图（design/23 §15 D3-8 + §8）。 +// 生产实现是 `*memory.Store`（`ListIngestReviews` / `RevertMerged`）。 +// +// 与 voiceColumnReader 同一手法的可选接口：未实现它的存储（未被组合根接线的测试替身） +// 得到的是**诚实**的 500 `memory_review_unavailable`，而不是一份空列表 —— +// 「没接线」与「一条留痕都没有」必须能分辨。 +type ingestReviewStore interface { + // ListIngestReviews 返回最近的审查留痕（最新在前，至多 limit 条）。 + ListIngestReviews(ctx context.Context, limit int) ([]memory.IngestReview, error) + // RevertMerged 撤销一次软合并。该行不在 merged 状态时返回 memory.ErrNotFound。 + RevertMerged(ctx context.Context, id string) error +} + +const ( + // memoryReviewDefaultLimit 是列表的默认页大小（未给/非正 limit）。 + memoryReviewDefaultLimit = 100 + // memoryReviewMaxLimit 是列表的硬上限：超出的 limit **钳到**它而不是报错 + // （与渠道事件查询面同一约定）。 + memoryReviewMaxLimit = 500 + // memoryReviewUnavailableCode 是「留痕存储未接线」的线上错误码。两条端点共用： + // 它们背后是同一份存储，未接线时不存在「一条能用一条不能用」。 + memoryReviewUnavailableCode = "memory_review_unavailable" + // memoryReviewRevertAction 是撤销动作的审计动作名（design/23 §16.4 的 `memory.*` + // 族；`memory.review.ingest` 是审查自身的动作名，本动作名不与之冲突）。 + memoryReviewRevertAction = "memory.review.revert" +) + +// AdminIngestReview 是一条审查留痕的线上结构（键名 camelCase，时间 UTC RFC3339， +// 与同族的管理 DTO —— 渠道事件行 —— 同一风格）。**不含记忆正文**。 +type AdminIngestReview struct { + // ID 是留痕自身的 id（面板据此做去重/翻页键）。 + ID string `json:"id"` + // ItemID 是被审的那条记忆（命中的既有 id 由调用方回表取原文）。 + ItemID string `json:"itemId"` + // PersonaID / Scope / ProjectID 是这次审查发生的**分区**（§15 D3-7 只在同一分区内判定）。 + PersonaID string `json:"personaId"` + Scope string `json:"scope"` + ProjectID string `json:"projectId,omitempty"` + // Decision ∈ skip | merge | modify | create（四态闭集）。 + Decision string `json:"decision"` + // Reason 是判定依据（规则名，如完全等价/真包含/无相似）。 + Reason string `json:"reason,omitempty"` + // Veto 是命中的否决规则；`merge`/`modify` 时为空。它是**结构化依据**， + // 面板据此聚合「为什么这一条没被合并」，而不是去解析 Reason 文案。 + Veto string `json:"veto,omitempty"` + // TargetID 是这次判定命中的**既有条目** id（create 时为空）。 + TargetID string `json:"targetId,omitempty"` + // MergedIDs 是被这次判定吸收（软合并）的行 id 列表；恒为非 null 数组。 + MergedIDs []string `json:"mergedIds"` + // Score 是候选召回命中的相似度；RecallMode 是这次召回的路径（语义/词项）。 + Score float64 `json:"score"` + RecallMode string `json:"recallMode,omitempty"` + // Degraded 报告这次判定是否**降级**完成（embedder 不可用 ⇒ 回落到词项余弦）。 + Degraded bool `json:"degraded"` + // RuleVersion 是判定规则集版本（改规则必须同时升它，否则新旧留痕无法区分）。 + RuleVersion string `json:"ruleVersion"` + // SessionID 是完成这次审查的记忆管理会话（独立的 sess-memorytidy，不碰主会话）。 + SessionID string `json:"sessionId,omitempty"` + // CreatedAt 是审查时刻（UTC RFC3339）。 + CreatedAt string `json:"createdAt"` +} + +// adminMemoryReviews 处理 GET /api/v1/admin/memories/reviews。它只读、不审计 +// （与同族只读管理端点同一档）。 +func (s *Server) adminMemoryReviews(w http.ResponseWriter, r *http.Request) { + store := s.ingestReviews() + if store == nil { + writeError(w, apperr.Wrap(apperr.ErrInternal, memoryReviewUnavailableCode, "入库前审查留痕未接线")) + return + } + limit, ok := parseMemoryReviewLimit(r) + if !ok { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid limit")) + return + } + reviews, err := store.ListIngestReviews(r.Context(), limit) + if err != nil { + s.log.Error("list ingest reviews", "err", err) + writeError(w, apperr.Wrap(apperr.ErrInternal, "reviews_failed", "读取审查留痕失败")) + return + } + out := make([]AdminIngestReview, 0, len(reviews)) + for _, rev := range reviews { + out = append(out, adminIngestReview(rev)) + } + writeJSON(w, http.StatusOK, out) +} + +// adminMemoryRevert 处理 POST /api/v1/admin/memories/{id}/revert：撤销一次软合并。 +// +// 三个分支都如实作答：未接线 ⇒ 500；该行不在 merged 状态（id 拼错 / 已被回滚过） +// ⇒ **404 not_found**（绝不假装回滚成功）；成功 ⇒ 200 + 被回滚的 id，并落审计。 +func (s *Server) adminMemoryRevert(w http.ResponseWriter, r *http.Request) { + store := s.ingestReviews() + if store == nil { + s.auditClaims(r, memoryReviewRevertAction, "error") + writeError(w, apperr.Wrap(apperr.ErrInternal, memoryReviewUnavailableCode, "入库前审查留痕未接线")) + return + } + id := r.PathValue("id") + if id == "" { + s.auditClaims(r, memoryReviewRevertAction, "error") + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "memory id is required")) + return + } + if err := store.RevertMerged(r.Context(), id); err != nil { + s.auditClaims(r, memoryReviewRevertAction, "error") + if errors.Is(err, memory.ErrNotFound) { + writeError(w, apperr.Wrap(apperr.ErrNotFound, "not_found", "记忆条目不在被合并状态，无法回滚")) + return + } + s.log.Error("revert merged memory", "err", err, "item", id) + writeError(w, apperr.Wrap(apperr.ErrInternal, "revert_failed", "回滚软合并失败")) + return + } + s.auditClaims(r, memoryReviewRevertAction, "ok") + writeJSON(w, http.StatusOK, map[string]any{"reverted": id}) +} + +// ingestReviews 返回当前接入的审查留痕存储；`Memories` 未实现该视图时为 nil。 +func (s *Server) ingestReviews() ingestReviewStore { + if s.memories == nil { + return nil + } + store, ok := s.memories.(ingestReviewStore) + if !ok { + return nil + } + return store +} + +// parseMemoryReviewLimit 解析并归一 limit：缺省/非正 ⇒ 默认页大小；超硬上限 ⇒ 钳位； +// 非整数 ⇒ ok=false（调用方回 400）。钳位而不是报错，与渠道事件查询面同一约定。 +func parseMemoryReviewLimit(r *http.Request) (int, bool) { + raw := r.URL.Query().Get("limit") + if raw == "" { + return memoryReviewDefaultLimit, true + } + v, err := strconv.Atoi(raw) + if err != nil { + return 0, false + } + if v <= 0 { + return memoryReviewDefaultLimit, true + } + if v > memoryReviewMaxLimit { + return memoryReviewMaxLimit, true + } + return v, true +} + +// adminIngestReview 把一条领域留痕投影成线上结构。MergedIDs 兜成非 null 数组 +// （面板按数组读），时间为 UTC RFC3339（与渠道事件行同口径）。 +func adminIngestReview(rev memory.IngestReview) AdminIngestReview { + ids := rev.MergedIDs + if ids == nil { + ids = []string{} + } + return AdminIngestReview{ + ID: rev.ID, + ItemID: rev.ItemID, + PersonaID: rev.PersonaID, + Scope: string(rev.Scope), + ProjectID: rev.ProjectID, + Decision: string(rev.Decision), + Reason: rev.Reason, + Veto: string(rev.Veto), + TargetID: rev.TargetID, + MergedIDs: ids, + Score: rev.Score, + RecallMode: rev.RecallMode, + Degraded: rev.Degraded, + RuleVersion: rev.RuleVer, + SessionID: rev.SessionID, + CreatedAt: rev.CreatedAt.UTC().Format(time.RFC3339), + } +} diff --git a/internal/services/gateway/admin_memory_review_test.go b/internal/services/gateway/admin_memory_review_test.go new file mode 100644 index 0000000..fe87922 --- /dev/null +++ b/internal/services/gateway/admin_memory_review_test.go @@ -0,0 +1,250 @@ +package gateway + +// 本文件锁定**入库前审查（D3）管理面**的两条端点（design/23 §15 D3-8 + §8；切片 L45）： +// ① 留痕列表的投影（camelCase、时间 UTC RFC3339、`mergedIds` 恒为非 null 数组）； +// ② `limit` 的默认值 / 钳位 / 非法值（400）； +// ③ 门与审计：owner|admin 200、viewer 403、匿名 401；回滚**成功与失败都落审计**； +// ④ 未接线时诚实作答 500 `memory_review_unavailable`（**绝不回空列表**假装「没有留痕」）； +// ⑤ 回滚的三种结果：成功 200、不在被合并状态 404 `not_found`、存储报错 500。 + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" +) + +// fakeIngestReviews 是审查留痕存储的替身：既满足 Deps.Memories（MemoryStore）， +// 又实现管理面要的窄视图（ingestReviewStore）。 +type fakeIngestReviews struct { + reviews []memory.IngestReview + listErr error + revertErr error + limitGot int + reverted []string +} + +func (f *fakeIngestReviews) List(context.Context) ([]memory.MemoryItem, error) { + return []memory.MemoryItem{{ID: "m1", Scope: memory.ScopeUser, Content: "一条记忆"}}, nil +} + +func (f *fakeIngestReviews) ListIngestReviews(_ context.Context, limit int) ([]memory.IngestReview, error) { + f.limitGot = limit + return f.reviews, f.listErr +} + +func (f *fakeIngestReviews) RevertMerged(_ context.Context, id string) error { + if f.revertErr != nil { + return f.revertErr + } + f.reverted = append(f.reverted, id) + return nil +} + +// listOnlyMemories 只实现 Deps.Memories 的最低要求（没有留痕视图的存储）。 +type listOnlyMemories struct{} + +func (listOnlyMemories) List(context.Context) ([]memory.MemoryItem, error) { + return nil, nil +} + +// recordingAudit 捕获审计链上的每一次 Append（谁/做了什么/结果）。 +type recordingAudit struct{ entries []AuditRecord } + +func (a *recordingAudit) Append(_ context.Context, who, what, result string, ts time.Time) error { + a.entries = append(a.entries, AuditRecord{Who: who, What: what, When: ts.Format(time.RFC3339), Result: result}) + return nil +} + +func (a *recordingAudit) List(context.Context, int) ([]AuditRecord, error) { return a.entries, nil } + +// fixtureIngestReview 是一条四态留痕（merge：命中了既有条目并把新行软合并进去）。 +func fixtureIngestReview() memory.IngestReview { + return memory.IngestReview{ + ID: "rev-1", + ItemID: "item-new", + PersonaID: "yuki", + Scope: memory.ScopeUser, + ProjectID: "proj-1", + Decision: memory.IngestMerge, + Reason: "subset_of_existing", + TargetID: "item-existing", + MergedIDs: []string{"item-new"}, + Score: 0.91, + RecallMode: "vector", + Degraded: false, + RuleVer: "d3.1", + SessionID: "sess-memorytidy", + CreatedAt: time.Date(2026, 10, 6, 3, 4, 5, 0, time.UTC), + } +} + +func TestAdminMemoryReviewsListProjection(t *testing.T) { + t.Parallel() + fake := &fakeIngestReviews{reviews: []memory.IngestReview{fixtureIngestReview()}} + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Memories: fake}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/memories/reviews", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + if fake.limitGot != memoryReviewDefaultLimit { + t.Fatalf("默认 limit = %d, want %d", fake.limitGot, memoryReviewDefaultLimit) + } + var rows []map[string]json.RawMessage + if err := json.Unmarshal(decodeBody(t, body)["data"], &rows); err != nil { + t.Fatalf("解析 data: %v (body=%s)", err, body) + } + if len(rows) != 1 { + t.Fatalf("留痕条数 = %d, want 1", len(rows)) + } + want := map[string]string{ + "id": `"rev-1"`, + "itemId": `"item-new"`, + "personaId": `"yuki"`, + "scope": `"user"`, + "projectId": `"proj-1"`, + "decision": `"merge"`, + "reason": `"subset_of_existing"`, + "targetId": `"item-existing"`, + "score": `0.91`, + "recallMode": `"vector"`, + "degraded": `false`, + "ruleVersion": `"d3.1"`, + "sessionId": `"sess-memorytidy"`, + "createdAt": `"2026-10-06T03:04:05Z"`, + } + for key, val := range want { + raw, ok := rows[0][key] + if !ok { + t.Fatalf("缺键 %q（row=%v）", key, rows[0]) + } + if string(raw) != val { + t.Fatalf("%s = %s, want %s", key, raw, val) + } + } + if string(rows[0]["mergedIds"]) != `["item-new"]` { + t.Fatalf("mergedIds = %s, want [\"item-new\"]", rows[0]["mergedIds"]) + } + // 无否决规则时不发 veto 键（omitempty）。 + if raw, ok := rows[0]["veto"]; ok { + t.Fatalf("未命中否决规则时不该有 veto 键：%s", raw) + } +} + +func TestAdminMemoryReviewsLimitAndEmpty(t *testing.T) { + t.Parallel() + fake := &fakeIngestReviews{} + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Memories: fake}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/memories/reviews?limit=3", nil, tok) + if resp.StatusCode != http.StatusOK || fake.limitGot != 3 { + t.Fatalf("limit=3 ⇒ status/limit = %d/%d, want 200/3 (body=%s)", resp.StatusCode, fake.limitGot, body) + } + // 超硬上限 ⇒ 钳位（不报错，与渠道事件查询面同一约定）。 + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/memories/reviews?limit=9999", nil, tok) + if resp.StatusCode != http.StatusOK || fake.limitGot != memoryReviewMaxLimit { + t.Fatalf("limit=9999 ⇒ status/limit = %d/%d, want 200/%d (body=%s)", + resp.StatusCode, fake.limitGot, memoryReviewMaxLimit, body) + } + // 非正 ⇒ 默认。 + resp, _ = doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/memories/reviews?limit=0", nil, tok) + if resp.StatusCode != http.StatusOK || fake.limitGot != memoryReviewDefaultLimit { + t.Fatalf("limit=0 ⇒ status/limit = %d/%d, want 200/%d", resp.StatusCode, fake.limitGot, memoryReviewDefaultLimit) + } + // 非整数 ⇒ 400 bad_request。 + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/memories/reviews?limit=abc", nil, tok) + if resp.StatusCode != http.StatusBadRequest || mediaErrorCode(t, body) != "bad_request" { + t.Fatalf("limit=abc ⇒ %d/%s, want 400/bad_request", resp.StatusCode, body) + } + // 空留痕 ⇒ 200 + 非 null 空数组（「算过了，这一条都没有」与「没接线」必须可区分）。 + resp, body = doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/memories/reviews", nil, tok) + if resp.StatusCode != http.StatusOK || string(decodeBody(t, body)["data"]) != "[]" { + t.Fatalf("空留痕 ⇒ %d/%s, want 200/[]", resp.StatusCode, body) + } +} + +func TestAdminMemoryReviewsRBACAndUnavailable(t *testing.T) { + t.Parallel() + fake := &fakeIngestReviews{} + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Memories: fake}) + + if resp, _ := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/memories/reviews", nil, ""); resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("匿名 status = %d, want 401", resp.StatusCode) + } + viewer := signAny(t, jwt, "acct", auth.RoleViewer) + if resp, _ := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/memories/reviews", nil, viewer); resp.StatusCode != http.StatusForbidden { + t.Fatalf("viewer status = %d, want 403（留痕属私有记忆面）", resp.StatusCode) + } + admin := signAny(t, jwt, "acct", auth.RoleAdmin) + if resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/memories/reviews", nil, admin); resp.StatusCode != http.StatusOK { + t.Fatalf("admin status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + + // 存储没有留痕视图 ⇒ 500 memory_review_unavailable（绝不回空列表）。 + _, bare, bareJWT := buildEnv(t, cfgDefault(), Deps{Memories: listOnlyMemories{}}) + tok := signAny(t, bareJWT, "acct", auth.RoleOwner) + resp, body := doReq(t, http.MethodGet, bare.URL+"/api/v1/admin/memories/reviews", nil, tok) + if resp.StatusCode != http.StatusInternalServerError || mediaErrorCode(t, body) != memoryReviewUnavailableCode { + t.Fatalf("未接线 ⇒ %d/%s, want 500/%s", resp.StatusCode, body, memoryReviewUnavailableCode) + } +} + +func TestAdminMemoryRevertOutcomesAndAudit(t *testing.T) { + t.Parallel() + fake := &fakeIngestReviews{} + audit := &recordingAudit{} + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Memories: fake, AuditStore: audit}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // 成功：200 + 被回滚的 id，并落一条 ok 审计。 + resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/memories/item-new/revert", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("revert status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + if len(fake.reverted) != 1 || fake.reverted[0] != "item-new" { + t.Fatalf("回滚调用 = %v, want [item-new]", fake.reverted) + } + var payload map[string]string + if err := json.Unmarshal(decodeBody(t, body)["data"], &payload); err != nil { + t.Fatalf("解析 data: %v", err) + } + if payload["reverted"] != "item-new" { + t.Fatalf("data.reverted = %q, want item-new", payload["reverted"]) + } + if len(audit.entries) != 1 || audit.entries[0].What != memoryReviewRevertAction || audit.entries[0].Result != "ok" { + t.Fatalf("成功后的审计 = %+v, want 一条 %s/ok", audit.entries, memoryReviewRevertAction) + } + + // 该行不在被合并状态 ⇒ 404 not_found，且**失败也落审计**。 + fake.revertErr = memory.ErrNotFound + resp, body = doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/memories/plain-item/revert", nil, tok) + if resp.StatusCode != http.StatusNotFound || mediaErrorCode(t, body) != "not_found" { + t.Fatalf("非合并行 ⇒ %d/%s, want 404/not_found", resp.StatusCode, body) + } + if len(audit.entries) != 2 || audit.entries[1].Result != "error" { + t.Fatalf("失败后的审计 = %+v, want 一条 error", audit.entries) + } + + // 存储报错 ⇒ 500 revert_failed（不谎报成功）。 + fake.revertErr = errors.New("db exploded") + resp, body = doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/memories/item-new/revert", nil, tok) + if resp.StatusCode != http.StatusInternalServerError || mediaErrorCode(t, body) != "revert_failed" { + t.Fatalf("存储报错 ⇒ %d/%s, want 500/revert_failed", resp.StatusCode, body) + } + + // 未接线 ⇒ 500 memory_review_unavailable。 + _, bare, bareJWT := buildEnv(t, cfgDefault(), Deps{Memories: listOnlyMemories{}}) + btok := signAny(t, bareJWT, "acct", auth.RoleOwner) + resp, body = doReq(t, http.MethodPost, bare.URL+"/api/v1/admin/memories/item-new/revert", nil, btok) + if resp.StatusCode != http.StatusInternalServerError || mediaErrorCode(t, body) != memoryReviewUnavailableCode { + t.Fatalf("未接线 revert ⇒ %d/%s, want 500/%s", resp.StatusCode, body, memoryReviewUnavailableCode) + } +} diff --git a/internal/services/gateway/history.go b/internal/services/gateway/history.go index 26efba6..eb1835c 100644 --- a/internal/services/gateway/history.go +++ b/internal/services/gateway/history.go @@ -119,7 +119,10 @@ type historyMessage struct { // 它**只由读接口填充**（`handleListMessages`）：其余共用本结构的出口 // （`message_persisted` / `message_edited` / 握手回放 / 渠道日志）逐字不变。 ToolCalls []historyToolCall `json:"toolCalls,omitempty"` - CreatedAt int64 `json:"createdAt"` + // Voice 是这条消息的**语音面**（契约 v0.67 §4.4；形状与缺席语义见 history_voice.go）： + // 非语音消息**整键缺席**（指针 nil，不是 `{}` 也不是 `null`）⇒ **🎤 判据 = 本键存在**。 + Voice *historyVoice `json:"voice,omitempty"` + CreatedAt int64 `json:"createdAt"` } // handleHistorySessions 先返回主会话，再返回每个子会话（最旧 @@ -299,6 +302,8 @@ func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) { // （按页内路标识集合，禁 N+1）。图取不到 ⇒ 字段整体缺席（attachToolCalls 只在 // 真的回看到工具调用时才写），绝不回空数组。 s.attachToolCalls(r.Context(), id, out) + // 语音面（契约 v0.67 §4.4）：同样一次批量读列，列里没有语音事实 ⇒ 整体缺席。 + s.attachVoice(r.Context(), out) writeJSON(w, http.StatusOK, out) } @@ -360,6 +365,8 @@ func (s *Server) handleChannelPlatformLog(w http.ResponseWriter, r *http.Request for _, m := range msgs { out = append(out, toHistoryMessage(m)) } + // 语音面（契约 v0.67 §4.4）：渠道日志页也是读取面，同样补（一次批量）。 + s.attachVoice(r.Context(), out) writeJSON(w, http.StatusOK, out) } @@ -417,6 +424,8 @@ func (s *Server) handleChannelMessages(w http.ResponseWriter, r *http.Request) { for _, m := range msgs { out = append(out, toHistoryMessage(m)) } + // 语音面（契约 v0.67 §4.4）：IM 适配器日志页也是读取面之一，同样补（一次批量）。 + s.attachVoice(r.Context(), out) writeJSON(w, http.StatusOK, out) } func (s *Server) handleDeleteSession(w http.ResponseWriter, r *http.Request) { diff --git a/internal/services/gateway/history_voice.go b/internal/services/gateway/history_voice.go new file mode 100644 index 0000000..300813b --- /dev/null +++ b/internal/services/gateway/history_voice.go @@ -0,0 +1,134 @@ +package gateway + +// 本文件是消息 DTO 的**语音面**（契约 v0.67 §4.4 的 `voice` 对象；design/24 §10.4）。 +// +// 它回答客户端那个问题：「这条消息来自一条语音吗？转写成不成？有没有可回放的副本？」 +// 三个答案都在 `chat_messages.voice` 列里（JSON 单对象，迁移 0020），本文件只做 +// **读侧投影**：解码 → 映射为 DTO → 拼出唯一的派生字段 `url`。它不写列、不改正文， +// 也不判身份（来源是渠道事实，与谁在说话无关 —— 身份永远由 actor 元数据决定）。 +// +// 四条硬口径： +// +// - **🎤 判据 = `voice` 键存在**（契约明写）：非语音消息整键缺席（`omitempty` 的 +// 指针 nil），**不是** `{}` 也不是 `null`。客户端绝不从正文里认 `[语音]` 字样。 +// - **副本四键同生共死**：`name`/`mime`/`size`/`url` 只在**真的有副本**时一起出现 +// （`Ref.Validate` 在解码时已保证，本文件再用 `HasCopy` 复核一次），绝不出现 +// 「`url` 指着一个不存在的 `name`」。 +// - **`url` 只由服务端拼**（`voice.MediaPath`）：客户端绝不自己拼路径。 +// - **坏列可见地缺席**：列被外部改坏（非法 JSON / 闭集破坏）时**不静默当「不是语音」** +// —— 记一条带来源与消息 id（**不含正文**）的 WARN，并让该字段缺席。契约的 `voice` +// 没有「读不出」这一态，因此缺席是唯一诚实的表达（与文档列回一条 `unreadable` +// 不同：文档有那个态，语音没有）。 +// +// 覆盖范围（**如实**）：本文件的补齐点挂在会用 DTO 的读侧出口上（会话消息页、渠道 +// 消息日志两个端点、编辑消息的响应与其广播帧）。**尚未覆盖**：`message_persisted` / +// `channel_persisted`（`broadcast.go`）与握手 `history` 回放（`ws_history.go`）—— +// 前者还额外需要「随 INSERT 写入」的写路径（设计 §10.5 的首选口径），见交付报告。 + +import ( + "context" + "strings" + + voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" +) + +// historyVoice 是消息 DTO 里的 `voice` 对象（契约 v0.67 §4.4 逐字形状）： +// `{source,status,name?,mime?,size?,durationMs?,url?}`。 +// +// `source` / `status` 恒在（属于闭集，见 `voice.Source` / `voice.Status`）；其余 +// 全部 `omitempty` —— 给不出那一项时它**缺席**，绝不用 `0` 或空串编一个值。 +type historyVoice struct { + // Source ∈ asr（服务端识别，含云端）| platform（渠道平台自带转写）。 + Source string `json:"source"` + // Status ∈ transcribed（正文即转写文本）| empty（转写成功但没识别出文字）| + // failed（没拿到转写文本，正文是一行中文诚实提示）。闭集三态。 + Status string `json:"status"` + // Name 是音频副本的存储名；无副本时**整键缺席**（与 mime/size/url 同生共死）。 + Name string `json:"name,omitempty"` + // MIME 是保存时成对裁决所得的下发类型（`audio/ogg` 等）。 + MIME string `json:"mime,omitempty"` + // Size 是副本字节数。 + Size int64 `json:"size,omitempty"` + // DurationMs 只有渠道如实给出、或转码产物可确证时才有；读不出来就缺席，绝不估算。 + DurationMs int64 `json:"durationMs,omitempty"` + // URL 是服务端相对路径 `/api/v1/media/`（唯一派生字段，由 MediaPath 一处拼出）。 + URL string `json:"url,omitempty"` +} + +// voiceColumnReader 是历史存储的**可选读侧视图**：按消息 id 批量取 voice 列原文 +// （一次 `IN (...)`，禁 N+1）。生产实现是 `*history.Store`（`VoiceColumns`）。 +// +// 它刻意是**可选**接口（与 PersonaMoodStore / voiceProviderReporter 同一手法）： +// 未实现它的存储（测试替身、只做基本读写的实现）不会因为缺少这一面而失败 —— +// 这条路径上「没有这一面」的诚实结果是**字段缺席**，而不是 500。 +type voiceColumnReader interface { + VoiceColumns(ctx context.Context, ids []string) (map[string]string, error) +} + +// attachVoice 给一页消息补上各自的语音面（就地写进 DTO 切片）。它只查页内**非空** +// 的消息 id，且**一次批量**（页内 id 集合交给存储侧一条 SQL 查完）。 +// +// 失败处理与工具回看面同一纪律：查不到就只让这一面缺席并记 WARN —— 消息本身已经 +// 读到了，让一个可选列的故障把整页历史变成 500 是反的。 +func (s *Server) attachVoice(ctx context.Context, msgs []historyMessage) { + ids := pageMessageIDs(msgs) + if len(ids) == 0 { + return + } + reader, ok := s.history.(voiceColumnReader) + if !ok { + // 存储未提供这一面（未接线 / 测试替身）：整体缺席，不是错误。 + return + } + cols, err := reader.VoiceColumns(ctx, ids) + if err != nil { + s.log.Warn("history voice unavailable", "err", err, "messages", len(ids)) + return + } + for i := range msgs { + raw, found := cols[msgs[i].ID] + if !found { + continue // 该行取不到（已删除）：缺席 + } + ref, err := voiceref.Decode(raw) + if err != nil { + // 坏列**可见地**缺席：绝不静默当成「不是语音」。日志只带消息 id 与错误， + // 绝不带列内容或正文。 + s.log.Warn("history voice column malformed", "err", err, "message", msgs[i].ID) + continue + } + msgs[i].Voice = historyVoiceOf(ref) + } +} + +// pageMessageIDs 收集页内非空且**互不相同**的消息 id（保持首次出现顺序）。去重让 +// 同一页的「提问 + 回答」或编辑场景里的两份 DTO 共用同一次查询。 +func pageMessageIDs(msgs []historyMessage) []string { + seen := make(map[string]bool, len(msgs)) + out := make([]string, 0, len(msgs)) + for _, m := range msgs { + id := strings.TrimSpace(m.ID) + if id == "" || seen[id] { + continue + } + seen[id] = true + out = append(out, id) + } + return out +} + +// historyVoiceOf 把一条已解码的语音事实投影成 DTO；不是语音消息（闭集外/零值）时 +// 返回 nil，使 `voice` 键整体缺席（契约：非语音消息与引入语音之前逐字相同）。 +func historyVoiceOf(ref voiceref.Ref) *historyVoice { + if !ref.IsVoice() { + return nil + } + out := &historyVoice{Source: string(ref.Source), Status: string(ref.Status)} + if ref.HasCopy() { + // 四键同生共死：有副本时 name/mime/size/url 一起在，没有时一起缺席。 + out.Name, out.MIME, out.Size = ref.Name, ref.MIME, ref.Size + out.URL = voiceref.MediaPath(ref.Name) + } + out.DurationMs = ref.DurationMs + return out +} diff --git a/internal/services/gateway/history_voice_test.go b/internal/services/gateway/history_voice_test.go new file mode 100644 index 0000000..4bf1183 --- /dev/null +++ b/internal/services/gateway/history_voice_test.go @@ -0,0 +1,303 @@ +package gateway + +// 本文件锁定**消息 DTO 的语音面**（契约 v0.67 §4.4；design/24 §10.4）在网关侧的 +// 投影口径： +// ① 有语音事实 ⇒ `voice` 键存在，字段逐一对上，`url` 由服务端拼出； +// ② 非语音消息 ⇒ 键**整体缺席**（不是 `{}` 也不是 `null`）； +// ③ 无副本（微信那条路）⇒ `source`/`status` 在，`name`/`mime`/`size`/`url` **全缺席** +// （四者同生共死）； +// ④ 列缺失 / 列被改坏 ⇒ **字段缺席但正文照常 200**（可选的读取面绝不把整页历史拖成 500）； +// ⑤ 它挂在**多个读取面**上（会话消息页、渠道日志页、编辑消息的响应）。 + +import ( + "context" + "database/sql" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" +) + +// voiceEnv 打包语音面测试所需的依赖：真实（内存）历史存储 + httptest 网关。 +// 与本包既有的 newHistoryEnv 的差别只有一点：它把 `*sql.DB` 也交出来，使「列被外部 +// 改坏」这类场景能直接落 SQL（那是唯一能造出坏列的途径）。 +type voiceEnv struct { + ts *httptest.Server + jwt *auth.JWTService + st *history.Store + conn *sql.DB +} + +func newVoiceEnv(t *testing.T) *voiceEnv { + t.Helper() + ctx := context.Background() + conn, err := db.Open(ctx, ":memory:", testLogger()) + if err != nil { + t.Fatalf("打开内存库: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + st, err := history.NewStore(ctx, conn, testLogger()) + if err != nil { + t.Fatalf("构造历史存储: %v", err) + } + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{History: st}) + return &voiceEnv{ts: ts, jwt: jwt, st: st, conn: conn} +} + +// seedVoiceRow 追加一条消息并（可选）把一条语音事实写进它的 voice 列。 +func (e *voiceEnv) seedVoiceRow(t *testing.T, m history.Message, raw string) history.Message { + t.Helper() + got, err := e.st.AppendFull(context.Background(), m) + if err != nil { + t.Fatalf("落库消息: %v", err) + } + if raw != "" { + if err := e.st.UpdateMessageVoice(context.Background(), got.ID, raw); err != nil { + t.Fatalf("写 voice 列: %v", err) + } + } + return got +} + +// messagePage 是一页消息的**原始 JSON 形态**（`{data:[…]}` 信封里的数组；用于断言 +// 键是否真的缺席 —— 解码进结构体是看不出「键缺席」与「零值」的差别的）。 +func messagePage(t *testing.T, body []byte) []map[string]json.RawMessage { + t.Helper() + var page []map[string]json.RawMessage + if err := json.Unmarshal(decodeBody(t, body)["data"], &page); err != nil { + t.Fatalf("解析消息页: %v (body=%s)", err, body) + } + return page +} + +// decodedVoice 从原始 JSON 里取出 `voice` 对象；键缺席时返回 (nil, false)。 +func decodedVoice(t *testing.T, row map[string]json.RawMessage) (*historyVoice, bool) { + t.Helper() + raw, ok := row["voice"] + if !ok { + return nil, false + } + var v historyVoice + if err := json.Unmarshal(raw, &v); err != nil { + t.Fatalf("解析 voice 对象: %v (raw=%s)", err, raw) + } + return &v, true +} + +func TestHistoryMessageVoiceFaceOnSessionPage(t *testing.T) { + t.Parallel() + env := newVoiceEnv(t) + tok := signAny(t, env.jwt, "acct", auth.RoleOwner) + name := "0123456789abcdef0123456789abcdef.ogg" + raw, err := voiceref.Encode(voiceref.Ref{ + Source: voiceref.SourceASR, + Status: voiceref.StatusTranscribed, + Name: name, + MIME: "audio/ogg", + Size: 4242, + DurationMs: 3200, + }) + if err != nil { + t.Fatalf("编码语音事实: %v", err) + } + voiceMsg := env.seedVoiceRow(t, history.Message{ + SessionID: history.MainSessionID, Role: history.RoleUser, Content: "转写文本", + }, raw) + plain := env.seedVoiceRow(t, history.Message{ + SessionID: history.MainSessionID, Role: history.RoleUser, Content: "普通文本", + }, "") + + resp, body := doReq(t, http.MethodGet, env.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) + } + page := messagePage(t, body) + if len(page) != 2 { + t.Fatalf("消息条数 = %d, want 2", len(page)) + } + byID := make(map[string]map[string]json.RawMessage, len(page)) + for _, row := range page { + var id string + if err := json.Unmarshal(row["id"], &id); err != nil { + t.Fatalf("解析 id: %v", err) + } + byID[id] = row + } + + got, ok := decodedVoice(t, byID[voiceMsg.ID]) + if !ok { + t.Fatalf("语音消息缺 voice 键（body=%s）", body) + } + if got.Source != "asr" || got.Status != "transcribed" { + t.Fatalf("source/status = %q/%q, want asr/transcribed", got.Source, got.Status) + } + if got.Name != name || got.MIME != "audio/ogg" || got.Size != 4242 || got.DurationMs != 3200 { + t.Fatalf("副本字段 = %+v", got) + } + if want := voiceref.MediaPathPrefix + name; got.URL != want { + t.Fatalf("url = %q, want %q（服务端唯一派生字段）", got.URL, want) + } + + // 非语音消息：键必须**真的缺席**（不是 `{}`、不是 `null`）。 + if _, ok := decodedVoice(t, byID[plain.ID]); ok { + t.Fatalf("普通消息带了 voice 键（body=%s）", body) + } +} + +func TestHistoryMessageVoiceWithoutCopyOmitsCopyFields(t *testing.T) { + t.Parallel() + env := newVoiceEnv(t) + tok := signAny(t, env.jwt, "acct", auth.RoleOwner) + // 微信那条路：平台自带转写、按既有裁决不取字节 ⇒ 没有可回放副本。 + raw, err := voiceref.Encode(voiceref.Ref{ + Source: voiceref.SourcePlatform, + Status: voiceref.StatusTranscribed, + }) + if err != nil { + t.Fatalf("编码语音事实: %v", err) + } + msg := env.seedVoiceRow(t, history.Message{ + SessionID: history.MainSessionID, Role: history.RoleUser, Content: "平台转写文本", + }, raw) + + resp, body := doReq(t, http.MethodGet, env.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) + } + page := messagePage(t, body) + if len(page) != 1 { + t.Fatalf("消息条数 = %d, want 1", len(page)) + } + var id string + _ = json.Unmarshal(page[0]["id"], &id) + if id != msg.ID { + t.Fatalf("消息 id = %q, want %q", id, msg.ID) + } + v, ok := decodedVoice(t, page[0]) + if !ok { + t.Fatal("缺 voice 键：没有副本不等于不是语音消息") + } + if v.Source != "platform" || v.Status != "transcribed" { + t.Fatalf("source/status = %q/%q, want platform/transcribed", v.Source, v.Status) + } + var obj map[string]json.RawMessage + if err := json.Unmarshal(page[0]["voice"], &obj); err != nil { + t.Fatalf("解析 voice 对象: %v", err) + } + for _, key := range []string{"name", "mime", "size", "url"} { + if _, has := obj[key]; has { + t.Fatalf("无副本时不该有 %q 键（四者同生共死）", key) + } + } +} + +func TestHistoryMessageVoiceAbsentWhenColumnOrValueBroken(t *testing.T) { + t.Parallel() + ctx := context.Background() + + // ① 列还不存在（迁移 0020 之前建出的表且从未写过语音）：字段缺席、正文照常 200。 + fresh := newVoiceEnv(t) + tok := signAny(t, fresh.jwt, "acct", auth.RoleOwner) + msg := fresh.seedVoiceRow(t, history.Message{ + SessionID: history.MainSessionID, Role: history.RoleUser, Content: "普通文本", + }, "") + resp, body := doReq(t, http.MethodGet, fresh.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) + } + page := messagePage(t, body) + var id, content string + _ = json.Unmarshal(page[0]["id"], &id) + _ = json.Unmarshal(page[0]["content"], &content) + if id != msg.ID || content != "普通文本" { + t.Fatalf("正文行不可用：id=%q content=%q, want %q/普通文本", id, content, msg.ID) + } + if _, ok := decodedVoice(t, page[0]); ok { + t.Fatal("列缺失时不该凭空产出 voice 键") + } + + // ② 列被外部改坏：同样只让该字段缺席，并记一条带来源的 WARN（可观测，不静默）。 + broken := newVoiceEnv(t) + btok := signAny(t, broken.jwt, "acct", auth.RoleOwner) + bmsg := broken.seedVoiceRow(t, history.Message{ + SessionID: history.MainSessionID, Role: history.RoleUser, Content: "坏列消息", + }, "") + valid, err := voiceref.Encode(voiceref.Ref{Source: voiceref.SourceASR, Status: voiceref.StatusEmpty}) + if err != nil { + t.Fatalf("编码: %v", err) + } + // 先用一次合法写入把列前滚出来，再把它改成坏内容。 + if err := broken.st.UpdateMessageVoice(ctx, bmsg.ID, valid); err != nil { + t.Fatalf("前滚 voice 列: %v", err) + } + if _, err := broken.conn.ExecContext(ctx, + `UPDATE chat_messages SET voice = ? WHERE id = ?`, "{not json", bmsg.ID); err != nil { + t.Fatalf("写坏列: %v", err) + } + resp, body = doReq(t, http.MethodGet, broken.ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, btok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("坏列时 status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + brokenPage := messagePage(t, body) + var brokenContent string + _ = json.Unmarshal(brokenPage[0]["content"], &brokenContent) + if brokenContent != "坏列消息" { + t.Fatalf("正文被坏列影响：%q", brokenContent) + } + if _, ok := decodedVoice(t, brokenPage[0]); ok { + t.Fatal("坏列不该被静默解释成一个语音事实") + } +} + +func TestHistoryMessageVoiceOnChannelLogAndEditResponse(t *testing.T) { + t.Parallel() + env := newVoiceEnv(t) + tok := signAny(t, env.jwt, "acct", auth.RoleOwner) + name := "ffffffffffffffffffffffffffffffff.mp3" + raw, err := voiceref.Encode(voiceref.Ref{ + Source: voiceref.SourceASR, Status: voiceref.StatusTranscribed, + Name: name, MIME: "audio/mpeg", Size: 99, + }) + if err != nil { + t.Fatalf("编码语音事实: %v", err) + } + adapter := "qq-obv11:10001" + msg := env.seedVoiceRow(t, history.Message{ + SessionID: history.MainSessionID, Role: history.RoleUser, Content: "IM 语音", + Platform: "qq-obv11", AdapterID: adapter, + }, raw) + + // 渠道日志页（管理面读取面之一）同样带 voice。 + resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/channels/messages?adapter="+adapter, nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("渠道日志 status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + page := messagePage(t, body) + if len(page) != 1 { + t.Fatalf("渠道日志条数 = %d, want 1", len(page)) + } + v, ok := decodedVoice(t, page[0]) + if !ok || v.Name != name { + t.Fatalf("渠道日志里的 voice = %+v (ok=%v)", v, ok) + } + + // 编辑消息的响应同样带 voice（同一次读取面口径）。 + resp, body = doReq(t, http.MethodPatch, env.ts.URL+"/api/v1/messages/"+msg.ID, + map[string]any{"content": "改过的正文"}, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("PATCH status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var patched map[string]json.RawMessage + if err := json.Unmarshal(decodeBody(t, body)["data"], &patched); err != nil { + t.Fatalf("解析 PATCH data: %v", err) + } + pv, ok := decodedVoice(t, patched) + if !ok || pv.Name != name { + t.Fatalf("PATCH 响应里的 voice = %+v (ok=%v)", pv, ok) + } +} diff --git a/internal/services/gateway/history_write.go b/internal/services/gateway/history_write.go index d46df7e..f71cc78 100644 --- a/internal/services/gateway/history_write.go +++ b/internal/services/gateway/history_write.go @@ -177,10 +177,15 @@ func (s *Server) handlePatchMessage(w http.ResponseWriter, r *http.Request) { // 气泡（message_persisted 实时同步契约，D3 编辑 // 路径）。thinking 会从帧中剥掉，以匹配持久化/重放 // 契约；REST 回复保留它，因为消息路由仅 owner/admin。 - edited := toHistoryMessage(msg) - edited.Thinking = "" - s.hub.broadcastMessageEdited(edited) - writeJSON(w, http.StatusOK, toHistoryMessage(msg)) + // + // 语音面（契约 v0.67 §4.4；切片 L45）：响应与广播帧**都要**带 voice（编辑不改变 + // 语音事实，但读取面必须一致）。两条 DTO 共用**一次**批量读列 —— attachVoice 按 + // id 去重，同一条消息只查一次。 + pair := []historyMessage{toHistoryMessage(msg), toHistoryMessage(msg)} + pair[0].Thinking = "" + s.attachVoice(r.Context(), pair) + s.hub.broadcastMessageEdited(pair[0]) + writeJSON(w, http.StatusOK, pair[1]) } // handleDeleteMessage 移除单条消息（D3）。未知 id 返回 404， diff --git a/internal/services/gateway/media.go b/internal/services/gateway/media.go new file mode 100644 index 0000000..a743675 --- /dev/null +++ b/internal/services/gateway/media.go @@ -0,0 +1,108 @@ +package gateway + +// 本文件承载**语音副本的回放端点**：`GET /api/v1/media/{name}`（design/24 §10.11； +// 契约 v0.67 已登记形状，本片落地）。 +// +// 它与文档读侧（`GET /api/v1/documents/{name}`）刻意分道，三处不同： +// +// - **下发类型恒为 `audio/*`**，按**存储名的扩展名**映射（`voice.MediaMIME`）， +// **不查库、不重新嗅探** —— 「扩展名 ↔ 字节」在保存那一刻已经成对裁决 +// （`documents.SaveAudio`：音频字节的嗅探结果恒为 `application/octet-stream`， +// 嗅探对音频没有意义）； +// - **内联而非附件**（`inline; filename=…`）：这份字节就是要被播放器消费的。 +// 因下发类型恒在 `audio/*` + `nosniff`，内联不构成文档侧那类 XSS 面（文档读侧 +// 之所以永不内联，是因为它可能装着 HTML/SVG 这类可渲染内容）； +// - **鉴权与文档同档**（Bearer `owner|admin`，门在路由表）：语音是私密内容，**不** +// 走图片那套「随机名即凭据」的免鉴权读法 —— 客户端因此必须带 token 取字节 +// （Web：`fetch` → `blob:` → 播放），音频 URL **不能**直接贴进 `