RigelBuild · rigel-mintaka · Aug 28, 2026 · Aug 28, 2026
@@ -2,19 +2,23 @@ package store
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
 )
 
-// The forge poll driver's durable state (SEA-1810 T2, design
-// docs/designs/product/compass-forge-poll-driver/design.md §T2): the repo-LIST
-// per-page FETCH cursor (forge_list_cursors) and the board's per-REPO poll
-// targets (forge_repo_subscriptions). The two DL-053 anticipatory tables
-// (agent_forge_subscriptions, forge_artifact_cursors) are writer-less this
-// slice and get their store surface with their writers.
+// The board arm's durable state (RIG-2883): the per-REPO poll targets and their
+// swept-updated-at watermark (forge_repo_subscriptions), plus the poll driver's
+// per-page FETCH cursor (forge_list_cursors) — the latter retires atomically
+// with its serve.go consumer in T5, so it survives this additive slice. The two
+// DL-053 anticipatory tables (agent_forge_subscriptions, forge_artifact_cursors)
+// are writer-less this slice and get their store surface with their writers.
 
 // ForgeListPageCursor is one durable page row of a repo's issue-LIST fetch
 // cursor (the DL-053 FETCH-cursor model at repo-LIST granularity). ETag ""
-// means never fetched (an unconditional GET).
+// means never fetched (an unconditional GET). Retires with the poll driver (T5).
 type ForgeListPageCursor struct {
 	Provider ForgeProvider // GITHUB(1)/GITLAB(2)/FORGEJO(3)/LINEAR(4); never 0
 	Host     string
@@ -24,9 +28,9 @@ type ForgeListPageCursor struct {
 	HasNext  bool
 }
 
-// ForgeRepoSubscription is one board poll target: a repo the poll driver walks
+// ForgeRepoSubscription is one board poll target: a repo the board arm walks
 // (OQ-C's table model). Enabled=false soft-disables the target without deleting
-// its cursor history.
+// its watermark history.
 type ForgeRepoSubscription struct {
 	Provider ForgeProvider
 	Host     string
@@ -133,8 +137,65 @@ func (s *Store) PruneForgeListCursorPages(ctx context.Context, provider ForgePro
 	return nil
 }
 
+// LoadForgeRepoWatermark reads the repo's swept_updated_at watermark and its
+// conditional-GET list_etag. A never-swept repo (swept_updated_at IS NULL)
+// returns the zero time.Time; an unknown coordinate is not an error — it too
+// returns the zero watermark and empty etag (the reconciler treats "no row" and
+// "never swept" identically, walking from the beginning). Zero/empty coordinate
+// fields -> ErrInvalidArgument.
+func (s *Store) LoadForgeRepoWatermark(ctx context.Context, provider ForgeProvider, host, repo string) (time.Time, string, error) {
+	if err := validCoordinate(provider, host, repo); err != nil {
+		return time.Time{}, "", err
+	}
+	var swept *time.Time
+	var etag string
+	err := s.pool.QueryRow(ctx,
+		`SELECT swept_updated_at, list_etag
+		   FROM forge_repo_subscriptions
+		  WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3`,
+		int32(provider), host, repo,
+	).Scan(&swept, &etag)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return time.Time{}, "", nil
+	}
+	if err != nil {
+		return time.Time{}, "", fmt.Errorf("store: load forge repo watermark: %w", err)
+	}
+	if swept == nil {
+		return time.Time{}, etag, nil
+	}
+	return *swept, etag, nil
+}
+
+// StoreForgeRepoWatermark writes the repo's swept_updated_at watermark and
+// list_etag, touching updated_at. An unknown coordinate -> ErrNotFound (the
+// subscription must exist — the seed/upsert path owns row creation). Zero/empty
+// coordinate fields -> ErrInvalidArgument.
+func (s *Store) StoreForgeRepoWatermark(ctx context.Context, provider ForgeProvider, host, repo string, mark time.Time, etag string) error {
+	if err := validCoordinate(provider, host, repo); err != nil {
+		return err
+	}
+	var swept *time.Time
+	if !mark.IsZero() {
+		swept = &mark
+	}
+	tag, err := s.pool.Exec(ctx,
+		`UPDATE forge_repo_subscriptions
+		    SET swept_updated_at = $4, list_etag = $5, updated_at = now()
+		  WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3`,
+		int32(provider), host, repo, swept, etag,
+	)
+	if err != nil {
+		return fmt.Errorf("store: store forge repo watermark: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		return fmt.Errorf("%w: forge repo subscription (%d, %q, %q)", ErrNotFound, provider, host, repo)
+	}
+	return nil
+}
+
 // EnsureForgeRepoSubscription inserts the target if absent; on conflict it DOES
-// NOTHING — the T4 seed reconcile is a bootstrap-only insert and the table is
+// NOTHING — the seed reconcile is a bootstrap-only insert and the table is
 // authoritative after the first insert (the seed never deletes, disables, or
 // re-enables an existing row). Zero/empty coordinate fields -> ErrInvalidArgument.
 func (s *Store) EnsureForgeRepoSubscription(ctx context.Context, sub ForgeRepoSubscription) error {
@@ -152,9 +213,66 @@ func (s *Store) EnsureForgeRepoSubscription(ctx context.Context, sub ForgeRepoSu
 	return nil
 }
 
+// ListEnabledForgeRepos reads every enabled target's repo, ascending — the board
+// reconciler's per-pass target enumeration across all coordinates. No rows is a
+// nil slice, not an error.
+//
+// Repo-only keyed (no provider/host), matching the frozen repo-keyed ingest
+// seam. In a github.com-only deployment repo is unambiguous; if multi-host is
+// ever enabled, two coordinates sharing a repo string (e.g. github.com and a GHE
+// host both carrying "a/b") would collapse to one entry here and to an ambiguous
+// watermark under the coordinate-keyed Load/Store methods — thread (provider,
+// host) through this seam before enabling multi-host.
+func (s *Store) ListEnabledForgeRepos(ctx context.Context) ([]string, error) {
+	rows, err := s.pool.Query(ctx,
+		`SELECT repo
+		   FROM forge_repo_subscriptions
+		  WHERE enabled = TRUE
+		  ORDER BY repo ASC`,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("store: list enabled forge repos: %w", err)
+	}
+	defer rows.Close()
+
+	var out []string
+	for rows.Next() {
+		var repo string
+		if err := rows.Scan(&repo); err != nil {
+			return nil, fmt.Errorf("store: scan forge repo: %w", err)
+		}
+		out = append(out, repo)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("store: iterate forge repos: %w", err)
+	}
+	return out, nil
+}
+
+// IsEnabledForgeRepo reports whether an enabled subscription exists for the repo
+// (the point membership check the webhook arm gates on). An empty repo ->
+// ErrInvalidArgument. Repo-only keyed like ListEnabledForgeRepos — returns true
+// if ANY coordinate's subscription for the repo is enabled; unambiguous in a
+// github.com-only deployment (see ListEnabledForgeRepos for the multi-host note).
+func (s *Store) IsEnabledForgeRepo(ctx context.Context, repo string) (bool, error) {
+	if repo == "" {
+		return false, fmt.Errorf("%w: repo is required", ErrInvalidArgument)
+	}
+	var exists bool
+	if err := s.pool.QueryRow(ctx,
+		`SELECT EXISTS (
+		   SELECT 1 FROM forge_repo_subscriptions
+		    WHERE repo = $1 AND enabled = TRUE)`,
+		repo,
+	).Scan(&exists); err != nil {
+		return false, fmt.Errorf("store: is enabled forge repo: %w", err)
+	}
+	return exists, nil
+}
+
 // ListEnabledForgeRepoSubscriptions reads the enabled targets for one (provider,
-// host), ascending repo — the driver's per-pass target enumeration. No rows is a
-// nil slice, not an error. Zero provider / empty host -> ErrInvalidArgument.
+// host), ascending repo. No rows is a nil slice, not an error. Zero provider /
+// empty host -> ErrInvalidArgument.
 func (s *Store) ListEnabledForgeRepoSubscriptions(ctx context.Context, provider ForgeProvider, host string) ([]ForgeRepoSubscription, error) {
 	if provider == ForgeProviderUnspecified {
 		return nil, fmt.Errorf("%w: forge provider is required", ErrInvalidArgument)

@@ -26,15 +26,19 @@ func TestMigration0016TablesExist(t *testing.T) {
 	ctx := context.Background()
 	s := newTestStore(t)
 
-	// forge_repo_subscriptions
-	mustExec(t, s, `INSERT INTO forge_repo_subscriptions (forge_provider, forge_host, repo) VALUES (1, 'github.com', 'a/b')`)
+	// forge_repo_subscriptions — including the RIG-2883 T4 columns
+	// (swept_updated_at watermark + list_etag), proving they applied.
+	mustExec(t, s, `INSERT INTO forge_repo_subscriptions (forge_provider, forge_host, repo, swept_updated_at, list_etag) VALUES (1, 'github.com', 'a/b', now(), '"e"')`)
 	// forge_list_cursors
 	mustExec(t, s, `INSERT INTO forge_list_cursors (forge_provider, forge_host, repo, page) VALUES (1, 'github.com', 'a/b', 1)`)
 	// forge_artifact_cursors — both legal kind values (1=issue, 2=pull_request)
 	// accept, proving the CHECK IN (1,2) upper bound, symmetric to the
 	// provider-domain 1..4 accept test.
 	mustExec(t, s, `INSERT INTO forge_artifact_cursors (forge_provider, forge_host, repo, kind, number) VALUES (1, 'github.com', 'a/b', 1, 7)`)
 	mustExec(t, s, `INSERT INTO forge_artifact_cursors (forge_provider, forge_host, repo, kind, number) VALUES (1, 'github.com', 'a/b', 2, 8)`)
+	// issues — the RIG-2883 T4 forge_updated_at column is present (INERT this
+	// slice; T4a threads its write path). Proven present by inserting it.
+	mustExec(t, s, `INSERT INTO issues (id, forge_provider, forge_host, repo, number, forge_updated_at) VALUES ('i-1', 1, 'github.com', 'a/b', 7, now())`)
 	// agent_forge_subscriptions needs a real agent_account_id (FK); seed one.
 	owner := mustUser(t, s, "forge-owner")
 	agent := mustAgent(t, s, owner.ID, "forge-agent")
@@ -373,6 +377,161 @@ func TestForgeCursorInvalidArgument(t *testing.T) {
 	sentinelIs(t, s.SetForgeRepoSubscriptionEnabled(ctx, ForgeProviderGitHub, "h", "", true), ErrInvalidArgument, "set empty repo")
 }
 
+// ── Test 8: forge_repo_subscriptions watermark round-trip (RIG-2883 T4) ───────
+
+// TestForgeRepoWatermarkRoundTrip proves the swept_updated_at + list_etag
+// watermark persists: a never-swept row reads zero/empty, an unknown coordinate
+// also reads zero/empty (not an error — the reconciler treats "no row" and
+// "never swept" identically), a store then reads back, and a store on an unknown
+// coordinate is ErrNotFound (the subscription must exist first).
+func TestForgeRepoWatermarkRoundTrip(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+
+	sub := ForgeRepoSubscription{Provider: ForgeProviderGitHub, Host: "github.com", Repo: "a/b", Enabled: true}
+	if err := s.EnsureForgeRepoSubscription(ctx, sub); err != nil {
+		t.Fatalf("ensure: %v", err)
+	}
+
+	// Never-swept row: zero time, empty etag.
+	mark, etag, err := s.LoadForgeRepoWatermark(ctx, ForgeProviderGitHub, "github.com", "a/b")
+	if err != nil {
+		t.Fatalf("load never-swept: %v", err)
+	}
+	if !mark.IsZero() || etag != "" {
+		t.Fatalf("never-swept = (%v, %q), want (zero, \"\")", mark, etag)
+	}
+
+	// Unknown coordinate (no row) also reads zero/empty, not an error.
+	mark, etag, err = s.LoadForgeRepoWatermark(ctx, ForgeProviderGitLab, "gitlab.com", "x/y")
+	if err != nil {
+		t.Fatalf("load unknown coordinate: %v", err)
+	}
+	if !mark.IsZero() || etag != "" {
+		t.Fatalf("unknown coordinate = (%v, %q), want (zero, \"\")", mark, etag)
+	}
+
+	// Store then round-trip.
+	want := time.Date(2026, 8, 27, 12, 0, 0, 0, time.UTC)
+	if err := s.StoreForgeRepoWatermark(ctx, ForgeProviderGitHub, "github.com", "a/b", want, `"v1"`); err != nil {
+		t.Fatalf("store watermark: %v", err)
+	}
+	mark, etag, err = s.LoadForgeRepoWatermark(ctx, ForgeProviderGitHub, "github.com", "a/b")
+	if err != nil {
+		t.Fatalf("load after store: %v", err)
+	}
+	if !mark.Equal(want) {
+		t.Fatalf("watermark = %v, want %v", mark, want)
+	}
+	if etag != `"v1"` {
+		t.Fatalf("etag = %q, want %q", etag, `"v1"`)
+	}
+
+	// Storing on an unknown coordinate -> ErrNotFound (the subscription must exist).
+	err = s.StoreForgeRepoWatermark(ctx, ForgeProviderGitHub, "github.com", "no/such", want, "")
+	sentinelIs(t, err, ErrNotFound, "store watermark on unknown coordinate")
+}
+
+// ── Test 9: watermark coordinate isolation across (provider, host) ────────────
+
+func TestForgeRepoWatermarkCoordinateIsolation(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+
+	const repo = "a/b"
+	subs := []struct {
+		provider ForgeProvider
+		host     string
+		mark     time.Time
+		etag     string
+	}{
+		{ForgeProviderGitHub, "github.com", time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), `"gh"`},
+		{ForgeProviderGitHub, "ghe.example.com", time.Date(2026, 2, 2, 0, 0, 0, 0, time.UTC), `"ghe"`},
+		{ForgeProviderGitLab, "github.com", time.Date(2026, 3, 3, 0, 0, 0, 0, time.UTC), `"gl"`},
+	}
+	for _, c := range subs {
+		if err := s.EnsureForgeRepoSubscription(ctx, ForgeRepoSubscription{Provider: c.provider, Host: c.host, Repo: repo, Enabled: true}); err != nil {
+			t.Fatalf("ensure (%d,%q): %v", c.provider, c.host, err)
+		}
+		if err := s.StoreForgeRepoWatermark(ctx, c.provider, c.host, repo, c.mark, c.etag); err != nil {
+			t.Fatalf("store (%d,%q): %v", c.provider, c.host, err)
+		}
+	}
+	// Each coordinate reads back exactly its own watermark and etag.
+	for _, c := range subs {
+		mark, etag, err := s.LoadForgeRepoWatermark(ctx, c.provider, c.host, repo)
+		if err != nil {
+			t.Fatalf("load (%d,%q): %v", c.provider, c.host, err)
+		}
+		if !mark.Equal(c.mark) || etag != c.etag {
+			t.Fatalf("coordinate (%d,%q) = (%v, %q), want (%v, %q)", c.provider, c.host, mark, etag, c.mark, c.etag)
+		}
+	}
+}
+
+// ── Test 10: enabled-repo enumeration + point membership (RIG-2883 T4) ────────
+
+func TestListAndIsEnabledForgeRepos(t *testing.T) {
+	ctx := context.Background()
+	s := newTestStore(t)
+
+	// No rows -> nil, and the point check is false.
+	repos, err := s.ListEnabledForgeRepos(ctx)
+	if err != nil {
+		t.Fatalf("list enabled repos empty: %v", err)
+	}
+	if repos != nil {
+		t.Fatalf("empty = %v, want nil", repos)
+	}
+	ok, err := s.IsEnabledForgeRepo(ctx, "a/b")
+	if err != nil {
+		t.Fatalf("is-enabled empty: %v", err)
+	}
+	if ok {
+		t.Fatal("is-enabled = true on empty, want false")
+	}
+
+	// Seed enabled repos out of lexical order across coordinates, plus one
+	// disabled — enumeration is ascending and excludes the disabled repo.
+	seed := []ForgeRepoSubscription{
+		{Provider: ForgeProviderGitHub, Host: "github.com", Repo: "z/z", Enabled: true},
+		{Provider: ForgeProviderGitLab, Host: "gitlab.com", Repo: "a/a", Enabled: true},
+		{Provider: ForgeProviderGitHub, Host: "ghe.example.com", Repo: "m/m", Enabled: true},
+	}
+	for _, sub := range seed {
+		if err := s.EnsureForgeRepoSubscription(ctx, sub); err != nil {
+			t.Fatalf("ensure %+v: %v", sub, err)
+		}
+	}
+	if err := s.EnsureForgeRepoSubscription(ctx, ForgeRepoSubscription{Provider: ForgeProviderGitHub, Host: "github.com", Repo: "d/d", Enabled: false}); err != nil {
+		t.Fatalf("ensure disabled: %v", err)
+	}
+
+	repos, err = s.ListEnabledForgeRepos(ctx)
+	if err != nil {
+		t.Fatalf("list enabled repos: %v", err)
+	}
+	if len(repos) != 3 || repos[0] != "a/a" || repos[1] != "m/m" || repos[2] != "z/z" {
+		t.Fatalf("list enabled repos = %v, want [a/a m/m z/z] ascending", repos)
+	}
+
+	// Point check: an enabled repo is true, the disabled repo is false.
+	ok, err = s.IsEnabledForgeRepo(ctx, "m/m")
+	if err != nil {
+		t.Fatalf("is-enabled m/m: %v", err)
+	}
+	if !ok {
+		t.Fatal("is-enabled m/m = false, want true")
+	}
+	ok, err = s.IsEnabledForgeRepo(ctx, "d/d")
+	if err != nil {
+		t.Fatalf("is-enabled d/d: %v", err)
+	}
+	if ok {
+		t.Fatal("is-enabled d/d (disabled) = true, want false")
+	}
+}
+
 // ── helpers ───────────────────────────────────────────────────────────────────
 
 func mustExec(t *testing.T, s *Store, sql string) {

@@ -586,6 +586,11 @@ CREATE TABLE issues (
     labels         TEXT[]   NOT NULL DEFAULT '{}',
     agent_handle   TEXT     NOT NULL DEFAULT '',  -- '' = non-Compass author
 
+    -- OQ-6(a) recency-guard column (RIG-2883 T4): the forge's last-updated
+    -- timestamp for the artifact. INERT until T4a threads the write path
+    -- (Issue.UpdatedAt reaches no writer today); the bare column is a no-op.
+    forge_updated_at TIMESTAMPTZ,
+
     -- Compass machinery (server-owned; none on the forge). state defaults to
     -- BACKLOG; CHECK 1..8: a persisted issue is NEVER UNSPECIFIED(0). The
     -- machinery columns get their writers in later slices.
@@ -618,6 +623,8 @@ CREATE TABLE forge_repo_subscriptions (
     forge_host     TEXT     NOT NULL,
     repo           TEXT     NOT NULL,
     enabled        BOOLEAN  NOT NULL DEFAULT TRUE,
+    swept_updated_at TIMESTAMPTZ,              -- last swept forge updated_at watermark; NULL = never swept
+    list_etag      TEXT     NOT NULL DEFAULT '', -- conditional-GET etag for the repo LIST walk
     created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
     updated_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
     PRIMARY KEY (forge_provider, forge_host, repo)
@@ -671,7 +678,8 @@ CREATE TABLE forge_artifact_cursors (
 -- A durable conditional-GET cache; etag advances ONLY after every row of that
 -- page's content is durably sunk. has_next persists the Link-chain fact so a 304
 -- can keep walking a multi-page repo. advanced_at records the last content
--- advance (an etag-storing 200+sink), NOT the last poll.
+-- advance (an etag-storing 200+sink), NOT the last poll. Retires with the poll
+-- driver (RIG-2883 T5), atomically with its serve.go consumer.
 CREATE TABLE forge_list_cursors (
     forge_provider SMALLINT NOT NULL CHECK (forge_provider IN (1, 2, 3, 4)),
     forge_host     TEXT     NOT NULL,