From 147de055e504b392b124759bd555a56a039be311 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sat, 12 Sep 2026 16:37:22 -0400 Subject: [PATCH 1/2] feat(ingest): self-origin notification suppression seam and predicate (RIG-3326) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the IdentityResolver seam and the suppression predicate at the notify-router fan-out, per the frozen self-delegate-suppression record. An event's actor handle is resolved once per route; each subscriber's handle is resolved through the seam and memoized by account id. A dispatch is skipped only on a positive owner-qualified match — both owners and both agents non-empty and equal. The owner leg is load-bearing: an agent handle is unique only per owner, so a bare-handle match would false-suppress a genuine cross-agent notification between two owners' agents both named the same thing. Every ambiguous state delivers: human commenter, either side unqualified, no ownership row at the coordinate, a store fault (logged at warn), or a nil resolver. CHECKS and UPDATE carry no actor and never suppress; their tests build an event that DOES carry a matching actor, so the arm rather than the fixture is what keeps the dispatch. The STATE arm resolves a zero handle and delivers: its actor rides a forge_state_transitions memo that is not reachable through this two-method seam, which is the record's documented interim. Subscription scope is projected out of both subscriber queries and threaded to the router so the cursor advance can stay artifact-scoped. Co-authored-by: Matt Wilkinson --- go/internal/ingest/notify_reconcile_test.go | 4 +- go/internal/ingest/notify_router.go | 114 ++++++- go/internal/ingest/notify_router_test.go | 290 +++++++++++++++++- .../store/db/forge_subscriptions.sql.go | 8 +- go/internal/store/forge_subscriptions.go | 14 +- .../store/forge_subscriptions_pgtest_test.go | 22 +- .../store/queries/forge_subscriptions.sql | 4 +- go/server/forge_notify_e2e_pgtest_test.go | 2 + go/server/forge_notify_matrix_test.go | 12 +- go/server/forge_notify_pgtest_test.go | 6 +- go/server/serve.go | 5 +- 11 files changed, 446 insertions(+), 35 deletions(-) diff --git a/go/internal/ingest/notify_reconcile_test.go b/go/internal/ingest/notify_reconcile_test.go index 4853ffd7c..f7f52646e 100644 --- a/go/internal/ingest/notify_reconcile_test.go +++ b/go/internal/ingest/notify_reconcile_test.go @@ -90,7 +90,7 @@ func (r *fakeReader) ListNewArtifacts(_ context.Context, _ string, _ compassv1in func newReconciler(t *testing.T, rd forge.NotifyReader, st *fakeNotifyStore, d *fakeDispatcher) *NotifyReconciler { t.Helper() - router := NewNotifyRouter(st, d, &fakeChecksRoller{}, nil, testRef(), nil) + router := NewNotifyRouter(st, d, &fakeChecksRoller{}, nil, nil, testRef(), nil) return NewNotifyReconciler(rd, st, router, compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, "github.com", ReconcileConfig{Pace: -1}) // pacing disabled: no real sleeps in tests @@ -411,7 +411,7 @@ func TestRunImmediateSweepThenCancel(t *testing.T) { } d := &fakeDispatcher{} synctest.Test(t, func(t *testing.T) { - rc := NewNotifyReconciler(rd, st, NewNotifyRouter(st, d, &fakeChecksRoller{}, nil, testRef(), nil), + rc := NewNotifyReconciler(rd, st, NewNotifyRouter(st, d, &fakeChecksRoller{}, nil, nil, testRef(), nil), compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, "github.com", ReconcileConfig{Backstop: time.Hour, Pace: -1}) // long backstop: only the immediate sweep fires ctx, cancel := context.WithCancel(context.Background()) diff --git a/go/internal/ingest/notify_router.go b/go/internal/ingest/notify_router.go index bd30f343f..8aa58e853 100644 --- a/go/internal/ingest/notify_router.go +++ b/go/internal/ingest/notify_router.go @@ -23,12 +23,16 @@ import ( // mirror of store.ForgeNotifySubscriber (the no-store rule keeps the store type // out of this package). SubscriptionID is the ack correlation key; Project is // the subscriber's own container project (set only for a Linear container sub, -// "" otherwise) so an OPENED event matches only its project's container subs. +// "" otherwise) so an OPENED event matches only its project's container subs; +// Scope is the subscription's own scope, which self-origin suppression reads to +// keep the cursor advance artifact-scope-only (a container-scope sub is skipped +// but its cursor is never advanced). type NotifySubscriber struct { SubscriptionID string AgentAccountID string DeliveredRevision string Project string + Scope compassv1internal.ForgeSubscriptionScope } // ArtifactCursor is the router's view of one shared per-artifact FETCH cursor: @@ -132,6 +136,31 @@ type PullNumberResolver interface { PullNumberForSHA(ctx context.Context, repo, headSHA string) (uint64, error) } +// Handle is an owner-qualified Compass identity: the owning user's handle plus +// the agent's handle. Two handles match iff both owners and both agents are +// non-empty and equal — a bare agent handle is unique only per owner, so the +// owner leg is load-bearing, not decorative. +type Handle struct{ Owner, Agent string } + +// qualified reports whether both components are non-empty — the precondition for +// any positive match (an unqualified handle on either side fails open). +func (h Handle) qualified() bool { return h.Owner != "" && h.Agent != "" } + +// IdentityResolver resolves owner-qualified Compass handles for self-origin +// suppression. A zero Handle with a nil error is a clean miss; the caller MUST +// fail open (deliver). A non-nil error is a store fault — log and fail open. A +// nil IdentityResolver disables suppression entirely (the zero value is the +// fail-open posture). +type IdentityResolver interface { + // HandleForAccount resolves an agent account id to its owner-qualified + // Compass handle. + HandleForAccount(ctx context.Context, accountID string) (Handle, error) + // AuthorHandle resolves the recorded authoring agent at a coordinate (the + // DL-055 ownership row) to its owner-qualified handle. (provider, host) are + // bound by the server adapter, like NotifyStore. + AuthorHandle(ctx context.Context, repo string, kind compassv1internal.ForgeArtifactKind, number uint64) (Handle, error) +} + // NotifyRouter routes one normalized event: load the coordinate's snapshot, // apply the event (snapshot mutation + new revision digest), upsert the cursor, // then notify each matched subscriber. It never advances delivered_revision @@ -141,20 +170,23 @@ type NotifyRouter struct { dispatcher NotifyDispatcher checksRoller ChecksRoller pullNumbers PullNumberResolver + identities IdentityResolver forgeRef *compassv1.ForgeRef log *slog.Logger } // NewNotifyRouter returns a router over the durable seam st, the notify seam -// disp, the roll-up seam checks, and the head_sha->number resolution seam pulls, -// stamping forgeRef on every notification. A nil pulls disables step 0 (a CHECKS -// event with no number then fails the guard, the pre-RIG-2869 behavior). A nil -// log defaults to slog.Default so the router never nil-panics on the log path. -func NewNotifyRouter(st NotifyStore, disp NotifyDispatcher, checks ChecksRoller, pulls PullNumberResolver, forgeRef *compassv1.ForgeRef, log *slog.Logger) *NotifyRouter { +// disp, the roll-up seam checks, the head_sha->number resolution seam pulls, and +// the identity seam ids, stamping forgeRef on every notification. A nil pulls +// disables step 0 (a CHECKS event with no number then fails the guard, the +// pre-RIG-2869 behavior). A nil ids disables self-origin suppression entirely +// (every dispatch delivered). A nil log defaults to slog.Default so the router +// never nil-panics on the log path. +func NewNotifyRouter(st NotifyStore, disp NotifyDispatcher, checks ChecksRoller, pulls PullNumberResolver, ids IdentityResolver, forgeRef *compassv1.ForgeRef, log *slog.Logger) *NotifyRouter { if log == nil { log = slog.Default() } - return &NotifyRouter{store: st, dispatcher: disp, checksRoller: checks, pullNumbers: pulls, forgeRef: forgeRef, log: log} + return &NotifyRouter{store: st, dispatcher: disp, checksRoller: checks, pullNumbers: pulls, identities: ids, forgeRef: forgeRef, log: log} } // Route runs the frozen algorithm (design.md:841-872) for one event: @@ -288,7 +320,16 @@ func (r *NotifyRouter) Route(ctx context.Context, ev forge.ForgeEvent) error { } // 6. Build + dispatch a notification per subscriber, carrying revision. + // Self-origin suppression: the event's actor handle is a property of the + // event, so resolve it once here; the per-subscriber handle is resolved in + // selfOrigin, memoized per route. A nil identity seam leaves actor a zero + // Handle and every selfOrigin call false (suppression disabled). + actor := r.actorHandle(ctx, ev) + subMemo := map[string]Handle{} for _, sub := range subs { + if r.selfOrigin(ctx, actor, sub, subMemo) { + continue + } n := r.notification(ev, sub.SubscriptionID, revision) if derr := r.dispatcher.Notify(ctx, sub.AgentAccountID, n); derr != nil { // A vanished subscription / dropped session is logged, not fatal — @@ -332,6 +373,65 @@ func (r *NotifyRouter) SynthesizeUpdate(ctx context.Context, sub NotifySubscribe } } +// actorHandle resolves the event's owner-qualified ACTOR handle per the +// suppress/keep matrix. A zero Handle means no actor evidence or an unresolvable +// one — the caller fails open. A nil identity seam short-circuits every arm to +// the zero Handle, so suppression is disabled wholesale. +// +// COMMENT/REVIEW read the actor straight off the header-stamped CommentRef +// (unset for a human commenter -> zero Handle). OPENED resolves the DL-055 +// ownership row's recorded author, which IS the actor by construction. STATE's +// actor rides RIG-3331's forge_state_transitions memo, which is not reachable +// through this two-method seam, so STATE resolves the zero Handle here and +// delivers (the safe interim documented in §STATE) until the memo consumer is +// wired. CHECKS and UPDATE carry no actor and never suppress. +func (r *NotifyRouter) actorHandle(ctx context.Context, ev forge.ForgeEvent) Handle { + if r.identities == nil { + return Handle{} + } + switch ev.Change { + case compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_COMMENT, + compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_REVIEW: + agent := ev.Comment.GetAgent() + return Handle{Owner: agent.GetOwnerHandle(), Agent: agent.GetAgentHandle()} + case compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_OPENED: + h, err := r.identities.AuthorHandle(ctx, ev.Repo, ev.Kind, ev.Number) + if err != nil { + r.log.WarnContext(ctx, "forge notify author handle resolve failed", + "repo", ev.Repo, "number", ev.Number, "error", err) + return Handle{} + } + return h + default: + // STATE (interim), CHECKS, UPDATE: no reachable actor. + return Handle{} + } +} + +// selfOrigin reports whether dispatch to sub must be skipped: the event's actor +// and this subscriber's handles are both fully owner-qualified and equal. An +// unqualified actor short-circuits before any store read (the common no-actor +// path). The subscriber handle is resolved through the identity seam, memoized +// per route by account id; a resolver fault is logged and treated as a miss +// (fail open — deliver). +func (r *NotifyRouter) selfOrigin(ctx context.Context, actor Handle, sub NotifySubscriber, memo map[string]Handle) bool { + if !actor.qualified() { + return false + } + subHandle, ok := memo[sub.AgentAccountID] + if !ok { + h, err := r.identities.HandleForAccount(ctx, sub.AgentAccountID) + if err != nil { + r.log.WarnContext(ctx, "forge notify subscriber handle resolve failed", + "account", sub.AgentAccountID, "error", err) + h = Handle{} + } + memo[sub.AgentAccountID] = h + subHandle = h + } + return subHandle.qualified() && actor == subHandle +} + // notification builds the wire ForgeNotification for one subscriber: the // coordinate + the per-kind payload (comment / checks / state) + the snapshot // revision the agent echoes in its ack (design.md:341-345). It never sets diff --git a/go/internal/ingest/notify_router_test.go b/go/internal/ingest/notify_router_test.go index 0be21bd5f..4171bfd33 100644 --- a/go/internal/ingest/notify_router_test.go +++ b/go/internal/ingest/notify_router_test.go @@ -138,14 +138,14 @@ func testRef() *compassv1.ForgeRef { // every pre-RIG-2869 case exercises (a zero-number event is rejected). func newRouter(t *testing.T, st *fakeNotifyStore, d *fakeDispatcher, c *fakeChecksRoller) *NotifyRouter { t.Helper() - return NewNotifyRouter(st, d, c, nil, testRef(), nil) + return NewNotifyRouter(st, d, c, nil, nil, testRef(), nil) } // newRouterWithPulls builds a router with the head_sha->number resolution seam // wired (the RIG-2869 shape the prod GitHub lane uses). func newRouterWithPulls(t *testing.T, st *fakeNotifyStore, d *fakeDispatcher, c *fakeChecksRoller, p PullNumberResolver) *NotifyRouter { t.Helper() - return NewNotifyRouter(st, d, c, p, testRef(), nil) + return NewNotifyRouter(st, d, c, p, nil, testRef(), nil) } const ( @@ -157,6 +157,12 @@ const ( chOpened = compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_OPENED chUpdate = compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_UPDATE chReview = compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_REVIEW + + scopeContainer = compassv1internal.ForgeSubscriptionScope_FORGE_SUBSCRIPTION_SCOPE_CONTAINER + + // selfAgent is the agent handle every suppression fixture uses; the owner + // leg is what the cases vary. + selfAgent = "atlas" ) func ghComment(url, body, account string) *compassv1internal.CommentRef { @@ -172,6 +178,53 @@ func commentEvent(url string) forge.ForgeEvent { } } +// fakeIdentityResolver scripts the two identity-seam reads: an account-id -> +// Handle map for HandleForAccount, and a single author Handle for AuthorHandle +// (the OPENED coordinate). Either read can be forced to fault. authorMiss makes +// AuthorHandle return a zero Handle with a nil error (the ErrNotFound / clean +// miss the router treats as fail-open). +type fakeIdentityResolver struct { + accounts map[string]Handle + author Handle + authorMiss bool + accountErr error + authorErr error + authorCalls int +} + +func (f *fakeIdentityResolver) HandleForAccount(_ context.Context, accountID string) (Handle, error) { + if f.accountErr != nil { + return Handle{}, f.accountErr + } + return f.accounts[accountID], nil +} + +func (f *fakeIdentityResolver) AuthorHandle(_ context.Context, _ string, _ compassv1internal.ForgeArtifactKind, _ uint64) (Handle, error) { + f.authorCalls++ + if f.authorErr != nil { + return Handle{}, f.authorErr + } + if f.authorMiss { + return Handle{}, nil + } + return f.author, nil +} + +// newRouterWithIDs builds a router with the identity seam wired — the shape the +// self-origin suppression cases exercise. +func newRouterWithIDs(t *testing.T, st *fakeNotifyStore, d *fakeDispatcher, ids IdentityResolver) *NotifyRouter { + t.Helper() + return NewNotifyRouter(st, d, &fakeChecksRoller{}, nil, ids, testRef(), nil) +} + +// compassComment is a COMMENT event whose commenter is a Compass agent, carrying +// the owner-qualified attribution the header parse stamps. +func compassComment(owner string) forge.ForgeEvent { + ev := commentEvent("https://gh/o/r/issues/7#c1") + ev.Comment.Agent = &compassv1.AgentAttribution{AgentHandle: selfAgent, OwnerHandle: owner} + return ev +} + // ---- tests ---- // TestRouteEachKindRoutesAndNotifies: each kind reaches the exact-coordinate @@ -686,3 +739,236 @@ func subIDs(ns []*compassv1internal.ForgeNotification) []string { } return out } + +// ---- self-origin suppression (T1) ---- + +// stateEvent is a GitHub STATE event on o/r#7 (no reachable actor until the +// RIG-3331 memo consumer lands). +func stateEvent() forge.ForgeEvent { + return forge.ForgeEvent{ + Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, + Host: "github.com", Repo: "o/r", Kind: kindPR, Number: 7, + URL: "u", Change: chState, State: "closed", + } +} + +// TestSelfOriginCommentSuppressedOnMatch: a COMMENT whose actor's +// owner-qualified handle equals the subscriber's is skipped. +func TestSelfOriginCommentSuppressedOnMatch(t *testing.T) { + st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}} + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), compassComment("own")); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 0 { + t.Errorf("notifications = %d, want 0 (self-origin suppressed)", len(d.sent)) + } +} + +// TestSelfOriginCommentDeliveredHumanCommenter: a human commenter (Agent unset) +// always delivers — there is no actor handle to match. +func TestSelfOriginCommentDeliveredHumanCommenter(t *testing.T) { + st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}} + // commentEvent leaves Comment.Agent nil (human commenter). + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), commentEvent("https://gh/c1")); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Errorf("notifications = %d, want 1 (human commenter delivers)", len(d.sent)) + } +} + +// TestSelfOriginCommentDeliveredCrossOwnerSameAgent: the load-bearing +// owner-namespace-collision case — atlas@owner-A acts, atlas@owner-B subscribes. +// The bare agent handle matches but the owner differs, so this MUST deliver (a +// bare-handle match would be a fail-CLOSED cross-agent suppression bug). +func TestSelfOriginCommentDeliveredCrossOwnerSameAgent(t *testing.T) { + st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-B"}}} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{accounts: map[string]Handle{"acct-B": {Owner: "owner-B", Agent: "atlas"}}} + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), compassComment("owner-A")); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Errorf("notifications = %d, want 1 (atlas@owner-A != atlas@owner-B, must deliver)", len(d.sent)) + } +} + +// TestSelfOriginReviewSuppressedOnMatch: REVIEW shares the CommentRef actor +// source, so a self-review is suppressed the same way COMMENT is. +func TestSelfOriginReviewSuppressedOnMatch(t *testing.T) { + st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}} + ev := compassComment("own") + ev.Kind = kindPR + ev.Change = chReview + ev.State = "approved" + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), ev); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 0 { + t.Errorf("notifications = %d, want 0 (self-review suppressed)", len(d.sent)) + } +} + +// TestSelfOriginOpenedSuppressedViaAuthorHandle: OPENED resolves the actor +// through AuthorHandle; a match with the (container) subscriber suppresses. +func TestSelfOriginOpenedSuppressedViaAuthorHandle(t *testing.T) { + st := &fakeNotifyStore{openedSub: []NotifySubscriber{ + {SubscriptionID: "s", AgentAccountID: "acct-self", Project: "proj-A", Scope: scopeContainer}, + }} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{ + author: Handle{Owner: "own", Agent: "atlas"}, + accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}, + } + ev := forge.ForgeEvent{ + Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, Host: "linear.app", + Repo: "RIG", Kind: kindIssue, Number: 42, Project: "proj-A", URL: "u", Change: chOpened, + } + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), ev); err != nil { + t.Fatalf("Route: %v", err) + } + if ids.authorCalls != 1 { + t.Errorf("AuthorHandle calls = %d, want 1 (OPENED resolves the actor once)", ids.authorCalls) + } + if len(d.sent) != 0 { + t.Errorf("notifications = %d, want 0 (self-opened suppressed)", len(d.sent)) + } +} + +// TestSelfOriginOpenedDeliveredOnAuthorMiss: the webhook-races-the-row case — +// AuthorHandle is a clean miss (the DL-055 row not yet committed), so the actor +// is unresolved and the OPENED dispatch delivers (fail open). +func TestSelfOriginOpenedDeliveredOnAuthorMiss(t *testing.T) { + st := &fakeNotifyStore{openedSub: []NotifySubscriber{ + {SubscriptionID: "s", AgentAccountID: "acct-self", Project: "proj-A", Scope: scopeContainer}, + }} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{ + authorMiss: true, + accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}, + } + ev := forge.ForgeEvent{ + Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, Host: "linear.app", + Repo: "RIG", Kind: kindIssue, Number: 42, Project: "proj-A", URL: "u", Change: chOpened, + } + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), ev); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Errorf("notifications = %d, want 1 (author-row miss delivers, fail open)", len(d.sent)) + } +} + +// TestSelfOriginChecksNeverSuppressed is the CHECKS invariant: CI results on an +// agent's own push are the point of watching CI. The event carries a matching +// Compass commenter, so the arm — not the absence of actor evidence — is what +// keeps the dispatch. +func TestSelfOriginChecksNeverSuppressed(t *testing.T) { + st := &fakeNotifyStore{ + cursor: &ArtifactCursor{Repo: "o/r", Kind: kindPR, Number: 7}, + artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}, + } + d := &fakeDispatcher{} + // A resolver that would match ANY subscriber — proving the CHECKS arm never + // consults it. + ids := &fakeIdentityResolver{accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}} + roller := &fakeChecksRoller{res: forge.ConditionalResult[forge.Checks]{ + V: forge.Checks{HeadSHA: "sha1", State: "success"}, + }} + r := NewNotifyRouter(st, d, roller, nil, ids, testRef(), nil) + ev := compassComment("own") + ev.Repo = "o/r" + ev.Kind = kindPR + ev.Number = 7 + ev.Change = chChecks + ev.HeadSHA = "sha1" + if err := r.Route(context.Background(), ev); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Errorf("notifications = %d, want 1 (CHECKS is never suppressed)", len(d.sent)) + } +} + +// TestSelfOriginUpdateNeverSuppressed: UPDATE never suppresses even when the +// event carries an actor matching the subscriber, so the arm is what delivers. +func TestSelfOriginUpdateNeverSuppressed(t *testing.T) { + st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}} + ev := compassComment("own") + ev.Kind = kindIssue + ev.Change = chUpdate + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), ev); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Errorf("notifications = %d, want 1 (UPDATE never suppressed)", len(d.sent)) + } +} + +// TestSelfOriginStateDeliversWithNoMemoConsumer: STATE has no reachable actor +// through the two-method seam (RIG-3331's memo consumer is not wired), so the +// actor resolves to a zero Handle and STATE delivers (the safe interim). +func TestSelfOriginStateDeliversWithNoMemoConsumer(t *testing.T) { + st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}} + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), stateEvent()); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Errorf("notifications = %d, want 1 (STATE interim-open, no memo consumer)", len(d.sent)) + } +} + +// TestSelfOriginUnqualifiedActorDelivers: an actor whose owner (or agent) is +// empty is unqualified, so no positive match is possible and the dispatch +// delivers even to an identically-named subscriber. +func TestSelfOriginUnqualifiedActorDelivers(t *testing.T) { + st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{accounts: map[string]Handle{"acct-self": {Owner: "", Agent: "atlas"}}} + // Actor has an empty owner (a body whose header carried no owner). + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), compassComment("")); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Errorf("notifications = %d, want 1 (unqualified actor -> fail open)", len(d.sent)) + } +} + +// TestSelfOriginResolverFaultDelivers: a store fault from HandleForAccount is +// logged and treated as a miss — the dispatch delivers (fail open). +func TestSelfOriginResolverFaultDelivers(t *testing.T) { + st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{accountErr: errors.New("resolver boom")} + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), compassComment("own")); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Errorf("notifications = %d, want 1 (resolver fault -> fail open)", len(d.sent)) + } +} + +// TestSelfOriginNilResolverDeliversEverything: a nil IdentityResolver disables +// suppression wholesale — even a self-comment that would otherwise match is +// delivered. +func TestSelfOriginNilResolverDeliversEverything(t *testing.T) { + st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}} + d := &fakeDispatcher{} + // nil ids via newRouter (the legacy shape). + if err := newRouter(t, st, d, &fakeChecksRoller{}).Route(context.Background(), compassComment("own")); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Errorf("notifications = %d, want 1 (nil resolver disables suppression)", len(d.sent)) + } +} diff --git a/go/internal/store/db/forge_subscriptions.sql.go b/go/internal/store/db/forge_subscriptions.sql.go index 3c7c918aa..82baad9b7 100644 --- a/go/internal/store/db/forge_subscriptions.sql.go +++ b/go/internal/store/db/forge_subscriptions.sql.go @@ -176,7 +176,7 @@ func (q *Queries) GCForgeArtifactCursorIfUnsubscribed(ctx context.Context, arg G const listForgeNotifyTargets = `-- name: ListForgeNotifyTargets :many SELECT s.repo, s.kind, (CASE WHEN s.scope = 2 THEN 0 ELSE s.number END)::BIGINT AS coord_number, - s.id, s.agent_account_id, s.delivered_revision, s.project, + s.id, s.agent_account_id, s.delivered_revision, s.project, s.scope, (c.forge_provider IS NOT NULL)::boolean AS has_cursor, c.etag, c.comments_etag, c.checks_etag, c.revision, c.snapshot, c.polled_at FROM agent_forge_subscriptions s @@ -203,6 +203,7 @@ type ListForgeNotifyTargetsRow struct { AgentAccountID string DeliveredRevision string Project string + Scope int16 HasCursor bool Etag pgtype.Text CommentsEtag pgtype.Text @@ -233,6 +234,7 @@ func (q *Queries) ListForgeNotifyTargets(ctx context.Context, arg ListForgeNotif &i.AgentAccountID, &i.DeliveredRevision, &i.Project, + &i.Scope, &i.HasCursor, &i.Etag, &i.CommentsEtag, @@ -295,7 +297,7 @@ func (q *Queries) LoadForgeArtifactCursor(ctx context.Context, arg LoadForgeArti } const subscribersForArtifact = `-- name: SubscribersForArtifact :many -SELECT id, agent_account_id, delivered_revision, project +SELECT id, agent_account_id, delivered_revision, project, scope FROM agent_forge_subscriptions WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND kind = $4 AND ( @@ -319,6 +321,7 @@ type SubscribersForArtifactRow struct { AgentAccountID string DeliveredRevision string Project string + Scope int16 } // Exact-artifact subscribers, plus (on an opened event) the container-scope @@ -345,6 +348,7 @@ func (q *Queries) SubscribersForArtifact(ctx context.Context, arg SubscribersFor &i.AgentAccountID, &i.DeliveredRevision, &i.Project, + &i.Scope, ); err != nil { return nil, err } diff --git a/go/internal/store/forge_subscriptions.go b/go/internal/store/forge_subscriptions.go index 10cf297e4..18e000caa 100644 --- a/go/internal/store/forge_subscriptions.go +++ b/go/internal/store/forge_subscriptions.go @@ -221,16 +221,18 @@ func (s *Store) AgentForgeSubscriptionsForArtifact(ctx context.Context, provider // ForgeNotifySubscriber is one subscriber the notify path fans a change out to: // the subscription id (the ack correlation key), the owning agent, that // subscriber's last-notified DeliveredRevision (the router suppresses a -// re-notify when the change's revision equals it), and — for a collapsed -// container target whose subscribers span multiple Linear projects — the -// subscriber's own Project, so the router matches a project-P change to only -// its project-P subscribers ("" for artifact/GitHub subs). Struct shape frozen -// by the design record (RIG-2732 T3, §ListForgeNotifyTargets). +// re-notify when the change's revision equals it), the subscriber's own Project +// (for a collapsed container target spanning multiple Linear projects, so the +// router matches a project-P change to only its project-P subscribers; "" for +// artifact/GitHub subs), and its subscription Scope, which the router's +// self-origin suppression consults to gate the artifact-scope-only cursor +// advance apart from a container-scope skip. type ForgeNotifySubscriber struct { SubscriptionID string AgentAccountID AccountID DeliveredRevision string Project string + Scope ForgeSubscriptionScope } // ForgeArtifactCursor is one row of forge_artifact_cursors: the shared @@ -304,6 +306,7 @@ func (s *Store) SubscribersForArtifact(ctx context.Context, provider ForgeProvid AgentAccountID: AccountID(r.AgentAccountID), DeliveredRevision: r.DeliveredRevision, Project: r.Project, + Scope: ForgeSubscriptionScope(r.Scope), }) } return out, nil @@ -373,6 +376,7 @@ func (s *Store) ListForgeNotifyTargets(ctx context.Context, provider ForgeProvid AgentAccountID: AccountID(r.AgentAccountID), DeliveredRevision: r.DeliveredRevision, Project: r.Project, + Scope: ForgeSubscriptionScope(r.Scope), }) } return out, nil diff --git a/go/internal/store/forge_subscriptions_pgtest_test.go b/go/internal/store/forge_subscriptions_pgtest_test.go index 6e8eae519..195c79cd9 100644 --- a/go/internal/store/forge_subscriptions_pgtest_test.go +++ b/go/internal/store/forge_subscriptions_pgtest_test.go @@ -444,18 +444,21 @@ func TestSubscribersForArtifactGitHub(t *testing.T) { if len(subs) != 1 || subs[0].AgentAccountID != exact { t.Fatalf("not-opened subs = %+v, want just exact agent %q", subs, exact) } + if subs[0].Scope != ForgeSubscriptionScopeArtifact { + t.Errorf("exact subscriber scope = %d, want ARTIFACT(%d)", subs[0].Scope, ForgeSubscriptionScopeArtifact) + } // Opened event: exact + container. subs, err = s.SubscribersForArtifact(ctx, provider, host, repo, kind, number, "", true) if err != nil { t.Fatalf("SubscribersForArtifact (opened): %v", err) } - got := map[AccountID]bool{} + got := map[AccountID]ForgeSubscriptionScope{} for _, sub := range subs { - got[sub.AgentAccountID] = true + got[sub.AgentAccountID] = sub.Scope } - if len(subs) != 2 || !got[exact] || !got[ctr] { - t.Fatalf("opened subs = %+v, want exact %q + container %q", subs, exact, ctr) + if len(subs) != 2 || got[exact] != ForgeSubscriptionScopeArtifact || got[ctr] != ForgeSubscriptionScopeContainer { + t.Fatalf("opened subs = %+v, want exact %q (ARTIFACT) + container %q (CONTAINER)", subs, exact, ctr) } } @@ -711,12 +714,23 @@ func TestListForgeNotifyTargetsMixedArtifactAndContainer(t *testing.T) { if len(artTarget.Subscribers) != 1 { t.Fatalf("artifact subscribers = %d, want 1", len(artTarget.Subscribers)) } + // The sweep lane must project each subscriber's real scope — the router's + // artifact-scope-only cursor advance depends on it, and a missing copy + // would silently arrive as ARTIFACT(1) for a CONTAINER row. + if artTarget.Subscribers[0].Scope != ForgeSubscriptionScopeArtifact { + t.Errorf("artifact subscriber scope = %d, want ARTIFACT(%d)", artTarget.Subscribers[0].Scope, ForgeSubscriptionScopeArtifact) + } if containerTarget == nil { t.Fatalf("no collapsed container target (number=0) in %+v", targets) } if len(containerTarget.Subscribers) != 2 { t.Fatalf("container subscribers = %d, want 2", len(containerTarget.Subscribers)) } + for _, sub := range containerTarget.Subscribers { + if sub.Scope != ForgeSubscriptionScopeContainer { + t.Errorf("container subscriber %s scope = %d, want CONTAINER(%d)", sub.AgentAccountID, sub.Scope, ForgeSubscriptionScopeContainer) + } + } } // ── T3: AdvanceForgeDeliveredRevision ───────────────────────────────────────── diff --git a/go/internal/store/queries/forge_subscriptions.sql b/go/internal/store/queries/forge_subscriptions.sql index 28d344ee6..1ac4150c3 100644 --- a/go/internal/store/queries/forge_subscriptions.sql +++ b/go/internal/store/queries/forge_subscriptions.sql @@ -44,7 +44,7 @@ SELECT count(*) FROM agent_forge_subscriptions -- name: SubscribersForArtifact :many -- Exact-artifact subscribers, plus (on an opened event) the container-scope -- subscribers for the same container/project. -SELECT id, agent_account_id, delivered_revision, project +SELECT id, agent_account_id, delivered_revision, project, scope FROM agent_forge_subscriptions WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND kind = $4 AND ( @@ -59,7 +59,7 @@ WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND kind = $4 -- (repo, kind) to coord_number 0. The Go groups the flat rows into targets. SELECT s.repo, s.kind, (CASE WHEN s.scope = 2 THEN 0 ELSE s.number END)::BIGINT AS coord_number, - s.id, s.agent_account_id, s.delivered_revision, s.project, + s.id, s.agent_account_id, s.delivered_revision, s.project, s.scope, (c.forge_provider IS NOT NULL)::boolean AS has_cursor, c.etag, c.comments_etag, c.checks_etag, c.revision, c.snapshot, c.polled_at FROM agent_forge_subscriptions s diff --git a/go/server/forge_notify_e2e_pgtest_test.go b/go/server/forge_notify_e2e_pgtest_test.go index eb4357092..5d80bfec6 100644 --- a/go/server/forge_notify_e2e_pgtest_test.go +++ b/go/server/forge_notify_e2e_pgtest_test.go @@ -158,6 +158,7 @@ func newNotifyE2EWire(t *testing.T) *notifyE2EWire { &forgeNotifyDispatcher{hub: hub}, &matrixChecksRoller{}, // a CHECKS cell here carries no head SHA, so step 0 never resolves and the roller is never reached; a trivially-scripted roller is correct. nil, // no pull-number resolver: this lane's fixtures carry explicit numbers. + nil, // no identity resolver: self-origin suppression is the T3 lane wiring, not this seam-level assembly. mxRef(), log, ) @@ -169,6 +170,7 @@ func newNotifyE2EWire(t *testing.T) *notifyE2EWire { &forgeNotifyDispatcher{hub: hub}, &matrixChecksRoller{}, nil, + nil, &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, Host: "linear.app"}, log, ) diff --git a/go/server/forge_notify_matrix_test.go b/go/server/forge_notify_matrix_test.go index efffcea46..d15bcf821 100644 --- a/go/server/forge_notify_matrix_test.go +++ b/go/server/forge_notify_matrix_test.go @@ -392,7 +392,7 @@ func TestForgeNotifyMatrix_Route(t *testing.T) { t.Run(tc.name, func(t *testing.T) { st := &matrixNotifyStore{artifactSub: []ingest.NotifySubscriber{sub}} d := &matrixDispatcher{} - r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, nil, mxRef(), nil) + r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, nil, nil, mxRef(), nil) if err := r.Route(t.Context(), tc.ev); err != nil { t.Fatalf("Route: %v", err) } @@ -433,7 +433,7 @@ func TestForgeNotifyMatrix_ContainerScope(t *testing.T) { container := ingest.NotifySubscriber{SubscriptionID: "repo-sub", AgentAccountID: "acct-repo"} st := &matrixNotifyStore{openedSub: []ingest.NotifySubscriber{container}} d := &matrixDispatcher{} - r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, nil, mxRef(), nil) + r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, nil, nil, mxRef(), nil) ev := forge.ForgeEvent{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, Host: "github.com", Repo: "octo/repo", Kind: mxIssue, Number: 99, URL: "u", Change: mxOpened} if err := r.Route(t.Context(), ev); err != nil { t.Fatalf("Route: %v", err) @@ -451,7 +451,7 @@ func TestForgeNotifyMatrix_ContainerScope(t *testing.T) { otherProj := ingest.NotifySubscriber{SubscriptionID: "other-sub", AgentAccountID: "acct-other", Project: "proj-beta"} st := &matrixNotifyStore{openedSub: []ingest.NotifySubscriber{inProj, otherProj}} d := &matrixDispatcher{} - r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, nil, &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, Host: "linear.app"}, nil) + r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, nil, nil, &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, Host: "linear.app"}, nil) ev := forge.ForgeEvent{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, Host: "linear.app", Repo: "SEA", Kind: mxIssue, Number: 42, Project: "proj-alpha", URL: "u", Change: mxOpened} if err := r.Route(t.Context(), ev); err != nil { t.Fatalf("Route: %v", err) @@ -466,7 +466,7 @@ func TestForgeNotifyMatrix_ContainerScope(t *testing.T) { container := ingest.NotifySubscriber{SubscriptionID: "repo-sub", AgentAccountID: "acct-repo"} st := &matrixNotifyStore{artifactSub: []ingest.NotifySubscriber{exact}, openedSub: []ingest.NotifySubscriber{container}} d := &matrixDispatcher{} - r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, nil, mxRef(), nil) + r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, nil, nil, mxRef(), nil) // A COMMENT (not OPENED) is artifact-scope: no container fan-in. ev := forge.ForgeEvent{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, Host: "github.com", Repo: "octo/repo", Kind: mxIssue, Number: 7, URL: "u#c", Change: mxComment, Comment: &compassv1internal.CommentRef{Url: "u#c", Body: "x", ForgeAccount: "a"}} if err := r.Route(t.Context(), ev); err != nil { @@ -517,7 +517,7 @@ func TestForgeNotifyMatrix_CheckSuiteResolvesPRNumber(t *testing.T) { st := &matrixNotifyStore{artifactSub: []ingest.NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "a"}}} d := &matrixDispatcher{} pulls := &matrixPullNumbers{number: 4242} - r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, pulls, mxRef(), nil) + r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, pulls, nil, mxRef(), nil) if err := r.Route(t.Context(), ev); err != nil { t.Fatalf("Route(check_suite): %v", err) } @@ -541,7 +541,7 @@ func TestForgeNotifyMatrix_CheckSuiteResolvesPRNumber(t *testing.T) { st := &matrixNotifyStore{artifactSub: []ingest.NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "a"}}} d := &matrixDispatcher{} pulls := &matrixPullNumbers{err: forge.ErrNoPullRequestForSHA} - r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, pulls, mxRef(), nil) + r := ingest.NewNotifyRouter(st, d, &matrixChecksRoller{}, pulls, nil, mxRef(), nil) if err := r.Route(t.Context(), ev); err == nil { t.Fatal("Route(check_suite, no PR for head sha) = nil error, want the route to fail closed") } diff --git a/go/server/forge_notify_pgtest_test.go b/go/server/forge_notify_pgtest_test.go index 0b9b77153..49dc9a423 100644 --- a/go/server/forge_notify_pgtest_test.go +++ b/go/server/forge_notify_pgtest_test.go @@ -180,7 +180,7 @@ func TestForgeNotifyRoutedAdvancesFetchCursorOnly(t *testing.T) { notifyStore := &forgeNotifyStore{st: st, provider: store.ForgeProviderGitHub, host: forgeTestHost} disp := &recordingDispatcher{} forgeRef := &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, Host: forgeTestHost} - router := ingest.NewNotifyRouter(notifyStore, disp, fixedChecksRoller{}, nil, forgeRef, nil) + router := ingest.NewNotifyRouter(notifyStore, disp, fixedChecksRoller{}, nil, nil, forgeRef, nil) arm := ingest.NewNotifyWebhookArm(router, ingest.NotifyArmConfig{}) // Precondition: never observed → no fetch cursor, empty delivered_revision. @@ -283,7 +283,7 @@ func TestForgeNotifyNoLiveSessionIsNonFatal(t *testing.T) { notifyStore := &forgeNotifyStore{st: st, provider: store.ForgeProviderGitHub, host: forgeTestHost} disp := &recordingDispatcher{noSession: true} forgeRef := &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, Host: forgeTestHost} - router := ingest.NewNotifyRouter(notifyStore, disp, fixedChecksRoller{}, nil, forgeRef, nil) + router := ingest.NewNotifyRouter(notifyStore, disp, fixedChecksRoller{}, nil, nil, forgeRef, nil) // The route must NOT fail: a per-subscriber dispatch error is logged and // skipped, never propagated (design.md:233-243). @@ -387,7 +387,7 @@ func TestLinearNotifyRoutedOpenedFansOutToProject(t *testing.T) { notifyStore := &forgeNotifyStore{st: st, provider: store.ForgeProviderLinear, host: host} disp := &recordingDispatcher{} forgeRef := &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, Host: host} - router := ingest.NewNotifyRouter(notifyStore, disp, fixedChecksRoller{}, nil, forgeRef, nil) + router := ingest.NewNotifyRouter(notifyStore, disp, fixedChecksRoller{}, nil, nil, forgeRef, nil) if err := router.Route(ctx, linearOpenedEvent(repo, number, alpha, url)); err != nil { t.Fatalf("Route: %v", err) diff --git a/go/server/serve.go b/go/server/serve.go index 52b579e36..134d3a462 100644 --- a/go/server/serve.go +++ b/go/server/serve.go @@ -1566,6 +1566,7 @@ func toIngestSubscribers(subs []store.ForgeNotifySubscriber) []ingest.NotifySubs AgentAccountID: string(s.AgentAccountID), DeliveredRevision: s.DeliveredRevision, Project: s.Project, + Scope: compassv1internal.ForgeSubscriptionScope(s.Scope), }) } return out @@ -1707,7 +1708,7 @@ func buildForgeNotifyLane( Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, Host: fc.Host, } - router := ingest.NewNotifyRouter(notifyStore, dispatcher, checks, pulls, forgeRef, log) + router := ingest.NewNotifyRouter(notifyStore, dispatcher, checks, pulls, nil, forgeRef, log) arm := ingest.NewNotifyWebhookArm(router, ingest.NotifyArmConfig{Log: log}) reconciler := ingest.NewNotifyReconciler(client, notifyStore, router, compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, fc.Host, ingest.ReconcileConfig{ @@ -1757,7 +1758,7 @@ func buildLinearNotifyLane( // Nil pull-number resolver: Linear is issues-only and never produces a // CHECKS event, so there is no head SHA to resolve (the router tolerates a // nil resolver and keeps the pre-RIG-2869 guard behavior). - router := ingest.NewNotifyRouter(notifyStore, dispatcher, checks, nil, forgeRef, log) + router := ingest.NewNotifyRouter(notifyStore, dispatcher, checks, nil, nil, forgeRef, log) arm := ingest.NewNotifyWebhookArm(router, ingest.NotifyArmConfig{Log: log}) reconciler := ingest.NewNotifyReconciler(client, notifyStore, router, compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, host, ingest.ReconcileConfig{ From 1d0eb03dc7e9e58de9fbb9d0bdf92f578fe84852 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sat, 12 Sep 2026 17:24:58 -0400 Subject: [PATCH 2/2] test(ingest): make the STATE and fault invariants prove the code, not the fixture (RIG-3326) The STATE interim-open test passed because its event carried no actor at all, so STATE resolved a zero handle through any arm. Routing STATE through the author row - the proxy the record forbids, because it would eat an agent's notification that a human closed its issue - left the test green. The STATE event now carries a matching Compass commenter and a matching author row, so delivering holds only if STATE reached the no-actor arm. Both misroutes now fail it. The two resolver-fault tests had the same shape: the fake returned a zero handle alongside its error, so a caller that used the value instead of failing open still delivered. The fake now returns a populated handle with the error and the account fixture matches the subscriber, so using it suppresses and the tests catch it. Adds the missing AuthorHandle fault case - the record requires a store fault there to log and deliver, and only the HandleForAccount side was covered. Drops the stateEvent helper, now unused. Co-authored-by: Matt Wilkinson --- go/internal/ingest/notify_router_test.go | 66 ++++++++++++++++++------ 1 file changed, 50 insertions(+), 16 deletions(-) diff --git a/go/internal/ingest/notify_router_test.go b/go/internal/ingest/notify_router_test.go index 4171bfd33..3c643ad1f 100644 --- a/go/internal/ingest/notify_router_test.go +++ b/go/internal/ingest/notify_router_test.go @@ -192,9 +192,12 @@ type fakeIdentityResolver struct { authorCalls int } +// The two error paths return a POPULATED handle alongside the error, so a +// caller that used the value instead of failing open would suppress — that is +// what makes the fault cases discriminating. func (f *fakeIdentityResolver) HandleForAccount(_ context.Context, accountID string) (Handle, error) { if f.accountErr != nil { - return Handle{}, f.accountErr + return f.accounts[accountID], f.accountErr } return f.accounts[accountID], nil } @@ -202,7 +205,7 @@ func (f *fakeIdentityResolver) HandleForAccount(_ context.Context, accountID str func (f *fakeIdentityResolver) AuthorHandle(_ context.Context, _ string, _ compassv1internal.ForgeArtifactKind, _ uint64) (Handle, error) { f.authorCalls++ if f.authorErr != nil { - return Handle{}, f.authorErr + return f.author, f.authorErr } if f.authorMiss { return Handle{}, nil @@ -742,16 +745,6 @@ func subIDs(ns []*compassv1internal.ForgeNotification) []string { // ---- self-origin suppression (T1) ---- -// stateEvent is a GitHub STATE event on o/r#7 (no reachable actor until the -// RIG-3331 memo consumer lands). -func stateEvent() forge.ForgeEvent { - return forge.ForgeEvent{ - Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, - Host: "github.com", Repo: "o/r", Kind: kindPR, Number: 7, - URL: "u", Change: chState, State: "closed", - } -} - // TestSelfOriginCommentSuppressedOnMatch: a COMMENT whose actor's // owner-qualified handle equals the subscriber's is skipped. func TestSelfOriginCommentSuppressedOnMatch(t *testing.T) { @@ -865,6 +858,32 @@ func TestSelfOriginOpenedDeliveredOnAuthorMiss(t *testing.T) { } } +// TestSelfOriginOpenedDeliveredOnAuthorFault: a store fault resolving the +// author row is logged and treated as a miss, so the dispatch delivers. The +// subscriber WOULD match the faulting coordinate's author, so a fault that +// suppressed instead would eat a real notification. +func TestSelfOriginOpenedDeliveredOnAuthorFault(t *testing.T) { + st := &fakeNotifyStore{openedSub: []NotifySubscriber{ + {SubscriptionID: "s", AgentAccountID: "acct-self", Project: "proj-A", Scope: scopeContainer}, + }} + d := &fakeDispatcher{} + ids := &fakeIdentityResolver{ + authorErr: errors.New("store unreachable"), + author: Handle{Owner: "own", Agent: selfAgent}, + accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: selfAgent}}, + } + ev := forge.ForgeEvent{ + Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, Host: "linear.app", + Repo: "RIG", Kind: kindIssue, Number: 42, Project: "proj-A", URL: "u", Change: chOpened, + } + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), ev); err != nil { + t.Fatalf("Route: %v", err) + } + if len(d.sent) != 1 { + t.Errorf("notifications = %d, want 1 (author fault delivers, fail open)", len(d.sent)) + } +} + // TestSelfOriginChecksNeverSuppressed is the CHECKS invariant: CI results on an // agent's own push are the point of watching CI. The event carries a matching // Compass commenter, so the arm — not the absence of actor evidence — is what @@ -915,12 +934,24 @@ func TestSelfOriginUpdateNeverSuppressed(t *testing.T) { // TestSelfOriginStateDeliversWithNoMemoConsumer: STATE has no reachable actor // through the two-method seam (RIG-3331's memo consumer is not wired), so the -// actor resolves to a zero Handle and STATE delivers (the safe interim). +// actor resolves to a zero Handle and STATE delivers (the safe interim). The +// event is loaded so that EVERY other arm would match — a matching Compass +// commenter and a matching author row — so delivering proves STATE fell +// through to the no-actor arm. Routing STATE through the author row is the +// fail-closed bug the record forbids: it would eat an agent's notification +// that a HUMAN closed its issue. func TestSelfOriginStateDeliversWithNoMemoConsumer(t *testing.T) { st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}} d := &fakeDispatcher{} - ids := &fakeIdentityResolver{accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: "atlas"}}} - if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), stateEvent()); err != nil { + ids := &fakeIdentityResolver{ + author: Handle{Owner: "own", Agent: selfAgent}, + accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: selfAgent}}, + } + ev := compassComment("own") + ev.Kind = kindPR + ev.Change = chState + ev.State = "closed" + if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), ev); err != nil { t.Fatalf("Route: %v", err) } if len(d.sent) != 1 { @@ -949,7 +980,10 @@ func TestSelfOriginUnqualifiedActorDelivers(t *testing.T) { func TestSelfOriginResolverFaultDelivers(t *testing.T) { st := &fakeNotifyStore{artifactSub: []NotifySubscriber{{SubscriptionID: "s", AgentAccountID: "acct-self"}}} d := &fakeDispatcher{} - ids := &fakeIdentityResolver{accountErr: errors.New("resolver boom")} + ids := &fakeIdentityResolver{ + accountErr: errors.New("resolver boom"), + accounts: map[string]Handle{"acct-self": {Owner: "own", Agent: selfAgent}}, + } if err := newRouterWithIDs(t, st, d, ids).Route(context.Background(), compassComment("own")); err != nil { t.Fatalf("Route: %v", err) }