From cb1bfa8acd587642275bbd4baeaa4018c75f62f5 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 30 Aug 2026 15:47:32 -0400 Subject: [PATCH] feat(server): wire the Linear agent-notification lane data sink (RIG-2732 T7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Linear webhook ingress (POST /webhooks/linear, #744) landed with its data-change sink injected-and-nil-for-now (DL-302): a verified Issue/Comment event acked-and-dropped. This builds the Linear-provider-bound notify lane and threads its sink into the handler, so a verified event now routes to subscribers at the LINEAR/linear.app coordinate. - buildLinearNotifyLane: the Linear sibling of buildForgeNotifyLane, gated App-INDEPENDENTLY on LINEAR_FORGE_TOKEN (forgeSecretDeclared) — the same credential the write path's Linear coordinate uses; undeclared -> nil lane (off-state), a resolve fault fails fast. Binds store.ForgeProviderLinear / "linear.app", client forge.NewLinear, reconciler Backstop 0 (-> ingest default). - forgeNotifyChecksRoller generalized to hold a forge.NotifyReader instead of a concrete *forge.GitHub, so one adapter serves both lanes (both *forge.GitHub and *forge.Linear satisfy NotifyReader). Behavior-preserving for GitHub. Linear's ChecksConditional returns ErrUnsupported but the router invokes RollUp only on a CHECKS event Linear (issues-only) never produces. - Serve builds the lane (same fail-fast cleanup path as the board wiring) and threads its sink through buildDoors (new linearDataSink param) into buildLinearWebhookWiring, replacing the nil sink. The two gates stay independent: the webhook secret gates the handler; LINEAR_FORGE_TOKEN gates the sink. - startForgeIngestLanes now takes the notify lanes variadically; the Linear lane's arm + reconciler start on the serve errgroup. A nil lane starts nothing. - The session arm (sessionSink) stays nil (RIG-2717's separate responder lane). Refs RIG-2732 Co-authored-by: Matt Wilkinson --- go/server/forge_notify_pgtest_test.go | 109 ++++++++++++++++++++++ go/server/linear_webhook_test.go | 40 +++++++++ go/server/serve.go | 124 +++++++++++++++++++++----- go/server/serve_forge_test.go | 42 +++++++++ go/server/sinks.go | 28 +++--- 5 files changed, 313 insertions(+), 30 deletions(-) diff --git a/go/server/forge_notify_pgtest_test.go b/go/server/forge_notify_pgtest_test.go index ac402cb2..8fd51ad8 100644 --- a/go/server/forge_notify_pgtest_test.go +++ b/go/server/forge_notify_pgtest_test.go @@ -303,3 +303,112 @@ func TestForgeNotifyNoLiveSessionIsNonFatal(t *testing.T) { t.Fatalf("recorded notifications = %d after no-session, want 0", len(disp.sent)) } } + +// --- test: Linear routed OPENED fans out to the matching project only --------- + +// seedLinearContainerSub creates an agent + owning user and a Linear +// container-scope subscription at (LINEAR, "linear.app", repo/team, ISSUE) bound +// to project, returning the agent account id and subscription id. A Linear +// container subscription REQUIRES a project (store enforces it, +// forge_subscriptions.go:108-111). +func seedLinearContainerSub(t *testing.T, st *store.Store, handle, repo, project string) (store.AccountID, string) { + t.Helper() + ctx := context.Background() // test root + owner, err := st.CreateUser(ctx, store.NewUser{Handle: handle + "-owner", DisplayName: "Owner"}) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + agent, err := st.CreateAgent(ctx, owner.ID, store.NewAgent{Handle: handle, DisplayName: "Agent"}) + if err != nil { + t.Fatalf("CreateAgent: %v", err) + } + subID, err := st.EnsureAgentForgeSubscription(ctx, store.AgentForgeSubscription{ + AgentAccountID: agent.ID, + Provider: store.ForgeProviderLinear, + Host: "linear.app", + Repo: repo, + Kind: store.ForgeArtifactKindIssue, + Scope: store.ForgeSubscriptionScopeContainer, + Project: project, + }) + if err != nil { + t.Fatalf("EnsureAgentForgeSubscription: %v", err) + } + return agent.ID, subID +} + +// linearOpenedEvent builds a Linear issue-OPENED ForgeEvent at the LINEAR +// coordinate carrying a project — the container-fan-out trigger. +func linearOpenedEvent(repo string, number uint64, project, url string) forge.ForgeEvent { + return forge.ForgeEvent{ + Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, + Host: "linear.app", + Repo: repo, + Kind: compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_ISSUE, + Number: number, + Project: project, + URL: url, + Change: compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_OPENED, + } +} + +// TestLinearNotifyRoutedOpenedFansOutToProject drives the Linear notify lane's +// assembled router over the REAL store adapters bound to (LINEAR, "linear.app"): +// an OPENED Issue in project-alpha fans out to ONLY the alpha container +// subscriber, never the beta one (W2 / DL-267: a Linear container is a PROJECT, +// so an OPENED matches only its project's subscribers). The dispatcher + checks +// roller are fakes; the store adapters + router are the real assembled seams. The +// shared FETCH cursor advances (fetch-side truth); delivered_revision stays +// unadvanced (W3). Runs in CI; compiles locally. +func TestLinearNotifyRoutedOpenedFansOutToProject(t *testing.T) { + st := forgeTestStore(t) + ctx := context.Background() // test root + const ( + repo = "RIG" + host = "linear.app" + number = uint64(42) + alpha = "proj-alpha" + beta = "proj-beta" + url = "https://linear.app/rig/issue/RIG-42" + ) + alphaAgent, alphaSub := seedLinearContainerSub(t, st, "lin-alpha", repo, alpha) + _, betaSub := seedLinearContainerSub(t, st, "lin-beta", repo, beta) + + 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{}, forgeRef, nil) + + if err := router.Route(ctx, linearOpenedEvent(repo, number, alpha, url)); err != nil { + t.Fatalf("Route: %v", err) + } + + // Exactly the alpha subscriber is notified — never beta. + if len(disp.sent) != 1 { + t.Fatalf("dispatched notifications = %d, want 1 (alpha only)", len(disp.sent)) + } + if disp.accounts[0] != string(alphaAgent) { + t.Errorf("notified account = %q, want the alpha agent %q", disp.accounts[0], alphaAgent) + } + n := disp.sent[0] + if n.GetSubscriptionId() != alphaSub { + t.Errorf("notification subscription_id = %q, want alpha %q (not beta %q)", n.GetSubscriptionId(), alphaSub, betaSub) + } + if n.GetForge().GetProvider() != compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR { + t.Errorf("notification provider = %v, want LINEAR", n.GetForge().GetProvider()) + } + if n.GetChange() != compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_OPENED { + t.Errorf("notification change = %v, want OPENED", n.GetChange()) + } + + // The shared FETCH cursor advanced at the container coordinate (number=0 for + // a container OPENED is NOT how the router keys it — the event carries the + // artifact number, so the cursor lands at (repo, ISSUE, number)). + cur, err := st.LoadForgeArtifactCursor(ctx, store.ForgeProviderLinear, host, repo, store.ForgeArtifactKindIssue, number) + if err != nil { + t.Fatalf("LoadForgeArtifactCursor: %v", err) + } + if cur == nil || cur.Revision == "" { + t.Fatal("fetch cursor did not advance after the Linear OPENED route") + } +} diff --git a/go/server/linear_webhook_test.go b/go/server/linear_webhook_test.go index 169c9941..33ae03d1 100644 --- a/go/server/linear_webhook_test.go +++ b/go/server/linear_webhook_test.go @@ -25,6 +25,7 @@ import ( compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/linearagent" + "github.com/RigelBuild/compass/go/internal/secrets" ) // recordingSessionSink records enqueued session events; enqErr (when set) is @@ -352,3 +353,42 @@ func TestLinearWebhookHandler_VerifiedUnparseable(t *testing.T) { t.Errorf("enqueued data=%d session=%d, want 0/0 (unparseable drop)", data.count(), session.count()) } } + +// TestBuildLinearWebhookWiring_DeliversToInjectedSink proves the RIG-2732 T7 +// wiring seam this slice adds: buildLinearWebhookWiring threads the notify lane's +// dataSink all the way to the mounted handler, so a verified Issue event routes +// to a NON-nil sink instead of the prior injected-nil ack-and-drop. The secret +// is resolved via the real newCachedWebhookSecret path off a declared fakeResolver +// secret, exactly as Serve wires it. +func TestBuildLinearWebhookWiring_DeliversToInjectedSink(t *testing.T) { + ctx := context.Background() // test root + const secretName = "LINEAR_WEBHOOK_SECRET" + secret := []byte("shh") + res := &fakeResolver{resolved: []secrets.ResolvedSecret{{Name: secretName, Value: string(secret)}}} + cfg := ServeConfig{Forge: ForgeConfig{LinearWebhookSecretName: secretName}} + sink := &recordingSink{} + + handler, err := buildLinearWebhookWiring(ctx, cfg, res, sink, nil) + if err != nil { + t.Fatalf("buildLinearWebhookWiring: %v", err) + } + if handler == nil { + t.Fatal("handler == nil with the webhook secret declared, want a mounted handler") + } + // Pin the handler's clock so the fresh timestamp check passes deterministically. + now := time.Unix(1_700_000_000, 0) + lh := handler.(*linearWebhookHandler) + lh.now = func() time.Time { return now } + + body := fmt.Appendf(nil, `{"type":"Issue","action":"create","webhookTimestamp":%d,"data":{"number":7,"team":{"key":"RIG"},"url":"https://linear.app/i/7"}}`, freshTS(now)) + rec := linPost(lh, linSign(secret, body), body) + if rec.Code != http.StatusOK { + t.Fatalf("code = %d, want 200", rec.Code) + } + if sink.count() != 1 { + t.Fatalf("data enqueued = %d, want 1 (verified Issue routes to the injected sink)", sink.count()) + } + if ev := sink.events[0]; ev.Provider != compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR { + t.Errorf("Provider = %v, want LINEAR", ev.Provider) + } +} diff --git a/go/server/serve.go b/go/server/serve.go index 68efac18..060c3cb6 100644 --- a/go/server/serve.go +++ b/go/server/serve.go @@ -558,7 +558,8 @@ func Serve(ctx context.Context, cfg ServeConfig) error { } // Assemble the three compass.v1 doors (shipped Unix socket, optional dev - // loopback, optional authenticated network). On a net-door build error the + // loopback, optional authenticated network) plus the Linear notify lane the + // net door's /webhooks/linear handler feeds. On a net-door build error the // listeners this Serve bound are still ours to close. doors, err := buildDoors(ctx, cfg, svc, commsSvc, secretsSvc, hub, st, admin.ID, resolver, devListener, netListener, netTLS, webhookSink, webhookSecret) @@ -598,9 +599,11 @@ func Serve(ctx context.Context, cfg ServeConfig) error { // each lane's webhook-arm drain and reconciler sweep join the SAME scoped // group so they inherit the doors' lifecycle exactly — cancelled on // SIGINT/SIGTERM via gctx, first-error-wins, drained with everything else. - // Both lanes are nil when the GitHub App is absent (they share the App gate), - // and a nil lane starts nothing. Both Runs return nil on ctx-cancel. - startForgeIngestLanes(gctx, g, lane, notifyLane) + // The board + GitHub notify lanes are nil when the GitHub App is absent (they + // share the App gate); the Linear notify lane is nil when LINEAR_FORGE_TOKEN + // is undeclared (its independent gate). A nil lane starts nothing. Every Run + // returns nil on ctx-cancel. + startForgeIngestLanes(gctx, g, lane, notifyLane, doors.linearNotify) // The comms-bus consumers (RIG-1569): the T3 delivery fan-out consumer and // the T8 presence projection, both tailing the comms bus with their bus-tail // goroutines on the serve group rooted on gctx (cancels at shutdown; each also @@ -636,6 +639,10 @@ type serveDoors struct { uds *http.Server dev *http.Server net *http.Server + // linearNotify is the Linear agent-notification lane (RIG-2732 T7), built + // beside the webhook handler it feeds; nil when LINEAR_FORGE_TOKEN is + // undeclared. Serve starts its arm + reconciler on the serve group. + linearNotify *forgeNotifyLane } // buildDoors assembles the three compass.v1 doors off the already-built service @@ -736,18 +743,37 @@ func buildDoors( devServer = &http.Server{Handler: devCORS().Handler(devMux), Protocols: cleartextHTTP2()} //nolint:gosec // G112: loopback dev-only door (off on the shipped path), so the Slowloris ReadHeaderTimeout does not apply here either } + // The Linear agent-notification lane (RIG-2732 T7): App-INDEPENDENT, gated on + // LINEAR_FORGE_TOKEN. Built here — beside the webhook handler it feeds — so a + // resolve fault fail-fasts door assembly, and its data-change sink threads + // straight into buildLinearWebhookWiring below, replacing the + // injected-and-nil-for-now sink so a verified /webhooks/linear Issue/Comment + // event routes to subscribers instead of ack-and-drop. Nil when the secret is + // undeclared (the handler's data branch then acks-and-drops). The lane is + // returned in serveDoors so Serve can start its arm + reconciler on the serve + // group. + linearNotifyLane, err := buildLinearNotifyLane(ctx, st, hub, resolver, slog.Default()) + if err != nil { + return serveDoors{}, err + } + var linearDataSink ForgeEventSink + if linearNotifyLane != nil { + linearDataSink = linearNotifyLane.sink + } + // The Linear webhook ingress (RIG-2732 T7d / RIG-2717): a shared // POST /webhooks/linear handler (DL-302) built iff the Linear webhook secret // is declared — an App-INDEPENDENT gate (a deployment can run Linear - // notifications without a GitHub App). Its data-change arm's sink is - // injected-and-nil-for-now (DL-302): feeding the GitHub-coordinate fanout - // would mis-route Linear events, so the data branch acks-and-drops until a - // Linear-provider-bound notify lane injects a real sink. Its session arm is - // left unwired (nil sessionSink -> logged-drop) until the RIG-2717 responder - // assembly wires a *linearagent.Dispatcher here. Built here (not gated on the - // net door) so a resolve fault fail-fasts startup regardless of --listen; the - // handler is mounted only on the net door below, when one exists. - linearWebhookHandler, err := buildLinearWebhookWiring(ctx, cfg, resolver, nil, slog.Default()) + // notifications without a GitHub App). Its data-change arm's sink is the + // Linear-provider-bound notify lane's sink (linearDataSink), so a verified + // Issue/Comment event routes to subscribers at the LINEAR/linear.app + // coordinate; nil when the notify lane is off (LINEAR_FORGE_TOKEN undeclared), + // and the handler's data branch then acks-and-drops. The two gates are + // independent: the webhook secret gates the handler; LINEAR_FORGE_TOKEN gates + // the sink. Its session arm is left unwired (nil sessionSink -> logged-drop) + // until the RIG-2717 responder assembly wires a *linearagent.Dispatcher here. + // The handler is mounted only on the net door below, when one exists. + linearWebhookHandler, err := buildLinearWebhookWiring(ctx, cfg, resolver, linearDataSink, slog.Default()) if err != nil { return serveDoors{}, err } @@ -769,7 +795,7 @@ func buildDoors( netServer = s } - return serveDoors{uds: udsServer, dev: devServer, net: netServer}, nil + return serveDoors{uds: udsServer, dev: devServer, net: netServer, linearNotify: linearNotifyLane}, nil } // drainSet is the shutdown-side view of what Serve built: the two buses whose @@ -1249,13 +1275,19 @@ func (d *forgeNotifyDispatcher) Notify(ctx context.Context, account string, n *c }) } -// forgeNotifyChecksRoller adapts *forge.GitHub to ingest.ChecksRoller: the -// combined checks roll-up a CHECKS event needs. GitHub's ChecksConditional has -// the exact RollUp signature (notify_router.go:111-113), so this is a one-method -// forwarding adapter rather than a method value, keeping the seam an explicit -// named type. +// forgeNotifyChecksRoller adapts a forge.NotifyReader to ingest.ChecksRoller: the +// combined checks roll-up a CHECKS event needs. NotifyReader's ChecksConditional +// has the exact RollUp signature (notify_router.go:111-113), so this is a +// one-method forwarding adapter rather than a method value, keeping the seam an +// explicit named type. It holds the NotifyReader interface (not a concrete +// *forge.GitHub) so ONE adapter serves both the GitHub lane (client is a +// *forge.GitHub) and the Linear lane (client is a *forge.Linear) — both satisfy +// forge.NotifyReader (notify_reader.go:77,367). Linear's ChecksConditional +// returns ErrUnsupported, but the router invokes RollUp only on a CHECKS-kind +// event, which Linear (issues-only) never produces — so it is correct and +// never-called (fail-closed if one somehow arrived). type forgeNotifyChecksRoller struct { - client *forge.GitHub + client forge.NotifyReader } // RollUp resolves the combined checks roll-up for the head SHA, forwarding to the @@ -1326,6 +1358,58 @@ func buildForgeNotifyLane( return &forgeNotifyLane{arm: arm, reconciler: reconciler, sink: arm}, nil } +// buildLinearNotifyLane assembles the Linear agent-notification lane (RIG-2732 +// T7), the Linear sibling of buildForgeNotifyLane. It gates App-INDEPENDENTLY on +// the LINEAR_FORGE_TOKEN read credential (forgeSecretDeclared) — the same secret +// the write path's Linear coordinate gates on (serve.go:1513-1517) — matching the +// house pattern that every forge lane gates as a unit on its own credential. An +// undeclared/absent secret returns (nil, nil), the off-state the caller reads as +// "mount no Linear notify sink"; a resolve FAULT returns the error (fail-fast, +// like forgeSecretDeclared elsewhere). +// +// Linear is issues-only and check-less (DL-051): its event alphabet is +// Issue/Comment, so no CHECKS/REVIEW arms ever fire. The checks roller is still +// wired (the router's ChecksRoller seam is non-optional) over the same +// forgeNotifyChecksRoller adapter the GitHub lane uses — the Linear client's +// ChecksConditional returns ErrUnsupported, but the router invokes RollUp only on +// a CHECKS event Linear never produces (correct and never-called). +func buildLinearNotifyLane( + ctx context.Context, + st *store.Store, + hub *runnerhub.Hub, + resolver secrets.Resolver, + log *slog.Logger, +) (*forgeNotifyLane, error) { + declared, err := forgeSecretDeclared(ctx, resolver, defaultForgeLinearSecretName) + if err != nil { + return nil, err + } + if !declared { + return nil, nil //nolint:nilnil // an undeclared LINEAR_FORGE_TOKEN is a valid off-state: a nil lane is the signal (the caller guards `if lane != nil`), not an ambiguous nil-nil — a sentinel error would force the caller to distinguish it from a real fault. + } + const ( + provider = store.ForgeProviderLinear + host = "linear.app" + ) + client := forge.NewLinear(forge.LinearConfig{Token: newForgeTokenSource(resolver, defaultForgeLinearSecretName), Log: log}) + + notifyStore := &forgeNotifyStore{st: st, provider: provider, host: host} + dispatcher := &forgeNotifyDispatcher{hub: hub} + checks := &forgeNotifyChecksRoller{client: client} + forgeRef := &compassv1.ForgeRef{ + Provider: compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, + Host: host, + } + router := ingest.NewNotifyRouter(notifyStore, dispatcher, checks, forgeRef, log) + arm := ingest.NewNotifyWebhookArm(router, ingest.NotifyArmConfig{Log: log}) + reconciler := ingest.NewNotifyReconciler(client, notifyStore, router, + compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR, host, ingest.ReconcileConfig{ + Backstop: 0, // no App config carries a Linear backstop; 0 -> ingest's defaultBackstop. + Log: log, + }) + return &forgeNotifyLane{arm: arm, reconciler: reconciler, sink: arm}, nil +} + // newDeclaredSecretResolver returns a func that resolves the declared server_only // secret NAME to its raw value bytes on each call — the lazy PEM/webhook-secret // seam the App token source and the webhook ingress consume. A resolve fault or diff --git a/go/server/serve_forge_test.go b/go/server/serve_forge_test.go index 81b0f065..2c2ad77e 100644 --- a/go/server/serve_forge_test.go +++ b/go/server/serve_forge_test.go @@ -149,6 +149,48 @@ func TestForgeReviewerSecretDefaultingAndWritesEnabled(t *testing.T) { }) } +// TestBuildLinearNotifyLaneGate pins the RIG-2732 T7 Linear notify lane's +// App-INDEPENDENT gate: buildLinearNotifyLane runs iff LINEAR_FORGE_TOKEN is +// declared (the read credential the reconciler needs), NOT the GitHub App gate. +// The gate short-circuits before any store/hub touch, so a nil store + nil hub +// suffice; the declared path binds the Linear coordinate +// (store.ForgeProviderLinear / "linear.app"). A resolve fault fails fast. +func TestBuildLinearNotifyLaneGate(t *testing.T) { + ctx := context.Background() // test root + t.Run("undeclared LINEAR_FORGE_TOKEN -> nil lane (off-state)", func(t *testing.T) { + lane, err := buildLinearNotifyLane(ctx, nil, nil, &fakeResolver{}, nil) + if err != nil { + t.Fatalf("buildLinearNotifyLane (undeclared): %v", err) + } + if lane != nil { + t.Fatal("lane != nil with LINEAR_FORGE_TOKEN undeclared, want nil (lane off)") + } + }) + t.Run("declared LINEAR_FORGE_TOKEN -> non-nil lane with a sink", func(t *testing.T) { + res := &fakeResolver{resolved: []secrets.ResolvedSecret{{Name: defaultForgeLinearSecretName, Value: "lin-tok"}}} + lane, err := buildLinearNotifyLane(ctx, nil, nil, res, nil) + if err != nil { + t.Fatalf("buildLinearNotifyLane (declared): %v", err) + } + if lane == nil { + t.Fatal("lane == nil with LINEAR_FORGE_TOKEN declared, want a non-nil lane") + } + if lane.arm == nil || lane.reconciler == nil || lane.sink == nil { + t.Fatalf("assembled lane has a nil member: %+v", lane) + } + }) + t.Run("resolve fault -> error (fail-fast)", func(t *testing.T) { + res := &fakeResolver{err: errors.New("boom")} + lane, err := buildLinearNotifyLane(ctx, nil, nil, res, nil) + if err == nil { + t.Fatal("buildLinearNotifyLane returned nil error on a resolve fault, want fail-fast") + } + if lane != nil { + t.Fatalf("lane = %+v on a resolve fault, want nil", lane) + } + }) +} + func TestForgeTokenSourceCachesUntilTTL(t *testing.T) { res := &fakeResolver{resolved: []secrets.ResolvedSecret{{Name: "GITHUB_FORGE_TOKEN", Value: "tok-1"}}} ts := newForgeTokenSource(res, "GITHUB_FORGE_TOKEN") diff --git a/go/server/sinks.go b/go/server/sinks.go index 3ca85de7..119ad208 100644 --- a/go/server/sinks.go +++ b/go/server/sinks.go @@ -173,20 +173,28 @@ func startCommsBusConsumers(gctx context.Context, g *errgroup.Group, commsBus *e } // startForgeIngestLanes starts the forge webhook-ingestion lanes' background -// goroutines on the serve group: the board lane (RIG-2883) and the agent- -// notification lane (RIG-2732 T7), each contributing its webhook-arm drain and -// its reconciler sweep. Both share the App gate, so each lane is nil-or-set as a -// unit; a nil lane starts nothing. Serve calls this one helper so the four -// Run starts, which share the serve group + gctx, stay one statement at the call -// site (mirroring startCommsBusConsumers). Both Runs return nil on ctx-cancel. -func startForgeIngestLanes(gctx context.Context, g *errgroup.Group, board *boardIngestLane, notify *forgeNotifyLane) { +// goroutines on the serve group: the board lane (RIG-2883) and the two agent- +// notification lanes (RIG-2732 T7) — the GitHub notify lane and the Linear notify +// lane — each contributing its webhook-arm drain and its reconciler sweep. The +// board and GitHub notify lanes share the App gate (nil-or-set together); the +// Linear notify lane gates INDEPENDENTLY on LINEAR_FORGE_TOKEN, so it is nil-or- +// set on its own. Every lane is nil-checked; a nil lane starts nothing. The +// notify lanes are the same *forgeNotifyLane type, taken variadically so a new +// notify lane is one more argument, not a new param. Serve calls this one helper +// so the Run starts, which share the serve group + gctx, stay one statement at +// the call site (mirroring startCommsBusConsumers). Every Run returns nil on +// ctx-cancel. +func startForgeIngestLanes(gctx context.Context, g *errgroup.Group, board *boardIngestLane, notify ...*forgeNotifyLane) { if board != nil { g.Go(func() error { return board.arm.Run(gctx) }) g.Go(func() error { return board.reconciler.Run(gctx) }) } - if notify != nil { - g.Go(func() error { return notify.arm.Run(gctx) }) - g.Go(func() error { return notify.reconciler.Run(gctx) }) + for _, lane := range notify { + if lane == nil { + continue + } + g.Go(func() error { return lane.arm.Run(gctx) }) + g.Go(func() error { return lane.reconciler.Run(gctx) }) } }