fix(telemetry): update stage rows count the stage a failed run died in

hermes.update.run blamed 971 failed CLI runs in Sep 28 - Oct 5 on the apply
stage while hermes.update.stage held zero failed apply rows. Stage marks are
END marks: every pre-apply exit (fetch, channel, branch, merge, HEAD checks)
leaves no mark for the stage it died in, so the run row inferred the stage and
no stage row said it failed. A failed run whose failed_stage names a stage it
never marked now gets one failed row for it.

Rebuilt on the updater overhaul's followups model (#132386): a committed run
that owes follow-ups is a run-level success and its failed stage already has
its own mark, so only failed runs get the row; refusals never do.

hermes_cli/observability/shared_metrics_update.py

@@ -11,6 +11,11 @@ Desktop's packaged updaters (electron-updater, App Installer, Store) never run `
 Desktop reports their outcome through the ``shared_metrics.update_run`` RPC instead. Desktop's
 source-checkout hand-off DOES run ``hermes update``; that receipt is tagged ``initiator=desktop``
 and counted here, never by the RPC, so no run is counted twice.
+
+Run and stage rows agree on where a FAILED run stopped: a run whose ``failed_stage`` names a stage it
+never marked (it exited inside it, before the stage's END mark) also gets one failed stage row for it.
+A run that passed the commit point and only owes follow-ups (contract C3) is a ``success`` run, while
+the stage that failed (``deps``, ``build``, ``restart``, ``verify``) still reads ``failed`` at stage level.
 """
 
 from __future__ import annotations
@@ -293,9 +298,31 @@ def update_receipt_fields(receipt: dict[str, Any]) -> tuple[dict[str, str], list
         }
         for stage in stages
     ]
+    if died_in := _unmarked_failed_stage(run, stages):
+        last_mark = next((s["at"] for s in reversed(stages) if s.get("at")), receipt.get("started_at"))
+        stage_rows.append({
+            "duration_bucket": update_duration_bucket(_elapsed_ms(last_mark, receipt.get("finished_at"))),
+            "outcome": "failed", "stage": died_in,
+        })
     return run, stage_rows
 
 
+def _unmarked_failed_stage(run: dict[str, str], stages: list[dict[str, Any]]) -> str | None:
+    """The stage a failed run stopped in when the receipt holds no mark for it, else None.
+
+    Stage marks are END marks, so a run that exits inside a stage (every pre-apply ``sys.exit``:
+    fetch, channel, branch, merge, HEAD checks) leaves none for it and the stage rows would show no
+    failure where the run row says it died. Only ``failed`` runs: a refusal is not a stage failure,
+    and a committed run that owes follow-ups is a ``success`` whose failed stage already has its mark.
+    """
+    from .shared_metrics_contract import UPDATE_STAGES
+
+    died_in = run["failed_stage"]
+    if run["outcome"] != "failed" or died_in not in UPDATE_STAGES or any(s["name"] == died_in for s in stages):
+        return None
+    return died_in
+
+
 def _collection_on() -> bool:
     """Cheap pre-gate so a disabled install never loads the Relay runtime; `_emit` re-checks."""
     from hermes_cli.config import read_raw_config_readonly

tests/hermes_cli/test_shared_metrics_install_failures.py

@@ -116,6 +116,23 @@ def test_a_partial_run_with_a_failed_stage_keeps_that_stage_and_its_verify_row_p
     assert [(s["stage"], s["outcome"]) for s in stage_rows if s["outcome"] == "failed"] == [(failed, "failed")]
 
 
+def test_a_run_that_died_before_a_stage_mark_has_a_failed_row_for_that_stage():
+    """Invariant: stage rows agree with the run row on where a FAILED run stopped, and a committed run
+    that owes follow-ups (C3) is a run-level success whose failed stage still shows at stage level."""
+    fetch_failed = _receipt("failed", [("plan", "success"), ("snapshot", "skipped")], exit_code=1,
+                            stop_reason="sys.exit(1)")
+    run, stage_rows = update_metrics.update_receipt_fields(fetch_failed)
+    failed_rows = [s for s in stage_rows if s["outcome"] == "failed"]
+    assert run["failed_stage"] == "apply" and [s["stage"] for s in failed_rows] == ["apply"]
+    assert all(contract.counter_dimensions_are_valid(contract.UPDATE_STAGE_METRIC, s) for s in stage_rows)
+
+    deps_owed = _receipt("success", [*_ALL_PASSED[:3], ("deps", "failed")],
+                         followups=[{"step": "dependencies", "reason": "uv sync exited 2"}])
+    run, stage_rows = update_metrics.update_receipt_fields(deps_owed)
+    assert (run["outcome"], run["failed_stage"]) == ("success", "none")
+    assert [s["stage"] for s in stage_rows if s["outcome"] == "failed"] == ["deps"]
+
+
 @pytest.mark.parametrize(("receipt", "expected"), [
     (_receipt("success", _ALL_PASSED), "none"),
     (_receipt("refused", [], steps=[{"name": "admission", "ok": False}], stop_reason="docker"), "managed_install"),

tests/hermes_cli/test_shared_metrics_reliability.py

@@ -75,8 +75,9 @@ def test_failed_update_reports_the_stage_it_never_reached(marks, monkeypatch):
         "apply_mode": "git", "duration_bucket": "lt_30s", "failed_stage": "deps",
         "from_version_age_bucket": "7d_to_30d", "kind": "cli", "outcome": "failed", "failure_class": "deps_failed",
     }]
+    # The stage the run died in carries a failed row, so run and stage counts agree.
     assert [(s["stage"], s["outcome"]) for s in stages] == [
-        ("plan", "success"), ("snapshot", "skipped"), ("apply", "success")]
+        ("plan", "success"), ("snapshot", "skipped"), ("apply", "success"), ("deps", "failed")]
     _assert_valid(marks.rows)