Ingestion pipeline: fetch, cluster, synthesize

The whole ingestion path — fetch, dedupe, enrichment, clustering, synthesis, caching — lives in one file: src/lib/services/rss-aggregation.ts. It runs inline, synchronously, inside whichever request triggered it. There is no worker, no job table, and no intermediate persistence between stages. The public entry point is listAggregatedStories() (src/lib/services/rss-aggregation.ts:3830), which returns cache.stories from loadFeedCache().

Two paths, one live

The rssQueuePipeline flag defaults to false (src/lib/contracts/feature-flags.ts:52) and is flipped by hand at /admin/flags, where it renders as "RSS queue pipeline" with the description "Routes RSS aggregation through the Vercel Queues fan-out instead of the inline path. Off by default; the inline aggregation remains authoritative until this is enabled." Even with it on, only the rss.ingest consumer does work, and that work is the same inline aggregation. See the queue subsystem below.

What triggers a build

The cron handler (src/app/api/admin/rss/refresh/route.ts, maxDuration = 300) does three things in order: publish an ingest message if the queue flag is on, then always run ingestAllTenantScopedFeeds(), then run evaluateInstantAlertRulesForAllTenants(). Tenant ingestion is deliberately not gated on the queue publish — the in-code comment states that the rss.ingest consumer never touches tenant scopes, so gating it would reopen the staleness bug it was written to close.

ingestAllTenantScopedFeeds() (src/lib/platform/local-platform-data.ts:1203) enumerates completed newsroom tenant scopes, processes them serially under TENANT_FEED_INGESTION_TIME_BUDGET_MS (ERMIS_TENANT_INGESTION_TIME_BUDGET_MS, default 240000), and starts at a rotating offset of floor(now / 10 min) % n so the tail is not permanently starved. Each scope is refreshed with { forceRefresh: true, bootstrapRefresh: 'never' } and is failure-isolated. It returns { tenantScopes, refreshed, failed, skippedTimeBudget, elapsedMs }.

Stage 1 — which sources are in scope

A global build reads listSourceCatalogFromRepository({ ingestEnabledOnly: true }) (rss-aggregation.ts:3571), i.e. rows that are both active and ingest_enabled. In the bundled catalog that is 13 of 1,427 entries.

A tenant build resolves its own list in resolveTenantScopedSourceCatalog (src/lib/platform/local-platform-data.ts:963):

Catalog mechanics, the active versus ingest_enabled distinction, and the manual sync route are covered in Source catalog operations.

Stage 2 — fetch

fetchFeed (rss-aggregation.ts:3290) fetches each feed through fetchPublicNetworkResource (rss-aggregation.ts:66), which validates the target against the shared public-network blocklist before connecting, follows redirects with redirect: 'manual', and re-validates every hop up to MAX_PUBLIC_FETCH_REDIRECTS = 4. DNS resolution is skipped when NODE_ENV === 'test'.

• Request headers: User-Agent: ErmisAI/1.0 RSS Aggregator, plus RSS/Atom/XML Accept.
• Timeout: FEED_REQUEST_TIMEOUT_MS, default 8000 ms, via AbortController.
• Concurrency: FEED_FETCH_CONCURRENCY, default 12.

Every fetch writes a status back to Postgres via updateSourceFetchStatus — ok, failed, or empty. Sources whose id starts with custom- are skipped: they have no source_catalog_entries row, so the update would match nothing (src/lib/db/source-catalog-repository.ts:690-694). Custom-source health therefore exists only inside the tenant's scoped feed cache, which is why the admin "Source health" table never lists them.

An empty parse is diagnosed by getEmptyFeedParseMessage (rss-aggregation.ts:1461) into one of three strings, and that string is what the admin table shows as the last error:

• Feed response looked like HTML and did not contain RSS or Atom items
• Feed response content-type {ct} did not contain RSS or Atom items
• Feed parsed with no RSS item or Atom entry elements

Failed and empty results are recorded with errorRate24h: 100.

Stage 3 — parse and normalize

