From f68eb9d905cbe9d18d9eacdb4c94e329fe26bf4a Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Tue, 21 Jul 2026 14:33:19 -0700 Subject: [PATCH] fix(takes): kill propose_takes rescan loop, keep every claim, and route cycle/brainstorm models through config (#2804 #2805) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Takeover of #2804 and #2805 (stacked), rebased onto current master. Rescan loop + dropped claims (#2804): - Zero-claim scans now write a status='empty' sentinel row so the (source_id, page_slug, content_hash, prompt_version) cache hits on every subsequent cycle instead of re-spending the extractor LLM call on the same unchanged page forever. - The idempotency index gains md5(claim_text): multi-claim pages keep every claim (the 4-column unique index silently dropped claims 2..N via ON CONFLICT DO NOTHING). proposals_inserted now counts rows that actually landed (RETURNING id), not insert attempts. - Migration renumbered 123 -> 125 (master shipped v123 configurable_fts_language and v124 after the PR was cut; the duplicate v123 would never have run under the version > current runner). Schema parity across schema.sql, pglite-schema.ts, schema-embedded.ts. Model-tier honoring (#2805), reworked for current master: - propose_takes, grade_takes, calibration_profile resolve their model through resolveModel(models. > models.default > env > getChatModel()) — per-phase config keys are now honored while the gateway chat model (already tier-resolved by reconfigureGatewayWithEngine) stays the default, preserving master's #2451 semantics (provider-prefixed stored ids, nested prefixes like openrouter:... intact) and the #2997 opts.model ?? getChatModel() idiom. The PR's original bare-tail-everywhere approach was dropped where it conflicted with #2451; grade_takes keeps the bare tail ONLY for its cache key / stored judge_model_id (continuity with historical rows), while the judge call itself gets the full resolved string — fixing the live label-vs-actual mismatch (call rode chat_model, telemetry recorded a hardcoded claude-sonnet-4-6). - brainstorm resolves via models.brainstorm > tier chain (was hardcoded anthropic:claude-sonnet-4-6). - gbrain models lists the four new per-task keys; output caps and context-window entries added for newer Claude models. Tests: test/propose-takes-rescan.test.ts (new, hermetic PGLite); config-key coverage in propose-takes/grade-takes tests; existing #2451 and gateway-chat-model tests pass unchanged. Co-authored-by: p3ob7o Co-Authored-By: Claude Fable 5 --- src/commands/models.ts | 4 + src/core/brainstorm/judges.ts | 3 + src/core/brainstorm/orchestrator.ts | 10 +- src/core/cycle/calibration-profile.ts | 16 ++- src/core/cycle/grade-takes.ts | 25 +++- src/core/cycle/propose-takes.ts | 74 +++++++++--- src/core/cycle/synthesize.ts | 3 + src/core/migrate.ts | 35 ++++++ src/core/pglite-schema.ts | 4 +- src/core/schema-embedded.ts | 14 ++- src/schema.sql | 14 ++- test/calibration-profile.test.ts | 1 + test/grade-takes-ensemble.test.ts | 1 + test/grade-takes.test.ts | 24 ++++ test/propose-takes-rescan.test.ts | 160 ++++++++++++++++++++++++++ test/propose-takes.test.ts | 34 +++++- 16 files changed, 388 insertions(+), 34 deletions(-) create mode 100644 test/propose-takes-rescan.test.ts diff --git a/src/commands/models.ts b/src/commands/models.ts index 8b444370b..42443dc9c 100644 --- a/src/commands/models.ts +++ b/src/commands/models.ts @@ -50,6 +50,10 @@ const PER_TASK_KEYS: Array<{ key: string; tier: ModelTier; description: string } { key: 'models.eval.contradictions_judge', tier: 'utility', description: 'Contradiction probe judge (v0.34 temporal-aware)' }, { key: 'models.expansion', tier: 'utility', description: 'Query expansion for hybrid search' }, { key: 'models.chat', tier: 'reasoning', description: 'Default `gateway.chat()` model' }, + { key: 'models.propose_takes', tier: 'reasoning', description: 'propose_takes claim extractor' }, + { key: 'models.grade_takes', tier: 'reasoning', description: 'grade_takes verdict judge' }, + { key: 'models.calibration_profile', tier: 'reasoning', description: 'Calibration profile generator' }, + { key: 'models.brainstorm', tier: 'reasoning', description: '`gbrain brainstorm` orchestrator' }, ]; interface ModelEntry { diff --git a/src/core/brainstorm/judges.ts b/src/core/brainstorm/judges.ts index 00cdad468..2b157c5eb 100644 --- a/src/core/brainstorm/judges.ts +++ b/src/core/brainstorm/judges.ts @@ -61,7 +61,10 @@ export const MAX_OUTPUT_TOKENS_CEIL = 32_000; * (with a readable error) instead of the provider's opaque HTTP 400. */ export const ANTHROPIC_OUTPUT_CAPS: Record = { + 'claude-fable-5': 64_000, + 'claude-opus-4-8': 32_000, 'claude-opus-4-7': 32_000, + 'claude-sonnet-5': 64_000, 'claude-sonnet-4-6': 64_000, 'claude-haiku-4-5': 64_000, 'claude-haiku-4-5-20251001': 64_000, diff --git a/src/core/brainstorm/orchestrator.ts b/src/core/brainstorm/orchestrator.ts index 3ac7caef2..cbe7bcd42 100644 --- a/src/core/brainstorm/orchestrator.ts +++ b/src/core/brainstorm/orchestrator.ts @@ -32,6 +32,7 @@ */ import type { BrainEngine } from '../engine.ts'; +import { resolveModel } from '../model-config.ts'; import { chat as defaultChat, embedQuery, type ChatResult, type ChatOpts } from '../ai/gateway.ts'; import { hybridSearch, hybridSearchCached } from '../search/hybrid.ts'; import { fetchFar, type CloseRef, type FarPage } from './domain-bank.ts'; @@ -538,7 +539,14 @@ async function _runBrainstormInner( const embedFn = opts.embedQueryFn ?? embedQuery; // ---- Phase 0: cost preview + TTY grace ---- - const modelStr = opts.modelOverride ?? 'anthropic:claude-sonnet-4-6'; + // Tier-resolved (mirrors the cycle phases): honors models.brainstorm > + // models.default > models.tier.reasoning; the fallback keeps stock + // behavior identical (reasoning tier default IS claude-sonnet-4-6). + const modelStr = opts.modelOverride ?? await resolveModel(engine, { + configKey: 'models.brainstorm', + tier: 'reasoning', + fallback: 'anthropic:claude-sonnet-4-6', + }); const { aborted, estimate } = await previewCostAndWait({ profile, model: modelStr, diff --git a/src/core/cycle/calibration-profile.ts b/src/core/cycle/calibration-profile.ts index 2555cb03e..f4fe62ba8 100644 --- a/src/core/cycle/calibration-profile.ts +++ b/src/core/cycle/calibration-profile.ts @@ -26,8 +26,8 @@ */ import { BaseCyclePhase, type ScopedReadOpts, type BasePhaseOpts } from './base-phase.ts'; -import { chat as gatewayChat } from '../ai/gateway.ts'; -import { TIER_DEFAULTS } from '../model-config.ts'; +import { chat as gatewayChat, getChatModel } from '../ai/gateway.ts'; +import { resolveModel } from '../model-config.ts'; import { gateVoice, type VoiceGateGenerator, type VoiceGateJudge } from '../calibration/voice-gate.ts'; import { patternStatementTemplate, type PatternStatementSlots } from '../calibration/templates.ts'; // v0.41 T10 — domain widening. The aggregator module resolves the active @@ -229,7 +229,16 @@ class CalibrationProfilePhase extends BaseCyclePhase { ): Promise<{ summary: string; details: Record; status?: PhaseStatus }> { const holder = opts.holder ?? 'garry'; const promptVersion = opts.promptVersion ?? CALIBRATION_PROFILE_PROMPT_VERSION; - const modelId = opts.model ?? TIER_DEFAULTS.reasoning; + // Resolved once (see propose-takes.ts for the chain): models.calibration_profile + // > models.default > env > the gateway's chat model (itself resolved + // through models.chat + the reasoning tier). Provider-prefixed per #2451 + // — a bare id would make gateway.chat() throw "missing a provider + // prefix". Drives the generator's chat call, the budget label, and the + // persisted model_id, so the three can never disagree. + const modelId = opts.model ?? await resolveModel(engine, { + configKey: 'models.calibration_profile', + fallback: getChatModel(), + }); const gradeCompletion = opts.gradeCompletion ?? 1.0; const patternsGenerator = opts.patternsGenerator ?? defaultPatternsGenerator; const biasTagsGenerator = opts.biasTagsGenerator ?? defaultBiasTagsGenerator; @@ -265,6 +274,7 @@ class CalibrationProfilePhase extends BaseCyclePhase { scorecard, holder, attempt, + modelHint: modelId, ...(feedback !== undefined ? { feedback } : {}), }); return lines.join('\n'); diff --git a/src/core/cycle/grade-takes.ts b/src/core/cycle/grade-takes.ts index 0f1c14c4b..3062bf0af 100644 --- a/src/core/cycle/grade-takes.ts +++ b/src/core/cycle/grade-takes.ts @@ -36,7 +36,9 @@ import { createHash } from 'node:crypto'; import { BaseCyclePhase, type ScopedReadOpts, type BasePhaseOpts } from './base-phase.ts'; -import { chat as gatewayChat } from '../ai/gateway.ts'; +import { chat as gatewayChat, getChatModel } from '../ai/gateway.ts'; +import { resolveModel } from '../model-config.ts'; +import { splitProviderModelId } from '../model-id.ts'; import { GBrainError } from '../types.ts'; import type { OperationContext } from '../operations.ts'; import type { BrainEngine, Take, TakeResolution } from '../engine.ts'; @@ -395,7 +397,24 @@ class GradeTakesPhase extends BaseCyclePhase { const autoResolve = opts.autoResolve ?? false; // D17 default OFF const autoResolveThreshold = opts.autoResolveThreshold ?? 0.95; // D12 conservative const resolvedByLabel = opts.resolvedByLabel ?? 'gbrain:grade_takes'; - const judgeModelId = opts.model ?? 'claude-sonnet-4-6'; + // Resolve the judge model ONCE (see propose-takes.ts for the chain — + // same label-vs-actual split fixed here: the judge call rode the + // gateway's chat_model while the grade cache key, evidence signature, + // and budget label recorded a hardcoded 4.6). + // NOTE: changing the resolved judge model invalidates the grade cache + // (judge_model_id is part of its key) — a one-time, budget-capped + // re-grade wave that is CORRECT, since the actual judge did change. + const judgeModelFull = opts.model ?? await resolveModel(engine, { + configKey: 'models.grade_takes', + fallback: getChatModel(), + }); + // Bare tail for cache keys / evidence signatures / stored ids — the + // grade cache has always been keyed on bare ids; the gateway default + // resolves provider-prefixed, and normalizing preserves cache continuity + // on stock installs (no spurious re-judge wave from a prefix change). A + // genuinely different configured judge still invalidates, which is + // correct. The FULL string drives the actual judge call. + const judgeModelId = splitProviderModelId(judgeModelFull).model || judgeModelFull; const useEnsemble = opts.useEnsemble ?? false; const ensembleThreshold = opts.ensembleThreshold ?? 0.85; @@ -468,7 +487,7 @@ class GradeTakesPhase extends BaseCyclePhase { // Call the single-model judge. Errors on a single take log warning + continue. let verdict: JudgeVerdict; try { - verdict = await judge({ take, evidence, modelHint: opts.model }); + verdict = await judge({ take, evidence, modelHint: judgeModelFull }); } catch (err) { const msg = err instanceof Error ? err.message : String(err); result.warnings.push(`judge failed on take ${take.id}: ${msg}`); diff --git a/src/core/cycle/propose-takes.ts b/src/core/cycle/propose-takes.ts index 63ada141e..ce56856cf 100644 --- a/src/core/cycle/propose-takes.ts +++ b/src/core/cycle/propose-takes.ts @@ -5,11 +5,16 @@ * a tuned LLM extractor, writes the extracted gradeable claims to the * `take_proposals` queue. User accepts/rejects via `gbrain takes propose`. * - * Idempotency contract (D17 schema spec): - * The unique index on (source_id, page_slug, content_hash, prompt_version) - * means an unchanged page never re-spends LLM tokens. Bumping - * PROPOSE_TAKES_PROMPT_VERSION cleanly invalidates the cache so a tuned - * prompt re-runs proposals on every page. + * Idempotency contract (D17 schema spec; per-claim rows since migration v125): + * Every scan of a (source_id, page_slug, content_hash, prompt_version) + * tuple leaves at least one row — one per extracted claim, or a single + * status='empty' sentinel when extraction yields nothing — so an unchanged + * page never re-spends LLM tokens. Pre-v125 only proposal rows were + * written: a zero-claim page never entered the cache and was re-extracted + * on EVERY cycle (observed live: ~60 such pages × every cycle ≈ 1,400 + * wasted extractor calls / ~$15 per day — ~90% of total autopilot spend). + * Bumping PROPOSE_TAKES_PROMPT_VERSION cleanly invalidates the cache so a + * tuned prompt re-runs proposals on every page. * * F2 fence dedup: * The phase reads the page's existing `` fence @@ -40,6 +45,7 @@ import { randomUUID, createHash } from 'node:crypto'; import { BaseCyclePhase, type ScopedReadOpts, type BasePhaseOpts } from './base-phase.ts'; import { chat as gatewayChat, getChatModel } from '../ai/gateway.ts'; +import { resolveModel } from '../model-config.ts'; import { writeReceipt } from '../extract/receipt-writer.ts'; import { upsertExtractRollup } from '../extract/rollup-writer.ts'; import { GBrainError } from '../types.ts'; @@ -307,6 +313,18 @@ class ProposeTakesPhase extends BaseCyclePhase { const promptVersion = opts.promptVersion ?? PROPOSE_TAKES_PROMPT_VERSION; const pageLimit = opts.pageLimit ?? 100; const skipPagesWithFence = opts.skipPagesWithFence ?? false; + // Resolve the extractor model ONCE: models.propose_takes > + // models.default > GBRAIN_MODEL env > the gateway's chat model (which + // reconfigureGatewayWithEngine already resolved through models.chat + + // the reasoning tier). One resolved provider-prefixed string drives the + // actual chat call, the budget estimate, AND the stored model_id — so + // the recorded model can never disagree with the model that ran (#2451 + // convention: stored ids are provider-prefixed, nested prefixes like + // openrouter:anthropic/... stay intact). + const extractorModelId = opts.model ?? await resolveModel(engine, { + configKey: 'models.propose_takes', + fallback: getChatModel(), + }); const proposalRunId = `propose-${new Date().toISOString().slice(0, 19).replace(/[-:T]/g, '')}-${randomUUID().slice(0, 8)}`; const result: ProposeTakesResult = { @@ -330,8 +348,6 @@ class ProposeTakesPhase extends BaseCyclePhase { opts.reporter.start('propose_takes.pages' as never, pages.length); } - const modelId = opts.model ?? getChatModel(); - for (const page of pages) { result.pages_scanned += 1; this.tick(opts); @@ -361,7 +377,7 @@ class ProposeTakesPhase extends BaseCyclePhase { // Budget pre-check before the LLM call. Estimate: ~1500 input tokens + 500 output. const budget = this.checkBudget({ - modelId, + modelId: extractorModelId, estimatedInputTokens: 1500, maxOutputTokens: 500, }); @@ -380,7 +396,7 @@ class ProposeTakesPhase extends BaseCyclePhase { pagePath: page.slug, pageBody: body, existingTakes, - modelHint: opts.model, + modelHint: extractorModelId, }); } catch (err) { const msg = err instanceof Error ? err.message : String(err); @@ -388,16 +404,42 @@ 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. - for (const p of proposals) { + // Zero-claim scans MUST still enter the idempotency cache. Pre-v125 + // only proposal rows were written, so a page whose extraction yielded + // no gradeable claims never got a row for its (page, content_hash) — + // the cache check above missed on every subsequent cycle and the LLM + // call was re-spent on the same unchanged page, forever. The sentinel + // row (status='empty', empty claim_text) is invisible to the review + // queue (pending_idx is partial on status='pending'); it exists only + // so the cache check hits. + if (proposals.length === 0) { await engine.executeRaw( + `INSERT INTO take_proposals + (source_id, page_slug, content_hash, prompt_version, proposal_run_id, + status, claim_text, kind, holder, weight, domain, dedup_against_fence_rows, model_id) + VALUES ($1, $2, $3, $4, $5, 'empty', '', 'none', 'brain', 0, NULL, NULL, $6) + ON CONFLICT (source_id, page_slug, content_hash, prompt_version, md5(claim_text)) DO NOTHING`, + [sourceId, page.slug, ch, promptVersion, proposalRunId, extractorModelId], + ); + continue; + } + + // Write proposals to take_proposals, one row per claim. The v125 + // idempotency index includes md5(claim_text), so a same-page + // multi-claim run keeps EVERY claim — the pre-v125 four-column unique + // index made claims 2..N conflict with claim 1 and ON CONFLICT DO + // NOTHING silently dropped them (the review queue only ever saw the + // first claim of each page version). RETURNING id keeps + // proposals_inserted honest: it counts rows that actually landed, + // not insert attempts. + for (const p of proposals) { + const landed = 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, @@ -410,10 +452,10 @@ class ProposeTakesPhase extends BaseCyclePhase { p.weight, p.domain ?? null, JSON.stringify(existingTakes), - modelId, + extractorModelId, ], ); - result.proposals_inserted += 1; + if (landed.length > 0) result.proposals_inserted += 1; } } diff --git a/src/core/cycle/synthesize.ts b/src/core/cycle/synthesize.ts index 2dc1a5f7a..5caeef0f0 100644 --- a/src/core/cycle/synthesize.ts +++ b/src/core/cycle/synthesize.ts @@ -58,8 +58,11 @@ const SUMMARY_SLUG_RE = /^[a-z0-9][a-z0-9\-]*(\/[a-z0-9][a-z0-9\-]*)*$/; * resolver returns for known Anthropic aliases. */ const MODEL_CONTEXT_TOKENS: Record = { + 'claude-fable-5': 1_000_000, + 'claude-opus-4-8': 1_000_000, 'claude-opus-4-7': 1_000_000, 'claude-opus-4-6': 1_000_000, + 'claude-sonnet-5': 1_000_000, 'claude-sonnet-4-6': 200_000, 'claude-sonnet-4-5': 200_000, 'claude-haiku-4-5-20251001': 200_000, diff --git a/src/core/migrate.ts b/src/core/migrate.ts index 124b79c8b..c717680f0 100644 --- a/src/core/migrate.ts +++ b/src/core/migrate.ts @@ -5671,6 +5671,41 @@ export const MIGRATIONS: Migration[] = [ `); }, }, + { + version: 125, + name: 'take_proposals_empty_scan_sentinels_and_per_claim_rows', + // v0.42.x — kill the propose_takes rescan loop + stop dropping claims. + // + // Two defects, one schema touch: + // 1. Zero-claim scans never entered the idempotency cache (only + // proposal rows were written), so pages whose extraction yielded + // nothing were re-extracted on EVERY cycle. Observed live: ~60 + // such pages per run ≈ 1,400 wasted extractor calls / ~$15 per + // day — ~90% of total autopilot LLM spend. Fix: the phase now + // writes a status='empty' sentinel row per zero-claim scan; the + // status CHECK gains the 'empty' value. Sentinels are excluded + // from the partial pending index, so the review queue never sees + // them. + // 2. The 4-column unique index collapsed a same-page multi-claim run + // to its FIRST claim: rows 2..N conflicted and ON CONFLICT DO + // NOTHING silently dropped them (verified live: exactly one row + // per (page, hash) across 3 days of runs). Fix: the idempotency + // index gains md5(claim_text) — per-claim rows, while the + // 4-column prefix still serves the per-scan cache lookup. + // + // Existing data is index-safe by construction: the old index guaranteed + // at most one row per 4-tuple, so the widened index has no duplicates + // to trip on. The DROP+ADD CONSTRAINT pair is idempotent as a unit. + idempotent: true, + sql: ` + ALTER TABLE take_proposals DROP CONSTRAINT IF EXISTS take_proposals_status_check; + ALTER TABLE take_proposals ADD CONSTRAINT take_proposals_status_check + CHECK (status IN ('pending','accepted','rejected','superseded','empty')); + 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..5b2985e92 100644 --- a/src/core/pglite-schema.ts +++ b/src/core/pglite-schema.ts @@ -762,7 +762,7 @@ CREATE TABLE IF NOT EXISTS take_proposals ( proposed_at TIMESTAMPTZ NOT NULL DEFAULT now(), proposal_run_id TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending' - CHECK (status IN ('pending','accepted','rejected','superseded')), + CHECK (status IN ('pending','accepted','rejected','superseded','empty')), claim_text TEXT NOT NULL, kind TEXT NOT NULL, holder TEXT NOT NULL, @@ -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..17b540069 100644 --- a/src/core/schema-embedded.ts +++ b/src/core/schema-embedded.ts @@ -1274,8 +1274,14 @@ CREATE INDEX IF NOT EXISTS calibration_profiles_published_idx 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. +-- composite unique index (source_id, page_slug, content_hash, prompt_version, +-- md5(claim_text)) — the 4-column prefix is the per-scan cache key (mirrors +-- v0.23 dream_verdicts); md5(claim_text) makes rows per-claim so multi-claim +-- pages keep every claim (v125). status='empty' rows are zero-claim scan +-- sentinels: they hold the cache slot for a page version whose extraction +-- yielded nothing — the phase never re-spends the LLM call on that page +-- version. Excluded from the partial pending index. proposal_run_id supports +-- --rollback by run. CREATE TABLE IF NOT EXISTS take_proposals ( id BIGSERIAL PRIMARY KEY, source_id TEXT NOT NULL REFERENCES sources(id) ON DELETE CASCADE, @@ -1286,7 +1292,7 @@ CREATE TABLE IF NOT EXISTS take_proposals ( proposed_at TIMESTAMPTZ NOT NULL DEFAULT now(), proposal_run_id TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending' - CHECK (status IN ('pending','accepted','rejected','superseded')), + CHECK (status IN ('pending','accepted','rejected','superseded','empty')), claim_text TEXT NOT NULL, kind TEXT NOT NULL, holder TEXT NOT NULL, @@ -1301,7 +1307,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..a708a3fc8 100644 --- a/src/schema.sql +++ b/src/schema.sql @@ -1270,8 +1270,14 @@ CREATE INDEX IF NOT EXISTS calibration_profiles_published_idx 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. +-- composite unique index (source_id, page_slug, content_hash, prompt_version, +-- md5(claim_text)) — the 4-column prefix is the per-scan cache key (mirrors +-- v0.23 dream_verdicts); md5(claim_text) makes rows per-claim so multi-claim +-- pages keep every claim (v125). status='empty' rows are zero-claim scan +-- sentinels: they hold the cache slot for a page version whose extraction +-- yielded nothing — the phase never re-spends the LLM call on that page +-- version. Excluded from the partial pending index. proposal_run_id supports +-- --rollback by run. CREATE TABLE IF NOT EXISTS take_proposals ( id BIGSERIAL PRIMARY KEY, source_id TEXT NOT NULL REFERENCES sources(id) ON DELETE CASCADE, @@ -1282,7 +1288,7 @@ CREATE TABLE IF NOT EXISTS take_proposals ( proposed_at TIMESTAMPTZ NOT NULL DEFAULT now(), proposal_run_id TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending' - CHECK (status IN ('pending','accepted','rejected','superseded')), + CHECK (status IN ('pending','accepted','rejected','superseded','empty')), claim_text TEXT NOT NULL, kind TEXT NOT NULL, holder TEXT NOT NULL, @@ -1297,7 +1303,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/calibration-profile.test.ts b/test/calibration-profile.test.ts index f92e516ed..8d3ee0f6c 100644 --- a/test/calibration-profile.test.ts +++ b/test/calibration-profile.test.ts @@ -38,6 +38,7 @@ function buildMockEngine(opts: { scorecard: TakesScorecard }): { } { const captured: CapturedSql[] = []; const engine = { + async getConfig() { return null; }, kind: 'pglite', async getScorecard() { return opts.scorecard; diff --git a/test/grade-takes-ensemble.test.ts b/test/grade-takes-ensemble.test.ts index d624c878b..558a2c4dd 100644 --- a/test/grade-takes-ensemble.test.ts +++ b/test/grade-takes-ensemble.test.ts @@ -47,6 +47,7 @@ function buildMockEngine(opts: { takes: Take[] }): { const captured: CapturedSql[] = []; const resolves: CapturedResolve[] = []; const engine = { + async getConfig() { return null; }, kind: 'pglite', async listTakes() { return opts.takes; diff --git a/test/grade-takes.test.ts b/test/grade-takes.test.ts index c8bd3e3d8..5ab4735d5 100644 --- a/test/grade-takes.test.ts +++ b/test/grade-takes.test.ts @@ -46,12 +46,14 @@ interface CapturedResolve { function buildMockEngine(opts: { takes: Take[]; cachedGrades?: Set; // composite-key strings already in take_grade_cache + config?: Record; // engine.getConfig plane (models.grade_takes etc.) }): { engine: BrainEngine; captured: CapturedSql[]; resolves: CapturedResolve[] } { const captured: CapturedSql[] = []; const resolves: CapturedResolve[] = []; const cached = opts.cachedGrades ?? new Set(); const engine = { + async getConfig(key: string) { return opts.config?.[key] ?? null; }, kind: 'pglite', async listTakes() { return opts.takes; @@ -224,6 +226,28 @@ describe('runPhaseGradeTakes — phase integration', () => { expect(resolves).toHaveLength(0); // no canonical mutation }); + test('models.grade_takes config drives the judge call; cache key stays bare-tailed', async () => { + // Pre-fix the judge call rode the gateway's chat_model while the cache + // key / budget label recorded a hardcoded 'claude-sonnet-4-6'. The phase + // now resolves models.grade_takes; the judge gets the FULL string and + // the cache row keys on the bare tail (continuity with historical rows). + const takes = [buildTake({ id: 1, sinceDate: '2023-01-01' })]; + const { engine, captured } = buildMockEngine({ + takes, + config: { 'models.grade_takes': 'anthropic:claude-sonnet-5' }, + }); + const hints: Array = []; + const judge: JudgeFn = async ({ modelHint }) => { + hints.push(modelHint); + return { verdict: 'correct', confidence: 0.9, reasoning: 'held' }; + }; + const result = await runPhaseGradeTakes(buildCtx(engine), { judge }); + expect(result.status).toBe('ok'); + expect(hints).toEqual(['anthropic:claude-sonnet-5']); // actual call gets the FULL string + const inserts = captured.filter(c => c.sql.includes('INSERT INTO take_grade_cache')); + expect(inserts[0]!.params[2]).toBe('claude-sonnet-5'); // judge_model_id is the bare tail + }); + test('D17: auto-resolve OFF by default — even high-confidence verdict does NOT mutate takes', async () => { const takes = [buildTake({ id: 1, sinceDate: '2023-01-01' })]; const { engine, resolves } = buildMockEngine({ takes }); diff --git a/test/propose-takes-rescan.test.ts b/test/propose-takes-rescan.test.ts new file mode 100644 index 000000000..6d9c46cf5 --- /dev/null +++ b/test/propose-takes-rescan.test.ts @@ -0,0 +1,160 @@ +/** + * propose_takes rescan-loop + dropped-claims regression tests (migration v125). + * + * Two live-observed defects, both fixed by the v125 schema + phase change: + * + * 1. RESCAN LOOP — a page whose extraction yielded zero claims never + * entered the idempotency cache (only proposal rows were written), so + * every cycle re-spent the extractor call on the same unchanged page. + * Live impact: ~60 such pages × every cycle ≈ 1,400 wasted LLM calls + * (~$15) per day — ~90% of total autopilot spend. Fix: status='empty' + * sentinel row per zero-claim scan. + * + * 2. DROPPED CLAIMS — the 4-column unique index collapsed a same-page + * multi-claim run to its first claim (rows 2..N conflicted, ON CONFLICT + * DO NOTHING dropped them silently; verified live: exactly 1 row per + * (page, hash) over 3 days). Fix: idempotency index gains + * md5(claim_text). + * + * Hermetic: PGLite engine + injected extractor; no gateway, no LLM. + */ +import { describe, test, expect, beforeAll, afterAll } from 'bun:test'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { runPhaseProposeTakes, type ProposeTakesExtractor, type ProposedTake } from '../src/core/cycle/propose-takes.ts'; +import type { OperationContext } from '../src/core/operations.ts'; + +let engine: PGLiteEngine; + +function ctx(): OperationContext { + return { + engine, + remote: false, + config: {} as OperationContext['config'], + logger: { info() {}, warn() {}, error() {}, debug() {} } as unknown as OperationContext['logger'], + } as unknown as OperationContext; +} + +/** Extractor stub that counts invocations per page slug. */ +function countingExtractor( + claimsBySlug: Record, +): { extractor: ProposeTakesExtractor; calls: string[] } { + const calls: string[] = []; + const extractor: ProposeTakesExtractor = async ({ pagePath }) => { + calls.push(pagePath); + return claimsBySlug[pagePath] ?? []; + }; + return { extractor, calls }; +} + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); + await engine.putPage('notes/zero-claims', { + type: 'note', + title: 'pure narrative', + compiled_truth: 'A quiet walk in the park. Nothing opinionated happened at all today.', + }); + await engine.putPage('notes/three-claims', { + type: 'note', + title: 'opinionated', + compiled_truth: 'I bet acme-example wins the market. widget-co will struggle. fund-a is overexposed.', + }); +}); + +afterAll(async () => { + await engine.disconnect(); +}); + +describe('rescan loop — zero-claim scans enter the cache', () => { + test('second run cache-hits: extractor is NOT called again on unchanged pages', async () => { + const claims = { + 'notes/three-claims': [ + { claim_text: 'acme-example wins the market', kind: 'bet' as const, holder: 'brain', weight: 0.7 }, + { claim_text: 'widget-co will struggle', kind: 'take' as const, holder: 'brain', weight: 0.6 }, + { claim_text: 'fund-a is overexposed', kind: 'take' as const, holder: 'brain', weight: 0.55 }, + ], + }; + + const first = countingExtractor(claims); + const r1 = await runPhaseProposeTakes(ctx(), { extractor: first.extractor }); + expect(r1.status).toBe('ok'); + // Both pages extracted on the first pass. + expect(first.calls).toContain('notes/zero-claims'); + expect(first.calls).toContain('notes/three-claims'); + + const second = countingExtractor(claims); + const r2 = await runPhaseProposeTakes(ctx(), { extractor: second.extractor }); + expect(r2.status).toBe('ok'); + // THE regression: pre-fix the zero-claim page missed the cache every + // run and was re-extracted here. (Run 1's receipt page legitimately + // appears once — it's a new page — and its zero-claim scan now caches + // too; pre-fix, receipts re-scanned forever as well.) + expect(second.calls).not.toContain('notes/zero-claims'); + expect(second.calls).not.toContain('notes/three-claims'); + + // Run 2 inserted nothing, so no new receipt page exists: run 3 must be + // fully quiescent — zero extractor calls, zero cache misses. + const third = countingExtractor(claims); + const r3 = await runPhaseProposeTakes(ctx(), { extractor: third.extractor }); + expect(r3.status).toBe('ok'); + expect(third.calls).toEqual([]); + expect((r3.details as Record).cache_misses).toBe(0); + }); + + test('zero-claim scan wrote an "empty" sentinel invisible to the pending queue', async () => { + const sentinel = await engine.executeRaw<{ status: string; claim_text: string }>( + `SELECT status, claim_text FROM take_proposals WHERE page_slug = 'notes/zero-claims'`, + [], + ); + expect(sentinel.length).toBe(1); + expect(sentinel[0].status).toBe('empty'); + expect(sentinel[0].claim_text).toBe(''); + const pending = await engine.executeRaw<{ n: number }>( + `SELECT COUNT(*)::int AS n FROM take_proposals WHERE page_slug = 'notes/zero-claims' AND status = 'pending'`, + [], + ); + expect(Number(pending[0].n)).toBe(0); + }); +}); + +describe('dropped claims — per-claim rows survive the idempotency index', () => { + test('a 3-claim page stores 3 rows and reports an honest inserted count', async () => { + const rows = await engine.executeRaw<{ claim_text: string }>( + `SELECT claim_text FROM take_proposals WHERE page_slug = 'notes/three-claims' AND status = 'pending' ORDER BY id`, + [], + ); + // Pre-fix the 4-column unique index kept only the FIRST claim. + expect(rows.length).toBe(3); + expect(rows.map(r => r.claim_text)).toEqual([ + 'acme-example wins the market', + 'widget-co will struggle', + 'fund-a is overexposed', + ]); + }); + + test('content change re-extracts and stores the new version separately', async () => { + await engine.putPage('notes/zero-claims', { + type: 'note', + title: 'pure narrative', + compiled_truth: 'Updated: I now believe acme-example is undervalued and will re-rate within a year.', + }); + const claims = { + 'notes/zero-claims': [ + { claim_text: 'acme-example is undervalued', kind: 'take' as const, holder: 'brain', weight: 0.6 }, + ], + }; + const run = countingExtractor(claims); + const r = await runPhaseProposeTakes(ctx(), { extractor: run.extractor }); + expect(r.status).toBe('ok'); + // Changed page re-extracts; the unchanged 3-claim page stays cached. + expect(run.calls).toEqual(['notes/zero-claims']); + expect((r.details as Record).proposals_inserted).toBe(1); + const all = await engine.executeRaw<{ status: string }>( + `SELECT status FROM take_proposals WHERE page_slug = 'notes/zero-claims' ORDER BY id`, + [], + ); + // Old hash's sentinel + new hash's pending claim coexist. + expect(all.map(r2 => r2.status).sort()).toEqual(['empty', 'pending']); + }); +}); diff --git a/test/propose-takes.test.ts b/test/propose-takes.test.ts index 3c0ccb68d..3c77ea346 100644 --- a/test/propose-takes.test.ts +++ b/test/propose-takes.test.ts @@ -41,12 +41,16 @@ interface CapturedSql { function buildMockEngine(opts: { pages: Page[]; existingProposals?: Set; // composite-key strings already in take_proposals + config?: Record; // engine.getConfig plane (models.tier.* etc.) }): { engine: BrainEngine; captured: CapturedSql[] } { const captured: CapturedSql[] = []; const existing = opts.existingProposals ?? new Set(); const engine = { kind: 'pglite', + async getConfig(key: string) { + return opts.config?.[key] ?? null; + }, async listPages() { return opts.pages; }, @@ -59,7 +63,12 @@ function buildMockEngine(opts: { if (existing.has(key)) return [{ id: 1 } as unknown as T]; return []; } - // INSERT — return nothing + // INSERT ... RETURNING id — emulate a successful insert so the + // honest proposals_inserted counter (counts RETURNING rows, not + // attempts) sees the row land. Conflicted inserts would return []. + if (sql.includes('INSERT INTO take_proposals') && sql.includes('RETURNING id')) { + return [{ id: 1 } as unknown as T]; + } return []; }, } as unknown as BrainEngine; @@ -267,6 +276,29 @@ describe('runPhaseProposeTakes — phase integration', () => { expect(inserts[0]!.params[9]).toBe('market'); // domain }); + test('extractor model resolves through models.propose_takes config', async () => { + // Pre-fix the phase's only model knob was the gateway chat model — there + // was no per-phase config key. The phase now resolves once via + // resolveModel(models.propose_takes > models.default > env > gateway + // chat model); the extractor hint and the stored model_id must both + // reflect the configured override (full provider-prefixed string, #2451). + const pages = [buildPage({ slug: 'wiki/concepts/tier-routing', body: 'Tier-routed models will win.' })]; + const { engine, captured } = buildMockEngine({ + pages, + config: { 'models.propose_takes': 'anthropic:claude-sonnet-5' }, + }); + const seen: Array = []; + const extractor: ProposeTakesExtractor = async ({ modelHint }) => { + seen.push(modelHint); + return [{ claim_text: 'tier-routed models win', kind: 'bet', holder: 'brain', weight: 0.7 }]; + }; + const result = await runPhaseProposeTakes(buildCtx(engine), { extractor }); + expect(result.status).toBe('ok'); + expect(seen).toEqual(['anthropic:claude-sonnet-5']); // chat call gets the FULL string + const inserts = captured.filter(c => c.sql.includes('INSERT INTO take_proposals')); + expect(inserts[0]!.params[11]).toBe('anthropic:claude-sonnet-5'); // stored model_id matches the call + }); + 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 })];