feat(sec): cap how long the fundamentals import can stay deferred

MISSING_XBRL_RETRY_DAYS bounds how long ONE filing blocks promotion. It does
not bound the import as a whole, and the two come apart because a blocking
filing is only queued by promote(), which a deferred run never reaches. During
a rolling supply of unresolvable filings — earnings season, when SEC's
Company-Facts aggregation lags furthest — each new arrival restarts the 3-day
clock before the previous one clears, and nothing is written at all: not the
good rows, not the gap rows that would stop those filings blocking again.

Add an aggregate ceiling. Once promotions have been stale for
PROMOTION_CEILING_DAYS (7), every unresolved filing is aged past the retry
window in place, so promote() queues them all through the path that already
exists, source_max_date advances, and _missing() keeps queued rows aged-out on
later runs. The import self-heals instead of compounding.

Deliberately not the alternative of queueing gap rows on a deferred run: that
would drop the grace period to a single run for every filing, including the
common case of a Company-Facts lag that resolves in a day, and it needs a write
on a run that failed validation.

The per-filing window is untouched, a never-promoted source never trips (that
is initial setup, not a wedge), and affected symbols stay barred from setups
either way since setup_blocked_ciks ignores the window. A forced promotion
raises promotion_ceiling_forced so the safety valve is never silent.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

app/services/sec_fundamentals_importer.py

@@ -51,10 +51,10 @@ import json
 import logging
 from collections import Counter, defaultdict
 from dataclasses import dataclass, field, replace
-from datetime import date, datetime, timedelta, timezone
+from datetime import date, datetime, time, timedelta, timezone
 from typing import Any, Callable
 
-from sqlalchemy import delete, select, update
+from sqlalchemy import delete, exists, select, update
 
 from app.database import insert_for_session
 from app.models.data_import_run import DataImportRun
@@ -81,6 +81,26 @@ MIN_BACKFILL_COVERAGE = 0.5
 # three); past that it is misfiled, not late, and blocking forever costs more
 # than the missing filing does — see the unresolved-filing guardrail below.
 MISSING_XBRL_RETRY_DAYS = 3
+
+# Aggregate ceiling on deferral. MISSING_XBRL_RETRY_DAYS bounds how long ONE
+# filing blocks; it does not bound how long the import as a whole can stay
+# deferred. Those differ because a blocking filing is only queued by promote(),
+# which a deferred run never reaches — so during a rolling supply of
+# unresolvable filings (earnings season, when SEC's Company-Facts aggregation is
+# furthest behind) each new arrival restarts the clock before the previous one
+# clears, and nothing is written at all: not the good rows, not the gap rows
+# that would stop those filings blocking again.
+#
+# Once promotions have been stale this long, every unresolved filing is treated
+# as past the window. promote() then queues them all (see the _past_retry_window
+# call there), source_max_date advances, and _missing() forces queued rows
+# aged-out on later runs so they never block again — the import self-heals
+# through the paths that already exist.
+#
+# Well above MISSING_XBRL_RETRY_DAYS so ordinary overlapping blocks never trip
+# it. Affected symbols stay barred from setups either way: setup_blocked_ciks is
+# built from every missing filing regardless of window.
+PROMOTION_CEILING_DAYS = 7
 FILING_GAP_ESCALATE_DAYS = 14
 # Share-count band a co-registrant-recovered row must land in, relative to the
 # issuer's own last snapshot. Wide enough for buybacks/issuance, nowhere near
@@ -157,6 +177,9 @@ class SecFundamentalsImporter:
         self._retry_rows: list[dict[str, Any]] = []
         self._latest_index_date: date | None = None
         self._backfill = False
+        # Set by validate() when the aggregate ceiling forced the block open;
+        # read by promote() to alert that it did.
+        self._ceiling_tripped: dict[str, Any] | None = None
 
     # -- SourceImporter protocol -------------------------------------------
 
