feat(host_agent): a dropped ingest says which step of the metric write it stalled in (#5149)

The first instrumented drop on dev read "dropped after 10.0s … (completed:
parse=0.1s)": the whole stall is inside record_metrics. It now marks its
three steps (flush of the rows, the rule and state reads, the per-sample
evaluation) on the caller's phases, and the drop line carries the request's
samples and metric row count.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

steward/core/alerts.py

@@ -141,7 +141,7 @@ def _after_commit(session: AsyncSession, fn, *args) -> None:
     info[_AFTER_COMMIT].append((fn, args))
 
 
-async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> None:
+async def record_metrics(session: AsyncSession, rows: list[PluginMetric], phases=None) -> None:
     """Write metric rows and evaluate the alert rules that watch them.
 
     Every ingest path comes through here, so a rule on any metric Steward
@@ -155,7 +155,11 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non
     happened. A maintenance window holds back notifications and actions, not
     evaluation: the state still moves, and the event says it was muted.
     Notifications and actions start only once the transaction commits.
+
+    `phases`, when given, is marked after each step (flush, rules, evaluate)
+    so a slow caller can say which one its time went to.
     """
+    mark = phases.mark if phases is not None else (lambda _name: None)
     if not rows:
         return
     now = datetime.now(timezone.utc)
@@ -164,6 +168,7 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non
             row.recorded_at = now
     session.add_all(rows)
     await session.flush()
+    mark("flush")
 
     by_key: dict[tuple[str, str, str], list[PluginMetric]] = {}
     for row in rows:
@@ -180,6 +185,7 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non
     )).scalars().all()
     rules = [r for r in rules if (r.source_module, r.resource_name, r.metric_name) in by_key]
     if not rules:
+        mark("rules")
         return
 
     windows = (await session.execute(
@@ -189,6 +195,7 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non
     states = {st.rule_id: st for st in (await session.execute(
         select(AlertState).where(AlertState.rule_id.in_([r.id for r in rules]))
     )).scalars()}
+    mark("rules")
 
     for rule in rules:
         alert_state = states.get(rule.id)
@@ -214,6 +221,7 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non
                 # Admin-configured Ansible action runs only on transition-to-firing.
                 if notif_type == "firing" and rule.ansible_action:
                     _after_commit(session, _run_ansible_action, rule, sample.value, event_id)
+    mark("evaluate")
 
 
 # A series is silent once it has gone this long without a sample, or three of