Recover SEC facts misfiled under a co-registrant CIK

The fundamentals import had been failing for three days on two tracked
filings the index listed but Company Facts appeared not to have. They were
not lagging: SEC filed the XBRL of NEE's and DOW's 2026-07-24 combined
parent/subsidiary 10-Qs under the co-registrant's CIK (Florida Power &
Light, Dow Chemical), so the ticker-carrying filer's own facts file never
receives that accession. This does not self-correct - an NEE filing
misattributed the same way in 2014 is still misfiled.

Because source_max_date advances only on a promoted run, the failure was
self-perpetuating: every later run re-walked the same index day and re-hit
the same two filings.

- Recover from the co-registrant file. The daily index lists every
  co-registrant of an accession, which is the only pointer to where the
  facts actually landed. Rows are re-stamped to the real filer, since
  parse_snapshots stamps the CIK of the payload it read.
- Guard the recovery with a share-count continuity check against the
  issuer's own history, so a subsidiary's standalone facts can never be
  stored as the parent's. No history, no recovery.
- Bound the blocking: a filing still unresolvable after
  MISSING_XBRL_RETRY_DAYS promotes with a named unresolved_filing warning
  instead of wedging every later import.
- Name the offending filings in the alert and record them in
  validation_json, separating not_in_companyfacts from not_in_submissions.
  The gate previously reported a count and discarded the accessions.

Verified against live SEC data: both filings recover with the correct CIK
and the guard rejects a mismatched reference.

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

app/services/sec_fundamentals_importer.py

@@ -18,6 +18,23 @@ Guardrails (design + reviews):
   a tracked XBRL index accession missing from Company Facts fails the run (the two
   are separate SEC products that can lag) so we retry rather than record a
   null/partial snapshot. Non-XBRL amendments are skipped with a recorded reason.
