test(memory): 媒体存储的压力、冒烟与长稳测试

在 a822674 的 CAS 层之上补齐三类验证。 ## 压力测试（stress_test.go，9 例）核心不是吞吐数字，而是并发下的不变量。为此写了 checkRefIntegrity：用 SQL 对比每个 digest 的 ref_count 与 media_refs 实际行数。这条对不上就意味着 GC 的判断依据是错的——计数偏低会误删有引用的内容，虚高会让孤儿永远清不掉。所有并发用例收尾都验它。 - 32 goroutine 并发 Put 同一内容 → digest 一致、磁盘只 1 份 - 400 个不同内容并发入库 → 无丢条目、逐条回读无内容串位 - 24 worker × 40 轮引用增删风暴（含故意重复 AddRef 验并发下的幂等） - GC 与读写并发 1.5s → 实测 11508 次 Put / 604 轮 GC，受保护内容零失败 - Describe 与 Search/Pending 并发 → 无 database is locked - 容量上限持续加压 → 上限 256KB 收尾 98KB，有引用项全存活 - 重度 churn 后重开 → 磁盘文件数 == 元数据条数，无双向孤儿 - 4MB 单文件往返（see_video 10 帧 × 2MB 是现实上限附近） - data URL 往返 ×50（SetToolBlocks 给出的实际形态） -race -count=3 干净。 ## 冒烟测试（smoke_test.go，6 场景）走真实数据路径：真 PNG（自建 IHDR/IDAT/IEND + zlib）、真 data URL、真 sha256、真 GC、真重启，而不是随机字节。 - 同一张截图连问 5 轮 → 磁盘 1 份、5 个 context 引用 - see_video 6 帧内容各异 → 各存一份、共享一个 owner - 描述落库后按关键词检索命中（方案 C 最关键的一环：blob 可被淘汰，描述会长期留在记忆里） - L0→L2 归档时引用从 context owner 转到 document owner，期间内容可读 - 别的工具留下的一次性图被 GC 清掉，被记忆引用的一个不少 - 全生命周期跨重启：描述、引用、内容、磁盘一致性全部完好第一次跑挂在「6 帧只搜到 5 条」，看着像存储丢帧，实际是夹具的 palette[(variant+y*3/h)%5] 只有 5 色，variant=0 与 5 产出逐字节相同的 PNG，被 CAS 正确去重。已把 variant 写进像素保证帧间真不同，并把这段经过记进注释——误报本身证明了去重在工作，也证明冒烟确实有能力发现「帧数对不上」这类问题。冒烟原先是 internal/memory/media/smoke/ 下带 //go:build smoke 的独立 main，得记着加 -tags smoke 才跑得到，那种早晚被忘掉。已搬成普通测试，随 go test ./... 一起跑，冒烟的意义才真正成立。 ## 长稳测试（soak_test.go，-short 下跳过） 60 秒五路混合负载。实测：put=281885 get=1644084 gc=9142 describe=53875 search=23268 refOps=187926，零失败。收尾 20 个受保护项内容字节一致、ref_count 全为 1；8MB 上限下实际占用 139KB / 48 条，说明 28 万次写入产生的孤儿被持续清理，无无界增长。描述者从 Pending() 取项再 Describe()，GC 随时可能在这两步之间清掉它。这是正常竞态，故忽略 unknown digest 并注明原因；5 万多次调用没把它升级成计数错位，印证了 Describe 对已删项返回错误而非静默建条目的选择。全仓 go build / go vet / go test 通过，SDK 接口冻结 diff 为 0。
2026-09-22 18:08:04 +00:00 · 2026-09-04 11:29:18 +08:00
parent a82267484a
commit 4ef3371701
3 changed files with 1097 additions and 0 deletions
--- a/internal/memory/media/smoke_test.go
+++ b/internal/memory/media/smoke_test.go
@ -0,0 +1,316 @@
+package media
+
+import (
+	"bytes"
+	"compress/zlib"
+	"encoding/binary"
+	"fmt"
+	"hash/crc32"
+	"testing"
+)
+
+// 冒烟测试：走真实数据路径的端到端场景，而非孤立的 API 单测。
+//
+// 之前这套场景是 internal/memory/media/smoke/ 下一个带 //go:build smoke 的
+// 独立 main，得记着加 -tags smoke 才跑得到——那种早晚会被忘掉。搬成普通
+// 测试后它随 go test ./... 一起跑，冒烟的意义（每次改动都过一遍真实链路）
+// 才真正成立。
+
+// makePNG 生成一张 w×h 的条带 PNG，用真 PNG 而不是随机字节，
+// 让入库/回读/digest 走的是与生产一致的数据形态。
+//
+// variant 注入到像素而不只用于选色：最初写的是
+// palette[(variant+y*3/h)%5]，调色盘只 5 色，于是 variant=0 与 5 产出
+// 逐字节相同的 PNG——冒烟跑出「6 帧只得 5 条」，看着像存储丢了一帧，
+// 实际是 CAS 正确去重了两张真同图。冒烟要验的是「不同帧各存一份」，
+// 夹具就必须保证帧间真的不同。
+func makePNG(w, h, variant int) []byte {
+	palette := [][3]byte{
+		{255, 0, 0}, {0, 192, 0}, {0, 0, 255}, {255, 220, 0}, {160, 0, 200},
+	}
+	var raw bytes.Buffer
+	for y := 0; y < h; y++ {
+		raw.WriteByte(0) // 每行的滤波器字节
+		c := palette[(variant+y*3/h)%len(palette)]
+		for x := 0; x < w; x++ {
+			raw.Write(c[:])
+		}
+	}
+	// 把 variant 写进首行头几个像素，确保不同 variant 字节必然不同。
+	b := raw.Bytes()
+	if len(b) > 8 {
+		b[1] = byte(variant)
+		b[2] = byte(variant >> 8)
+	}
+
+	var z bytes.Buffer
+	zw := zlib.NewWriter(&z)
+	zw.Write(b)
+	zw.Close()
+
+	chunk := func(typ string, data []byte) []byte {
+		var out bytes.Buffer
+		binary.Write(&out, binary.BigEndian, uint32(len(data)))
+		out.WriteString(typ)
+		out.Write(data)
+		binary.Write(&out, binary.BigEndian, crc32.ChecksumIEEE(append([]byte(typ), data...)))
+		return out.Bytes()
+	}
+	var ihdr bytes.Buffer
+	binary.Write(&ihdr, binary.BigEndian, uint32(w))
+	binary.Write(&ihdr, binary.BigEndian, uint32(h))
+	ihdr.Write([]byte{8, 2, 0, 0, 0}) // 8bit 深度、truecolor
+
+	var out bytes.Buffer
+	out.Write([]byte{0x89, 'P', 'N', 'G', '\r', '\n', 0x1a, '\n'})
+	out.Write(chunk("IHDR", ihdr.Bytes()))
+	out.Write(chunk("IDAT", z.Bytes()))
+	out.Write(chunk("IEND", nil))
+	return out.Bytes()
+}
+
+func TestSmoke_SamePictureAcrossTurns(t *testing.T) {
+	// 场景：用户连问几轮同一张截图。multimodal 每轮都会重新注入，
+	// 磁盘上应该只有一份，但每轮的 context 事件各持一个引用。
+	s := newTestStore(t, 50*1024*1024)
+	png := makePNG(400, 400, 0)
+
+	var d0 string
+	for turn := 1; turn <= 5; turn++ {
+		// 走 data URL：这是 SetToolBlocks 实际给出的形态
+		url := DataURL("image/png", png)
+		mime, data, ok := ParseDataURL(url)
+		if !ok {
+			t.Fatalf("第 %d 轮 data URL 解析失败", turn)
+		}
+		d, err := s.Put(data, Item{
+			MIME: mime, Width: 400, Height: 400,
+			OriginPath: fmt.Sprintf("/tmp/probe_%d.png", turn),
+			Tool:       "multimodal_see_picture",
+		})
+		if err != nil {
+			t.Fatalf("第 %d 轮 Put: %v", turn, err)
+		}
+		if d0 == "" {
+			d0 = d
+		} else if d != d0 {
+			t.Fatalf("同一张图第 %d 轮 digest 变了", turn)
+		}
+		if err := s.AddRef(d, "context", fmt.Sprintf("evt-%d", turn)); err != nil {
+			t.Fatalf("第 %d 轮 AddRef: %v", turn, err)
+		}
+	}
+
+	st := s.Stats()
+	if st["count"].(int) != 1 {
+		t.Fatalf("5 轮同图应只存 1 份，实际 %v 条", st["count"])
+	}
+	if total := st["total_bytes"].(int64); total != int64(len(png)) {
+		t.Fatalf("字节数应等于单张原图 %d，实际 %d", len(png), total)
+	}
+	it, _ := s.Stat(d0)
+	if it.RefCount != 5 {
+		t.Fatalf("应有 5 个引用，实际 %d", it.RefCount)
+	}
+	t.Logf("同图 5 轮：条目=1 字节=%d refcount=%d", len(png), it.RefCount)
+	checkRefIntegrity(t, s)
+}
+
+func TestSmoke_VideoFramesDistinct(t *testing.T) {
+	// 场景：see_video 抽 6 帧，帧间内容不同，应各存一份并共享一个 owner。
+	s := newTestStore(t, 50*1024*1024)
+	var frames []string
+	for i := 0; i < 6; i++ {
+		d, err := s.Put(makePNG(320, 240, i), Item{
+			MIME: "image/jpeg", Width: 320, Height: 240, Tool: "multimodal_see_video",
+		})
+		if err != nil {
+			t.Fatalf("第 %d 帧: %v", i, err)
+		}
+		frames = append(frames, d)
+		if err := s.AddRef(d, "context", "evt-video"); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	st := s.Stats()
+	if st["count"].(int) != 6 {
+		t.Fatalf("6 帧应各存一份，实际 %v 条", st["count"])
+	}
+	refs, err := s.Refs("context", "evt-video")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(refs) != 6 {
+		t.Fatalf("evt-video 应引用 6 帧，实际 %d", len(refs))
+	}
+	checkRefIntegrity(t, s)
+}
+
+func TestSmoke_DescribeThenRetrieve(t *testing.T) {
+	// 场景 C：视觉模型描述落库后，描述文字成为可检索的语义入口。
+	// 这是本方案最关键的一环——blob 可能被淘汰，描述会长期留在记忆里。
+	s := newTestStore(t, 50*1024*1024)
+
+	pic, _ := s.Put(makePNG(400, 400, 0), Item{MIME: "image/png", Tool: "multimodal_see_picture"})
+	if err := s.Describe(pic, "一张 400x400 的三色带图：上红、中绿、下蓝", "visionllm"); err != nil {
+		t.Fatal(err)
+	}
+	var frames []string
+	for i := 0; i < 6; i++ {
+		d, _ := s.Put(makePNG(320, 240, i), Item{MIME: "image/jpeg", Tool: "multimodal_see_video"})
+		frames = append(frames, d)
+		if err := s.Describe(d, fmt.Sprintf("视频第 %d 帧：测试图卡，含彩条与计数器", i+1), "visionllm"); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if hits, _ := s.Search("三色带", KindImage, 10); len(hits) != 1 {
+		t.Fatalf("搜「三色带」应命中 1 条，实际 %d", len(hits))
+	}
+	if hits, _ := s.Search("计数器", KindImage, 10); len(hits) != 6 {
+		t.Fatalf("搜「计数器」应命中 6 帧，实际 %d", len(hits))
+	}
+	pend, _ := s.Pending(100)
+	if len(pend) != 0 {
+		t.Fatalf("应全部已描述，仍有 %d 条待描述", len(pend))
+	}
+	_ = frames
+}
+
+func TestSmoke_ArchiveTransfersOwnership(t *testing.T) {
+	// 场景：L0 的 context 事件被 Prune 归档进 L2 文档，
+	// 媒体引用需从 context owner 转到 document owner，期间内容不能被 GC 掉。
+	s := newTestStore(t, 50*1024*1024)
+	png := makePNG(400, 400, 0)
+	d, _ := s.Put(png, Item{MIME: "image/png", Tool: "multimodal_see_picture"})
+	for turn := 1; turn <= 5; turn++ {
+		s.AddRef(d, "context", fmt.Sprintf("evt-%d", turn))
+	}
+
+	// evt-1 被淘汰，其内容归档为一篇文档
+	n, err := s.DropOwner("context", "evt-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != 1 {
+		t.Fatalf("应注销 1 条引用，实际 %d", n)
+	}
+	if err := s.AddRef(d, "document", "doc_archived_001"); err != nil {
+		t.Fatal(err)
+	}
+
+	it, _ := s.Stat(d)
+	if it.RefCount != 5 {
+		t.Fatalf("引用转移后总数应仍为 5（4 context + 1 document），实际 %d", it.RefCount)
+	}
+	// 归档过程中内容必须始终可读
+	if got, err := s.Get(d); err != nil || !bytes.Equal(got, png) {
+		t.Fatalf("归档后内容应完好: %v", err)
+	}
+	checkRefIntegrity(t, s)
+}
+
+func TestSmoke_GCSweepsToolLeftovers(t *testing.T) {
+	// 场景：别的工具（cmd_run 之类）产出的一次性图片没人引用，
+	// 应被 GC 清掉；而被记忆引用的媒体一个都不能少。
+	s := newTestStore(t, 50*1024*1024)
+
+	keep, _ := s.Put(makePNG(400, 400, 0), Item{MIME: "image/png"})
+	s.AddRef(keep, "document", "doc-1")
+	var frames []string
+	for i := 0; i < 6; i++ {
+		d, _ := s.Put(makePNG(320, 240, i), Item{MIME: "image/jpeg"})
+		s.AddRef(d, "context", "evt-video")
+		frames = append(frames, d)
+	}
+	// 1000+i 保证与上面的帧、以及彼此都不重复
+	for i := 0; i < 20; i++ {
+		s.Put(makePNG(100, 100, 1000+i), Item{MIME: "image/png", Tool: "cmd_run"})
+	}
+
+	before := s.Stats()["count"].(int)
+	removed, freed, err := s.GC(0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	after := s.Stats()["count"].(int)
+	if removed != 20 {
+		t.Fatalf("应清 20 条孤儿，实际 %d", removed)
+	}
+	if after != before-20 {
+		t.Fatalf("条目数应从 %d 降到 %d，实际 %d", before, before-20, after)
+	}
+	if _, err := s.Get(keep); err != nil {
+		t.Fatalf("被文档引用的图被误删: %v", err)
+	}
+	for i, f := range frames {
+		if _, err := s.Get(f); err != nil {
+			t.Fatalf("第 %d 帧被误删: %v", i, err)
+		}
+	}
+	t.Logf("GC: %d 条 → 清 %d 条（%d 字节）→ %d 条", before, removed, freed, after)
+	checkRefIntegrity(t, s)
+}
+
+func TestSmoke_FullLifecycleAcrossRestart(t *testing.T) {
+	// 端到端：入库 → 描述 → 引用 → GC → 重启 → 检索，
+	// 并确认磁盘与元数据不出现双向孤儿。记忆的意义就在于跨重启还在。
+	dir := t.TempDir()
+	s, err := New(dir, 50*1024*1024)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	png := makePNG(400, 400, 0)
+	pic, _ := s.Put(png, Item{MIME: "image/png", Width: 400, Height: 400, Tool: "multimodal_see_picture"})
+	s.Describe(pic, "一张 400x400 的三色带图：上红、中绿、下蓝", "visionllm")
+	s.AddRef(pic, "graph_sentence", "sent-42")
+	for i := 0; i < 6; i++ {
+		d, _ := s.Put(makePNG(320, 240, i), Item{MIME: "image/jpeg", Tool: "multimodal_see_video"})
+		s.Describe(d, fmt.Sprintf("视频第 %d 帧", i+1), "visionllm")
+		s.AddRef(d, "context", "evt-video")
+	}
+	for i := 0; i < 10; i++ {
+		s.Put(makePNG(64, 64, 2000+i), Item{MIME: "image/png", Tool: "cmd_run"})
+	}
+	if _, _, err := s.GC(0); err != nil {
+		t.Fatal(err)
+	}
+	beforeCount := s.Stats()["count"].(int)
+	s.Close()
+
+	s2, err := New(dir, 50*1024*1024)
+	if err != nil {
+		t.Fatalf("重开失败: %v", err)
+	}
+	defer s2.Close()
+
+	if got := s2.Stats()["count"].(int); got != beforeCount {
+		t.Fatalf("重开后条目数变了: %d → %d", beforeCount, got)
+	}
+	it, err := s2.Stat(pic)
+	if err != nil {
+		t.Fatalf("重开后查不到: %v", err)
+	}
+	if it.Description == "" || it.RefCount != 1 {
+		t.Fatalf("元数据未持久化: %+v", it)
+	}
+	data, err := s2.Get(pic)
+	if err != nil || !bytes.Equal(data, png) {
+		t.Fatalf("重开后内容不一致: %v", err)
+	}
+	if refs, _ := s2.Refs("context", "evt-video"); len(refs) != 6 {
+		t.Fatalf("重开后视频帧引用应为 6，实际 %d", len(refs))
+	}
+	if hits, _ := s2.Search("三色带", KindImage, 10); len(hits) != 1 {
+		t.Fatal("重开后描述应仍可检索")
+	}
+
+	// 磁盘文件数 == 元数据条数：无「元数据在文件没了」也无「文件在元数据没了」
+	if n := blobFileCount(t, s2); n != beforeCount {
+		t.Fatalf("磁盘 blob=%d 与元数据=%d 不一致", n, beforeCount)
+	}
+	checkRefIntegrity(t, s2)
+	t.Logf("跨重启：%d 条目、描述与引用全部完好", beforeCount)
+}
--- a/internal/memory/media/soak_test.go
+++ b/internal/memory/media/soak_test.go
@ -0,0 +1,180 @@
+package media
+
+import (
+	"bytes"
+	"crypto/rand"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// TestSoak_SustainedMixedLoad 长稳测试：持续混合负载下不变量不破。
+// 用 -run TestSoak -timeout 300s 单独跑，默认 short 模式跳过。
+func TestSoak_SustainedMixedLoad(t *testing.T) {
+	if testing.Short() {
+		t.Skip("long soak test; run with -run TestSoak")
+	}
+	dur := 60 * time.Second
+	s := newTestStore(t, 8*1024*1024) // 8MB 上限，逼 GC 频繁工作
+
+	// 常驻受保护集
+	const keepN = 20
+	keep := make([]string, keepN)
+	keepData := make([][]byte, keepN)
+	for i := range keep {
+		d := make([]byte, 4096)
+		rand.Read(d)
+		d = append([]byte(fmt.Sprintf("keep-%d-", i)), d...)
+		dg, err := s.Put(d, Item{MIME: "image/png"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := s.AddRef(dg, "graph_sentence", fmt.Sprintf("s-%d", i)); err != nil {
+			t.Fatal(err)
+		}
+		keep[i] = dg
+		keepData[i] = d
+	}
+
+	stop := make(chan struct{})
+	var wg sync.WaitGroup
+	var puts, gets, gcs, describes, searches, refOps atomic.Int64
+	var fatal atomic.Int64
+
+	worker := func(name string, fn func(iter int) error) {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; ; i++ {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				if err := fn(i); err != nil {
+					fatal.Add(1)
+					t.Errorf("%s 第 %d 次失败: %v", name, i, err)
+					return
+				}
+			}
+		}()
+	}
+
+	// 写入者 ×3
+	for w := 0; w < 3; w++ {
+		wid := w
+		worker(fmt.Sprintf("put-%d", wid), func(i int) error {
+			b := make([]byte, 2048)
+			rand.Read(b)
+			b = append([]byte(fmt.Sprintf("eph-%d-%d-", wid, i)), b...)
+			d, err := s.Put(b, Item{MIME: "image/png", Tool: "cmd_run"})
+			if err != nil {
+				return err
+			}
+			puts.Add(1)
+			// 三分之一挂上引用再立刻注销，模拟短命引用
+			if i%3 == 0 {
+				own := fmt.Sprintf("tmp-%d-%d", wid, i)
+				if err := s.AddRef(d, "context", own); err != nil {
+					return err
+				}
+				if err := s.DropRef(d, "context", own); err != nil {
+					return err
+				}
+				refOps.Add(2)
+			}
+			return nil
+		})
+	}
+
+	// 读取者 ×3：受保护集必须始终完好
+	for r := 0; r < 3; r++ {
+		worker("get", func(i int) error {
+			idx := i % keepN
+			got, err := s.Get(keep[idx])
+			if err != nil {
+				return err
+			}
+			if !bytes.Equal(got, keepData[idx]) {
+				return fmt.Errorf("内容被改 %s", shortDigest(keep[idx]))
+			}
+			gets.Add(1)
+			return nil
+		})
+	}
+
+	// GC 者
+	worker("gc", func(i int) error {
+		if _, _, err := s.GC(0); err != nil {
+			return err
+		}
+		gcs.Add(1)
+		time.Sleep(5 * time.Millisecond)
+		return nil
+	})
+
+	// 描述者
+	worker("describe", func(i int) error {
+		pend, err := s.Pending(5)
+		if err != nil {
+			return err
+		}
+		for _, it := range pend {
+			// 忽略 unknown digest：GC 可能在 Pending 与 Describe 之间清掉它，
+			// 这是正常竞态而非缺陷。
+			_ = s.Describe(it.Digest, fmt.Sprintf("描述 %d 含图表与文字", i), "vis")
+			describes.Add(1)
+		}
+		time.Sleep(2 * time.Millisecond)
+		return nil
+	})
+
+	// 检索者
+	worker("search", func(i int) error {
+		if _, err := s.Search("图表", KindImage, 20); err != nil {
+			return err
+		}
+		if _, err := s.Stat(keep[i%keepN]); err != nil {
+			return err
+		}
+		searches.Add(1)
+		time.Sleep(2 * time.Millisecond)
+		return nil
+	})
+
+	time.Sleep(dur)
+	close(stop)
+	wg.Wait()
+
+	if n := fatal.Load(); n > 0 {
+		t.Fatalf("%d 个 worker 报致命错误", n)
+	}
+
+	t.Logf("%v 内: put=%d get=%d gc=%d describe=%d search=%d refOps=%d",
+		dur, puts.Load(), gets.Load(), gcs.Load(), describes.Load(), searches.Load(), refOps.Load())
+
+	// 收尾断言
+	for i, d := range keep {
+		got, err := s.Get(d)
+		if err != nil {
+			t.Fatalf("受保护项丢失 %s: %v", shortDigest(d), err)
+		}
+		if !bytes.Equal(got, keepData[i]) {
+			t.Fatalf("受保护项内容变了 %s", shortDigest(d))
+		}
+		it, err := s.Stat(d)
+		if err != nil || it.RefCount != 1 {
+			t.Fatalf("受保护项引用计数应为 1: %+v", it)
+		}
+	}
+	checkRefIntegrity(t, s)
+
+	st := s.Stats()
+	t.Logf("收尾: 条目=%v 字节=%v 未引用=%v 已描述=%v",
+		st["count"], st["total_bytes"], st["unreferenced"], st["described"])
+	if total := st["total_bytes"].(int64); total > 8*1024*1024*3 {
+		t.Fatalf("容量失控: %d 远超上限", total)
+	}
+}
--- a/internal/memory/media/stress_test.go
+++ b/internal/memory/media/stress_test.go
@ -0,0 +1,601 @@
+package media
+
+import (
+	"bytes"
+	"crypto/rand"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// 压力测试与冒烟测试。
+//
+// 关注点不是吞吐数字，而是并发下的不变量是否被破坏：
+//   1. ref_count 与 media_refs 表的行数必须始终一致（错位会让 GC 误删或永不清）
+//   2. GC 与读写并发时，有引用的内容绝不能被删
+//   3. 同内容并发 Put 只落一份磁盘、digest 一致
+//   4. SQLite 在多 goroutine 下不出现 "database is locked"
+
+func randBytes(t *testing.T, n int) []byte {
+	t.Helper()
+	b := make([]byte, n)
+	if _, err := rand.Read(b); err != nil {
+		t.Fatalf("rand: %v", err)
+	}
+	return b
+}
+
+// checkRefIntegrity 校验核心不变量：每个 digest 的 ref_count 等于
+// media_refs 里指向它的行数。这条对不上就意味着 GC 的判断依据是错的。
+func checkRefIntegrity(t *testing.T, s *Store) {
+	t.Helper()
+	rows, err := s.db.Query(`
+		SELECT m.digest, m.ref_count, COUNT(r.digest)
+		FROM media m LEFT JOIN media_refs r ON m.digest = r.digest
+		GROUP BY m.digest, m.ref_count`)
+	if err != nil {
+		t.Fatalf("integrity query: %v", err)
+	}
+	defer rows.Close()
+	var bad int
+	for rows.Next() {
+		var d string
+		var stored, actual int
+		if err := rows.Scan(&d, &stored, &actual); err != nil {
+			continue
+		}
+		if stored != actual {
+			bad++
+			if bad <= 5 {
+				t.Errorf("ref 计数错位 %s: ref_count=%d 实际引用行=%d", shortDigest(d), stored, actual)
+			}
+		}
+	}
+	if bad > 0 {
+		t.Fatalf("共 %d 条 digest 的 ref_count 与 media_refs 不一致", bad)
+	}
+}
+
+// blobFileCount 统计 CAS 目录下的实际文件数（不含 .tmp）。
+func blobFileCount(t *testing.T, s *Store) int {
+	t.Helper()
+	n := 0
+	filepath.Walk(s.blobDir, func(path string, info os.FileInfo, err error) error {
+		if err != nil || info == nil || info.IsDir() {
+			return nil
+		}
+		if strings.HasSuffix(path, ".tmp") {
+			t.Errorf("残留临时文件: %s", path)
+			return nil
+		}
+		n++
+		return nil
+	})
+	return n
+}
+
+func TestStress_ConcurrentPutSameContent(t *testing.T) {
+	// 同一内容被 N 个 goroutine 同时 Put：digest 必须一致，磁盘只一份。
+	// 现实对应：see_video 抽出的相邻帧、用户连发同一张图。
+	s := newTestStore(t, 0)
+	data := randBytes(t, 64*1024)
+
+	const workers = 32
+	var wg sync.WaitGroup
+	digests := make([]string, workers)
+	errs := make([]error, workers)
+	for i := 0; i < workers; i++ {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			d, err := s.Put(data, Item{MIME: "image/png", OriginPath: fmt.Sprintf("/tmp/%d.png", idx)})
+			digests[idx] = d
+			errs[idx] = err
+		}(i)
+	}
+	wg.Wait()
+
+	for i, err := range errs {
+		if err != nil {
+			t.Fatalf("worker %d Put 失败: %v", i, err)
+		}
+	}
+	first := digests[0]
+	for i, d := range digests {
+		if d != first {
+			t.Fatalf("worker %d digest 不一致: %s vs %s", i, d, first)
+		}
+	}
+	if n := blobFileCount(t, s); n != 1 {
+		t.Fatalf("同一内容应只落一份 blob，实际 %d 个文件", n)
+	}
+	if got, err := s.Get(first); err != nil || !bytes.Equal(got, data) {
+		t.Fatalf("内容应可完整读回: err=%v len=%d", err, len(got))
+	}
+	checkRefIntegrity(t, s)
+}
+
+func TestStress_ConcurrentPutDistinctContent(t *testing.T) {
+	// 大量不同内容并发入库：不丢条目、不串内容。
+	s := newTestStore(t, 0)
+	const workers = 16
+	const perWorker = 25
+
+	var wg sync.WaitGroup
+	var failed atomic.Int64
+	type rec struct {
+		digest string
+		data   []byte
+	}
+	recCh := make(chan rec, workers*perWorker)
+
+	for w := 0; w < workers; w++ {
+		wg.Add(1)
+		go func(wid int) {
+			defer wg.Done()
+			for i := 0; i < perWorker; i++ {
+				data := []byte(fmt.Sprintf("w%d-i%d-", wid, i))
+				data = append(data, randBytes(t, 512)...)
+				d, err := s.Put(data, Item{MIME: "image/png"})
+				if err != nil {
+					failed.Add(1)
+					continue
+				}
+				recCh <- rec{digest: d, data: data}
+			}
+		}(w)
+	}
+	wg.Wait()
+	close(recCh)
+
+	if n := failed.Load(); n > 0 {
+		t.Fatalf("%d 次 Put 失败", n)
+	}
+
+	var records []rec
+	for r := range recCh {
+		records = append(records, r)
+	}
+	if len(records) != workers*perWorker {
+		t.Fatalf("应有 %d 条记录，实际 %d", workers*perWorker, len(records))
+	}
+
+	// 逐条回读校验内容没串
+	for _, r := range records {
+		got, err := s.Get(r.digest)
+		if err != nil {
+			t.Fatalf("读 %s 失败: %v", shortDigest(r.digest), err)
+		}
+		if !bytes.Equal(got, r.data) {
+			t.Fatalf("内容串了: %s", shortDigest(r.digest))
+		}
+	}
+
+	st := s.Stats()
+	if st["count"].(int) != len(records) {
+		t.Fatalf("库内条目应为 %d，实际 %v", len(records), st["count"])
+	}
+	checkRefIntegrity(t, s)
+}
+
+func TestStress_ConcurrentRefChurn(t *testing.T) {
+	// 引用增删风暴：多 owner 对少量 digest 反复 AddRef/DropRef。
+	// 核心断言是最终 ref_count 与 media_refs 行数一致——错位就意味着
+	// GC 会误删（计数偏低）或永不清（计数虚高）。
+	s := newTestStore(t, 0)
+
+	const digestCount = 8
+	digests := make([]string, digestCount)
+	for i := range digests {
+		d, err := s.Put([]byte(fmt.Sprintf("payload-%d", i)), Item{MIME: "image/png"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		digests[i] = d
+	}
+
+	const workers = 24
+	const rounds = 40
+	var wg sync.WaitGroup
+	var addErr, dropErr atomic.Int64
+
+	for w := 0; w < workers; w++ {
+		wg.Add(1)
+		go func(wid int) {
+			defer wg.Done()
+			owner := fmt.Sprintf("evt-%d", wid)
+			for r := 0; r < rounds; r++ {
+				d := digests[(wid+r)%digestCount]
+				if err := s.AddRef(d, "context", owner); err != nil {
+					addErr.Add(1)
+				}
+				// 故意重复 AddRef：幂等性在并发下也必须成立
+				if err := s.AddRef(d, "context", owner); err != nil {
+					addErr.Add(1)
+				}
+				if r%2 == 0 {
+					if err := s.DropRef(d, "context", owner); err != nil {
+						dropErr.Add(1)
+					}
+				}
+			}
+		}(w)
+	}
+	wg.Wait()
+
+	if n := addErr.Load(); n > 0 {
+		t.Fatalf("AddRef 失败 %d 次", n)
+	}
+	if n := dropErr.Load(); n > 0 {
+		t.Fatalf("DropRef 失败 %d 次", n)
+	}
+	checkRefIntegrity(t, s)
+}
+
+func TestStress_GCConcurrentWithWrites(t *testing.T) {
+	// GC 与读写并发。最重要的断言：有引用的内容在整个过程中始终可读。
+	// 这条一旦破，记忆里的 digest 就成了悬空指针。
+	s := newTestStore(t, 0)
+
+	// 一批"受保护"的内容，全程持有引用
+	const protectedCount = 10
+	protected := make([]string, protectedCount)
+	protectedData := make([][]byte, protectedCount)
+	for i := range protected {
+		data := append([]byte(fmt.Sprintf("protected-%d-", i)), randBytes(t, 256)...)
+		d, err := s.Put(data, Item{MIME: "image/png"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := s.AddRef(d, "document", fmt.Sprintf("doc-%d", i)); err != nil {
+			t.Fatal(err)
+		}
+		protected[i] = d
+		protectedData[i] = data
+	}
+
+	stop := make(chan struct{})
+	var wg sync.WaitGroup
+	var readErr atomic.Int64
+	var gcRuns atomic.Int64
+	var putCount atomic.Int64
+
+	// 写入者：持续 Put 一次性内容（不加引用，是 GC 的正常目标）
+	for w := 0; w < 4; w++ {
+		wg.Add(1)
+		go func(wid int) {
+			defer wg.Done()
+			i := 0
+			for {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				data := append([]byte(fmt.Sprintf("ephemeral-%d-%d-", wid, i)), randBytes(t, 128)...)
+				if _, err := s.Put(data, Item{MIME: "image/png"}); err == nil {
+					putCount.Add(1)
+				}
+				i++
+			}
+		}(w)
+	}
+
+	// 读取者：反复读受保护内容，任何一次失败都是致命的
+	for r := 0; r < 4; r++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				for i, d := range protected {
+					got, err := s.Get(d)
+					if err != nil {
+						readErr.Add(1)
+						t.Errorf("受保护内容读失败 %s: %v", shortDigest(d), err)
+						return
+					}
+					if !bytes.Equal(got, protectedData[i]) {
+						readErr.Add(1)
+						t.Errorf("受保护内容被改 %s", shortDigest(d))
+						return
+					}
+				}
+			}
+		}()
+	}
+
+	// GC 者：minAge=0 让所有无引用项立刻可清，最大化与写入的冲突
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			if _, _, err := s.GC(0); err != nil {
+				t.Errorf("GC 报错: %v", err)
+				return
+			}
+			gcRuns.Add(1)
+			time.Sleep(time.Millisecond)
+		}
+	}()
+
+	time.Sleep(1500 * time.Millisecond)
+	close(stop)
+	wg.Wait()
+
+	if n := readErr.Load(); n > 0 {
+		t.Fatalf("受保护内容读取失败 %d 次——GC 误删了有引用的项", n)
+	}
+	t.Logf("并发窗口内: Put=%d GC=%d 轮", putCount.Load(), gcRuns.Load())
+
+	// 收尾确认：受保护的一个都没少
+	for i, d := range protected {
+		got, err := s.Get(d)
+		if err != nil || !bytes.Equal(got, protectedData[i]) {
+			t.Fatalf("收尾检查失败 %s: %v", shortDigest(d), err)
+		}
+		it, err := s.Stat(d)
+		if err != nil || it.RefCount != 1 {
+			t.Fatalf("受保护项引用计数应为 1: %+v err=%v", it, err)
+		}
+	}
+	checkRefIntegrity(t, s)
+}
+
+func TestStress_DescribeConcurrentWithSearch(t *testing.T) {
+	// 描述写入与检索并发。C 部分的后台描述任务会长期这样跑。
+	s := newTestStore(t, 0)
+	const n = 60
+	digests := make([]string, n)
+	for i := range digests {
+		d, err := s.Put([]byte(fmt.Sprintf("img-%d", i)), Item{MIME: "image/png"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		digests[i] = d
+	}
+
+	var wg sync.WaitGroup
+	var descErr, searchErr atomic.Int64
+
+	// 描述写入者
+	for w := 0; w < 4; w++ {
+		wg.Add(1)
+		go func(wid int) {
+			defer wg.Done()
+			for i := wid; i < n; i += 4 {
+				desc := fmt.Sprintf("第 %d 张图，含蓝色图表与文字", i)
+				if err := s.Describe(digests[i], desc, "vis-src"); err != nil {
+					descErr.Add(1)
+				}
+			}
+		}(w)
+	}
+
+	// 检索者 + Pending 消费者
+	for r := 0; r < 3; r++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < 50; i++ {
+				if _, err := s.Search("图表", KindImage, 20); err != nil {
+					searchErr.Add(1)
+				}
+				if _, err := s.Pending(10); err != nil {
+					searchErr.Add(1)
+				}
+			}
+		}()
+	}
+	wg.Wait()
+
+	if v := descErr.Load(); v > 0 {
+		t.Fatalf("Describe 失败 %d 次", v)
+	}
+	if v := searchErr.Load(); v > 0 {
+		t.Fatalf("Search/Pending 失败 %d 次", v)
+	}
+
+	// 全部应已描述完
+	pending, err := s.Pending(1000)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(pending) != 0 {
+		t.Fatalf("应全部描述完，仍有 %d 条未描述", len(pending))
+	}
+	got, err := s.Search("图表", KindImage, 1000)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != n {
+		t.Fatalf("应检索到 %d 条，实际 %d", n, len(got))
+	}
+}
+
+func TestStress_CapacityGCUnderLoad(t *testing.T) {
+	// 容量上限在持续写入下必须真正生效，且不碰有引用的项。
+	const cap = 256 * 1024 // 256KB
+	s := newTestStore(t, cap)
+
+	// 先放 3 个有引用的大项（合计约 96KB），它们永不可删
+	const keepN = 3
+	keep := make([]string, keepN)
+	for i := range keep {
+		data := append([]byte(fmt.Sprintf("keep-%d-", i)), randBytes(t, 32*1024)...)
+		d, err := s.Put(data, Item{MIME: "image/png"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := s.AddRef(d, "graph_sentence", fmt.Sprintf("sent-%d", i)); err != nil {
+			t.Fatal(err)
+		}
+		keep[i] = d
+	}
+
+	// 持续写入无引用内容，交替 GC
+	for round := 0; round < 30; round++ {
+		for i := 0; i < 3; i++ {
+			data := append([]byte(fmt.Sprintf("tmp-%d-%d-", round, i)), randBytes(t, 16*1024)...)
+			if _, err := s.Put(data, Item{MIME: "image/png"}); err != nil {
+				t.Fatalf("round %d Put: %v", round, err)
+			}
+		}
+		if _, _, err := s.GC(0); err != nil {
+			t.Fatalf("round %d GC: %v", round, err)
+		}
+	}
+
+	st := s.Stats()
+	total := st["total_bytes"].(int64)
+	t.Logf("上限 %d，收尾总量 %d，条目 %v", cap, total, st["count"])
+
+	// 有引用的项必须都在
+	for _, d := range keep {
+		if _, err := s.Get(d); err != nil {
+			t.Fatalf("有引用项被容量 GC 删了 %s: %v", shortDigest(d), err)
+		}
+	}
+	// 无引用项应被压到上限附近：允许略超（有引用项本身可能就占了大头），
+	// 但不该无界增长——30 轮 × 3 × 16KB = 1.4MB 若全留下就是失控。
+	if total > cap*2 {
+		t.Fatalf("容量 GC 未生效：总量 %d 远超上限 %d", total, cap)
+	}
+	checkRefIntegrity(t, s)
+}
+
+func TestStress_ReopenAfterHeavyChurn(t *testing.T) {
+	// 大量写入 + GC 之后重开：元数据与磁盘不该出现互相不认的孤儿。
+	dir := t.TempDir()
+	s1, err := New(dir, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var kept []string
+	for i := 0; i < 100; i++ {
+		data := append([]byte(fmt.Sprintf("churn-%d-", i)), randBytes(t, 256)...)
+		d, err := s1.Put(data, Item{MIME: "image/png"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if i%5 == 0 {
+			if err := s1.AddRef(d, "context", fmt.Sprintf("e-%d", i)); err != nil {
+				t.Fatal(err)
+			}
+			kept = append(kept, d)
+		}
+	}
+	if _, _, err := s1.GC(0); err != nil {
+		t.Fatal(err)
+	}
+	beforeStats := s1.Stats()
+	s1.Close()
+
+	s2, err := New(dir, 0)
+	if err != nil {
+		t.Fatalf("重开失败: %v", err)
+	}
+	defer s2.Close()
+
+	afterStats := s2.Stats()
+	if beforeStats["count"] != afterStats["count"] {
+		t.Fatalf("重开后条目数变了: %v → %v", beforeStats["count"], afterStats["count"])
+	}
+
+	// 每条元数据都应有对应磁盘文件（无「元数据在文件没了」的孤儿）
+	rows, err := s2.db.Query(`SELECT digest FROM media`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var missing int
+	for rows.Next() {
+		var d string
+		if rows.Scan(&d) != nil {
+			continue
+		}
+		if _, err := os.Stat(s2.blobPath(d)); err != nil {
+			missing++
+			if missing <= 3 {
+				t.Errorf("元数据存在但 blob 缺失: %s", shortDigest(d))
+			}
+		}
+	}
+	rows.Close()
+	if missing > 0 {
+		t.Fatalf("%d 条元数据没有对应文件", missing)
+	}
+
+	// 磁盘文件数应等于元数据条数（无「文件在元数据没了」的孤儿）
+	if n := blobFileCount(t, s2); n != afterStats["count"].(int) {
+		t.Fatalf("磁盘文件 %d 与元数据 %v 不一致", n, afterStats["count"])
+	}
+
+	for _, d := range kept {
+		if _, err := s2.Get(d); err != nil {
+			t.Fatalf("有引用项重开后读不到 %s: %v", shortDigest(d), err)
+		}
+	}
+	checkRefIntegrity(t, s2)
+}
+
+func TestStress_LargeBlob(t *testing.T) {
+	// 单个大文件：see_video 10 帧 × 2MB 是现实上限附近。
+	s := newTestStore(t, 0)
+	data := randBytes(t, 4*1024*1024) // 4MB
+
+	d, err := s.Put(data, Item{MIME: "image/jpeg", Width: 1920, Height: 1080})
+	if err != nil {
+		t.Fatalf("4MB Put 失败: %v", err)
+	}
+	got, err := s.Get(d)
+	if err != nil {
+		t.Fatalf("4MB Get 失败: %v", err)
+	}
+	if !bytes.Equal(got, data) {
+		t.Fatal("4MB 内容回读不一致")
+	}
+	it, _ := s.Stat(d)
+	if it.Size != int64(len(data)) {
+		t.Fatalf("Size 记录错: %d vs %d", it.Size, len(data))
+	}
+}
+
+func TestStress_DataURLRoundTripAtScale(t *testing.T) {
+	// data URL 往返是插件注入的实际路径（SetToolBlocks 给的就是 data URL）。
+	s := newTestStore(t, 0)
+	for i := 0; i < 50; i++ {
+		raw := randBytes(t, 2048)
+		url := DataURL("image/png", raw)
+		mime, decoded, ok := ParseDataURL(url)
+		if !ok {
+			t.Fatalf("第 %d 次解析失败", i)
+		}
+		if mime != "image/png" || !bytes.Equal(decoded, raw) {
+			t.Fatalf("第 %d 次往返不一致", i)
+		}
+		d, err := s.Put(decoded, Item{MIME: mime})
+		if err != nil {
+			t.Fatal(err)
+		}
+		back, err := s.Get(d)
+		if err != nil || !bytes.Equal(back, raw) {
+			t.Fatalf("第 %d 次入库回读不一致: %v", i, err)
+		}
+	}
+	checkRefIntegrity(t, s)
+}