fix(sec): tell an attribution collision apart from a changed reconstruction

snapshot_discrepancy named the accessions but not the columns, so it could not
distinguish "our numbers moved" from "the same filing is attributed twice".
The fields were already computed for validation_json and simply dropped from
the message; they are now in it.

A difference in cik ALONE is no longer reported as a reconstruction change at
all. Every fact matched, so two tracked CIKs are claiming one filing and the
fix is the universe, not the parser: it raises accession_cik_collision naming
both CIKs and sec_cik_overrides. It also never self-heals — the losing CIK
stores no row, so _ciks_with_snapshots never sees it and it is full-history
backfilled and re-reported every run until its ticker is re-pointed or retired.

Observed 2026-08-19 for EQR: after Equity Residential renamed to Vivmark
Residential (VMRK, CIK 906107), SEC's own company_tickers.json left the old
symbol on ERP Operating LP (CIK 931182), the non-traded co-registrant of their
combined 10-Qs. Both were tracked, both reconstructed the same two filings.

The reparse path now excludes cik-only differences from its rewrite set:
rewriting one would re-stamp the filing onto the co-registrant, taking it from
the issuer that actually filed it, which no parser fix asks for.

No stored value was wrong in that incident — reports/ carries the full
reproduction for this and for the companyfacts staleness behind the gate fix.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Lo99z3jWqu9X9ueBU3Z3D

app/services/fundamentals_quality_service.py

@@ -3,7 +3,9 @@
 from __future__ import annotations
 
 import json
+from collections import defaultdict
 from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import exists, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -15,6 +17,23 @@ from app.models.ticker import Ticker
 
 _SEC_FORMS = ("10-K", "10-Q", "10-K/A", "10-Q/A")
 
+# How recent the issuer's own newest filing must be for an *escalated* gap to
+# stop pausing setups. A gap pauses an issuer until it is either resolved or
+# superseded by a later ingested filing — which assumes the gap is temporary.
+# It is not always: SEC's per-company Company-Facts files can go stale
+# indefinitely (2026-08, 43 large caps whose Q2 10-Qs the frames API carried but
+# whose companyfacts files never received), and since the supersede rule needs a
+# *successfully ingested* later filing, a stale file also swallows the next
+# quarter. The pause is then open-ended rather than seasonal.
+#
+# So the pause hands off to the alert: once `filing_gap_aged` has escalated a gap
+# to an operator (`escalated_at`), the issuer resumes on the fundamentals it does
+# have — provided those are recent. An issuer with nothing this fresh has no
+# usable fundamentals at all and stays paused, which is the case the gate was
+# built for. The retry queue is untouched: `active_gaps` still returns these, so
+# the importer keeps retrying and a recovered filing still resolves normally.
+GAP_GATE_RECENT_FILING_DAYS = 180
+
 
 @dataclass(frozen=True)
 class SetupQuality:
