feat(server): mount the Linear webhook ingress /webhooks/linear (RIG-2732 T7)

go/cmd/compass-server/main.go

@@ -382,6 +382,7 @@ type forgeFlags struct {
 	installationID *string
 	appKeySecret   *string
 	appWebhook     *string
+	linearWebhook  *string
 }
 
 // registerForgeFlags declares the forge flags on the given FlagSet and returns
@@ -414,6 +415,12 @@ func registerForgeFlags(fs *flag.FlagSet) forgeFlags {
 			"Declared server_only secret NAME holding the webhook signing secret the "+
 				"ingress verifies deliveries against. Defaults to "+
 				"$COMPASS_FORGE_APP_WEBHOOK_SECRET."),
+		linearWebhook: fs.String("forge-linear-webhook-secret", "",
+			"Declared server_only secret NAME holding the Linear webhook signing "+
+				"secret the shared POST /webhooks/linear ingress verifies deliveries "+
+				"against. Defaults to $COMPASS_FORGE_LINEAR_WEBHOOK_SECRET. The Linear "+
+				"data-change/session arm runs iff this is declared, independent of the "+
+				"GitHub App gate."),
 	}
 }
 
@@ -428,6 +435,7 @@ func (f forgeFlags) resolve() (server.ForgeConfig, error) {
 		firstNonEmpty(*f.installationID, os.Getenv("COMPASS_FORGE_INSTALLATION_ID")),
 		firstNonEmpty(*f.appKeySecret, os.Getenv("COMPASS_FORGE_APP_KEY_SECRET")),
 		firstNonEmpty(*f.appWebhook, os.Getenv("COMPASS_FORGE_APP_WEBHOOK_SECRET")),
+		firstNonEmpty(*f.linearWebhook, os.Getenv("COMPASS_FORGE_LINEAR_WEBHOOK_SECRET")),
 	)
 }
 
