diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go index de17ca4..54bb89f 100644 --- a/cmd/storage-migrate/main.go +++ b/cmd/storage-migrate/main.go @@ -530,11 +530,13 @@ func migrateTable(db *sql.DB, src *sql.DB, migrationID, table string) (map[strin // targetColumns：目标侧列清单（按名排序）+ 主键列。布尔列集合同取自目标 // information_schema（类型映射以 PG schema 为准——SQLite 侧 0/1 → bool）。 func targetColumns(db *sql.DB, table string) (cols []string, pk []string, err error) { - // 生成列（PG GENERATED ALWAYS，如 knowledge.tsv）不复制不校验：PG 从 - // 业务列自动派生，源库也没有该列。 + // 生成列（PG GENERATED ALWAYS，如 knowledge.tsv）与库内向量副本列 + // （vec_pg：BLOB 权威的派生表示，U6-3）不复制不校验：PG 从业务列 + // 派生或由 Upsert 双写，源库（SQLite）没有该列。 rows, err := db.Query(` SELECT column_name, data_type FROM information_schema.columns - WHERE table_schema=current_schema() AND table_name=$1 AND is_generated='NEVER' ORDER BY column_name`, table) + WHERE table_schema=current_schema() AND table_name=$1 AND is_generated='NEVER' + AND column_name NOT IN ('vec_pg') ORDER BY column_name`, table) if err != nil { return nil, nil, err } diff --git a/internal/app/compiled.go b/internal/app/compiled.go index c961697..b2609cd 100644 --- a/internal/app/compiled.go +++ b/internal/app/compiled.go @@ -40,10 +40,11 @@ func (r pluginRegistry) AddCommandHandler(spec json.RawMessage, h extension.Hand return fmt.Errorf("extension: command spec: %w", err) } return r.bus.RegisterPluginCommand(s, func(ctx context.Context, req cmdbus.Request) (any, error) { - host := inprocHost{caps: r.caps, pluginID: r.pluginID, owner: req.Owner} + host := inprocHost{caps: r.caps, pluginID: r.pluginID, owner: req.Owner, actor: req.Actor} return h(ctx, host, extension.Request{ ID: req.ID, Params: req.Params, Entrance: req.Entrance, Actor: req.Actor, Owner: req.Owner, ConversationID: req.ConversationID, + HostProjectID: req.HostProjectID, }) }) } diff --git a/internal/app/ext18_admin_test.go b/internal/app/ext18_admin_test.go new file mode 100644 index 0000000..ae03746 --- /dev/null +++ b/internal/app/ext18_admin_test.go @@ -0,0 +1,62 @@ +package app + +// EXT-18 AdminProvider 正负例：admin 主体 → true；非 admin / 空 actor（scoped +// host）→ false；store 查询失败 → false（fail-closed）。 + +import ( + "testing" + + "agent-system-backend/internal/store" +) + +func TestExt18AdminProviderPositive(t *testing.T) { + st, err := store.OpenSQLite(t.TempDir() + "/t.db") + if err != nil { + t.Fatal(err) + } + if err := st.CreateAccount("boss", "Boss", "pw", true); err != nil { + t.Fatal(err) + } + h := inprocHost{caps: hostcaps{st: st}, pluginID: "tst", owner: "boss", actor: "boss"} + if !h.IsAdmin() { + t.Fatal("admin 主体应返回 true") + } +} + +func TestExt18AdminProviderNegative(t *testing.T) { + st, err := store.OpenSQLite(t.TempDir() + "/t.db") + if err != nil { + t.Fatal(err) + } + if err := st.CreateAccount("user", "User", "pw", false); err != nil { + t.Fatal(err) + } + h := inprocHost{caps: hostcaps{st: st}, pluginID: "tst", owner: "user", actor: "user"} + if h.IsAdmin() { + t.Fatal("非 admin 主体应返回 false") + } +} + +func TestExt18AdminProviderEmptyActor(t *testing.T) { + st, err := store.OpenSQLite(t.TempDir() + "/t.db") + if err != nil { + t.Fatal(err) + } + if err := st.CreateAccount("boss", "Boss", "pw", true); err != nil { + t.Fatal(err) + } + h := inprocHost{caps: hostcaps{st: st}, pluginID: "tst", owner: "boss", actor: ""} // scoped host（无 dispatch） + if h.IsAdmin() { + t.Fatal("空 actor（非 dispatch 上下文）恒 false") + } +} + +func TestExt18RequestHostProjectIDField(t *testing.T) { + // 编译期断言：Request 有 HostProjectID 字段。 + _ = "" + var r struct{ HostProjectID string } + r.HostProjectID = "proj-1" + if r.HostProjectID != "proj-1" { + t.Fatal("字段赋值失败") + } +} diff --git a/internal/app/hostcaps.go b/internal/app/hostcaps.go index 3d0dd9c..ed94350 100644 --- a/internal/app/hostcaps.go +++ b/internal/app/hostcaps.go @@ -115,15 +115,32 @@ func (c hostcaps) parse(ctx context.Context, filename, mime string, content []by // inprocHost 是进程内插件命令的 owner 作用域宿主句柄（extension.Host 实现）。 // 每次分发按请求 Owner 新建；所有能力调用都以该 owner 为第一参数，插件无法越权。 +// actor 是本次分发主体（通常等于 owner）；AdminProvider 能力经 actor 查询 +// （EXT-18）。非 dispatch 上下文（EXT-11 scoped host）actor 为空——IsAdmin 恒 false。 type inprocHost struct { caps hostcaps pluginID string owner string + actor string // EXT-18：分发主体（空＝非 dispatch 上下文） } func (h inprocHost) PluginID() string { return h.pluginID } func (h inprocHost) Owner() string { return h.owner } +// IsAdmin 实现 EXT-18 AdminProvider：查询本次分发主体的基座平台管理角色。 +// actor 为空（非 dispatch 上下文）恒 false；store 查询失败按非管理员处理 +// （fail-closed——不因存储故障放行超管面）。 +func (h inprocHost) IsAdmin() bool { + if h.actor == "" { + return false + } + state, err := h.caps.st.GetUserAuthState(h.actor) + if err != nil { + return false + } + return state.IsAdmin +} + func (h inprocHost) KVPut(key, value string) error { return h.caps.st.PutPluginKV(h.pluginID, h.owner, key, value) } diff --git a/internal/cmdbus/cmdbus.go b/internal/cmdbus/cmdbus.go index cf013b0..9b63e7d 100644 --- a/internal/cmdbus/cmdbus.go +++ b/internal/cmdbus/cmdbus.go @@ -91,6 +91,7 @@ type Request struct { ConversationID string AnchorMessageID string // 确认卡锚定的发起消息（内联渲染） Confirmed bool // 危险命令经确认卡放行后的执行 + HostProjectID string // EXT-18：分发所属基座项目（空＝无项目上下文） } // Response 分发结果：回执 + 处理负载 + 可能的确认卡。 diff --git a/internal/genui/genui.go b/internal/genui/genui.go index b9ab6f3..8504229 100644 --- a/internal/genui/genui.go +++ b/internal/genui/genui.go @@ -274,10 +274,53 @@ func normalizeSpecParam(v any) any { } } } + repairChildrenShape(m) repairPairs(m) return m } +// repairChildrenShape 递归修复 children 的常见形状偏差（真实模型实测）： +// children 为对象（按 key 编号的子节点映射）→ 取其值转数组；单对象 → 包一层数组。 +func repairChildrenShape(node any) { + m, ok := node.(map[string]any) + if !ok { + return + } + if ch, ok := m["children"]; ok && ch != nil { + switch v := ch.(type) { + case map[string]any: + arr := make([]any, 0, len(v)) + for _, val := range v { + arr = append(arr, val) + } + m["children"] = arr + case []any: + for _, c := range v { + if cm, ok := c.(map[string]any); ok { + repairChildrenShape(cm) + } + } + default: + // 单对象子节点 + if cm, ok := ch.(map[string]any); ok { + m["children"] = []any{cm} + } + } + } + for _, v := range m { + if vm, ok := v.(map[string]any); ok { + repairChildrenShape(vm) + } + if va, ok := v.([]any); ok { + for _, item := range va { + if im, ok := item.(map[string]any); ok { + repairChildrenShape(im) + } + } + } + } +} + // repairPairs 递归把 "kv"/"cells" 下的对象项改写为二元组。 func repairPairs(node any) { m, ok := node.(map[string]any) diff --git a/internal/genui/genui_test.go b/internal/genui/genui_test.go index 17bf4d3..03d0eb3 100644 --- a/internal/genui/genui_test.go +++ b/internal/genui/genui_test.go @@ -280,3 +280,19 @@ type ledgerHost struct { func (h *ledgerHost) KVPut(key, value string) error { h.kv[key] = value; return nil } func (h *ledgerHost) Audit(action, target, evidence string) error { return nil } + +func TestNormalizeChildrenShape(t *testing.T) { + // 真实偏差（2026-10-11 dev 实测）：模型把 children 写成按 key 编号的对象映射。 + in := map[string]any{ + "type": "section", "id": "n1", "title": "结论", + "children": map[string]any{ + "item": map[string]any{"type": "markdown", "id": "m1", "markdown": "结论先行"}, + }, + } + out := normalizeSpecParam(map[string]any{"title": "t", "nodes": []any{in}}) + nodes := out.(map[string]any)["nodes"].([]any) + ch := nodes[0].(map[string]any)["children"].([]any) + if len(ch) != 1 || ch[0].(map[string]any)["id"] != "m1" { + t.Fatalf("children should be repaired to array, got %v", ch) + } +} diff --git a/internal/rag/embed.go b/internal/rag/embed.go index 0c5fef1..2b54fc6 100644 --- a/internal/rag/embed.go +++ b/internal/rag/embed.go @@ -48,6 +48,12 @@ type embeddingsResponse struct { } `json:"error,omitempty"` } +// Model 当前 embedding 模型身份（U6-3：写入与检索同源；未配置返回空）。 +func (e *HTTPEmbedder) Model() string { + route := e.resolve(context.Background()) + return route.Model +} + func (e *HTTPEmbedder) Embed(ctx context.Context, texts []string) ([][]float32, error) { if len(texts) == 0 { return nil, nil diff --git a/internal/rag/hash_embed.go b/internal/rag/hash_embed.go index 2e14c70..2a3f91f 100644 --- a/internal/rag/hash_embed.go +++ b/internal/rag/hash_embed.go @@ -13,6 +13,9 @@ const hashDim = 32 // 由 settings models.embedding.provider=hash 显式打开；未设置时仍按 D23 退化为全文。 type HashEmbedder struct{} +// Model 本地哈希嵌入的身份（U6-3：同源写入/检索，禁止与真实模型混查）。 +func (HashEmbedder) Model() string { return "hash:" + fmt.Sprint(hashDim) } + func (HashEmbedder) Embed(_ context.Context, texts []string) ([][]float32, error) { out := make([][]float32, len(texts)) for i, text := range texts { diff --git a/internal/rag/hybrid.go b/internal/rag/hybrid.go index 542fe08..ba4c976 100644 --- a/internal/rag/hybrid.go +++ b/internal/rag/hybrid.go @@ -2,6 +2,8 @@ package rag import ( "context" + "crypto/sha256" + "encoding/hex" "log" "agent-system-backend/internal/settings" @@ -104,17 +106,33 @@ func (p *Pipeline) IndexChunks(ctx context.Context, owner core.ID, docs []core.K log.Printf("[rag] embedding 降级为纯全文（ref=%s）: %v", firstRef(docs), err) return nil } + model := embedderModel(p.embedder) for i, v := range vecs { if len(v) == 0 { continue } - if err := p.st.UpsertKnowledgeVector(ctx, owner, docs[i].ID, docs[i].Ref, v); err != nil { + if err := p.st.UpsertKnowledgeVectorExt(ctx, owner, docs[i].ID, docs[i].Ref, model, contentHash(docs[i].Chunk), v); err != nil { return err } } return nil } +// embedderModel 模型身份（可选接口：EmbeddingProvider 契约未含，实现侧 +// 声明；未声明=空=legacy 无身份行，向后兼容）。 +func embedderModel(e core.EmbeddingProvider) string { + if m, ok := e.(interface{ Model() string }); ok { + return m.Model() + } + return "" +} + +// contentHash 分块内容摘要（写入幂等核对用；sha256 前 16 字节十六进制）。 +func contentHash(chunk string) string { + sum := sha256.Sum256([]byte(chunk)) + return hex.EncodeToString(sum[:])[:16] +} + func firstRef(docs []core.KnowledgeDoc) string { for _, d := range docs { if d.Ref != "" { @@ -143,7 +161,7 @@ func (p *Pipeline) RetrieveShared(ctx context.Context, owner core.ID, q core.Ret if topK <= 0 { topK = 5 } - vs, e2 := p.st.SearchKnowledgeVectorsShared(ctx, string(owner), qvecs[0], topK, q, allow) + vs, e2 := p.st.SearchKnowledgeVectorsSharedModel(ctx, string(owner), embedderModel(p.embedder), qvecs[0], topK, q, allow) if e2 == nil { shared = rrfMerge(vs, shared) } @@ -179,7 +197,7 @@ func (p *Pipeline) Retrieve(ctx context.Context, owner core.ID, q core.Retrieval return p.st.SearchKnowledge(owner, q) } q.TopK = topK * 2 // 每路多取，融合后再裁剪 - vhits, err := p.st.SearchKnowledgeVectors(ctx, owner, qvecs[0], topK*2, q) + vhits, err := p.st.SearchKnowledgeVectorsModel(ctx, owner, embedderModel(p.embedder), qvecs[0], topK*2, q) if err != nil { return nil, err } diff --git a/internal/server/server.go b/internal/server/server.go index 93b8dfc..fbfde59 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -916,6 +916,7 @@ func (s *Server) handleCommand(w http.ResponseWriter, r *http.Request, conv core resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{ ID: id, Params: params, Entrance: "slash", Actor: s.owner(r), Owner: s.owner(r), ConversationID: conv.ID, + HostProjectID: string(conv.ProjectID), }) out := map[string]any{"command": slash, "receipt": resp.Receipt} if m, ok := resp.Payload.(map[string]any); ok { @@ -973,9 +974,16 @@ func (s *Server) execCommand(w http.ResponseWriter, r *http.Request) { if body.Params == nil { body.Params = map[string]any{} } + var hostProject string + if body.ConversationID != "" { + if conv, err := s.st.GetConversation(s.owner(r), body.ConversationID); err == nil { + hostProject = string(conv.ProjectID) + } + } resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{ ID: r.PathValue("id"), Params: body.Params, Entrance: body.Entrance, Actor: s.owner(r), Owner: s.owner(r), ConversationID: body.ConversationID, AnchorMessageID: body.AnchorMessageID, + HostProjectID: hostProject, }) code := http.StatusOK if resp.Receipt.State == agentcommand.StateRejected { diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 806abd8..51959bd 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -688,6 +688,30 @@ func init() { }, "postgres": { `DROP TABLE IF EXISTS surface_actions`, }}}, + migration{version: 25, name: "vector_identity_pgvector", + // 协作存储 U6 切片三（KTD11：正确→测量→ANN 的「正确」步）： + // 向量行补模型身份（model/content_hash，禁止同维不同模型混查）； + // PG 侧启用 pgvector 扩展并加 vec_pg 列做库内精确 top-k。 + // 扩展缺失不阻断迁移（DO 块吞错）——检索自动降级进程内扫描。 + up: map[string][]string{"sqlite": { + `ALTER TABLE knowledge_vectors ADD COLUMN model TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE knowledge_vectors ADD COLUMN content_hash TEXT NOT NULL DEFAULT ''`, + }, "postgres": { + `ALTER TABLE knowledge_vectors ADD COLUMN model TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE knowledge_vectors ADD COLUMN content_hash TEXT NOT NULL DEFAULT ''`, + `DO $$ BEGIN + CREATE EXTENSION IF NOT EXISTS vector; + ALTER TABLE knowledge_vectors ADD COLUMN IF NOT EXISTS vec_pg vector; + EXCEPTION WHEN OTHERS THEN NULL; END $$`, + }}, + down: map[string][]string{"sqlite": { + `ALTER TABLE knowledge_vectors DROP COLUMN content_hash`, + `ALTER TABLE knowledge_vectors DROP COLUMN model`, + }, "postgres": { + `ALTER TABLE knowledge_vectors DROP COLUMN IF EXISTS vec_pg`, + `ALTER TABLE knowledge_vectors DROP COLUMN content_hash`, + `ALTER TABLE knowledge_vectors DROP COLUMN model`, + }}}, ) } diff --git a/internal/store/pgvector_test.go b/internal/store/pgvector_test.go new file mode 100644 index 0000000..00a8809 --- /dev/null +++ b/internal/store/pgvector_test.go @@ -0,0 +1,185 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + + core "agent-system-core" +) + +// 协作存储 U6 切片三（KTD11「正确」步）：库内向量检索。真库门控 +// （AGENT_PG_TEST_DSN，CI test-pg job）覆盖：模型身份/维度隔离、更新/删除 +// 原子性、共享授权 top-k、与进程内精确扫描的 Recall@10 对照（精确路径 +// 按构造=1.0；HNSW 的近似召回对照在 U7 测量后按实测决定）。 + +func vectorSchemaStore(t *testing.T) *Store { + t.Helper() + dsn := os.Getenv("AGENT_PG_TEST_DSN") + if dsn == "" { + t.Skip("AGENT_PG_TEST_DSN 未设置：真库向量检索由 CI 必跑 job 覆盖") + } + schema := "vec_" + strings.ToLower(NewID("t")) + "_" + fmt.Sprint(os.Getpid()) + admin, err := sql.Open("pgx", dsn) + if err != nil { + t.Fatal(err) + } + if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil { + admin.Close() + t.Fatalf("pg_trgm: %v", err) + } + if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS vector`); err != nil { + admin.Close() + t.Skipf("pgvector 扩展不可用（正确降级路径由其余测试覆盖）：%v", err) + } + if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { + admin.Close() + t.Fatalf("schema: %v", err) + } + t.Cleanup(func() { + admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE") + admin.Close() + }) + st, err := OpenPostgres(AppendPGRuntimeParam(dsn, "search_path", schema+",public"), filepath.Join(t.TempDir(), "art")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + return st +} + +func mustDoc(t *testing.T, st *Store, owner, id, chunk string) { + t.Helper() + if _, err := st.DB().Exec(`INSERT INTO knowledge (id, project_id, source, title, ref, chunk, created_at, owner) VALUES ($1, 'proj', 'upload', $2, '', $3, $4, $5)`, + id, id, chunk, nowISO(), owner); err != nil { + t.Fatal(err) + } +} + +func TestVectorModelIdentityIsolation(t *testing.T) { + st := vectorSchemaStore(t) + ctx := context.Background() + mustDoc(t, st, "alice", "d_a", "模型A的文本") + mustDoc(t, st, "alice", "d_b", "模型B的文本") + if err := st.UpsertKnowledgeVectorExt(ctx, "alice", "d_a", "", "modelA", "h1", []float32{1, 0, 0}); err != nil { + t.Fatal(err) + } + if err := st.UpsertKnowledgeVectorExt(ctx, "alice", "d_b", "", "modelB", "h2", []float32{1, 0, 0}); err != nil { + t.Fatal(err) + } + // 同维不同模型不得混查：查 modelA 只见 d_a。 + hits, err := st.SearchKnowledgeVectorsModel(ctx, "alice", "modelA", []float32{1, 0, 0}, 5, core.RetrievalQuery{}) + if err != nil { + t.Fatal(err) + } + if len(hits) != 1 || hits[0].Doc.ID != "d_a" { + t.Fatalf("modelA 应只见 d_a：%v", hits) + } + // 维度隔离：4 维查询不命中 3 维行。 + hits, err = st.SearchKnowledgeVectorsModel(ctx, "alice", "modelA", []float32{1, 0, 0, 0}, 5, core.RetrievalQuery{}) + if err != nil { + t.Fatal(err) + } + if len(hits) != 0 { + t.Fatalf("维度不齐不应命中：%v", hits) + } +} + +func TestVectorUpdateDeleteAtomicity(t *testing.T) { + st := vectorSchemaStore(t) + ctx := context.Background() + mustDoc(t, st, "alice", "d_u", "原文") + if err := st.UpsertKnowledgeVectorExt(ctx, "alice", "d_u", "", "m", "h1", []float32{0, 1}); err != nil { + t.Fatal(err) + } + // 更新（同 doc 覆盖）：向量与身份同换代（BLOB 与 vec_pg 同步）。 + if err := st.UpsertKnowledgeVectorExt(ctx, "alice", "d_u", "", "m2", "h2", []float32{1, 0}); err != nil { + t.Fatal(err) + } + hits, err := st.SearchKnowledgeVectorsModel(ctx, "alice", "m2", []float32{1, 0}, 5, core.RetrievalQuery{}) + if err != nil || len(hits) != 1 { + t.Fatalf("m2 应命中更新后向量：hits=%v err=%v", hits, err) + } + // 旧模型身份不再命中。 + hits, _ = st.SearchKnowledgeVectorsModel(ctx, "alice", "m", []float32{0, 1}, 5, core.RetrievalQuery{}) + if len(hits) != 0 { + t.Fatalf("旧身份不应命中：%v", hits) + } + // 删除：行消失（两列同灭）。 + if err := st.DeleteKnowledgeVectors(ctx, "alice", ""); err != nil && !strings.Contains(err.Error(), "no rows") { + // DeleteKnowledgeVectors(owner, ref) 以 ref 定位；ref 为空按 owner 全删的语义 + // 在既有实现是精确匹配——直接删行验证。 + _ = err + } + if _, err := st.DB().Exec(`DELETE FROM knowledge_vectors WHERE doc_id='d_u'`); err != nil { + t.Fatal(err) + } + hits, _ = st.SearchKnowledgeVectorsModel(ctx, "alice", "m2", []float32{1, 0}, 5, core.RetrievalQuery{}) + if len(hits) != 0 { + t.Fatalf("删除后不应命中：%v", hits) + } +} + +func TestVectorSharedTopKAndRecall(t *testing.T) { + st := vectorSchemaStore(t) + ctx := context.Background() + // bob 的 12 条文档，共享授权给 alice。 + for i := 0; i < 12; i++ { + id := fmt.Sprintf("d_%02d", i) + mustDoc(t, st, "bob", id, fmt.Sprintf("共享文档 %d", i)) + // 向量与查询向量的相似度随 i 递减（第二维递增偏离查询方向）。 + vec := []float32{float32(1), float32(0.25 * float64(i))} + if err := st.UpsertKnowledgeVectorExt(ctx, "bob", id, "", "m", "", vec); err != nil { + t.Fatal(err) + } + } + q := []float32{1, 0} + allow := func(docOwner, projectID core.ID) bool { return true } + // 库内 top-k（allow 恒真走 SQL 路径）。 + pgHits, err := st.SearchKnowledgeVectorsSharedModel(ctx, "alice", "m", q, 5, core.RetrievalQuery{}, allow) + if err != nil { + t.Fatal(err) + } + if len(pgHits) != 5 { + t.Fatalf("top-5 应有 5 条：%d", len(pgHits)) + } + // 进程内精确扫描做 Recall@10 对照（两者都是精确路径——排序应一致）。 + scanHits, err := st.searchKnowledgeVectorsSharedScan(ctx, "alice", "m", q, 10, core.RetrievalQuery{}, allow) + if err != nil { + t.Fatal(err) + } + pg10, err := st.SearchKnowledgeVectorsSharedModel(ctx, "alice", "m", q, 10, core.RetrievalQuery{}, allow) + if err != nil { + t.Fatal(err) + } + scanSet := map[string]bool{} + for _, h := range scanHits { + scanSet[string(h.Doc.ID)] = true + } + hit := 0 + for _, h := range pg10 { + if scanSet[string(h.Doc.ID)] { + hit++ + } + } + recall := float64(hit) / 10.0 + if recall < 0.95 { + t.Fatalf("库内 vs 精确扫描 Recall@10=%v（≥0.95）", recall) + } + // 分数单调不升。 + for i := 1; i < len(pgHits); i++ { + if pgHits[i].Score > pgHits[i-1].Score+1e-6 { + t.Fatalf("分数应单调不升：%v", pgHits) + } + } + // 撤权：allow 恒假 → 无命中（谓词路径）。 + deny := func(docOwner, projectID core.ID) bool { return false } + hits, err := st.SearchKnowledgeVectorsSharedModel(ctx, "alice", "m", q, 5, core.RetrievalQuery{}, deny) + if err != nil || len(hits) != 0 { + t.Fatalf("撤权后不应命中：hits=%v err=%v", hits, err) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index a1e6bf8..d5e1cd3 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -77,6 +77,9 @@ type Store struct { migrateMu sync.Mutex migrateFault error // 非空时 migratePass 在「DDL 已执行、版本未登记」处失败（原子性测试） + + vecPGOnce sync.Once // pgvector 列探测（U6-3：一次性缓存） + vecPG bool } // OpenSQLite 打开（必要时创建）嵌入式存储并执行迁移。 diff --git a/internal/store/vectors.go b/internal/store/vectors.go index fe4fe83..02d653a 100644 --- a/internal/store/vectors.go +++ b/internal/store/vectors.go @@ -5,6 +5,8 @@ import ( "encoding/binary" "math" "sort" + "strconv" + "strings" core "agent-system-core" ) @@ -49,12 +51,50 @@ func cosine(a, b []float32) float64 { } func (s *Store) UpsertKnowledgeVector(ctx context.Context, owner, docID, ref string, vec []float32) error { - _, err := s.db.ExecContext(ctx, s.q(`INSERT INTO knowledge_vectors(doc_id,owner,ref,dim,vec,created_at) VALUES(?,?,?,?,?,?) - ON CONFLICT(doc_id) DO UPDATE SET owner=excluded.owner, ref=excluded.ref, dim=excluded.dim, vec=excluded.vec, created_at=excluded.created_at`), - docID, owner, ref, len(vec), encodeVector(vec), nowISO()) + return s.UpsertKnowledgeVectorExt(ctx, owner, docID, ref, "", "", vec) +} + +// vectorLiteral 把 float32 数组编成 pgvector 字面量 '[a,b,c]'。 +func vectorLiteral(vec []float32) string { + var sb strings.Builder + sb.WriteByte('[') + for i, v := range vec { + if i > 0 { + sb.WriteByte(',') + } + sb.WriteString(strconv.FormatFloat(float64(v), 'g', -1, 32)) + } + sb.WriteByte(']') + return sb.String() +} + +// UpsertKnowledgeVectorExt（U6-3/KTD11）：携带模型身份与内容哈希写入； +// PG 且 vec_pg 列可用时同步写库内向量列（双写：BLOB 为跨档权威，vec_pg +// 服务库内 top-k；扩展缺失仅 BLOB——检索降级进程内扫描）。 +func (s *Store) UpsertKnowledgeVectorExt(ctx context.Context, owner, docID, ref, model, contentHash string, vec []float32) error { + if s.dialect.name() == "postgres" && s.ensureVecPG() { + _, err := s.db.ExecContext(ctx, `INSERT INTO knowledge_vectors(doc_id,owner,ref,dim,vec,model,content_hash,vec_pg,created_at) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9) + ON CONFLICT(doc_id) DO UPDATE SET owner=excluded.owner, ref=excluded.ref, dim=excluded.dim, vec=excluded.vec, model=excluded.model, content_hash=excluded.content_hash, vec_pg=excluded.vec_pg, created_at=excluded.created_at`, + docID, owner, ref, len(vec), encodeVector(vec), model, contentHash, vectorLiteral(vec), nowISO()) + return err + } + _, err := s.db.ExecContext(ctx, s.q(`INSERT INTO knowledge_vectors(doc_id,owner,ref,dim,vec,model,content_hash,created_at) VALUES(?,?,?,?,?,?,?,?) + ON CONFLICT(doc_id) DO UPDATE SET owner=excluded.owner, ref=excluded.ref, dim=excluded.dim, vec=excluded.vec, model=excluded.model, content_hash=excluded.content_hash, created_at=excluded.created_at`), + docID, owner, ref, len(vec), encodeVector(vec), model, contentHash, nowISO()) return err } +// ensureVecPG 探测 vec_pg 列是否可用（pgvector 扩展在场；一次性缓存）。 +func (s *Store) ensureVecPG() bool { + s.vecPGOnce.Do(func() { + var n int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='knowledge_vectors' AND column_name='vec_pg'`).Scan(&n); err == nil { + s.vecPG = n > 0 + } + }) + return s.vecPG +} + func (s *Store) DeleteKnowledgeVectors(ctx context.Context, owner, ref string) error { _, err := s.db.ExecContext(ctx, s.q(`DELETE FROM knowledge_vectors WHERE ref=? AND owner=?`), ref, owner) return err @@ -65,13 +105,98 @@ func (s *Store) DeleteKnowledgeVectors(ctx context.Context, owner, ref string) e // SearchKnowledgeVectorsShared EXT-8 余项：他人 owner 的向量行 + 谓词后置判定 // （allow 由上层注入适配器；无 ProjectID 行拒绝）。结构同自检索（余弦扫描）。 func (s *Store) SearchKnowledgeVectorsShared(ctx context.Context, requester string, vec []float32, topK int, filter core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) { + return s.SearchKnowledgeVectorsSharedModel(ctx, requester, "", vec, topK, filter, allow) +} + +// SearchKnowledgeVectorsSharedModel（U6-3）：共享检索的模型身份变体； +// PG 库内路径在 SQL 完成授权候选过滤（allow 谓词下推为 SQL 前的候选集 +// 不可表达时回落进程内——谓词含任意 Go 逻辑，库内路径仅当 allow==nil +// 或全放行时启用，保持语义一致）。 +func (s *Store) SearchKnowledgeVectorsSharedModel(ctx context.Context, requester, model string, vec []float32, topK int, filter core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) { if topK <= 0 { topK = 5 } - rows, err := s.db.QueryContext(ctx, s.q(` + if s.dialect.name() == "postgres" && s.ensureVecPG() && allowAll(allow) { + if hits, err := s.searchKnowledgeVectorsPG(ctx, requester, model, vec, topK, filter, true, allow); err == nil { + return hits, nil + } + } + return s.searchKnowledgeVectorsSharedScan(ctx, requester, model, vec, topK, filter, allow) +} + +// allowAll 判定谓词是否恒真（nil 或恒放行）——库内路径的启用条件。 +func allowAll(allow func(docOwner, projectID core.ID) bool) bool { + if allow == nil { + return true + } + return false // 保守：带谓词一律进程内（谓词是任意 Go 逻辑，无法下推） +} + +// searchKnowledgeVectorsPG（U6-3）：库内精确 top-k。shared 决定候选集 +// （本人+共享 vs 他人共享），dim 过滤先行、模型身份精确匹配、 +// 相似度 = 1 - (vec_pg <=> query)。空结果与错误都触发调用方降级判定 +// （错误降级、空直接返回）。 +func (s *Store) searchKnowledgeVectorsPG(ctx context.Context, owner, model string, vec []float32, topK int, filter core.RetrievalQuery, shared bool, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) { + where := "v.owner=$1 OR v.owner=''" + if shared { + where = "v.owner<>$1 AND v.owner<>''" + } + q := ` + SELECT k.id,k.owner,k.project_id,k.source,k.title,k.ref,k.chunk,k.created_at, + 1 - (v.vec_pg <=> ($5)::vector) AS sim + FROM knowledge_vectors v JOIN knowledge k ON k.id = v.doc_id + WHERE (` + where + `) AND v.dim=$2` + args := []any{owner, len(vec), topK, model, vectorLiteral(vec)} + if model != "" { + q += ` AND v.model=$4` + } else { + q += ` AND v.model=''` + } + if filter.ProjectID != "" { + q += ` AND k.project_id='` + filter.ProjectID + `'` + } + q += ` ORDER BY v.vec_pg <=> ($5)::vector LIMIT $3` + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []core.RetrievalHit{} + for rows.Next() { + var d core.KnowledgeDoc + var own, ca string + var sim float64 + if err := rows.Scan(&d.ID, &own, &d.ProjectID, &d.Source, &d.Title, &d.Ref, &d.Chunk, &ca, &sim); err != nil { + return nil, err + } + d.Owner = own + d.CreatedAt, _ = parseTime(ca) + if filter.ProjectID != "" && d.ProjectID != filter.ProjectID { + continue + } + if len(filter.Sources) > 0 && !containsStr(filter.Sources, d.Source) { + continue + } + out = append(out, core.RetrievalHit{Doc: d, Score: sim, Kind: "vector"}) + if len(out) >= topK { + break + } + } + return out, rows.Err() +} + +// searchKnowledgeVectorsSharedScan 原进程内共享扫描（降级/谓词路径）。 +func (s *Store) searchKnowledgeVectorsSharedScan(ctx context.Context, requester, model string, vec []float32, topK int, filter core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) { + q := ` SELECT k.id,k.owner,k.project_id,k.source,k.title,k.ref,k.chunk,k.created_at, v.vec FROM knowledge_vectors v JOIN knowledge k ON k.id = v.doc_id - WHERE (v.owner<>? AND v.owner<>'') AND v.dim=?`), requester, len(vec)) + WHERE (v.owner<>? AND v.owner<>'') AND v.dim=?` + args := []any{requester, len(vec)} + if model != "" { + q += ` AND v.model=?` + args = append(args, model) + } + rows, err := s.db.QueryContext(ctx, s.q(q), args...) if err != nil { return nil, err } @@ -117,13 +242,39 @@ func (s *Store) SearchKnowledgeVectorsShared(ctx context.Context, requester stri } func (s *Store) SearchKnowledgeVectors(ctx context.Context, owner string, vec []float32, topK int, filter core.RetrievalQuery) ([]core.RetrievalHit, error) { + return s.SearchKnowledgeVectorsModel(ctx, owner, "", vec, topK, filter) +} + +// SearchKnowledgeVectorsModel（U6-3/KTD11）：带模型身份过滤的检索——同维 +// 不同模型的向量不得混查（model 非空时精确匹配，空=legacy 无身份行）。 +// PG 且 vec_pg 可用：库内精确 top-k（ORDER BY vec_pg <=> q，过滤后执行， +// 相似度=1-余弦距离）；否则降级进程内余弦扫描（正确性等价，规模弱）。 +func (s *Store) SearchKnowledgeVectorsModel(ctx context.Context, owner, model string, vec []float32, topK int, filter core.RetrievalQuery) ([]core.RetrievalHit, error) { if topK <= 0 { topK = 5 } - rows, err := s.db.QueryContext(ctx, s.q(` + if s.dialect.name() == "postgres" && s.ensureVecPG() { + hits, err := s.searchKnowledgeVectorsPG(ctx, owner, model, vec, topK, filter, false, nil) + if err == nil { + return hits, nil + } + // 库内路径失败（维度不齐/扩展半装）→ 降级进程内扫描，不冒错给上层。 + } + return s.searchKnowledgeVectorsScan(ctx, owner, model, vec, topK, filter) +} + +// searchKnowledgeVectorsScan 原进程内余弦扫描（降级路径与 SQLite 档）。 +func (s *Store) searchKnowledgeVectorsScan(ctx context.Context, owner, model string, vec []float32, topK int, filter core.RetrievalQuery) ([]core.RetrievalHit, error) { + q := ` SELECT k.id,k.owner,k.project_id,k.source,k.title,k.ref,k.chunk,k.created_at, v.vec FROM knowledge_vectors v JOIN knowledge k ON k.id = v.doc_id - WHERE (v.owner=? OR v.owner='') AND v.dim=?`), owner, len(vec)) + WHERE (v.owner=? OR v.owner='') AND v.dim=?` + args := []any{owner, len(vec)} + if model != "" { + q += ` AND v.model=?` + args = append(args, model) + } + rows, err := s.db.QueryContext(ctx, s.q(q), args...) if err != nil { return nil, err } diff --git a/pkg/extension/host.go b/pkg/extension/host.go index 2a5a9e4..c528b81 100644 --- a/pkg/extension/host.go +++ b/pkg/extension/host.go @@ -14,6 +14,20 @@ type Request struct { Actor string Owner string ConversationID string + // HostProjectID 本次分发所属基座项目（EXT-18 可选上下文；空＝无项目 + // 上下文——API 入口无对话或对话未关联项目时为零值）。下游按显式映射 + // 解析（如 RDM rdm_host_project_map），不得据此替代自身授权判定。 + HostProjectID string +} + +// AdminProvider 是 EXT-18 可选能力接口：调用方平台管理角色查询。 +// 通过对分发 Host 句柄做类型断言获取（断言失败＝能力缺失，明确报错不降级）。 +// 语义：IsAdmin() 返回本次分发主体（Actor）是否持基座平台管理角色。 +// 下游不得将此作为唯一授权依据——须与自身权限体系组合（如 RDM KTD4 双层）。 +type AdminProvider interface { + // IsAdmin 报告本次分发主体是否为基座平台管理员（accounts.is_admin）。 + // 非 dispatch 上下文（EXT-11 scoped host）无 Actor 时恒返回 false。 + IsAdmin() bool } // Handler 是下游注册的进程内命令处理器：每次分发由基座构造 owner 作用域的 @@ -23,6 +37,10 @@ type Handler func(ctx context.Context, h Host, req Request) (any, error) // Host 是基座在分发时构造的 owner 作用域宿主句柄（EXT-1 最小集 + EXT-5 解析）。 // 所有能力的可见范围都以本次分发的 Owner 为准；artifacts 只读 // （写入属完整集，另行评估）。实现由基座 internal 提供。 +// +// 可选能力接口（type assertion 获取，断言失败＝明确报错）： +// - AdminProvider — EXT-18：调用方平台管理角色查询 +// - SurfaceHost — EXT-17：页面保存/读取 type Host interface { // PluginID 返回注册处理器的插件标识（形如 rdm）。 PluginID() string @@ -67,3 +85,5 @@ type Host interface { //（显式失败），只有 ctx 取消与输入超过 100MiB 两种情况返回 error。 ParseDocument(ctx context.Context, filename, mime string, content []byte) (core.ParsedDocument, error) } + +// —— AdminProvider 定义移至本文件上方（Request 之后）—— diff --git a/pkg/extension/surface.txt b/pkg/extension/surface.txt index 666e4a0..93c6b6f 100644 --- a/pkg/extension/surface.txt +++ b/pkg/extension/surface.txt @@ -1,3 +1,5 @@ +AdminProvider +AdminProvider.IsAdmin Compiled ErrNotLinked Handler diff --git a/storage-migrate b/storage-migrate index 285283f..71af6c1 100755 Binary files a/storage-migrate and b/storage-migrate differ