diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py
index 7ed2ac2..01807dc 100644
--- a/tests/unit/test_sec_fundamentals_importer.py
+++ b/tests/unit/test_sec_fundamentals_importer.py
@@ -1204,3 +1204,82 @@ async def test_ceiling_never_fires_before_a_first_promotion(engine):
 
     assert result.ok is False
     assert result.summary["promotion_ceiling_tripped"] is None
+
+
+async def test_ceiling_promotes_queues_and_alerts_end_to_end(engine, monkeypatch):
+    """The self-heal claim, end to end: a filing that would block forever gets
+    promoted through, queued for retry, and announced."""
+    from app.models.data_import_run import DataImportRun
+    from app.services.sec_facts_parser import ParseResult
+    from sqlalchemy import update as sa_update
+
+    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),
+    )
+    assert (await run_import(_importer(backfill), engine=engine)).status == STATUS_PROMOTED
+
+    # Age the only promotion past the ceiling: this is the wedge the ceiling exists
+    # for — the filing below stays young, so nothing else would ever release it.
+    async with factory() as db:
+        await db.execute(
+            sa_update(DataImportRun)
+            .where(DataImportRun.source == "sec_facts")
+            .values(started_at=datetime(2026, 4, 20, tzinfo=timezone.utc))
+        )
+        await db.commit()
+
+    # Present in Company Facts but unparseable — one missing_xbrl entry, not the
+    # two an absent-from-facts accession would also raise.
+    stuck_fact = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "STUCK")
+    stuck_share = _shares("2026-04-17", 14687, "STUCK", 2026, "Q2")
+    client = FakeSecClient(
+        tickers={"AAPL": 320193},
+        companyfacts={
+            320193: _companyfacts(
+                [CF_K, CF_Q1, stuck_fact], [SH_K, SH_Q1, stuck_share]
+            )
+        },
+        submissions={320193: _submissions(SUB_FILINGS + [
+            _filing("STUCK", "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": "STUCK"},
+        ]},
+    )
+    monkeypatch.setattr(
+        "app.services.sec_facts_parser.parse_snapshots",
+        lambda *a, **k: ParseResult(
+            skipped_filings=[{"accession": "STUCK", "reason": "unparseable"}]
+        ),
+    )
+
+    # index_date 2026-05-01 vs today 2026-05-03 => 2 days old, still inside the
+    # per-filing window, so only the aggregate ceiling can let this through.
+    run = await run_import(_importer(client, today=date(2026, 5, 3)), engine=engine)
+    assert run.status == STATUS_PROMOTED
+
+    summary = json.loads(run.validation_json)
+    assert summary["promotion_ceiling_tripped"] == {"forced": 1, "unresolved": 1}
+
+    async with factory() as db:
+        gap = (await db.execute(select(SecFilingGap))).scalar_one()
+        events = (
+            await db.execute(
+                select(SystemEvent).where(
+                    SystemEvent.code == "promotion_ceiling_forced"
+                )
+            )
+        ).scalars().all()
+    # Queued, so later runs retry it without it ever blocking again...
+    assert gap.accession == "STUCK"
+    # ...and the safety valve firing is visible, not silent.
+    assert len(events) == 1
+    assert events[0].severity == "warning"
+    assert "7 days" in events[0].message