+  A failure here blocks every later run (``source_max_date`` only advances on a
+  promoted run), so it names the offending filings in the alert and separates the
+  causes — ``not_in_companyfacts`` (facts lag) vs ``not_in_submissions`` (the
+  index row is absent from the issuer's own filing list, which no retry fixes).
+- **Co-registrant recovery**, because "missing from Company Facts" is often not
+  missing at all: SEC files some combined parent/subsidiary filings' XBRL under
+  the co-registrant's CIK, so the ticker-carrying parent's own file never gets
+  that accession. The daily index lists every co-registrant of an accession, so
+  the facts are found there and re-stamped to the real filer — guarded by a
+  share-count continuity check so a subsidiary's standalone numbers can never be
+  stored as the parent's. Confirmed 2026-07-27 (NEE via FPL, DOW via Dow Chemical)
+  and it is not transient: an NEE filing misattributed in 2014 is still misfiled.
+- **Bounded blocking.** Anything still unresolvable after ``MISSING_XBRL_RETRY_DAYS``
+  stops failing the run and is promoted around, with a named ``unresolved_filing``
+  warning. One filing SEC misfiled must not wedge every later import; the index
+  only moves forward, so a tolerated accession returns only via a reparse, and
+  only once SEC has re-filed it under the filer's own CIK.
 - ``promote`` inserts snapshots ``ON CONFLICT (accession) DO NOTHING`` (immutable),
   reports differing existing accessions, and applies ticker updates in the same
   transaction.
@@ -32,8 +49,8 @@ Guardrails (design + reviews):
 from __future__ import annotations
 
 import logging
-from collections import defaultdict
-from dataclasses import dataclass, field
+from collections import Counter, defaultdict
+from dataclasses import dataclass, field, replace
 from datetime import date, datetime, timedelta, timezone
 from typing import Any, Callable
 
@@ -57,6 +74,16 @@ _XBRL_FORMS = {"10-K", "10-Q", "10-K/A", "10-Q/A"}
 # On the one-time backfill, require this fraction of tracked issuers to yield at
 # least one snapshot (guards a broken fetch/parse from promoting a hollow table).
 MIN_BACKFILL_COVERAGE = 0.5
+# How long an index accession may stay unresolvable before the run stops failing
+# on it. Genuine index↔facts lag clears within a day (a weekend stretches it to
+# 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
+# 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
+# wide enough to let a subsidiary shell's token float through (see _shares_continuous).
+RECOVERY_SHARES_MIN = 0.5
+RECOVERY_SHARES_MAX = 2.0
 
 _SNAPSHOT_COLS = (
     "cik", "accession", "form", "filed_date", "accepted_at", "period_start",
@@ -78,7 +105,11 @@ class StagedFundamentals:
     skipped_filings: list[dict[str, str]] = field(default_factory=list)
     field_issues: list[dict[str, str]] = field(default_factory=list)
     skipped_non_xbrl: list[dict[str, str]] = field(default_factory=list)
-    missing_xbrl: list[dict[str, str]] = field(default_factory=list)
+    # Index rows we could not resolve to Company Facts, with a per-row reason
+    # (not_in_companyfacts | not_in_submissions | ...) — see _missing().
+    missing_xbrl: list[dict[str, Any]] = field(default_factory=list)
+    # Accessions parsed out of a co-registrant's Company Facts file.
+    recovered: list[dict[str, Any]] = field(default_factory=list)
     invalid_payloads: list[dict[str, str]] = field(default_factory=list)
     existing_accessions: set[str] = field(default_factory=set)
     # Tracked issuers whose registrant has NO XBRL 10-K/10-Q at all: they can
@@ -117,6 +148,9 @@ class SecFundamentalsImporter:
         # cached by detect_revision, consumed by stage:
         self._resolved: ResolvedUniverse | None = None
         self._index_rows: list[dict[str, Any]] = []
+        # accession -> the OTHER CIKs the daily index lists it under (co-registrants
+        # of a combined filing). Only populated for accessions a tracked issuer filed.
+        self._coregistrants: dict[str, list[int]] = {}
         self._latest_index_date: date | None = None
         self._backfill = False
 
@@ -132,6 +166,7 @@ class SecFundamentalsImporter:
             # Reparse needs every accession restaged, not just those filed since
             # the last run — the facts a fixed parser now accepts were never
             # stored, so a reparse cannot be served from the database.
+            self._coregistrants = {}
             if last_processed is None or self.reparse:
                 self._backfill = True
                 self._index_rows = []
@@ -151,10 +186,12 @@ class SecFundamentalsImporter:
         staged = StagedFundamentals(resolved=resolved, backfill=self._backfill)
 
         cik_to_tids = resolved.cik_to_ticker_ids
-        filed_by_cik: dict[int, list[str]] = defaultdict(list)
+        # Whole index rows (not bare accessions): form + index date are what make
+        # an unresolvable filing diagnosable without re-walking the index by hand.
+        filed_by_cik: dict[int, list[dict[str, Any]]] = defaultdict(list)
         for r in self._index_rows:
             if r["cik"] in cik_to_tids:
-                filed_by_cik[r["cik"]].append(r["accession"])
+                filed_by_cik[r["cik"]].append(r)
 
         existing = await self._ciks_with_snapshots(db, set(cik_to_tids))
         if self._backfill:
@@ -165,10 +202,15 @@ class SecFundamentalsImporter:
             backfill_ciks = {c for c in cik_to_tids if c not in existing}
         incremental_ciks = set(filed_by_cik) - backfill_ciks
 
+        # Continuity reference for co-registrant recovery, read once up front.
+        last_shares = await self._last_shares_outstanding(db, set(cik_to_tids))
+
         async with self._client_factory() as client:
             for cik in sorted(backfill_ciks | incremental_ciks):
                 is_backfill = cik in backfill_ciks
-                await self._stage_issuer(client, cik, is_backfill, filed_by_cik, staged)
+                await self._stage_issuer(
+                    client, cik, is_backfill, filed_by_cik, staged, last_shares
+                )
 
         # Read-only discrepancy detection: an accession we reconstructed that is
         # already stored, differing in ANY source field (immutable → report in
@@ -185,7 +227,9 @@ class SecFundamentalsImporter:
                         staged.discrepancies.append({"accession": row.accession, "fields": fields})
         return staged
 
-    async def _stage_issuer(self, client, cik, is_backfill, filed_by_cik, staged) -> None:
+    async def _stage_issuer(
+        self, client, cik, is_backfill, filed_by_cik, staged, last_shares
+    ) -> None:
         cf = await client.companyfacts(cik)
         bad = _companyfacts_structure_error(cf)
         if bad is not None:
@@ -201,31 +245,61 @@ class SecFundamentalsImporter:
                 {"cik": cik10(cik), "name": sub.get("name"), "tickers": sub.get("tickers")}
             )
 
+        fiscal_year_end = sub.get("fiscal_year_end")
+        recovered_rows: list[SnapshotRow] = []
         if is_backfill:
             accns = set(xbrl_meta)
         else:
             present = parser.companyfacts_accessions(cf)
             accns = set()
-            for accn in filed_by_cik.get(cik, []):
+            for index_row in filed_by_cik.get(cik, []):
+                accn = index_row["accession"]
                 if accn in nonxbrl:
                     staged.skipped_non_xbrl.append({"cik": cik10(cik), "accession": accn})
-                elif accn in xbrl_meta and accn in present:
+                elif accn not in xbrl_meta:
+                    # The daily index lists it but the issuer's own filing list does
+                    # not (submissions lagging the index, no usable period metadata,
+                    # or a co-registrant filing). NOT a Company-Facts lag — separate
+                    # cause, separate fix, so it gets its own reason.
+                    staged.missing_xbrl.append(
+                        _missing(cik, index_row, "not_in_submissions", self.today)
+                    )
+                elif accn in present:
                     accns.add(accn)
                 else:
-                    # XBRL (or unknown) filing not yet in Company Facts → the products
-                    # have lagged; fail+retry rather than record nothing for it.
-                    staged.missing_xbrl.append({"cik": cik10(cik), "accession": accn})
+                    # Filed, XBRL, but absent from this issuer's Company Facts. Try
+                    # the co-registrant file before treating it as missing data.
+                    row, source_cik = await self._recover_from_coregistrant(
+                        client, cik, accn, xbrl_meta, fiscal_year_end,
+                        last_shares.get(cik10(cik)), staged,
+                    )
+                    if row is not None:
+                        recovered_rows.append(row)
+                        staged.recovered.append({
+                            "cik": cik10(cik),
+                            "accession": accn,
+                            "source_cik": source_cik,
+                            "form": index_row.get("form"),
+                        })
+                    else:
+                        staged.missing_xbrl.append(_missing(
+                            cik, index_row,
+                            # Found, but it did not look like this issuer's own
+                            # numbers — say so; it is not the same as absent.
+                            "coregistrant_facts_rejected" if source_cik
+                            else "not_in_companyfacts",
+                            self.today,
+                        ))
 
         # fiscalYearEnd (MMDD) is what lets the parser derive period identity from
         # reportDate instead of SEC's unreliable fy/fp fields.
-        result = parser.parse_snapshots(
-            cf, xbrl_meta, accns, fiscal_year_end=sub.get("fiscal_year_end")
-        )
+        result = parser.parse_snapshots(cf, xbrl_meta, accns, fiscal_year_end=fiscal_year_end)
         staged.rows.extend(result.rows)
+        staged.rows.extend(recovered_rows)
         staged.skipped_filings.extend(result.skipped_filings)
         staged.field_issues.extend(result.field_issues)
         staged.issuers_fetched += 1
-        if result.rows:
+        if result.rows or recovered_rows:
             staged.issuers_with_rows += 1
 
         # SIC proposal for this issuer's tickers (read-only; applied in promote).
@@ -234,14 +308,65 @@ class SecFundamentalsImporter:
         for tid in staged.resolved.cik_to_ticker_ids.get(cik, []):
             staged.sic_updates.append((tid, sic, desc))
 
+    async def _recover_from_coregistrant(
+        self, client, cik: int, accn: str, xbrl_meta, fiscal_year_end, reference, staged,
+    ) -> tuple[SnapshotRow | None, str | None]:
+        """Look for ``accn``'s facts in a co-registrant's Company Facts file.
+
+        SEC sometimes files a combined parent/subsidiary filing's XBRL under the
+        co-registrant's CIK rather than the filer's — the ticker-carrying parent's
+        own file simply never gets that accession. Verified 2026-07-27 for NEE
+        (facts under Florida Power & Light) and DOW (under Dow Chemical); an NEE
+        filing misattributed the same way in **2014** is still misattributed, so
+        this does not self-correct and no amount of retrying recovers it.
+
+        Returns ``(row, source_cik)`` on success, ``(None, source_cik)`` when the
+        facts were found but rejected by the continuity guard, ``(None, None)``
+        when no co-registrant has them.
+
+        Incremental path only: the co-registrant map comes from the daily index,
+        which a backfill/reparse does not walk. A reparse therefore recovers a
+        filing only once SEC re-files it under the filer's own CIK.
+        """
+        for co in self._coregistrants.get(accn, []):
+            try:
+                cf_co = await client.companyfacts(co)
+            except SecError:
+                continue  # a co-registrant shell often has no facts file at all
+            if _companyfacts_structure_error(cf_co) is not None:
+                continue
+            result = parser.parse_snapshots(
+                cf_co, xbrl_meta, {accn}, fiscal_year_end=fiscal_year_end
+            )
+            if not result.rows:
+                continue
+            row = result.rows[0]
+            if not _shares_continuous(row.shares_outstanding, reference):
+                return None, cik10(co)
+            # A recovered row is the one most worth flagging, so its parser caveats
+            # travel with it rather than being dropped on the way out.
+            staged.field_issues.extend(result.field_issues)
+            # parse_snapshots stamps the CIK of the payload it read — re-stamp to
+            # the issuer that actually filed, or the row lands under the shell.
+            return replace(row, cik=cik10(cik)), cik10(co)
+        return None, None
+
     async def validate(self, db, staged: StagedFundamentals) -> ValidationResult:
         messages: list[str] = []
 
-        # Consistency gate — before any write.
-        if staged.missing_xbrl:
+        # Consistency gate — before any write. Only filings still inside the retry
+        # window block: a failure here stops every later run too (source_max_date
+        # advances on promotion alone), so blocking forever on a filing SEC has
+        # misfiled would cost far more than the one filing it withholds. Older
+        # ones are carried by promote() as a warning instead. The message names
+        # the filings: "which ones" has to be in the alert itself, not merely
+        # reconstructible by re-walking the index.
+        blocking = _within_retry_window(staged.missing_xbrl)
+        if blocking:
             messages.append(
-                f"{len(staged.missing_xbrl)} tracked XBRL filing(s) not yet in "
-                "Company Facts (index/facts lag) — retry"
+                f"{len(blocking)} tracked XBRL filing(s) unresolved within the "
+                f"{MISSING_XBRL_RETRY_DAYS}-day retry window "
+                f"({_reason_counts(blocking)}) — retry: {_missing_detail(blocking)}"
             )
         # Malformed companyfacts payloads must fail, not degrade to skipped rows.
         if staged.invalid_payloads:
@@ -272,7 +397,11 @@ class SecFundamentalsImporter:
             "skipped_non_xbrl": len(staged.skipped_non_xbrl),
             "no_xbrl_filings": staged.no_xbrl_filings[:50],
             "no_xbrl_filings_count": len(staged.no_xbrl_filings),
-            "missing_xbrl": len(staged.missing_xbrl),
+            "missing_xbrl": staged.missing_xbrl[:50],
+            "missing_xbrl_count": len(staged.missing_xbrl),
+            "missing_xbrl_blocking": len(blocking),
+            "recovered_from_coregistrant": staged.recovered[:50],
+            "recovered_count": len(staged.recovered),
             "invalid_payloads": staged.invalid_payloads,
             "cik_updates": len(staged.resolved.cik_updates),
             # differing existing accessions (immutable — kept, reported here)
@@ -332,6 +461,48 @@ class SecFundamentalsImporter:
                 created_at=_now(),
             ))
 