@@ -421,6 +444,24 @@ class SecFundamentalsImporter:
         # reconstructible by re-walking the index.
         blocking = _within_retry_window(staged.missing_xbrl)
         aged_out = _past_retry_window(staged.missing_xbrl)
+
+        # ...unless promotions have been stale past the aggregate ceiling, in
+        # which case the deferral has cost more than the filings it withholds.
+        # Ageing them here (not just locally) is deliberate: promote() re-derives
+        # the queue from the same list, so this is what gets them queued.
+        self._ceiling_tripped = None
+        if blocking and db is not None and await self._promotions_stale(db):
+            for item in staged.missing_xbrl:
+                item["age_days"] = max(
+                    item.get("age_days", 0), MISSING_XBRL_RETRY_DAYS + 1
+                )
+            self._ceiling_tripped = {
+                "forced": len(blocking),
+                "unresolved": len(staged.missing_xbrl),
+            }
+            blocking = _within_retry_window(staged.missing_xbrl)
+            aged_out = _past_retry_window(staged.missing_xbrl)
+
         if blocking:
             messages.append(
                 f"{len(blocking)} tracked XBRL filing(s) unresolved within the "
@@ -464,6 +505,9 @@ class SecFundamentalsImporter:
             "missing_xbrl": staged.missing_xbrl[:50],
             "missing_xbrl_count": len(staged.missing_xbrl),
             "missing_xbrl_blocking": len(blocking),
+            # Present only when the aggregate ceiling forced this run through, so
+            # a promoted run that carries known-unresolved filings says so.
+            "promotion_ceiling_tripped": self._ceiling_tripped,
             "recovered_from_coregistrant": staged.recovered[:50],
             "recovered_count": len(staged.recovered),
             # Complete compact gate input; detailed audit lists above stay capped.
@@ -616,6 +660,25 @@ class SecFundamentalsImporter:
                 created_at=_now(),
             ))
 
+        # A ceiling-forced promotion is the safety valve firing — it must be
+        # visible, or the import silently starts carrying known-unresolved
+        # filings. The affected symbols stay barred from setups regardless.
+        if self._ceiling_tripped:
+            db.add(SystemEvent(
+                severity="warning",
+                source="sec_facts",
+                code="promotion_ceiling_forced",
+                message=(
+                    f"Promoted with {self._ceiling_tripped['unresolved']} unresolved "
+                    f"filing(s) — {self._ceiling_tripped['forced']} still inside the "
+                    f"{MISSING_XBRL_RETRY_DAYS}-day retry window — because nothing had "
+                    f"promoted in {PROMOTION_CEILING_DAYS} days. They are queued for "
+                    "retry and their symbols remain blocked from setups."
+                )[:4000],
+                dedup_key=f"sec_facts:promotion_ceiling_forced:{run_id}",
+                created_at=now,
+            ))
+
         # Persistent current gaps get one actionable escalation rather than a
         # daily warning. The nullable marker makes this durable and noise-free.
         escalation_cutoff = now - timedelta(days=FILING_GAP_ESCALATE_DAYS)
@@ -757,6 +820,38 @@ class SecFundamentalsImporter:
             if accession not in resolved
         ]
 
