diff --git a/internal/app/app.go b/internal/app/app.go index 714f1be..9054420 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -69,6 +69,9 @@ type Options struct { // DisableUploadIndexing EXT-9：true 时上传端点不自动写检索索引（发布决定 // 移交下游命令层）；零值=false=现状。由 link 层自 extension.Options 透传。 DisableUploadIndexing bool + // DisableConvCardMirror EXT-12：true 时会话卡不镜像进项目公共流（不发布 + // 即不持久，重放不含会话卡）；零值=false=现状全量镜像。 + DisableConvCardMirror bool } // Run 装配并阻塞服务直至 SIGINT/SIGTERM（优雅停机：中断在途 run、冲净事件缓冲）。 @@ -407,6 +410,8 @@ func Run(opt Options) error { srv.SetProjectAuthorizer(authzSlot.Authorize) // EXT-9：宿主可关闭上传自动索引（发布决定移交下游命令层）。 srv.DisableUploadIndexing = opt.DisableUploadIndexing + // EXT-12 最小开关：会话卡不镜像进项目公共流（cardSvc 在上方已建）。 + cardSvc.MirrorConvCards = !opt.DisableConvCardMirror srv.Workflows = wfReg mgr.SetWorkflows(wfReg) diff --git a/internal/cardsvc/cardsvc.go b/internal/cardsvc/cardsvc.go index 168061a..f000934 100644 --- a/internal/cardsvc/cardsvc.go +++ b/internal/cardsvc/cardsvc.go @@ -29,13 +29,16 @@ const cardAppendWindow = 250 * time.Millisecond type Service struct { st *store.Store emit Emitter + // MirrorConvCards EXT-12：会话卡是否镜像进项目公共流（缺省 true=现状全量 + // 镜像；false 时会话卡只进本人会话流，项目级卡不受影响）。 + MirrorConvCards bool now func() time.Time emitMu sync.Mutex lastEmit map[string]time.Time // 流式追加的最近发布时刻（按卡） } func New(st *store.Store, emit Emitter) *Service { - return &Service{st: st, emit: emit, now: time.Now, lastEmit: map[string]time.Time{}} + return &Service{st: st, emit: emit, now: time.Now, lastEmit: map[string]time.Time{}, MirrorConvCards: true} } // emitCard 按卡归属发布卡片事件（D50）： @@ -52,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/cardsvc/mirror_switch_test.go b/internal/cardsvc/mirror_switch_test.go new file mode 100644 index 0000000..91f04e5 --- /dev/null +++ b/internal/cardsvc/mirror_switch_test.go @@ -0,0 +1,56 @@ +package cardsvc + +import ( + "testing" + + "agent-system-backend/internal/store" + core "agent-system-core" +) + +// EXT-12 最小开关（红测）：MirrorConvCards=false 时会话卡不镜像进项目公共流 +// （不发布即不持久，重放天然不含会话卡）；项目级卡不受影响；缺省 true 保持现状。 +func TestConvCardMirrorSwitch(t *testing.T) { + newSvc := func(mirror bool) (*Service, *[]string) { + chans := []string{} + emit := func(convID, runID string, typ core.StreamEventType, payload any) error { + chans = append(chans, convID) + return nil + } + st, err := store.OpenSQLite(t.TempDir() + "/cards.db") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + svc := New(st, emit) + svc.MirrorConvCards = mirror + return svc, &chans + } + card := core.ActivityCard{ConversationID: "c-priv", ProjectID: "p-1", Kind: core.CardKindProgress, Title: "私有过程卡"} + + // 开（缺省现状）：会话流 + 项目流双发。 + svc, chans := newSvc(true) + if _, err := svc.Create("alice", card, ""); err != nil { + t.Fatal(err) + } + if len(*chans) != 2 { + t.Fatalf("mirror on: emitted %v, want 会话流+项目流", *chans) + } + + // 关：只进本人会话流，项目公共流零事件。 + svc2, chans2 := newSvc(false) + if _, err := svc2.Create("alice", card, ""); err != nil { + t.Fatal(err) + } + if len(*chans2) != 1 || (*chans2)[0] != "c-priv" { + t.Fatalf("mirror off: emitted %v, want 仅会话流 c-priv（项目流不得发布）", *chans2) + } + + // 项目级卡（无会话归属）不受开关影响。 + pc := core.ActivityCard{ProjectID: "p-1", Kind: core.CardKindProgress, Title: "项目级卡"} + if _, err := svc2.Create("alice", pc, ""); err != nil { + t.Fatal(err) + } + if len(*chans2) != 2 || (*chans2)[1] != string(core.ProjectChannelKey("p-1")) { + t.Fatalf("project-level card must still publish: %v", *chans2) + } +} 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 new file mode 100644 index 0000000..5102b42 --- /dev/null +++ b/internal/server/shared_retrieval_test.go @@ -0,0 +1,142 @@ +package server + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + core "agent-system-core" +) + +// EXT-8 切片 2（红测）：适配器放行后，他人 owner 的项目知识对本人可检索； +// 未放行/第三方保持零命中（不泄露存在性）。 +func TestSharedKnowledgeRetrieval(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) + } + + // alice 建项目并写入带 ProjectID 的知识（模拟其已发布内容）。 + req := httptest.NewRequest(http.MethodPost, "/api/projects", bytes.NewReader(mustJSON2(t, map[string]string{"name": "S2"}))) + req.Header.Set("Authorization", "Bearer "+alice) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("create project: %d %s", rec.Code, rec.Body.String()) + } + var pj map[string]any + readJSON(t, rec.Body.Bytes(), &pj) + pid := fmt.Sprint(pj["id"]) + if err := st.AddKnowledge(core.KnowledgeDoc{Owner: "alice", ProjectID: core.ID(pid), + Source: "plugin:rdm", Ref: "kb-1", Chunk: "共享检索独占词 甲"}); err != nil { + t.Fatal(err) + } + + search := func(key string) int { + req := httptest.NewRequest(http.MethodGet, "/api/knowledge/search?q=%E5%85%B1%E4%BA%AB%E6%A3%80%E7%B4%A2%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) + } + + if n := search(alice); n != 1 { + t.Fatalf("alice own search = %d, want 1", n) + } + if n := search(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 := search(bob); n != 1 { + t.Fatalf("bob after grant = %d, want 1（适配器放行后共享知识可检索）", n) + } +} + +func readJSON(t *testing.T, b []byte, v any) { + t.Helper() + if err := json.Unmarshal(b, v); err != nil { + t.Fatalf("decode: %v (%s)", err, b) + } +} + +func mustJSON2(t *testing.T, v any) []byte { + t.Helper() + b, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + return b +} + +// EXT-8 切片 2（红测）：SSE 项目流订阅门——owner 自有未命中时咨询适配器； +// 未放行 404（不泄露存在性），放行可订阅（事件头即证）。 +func TestProjectSSEScopeGate(t *testing.T) { + srv, st := newAuthTestServer(t) + _ = st + 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": "SSE"}))) + 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"]) + + srvSSE := httptest.NewServer(h) + t.Cleanup(srvSSE.Close) + sub := func(key string) int { + 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) + 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) + } + srv.SetProjectAuthorizer(func(owner string, projectID core.ID) (bool, error) { + return owner == "bob" && string(projectID) == pid, nil + }) + if c := sub(bob); c != http.StatusOK { + t.Fatalf("bob after grant: %d, want 200（适配器放行后可订阅项目流）", c) + } + if c := sub(alice); c != http.StatusOK { + t.Fatalf("alice own project: %d, want 200（现状不受影响）", 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() +} diff --git a/pkg/extension/extension.go b/pkg/extension/extension.go index d27b570..f0860f9 100644 --- a/pkg/extension/extension.go +++ b/pkg/extension/extension.go @@ -14,7 +14,9 @@ import ( // 2.1.0-experimental：EXT-8 新增 ResourceAuthorizer 与 Registry.RegisterResourceAuthorizer。 // 2.2.0-experimental：EXT-9 新增 Host.KnowledgePublish/KnowledgeDelete 与 // Options.DisableUploadIndexing。 -const SurfaceVersion = "2.2.0-experimental" +// 2.3.0-experimental：EXT-12 最小开关 Options.DisableConvCardMirror（会话卡不镜像 +// 进项目公共流）。 +const SurfaceVersion = "2.3.0-experimental" // ErrNotLinked 表示进程没有空白导入 link 包，Main 无法启动服务。 var ErrNotLinked = errors.New("extension: runner not linked; blank-import agent-system-backend/pkg/extension/link") @@ -55,6 +57,10 @@ type Options struct { // DisableUploadIndexing 关闭上传端点的自动检索索引（EXT-9；零值=false=现状： // 可解析文本照旧进索引。下游以发布状态机自管索引时置 true）。 DisableUploadIndexing bool + // DisableConvCardMirror 关闭会话卡向项目公共流的镜像（EXT-12 最小开关； + // 零值=false=现状全量镜像。true 时会话卡只进本人会话流，不发布即不持久， + // 项目流重放不含会话卡；项目级卡不受影响）。 + DisableConvCardMirror bool } var ( diff --git a/pkg/extension/link/link.go b/pkg/extension/link/link.go index ef0be23..fac95f3 100644 --- a/pkg/extension/link/link.go +++ b/pkg/extension/link/link.go @@ -13,6 +13,7 @@ func init() { Addr: opt.Addr, DataDir: opt.DataDir, PluginsDir: opt.PluginsDir, ForceMock: opt.Mock, StorageDSN: opt.StorageDSN, ArtifactsDir: opt.ArtifactsDir, AuthMode: opt.AuthMode, DisableUploadIndexing: opt.DisableUploadIndexing, + DisableConvCardMirror: opt.DisableConvCardMirror, }) }) }