refactor(forge_client): Phase B language-advantage pass — dichotomy wave

src/go/forge_client/forge_client.go

@@ -364,26 +364,11 @@ func (a *GiteaForgejoAdapter) ListOpenIssues() ([]any, error) {
 // paginated: q matches titles (and keywords), so the full marker is sent
 // as q and a client-side contains-check keeps exactly the issues whose
 // title OR body contains the marker.
+//
+// Phase B: pagination — the module's one serial IO seam — is pipelined
+// (see phaseb.go); the request sequence is identical to the serial path.
 func (a *GiteaForgejoAdapter) SearchIssues(query string) ([]any, error) {
-	results := []any{}
-	page := 1
-	for {
-		r, err := Request(a.Sender, "GET",
-			a.BaseURL+"/issues?"+searchQuery(query, page), a.opts())
-		if err != nil {
-			return nil, err
-		}
-		v, err := r.JSON()
-		if err != nil {
-			return nil, err
-		}
-		batch, _ := v.([]any)
-		results = append(results, batch...)
-		if len(batch) < SearchPageSize {
-			return filterMarker(results, query), nil
-		}
-		page++
-	}
+	return a.SearchIssuesPhaseB(query)
 }
 
 // searchQuery — Python urlencode preserves insertion order; so does this.

src/go/forge_client/phaseb.go

@@ -0,0 +1,70 @@
+package main
+
+// Phase B — language-advantage pass. The parity port kept SearchIssues
+// serial (page fetched → appended → next page fetched). Pagination is the
+// module's one serial IO seam, so Phase B pipelines it: a producer
+// goroutine fetches pages strictly in order (stopping after a short page,
+// exactly like the Python oracle) while the consumer filters markers from
+// already-received pages. The request sequence is byte-identical to the
+// serial path — a page N+1 is only requested after page N arrived and was
+// full — so parity is preserved while filtering CPU overlaps network wait.
+
+// searchPages streams pages from the paginated list endpoint. The channel
+// is unbuffered: the producer blocks until the consumer has received page
+// N before fetching page N+1, so no speculative request is ever made that
+// the serial oracle would not have made.
+func (a *GiteaForgejoAdapter) searchPages(query string, pages chan<- pageResult) {
+	defer close(pages)
+	for pageNum := 1; ; pageNum++ {
+		r, err := Request(a.Sender, "GET",
+			a.BaseURL+"/issues?"+searchQuery(query, pageNum), a.opts())
+		if err != nil {
+			pages <- pageResult{err: err}
+			return
+		}
+		v, err := r.JSON()
+		if err != nil {
+			pages <- pageResult{err: err}
+			return
+		}
+		batch, _ := v.([]any)
+		if batch == nil {
+			batch = []any{}
+		}
+		pages <- pageResult{items: batch}
+		if len(batch) < SearchPageSize {
+			return
+		}
+	}
+}
+
+type pageResult struct {
+	items []any
+	err   error
+}
+
+// SearchIssuesPhaseB — the pipelined variant. Same result, same request
+// sequence, same error semantics (the first transport error propagates and
+// discards partial results, as in the oracle). Filtering is per-item, so
+// filtering each page as it arrives is identical to the oracle's
+// end-of-loop filter, only overlapped with the next page's network wait.
+func (a *GiteaForgejoAdapter) SearchIssuesPhaseB(query string) ([]any, error) {
+	pages := make(chan pageResult)
+	go a.searchPages(query, pages)
+	out := []any{}
+	for p := range pages {
+		if p.err != nil {
+			return nil, p.err
+		}
+		out = append(out, filterMarker(p.items, query)...)
+	}
+	return out, nil
+}
+
+// Compile-time safety: both adapters satisfy the protocol, the default
+// transport satisfies the seam.
+var (
+	_ ForgeClient = (*GiteaForgejoAdapter)(nil)
+	_ ForgeClient = NullAdapter{}
+	_ Sender      = (*HTTPXSender)(nil)
+)

src/go/forge_client/phaseb_test.go

@@ -0,0 +1,186 @@
+package main
+
+// Phase B battery — the goroutine pipeline at the pagination seam must be
+// indistinguishable from the serial oracle path: same results, same request
+// sequence, no speculative fetches, same error semantics. Race-clean under
+// -race.
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+)
+
+func newPageSender(pages [][]any) *scriptedSender {
+	entries := make([]scriptEntry, 0, len(pages))
+	for _, p := range pages {
+		entries = append(entries, scriptEntry{Status: 200, Payload: p})
+	}
+	return &scriptedSender{entries: entries, requests: &[]any{}}
+}
+
+func markerItem(n int, marker bool) map[string]any {
+	title := fmt.Sprintf("filler issue %d", n)
+	if marker {
+		title = fmt.Sprintf("[auto-failure:cron:Timeout] %d", n)
+	}
+	return map[string]any{"number": n, "title": title, "body": ""}
+}
+
+// TestPhaseBPipelineParity — a 3-page search through the pipelined path
+// yields exactly the oracle's result: markers kept in order, noise dropped.
+func TestPhaseBPipelineParity(t *testing.T) {
+	pages := [][]any{
+		fullPageWith(2),
+		fullPageWith(4),
+		{markerItem(5, true)},
+	}
+	sender := newPageSender(pages)
+	a := NewGiteaForgejoAdapter("https://forge.example.com", "tok", "o", "r")
+	a.Sender = sender
+	got, err := a.SearchIssues("[auto-failure:cron:Timeout]")
+	if err != nil {
+		t.Fatalf("pipelined search failed: %v", err)
+	}
+	var nums []int
+	for _, item := range got {
+		m, _ := item.(map[string]any)
+		nums = append(nums, numInt(m["number"]))
+	}
+	if len(nums) != 3 || nums[0] != 2 || nums[1] != 4 || nums[2] != 5 {
+		t.Fatalf("pipelined filter diverged: %v", nums)
+	}
+	if len(*sender.requests) != 3 {
+		t.Fatalf("expected 3 page requests, got %d", len(*sender.requests))
+	}
+	for i, req := range *sender.requests {
+		url := fmt.Sprint(req.(map[string]any)["url"])
+		if !strings.Contains(url, fmt.Sprintf("page=%d", i+1)) {
+			t.Fatalf("page %d requested out of order: %s", i+1, url)
+		}
+	}
+}
+
+// TestPhaseBNoSpeculativeRequests — a short first page must terminate the
+// producer with exactly one request; the goroutine must not prefetch a
+// page the serial oracle would never have asked for.
+func TestPhaseBNoSpeculativeRequests(t *testing.T) {
+	sender := newPageSender([][]any{{markerItem(1, true)}})
+	a := NewGiteaForgejoAdapter("https://forge.example.com", "tok", "o", "r")
+	a.Sender = sender
+	got, err := a.SearchIssues("[auto-failure:cron:Timeout]")
+	if err != nil {
+		t.Fatalf("search failed: %v", err)
+	}
+	if len(got) != 1 {
+		t.Fatalf("expected the single marker hit, got %d items", len(got))
+	}
+	if len(*sender.requests) != 1 {
+		t.Fatalf("speculative fetch detected: %d requests for a short single page", len(*sender.requests))
+	}
+}
+
+// TestPhaseBErrorPropagation — a transport error on page 2 surfaces with
+// the frozen httpx-kind vocabulary and discards partial results, exactly
+// like the serial path. The connect error is retryable NOISE per _request
+// (retries=2 → 3 attempts), so the script must carry all three before the
+// error finally propagates.
+func TestPhaseBErrorPropagation(t *testing.T) {
+	sender := &scriptedSender{
+		entries: []scriptEntry{
+			{Status: 200, Payload: fullPage()},
+			{Exc: "connect"},
+			{Exc: "connect"},
+			{Exc: "connect"},
+		},
+		requests: &[]any{},
+	}
+	a := NewGiteaForgejoAdapter("https://forge.example.com", "tok", "o", "r")
+	a.Sender = sender
+	_, err := a.SearchIssues("[auto-failure:cron:Timeout]")
+	if err == nil {
+		t.Fatal("page-2 connect error must propagate")
+	}
+	if errorKind(err) != "httpx.ConnectError" {
+		t.Fatalf("frozen error vocabulary changed: %v", errorKind(err))
+	}
+	if len(*sender.requests) != 4 {
+		t.Fatalf("expected 4 requests (page 1 + 3 page-2 attempts), got %d", len(*sender.requests))
+	}
+}
+
+// TestPhaseBPipelineOverlapsFilter — the pipeline is not decorative: while
+// the consumer filters page N, the producer is already fetching page N+1.
+// Proven by a sender that blocks in-flight until the consumer has begun
+// filtering the previous page (unbuffered channel rendezvous).
+func TestPhaseBPipelineOverlapsFilter(t *testing.T) {
+	pageFetchStarted := make(chan int, 4)
+	sender := &countingSender{scriptedSender: scriptedSender{
+		entries: []scriptEntry{
+			{Status: 200, Payload: fullPage()},
+			{Status: 200, Payload: []any{}},
+		},
+		requests: &[]any{},
+	}, started: pageFetchStarted}
+	a := NewGiteaForgejoAdapter("https://forge.example.com", "tok", "o", "r")
+	a.Sender = sender
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		if _, err := a.SearchIssues("x"); err != nil {
+			t.Errorf("search failed: %v", err)
+		}
+	}()
+	// page 1 fetch starts immediately
+	<-pageFetchStarted
+	// page 2 fetch must start while page 1 is still being consumed —
+	// wait for it with a generous timeout instead of deadlocking.
+	select {
+	case <-pageFetchStarted:
+	case <-time.After(5 * time.Second):
+		t.Fatal("producer did not overlap: page 2 not fetched while page 1 was being consumed")
+	}
+	<-done
+}
+
+func fullPage() []any {
+	p := make([]any, SearchPageSize)
+	for i := range p {
+		p[i] = map[string]any{"number": i, "title": "x", "body": ""}
+	}
+	return p
+}
+
+// fullPageWith — a full 50-item page whose item #markerAt carries the
+// search marker (forcing pagination exactly like the live forge).
+func fullPageWith(markerAt int) []any {
+	p := fullPage()
+	p[markerAt] = markerItem(markerAt, true)
+	return p
+}
+
+// countingSender counts started requests so the test can observe the
+// producer/consumer overlap.
+type countingSender struct {
+	scriptedSender
+	started chan int
+	n       int
+}
+
+func (s *countingSender) record(method, url string, headers map[string]string, jsonBody any, timeout time.Duration) *scriptEntry {
+	s.n++
+	select {
+	case s.started <- s.n:
+	default:
+	}
+	return s.scriptedSender.record(method, url, headers, jsonBody, timeout)
+}
+
+func (s *countingSender) Post(url string, headers map[string]string, jsonBody any, timeout time.Duration) (*Response, error) {
+	return s.respond(s.record("POST", url, headers, jsonBody, timeout))
+}
+
+func (s *countingSender) Get(url string, headers map[string]string, timeout time.Duration) (*Response, error) {
+	return s.respond(s.record("GET", url, headers, nil, timeout))
+}