+        # Recovered rows are real data from an unexpected place — record where they
+        # came from, so a wrong recovery is auditable rather than invisible.
+        if staged.recovered:
+            named = ", ".join(
+                f"{r['accession']} <- CIK {r['source_cik']}" for r in staged.recovered[:10]
+            )
+            db.add(SystemEvent(
+                severity="warning",
+                source="sec_facts",
+                code="coregistrant_recovery",
+                message=(
+                    f"{len(staged.recovered)} filing(s) were absent from the filer's "
+                    f"own Company Facts and were parsed from a co-registrant's file "
+                    f"instead (share count checked against the issuer's history): {named}"
+                )[:4000],
+                dedup_key=f"sec_facts:coregistrant_recovery:{run_id}",
+                created_at=_now(),
+            ))
+
+        # Filings past the retry window: promoted WITHOUT them so one misfiled
+        # filing cannot wedge every later import. This is the deliberate trade —
+        # loud and named, because the index only moves forward and nothing will
+        # revisit them on its own.
+        tolerated = _past_retry_window(staged.missing_xbrl)
+        if tolerated:
+            db.add(SystemEvent(
+                severity="warning",
+                source="sec_facts",
+                code="unresolved_filing",
+                message=(
+                    f"{len(tolerated)} tracked filing(s) still unresolvable after "
+                    f"{MISSING_XBRL_RETRY_DAYS} days; promoting without them rather "
+                    f"than blocking every later import. They are NOT retried. "
+                    f"scripts/reparse_fundamentals.py recovers them ONLY once SEC "
+                    f"re-files under the filer's own CIK — it walks no index, so it "
+                    f"cannot reach facts still sitting under a co-registrant: "
+                    f"{_missing_detail(tolerated)}"
+                )[:4000],
+                dedup_key=f"sec_facts:unresolved_filing:{run_id}",
+                created_at=_now(),
+            ))
+
         # A tracked issuer whose registrant has no XBRL filings can never produce a
         # snapshot, and it is restaged on every run forever. That is a resolution
         # problem, not missing data, and it is silent without this.
