diff --git a/internal/cardsvc/cardsvc.go b/internal/cardsvc/cardsvc.go index 48930e0..f000934 100644 --- a/internal/cardsvc/cardsvc.go +++ b/internal/cardsvc/cardsvc.go @@ -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) diff --git a/internal/server/server.go b/internal/server/server.go index 930429e..6b9127b 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -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) } diff --git a/internal/server/shared_retrieval_test.go b/internal/server/shared_retrieval_test.go index 8572551..5102b42 100644 --- a/internal/server/shared_retrieval_test.go +++ b/internal/server/shared_retrieval_test.go @@ -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) diff --git a/internal/store/content.go b/internal/store/content.go index bdc17ef..10331c2 100644 --- a/internal/store/content.go +++ b/internal/store/content.go @@ -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) { diff --git a/internal/store/search_postgres.go b/internal/store/search_postgres.go index 37c47d7..0620c9f 100644 --- a/internal/store/search_postgres.go +++ b/internal/store/search_postgres.go @@ -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() +}