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

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