[Bug + tested fix] dsh-superagent-prototype: GitHub observation stalls after 100 same-time events

[ericcaiwx-star]

The repository-wide observation streams in deepseek-ai/dsh-superagent-prototype read only their first page. More than 100 items sharing an update timestamp can keep later items unread while the Source reports a successful poll.

Posting here because the prototype repository currently has Issues/Discussions disabled and its PR endpoint returns 404. Harness's contribution guide directs bug reports to Discussions; #11700 already reports prototype defects here.

Baseline: 6337fc823ce20251c8de8085948165c81aaa701e. This concerns readStream, not #11700's delayed-notification cursor fix.

Reproduction and cause

Enable observeIssues/observePulls or observeComments, with no notification or subscribed-thread path supplying the missing events. Start with a stream cursor before T. Return 201 valid items updated at T, across pages of 100, 100 and 1.

readStream requests per_page=100 with ascending update order, ignores Link: ... rel="next", and advances to the first page's newest update minus five seconds. Subsequent polls see those same first 100 items. Once the cursor stops moving, the first-page ETag can yield 304 without the Source ever reading later pages.

The offline regression observes 100 of 201 items on the baseline in each of the issues/PR, conversation-comment and inline-comment streams. Other intake paths can sometimes recover an event; this report isolates repository observation rather than claiming every missing item is lost by the entire system.

Fix

Three files, +115/−12 lines, no dependency/configuration-setting changes:

• When a response has another page, reuse the existing repository-scoped Commands.pages helper to read the complete list before advancing the cursor.
• Cache an ETag only for a short, single-page response. Exactly 100 items must remain unconditional: a new page can appear while the first 100 items stay unchanged.
• Mark newly verified cache entries with singlePage; legacy ETags are refreshed unconditionally once because they may represent an incomplete first page.
• Keep the cursor unchanged after later-page failure or cancellation; the existing Trigger IDs deduplicate overlap/retry delivery.

The multi-page case repeats the first request through the existing pagination helper. This keeps the patch small and consistent with notification polling; it costs one extra request on paginated polls. No arbitrary Link target is executed by this code.

Verification

Mac Studio / macOS arm64, Node 26.7.0; dependencies installed with pnpm 11.7.0, --frozen-lockfile --ignore-scripts. Empty test HOME/configuration, no real account credentials or paid models. The test processes ran with external networking denied; existing HTTP tests used local fixtures.

• Seven new focused cases fail on the baseline and pass with the patch: three streams with 201 same-time items and duplicate-safe repeated polls; legacy ETag; full-page growth to item 101; later-page failure and cancellation/retry.
• Typecheck: pass.
• Full Vitest suite: 59 files, 1204 tests passed.
• Build: pass.
• Independent read-only review: no blocking findings in the changed scope.
• Attached patch applies cleanly to the baseline (git apply --check --whitespace=error-all).

The check script's three commands were run through their locked local executables (tsc --noEmit, vitest run, tsdown); the pnpm version launcher was stopped before checks began when it waited under the offline sandbox. The new tests mock the CLI transport while exercising pollRound, Engine and SQLite; they do not claim a live GitHub reproduction or real gh process cancellation proof.

Patch

Local commit 64a4efe2477403dec00db6e3bb3bd729fb7413fb, Eric Cai <ericcaiwx@gmail.com>. The complete mail-format patch is included below so no fork is required. Save the code block as github-stream-pagination.patch; it supports git am or inspection with git apply --check against the baseline. MIT terms remain unchanged.

Complete patch: fix(github): read all observed repository stream pages

From 64a4efe2477403dec00db6e3bb3bd729fb7413fb Mon Sep 17 00:00:00 2001
From: Eric Cai <ericcaiwx@gmail.com>
Date: Sun, 1 Nov 2026 20:56:12 +0800
Subject: [PATCH] fix(github): read all observed repository stream pages

---
 src/kv.ts             |   4 +-
 src/sources/github.ts |  21 +++++----
 tests/sources.spec.ts | 102 +++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 115 insertions(+), 12 deletions(-)

