worldmonitor/scripts/lib/digest-orchestration-helpers.mjs

// Pure helpers for the digest cron's per-user compose loop.
//
// Extracted from scripts/seed-digest-notifications.mjs so they can be
// unit-tested without dragging the cron's env-checking side effects
// (DIGEST_CRON_ENABLED check, Upstash REST helper, Convex relay
// auth) into the test runtime. The cron imports back from here.

import { compareRules, MAX_STORIES_PER_USER } from './brief-compose.mjs';
import { generateDigestProse } from './brief-llm.mjs';

/**
 * Build the email subject string. Extracted so the synthesis-level
 * → subject ternary can be unit-tested without standing up the whole
 * cron loop. (Plan acceptance criterion A6.i.)
 *
 * Rules:
 *   - synthesisLevel 1 or 2 + non-empty briefLead → "Intelligence Brief"
 *   - synthesisLevel 3 OR empty/null briefLead → "Digest"
 *
 * Mirrors today's UX where the editorial subject only appeared when
 * a real LLM-produced lead was available; the L3 stub falls back to
 * the plain "Digest" subject to set reader expectations correctly.
 *
 * @param {{ briefLead: string | null | undefined; synthesisLevel: number; shortDate: string }} input
 * @returns {string}
 */
export function subjectForBrief({ briefLead, synthesisLevel, shortDate }) {
  if (briefLead && synthesisLevel >= 1 && synthesisLevel <= 2) {
    return `WorldMonitor Intelligence Brief — ${shortDate}`;
  }
  return `WorldMonitor Digest — ${shortDate}`;
}

/**
 * Single source of truth for the digest's story window. Used by BOTH
 * the compose path (digestFor closure in the cron) and the send loop.
 * Without this, the brief lead can be synthesized from a 24h pool
 * while the channel body ships 7d / 12h of stories — reintroducing
 * the cross-surface divergence the canonical-brain refactor is meant
 * to eliminate, just in a different shape.
 *
 * `lastSentAt` is the rule's previous successful send timestamp (ms
 * since epoch) or null on first send. `defaultLookbackMs` is the
 * first-send fallback (today: 24h).
 *
 * @param {number | null | undefined} lastSentAt
 * @param {number} nowMs
 * @param {number} defaultLookbackMs
 * @returns {number}
 */
export function digestWindowStartMs(lastSentAt, nowMs, defaultLookbackMs) {
  return lastSentAt ?? (nowMs - defaultLookbackMs);
}

/**
 * Walk an annotated rule list and return the winning candidate +
 * its non-empty story pool. Two-pass: due rules first (so the
 * synthesis comes from a rule that's actually sending), then ALL
 * eligible rules (compose-only tick — keeps the dashboard brief
 * fresh for weekly/twice_daily users). Within each pass, walk by
 * compareRules priority and pick the FIRST candidate whose pool is
 * non-empty AND survives `tryCompose` (when provided).
 *
 * Returns null when every candidate is rejected — caller skips the
 * user (same as today's behavior on empty-pool exhaustion).
 *
 * Plan acceptance criteria A6.l (compose-only tick still works for
 * weekly user) + A6.m (winner walks past empty-pool top-priority
 * candidate). Codex Round-3 High #1 + Round-4 High #1 + Round-4
 * Medium #2.
 *
 * `tryCompose` (optional): called with `(cand, stories)` after a
 * non-empty pool is found. Returning a truthy value claims the
 * candidate as winner and the value is forwarded as `composeResult`.
 * Returning a falsy value (e.g. composeBriefFromDigestStories
 * dropped every story via its URL/headline/shape filters) walks to
 * the next candidate. Without this callback, the helper preserves
 * the original "first non-empty pool wins" semantics, which let a
 * filter-rejected top-priority candidate suppress the brief for the
 * user even when a lower-priority candidate would have shipped one.
 *
 * `digestFor` receives the full annotated candidate (not just the
 * rule) so callers can derive a per-candidate story window from
 * `cand.lastSentAt` — see `digestWindowStartMs`.
 *
 * `log` is the per-rejected-candidate log emitter — passed in so
 * tests can capture lines without reaching for console.log.
 *
 * @param {Array<{ rule: object; lastSentAt: number | null; due: boolean }>} annotated
 * @param {(cand: { rule: object; lastSentAt: number | null; due: boolean }) => Promise<unknown[] | null | undefined>} digestFor
 * @param {(line: string) => void} log
 * @param {string} userId
 * @param {((cand: { rule: object; lastSentAt: number | null; due: boolean }, stories: unknown[]) => Promise<unknown> | unknown)} [tryCompose]
 * @returns {Promise<{ winner: { rule: object; lastSentAt: number | null; due: boolean }; stories: unknown[]; composeResult?: unknown } | null>}
 */