@@ -390,8 +561,24 @@ class SecFundamentalsImporter:
         rows: list[dict[str, Any]] = []
         day = last_processed + timedelta(days=1)
         while day <= latest:
+            # Group the whole day first: a combined filing is listed once per
+            # co-registrant CIK, and those sibling CIKs are the only pointer to
+            # where SEC may have put the XBRL (see _recover_from_coregistrant).
+            by_accession: dict[str, list[dict[str, Any]]] = defaultdict(list)
             for r in await client.daily_index(day):
-                if r["form"] in _XBRL_FORMS and r["cik"] in tracked:
+                if r["form"] in _XBRL_FORMS:
+                    by_accession[r["accession"]].append(r)
+            for accession, group in by_accession.items():
+                filers = {r["cik"] for r in group}
+                tracked_filers = filers & tracked
+                if not tracked_filers:
+                    continue
+                siblings = sorted(filers - tracked_filers)
+                if siblings:
+                    self._coregistrants[accession] = siblings
+                for r in group:
+                    if r["cik"] in tracked_filers:
+                        r["index_date"] = day  # not hashed (revision uses cik/accession)
                         rows.append(r)
             day += timedelta(days=1)
         return rows
@@ -409,6 +596,30 @@ class SecFundamentalsImporter:
         ).scalars().all()
         return {int(c) for c in found}
 
