fix(gdelt-intel): share one timestamp validator between the fetch-ordering and health paths (#5858)

scripts/_seed-contract.mjs

@@ -45,9 +45,9 @@ const OPTIONAL_FIELDS = new Set([
   'recordCount',     // legacy — kept optional through PR 2, removed in PR 3 in favor of declareRecords
   'metaTtlSeconds',  // legacy — used today by writeSeedMeta / writeExtraKeyWithMeta (e.g. scripts/seed-jodi-gas.mjs); removed in PR 3 when legacy meta writes go away
   // Content-age contract (2026-05-04 health-readiness plan).
-  // `contentMeta` is a function `(rawData) => {newestItemAt, oldestItemAt} | null`
-  // invoked by runSeed BEFORE publishTransform so seeders can compute item-age
-  // metadata from helper fields that are stripped before publish.
+  // `contentMeta` is a function `(rawData, runStartedAtMs) => {newestItemAt, oldestItemAt} | null`
+  // invoked by runSeed BEFORE publishTransform with the immutable run clock, so
+  // seeders can compute item-age metadata from helper fields that are stripped before publish.
   // `maxContentAgeMin` is the seeder's content-staleness budget in minutes.
   // The two opt in TOGETHER: declaring contentMeta without maxContentAgeMin
   // (or vice-versa) is a contract violation — see the cross-field check below.

scripts/_seed-utils.mjs

@@ -1961,7 +1961,7 @@ export async function runSeed(domain, resource, canonicalKey, fetchFn, opts = {}
     sourceVersion,         // new — required when declareRecords is passed
     schemaVersion,         // new — required when declareRecords is passed
     zeroIsValid = false,   // new — when true, recordCount=0 is OK_ZERO, not RETRY
-    contentMeta,           // (rawData) => {newestItemAt, oldestItemAt} | null
+    contentMeta,           // (rawData, runStartedAtMs) => {newestItemAt, oldestItemAt} | null
     maxContentAgeMin,      // positive integer minutes — opts in together with contentMeta
     fetchPhaseTimeoutMs,   // hard ceiling on the fetch phase; defaults to lockTtlMs + margin (#4786)
   } = opts;
@@ -2128,7 +2128,11 @@ export async function runSeed(domain, resource, canonicalKey, fetchFn, opts = {}
     : lockTtlMs + FETCH_PHASE_DEADLINE_MARGIN_MS;
   let data;
   try {
-    data = await raceFetchDeadline(withRetry(fetchFn), fetchDeadlineMs, `${domain}:${resource}`);
+    data = await raceFetchDeadline(
+      withRetry(() => fetchFn({ runStartedAtMs: startMs })),
+      fetchDeadlineMs,
+      `${domain}:${resource}`,
+    );
   } catch (err) {
     // Keep the SIGTERM handler installed across the fetch-failure
     // cleanup. Earlier code did `process.off('SIGTERM', sigTermHandler)`
@@ -2172,7 +2176,7 @@ export async function runSeed(domain, resource, canonicalKey, fetchFn, opts = {}
     let contentOldestAt = null;
     if (contentAgeOptedIn) {
       try {
-        const result = contentMeta(data);
+        const result = contentMeta(data, startMs);
         if (result && typeof result === 'object'
             && Number.isFinite(result.newestItemAt) && result.newestItemAt > 0
             && Number.isFinite(result.oldestItemAt) && result.oldestItemAt > 0) {

scripts/seed-gdelt-intel.mjs

@@ -252,6 +252,35 @@ function withBudget(operation, budgetMs, fallback, onTimeout) {
 // because the fetch order was only ever reconstructible from a sequence of
 // `Fetching x...` lines, never stated; emitting the ranking keys makes "why is
 // topic X still stale" answerable from the run log alone.
+const CLOCK_SKEW_TOLERANCE_MS = 60 * 60 * 1000;
+
+/**
+ * Parse a stored topic stamp (`fetchedAt` / `attemptedAt`) against the run clock.
+ *
+ * One validator for both readers of these stamps, per #5858. The fetch-ordering
+ * path clamped forward skew while the health path did not, so the same stored
+ * value produced two different numbers depending on who read it — and the
+ * unclamped one is the one `maxContentAgeMin` is evaluated against.
+ *
+ * Returns null for anything unusable (unparseable, non-finite, at or before the
+ * epoch) so each caller can pick its own sentinel. Every finite run clock
+ * clamps a tolerated future value to that clock so health never receives a
+ * future timestamp. Health callers can also reject values beyond the one-hour
+ * clock-skew tolerance instead of turning them into fresh evidence.
+ *
+ * @param {unknown} value
+ * @param {number} nowMs run clock; a non-finite value disables clock handling
+ * @param {{rejectBeyondTolerance?: boolean}} options
+ * @returns {number | null}
+ */
+export function parseStampMs(value, nowMs, { rejectBeyondTolerance = false } = {}) {
+  const parsed = Date.parse(value);
+  if (!Number.isFinite(parsed) || parsed <= 0) return null;
+  if (!Number.isFinite(nowMs)) return parsed;
+  if (rejectBeyondTolerance && parsed > nowMs + CLOCK_SKEW_TOLERANCE_MS) return null;
+  return Math.min(parsed, nowMs);
+}
+
 export function rankTopicsForFetch(topics, previous, nowMs) {
   const previousById = new Map();
   // Array.isArray, not `?? []`: the cache-merge below already treats this cached
@@ -260,11 +289,7 @@ export function rankTopicsForFetch(topics, previous, nowMs) {
   for (const topic of Array.isArray(previous?.topics) ? previous.topics : []) {
     if (topic?.id) previousById.set(topic.id, topic);
   }
-  const stampMs = (value) => {
-    const parsed = Date.parse(value);
-    if (!Number.isFinite(parsed)) return Number.NEGATIVE_INFINITY;
-    return Number.isFinite(nowMs) ? Math.min(parsed, nowMs) : parsed;
-  };
+  const stampMs = (value) => parseStampMs(value, nowMs) ?? Number.NEGATIVE_INFINITY;
   return topics
     .map((topic, index) => {
       const prev = previousById.get(topic.id);
@@ -301,6 +326,7 @@ function rankStampIso(ms) {
 export async function fetchAllTopics(deps = {}) {
   const {
     _now = () => Date.now(),
+    runStartedAtMs,
     _sleep = sleep,
     _fetchArticles = fetchArticlesOnce,
     _fetchTimeline = fetchTopicTimelineResult,
@@ -325,7 +351,7 @@ export async function fetchAllTopics(deps = {}) {
     _minRequestBudgetMs = MIN_REQUEST_BUDGET_MS,
     _interRequestDelayMs = GDELT_REQUEST_DELAY_MS,
   } = deps;
-  const runStartedAt = _now();
+  const runStartedAt = Number.isFinite(runStartedAtMs) ? runStartedAtMs : _now();
   const deadlineAt = runStartedAt + _softBudgetMs;
   const remaining = () => deadlineAt - _now();
 
@@ -934,15 +960,20 @@ export function declareRecords(data) {
 //                  is coasting (a topic is always attempted first, so any
 //                  GDELT success at all keeps this fresh);
 //   oldestItemAt = most starved topic, for operator visibility.
-export function contentMeta(data) {
+export function contentMeta(data, nowMs = Date.now()) {
   // Only topics that actually carry articles count: an articleless topic keeps
   // fetchedAt=now (the empty-topic placeholder), which would hold newestItemAt
   // fresh precisely in the total-death scenario — brownout + expired canonical,
   // nothing to backfill — where STALE_CONTENT matters most.
+  //
+  // Stamps go through the same parseStampMs the fetch ordering uses (#5858).
+  // The health mode keeps the shared run-clock clamp for tolerated skew but
+  // rejects a far-future stored stamp, so cache merge cannot mint fresh health
+  // evidence from a poisoned persisted value.
   const times = (data?.topics ?? [])
     .filter((t) => Array.isArray(t?.articles) && t.articles.length > 0)
-    .map((t) => Date.parse(t?.fetchedAt))
-    .filter((ms) => Number.isFinite(ms) && ms > 0);
+    .map((t) => parseStampMs(t?.fetchedAt, nowMs, { rejectBeyondTolerance: true }))
+    .filter((ms) => ms != null);
   if (times.length === 0) return null;
   return { newestItemAt: Math.max(...times), oldestItemAt: Math.min(...times) };
 }

tests/seed-content-age-contract.test.mjs

@@ -6,7 +6,7 @@
 //     violation that hard-fails at config time, not at write time).
 //   - maxContentAgeMin must be a positive integer (rejects 0, negatives,
 //     non-integer, undefined, null, strings).
-//   - contentMeta(rawData) runs BEFORE publishTransform(rawData) so seeders
+//   - contentMeta(rawData, runStartedAtMs) runs BEFORE publishTransform(rawData) so seeders
 //     can use pre-publish helper fields (e.g. _publishedAtIsSynthetic) for
 //     timestamp computation, then strip those helpers from the public payload.
 //   - contentMeta returning null OR throwing both result in newestItemAt:null
@@ -19,6 +19,10 @@ import { test, beforeEach, afterEach } from 'node:test';
 import assert from 'node:assert/strict';
 
 import { runSeed } from '../scripts/_seed-utils.mjs';
+import {
+  contentMeta as gdeltContentMeta,
+  fetchAllTopics,
+} from '../scripts/seed-gdelt-intel.mjs';
 
 const ORIGINAL_FETCH = globalThis.fetch;
 const ORIGINAL_EXIT = process.exit;
@@ -298,6 +302,67 @@ test('content-meta with valid timestamps → newestItemAt/oldestItemAt populated
   assert.equal(meta.maxContentAgeMin, 1440);
 });
 
+test('runSeed passes one run clock and publishes null for a persisted future GDELT stamp', async () => {
+  const FUTURE = '2099-01-01T00:00:00.000Z';
+  const TOPIC_IDS = ['military', 'cyber', 'nuclear', 'sanctions', 'intelligence', 'maritime'];
+  const previous = {
+    topics: TOPIC_IDS.map((id) => ({
+      id,
+      articles: [{ url: 'https://example.test/' + id }],
+      fetchedAt: FUTURE,
+      attemptedAt: FUTURE,
+    })),
+  };
+  let observedRunClock;
+  let fetchRunClock;
+
+  await runWithExitTrap(() =>
+    runSeed(
+      'test',
+      'gdelt-future-cache',
+      'test:gdelt-future-cache:v1',
+      (runContext) => {
+        fetchRunClock = runContext.runStartedAtMs;
+        return fetchAllTopics({
+          runStartedAtMs: fetchRunClock,
+          _now: () => fetchRunClock,
+          _loadPrevious: async () => previous,
+          _fetchArticles: async (topic) => ({
+            id: topic.id,
+            articles: [],
+            fetchedAt: FUTURE,
+            failureCode: 'GDELT_TEST_OUTAGE',
+          }),
+          _softBudgetMs: 60_000,
+          _minRequestBudgetMs: 1,
+          _interRequestDelayMs: 0,
+        });
+      },
+      {
+        validateFn: (data) => data?.topics?.filter((topic) => topic.articles.length > 0).length >= 3,
+        ttlSeconds: 3600,
+        declareRecords: (data) => data.topics.length,
+        sourceVersion: 'gdelt-future-cache-v1',
+        schemaVersion: 1,
+        maxStaleMin: 1440,
+        contentMeta: (data, runClock) => {
+          observedRunClock = runClock;
+          return gdeltContentMeta(data, runClock);
+        },
+        maxContentAgeMin: 1440,
+      },
+    ),
+  );
+
+  assert.ok(Number.isFinite(observedRunClock), 'runSeed must pass its immutable run clock to contentMeta');
+  assert.equal(observedRunClock, fetchRunClock, 'fetch ordering and content health must share the same run clock');
+  const meta = lastMetaSetBody('gdelt-future-cache');
+  assert.ok(meta, 'seed-meta must be written through the publish path');
+  assert.equal(meta.newestItemAt, null, 'a persisted far-future cache stamp must fail closed');
+  assert.equal(meta.oldestItemAt, null, 'a persisted far-future cache stamp must not renew the oldest edge');
+  assert.equal(meta.maxContentAgeMin, 1440);
+});
+
 // ── Greptile PR #3596 P1 regression: non-contract-mode seeders ──────────
 //
 // Pre-fix the seed-meta mirror gated on `envelopeMeta` (which is null for