From 92ede3b970dd8bc0a40b3359a04a6e35b812a9d1 Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 8 Oct 2026 17:16:10 +0000 Subject: [PATCH] Close the EXT-8 remaining blocks: vector sharing, hostapi face, citation consistency Why: EXT-8 closure needs the vector channel shared (not just fulltext), the /host/v1 knowledge face honoring the authorizer, and citation-ref-to-artifact consistency pinned as an acceptance test. Evidence: TestRetrieveSharedVectorChannel (hash embedder: authorized shared vector hit retrievable, denied yields zero), TestSharedSearchFace (hostapi switches to SharedSearch when wired), TestSharedCitationRefDownloadable (shared hit -> Ref -> artifact content download for the authorized member); go test ./... 32/32 packages ok. Impact: store.SearchKnowledgeVectorsShared; rag.Pipeline.RetrieveShared (own + shared fulltext + shared vector, RRF-merged); server searchKnowledge prefers RetrieveShared when the retriever implements it; hostapi Server.SharedSearch wired in the composition root through the same authorizer slot. Default behavior unchanged. Refs: EXT-8 proposal acceptance; slices 1-2. --- internal/app/app.go | 15 +++++ internal/hostapi/helper_test.go | 5 ++ internal/hostapi/server.go | 12 +++- internal/hostapi/shared_search_test.go | 49 ++++++++++++++ internal/rag/hybrid.go | 32 ++++++++++ internal/rag/shared_retrieval_test.go | 53 ++++++++++++++++ internal/server/server.go | 18 ++++-- internal/server/shared_retrieval_test.go | 81 ++++++++++++++++++++++++ internal/store/vectors.go | 55 ++++++++++++++++ 9 files changed, 313 insertions(+), 7 deletions(-) create mode 100644 internal/hostapi/helper_test.go create mode 100644 internal/hostapi/shared_search_test.go create mode 100644 internal/rag/shared_retrieval_test.go diff --git a/internal/app/app.go b/internal/app/app.go index a2b493e..6d1e1cb 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -403,6 +403,21 @@ func Run(opt Options) error { hostSrv.Search = func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) { return caps.searchDefault(ctx, owner, text) } + // EXT-8 余项：hostapi 检索面同样吃适配器共享（本人 + 放行的他人项目知识）。 + hostSrv.SharedSearch = func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) { + own, err := caps.searchDefault(ctx, owner, text) + if err != nil { + return nil, err + } + shared, err := st.SearchKnowledgeShared(owner, core.RetrievalQuery{Text: text, TopK: 5}, func(docOwner, projectID core.ID) bool { + ok, aerr := authzSlot.Authorize(owner, projectID) + return aerr == nil && ok + }) + if err != nil { + return nil, err + } + return append(own, shared...), nil + } hostSrv.Complete = func(ctx context.Context, owner, prompt string) (string, error) { return caps.complete(ctx, owner, prompt) } diff --git a/internal/hostapi/helper_test.go b/internal/hostapi/helper_test.go new file mode 100644 index 0000000..359f2af --- /dev/null +++ b/internal/hostapi/helper_test.go @@ -0,0 +1,5 @@ +package hostapi + +import "encoding/json" + +func jsonUnmarshal(b []byte, v any) error { return json.Unmarshal(b, v) } diff --git a/internal/hostapi/server.go b/internal/hostapi/server.go index 84bb91f..53aaa08 100644 --- a/internal/hostapi/server.go +++ b/internal/hostapi/server.go @@ -18,7 +18,10 @@ type Server struct { Store *store.Store Dispatch func(ctx context.Context, pluginID, owner, commandID string, params map[string]any) error Search func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) - Complete func(ctx context.Context, owner, prompt string) (string, error) + // SharedSearch EXT-8 余项：非 nil 时 search 端点改用之（组合根传入 + // 「本人结果 + 适配器放行的共享追加」组合；nil=现状仅本人）。 + SharedSearch func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) + Complete func(ctx context.Context, owner, prompt string) (string, error) } func (s *Server) Handler() http.Handler { @@ -143,7 +146,12 @@ func (s *Server) search(w http.ResponseWriter, r *http.Request) { http.Error(w, "retriever unavailable", http.StatusServiceUnavailable) return } - hits, err := s.Search(r.Context(), s.owner(r), r.URL.Query().Get("q")) + // EXT-8 余项：共享版可用时整体切换（本人+共享追加）。 + fn := s.Search + if s.SharedSearch != nil { + fn = s.SharedSearch + } + hits, err := fn(r.Context(), s.owner(r), r.URL.Query().Get("q")) if err != nil { http.Error(w, err.Error(), http.StatusBadGateway) return diff --git a/internal/hostapi/shared_search_test.go b/internal/hostapi/shared_search_test.go new file mode 100644 index 0000000..6cfe340 --- /dev/null +++ b/internal/hostapi/shared_search_test.go @@ -0,0 +1,49 @@ +package hostapi + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + core "agent-system-core" +) + +// EXT-8 余项：hostapi 检索面——SharedSearch 可用时整体切换（本人+适配器放行的 +// 共享追加）；不可用时维持现状（仅本人回调）。 +func TestSharedSearchFace(t *testing.T) { + own := []core.RetrievalHit{{Doc: core.KnowledgeDoc{Owner: "bob", Chunk: "本人命中"}}} + sharedHit := core.RetrievalHit{Doc: core.KnowledgeDoc{Owner: "alice", ProjectID: "prj-s", Chunk: "共享命中"}} + + srv := &Server{Token: "t", Search: func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) { + return own, nil + }} + run := func() int { + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/host/v1/knowledge/search?q=x", nil) + req.Header.Set("Authorization", "Bearer "+srv.Token) + srv.Handler().ServeHTTP(rec, req) + return rec.Code + } + if code := run(); code != http.StatusOK { + t.Fatalf("plain search: %d", code) + } + + srv.SharedSearch = func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) { + return append(own, sharedHit), nil + } + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/host/v1/knowledge/search?q=x", nil) + req.Header.Set("Authorization", "Bearer "+srv.Token) + srv.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("shared search: %d %s", rec.Code, rec.Body.String()) + } + var hits []core.RetrievalHit + if err := jsonUnmarshal(rec.Body.Bytes(), &hits); err != nil { + t.Fatal(err) + } + if len(hits) != 2 || hits[1].Doc.Owner != "alice" { + t.Fatalf("hits = %+v, want 本人+共享", hits) + } +} diff --git a/internal/rag/hybrid.go b/internal/rag/hybrid.go index 2f8a502..542fe08 100644 --- a/internal/rag/hybrid.go +++ b/internal/rag/hybrid.go @@ -124,6 +124,38 @@ func firstRef(docs []core.KnowledgeDoc) string { return "" } +// RetrieveShared EXT-8 余项：本人两路如常 + 追加共享两路（向量 Shared + 全文 +// Shared，allow=适配器判定），RRF 融合后裁剪 topK。embedder 不可用时共享仅全文 +// （与切片 2 行为一致）。 +func (p *Pipeline) RetrieveShared(ctx context.Context, owner core.ID, q core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) { + own, err := p.Retrieve(ctx, owner, q) + if err != nil { + return nil, err + } + shared, err := p.st.SearchKnowledgeShared(string(owner), q, allow) + if err != nil { + return nil, err + } + // 向量共享路：embedder 可用时并入。 + if p.embedder != nil && p.embeddingRoute(ctx).OK() { + if qvecs, e := p.embedder.Embed(ctx, []string{q.Text}); e == nil && len(qvecs) > 0 && len(qvecs[0]) > 0 { + topK := q.TopK + if topK <= 0 { + topK = 5 + } + vs, e2 := p.st.SearchKnowledgeVectorsShared(ctx, string(owner), qvecs[0], topK, q, allow) + if e2 == nil { + shared = rrfMerge(vs, shared) + } + } + } + merged := rrfMerge(own, shared) + if limit := q.TopK; limit > 0 && len(merged) > limit { + merged = merged[:limit] + } + return merged, nil +} + // Retrieve 统一检索（core.Retriever）。 func (p *Pipeline) Retrieve(ctx context.Context, owner core.ID, q core.RetrievalQuery) ([]core.RetrievalHit, error) { topK := q.TopK diff --git a/internal/rag/shared_retrieval_test.go b/internal/rag/shared_retrieval_test.go new file mode 100644 index 0000000..9118d69 --- /dev/null +++ b/internal/rag/shared_retrieval_test.go @@ -0,0 +1,53 @@ +package rag + +import ( + "context" + "path/filepath" + "testing" + + "agent-system-backend/internal/settings" + "agent-system-backend/internal/store" + core "agent-system-core" +) + +// EXT-8 余项红/绿：向量通道共享——他人 owner 的向量命中经 allow 放行可检。 +// 用 hash 档 embedder（确定性向量），owner=alice 建项目知识并嵌入；bob 经 +// RetrieveShared 放行后命中（含 vector Kind），拒绝时零命中。 +func TestRetrieveSharedVectorChannel(t *testing.T) { + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "t.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + set := settings.NewService(st) + p := NewPipeline(st, set, nil, nil) + p.SetEmbedder(SwitchingEmbedder{Provider: func() string { return "hash" }}) + + docs := []core.KnowledgeDoc{{Owner: "alice", ProjectID: "prj-s", Source: "plugin:rdm", + Ref: "kb-v", Chunk: "向量共享独占词"}} + // 只走管线写入（分块行+向量一体；不再手工 AddKnowledge 避免主键冲突）。 + if err := indexForTest(t, st, p, docs); err != nil { + t.Fatal(err) + } + + allow := func(docOwner, projectID core.ID) bool { + return docOwner == "alice" && projectID == "prj-s" + } + hits, err := p.RetrieveShared(context.Background(), "bob", core.RetrievalQuery{Text: "向量共享独占词", TopK: 5}, allow) + if err != nil { + t.Fatal(err) + } + if len(hits) == 0 { + t.Fatal("authorized shared vector hit must be retrievable（向量通道共享）") + } + deny := func(docOwner, projectID core.ID) bool { return false } + if hits2, _ := p.RetrieveShared(context.Background(), "bob", core.RetrievalQuery{Text: "向量共享独占词", TopK: 5}, deny); len(hits2) != 0 { + t.Fatalf("denied shared search = %d hits, want 0", len(hits2)) + } +} + +// indexForTest：把 docs 直接嵌入并落向量（复用管线的公开写入路径）。 +func indexForTest(t *testing.T, st *store.Store, p *Pipeline, docs []core.KnowledgeDoc) error { + t.Helper() + return p.IndexChunks(context.Background(), "alice", docs) +} diff --git a/internal/server/server.go b/internal/server/server.go index 6b9127b..40a0e95 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1474,6 +1474,18 @@ func (s *Server) searchKnowledge(w http.ResponseWriter, r *http.Request) { topK, _ := strconv.Atoi(r.URL.Query().Get("topK")) query := core.RetrievalQuery{ProjectID: r.URL.Query().Get("project"), Text: q, TopK: topK} owner := s.owner(r) + // EXT-8 余项：Retriever 支持共享版（向量+全文两路追加）时整体走共享版； + // 否则维持切片 2 行为（本人检索 + 全文共享追加）。 + allow := func(docOwner, projectID core.ID) bool { + return s.authorizeSharedProject(owner, projectID) + } + if rs, ok := s.Retriever.(interface { + RetrieveShared(ctx context.Context, owner core.ID, q core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) + }); ok && s.Retriever != nil { + hits, err := rs.RetrieveShared(r.Context(), core.ID(owner), query, allow) + writeJSONOr(w, hits, err) + return + } var hits []core.RetrievalHit var err error if s.Retriever != nil { @@ -1485,11 +1497,7 @@ func (s *Server) searchKnowledge(w http.ResponseWriter, r *http.Request) { writeStoreErr(w, err) return } - // EXT-8 切片 2：追加「他人 owner 且其 ProjectID 经适配器放行」的共享命中 - //（全文通道；本人命中在前，合计截断到 topK）。 - shared, serr := s.st.SearchKnowledgeShared(owner, query, func(docOwner, projectID core.ID) bool { - return s.authorizeSharedProject(owner, projectID) - }) + shared, serr := s.st.SearchKnowledgeShared(owner, query, allow) if serr != nil { writeStoreErr(w, serr) return diff --git a/internal/server/shared_retrieval_test.go b/internal/server/shared_retrieval_test.go index 5102b42..f4a1327 100644 --- a/internal/server/shared_retrieval_test.go +++ b/internal/server/shared_retrieval_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "mime/multipart" "net/http" "net/http/httptest" "testing" @@ -140,3 +141,83 @@ func TestProjectSSEScopeGate(t *testing.T) { t.Fatalf("alice own project: %d, want 200（现状不受影响）", c) } } + +// EXT-8 余项（引用一致性）：适配器放行后，共享检索命中的 Ref（工件 ID）可经 +// /api/artifacts/{id}/content 取回——原件/搜索/引用/下载四口径一致。 +func TestSharedCitationRefDownloadable(t *testing.T) { + srv, st := newAuthTestServer(t) + h := srv.Handler() + alice, err := st.CreateAPIKey("alice", "e2e") + if err != nil { + t.Fatal(err) + } + bob, err := st.CreateAPIKey("bob", "e2e") + if err != nil { + t.Fatal(err) + } + req := httptest.NewRequest(http.MethodPost, "/api/projects", bytes.NewReader(mustJSON2(t, map[string]string{"name": "引用链"}))) + req.Header.Set("Authorization", "Bearer "+alice) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + var pj map[string]any + readJSON(t, rec.Body.Bytes(), &pj) + pid := fmt.Sprint(pj["id"]) + + // alice 上传原件（走真实 upload，含索引），并以其 artifactId 为共享知识的 Ref。 + up := uploadMultipart(t, h, alice, pid, "引用源.txt", "引用链独占词 原文内容") + // 上传端点已把可解析文本入索引（Retriever=nil 走 AddKnowledge，Source=upload， + // owner=alice、project=pid）；bob 未授权时零命中。 + if n := searchCount(t, h, bob, "引用链独占词"); n != 0 { + t.Fatalf("bob before grant = %d, want 0", n) + } + srv.SetProjectAuthorizer(func(owner string, projectID core.ID) (bool, error) { + return owner == "bob" && string(projectID) == pid, nil + }) + if n := searchCount(t, h, bob, "引用链独占词"); n != 1 { + t.Fatalf("bob after grant = %d, want 1（共享命中，Ref=工件）", n) + } + // 共享命中的 Ref（=上传工件 ID）可下载（引用→原件一致）。 + reqC := httptest.NewRequest(http.MethodGet, "/api/artifacts/"+up+"/content", nil) + reqC.Header.Set("Authorization", "Bearer "+bob) + recC := httptest.NewRecorder() + h.ServeHTTP(recC, reqC) + if recC.Code != http.StatusOK || !bytes.Contains(recC.Body.Bytes(), []byte("原文内容")) { + t.Fatalf("shared citation ref download: %d %q", recC.Code, recC.Body.String()) + } +} + +func uploadMultipart(t *testing.T, h http.Handler, key, pid, name, text string) string { + t.Helper() + var buf bytes.Buffer + mw := multipart.NewWriter(&buf) + fw, _ := mw.CreateFormFile("file", name) + fw.Write([]byte(text)) + mw.Close() + req := httptest.NewRequest(http.MethodPost, "/api/projects/"+pid+"/uploads", &buf) + req.Header.Set("Authorization", "Bearer "+key) + req.Header.Set("Content-Type", mw.FormDataContentType()) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("upload: %d %s", rec.Code, rec.Body.String()) + } + var up struct { + ArtifactID string `json:"artifactId"` + } + readJSON(t, rec.Body.Bytes(), &up) + return up.ArtifactID +} + +func searchCount(t *testing.T, h http.Handler, key, q string) int { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/api/knowledge/search?q=%E5%BC%95%E7%94%A8%E9%93%BE%E7%8B%AC%E5%8D%A0%E8%AF%8D", nil) + req.Header.Set("Authorization", "Bearer "+key) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("search: %d %s", rec.Code, rec.Body.String()) + } + var hits []any + readJSON(t, rec.Body.Bytes(), &hits) + return len(hits) +} diff --git a/internal/store/vectors.go b/internal/store/vectors.go index ba1a5cd..fe4fe83 100644 --- a/internal/store/vectors.go +++ b/internal/store/vectors.go @@ -4,6 +4,7 @@ import ( "context" "encoding/binary" "math" + "sort" core "agent-system-core" ) @@ -61,6 +62,60 @@ func (s *Store) DeleteKnowledgeVectors(ctx context.Context, owner, ref string) e // SearchKnowledgeVectors 余弦扫描：候选 = 租户本人 + 共享（owner=”），dim 匹配， // 再按 RetrievalQuery 的项目/来源过滤。 +// 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) { + if topK <= 0 { + topK = 5 + } + rows, err := s.db.QueryContext(ctx, s.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)) + if err != nil { + return nil, err + } + defer rows.Close() + type cand struct { + hit core.RetrievalHit + sim float64 + } + var cands []cand + for rows.Next() { + var d core.KnowledgeDoc + var own, ca string + var blob []byte + if err := rows.Scan(&d.ID, &own, &d.ProjectID, &d.Source, &d.Title, &d.Ref, &d.Chunk, &ca, &blob); err != nil { + return nil, err + } + d.Owner = own + d.CreatedAt, _ = parseTime(ca) + if d.ProjectID == "" || !allow(d.Owner, d.ProjectID) { + continue + } + if filter.ProjectID != "" && d.ProjectID != filter.ProjectID { + continue + } + if len(filter.Sources) > 0 && !containsStr(filter.Sources, d.Source) { + continue + } + sim := cosine(vec, decodeVector(blob)) + cands = append(cands, cand{hit: core.RetrievalHit{Doc: d, Score: sim, Kind: "vector"}, sim: sim}) + } + if err := rows.Err(); err != nil { + return nil, err + } + sort.SliceStable(cands, func(i, j int) bool { return cands[i].sim > cands[j].sim }) + out := []core.RetrievalHit{} + for _, c := range cands { + out = append(out, c.hit) + if len(out) >= topK { + break + } + } + return out, nil +} + func (s *Store) SearchKnowledgeVectors(ctx context.Context, owner string, vec []float32, topK int, filter core.RetrievalQuery) ([]core.RetrievalHit, error) { if topK <= 0 { topK = 5