From 4f72aaf3d8eaef1ae08e3df2d3c94f7bf913d615 Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 13 Oct 2026 09:04:06 +0000 Subject: [PATCH] Recheck resumed project work and isolate publication replacement Why: Membership could be revoked while a tool waited for confirmation or a queued HITL reply waited for consumption. Publication replacement also crossed an empty project scope or another plugin's source namespace for the same Ref. Evidence: SELF replay of all four independent review reproductions failed before the fix and passed unchanged afterward. Committed regressions cover direct and queued HITL, yes/session confirmation, direct artifact writes, protected vectors, exact empty project scopes, literal wildcard IDs and unregistered compatibility. SELF uncached go test ./..., isolated PostgreSQL Store tests and static build passed. Impact: Recheck trusted live owner/project authorization when HITL resumes and at write boundaries. Registered publishers replace only exact owner/project/ref/plugin rows and vectors; ordinary unregistered publication semantics are preserved. Refs: Dev project-member-access implementation plan r1 Task 1; independent Backend review findings against b9bea5bee080b1e23ea5fa33af1878146ecd5e04 --- internal/app/hostcaps.go | 15 +++- internal/app/knowledge_authorization_test.go | 61 +++++++++++++++ internal/runmgr/knowledge_resource_test.go | 77 +++++++++++++++++++ internal/runmgr/runmgr.go | 32 +++++++- internal/store/content.go | 30 ++++++-- .../store/knowledge_authorization_test.go | 48 ++++++++++++ 6 files changed, 249 insertions(+), 14 deletions(-) diff --git a/internal/app/hostcaps.go b/internal/app/hostcaps.go index 62cebe5..866a4f4 100644 --- a/internal/app/hostcaps.go +++ b/internal/app/hostcaps.go @@ -322,9 +322,9 @@ func (h inprocHost) KnowledgeWrite(ctx context.Context, docs []core.KnowledgeDoc return h.caps.pipe.IndexChunks(ctx, core.ID(h.owner), docs) } -// KnowledgePublish 幂等发布（EXT-9）：按 docs 的 ref 整批替换——先清该 owner -// 下同 ref 旧行与旧向量（跨来源也清：ref 是发布单元），再走 KnowledgeWrite -// 语义（Owner/Source 强制同彼）。重试产生相同结果。 +// KnowledgePublish replaces refs and their vectors within the publication scope. +// A registered lifecycle scopes replacement to the exact owner/project/plugin; +// ordinary unregistered Hosts retain the legacy cross-source ref semantics. func (h inprocHost) KnowledgePublish(ctx context.Context, docs []core.KnowledgeDoc, embed bool) error { if len(docs) == 0 { return nil @@ -343,8 +343,15 @@ func (h inprocHost) KnowledgePublish(ctx context.Context, docs []core.KnowledgeD refs[publicationRef{docs[i].ProjectID, docs[i].Ref}] = true } } + scoped := h.caps.projects != nil && h.caps.projects.active() for ref := range refs { - if err := h.caps.st.DeleteProjectKnowledgeByRef(ctx, h.owner, ref.projectID, ref.ref); err != nil { + var err error + if scoped { + err = h.caps.st.DeletePluginProjectKnowledgeByRef(ctx, h.owner, ref.projectID, ref.ref, h.pluginID) + } else { + err = h.caps.st.DeleteProjectKnowledgeByRef(ctx, h.owner, ref.projectID, ref.ref) + } + if err != nil { return err } } diff --git a/internal/app/knowledge_authorization_test.go b/internal/app/knowledge_authorization_test.go index 7df87d3..e28dff3 100644 --- a/internal/app/knowledge_authorization_test.go +++ b/internal/app/knowledge_authorization_test.go @@ -103,3 +103,64 @@ func TestRegisteredProjectHostRejectsForeignKnowledgeSource(t *testing.T) { t.Fatalf("ordinary source semantics changed: %v", err) } } + +func TestProjectPublishPreservesForeignPluginAndScope(t *testing.T) { + st, err := store.OpenSQLite(t.TempDir() + "/t.db") + if err != nil { + t.Fatal(err) + } + defer st.Close() + slot := &projectLifecycleSlot{} + caps := hostcaps{st: st, projects: slot} + if err := (pluginRegistry{pluginID: "rdm", caps: caps}).RegisterProjectLifecycle(completeLifecycle()); err != nil { + t.Fatal(err) + } + original := core.KnowledgeDoc{ID: "rdm-published", Owner: "alice", ProjectID: "domain", Source: "plugin:rdm:published", Ref: "notes.md", Chunk: "originalneedle"} + if err := st.AddKnowledge(original); err != nil { + t.Fatal(err) + } + if err := st.UpsertKnowledgeVector(context.Background(), "alice", original.ID, original.Ref, []float32{1, 0}); err != nil { + t.Fatal(err) + } + h := inprocHost{owner: "alice", pluginID: "other", caps: caps} + replacement := []core.KnowledgeDoc{{ID: "other", ProjectID: "domain", Source: "plugin:other", Ref: "notes.md", Chunk: "replacementneedle"}} + for i := 0; i < 2; i++ { + if err := h.KnowledgePublish(context.Background(), replacement, false); err != nil { + t.Fatal(err) + } + } + if hits, err := st.SearchKnowledge("alice", core.RetrievalQuery{Text: "originalneedle"}); err != nil || len(hits) != 1 { + t.Fatalf("foreign plugin deleted RDM publication: %+v %v", hits, err) + } + if hits, err := st.SearchKnowledgeVectors(context.Background(), "alice", []float32{1, 0}, 5, core.RetrievalQuery{ProjectID: "domain"}); err != nil || len(hits) != 1 || hits[0].Doc.ID != original.ID { + t.Fatalf("foreign plugin deleted RDM vector: %+v %v", hits, err) + } + h.pluginID = "rdm" + if err := h.KnowledgePublish(context.Background(), []core.KnowledgeDoc{{ID: "unscoped", Source: "plugin:rdm", Ref: "notes.md", Chunk: "privateneedle"}}, false); err != nil { + t.Fatal(err) + } + if hits, err := st.SearchKnowledge("alice", core.RetrievalQuery{Text: "originalneedle", ProjectID: "domain"}); err != nil || len(hits) != 1 { + t.Fatalf("unscoped publication deleted scoped RDM data: %+v %v", hits, err) + } +} + +func TestUnregisteredPublishRetainsLegacyRefScope(t *testing.T) { + st, err := store.OpenSQLite(t.TempDir() + "/t.db") + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.AddKnowledge(core.KnowledgeDoc{ID: "legacy", Owner: "alice", ProjectID: "old-project", Source: "manual", Ref: "same", Chunk: "legacyneedle"}); err != nil { + t.Fatal(err) + } + h := inprocHost{owner: "alice", pluginID: "other", caps: hostcaps{st: st}} + if err := h.KnowledgePublish(context.Background(), []core.KnowledgeDoc{{ID: "replacement", Source: "manual", Ref: "same", Chunk: "replacementneedle"}}, false); err != nil { + t.Fatal(err) + } + if hits, err := st.SearchKnowledge("alice", core.RetrievalQuery{Text: "legacyneedle"}); err != nil || len(hits) != 0 { + t.Fatalf("unregistered legacy scope changed: %+v %v", hits, err) + } + if hits, err := st.SearchKnowledge("alice", core.RetrievalQuery{Text: "replacementneedle"}); err != nil || len(hits) != 1 { + t.Fatalf("unregistered replacement missing: %+v %v", hits, err) + } +} diff --git a/internal/runmgr/knowledge_resource_test.go b/internal/runmgr/knowledge_resource_test.go index 70c2a59..2cbd96e 100644 --- a/internal/runmgr/knowledge_resource_test.go +++ b/internal/runmgr/knowledge_resource_test.go @@ -64,3 +64,80 @@ func TestRunKnowledgeResolvesTrustedProjectScope(t *testing.T) { t.Fatalf("model current scope: %+v %v", hits, err) } } + +func TestProjectWorkRevocationBlocksQueuedHITL(t *testing.T) { + c, st := newExecEnv(t) + c.run.ID = "revoked-run" + c.run.Status = core.RunWaitingHITL + if err := st.CreateRun(&c.run); err != nil { + t.Fatal(err) + } + h := &runHandle{hitlCh: make(chan hitlAnswer, 1)} + c.mgr.active = map[core.ID]*runHandle{c.run.ID: h} + card := core.HITLCard{ID: "revoked-card", RunID: c.run.ID, ConversationID: c.run.ConversationID} + if err := st.SaveHITLCard("alice", &card); err != nil { + t.Fatal(err) + } + c.mgr.SetProjectWorkAuthorization(func(_ context.Context, owner string, id core.ID) (bool, error) { + if owner != "alice" || id != c.run.ProjectID { + t.Fatalf("untrusted resolution identity owner=%s project=%s", owner, id) + } + return false, nil + }) + if _, _, err := c.mgr.ResolveHITL("alice", card.ID, core.HITLResponse{ChoiceID: "yes"}); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("revoked confirmation should reject: %v", err) + } + persisted, err := st.GetHITLCard("alice", card.ID) + if err != nil || persisted.Resolved || len(h.hitlCh) != 0 { + t.Fatalf("revoked confirmation changed gate: card=%+v channel=%d error=%v", persisted, len(h.hitlCh), err) + } + run, err := st.GetRunByID(c.run.ID) + if err != nil || run.Status != core.RunWaitingHITL { + t.Fatalf("revoked confirmation resumed run: %+v %v", run, err) + } + payload, _ := json.Marshal(HITLCommandPayload{Owner: "alice", CardID: card.ID, Kind: "yes"}) + c.mgr.applyHITLCommand(ownedCommand{CommandID: "queued-approval", Payload: string(payload)}) + if len(h.hitlCh) != 0 { + t.Fatal("queued approval delivered a revoked answer") + } + run, err = st.GetRunByID(c.run.ID) + if err != nil || run.Status != core.RunWaitingHITL { + t.Fatalf("queued approval resumed run: %+v %v", run, err) + } +} + +func TestProjectWorkRevocationAfterToolConfirmation(t *testing.T) { + for _, choice := range []string{"yes", "session"} { + t.Run(choice, func(t *testing.T) { + c, st := newExecEnv(t) + c.h = &runHandle{hitlCh: make(chan hitlAnswer, 1)} + c.h.hitlCh <- hitlAnswer{resp: core.HITLResponse{ChoiceID: choice}} + c.prof.Tools.Confirm = []string{"write_artifact"} + ready := true + c.mgr.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { ok := ready; ready = false; return ok, nil }) + res := c.ExecuteTool(core.ToolCallRequest{CallID: "revoked-call", Name: "write_artifact", Arguments: json.RawMessage(`{"name":"revoked.txt","content":"x"}`)}) + artifacts, err := st.ListArtifacts("alice", c.run.ProjectID) + if err != nil { + t.Fatal(err) + } + if res.Status != "error" || len(artifacts) != 0 { + t.Fatalf("tool executed after revoked confirmation: status=%s artifacts=%d", res.Status, len(artifacts)) + } + if c.mgr.sessionAllows(c.run.ConversationID, "write_artifact") { + t.Fatal("revoked confirmation granted session permission") + } + }) + } +} + +func TestProjectWorkRevocationBlocksDirectArtifactWrite(t *testing.T) { + c, st := newExecEnv(t) + c.mgr.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return false, nil }) + if _, err := c.WriteArtifact("revoked.txt", "data", "text/plain", []byte("private")); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("direct artifact should reject: %v", err) + } + artifacts, err := st.ListArtifacts("alice", c.run.ProjectID) + if err != nil || len(artifacts) != 0 { + t.Fatalf("revoked artifact was written: %+v %v", artifacts, err) + } +} diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index b9d7ddb..e5fe6a3 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -630,7 +630,7 @@ func (c *runCtx) SetAssistantParts(parts []core.Part) error { return nil } -func (c *runCtx) AskHITL(card core.HITLCard) (core.HITLResponse, error) { +func (c *runCtx) AskHITL(card core.HITLCard) (answer core.HITLResponse, err error) { card.ID = store.NewID("hitl") card.RunID = c.run.ID card.ConversationID = c.run.ConversationID @@ -672,8 +672,13 @@ func (c *runCtx) AskHITL(card core.HITLCard) (core.HITLResponse, error) { // 排队，取消语境下重取失败由取消路径收口）。 c.mgr.adm.release(c.mgr.st, string(c.run.ID)) defer func() { - if err := c.mgr.adm.acquire(c.ctx, c.mgr.st, string(c.run.ID), c.run.Owner); err != nil { - log.Printf("[runmgr] admission reacquire after HITL %s: %v", c.run.ID, err) + if acquireErr := c.mgr.adm.acquire(c.ctx, c.mgr.st, string(c.run.ID), c.run.Owner); acquireErr != nil { + log.Printf("[runmgr] admission reacquire after HITL %s: %v", c.run.ID, acquireErr) + } + // Permission can change during both confirmation and admission waiting. + // Every HITL caller receives the live decision before it resumes work. + if err == nil { + err = c.mgr.authorizeProjectWork(c.ctx, c.run.Owner, c.run.ProjectID) } }() return waitHITL(c.ctx, c.h.hitlCh, timeout) @@ -876,6 +881,9 @@ func (c *runCtx) Artifact(id string) (core.Artifact, []byte, bool) { } func (c *runCtx) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) { + if err := c.mgr.authorizeProjectWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil { + return core.Artifact{}, err + } art := core.Artifact{ProjectID: c.run.ProjectID, Owner: c.run.Owner, Name: name, Kind: kind, MIME: mime, SourceRun: c.run.ID} if err := c.mgr.st.CreateArtifact(&art, content); err != nil { return art, err @@ -1008,12 +1016,18 @@ func (c *runCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult { if err != nil { return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: "hitl aborted: " + err.Error()} } + if err := c.mgr.authorizeProjectWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil { + return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: err.Error()} + } if resp.ChoiceID == "session" { c.mgr.rememberSessionAllow(c.run.ConversationID, call.Name) } else if resp.ChoiceID != "yes" { return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: "用户取消了本次工具调用"} } } + if err := c.mgr.authorizeProjectWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil { + return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: err.Error()} + } // EXT-14：命令即工具——命中命令投影时经命令总分发（Entrance=agent、 // Actor/Owner=会话 owner、带会话上下文），同审计信封与高危确认卡收口。 if c.mgr.commands != nil { @@ -1662,6 +1676,12 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv if err != nil { return "", false, err } + if m.projectWorkAuthorization != nil && (run.Owner != owner || run.ConversationID != card.ConversationID) { + return "", false, store.ErrNotFound + } + if err := m.authorizeProjectWork(context.Background(), run.Owner, run.ProjectID); err != nil { + return "", false, err + } // U7 双进程验收实测缺陷修复：run 不在本进程执行时不得先落 resolve—— // 卡一旦标记 resolved，另一实例 server 层的命令 fallback（resolveHITL 的 // !card.Resolved 判定）即被跳过，执行实例的 runner 永远等不到答案， @@ -1706,6 +1726,9 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv } } if pivot { + if err := m.authorizeProjectWork(context.Background(), run.Owner, run.ProjectID); err != nil { + return "", false, err + } _ = m.CancelRun(run.ID) m.st.ResolveHITLCard(owner, cardID, core.HITLResolvedPayload{CardID: core.ID(cardID), Kind: "cancelled", Answer: answer}) m.emit(run.ConversationID, run.ID, core.EvHITLResolved, core.HITLResolvedPayload{CardID: core.ID(cardID), Kind: "cancelled", Answer: answer}) @@ -1716,6 +1739,9 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv return trimmed, true, nil } } + if err := m.authorizeProjectWork(context.Background(), run.Owner, run.ProjectID); err != nil { + return "", false, err + } resolution := core.HITLResolvedPayload{CardID: core.ID(cardID), Answer: answer, Kind: kind} if err := m.st.ResolveHITLCard(owner, cardID, resolution); err != nil { return "", false, err diff --git a/internal/store/content.go b/internal/store/content.go index f203824..c740f98 100644 --- a/internal/store/content.go +++ b/internal/store/content.go @@ -409,20 +409,36 @@ func (s *Store) SearchKnowledgeSharedAuthorized(requester string, q core.Retriev return out, nil } -// DeleteProjectKnowledgeByRef replaces one publication scope and its vectors in -// a single transaction. Empty project is the legacy unscoped publication domain. +// DeleteProjectKnowledgeByRef preserves the ordinary legacy replacement scope. +// Registered lifecycle publishers use DeletePluginProjectKnowledgeByRef instead. func (s *Store) DeleteProjectKnowledgeByRef(ctx context.Context, owner, projectID, ref string) error { - tx, err := s.db.BeginTx(ctx, nil) - if err != nil { - return err - } - defer tx.Rollback() args := []any{owner, ref} pred := "owner=? AND ref=?" if projectID != "" { pred += " AND project_id=?" args = append(args, projectID) } + return s.deleteKnowledgeAndVectors(ctx, pred, args...) +} + +// DeletePluginProjectKnowledgeByRef replaces only this actor/plugin publication +// scope. Empty project is an exact unscoped domain, never a wildcard. The source +// prefix is compared literally, including any percent/underscore in plugin IDs. +func (s *Store) DeletePluginProjectKnowledgeByRef(ctx context.Context, owner, projectID, ref, pluginID string) error { + if owner == "" || pluginID == "" { + return errors.New("store: publication owner and plugin required") + } + prefix := "plugin:" + pluginID + pred := `owner=? AND project_id=? AND ref=? AND (source=? OR substr(source,1,length(?))=?)` + return s.deleteKnowledgeAndVectors(ctx, pred, owner, projectID, ref, prefix, prefix+":", prefix+":") +} + +func (s *Store) deleteKnowledgeAndVectors(ctx context.Context, pred string, args ...any) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() if _, err := tx.ExecContext(ctx, s.q("DELETE FROM knowledge_vectors WHERE doc_id IN (SELECT id FROM knowledge WHERE "+pred+")"), args...); err != nil { return err } diff --git a/internal/store/knowledge_authorization_test.go b/internal/store/knowledge_authorization_test.go index 2ef80c1..b765e96 100644 --- a/internal/store/knowledge_authorization_test.go +++ b/internal/store/knowledge_authorization_test.go @@ -2,6 +2,7 @@ package store import ( core "agent-system-core" + "context" "testing" ) @@ -20,3 +21,50 @@ func TestAuthorizedKnowledgeFiltersBeforeTopK(t *testing.T) { t.Fatalf("denied rows consumed top-k: %+v %v", hits, err) } } + +func TestPluginProjectPublicationDeleteUsesExactScope(t *testing.T) { + st := openTestStore(t) + ctx := context.Background() + docs := []core.KnowledgeDoc{{ID: "unscoped", Owner: "alice", Source: "plugin:rdm", Ref: "same", Chunk: "unscoped"}, {ID: "scoped", Owner: "alice", ProjectID: "domain", Source: "plugin:rdm:published", Ref: "same", Chunk: "scoped"}, {ID: "foreign", Owner: "alice", Source: "plugin:other", Ref: "same", Chunk: "foreign"}, {ID: "other_owner", Owner: "bob", Source: "plugin:rdm", Ref: "same", Chunk: "other owner"}} + if err := st.AddKnowledge(docs...); err != nil { + t.Fatal(err) + } + for _, d := range docs { + if err := st.UpsertKnowledgeVector(ctx, d.Owner, d.ID, d.Ref, []float32{1, 0}); err != nil { + t.Fatal(err) + } + } + if err := st.DeletePluginProjectKnowledgeByRef(ctx, "alice", "", "same", "rdm"); err != nil { + t.Fatal(err) + } + for _, table := range []string{"knowledge", "knowledge_vectors"} { + var count int + if err := st.db.QueryRow("SELECT COUNT(*) FROM " + table).Scan(&count); err != nil || count != 3 { + t.Fatalf("exact empty-project namespace %s count=%d error=%v", table, count, err) + } + } + if err := st.DeletePluginProjectKnowledgeByRef(ctx, "alice", "domain", "same", "rdm"); err != nil { + t.Fatal(err) + } + for _, id := range []string{"foreign", "other_owner"} { + var count int + if err := st.db.QueryRow(st.q(`SELECT COUNT(*) FROM knowledge WHERE id=?`), id).Scan(&count); err != nil || count != 1 { + t.Fatalf("protected row %s count=%d error=%v", id, count, err) + } + } +} + +func TestPluginPublicationNamespaceIsLiteral(t *testing.T) { + st := openTestStore(t) + docs := []core.KnowledgeDoc{{ID: "owned", Owner: "alice", Source: "plugin:rdm_%", Ref: "same", Chunk: "owned"}, {ID: "owned_suffix", Owner: "alice", Source: "plugin:rdm_%:published", Ref: "same", Chunk: "owned suffix"}, {ID: "protected", Owner: "alice", Source: "plugin:rdmXYZ:published", Ref: "same", Chunk: "protected"}} + if err := st.AddKnowledge(docs...); err != nil { + t.Fatal(err) + } + if err := st.DeletePluginProjectKnowledgeByRef(context.Background(), "alice", "", "same", "rdm_%"); err != nil { + t.Fatal(err) + } + var id string + if err := st.db.QueryRow(`SELECT id FROM knowledge`).Scan(&id); err != nil || id != "protected" { + t.Fatalf("wildcards broadened namespace: id=%s error=%v", id, err) + } +}