lib/batch.js

// Anthropic writing helpers for the 3-round pipeline.
// Highest-quality model is Fable 5; we probe it once and fall back to Opus 4.8 if
// the org's data-retention setting (or any unavailability) rejects it.
//
// NIR runs the LIVE Messages API (streamed, bounded concurrency) rather than the
// Message Batches API, to speed turnaround — `runRound` is the live implementation.
// The batch path is kept as `runRoundBatch` (+ submit/await/collect) for fallback.

const Anthropic = require("@anthropic-ai/sdk");
const crypto = require("crypto");
const { env } = require("./env");

// Anthropic batch custom_id max is 64 chars; several slugs are longer. Map a slug
// to a short, deterministic, unique id and translate results back.
function shortId(slug) {
  if (slug.length <= 64) return slug;
  const h = crypto.createHash("sha1").update(slug).digest("hex").slice(0, 8);
  return slug.slice(0, 55) + "-" + h; // 55 + 1 + 8 = 64
}

const FABLE = "claude-fable-5";
const OPUS = "claude-opus-4-8";

// Server-side web search (dynamic-filtering variant — supported on Fable 5 / Opus 4.8).
// Cap uses to bound cost per post.
const WEB_SEARCH = { type: "web_search_20260209", name: "web_search", max_uses: 10 };

let _client = null;
function client() {
  if (!_client) {
    _client = new Anthropic({ apiKey: env("ANTHROPIC_API_KEY", { required: true }) });
  }
  return _client;
}

function sleep(ms) {
  return new Promise((r) => setTimeout(r, ms));
}

// Model-specific request knobs. Fable 5 has always-on thinking (omit the param);
// Opus 4.8 takes adaptive thinking. Both take effort via output_config.
function modelParams(model) {
  const p = { model, output_config: { effort: "high" } };
  if (model !== FABLE) p.thinking = { type: "adaptive" };
  return p;
}

// Probe Fable 5 with one tiny request. Returns the model id to use for the run.
// Falls back to Opus 4.8 on a 400 (retention/availability) or any error.
async function probeModel(preferred = FABLE) {
  if (preferred !== FABLE) return preferred;
  try {
    const res = await client().messages.create({
      ...modelParams(FABLE),
      max_tokens: 16,
      messages: [{ role: "user", content: "Reply with the single word: ok" }],
    });
    if (res.stop_reason === "refusal") {
      console.warn("[probe] Fable 5 refused the probe — falling back to Opus 4.8");
      return OPUS;
    }
    console.log("[probe] Fable 5 available — using it.");
    return FABLE;
  } catch (e) {
    const status = e?.status || e?.statusCode;
    console.warn(
      `[probe] Fable 5 unavailable (${status || ""} ${e?.message || e}) — falling back to Opus 4.8`,
    );
    return OPUS;
  }
}

// Build a batch request object. `tools` is optional (e.g. [WEB_SEARCH]).
function request(custom_id, { model, system, userContent, max_tokens = 16000, tools }) {
  const params = {
    ...modelParams(model),
    max_tokens,
    system: [{ type: "text", text: system, cache_control: { type: "ephemeral" } }],
    messages: [{ role: "user", content: userContent }],
  };
  if (tools && tools.length) params.tools = tools;
  return { custom_id, params };
}

async function submitBatch(requests) {
  const batch = await client().messages.batches.create({ requests });
  console.log(`[batch] submitted ${batch.id} (${requests.length} requests)`);
  return batch.id;
}

// Poll until processing_status === "ended".
async function awaitBatch(id, { intervalMs = 45000 } = {}) {
  for (;;) {
    const b = await client().messages.batches.retrieve(id);
    if (b.processing_status === "ended") {
      console.log(
        `[batch] ${id} ended — succeeded:${b.request_counts.succeeded} errored:${b.request_counts.errored} canceled:${b.request_counts.canceled} expired:${b.request_counts.expired}`,
      );
      return b;
    }
    console.log(
      `[batch] ${id} ${b.processing_status} — processing:${b.request_counts.processing}`,
    );
    await sleep(intervalMs);
  }
}

// Stream results -> map keyed by custom_id. Each value: {ok, text, stop_reason, model, error}.
async function collectResults(id) {
  const out = new Map();
  for await (const r of await client().messages.batches.results(id)) {
    const cid = r.custom_id;
    if (r.result.type === "succeeded") {
      const msg = r.result.message;
      const text = (msg.content || [])
        .filter((b) => b.type === "text")
        .map((b) => b.text)
        .join("");
      out.set(cid, {
        ok: msg.stop_reason !== "refusal" && !!text,
        text,
        stop_reason: msg.stop_reason,
        model: msg.model,
      });
    } else {
      out.set(cid, {
        ok: false,
        error: r.result.type,
        detail: r.result.error ? JSON.stringify(r.result.error) : null,
      });
    }
  }
  return out;
}