diff --git a/src/kv.ts b/src/kv.ts
index e7ed695..7fd8cd0 100644
--- a/src/kv.ts
+++ b/src/kv.ts
@@ -40,12 +40,12 @@ export const KV = {
   sources: { labels: family<Record<string, string>, [source: string]>('source-labels:') },
   /**
    * GitHub Source: notification cursor and pending retries; per-Issue/PR read cursor shared by notification and polled reads;
-   * per-stream cursor, start times by kind, and ETag; default-branch CI state; and per-PR CI fingerprint and revision (`revision`
+   * per-stream cursor, start times by kind, and verified short single-page ETag; default-branch CI state; and per-PR CI fingerprint and revision (`revision`
    * counts reported CI changes so each produces a distinct Trigger ID).
    */
   github: { since: record<string>('github-since'), retries: record<Record<string, Retry>>('github-retry'),
     cursor: family<string, [number: number]>('github-cursor:'),
-    stream: family<{ since: string; started: Record<string, string>; etag?: string }, [stream: string]>('github-stream:'),
+    stream: family<{ since: string; started: Record<string, string>; etag?: string; singlePage?: boolean }, [stream: string]>('github-stream:'),
     branchCi: record<{ branch?: string; failures?: string[]; revision: number }>('github-branch-ci'),
     ci: family<{ fingerprint: string; revision: number }, [number: number]>('github-ci:') },
   /**
diff --git a/src/sources/github.ts b/src/sources/github.ts
index f9b80b6..9ae7946 100644
--- a/src/sources/github.ts
+++ b/src/sources/github.ts
@@ -195,10 +195,9 @@ async function pollRepository(ctx: SourceContext, round: Round): Promise<void> {
   catch (error) { if (!stopped(ctx, round)) round.failures.push(`default branch CI: ${String(error)}`) }
 }
 /**
- * Polls one repository-wide event stream (oldest first) since its cursor and steps the cursor to five seconds before the
- * newest item (`cursorAt`), relying on Trigger ID deduplication so repeated messages are no-ops. While the cursor stays
- * unchanged (including when clamped to the hourly horizon), sending the last page's ETag makes an unchanged list return
- * HTTP 304 and cost no rate limit. Items from before their kind was switched on or beyond the horizon are ignored.
+ * Reads every page of a repository-wide stream before advancing its overlap-safe cursor. Trigger IDs deduplicate retries.
+ * Only a short single-page response can be cached: a full first page may stay unchanged while later pages gain items.
+ * Items from before their kind was switched on or beyond the horizon are ignored.
  */
 async function readStream(ctx: SourceContext, round: Round, id: string, path: string, config: GithubConfig): Promise<void> {
   const name = KV.github.stream(id)
@@ -208,7 +207,7 @@ async function readStream(ctx: SourceContext, round: Round, id: string, path: st
   const started: Record<string, string> = Object.fromEntries(kinds.map(kind => [kind, before[kind] ?? round.started]))
   const horizon = Math.floor((Date.parse(round.started) - POLL_HORIZON_MS) / 3_600_000) * 3_600_000
   const since = new Date(Math.max(Date.parse(stored?.since ?? round.started), horizon)).toISOString()
-  const conditional = stored?.etag && stored.since === since ? ['-H', `If-None-Match: ${stored.etag}`] : []
+  const conditional = stored?.singlePage && stored.etag && stored.since === since ? ['-H', `If-None-Match: ${stored.etag}`] : []
   let raw: string
   const listed = `${path}&sort=updated&direction=asc&per_page=100&since=${encodeURIComponent(since)}`
   try { raw = await ctx.commands.withSignal(round.signal).api(listed, '--include', ...conditional) }
@@ -217,7 +216,10 @@ async function readStream(ctx: SourceContext, round: Round, id: string, path: st
     throw error
   }
   const { headers, body } = included(raw)
-  const items = validItems(ListedSchema, z.array(z.unknown()).parse(JSON.parse(body)), round.tally)
+  const first = z.array(z.unknown()).parse(JSON.parse(body))
+  const more = /rel="next"/.test(headers)
+  const items = more ? await ctx.commands.withSignal(round.signal).pages(ListedSchema, listed, round.tally)
+    : validItems(ListedSchema, first, round.tally)
   const { login } = ctx.deployment.github
   const fresh = (at: string | null | undefined, kind: string) =>
     kind in started && Date.parse(at ?? '') >= Math.max(Date.parse(started[kind]!), horizon)
@@ -240,12 +242,13 @@ async function readStream(ctx: SourceContext, round: Round, id: string, path: st
     const trigger = commentTrigger(item, ref, id === 'comments' ? 'conversation' : 'inline', login)
     if (trigger.text) await ctx.receive(trigger)
   }
+  if (stopped(ctx, round)) throw new Error('The GitHub Source stopped during the poll')
   const newest = Math.max(...items.map(item => Date.parse(item.updated_at ?? '')).filter(Number.isFinite))
   const moved = Number.isFinite(newest) ? cursorAt(new Date(newest).toISOString()) : since
   const next = Date.parse(moved) > Date.parse(since) ? moved : since
-  const etag = /^etag:\s*(.+)$/im.exec(headers)?.[1]?.trim()
-  // The ETag belongs to this request, which the next round repeats only while the cursor stays unchanged.
-  ctx.state.set(name, { since: next, started, ...etag && next === since ? { etag } : {} })
+  const etag = !more && first.length < 100 ? /^etag:\s*(.+)$/im.exec(headers)?.[1]?.trim() : undefined
+  // Legacy ETags lack this marker and may describe an incomplete first page; refresh them unconditionally once.
+  ctx.state.set(name, { since: next, started, ...etag && next === since ? { etag, singlePage: true } : {} })
 }
 /** Polls default-branch CI checks, emitting a trigger only when the set of failing checks changes after the initial baseline. */
 async function pollBranchCi(ctx: SourceContext, round: Round): Promise<void> {
diff --git a/tests/sources.spec.ts b/tests/sources.spec.ts
index f4d0332..c7eba52 100644
--- a/tests/sources.spec.ts
+++ b/tests/sources.spec.ts
@@ -1233,7 +1233,7 @@ describe('observing the GitHub repository', () => {
     expect(list()[0]![1]).toContain(`since=${encodeURIComponent(at(0))}`)
     // The next round repeats the request at the newest Issue, whose ETag the round after sends, which GitHub answers 304 for.
     await pollRound(h)
-    expect(h.store.get('github-stream:issues')).toEqual({ since: cursorAt(at(4)), started, etag: '"same"' })
+    expect(h.store.get('github-stream:issues')).toEqual({ since: cursorAt(at(4)), started, etag: '"same"', singlePage: true })
     h.receive.mockClear()
     await pollRound(h)
     expect(list().at(-1)).toEqual(expect.arrayContaining(['-H', 'If-None-Match: "same"']))
@@ -1317,6 +1317,106 @@ describe('observing the GitHub repository', () => {
     expect(h.store.get('github-branch-ci')).toEqual({ branch: 'main', failures: ['lint'], revision: 2 })
     h.store.close()
   })
+  /** A paginated CLI response: normal calls return one HTTP page; --paginate --slurp returns all JSON pages. */
+  function streamPages(h: ReturnType<typeof host>, stream: string, pages: unknown[][], beforeRead = (_page: number) => {}) {
+    const calls = repository(h, {})
+    vi.spyOn(h.commands, 'withSignal').mockReturnValue(h.commands)
+    const fallback = vi.mocked(h.commands.run).getMockImplementation()!
+    vi.mocked(h.commands.run).mockImplementation(async (program, args) => {
+      const url = new URL(args[1]!, 'https://api.github.com/')
+      if (url.pathname !== `/repos/acme/widgets/${stream}`) return fallback(program, args)
+      calls.push(args)
+      const page = Number(url.searchParams.get('page') ?? 1)
+      if (args.includes('--paginate')) {
+        pages.forEach((_items, index) => { beforeRead(index + 1) })
+        return JSON.stringify(pages)
+      }
+      beforeRead(page)
+      if (args.includes('If-None-Match: "same"')) throw refused(304, 'Not Modified')
+      const link = page < pages.length ? `Link: <https://api.github.com${url.pathname}?page=${page + 1}>; rel="next"\r\n` : ''
+      return `HTTP/2.0 200 OK\r\nETag: "same"\r\n${link}\r\n${JSON.stringify(pages[page - 1] ?? [])}`
+    })
+    return calls
+  }
+  it.each(['issues', 'comments', 'review-comments'])('reads every page of the %s stream when more than 100 items share a timestamp', async stream => {
+    const h = host()
+    try {
+      observe(h, { observeIssues: true, observePulls: true, observeComments: true })
+      h.store.set('github-since', at(-24))
+      h.store.set(`github-stream:${stream}`, { since: at(0), started: { issues: at(0), pulls: at(0), comments: at(0) } })
+      const path = stream === 'issues' ? 'issues' : stream === 'comments' ? 'issues/comments' : 'pulls/comments'
+      const items = Array.from({ length: 201 }, (_, i) => stream === 'issues' ? issue(i + 1, { created_at: at(1), updated_at: at(1) })
+        : { id: i + 1, body: `comment ${i + 1}`, user: { login: 'alice', type: 'User' }, created_at: at(1), updated_at: at(1),
+          issue_url: 'https://api.github.com/repos/acme/widgets/issues/7', pull_request_url: 'https://api.github.com/repos/acme/widgets/pulls/7',
+          html_url: `https://github.com/acme/widgets/pull/7#comment-${i + 1}` })
+      const pages = [items.slice(0, 100), items.slice(100, 200), items.slice(200)]
+      const calls = streamPages(h, path, pages)
+      await pollRound(h)
+      expect(new Set(h.receive.mock.calls.map(([trigger]) => trigger.id)).size).toBe(201)
+      const ids = h.receive.mock.calls.map(([trigger]) => trigger.id)
+      // Re-reading overlapping pages produces no additional domain effects, even once the cursor stops moving.
+      for (let round = 0; round < 2; round++) {
+        h.receive.mockClear()
+        await pollRound(h)
+        expect(h.receive.mock.results.map(result => result.value)).toHaveLength(201)
+        expect(await Promise.all(h.receive.mock.results.map(result => result.value))).toEqual(Array(201).fill('duplicate'))
+        expect(h.receive.mock.calls.map(([trigger]) => trigger.id)).toEqual(ids)
+      }
+      expect(calls.filter(args => args[1]!.startsWith(`repos/acme/widgets/${path}?`)).every(args =>
+        !args.includes('If-None-Match: "same"'))).toBe(true)
+    } finally { h.store.close() }
+  })
+  it('does not reuse a legacy stream ETag which may describe only the first page', async () => {
+    const h = host()
+    try {
+      observe(h, { observeIssues: true })
+      h.store.set('github-since', at(-24))
+      h.store.set('github-stream:issues', { since: at(0), started: { issues: at(0) }, etag: '"same"' })
+      const items = Array.from({ length: 101 }, (_, i) => issue(i + 1, { created_at: at(1), updated_at: at(1) }))
+      const calls = streamPages(h, 'issues', [items.slice(0, 100), items.slice(100)])
+      await pollRound(h)
+      expect(h.receive.mock.calls.map(([trigger]) => trigger.meta.number)).toEqual(items.map(item => item.number))
+      expect(calls.every(args => !args.includes('If-None-Match: "same"'))).toBe(true)
+    } finally { h.store.close() }
+  })
+  it('does not cache a full last page which can later grow another page without changing its first 100 items', async () => {
+    const h = host()
+    try {
+      observe(h, { observeIssues: true })
+      h.store.set('github-since', at(-24))
+      h.store.set('github-stream:issues', { since: at(0), started: { issues: at(0) } })
+      const pages = [Array.from({ length: 100 }, (_, i) => issue(i + 1, { created_at: at(1), updated_at: at(1) }))]
+      streamPages(h, 'issues', pages)
+      await pollRound(h); await pollRound(h)
+      pages.push([issue(101, { created_at: at(1), updated_at: at(1) })])
+      h.receive.mockClear()
+      await pollRound(h)
+      expect(h.receive.mock.calls.map(([trigger]) => trigger.meta.number)).toContain(101)
+    } finally { h.store.close() }
+  })
+  it.each(['failure', 'abort'])('keeps a stream cursor unchanged after a later-page %s and retries its missing items', async mode => {
+    const h = host()
+    try {
+      observe(h, { observeIssues: true })
+      h.store.set('github-since', at(-24))
+      const original = { since: at(0), started: { issues: at(0) } }
+      h.store.set('github-stream:issues', original)
+      const signal = new AbortController()
+      let interrupt = true
+      const items = Array.from({ length: 101 }, (_, i) => issue(i + 1, { created_at: at(1), updated_at: at(1) }))
+      streamPages(h, 'issues', [items.slice(0, 100), items.slice(100)], page => {
+        if (page !== 2 || !interrupt) return
+        if (mode === 'abort') signal.abort()
+        else throw refused(502, 'Bad Gateway on page 2')
+      })
+      if (mode === 'failure') await expect(pollRound(h, signal.signal)).rejects.toThrow('Bad Gateway on page 2')
+      else await pollRound(h, signal.signal)
+      expect(h.store.get('github-stream:issues')).toEqual(original)
+      interrupt = false
+      await pollRound(h)
+      expect(new Set(h.receive.mock.calls.map(([trigger]) => trigger.meta.number))).toEqual(new Set(items.map(item => item.number)))
+    } finally { h.store.close() }
+  })
   it('takes in new PRs only from when their switch turned on, apart from when new Issues were', async () => {
     const h = host()
     observe(h, { observeIssues: true, observePulls: true })