Per Council R3 Lock R3-4 item #10. Data layer for all future tuning
decisions — escalation-rate, classifier, route-shape. Without
per-route counters we're flying blind on whether cheap-substrate
work is saving what we expect.

# New module: kora_cli/telemetry/

  * ``cost_telemetry.py`` — ``CostRouteTelemetry`` singleton with
    threading.RLock-protected counters across 3 windows
    (process_lifetime / rolling_24h / monthly). Per-route counters:
    calls_count, input/output/cache_read/cache_creation_tokens_total,
    cost_estimate_usd_total, escalation_count, model_breakdown.
  * ``__init__.py`` — re-exports + canonical route + window literals.

Route taxonomy (v1, spec §2):
  slack_dm, email_inbound, email_outbound_compose, mcp_tool,
  alert_investigation, probe_investigation, tool_loop_iteration,
  scheduled_task, unknown.

Fail-soft: unknown route strings bucket to "unknown" rather than
raising. ``record_call`` wraps the inner write in try/except so
hot-path inference handlers never see a telemetry failure.

# Wire route through record_inference

``CostStateHolder.record_inference`` gains optional kwargs:
  - ``route: str = "unknown"`` — canonical taxonomy literal
  - ``escalated_to_opus: bool = False`` — Lock R3-3 tunable signal

Both are additive + backwards-compatible. After successful pricing
estimation, telemetry's ``record_call`` is invoked alongside the
existing billing accumulation. Telemetry write is READ-side (does
NOT affect billing) and fail-soft on import / record errors.

``cost_ladder_wire.record_inference_from_response`` forwards both
kwargs verbatim.

No source= harmonization needed (existing record_inference had no
source kwarg; the IncomingMessage.source field is at a different
layer — the engine input).

# Route wiring this PR (wired vs deferred)

| Route | Call site | Wired? |
|---|---|---|
| slack_dm | slack_dm_handler.py:676 (Kora reply-bill) | ✅ wired |
| email_inbound | (handler doesn't yet write a bill) | ⏸️ reserved |
| email_outbound_compose | (no consumer) | ⏸️ reserved |
| mcp_tool | (no consumer) | ⏸️ reserved |
| alert_investigation | (consumer pending KR-PLUGIN-AUDIT) | ⏸️ reserved |
| probe_investigation | (consumer pending probe-audit) | ⏸️ reserved |
| tool_loop_iteration | (engine doesn't yet surface iteration tag) | ⏸️ reserved |
| scheduled_task | (no consumer) | ⏸️ reserved |
| (agent main loop) | conversation_loop.py:1603 | leaves default "unknown" — agent-side, not Kora-side; not in spec taxonomy |
| (auxiliary client) | auxiliary_client.py:5341/5360 | leaves default "unknown" — agent-side |

Every reserved route accepts the literal today; consumer wiring is
a follow-on bucket per the spec §4 STOP-ASK guidance ("don't fail
the bucket on missing call sites").

# Periodic-task wiring (cost_telemetry_listener)

3 tasks registered with the heartbeat scheduler:
  * ``cost_telemetry.persist`` — 5min cadence; atomic-writes
    counter snapshot to ``${KORA_HOME}/cache/cost_telemetry.json``
  * ``cost_telemetry.rolling_24h_reset`` — 1h watch-and-act;
    fires reset on UTC date crossover
  * ``cost_telemetry.monthly_reset`` — 1h watch-and-act; fires
    reset on UTC month rollover

First-tick stamping pattern: the reset checks stamp "today's
date" on first call after boot without firing a reset (counters
at zero anyway), so resets only fire on subsequent boundary
crossings. Avoids the trickiness of exact-midnight asyncio
scheduling.

# Web endpoint

``GET /api/cost_telemetry`` returns the in-memory counter
snapshot across all 3 windows (no disk roundtrip; no LLM cost).

# Snapshot v2 (KR-CHEAP-PRE-WARMED-SNAPSHOT extension)

Bumped ``SCHEMA_VERSION`` 1 → 2. ``compute_snapshot()`` adds a
``cost_telemetry`` section exposing the two operator-facing
windows (``rolling_24h`` + ``monthly``); ``process_lifetime``
intentionally excluded from the on-disk snapshot to keep file
size bounded — operator hits ``/api/cost_telemetry`` for the
full window set. Section degrades to empty dicts on telemetry
unavailability (fail-soft).

# Concurrency

threading.RLock on CostRouteTelemetry protects counter mutations
+ snapshot reads. Same shape as cron/jobs.py + suitable for
asyncio loop + cron + reasoning all potentially racing. Test
``test_concurrent_record_call_no_lost_counts`` proves 1000 calls
across 10 threads → exact total (no lost increments).

# Read-only contract preserved

This module is a READ-side observer of cost-ladder accounting. It
does NOT change billing accumulation logic. Telemetry can be
disabled (singleton swap) without affecting the cost-ladder's
$200/mo budget enforcement.

# Tests

141/141 pass:
  * 27 cost_telemetry (counter shape + route taxonomy + windows +
    concurrency + singleton + JSON-serializability + fail-soft)
  * 20 listener (registration + cadence resolution + persist
    cycle + window-reset checks + read-back + lifecycle log)
  * 9 web endpoint + snapshot v2 (endpoint shape + version bump
    + cost_telemetry section + degradation + end-to-end)
  * 51 snapshot tests pass (1 updated for schema v2 + new top-
    level key)
  * 16 existing cost_state_holder + cost_ladder_wire tests pass
    (backwards-compat preserved across new optional kwargs)

Ruff clean.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>