export async function pickWinningCandidateWithPool(annotated, digestFor, log, userId, tryCompose) {
  if (!Array.isArray(annotated) || annotated.length === 0) return null;
  const sortedDue = annotated.filter((a) => a.due).sort((a, b) => compareRules(a.rule, b.rule));
  const sortedAll = [...annotated].sort((a, b) => compareRules(a.rule, b.rule));
  // Build the walk order, deduping by rule reference so the same
  // rule isn't tried twice (a due rule appears in both sortedDue
  // and sortedAll).
  const seen = new Set();
  const walkOrder = [];
  for (const cand of [...sortedDue, ...sortedAll]) {
    if (seen.has(cand.rule)) continue;
    seen.add(cand.rule);
    walkOrder.push(cand);
  }
  for (const cand of walkOrder) {
    const stories = await digestFor(cand);
    if (!stories || stories.length === 0) {
      log(
        `[digest] brief filter drops user=${userId} ` +
          `sensitivity=${cand.rule.sensitivity ?? 'high'} ` +
          `variant=${cand.rule.variant ?? 'full'} ` +
          `due=${cand.due} ` +
          `outcome=empty-pool ` +
          `in=0 dropped_severity=0 dropped_url=0 dropped_headline=0 dropped_shape=0 dropped_cap=0 out=0`,
      );
      continue;
    }
    if (typeof tryCompose === 'function') {
      const composeResult = await tryCompose(cand, stories);
      if (!composeResult) {
        log(
          `[digest] brief filter drops user=${userId} ` +
            `sensitivity=${cand.rule.sensitivity ?? 'high'} ` +
            `variant=${cand.rule.variant ?? 'full'} ` +
            `due=${cand.due} ` +
            `outcome=filter-rejected ` +
            `in=${stories.length} out=0`,
        );
        continue;
      }
      return { winner: cand, stories, composeResult };
    }
    return { winner: cand, stories };
  }
  return null;
}

/**
 * Run the three-level canonical synthesis fallback chain.
 *   L1: full pre-cap pool + ctx (profile, greeting, !public) — canonical.
 *   L2: envelope-sized slice + empty ctx — degraded fallback (mirrors
 *       today's enrichBriefEnvelopeWithLLM behaviour).
 *   L3: null synthesis — caller composes from stub.
 *
 * Returns { synthesis, level } with `synthesis` matching
 * generateDigestProse's output shape (or null on L3) and `level`
 * one of {1, 2, 3}.
 *
 * Pure helper — no I/O beyond the deps.callLLM the inner functions
 * already perform. Errors at L1 propagate to L2; L2 errors propagate
 * to L3 (null/stub). `trace` callback fires per level transition so
 * callers can quantify failure-mode distribution in production logs.
 *
 * Plan acceptance criterion A6.h (3-level fallback triggers).
 *
 * @param {string} userId
 * @param {Array} stories — full pre-cap pool
 * @param {string} sensitivity
 * @param {{ profile: string | null; greeting: string | null }} ctx
 * @param {{ callLLM: Function; cacheGet: Function; cacheSet: Function }} deps
 * @param {(level: 1 | 2 | 3, kind: 'success' | 'fall' | 'throw', err?: unknown) => void} [trace]
 * @returns {Promise<{ synthesis: object | null; level: 1 | 2 | 3 }>}
 */
export async function runSynthesisWithFallback(userId, stories, sensitivity, ctx, deps, trace) {
  const noteTrace = typeof trace === 'function' ? trace : () => {};
  // L1 — canonical
  try {
    const l1 = await generateDigestProse(userId, stories, sensitivity, deps, {
      profile: ctx?.profile ?? null,
      greeting: ctx?.greeting ?? null,
      isPublic: false,
    });
    if (l1) {
      noteTrace(1, 'success');
      return { synthesis: l1, level: 1 };
    }
    noteTrace(1, 'fall');
  } catch (err) {
    noteTrace(1, 'throw', err);
  }
  // L2 — degraded fallback
  try {
    const cappedSlice = (Array.isArray(stories) ? stories : []).slice(0, MAX_STORIES_PER_USER);
    const l2 = await generateDigestProse(userId, cappedSlice, sensitivity, deps);
    if (l2) {
      noteTrace(2, 'success');
      return { synthesis: l2, level: 2 };
    }
    noteTrace(2, 'fall');
  } catch (err) {
    noteTrace(2, 'throw', err);
  }
  // L3 — stub
  noteTrace(3, 'success');
  return { synthesis: null, level: 3 };
}

