/** * Single submission entry point for the `embed-backfill` minion job * (v0.40 Federated Sync v2 — D19). * * Every caller routes through `submitEmbedBackfill`: * - Parallel sync --all completion (D18) * - Extended `sync` handler (D22, auto_embed_backfill) * - POST /webhooks/github * - `sources federate` / `unfederate` flip hook * - `gbrain sync trigger` * - autopilot per-source dispatch (when source is stale and degraded) * * Why centralize: D2 added a per-source DB lock at handler entry. That * protects against double-RUN but not double-SUBMIT — a webhook storm could * still queue 20 embed-backfill jobs, each one waking, acquiring the lock, * finding it busy, completing as `already_in_progress`. Net effect: zero * wasted Voyage spend (D6 budget cap), but high queue churn. Worse, even if * the lock held, a 30-second idempotency bucket only coalesces SUBMITS within * the same window. Multi-hour push activity racks up unbounded calls. * * D19 layered defenses (composed here): * 1. Per-source cooldown (default 10min). Refuses submission if the most * recent embed-backfill for this source finished or is still active * inside the window. * 2. Per-source 24h rolling spend cap (default $25). Computed from the * embed-backfill-tagged rows in the budget audit JSONL. Refuses * submission when spend has hit the cap. * * Both bounds are config-overridable: * - `embed.backfill_cooldown_min` (default 10) * - `embed.backfill_max_usd_per_source_24h` (default 25) * * Returns a tagged-union status so callers can render the right user signal * (`gbrain sources status`, webhook response body, sync completion banner). */ import type { BrainEngine } from './engine.ts'; import { MinionQueue } from './minions/queue.ts'; import { parseUsdLimit, resolveSpendPosture, type SpendPosture } from './spend-posture.ts'; export const COOLDOWN_CONFIG_KEY = 'embed.backfill_cooldown_min'; export const SPEND_CAP_CONFIG_KEY = 'embed.backfill_max_usd_per_source_24h'; const DEFAULT_COOLDOWN_MIN = 10; const DEFAULT_SPEND_CAP_USD = 25; export type SubmitEmbedBackfillStatus = | 'submitted' | 'cooldown' | 'spend_capped'; export interface SubmitEmbedBackfillResult { status: SubmitEmbedBackfillStatus; /** Set when status === 'submitted'. */ jobId?: number; /** Set when status === 'cooldown'. Seconds remaining until cooldown lifts. */ cooldownRemainingSeconds?: number; /** Set when status === 'spend_capped'. Dollars spent in the 24h window. */ spend24hUsd?: number; /** Set when status === 'spend_capped'. Active cap. */ spendCapUsd?: number; /** * Set true when `spend.posture=tokenmax` waved the job past the 24h spend * cap (#2139). The spend is still LEDGERED by the per-job BudgetTracker — * posture removes the ceiling, not the accounting. Cooldown is NOT bypassed * (it's queue-churn protection, not a spend gate). */ spendCapBypassed?: boolean; } export interface SubmitEmbedBackfillOpts { /** Logged into the job's data row for audit. */ reason: string; /** Override cooldown lookup (tests). */ cooldownMinOverride?: number; /** Override spend-cap lookup (tests). */ spendCapUsdOverride?: number; /** Override the 24h spend aggregator (tests). Returns spend in USD. */ spend24hFn?: (engine: BrainEngine, sourceId: string) => Promise; /** Override `Date.now` (tests). */ nowMs?: number; /** Job priority. Default 5 (lower than autopilot's 0; above default jobs). */ priority?: number; /** Override the resolved spend posture (tests). Default: read from config. */ postureOverride?: SpendPosture; } /** * Default 24h-spend aggregator. Reads completed embed-backfill rows in * `minion_jobs` and sums their token usage × known-price proxy. We * deliberately do NOT read the budget-tracker audit JSONL — the file * lives on the worker's local disk and may not be present on the * machine submitting the job (e.g. a remote MCP serve-http process). * * The minion_jobs.tokens_input/output columns ARE populated by the * subagent flow but NOT by the embed flow (gateway.embed doesn't go * through the chat-completion roll-up). For v0.40 we use a job-COUNT * proxy capped at the daily limit: 1 job ≈ default-cap-share. Accepts * imprecision in exchange for cross-process visibility. * * A precise rolling-spend tracker is filed as a v0.41 TODO. */ async function defaultSpend24hForSource( engine: BrainEngine, sourceId: string, ): Promise { // Conservative proxy: count jobs that completed (or are running) in the // 24h window. Each is treated as worth `DEFAULT_SPEND_CAP_USD / 25` ($1) // toward the cap — i.e. 25 jobs in 24h saturate the default cap. const rows = await engine.executeRaw<{ n: number }>( `SELECT COUNT(*)::int AS n FROM minion_jobs WHERE name = 'embed-backfill' AND data->>'sourceId' = $1 AND status IN ('active', 'completed') AND created_at > NOW() - INTERVAL '24 hours'`, [sourceId], ); const jobCount = rows[0]?.n ?? 0; return jobCount * 1; // $1 / job placeholder; configurable in a later wave. } /** * Look up an integer-valued config key with sane defaults. * Returns `def` on missing / NaN / non-positive. */ async function readIntConfig( engine: BrainEngine, key: string, def: number, ): Promise { const raw = await engine.getConfig(key); if (raw === null || raw === undefined) return def; const n = Number(raw); return Number.isFinite(n) && n > 0 ? n : def; } export async function submitEmbedBackfill( engine: BrainEngine, sourceId: string, opts: SubmitEmbedBackfillOpts, ): Promise { const now = opts.nowMs ?? Date.now(); const cooldownMin = opts.cooldownMinOverride ?? (await readIntConfig(engine, COOLDOWN_CONFIG_KEY, DEFAULT_COOLDOWN_MIN)); // v0.42.42.0 (#2139): spend cap honors `off`/`unlimited`/`none` → Infinity. // `0` still falls back to the default (off semantics ≠ 0). const spendCap = opts.spendCapUsdOverride ?? (raw => parseUsdLimit(raw, DEFAULT_SPEND_CAP_USD))(await engine.getConfig(SPEND_CAP_CONFIG_KEY)); const posture = opts.postureOverride ?? (await resolveSpendPosture(engine)); // ── Source-level cooldown ───────────────────────────────────── // Block re-submission if (a) an embed-backfill is currently active for this // source, OR (b) the most-recent embed-backfill finished within the // cooldown window. const lastJob = await engine.executeRaw<{ finished_at: Date | null; status: string; }>( `SELECT finished_at, status FROM minion_jobs WHERE name = 'embed-backfill' AND data->>'sourceId' = $1 ORDER BY id DESC LIMIT 1`, [sourceId], ); if (lastJob[0]) { if (lastJob[0].status === 'active' || lastJob[0].status === 'waiting') { // Active or waiting: no cooldown-remaining number (would be misleading). return { status: 'cooldown' }; } if (lastJob[0].finished_at) { const finishedMs = new Date(lastJob[0].finished_at).getTime(); const ageMs = now - finishedMs; const cooldownMs = cooldownMin * 60 * 1000; if (ageMs < cooldownMs) { return { status: 'cooldown', cooldownRemainingSeconds: Math.ceil((cooldownMs - ageMs) / 1000), }; } } } // ── 24h rolling spend cap ───────────────────────────────────── // v0.42.42.0 (#2139): `spend.posture=tokenmax` waves past the cap (the // operator declared cost isn't the constraint). The per-job BudgetTracker // still ledgers the spend — posture removes the ceiling, not the accounting. // An `off`/`unlimited` cap (Infinity) is likewise never tripped. const spend24hFn = opts.spend24hFn ?? defaultSpend24hForSource; const spend24h = await spend24hFn(engine, sourceId); const spendCapBypassed = posture === 'tokenmax' && spend24h >= spendCap; if (spend24h >= spendCap && !spendCapBypassed) { return { status: 'spend_capped', spend24hUsd: spend24h, spendCapUsd: spendCap, }; } // ── Submission ──────────────────────────────────────────────── const queue = new MinionQueue(engine); const job = await queue.add( 'embed-backfill', { sourceId, batchSize: 500, reason: opts.reason }, { priority: opts.priority ?? 5, idempotency_key: `embed-backfill:${sourceId}:${bucketize(now, 5 * 60_000)}`, maxWaiting: 1, }, ); return spendCapBypassed ? { status: 'submitted', jobId: job.id, spendCapBypassed: true, spend24hUsd: spend24h } : { status: 'submitted', jobId: job.id }; } /** Round timestamp down to the nearest `bucketMs` boundary. */ function bucketize(ms: number, bucketMs: number): number { return Math.floor(ms / bucketMs); }