feat(takes): review-first take proposal list/accept/reject over MCP + CLI (#2269 )

Takeover of PR #2418. Exposes the take_proposals review queue as three contract-first operations — takes_propose_list (scope: read), takes_propose_accept / takes_propose_reject (scope: write) — plus the matching `gbrain takes propose list|accept|reject` CLI subcommands, so server-side/no-TTY deployments can drain the proposal queue natively in chat. Preserves the D17 no-auto-accept posture: accept is the only path from queue to canonical fence, and it stays an explicit reviewed action. Repairs over the original PR: - Source isolation routes through sourceScopeOpts(ctx) (federated allowedSources array > scalar ctx.sourceId > nothing) instead of the fail-open `ctx.remote ? ctx.sourceId : undefined`, which ignored federated grants and treated an unset remote flag as trusted. - D4 holder privacy: all three ops thread ctx.takesHoldersAllowList so an allow-listed MCP token can neither see nor act on proposals for holders outside its grant (empty list matches nothing, fail-closed). - Containment guard on the accept path: a proposal's page_slug can never resolve a markdown write outside the brain directory. - Invalid status values error instead of silently returning nothing. - Rebased onto master (takes.ts source-scope refactor). Tests run against a real in-memory PGLite engine (not a hand-rolled SQL fake), covering scope isolation, holder allow-list, promote + markdown mirror + idempotency, rollback on DB failure, and reject lifecycle. Co-authored-by: Mr-B-1 <Mr-B-1@users.noreply.github.com> Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-18 09:48:17 +00:00 · 2026-07-21 15:01:18 -07:00
4 changed files with 761 additions and 1 deletions
@@ -29,6 +29,11 @@ import {
 } from '../core/takes-fence.ts';
 import { withPageLock } from '../core/page-lock.ts';
 import { resolveSourceId } from '../core/source-resolver.ts';
+import {
+  acceptTakeProposal,
+  listTakeProposals,
+  rejectTakeProposal,
+} from '../core/take-proposals.ts';

 // --- Helpers ---

@@ -189,6 +194,81 @@ async function cmdSearch(engine: BrainEngine, args: string[]): Promise<void> {
  }
 }

