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 b9bea5bee0

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
 		}
 	}

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)
+	}
+}

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)
+	}
+}

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

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
 	}

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)
+	}
+}