+    async def _promotions_stale(self, db) -> bool:
+        """Has nothing promoted within ``PROMOTION_CEILING_DAYS``?
+
+        Only true for a source that HAS promoted before. A never-promoted import
+        is initial setup, not a wedge: forcing its first promotion through would
+        mask a misconfiguration rather than recover from a transient SEC gap.
+
+        Measured from ``self.today`` rather than the wall clock, so the ceiling
+        honors the same injected date that ages the filings it releases.
+        """
+        cutoff = datetime.combine(
+            self.today - timedelta(days=PROMOTION_CEILING_DAYS),
+            time.min,
+            tzinfo=timezone.utc,
+        )
+        ever, recent = (
+            await db.execute(
+                select(
+                    exists().where(
+                        DataImportRun.source == SOURCE,
+                        DataImportRun.status == STATUS_PROMOTED,
+                    ),
+                    exists().where(
+                        DataImportRun.source == SOURCE,
+                        DataImportRun.status == STATUS_PROMOTED,
+                        DataImportRun.started_at >= cutoff,
+                    ),
+                )
+            )
+        ).one()
+        return bool(ever) and not bool(recent)
+
     async def _last_processed_index_date(self, db) -> date | None:
         return (
             await db.execute(

tests/unit/test_sec_fundamentals_importer.py

@@ -6,7 +6,7 @@ from __future__ import annotations
 import json
 import os
 import tempfile
-from datetime import date, datetime, timezone
+from datetime import date, datetime, timedelta, timezone
 
 import pytest
 from sqlalchemy import func, select
@@ -1128,3 +1128,79 @@ async def test_malformed_cik_override_is_ignored_not_fatal(engine):
         resolved = await resolve_ciks(db, client)
 
     assert resolved.symbol_to_cik["AAPL"] == 320193  # fell back to company_tickers
+
+
+# --- aggregate deferral ceiling -------------------------------------------
+
+def _blocking_staged(today: date) -> StagedFundamentals:
+    """One filing still inside the per-filing retry window."""
+    return StagedFundamentals(
+        resolved=ResolvedUniverse(),
+        missing_xbrl=[{
+            "cik": "0000320193",
+            "accession": "YOUNG-1",
+            "form": "10-Q",
+            "index_date": today,
+            "age_days": 0,
+            "reason": "not_in_companyfacts",
+        }],
+    )
+
+
+async def _add_run(factory, *, status: str, started_at: datetime) -> None:
+    from app.models.data_import_run import DataImportRun
+
+    async with factory() as db:
+        db.add(DataImportRun(
+            source="sec_facts", status=status, started_at=started_at,
+        ))
+        await db.commit()
+
+
+async def _validate_with_history(engine, *, promoted_days_ago: int | None):
+    factory = _factory(engine)
+    today = date(2026, 5, 20)
+    now = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc)
+    if promoted_days_ago is not None:
+        await _add_run(
+            factory,
+            status=STATUS_PROMOTED,
+            started_at=now - timedelta(days=promoted_days_ago),
+        )
+    importer = SecFundamentalsImporter(today=today)
+    importer._latest_index_date = date(2026, 5, 19)
+    staged = _blocking_staged(today)
+    async with factory() as db:
+        return await importer.validate(db, staged), staged, importer
+
+
+async def test_ceiling_forces_a_promotion_once_deferral_outlasts_it(engine):
+    """The per-filing window bounds one filing; this bounds the whole import."""
+    result, staged, importer = await _validate_with_history(engine, promoted_days_ago=10)
+
+    assert result.ok is True
+    assert result.summary["missing_xbrl_blocking"] == 0
+    assert result.summary["promotion_ceiling_tripped"] == {"forced": 1, "unresolved": 1}
+    # Aged in place, so promote() re-derives the same verdict and queues it.
+    assert staged.missing_xbrl[0]["age_days"] > 3
+    # The symbol stays barred from setups — promoting is not trusting the data.
+    assert result.summary["setup_blocked_ciks"] == ["0000320193"]
+
+
+async def test_a_recent_promotion_keeps_the_normal_block(engine):
+    result, staged, _ = await _validate_with_history(engine, promoted_days_ago=1)
+
+    assert result.ok is False
+    assert result.summary["missing_xbrl_blocking"] == 1
+    assert result.summary["promotion_ceiling_tripped"] is None
+    assert result.retryable is True
+    assert staged.missing_xbrl[0]["age_days"] == 0
+
+
+async def test_ceiling_never_fires_before_a_first_promotion(engine):
+    """No baseline means initial setup, not a wedge — forcing it through would
+    mask a misconfiguration instead of recovering from an SEC gap."""
+    result, _, _ = await _validate_with_history(engine, promoted_days_ago=None)
+
+    assert result.ok is False
+    assert result.summary["promotion_ceiling_tripped"] is None