openvibers/openvibe.trade Commits

AI briefs: the worker's report briefs wait in the batch class (#234)

OP OpenVibers committed 5 days ago · d49a909555 · 8 files, +472 −47

* AI briefs: the worker's report briefs wait in the batch class

A report brief now asks OpenVibe.AI for options.latency 'batch' with a deadline, keeps the pending run id and the exact inputs in trade_ai_brief_jobs (migration 0009), and collects the answer on a later tick. The run id is the idempotency key, so a restart, a retried pass or a second poller writes one brief. A batch refused at the door or ended without an answer falls back to the background class once; a reader's ask stays synchronous. Tests cover the pending run, a restart, an expired batch, a refused start and no double write.

* A market brief waits at most 3 h in the batch class by default (staler than that, it falls back to background)

Changes

ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex 69efc3a..53d95d9 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -2,6 +2,7 @@
  
  ## Unreleased
  
++- Worker-written AI briefs ask OpenVibe.AI for the `batch` latency class: the pending run id and the exact inputs live in `trade_ai_brief_jobs` (migration `0009`), the worker collects the answer on a later tick, a run id is written once, and a refused or expired batch falls back to `background` once; a reader's ask stays synchronous. `test/ai-briefs.test.js`.
  - The instrument page carries the market-research outcomes the data backs: every observation shows the source URL and the source's own reference (`source_ref`), when the value was true, when it was fetched, and a stale verdict with the value's `max_age_sec` and a plain reason; a "Free, always" line; one-tap watchlist (new `POST /watchlists/watch`) and new-filing alert actions, with the configured limits stated instead of a no-cap claim; price rows are labelled source-stated, not live, and no delay is invented. Observation rows are compact on phones, with the long provenance in a closed "Source details" fold; the source table uses the OpenVibe Kit and stacks on phones. `test/market-outcomes.test.js`.
  - The front page is curated (`server/render/home.js`): a company search and the numbers beside the hero (companies, filings in 24 hours, filings recorded, companies with figures, AI briefs, currencies), then AI briefs as cards, euro reference rates as cards with the change from the same source, new 10-K/10-Q reports, the week's filings by form in plain words, the newest filings, companies with something new, what is here and six questions (FAQPage JSON-LD). The instruments list moved to `/instruments` (301 from `/?page=N`); `/filings[?form=]` is new; `/sitemaps/pages.xml` lists `/`, `/instruments`, `/filings` and `/sources` (the sitemap listed nothing before: instrument pages stay noindex until a person reviews their context). The navbar reads Companies, Filings, Sources, Watchlists.
  - Account export and deletion (ADR-033) through `server/domain/account-data.js` over `openvibe-sdk/account-data` (SDK v0.37.0, was v0.35.0):
ddiff --git a/README.md b/README.md
iindex 84e4356..92b3b75 100644
---- a/README.md
++++ b/README.md
@@@ -139,6 +139,10 @@ Trade shows only what a source stated:
    - written from a company's numbers and filings list only, when a report brings new numbers (the worker,
      `TRADE_BRIEFS_PER_TICK`) or when a reader asks (`POST /i/:symbol/brief`: same origin, 5 per address per 10
      minutes), within `TRADE_BRIEFS_PER_HOUR` for the whole site;
++  - the worker's report brief asks OpenVibe.AI for the `batch` latency class: the run id and the exact inputs are
++    kept in `trade_ai_brief_jobs` (migration `0009`), the answer is collected on a later tick, and a refused or
++    expired batch falls back to the `background` class once (`TRADE_BRIEFS_BATCH=off` keeps it synchronous;
++    `TRADE_BRIEFS_BATCH_DEADLINE_HOURS` defaults to 3: a market brief older than that is stale, so it falls back to background). A reader's ask stays synchronous;
    - every kept point cites Trade's own observation or document rows, and an uncited point is dropped and listed;
    - text that reads as advice is refused, and so is a placeholder run;
    - shown labelled "AI-written, not reviewed by a person". They do not make a page indexable: that still needs a