The parser is regex-based, not an XML library. parseFeedItems (rss-aggregation.ts:1419) matches <item>…</item> and <entry>…</entry> blocks. Body preference is content:encoded / content, falling back to description / summary. Categories come from <category> and <dc:subject>.

• Per-source cap: MAX_ITEMS_PER_SOURCE, default 30.
• URL normalization strips utm_*, fbclid, gclid and the hash fragment.
• The article id is a deterministic hash of sourceId|url|publishedAt|headline, prefixed story- (createArticleFromFeedItem, rss-aggregation.ts:1577).
• breaking is a heuristic: the headline contains "breaking" or "live", or the item is 90 minutes old or less (toBreaking, rss-aggregation.ts:1562).
• Dedupe is by normalized URL, ties broken by longer body then newer publishedAt (dedupeArticles, rss-aggregation.ts:1756).

Stage 4 — article enrichment

Most feeds ship teasers. isLikelyTeaserBody (rss-aggregation.ts:1598) treats a body as a teaser when it is empty, equals the headline, is under 55 words, or matches teaserMarkerRegex (rss-aggregation.ts:559):

/read more|continue reading|click here|περισσότερα|συνέχεια|full story|learn more|view on site/i

Teasers are re-fetched from the article URL through the same SSRF-guarded path, with User-Agent: ErmisAI/1.0 Article Fetcher and ARTICLE_REQUEST_TIMEOUT_MS (default 5000 ms). Extraction prefers <article> / <main> blocks of 120 words or more, else joined <p> blocks of 10 words or more each; the result is discarded below 70 words. Successful bodies are cached in a process-global map for ARTICLE_BODY_CACHE_TTL_MS (default 45 minutes).

Enrichment is capped at MAX_ARTICLE_ENRICHMENT_PER_REFRESH (default 48) with ARTICLE_ENRICHMENT_CONCURRENCY (default 6). When it succeeds, bodyOrigin flips from 'feed' to 'article-fetch'.

Stage 5 — clustering

Articles are sorted newest-first and truncated to MAX_CLUSTER_INPUT_ARTICLES (default 48) before clustering.

Grouping is always heuristic. clusterArticlesHeuristically (rss-aggregation.ts:2043) does greedy grouping on articleSimilarity (rss-aggregation.ts:2013):

headlineSimilarity * 0.75 + bodySimilarity * 0.25 + (categoryOverlap ? 0.08 : 0)

Both similarities are Jaccard over tokenized text; the body comparison uses only the first 320 characters. The join threshold is HEURISTIC_CLUSTER_SIMILARITY_THRESHOLD = 0.24. Groups are seeded in source-tier order — wire 1 > tier1 0.9 > tier2 0.72 > tier3 0.55 — then by recency.

The AI pass refines metadata; it never decides membership. clusterArticlesWithAi (rss-aggregation.ts:2664) sends only clusters with two or more articles, and the model returns exactly eventSummary, category, breaking, and one uniqueAngle per article (aiClusterRefinementSchema, rss-aggregation.ts:492). Article-to-cluster assignment is entirely the heuristic's.

Two details worth knowing:

• maxOutputTokens for refinement scales with cluster size: min(8192, max(2048, 256 + n × 96)) (resolveClusterRefinementMaxOutputTokens, rss-aggregation.ts:212). A flat 2048 cap structurally truncated the JSON at roughly 35 articles, which clusters reach during major breaking events.
• The refinement prompt instructs the model to write eventSummary and category in the dominant language of the source material (rss-aggregation.ts:2576). This is the one stage that does not follow the newsroom's content locale. The returned category is then normalized through the canonical registry (src/lib/i18n/categories.ts), falling back to keyword inference and finally world.

Clusters are prioritized by member count, then newest member, and truncated to MAX_CLUSTERS_PER_REFRESH (default 12) before synthesis.

Stage 6 — synthesis

synthesizeClusterWithAi (rss-aggregation.ts:2878) emits a structured object validated by aiSynthesisSchema (rss-aggregation.ts:522): headline (5–300 chars), body (100–15000 chars), eventSummary, uniqueAngles (max 8), conflicts (max 8), and sensitive.