/**
 * READ-time freshness predicate. Returns true if the story:track:v1 row
 * should be dropped because its source `publishedAt` is older than the
 * cutoff. Used by buildDigest to keep pre-deploy residue (whose ingest
 * gate has since been tightened) from shipping in briefs.
 *
 * Behaviour matrix:
 *   - publishedAt is a positive integer epoch-ms AND < cutoff → drop (true).
 *   - publishedAt is a positive integer epoch-ms AND ≥ cutoff → keep (false).
 *   - publishedAt is missing/unparseable/zero/negative → keep (false).
 *
 * The "missing → keep" branch is back-compat for legacy story:track:v1
 * rows written before publishedAt was persisted. Pre-deploy residue
 * with no publishedAt is NOT caught here — handle it via the audit
 * script's `--mode=residue` (one-shot eviction). Once that has run AND
 * ≥1 cron cycle has refreshed publishedAt on still-active rows, any
 * row reaching this predicate without publishedAt is anomalous, not
 * residue.
 *
 * See: skill ingest-gate-tightening-leaves-residue-in-read-path.
 *
 * @param {Record<string, string> | null | undefined} track
 * @param {number} ageCutoffMs — drop rows with publishedAt strictly less than this
 * @returns {boolean}
 */
export function shouldDropTrackByAge(track, ageCutoffMs) {
  const pubMs = Number.parseInt(track?.publishedAt ?? '', 10);
  if (!Number.isInteger(pubMs) || pubMs <= 0) return false;
  return pubMs < ageCutoffMs;
}

/**
 * Compute the READ-time freshness cutoff for a given digest window.
 * Cutoff is anchored to `windowStartMs` (from `digestWindowStartMs`)
 * minus a 24h buffer that accommodates sustained stories whose first
 * mention sits just before the window edge.
 *
 * Daily user (24h window) → 48h-ago cutoff.
 * Weekly user (7d window) → 8d-ago cutoff.
 *
 * @param {number} windowStartMs
 * @returns {number}
 */
export function readTimeAgeCutoffMs(windowStartMs) {
  const STALE_BUFFER_MS = 24 * 60 * 60 * 1000;
  return windowStartMs - STALE_BUFFER_MS;
}

/**
 * Sprint 1 / U2 — option (a) canonical-send mapping.
 *
 * Given the per-user winner record from the compose phase
 * (briefByUser.get(userId), shape: `{ envelope, magazineUrl, chosenVariant,
 * synthesisLevel }`) and the candidate rule list for that user, return
 * the SINGLE rule whose channel body the send loop should use for this
 * user-slot. Under option (a), the email body and the magazine URL
 * BOTH come from this rule's pool — no per-rule fan-out, no
 * winner-vs-non-winner channel divergence.
 *
 * Returns null when:
 *   - briefByUser has no entry for this user (compose found no
 *     non-empty pool — nothing to send).
 *   - briefByUser entry has no chosenVariant (defensive; treat as
 *     "no canonical winner identified" and skip the send).
 *   - candidate list contains no rule whose `variant` matches
 *     `chosenVariant` (rule was deleted between compose and send;
 *     skip rather than misroute to a sibling rule).
 *
 * The send loop calls this BEFORE the per-rule isDue check — non-
 * winner rules are dropped here, so the cron's `digest:last-sent:v1:
 * ${userId}:${variant}` write only ever happens for the winner-variant
 * key. Old non-winner-variant keys from pre-option-(a) sends remain in
 * Redis until their 8d TTL elapses; they are orphan but harmless
 * (nothing reads them after this change).
 *
 * Pure helper — no I/O.
 *
 * @param {{ chosenVariant?: string } | null | undefined} brief
 * @param {Array<{ userId?: string; variant?: string }>} userRules — all rules for ONE user
 * @returns {{ userId?: string; variant?: string } | null}
 */
export function selectCanonicalSendRule(brief, userRules) {
  if (!brief || typeof brief.chosenVariant !== 'string' || brief.chosenVariant === '') {
    return null;
  }
  if (!Array.isArray(userRules) || userRules.length === 0) return null;
  for (const rule of userRules) {
    if (!rule || typeof rule.variant !== 'string') continue;
    if (rule.variant === brief.chosenVariant) return rule;
  }
  return null;
}