diff --git a/migrations/0137_x_discovery_continuation.sql b/migrations/0137_x_discovery_continuation.sql new file mode 100644 index 00000000..a27b59d7 --- /dev/null +++ b/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) +); diff --git a/shared/checkin-external.ts b/shared/checkin-external.ts index 0fec5d3d..584eac6e 100644 --- a/shared/checkin-external.ts +++ b/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/ 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/ 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', diff --git a/shared/server/x-ingest.ts b/shared/server/x-ingest.ts index a499a3e0..c7f9ea45 100644 --- a/shared/server/x-ingest.ts +++ b/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 { 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>) { const sinceIds: Array = [] const maxResults: Array = [] + const nextTokens: Array = [] 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>) { 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