+async function cmdPropose(engine: BrainEngine, args: string[]): Promise<void> {
+  const sub = args[0];
+  const rest = args.slice(1);
+  const json = flagPresent(rest, '--json') || flagPresent(args, '--json');
+  if (!sub || sub === 'list') {
+    const status = flagValue(rest, '--status') as 'pending' | 'accepted' | 'rejected' | 'superseded' | undefined;
+    const limit = parseInt(flagValue(rest, '--limit') ?? '30', 10);
+    const offset = parseInt(flagValue(rest, '--offset') ?? '0', 10);
+    const rows = await listTakeProposals(engine, { status: status ?? 'pending', limit, offset });
+    if (json) {
+      console.log(JSON.stringify(rows, null, 2));
+      return;
+    }
+    if (rows.length === 0) {
+      console.log(`No ${status ?? 'pending'} take proposals.`);
+      return;
+    }
+    for (const row of rows) {
+      const date = row.effective_date ? row.effective_date.slice(0, 10) : '';
+      const dateSource = row.effective_date_source ? ` ${row.effective_date_source}` : '';
+      const brier = row.predicted_brier === null || row.predicted_brier === undefined
+        ? ''
+        : ` • brier=${row.predicted_brier.toFixed(3)}`;
+      console.log(
+        `#${row.id} [${row.kind} • ${row.holder} • w=${Number(row.weight).toFixed(2)}${date ? ` • ${date}${dateSource}` : ''}${brier}]\n` +
+        `  ${row.page_slug}\n` +
+        `  ${row.claim_text}\n`,
+      );
+    }
+    return;
+  }
+
+  if (sub === 'accept') {
+    const id = parseInt(rest[0] ?? flagValue(rest, '--id') ?? '', 10);
+    if (!Number.isFinite(id)) {
+      console.error('Usage: gbrain takes propose accept <proposal_id> [--by <actor>] [--dir <path>] [--json]');
+      process.exit(1);
+    }
+    const result = await acceptTakeProposal(engine, id, {
+      actedBy: flagValue(rest, '--by') ?? 'gbrain-cli',
+      brainDir: flagValue(rest, '--dir'),
+    });
+    if (json) {
+      console.log(JSON.stringify(result, null, 2));
+      return;
+    }
+    const since = result.since_date ? ` since=${result.since_date}` : ' since=(unset: no real claim-date)';
+    const suffix = result.idempotent ? ' (already accepted)' : '';
+    console.log(`Accepted proposal #${result.proposal_id} → ${result.page_slug}#${result.row_num}${since}${suffix}.`);
+    return;
+  }
+
+  if (sub === 'reject') {
+    const id = parseInt(rest[0] ?? flagValue(rest, '--id') ?? '', 10);
+    if (!Number.isFinite(id)) {
+      console.error('Usage: gbrain takes propose reject <proposal_id> [--reason "..."] [--by <actor>] [--json]');
+      process.exit(1);
+    }
+    const result = await rejectTakeProposal(engine, id, {
+      actedBy: flagValue(rest, '--by') ?? 'gbrain-cli',
+      reason: flagValue(rest, '--reason'),
+    });
+    if (json) {
+      console.log(JSON.stringify(result, null, 2));
+      return;
+    }
+    const suffix = result.idempotent ? ' (already rejected)' : '';
+    console.log(`Rejected proposal #${result.proposal_id}${suffix}.`);
+    return;
+  }
+
+  console.error(`Unknown takes propose subcommand: ${sub}`);
+  process.exit(1);
+}
+
 async function cmdAdd(engine: BrainEngine, args: string[], sourceId?: string): Promise<void> {
  const slug = args[0];
  if (!slug) {
@@ -556,6 +636,12 @@ Subcommands:
                                          List takes for a page
  takes search "<query>" [--limit N] [--json]
                                          Keyword search across all takes
+  takes propose list [--status pending] [--limit N] [--offset N] [--json]
+                                          Review pending take proposals
+  takes propose accept <proposal_id> [--by <actor>] [--json]
+                                          Promote a reviewed proposal into the page's takes fence
+  takes propose reject <proposal_id> [--reason "..."] [--by <actor>] [--json]
+                                          Reject a proposal so it is not re-proposed
  takes add <slug> --claim "..." --kind <fact|take|bet|hunch> --who <holder>
                   [--weight 0.5] [--source "..."] [--since YYYY-MM]
                                          Append a take (markdown + DB)
@@ -584,6 +670,7 @@ Common flags:

  switch (sub) {
    case 'search':      return cmdSearch(engine, rest);
+    case 'propose':     return cmdPropose(engine, rest);
    case 'add':         return cmdAdd(engine, rest, await resolveTakesSourceId(engine));
    case 'update':      return cmdUpdate(engine, rest, await resolveTakesSourceId(engine));
    case 'supersede':   return cmdSupersede(engine, rest, await resolveTakesSourceId(engine));
@@ -26,6 +26,11 @@ import { isSearchMode } from './search/mode.ts';
 import { stampEvidence } from './search/evidence.ts';
 import type { SearchResult } from './types.ts';
 import { CJK_SLUG_CHARS } from './cjk.ts';
+import {
+  acceptTakeProposal,
+  listTakeProposals,
+  rejectTakeProposal,
+} from './take-proposals.ts';
 import * as db from './db.ts';
 import { VERSION } from '../version.ts';
 import {
@@ -1782,6 +1787,63 @@ const takes_search: Operation = {
  cliHints: { name: 'takes-search', positional: ['query'] },
 };

+const takes_propose_list: Operation = {
+  name: 'takes_propose_list',
+  description: 'List pending reviewed take proposals before promotion into canonical takes.',
+  scope: 'read',
+  params: {
+    limit: { type: 'number', description: 'Max rows (default 50, cap 500)' },
+    offset: { type: 'number', description: 'Skip first N rows' },
+    status: { type: 'string', description: 'pending | accepted | rejected | superseded (default pending)' },
+  },
+  handler: async (ctx, p) => {
+    // Source isolation + D4 holder privacy: same posture as takes_list /
+    // takes_search — federated grant > scalar source > nothing, and remote
+    // callers only see holders on their allow-list.
+    return listTakeProposals(ctx.engine, {
+      limit: p.limit as number | undefined,
+      offset: p.offset as number | undefined,
+      status: p.status as 'pending' | 'accepted' | 'rejected' | 'superseded' | undefined,
+      ...sourceScopeOpts(ctx),
+      holdersAllowList: ctx.takesHoldersAllowList,
+    });
+  },
+};
+
+const takes_propose_accept: Operation = {
+  name: 'takes_propose_accept',
+  description: 'Accept one reviewed take proposal, append it to the source markdown takes fence, mirror it to DB, and stamp the proposal accepted.',
+  scope: 'write',
+  params: {
+    proposal_id: { type: 'number', required: true },
+  },
+  handler: async (ctx, p) => {
+    return acceptTakeProposal(ctx.engine, p.proposal_id as number, {
+      actedBy: ctx.auth?.clientName ?? 'mcp',
+      ...sourceScopeOpts(ctx),
+      holdersAllowList: ctx.takesHoldersAllowList,
+    });
+  },
+};
+
+const takes_propose_reject: Operation = {
+  name: 'takes_propose_reject',
+  description: 'Reject one take proposal so review-first queues do not re-offer it.',
+  scope: 'write',
+  params: {
+    proposal_id: { type: 'number', required: true },
+    reason: { type: 'string', description: 'Operator note. Current schema records acted_by/acted_at, not a reason column.' },
+  },
+  handler: async (ctx, p) => {
+    return rejectTakeProposal(ctx.engine, p.proposal_id as number, {
+      actedBy: ctx.auth?.clientName ?? 'mcp',
+      reason: p.reason as string | undefined,
+      ...sourceScopeOpts(ctx),
+      holdersAllowList: ctx.takesHoldersAllowList,
+    });
+  },
+};
+
 /**
 * v0.30.0 (Slice A1): aggregate calibration scorecard. Pure SQL aggregation.
 *
@@ -5402,7 +5464,7 @@ export const operations: Operation[] = [
  // v0.36.1.0 (T7) — Hindsight calibration wave: read profile via MCP
  get_calibration_profile,
  // v0.28: Takes + think
-  takes_list, takes_search, think,
+  takes_list, takes_search, takes_propose_list, takes_propose_accept, takes_propose_reject, think,
  // v0.30: calibration aggregates over takes
  takes_scorecard, takes_calibration,
  // v0.28: whoami + scoped sources management
@@ -0,0 +1,356 @@
+import { existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs';
+import { dirname, join, resolve, sep } from 'node:path';
+import type { BrainEngine, TakeKind } from './engine.ts';
+import { upsertTakeRow } from './takes-fence.ts';
+import { withPageLock } from './page-lock.ts';
+
+/**
+ * Caller scope for the proposal review surface. Mirrors `sourceScopeOpts`
+ * precedence (federated `sourceIds` array > scalar `sourceId` > nothing) plus
+ * the D4 takes-holder allow-list: an MCP token restricted to certain holders
+ * must not see — or act on — proposals for other holders. An empty
+ * `holdersAllowList` array matches nothing (fail-closed), same as the
+ * engine's `holder = ANY($allowList)` behavior.
+ */
+export interface ProposalScope {
+  sourceId?: string;
+  sourceIds?: string[];
+  holdersAllowList?: string[];
+}
+
+function assertProposalInScope(
+  row: Record<string, unknown>,
+  scope: ProposalScope,
+  proposalId: number,
+): void {
+  const sourceOk =
+    scope.sourceIds && scope.sourceIds.length > 0
+      ? scope.sourceIds.includes(String(row.source_id))
+      : scope.sourceId
+        ? String(row.source_id) === scope.sourceId
+        : true;
+  if (!sourceOk) {
+    throw new Error(`take proposal ${proposalId} is outside your source scope`);
+  }
+  if (scope.holdersAllowList && !scope.holdersAllowList.includes(String(row.holder))) {
+    throw new Error(`take proposal ${proposalId} is outside your holder allow-list`);
+  }
+}
+
+const PROPOSAL_STATUSES = ['pending', 'accepted', 'rejected', 'superseded'] as const;
+
+export interface TakeProposalRow {
+  id: number;
+  source_id: string;
+  page_slug: string;
+  status: 'pending' | 'accepted' | 'rejected' | 'superseded';
+  claim_text: string;
+  kind: TakeKind;
+  holder: string;
+  weight: number;
+  domain?: string | null;
+  dedup_against_fence_rows?: unknown;
+  model_id: string;
+  proposed_at: string;
+  acted_at?: string | null;
+  acted_by?: string | null;
+  promoted_row_num?: number | null;
+  predicted_brier?: number | null;
+  predicted_brier_bucket_n?: number | null;
+  effective_date?: string | null;
+  effective_date_source?: string | null;
+}
+
+export interface TakeProposalAcceptResult {
+  ok: true;
+  proposal_id: number;
+  page_slug: string;
+  row_num: number;
+  status: 'accepted';
+  idempotent: boolean;
+  since_date?: string;
+}
+
+export interface TakeProposalRejectResult {
+  ok: true;
+  proposal_id: number;
+  status: 'rejected';
+  idempotent: boolean;
+  reason?: string;
+}
+
+function isoOrNull(value: unknown): string | null {
+  if (value === null || value === undefined) return null;
+  if (value instanceof Date) return value.toISOString();
+  return String(value);
+}
+
+function dateOnlyOrUndefined(value: unknown, source: unknown): string | undefined {
+  if (value === null || value === undefined) return undefined;
+  if (source === 'fallback') return undefined;
+  if (value instanceof Date) return value.toISOString().slice(0, 10);
+  const raw = String(value);
+  return /^\d{4}-\d{2}-\d{2}/.test(raw) ? raw.slice(0, 10) : undefined;
+}
+
+function numberOrNull(value: unknown): number | null {
+  if (value === null || value === undefined) return null;
+  const n = Number(value);
+  return Number.isFinite(n) ? n : null;
+}
+
+function mapProposalRow(row: Record<string, unknown>): TakeProposalRow {
+  return {
+    id: Number(row.id),
+    source_id: String(row.source_id),
+    page_slug: String(row.page_slug),
+    status: row.status as TakeProposalRow['status'],
+    claim_text: String(row.claim_text),
+    kind: String(row.kind) as TakeKind,
+    holder: String(row.holder),
+    weight: Number(row.weight),
+    domain: row.domain === undefined ? null : row.domain as string | null,
+    dedup_against_fence_rows: row.dedup_against_fence_rows,
+    model_id: String(row.model_id),
+    proposed_at: isoOrNull(row.proposed_at) ?? '',
+    acted_at: isoOrNull(row.acted_at),
+    acted_by: row.acted_by === undefined ? null : row.acted_by as string | null,
+    promoted_row_num: numberOrNull(row.promoted_row_num),
+    predicted_brier: numberOrNull(row.predicted_brier),
+    predicted_brier_bucket_n: numberOrNull(row.predicted_brier_bucket_n),
+    effective_date: isoOrNull(row.effective_date),
+    effective_date_source: row.effective_date_source === undefined ? null : row.effective_date_source as string | null,
+  };
+}
+
+async function resolveBrainDir(engine: BrainEngine, explicitDir?: string): Promise<string> {
+  if (explicitDir) return explicitDir;
+  const configured = await engine.getConfig('sync.repo_path');
+  if (configured) return configured;
+  throw new Error('No brain directory configured. Pass brainDir or set sync.repo_path.');
+}
+
+function pageFilePath(brainDir: string, slug: string): string {
+  return join(brainDir, `${slug}.md`);
+}
+
+export async function listTakeProposals(
+  engine: BrainEngine,
+  opts: {
+    limit?: number;
+    offset?: number;
+    status?: TakeProposalRow['status'];
+  } & ProposalScope = {},
+): Promise<TakeProposalRow[]> {
+  const limit = Math.max(1, Math.min(500, Math.floor(opts.limit ?? 50)));
+  const offset = Math.max(0, Math.floor(opts.offset ?? 0));
+  const status = opts.status ?? 'pending';
+  if (!PROPOSAL_STATUSES.includes(status)) {
+    throw new Error(`invalid proposal status '${status}'. Expected: ${PROPOSAL_STATUSES.join(' | ')}`);
+  }
+  const params: unknown[] = [status];
+  const where: string[] = ['tp.status = $1'];
+  if (opts.sourceIds && opts.sourceIds.length > 0) {
+    params.push(opts.sourceIds);
+    where.push(`tp.source_id = ANY($${params.length}::text[])`);
+  } else if (opts.sourceId) {
+    params.push(opts.sourceId);
+    where.push(`tp.source_id = $${params.length}`);
+  }
+  if (opts.holdersAllowList) {
+    params.push(opts.holdersAllowList);
+    where.push(`tp.holder = ANY($${params.length}::text[])`);
+  }
+  params.push(limit, offset);
+  const rows = await engine.executeRaw(
+    `SELECT
+       tp.id, tp.source_id, tp.page_slug, tp.status, tp.claim_text,
+       tp.kind, tp.holder, tp.weight, tp.domain, tp.dedup_against_fence_rows,
+       tp.model_id, tp.proposed_at, tp.acted_at, tp.acted_by,
+       tp.promoted_row_num, tp.predicted_brier, tp.predicted_brier_bucket_n,
+       p.effective_date, p.effective_date_source
+     FROM take_proposals tp
+     LEFT JOIN pages p ON p.slug = tp.page_slug AND p.source_id = tp.source_id
+     WHERE ${where.join(' AND ')}
+     ORDER BY
+       CASE WHEN tp.predicted_brier IS NULL THEN 1 ELSE 0 END,
+       tp.predicted_brier ASC NULLS LAST,
+       tp.proposed_at DESC
+     LIMIT $${params.length - 1} OFFSET $${params.length}`,
+    params,
+  );
+  return rows.map((r) => mapProposalRow(r as Record<string, unknown>));
+}
+
+export async function acceptTakeProposal(
+  engine: BrainEngine,
+  proposalId: number,
+  opts: { brainDir?: string; actedBy?: string } & ProposalScope = {},
+): Promise<TakeProposalAcceptResult> {
+  const proposalLookup = await engine.executeRaw<{ page_slug: string; source_id: string; holder: string; status: string; promoted_row_num: number | null }>(
+    `SELECT page_slug, source_id, holder, status, promoted_row_num
+     FROM take_proposals WHERE id = $1 LIMIT 1`,
+    [proposalId],
+  );
+  const existing = proposalLookup[0];
+  if (!existing) throw new Error(`take proposal not found: ${proposalId}`);
+  assertProposalInScope(existing, opts, proposalId);
+
+  const actedBy = opts.actedBy ?? 'gbrain-cli';
+  const brainDir = await resolveBrainDir(engine, opts.brainDir);
+
+  return withPageLock(existing.page_slug, async () => {
+    const path = pageFilePath(brainDir, existing.page_slug);
+    // Containment guard: page_slug comes from the DB, but this function is
+    // reachable from remote (MCP write-scope) callers — never let a
+    // traversal-shaped slug escape the brain directory.
+    if (!resolve(path).startsWith(resolve(brainDir) + sep)) {
+      throw new Error(`take proposal ${proposalId} resolves outside the brain directory`);
+    }
+    if (!existsSync(path)) {
+      throw new Error(`source markdown page not found: ${path}`);
+    }
+
+    let originalBody: string | null = null;
+    let wroteBody = false;
+
+    try {
+      return await engine.transaction(async (tx) => {
+        const rows = await tx.executeRaw<Record<string, unknown>>(
+          `SELECT
+             tp.id, tp.source_id, tp.page_slug, tp.status, tp.claim_text,
+             tp.kind, tp.holder, tp.weight, tp.model_id, tp.promoted_row_num,
+             p.id AS page_id, p.effective_date, p.effective_date_source
+           FROM take_proposals tp
+           JOIN pages p ON p.slug = tp.page_slug AND p.source_id = tp.source_id
+           WHERE tp.id = $1
+           FOR UPDATE OF tp`,
+          [proposalId],
+        );
+        const row = rows[0];
+        if (!row) throw new Error(`take proposal not found: ${proposalId}`);
+        assertProposalInScope(row, opts, proposalId);
+
+        const promotedRowNum = numberOrNull(row.promoted_row_num);
+        if (row.status === 'accepted' && promotedRowNum !== null) {
+          return {
+            ok: true,
+            proposal_id: proposalId,
+            page_slug: String(row.page_slug),
+            row_num: promotedRowNum,
+            status: 'accepted',
+            idempotent: true,
+            since_date: dateOnlyOrUndefined(row.effective_date, row.effective_date_source),
+          } satisfies TakeProposalAcceptResult;
+        }
+        if (row.status !== 'pending') {
+          throw new Error(`take proposal ${proposalId} is ${row.status}; only pending proposals can be accepted`);
+        }
+
+        const pageId = Number(row.page_id);
+        await tx.executeRaw('SELECT pg_advisory_xact_lock($1::bigint)', [pageId]);
+
+        const dupes = await tx.executeRaw<{ row_num: number }>(
+          `SELECT row_num
+           FROM takes
+           WHERE page_id = $1 AND active = true
+             AND lower(trim(claim)) = lower(trim($2))
+           LIMIT 1`,
+          [pageId, row.claim_text],
+        );
+        if (dupes.length > 0) {
+          throw new Error(`take proposal ${proposalId} duplicates existing take row #${dupes[0].row_num}`);
+        }
+
+        originalBody = readFileSync(path, 'utf-8');
+        const sinceDate = dateOnlyOrUndefined(row.effective_date, row.effective_date_source);
+        const { body: nextBody, rowNum } = upsertTakeRow(originalBody, {
+          claim: String(row.claim_text),
+          kind: String(row.kind),
+          holder: String(row.holder),
+          weight: Number(row.weight),
+          source: `proposal:${proposalId}`,
+          sinceDate,
+          active: true,
+        });
+        mkdirSync(dirname(path), { recursive: true });
+        writeFileSync(path, nextBody, 'utf-8');
+        wroteBody = true;
+
+        await tx.addTakesBatch([{
+          page_id: pageId,
+          row_num: rowNum,
+          claim: String(row.claim_text),
+          kind: String(row.kind),
+          holder: String(row.holder),
+          weight: Number(row.weight),
+          since_date: sinceDate,
+          source: `proposal:${proposalId}`,
+          active: true,
+          superseded_by: null,
+        }]);
+        const stamped = await tx.executeRaw<{ promoted_row_num: number }>(
+          `UPDATE take_proposals
+           SET status = 'accepted',
+               acted_at = now(),
+               acted_by = $2,
+               promoted_row_num = $3
+           WHERE id = $1 AND status = 'pending'
+           RETURNING promoted_row_num`,
+          [proposalId, actedBy, rowNum],
+        );
+        if (stamped.length === 0) {
+          throw new Error(`take proposal ${proposalId} was not stamped accepted`);
+        }
+        return {
+          ok: true,
+          proposal_id: proposalId,
+          page_slug: String(row.page_slug),
+          row_num: rowNum,
+          status: 'accepted',
+          idempotent: false,
+          since_date: sinceDate,
+        } satisfies TakeProposalAcceptResult;
+      });
+    } catch (err) {
+      if (wroteBody && originalBody !== null) {
+        writeFileSync(path, originalBody, 'utf-8');
+      }
+      throw err;
+    }
+  });
+}
+
+export async function rejectTakeProposal(
+  engine: BrainEngine,
+  proposalId: number,
+  opts: { actedBy?: string; reason?: string } & ProposalScope = {},
+): Promise<TakeProposalRejectResult> {
+  const actedBy = opts.actedBy ?? 'gbrain-cli';
+  return engine.transaction(async (tx) => {
+    const rows = await tx.executeRaw<Record<string, unknown>>(
+      `SELECT id, source_id, holder, status, promoted_row_num
+       FROM take_proposals WHERE id = $1
+       FOR UPDATE`,
+      [proposalId],
+    );
+    const row = rows[0];
+    if (!row) throw new Error(`take proposal not found: ${proposalId}`);
+    assertProposalInScope(row, opts, proposalId);
+    if (row.status === 'rejected') {
+      return { ok: true, proposal_id: proposalId, status: 'rejected', idempotent: true, reason: opts.reason };
+    }
+    if (row.status === 'accepted' || numberOrNull(row.promoted_row_num) !== null) {
+      throw new Error(`take proposal ${proposalId} is already accepted and cannot be rejected`);
+    }
+    await tx.executeRaw(
+      `UPDATE take_proposals
+       SET status = 'rejected',
+           acted_at = now(),
+           acted_by = $2
+       WHERE id = $1`,
+      [proposalId, actedBy],
+    );
+    return { ok: true, proposal_id: proposalId, status: 'rejected', idempotent: false, reason: opts.reason };
+  });
+}
@@ -0,0 +1,255 @@
+/**
+ * Review-first take proposal lifecycle (#2269, takeover of PR #2418):
+ * list / accept / reject over the shared operation registry, plus the
+ * source-isolation + holder-allow-list repairs.
+ *
+ * Real PGLite engine (in-memory, no DATABASE_URL) so the SQL shapes
+ * (FOR UPDATE OF, pg_advisory_xact_lock, ANY($::text[])) run for real.
+ */
+import { afterAll, beforeAll, describe, expect, test } from 'bun:test';
+import { mkdtempSync, readFileSync, rmSync, writeFileSync, mkdirSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { dirname, join } from 'node:path';
+import { PGLiteEngine } from '../src/core/pglite-engine.ts';
+import type { BrainEngine } from '../src/core/engine.ts';
+import { dispatchToolCall } from '../src/mcp/dispatch.ts';
+import { operations } from '../src/core/operations.ts';
+import { buildToolDefs } from '../src/mcp/tool-defs.ts';
+import {
+  acceptTakeProposal,
+  listTakeProposals,
+  rejectTakeProposal,
+} from '../src/core/take-proposals.ts';
+
+let engine: PGLiteEngine;
+let brainDir: string;
+
+async function addSource(id: string): Promise<void> {
+  await engine.executeRaw(
+    `INSERT INTO sources (id, name, local_path, config, created_at)
+     VALUES ($1, $1, NULL, '{}'::jsonb, NOW()) ON CONFLICT (id) DO NOTHING`,
+    [id],
+  );
+}
+
+async function insertProposal(p: {
+  source_id: string;
+  page_slug: string;
+  claim: string;
+  holder?: string;
+  kind?: string;
+  weight?: number;
+}): Promise<number> {
+  const rows = 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, model_id)
+     VALUES ($1, $2, md5($6 || random()::text), 'v1', 'run-test', $6, $3, $4, $5, 'test-model')
+     RETURNING id`,
+    [p.source_id, p.page_slug, p.kind ?? 'take', p.holder ?? 'garry', p.weight ?? 0.7, p.claim],
+  );
+  return Number(rows[0].id);
+}
+
+function writePage(slug: string, body = '# Page\n'): string {
+  const path = join(brainDir, `${slug}.md`);
+  mkdirSync(dirname(path), { recursive: true });
+  writeFileSync(path, body, 'utf-8');
+  return path;
+}
+
+beforeAll(async () => {
+  brainDir = mkdtempSync(join(tmpdir(), 'gbrain-take-proposals-'));
+  engine = new PGLiteEngine();
+  await engine.connect({});
+  await engine.initSchema();
+  await engine.setConfig('sync.repo_path', brainDir);
+  await addSource('tenant-a');
+  await addSource('tenant-b');
+  await engine.putPage('topics/a', { title: 'A', type: 'concept', compiled_truth: 'Body' }, { sourceId: 'tenant-a' });
+  await engine.putPage('topics/b', { title: 'B', type: 'concept', compiled_truth: 'Body' }, { sourceId: 'tenant-b' });
+  // Real content date on topics/a so accept threads since_date.
+  await engine.executeRaw(
+    `UPDATE pages SET effective_date = '2024-03-02', effective_date_source = 'frontmatter' WHERE slug = 'topics/a'`,
+  );
+}, 30000);
+
+afterAll(async () => {
+  await engine.disconnect();
+  rmSync(brainDir, { recursive: true, force: true });
+});
+
+describe('listTakeProposals — scope isolation', () => {
+  let idA: number;
+  let idB: number;
+  let idWorld: number;
+
+  beforeAll(async () => {
+    idA = await insertProposal({ source_id: 'tenant-a', page_slug: 'topics/a', claim: 'A will happen', holder: 'garry' });
+    idB = await insertProposal({ source_id: 'tenant-b', page_slug: 'topics/b', claim: 'B will happen', holder: 'garry' });
+    idWorld = await insertProposal({ source_id: 'tenant-a', page_slug: 'topics/a', claim: 'Public claim', holder: 'world' });
+  });
+
+  test('unscoped (trusted local) sees all pending proposals', async () => {
+    const rows = await listTakeProposals(engine);
+    const ids = rows.map(r => r.id);
+    expect(ids).toContain(idA);
+    expect(ids).toContain(idB);
+    expect(ids).toContain(idWorld);
+  });
+
+  test('scalar sourceId filters to that source', async () => {
+    const rows = await listTakeProposals(engine, { sourceId: 'tenant-b' });
+    expect(rows.map(r => r.id)).toEqual([idB]);
+  });
+
+  test('federated sourceIds array filters to the grant', async () => {
+    const rows = await listTakeProposals(engine, { sourceIds: ['tenant-a'] });
+    const ids = rows.map(r => r.id).sort();
+    expect(ids).toEqual([idA, idWorld].sort());
+  });
+
+  test('holdersAllowList hides other holders; empty list matches nothing', async () => {
+    const world = await listTakeProposals(engine, { holdersAllowList: ['world'] });
+    expect(world.map(r => r.id)).toEqual([idWorld]);
+    expect(await listTakeProposals(engine, { holdersAllowList: [] })).toEqual([]);
+  });
+
+  test('invalid status is rejected', async () => {
+    await expect(listTakeProposals(engine, { status: 'garbage' as never })).rejects.toThrow('invalid proposal status');
+  });
+
+  test('takes_propose_list op threads source scope + holder allow-list (remote)', async () => {
+    const result = await dispatchToolCall(engine, 'takes_propose_list', {}, {
+      remote: true,
+      sourceId: 'tenant-a',
+      takesHoldersAllowList: ['world'],
+    });
+    expect(result.isError).toBeFalsy();
+    const rows = JSON.parse(result.content[0].text) as Array<{ id: number; holder: string; source_id: string }>;
+    expect(rows.map(r => r.id)).toEqual([idWorld]);
+  });
+
+  test('list surfaces page effective-date metadata', async () => {
+    const rows = await listTakeProposals(engine, { sourceIds: ['tenant-a'] });
+    const a = rows.find(r => r.id === idA)!;
+    expect(a.effective_date).toContain('2024-03-02');
+    expect(a.effective_date_source).toBe('frontmatter');
+  });
+});
+
+describe('acceptTakeProposal', () => {
+  test('promotes: writes markdown fence, mirrors DB, stamps proposal; idempotent re-accept', async () => {
+    const pagePath = writePage('topics/a', '# A\n\nBody\n');
+    const id = await insertProposal({ source_id: 'tenant-a', page_slug: 'topics/a', claim: 'Promote me', holder: 'garry' });
+
+    const result = await acceptTakeProposal(engine, id, { actedBy: 'test', brainDir });
+    expect(result).toMatchObject({ ok: true, proposal_id: id, page_slug: 'topics/a', status: 'accepted', idempotent: false, since_date: '2024-03-02' });
+
+    const body = readFileSync(pagePath, 'utf-8');
+    expect(body).toContain('Promote me');
+    expect(body).toContain(`proposal:${id}`);
+
+    const takes = await engine.listTakes({ page_slug: 'topics/a', sourceId: 'tenant-a' });
+    const promoted = takes.find(t => t.claim === 'Promote me')!;
+    expect(promoted).toBeDefined();
+    expect(promoted.row_num).toBe(result.row_num);
+    expect(promoted.since_date).toContain('2024-03-02');
+
+    const [stamped] = await engine.executeRaw<{ status: string; acted_by: string; promoted_row_num: number }>(
+      `SELECT status, acted_by, promoted_row_num FROM take_proposals WHERE id = $1`, [id],
+    );
+    expect(stamped).toMatchObject({ status: 'accepted', acted_by: 'test' });
+    expect(Number(stamped.promoted_row_num)).toBe(result.row_num);
+
+    const again = await acceptTakeProposal(engine, id, { actedBy: 'test', brainDir });
+    expect(again).toMatchObject({ ok: true, idempotent: true, row_num: result.row_num });
+  });
+
+  test('refuses out-of-scope source and out-of-allow-list holder', async () => {
+    writePage('topics/a', '# A\n');
+    const id = await insertProposal({ source_id: 'tenant-a', page_slug: 'topics/a', claim: 'Scoped claim', holder: 'garry' });
+    await expect(acceptTakeProposal(engine, id, { brainDir, sourceIds: ['tenant-b'] })).rejects.toThrow('outside your source scope');
+    await expect(acceptTakeProposal(engine, id, { brainDir, sourceId: 'tenant-b' })).rejects.toThrow('outside your source scope');
+    await expect(acceptTakeProposal(engine, id, { brainDir, holdersAllowList: ['world'] })).rejects.toThrow('outside your holder allow-list');
+    const [row] = await engine.executeRaw<{ status: string }>(`SELECT status FROM take_proposals WHERE id = $1`, [id]);
+    expect(row.status).toBe('pending');
+  });
+
+  test('refuses a duplicate of an existing active take', async () => {
+    writePage('topics/a', '# A\n');
+    const id = await insertProposal({ source_id: 'tenant-a', page_slug: 'topics/a', claim: '  PROMOTE ME  ', holder: 'garry' });
+    await expect(acceptTakeProposal(engine, id, { brainDir })).rejects.toThrow('duplicates existing take row');
+  });
+
+  test('rolls back markdown and proposal stamp when the DB mirror fails', async () => {
+    const original = '# Rollback\n';
+    const pagePath = writePage('topics/a-rollback', original);
+    await engine.putPage('topics/a-rollback', { title: 'R', type: 'concept', compiled_truth: 'Body' }, { sourceId: 'tenant-a' });
+    const id = await insertProposal({ source_id: 'tenant-a', page_slug: 'topics/a-rollback', claim: 'Should not persist', holder: 'garry' });
+
+    // Wrap the real engine: same transaction machinery, injected batch failure.
+    const failing = Object.create(engine) as BrainEngine;
+    failing.transaction = <T>(fn: (tx: BrainEngine) => Promise<T>) =>
+      engine.transaction((tx) => {
+        const failingTx = Object.create(tx) as BrainEngine;
+        failingTx.addTakesBatch = async () => { throw new Error('injected addTakesBatch failure'); };
+        return fn(failingTx);
+      });
+
+    await expect(acceptTakeProposal(failing, id, { brainDir })).rejects.toThrow('injected addTakesBatch failure');
+    expect(readFileSync(pagePath, 'utf-8')).toBe(original);
+    const [row] = await engine.executeRaw<{ status: string; promoted_row_num: number | null }>(
+      `SELECT status, promoted_row_num FROM take_proposals WHERE id = $1`, [id],
+    );
+    expect(row.status).toBe('pending');
+    expect(row.promoted_row_num).toBeNull();
+  });
+
+  test('accept without a real content date leaves since_date unset', async () => {
+    writePage('topics/b', '# B\n');
+    const id = await insertProposal({ source_id: 'tenant-b', page_slug: 'topics/b', claim: 'No date here', holder: 'garry' });
+    const result = await acceptTakeProposal(engine, id, { brainDir });
+    expect(result.since_date).toBeUndefined();
+  });
+});
+
+describe('rejectTakeProposal', () => {
+  test('stamps pending → rejected; idempotent; refuses accepted; enforces scope', async () => {
+    const id = await insertProposal({ source_id: 'tenant-a', page_slug: 'topics/a', claim: 'Reject me', holder: 'garry' });
+
+    await expect(rejectTakeProposal(engine, id, { sourceId: 'tenant-b' })).rejects.toThrow('outside your source scope');
+    await expect(rejectTakeProposal(engine, id, { holdersAllowList: ['world'] })).rejects.toThrow('outside your holder allow-list');
+
+    const first = await rejectTakeProposal(engine, id, { actedBy: 'reviewer', reason: 'not supported' });
+    expect(first).toEqual({ ok: true, proposal_id: id, status: 'rejected', idempotent: false, reason: 'not supported' });
+    const [row] = await engine.executeRaw<{ status: string; acted_by: string }>(
+      `SELECT status, acted_by FROM take_proposals WHERE id = $1`, [id],
+    );
+    expect(row).toMatchObject({ status: 'rejected', acted_by: 'reviewer' });
+
+    const second = await rejectTakeProposal(engine, id, { actedBy: 'reviewer' });
+    expect(second.idempotent).toBe(true);
+
+    writePage('topics/a', '# A\n');
+    const acceptedId = await insertProposal({ source_id: 'tenant-a', page_slug: 'topics/a', claim: 'Accepted already', holder: 'garry' });
+    await acceptTakeProposal(engine, acceptedId, { brainDir });
+    await expect(rejectTakeProposal(engine, acceptedId)).rejects.toThrow('already accepted');
+  });
+});
+
+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);
+
+    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']);
+  });
+});