Add imap-ingest CLI + ImapProvider: route emails to IE/Schwab parsers

Wires the IE + Schwab email parsers into an actual runnable sync. Walks
the IMAP mailbox, routes each message by sender domain:
  - *@investengine.com → invest_engine.parse_invest_engine_email
  - *@schwab.com       → schwab.parse_schwab_email
then pushes the resulting Activities through the shared pipeline.

broker-sync imap-ingest — new CLI command taking IMAP_HOST/USER/PASSWORD/
DIRECTORY (mirrors the old wealthfolio-sync image's env shape so the
Terraform CronJob's existing env wiring works unchanged).

Verified: poetry run pytest -q → 109 passed + 1 skipped; mypy strict
clean (37 files); ruff + yapf clean.

broker_sync/cli.py

@@ -230,6 +230,69 @@ def invest_engine(
     asyncio.run(_run())
 
 
+@app.command("imap-ingest")
+def imap_ingest(
+        wf_base_url: str = typer.Option(..., envvar="WF_BASE_URL"),
+        wf_username: str = typer.Option(..., envvar="WF_USERNAME"),
+        wf_password: str = typer.Option(..., envvar="WF_PASSWORD"),
+        wf_session_path: str = typer.Option("/data/wealthfolio_session.json",
+                                            envvar="WF_SESSION_PATH"),
+        imap_host: str = typer.Option(..., envvar="IMAP_HOST"),
+        imap_user: str = typer.Option(..., envvar="IMAP_USER"),
+        imap_password: str = typer.Option(..., envvar="IMAP_PASSWORD"),
+        imap_directory: str = typer.Option("INBOX", envvar="IMAP_DIRECTORY"),
+        data_dir: str = typer.Option("/data", envvar="BROKER_SYNC_DATA_DIR"),
+) -> None:
+    """Phase 2/3 — ingest InvestEngine + Schwab confirmation emails via IMAP.
+
+    Walks the mailbox, routes each message by `From:` sender domain to the
+    matching parser, pushes any resulting activities through the shared
+    pipeline (dedup → Wealthfolio CSV-free JSON import).
+    """
+    from broker_sync.dedup import SyncRecordStore
+    from broker_sync.pipeline import sync_provider_to_wealthfolio
+    from broker_sync.providers.imap import ImapCreds, ImapProvider
+    from broker_sync.sinks.wealthfolio import WealthfolioSink
+
+    _setup_logging()
+    data = Path(data_dir)
+    data.mkdir(parents=True, exist_ok=True)
+
+    async def _run() -> None:
+        sink = WealthfolioSink(
+            base_url=wf_base_url,
+            username=wf_username,
+            password=wf_password,
+            session_path=wf_session_path,
+        )
+        provider = ImapProvider(
+            ImapCreds(
+                host=imap_host,
+                user=imap_user,
+                password=imap_password,
+                directory=imap_directory,
+            ))
+        dedup = SyncRecordStore(data / "sync.db")
+        try:
+            if not Path(wf_session_path).exists():
+                await sink.login()
+            result = await sync_provider_to_wealthfolio(
+                provider=provider,
+                sink=sink,
+                dedup=dedup,
+            )
+        finally:
+            await sink.close()
+        typer.echo(f"imap-ingest: fetched={result.fetched} "
+                   f"new={result.new_after_dedup} "
+                   f"imported={result.imported} "
+                   f"failed={result.failed}")
+        if result.failed > 0:
+            sys.exit(1)
+
+    asyncio.run(_run())
+
+
 def _setup_logging() -> None:
     logging.basicConfig(
         level=logging.INFO,