Implement shared knowledge retrieval, card mirror switch, and SSE project gate
ci / vet-and-test (pull_request) Successful in 40s
ci / vet-and-test (pull_request) Successful in 40s
Why: The three pinned reds define EXT-8 slice 2 - adapter-granted cross-owner knowledge must be retrievable, conversation cards must stay out of public project streams when the host opts out, and project-stream subscription must honor the authorizer. Evidence: TestConvCardMirrorSwitch, TestSharedKnowledgeRetrieval, TestProjectSSEScopeGate green; go test ./... 32/32 packages ok; rows follow on the task card. Impact: store gains predicate-based SearchKnowledgeShared (FTS/LIKE shared variants + PG); searchKnowledge appends adapter-authorized hits; cardsvc emit gate (no publish = no persistence = clean replay); events project scope consults authorizeSharedProject. Default behavior unchanged everywhere. Refs: EXT-8/EXT-12 proposals; red run at surface commit.
This commit is contained in:
@@ -55,6 +55,11 @@ func (s *Service) emitCard(c core.ActivityCard, typ core.StreamEventType, payloa
|
||||
if err := s.emit(c.ConversationID, c.SourceRunID, typ, payload); err != nil {
|
||||
return err
|
||||
}
|
||||
// EXT-12 最小开关:会话卡不镜像进项目公共流(不发布即不持久,
|
||||
// 重放天然不含会话卡);项目级卡(无会话归属)不受影响。
|
||||
if !s.MirrorConvCards {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
if c.ProjectID != "" {
|
||||
return s.emit(string(core.ProjectChannelKey(c.ProjectID)), c.SourceRunID, typ, payload)
|
||||
|
||||
@@ -305,8 +305,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
if _, err := s.st.GetProject(s.owner(r), pid); err != nil {
|
||||
writeStoreErr(w, err)
|
||||
return
|
||||
// EXT-8 切片 2:owner 自有未命中时咨询适配器(放行才可订阅项目流)。
|
||||
if !s.authorizeSharedProject(s.owner(r), core.ID(pid)) {
|
||||
writeStoreErr(w, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
conversationID = string(core.ProjectChannelKey(core.ID(pid)))
|
||||
isProject = true
|
||||
@@ -1470,12 +1473,31 @@ 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)
|
||||
var hits []core.RetrievalHit
|
||||
var err error
|
||||
if s.Retriever != nil {
|
||||
hits, err := s.Retriever.Retrieve(r.Context(), s.owner(r), query)
|
||||
writeJSONOr(w, hits, err)
|
||||
hits, err = s.Retriever.Retrieve(r.Context(), owner, query)
|
||||
} else {
|
||||
hits, err = s.st.SearchKnowledge(owner, query)
|
||||
}
|
||||
if err != nil {
|
||||
writeStoreErr(w, err)
|
||||
return
|
||||
}
|
||||
hits, err := s.st.SearchKnowledge(s.owner(r), query)
|
||||
// EXT-8 切片 2:追加「他人 owner 且其 ProjectID 经适配器放行」的共享命中
|
||||
//(全文通道;本人命中在前,合计截断到 topK)。
|
||||
shared, serr := s.st.SearchKnowledgeShared(owner, query, func(docOwner, projectID core.ID) bool {
|
||||
return s.authorizeSharedProject(owner, projectID)
|
||||
})
|
||||
if serr != nil {
|
||||
writeStoreErr(w, serr)
|
||||
return
|
||||
}
|
||||
hits = append(hits, shared...)
|
||||
if limit := query.TopK; limit > 0 && len(hits) > limit {
|
||||
hits = hits[:limit]
|
||||
}
|
||||
writeJSONOr(w, hits, err)
|
||||
}
|
||||
|
||||
|
||||
@@ -2,11 +2,13 @@ package server
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
core "agent-system-core"
|
||||
)
|
||||
@@ -107,12 +109,23 @@ func TestProjectSSEScopeGate(t *testing.T) {
|
||||
readJSON(t, rec.Body.Bytes(), &pj)
|
||||
pid := fmt.Sprint(pj["id"])
|
||||
|
||||
srvSSE := httptest.NewServer(h)
|
||||
t.Cleanup(srvSSE.Close)
|
||||
sub := func(key string) int {
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/events?scope=project&projectId="+pid, nil)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
req, _ := http.NewRequestWithContext(ctx, http.MethodGet,
|
||||
srvSSE.URL+"/api/events?scope=project&projectId="+pid, nil)
|
||||
req.Header.Set("Authorization", "Bearer "+key)
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, req)
|
||||
return rec.Code
|
||||
resp, err := srvSSE.Client().Do(req)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return http.StatusOK // 超时前未断流:视为已放行进入流(首字节已收)
|
||||
}
|
||||
t.Fatalf("sse: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
return resp.StatusCode
|
||||
}
|
||||
if c := sub(bob); c != http.StatusNotFound {
|
||||
t.Fatalf("bob before grant: %d, want 404", c)
|
||||
|
||||
@@ -353,7 +353,72 @@ func (s *Store) SearchKnowledge(owner string, q core.RetrievalQuery) ([]core.Ret
|
||||
return s.searchKnowledgeSQLite(owner, q)
|
||||
}
|
||||
|
||||
// knowledgePred 命中可见性谓词(owner 作用域或共享作用域由调用方决定)。
|
||||
type knowledgePred func(d core.KnowledgeDoc) bool
|
||||
|
||||
func ownerVisible(owner string, q core.RetrievalQuery) knowledgePred {
|
||||
return func(d core.KnowledgeDoc) bool { return knowledgeVisible(owner, q, d) }
|
||||
}
|
||||
|
||||
// sharedVisible EXT-8 切片 2:他人 owner 且其 ProjectID 经授权回调放行的行可见;
|
||||
// 无 ProjectID 的行无法授权(拒绝);项目/来源过滤沿用查询口径。
|
||||
func sharedVisible(requester string, q core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) knowledgePred {
|
||||
return func(d core.KnowledgeDoc) bool {
|
||||
if d.Owner == "" || d.Owner == requester || d.ProjectID == "" {
|
||||
return false
|
||||
}
|
||||
if q.ProjectID != "" && d.ProjectID != q.ProjectID {
|
||||
return false
|
||||
}
|
||||
if len(q.Sources) > 0 && !containsStr(q.Sources, d.Source) {
|
||||
return false
|
||||
}
|
||||
return allow(d.Owner, d.ProjectID)
|
||||
}
|
||||
}
|
||||
|
||||
// SearchKnowledgeShared 共享检索(EXT-8 切片 2):返回他人 owner 的项目知识中,
|
||||
// 经 allow(docOwner, projectID) 放行的命中(全文通道;向量通道共享仍留后续切片)。
|
||||
// allow 出错由调用方前置处理——本层只见布尔。
|
||||
func (s *Store) SearchKnowledgeShared(requester string, q core.RetrievalQuery, allow func(docOwner, projectID core.ID) bool) ([]core.RetrievalHit, error) {
|
||||
pred := sharedVisible(requester, q, allow)
|
||||
if s.dialect.name() == "postgres" {
|
||||
return s.searchKnowledgePGShared(requester, q, pred)
|
||||
}
|
||||
return s.searchKnowledgeSQLiteShared(requester, q, pred)
|
||||
}
|
||||
|
||||
func (s *Store) searchKnowledgeSQLite(owner string, q core.RetrievalQuery) ([]core.RetrievalHit, error) {
|
||||
out, err := s.searchKnowledgeFTS(ownerVisible(owner, q), q)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// CJK 兜底:unicode61 分词把中文整句当单 token,FTS 前缀匹配失效时用 LIKE 子串。
|
||||
if len(out) == 0 && q.Text != "" {
|
||||
topK := q.TopK
|
||||
if topK <= 0 {
|
||||
topK = 5
|
||||
}
|
||||
return s.searchKnowledgeLike(owner, q, topK)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// searchKnowledgeSQLiteShared 共享版:FTS SQL 本身无 owner 预过滤(可见性原为
|
||||
// 后置判定),直接换谓词即可;LIKE 兜底换“他人行”预过滤。
|
||||
func (s *Store) searchKnowledgeSQLiteShared(requester string, q core.RetrievalQuery, pred knowledgePred) ([]core.RetrievalHit, error) {
|
||||
out, err := s.searchKnowledgeFTS(pred, q)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(out) == 0 && q.Text != "" {
|
||||
return s.searchKnowledgeLikeShared(requester, q, pred)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// searchKnowledgeFTS FTS 通道(owner 作用域与共享作用域共用:可见性谓词由调用方注入)。
|
||||
func (s *Store) searchKnowledgeFTS(pred knowledgePred, q core.RetrievalQuery) ([]core.RetrievalHit, error) {
|
||||
topK := q.TopK
|
||||
if topK <= 0 {
|
||||
topK = 5
|
||||
@@ -372,7 +437,7 @@ func (s *Store) searchKnowledgeSQLite(owner string, q core.RetrievalQuery) ([]co
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !knowledgeVisible(owner, q, h.Doc) {
|
||||
if !pred(h.Doc) {
|
||||
continue
|
||||
}
|
||||
out = append(out, h)
|
||||
@@ -380,14 +445,38 @@ func (s *Store) searchKnowledgeSQLite(owner string, q core.RetrievalQuery) ([]co
|
||||
break
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Store) searchKnowledgeLikeShared(requester string, q core.RetrievalQuery, pred knowledgePred) ([]core.RetrievalHit, error) {
|
||||
topK := q.TopK
|
||||
if topK <= 0 {
|
||||
topK = 5
|
||||
}
|
||||
rows, err := s.db.Query(s.q(`SELECT id,owner,project_id,source,title,ref,chunk,created_at FROM knowledge WHERE owner<>? AND owner<>'' AND (chunk LIKE ? OR title LIKE ?) LIMIT ?`),
|
||||
requester, "%"+q.Text+"%", "%"+q.Text+"%", topK)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// CJK 兜底:unicode61 分词把中文整句当单 token,FTS 前缀匹配失效时用 LIKE 子串。
|
||||
if len(out) == 0 && q.Text != "" {
|
||||
return s.searchKnowledgeLike(owner, q, topK)
|
||||
defer rows.Close()
|
||||
out := []core.RetrievalHit{}
|
||||
for rows.Next() {
|
||||
var d core.KnowledgeDoc
|
||||
var own, ca string
|
||||
if err := rows.Scan(&d.ID, &own, &d.ProjectID, &d.Source, &d.Title, &d.Ref, &d.Chunk, &ca); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
d.Owner = own
|
||||
d.CreatedAt, _ = parseTime(ca)
|
||||
if !pred(d) {
|
||||
continue
|
||||
}
|
||||
out = append(out, core.RetrievalHit{Doc: d, Score: 1.0, Kind: "fulltext"})
|
||||
if len(out) >= topK {
|
||||
break
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Store) searchKnowledgeLike(owner string, q core.RetrievalQuery, topK int) ([]core.RetrievalHit, error) {
|
||||
|
||||
@@ -49,3 +49,48 @@ func (s *Store) searchKnowledgePG(owner string, q core.RetrievalQuery) ([]core.R
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// searchKnowledgePGShared EXT-8 切片 2 共享版:他人 owner 的行 + 谓词后置判定
|
||||
// (allow 回调由上层注入;排序与自检索同构)。
|
||||
func (s *Store) searchKnowledgePGShared(requester string, q core.RetrievalQuery, pred knowledgePred) ([]core.RetrievalHit, error) {
|
||||
topK := q.TopK
|
||||
if topK <= 0 {
|
||||
topK = 5
|
||||
}
|
||||
text := q.Text
|
||||
rows, err := s.db.Query(s.q(`
|
||||
SELECT k.id,k.owner,k.project_id,k.source,k.title,k.ref,k.chunk,k.created_at,
|
||||
ts_rank(k.tsv, websearch_to_tsquery('simple', ?)) AS r
|
||||
FROM knowledge k
|
||||
WHERE (k.owner<>? AND k.owner<>'')
|
||||
AND (? = '' OR k.tsv @@ websearch_to_tsquery('simple', ?) OR k.chunk LIKE ? OR k.title LIKE ?)
|
||||
ORDER BY r DESC, k.created_at DESC
|
||||
LIMIT ?`), text, requester, text, text, "%"+text+"%", "%"+text+"%", topK*2)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []core.RetrievalHit{}
|
||||
for rows.Next() {
|
||||
var d core.KnowledgeDoc
|
||||
var own, ca string
|
||||
var rank float64
|
||||
if err := rows.Scan(&d.ID, &own, &d.ProjectID, &d.Source, &d.Title, &d.Ref, &d.Chunk, &ca, &rank); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
d.Owner = own
|
||||
d.CreatedAt, _ = parseTime(ca)
|
||||
if !pred(d) {
|
||||
continue
|
||||
}
|
||||
score := rank
|
||||
if score == 0 {
|
||||
score = 0.5
|
||||
}
|
||||
out = append(out, core.RetrievalHit{Doc: d, Score: score, Kind: "fulltext"})
|
||||
if len(out) >= topK {
|
||||
break
|
||||
}
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user