Inputs are bounded hard: only the top AI_SYNTHESIS_MAX_SOURCE_ARTICLES = 8 articles by tier and recency are sent, each body truncated to AI_SYNTHESIS_SOURCE_BODY_CHAR_LIMIT = 700 characters.

Voice rules are baked into the prompt (rss-aggregation.ts:2979-2991): first-party original reporting, the event as the subject rather than an outlet's coverage of it, explicit forbidden phrasings ("Al Jazeera published…", "the BBC reported…", "according to the description"), inline attribution only for specific or contested facts, and a length constraint of 220–900 words with "a tight 400-600 word article is ideal".

Quality gate. If the returned body is under MIN_SYNTHESIZED_BODY_WORDS = 220 words, one bounded regeneration runs with explicit revision feedback naming the actual word count; the revision is kept only if it is longer (rss-aggregation.ts:3038-3058). A failure of the revision pass is swallowed and the short draft is kept.

Synthesis runs across clusters at FEED_SYNTHESIS_CONCURRENCY (default 3).

Stage 7 — confidence and pipeline status

calculateConfidence (rss-aggregation.ts:2796) is a weighted sum clamped to 40–99:

sourceScore × 30 + averageTierWeight × 25 + agreementScore × 25 + recencyScore × 20

• sourceScore = min(1, uniqueSources / 5).
• averageTierWeight uses the tier weights above.
• agreementScore is the mean pairwise Jaccard of headlines, floored at 0.25 (0.45 for a single article). It is lexical headline overlap, not fact comparison.
• recencyScore is a step function on the newest member: ≤ 30 min → 1, ≤ 2 h → 0.86, ≤ 8 h → 0.68, ≤ 24 h → 0.52, otherwise 0.4. A missing timestamp scores 0.35.

resolveStoryPhase (rss-aggregation.ts:3107) then assigns the pipeline status:

if (input.confidence >= STORY_AUTOPUBLISH_CONFIDENCE && !input.sensitive) {
  return 'published' as const
}

return 'review' as const

STORY_AUTOPUBLISH_CONFIDENCE is 85. The pipeline only ever emits review or published. It never emits collecting, clustering, or synthesizing — those values exist in the contract enum but no code path produces them.

sensitive comes from the model. In the heuristic path it comes from sensitiveContentRegex (rss-aggregation.ts:2121), which covers English and Greek terms only.

Stage 8 — materialization into editorial drafts

listTenantScopedArticleDrafts (src/lib/platform/drafts/tenant-draft-store.ts:239) reads the tenant's stories and its persisted editorial_story_drafts rows, then runs synchronizeDraftCollectionWithStories, creating a draft for every story id it has not seen.

The English placeholder (src/lib/stories/draft-state.ts:4-11):

• headline — Draft in composition
• body — Initial article composition is pending in English.\n\nUse the compose flow to generate the first publishable draft.
• source line — The initial article will be composed from {n} sources once the first compose action runs.

Localized variants exist for all ten content locales in the same file.

The synthesized text is not lost: it is preserved in the draft's story_snapshot jsonb column and is what the compose flow consumes through buildComposeStoryContextBody (local-platform-data.ts:441-468), which truncates the source body at 8000 characters.

Other materialization behaviour:

• Initial editorial state (resolveInitialEditorialState, local-platform-data.ts:490): pipeline review → in_review with submittedForReviewAt set to story.updatedAt; pipeline published → draft. A new draft never starts at editorial published; publishedAt is owned exclusively by the human approve transition.
• Initial priority: high when the story is breaking or sensitive, otherwise standard.
• Claim provenance is synthesized deterministically from paragraphs and source attributions, max 8 claims (buildDraftClaimProvenance, local-platform-data.ts:551).
• Persistence: editorial_story_drafts, primary key (tenant_scope_id, story_id), with a version column for optimistic concurrency and CHECK constraints enforcing in_review ⇒ submitted_for_review_at, published ⇒ published_at, rejected ⇒ reviewed_at (src/lib/db/schema.ts:530-615). Sync writes are compare-and-set guarded in passive mode: a version conflict means a human edit landed, so the sync copy is skipped rather than clobbering it.
• New stories fire onNewStoriesSynchronized → evaluateAlertRulesForTenant, scheduled with runAfterResponse so the feed request never blocks on alert evaluation.