@@ -438,7 +446,7 @@ func (f forgeFlags) resolve() (server.ForgeConfig, error) {
 // so Owner/Name and owner/name collapse to one target. appID/installationID are
 // parsed as int64 when set (garbage is a startup error); empty leaves them zero.
 // Empty host/secret/App-secret NAMEs default server-side.
-func resolveForge(repos, secret, host, appID, installationID, appKeySecret, appWebhook string) (server.ForgeConfig, error) {
+func resolveForge(repos, secret, host, appID, installationID, appKeySecret, appWebhook, linearWebhook string) (server.ForgeConfig, error) {
 	seed, err := parseForgeRepos(repos)
 	if err != nil {
 		return server.ForgeConfig{}, err
@@ -452,9 +460,10 @@ func resolveForge(repos, secret, host, appID, installationID, appKeySecret, appW
 		return server.ForgeConfig{}, err
 	}
 	return server.ForgeConfig{
-		Host:       host,
-		SeedRepos:  seed,
-		SecretName: secret,
+		Host:                    host,
+		SeedRepos:               seed,
+		SecretName:              secret,
+		LinearWebhookSecretName: linearWebhook,
 		App: server.ForgeAppConfig{
 			AppID:                id,
 			InstallationID:       instID,

go/cmd/compass-server/main_forge_test.go

@@ -19,7 +19,7 @@ import (
 
 func TestResolveForgeMapping(t *testing.T) {
 	t.Run("disabled default: no repos, no App", func(t *testing.T) {
-		fc, err := resolveForge("", "", "", "", "", "", "")
+		fc, err := resolveForge("", "", "", "", "", "", "", "")
 		if err != nil {
 			t.Fatalf("resolveForge: %v", err)
 		}
@@ -40,7 +40,7 @@ func TestResolveForgeMapping(t *testing.T) {
 
 	t.Run("full flag mapping", func(t *testing.T) {
 		fc, err := resolveForge("owner/repo, foo/bar", "MY_TOKEN", "ghe.example.com",
-			"12345", "678", "APP_KEY", "WEBHOOK_SECRET")
+			"12345", "678", "APP_KEY", "WEBHOOK_SECRET", "LINEAR_WEBHOOK_SECRET")
 		if err != nil {
 			t.Fatalf("resolveForge: %v", err)
 		}
@@ -57,6 +57,9 @@ func TestResolveForgeMapping(t *testing.T) {
 			t.Fatalf("App secrets = %q/%q, want APP_KEY/WEBHOOK_SECRET",
 				fc.App.AppPrivateKeySecret, fc.App.AppWebhookSecretName)
 		}
+		if fc.LinearWebhookSecretName != "LINEAR_WEBHOOK_SECRET" {
+			t.Fatalf("LinearWebhookSecretName = %q, want LINEAR_WEBHOOK_SECRET", fc.LinearWebhookSecretName)
+		}
 		want := []string{"owner/repo", "foo/bar"}
 		if len(fc.SeedRepos) != len(want) {
 			t.Fatalf("SeedRepos = %v, want %v", fc.SeedRepos, want)
@@ -69,7 +72,7 @@ func TestResolveForgeMapping(t *testing.T) {
 	})
 
 	t.Run("case normalization: Owner/Name lowercases to one target", func(t *testing.T) {
-		fc, err := resolveForge("Owner/Name", "", "", "", "", "", "")
+		fc, err := resolveForge("Owner/Name", "", "", "", "", "", "", "")
 		if err != nil {
 			t.Fatalf("resolveForge: %v", err)
 		}
@@ -91,7 +94,7 @@ func TestResolveForgeRejectsGarbage(t *testing.T) {
 	}
 	for _, tc := range garbage {
 		t.Run(tc.name, func(t *testing.T) {
-			_, err := resolveForge(tc.repos, "", "", "", "", "", "")
+			_, err := resolveForge(tc.repos, "", "", "", "", "", "", "")
 			if err == nil {
 				t.Fatalf("resolveForge(%q) = nil error, want a startup error", tc.repos)
 			}
@@ -110,7 +113,7 @@ func TestResolveForgeRejectsBadAppID(t *testing.T) {
 		{"non-numeric installation id", "123", "nope", "--forge-installation-id"},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			_, err := resolveForge("owner/repo", "", "", tc.appID, tc.installID, "", "")
+			_, err := resolveForge("owner/repo", "", "", tc.appID, tc.installID, "", "", "")
 			if err == nil {
 				t.Fatalf("resolveForge = nil error, want a startup error")
 			}

go/server/cors_pgtest_test.go

@@ -73,7 +73,7 @@ func buildDoorHandler(t *testing.T, corsOrigin string) http.Handler {
 	if err != nil {
 		t.Fatalf("otelconnect.NewInterceptor: %v", err)
 	}
-	srv, err := buildNetworkServer(ctx, cfg, svc, commsSvc, secretsSvc, nil, st, admin, nil, nil, otelIC, nil, nil)
+	srv, err := buildNetworkServer(ctx, cfg, svc, commsSvc, secretsSvc, nil, st, admin, nil, nil, otelIC, nil, nil, nil)
 	if err != nil {
 		t.Fatalf("buildNetworkServer: %v", err)
 	}

go/server/linear_webhook.go

@@ -0,0 +1,239 @@
+//go:build unix
+
+// The Linear webhook ingress: a single shared http.Handler for
+// POST /webhooks/linear (symmetric with /webhooks/github, DL-302) that
+// type-dispatches the top-level envelope. "AgentSessionEvent" routes to the
+// RIG-2717 responder seam (a local SessionEventSink satisfied by
+// *linearagent.Dispatcher); "Issue"/"Comment" data-change events route to an
+// injected ForgeEventSink. It mirrors github_webhook.go's fail-closed order
+// (secret -> HMAC -> parse) and its ack-fast discipline; T7d mounts it and
+// supplies the sinks. The data sink is injected-and-nil-for-now (DL-302): a
+// Linear-provider-bound notify lane wires the real sink later, so a nil sink
+// acks-and-drops rather than mis-routing into the GitHub-coordinate fanout.
+//
+// Dedup is NOT done here: RIG-2717's design (design.md:167-170) rides the comms
+// rail's own idempotency for sessions (the dispatcher's client_request_id) and
+// the notify router's idempotency for data, so — unlike the GitHub handler's
+// X-GitHub-Delivery LRU — this mount adds no delivery-id LRU.
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"log/slog"
+	"net/http"
+	"time"
+
+	"github.com/RigelBuild/compass/go/internal/linearagent"
+)
+
+const (
+	// linearWebhookPath is the shared Linear ingress path this handler mounts at
+	// — /webhooks/linear, for symmetry with /webhooks/github (DL-302, Matt's
+	// 2026-08-30 amendment).
+	linearWebhookPath = "/webhooks/linear"
+
+	// linearSignatureHeader carries the HMAC-SHA256 hex of the raw body under
+	// the webhook secret (RIG-2717 design §134).
+	linearSignatureHeader = "Linear-Signature"
+
+	// linearWebhookSkew bounds how stale a webhookTimestamp may be before the
+	// delivery is acked-and-dropped (RIG-2717 §145): the timestamp is inside the
+	// signed body, so a replay is already signature-valid; a stale timestamp is
+	// dropped with a 200, never a 400.
+	linearWebhookSkew = 60 * time.Second
+
+	// Envelope type discriminants (RIG-2717 §134, data arm design.md:660-666).
+	linearTypeSession = "AgentSessionEvent"
+	linearTypeIssue   = "Issue"
+	linearTypeComment = "Comment"
+)
+
+// SessionEventSink receives a verified Linear session event for asynchronous
+// dispatch. *linearagent.Dispatcher satisfies it via Enqueue (dispatcher.go:145).
+// Defined locally so the server can wire a real dispatcher OR pass nil when the
+// RIG-2717 responder is not assembled — the handler logs-and-drops in that case.
+type SessionEventSink interface {
+	// Enqueue offers a verified session event to the dispatcher without
+	// blocking; a full queue returns linearagent.ErrQueueFull, which the handler
+	// maps to a 500 so Linear retries the delivery.
+	Enqueue(ev *linearagent.SessionEvent) error
+}
+
+// linearWebhookHandler serves POST /webhooks/linear.
+type linearWebhookHandler struct {
+	secret      func(ctx context.Context) ([]byte, error)
+	dataSink    ForgeEventSink
+	sessionSink SessionEventSink
+	maxBody     int64
+	skew        time.Duration
+	now         func() time.Time
+	log         *slog.Logger
+}
+
+// NewLinearWebhookHandler returns the POST /webhooks/linear handler and the
+// path it mounts at. secret lazily resolves the Linear webhook secret (TTL-cached by the
+// caller); dataSink receives every accepted data-change event; sessionSink
+// receives session events (nil when the responder is unassembled — session
+// events are then logged-and-dropped). Mirrors NewGitHubWebhookHandler's
+// nil-log default and field init.
+func NewLinearWebhookHandler(
+	secret func(ctx context.Context) ([]byte, error),
+	dataSink ForgeEventSink,
+	sessionSink SessionEventSink,
+	log *slog.Logger,
+) (string, http.Handler) {
+	if log == nil {
+		log = slog.Default()
+	}
+	return linearWebhookPath, &linearWebhookHandler{
+		secret:      secret,
+		dataSink:    dataSink,
+		sessionSink: sessionSink,
+		maxBody:     githubWebhookMaxBody,
+		skew:        linearWebhookSkew,
+		now:         time.Now,
+		log:         log,
+	}
+}
+
+// linearEnvelope is the minimal peek this handler routes on: the top-level type
+// (Issue/Comment/AgentSessionEvent) and the webhookTimestamp both envelopes
+// carry. The full body is re-parsed by the branch parser after routing.
+type linearEnvelope struct {
+	Type             string `json:"type"`
+	WebhookTimestamp int64  `json:"webhookTimestamp"`
+}
+
+func (h *linearWebhookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+	ctx := r.Context()
+
+	// Bound the body read before buffering it (oversized-body rejection). A body
+	// over the cap trips MaxBytesReader mid-read, so a hostile Content-Length
+	// never forces the full allocation.
+	r.Body = http.MaxBytesReader(w, r.Body, h.maxBody)
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		var mbe *http.MaxBytesError
+		if errors.As(err, &mbe) {
+			http.Error(w, "payload too large", http.StatusRequestEntityTooLarge)
+			return
+		}
+		http.Error(w, "bad request", http.StatusBadRequest)
+		return
+	}
+
+	// Fail-closed order: secret -> HMAC -> parse. An attacker who reaches this
+	// unauthenticated endpoint must clear the signature before any body parsing.
+	secret, err := h.secret(ctx)
+	if err != nil {
+		h.log.ErrorContext(ctx, "linear webhook secret unavailable", "err", err)
+		http.Error(w, "webhook unavailable", http.StatusServiceUnavailable)
+		return
+	}
+	if !linearagent.VerifySignature(secret, body, r.Header.Get(linearSignatureHeader)) {
+		// Fail-closed: an unverifiable delivery is a 400, never processed.
+		http.Error(w, "invalid signature", http.StatusBadRequest)
+		return
+	}
+
+	// Peek just enough to route (type) and freshness-check (timestamp); the
+	// branch parser re-parses the full body.
+	var env linearEnvelope
+	if err := json.Unmarshal(body, &env); err != nil {
+		h.log.WarnContext(ctx, "linear webhook envelope parse error", "err", err)
+		w.WriteHeader(http.StatusOK) // verified-but-unparseable: ack-and-drop.
+		return
+	}
+
+	// Stale timestamp -> 200-with-drop, NOT 400 (RIG-2717 §145-155): the
+	// timestamp is inside the signed body, so a replay is signature-valid; a 400
+	// would burn Linear's retry ladder on a legitimate late retry.
+	if !linearagent.CheckTimestamp(env.WebhookTimestamp, h.now(), h.skew) {
+		h.log.WarnContext(ctx, "linear webhook stale timestamp, dropped",
+			"type", env.Type, "webhook_timestamp", env.WebhookTimestamp)
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+
+	switch env.Type {
+	case linearTypeSession:
+		h.serveSession(ctx, w, body)
+	case linearTypeIssue, linearTypeComment:
+		h.serveData(ctx, w, body)
+	default:
+		// An envelope type this mount does not handle (e.g. "Reaction"): ignored.
+		w.WriteHeader(http.StatusOK)
+	}
+}
+
+// serveSession routes a verified AgentSessionEvent to the responder seam. Unlike
+// the data branch, Enqueue is itself non-blocking (a bounded try-send) and its
+// return maps to the status code, so the status is written AFTER Enqueue: a full
+// queue is a 500 (Linear retries), everything else a 200.
+func (h *linearWebhookHandler) serveSession(ctx context.Context, w http.ResponseWriter, body []byte) {
+	ev, err := linearagent.ParseSessionEvent(body)
+	if err != nil {
+		h.log.WarnContext(ctx, "linear session event parse error", "err", err)
+		w.WriteHeader(http.StatusOK) // verified-but-malformed: ack-and-drop.
+		return
+	}
+	if h.sessionSink == nil {
+		// The RIG-2717 responder assembly wires a real dispatcher here; until
+		// then session events are acked-and-dropped so Linear does not retry.
+		h.log.WarnContext(ctx, "linear session responder not wired, dropping event",
+			"action", ev.Action, "session", ev.AgentSession.ID)
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+	if err := h.sessionSink.Enqueue(ev); err != nil {
+		if errors.Is(err, linearagent.ErrQueueFull) {
+			// A full queue is a 500 so Linear retries rather than the event
+			// being silently dropped (dispatcher.go:30-33).
+			http.Error(w, "queue full", http.StatusInternalServerError)
+			return
+		}
+		h.log.ErrorContext(ctx, "linear session enqueue failed", "err", err)
+		http.Error(w, "enqueue failed", http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+}
+
+// serveData routes a verified Issue/Comment data-change event to the shared
+// fanout sink. It acks 200 (and flushes) BEFORE the non-blocking Enqueue so the
+// ack is never on the sink's latency path (mirrors github_webhook.go:181-192).
+func (h *linearWebhookHandler) serveData(ctx context.Context, w http.ResponseWriter, body []byte) {
+	ev, ok, perr := linearagent.ParseLinearDataEvent(body)
+	if perr != nil {
+		h.log.WarnContext(ctx, "linear data event parse error", "err", perr)
+		w.WriteHeader(http.StatusOK) // verified-but-malformed: ack-and-drop.
+		return
+	}
+	// Ack fast, enqueue after: the sink must never be on the ack's latency path.
+	w.WriteHeader(http.StatusOK)
+	if !ok {
+		return // ignored action (e.g. Issue remove): counted-and-dropped.
+	}
+	// Flush the ack onto the wire before handing off: WriteHeader only records
+	// the status, so a blocking Enqueue would otherwise delay the client-visible
+	// 200 until ServeHTTP returns.
+	if f, ok := w.(http.Flusher); ok {
+		f.Flush()
+	}
+	if h.dataSink == nil {
+		// dataSink is injected-and-nil-for-now by driver decision (DL-302): the
+		// GitHub-bound fanout sink would mis-route Linear events (Linear subs
+		// looked up under a GitHub coordinate), so the data branch acks-and-drops
+		// until a Linear-provider-bound notify lane injects a real sink here.
+		h.log.WarnContext(ctx, "linear data-change routing pending the Linear notify lane, dropping event")
+		return
+	}
+	h.dataSink.Enqueue(ctx, ev)
+}