feat(notifications): Slice 3 — governance action queue (RBAC-scoped) (#686)

Turn the bell into an action queue: route governance + remediation events to
the users who can act on them, not the whole fleet.

New primitives:
- auth.RolesWithPermission(p) — resolve a permission to the built-in roles that
  grant it (pure helper over BuiltInRoles).
- notifyfeed.Store.RecordForRoles — fan one row per active user holding any of
  the given roles (EXISTS, so a multi-role user gets one row), same
  upsert/collapse as RecordFanout.

Producer: notifyfeed.GovernanceProjector
- ExceptionRequested -> users who can approve (roles granting exception:approve:
  auditor, security_admin, admin), kind exception_pending, high, deep-link
  /settings/policies, grouped per exception.
- ExceptionDecided -> the requester only, exception_approved/_rejected, medium.
- RemediationFailed -> users who can act (roles granting remediation:execute:
  ops_lead, security_admin, admin), kind remediation_failed, high, deep-link
  /hosts/{id}, grouped per host+rule. Fires on a TERMINAL failure (execute that
  failed, or a rollback that did not restore) — NOT a successful user-initiated
  rollback (that is intended, not an alarm).

Producer hooks (best-effort, never fail the transition/job; nil-safe):
- exception.Service.WithNotifier + calls in Request/Approve/Reject.
- worker.RemediationWorker GovernanceNotifier + calls on the two failure sites.
- wired in cmd/openwatch/main.go (serve) and worker.go (dedicated worker).

Both producers hold interfaces in their own packages, so neither imports
notifyfeed. Spec system-notifications v1.5.0: C-08 + AC-15/16/17. Tests:
auth.RolesWithPermission unit test; DB-backed governance fan-out tests
(approver-scoped, requester-only, operator-scoped). No frontend change — rows
render in the existing bell drawer.

Author: remyluslosius

CHANGELOG.md

@@ -16,6 +16,12 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   regressed (1 critical)"), severity-ranked and deep-linked to the host. A
   host's first scan (all-new baseline) does not flood the bell, and repeat
   failures collapse onto a single re-surfacing entry. (notifications Slice 2)
+- The notification bell is now an action queue for governance: an exception
+  awaiting approval reaches the users who can approve it (auditors and
+  security/admin roles), an approve or reject reaches the requester, and a
+  remediation that fails (or a rollback that does not restore) reaches the
+  operators who can re-run it. Each item is severity-ranked and deep-links to
+  where you act on it. (notifications Slice 3)
 
 ---
 

cmd/openwatch/main.go

@@ -588,9 +588,13 @@ func cmdServe(cfg *config.Config, _ []string, stdout, stderr *os.File) int {
 	// a fresh boot has today's row. Spec system-posture-snapshots.
 	posture.Run(ctx, pool, 0)
 
+	// Governance action-queue notifications (Slice 3): exceptions pending /
+	// decided, remediation failures. Reuses the Slice-1 feed store.
+	govProjector := notifyfeed.NewGovernanceProjector(notifFeedStore)
+
 	// Compliance exception governance + its hourly expiry sweep.
 	// Spec api-compliance-exceptions.
-	exceptionSvc := exception.NewService(pool, audit.Emit)
+	exceptionSvc := exception.NewService(pool, audit.Emit).WithNotifier(govProjector)
 	exceptionSvc.Run(ctx, 0)
 
 	// Remediation governance: request/approve/reject + projected lift (free
@@ -628,13 +632,14 @@ func cmdServe(cfg *config.Config, _ []string, stdout, stderr *os.File) int {
 		remExecutor = remExecutor.WithRemediateFunc(remFn, rbFn)
 	}
 	remediationWorker := worker.NewRemediationWorker(worker.RemediationConfig{
-		Pool:     pool,
-		Executor: remExecutor,
-		Service:  remediationSvc,
-		Writer:   remTxWriter,
-		QueueKey: scanQueueKey,
-		Bus:      bus,
-		Emit:     audit.Emit,
+		Pool:       pool,
+		Executor:   remExecutor,
+		Service:    remediationSvc,
+		Writer:     remTxWriter,
+		QueueKey:   scanQueueKey,
+		Bus:        bus,
+		Emit:       audit.Emit,
+		Governance: govProjector,
 	})
 
 	scanWorker := worker.NewScanWorker(worker.Config{

cmd/openwatch/worker.go

@@ -230,13 +230,18 @@ func cmdWorker(cfg *config.Config, args []string, stdout, stderr *os.File) int {
 	} else {
 		remExecutor = remExecutor.WithRemediateFunc(remFn, rbFn)
 	}
+	// One in-app feed store, shared by the remediation-failure governance
+	// notifier and the scan worker's regression projector.
+	notifFeedStore := notifyfeed.NewStore(pool)
+
 	remediationWorker := worker.NewRemediationWorker(worker.RemediationConfig{
-		Pool:     pool,
-		Executor: remExecutor,
-		Service:  remediation.NewService(pool, audit.Emit),
-		Writer:   writer,
-		QueueKey: queueKey,
-		Emit:     audit.Emit,
+		Pool:       pool,
+		Executor:   remExecutor,
+		Service:    remediation.NewService(pool, audit.Emit),
+		Writer:     writer,
+		QueueKey:   queueKey,
+		Emit:       audit.Emit,
+		Governance: notifyfeed.NewGovernanceProjector(notifFeedStore),
 		// Bus nil: the dedicated worker has no SSE subscribers (cross-process
 		// delivery is a known non-goal, same as scan.completed).
 	})
@@ -267,7 +272,7 @@ func cmdWorker(cfg *config.Config, args []string, stdout, stderr *os.File) int {
 		Emit:                 audit.Emit,
 		Sched:                sched,
 		RemediationProcessor: remediationWorker,
-		Regressions:          notifyfeed.NewProjector(notifyfeed.NewStore(pool)),
+		Regressions:          notifyfeed.NewProjector(notifFeedStore),
 	})
 
 	ctx, stop := signal.NotifyContext(bootCtx, syscall.SIGINT, syscall.SIGTERM)

internal/auth/grant.go

@@ -22,3 +22,22 @@ func RoleGrantsWithin(caller Identity, requested RoleID) bool {
 	}
 	return true
 }
+
+// RolesWithPermission returns the built-in role IDs whose definition grants p,
+// in declaration order. It is the recipient-resolution primitive for
+// permission-scoped fan-out (e.g. "every role that can approve an exception"):
+// callers translate a permission into the set of roles, then query user_roles
+// for the holders. An unknown/empty permission yields no roles.
+func RolesWithPermission(p Permission) []RoleID {
+	var out []RoleID
+	for _, id := range BuiltInRoleIDs() {
+		def := BuiltInRoles[id]
+		for _, granted := range def.Permissions {
+			if granted == p {
+				out = append(out, id)
+				break
+			}
+		}
+	}
+	return out
+}

internal/auth/grant_test.go

@@ -0,0 +1,58 @@
+package auth
+
+import (
+	"sort"
+	"testing"
+)
+
+func roleSet(ids []RoleID) map[RoleID]bool {
+	m := make(map[RoleID]bool, len(ids))
+	for _, id := range ids {
+		m[id] = true
+	}
+	return m
+}
+
+func TestRolesWithPermission(t *testing.T) {
+	cases := []struct {
+		perm Permission
+		want []RoleID
+	}{
+		// exception:approve is held by the auditor (review role) and the two
+		// security roles — but NOT ops_lead (who can only request).
+		{ExceptionApprove, []RoleID{RoleAuditor, RoleSecurityAdmin, RoleAdmin}},
+		// remediation:execute is held by the operator tier and up.
+		{RemediationExecute, []RoleID{RoleOpsLead, RoleSecurityAdmin, RoleAdmin}},
+		// host:read is universal across the five built-in roles.
+		{HostRead, []RoleID{RoleViewer, RoleAuditor, RoleOpsLead, RoleSecurityAdmin, RoleAdmin}},
+		// admin-only verb.
+		{SystemConfigWrite, []RoleID{RoleAdmin}},
+	}
+	for _, c := range cases {
+		got := roleSet(RolesWithPermission(c.perm))
+		want := roleSet(c.want)
+		if len(got) != len(want) {
+			t.Errorf("RolesWithPermission(%s): got %v, want %v", c.perm, sortedRoles(got), c.want)
+			continue
+		}
+		for r := range want {
+			if !got[r] {
+				t.Errorf("RolesWithPermission(%s): missing %s (got %v)", c.perm, r, sortedRoles(got))
+			}
+		}
+	}
+
+	// An unknown permission yields no roles.
+	if got := RolesWithPermission(Permission("does:not_exist")); len(got) != 0 {
+		t.Errorf("unknown permission must yield no roles, got %v", got)
+	}
+}
+
+func sortedRoles(m map[RoleID]bool) []RoleID {
+	out := make([]RoleID, 0, len(m))
+	for r := range m {
+		out = append(out, r)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
+	return out
+}

internal/exception/service.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log/slog"
 	"strings"
 	"time"
 
@@ -19,17 +20,35 @@ import (
 // pass a fake.
 type EmitFunc func(ctx context.Context, code audit.Code, ev audit.Event)
 
+// Notifier receives exception lifecycle signals for the in-app notification
+// feed. notifyfeed.GovernanceProjector implements it; the service holds the
+// interface so it does not import the feed package. Methods are best-effort:
+// the service logs (never fails the transition on) a notifier error. nil
+// disables in-app exception notifications.
+type Notifier interface {
+	ExceptionRequested(ctx context.Context, exceptionID, hostID uuid.UUID, ruleID string) error
+	ExceptionDecided(ctx context.Context, exceptionID, requestedBy uuid.UUID, ruleID string, approved bool) error
+}
+
 // Service is the exception governance service.
 type Service struct {
-	pool *pgxpool.Pool
-	emit EmitFunc
+	pool     *pgxpool.Pool
+	emit     EmitFunc
+	notifier Notifier
 }
 
 // NewService wires the service. emit is audit.Emit in production.
 func NewService(pool *pgxpool.Pool, emit EmitFunc) *Service {
 	return &Service{pool: pool, emit: emit}
 }
 
+// WithNotifier attaches the in-app notification projector and returns the
+// service for chaining at boot. nil leaves in-app notifications disabled.
+func (s *Service) WithNotifier(n Notifier) *Service {
+	s.notifier = n
+	return s
+}
+
 const selectCols = `id, host_id, rule_id, reason, status, requested_by,
 	reviewed_by, COALESCE(review_note, ''), expires_at, requested_at, reviewed_at`
 
@@ -91,23 +110,32 @@ func (s *Service) Request(ctx context.Context, hostID uuid.UUID, ruleID, reason
 	}
 
 	s.emitEvent(ctx, audit.ComplianceExceptionRequested, e, requestedBy, "requested")
+	s.notifyRequested(ctx, e)
 	return e, nil
 }
 
 // Approve transitions a 'requested' exception to 'approved'. The
 // reviewer must differ from the requester (separation of duties).
 // Emits compliance.exception.approved.
 func (s *Service) Approve(ctx context.Context, id, reviewedBy uuid.UUID, note string) (Exception, error) {
-	return s.review(ctx, id, reviewedBy, note, StatusRequested, StatusApproved,
+	e, err := s.review(ctx, id, reviewedBy, note, StatusRequested, StatusApproved,
 		audit.ComplianceExceptionApproved, true)
+	if err == nil {
+		s.notifyDecided(ctx, e, true)
+	}
+	return e, err
 }
 
 // Reject transitions a 'requested' exception to 'rejected'. Like
 // Approve, the reviewer must differ from the requester. Emits
 // compliance.exception.rejected.
 func (s *Service) Reject(ctx context.Context, id, reviewedBy uuid.UUID, note string) (Exception, error) {
-	return s.review(ctx, id, reviewedBy, note, StatusRequested, StatusRejected,
+	e, err := s.review(ctx, id, reviewedBy, note, StatusRequested, StatusRejected,
 		audit.ComplianceExceptionRejected, true)
+	if err == nil {
+		s.notifyDecided(ctx, e, false)
+	}
+	return e, err
 }
 
 // Revoke transitions an 'approved' exception to 'revoked' before its
@@ -311,6 +339,31 @@ func (s *Service) emitEvent(ctx context.Context, code audit.Code, e Exception, a
 	})
 }
 
+// notifyRequested fans an "exception pending approval" notification to
+// approvers. Best-effort: a notifier error is logged, never propagated (the
+// exception was already persisted). No-op when no notifier is wired.
+func (s *Service) notifyRequested(ctx context.Context, e Exception) {
+	if s.notifier == nil {
+		return
+	}
+	if err := s.notifier.ExceptionRequested(ctx, e.ID, e.HostID, e.RuleID); err != nil {
+		slog.WarnContext(ctx, "exception requested notification failed",
+			slog.String("exception_id", e.ID.String()), slog.String("error", err.Error()))
+	}
+}
+
+// notifyDecided notifies the requester that their exception was approved or
+// rejected. Best-effort, like notifyRequested.
+func (s *Service) notifyDecided(ctx context.Context, e Exception, approved bool) {
+	if s.notifier == nil {
+		return
+	}
+	if err := s.notifier.ExceptionDecided(ctx, e.ID, e.RequestedBy, e.RuleID, approved); err != nil {
+		slog.WarnContext(ctx, "exception decided notification failed",
+			slog.String("exception_id", e.ID.String()), slog.String("error", err.Error()))
+	}
+}
+
 // isUniqueViolation reports whether err is a Postgres unique-violation
 // (SQLSTATE 23505) - the partial-unique one-open-per-host+rule index.
 // Same errors.As idiom as internal/host (robust to wrapping).

internal/notifyfeed/governance.go

@@ -0,0 +1,120 @@
+package notifyfeed
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/uuid"
+
+	"github.com/Hanalyx/openwatch/internal/auth"
+)
+
+// GovernanceProjector turns governance + remediation lifecycle events into
+// RBAC-scoped in-app notifications — the bell's "action queue" (Slice 3). Unlike
+// the alert Channel and the regression Projector (which fan to every active
+// user), governance items reach only the users who can act on them: an
+// exception pending approval reaches approvers, a decision reaches the
+// requester, a failed remediation reaches the operators who can re-run it.
+//
+// It implements the Notifier interface the exception service and the
+// GovernanceNotifier the remediation worker hold, so neither producer imports
+// this package. Spec: system-notifications (Slice 3).
+type GovernanceProjector struct {
+	store *Store
+}
+
+// NewGovernanceProjector returns a governance projector over the feed store.
+func NewGovernanceProjector(store *Store) *GovernanceProjector {
+	return &GovernanceProjector{store: store}
+}
+
+// exceptionQueueLink is where the exception queue lives in the UI
+// (frontend/src/components/settings/ExceptionQueue.tsx, mounted on the policies
+// settings page). Approvers act on the queue here.
+const exceptionQueueLink = "/settings/policies"
+
+// ExceptionRequested records an "exception pending approval" notification for
+// every user whose role can approve it (exception:approve → security_admin,
+// admin). Grouped per exception so a re-surfaced request collapses onto one
+// row. Best-effort: an error is returned for the caller to log, never to fail
+// the request.
+func (g *GovernanceProjector) ExceptionRequested(ctx context.Context, exceptionID, hostID uuid.UUID, ruleID string) error {
+	host := g.store.hostName(ctx, hostID)
+	h := hostID
+	n := Notification{
+		Kind:     "exception_pending",
+		Severity: "high",
+		Title:    fmt.Sprintf("Exception awaiting approval: %s on %s", ruleID, host),
+		Body:     "A compliance exception request needs review and approval or rejection.",
+		HostID:   &h,
+		Link:     exceptionQueueLink,
+		GroupKey: "exception_pending:" + exceptionID.String(),
+	}
+	if err := g.store.RecordForRoles(ctx, roleStrings(auth.RolesWithPermission(auth.ExceptionApprove)), n); err != nil {
+		return fmt.Errorf("notifyfeed: exception requested: %w", err)
+	}
+	return nil
+}
+
+// ExceptionDecided records the outcome notification for the requester (only),
+// closing the loop for the person who asked. Grouped per exception.
+func (g *GovernanceProjector) ExceptionDecided(ctx context.Context, exceptionID, requestedBy uuid.UUID, ruleID string, approved bool) error {
+	if requestedBy == uuid.Nil {
+		return nil
+	}
+	verb, kind := "rejected", "exception_rejected"
+	if approved {
+		verb, kind = "approved", "exception_approved"
+	}
+	// Requester-facing: the rule id identifies the request; no host lookup.
+	n := Notification{
+		UserID:   requestedBy,
+		Kind:     kind,
+		Severity: "medium",
+		Title:    fmt.Sprintf("Exception %s: %s", verb, ruleID),
+		Body:     fmt.Sprintf("Your compliance exception request for %s was %s.", ruleID, verb),
+		Link:     exceptionQueueLink,
+		GroupKey: "exception_decided:" + exceptionID.String(),
+	}
+	if err := g.store.Record(ctx, n); err != nil {
+		return fmt.Errorf("notifyfeed: exception decided: %w", err)
+	}
+	return nil
+}
+
+// RemediationFailed records a "remediation failed / rolled back" notification
+// for the operators who can act on it (remediation:execute → ops_lead,
+// security_admin, admin). Grouped per (host, rule). finalStatus is the
+// terminal remediation status ("failed" | "rolled_back"); action is
+// "execute" | "rollback".
+func (g *GovernanceProjector) RemediationFailed(ctx context.Context, hostID uuid.UUID, ruleID, action, finalStatus string) error {
+	host := g.store.hostName(ctx, hostID)
+	h := hostID
+	lead := "Remediation failed"
+	if finalStatus == "rolled_back" {
+		lead = "Remediation rolled back"
+	}
+	n := Notification{
+		Kind:     "remediation_failed",
+		Severity: "high",
+		Title:    fmt.Sprintf("%s: %s on %s", lead, ruleID, host),
+		Body:     "An automated fix did not complete successfully. Review the host and re-run or remediate manually.",
+		HostID:   &h,
+		Link:     "/hosts/" + hostID.String(),
+		GroupKey: "remediation_failed:" + hostID.String() + ":" + ruleID,
+	}
+	if err := g.store.RecordForRoles(ctx, roleStrings(auth.RolesWithPermission(auth.RemediationExecute)), n); err != nil {
+		return fmt.Errorf("notifyfeed: remediation failed: %w", err)
+	}
+	return nil
+}
+
+// roleStrings adapts auth.RoleID values to the plain strings RecordForRoles
+// queries user_roles.role_id with.
+func roleStrings(ids []auth.RoleID) []string {
+	out := make([]string, len(ids))
+	for i, id := range ids {
+		out[i] = string(id)
+	}
+	return out
+}