fix(takes): persist think --take rows + review-first take proposals (#2556, #2411)

Takeover of two community PRs, rebased onto master and repaired: PR #2618 (think --take, fixes #2556): runThink declared `take` but never consumed it — `--take` silently wrote nothing ("Takes: 0"). Adds persistThinkTake (anchor-page take append, refuses empty/no-LLM synthesis, source-scope forwarded to getPage), explicit CLI error when no row is written, REMOTE_PERSISTED_BLOCKED warning at the trust boundary. Rebased onto master's thinkSourceScopeOpts (#2739) split. PR #2418 (takes propose CLI + ops, fixes #2411): implements the documented `gbrain takes propose list/accept/reject` promotion path (D17) plus takes_propose_* MCP ops. Repairs from review: - Source scoping now routes through sourceScopeOpts(ctx) (federated array > scalar > nothing, fail-closed) instead of the fail-open `ctx.remote ? ctx.sourceId : undefined`; listTakeProposals/accept/reject accept sourceIds[]. - takes_propose_accept is localOnly (writes source markdown under sync.repo_path — same class as sync; D17 operator-only promotion) and both write ops carry mutating: true. - Rebased onto master's source-resolver-threaded takes dispatcher. Also corrects the propose-takes.ts D17 doc comment to the shipped syntax (`takes propose accept <proposal_id>`). Co-authored-by: javieraldape <javieraldape@users.noreply.github.com> Co-authored-by: Mr-B-1 <Mr-B-1@users.noreply.github.com> Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-29 19:01:39 +00:00 · 2026-07-21 14:58:16 -07:00
parent 0612b0daa8
commit f23f964a5c
9 changed files with 1020 additions and 6 deletions
@@ -0,0 +1,386 @@
+import { afterEach, describe, expect, test } from 'bun:test';
+import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { dirname, join } from 'node:path';
+import { operations } from '../src/core/operations.ts';
+import { buildToolDefs } from '../src/mcp/tool-defs.ts';
+import {
+  acceptTakeProposal,
+  listTakeProposals,
+  rejectTakeProposal,
+  type TakeProposalRow,
+} from '../src/core/take-proposals.ts';
+import type { BrainEngine } from '../src/core/engine.ts';
+
+type ProposalStatus = TakeProposalRow['status'];
+
+interface ProposalRecord {
+  id: number;
+  source_id: string;
+  page_slug: string;
+  status: ProposalStatus;
+  claim_text: string;
+  kind: string;
+  holder: string;
+  weight: number;
+  domain?: string | null;
+  dedup_against_fence_rows?: unknown;
+  model_id: string;
+  proposed_at: Date;
+  acted_at?: Date | null;
+  acted_by?: string | null;
+  promoted_row_num?: number | null;
+  predicted_brier?: number | null;
+  predicted_brier_bucket_n?: number | null;
+}
+
+interface PageRecord {
+  id: number;
+  source_id: string;
+  slug: string;
+  effective_date?: Date | null;
+  effective_date_source?: string | null;
+}
+
+const tmpRoots: string[] = [];
+
+afterEach(() => {
+  for (const root of tmpRoots.splice(0)) {
+    rmSync(root, { recursive: true, force: true });
+  }
+});
+
+function makeBrainDir(): string {
+  const root = mkdtempSync(join(tmpdir(), 'gbrain-take-proposals-'));
+  tmpRoots.push(root);
+  return root;
+}
+
+function writePage(brainDir: string, slug: string, body = '# Page\n'): string {
+  const path = join(brainDir, `${slug}.md`);
+  mkdirSync(dirname(path), { recursive: true });
+  writeFileSync(path, body, 'utf-8');
+  return path;
+}
+
+class FakeProposalEngine {
+  readonly kind = 'postgres' as const;
+  readonly proposals = new Map<number, ProposalRecord>();
+  readonly pages = new Map<string, PageRecord>();
+  readonly config = new Map<string, string>();
+  readonly addedBatches: unknown[][] = [];
+  readonly queries: Array<{ sql: string; params: unknown[] }> = [];
+  failAddTakes = false;
+  activeTakes: Array<{ page_id: number; row_num: number; claim: string; active: boolean }> = [];
+
+  constructor(brainDir: string) {
+    this.config.set('sync.repo_path', brainDir);
+  }
+
+  addPage(page: PageRecord): void {
+    this.pages.set(`${page.source_id}:${page.slug}`, page);
+  }
+
+  addProposal(record: Partial<ProposalRecord> & Pick<ProposalRecord, 'id' | 'page_slug' | 'claim_text'>): void {
+    this.proposals.set(record.id, {
+      source_id: 'default',
+      status: 'pending',
+      kind: 'take',
+      holder: 'people/garry-tan',
+      weight: 0.7,
+      model_id: 'test-model',
+      proposed_at: new Date('2026-06-25T00:00:00Z'),
+      acted_at: null,
+      acted_by: null,
+      promoted_row_num: null,
+      predicted_brier: null,
+      predicted_brier_bucket_n: null,
+      ...record,
+    });
+  }
+
+  async getConfig(key: string): Promise<string | null> {
+    return this.config.get(key) ?? null;
+  }
+
+  async transaction<T>(fn: (engine: BrainEngine) => Promise<T>): Promise<T> {
+    return fn(this as unknown as BrainEngine);
+  }
+
+  async addTakesBatch(rows: unknown[]): Promise<number> {
+    if (this.failAddTakes) throw new Error('addTakesBatch failed');
+    this.addedBatches.push(rows);
+    for (const row of rows as Array<{ page_id: number; row_num: number; claim: string; active: boolean }>) {
+      this.activeTakes.push(row);
+    }
+    return rows.length;
+  }
+
+  async executeRaw<T = Record<string, unknown>>(sql: string, params: unknown[] = []): Promise<T[]> {
+    this.queries.push({ sql, params });
+    const compact = sql.replace(/\s+/g, ' ').trim();
+
+    if (compact.includes('FROM take_proposals tp LEFT JOIN pages p')) {
+      const [status, scopeIds, limitRaw, offsetRaw] = params as [string | null, string[] | null, number, number];
+      const limit = Number(limitRaw ?? 50);
+      const offset = Number(offsetRaw ?? 0);
+      const rows = [...this.proposals.values()]
+        .filter((p) => status === null || p.status === status)
+        .filter((p) => scopeIds === null || scopeIds.includes(p.source_id))
+        .sort((a, b) => b.proposed_at.getTime() - a.proposed_at.getTime())
+        .slice(offset, offset + limit)
+        .map((p) => ({
+          ...p,
+          ...this.pageFields(p.source_id, p.page_slug),
+        }));
+      return rows as T[];
+    }
+
+    if (compact.startsWith('SELECT page_slug, source_id, status, promoted_row_num FROM take_proposals WHERE id')) {
+      const p = this.proposals.get(Number(params[0]));
+      return (p ? [{
+        page_slug: p.page_slug,
+        source_id: p.source_id,
+        status: p.status,
+        promoted_row_num: p.promoted_row_num ?? null,
+      }] : []) as T[];
+    }
+
+    if (compact.includes('FROM take_proposals tp JOIN pages p')) {
+      const p = this.proposals.get(Number(params[0]));
+      if (!p) return [] as T[];
+      const page = this.pages.get(`${p.source_id}:${p.page_slug}`);
+      if (!page) return [] as T[];
+      return [{
+        ...p,
+        page_id: page.id,
+        effective_date: page.effective_date ?? null,
+        effective_date_source: page.effective_date_source ?? null,
+      }] as T[];
+    }
+
+    if (compact.startsWith('SELECT pg_advisory_xact_lock')) {
+      return [] as T[];
+    }
+
+    if (compact.includes('FROM takes WHERE page_id')) {
+      const [pageId, claim] = params;
+      const normalizedClaim = String(claim).trim().toLowerCase();
+      const duplicate = this.activeTakes.find((t) =>
+        t.page_id === Number(pageId) &&
+        t.active === true &&
+        t.claim.trim().toLowerCase() === normalizedClaim
+      );
+      return (duplicate ? [{ row_num: duplicate.row_num }] : []) as T[];
+    }
+
+    if (compact.startsWith("UPDATE take_proposals SET status = 'accepted'")) {
+      const [id, actedBy, rowNum] = params;
+      const p = this.proposals.get(Number(id));
+      if (!p || p.status !== 'pending') return [] as T[];
+      p.status = 'accepted';
+      p.acted_at = new Date('2026-06-25T01:00:00Z');
+      p.acted_by = String(actedBy);
+      p.promoted_row_num = Number(rowNum);
+      return [{ promoted_row_num: p.promoted_row_num }] as T[];
+    }
+
+    if (compact.startsWith('SELECT id, source_id, status, promoted_row_num FROM take_proposals WHERE id')) {
+      const p = this.proposals.get(Number(params[0]));
+      return (p ? [{
+        id: p.id,
+        source_id: p.source_id,
+        status: p.status,
+        promoted_row_num: p.promoted_row_num ?? null,
+      }] : []) as T[];
+    }
+
+    if (compact.startsWith("UPDATE take_proposals SET status = 'rejected'")) {
+      const [id, actedBy] = params;
+      const p = this.proposals.get(Number(id));
+      if (p) {
+        p.status = 'rejected';
+        p.acted_at = new Date('2026-06-25T01:00:00Z');
+        p.acted_by = String(actedBy);
+      }
+      return [] as T[];
+    }
+
+    throw new Error(`unhandled SQL in fake engine: ${compact}`);
+  }
+
+  private pageFields(sourceId: string, slug: string) {
+    const page = this.pages.get(`${sourceId}:${slug}`);
+    return {
+      effective_date: page?.effective_date ?? null,
+      effective_date_source: page?.effective_date_source ?? null,
+    };
+  }
+}
+
+describe('take proposal review helpers', () => {
+  test('lists pending proposals with page effective-date metadata', async () => {
+    const brainDir = makeBrainDir();
+    const engine = new FakeProposalEngine(brainDir);
+    engine.addPage({ id: 10, source_id: 'default', slug: 'topics/a', effective_date: new Date('2024-03-02T00:00:00Z'), effective_date_source: 'frontmatter' });
+    engine.addProposal({ id: 1, page_slug: 'topics/a', claim_text: 'A will happen' });
+
+    const rows = await listTakeProposals(engine as unknown as BrainEngine, { status: 'pending', sourceId: 'default' });
+    expect(rows).toHaveLength(1);
+    expect(rows[0]).toMatchObject({
+      id: 1,
+      status: 'pending',
+      page_slug: 'topics/a',
+      effective_date_source: 'frontmatter',
+    });
+    expect(rows[0].effective_date).toContain('2024-03-02');
+  });
+
+  test('accept writes markdown, mirrors DB, stamps proposal, and uses real effective date', async () => {
+    const brainDir = makeBrainDir();
+    const pagePath = writePage(brainDir, 'topics/a', '# A\n\nBody\n');
+    const engine = new FakeProposalEngine(brainDir);
+    engine.addPage({ id: 10, source_id: 'default', slug: 'topics/a', effective_date: new Date('2024-03-02T00:00:00Z'), effective_date_source: 'frontmatter' });
+    engine.addProposal({ id: 1, page_slug: 'topics/a', claim_text: 'A will happen' });
+
+    const result = await acceptTakeProposal(engine as unknown as BrainEngine, 1, { actedBy: 'test' });
+
+    expect(result).toMatchObject({ ok: true, proposal_id: 1, page_slug: 'topics/a', status: 'accepted', idempotent: false, since_date: '2024-03-02' });
+    expect(result.row_num).toBe(1);
+    expect(engine.addedBatches).toHaveLength(1);
+    expect((engine.addedBatches[0][0] as Record<string, unknown>)).toMatchObject({
+      page_id: 10,
+      row_num: 1,
+      claim: 'A will happen',
+      since_date: '2024-03-02',
+      source: 'proposal:1',
+      active: true,
+    });
+    expect(engine.proposals.get(1)).toMatchObject({ status: 'accepted', acted_by: 'test', promoted_row_num: 1 });
+    const body = readFileSync(pagePath, 'utf-8');
+    expect(body).toContain('A will happen');
+    expect(body).toContain('2024-03-02');
+    expect(body).toContain('proposal:1');
+  });
+
+  test('accept omits since_date when only fallback effective date is available', async () => {
+    const brainDir = makeBrainDir();
+    writePage(brainDir, 'topics/fallback', '# Fallback\n');
+    const engine = new FakeProposalEngine(brainDir);
+    engine.addPage({ id: 11, source_id: 'default', slug: 'topics/fallback', effective_date: new Date('2024-03-02T00:00:00Z'), effective_date_source: 'fallback' });
+    engine.addProposal({ id: 2, page_slug: 'topics/fallback', claim_text: 'Fallback date should not be trusted' });
+
+    const result = await acceptTakeProposal(engine as unknown as BrainEngine, 2, { actedBy: 'test' });
+
+    expect(result.since_date).toBeUndefined();
+    expect((engine.addedBatches[0][0] as Record<string, unknown>).since_date).toBeUndefined();
+  });
+
+  test('accept rolls back markdown if DB mirror fails', async () => {
+    const brainDir = makeBrainDir();
+    const original = '# Rollback\n';
+    const pagePath = writePage(brainDir, 'topics/rollback', original);
+    const engine = new FakeProposalEngine(brainDir);
+    engine.failAddTakes = true;
+    engine.addPage({ id: 12, source_id: 'default', slug: 'topics/rollback', effective_date: new Date('2024-03-02T00:00:00Z'), effective_date_source: 'frontmatter' });
+    engine.addProposal({ id: 3, page_slug: 'topics/rollback', claim_text: 'Should not persist' });
+
+    await expect(acceptTakeProposal(engine as unknown as BrainEngine, 3, { actedBy: 'test' })).rejects.toThrow('addTakesBatch failed');
+    expect(readFileSync(pagePath, 'utf-8')).toBe(original);
+    expect(engine.proposals.get(3)).toMatchObject({ status: 'pending', promoted_row_num: null });
+  });
+
+  test('accept is idempotent after a proposal has already been promoted', async () => {
+    const brainDir = makeBrainDir();
+    writePage(brainDir, 'topics/done', '# Done\n');
+    const engine = new FakeProposalEngine(brainDir);
+    engine.addPage({ id: 13, source_id: 'default', slug: 'topics/done', effective_date: new Date('2024-03-02T00:00:00Z'), effective_date_source: 'frontmatter' });
+    engine.addProposal({ id: 4, page_slug: 'topics/done', claim_text: 'Already accepted', status: 'accepted', promoted_row_num: 9 });
+
+    const result = await acceptTakeProposal(engine as unknown as BrainEngine, 4, { actedBy: 'test' });
+    expect(result).toMatchObject({ ok: true, proposal_id: 4, row_num: 9, status: 'accepted', idempotent: true, since_date: '2024-03-02' });
+    expect(engine.addedBatches).toHaveLength(0);
+  });
+
+  test('reject stamps pending proposals and is idempotent for rejected proposals', async () => {
+    const brainDir = makeBrainDir();
+    const engine = new FakeProposalEngine(brainDir);
+    engine.addProposal({ id: 5, page_slug: 'topics/reject', claim_text: 'Reject me' });
+
+    const first = await rejectTakeProposal(engine as unknown as BrainEngine, 5, { actedBy: 'reviewer', reason: 'not supported' });
+    expect(first).toEqual({ ok: true, proposal_id: 5, status: 'rejected', idempotent: false, reason: 'not supported' });
+    expect(engine.proposals.get(5)).toMatchObject({ status: 'rejected', acted_by: 'reviewer' });
+
+    const second = await rejectTakeProposal(engine as unknown as BrainEngine, 5, { actedBy: 'reviewer', reason: 'not supported' });
+    expect(second).toEqual({ ok: true, proposal_id: 5, status: 'rejected', idempotent: true, reason: 'not supported' });
+  });
+
+  test('reject refuses accepted proposals', async () => {
+    const brainDir = makeBrainDir();
+    const engine = new FakeProposalEngine(brainDir);
+    engine.addProposal({ id: 6, page_slug: 'topics/accepted', claim_text: 'Accepted', status: 'accepted', promoted_row_num: 1 });
+
+    await expect(rejectTakeProposal(engine as unknown as BrainEngine, 6, { actedBy: 'reviewer' })).rejects.toThrow('already accepted');
+  });
+
+  test('source isolation: federated sourceIds scope filters list and blocks accept/reject outside scope', async () => {
+    const brainDir = makeBrainDir();
+    writePage(brainDir, 'topics/other', '# Other\n');
+    const engine = new FakeProposalEngine(brainDir);
+    engine.addPage({ id: 20, source_id: 'other', slug: 'topics/other', effective_date: null, effective_date_source: null });
+    engine.addProposal({ id: 7, page_slug: 'topics/other', claim_text: 'Foreign claim', source_id: 'other' });
+    engine.addProposal({ id: 8, page_slug: 'topics/mine', claim_text: 'My claim', source_id: 'mine' });
+
+    // Federated array scope (ctx.auth.allowedSources) filters the list.
+    const scoped = await listTakeProposals(engine as unknown as BrainEngine, { sourceIds: ['mine'] });
+    expect(scoped.map((r) => r.id)).toEqual([8]);
+    // Array takes precedence over scalar.
+    const arrayWins = await listTakeProposals(engine as unknown as BrainEngine, { sourceIds: ['mine'], sourceId: 'other' });
+    expect(arrayWins.map((r) => r.id)).toEqual([8]);
+
+    // Accept and reject refuse proposals outside the federated scope.
+    await expect(acceptTakeProposal(engine as unknown as BrainEngine, 7, { sourceIds: ['mine'] })).rejects.toThrow('outside source scope');
+    await expect(rejectTakeProposal(engine as unknown as BrainEngine, 7, { sourceIds: ['mine'] })).rejects.toThrow('outside source scope');
+    // Scalar scope still enforced too.
+    await expect(rejectTakeProposal(engine as unknown as BrainEngine, 7, { sourceId: 'mine' })).rejects.toThrow('outside source scope');
+    expect(engine.proposals.get(7)).toMatchObject({ status: 'pending' });
+  });
+});
+
+describe('take proposal MCP operation schema', () => {
+  test('exposes list/accept/reject through the shared operation registry with correct scopes', () => {
+    const byName = Object.fromEntries(operations.map((op) => [op.name, op]));
+    expect(byName.takes_propose_list?.scope).toBe('read');
+    expect(byName.takes_propose_accept?.scope).toBe('write');
+    expect(byName.takes_propose_reject?.scope).toBe('write');
+    expect(byName.takes_propose_accept.params.proposal_id.required).toBe(true);
+    expect(byName.takes_propose_reject.params.proposal_id.required).toBe(true);
+
+    // accept writes the source markdown under sync.repo_path — local CLI only (D17).
+    expect(byName.takes_propose_accept?.localOnly).toBe(true);
+
+    const defs = Object.fromEntries(buildToolDefs(operations).map((def) => [def.name, def]));
+    expect(defs.takes_propose_accept.inputSchema.required).toEqual(['proposal_id']);
+    expect(defs.takes_propose_reject.inputSchema.required).toEqual(['proposal_id']);
+    expect(Object.keys(defs.takes_propose_list.inputSchema.properties)).toEqual(['limit', 'offset', 'status']);
+  });
+
+  test('takes_propose_list handler scopes via sourceScopeOpts — fail-closed when ctx.remote is undefined', async () => {
+    const engine = new FakeProposalEngine(makeBrainDir());
+    engine.addProposal({ id: 1, page_slug: 'topics/a', claim_text: 'mine', source_id: 'mine' });
+    engine.addProposal({ id: 2, page_slug: 'topics/b', claim_text: 'other', source_id: 'other' });
+    const op = operations.find((o) => o.name === 'takes_propose_list')!;
+
+    // ctx.remote undefined (the invariant's fail-closed case) with a scalar sourceId:
+    // scoping must still apply — the old `ctx.remote ? ctx.sourceId : undefined` dropped it.
+    const scalar = await op.handler({ engine, sourceId: 'mine', remote: undefined } as never, {});
+    expect((scalar as TakeProposalRow[]).map((r) => r.id)).toEqual([1]);
+
+    // Federated array (ctx.auth.allowedSources) wins over scalar.
+    const federated = await op.handler(
+      { engine, sourceId: 'mine', remote: true, auth: { allowedSources: ['other'] } } as never,
+      {},
+    );
+    expect((federated as TakeProposalRow[]).map((r) => r.id)).toEqual([2]);
+  });
+});
@@ -218,9 +218,10 @@ describe('think op — read-only on remote callers (Lane D landed)', () => {
      saved_slug: string | null;
      warnings: string[];
    };
-    // Codex P1 #7: remote save/take is silently disabled.
+    // Codex P1 #7: remote save/take is blocked explicitly at the trust boundary.
    expect(env.remote_persisted_blocked).toBe(true);
    expect(env.saved_slug).toBeNull();
+    expect(env.warnings).toContain('REMOTE_PERSISTED_BLOCKED');
    // Without API key, gather succeeds but synthesis is skipped.
    expect(env.warnings).toContain('NO_ANTHROPIC_API_KEY');
  });