// Run one full round: submit -> await -> collect. If the model was Fable 5 and any
// items refused/errored, re-queue just those on Opus 4.8.
async function runRoundBatch(model, items, buildSystemAndUser) {
  // custom_id = shortId(slug); keep a map to translate results back to slugs.
  const buildRequests = (its, m) => {
    const cidToSlug = new Map();
    const reqs = its.map((it) => {
      const cid = shortId(it.slug);
      cidToSlug.set(cid, it.slug);
      const { system, userContent, max_tokens, tools } = buildSystemAndUser(it);
      return request(cid, { model: m, system, userContent, max_tokens, tools });
    });
    return { reqs, cidToSlug };
  };

  const { reqs, cidToSlug } = buildRequests(items, model);
  const id = await submitBatch(reqs);
  await awaitBatch(id);
  const raw = await collectResults(id); // keyed by custom_id
  const results = new Map();
  for (const [cid, v] of raw) results.set(cidToSlug.get(cid) || cid, v);

  // Re-queue Fable 5 refusals on Opus 4.8 (fallbacks aren't available in batch).
  if (model === FABLE) {
    const retry = items.filter((it) => {
      const r = results.get(it.slug);
      return !r || !r.ok;
    });
    if (retry.length) {
      console.warn(`[round] re-queueing ${retry.length} failed item(s) on Opus 4.8`);
      const { reqs: rreq, cidToSlug: rmap } = buildRequests(retry, OPUS);
      const rid = await submitBatch(rreq);
      await awaitBatch(rid);
      const rraw = await collectResults(rid);
      for (const [cid, v] of rraw) results.set(rmap.get(cid) || cid, v);
    }
  }
  return results; // keyed by slug
}

// ---- Live Messages API path (default for NIR) ------------------------------

// Bounded-concurrency async map (preserves input order in the returned array).
async function mapLimit(items, limit, fn) {
  const ret = new Array(items.length);
  let next = 0;
  async function worker() {
    while (next < items.length) {
      const idx = next++;
      ret[idx] = await fn(items[idx], idx);
    }
  }
  const n = Math.max(1, Math.min(limit, items.length));
  await Promise.all(Array.from({ length: n }, worker));
  return ret;
}

// One live streamed generation. Streaming is used because high max_tokens + adaptive
// thinking + web search can exceed the non-streaming request window. Same result shape
// as collectResults' values: {ok, text, stop_reason, model} or {ok:false, error, detail}.
async function createLive(model, { system, userContent, max_tokens = 16000, tools }) {
  const params = {
    ...modelParams(model),
    max_tokens,
    system: [{ type: "text", text: system, cache_control: { type: "ephemeral" } }],
    messages: [{ role: "user", content: userContent }],
  };
  if (tools && tools.length) params.tools = tools;
  const stream = client().messages.stream(params);
  const msg = await stream.finalMessage();
  const text = (msg.content || [])
    .filter((b) => b.type === "text")
    .map((b) => b.text)
    .join("");
  return {
    ok: msg.stop_reason !== "refusal" && !!text,
    text,
    stop_reason: msg.stop_reason,
    model: msg.model,
  };
}

// Concurrency cap for live calls (override via DL_CONCURRENCY). Modest by default to
// stay under web-search / token rate limits while still fanning the 15 briefs out.
const LIVE_CONCURRENCY = Number(process.env.DL_CONCURRENCY || 4);

// Live equivalent of runRoundBatch: same signature, same slug-keyed return Map.
// Per-item Fable 5 -> Opus 4.8 fallback on refusal/error.
async function runRoundLive(model, items, buildSystemAndUser) {
  const results = new Map();
  await mapLimit(items, LIVE_CONCURRENCY, async (it) => {
    const spec = buildSystemAndUser(it);
    let r;
    try {
      r = await createLive(model, spec);
    } catch (e) {
      r = { ok: false, error: "api_error", detail: String(e?.message || e) };
    }
    if (model === FABLE && (!r || !r.ok)) {
      console.warn(`[live] ${it.slug} failed on Fable 5 — retrying on Opus 4.8`);
      try {
        r = await createLive(OPUS, spec);
      } catch (e) {
        r = { ok: false, error: "api_error", detail: String(e?.message || e) };
      }
    }
    results.set(it.slug, r);
    console.log(
      `[live] ${it.slug} ${r.ok ? "ok" : "FAIL"} (${r.model || model}${r.ok ? "" : " — " + (r.error || r.stop_reason)})`,
    );
    return r;
  });
  return results;
}

// Default export: the live path. Swap to runRoundBatch to use Message Batches instead.
const runRound = runRoundLive;

module.exports = {
  FABLE,
  OPUS,
  WEB_SEARCH,
  shortId,
  client,
  probeModel,
  modelParams,
  request,
  submitBatch,
  awaitBatch,
  collectResults,
  createLive,
  runRoundLive,
  runRoundBatch,
  runRound,
};