Merge authoritative main into the task branch
ci / vet-and-test (pull_request) Successful in 42s

This commit is contained in:
2026-10-11 17:37:51 +00:00
18 changed files with 576 additions and 14 deletions
+5 -3
View File
@@ -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
}
+2 -1
View File
@@ -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,
})
})
}
+62
View File
@@ -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("字段赋值失败")
}
}
+17
View File
@@ -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)
}
+1
View File
@@ -91,6 +91,7 @@ type Request struct {
ConversationID string
AnchorMessageID string // 确认卡锚定的发起消息(内联渲染)
Confirmed bool // 危险命令经确认卡放行后的执行
HostProjectID string // EXT-18:分发所属基座项目(空=无项目上下文)
}
// Response 分发结果:回执 + 处理负载 + 可能的确认卡。
+43
View File
@@ -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)
+16
View File
@@ -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)
}
}
+6
View File
@@ -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
+3
View File
@@ -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 {
+21 -3
View File
@@ -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
}
+8
View File
@@ -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 {
+24
View File
@@ -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`,
}}},
)
}
+185
View File
@@ -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)
}
}
+3
View File
@@ -77,6 +77,9 @@ type Store struct {
migrateMu sync.Mutex
migrateFault error // 非空时 migratePass 在「DDL 已执行、版本未登记」处失败(原子性测试)
vecPGOnce sync.Once // pgvector 列探测(U6-3:一次性缓存)
vecPG bool
}
// OpenSQLite 打开(必要时创建)嵌入式存储并执行迁移。
+158 -7
View File
@@ -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
}
+20
View File
@@ -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 之后)——
+2
View File
@@ -1,3 +1,5 @@
AdminProvider
AdminProvider.IsAdmin
Compiled
ErrNotLinked
Handler
BIN
View File
Binary file not shown.