@@ -1,7 +1,7 @@
 import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
 import { PGLiteEngine } from '../src/core/pglite-engine.ts';
 import { operationsByName } from '../src/core/operations.ts';
-import { runThink, persistSynthesis, type ThinkLLMClient } from '../src/core/think/index.ts';
+import { runThink, persistSynthesis, persistThinkTake, type ThinkLLMClient } from '../src/core/think/index.ts';
 import { sanitizeTakeForPrompt, renderTakesBlock } from '../src/core/think/sanitize.ts';
 import { resolveCitations, parseInlineCitations, normalizeStructuredCitations } from '../src/core/think/cite-render.ts';
 import { runGather } from '../src/core/think/gather.ts';
@@ -369,6 +369,68 @@ describe('runThink + persistSynthesis — #1698 never persist empty', () => {
    const saved = await persistSynthesis(engine, legacy);
    expect(saved.slug).toContain('synthesis/legacy-backcompat');
  });
+
+  test('persistThinkTake appends the synthesis answer as the next anchor take (#2556)', async () => {
+    const target = await engine.putPage('notes/think-take-target-example', {
+      title: 'Think take target',
+      type: 'note',
+      compiled_truth: 'A safe placeholder page for think take persistence.',
+    });
+    const result: any = {
+      question: 'what should this page remember?',
+      answer: 'This page should remember the synthesized placeholder insight.',
+      citations: [],
+      gaps: [],
+      pagesGathered: 0,
+      takesGathered: 0,
+      graphHits: 0,
+      modelUsed: 'stub',
+      rounds: 1,
+      warnings: [],
+      synthesisOk: true,
+      diagnostics: { pagesFromHybrid: 0, takesFromKeyword: 0, takesFromVector: 0, graphHits: 0 },
+    };
+
+    const persisted = await persistThinkTake(engine, result, { anchor: target.slug });
+
+    expect(persisted).toEqual({ rowNum: 1, inserted: 1, warnings: [] });
+    const takes = await engine.listTakes({ page_id: target.id });
+    expect(takes).toHaveLength(1);
+    expect(takes[0]).toMatchObject({
+      row_num: 1,
+      claim: 'This page should remember the synthesized placeholder insight.',
+      kind: 'take',
+      holder: 'brain',
+      source: 'gbrain think',
+    });
+  });
+
+  test('persistThinkTake refuses empty/no-LLM synthesis instead of writing a blank take (#2556)', async () => {
+    const target = await engine.putPage('notes/think-take-empty-example', {
+      title: 'Think take empty target',
+      type: 'note',
+      compiled_truth: 'A safe placeholder page for empty think take persistence.',
+    });
+    const result: any = {
+      question: 'empty synthesis',
+      answer: '(no LLM available)',
+      citations: [],
+      gaps: [],
+      pagesGathered: 0,
+      takesGathered: 0,
+      graphHits: 0,
+      modelUsed: 'stub',
+      rounds: 0,
+      warnings: [],
+      synthesisOk: false,
+      diagnostics: { pagesFromHybrid: 0, takesFromKeyword: 0, takesFromVector: 0, graphHits: 0 },
+    };
+
+    const persisted = await persistThinkTake(engine, result, { anchor: target.slug });
+
+    expect(persisted).toEqual({ rowNum: null, inserted: 0, warnings: ['TAKE_EMPTY_NOT_PERSISTED'] });
+    expect(await engine.listTakes({ page_id: target.id })).toHaveLength(0);
+  });
 });

 describe('think MCP op — #1698 C3 + #10', () => {