What the feed derives from the draft

mapDraftToFeedItem (src/lib/platform/feed/feed-lifecycle.ts:60) calls deriveStoryWorkflowState (src/lib/platform/feed/publishability.ts:42):

• articleReady = !doesDraftRequireInitialCompose({ headline, body }) — false while the placeholder is present.
• When articleReady is false, any phase outside {collecting, clustering, synthesizing} collapses to synthesizing (resolvePreEditorialPhase, publishability.ts:24-26). Since the pipeline only emits review or published, a freshly materialized draft always shows the "Ready to compose" chip.
• deliverable = articleReady && editorialState === 'published'; reviewQueueEligible = articleReady && editorialState === 'in_review'.

The confusing consequence, and it is the normal case: a story with pipeline status review shows an "In review" editorial badge next to a "Ready to compose" pipeline chip, and does not appear on the editorial board, because board eligibility requires articleReady.

Caching, scoping and locking

This is where most operational behaviour lives.

Scope key. getFeedScopeKey (rss-aggregation.ts:3438) returns global when no source catalog is passed, sources:<comma-joined ids> for a tenant, sources:none for an explicit empty catalog, plus a |locale:<code> suffix when the content locale is not the platform default. Two tenants with the same source set and the same content locale share one cache entry and therefore one AI build.

Three layers, checked in order:

1. Per-instance globals (globalThis.__ermisStoryPipeline*, rss-aggregation.ts:598-607).
2. A Postgres platform_state row keyed platform:story-feed-cache:v1:<encoded scope> with expires_at.
3. A rebuild.

TTL is FEED_CACHE_TTL_MS (default 3 minutes); the persisted row's retention is FEED_CACHE_PERSIST_RETENTION_MS (default 3 hours). Persisted caches are zod- and predicate-validated on read, and any malformed record is discarded wholesale (coercePersistedFeedCache, rss-aggregation.ts:699). A Postgres error disables the persisted layer process-wide for the rest of that instance's life (disableFeedCacheDb). Empty caches are never persisted.

Serve-stale-then-refresh versus synchronous cold build. A warm, non-empty cache is returned immediately and a background refresh is kicked when it is stale. A cold, empty, or forced path builds synchronously and awaits. The comment at rss-aggregation.ts:3823-3827 records why: a fire-and-forget bootstrap build never converged on serverless, because the instance freezes after the HTTP response — it fetched sources but never synthesized or persisted, leaving cold tenants stuck.

Throttles (per instance): FEED_AUTO_REFRESH_MIN_INTERVAL_MS 45 s and FEED_FORCE_REFRESH_MIN_INTERVAL_MS 20 s. The throttle is honoured only when there is real content to serve, so a cold tenant always converges.

Cross-instance build lock. Redis SET NX PX on ermis:feed:build-lock:<scope> with FEED_BUILD_LOCK_TTL_MS = 7 minutes, deliberately longer than the 300 s function budget.

In-flight promise dedupe has a staleness escape hatch: a load promise older than FEED_BUILD_PROMISE_STALE_MS = 120 s is treated as dead (a frozen serverless instance) and a rebuild starts.

On build failure with allowCacheFallback, the previous cache is re-served without re-stamping expiresAt (rss-aggregation.ts:3738-3756) — deliberately, so a repeatedly failing build cannot indefinitely extend a cache that claims to be fresh.

Strict mode and the fallback matrix

isStrictAiPipelineMode() is ERMIS_AI_PIPELINE_STRICT_MODE === 'true', default false. shouldAllowAiPipelineFallback() is !strictMode && NODE_ENV !== 'production'. shouldAllowFeedCacheFallback() is !strictMode (rss-aggregation.ts:630-650).

