Protect private project event replay and live delivery

Why:
Legacy project streams replayed upload filenames and parser metadata to newly
authorized members even though the underlying cards remained owner-only.

Evidence:
SELF real HTTP red regressions exposed legacy replay, spoofed sharing claims,
private-history pagination, and live delivery after membership removal. The
affected packages, uncached full Backend suite, isolated PostgreSQL Store suite,
and agent-system build pass after applying the event boundary.

Impact:
Registered lifecycle streams resolve trusted resource ownership and require an
internal publication proof for cross-owner delivery. Event and proof commit in
one transaction; v30 adds an additive SQLite/PostgreSQL sidecar without granting
access to historical events. Seen and visible cursors remain separate, revoked
members leave live streams, and unregistered base behavior remains compatible.
The storage migration tool preserves the new proof rows.

Refs:
project-member-access-implementation-plan.md Task 1 / R5; authorized legacy SSE
privacy follow-up; project-member-composed-legacy-project-history.json

cmd/storage-migrate/main.go

@@ -284,7 +284,7 @@ var copyOrder = []string{
 	"projects", "users", "profiles", "settings", "plugins",
 	"conversations", "runs", "messages", "events", "artifacts",
 	"hitl_cards", "plans", "todos", "knowledge", "knowledge_vectors",
-	"activity_cards", "command_audit", "prompt_modules",
+	"activity_cards", "project_event_shares", "command_audit", "prompt_modules",
 	"sessions", "api_keys",
 	"plugin_grants", "plugin_schema", "plugin_kv", "background_tasks",
 	"memory_entries", "audit_events", "pending_confirms", "llm_usage_daily",
@@ -784,7 +784,7 @@ func advanceSequences(db *sql.DB, table string) error {
 // checkOwners：业务行引用的 owner 必须存在于 users（单租户默认 ”/'local'
 // 除外）。未知 owner 阻塞迁移（U4/R7：不静默带过未知归属）。
 func checkOwners(db *sql.DB, src *sql.DB) error {
-	ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards", "command_audit"}
+	ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards", "project_event_shares", "command_audit"}
 	for _, t := range ownerTables {
 		if !sqliteHasTable(src, t) {
 			continue

internal/app/app.go

@@ -528,6 +528,7 @@ func Run(opt Options) error {
 
 	// 收件箱任务卡：事件走会话事件管道（持久化/seq/重放）；关闭即 T0 基线。
 	cardSvc := cardsvc.New(st, mgr.EmitCardEvent)
+	cardSvc.SetProjectSharedEmitter(mgr.EmitProjectSharedCardEvent)
 	surfaceSvc := surfacesvc.New(st, mgr.EmitCardEvent)
 	go func() { // 启动恢复：重发「快照已落盘、事件未发出」的待恢复投影（KTD）。
 		if n, err := surfaceSvc.RecoverPendingEvents(context.Background()); err == nil && n > 0 {

internal/cardsvc/cardsvc.go

@@ -27,8 +27,9 @@ var (
 const cardAppendWindow = 250 * time.Millisecond
 
 type Service struct {
-	st   *store.Store
-	emit Emitter
+	st                *store.Store
+	emit              Emitter
+	projectSharedEmit ProjectSharedEmitter
 	// MirrorConvCards EXT-12：会话卡是否镜像进项目公共流（缺省 true=现状全量
 	// 镜像；false 时会话卡只进本人会话流，项目级卡不受影响）。
 	MirrorConvCards bool
@@ -41,6 +42,42 @@ func New(st *store.Store, emit Emitter) *Service {
 	return &Service{st: st, emit: emit, now: time.Now, lastEmit: map[string]time.Time{}, MirrorConvCards: true}
 }
 
+// ProjectSharedEmitter is an internal trusted publication adapter. It is never
+// bound to model arguments or exposed as a mandatory extension capability.
+type ProjectSharedEmitter func(owner, producer, channel, runID string, typ core.StreamEventType, payload any) error
+
+func (s *Service) SetProjectSharedEmitter(emit ProjectSharedEmitter) { s.projectSharedEmit = emit }
+
+// PublishProjectCard explicitly publishes a project-level structural snapshot.
+// Ordinary Create/Update remain private unless this entry issues a new proof.
+func (s *Service) PublishProjectCard(owner, producer string, card core.ActivityCard, ref string) (core.ActivityCard, error) {
+	if s.projectSharedEmit == nil || owner == "" || producer == "" || card.ProjectID == "" || card.ConversationID != "" {
+		return card, errors.New("card: trusted project publisher required")
+	}
+	if card.ID != "" {
+		current, err := s.st.GetCard(owner, card.ID)
+		if err != nil {
+			return card, err
+		}
+		if current.ProjectID != card.ProjectID || current.ConversationID != "" {
+			return card, store.ErrNotFound
+		}
+	} else {
+		card.ID = store.NewID("card")
+	}
+	now := s.now().UTC()
+	card.CreatedAt = now
+	card.UpdatedAt = now
+	if card.Status == "" {
+		card.Status = core.CardActive
+	}
+	if err := s.st.SaveCard(owner, &card, ref); err != nil {
+		return card, err
+	}
+	err := s.projectSharedEmit(owner, producer, string(core.ProjectChannelKey(card.ProjectID)), card.SourceRunID, core.EvCardCreated, core.CardPayload{Card: card})
+	return card, err
+}
+
 // emitCard 按卡归属发布卡片事件（D50）：
 //   - 会话卡 → 会话流；同时镜像项目流（伪会话键 "project:<projectId>"）——
 //     项目收件箱聚合视图（项目级卡 ∪ 各会话未决卡）的实时性依赖它；卡片事件

internal/runmgr/events.go

@@ -116,6 +116,34 @@ func (eb *eventBuffer) emit(conversationID, runID string, typ core.StreamEventTy
 	return 0, nil
 }
 
+// emitProjectShared persists an explicit internal publication and its proof
+// before fan-out. A client-visible payload cannot opt itself into this path.
+func (eb *eventBuffer) emitProjectShared(owner, producer, channel, runID string, typ core.StreamEventType, payload any) error {
+	project, ok := core.ProjectFromChannel(channel)
+	if !ok {
+		return store.ErrNotFound
+	}
+	encoded, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	event := core.StreamEvent{ConversationID: channel, ProjectID: project, RunID: runID, Type: typ, Payload: encoded, TS: time.Now().UTC().Format(time.RFC3339Nano), V: core.ProtocolVersion}
+	eb.flushConversation(channel)
+	eb.flushing.Lock()
+	defer eb.flushing.Unlock()
+	events := []core.StreamEvent{event}
+	if _, err := eb.store.AppendProjectSharedEvents(owner, producer, channel, events); err != nil {
+		return err
+	}
+	for _, event := range events {
+		eb.bus.Publish(event)
+	}
+	if eb.onPersisted != nil {
+		eb.onPersisted(channel, events)
+	}
+	return nil
+}
+
 // append 落盘并按 seq 顺序发布。
 func (eb *eventBuffer) append(conversationID string, evs []core.StreamEvent) error {
 	eb.flushing.Lock()

internal/runmgr/runmgr.go

@@ -179,6 +179,12 @@ type IntentGate interface {
 	Match(ctx context.Context, input string) (branch, topID string)
 }
 
+// EmitProjectSharedCardEvent is an internal producer-authorized structural
+// publication entry, separate from the ordinary private card event emitter.
+func (m *Manager) EmitProjectSharedCardEvent(owner, producer, channel, runID string, typ core.StreamEventType, payload any) error {
+	return m.events.emitProjectShared(owner, producer, channel, runID, typ, payload)
+}
+
 // SetGate 注入意图闸门（装配期一次；插件意图 overlay 经 pluginhost 回调重建）。
 func (m *Manager) SetGate(g IntentGate) { m.gate = g }
 

internal/server/project_events.go

@@ -0,0 +1,30 @@
+package server
+
+import (
+	core "agent-system-core"
+)
+
+// projectStreamMembership consults the registered resource authorizer using the
+// physical project's trusted owner, never the event publisher or payload claims.
+func (s *Server) projectStreamMembership(actor, project string) bool {
+	p, err := s.st.ProjectMetadata(project)
+	if err != nil {
+		return false
+	}
+	return s.authorizeSharedResource(actor, "stream", "project", project, p.Owner, project)
+}
+
+func (s *Server) projectEventVisible(actor string, event core.StreamEvent) bool {
+	if s.projectLifecycle == nil {
+		return true
+	}
+	access, err := s.st.ResolveProjectEventAccess(event)
+	if err != nil || access.Owner == "" {
+		return false
+	}
+	if access.Owner == actor {
+		return true
+	}
+	project, ok := core.ProjectFromChannel(event.ConversationID)
+	return ok && access.Shared && s.projectStreamMembership(actor, project)
+}

internal/server/project_events_test.go

@@ -0,0 +1,163 @@
+package server
+
+import (
+	"agent-system-backend/internal/cardsvc"
+	"agent-system-backend/pkg/extension"
+	core "agent-system-core"
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func projectStreamReplay(t *testing.T, s *Server, key, project string) *httptest.ResponseRecorder {
+	t.Helper()
+	r := httptest.NewRequest("GET", "/api/events?scope=project&projectId="+project+"&since=0", nil)
+	r.Header.Set("Authorization", "Bearer "+key)
+	ctx, cancel := context.WithTimeout(r.Context(), 200*time.Millisecond)
+	defer cancel()
+	w := httptest.NewRecorder()
+	s.Handler().ServeHTTP(w, r.WithContext(ctx))
+	return w
+}
+func projectEventFixture(t *testing.T) (*Server, string, string, string, *atomic.Bool) {
+	t.Helper()
+	s, st := newAuthTestServer(t)
+	alice, _ := st.CreateAPIKey("alice", "test")
+	bob, _ := st.CreateAPIKey("bob", "test")
+	p, _ := st.CreateProject("alice", "Project")
+	s.Cards = cardsvc.New(st, s.mgr.EmitCardEvent)
+	s.Cards.SetProjectSharedEmitter(s.mgr.EmitProjectSharedCardEvent)
+	member := &atomic.Bool{}
+	member.Store(true)
+	s.SetResourceAuthorizerV2(func(q extension.ResourceAuthzQuery) (bool, error) {
+		return q.ResourceType == "project" && q.ProjectID == p.ID && (q.Actor == "alice" || (q.Actor == "bob" && member.Load())), nil
+	})
+	return s, alice, bob, p.ID, member
+}
+func TestProjectEventsLegacyReplayIsPrivate(t *testing.T) {
+	s, alice, bob, p, _ := projectEventFixture(t)
+	s.DisableUploadIndexing = true
+	if w := lifecycleUpload(t, s, alice, p, "legacy-owner-private.txt", "PRIVATE"); w.Code != 200 {
+		t.Fatal(w.Code)
+	}
+	s.SetProjectLifecycle(extension.ProjectLifecycle{Work: func(context.Context, string, core.ID) (bool, error) { return true, nil }}, nil)
+	member := projectStreamReplay(t, s, bob, p)
+	if member.Code != 200 || bytes.Contains(member.Body.Bytes(), []byte("legacy-owner-private.txt")) || bytes.Contains(member.Body.Bytes(), []byte("builtin:txt")) {
+		t.Fatal("legacy private upload leaked to member")
+	}
+	owner := projectStreamReplay(t, s, alice, p)
+	if owner.Code != 200 || !bytes.Contains(owner.Body.Bytes(), []byte("legacy-owner-private.txt")) {
+		t.Fatal("owner lost private history")
+	}
+}
+func TestProjectEventsSharedProofAndPrivatePagination(t *testing.T) {
+	s, _, bob, p, _ := projectEventFixture(t)
+	s.SetProjectLifecycle(extension.ProjectLifecycle{}, nil)
+	card, err := s.Cards.Create("alice", core.ActivityCard{ProjectID: p, Title: "PRIVATE HISTORY"}, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	payload, _ := json.Marshal(map[string]any{"card": card, "owner": "bob", "shared": true, "producer": "plugin:rdm:structure"})
+	private := make([]core.StreamEvent, 1001)
+	for i := range private {
+		private[i] = core.StreamEvent{Type: core.EvCardUpdated, Payload: payload, TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1}
+	}
+	if _, err := s.st.AppendEvents(core.ProjectChannelKey(p), private); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Cards.PublishProjectCard("alice", "plugin:rdm:structure", core.ActivityCard{ProjectID: p, Title: "EXPLICIT SHARED"}, ""); err != nil {
+		t.Fatal(err)
+	}
+	w := projectStreamReplay(t, s, bob, p)
+	if w.Code != 200 || bytes.Contains(w.Body.Bytes(), []byte("PRIVATE HISTORY")) || !bytes.Contains(w.Body.Bytes(), []byte("EXPLICIT SHARED")) {
+		t.Fatal("private pagination/spoof proof failed")
+	}
+}
+func TestProjectEventsLivePrivacyAndRevocation(t *testing.T) {
+	s, _, bob, p, member := projectEventFixture(t)
+	s.SetProjectLifecycle(extension.ProjectLifecycle{}, nil)
+	httpServer := httptest.NewServer(s.Handler())
+	defer httpServer.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	req, _ := http.NewRequestWithContext(ctx, "GET", httpServer.URL+"/api/events?scope=project&projectId="+p, nil)
+	req.Header.Set("Authorization", "Bearer "+bob)
+	response, err := httpServer.Client().Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	if _, err := s.Cards.Create("alice", core.ActivityCard{ProjectID: p, Title: "LIVE PRIVATE"}, ""); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Cards.PublishProjectCard("alice", "plugin:rdm:structure", core.ActivityCard{ProjectID: p, Title: "LIVE SHARED"}, ""); err != nil {
+		t.Fatal(err)
+	}
+	reader := bufio.NewReader(response.Body)
+	var before bytes.Buffer
+	for {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			t.Fatalf("live shared event did not arrive: %v", err)
+		}
+		before.WriteString(line)
+		if bytes.Contains([]byte(line), []byte("LIVE SHARED")) {
+			break
+		}
+	}
+	if bytes.Contains(before.Bytes(), []byte("LIVE PRIVATE")) {
+		t.Fatal("live private frame leaked before revocation")
+	}
+	member.Store(false)
+	if _, err := s.Cards.PublishProjectCard("alice", "plugin:rdm:structure", core.ActivityCard{ProjectID: p, Title: "AFTER REMOVAL"}, ""); err != nil {
+		t.Fatal(err)
+	}
+	body, err := io.ReadAll(reader)
+	if err != nil {
+		t.Fatalf("revoked stream did not stop: %v", err)
+	}
+	if bytes.Contains(body, []byte("LIVE PRIVATE")) || bytes.Contains(body, []byte("AFTER REMOVAL")) {
+		t.Fatal("live private/revoked event leaked")
+	}
+}
+func TestUnregisteredProjectEventsRetainLegacyReads(t *testing.T) {
+	s, _, bob, p, _ := projectEventFixture(t)
+	if _, err := s.Cards.Create("alice", core.ActivityCard{ProjectID: p, Title: "LEGACY BASE"}, ""); err != nil {
+		t.Fatal(err)
+	}
+	w := projectStreamReplay(t, s, bob, p)
+	if w.Code != 200 || !bytes.Contains(w.Body.Bytes(), []byte("LEGACY BASE")) {
+		t.Fatal("unregistered project stream changed")
+	}
+}
+
+func TestProjectEventsIdleStreamStopsAfterMembershipRemoval(t *testing.T) {
+	previous := sseAuthRecheck
+	sseAuthRecheck = 20 * time.Millisecond
+	defer func() { sseAuthRecheck = previous }()
+	s, _, bob, p, member := projectEventFixture(t)
+	s.SetProjectLifecycle(extension.ProjectLifecycle{}, nil)
+	httpServer := httptest.NewServer(s.Handler())
+	defer httpServer.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	request, _ := http.NewRequestWithContext(ctx, "GET", httpServer.URL+"/api/events?scope=project&projectId="+p, nil)
+	request.Header.Set("Authorization", "Bearer "+bob)
+	response, err := httpServer.Client().Do(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	member.Store(false)
+	body, err := io.ReadAll(response.Body)
+	if err != nil || !bytes.Contains(body, []byte("project-access-revoked")) {
+		t.Fatal("idle project stream kept revoked member connected")
+	}
+}

internal/server/project_lifecycle_test.go

@@ -409,8 +409,9 @@ func TestProjectLifecycleUploadSuppressesPrivateProjectCards(t *testing.T) {
 		return q.ResourceType == "project" && q.ProjectID == p.ID && (q.Actor == "alice" || q.Actor == "bob"), nil
 	})
 	s.Cards = cardsvc.New(st, s.mgr.EmitCardEvent)
+	s.Cards.SetProjectSharedEmitter(s.mgr.EmitProjectSharedCardEvent)
 	s.DisableUploadIndexing = true
-	if _, err := s.Cards.Create("alice", core.ActivityCard{ProjectID: p.ID, Title: "Shared structure", Kind: core.CardKindProgress}, ""); err != nil {
+	if _, err := s.Cards.PublishProjectCard("alice", "plugin:rdm:structure", core.ActivityCard{ProjectID: p.ID, Title: "Shared structure", Kind: core.CardKindProgress}, ""); err != nil {
 		t.Fatal(err)
 	}
 	// The owner path already admitted upload before the member authorization fix.

internal/server/server.go

@@ -431,6 +431,7 @@ var sseAuthRecheck = 15 * time.Second
 func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 	conversationID := r.URL.Query().Get("conversationId")
 	isProject := false
+	streamProject := ""
 	// 项目流（D50）：scope=project&projectId=<id> 订阅项目频道——伪会话键
 	// "project:<id>" 复用会话流的持久化/seq/重放机制；租户边界按项目校验。
 	if r.URL.Query().Get("scope") == "project" {
@@ -448,6 +449,7 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 		}
 		conversationID = string(core.ProjectChannelKey(core.ID(pid)))
 		isProject = true
+		streamProject = pid
 	} else if conversationID == "" {
 		http.Error(w, "conversationId required", http.StatusBadRequest)
 		return
@@ -509,28 +511,38 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 	live, stop := s.Subscribe(conversationID)
 	defer stop()
 
+	lastVisible := since
 	send := func(e core.StreamEvent) bool {
+		if e.Seq <= lastVisible {
+			return true
+		}
+		if isProject && !s.projectEventVisible(wantOwner, e) {
+			return r.Context().Err() == nil
+		}
 		buf, _ := json.Marshal(e)
 		fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", e.Seq, e.Type, buf)
 		flusher.Flush()
+		lastVisible = e.Seq
 		return r.Context().Err() == nil
 	}
 
-	lastSent := since
+	// Seen progresses even when a private frame is filtered. Visible only tracks
+	// frames delivered to this actor, preserving authoritative seq and resumption.
+	lastSeen := since
 	for {
-		replay, err := s.st.EventsAfter(conversationID, lastSent, 1000)
+		replay, err := s.st.EventsAfter(conversationID, lastSeen, 1000)
 		if err != nil || len(replay) == 0 {
 			break
 		}
 		for _, e := range replay {
-			if e.Seq <= lastSent {
+			if e.Seq <= lastSeen {
 				continue
 			}
 			if !send(e) {
 				stop()
 				return
 			}
-			lastSent = e.Seq
+			lastSeen = e.Seq
 		}
 		if len(replay) < 1000 {
 			break
@@ -548,6 +560,11 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 			stop()
 			return
 		case <-recheck.C:
+			if isProject && s.projectLifecycle != nil && !s.projectStreamMembership(wantOwner, streamProject) {
+				fmt.Fprint(w, ": project-access-revoked\n\n")
+				flusher.Flush()
+				return
+			}
 			if !stillAuthed() {
 				fmt.Fprint(w, ": session-expired\n\n")
 				flusher.Flush()
@@ -562,14 +579,19 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 				stop()
 				return
 			}
-			if e.ConversationID != conversationID || e.Seq <= lastSent {
+			if e.ConversationID != conversationID || e.Seq <= lastSeen {
 				continue
 			}
+			if isProject && s.projectLifecycle != nil && !s.projectStreamMembership(wantOwner, streamProject) {
+				fmt.Fprint(w, ": project-access-revoked\n\n")
+				flusher.Flush()
+				return
+			}
 			if !send(e) {
 				stop()
 				return
 			}
-			lastSent = e.Seq
+			lastSeen = e.Seq
 		}
 	}
 }

internal/store/messages.go

@@ -266,6 +266,10 @@ func (s *Store) ChildRuns(runID string) ([]core.Run, error) {
 // 成功后回填 evs[i].Seq。事务内锁定会话序列：PG 用事务级咨询锁（跨实例安全），
 // SQLite 单写连接天然串行。空批次为空操作。
 func (s *Store) AppendEvents(conversationID string, evs []core.StreamEvent) (int64, error) {
+	return s.appendEvents(conversationID, evs, "", "")
+}
+
+func (s *Store) appendEvents(conversationID string, evs []core.StreamEvent, owner, producer string) (int64, error) {
 	if len(evs) == 0 {
 		return 0, nil
 	}
@@ -298,6 +302,13 @@ func (s *Store) AppendEvents(conversationID string, evs []core.StreamEvent) (int
 	if _, err := tx.Exec(s.q(sb.String()), args...); err != nil {
 		return 0, err
 	}
+	if producer != "" {
+		for i, e := range evs {
+			if err := s.saveProjectEventProof(tx, conversationID, base.Int64+int64(i)+1, e, owner, producer); err != nil {
+				return 0, err
+			}
+		}
+	}
 	// KTD6：事件提交与 NOTIFY 同事务（outbox 语义的前半——通知不早于事件
 	// 可见；后半的命令 outbox 走 EnqueueCommandInTx 由批准路径使用）。
 	if s.dialect.name() == "postgres" {

internal/store/migrate.go

@@ -841,6 +841,13 @@ func init() {
 				`DROP INDEX IF EXISTS idx_redeem_attempts_ip_at`,
 				`DROP TABLE IF EXISTS redeem_attempts`,
 			}}},
+		migration{version: 30, name: "project_event_publication_proofs",
+			up: map[string][]string{
+				"sqlite":   {`CREATE TABLE project_event_shares(channel TEXT NOT NULL,seq INTEGER NOT NULL,event_type TEXT NOT NULL,payload_sha256 TEXT NOT NULL,owner TEXT NOT NULL,resource_type TEXT NOT NULL,resource_id TEXT NOT NULL,producer TEXT NOT NULL,created_at TEXT NOT NULL,PRIMARY KEY(channel,seq),FOREIGN KEY(channel,seq) REFERENCES events(conversation_id,seq) ON DELETE CASCADE)`},
+				"postgres": {`CREATE TABLE project_event_shares(channel TEXT NOT NULL,seq BIGINT NOT NULL,event_type TEXT NOT NULL,payload_sha256 TEXT NOT NULL,owner TEXT NOT NULL,resource_type TEXT NOT NULL,resource_id TEXT NOT NULL,producer TEXT NOT NULL,created_at TEXT NOT NULL,PRIMARY KEY(channel,seq),FOREIGN KEY(channel,seq) REFERENCES events(conversation_id,seq) ON DELETE CASCADE)`},
+			},
+			down: map[string][]string{"sqlite": {`DROP TABLE IF EXISTS project_event_shares`}, "postgres": {`DROP TABLE IF EXISTS project_event_shares`}},
+		},
 	)
 }
 

internal/store/migrate_compat.go

@@ -34,6 +34,7 @@ var migrationBreaksOldReaders = map[int]bool{
 	27: false,
 	28: false, // auth_version_account_events：仅加列与新表，旧读者不受影响
 	29: false, // redeem_attempts：新表，旧读者不受影响
+	30: false, // project_event_publication_proofs：新表，旧读者不受影响
 }
 
 // SchemaTarget 是本二进制会迁移到的 schema 版本。

internal/store/project_events.go

@@ -0,0 +1,147 @@
+package store
+
+import (
+	core "agent-system-core"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// ProjectEventAccess is trusted read metadata; payload claims never establish it.
+type ProjectEventAccess struct {
+	Owner    string
+	Shared   bool
+	Producer string
+}
+type eventResource struct{ owner, project, conversation, kind, id string }
+type eventQuery interface{ QueryRow(string, ...any) *sql.Row }
+
+func cardEventID(e core.StreamEvent) (string, error) {
+	switch e.Type {
+	case core.EvCardCreated, core.EvCardUpdated, core.EvCardStatus, core.EvCardResolved:
+	default:
+		return "", nil
+	}
+	var payload struct {
+		Card struct {
+			ID string `json:"id"`
+		} `json:"card"`
+		CardID string `json:"cardId"`
+	}
+	if err := json.Unmarshal(e.Payload, &payload); err != nil {
+		return "", err
+	}
+	switch e.Type {
+	case core.EvCardCreated, core.EvCardUpdated:
+		return payload.Card.ID, nil
+	case core.EvCardStatus, core.EvCardResolved:
+		return payload.CardID, nil
+	}
+	return "", nil
+}
+func eventPayloadHash(payload []byte) string {
+	hash := sha256.Sum256(payload)
+	return hex.EncodeToString(hash[:])
+}
+
+func (s *Store) projectEventResource(q eventQuery, e core.StreamEvent) (eventResource, error) {
+	var resource eventResource
+	id, err := cardEventID(e)
+	if err != nil {
+		return resource, err
+	}
+	switch e.Type {
+	case core.EvCardCreated, core.EvCardUpdated, core.EvCardStatus, core.EvCardResolved:
+		if id == "" {
+			return resource, ErrNotFound
+		}
+	}
+	if id != "" {
+		resource.kind, resource.id = "card", id
+		err = q.QueryRow(s.q(`SELECT owner,project_id,conversation_id FROM activity_cards WHERE id=?`), id).Scan(&resource.owner, &resource.project, &resource.conversation)
+	} else if e.Type == core.EvArtifactCreated {
+		var payload struct {
+			Artifact struct {
+				ID string `json:"id"`
+			} `json:"artifact"`
+		}
+		if err := json.Unmarshal(e.Payload, &payload); err != nil {
+			return resource, err
+		}
+		if payload.Artifact.ID == "" {
+			return resource, ErrNotFound
+		}
+		resource.kind, resource.id = "artifact", payload.Artifact.ID
+		err = q.QueryRow(s.q(`SELECT owner,project_id FROM artifacts WHERE id=?`), resource.id).Scan(&resource.owner, &resource.project)
+	} else if e.RunID != "" {
+		resource.kind, resource.id = "run", e.RunID
+		err = q.QueryRow(s.q(`SELECT owner,project_id,conversation_id FROM runs WHERE id=?`), e.RunID).Scan(&resource.owner, &resource.project, &resource.conversation)
+	} else {
+		return resource, ErrNotFound
+	}
+	if errors.Is(err, sql.ErrNoRows) {
+		return resource, ErrNotFound
+	}
+	if err != nil {
+		return resource, err
+	}
+	project, ok := core.ProjectFromChannel(e.ConversationID)
+	if !ok || resource.owner == "" || resource.project != project || (e.ProjectID != "" && e.ProjectID != project) {
+		return resource, ErrNotFound
+	}
+	return resource, nil
+}
+
+// ResolveProjectEventAccess resolves immutable resource ownership and verifies an
+// exact internal publication record. Old events retain their private default.
+func (s *Store) ResolveProjectEventAccess(e core.StreamEvent) (ProjectEventAccess, error) {
+	resource, err := s.projectEventResource(s.db, e)
+	if err != nil {
+		return ProjectEventAccess{}, err
+	}
+	access := ProjectEventAccess{Owner: resource.owner}
+	var owner, kind, id, producer, typ, hash string
+	err = s.db.QueryRow(s.q(`SELECT owner,resource_type,resource_id,producer,event_type,payload_sha256 FROM project_event_shares WHERE channel=? AND seq=?`), e.ConversationID, e.Seq).Scan(&owner, &kind, &id, &producer, &typ, &hash)
+	if errors.Is(err, sql.ErrNoRows) {
+		return access, nil
+	}
+	if err != nil {
+		return ProjectEventAccess{}, err
+	}
+	if owner != resource.owner || kind != resource.kind || id != resource.id || strings.TrimSpace(producer) == "" || typ != string(e.Type) || hash != eventPayloadHash(e.Payload) || resource.conversation != "" {
+		return ProjectEventAccess{}, ErrNotFound
+	}
+	access.Shared, access.Producer = true, producer
+	return access, nil
+}
+
+// AppendProjectSharedEvents is an internal publication boundary, not an API or
+// model tool. Events and their exact sharing proofs commit together. Caller
+// identity must own a project-level card; private conversation cards cannot be
+// published through this structural-card path.
+func (s *Store) AppendProjectSharedEvents(owner, producer, channel string, events []core.StreamEvent) (int64, error) {
+	if owner == "" || strings.TrimSpace(producer) == "" {
+		return 0, fmt.Errorf("store: shared event requires owner and producer")
+	}
+	if _, ok := core.ProjectFromChannel(channel); !ok {
+		return 0, ErrNotFound
+	}
+	return s.appendEvents(channel, events, owner, producer)
+}
+
+func (s *Store) saveProjectEventProof(tx *sql.Tx, channel string, seq int64, e core.StreamEvent, owner, producer string) error {
+	e.ConversationID = channel
+	resource, err := s.projectEventResource(tx, e)
+	if err != nil {
+		return err
+	}
+	if resource.kind != "card" || resource.owner != owner || resource.conversation != "" {
+		return ErrNotFound
+	}
+	_, err = tx.Exec(s.q(`INSERT INTO project_event_shares(channel,seq,event_type,payload_sha256,owner,resource_type,resource_id,producer,created_at) VALUES(?,?,?,?,?,?,?,?,?)`), channel, seq, string(e.Type), eventPayloadHash(e.Payload), resource.owner, resource.kind, resource.id, producer, nowISO())
+	return err
+}

internal/store/project_events_test.go

@@ -0,0 +1,140 @@
+package store
+
+import (
+	core "agent-system-core"
+	"encoding/json"
+	"testing"
+	"time"
+)
+
+func storedProjectCardEvent(t *testing.T, st *Store, owner, project, id string) core.StreamEvent {
+	t.Helper()
+	card := core.ActivityCard{ID: id, ProjectID: project, Title: "PRIVATE", CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC()}
+	if err := st.SaveCard(owner, &card, ""); err != nil {
+		t.Fatal(err)
+	}
+	payload, _ := json.Marshal(core.CardPayload{Card: card})
+	return core.StreamEvent{ConversationID: core.ProjectChannelKey(project), ProjectID: project, Type: core.EvCardCreated, Payload: payload, TS: time.Now().UTC().Format(time.RFC3339Nano), V: 1}
+}
+func TestProjectEventSharedProofBindsExactEvent(t *testing.T) {
+	st := openTestStore(t)
+	p, _ := st.CreateProject("alice", "Project")
+	e := storedProjectCardEvent(t, st, "alice", p.ID, "card-a")
+	private := []core.StreamEvent{e}
+	if _, err := st.AppendEvents(e.ConversationID, private); err != nil {
+		t.Fatal(err)
+	}
+	access, err := st.ResolveProjectEventAccess(private[0])
+	if err != nil || access.Owner != "alice" || access.Shared {
+		t.Fatalf("private access=%+v %v", access, err)
+	}
+	shared := []core.StreamEvent{e}
+	if _, err := st.AppendProjectSharedEvents("alice", "plugin:rdm:structure", e.ConversationID, shared); err != nil {
+		t.Fatal(err)
+	}
+	access, err = st.ResolveProjectEventAccess(shared[0])
+	if err != nil || !access.Shared || access.Owner != "alice" {
+		t.Fatalf("shared proof=%+v %v", access, err)
+	}
+	access, err = st.ResolveProjectEventAccess(private[0])
+	if err != nil || access.Shared {
+		t.Fatal("publication retroactively shared private history")
+	}
+	for _, mutate := range []func(*core.StreamEvent){func(e *core.StreamEvent) { e.Type = core.EvCardUpdated }, func(e *core.StreamEvent) { e.Payload = append(e.Payload, ' ') }, func(e *core.StreamEvent) { e.Seq += 100 }, func(e *core.StreamEvent) { e.ConversationID = "project:another" }, func(e *core.StreamEvent) { e.ProjectID = "another" }} {
+		changed := shared[0]
+		mutate(&changed)
+		access, err = st.ResolveProjectEventAccess(changed)
+		if err == nil && access.Shared {
+			t.Fatal("mutated event retained sharing")
+		}
+	}
+	bad := []core.StreamEvent{e}
+	before, _ := st.LastEventSeq(e.ConversationID)
+	if _, err := st.AppendProjectSharedEvents("bob", "plugin:rdm:structure", e.ConversationID, bad); err == nil {
+		t.Fatal("other owner issued publication")
+	}
+	after, _ := st.LastEventSeq(e.ConversationID)
+	if after != before {
+		t.Fatal("rejected publication appended event")
+	}
+}
+func TestProjectEventMigrationPreservesPrivateHistory(t *testing.T) {
+	st := openTestStore(t)
+	if err := st.MigrateDown(29); err != nil {
+		t.Fatal(err)
+	}
+	p, _ := st.CreateProject("alice", "Legacy")
+	e := storedProjectCardEvent(t, st, "alice", p.ID, "legacy-card")
+	events := []core.StreamEvent{e}
+	if _, err := st.AppendEvents(e.ConversationID, events); err != nil {
+		t.Fatal(err)
+	}
+	before, _ := st.EventsAfter(e.ConversationID, 0, 100)
+	if err := st.migrate(); err != nil {
+		t.Fatal(err)
+	}
+	after, _ := st.EventsAfter(e.ConversationID, 0, 100)
+	b, _ := json.Marshal(before)
+	a, _ := json.Marshal(after)
+	if string(a) != string(b) {
+		t.Fatal("upgrade changed private event history")
+	}
+	access, err := st.ResolveProjectEventAccess(after[0])
+	if err != nil || access.Owner != "alice" || access.Shared {
+		t.Fatalf("legacy privacy=%+v %v", access, err)
+	}
+}
+
+func TestProjectSharedEventBatchRollsBackProofAndEvents(t *testing.T) {
+	st := openTestStore(t)
+	p, _ := st.CreateProject("alice", "Project")
+	a := storedProjectCardEvent(t, st, "alice", p.ID, "card-a")
+	b := storedProjectCardEvent(t, st, "bob", p.ID, "card-b")
+	if _, err := st.AppendProjectSharedEvents("alice", "plugin:rdm:structure", a.ConversationID, []core.StreamEvent{a, b}); err == nil {
+		t.Fatal("mixed owner batch published")
+	}
+	var count int
+	for _, table := range []string{"events", "project_event_shares"} {
+		if err := st.db.QueryRow("SELECT COUNT(*) FROM " + table).Scan(&count); err != nil || count != 0 {
+			t.Fatalf("partial %s publication: count=%d error=%v", table, count, err)
+		}
+	}
+}
+
+func TestProjectEventInvalidProofAndMissingOwnerFailClosed(t *testing.T) {
+	st := openTestStore(t)
+	p, _ := st.CreateProject("alice", "Project")
+	e := storedProjectCardEvent(t, st, "alice", p.ID, "card-a")
+	shared := []core.StreamEvent{e}
+	if _, err := st.AppendProjectSharedEvents("alice", "plugin:rdm:structure", e.ConversationID, shared); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.db.Exec(`UPDATE project_event_shares SET producer=''`); err != nil {
+		t.Fatal(err)
+	}
+	if access, err := st.ResolveProjectEventAccess(shared[0]); err == nil || access.Shared {
+		t.Fatal("corrupted publication proof was accepted")
+	}
+	if _, err := st.db.Exec(`UPDATE activity_cards SET owner=''`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.ResolveProjectEventAccess(shared[0]); err == nil {
+		t.Fatal("missing resource owner was accepted")
+	}
+}
+
+func TestProjectSharingRejectsPrivateConversationCards(t *testing.T) {
+	st := openTestStore(t)
+	p, _ := st.CreateProject("alice", "Project")
+	e := storedProjectCardEvent(t, st, "alice", p.ID, "card-a")
+	if _, err := st.db.Exec(`UPDATE activity_cards SET conversation_id='private-conversation'`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.AppendProjectSharedEvents("alice", "plugin:rdm:structure", e.ConversationID, []core.StreamEvent{e}); err == nil {
+		t.Fatal("private conversation card publicly marked")
+	}
+	events, err := st.EventsAfter(e.ConversationID, 0, 100)
+	if err != nil || len(events) != 0 {
+		t.Fatal("rejected conversation publication appended events")
+	}
+}