provider=adapter.name,
        bars=len(bars),
        inserted=written,
    )
    return written


def ingest_security(db: Session, adapter: ProviderAdapter, ticker: str,
                     secondary_adapter: ProviderAdapter | None = None,
                     include_quarterly: bool = False) -> str:
    """Ingest one security's financials from `adapter` (the primary/only source for this call).

    AUDIT FIX (StockLab overhaul, final engineering pass, Part A6): `secondary_adapter` is new and
    optional. When None -- the default, and what every caller in this codebase passed before this
    pass -- this function's behavior is completely unchanged: every field comes straight from
    `adapter`, exactly as before. When given a second adapter, each period's line items are
    resolved field-by-field against the primary's via `resolve_line_items()` (source-hierarchy
    tie-break, real disagreements flagged) instead of blindly overwritten by whichever ran last.
    See docs/AUDIT_PROVIDER_CONFLICT_A6.md for the full design, the period-alignment limitation,
    and honest TESTED/NOT TESTED status -- none of the dual-source path has been executed against
    real provider data or a real database in this build environment.
    """
    is_demo = adapter.tier == "DEMO"
    security = get_or_create_company(db, adapter, ticker)
    _assert_no_demo_live_mix(db, security.id, is_demo)

    source = _get_or_create_source(db, adapter.name, adapter.tier, is_demo, f"ingest:{ticker}")

    # AUDIT FIX (StockLab final engineering pass, Part A9 -- CRITICAL, see
    # docs/AUDIT_INGEST_STATEMENTS_A9.md). Before this fix, this function fetched ONLY the income
    # statement and then wrote the BalanceSheet, CashFlow and Shares rows out of that same
    # response's line items. adapter.get_balance_sheets() and adapter.get_cash_flows() were never
    # called anywhere in the application. That is invisible in DEMO mode (DemoDataAdapter returns
    # one fully-merged dict from all three methods) and catastrophic against FMP, whose income
    # response carries only _INCOME_MAP's fields -- every balance-sheet, cash-flow and share-count
    # column would have been written NULL for real provider data, taking most metrics, all five
    # pillar scores and every valuation down with it.
    income_periods = adapter.get_income_statements(ticker)
    balance_by_period = _fetch_statements_by_period(adapter, adapter.get_balance_sheets, ticker, "balance_sheets")
    cash_by_period = _fetch_statements_by_period(adapter, adapter.get_cash_flows, ticker, "cash_flows")

    # Part A9: optional quarterly ingestion. Off by default (Settings.INGEST_QUARTERLY_PERIODS).
    # Quarterly rows are stored as ordinary FinancialPeriod rows with period_type Q1..Q4 -- the
    # (security_id, period_end, period_type, filing_date) unique constraint keeps them distinct
    # from the annual rows, and every existing query that filters period_type == "FY" is
    # unaffected. app/engines/ttm.py turns four of them into a TTM basis at recompute time.
    if include_quarterly:
        try:
            quarterly_income = adapter.get_income_statements(ticker, period="quarter")
        except NotImplementedError:
            logger.warning("ingest.security.quarterly_not_implemented", ticker=ticker,
                           provider=adapter.name)
            quarterly_income = []
        if quarterly_income:
            income_periods = list(income_periods) + list(quarterly_income)
            balance_by_period.update(_fetch_statements_by_period(
                adapter, lambda t: adapter.get_balance_sheets(t, period="quarter"), ticker,
                "balance_sheets_quarterly"))
            cash_by_period.update(_fetch_statements_by_period(
                adapter, lambda t: adapter.get_cash_flows(t, period="quarter"), ticker,
                "cash_flows_quarterly"))

    secondary_source = None
    secondary_by_period: dict = {}
    secondary_balance: dict = {}
    secondary_cash: dict = {}
    if secondary_adapter is not None:
        if secondary_adapter.tier == "DEMO" or is_demo:
            # Mirrors _assert_no_demo_live_mix's own rule at the single-call level: a demo source
            # can never be one half of a dual-source resolution, since demo data is synthetic and
            # was never meant to be cross-checked against (or corrupt) real provider data.
            raise RuntimeError("Refusing dual-source ingestion: DEMO cannot be paired with a second provider.")
        secondary_source = _get_or_create_source(db, secondary_adapter.name, secondary_adapter.tier, is_demo, f"ingest:{ticker}")
        # AUDIT (Part A6): period alignment across two providers is keyed on (period_end,
        # period_type) only -- NOT filing_date, which two providers can legitimately report a day
        # or more apart for the "same" filing. This is a real, documented limitation: a provider
        # that reports a fiscal period boundary a few days off from the other (a genuine, if rare,
        # real-world occurrence) will not be matched and that period is silently treated as
        # primary-only for this run, not flagged as a mismatch. See docs/AUDIT_PROVIDER_CONFLICT_A6.md.
        try:
            secondary_by_period = {
                (p.period_end, p.period_type): p for p in secondary_adapter.get_income_statements(ticker)
            }
            # Part A9: the secondary provider's balance-sheet and cash-flow responses are fetched
            # too, so cross-source conflict resolution covers all four statement groups rather
            # than only the income statement. _fetch_statements_by_period returns {} for an
            # adapter that has not implemented a statement, so this adds no new failure mode.
            secondary_balance = _fetch_statements_by_period(
                secondary_adapter, secondary_adapter.get_balance_sheets, ticker, "balance_sheets")
            secondary_cash = _fetch_statements_by_period(
                secondary_adapter, secondary_adapter.get_cash_flows, ticker, "cash_flows")
        except NotImplementedError:
            # AUDIT FIX (Part A6, real bug caught before this shipped): EODHDAdapter.get_income_
            # statements/get_balance_sheets/get_cash_flows are NotImplementedError today
            # (app/adapters/eodhd.py's own module docstring -- statement-level field mapping was
            # deliberately deferred until verified against a live response). EODHD is the only
            # other real (non-DEMO) adapter besides FMP, so PROVIDER_SECONDARY="EODHD" is the one
            # realistic dual-source configuration this codebase can actually be set to today -- and
            # without this except clause, setting it would crash every single ingestion task the
            # instant this function tried to call the secondary adapter's statement method. Treated
            # as "no secondary statement data available this run", not a crash: falls back to
            # exactly the primary-only behavior secondary_adapter=None already has for every period.
            logger.warning("ingest.security.secondary_statements_not_implemented", ticker=ticker,
                            secondary_provider=secondary_adapter.name)

    for p in income_periods:
        fp = db.query(FinancialPeriod).filter_by(
            security_id=security.id, period_end=p.period_end, period_type=p.period_type, filing_date=p.filing_date,
        ).one_or_none()
        if fp is None:
            fp = FinancialPeriod(
                security_id=security.id, period_end=p.period_end, period_type=p.period_type,
                filing_date=p.filing_date, currency=p.currency, source_id=source.id,
            )
            db.add(fp)
            db.flush()

        key = (p.period_end, p.period_type)
        li = merge_statement_line_items(
            p.line_items,
            balance_by_period.get(key),
            cash_by_period.get(key),
        )
        secondary_p = secondary_by_period.get(key)
        li_b = (
            merge_statement_line_items(
                secondary_p.line_items, secondary_balance.get(key), secondary_cash.get(key)
            )
            if secondary_p is not None
            else None
        )
        secondary_tier = secondary_adapter.tier if (secondary_adapter is not None and li_b is not None) else None

        income = fp.income_statement or IncomeStatement(financial_period_id=fp.id)
        resolved, conflicts = resolve_line_items(
            li, li_b, _INCOME_FIELDS, adapter.tier, secondary_tier
        )
        merged_report = ValidationReport()
        resolved = validate_line_items(
            resolved,
            merged_report,
            row_key=f"{ticker}:{p.period_end}:{p.period_type}:income",
        )
        for field, value in resolved.items():
            setattr(income, field, value)
        db.add(income)
        if conflicts:
            _record_conflicts(db, "income_statements", fp.id, conflicts, source, secondary_source)

        balance = fp.balance_sheet or BalanceSheet(financial_period_id=fp.id)
        resolved, conflicts = resolve_line_items(
            li, li_b, _BALANCE_FIELDS, adapter.tier, secondary_tier
        )
        merged_report = ValidationReport()
        resolved = validate_line_items(
            resolved,
            merged_report,
            row_key=f"{ticker}:{p.period_end}:{p.period_type}:balance",
        )
        for field, value in resolved.items():
            setattr(balance, field, value)
        db.add(balance)
        if conflicts:
            _record_conflicts(db, "balance_sheets", fp.id, conflicts, source, secondary_source)

        cash_flow = fp.cash_flow or CashFlow(financial_period_id=fp.id)
        resolved, conflicts = resolve_line_items(
            li, li_b, _CASH_FLOW_FIELDS, adapter.tier, secondary_tier
        )
        merged_report = ValidationReport()
        resolved = validate_line_items(
            resolved,
            merged_report,
            row_key=f"{ticker}:{p.period_end}:{p.period_type}:cash_flow",
        )
        for field, value in resolved.items():
            setattr(cash_flow, field, value)
        db.add(cash_flow)
        if conflicts:
            _record_conflicts(db, "cash_flows", fp.id, conflicts, source, secondary_source)

        shares = fp.shares or Shares(financial_period_id=fp.id)
        resolved, conflicts = resolve_line_items(
            li, li_b, _SHARES_FIELDS, adapter.tier, secondary_tier
        )
        merged_report = ValidationReport()
        resolved = validate_line_items(
            resolved,
            merged_report,
            row_key=f"{ticker}:{p.period_end}:{p.period_type}:shares",
        )
        for field, value in resolved.items():
            setattr(shares, field, value)
        db.add(shares)
        if conflicts:
            _record_conflicts(db, "shares", fp.id, conflicts, source, secondary_source)

    # AUDIT FIX (final master pass, §21) — the fourth "engine built, never wired" defect.
    # `ProviderAdapter.get_estimates()` is implemented by FMP and by the demo adapter,
    # `ProviderEstimateRow` exists, and the `estimates` table exists with a unique constraint on
    # (security_id, period_end, metric, as_of_date). Nothing ever called it, nothing ever wrote a