feat(forge): ListUpdatedIssues updated-order conditional walk (RIG-2883 T2)

go/internal/forge/github.go

@@ -143,6 +143,12 @@ type ghIssue struct {
 	PullRequest *json.RawMessage `json:"pull_request"`
 }
 
+// jsonNull is the literal a *json.RawMessage holds when GitHub sends an explicit
+// `"pull_request": null` (a non-nil RawMessage wrapping the four bytes), as
+// opposed to omitting the key for a plain issue. The interleaved-PR guards test
+// against it so a null marker never drops a real issue.
+const jsonNull = "null"
+
 // ghError is the wire shape of a GitHub error body (the message field feeds
 // StatusError.Message).
 type ghError struct {
@@ -223,7 +229,7 @@ func (g *GitHub) ListIssuesPage(ctx context.Context, repo string, f IssueFilter,
 		// OMITS the pull_request key for a plain issue, but a *json.RawMessage
 		// unmarshals an explicit "pull_request": null to a non-nil
 		// RawMessage("null") — guard that so a null never drops a real issue.
-		if raw := r.PullRequest; raw != nil && len(*raw) > 0 && string(*raw) != "null" {
+		if raw := r.PullRequest; raw != nil && len(*raw) > 0 && string(*raw) != jsonNull {
 			continue
 		}
 		issues = append(issues, r.toIssue())

go/internal/forge/notify_reader.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"net/http"
 	"strconv"
+	"time"
 
 	compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1"
 )
@@ -238,7 +239,7 @@ func (g *GitHub) ListNewArtifacts(ctx context.Context, repo string, kind compass
 				// Drop the PR rows GitHub interleaves into /issues (a *json.Raw
 				// unmarshals an explicit "pull_request": null to a non-nil
 				// RawMessage("null") — guard that, mirroring ListIssuesPage).
-				if raw := r.PullRequest; raw != nil && len(*raw) > 0 && string(*raw) != "null" {
+				if raw := r.PullRequest; raw != nil && len(*raw) > 0 && string(*raw) != jsonNull {
 					return Issue{}, false
 				}
 				return r.toIssue(), true
@@ -297,6 +298,69 @@ func ghWalkNewArtifacts[R any](ctx context.Context, g *GitHub, base string, sinc
 	return ConditionalResult[[]Issue]{V: out, ETag: pageETag}, nil
 }
 
+// ListUpdatedIssues walks /repos/{repo}/issues?state=all&sort=updated&
+// direction=desc newest-updated-first (page 1 conditioned on etag; a 304 =>
+// NotModified), collecting issue rows (PR rows dropped by the pull_request
+// marker, mirroring ListNewArtifacts) until a page's oldest updated_at is
+// strictly < since, or no rel="next" remains. Rows with updated_at == since are
+// RE-included: GitHub's updated_at is second-granularity, so a <= stop would
+// permanently exclude an issue updated in the same second as the stored
+// watermark after the sweep read it; the duplicates are free by coordinate
+// idempotency. A zero since walks ALL pages (cold start). It returns page 1's
+// ETag to re-store. This is the updated-order sibling of ListNewArtifacts
+// (created-order, number-keyed): the reconcile/backfill read cannot see updates
+// to existing issues via the created-order walk.
+func (g *GitHub) ListUpdatedIssues(ctx context.Context, repo string, since time.Time, etag string) (ConditionalResult[[]Issue], error) {
+	base := g.apiBase() + "/repos/" + repo + "/issues?state=all&sort=updated&direction=desc"
+	var out []Issue
+	pageETag := ""
+	for page := 1; ; page++ {
+		u := base + "&per_page=" + strconv.Itoa(perPage) + "&page=" + strconv.Itoa(page)
+		sendETag := ""
+		if page == 1 {
+			sendETag = etag
+		}
+		var rows []ghIssue
+		notMod, e, hasNext, err := g.getJSONCond(ctx, u, sendETag, &rows)
+		if err != nil {
+			return ConditionalResult[[]Issue]{}, fmt.Errorf("forge: github list updated issues %q: %w", repo, err)
+		}
+		if page == 1 {
+			if notMod {
+				return ConditionalResult[[]Issue]{NotModified: true}, nil
+			}
+			pageETag = e
+		}
+		reachedOld := false
+		for _, r := range rows {
+			iss := r.toIssue()
+			if !iss.UpdatedAt.IsZero() && iss.UpdatedAt.Before(since) {
+				// Newest-updated-first: strictly older than the watermark, so this
+				// and everything after it is old. A row == since is NOT Before it,
+				// so it is re-included (second-granularity dedup safety). A row
+				// whose updated_at failed to parse (zero time) is NOT a stop
+				// signal — treating it as one would let a single malformed row
+				// truncate the whole sweep persistently; skip it and keep walking.
+				reachedOld = true
+				continue
+			}
+			if iss.UpdatedAt.IsZero() {
+				continue
+			}
+			// Drop the PR rows GitHub interleaves into /issues (mirroring
+			// ListNewArtifacts / ListIssuesPage's pull_request-marker guard).
+			if raw := r.PullRequest; raw != nil && len(*raw) > 0 && string(*raw) != jsonNull {
+				continue
+			}
+			out = append(out, iss)
+		}
+		if reachedOld || !hasNext {
+			break
+		}
+	}
+	return ConditionalResult[[]Issue]{V: out, ETag: pageETag}, nil
+}
+
 // --- Linear arm --------------------------------------------------------------
 
 // Compile-time proof the Linear client satisfies the conditional-read surface.

go/internal/forge/notify_reader_test.go

@@ -12,6 +12,7 @@ import (
 	"errors"
 	"net/http"
 	"testing"
+	"time"
 
 	compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1"
 )
@@ -274,6 +275,179 @@ func TestListNewArtifactsPage1_304(t *testing.T) {
 	}
 }
 
+// --- GitHub: ListUpdatedIssues contract points -------------------------------
+
+// TestListUpdatedIssuesStopsStrictlyBelowSince: a multi-page updated-order walk
+// stops when a page's oldest updated_at is strictly < since. A row whose
+// updated_at == since is RE-included (second-granularity dedup safety — a <=
+// stop would permanently drop a same-second issue).
+func TestListUpdatedIssuesStopsStrictlyBelowSince(t *testing.T) {
+	since := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC)
+	// Page 1 all >= since (newest-updated-first); its oldest (== since) is NOT
+	// below since, so the walk continues to page 2.
+	page1 := `[
+		{"number":50,"state":"open","html_url":"u50","updated_at":"2026-08-01T12:00:02Z"},
+		{"number":49,"state":"open","html_url":"u49","updated_at":"2026-08-01T12:00:00Z"}
+	]`
+	// Page 2's first row == since (re-included), its second is strictly below
+	// (dropped, and stops the walk).
+	page2 := `[
+		{"number":48,"state":"open","html_url":"u48","updated_at":"2026-08-01T12:00:00Z"},
+		{"number":40,"state":"open","html_url":"u40","updated_at":"2026-07-31T23:59:59Z"}
+	]`
+	rt := &scriptedRoundTripper{responses: []scriptedResponse{
+		{status: 200, body: page1, headers: map[string]string{"ETag": `"n1"`, "Link": `<https://api.github.com/x?page=2>; rel="next"`}},
+		{status: 200, body: page2, headers: map[string]string{"Link": `<https://api.github.com/x?page=3>; rel="next"`}},
+	}}
+	g := newTestGitHub(rt, &fakeTokenSource{token: "t"})
+
+	res, err := g.ListUpdatedIssues(context.Background(), "org/repo", since, "")
+	if err != nil {
+		t.Fatalf("ListUpdatedIssues: %v", err)
+	}
+	// 50,49 (page 1) + 48 (== since, re-included) = 3; 40 is strictly below and
+	// stops the walk before any page 3 is requested.
+	if len(res.V) != 3 {
+		t.Fatalf("kept %d (%+v), want 3 (50,49,48; 48 re-included at == since, 40 stops the walk)", len(res.V), res.V)
+	}
+	if res.V[2].Number != 48 {
+		t.Errorf("V[2].Number = %d, want 48 (the == since row is re-included)", res.V[2].Number)
+	}
+	if rt.calls != 2 {
+		t.Errorf("calls = %d, want 2 (the walk stopped once a page's oldest was strictly < since)", rt.calls)
+	}
+	if res.ETag != `"n1"` {
+		t.Errorf("ETag = %q, want page 1's %q (re-store)", res.ETag, `"n1"`)
+	}
+}
+
+// TestListUpdatedIssuesMalformedRowDoesNotTruncate: a row whose updated_at fails
+// to parse (zero time) must NOT be treated as the strictly-below-since stop
+// signal — otherwise one malformed row truncates the whole sweep persistently.
+// It is skipped (no output), and the walk continues to the valid rows behind it.
+func TestListUpdatedIssuesMalformedRowDoesNotTruncate(t *testing.T) {
+	since := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC)
+	// A malformed updated_at sits BETWEEN two valid >= since rows on page 1. If
+	// the zero time were a stop signal it would drop #48 (a real, fresh issue).
+	page1 := `[
+		{"number":50,"state":"open","html_url":"u50","updated_at":"2026-08-01T12:00:05Z"},
+		{"number":49,"state":"open","html_url":"u49","updated_at":"not-a-timestamp"},
+		{"number":48,"state":"open","html_url":"u48","updated_at":"2026-08-01T12:00:03Z"}
+	]`
+	rt := &scriptedRoundTripper{responses: []scriptedResponse{
+		{status: 200, body: page1, headers: map[string]string{"ETag": `"n1"`}},
+	}}
+	g := newTestGitHub(rt, &fakeTokenSource{token: "t"})
+
+	res, err := g.ListUpdatedIssues(context.Background(), "org/repo", since, "")
+	if err != nil {
+		t.Fatalf("ListUpdatedIssues: %v", err)
+	}
+	// 50 and 48 are collected; the malformed 49 is skipped, not a stop signal.
+	if len(res.V) != 2 {
+		t.Fatalf("kept %d (%+v), want 2 (50,48; the malformed 49 is skipped without truncating)", len(res.V), res.V)
+	}
+	if res.V[0].Number != 50 || res.V[1].Number != 48 {
+		t.Errorf("kept %d,%d, want 50,48 (the row behind the malformed one still collected)", res.V[0].Number, res.V[1].Number)
+	}
+}
+
+// TestListUpdatedIssuesPage1_304: a 304 on page 1 short-circuits to NotModified
+// without walking (nothing updated since the last sweep).
+func TestListUpdatedIssuesPage1_304(t *testing.T) {
+	rt := &scriptedRoundTripper{responses: []scriptedResponse{
+		{status: http.StatusNotModified, headers: map[string]string{"ETag": `"u0"`}},
+	}}
+	g := newTestGitHub(rt, &fakeTokenSource{token: "t"})
+
+	res, err := g.ListUpdatedIssues(context.Background(), "org/repo", time.Time{}, `"u0"`)
+	if err != nil {
+		t.Fatalf("ListUpdatedIssues: %v", err)
+	}
+	if !res.NotModified || rt.calls != 1 {
+		t.Errorf("NotModified=%v calls=%d, want true/1", res.NotModified, rt.calls)
+	}
+}
+
+// TestListUpdatedIssuesZeroSinceWalksAll: a zero since (cold start) walks every
+// page to the last (no rel="next"), collecting all issue rows.
+func TestListUpdatedIssuesZeroSinceWalksAll(t *testing.T) {
+	page1 := `[{"number":50,"state":"open","html_url":"u50","updated_at":"2026-08-01T12:00:02Z"}]`
+	page2 := `[{"number":10,"state":"closed","html_url":"u10","updated_at":"2020-01-01T00:00:00Z"}]`
+	rt := &scriptedRoundTripper{responses: []scriptedResponse{
+		{status: 200, body: page1, headers: map[string]string{"ETag": `"n1"`, "Link": `<https://api.github.com/x?page=2>; rel="next"`}},
+		{status: 200, body: page2},
+	}}
+	g := newTestGitHub(rt, &fakeTokenSource{token: "t"})
+
+	res, err := g.ListUpdatedIssues(context.Background(), "org/repo", time.Time{}, "")
+	if err != nil {
+		t.Fatalf("ListUpdatedIssues: %v", err)
+	}
+	// A zero since is never strictly greater than any updated_at, so nothing
+	// stops the walk short — both pages collected, walk ends at no rel="next".
+	if len(res.V) != 2 {
+		t.Fatalf("kept %d (%+v), want 2 (zero since walks to the last page)", len(res.V), res.V)
+	}
+	if rt.calls != 2 {
+		t.Errorf("calls = %d, want 2 (walk ended only at no rel=\"next\")", rt.calls)
+	}
+}
+
+// TestListUpdatedIssuesFiltersPRs: /repos/{repo}/issues interleaves PR rows
+// (pull_request marker); the updated-order walk drops them, keeping only
+// issue-shaped rows, on the updated-order endpoint.
+func TestListUpdatedIssuesFiltersPRs(t *testing.T) {
+	body := `[
+		{"number":45,"state":"open","html_url":"u45","updated_at":"2026-08-01T12:00:02Z","pull_request":{"url":"pr"}},
+		{"number":44,"state":"open","html_url":"u44","updated_at":"2026-08-01T12:00:01Z"}
+	]`
+	rt := &scriptedRoundTripper{responses: []scriptedResponse{
+		{status: 200, body: body, headers: map[string]string{"ETag": `"n1"`}},
+	}}
+	g := newTestGitHub(rt, &fakeTokenSource{token: "t"})
+
+	res, err := g.ListUpdatedIssues(context.Background(), "org/repo", time.Time{}, "")
+	if err != nil {
+		t.Fatalf("ListUpdatedIssues: %v", err)
+	}
+	if len(res.V) != 1 || res.V[0].Number != 44 {
+		t.Fatalf("kept %+v, want only issue #44 (PR #45 filtered)", res.V)
+	}
+	if got := rt.requests[0].URL.Path; got != "/repos/org/repo/issues" {
+		t.Errorf("path = %q, want the issues endpoint", got)
+	}
+	if q := rt.requests[0].URL.Query(); q.Get("sort") != "updated" || q.Get("direction") != "desc" {
+		t.Errorf("query sort=%q direction=%q, want updated/desc", q.Get("sort"), q.Get("direction"))
+	}
+}
+
+// TestListUpdatedIssuesBudgetErrorPropagates: an armed budget gate fails the
+// page-1 read fast with ErrBudgetExhausted, propagated to the caller.
+func TestListUpdatedIssuesBudgetErrorPropagates(t *testing.T) {
+	base := time.Date(2026, 8, 9, 12, 0, 0, 0, time.UTC)
+	// Page 1 arms the gate (remaining==0); page 2's read then fails fast.
+	page1 := `[{"number":50,"state":"open","html_url":"u50","updated_at":"2026-08-01T12:00:02Z"}]`
+	rt := &scriptedRoundTripper{responses: []scriptedResponse{
+		{status: 200, body: page1, headers: map[string]string{
+			"ETag":                  `"n1"`,
+			"Link":                  `<https://api.github.com/x?page=2>; rel="next"`,
+			"x-ratelimit-remaining": "0",
+			"x-ratelimit-reset":     "9999999999",
+		}},
+	}}
+	g := newTestGitHub(rt, &fakeTokenSource{token: "t"})
+	g.now = func() time.Time { return base }
+
+	_, err := g.ListUpdatedIssues(context.Background(), "org/repo", time.Time{}, "")
+	if !errors.Is(err, ErrBudgetExhausted) {
+		t.Fatalf("err = %v, want ErrBudgetExhausted (gate armed on page 1 fails page 2 fast)", err)
+	}
+	if rt.calls != 1 {
+		t.Errorf("calls = %d, want 1 (page 2 fails fast without a request)", rt.calls)
+	}
+}
+
 // --- Linear arms -------------------------------------------------------------
 
 // TestLinearReaderNoETags: a Linear issue read returns a 200-equivalent with an