One consequence of that padding: ensureSubstantiveSynthesis (rss-aggregation.ts:2364-2392) can replace a valid-but-short AI body with heuristic prose while keeping the AI headline and the synthesisModel id, so a story labelled with a model id may contain template text. The heuristic narrative templates ship in English and Greek only and pick whichever script dominates the source material; they do not honour the tenant's content locale.

isAiPipelineConfigured() (rss-aggregation.ts:1991) returns false when ERMIS_DISABLE_AI_PIPELINE === 'true', and otherwise checks hasAzureLanguageModelConfiguration() in azure mode or hasGatewayLanguageModelConfiguration() in gateway mode.

How the pipeline calls models

Per-stage execution policies (src/lib/ai/language-model-policies.ts:132-170):

maxRetries: 0 is set on every generateText call, so the AI SDK's own retry is off and the pipeline's ladder is the only retry mechanism.

Two distinct fallback mechanisms exist and they are easy to confuse.

1. The gateway in-request chain (stageFallbackChain, language-model-policies.ts:241-254) is executed by the Vercel AI Gateway inside a single request on 5xx, rate limit, or timeout: cluster: gemini-3-flash → claude-haiku-4.5 → grok-4-fast-non-reasoning; synthesis: claude-sonnet-4.6 → gemini-3-flash → grok-4-fast-non-reasoning.
2. The application-level ladder (pipelineStageRetryModelIds, language-model-policies.ts:226-229) runs when a whole attempt fails: cluster: openai/gpt-4.1-nano → anthropic/claude-haiku-4.5 → anthropic/claude-sonnet-4.6; synthesis: openai/gpt-4.1-nano → anthropic/claude-sonnet-4.6 → anthropic/claude-opus-4.6.

Attempt order is [primary, configuredPrimary, ...retryIds] deduped and truncated to MAX_CLUSTER_AI_ATTEMPTS / MAX_SYNTHESIS_AI_ATTEMPTS, both 3, padded by repeating the last model when the ladder is short. resolveStagePrimaryModelId (rss-aggregation.ts:1165) reads the diagnostics snapshot: if the last recorded failure was on this stage with the configured model, attempt one starts on a different ladder model. In azure provider mode the app-level ladder is empty, so all three attempts reuse the configured Azure model.

Backoff between attempts is min((i + 1) × 750 ms, 3000 ms). Error classes configuration, authentication, authorization, billing, and quota short-circuit the ladder immediately (rss-aggregation.ts:1094-1104); content-filter, context-limit, and schema failures deliberately do not, because a different model may succeed.

Pipeline invocations run under tenantScopeId: 'platform:shared', actorType: 'system', billingScope: 'platform', billable: false, on surfaces pipeline_cluster and pipeline_synthesis. They are platform cost, not tenant cost — but assertAiInvocationAllowed still runs first, and the aiPaused kill switch and the platform spend cap hard-block them even when per-tenant guardrails are disabled. See AI usage accounting, guardrails and metering.

Telemetry spans are enabled only in production, with recordInputs: false and recordOutputs: false, so source material and generated bodies never enter spans.

Diagnostics

getNewsPipelineDiagnosticsSnapshot() (rss-aggregation.ts:1041) exposes status (idle | building | healthy | error), strictMode, aiConfigured, build timestamps and duration, lastBuildArticleCount, lastBuildClusterCount, lastBuildStoryCount, and a lastError object carrying stage, operation, message, model id and provider id. It is served by GET /api/admin/ai/diagnostics and rendered in the "News pipeline diagnostics" card on /admin/ai (super_admin only).

Two limits make this less useful than it looks:

• Only global scope builds are recorded. Tenant-scope builds — which is what the cron actually runs — are invisible here.
• The snapshot lives in a process global, so it is per-instance and not shared across lambdas.

The queue subsystem: what is actually wired