@@@ -300,8 +304,8 @@ services need to call Trade, is under [Grants the Network must hold](#grants-the
  - **OpenVibe.Network notifications:** Network consumes `trade.alert.triggered` (its `network`
    subscription) and puts a "Trade alert: <symbol>" notification in the rule owner's inbox, linking the
    instrument page; deliveries are also listed at `/watchlists`.
--- **OpenVibe.AI:** `ai.run.create` (namespace `trade`) for the AI briefs (`OV_AI_INTERNAL_URL`; off without it), and
--  optionally the drafts below.
++- **OpenVibe.AI:** `ai.run.create` (namespace `trade`) for the AI briefs, including the `batch` class for the
++  worker's report briefs (`OV_AI_INTERNAL_URL`; off without it), and optionally the drafts below.
  - **SEC EDGAR** (`data.sec.gov`, read directly, no key): XBRL company facts, public domain. Fair access: the
    User-Agent `TRADE_SEC_USER_AGENT` names us and a contact; requests are spaced `TRADE_SEC_MIN_INTERVAL_MS` (1 s)
    apart, at most `TRADE_SEC_PER_TICK` companies a tick. `TRADE_SEC_FACTS=off` turns it off.
ddiff --git a/migrations/0009_ai_brief_batch.sql b/migrations/0009_ai_brief_batch.sql
nnew file mode 100644
iindex 0000000..9118b91
---- /dev/null
++++ b/migrations/0009_ai_brief_batch.sql
@@@ -0,0 +1,19 @@
++-- phase: expand
++-- The worker's AI briefs in the batch class (server/domain/briefs.js, server/clients/ai.js): a report brief is
++-- queued at OpenVibe.AI instead of waited for, and the worker collects the answer on a later tick. The work item
++-- keeps the pending run id and the exact inputs it was made from, so a restart loses nothing and the citations
++-- written later still point at the rows the model saw.
++CREATE TABLE IF NOT EXISTS trade_ai_brief_jobs (
++    instrument_id text COLLATE "C" PRIMARY KEY REFERENCES trade_instruments(id),
++    run_id        text COLLATE "C" NOT NULL,                    -- the OpenVibe.AI run id the worker polls
++    inputs        jsonb NOT NULL,                               -- { input: <the exact workflow input>, refs: <citation refs> }
++    inputs_hash   text NOT NULL,                                -- the same hash as trade_ai_briefs.inputs_hash
++    status        text NOT NULL CHECK (status IN ('pending', 'fallback', 'done', 'failed')),
++    attempts      bigint NOT NULL DEFAULT 1,                    -- 1 = batch; 2 = its one background fallback
++    deadline_at   bigint NOT NULL,                              -- when the batch stops being worth waiting for
++    created_at    bigint NOT NULL,
++    updated_at    bigint NOT NULL
++);
++CREATE INDEX IF NOT EXISTS trade_ai_brief_jobs_open ON trade_ai_brief_jobs (status, created_at);
++-- One brief per run: a collected answer written twice is refused by the database, not only by the worker.
++CREATE UNIQUE INDEX IF NOT EXISTS trade_ai_briefs_run ON trade_ai_briefs (run_id) WHERE run_id IS NOT NULL;
ddiff --git a/server/clients/ai.js b/server/clients/ai.js
iindex eb64eb0..a880522 100644
---- a/server/clients/ai.js
++++ b/server/clients/ai.js
@@@ -6,12 +6,19 @@
   *   POST {OV_AI_INTERNAL_URL}/api/v1/runs?wait=…  { workflow, input: { instrument, stale_after_hours, sources } }
   *   (this service's token for audience openvibe.ai, capability ai.run.create, namespace trade)
   *
++ * The worker's report briefs ask for the `batch` latency class (options.latency 'batch', options.deadline_hours):
++ * the run is created now and answered through OpenVibe.AI's provider batch queue, so Trade keeps the run id and
++ * collects the answer on a later tick. A reader's ask stays synchronous: the person is waiting on the page.
++ * A batch refused at the door, or a run that ends without an answer, falls back to the background class once.
++ *
   * Only Trade's own rows go to the model: SEC's filed numbers (observations) and the filings list (documents),
   * each with its dates and source. A failed or unfinished run produces nothing.
   */
  const { serviceAuth } = require('openvibe-contracts');
  
  const WORKFLOW = 'trade.summarize_market_context';
++const DONE = new Set(['succeeded', 'cached']);
++const FAILED = new Set(['failed', 'cancelled', 'expired']);
  
  class AiError extends Error {
      constructor(code, message, status = 502) { super(message); this.code = code; this.status = status; }
@@@ -24,30 +31,101 @@ function createAi({ config, fetchImpl = globalThis.fetch }) {
          audience: 'openvibe.ai', scope: 'ai.run.create', fetchImpl,
      }) : null;
  
--    /** input: { instrument: { symbol, name, exchange }, stale_after_hours, sources: [AI source] } → { runId, model, stub, output } */
--    async function run(input) {
++    /**
++     * One JSON request to OpenVibe.AI. → { status, ok, body }. A stale service token is refetched once.
++     * The body is kept even on an error response: the AI service's problem+json names the refusal.
++     */
++    async function request(method, path, body, timeoutMs, retried = false) {
          if (!enabled) throw new AiError('ai.not_configured', 'OpenVibe.AI is not configured (OV_AI_INTERNAL_URL)', 503);
++        let headers;
++        try { headers = await tokens.authHeaders(); } catch (err) { throw new AiError('ai.auth', `OpenVibe.AI token: ${err.message}`, 503); }
          let res;
          try {
--            res = await fetchImpl(`${config.ai.internalUrl}/api/v1/runs?wait=${config.ai.waitMs}`, {
--                method: 'POST',
--                headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...(await tokens.authHeaders()) },
--                body: JSON.stringify({ workflow: WORKFLOW, input }),
--                signal: AbortSignal.timeout(config.ai.waitMs + 5000),
++            res = await fetchImpl(`${config.ai.internalUrl}${path}`, {
++                method,
++                headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...headers },
++                body: body ? JSON.stringify(body) : undefined,
++                signal: AbortSignal.timeout(timeoutMs),
              });
          } catch (err) {
              throw new AiError('ai.unavailable', `OpenVibe.AI is unreachable: ${err.message}`);
          }
--        const body = await res.json().catch(() => null);
--        if (!res.ok && res.status !== 202) throw new AiError(`ai.http_${res.status}`, `OpenVibe.AI answered ${res.status}${body && body.code ? ` (${body.code})` : ''}`);
--        const r = body && body.run;
--        if (!r || r.status !== 'succeeded' || !r.output || typeof r.output !== 'object') {
++        const json = await res.json().catch(() => null);
++        if (res.status === 401 && !retried) { tokens.invalidate(); return await request(method, path, body, timeoutMs, true); }
++        return { status: res.status, ok: res.ok, body: json };
++    }
++
++    /** A run as Trade reads it: the id, its status, its answer, and the model or error it carries. */
++    function runView(r) {
++        if (!r || !r.id) return null;
++        return {
++            runId: String(r.id), status: String(r.status || 'queued'),
++            model: r.provenance && r.provenance.model ? String(r.provenance.model) : null,
++            stub: Boolean(r.synthetic), output: r.output && typeof r.output === 'object' ? r.output : null,
++            error: r.error || null,
++        };
++    }
++
++    /**
++     * Run a workflow and wait for it. `latency: 'background'` asks for the class that may wait minutes (the
++     * workflow's own class is background); it is how a refused or expired batch falls back once. An
++     * idempotencyKey makes a retried fallback return the same run instead of a second one.
++     */
++    async function run(input, { latency = null, idempotencyKey = null } = {}) {
++        const body = { workflow: WORKFLOW, input };
++        if (latency) body.options = { latency };
++        if (idempotencyKey) body.idempotency_key = idempotencyKey;
++        const send = () => request('POST', `/api/v1/runs?wait=${config.ai.waitMs}`, body, config.ai.waitMs + 5000);
++        let out;
++        try { out = await send(); } catch (err) {
++            // A retried create with the same idempotency key is the same run: a timeout that lost the answer
++            // must not become a second paid run.
++            if (err.code !== 'ai.unavailable' || !idempotencyKey) throw err;
++            out = await send();
++        }
++        if (!out.ok && out.status !== 202) {
++            throw new AiError(`ai.http_${out.status}`, `OpenVibe.AI answered ${out.status}${out.body && out.body.code ? ` (${out.body.code})` : ''}`);
++        }
++        const r = runView(out.body && out.body.run);
++        if (!r || !DONE.has(r.status) || !r.output) {
              throw new AiError('ai.no_result', `The ${WORKFLOW} run did not finish with a result (${r ? r.status : 'no run'})`);
          }
--        return { runId: String(r.id), model: r.provenance && r.provenance.model ? String(r.provenance.model) : null, stub: Boolean(r.synthetic), output: r.output };
++        return r;
++    }
++
++    /**
++     * Queue a workflow run in the batch class and return its id at once; the answer is collected later.
++     * → { runId, status, model, stub, output, error, done } (done: it answered inside the call, e.g. a cache hit).
++     */
++    async function start(input, { deadlineHours = 24, idempotencyKey = null } = {}) {
++        const hours = Math.max(1, Math.min(72, Math.trunc(Number(deadlineHours) || 24)));
++        const body = { workflow: WORKFLOW, input, options: { latency: 'batch', deadline_hours: hours } };
++        if (idempotencyKey) body.idempotency_key = idempotencyKey;
++        const send = () => request('POST', '/api/v1/runs?wait=0', body, Math.max(20000, config.ai.waitMs));
++        let out;
++        try { out = await send(); } catch (err) {
++            // The run may have been created before the connection broke. The same key replays it, so retry once
++            // before the caller falls back and pays for a second run.
++            if (err.code !== 'ai.unavailable') throw err;
++            out = await send();
++        }
++        if (!out.ok && out.status !== 202) {
++            throw new AiError(`ai.http_${out.status}`, `OpenVibe.AI answered ${out.status}${out.body && out.body.code ? ` (${out.body.code})` : ''}`);
++        }
++        const r = runView(out.body && out.body.run);
++        if (!r) throw new AiError('ai.no_run', 'OpenVibe.AI did not answer a run for the batch');
++        return { ...r, done: DONE.has(r.status) };
++    }
++
++    /** One run by id; null when OpenVibe.AI has no such run (404). */
++    async function get(runId) {
++        const out = await request('GET', `/api/v1/runs/${encodeURIComponent(runId)}`, null, 15000);
++        if (out.status === 404) return null;
++        if (!out.ok) throw new AiError(`ai.http_${out.status}`, `OpenVibe.AI answered ${out.status}`);
++        return runView(out.body && out.body.run);
      }
  
--    return { enabled, WORKFLOW, run };
++    return { enabled, WORKFLOW, run, start, get, DONE, FAILED };
  }
  
  module.exports = { createAi, AiError, WORKFLOW };
ddiff --git a/server/config.js b/server/config.js
iindex eeb7ca9..0355d33 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -121,6 +121,10 @@ function load(env = process.env) {
              intervalMs: Math.max(int(env.TRADE_BRIEFS_INTERVAL_MS, 10 * 60_000), 10_000),
              perTick: Math.max(int(env.TRADE_BRIEFS_PER_TICK, 6), 0),
              perHour: Math.max(int(env.TRADE_BRIEFS_PER_HOUR, 40), 0),
++            // The worker's report briefs ask OpenVibe.AI for the batch class. TRADE_BRIEFS_BATCH=off keeps them
++            // synchronous (an incident switch); the deadline is 1 to 72 hours and OpenVibe.AI's own default is 24.
++            batch: String(env.TRADE_BRIEFS_BATCH || 'on').trim().toLowerCase() !== 'off',
++            batchDeadlineHours: Math.max(1, Math.min(72, int(env.TRADE_BRIEFS_BATCH_DEADLINE_HOURS, 3))),
          },
          // The market-wide brief (domain/market-briefs.js): one UTC day at most, from the day's filings, euro rates and
          // companies with new filings or figures. Off when TRADE_MARKET_BRIEFS=off, and always off without OpenVibe.AI.
ddiff --git a/server/domain/briefs.js b/server/domain/briefs.js
iindex 0a4d4f6..e66b93c 100644
---- a/server/domain/briefs.js
++++ b/server/domain/briefs.js
@@@ -15,6 +15,12 @@
   *
   * Briefs are made after a company's report brings new numbers (the worker, trigger 'report') or when a reader asks
   * (POST /i/:symbol/brief, trigger 'reader'), within TRADE_BRIEFS_PER_HOUR for the whole site.
++ *
++ * A report brief asks OpenVibe.AI for the batch class: the run is queued now and collected on a later worker tick
++ * (trade_ai_brief_jobs keeps the pending run id and the exact inputs). The worker survives a restart because the
++ * queue is a table, not a promise. A batch refused at the door, or a run that ends without an answer, falls back
++ * to the background class once; a second failure sends the company back through the worker on a later tick. The
++ * reader's ask stays synchronous: the person is waiting on the page.
   */
  const crypto = require('crypto');
  const { ADVICE_RE } = require('./context');
@@@ -25,15 +31,20 @@ const { ApiError } = require('../http/errors');
  const MAX_SOURCES = 50;
  const MAX_DOCS = 12;
  const PER_SERIES = 7;
++const FALLBACK_GRACE_MS = 30 * 60 * 1000;   // after the batch deadline, give OpenVibe.AI's reroute half an hour
  
  function createBriefs({ store, config, ctx, ai, log = console }) {
      const { db } = store;
      const q = {
          latest: db.prepare('SELECT * FROM trade_ai_briefs WHERE instrument_id = ? ORDER BY created_at DESC, id DESC LIMIT 1'),
++        byRun: db.prepare('SELECT * FROM trade_ai_briefs WHERE run_id = ? LIMIT 1'),
          recent: db.prepare(`SELECT b.*, i.symbol, i.name FROM trade_ai_briefs b JOIN trade_instruments i ON i.id = b.instrument_id
                              WHERE i.status = 'active' AND b.id = (SELECT id FROM trade_ai_briefs x WHERE x.instrument_id = b.instrument_id ORDER BY created_at DESC, id DESC LIMIT 1)
                              ORDER BY b.created_at DESC LIMIT ?`),
--        lastHour: db.prepare('SELECT COUNT(*) AS n FROM trade_ai_briefs WHERE created_at > ?'),
++        // The site's hourly budget counts a brief when it is written and a report brief when its batch is started
++        // (an open job has not been written yet, but it has been asked for and will be paid for).
++        lastHour: db.prepare(`SELECT (SELECT COUNT(*) FROM trade_ai_briefs WHERE created_at > ?)
++                            + (SELECT COUNT(*) FROM trade_ai_brief_jobs WHERE status IN ('pending', 'fallback') AND created_at > ?) AS n`),
          stats: db.prepare('SELECT COUNT(*) AS n, COUNT(DISTINCT instrument_id) AS companies FROM trade_ai_briefs'),
          insert: db.prepare(`INSERT INTO trade_ai_briefs (id, instrument_id, summary, points, cites, gaps, as_of, inputs_hash, run_id, model, trigger, created_at)
                              VALUES (@id, @instrument_id, @summary, @points, @cites, @gaps, @as_of, @inputs_hash, @run_id, @model, @trigger, @created_at)`),
@@@ -41,11 +52,25 @@ function createBriefs({ store, config, ctx, ai, log = console }) {
                              ORDER BY metric, observed_at DESC, recorded_at DESC`),
          docs: db.prepare(`SELECT * FROM trade_source_documents WHERE instrument_id = ? AND removed_at IS NULL
                            ORDER BY COALESCE(published_at, retrieved_at) DESC LIMIT ?`),
--        // Companies whose numbers were read after their last brief (or that have none): a report brought new numbers.
++        // Companies whose numbers were read after their last brief (or that have none): a report brought new
++        // numbers. A company whose batch is still open is not due again: its work is already in flight.
          due: db.prepare(`SELECT i.* FROM trade_instruments i JOIN trade_facts_state s ON s.instrument_id = i.id
                            WHERE i.status = 'active' AND s.ok = 1 AND s.observations > 0
                              AND NOT EXISTS (SELECT 1 FROM trade_ai_briefs b WHERE b.instrument_id = i.id AND b.created_at >= s.fetched_at)
++                            AND NOT EXISTS (SELECT 1 FROM trade_ai_brief_jobs j WHERE j.instrument_id = i.id AND j.status IN ('pending', 'fallback'))
                            ORDER BY s.fetched_at ASC LIMIT ?`),
++        instrument: db.prepare('SELECT * FROM trade_instruments WHERE id = ?'),
++        job: db.prepare('SELECT * FROM trade_ai_brief_jobs WHERE instrument_id = ?'),
++        openJobs: db.prepare("SELECT * FROM trade_ai_brief_jobs WHERE status IN ('pending', 'fallback') ORDER BY created_at ASC LIMIT ?"),
++        upsertJob: db.prepare(`INSERT INTO trade_ai_brief_jobs (instrument_id, run_id, inputs, inputs_hash, status, attempts, deadline_at, created_at, updated_at)
++                               VALUES (?, ?, ?::jsonb, ?, 'pending', ?, ?, ?, ?)
++                               ON CONFLICT (instrument_id) DO UPDATE SET run_id = excluded.run_id, inputs = excluded.inputs, inputs_hash = excluded.inputs_hash,
++                                 status = 'pending', attempts = excluded.attempts, deadline_at = excluded.deadline_at, created_at = excluded.created_at, updated_at = excluded.updated_at
++                               WHERE trade_ai_brief_jobs.status IN ('done', 'failed') RETURNING *`),
++        claimFallback: db.prepare(`UPDATE trade_ai_brief_jobs SET status = 'fallback', attempts = attempts + 1, updated_at = ?
++                                   WHERE instrument_id = ? AND run_id = ? AND status = 'pending' RETURNING *`),
++        finishJob: db.prepare(`UPDATE trade_ai_brief_jobs SET status = 'done', updated_at = ? WHERE instrument_id = ? AND run_id = ? AND status IN ('pending', 'fallback')`),
++        failJob: db.prepare(`UPDATE trade_ai_brief_jobs SET status = 'failed', updated_at = ? WHERE instrument_id = ? AND run_id = ? AND status IN ('pending', 'fallback')`),
      };
  
      const fmtDay = (t) => (t == null ? 'no date' : new Date(Number(t)).toISOString().slice(0, 10));
@@@ -94,34 +119,31 @@ function createBriefs({ store, config, ctx, ai, log = console }) {
              refs.push([{ kind: 'document', id: d.id, at: d.published_at || d.retrieved_at }]);
          }
          const hash = crypto.createHash('sha256').update(JSON.stringify(sources.map((s) => [s.source_id, s.content || s.snippet]))).digest('hex');
--        return { sources, refs, hash };
++        const input = {
++            instrument: { symbol: instrument.symbol, name: String(instrument.name).slice(0, 200), ...(instrument.exchange ? { exchange: instrument.exchange } : {}) },
++            stale_after_hours: 24 * 120,
++            sources,
++        };
++        return { sources, refs, hash, input };
      }
  
--    const lastHour = async () => (await q.lastHour.get(store.now() - 3600_000)).n;
++    const lastHour = async () => (await q.lastHour.get(store.now() - 3600_000, store.now() - 3600_000)).n;
++
++    const parse = (v, d) => { if (v == null) return d; if (typeof v !== 'string') return v; try { return JSON.parse(v); } catch { return d; } };
++    const isUnique = (err) => String((err && err.code) || '') === '23505' || /unique|duplicate key/i.test(String((err && err.message) || ''));
  
      /**
--     * Write a brief for the instrument now. → { brief, created } (created false: the inputs are unchanged since the
--     * last brief). Throws ApiError: 409 nothing to write from, 429 the hourly budget is spent, 422 advice refused,
--     * 502/503 OpenVibe.AI.
++     * Write one run's output as this instrument's next brief. The run id is the idempotency key: an answer
++     * collected twice (a retried tick, a restart, another poller) is read once and never written twice.
++     * Throws ApiError when the answer is a stub, empty, or advice. → { brief, created }
       */
--    async function make(instrument, { trigger = 'report' } = {}) {
--        const { sources, refs, hash } = await inputsFor(instrument);
--        if (!sources.length) throw new ApiError(409, 'brief.no_inputs', `Trade has no SEC numbers or filings for ${instrument.symbol} yet, so there is nothing to write a brief from`);
--        const prior = await q.latest.get(instrument.id);
--        if (prior && prior.inputs_hash === hash) return { brief: view(prior), created: false };
--        if (await lastHour() >= config.briefs.perHour) throw new ApiError(429, 'brief.budget', 'The AI briefs budget for this hour is used up; try again later');
--
--        let r;
--        try {
--            r = await ai.run({
--                instrument: { symbol: instrument.symbol, name: String(instrument.name).slice(0, 200), ...(instrument.exchange ? { exchange: instrument.exchange } : {}) },
--                stale_after_hours: 24 * 120,
--                sources,
--            });
--        } catch (err) {
--            throw new ApiError(err.status || 502, err.code || 'ai.error', `No AI brief: ${err.message}`);
--        }
++    async function keep(instrument, r, { refs, hash }, trigger) {
++        if (!r) throw new ApiError(502, 'brief.no_run', 'OpenVibe.AI returned no run for the brief');
          if (r.stub) throw new ApiError(503, 'brief.stub', 'OpenVibe.AI answered with a placeholder (no model is configured); no brief was made');
++        if (r.runId) {
++            const previous = await q.byRun.get(r.runId);
++            if (previous) return { brief: view(previous), created: false, duplicate: true };
++        }
          const out = r.output || {};
          const gaps = Array.isArray(out.gaps) ? out.gaps.map((g) => String(g).slice(0, 300)).slice(0, 20) : [];
          const citeRefs = (list) => {
@@@ -165,11 +187,162 @@ function createBriefs({ store, config, ctx, ai, log = console }) {
              id: newId('tbr', store.now()), instrument_id: instrument.id, summary, points: JSON.stringify(points), cites: JSON.stringify(cites),
              gaps: JSON.stringify(gaps.slice(0, 20)), as_of: asOf, inputs_hash: hash, run_id: r.runId, model: r.model, trigger, created_at: store.now(),
          };
--        await q.insert.run(row);
++        try {
++            await q.insert.run(row);
++        } catch (err) {
++            if (!isUnique(err)) throw err;
++            const previous = r.runId ? await q.byRun.get(r.runId) : null;
++            if (!previous) throw err;
++            return { brief: view(previous), created: false, duplicate: true };
++        }
          return { brief: view(await q.latest.get(instrument.id)), created: true };
      }
  
--    const parse = (v, d) => { if (v == null) return d; if (typeof v !== 'string') return v; try { return JSON.parse(v); } catch { return d; } };
++    /** The run input and the Trade rows its citations will point at, for one instrument. */
++    async function prepare(instrument) {
++        const { sources, refs, hash, input } = await inputsFor(instrument);
++        if (!sources.length) throw new ApiError(409, 'brief.no_inputs', `Trade has no SEC numbers or filings for ${instrument.symbol} yet, so there is nothing to write a brief from`);
++        return { sources, refs, hash, input };
++    }
++
++    /** A batch run refused at the door: answer it once in the background, with the same input. */
++    async function startBackground(instrument, prepared, attempt) {
++        const { input, refs, hash } = prepared;
++        try {
++            const r = await ai.run(input, { latency: 'background', idempotencyKey: `trade-brief:${instrument.id}:${hash}:bg:${attempt}` });
++            return await keep(instrument, r, { refs, hash }, 'report');
++        } catch (err) {
++            throw new ApiError(err.status || 502, err.code || 'ai.error', `No AI brief: ${err.message}`);
++        }
++    }
++
++    /**
++     * Ask for one report brief in the batch class. → { brief, created, pending, runId }.
++     * The pending run id and the exact inputs are stored before this returns, so the answer survives a restart.
++     */
++    async function startBatch(instrument, prepared) {
++        const { input, refs, hash } = prepared;
++        const open = await q.job.get(instrument.id);
++        if (open && ['pending', 'fallback'].includes(open.status)) {
++            return { brief: null, created: false, pending: true, runId: open.run_id };
++        }
++        const attempt = open ? Math.max(1, Number(open.attempts) || 1) + 1 : 1;
++        let started;
++        try {
++            started = await ai.start(input, {
++                deadlineHours: config.briefs.batchDeadlineHours,
++                idempotencyKey: `trade-brief:${instrument.id}:${hash}:batch:${attempt}`,
++            });
++        } catch (err) {
++            // The batch class was refused (a batch deadline/limit, no native route, a quota, an unreachable
++            // service): the same run goes to the background class once, never in a loop.
++            log.warn(`[Trade] AI batch brief for ${instrument.symbol} refused (${err.code || err.status || 'error'}); falling back to background`);
++            return await startBackground(instrument, prepared, attempt);
++        }
++        if (started.done) return await keep(instrument, started, { refs, hash }, 'report');
++        const now = store.now();
++        const stored = await q.upsertJob.get(instrument.id, started.runId, JSON.stringify({ input, refs }), hash, attempt,
++            now + Math.max(1, Math.min(72, Number(config.briefs.batchDeadlineHours) || 24)) * 3600_000, now, now);
++        if (!stored) {
++            // Another start won the row between the read and the insert: it is the pending work item.
++            const current = await q.job.get(instrument.id);
++            return { brief: null, created: false, pending: true, runId: current ? current.run_id : started.runId };
++        }
++        return { brief: null, created: true, pending: true, runId: started.runId };
++    }
++
++    /**
++     * Write a brief for the instrument now. → { brief, created } (created false: the inputs are unchanged since the
++     * last brief). A report-triggered brief returns { pending: true } when its batch is queued. Throws ApiError: 409
++     * nothing to write from, 429 the hourly budget is spent, 422 advice refused, 502/503 OpenVibe.AI.
++     */
++    async function make(instrument, { trigger = 'report' } = {}) {
++        const prepared = await prepare(instrument);
++        const prior = await q.latest.get(instrument.id);
++        if (prior && prior.inputs_hash === prepared.hash) return { brief: view(prior), created: false };
++        if (await lastHour() >= config.briefs.perHour) throw new ApiError(429, 'brief.budget', 'The AI briefs budget for this hour is used up; try again later');
++        if (trigger === 'report' && config.briefs.batch !== false) return await startBatch(instrument, prepared);
++
++        let r;
++        try {
++            r = await ai.run(prepared.input);
++        } catch (err) {
++            throw new ApiError(err.status || 502, err.code || 'ai.error', `No AI brief: ${err.message}`);
++        }
++        return await keep(instrument, r, prepared, trigger);
++    }
++
++    /** One background attempt for a batch that ended without an answer, or passed its deadline. */
++    async function fallBack(job, instrument, prepared) {
++        const now = store.now();
++        const claimed = await q.claimFallback.get(now, instrument.id, job.run_id);
++        if (!claimed) return { brief: null, created: false, pending: true };   // another pass took it
++        const attempt = Number(claimed.attempts) || 2;
++        try {
++            const r = await ai.run(prepared.input, { latency: 'background', idempotencyKey: `trade-brief:${instrument.id}:${prepared.hash}:bg:${attempt}` });
++            const kept = await keep(instrument, r, prepared, 'report');
++            await q.finishJob.run(now, instrument.id, job.run_id);
++            return kept;
++        } catch (err) {
++            await q.failJob.run(now, instrument.id, job.run_id);
++            log.warn(`[Trade] AI brief for ${instrument.symbol} fell back and did not answer: ${err.message}`);
++            throw err;
++        }
++    }
++
++    /**
++     * The later pass: read every pending batch's answer (or fall back once when it is refused/expired), and write
++     * each brief at most once. → { briefs, pending, failed } (briefs written, still waiting, errors this pass).
++     */
++    async function collect(limit = 6) {
++        const out = { briefs: 0, pending: 0, failed: 0 };
++        const jobs = await q.openJobs.all(Math.max(1, Number(limit) || 1));
++        for (const job of jobs) {
++            const instrument = await q.instrument.get(job.instrument_id);
++            const saved = parse(job.inputs, null);
++            if (!instrument || !saved || !saved.input) { await q.failJob.run(store.now(), job.instrument_id, job.run_id); out.failed++; continue; }
++            const prepared = { input: saved.input, refs: Array.isArray(saved.refs) ? saved.refs : [], hash: job.inputs_hash };
++            // A fallback that was in progress when the process stopped: the same idempotency key reads the same run.
++            if (job.status === 'fallback') {
++                try {
++                    const r = await ai.run(prepared.input, { latency: 'background', idempotencyKey: `trade-brief:${instrument.id}:${prepared.hash}:bg:${job.attempts}` });
++                    const kept = await keep(instrument, r, prepared, 'report');
++                    await q.finishJob.run(store.now(), job.instrument_id, job.run_id);
++                    if (kept.created) out.briefs++;
++                } catch (err) {
++                    await q.failJob.run(store.now(), job.instrument_id, job.run_id);
++                    out.failed++;
++                    log.warn(`[Trade] AI brief for ${instrument.symbol} did not finish its fallback: ${err.message}`);
++                }
++                continue;
++            }
++            let run;
++            try { run = await ai.get(job.run_id); } catch (err) { out.failed++; log.warn(`[Trade] AI brief run ${job.run_id} not read: ${err.message}`); continue; }
++            if (run && ai.DONE.has(run.status)) {
++                try {
++                    const kept = await keep(instrument, run, prepared, 'report');
++                    await q.finishJob.run(store.now(), job.instrument_id, job.run_id);
++                    if (kept.created) out.briefs++;
++                } catch (err) {
++                    await q.failJob.run(store.now(), job.instrument_id, job.run_id);
++                    out.failed++;
++                    log.warn(`[Trade] AI brief for ${instrument.symbol} was refused: ${err.message}`);
++                }
++                continue;
++            }
++            const dead = !run || ai.FAILED.has(run.status) || store.now() >= Number(job.deadline_at) + FALLBACK_GRACE_MS;
++            if (dead) {
++                try {
++                    const kept = await fallBack(job, instrument, prepared);
++                    if (kept.created) out.briefs++;
++                    if (kept.pending) out.pending++;
++                } catch { out.failed++; }
++                continue;
++            }
++            out.pending++;
++        }
++        return out;
++    }
  
      function view(row) {
          if (!row) return null;
@@@ -185,6 +358,7 @@ function createBriefs({ store, config, ctx, ai, log = console }) {
      return {
          enabled: ai.enabled,
          make,
++        collect,
          inputsFor,
          latest: async (instrument) => view(await q.latest.get(instrument.id)),
          recent: async (limit = 8) => (await q.recent.all(limit)).map(view),
ddiff --git a/server/worker.js b/server/worker.js
iindex 0c4f3b3..b5666f4 100644
---- a/server/worker.js
++++ b/server/worker.js
@@@ -41,11 +41,18 @@ function createWorker({ config, sync, freshness, indexing, outbox, facts = null,
                  }
              }
              if (briefs && briefs.enabled && config.briefs.perTick > 0) {
++                // First the batches that already answered (or fell due): a restart loses nothing, because the
++                // pending run ids are rows in trade_ai_brief_jobs, not promises in this process.
++                const collected = await briefs.collect(config.briefs.perTick);
++                out.briefs += collected.briefs;
++                out.failed += collected.failed;
++                // Then the companies whose report brought new numbers, within the hourly budget. A batch that
++                // was just started counts as this hour's work, so room is read after the collection.
                  const room = Math.max(0, Math.min(config.briefs.perTick, config.briefs.perHour - await briefs.lastHour()));
                  for (const i of room ? await briefs.due(room) : []) {
                      try {
                          const r = await briefs.make(i, { trigger: 'report' });
--                        if (r.created) out.briefs++;
++                        if (r.created && !r.pending) out.briefs++;
                      } catch (err) {
                          out.failed++;
                          log.warn(`[Trade] AI brief for ${i.symbol} not made:`, err.message);
ddiff --git a/test/ai-briefs.test.js b/test/ai-briefs.test.js
iindex 4615026..6965055 100644
---- a/test/ai-briefs.test.js
++++ b/test/ai-briefs.test.js
@@@ -12,6 +12,7 @@
  const assert = require('assert');
  const { boot, check, done } = require('./helpers/boot');
  const { extract } = require('../server/domain/facts');
++const { createAi } = require('../server/clients/ai');
  
  const q = (start, end, val, accn, frame, form = '10-Q') => ({ start, end, val, accn, fy: 2026, fp: 'Q3', form, filed: '2026-07-31', ...(frame ? { frame } : {}) });
  const i = (end, val, accn, frame) => ({ end, val, accn, fy: 2026, fp: 'Q3', form: '10-Q', filed: '2026-07-31', ...(frame ? { frame } : {}) });
@@@ -41,14 +42,37 @@ const quiet = { log() {}, warn() {}, error() {} };
      const secCalls = [];
      const sec = { enabled: true, async companyFacts(cik) { secCalls.push(cik); return { doc: FACTS, retrievedAt: t.clock.now() }; } };
      const aiCalls = [];
++    const aiRuns = new Map();
++    let aiSeq = 0;
      let aiOutput = null;
      let aiStub = false;
++    let aiStartError = null;
++    let backgroundCalls = 0;
      const ai = {
          enabled: true, WORKFLOW: 'trade.summarize_market_context',
--        async run(input) { aiCalls.push(input); return { runId: `run_${aiCalls.length}`, model: 'test-model', stub: aiStub, output: aiOutput(input) }; },
++        DONE: new Set(['succeeded', 'cached']), FAILED: new Set(['failed', 'cancelled', 'expired']),
++        // A report brief asks for the batch class and gets a pending run id; the answer comes later through get().
++        async start(input, opts = {}) {
++            aiCalls.push(input);
++            if (aiStartError) { const err = aiStartError; aiStartError = null; throw err; }
++            const id = `run_batch_${++aiSeq}`;
++            aiRuns.set(id, { runId: id, status: 'queued', model: 'test-model', stub: false, output: null, error: null, input, opts });
++            return { ...aiRuns.get(id), done: false };
++        },
++        async get(id) { const r = aiRuns.get(id); return r ? { ...r } : null; },
++        async run(input, opts = {}) { aiCalls.push(input); backgroundCalls++; return { runId: `run_bg_${++aiSeq}`, model: 'test-model', stub: aiStub, output: aiOutput(input), options: opts }; },
      };
++    const answerBatch = (id) => { const r = aiRuns.get(id); r.status = 'succeeded'; r.stub = aiStub; r.output = aiOutput(r.input); return r; };
++    const failBatch = (id, status = 'failed') => { const r = aiRuns.get(id); r.status = status; r.error = { code: 'batch.failed', detail: 'test' }; return r; };
      const t = await boot({ sec, ai, log: quiet, env: { TRADE_BRIEFS_PER_HOUR: '3' } });
      const db = t.ctx.store.db;
++    // These checks isolate one instrument at a time: close any batch left by an earlier check, and take every
++    // other company out of the briefs queue (its facts row stays, but s.ok = 0 is not due).
++    const resetBriefs = async () => {
++        await db.prepare('DELETE FROM trade_ai_brief_jobs').run();
++        await db.prepare('UPDATE trade_facts_state SET ok = 0').run();
++        t.clock.advance(3600_000 + 1);
++    };
      const apple = await t.instrument({ symbol: 'AAPL', name: 'Apple Inc.', cik: '320193', exchange: 'Nasdaq' });
      const rev = (input) => input.sources.findIndex((s) => /Revenue \(quarter\)/.test(s.title));
      aiOutput = (input) => ({
@@@ -94,18 +118,27 @@ const quiet = { log() {}, warn() {}, error() {} };
              kind: 'filing', form_type: '10-Q', filer_name: 'Apple Inc.', cik: '0000320193', accession: '0000320193-26-000090',
          }, apple);
          const calls = secCalls.length;
--        const tick = await t.ctx.worker.briefsTick();
--        assert.deepStrictEqual(tick, { facts: 1, briefs: 1, failed: 0 });
++        const started = await t.ctx.worker.briefsTick();
++        assert.deepStrictEqual(started, { facts: 1, briefs: 0, failed: 0 }, 'the batch was started, not waited for');
          assert.strictEqual(secCalls.length, calls + 1, 'the numbers were read again after the report');
          const input = aiCalls[aiCalls.length - 1];
          assert.deepStrictEqual(input.instrument, { symbol: 'AAPL', name: 'Apple Inc.', exchange: 'Nasdaq' });
          assert.ok(input.sources.length <= 50 && input.sources.some((s) => s.source_type === 'trade.sec_fact') && input.sources.some((s) => s.source_type === 'trade.sec_filing'));
          assert.match(input.sources[rev(input)].content, /3 months to 2026-06-27: 109417000000 USD \(filing 0000320193-26-000020\)/);
++        const job = await db.prepare('SELECT * FROM trade_ai_brief_jobs WHERE instrument_id = ?').get(apple.id);
++        assert.ok(job && job.status === 'pending' && /^run_batch_/.test(job.run_id), 'the pending run id is kept with the work item');
++        assert.strictEqual(job.inputs_hash.length, 64);
++        assert.strictEqual(aiRuns.get(job.run_id).opts.deadlineHours, 3, 'the run asks for the batch class with a deadline (a market brief is stale after 3 h)');
++        assert.match(aiRuns.get(job.run_id).opts.idempotencyKey, /:batch:1$/, 'a retried start replays the same run');
++        assert.strictEqual(await t.ctx.briefs.latest(apple), null, 'no brief before the answer is read');
++        answerBatch(job.run_id);
++        assert.deepStrictEqual(await t.ctx.worker.briefsTick(), { facts: 0, briefs: 1, failed: 0 }, 'the answer is collected on a later pass');
          const b = await t.ctx.briefs.latest(apple);
          assert.strictEqual(b.points.length, 1, 'the uncited point is dropped');
          assert.ok(b.gaps.some((g) => /nothing behind it/.test(g)), 'and listed as a gap');
          assert.ok(b.points[0].cites.every((c) => c.kind === 'observation' && /^obs_/.test(c.id)));
          assert.strictEqual(b.authorship.reviewed, false);
++        assert.strictEqual(b.authorship.run_id, job.run_id);
          assert.strictEqual(b.as_of, '2026-06-27T00:00:00.000Z');
          assert.deepStrictEqual(await t.ctx.worker.briefsTick(), { facts: 0, briefs: 0, failed: 0 }, 'nothing new, nothing written');
      });
@@@ -137,10 +170,10 @@ const quiet = { log() {}, warn() {}, error() {} };
          await t.ctx.facts.record(msft);
          const keep = aiOutput;
          aiOutput = () => ({ summary: 'Strong buy: you should buy before the price target is reached.', observations: [], citations: [0], gaps: [] });
--        await assert.rejects(t.ctx.briefs.make(msft), (e) => e.code === 'brief.advice_refused');
++        await assert.rejects(t.ctx.briefs.make(msft, { trigger: 'reader' }), (e) => e.code === 'brief.advice_refused');
          aiOutput = keep;
          aiStub = true;
--        await assert.rejects(t.ctx.briefs.make(msft), (e) => e.code === 'brief.stub');
++        await assert.rejects(t.ctx.briefs.make(msft, { trigger: 'reader' }), (e) => e.code === 'brief.stub');
          aiStub = false;
          assert.strictEqual(await t.ctx.briefs.latest(msft), null, 'nothing kept');
      });
@@@ -165,6 +198,111 @@ const quiet = { log() {}, warn() {}, error() {} };
          assert.strictEqual((await t.ctx.briefs.make(goog)).created, true, 'the next hour has room again');
      });
  
++    aiOutput = (input) => ({ summary: `Figures for ${input.instrument.symbol}.`, observations: [], citations: [0], gaps: [] });
++    aiStub = false;
++    await resetBriefs();
++
++    await check('a pending batch survives a restart and its answer is collected after it', async () => {
++        const tsla = await t.instrument({ symbol: 'TSLA', name: 'Tesla, Inc.', cik: '1318605' });
++        await t.ctx.facts.record(tsla);
++        const started = await t.ctx.worker.briefsTick();
++        assert.strictEqual(started.briefs, 0, 'nothing written while the batch waits');
++        const job = await db.prepare('SELECT * FROM trade_ai_brief_jobs WHERE instrument_id = ?').get(tsla.id);
++        assert.ok(job && job.status === 'pending' && job.run_id, 'the run id is a row, not a promise');
++        await t.restart();
++        answerBatch(job.run_id);
++        const collected = await t.ctx.worker.briefsTick();
++        assert.strictEqual(collected.briefs, 1, 'the restarted worker reads the same pending run');
++        const b = await t.ctx.briefs.latest(tsla);
++        assert.ok(b && b.authorship.run_id === job.run_id);
++        assert.strictEqual((await db.prepare('SELECT status FROM trade_ai_brief_jobs WHERE instrument_id = ?').get(tsla.id)).status, 'done');
++    });
++
++    await check('an expired batch falls back to the background class once', async () => {
++        await resetBriefs();
++        const amd = await t.instrument({ symbol: 'AMD', name: 'Advanced Micro Devices', cik: '2488' });
++        await t.ctx.facts.record(amd);
++        await t.ctx.worker.briefsTick();
++        const job = await db.prepare('SELECT * FROM trade_ai_brief_jobs WHERE instrument_id = ?').get(amd.id);
++        assert.ok(job && job.status === 'pending');
++        const before = backgroundCalls;
++        t.clock.advance(4 * 3600_000);                          // past the 3 h deadline and the 30 min grace
++        const tick = await t.ctx.worker.briefsTick();           // the run is still 'queued' in the mock
++        assert.strictEqual(backgroundCalls, before + 1, 'one background run, not a retry loop');
++        assert.strictEqual(tick.briefs, 1);
++        const b = await t.ctx.briefs.latest(amd);
++        assert.ok(b && /^run_bg_/.test(b.authorship.run_id), 'the fallback answer is the brief');
++        const count = async () => Number((await db.prepare('SELECT COUNT(*) AS n FROM trade_ai_briefs WHERE instrument_id = ?').get(amd.id)).n);
++        assert.strictEqual(await count(), 1);
++        await t.ctx.worker.briefsTick();
++        assert.strictEqual(backgroundCalls, before + 1, 'the fallback is not repeated on the next tick');
++        assert.strictEqual(await count(), 1, 'and the answer is not written twice');
++        t.clock.advance(3600_000 + 1);
++    });
++
++    await check('a run collected twice writes one brief', async () => {
++        await resetBriefs();
++        const intc = await t.instrument({ symbol: 'INTC', name: 'Intel Corporation', cik: '50863' });
++        await t.ctx.facts.record(intc);
++        await t.ctx.worker.briefsTick();
++        const job = await db.prepare('SELECT * FROM trade_ai_brief_jobs WHERE instrument_id = ?').get(intc.id);
++        answerBatch(job.run_id);
++        assert.strictEqual((await t.ctx.worker.briefsTick()).briefs, 1);
++        const count = async () => Number((await db.prepare('SELECT COUNT(*) AS n FROM trade_ai_briefs WHERE instrument_id = ?').get(intc.id)).n);
++        assert.strictEqual(await count(), 1);
++        // The same work item and answer come round again (a retried pass): the run id refuses the second write.
++        await db.prepare("UPDATE trade_ai_brief_jobs SET status = 'pending' WHERE instrument_id = ?").run(intc.id);
++        const again = await t.ctx.briefs.collect(6);
++        assert.strictEqual(again.briefs, 0, 'the answer was already kept');
++        assert.strictEqual(await count(), 1);
++        assert.strictEqual((await db.prepare('SELECT status FROM trade_ai_brief_jobs WHERE instrument_id = ?').get(intc.id)).status, 'done');
++        t.clock.advance(3600_000 + 1);
++    });
++
++    await check('a batch refused at the door falls back to background once', async () => {
++        await resetBriefs();
++        const ba = await t.instrument({ symbol: 'BA', name: 'The Boeing Company', cik: '12927' });
++        await t.ctx.facts.record(ba);
++        aiStartError = Object.assign(new Error('batch disabled'), { code: 'ai.http_422', status: 422 });
++        const before = backgroundCalls;
++        const tick = await t.ctx.worker.briefsTick();
++        assert.strictEqual(backgroundCalls, before + 1, 'the refused batch answered in the background');
++        assert.strictEqual(tick.briefs, 1);
++        assert.strictEqual(Number((await db.prepare('SELECT COUNT(*) AS n FROM trade_ai_brief_jobs WHERE instrument_id = ?').get(ba.id)).n), 0, 'no batch is left open');
++        const b = await t.ctx.briefs.latest(ba);
++        assert.ok(b && /^run_bg_/.test(b.authorship.run_id));
++    });
++
++    await check('the client sends the batch class and its deadline to OpenVibe.AI', async () => {
++        const calls = [];
++        const fakeFetch = async (url, opts) => {
++            calls.push({ url: String(url), body: opts && opts.body ? String(opts.body) : null });
++            if (String(url).includes('/oauth/token')) {
++                return { ok: true, status: 200, json: async () => ({ access_token: 'test-token', token_type: 'Bearer', expires_in: 3600 }) };
++            }
++            if (String(url).includes('wait=0')) {
++                return { ok: true, status: 202, json: async () => ({ run: { id: 'run_batch_client', status: 'queued' } }) };
++            }
++            return { ok: true, status: 201, json: async () => ({ run: { id: 'run_bg_client', status: 'succeeded', output: { summary: 'ok' }, provenance: { model: 'm' } } }) };
++        };
++        const client = createAi({
++            config: { ai: { internalUrl: 'http://ai.test', waitMs: 1000 }, oauth: { clientId: 'trade', clientSecret: 'secret' }, networkInternalUrl: 'http://network.test' },
++            fetchImpl: fakeFetch,
++        });
++        const started = await client.start({ sources: [] }, { deadlineHours: 6, idempotencyKey: 'trade-brief:ins_test:hash:batch:1' });
++        assert.strictEqual(started.runId, 'run_batch_client');
++        assert.strictEqual(started.done, false);
++        const batchCall = calls.find((c) => c.url.includes('wait=0'));
++        const batchBody = JSON.parse(batchCall.body);
++        assert.deepStrictEqual(batchBody.options, { latency: 'batch', deadline_hours: 6 });
++        assert.strictEqual(batchBody.idempotency_key, 'trade-brief:ins_test:hash:batch:1');
++        const bg = await client.run({ sources: [] }, { latency: 'background', idempotencyKey: 'trade-brief:ins_test:hash:bg:2' });
++        assert.strictEqual(bg.runId, 'run_bg_client');
++        const bgBody = JSON.parse(calls[calls.length - 1].body);
++        assert.deepStrictEqual(bgBody.options, { latency: 'background' });
++        assert.strictEqual(bgBody.idempotency_key, 'trade-brief:ins_test:hash:bg:2');
++    });
++
      await t.close();
      done();
  })();