fix(cycle): preserve per-page multi-claim proposals (#3297)

Co-authored-by: David Guidry <hairpie@mac.com>

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;
       }
     }
 

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

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';

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';

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';

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<number> {
+  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<void> {
+  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<string, unknown>).proposals_inserted).toBe(2);
+    expect(await countProposals('wiki/essays/thesis')).toBe(2);
+
+    const rerun = await runPhaseProposeTakes(context(), { extractor: proposals });
+    expect((rerun.details as Record<string, unknown>).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<string, unknown>).proposals_inserted).toBe(2);
+    expect(await countProposals('wiki/essays/thesis')).toBe(2);
+  });
+});

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<string, unknown>).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 })];