Four topics are declared in src/lib/platform/queues-adapter.ts:28-33 — rss.ingest, rss.enrich, rss.cluster, rss.synthesize — and bound to consumer routes by experimentalTriggers of type queue/v2beta in vercel.json. Delivery is push-based; the app never polls.

The ingest visibility timeout is 360 s on purpose, strictly greater than the 300 s maxDuration: an equal window redelivers a message whose handler was killed at exactly 300 s, "stacking up to maxAttempts whole-catalog AI builds from one message" (ingest/route.ts:34-38).

Be blunt about the rest:

• rss.ingest is not a fan-out. publishRssAggregationRefresh publishes exactly one message per refresh with sourceId: 'all-sources' and feedUrl: ''. Cross-source clustering makes aggregation a whole-catalog operation, so per-source and per-stage staging is explicitly reserved for a future refactor backed by a persistence layer.
• Enabling the flag does not offload tenant ingestion. The consumer refreshes the global catalog scope only. Tenant scopes stay on the cron's ingestAllTenantScopedFeeds.
• /admin/queues shows fabricated metrics. listQueueHealth (src/lib/platform/admin/dashboards.ts:149-196) derives four synthetic rows from the global draft feed and the cross-tenant review queue, with several fields hardcoded to 0 and the ingest, cluster and synthesis cards all sharing one lagSeconds (the age of the newest story), while the review-queue card hardcodes lagSeconds: 0 along with its retries and dead-letter counts. Nothing in the codebase reads Vercel Queue depth, retries, or the dead-letter topics.
• There is no platform DLQ. On the final attempt, createRssConsumer publishes a summary to a sibling <topic>-dlq topic, awaits it, acks, and swallows the error. Nothing reads those topics.

Messages carry an HMAC-SHA256 signature over a canonical JSON of every non-signature field sorted by key, keyed by ERMIS_QUEUE_SIGNING_SECRET. This exists because @vercel/queue's handleCallback performs no inbound verification and the consumer routes are allow-listed past Clerk in src/proxy.ts — without the HMAC, an anonymous POST to /api/queues/rss/ingest would trigger a full AI aggregation run. Verification is timingSafeEqual and fails closed: no secret configured means nothing verifies, so nothing is processed. Publishing without the secret throws rather than emitting a message that would be rejected.

More on scheduling and the operator view in Scheduled jobs and the queue subsystem.

Debugging an empty or stale feed

isAuthorizedCronRequest compares Authorization: Bearer ${CRON_SECRET} with timingSafeEqual and returns false when CRON_SECRET is unset, so a missing environment variable can never open the endpoint — it can only silence the schedule.

Tuning knobs

Every value below goes through readBoundedRuntimeNumber (rss-aggregation.ts:107-126): out-of-range values are clamped, and an unparseable value silently falls back to the default.

Non-numeric switches read by this subsystem. CRON_SECRET, ERMIS_CONTENT_LOCALE and ERMIS_TENANT_INGESTION_TIME_BUDGET_MS are documented in the README (lines 65-67); the strict-mode, disable-pipeline, disable-persisted-cache and queue-signing switches are not documented anywhere except this page and docs/ops/launch-checklist.md:

Constants that are not configurable at all: MIN_SYNTHESIZED_BODY_WORDS = 220, STORY_AUTOPUBLISH_CONFIDENCE = 85, HEURISTIC_CLUSTER_SIMILARITY_THRESHOLD = 0.24, AI_CLUSTER_SUMMARY_CHAR_LIMIT = 160, AI_SYNTHESIS_MAX_SOURCE_ARTICLES = 8, AI_SYNTHESIS_SOURCE_BODY_CHAR_LIMIT = 700, MAX_CLUSTER_AI_ATTEMPTS = MAX_SYNTHESIS_AI_ATTEMPTS = 3, FEED_BUILD_LOCK_TTL_MS = 7 min, FEED_BUILD_PROMISE_STALE_MS = 120 s, MAX_PUBLIC_FETCH_REDIRECTS = 4, FEED_CACHE_STORAGE_VERSION = 1.