diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go index bdb1561..7ee9cb2 100644 --- a/cmd/storage-migrate/main.go +++ b/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 diff --git a/internal/app/app.go b/internal/app/app.go index 67587f3..33749b7 100644 --- a/internal/app/app.go +++ b/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 { diff --git a/internal/cardsvc/cardsvc.go b/internal/cardsvc/cardsvc.go index a17b6dc..6b4d0ab 100644 --- a/internal/cardsvc/cardsvc.go +++ b/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:")—— // 项目收件箱聚合视图(项目级卡 ∪ 各会话未决卡)的实时性依赖它;卡片事件 diff --git a/internal/runmgr/events.go b/internal/runmgr/events.go index 9e41162..971e0c2 100644 --- a/internal/runmgr/events.go +++ b/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() diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index e5fe6a3..b3f3d99 100644 --- a/internal/runmgr/runmgr.go +++ b/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 } diff --git a/internal/server/project_events.go b/internal/server/project_events.go new file mode 100644 index 0000000..7058cec --- /dev/null +++ b/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) +} diff --git a/internal/server/project_events_test.go b/internal/server/project_events_test.go new file mode 100644 index 0000000..0360ded --- /dev/null +++ b/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") + } +} diff --git a/internal/server/project_lifecycle_test.go b/internal/server/project_lifecycle_test.go index 47ea686..a58553e 100644 --- a/internal/server/project_lifecycle_test.go +++ b/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. diff --git a/internal/server/server.go b/internal/server/server.go index 426070e..d90b3c3 100644 --- a/internal/server/server.go +++ b/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= 订阅项目频道——伪会话键 // "project:" 复用会话流的持久化/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 } } } diff --git a/internal/store/messages.go b/internal/store/messages.go index 36c2042..58724f2 100644 --- a/internal/store/messages.go +++ b/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" { diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 4428563..c31e0e8 100644 --- a/internal/store/migrate.go +++ b/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`}}, + }, ) } diff --git a/internal/store/migrate_compat.go b/internal/store/migrate_compat.go index ac0347d..86c692c 100644 --- a/internal/store/migrate_compat.go +++ b/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 版本。 diff --git a/internal/store/project_events.go b/internal/store/project_events.go new file mode 100644 index 0000000..ea6e7c4 --- /dev/null +++ b/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 +} diff --git a/internal/store/project_events_test.go b/internal/store/project_events_test.go new file mode 100644 index 0000000..2c69aab --- /dev/null +++ b/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") + } +}