fix(x-ingest): resume unread pages across discovery runs

migrations/0137_x_discovery_continuation.sql

@@ -0,0 +1,7 @@
+-- A bounded run resumes the unread tail instead of buying page one again.
+ALTER TABLE x_ingest_cursor ADD COLUMN pending_newest_id TEXT;
+ALTER TABLE x_ingest_cursor ADD COLUMN pending_started_at INTEGER;
+ALTER TABLE x_ingest_cursor ADD COLUMN next_token TEXT CHECK (
+  (next_token IS NULL AND pending_newest_id IS NULL AND pending_started_at IS NULL)
+  OR (next_token IS NOT NULL AND pending_started_at IS NOT NULL)
+);

shared/checkin-external.ts

@@ -22,15 +22,16 @@ Interpret this site evidence:
     },
     {
       id: 'skilld.spend',
-      prompt: `- X API is pay-per-use at $0.005 per post read, with no included allowance and a $20/month target. Accepted 2026-08-31 at a projected $18.15/month, which raised the target from $10. Gate on \`d1.cost\`:
-     - \`x_over_budget\` true is AMBER. The daily discovery budget failed to hold and the month will overrun.
+      prompt: `- X API is pay-per-use at $0.005 per post read, with no included allowance and a $40/month accepted line. Gate on \`d1.cost\`:
+     - \`x_over_budget\` true is AMBER. The daily discovery read ceiling failed to hold. Investigate before inferring billed spend.
      - \`x_projected_monthly_usd\` above 40 is AMBER, above 60 is RED, even when \`x_over_budget\` is false. Harlan raised the accepted line from $30 to $40 on 2026-09-07 (issue #160). Report the figure every run, not only when it breaches.
-     - \`x_budget_used_pct\` at 100 on consecutive days means discovery is truncating. That is the designed steady state, not a fault. Say so rather than reporting it as a failure.
-     - \`x_discovery_reads_today\` is the billed number. \`x_posts_24h\` counts only posts that survived repo extraction, so a large gap between them means the search query is paying for posts it discards. Flag a gap above 50% as a query-tuning action.`,
+     - \`x_budget_used_pct\` at 100 on consecutive days means discovery is truncating. Check cursor progress and repeated pages before calling it steady state.
+     - \`x_discovery_reads_today\` counts returned posts, including repeats. X normally charges once per resource per UTC day. The monthly projection extrapolates today's returns plus hot posts for 30 days. Report it as an upper-bound estimate, never billed spend. A catch-up burst does not establish the steady rate.
+     - \`x_posts_24h\` counts newly stored posts, including posts without repo references. A large gap from returned reads can indicate replay. Check continuation and cursor progress before proposing query tuning.`,
     },
     {
       id: 'skilld.pulse',
-      prompt: `Include a Pulse section with users, skills, repo changes, command copies split as run and install, digests, known AI cost, and X spend as "$X.XX/mo projected (N/<x_budget_target> reads today)". Read these values from collected evidence. Include deploy and CI changes in Overnight. Include migration, schema, deploy SHA, stale task, and observability gaps in Drift.`,
+      prompt: `Include a Pulse section with users, skills, repo changes, command copies split as run and install, digests, known AI cost, and X as "$X.XX/mo upper-bound estimate (N/<x_budget_target> returned reads today)". Read these values from collected evidence. Include deploy and CI changes in Overnight. Include migration, schema, deploy SHA, stale task, and observability gaps in Drift.`,
     },
     {
       id: 'skilld.seo-recovery',

shared/server/x-ingest.ts

@@ -90,26 +90,13 @@ export const CURSOR_STALE_MARGIN_SECONDS = 24 * 60 * 60
 const MAX_PAGES_PER_RUN = 5
 
 /**
- * Charged post reads discovery may spend per UTC day.
+ * Returned post reads discovery may consume per UTC day.
  *
- * A CEILING, NOT A SPEND. Only posts the query actually matches are charged,
- * so raising this does not raise the bill on a quiet day; it only stops a
- * runaway.
- *
- * RAISED FROM 400 BECAUSE 400 WAS THE BINDING CONSTRAINT, NOT THE COST.
- * Measured against the live project on 2026-08-14: `project_cap` is 3,000,000
- * posts/month and `project_usage` was 798. The old ceiling allowed 12,000 a
- * month, which is 0.4% of a cap that was already bought, while the corpus it
- * produced held 69 posts and topped out at 31 likes. 2,000/day is 60,000 a
- * month, still only 2% of the cap, and leaves room for the broadened query to
- * find the posts the old one could not see.
- *
- * IT MUST EXCEED THE STREAM RATE. The cursor now holds position when a run is
- * cut short, so a small budget no longer discards posts, but it does make the
- * ingest fall behind by the shortfall every day, and that lag never recovers.
- * At the previous 22/day against a ~250/day stream the backlog grew by more
- * than 200 posts daily and the feed was permanently a week stale. 400 leaves
- * roughly 60% headroom over the observed rate for spikes.
+ * This is a request ceiling, not a prepaid allowance or billed usage.
+ * X normally deduplicates resource charges within each UTC day. Returned
+ * reads still consume our ceiling, including repeats after cursor recovery.
+ * Bounded runs retain their continuation so the ceiling pays for the unread
+ * tail before discovery polls for newer posts.
  */
 export const DAILY_DISCOVERY_READ_BUDGET = 2000
 
@@ -186,42 +173,51 @@ function postUrl(post: XPost): string {
   return `https://x.com/${post.authorHandle}/status/${post.id}`
 }
 
+type SearchWindow = { _tag: 'idle' } | { _tag: 'paused', nextToken: string, newestId: string | null, startedAt: number }
+
 interface StoredCursor {
   /** The id to search from, or null for a cold start. */
   sinceId: string | null
   /** True when a stored cursor was dropped for predating the retention window. */
   staleReset: boolean
+  window: SearchWindow
 }
 
 async function readCursor(db: D1Database, key: string, now: number): Promise<StoredCursor> {
   const row = await db
-    .prepare(`SELECT since_id, since_id_written_at FROM x_ingest_cursor WHERE query_key = ?1`)
+    .prepare(`SELECT since_id, since_id_written_at, next_token, pending_newest_id, pending_started_at FROM x_ingest_cursor WHERE query_key = ?1`)
     .bind(key)
-    .first<{ since_id: string | null, since_id_written_at: number | null }>()
+    .first<{ since_id: string | null, since_id_written_at: number | null, next_token: string | null, pending_newest_id: string | null, pending_started_at: number | null }>()
   const sinceId = row?.since_id ?? null
   const writtenAt = row?.since_id_written_at ?? null
-  if (!sinceId)
-    return { sinceId: null, staleReset: false }
+  const window: SearchWindow = row?.next_token && row.pending_started_at !== null
+    ? { _tag: 'paused', nextToken: row.next_token, newestId: row.pending_newest_id, startedAt: row.pending_started_at }
+    : { _tag: 'idle' }
+  if (!sinceId && window._tag === 'idle')
+    return { sinceId: null, staleReset: false, window }
 
   // `since_id_written_at` dates the cursor by the run that wrote it. Dating it
   // by `last_run_at` instead was the bug this column fixed: held and failing
   // runs refresh `last_run_at` while the COALESCE keeps the old id, so a dead
   // id read as fresh forever on the 15-minute schedule. A cursor with no
   // timestamp cannot be dated and is dropped with the rest.
-  const fresh = writtenAt !== null
-    && now - writtenAt <= X_SEARCH_RETENTION_SECONDS - CURSOR_STALE_MARGIN_SECONDS
+  const datedAt = window._tag === 'paused'
+    ? (writtenAt === null ? window.startedAt : Math.min(writtenAt, window.startedAt))
+    : writtenAt
+  const fresh = datedAt !== null
+    && now - datedAt <= X_SEARCH_RETENTION_SECONDS - CURSOR_STALE_MARGIN_SECONDS
   if (fresh)
-    return { sinceId, staleReset: false }
+    return { sinceId, staleReset: false, window }
 
   // Cleared here, not only in the end-of-run upsert: that upsert COALESCEs a
   // held (null) advance into the stored value, so a run that later fails or
   // truncates would otherwise hand the same dead cursor to the next run and
   // the 400s would never stop.
   await db
-    .prepare(`UPDATE x_ingest_cursor SET since_id = NULL, since_id_written_at = NULL WHERE query_key = ?1`)
+    .prepare(`UPDATE x_ingest_cursor SET since_id = NULL, since_id_written_at = NULL, next_token = NULL, pending_newest_id = NULL, pending_started_at = NULL WHERE query_key = ?1`)
     .bind(key)
     .run()
-  return { sinceId: null, staleReset: true }
+  return { sinceId: null, staleReset: true, window: { _tag: 'idle' } }
 }
 
 /**
@@ -244,6 +240,8 @@ async function writeCursor(
   key: string,
   input: {
     sinceId: string | null
+    cursorWrittenAt: number
+    window: SearchWindow
     now: number
     resultCount: number
     postsRead: number
@@ -255,9 +253,10 @@ async function writeCursor(
     .prepare(
       `INSERT INTO x_ingest_cursor (
          query_key, since_id, since_id_written_at, last_run_at, last_result_count,
-         posts_read_total, budget_day, budget_spent
+         posts_read_total, budget_day, budget_spent,
+         next_token, pending_newest_id, pending_started_at
        )
-       VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
+       VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
        ON CONFLICT (query_key) DO UPDATE SET
          since_id = COALESCE(excluded.since_id, x_ingest_cursor.since_id),
          -- The timestamp must move only when the id does: a held advance keeps
@@ -273,17 +272,23 @@ async function writeCursor(
          -- Absolute, not additive: the caller already folded in what today had
          -- spent before this run, and rolled it to 0 on a new UTC day.
          budget_day = excluded.budget_day,
-         budget_spent = excluded.budget_spent`,
+         budget_spent = excluded.budget_spent,
+         next_token = excluded.next_token,
+         pending_newest_id = excluded.pending_newest_id,
+         pending_started_at = excluded.pending_started_at`,
     )
     .bind(
       key,
       input.sinceId,
-      input.sinceId === null ? null : input.now,
+      input.sinceId === null ? null : input.cursorWrittenAt,
       input.now,
       input.resultCount,
       input.postsRead,
       input.budgetDay,
       input.budgetSpent,
+      input.window._tag === 'paused' ? input.window.nextToken : null,
+      input.window._tag === 'paused' ? input.window.newestId : null,
+      input.window._tag === 'paused' ? input.window.startedAt : null,
     )
     .run()
 }
@@ -408,8 +413,9 @@ export async function ingestXMentions(deps: XIngestDeps): Promise<XIngestSummary
   let spent = spentBefore
 
   const collected: XPost[] = []
-  let nextToken: string | null = null
-  let newestId: string | null = null
+  let nextToken = cursor.window._tag === 'paused' ? cursor.window.nextToken : null
+  let newestId = cursor.window._tag === 'paused' ? cursor.window.newestId : null
+  const windowStartedAt = cursor.window._tag === 'paused' ? cursor.window.startedAt : now
   // Only true once pagination reached the end of the window, meaning every
   // post newer than the cursor has been seen.
   let windowExhausted = false
@@ -525,12 +531,16 @@ export async function ingestXMentions(deps: XIngestDeps): Promise<XIngestSummary
   // the first run after UTC midnight, that discarded almost a full day of the
   // stream, every day, and biased what survived towards the youngest posts.
   //
-  // Holding the cursor makes the next run re-paginate from the same point.
-  // Same-day re-reads are deduplicated by X, so resuming is free until
-  // midnight; only a window still unfinished at the rollover costs anything.
+  // Keep the unread page and the first page's newest id together. The next
+  // run resumes the tail, then advances to that original id. Restarting page
+  // one on every run exhausted the budget without ever reaching the tail.
   const advanceTo = windowExhausted ? newestId : null
   await writeCursor(db, X_DISCOVERY_CURSOR_KEY, {
     sinceId: advanceTo,
+    cursorWrittenAt: windowStartedAt,
+    window: nextToken === null
+      ? { _tag: 'idle' }
+      : { _tag: 'paused', nextToken, newestId, startedAt: windowStartedAt },
     now,
     resultCount: collected.length,
     postsRead: summary.postsRead,

test/unit/x-ingest.test.ts

@@ -25,6 +25,7 @@ const MIGRATIONS = [
   // Dates `since_id` by the run that wrote it, so the retention drop fires
   // even when later runs keep finishing.
   'migrations/0134_x_cursor_written_at.sql',
+  'migrations/0137_x_discovery_continuation.sql',
 ]
 const NOW = 1_760_000_000
 
@@ -58,10 +59,12 @@ function page(posts: XPost[], nextToken: string | null = null): XPage {
 function stubClient(pages: Array<XResult<XPage>>) {
   const sinceIds: Array<string | null> = []
   const maxResults: Array<number | undefined> = []
+  const nextTokens: Array<string | null | undefined> = []
   let call = 0
   const client: XClient = {
-    async searchRecent({ sinceId, maxResults: max }) {
+    async searchRecent({ sinceId, nextToken, maxResults: max }) {
       sinceIds.push(sinceId)
+      nextTokens.push(nextToken)
       maxResults.push(max)
       return pages[call++] ?? { _tag: 'ok', value: page([]) }
     },
@@ -72,7 +75,7 @@ function stubClient(pages: Array<XResult<XPage>>) {
       throw new Error('usage is not part of discovery')
     },
   }
-  return { client, sinceIds, maxResults, callCount: () => call }
+  return { client, sinceIds, nextTokens, maxResults, callCount: () => call }
 }
 
 let harness: SqliteD1 | null = null
@@ -462,6 +465,63 @@ describe('ingestXMentions failure handling', () => {
 })
 
 describe('ingestXMentions cursor safety', () => {
+  it('resumes a truncated window without rereading its first pages', async () => {
+    const first = stubClient(Array.from({ length: 5 }, (_, i) => ({
+      _tag: 'ok' as const,
+      value: page([post({ id: String(9000 - i) })], `token-${i + 1}`),
+    })))
+    await ingestXMentions({ db: db().db, client: first.client, now: NOW })
+
+    const resumed = stubClient([{ _tag: 'ok', value: page([post({ id: '8000' })]) }])
+    const summary = await ingestXMentions({ db: db().db, client: resumed.client, now: NOW + 900 })
+    expect(resumed.nextTokens).toEqual(['token-5'])
+    expect(resumed.sinceIds).toEqual([null])
+    expect(summary.cursorAdvancedTo).toBe('9000')
+
+    const polling = stubClient([{ _tag: 'ok', value: page([]) }])
+    await ingestXMentions({ db: db().db, client: polling.client, now: NOW + 1800 })
+    expect(polling.sinceIds).toEqual(['9000'])
+    expect(polling.nextTokens).toEqual([null])
+  })
+
+  it('keeps its next page through a failed request and a UTC budget rollover', async () => {
+    const first = stubClient([{
+      _tag: 'ok',
+      value: { ...page([post({ id: '9000' })], 'remaining'), postsRead: DAILY_DISCOVERY_READ_BUDGET },
+    }])
+    await ingestXMentions({ db: db().db, client: first.client, now: NOW })
+    const blocked = stubClient([])
+    await ingestXMentions({ db: db().db, client: blocked.client, now: NOW + 60 })
+    expect(blocked.callCount()).toBe(0)
+
+    const failing = stubClient([{ _tag: 'err', error: { _tag: 'rate-limited', resetAt: NOW + 86460 } }])
+    await ingestXMentions({ db: db().db, client: failing.client, now: NOW + 86400 })
+    expect(failing.nextTokens).toEqual(['remaining'])
+
+    const resumed = stubClient([{ _tag: 'ok', value: page([post({ id: '8000' })]) }])
+    const summary = await ingestXMentions({ db: db().db, client: resumed.client, now: NOW + 87300 })
+    expect(resumed.nextTokens).toEqual(['remaining'])
+    expect(summary.cursorAdvancedTo).toBe('9000')
+    expect(summary.budgetSpentToday).toBe(1)
+  })
+
+  it('expires a paused cold start before its search window leaves retention', async () => {
+    const first = stubClient(Array.from({ length: 5 }, (_, i) => ({
+      _tag: 'ok' as const,
+      value: page([post({ id: String(9000 - i) })], `token-${i + 1}`),
+    })))
+    await ingestXMentions({ db: db().db, client: first.client, now: NOW })
+    const next = stubClient([{ _tag: 'ok', value: page([post({ id: '10000' })]) }])
+    const summary = await ingestXMentions({
+      db: db().db,
+      client: next.client,
+      now: NOW + X_SEARCH_RETENTION_SECONDS - CURSOR_STALE_MARGIN_SECONDS + 1,
+    })
+    expect(summary.cursorReset).toBe(true)
+    expect(next.nextTokens).toEqual([null])
+    expect(summary.cursorAdvancedTo).toBe('10000')
+  })
+
   it('holds the cursor when the budget stops a run mid-window', async () => {
     // Advancing to the newest id here jumped past every post the run could not
     // afford, losing them permanently. Because the daily budget is spent by the