+    async def _last_shares_outstanding(self, db, ciks: set[int]) -> dict[str, float]:
+        """Latest known shares outstanding per tracked issuer — the continuity
+        reference co-registrant recovery is checked against."""
+        if not ciks:
+            return {}
+        rows = (
+            await db.execute(
+                select(FundamentalSnapshot.cik, FundamentalSnapshot.shares_outstanding)
+                .where(
+                    FundamentalSnapshot.cik.in_([cik10(c) for c in ciks]),
+                    FundamentalSnapshot.shares_outstanding.is_not(None),
+                )
+                # Last write per cik wins, so the sort must be total: an amendment
+                # and its original share a period_end, and an undefined tie there
+                # would make recovery non-deterministic across runs and dialects.
+                .order_by(
+                    FundamentalSnapshot.period_end,
+                    FundamentalSnapshot.filed_date,
+                    FundamentalSnapshot.accession,
+                )
+            )
+        ).all()
+        return {cik: float(shares) for cik, shares in rows}
+
     async def _existing_by_accession(self, db, accessions: list[str]) -> dict[str, FundamentalSnapshot]:
         if not accessions:
             return {}
@@ -435,6 +646,60 @@ def _companyfacts_structure_error(cf: Any) -> str | None:
     return None
 
 
+def _missing(cik: int, row: dict[str, Any], reason: str, today: date) -> dict[str, Any]:
+    """One unresolvable index row, carrying everything needed to look the filing
+    up by hand (EDGAR accession + the index date it was seen on) and to decide
+    whether it is still young enough to be worth blocking on."""
+    index_date = row.get("index_date")
+    return {
+        "cik": cik10(cik),
+        "accession": row["accession"],
+        "form": row.get("form"),
+        "index_date": index_date,
+        # No index date (older cached rows) => age 0 => blocks, the safe default.
+        "age_days": (today - index_date).days if isinstance(index_date, date) else 0,
+        "reason": reason,
+    }
+
+
+def _within_retry_window(missing: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [m for m in missing if m.get("age_days", 0) <= MISSING_XBRL_RETRY_DAYS]
+
+
+def _past_retry_window(missing: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [m for m in missing if m.get("age_days", 0) > MISSING_XBRL_RETRY_DAYS]
+
+
+def _shares_continuous(shares: float | None, reference: float | None) -> bool:
+    """Does a co-registrant-recovered share count look like this issuer's own?
+
+    The failure worth preventing is storing a subsidiary's standalone facts as the
+    parent's. A co-registrant shell holds a token float — Florida Power & Light
+    against NextEra's 2.09bn shares — so any sane band separates them while still
+    tolerating buybacks and issuance. With no history to compare against (a newly
+    tracked issuer) or no share count at all, recovery is refused, not guessed.
+    """
+    if not shares or not reference:
+        return False
+    return RECOVERY_SHARES_MIN <= shares / reference <= RECOVERY_SHARES_MAX
+
+
+def _reason_counts(missing: list[dict[str, Any]]) -> str:
+    counts = Counter(m["reason"] for m in missing)
+    return ", ".join(f"{reason}={n}" for reason, n in sorted(counts.items()))
+
+
+def _missing_detail(missing: list[dict[str, Any]], limit: int = 10) -> str:
+    detail = ", ".join(
+        f"{m['cik']}/{m['accession']} {m.get('form') or '?'} "
+        f"[{m.get('index_date') or '?'}] {m['reason']}"
+        for m in missing[:limit]
+    )
+    if len(missing) > limit:
+        detail += f", +{len(missing) - limit} more"
+    return detail
+
+
 def _filing_meta(sub: dict[str, Any]) -> tuple[dict[str, FilingMeta], set[str]]:
     """(xbrl_meta, nonxbrl_accessions) from a submissions payload. xbrl_meta only
     includes 10-K/10-Q(/A) filings that are XBRL and have full period metadata."""

tests/unit/test_sec_fundamentals_importer.py

@@ -3,6 +3,7 @@ import framework with a fake SEC client (no network)."""
 
 from __future__ import annotations
 
+import json
 import os
 import tempfile
 from datetime import date, datetime, timezone
@@ -14,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
 from app.database import Base
 import app.models  # noqa: F401
 from app.models.fundamental_snapshot import FundamentalSnapshot
+from app.models.system_event import SystemEvent
 from app.models.ticker import Ticker
 from app.services.data_import import STATUS_FAILED, STATUS_PROMOTED, run_import
 from app.services.sec_fundamentals_importer import SecFundamentalsImporter
@@ -212,10 +214,155 @@ async def test_consistency_gate_fails_when_facts_lag_index(engine):
     run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine)
 
     assert run.status == STATUS_FAILED
-    assert "Company Facts" in (run.error_details or "")
+    # The gate blocks every later run until it clears, so the alert itself has to
+    # name the filing and say why it could not be resolved.
+    details = run.error_details or ""
+    assert "GHOST" in details and "not_in_companyfacts" in details
+    assert "2026-05-01" in details  # index date the filing was seen on
+    summary = json.loads(run.validation_json or "{}")
+    assert summary["missing_xbrl_count"] == 1
+    assert summary["missing_xbrl"][0]["accession"] == "GHOST"
+    assert summary["missing_xbrl"][0]["form"] == "10-Q"
     assert await _count(factory, FundamentalSnapshot) == 2  # nothing new written
 
 
+async def test_gate_separates_missing_submissions_from_missing_facts(engine):
+    """An index row the issuer's own filing list does not carry is a different
+    failure from a Company-Facts lag, and must not be reported as one."""
+    factory = _factory(engine)
+    await _seed(factory, ["AAPL"])
+    backfill = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
+        submissions={320193: _submissions(SUB_FILINGS)},
+        latest_index=date(2026, 1, 31),
+    )
+    await run_import(_importer(backfill), engine=engine)
+
+    incr = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
+        submissions={320193: _submissions(SUB_FILINGS)},  # submissions never lists ORPHAN
+        latest_index=date(2026, 5, 2),
+        daily={date(2026, 5, 1): [{"form": "10-Q", "cik": 320193, "accession": "ORPHAN"}]},
+    )
+    run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine)
+
+    assert run.status == STATUS_FAILED
+    assert "ORPHAN" in (run.error_details or "")
+    assert "not_in_submissions" in (run.error_details or "")
+    summary = json.loads(run.validation_json or "{}")
+    assert summary["missing_xbrl"][0]["reason"] == "not_in_submissions"
+
+
+def _coregistrant_client(share_fact):
+    """Incremental client where Q2A's facts landed in co-registrant 99999's file
+    instead of the filer's own — the NEE-via-FPL / DOW-via-Dow-Chemical shape."""
+    cf_q2 = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "Q2A")
+    return FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={
+            320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1]),  # filer's own: no Q2A
+            99999: _companyfacts([cf_q2], [share_fact], cik=99999),
+        },
+        submissions={320193: _submissions(SUB_FILINGS + [
+            _filing("Q2A", "10-Q", "2026-03-28", "2026-05-01", "2026-05-01T10:01:00.000Z")])},
+        latest_index=date(2026, 5, 2),
+        daily={date(2026, 5, 1): [
+            # one combined filing, listed by the index under both co-registrants
+            {"form": "10-Q", "cik": 320193, "accession": "Q2A"},
+            {"form": "10-Q", "cik": 99999, "accession": "Q2A"},
+        ]},
+    )
+
+
+async def test_recovers_facts_misfiled_under_coregistrant(engine):
+    factory = _factory(engine)
+    await _seed(factory, ["AAPL"])
+    backfill = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
+        submissions={320193: _submissions(SUB_FILINGS)},
+        latest_index=date(2026, 1, 31),
+    )
+    await run_import(_importer(backfill), engine=engine)
+
+    # 14687 shares is continuous with the issuer's own history (14681 last quarter).
+    incr = _coregistrant_client(_shares("2026-04-17", 14687, "Q2A", 2026, "Q2"))
+    run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine)
+
+    assert run.status == STATUS_PROMOTED
+    summary = json.loads(run.validation_json or "{}")
+    assert summary["recovered_count"] == 1
+    assert summary["recovered_from_coregistrant"][0]["source_cik"] == "0000099999"
+    assert summary["missing_xbrl_count"] == 0
+
+    async with factory() as s:
+        q2 = (await s.execute(
+            select(FundamentalSnapshot).where(FundamentalSnapshot.accession == "Q2A")
+        )).scalar_one()
+        codes = (await s.execute(select(SystemEvent.code))).scalars().all()
+    # Stamped to the issuer that filed, NOT the co-registrant whose file it came from.
+    assert q2.cik == "0000320193"
+    assert q2.revenue == 254940 and q2.shares_outstanding == 14687
+    assert "coregistrant_recovery" in codes
+
+
+async def test_coregistrant_recovery_rejects_discontinuous_share_count(engine):
+    """A co-registrant shell's standalone facts must never be stored as the
+    parent's — a token float is the signature and it has to be refused."""
+    factory = _factory(engine)
+    await _seed(factory, ["AAPL"])
+    backfill = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
+        submissions={320193: _submissions(SUB_FILINGS)},
+        latest_index=date(2026, 1, 31),
+    )
+    await run_import(_importer(backfill), engine=engine)
+
+    incr = _coregistrant_client(_shares("2026-04-17", 100, "Q2A", 2026, "Q2"))
+    run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine)
+
+    assert run.status == STATUS_FAILED
+    assert "coregistrant_facts_rejected" in (run.error_details or "")
+    assert await _count(factory, FundamentalSnapshot) == 2  # nothing recovered
+
+
+async def test_unresolved_filing_stops_blocking_after_retry_window(engine):
+    """A filing SEC has misfiled must not wedge every later import forever."""
+    factory = _factory(engine)
+    await _seed(factory, ["AAPL"])
+    backfill = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},
+        submissions={320193: _submissions(SUB_FILINGS)},
+        latest_index=date(2026, 1, 31),
+    )
+    await run_import(_importer(backfill), engine=engine)
+
+    incr = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])},  # no GHOST
+        submissions={320193: _submissions(SUB_FILINGS + [
+            _filing("GHOST", "10-Q", "2026-03-28", "2026-05-01", "2026-05-01T10:01:00.000Z")])},
+        latest_index=date(2026, 5, 2),
+        daily={date(2026, 5, 1): [{"form": "10-Q", "cik": 320193, "accession": "GHOST"}]},
+    )
+    # 9 days after the index date — well past the retry window.
+    run = await run_import(_importer(incr, today=date(2026, 5, 10)), engine=engine)
+
+    assert run.status == STATUS_PROMOTED  # promoted around it, not blocked by it
+    assert run.source_max_date == date(2026, 5, 2)  # and the index advances
+    summary = json.loads(run.validation_json or "{}")
+    assert summary["missing_xbrl_count"] == 1 and summary["missing_xbrl_blocking"] == 0
+
+    async with factory() as s:
+        events = (await s.execute(select(SystemEvent))).scalars().all()
+    unresolved = [e for e in events if e.code == "unresolved_filing"]
+    assert len(unresolved) == 1 and "GHOST" in unresolved[0].message
+
+
 async def test_non_xbrl_amendment_skipped_not_failed(engine):
     factory = _factory(engine)
     await _seed(factory, ["AAPL"])