@@ -51,6 +70,39 @@ async def active_gaps(
     return list((await db.execute(stmt)).scalars().all())
 
 
+async def _gap_exempt_ciks(
+    db: AsyncSession, gaps: list[SecFilingGap]
+) -> set[str]:
+    """CIKs whose gaps have stopped pausing setups (see GAP_GATE_RECENT_FILING_DAYS).
+
+    Every one of a CIK's active gaps must be escalated: one fresh gap alongside an
+    old one still means a filing we might yet ingest, which is worth pausing for.
+    """
+    by_cik: dict[str, list[SecFilingGap]] = defaultdict(list)
+    for gap in gaps:
+        by_cik[gap.cik].append(gap)
+    escalated = {
+        cik
+        for cik, items in by_cik.items()
+        if all(gap.escalated_at is not None for gap in items)
+    }
+    if not escalated:
+        return set()
+    cutoff = (
+        datetime.now(timezone.utc) - timedelta(days=GAP_GATE_RECENT_FILING_DAYS)
+    ).date()
+    rows = await db.execute(
+        select(FundamentalSnapshot.cik)
+        .where(
+            FundamentalSnapshot.cik.in_(escalated),
+            FundamentalSnapshot.form.in_(_SEC_FORMS),
+            FundamentalSnapshot.filed_date >= cutoff,
+        )
+        .distinct()
+    )
+    return set(rows.scalars())
+
+
 async def _latest_validation(db: AsyncSession) -> dict:
     payload = (
         await db.execute(
@@ -80,8 +132,12 @@ async def blocked_reasons_by_cik(
     if ciks is not None and not ciks:
         return {}
 
+    gaps = await active_gaps(db, ciks)
+    # Escalated gaps on issuers that still have recent fundamentals no longer
+    # pause setups, on either path below — the summary mirrors the same filings.
+    exempt = await _gap_exempt_ciks(db, gaps)
     reasons = {
-        gap.cik: "sec_filing_gap" for gap in await active_gaps(db, ciks)
+        gap.cik: "sec_filing_gap" for gap in gaps if gap.cik not in exempt
     }
     summary = await _latest_validation(db)
 
@@ -92,11 +148,11 @@ async def blocked_reasons_by_cik(
     # stay capped for audit readability. Detailed entries supply the reason.
     for cik in summary.get("setup_blocked_ciks") or []:
         normalized = str(cik) if cik else ""
-        if normalized and wanted(normalized):
+        if normalized and wanted(normalized) and normalized not in exempt:
             reasons.setdefault(normalized, "sec_filing_gap")
     for item in summary.get("missing_xbrl") or []:
         normalized = str(item.get("cik") or "")
-        if normalized and wanted(normalized):
+        if normalized and wanted(normalized) and normalized not in exempt:
             reasons.setdefault(normalized, "sec_filing_gap")
     for cik in summary.get("no_xbrl_ciks") or []:
         normalized = str(cik) if cik else ""

app/services/sec_fundamentals_importer.py

@@ -36,7 +36,11 @@ Guardrails (design + reviews):
   excluded from actionable setups until its filing is recovered.
 - ``promote`` inserts snapshots ``ON CONFLICT (accession) DO NOTHING`` (immutable),
   reports differing existing accessions, and applies ticker updates in the same
-  transaction.
+  transaction. A difference in ``cik`` **alone** is reported separately as an
+  ``accession_cik_collision``: every fact matched, so two tracked CIKs are
+  claiming one filing and the fix is the universe, not the parser. It never
+  self-heals on its own — the losing CIK stores no row, so it is backfilled and
+  re-reported every run until its ticker is re-pointed or retired.
 - ``reparse=True`` is the one exception to immutability, and it is deliberate:
   it restages every accession with the current parser and **rewrites** the rows
   that now reconstruct differently. Immutability protects SEC's record (one row
@@ -67,7 +71,7 @@ from app.services import sec_universe
 from app.services.data_import import STATUS_PROMOTED, ValidationResult
 from app.services.sec_client import SecClient, SecError, cik10
 from app.services.sec_facts_parser import FilingMeta, SnapshotRow
-from app.services.sec_universe import ResolvedUniverse
+from app.services.sec_universe import CIK_OVERRIDES_KEY, ResolvedUniverse
 
 logger = logging.getLogger(__name__)
 
@@ -281,7 +285,18 @@ class SecFundamentalsImporter:
                 if old is not None:
                     fields = _diff_fields(row, old)
                     if fields:
-                        staged.discrepancies.append({"accession": row.accession, "fields": fields})
+                        # Carry both CIKs. promote() reads a bare ["cik"] as an
+                        # attribution collision rather than a changed
+                        # reconstruction, which holds only because _COMPARE_COLS
+                        # spans every stored fact: a fact column added to the
+                        # model but not to _SNAPSHOT_COLS would go uncompared and
+                        # let a real difference through as a collision.
+                        staged.discrepancies.append({
+                            "accession": row.accession,
+                            "fields": fields,
+                            "cik": row.cik,
+                            "stored_cik": old.cik,
+                        })
         return staged
 
     async def _stage_issuer(
@@ -555,8 +570,15 @@ class SecFundamentalsImporter:
         inserted = 0
         updated = 0
         # Only accessions whose reconstruction actually changed are rewritten;
-        # an unchanged stored row is left completely alone.
+        # an unchanged stored row is left completely alone. A cik-only difference
-        changed = {d["accession"] for d in staged.discrepancies} if self.reparse else set()
+        # is excluded on purpose: the facts are identical there, so rewriting
+        # would re-stamp the filing onto the colliding co-registrant — taking it
+        # from the issuer that actually filed it, which no parser fix asks for.
+        changed = (
+            {d["accession"] for d in staged.discrepancies if d["fields"] != ["cik"]}
+            if self.reparse
+            else set()
+        )
         for row in staged.rows:
             if row.accession in staged.existing_accessions:
                 if row.accession in changed:
@@ -641,10 +663,51 @@ class SecFundamentalsImporter:
             if gap["accession"] not in existing_gap_accessions
         ]
 
+        # Two tracked issuers claiming one filing is not a reconstruction change:
+        # every fact matched and only the CIK stamp differs, so re-parsing or
+        # reparsing fixes nothing — the universe resolution does. It is reported
+        # separately because it also does not self-heal: the loser of the
+        # collision never stores a row, so `_ciks_with_snapshots` never sees it,
+        # and it is full-history backfilled (and re-reported) on every run until
+        # a human re-points or retires the ticker. Observed 2026-08 for EQR,
+        # which SEC's own company_tickers.json maps to ERP Operating LP, the
+        # non-traded co-registrant of the issuer now trading as VMRK.
+        collisions = [d for d in staged.discrepancies if d["fields"] == ["cik"]]
+        if collisions:
+            named = ", ".join(
+                f"{d['accession']} (stored {d['stored_cik']}, parsed {d['cik']})"
+                for d in collisions[:10]
+            )
+            db.add(SystemEvent(
+                severity="warning",
+                source="sec_facts",
+                code="accession_cik_collision",
+                message=(
+                    f"{len(collisions)} filing(s) are claimed by two tracked CIKs — "
+                    "the reconstruction is identical, only the attribution differs, "
+                    "so one of the two is a co-registrant the universe should not "
+                    f"track. Re-point or retire the ticker (see {CIK_OVERRIDES_KEY}); "
+                    f"this repeats every run until then: {named}"
+                )[:4000],
+                dedup_key=f"sec_facts:accession_cik_collision:{run_id}",
+                created_at=_now(),
+            ))
+
         # Warn (in-transaction, so it commits atomically with the promotion) when
         # any existing accession reconstructed differently — kept immutable.
-        if staged.discrepancies:
+        reconstruction_diffs = [
-            accns = ", ".join(d["accession"] for d in staged.discrepancies[:10])
+            d for d in staged.discrepancies if d["fields"] != ["cik"]
+        ]
+        if reconstruction_diffs:
+            # Name the columns, not just the accession: "differs in revenue"
+            # (our numbers moved) and "differs in period_start" (the filing was
+            # re-placed in the calendar) need different responses, and the alert
+            # is where that call gets made. The fields are already computed for
+            # validation_json — they were simply dropped from the message.
+            accns = ", ".join(
+                f"{d['accession']} ({', '.join(d['fields'])})"
+                for d in reconstruction_diffs[:10]
+            )
             disposition = (
                 f"REWRITTEN by reparse run {run_id}" if self.reparse else "kept immutable"
             )
@@ -653,7 +716,7 @@ class SecFundamentalsImporter:
                 source="sec_facts",
                 code="snapshot_reparse" if self.reparse else "snapshot_discrepancy",
                 message=(
-                    f"{len(staged.discrepancies)} stored accession(s) reconstructed "
+                    f"{len(reconstruction_diffs)} stored accession(s) reconstructed "
                     f"differently; {disposition}: {accns}"
                 )[:4000],
                 dedup_key=f"sec_facts:discrepancy:{run_id}",

docs/fundamentals-deployment.md

@@ -24,6 +24,17 @@ entries: the application scheduler owns both jobs.
   tickers are excluded from actionable setups until a snapshot is recovered or
   a later valid 10-K/10-Q supersedes the gap. Migration `028` materializes older
   promoted gaps into this queue once, so setup reads never scan import history.
+- A gap that survives 14 days raises `filing_gap_aged` and, from that point,
+  stops pausing setups **if** the issuer's own newest stored 10-K/10-Q is less
+  than `GAP_GATE_RECENT_FILING_DAYS` (180) old. This is the hand-off from pause
+  to alert, and it exists because the pause would otherwise be open-ended:
+  SEC's per-company Company-Facts files can go stale indefinitely (2026-08: 43
+  large caps whose Q2 10-Qs the `frames` API carried but whose
+  `companyfacts/CIK*.json` never received), and the supersede rule needs a
+  *successfully ingested* later filing, so a stale file swallows the next
+  quarter too. Retrying is unaffected — the gap stays queued and a recovered
+  filing still resolves it normally. An issuer with no filing that recent has no
+  usable fundamentals at all and stays paused.
 
 The systemd service uses one application worker. The import framework also holds
 a PostgreSQL advisory lock per source, so an overlapping manual/scheduled run is

reports/sec-companyfacts-stale-20260821-findings.md

@@ -0,0 +1,150 @@
+# SEC fundamentals alerts, 2026-08-21
+
+Two `sec_facts` warnings, investigated against live SEC data. Both originate in SEC's
+own published data — a stale per-company Company-Facts file (1) and a stale
+ticker→CIK mapping (2) — and neither is a parser defect: no stored fundamental value
+is wrong. Every SEC-side probe below reproduces offline from public endpoints; the
+four database facts used are quoted where they appear.
+
+## 1. `filing_gap_aged` — 43 gaps, all `not_in_companyfacts`
+
+**Root cause: SEC's per-company Company-Facts files are stale for these issuers,
+while the same filings are present in SEC's own `frames` aggregation.**
+
+All ten named filings are real 10-Qs filed 2026-07-28/29, present in the issuer's
+`submissions` with `isXBRL=1`, with complete R-files and XBRL in the EDGAR archive
+— and absent from `companyfacts/CIK*.json`:
+
+| CIK | issuer | accession | filed | in `companyfacts` | newest fact in file |
+|---|---|---|---|---|---|
+| 0000001800 | Abbott | 0001628280-26-050134 | 2026-07-28 | no | 2026-04-29 |
+| 0000021344 | Coca-Cola | 0001628280-26-050503 | 2026-07-29 | no | 2026-04-30 |
+| 0000024741 | Corning | 0000024741-26-000255 | 2026-07-29 | no | 2026-05-01 |
+| 0000029989 | Omnicom | 0000029989-26-000019 | 2026-07-29 | no | 2026-04-29 |
+| 0000037996 | Ford | 0000037996-26-000156 | 2026-07-29 | no | 2026-04-30 |
+| 0000040533 | General Dynamics | 0000040533-26-000032 | 2026-07-29 | no | 2026-07-01 |
+| 0000048898 | Hubbell | 0001628280-26-050405 | 2026-07-29 | no | 2026-06-04 |
+| 0000049071 | Humana | 0000049071-26-000050 | 2026-07-29 | no | 2026-04-29 |
+| 0000049196 | Huntington Bancshares | 0000049196-26-000066 | 2026-07-28 | no | 2026-04-30 |
+| 0000062996 | Masco | 0000062996-26-000027 | 2026-07-29 | no | 2026-04-22 |
+
+Ruled out, with evidence:
+
+- **Not a global SEC outage.** Company Facts is current for other issuers filing the
+  same days — MSFT `0001193125-26-323660` @2026-07-29, AAPL @2026-07-31, P&G
+  @2026-08-04, Chevron @2026-08-06, JPMorgan @2026-08-20.
+- **Not a CDN/cache artifact.** A cache-busted request with `Cache-Control: no-cache`
+  returns the identical stale 3.39 MB payload; the response carries no cache headers.
+- **Not our filter.** The scan covers every taxonomy/concept/unit in the payload.
+- **Not a metadata discriminator.** Gap and non-gap filings are identical on
+  `isXBRL`, `isInlineXBRL`, `reportDate`, `primaryDocDescription`.
+- **SEC does have the facts.** `frames/us-gaap/Assets/USD/CY2026Q2I.json` lists
+  Abbott at exactly the missing accession `0001628280-26-050134`, and Coca-Cola and
+  Ford at theirs. The per-company endpoints are the degraded ones:
+  `companyconcept/CIK0000001800/us-gaap/Assets.json` returns `"units":{"USD":{}}`.
+
+**Consequence, and why the gate changed.** Retrying `companyfacts` cannot recover
+these — Abbott's file has been stale since April. And because `active_gaps`
+supersedes a gap only on a *successfully ingested later* filing, a stale file also
+swallows Q3: the pause was open-ended, not seasonal, on 43 large caps.
+
+**Fix** (`app/services/fundamentals_quality_service.py`): once `filing_gap_aged` has
+escalated a gap (`escalated_at`), it stops pausing setups **if** the issuer's own
+newest stored 10-K/10-Q is under `GAP_GATE_RECENT_FILING_DAYS` (180) old. Pause hands
+off to the alert; an issuer with nothing that recent stays paused. `active_gaps` is
+deliberately untouched, so `_retry_backlog` keeps retrying and a recovered filing
+still resolves normally. The bound is applied to the queue path *and* the
+`validation_json` summary path, which mirrors the same filings — bounding only one
+leaves the behaviour unchanged in production.
+
+**This is a bounded reprieve, not a removal — know the two ways it ends.** Abbott's
+newest ingested filing is `0001628280-26-028357`, filed 2026-04-29, so its recency
+window closes around **2026-10-26**; most of the 43 sit on late-April filings and
+turn back to paused within days of each other. That crossing is **silent**: the
+importer escalates only gaps with `escalated_at IS NULL`, so `filing_gap_aged` does
+not re-fire for a gap it has already reported. Separately, a Q3 10-Q that also fails
+to ingest creates a *new* un-escalated gap on the same CIK, which re-pauses it at
+once (that one does raise its own `filing_gap_aged` 14 days later). Whether the
+silent re-block deserves a re-escalation signal is an open call, deliberately not
+made here — "one actionable escalation rather than a daily warning" is the existing
+design intent.
+
+**Not done, with reasons.** A `frames`-backed recovery source was considered and
+rejected: frames are calendar-aligned with a tolerance (off-fiscal filers drop out)
+and carry one fact per issuer per period, so amendment/restatement semantics differ
+from Company Facts — lossy as a snapshot source, not merely expensive. Parsing the
+filing's own inline-XBRL instance is the authoritative alternative but is a new
+subsystem (contexts, dimensions, unit refs) duplicating the parser's fact model.
+
+## 2. `snapshot_discrepancy` — 0000906107-15-000012 / -000016
+
+**Root cause: two tracked tickers claim the same filing, because SEC's
+`company_tickers.json` still points the old symbol at a non-traded co-registrant.
+No stored value is wrong and no reparse is warranted.**
+
+CIK 0000906107 is **Vivmark Residential** (VMRK, formerly Equity Residential). Both
+alerted accessions are **combined EQR + ERP Operating LP 10-Qs** — one accession, two
+registrants (0000906107 and 0000931182) — the pattern behind the existing
+co-registrant recovery path.
+
+The stored rows are **byte-identical** to what the current parser reconstructs from
+EQR's own Company Facts — every column, verified: `cik` (`0000906107`), `form`,
+`filed_date`, `accepted_at`, both period dates, `fiscal_year`/`fiscal_period`,
+`revenue`, `net_income`, `operating_income`, `diluted_eps`, `cfo`, the two nulls,
+`cash_and_st_investments`, `total_debt` (340,900,000 / null),
+`shares_outstanding`, `shares_outstanding_date`, `weighted_avg_diluted_shares`. Both
+carry `import_run_id = 6`, and CIK 0000906107 holds all 69 of its filings across runs
+6–30, so the issuer's own history is complete.
+
+Run 63 (2026-08-19) recorded
+`fields: ["cik"]` for both accessions, and the universe explains it:
+
+```
+tickers:  VMRK -> 0000906107      (Vivmark Residential, ex-Equity Residential)
+          EQR  -> 0000931182      (ERP Operating Ltd Partnership)
+```
+
+SEC's own `company_tickers.json` carries `{"cik_str": 931182, "ticker": "EQR",
+"title": "ERP OPERATING LTD PARTNERSHIP"}` — after the rename, the old symbol stayed
+attached to the **non-traded operating partnership**, the co-registrant on those
+combined 10-Qs. `resolve_ciks` reads `active_only` tickers and follows SEC, so
+0000931182 is tracked. Its Company Facts holds 7 accessions, exactly 2 of them
+EQR-prefixed, so its backfill reconstructs exactly those two rows, stamps them
+`cik=0000931182`, and collides with the rows already stored under 0000906107 —
+identical in every fact, differing only in attribution.
+
+It cannot self-heal. The collision loser never stores a row (the insert is skipped as
+immutable), so `_ciks_with_snapshots` never sees 0000931182, and it is full-history
+backfilled — refetching every submissions shard and its companyfacts — **on every
+run**, re-raising the warning each time. `fundamental_snapshots` for 0000906107 holds
+all 69 filings across runs 6–30, so the issuer's own history is complete and correct.
+
+### Fixes
+
+**Code** (`sec_fundamentals_importer.py`): a `cik`-only difference is no longer
+reported as a reconstruction discrepancy. It raises `accession_cik_collision`, naming
+both CIKs and pointing at `sec_cik_overrides`, because the fix is the universe, not
+the parser. The reparse path also excludes these from its rewrite set — rewriting a
+cik-only difference would re-stamp the filing onto the co-registrant and take it from
+the issuer that filed it. (A reparse run while both CIKs are tracked fails validation
+on `duplicate accession in staged snapshots` instead, which is a safe stop.)
+
+**Data — needs an operator, and the alert repeats daily until then.** `EQR` is a stale
+symbol: the security now trades as `VMRK`, which is already tracked at the correct
+CIK. Retiring the `EQR` ticker ends the loop. A `sec_cik_overrides` pin of
+`EQR -> 906107` would silence the collision but leave two tickers on one security,
+double-counting the issuer in scans — retirement is the right action.
+
+**Not fixed, deliberately:** the permanent-backfill loop itself. A tracked CIK whose
+only parseable filings belong to another CIK is re-backfilled every run; ending that
+in code means teaching `_ciks_with_snapshots` about foreign-owned accessions, which is
+more state for a condition that is now loudly and specifically reported.
+
+### Separate observation: `total_debt` on this issuer looks wrong
+
+Independent of the alert, and unchanged by any fix here: the parser reconstructs
+`total_debt = 340,900,000` for EQR's 2015 Q1 and `null` for Q2, while the REIT carried
+roughly $10bn of debt. `_compose_debt` returns the short-term component alone when
+every `_LONG_TERM_DEBT_AGG` concept **and** the `LongTermDebtNoncurrent`/`Current`
+pair miss — which is what happened here, and Q2 matched neither. Worth checking
+against a current REIT filer before trusting `total_debt` for that sector.

tests/unit/test_fundamentals_quality_service.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import json
-from datetime import date, datetime, timezone
+from datetime import date, datetime, timedelta, timezone
 
 from app.models.data_import_run import DataImportRun
 from app.models.fundamental_snapshot import FundamentalSnapshot
@@ -139,3 +139,121 @@ async def test_ticker_quality_explains_no_xbrl_block(db_session):
     assert await fundamentals_quality_service.ticker_is_eligible(
         db_session, ticker.id
     ) is False
+
+
+def _escalated_gap(cik: str, *, escalated: bool = True) -> SecFilingGap:
+    first_seen = datetime.now(timezone.utc) - timedelta(days=24)
+    return SecFilingGap(
+        cik=cik,
+        accession=f"{cik}-STALE-Q",
+        form="10-Q",
+        index_date=(first_seen.date()),
+        reason="not_in_companyfacts",
+        first_seen_at=first_seen,
+        last_attempted_at=datetime.now(timezone.utc),
+        escalated_at=(
+            datetime.now(timezone.utc) - timedelta(days=10) if escalated else None
+        ),
+    )
+
+
+def _prior_quarter(cik: str, *, age_days: int) -> FundamentalSnapshot:
+    """The issuer's last successfully ingested filing, older than the gap so it
+    cannot supersede it — exactly the production shape of a stale companyfacts
+    file: Q1 stored, Q2 missing."""
+    filed = date.today() - timedelta(days=age_days)
+    return FundamentalSnapshot(
+        cik=cik,
+        accession=f"{cik}-PRIOR-Q",
+        form="10-Q",
+        filed_date=filed,
+        accepted_at=datetime.now(timezone.utc) - timedelta(days=age_days),
+        period_end=filed,
+        fiscal_year=filed.year,
+        fiscal_period="Q1",
+    )
+
+
+async def test_escalated_gap_stops_blocking_when_fundamentals_are_recent(db_session):
+    ticker = Ticker(symbol="STALEFACTS", cik="0000000046")
+    db_session.add(ticker)
+    db_session.add(_escalated_gap(ticker.cik))
+    await db_session.flush()
+    assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == {
+        ticker.id
+    }
+
+    # The alert has run and the issuer still has last quarter to score on.
+    db_session.add(_prior_quarter(ticker.cik, age_days=120))
+    await db_session.flush()
+
+    assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set()
+    # ...but the filing is still queued, so the importer keeps retrying it.
+    assert len(await fundamentals_quality_service.active_gaps(db_session)) == 1
+
+
+async def test_escalated_gap_keeps_blocking_when_fundamentals_are_stale(db_session):
+    ticker = Ticker(symbol="NOTHINGFRESH", cik="0000000047")
+    db_session.add_all([
+        ticker,
+        _escalated_gap(ticker.cik),
+        _prior_quarter(
+            ticker.cik,
+            age_days=fundamentals_quality_service.GAP_GATE_RECENT_FILING_DAYS + 30,
+        ),
+    ])
+    await db_session.flush()
+
+    assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == {
+        ticker.id
+    }
+
+
+async def test_unescalated_gap_still_blocks_alongside_an_escalated_one(db_session):
+    ticker = Ticker(symbol="TWOGAPS", cik="0000000048")
+    fresh = datetime.now(timezone.utc)
+    db_session.add_all([
+        ticker,
+        _escalated_gap(ticker.cik),
+        SecFilingGap(
+            cik=ticker.cik,
+            accession="TWOGAPS-FRESH-Q",
+            form="10-Q",
+            index_date=date.today(),
+            reason="not_in_companyfacts",
+            first_seen_at=fresh,
+            last_attempted_at=fresh,
+        ),
+        _prior_quarter(ticker.cik, age_days=120),
+    ])
+    await db_session.flush()
+
+    assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == {
+        ticker.id
+    }
+
+
+async def test_summary_path_does_not_reblock_an_exempt_cik(db_session):
+    """The run summary mirrors the same filings as the queue — it must honour the
+    same hand-off, or the bound is inert in production."""
+    ticker = Ticker(symbol="MIRRORED", cik="0000000049")
+    db_session.add(ticker)
+    db_session.add(_escalated_gap(ticker.cik))
+    db_session.add(_prior_quarter(ticker.cik, age_days=120))
+    await db_session.flush()
+    db_session.add(
+        DataImportRun(
+            source="sec_facts",
+            status="promoted",
+            validation_json=json.dumps({
+                "setup_blocked_ciks": [ticker.cik],
+                "missing_xbrl": [
+                    {"cik": ticker.cik, "accession": f"{ticker.cik}-STALE-Q"}
+                ],
+            }),
+            started_at=datetime.now(timezone.utc),
+        )
+    )
+    await db_session.flush()
+
+    assert await fundamentals_quality_service.blocked_ticker_ids(db_session) == set()

tests/unit/test_sec_fundamentals_importer.py

@@ -950,6 +950,9 @@ async def test_discrepancy_in_shares_is_detected_and_reported(engine):
         assert k.shares_outstanding == 999.0 and k.import_run_id == 1  # immutable — not overwritten
         events = (await s.execute(select(SystemEvent).where(SystemEvent.code == "snapshot_discrepancy"))).scalars().all()
     assert len(events) == 1 and events[0].severity == "warning"
+    # The alert has to say WHICH column moved: a differing cik is a co-registrant
+    # attribution, a differing revenue is our numbers changing.
+    assert "K (shares_outstanding, shares_outstanding_date)" in events[0].message
 
 
 # --- reparse: rewriting rows a fixed parser reconstructs differently --------
@@ -1283,3 +1286,105 @@ async def test_ceiling_promotes_queues_and_alerts_end_to_end(engine, monkeypatch
     assert len(events) == 1
     assert events[0].severity == "warning"
     assert "7 days" in events[0].message
+
+
+# --- attribution collisions: two tracked CIKs claiming one filing ----------
+
+# A REIT and its operating partnership co-file one 10-K, and SEC's
+# company_tickers.json points the old symbol at the partnership (EQR ->
+# ERP Operating LP) while the issuer itself trades under a new one (VMRK).
+_COMBINED = [_filing("COMBINED-K", "10-K", "2025-12-31", "2026-02-13",
+                     "2026-02-13T21:00:00.000Z")]
+_CF_COMBINED = _rev("2025-01-01", "2025-12-31", 2900000, 2025, "FY", "COMBINED-K")
+_SH_COMBINED = _shares("2026-02-01", 380000, "COMBINED-K", 2025, "FY")
+
+
+def _reit_submissions(cik, tickers):
+    return {"cik": cik, "sic": "6798", "sic_description": "REIT",
+            "fiscal_year_end": "1231", "tickers": tickers, "filings": _COMBINED}
+
+
+def _reit_client(tickers):
+    return FakeSecClient(
+        tickers=tickers,
+        companyfacts={
+            cik: _companyfacts([_CF_COMBINED], [_SH_COMBINED], cik=cik)
+            for cik in tickers.values()
+        },
+        submissions={
+            cik: _reit_submissions(cik, [sym]) for sym, cik in tickers.items()
+        },
+        latest_index=date(2026, 3, 1),
+    )
+
+
+async def test_cik_collision_is_reported_as_attribution_not_discrepancy(engine):
+    """Only `cik` differs, so nothing was re-parsed differently — the universe
+    resolves a co-registrant it should not track, and the alert must say that."""
+    factory = _factory(engine)
+    await _seed(factory, ["VMRK"])
+    run = await run_import(
+        _importer(_reit_client({"VMRK": 906107}), today=date(2026, 3, 2)), engine=engine
+    )
+    assert run.status == STATUS_PROMOTED
+
+    # The stale symbol is added, resolving to the partnership's CIK.
+    await _seed(factory, ["EQR"])
+    run = await run_import(
+        _importer(_reit_client({"VMRK": 906107, "EQR": 931182}), today=date(2026, 3, 2)),
+        engine=engine,
+    )
+    assert run.status == STATUS_PROMOTED
+    # Production's shape: a run-level incremental in which the untracked-until-now
+    # CIK is individually backfilled (run 63 recorded exactly this).
+    assert '"backfill": false' in (run.validation_json or "")
+
+    async with factory() as s:
+        rows = (await s.execute(select(FundamentalSnapshot))).scalars().all()
+        events = (await s.execute(select(SystemEvent))).scalars().all()
+    # The filing stays with the issuer that filed it, stored once.
+    assert [(r.accession, r.cik) for r in rows] == [("COMBINED-K", "0000906107")]
+
+    codes = {e.code for e in events}
+    assert "accession_cik_collision" in codes
+    assert "snapshot_discrepancy" not in codes  # not a reconstruction change
+    collision = next(e for e in events if e.code == "accession_cik_collision")
+    assert "stored 0000906107, parsed 0000931182" in collision.message
+    assert "sec_cik_overrides" in collision.message  # names the actual fix
+
+
+async def test_reparse_never_restamps_a_collision_onto_the_co_registrant(engine):
+    """A reparse rewrites rows a fixed parser reconstructs differently. A cik-only
+    difference is not that: rewriting would hand the filing to the co-registrant."""
+    from app.services.sec_facts_parser import SnapshotRow
+
+    factory = _factory(engine)
+    await _seed(factory, ["VMRK"])
+    assert (await run_import(
+        _importer(_reit_client({"VMRK": 906107}), today=date(2026, 3, 2)), engine=engine
+    )).status == STATUS_PROMOTED
+
+    importer = _importer(_reit_client({"VMRK": 906107}), today=date(2026, 3, 2))
+    importer.reparse = True
+    staged = StagedFundamentals(
+        resolved=ResolvedUniverse(),
+        rows=[SnapshotRow(
+            cik="0000931182", accession="COMBINED-K", form="10-K",
+            filed_date=date(2026, 2, 13),
+            accepted_at=datetime(2026, 2, 13, 21, tzinfo=timezone.utc),
+            period_end=date(2025, 12, 31), fiscal_year=2025, fiscal_period="FY",
+        )],
+        existing_accessions={"COMBINED-K"},
+        discrepancies=[{
+            "accession": "COMBINED-K", "fields": ["cik"],
+            "cik": "0000931182", "stored_cik": "0000906107",
+        }],
+    )
+    async with _factory(engine)() as db:
+        counts = await importer.promote(db, staged, run_id=999)
+        await db.commit()
+
+    assert counts["updated"] == 0
+    async with factory() as s:
+        row = (await s.execute(select(FundamentalSnapshot))).scalar_one()
+    assert row.cik == "0000906107"  # still the issuer that filed it