From ca4cf2a0c6a4e0a53f20c49cbe77a64ada01a73e Mon Sep 17 00:00:00 2001 From: RP-AGENT-BOT Date: Thu, 23 Jul 2026 19:12:57 -0500 Subject: [PATCH] fix(cycle): preserve per-page multi-claim proposals (#3297) Co-authored-by: David Guidry --- src/core/cycle/propose-takes.ts | 14 +++-- src/core/migrate.ts | 14 +++++ src/core/pglite-schema.ts | 2 +- src/core/schema-embedded.ts | 7 +-- src/schema.sql | 7 +-- test/propose-takes-per-claim.test.ts | 93 ++++++++++++++++++++++++++++ test/propose-takes.test.ts | 21 ++++++- 7 files changed, 142 insertions(+), 16 deletions(-) create mode 100644 test/propose-takes-per-claim.test.ts diff --git a/src/core/cycle/propose-takes.ts b/src/core/cycle/propose-takes.ts index 4fe1f1b2a..e6117fc02 100644 --- a/src/core/cycle/propose-takes.ts +++ b/src/core/cycle/propose-takes.ts @@ -484,16 +484,18 @@ class ProposeTakesPhase extends BaseCyclePhase { continue; } - // Write proposals to take_proposals. Each row is a separate INSERT - // because the composite idempotency key is on the per-page tuple — a - // bulk UPSERT would collapse a same-page-multi-claim run into one row. + // Write proposals to take_proposals. #2138: the idempotency key is + // per-CLAIM — take_proposals_idempotency_idx folds md5(claim_text) into + // the per-page tuple (migration v125), so a multi-claim page keeps every + // claim. RETURNING id prevents a repeated claim from inflating the count. for (const p of proposals) { - await engine.executeRaw( + const inserted = await engine.executeRaw<{ id: number }>( `INSERT INTO take_proposals (source_id, page_slug, content_hash, prompt_version, proposal_run_id, claim_text, kind, holder, weight, domain, dedup_against_fence_rows, model_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) - ON CONFLICT (source_id, page_slug, content_hash, prompt_version) DO NOTHING`, + ON CONFLICT (source_id, page_slug, content_hash, prompt_version, md5(claim_text)) DO NOTHING + RETURNING id`, [ sourceId, page.slug, @@ -509,7 +511,7 @@ class ProposeTakesPhase extends BaseCyclePhase { modelId, ], ); - result.proposals_inserted += 1; + result.proposals_inserted += inserted.length; } } diff --git a/src/core/migrate.ts b/src/core/migrate.ts index e0caafbe2..085368246 100644 --- a/src/core/migrate.ts +++ b/src/core/migrate.ts @@ -5710,6 +5710,20 @@ export const MIGRATIONS: Migration[] = [ `); }, }, + { + version: 125, + name: 'take_proposals_per_claim_idempotency', + // #2138: the original idempotency key was per page, so each INSERT after + // the first claim silently conflicted. md5(claim_text) makes it per claim + // without adding/backfilling a column. The new key is strictly finer than + // the old key and therefore preserves all existing rows. + idempotent: true, + sql: ` + DROP INDEX IF EXISTS take_proposals_idempotency_idx; + CREATE UNIQUE INDEX IF NOT EXISTS take_proposals_idempotency_idx + ON take_proposals (source_id, page_slug, content_hash, prompt_version, md5(claim_text)); + `, + }, ]; export const LATEST_VERSION = MIGRATIONS.length > 0 diff --git a/src/core/pglite-schema.ts b/src/core/pglite-schema.ts index a42837a7c..db1c762d5 100644 --- a/src/core/pglite-schema.ts +++ b/src/core/pglite-schema.ts @@ -777,7 +777,7 @@ CREATE TABLE IF NOT EXISTS take_proposals ( predicted_brier_bucket_n INTEGER ); CREATE UNIQUE INDEX IF NOT EXISTS take_proposals_idempotency_idx - ON take_proposals (source_id, page_slug, content_hash, prompt_version); + ON take_proposals (source_id, page_slug, content_hash, prompt_version, md5(claim_text)); CREATE INDEX IF NOT EXISTS take_proposals_pending_idx ON take_proposals (source_id, status, proposed_at DESC) WHERE status = 'pending'; diff --git a/src/core/schema-embedded.ts b/src/core/schema-embedded.ts index dd0362bd5..0e7e2c48e 100644 --- a/src/core/schema-embedded.ts +++ b/src/core/schema-embedded.ts @@ -1273,9 +1273,8 @@ CREATE INDEX IF NOT EXISTS calibration_profiles_published_idx ON calibration_profiles (source_id, published, holder) WHERE published = true; --- take_proposals: propose_takes phase queue. Idempotency cache via the --- composite unique index (source_id, page_slug, content_hash, prompt_version) --- mirrors v0.23 dream_verdicts. proposal_run_id supports --rollback by run. +-- take_proposals: per-claim idempotency via source/page/content/prompt plus +-- md5(claim_text). The old per-page key silently dropped claim #2+ (#2138). CREATE TABLE IF NOT EXISTS take_proposals ( id BIGSERIAL PRIMARY KEY, source_id TEXT NOT NULL REFERENCES sources(id) ON DELETE CASCADE, @@ -1301,7 +1300,7 @@ CREATE TABLE IF NOT EXISTS take_proposals ( predicted_brier_bucket_n INTEGER ); CREATE UNIQUE INDEX IF NOT EXISTS take_proposals_idempotency_idx - ON take_proposals (source_id, page_slug, content_hash, prompt_version); + ON take_proposals (source_id, page_slug, content_hash, prompt_version, md5(claim_text)); CREATE INDEX IF NOT EXISTS take_proposals_pending_idx ON take_proposals (source_id, status, proposed_at DESC) WHERE status = 'pending'; diff --git a/src/schema.sql b/src/schema.sql index 8c8faa224..084af8d7e 100644 --- a/src/schema.sql +++ b/src/schema.sql @@ -1269,9 +1269,8 @@ CREATE INDEX IF NOT EXISTS calibration_profiles_published_idx ON calibration_profiles (source_id, published, holder) WHERE published = true; --- take_proposals: propose_takes phase queue. Idempotency cache via the --- composite unique index (source_id, page_slug, content_hash, prompt_version) --- mirrors v0.23 dream_verdicts. proposal_run_id supports --rollback by run. +-- take_proposals: per-claim idempotency via source/page/content/prompt plus +-- md5(claim_text). The old per-page key silently dropped claim #2+ (#2138). CREATE TABLE IF NOT EXISTS take_proposals ( id BIGSERIAL PRIMARY KEY, source_id TEXT NOT NULL REFERENCES sources(id) ON DELETE CASCADE, @@ -1297,7 +1296,7 @@ CREATE TABLE IF NOT EXISTS take_proposals ( predicted_brier_bucket_n INTEGER ); CREATE UNIQUE INDEX IF NOT EXISTS take_proposals_idempotency_idx - ON take_proposals (source_id, page_slug, content_hash, prompt_version); + ON take_proposals (source_id, page_slug, content_hash, prompt_version, md5(claim_text)); CREATE INDEX IF NOT EXISTS take_proposals_pending_idx ON take_proposals (source_id, status, proposed_at DESC) WHERE status = 'pending'; diff --git a/test/propose-takes-per-claim.test.ts b/test/propose-takes-per-claim.test.ts new file mode 100644 index 000000000..79e551fb9 --- /dev/null +++ b/test/propose-takes-per-claim.test.ts @@ -0,0 +1,93 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, test } from 'bun:test'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; +import { + runPhaseProposeTakes, + type ProposeTakesExtractor, +} from '../src/core/cycle/propose-takes.ts'; +import { MIGRATIONS } from '../src/core/migrate.ts'; +import type { OperationContext } from '../src/core/operations.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); +}); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await resetPgliteState(engine); +}); + +function context(): OperationContext { + return { + engine, + config: {} as never, + logger: { info() {}, warn() {}, error() {} } as never, + dryRun: false, + remote: false, + sourceId: 'default', + }; +} + +async function countProposals(slug: string): Promise { + const rows = await engine.executeRaw<{ n: string }>( + `SELECT COUNT(*)::text AS n + FROM take_proposals + WHERE page_slug = $1 AND source_id = 'default'`, + [slug], + ); + return Number(rows[0]!.n); +} + +const proposals: ProposeTakesExtractor = async () => [ + { claim_text: 'Claim one', kind: 'take', holder: 'brain', weight: 0.6 }, + { claim_text: 'Claim two', kind: 'bet', holder: 'brain', weight: 0.8 }, + { claim_text: 'Claim one', kind: 'take', holder: 'brain', weight: 0.6 }, +]; + +async function putThesis(): Promise { + await engine.putPage('wiki/essays/thesis', { + title: 'thesis', + type: 'analysis' as never, + compiled_truth: 'Two strong claims live in this essay.', + frontmatter: {}, + timeline: '', + }); +} + +describe('#2138 per-claim proposal idempotency', () => { + test('keeps distinct claims, drops repeated claim, then page-cache hits', async () => { + await putThesis(); + const result = await runPhaseProposeTakes(context(), { extractor: proposals }); + expect((result.details as Record).proposals_inserted).toBe(2); + expect(await countProposals('wiki/essays/thesis')).toBe(2); + + const rerun = await runPhaseProposeTakes(context(), { extractor: proposals }); + expect((rerun.details as Record).cache_hits).toBe(1); + expect(await countProposals('wiki/essays/thesis')).toBe(2); + }); + + test('migration v125 replaces the old-shaped index', async () => { + await engine.executeRaw('DROP INDEX IF EXISTS take_proposals_idempotency_idx'); + await engine.executeRaw( + `CREATE INDEX take_proposals_idempotency_idx + ON take_proposals (source_id, page_slug, content_hash, prompt_version)`, + ); + const migration = MIGRATIONS.find((entry) => entry.version === 125); + expect(migration).toBeDefined(); + for (const statement of migration!.sql!.split(';').map(value => value.trim()).filter(Boolean)) { + await engine.executeRaw(statement); + } + + await putThesis(); + const result = await runPhaseProposeTakes(context(), { extractor: proposals }); + expect((result.details as Record).proposals_inserted).toBe(2); + expect(await countProposals('wiki/essays/thesis')).toBe(2); + }); +}); diff --git a/test/propose-takes.test.ts b/test/propose-takes.test.ts index fbe8deb26..17a2e1d47 100644 --- a/test/propose-takes.test.ts +++ b/test/propose-takes.test.ts @@ -68,7 +68,10 @@ function buildMockEngine(opts: { if (existing.has(key)) return [{ id: 1 } as unknown as T]; return []; } - // INSERT — return nothing + // INSERT ... RETURNING id — one row per successful insert (#2138). + if (sql.includes('INSERT INTO take_proposals')) { + return [{ id: captured.length } as unknown as T]; + } return []; }, } as unknown as BrainEngine; @@ -276,6 +279,22 @@ describe('runPhaseProposeTakes — phase integration', () => { expect(inserts[0]!.params[9]).toBe('market'); // domain }); + test('#2138: multi-claim page inserts every claim with a per-claim conflict target', async () => { + const pages = [buildPage({ slug: 'wiki/essays/thesis', body: 'Two strong claims live here.' })]; + const { engine, captured } = buildMockEngine({ pages }); + const extractor: ProposeTakesExtractor = async () => [ + { claim_text: 'Claim one', kind: 'take', holder: 'brain', weight: 0.6 }, + { claim_text: 'Claim two', kind: 'bet', holder: 'brain', weight: 0.8 }, + ]; + const result = await runPhaseProposeTakes(buildCtx(engine), { extractor }); + + expect((result.details as Record).proposals_inserted).toBe(2); + const inserts = captured.filter(c => c.sql.includes('INSERT INTO take_proposals')); + expect(inserts).toHaveLength(2); + for (const insert of inserts) expect(insert.sql).toContain('md5(claim_text)'); + expect(inserts.map(i => i.params[5])).toEqual(['Claim one', 'Claim two']); + }); + test('cache hit: page already in take_proposals is skipped', async () => { const body = 'A page that was already processed.'; const pages = [buildPage({ slug: 'wiki/old-page', body })];