From 1ee32822c30a9ce0518e9f4ae921b439fb104b74 Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 8 Oct 2026 15:53:39 +0000 Subject: [PATCH] Add the EXT-9 publish/revoke surface and red acceptance tests Why: EXT-9 is the other step-4 hard prerequisite; the surface (Options.DisableUploadIndexing passthrough, Host.KnowledgePublish idempotent replace, Host.KnowledgeDelete prefix-enforced revoke, SurfaceVersion 2.2.0-experimental) must land with the acceptance reds pinned before implementation. Evidence: app red = publish not implemented; server red = switch-on still indexes (hits=1); rows recorded on the task card. Impact: pkg/extension additive surface (surface.txt updated); internal Options passthrough; Server.DisableUploadIndexing field; host method stubs. Behavior unchanged. Refs: EXT-9 proposal; task card. --- internal/app/app.go | 6 ++ internal/app/ext9_publish_test.go | 81 ++++++++++++++++ internal/app/hostcaps.go | 11 +++ internal/server/ext9_upload_switch_test.go | 102 +++++++++++++++++++++ internal/server/server.go | 3 + pkg/extension/extension.go | 7 +- pkg/extension/host.go | 8 ++ pkg/extension/link/link.go | 1 + pkg/extension/surface.txt | 2 + 9 files changed, 220 insertions(+), 1 deletion(-) create mode 100644 internal/app/ext9_publish_test.go create mode 100644 internal/server/ext9_upload_switch_test.go diff --git a/internal/app/app.go b/internal/app/app.go index 9f3cf9d..714f1be 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -66,6 +66,9 @@ type Options struct { InstanceID string // 空 = 主机名:监听地址；写入 runs.instance_id，启动对账只收口本实例 InsecureListen bool AllowedHosts []string + // DisableUploadIndexing EXT-9：true 时上传端点不自动写检索索引（发布决定 + // 移交下游命令层）；零值=false=现状。由 link 层自 extension.Options 透传。 + DisableUploadIndexing bool } // Run 装配并阻塞服务直至 SIGINT/SIGTERM（优雅停机：中断在途 run、冲净事件缓冲）。 @@ -402,6 +405,9 @@ func Run(opt Options) error { srv := server.New(st, mgr, reg, profSvc, setSvc, pluginHost) // EXT-8：装载期注册的资源授权适配器注入读取入口（未注册=默认拒绝，行为不变）。 srv.SetProjectAuthorizer(authzSlot.Authorize) + // EXT-9：宿主可关闭上传自动索引（发布决定移交下游命令层）。 + srv.DisableUploadIndexing = opt.DisableUploadIndexing + srv.Workflows = wfReg mgr.SetWorkflows(wfReg) srv.Cards = cardSvc diff --git a/internal/app/ext9_publish_test.go b/internal/app/ext9_publish_test.go new file mode 100644 index 0000000..62b2ff6 --- /dev/null +++ b/internal/app/ext9_publish_test.go @@ -0,0 +1,81 @@ +package app + +import ( + "context" + "path/filepath" + "testing" + + "agent-system-backend/internal/rag" + "agent-system-backend/internal/settings" + "agent-system-backend/internal/store" + core "agent-system-core" +) + +// newExt9Host EXT-9 测试装配：真实 store + rag 管线 + inprocHost（无 embedder， +// embed=false 走全文档路径）。 +func newExt9Host(t *testing.T) (interface { + KnowledgePublish(ctx context.Context, docs []core.KnowledgeDoc, embed bool) error + KnowledgeDelete(ctx context.Context, sourcePrefix string) error +}, *store.Store, *rag.Pipeline) { + t.Helper() + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + set := settings.NewService(st) + pipe := rag.NewPipeline(st, set, nil, nil) + h := inprocHost{caps: hostcaps{st: st, pipe: pipe, set: set}, pluginID: "rdm", owner: "alice"} + return h, st, pipe +} + +func searchAll(t *testing.T, pipe *rag.Pipeline, owner, text string) []core.RetrievalHit { + t.Helper() + hits, err := pipe.Retrieve(context.Background(), core.ID(owner), core.RetrievalQuery{Text: text, TopK: 50}) + if err != nil { + t.Fatalf("retrieve: %v", err) + } + return hits +} + +// EXT-9 验收：publish 按 ref 幂等（重试不重复）；delete 后检索即停且幂等； +// manual 来源不经 host 可删（前缀强制 plugin:<自身>）。 +func TestKnowledgePublishIdempotentAndRevoke(t *testing.T) { + h, st, pipe := newExt9Host(t) + ctx := context.Background() + docs := []core.KnowledgeDoc{ + {Ref: "batch-001", Chunk: "EXT9甲 独占令牌之一陈述"}, + {Ref: "batch-001", Chunk: "EXT9乙 独占令牌之二陈述"}, + } + if err := h.KnowledgePublish(ctx, docs, false); err != nil { + t.Fatalf("publish#1: %v", err) + } + if err := h.KnowledgePublish(ctx, docs, false); err != nil { // 重试幂等 + t.Fatalf("publish#2: %v", err) + } + hits := searchAll(t, pipe, "alice", "独占令牌") + if len(hits) != 2 { + t.Fatalf("after double publish: hits=%d, want 2（幂等不重复；got %+v）", len(hits), hits) + } + + if err := h.KnowledgeDelete(ctx, "plugin:rdm"); err != nil { + t.Fatalf("delete: %v", err) + } + if hits = searchAll(t, pipe, "alice", "独占令牌"); len(hits) != 0 { + t.Fatalf("after delete: hits=%d, want 0（撤销即停）", len(hits)) + } + if err := h.KnowledgeDelete(ctx, "plugin:rdm"); err != nil { // 幂等 + t.Fatalf("delete#2: %v", err) + } + + // manual 来源：host 前缀强制拒绝，且内容不受插件删除影响。 + if err := st.AddKnowledge(core.KnowledgeDoc{Owner: "alice", Source: "manual", Ref: "m-1", Chunk: "manual 手册独占词"}); err != nil { + t.Fatal(err) + } + if err := h.KnowledgeDelete(ctx, "manual"); err == nil { + t.Fatal("delete(manual) must be rejected: plugins may only revoke their own prefix") + } + if hits = searchAll(t, pipe, "alice", "手册独占词"); len(hits) != 1 { + t.Fatalf("manual row must survive: hits=%d", len(hits)) + } +} diff --git a/internal/app/hostcaps.go b/internal/app/hostcaps.go index 8ef6857..c4c90f8 100644 --- a/internal/app/hostcaps.go +++ b/internal/app/hostcaps.go @@ -3,6 +3,7 @@ package app import ( "agent-system-backend/pkg/extension" "context" + "errors" "fmt" "sync" @@ -139,6 +140,16 @@ func (h inprocHost) KnowledgeWrite(ctx context.Context, docs []core.KnowledgeDoc return h.caps.pipe.IndexChunks(ctx, core.ID(h.owner), docs) } +// KnowledgePublish/KnowledgeDelete EXT-9 幂等发布与撤销（本提交为占位—— +// 红测先行，实现在后续提交落地）。 +func (h inprocHost) KnowledgePublish(ctx context.Context, docs []core.KnowledgeDoc, embed bool) error { + return errors.New("ext-9: knowledge publish not implemented yet") +} + +func (h inprocHost) KnowledgeDelete(ctx context.Context, sourcePrefix string) error { + return errors.New("ext-9: knowledge delete not implemented yet") +} + // TaskCreate/TaskUpdate 后台任务台账：基座只记进度/状态，执行体由下游自跑（H5）。 func (h inprocHost) TaskCreate(kind, title, conversationID string) (string, error) { t, err := h.caps.st.CreateBackgroundTask(h.owner, kind, title, conversationID) diff --git a/internal/server/ext9_upload_switch_test.go b/internal/server/ext9_upload_switch_test.go new file mode 100644 index 0000000..e4247ca --- /dev/null +++ b/internal/server/ext9_upload_switch_test.go @@ -0,0 +1,102 @@ +package server + +import ( + "bytes" + "encoding/json" + "fmt" + "mime/multipart" + "net/http" + "net/http/httptest" + "testing" +) + +// EXT-9 验收：DisableUploadIndexing=true 时上传不再自动进检索（原件与摄取卡 +// 行为不变）；=false 恢复现状（可解析文本入索引）。 +func TestUploadPreIndexSwitch(t *testing.T) { + srv, st := newAuthTestServer(t) + h := srv.Handler() + alice, err := st.CreateAPIKey("alice", "e2e") + if err != nil { + t.Fatal(err) + } + + req := httptest.NewRequest(http.MethodPost, "/api/projects", bytes.NewReader(mustJSON(t, map[string]string{"name": "EXT9"}))) + req.Header.Set("Authorization", "Bearer "+alice) + 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 + json.Unmarshal(rec.Body.Bytes(), &pj) + pid := fmt.Sprint(pj["id"]) + + upload := func(name, text string) string { + 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 "+alice) + req.Header.Set("Content-Type", mw.FormDataContentType()) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("upload %s: %d %s", name, rec.Code, rec.Body.String()) + } + var up map[string]any + json.Unmarshal(rec.Body.Bytes(), &up) + return fmt.Sprint(up["artifactId"]) + } + search := func(q string) int { + req := httptest.NewRequest(http.MethodGet, "/api/knowledge/search?q="+q, nil) + req.Header.Set("Authorization", "Bearer "+alice) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("search: %d %s", rec.Code, rec.Body.String()) + } + var arr []any + if err := json.Unmarshal(rec.Body.Bytes(), &arr); err == nil { + return len(arr) // 直接数组形态 + } + var out struct { + Hits []any `json:"hits"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil { + t.Fatalf("search decode: %v (%s)", err, rec.Body.String()) + } + return len(out.Hits) + } + + // 开关开：上传不入检索，但原件完整可取回。 + srv.DisableUploadIndexing = true + aid := upload("ext9-off.txt", "EXT9OFF 索引开关独占词甲") + if n := search("索引开关独占词甲"); n != 0 { + t.Fatalf("switch on: search hits=%d, want 0（存疑确认前不入正式检索）", n) + } + reqC := httptest.NewRequest(http.MethodGet, "/api/artifacts/"+aid+"/content", nil) + reqC.Header.Set("Authorization", "Bearer "+alice) + recC := httptest.NewRecorder() + h.ServeHTTP(recC, reqC) + if recC.Code != http.StatusOK || !bytes.Contains(recC.Body.Bytes(), []byte("索引开关独占词甲")) { + t.Fatalf("original must remain retrievable: %d %q", recC.Code, recC.Body.String()) + } + + // 开关关（现状）：同一上传路径恢复自动索引。 + srv.DisableUploadIndexing = false + upload("ext9-on.txt", "EXT9ON 索引开关独占词乙") + if n := search("索引开关独占词乙"); n != 1 { + t.Fatalf("switch off: search hits=%d, want 1（现状自动索引）", n) + } +} + +func mustJSON(t *testing.T, v any) []byte { + t.Helper() + b, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + return b +} diff --git a/internal/server/server.go b/internal/server/server.go index 00fe705..4bce8da 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -57,6 +57,9 @@ type Server struct { static fs.FS // Subscribe 注入 bus 订阅（事件在线转发；重放走 store 游标）。 Subscribe func(conversationID string) (<-chan core.StreamEvent, func()) + // DisableUploadIndexing EXT-9：true 时上传端点不再自动写检索索引（原件与 + // 摄取卡行为不变）；组合根自 extension.Options 注入，零值=false=现状。 + DisableUploadIndexing bool // ProjectAuthz EXT-8 资源授权适配器（组合根注入；nil=未注册，读取入口 // owner 自有判定未命中即 NotFound——与未扩展基座行为一致）。 ProjectAuthz func(owner string, projectID core.ID) (bool, error) diff --git a/pkg/extension/extension.go b/pkg/extension/extension.go index 642d47f..d27b570 100644 --- a/pkg/extension/extension.go +++ b/pkg/extension/extension.go @@ -12,7 +12,9 @@ import ( // SurfaceVersion 是本包导出面的实验版本。变更导出符号必须同时更新 surface.txt。 // 2.1.0-experimental：EXT-8 新增 ResourceAuthorizer 与 Registry.RegisterResourceAuthorizer。 -const SurfaceVersion = "2.1.0-experimental" +// 2.2.0-experimental：EXT-9 新增 Host.KnowledgePublish/KnowledgeDelete 与 +// Options.DisableUploadIndexing。 +const SurfaceVersion = "2.2.0-experimental" // ErrNotLinked 表示进程没有空白导入 link 包，Main 无法启动服务。 var ErrNotLinked = errors.New("extension: runner not linked; blank-import agent-system-backend/pkg/extension/link") @@ -50,6 +52,9 @@ type Options struct { StorageDSN string ArtifactsDir string AuthMode string + // DisableUploadIndexing 关闭上传端点的自动检索索引（EXT-9；零值=false=现状： + // 可解析文本照旧进索引。下游以发布状态机自管索引时置 true）。 + DisableUploadIndexing bool } var ( diff --git a/pkg/extension/host.go b/pkg/extension/host.go index 8e880ee..2a5a9e4 100644 --- a/pkg/extension/host.go +++ b/pkg/extension/host.go @@ -39,6 +39,14 @@ type Host interface { // Complete 经基座 LLM 路由补全（owner 的模型配置生效）。 Complete(ctx context.Context, prompt string) (string, error) + // KnowledgePublish 幂等发布（EXT-9）：按 docs 的 ref 整批替换——先清该 owner + // 下相同 ref 的旧行与旧向量，再走 KnowledgeWrite 语义（Owner/Source 强制同彼）。 + // 重试产生相同结果；embed=true 时重建向量。 + KnowledgePublish(ctx context.Context, docs []core.KnowledgeDoc, embed bool) error + // KnowledgeDelete 撤销本插件来源的知识索引（EXT-9）：sourcePrefix 必须是 + // "plugin:<本插件ID>"——manual 或他人来源拒绝；删除行与向量，幂等（无匹配不报错）。 + KnowledgeDelete(ctx context.Context, sourcePrefix string) error + // KnowledgeSearch 混合检索（owner 可见范围）。 KnowledgeSearch(ctx context.Context, projectID core.ID, text string, topK int) ([]core.RetrievalHit, error) // KnowledgeWrite 写知识分块；embed=true 时同步建向量（失败降级纯全文）。 diff --git a/pkg/extension/link/link.go b/pkg/extension/link/link.go index ebcc0b2..ef0be23 100644 --- a/pkg/extension/link/link.go +++ b/pkg/extension/link/link.go @@ -12,6 +12,7 @@ func init() { return app.Run(app.Options{ Addr: opt.Addr, DataDir: opt.DataDir, PluginsDir: opt.PluginsDir, ForceMock: opt.Mock, StorageDSN: opt.StorageDSN, ArtifactsDir: opt.ArtifactsDir, AuthMode: opt.AuthMode, + DisableUploadIndexing: opt.DisableUploadIndexing, }) }) } diff --git a/pkg/extension/surface.txt b/pkg/extension/surface.txt index c7f5658..f056733 100644 --- a/pkg/extension/surface.txt +++ b/pkg/extension/surface.txt @@ -8,6 +8,8 @@ Host.Audit Host.Complete Host.KVGet Host.KVPut +Host.KnowledgeDelete +Host.KnowledgePublish Host.KnowledgeSearch Host.KnowledgeWrite Host.Owner