diff --git a/hermes_cli/observability/__init__.py b/hermes_cli/observability/__init__.py
index 1c4e70300c..3f7e611bc8 100644
--- a/hermes_cli/observability/__init__.py
+++ b/hermes_cli/observability/__init__.py
@@ -12,7 +12,10 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None:
     """Dispatch a Hermes lifecycle event to built-in observability features."""
     from . import relay_shared_metrics
 
-    _safe_observe(relay_shared_metrics.observe_lifecycle, hook_name, kwargs)
+    try:
+        relay_shared_metrics.observe_lifecycle(hook_name, **kwargs)
+    except Exception:
+        logger.warning("Built-in observability hook failed: %s", hook_name, exc_info=True)
 
 
 def handles_hook(hook_name: str) -> bool:
@@ -20,12 +23,3 @@ def handles_hook(hook_name: str) -> bool:
     from . import relay_shared_metrics
 
     return relay_shared_metrics.handles_hook(hook_name)
-
-
-def _safe_observe(callback: Any, hook_name: str, kwargs: dict[str, Any]) -> None:
-    try:
-        callback(hook_name, **kwargs)
-    except Exception:
-        logger.warning(
-            "Built-in observability hook failed: %s", hook_name, exc_info=True
-        )
diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py
index d7cc71cc6a..a382bf18e9 100644
--- a/hermes_cli/observability/relay_shared_metrics.py
+++ b/hermes_cli/observability/relay_shared_metrics.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import atexit
+import contextlib
 import contextvars
 import logging
 import threading
@@ -41,34 +42,21 @@ from .shared_metrics_subscriber import SharedMetricsSubscriber
 
 logger = logging.getLogger(__name__)
 
-HANDLED_HOOKS = frozenset({
-    "on_session_start",
-    "on_session_end",
-    "on_session_finalize",
-    "on_session_reset",
-    "pre_llm_call",
-    "pre_api_request",
-    "pre_tool_call",
-    "post_tool_call",
-    "post_approval_response",
-    "post_api_request",
-    "api_request_error",
-    "on_skill_lifecycle",
-    "subagent_stop",
-})
-
 _RUNTIME_FAILED = object()
 _RUNTIMES: dict[str, _Runtime | object] = {}
 _RUNTIME_LOCK = threading.RLock()
 
+_ABORTED = {"failed": True, "turn_exit_reason": "system_aborted"}
+
+
+def _text(event: dict[str, Any], key: str) -> str:
+    return str(event.get(key) or "")
+
 
 def _session_pair(event: dict[str, Any], key: str) -> tuple[str, str] | None:
     """(session_id, event[key]) when both are non-empty."""
-    session_id = str(event.get("session_id") or "")
-    value = str(event.get(key) or "")
-    if not session_id or not value:
-        return None
-    return session_id, value
+    session_id, value = _text(event, "session_id"), _text(event, key)
+    return (session_id, value) if session_id and value else None
 
 
 def _retry_ordinal(event: dict[str, Any]) -> int | None:
@@ -78,6 +66,12 @@ def _retry_ordinal(event: dict[str, Any]) -> int | None:
     return None
 
 
+def _sole(items: Any) -> Any:
+    """The single distinct element of ``items`` (identity-deduplicated), else None."""
+    unique = {id(item): item for item in items}
+    return next(iter(unique.values())) if len(unique) == 1 else None
+
+
 @dataclass
 class _ModelCall:
     handle: Any
@@ -119,12 +113,8 @@ class _MetricsSession:
     closing: bool = False
     model_calls: dict[tuple[str, str], _ModelCall] = field(default_factory=dict)
     tasks: dict[str, _TaskRun] = field(default_factory=dict)
-    tool_calls: dict[tuple[str, str, str, str], _ToolCall] = field(
-        default_factory=dict
-    )
-    retired_turn_ids: deque[str] = field(
-        default_factory=lambda: deque(maxlen=256),
-    )
+    tool_calls: dict[tuple[str, str, str, str], _ToolCall] = field(default_factory=dict)
+    retired_turn_ids: deque[str] = field(default_factory=lambda: deque(maxlen=256))
 
 
 class _Runtime:
@@ -148,9 +138,7 @@ class _Runtime:
         self._turn_sessions: dict[tuple[str, str], _MetricsSession] = {}
         self._subscriber_name = f"{SUBSCRIBER_NAME}.{self.host.runtime_id}"
         self.subscriber = SharedMetricsSubscriber(
-            SharedMetricsStore(),
-            __version__,
-            runtime_id=self.host.runtime_id,
+            SharedMetricsStore(), __version__, runtime_id=self.host.runtime_id
         )
         self.relay.subscribers.register(self._subscriber_name, self.subscriber)
         self.host.retain_managed_execution(self._subscriber_name)
@@ -158,7 +146,7 @@ class _Runtime:
         atexit.register(self.shutdown)
 
     def ensure_session(self, event: dict[str, Any]) -> _MetricsSession | None:
-        session_id = str(event.get("session_id") or "")
+        session_id = _text(event, "session_id")
         if not session_id:
             return None
         with self._sessions_lock:
@@ -169,10 +157,7 @@ class _Runtime:
                 return None
             session = self._sessions.get(session_id)
             if session is None:
-                session = _MetricsSession(
-                    session_id=session_id,
-                    relay_session=relay_session,
-                )
+                session = _MetricsSession(session_id=session_id, relay_session=relay_session)
                 self._sessions[session_id] = session
         with session.lock:
             if session.closing:
@@ -182,40 +167,32 @@ class _Runtime:
     def record_client_active(self, event: dict[str, Any]) -> None:
         """Emit one payload-free activation attempt under the session scope."""
         session = self.ensure_session(event)
-        if session is None:
-            return
-        self._emit_client_active(session)
+        if session is not None:
+            self._emit_client_active(session)
 
     def _emit_client_active(self, session: _MetricsSession) -> None:
         with session.lock:
-            if session.closing:
-                return
-            self._run_in_session(
-                session,
-                self.relay.scope.event,
-                CLIENT_ACTIVE_MARK,
-                handle=session.relay_session.handle,
-                data={},
-                metadata=self._event_metadata(),
-            )
+            if not session.closing:
+                self._mark(session, None, CLIENT_ACTIVE_MARK, {})
+
+    def _mark(
+        self, session: _MetricsSession, task: _TaskRun | None, name: str, data: dict[str, str]
+    ) -> None:
+        """Emit one Relay mark under the task scope when given, else the session scope."""
+        handle = task.handle if task is not None else session.relay_session.handle
+        self._run_scoped(
+            session, task, self.relay.scope.event, name,
+            handle=handle, data=data, metadata=self._event_metadata(),
+        )
 
     def _run_in_session(
-        self,
-        session: _MetricsSession,
-        callback: Callable[..., Any],
-        *args: Any,
-        **kwargs: Any,
+        self, session: _MetricsSession, callback: Callable[..., Any], *args: Any, **kwargs: Any
     ) -> Any:
-        return self.host.run_in_session(
-            session.relay_session,
-            callback,
-            *args,
-            **kwargs,
-        )
+        return self.host.run_in_session(session.relay_session, callback, *args, **kwargs)
 
     def start_task(self, event: dict[str, Any]) -> _TaskRun | None:
         """Open one Relay function scope for a Hermes task run."""
-        task_key = self._task_key(event)
+        task_key = _session_pair(event, "task_id")
         if task_key is None:
             return None
         _, task_id = task_key
@@ -226,17 +203,15 @@ class _Runtime:
                     if owner.closing:
                         return None
                     task = owner.tasks.get(task_id)
-                    if task is not None:
-                        if not self._event_matches_task_turn(task, event):
-                            return None
-                        self._remember_turn(owner, task, event)
+                    if task is not None and not self._admits(owner, task, event):
+                        return None
                     return task
 
             session = self.ensure_session(event)
             if session is None:
                 return None
             with session.lock:
-                turn_id = str(event.get("turn_id") or "")
+                turn_id = _text(event, "turn_id")
                 if (
                     session.closing
                     or (turn_id and turn_id in session.retired_turn_ids)
@@ -256,17 +231,11 @@ class _Runtime:
                 ):
                     parent_handle = active_turn.handle
 
-                def push_task() -> Any:
-                    self.relay.get_scope_stack()
-                    return self.relay.scope.push(
-                        TASK_SCOPE,
-                        self.relay.ScopeType.Function,
-                        handle=parent_handle,
-                        input=start_fields,
-                        metadata=self._event_metadata(),
-                    )
-
-                handle = task_context.run(push_task)
+                handle = task_context.run(
+                    self._with_scope_stack, self.relay.scope.push,
+                    TASK_SCOPE, self.relay.ScopeType.Function,
+                    handle=parent_handle, input=start_fields, metadata=self._event_metadata(),
+                )
                 task = _TaskRun(
                     task_id=task_id,
                     handle=handle,
@@ -282,20 +251,16 @@ class _Runtime:
                 return task
 
     def _run_in_task(
-        self,
-        task: _TaskRun,
-        callback: Callable[..., Any],
-        *args: Any,
-        **kwargs: Any,
+        self, task: _TaskRun, callback: Callable[..., Any], *args: Any, **kwargs: Any
     ) -> Any:
-        def invoke() -> Any:
-            self.relay.get_scope_stack()
-            return callback(*args, **kwargs)
+        return task.context.copy().run(self._with_scope_stack, callback, *args, **kwargs)
 
-        return task.context.copy().run(invoke)
+    def _with_scope_stack(self, callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+        self.relay.get_scope_stack()
+        return callback(*args, **kwargs)
 
     def start_model_call(self, event: dict[str, Any]) -> None:
-        task_id = str(event.get("task_id") or "")
+        task_id = _text(event, "task_id")
         session, task = self._task_for(event, start=True)
         if task_id and task is None:
             return
@@ -303,88 +268,77 @@ class _Runtime:
             session = self.ensure_session(event)
         if session is None:
             return
-        model_call_key = self._new_model_call_key(event)
-        if model_call_key is None:
+        request_id = _text(event, "api_request_id")
+        if not request_id:
             return
-        _, request_id = model_call_key
+        model_call_key = (task_id, request_id)
         fields = model_call_fields(event)
         retry_ordinal = _retry_ordinal(event)
         with session.lock:
             if session.closing:
                 return
-            if task is not None:
-                if (
-                    session.tasks.get(task.task_id) is not task
-                    or not self._event_matches_task_turn(task, event)
-                ):
-                    return
-                self._remember_turn(session, task, event)
+            if task is not None and not self._admits(session, task, event, current=True):
+                return
             existing = session.model_calls.get(model_call_key)
             if existing is not None:
                 existing.fields = fields
                 if task is not None:
-                    # Every repeated start for one logical request is another
-                    # physical attempt. Provider fallback resets Hermes's
-                    # provider-local retry ordinal, so ordinal deltas are not a
-                    # reliable task-level retry counter.
+                    # Every repeated start for one logical request is another physical
+                    # attempt. Provider fallback resets Hermes's provider-local retry
+                    # ordinal, so ordinal deltas are not a reliable task-level counter.
                     task.retry_count += 1
                 if retry_ordinal is not None:
-                    existing.retry_ordinal = max(
-                        existing.retry_ordinal or 0,
-                        retry_ordinal,
-                    )
+                    existing.retry_ordinal = max(existing.retry_ordinal or 0, retry_ordinal)
                 return
             if task is not None:
                 task.model_call_ids.add(request_id)
                 if retry_ordinal is not None and retry_ordinal > 0:
-                    # A real Hermes retry can advance api_request_id while
-                    # carrying the retry ordinal. Count that physical attempt.
+                    # A real Hermes retry can advance api_request_id while carrying the
+                    # retry ordinal. Count that physical attempt.
                     task.retry_count += 1
             handle = self._run_scoped(
-                session,
-                task,
-                self.relay.llm.call,
-                MODEL_CALL_SCOPE,
-                self.relay.LLMRequest({}, {}),
+                session, task, self.relay.llm.call, MODEL_CALL_SCOPE, self.relay.LLMRequest({}, {}),
                 handle=task.handle if task is not None else session.relay_session.handle,
                 metadata=self._event_metadata(),
                 model_name=MODEL_CALL_PROFILE_MODEL,
             )
             session.model_calls[model_call_key] = _ModelCall(
-                handle=handle,
-                task_id=task_id,
-                fields=fields,
-                retry_ordinal=retry_ordinal,
+                handle=handle, task_id=task_id, fields=fields, retry_ordinal=retry_ordinal
             )
 
     def record_model_call_error(self, event: dict[str, Any]) -> None:
         """Retain the latest attempt error without closing the logical call."""
+        self._update_model_call(event, finish=False)
+
+    def end_model_call(self, event: dict[str, Any]) -> None:
+        self._update_model_call(event, finish=True)
+
+    def _update_model_call(self, event: dict[str, Any], *, finish: bool) -> None:
+        """Refresh the located model call's fields from ``event``, optionally closing it."""
         session = self._any_session(event)
         if session is None:
             return
         with session.lock:
             if session.closing:
                 return
-            located = self._model_call_for(session, event)
-            if located is not None:
-                located[1].fields = model_call_fields(event)
+            model_call_key = self._existing_model_call_key(session, event)
+            model_call = session.model_calls.get(model_call_key) if model_call_key else None
+            if model_call is None:
+                return
+            model_call.fields = model_call_fields(event)
+            if finish:
+                self._finish_model_call(session, model_call_key)
 
     def start_tool_call(self, event: dict[str, Any]) -> None:
         """Open one privacy-safe Relay tool lifecycle under its task."""
-        task_id = str(event.get("task_id") or "")
+        task_id = _text(event, "task_id")
         session, task = self._task_for(event, start=True)
-        if session is None or task is None:
-            return
-        tool_call_id = str(event.get("tool_call_id") or "")
-        if not tool_call_id:
+        if session is None or task is None or not _text(event, "tool_call_id"):
             return
         identity = self._tool_call_identity(event)
         with session.lock:
-            if session.closing:
+            if not self._admits(session, task, event):
                 return
-            if not self._event_matches_task_turn(task, event):
-                return
-            self._remember_turn(session, task, event)
             key = (task_id, *identity)
             if identity in task.completed_tool_call_ids or key in session.tool_calls:
                 return
@@ -397,51 +351,34 @@ class _Runtime:
         if session is None or task is None:
             return
         outcome = tool_approval_outcome(event)
-        tool_call_id = str(event.get("tool_call_id") or "")
         attribution = "unattributed"
         with session.lock:
-            if session.closing:
+            if session.closing or not self._event_matches_task_turn(task, event):
                 return
-            if not self._event_matches_task_turn(task, event):
-                return
-            if tool_call_id:
+            if _text(event, "tool_call_id"):
                 identity = self._tool_call_identity(event)
                 tool_call = session.tool_calls.get((task.task_id, *identity))
                 if tool_call is None:
-                    matching_keys = self._compatible_tool_call_keys(
-                        session, task.task_id, identity
-                    )
-                    tool_call = (
-                        session.tool_calls[matching_keys[0]]
-                        if len(matching_keys) == 1
-                        else None
-                    )
+                    key = _sole(self._compatible_tool_call_keys(session, task.task_id, identity))
+                    tool_call = session.tool_calls[key] if key is not None else None
                 if tool_call is not None:
                     tool_call.approval_outcome = outcome
                     attribution = "tool_call"
-            self._run_in_task(
-                task,
-                self.relay.scope.event,
-                TOOL_APPROVAL_MARK,
-                handle=task.handle,
-                data={"attribution": attribution, "outcome": outcome},
-                metadata=self._event_metadata(),
+            self._mark(
+                session, task, TOOL_APPROVAL_MARK, {"attribution": attribution, "outcome": outcome}
             )
 
     def record_tool_call(self, event: dict[str, Any]) -> None:
         """Close and count one unique privacy-safe tool lifecycle."""
-        task_id = str(event.get("task_id") or "")
+        task_id = _text(event, "task_id")
         session, task = self._task_for(event, start=False)
         if session is None or task is None:
             return
-        tool_call_id = str(event.get("tool_call_id") or "")
         with session.lock:
-            if session.closing:
+            if not self._admits(session, task, event):
                 return
-            if not self._event_matches_task_turn(task, event):
-                return
-            self._remember_turn(session, task, event)
-            if tool_call_id:
+            tool_call = None
+            if _text(event, "tool_call_id"):
                 observed_identity = self._tool_call_identity(event)
                 if observed_identity in task.completed_tool_call_ids:
                     return
@@ -449,109 +386,67 @@ class _Runtime:
                 tool_call = session.tool_calls.pop((task_id, *identity), None)
                 if tool_call is None:
                     if any(
-                        self._tool_call_identities_are_compatible(
-                            completed_identity,
-                            observed_identity,
-                        )
-                        for completed_identity in task.completed_tool_call_ids
+                        self._tool_call_identities_are_compatible(completed, observed_identity)
+                        for completed in task.completed_tool_call_ids
                     ):
                         return
                     matching_keys = self._compatible_tool_call_keys(
                         session, task_id, observed_identity
                     )
                     if len(matching_keys) > 1:
-                        # Partial context cannot safely choose between
-                        # concurrent calls that reused the provider-local ID.
+                        # Partial context cannot safely choose between concurrent calls
+                        # that reused the provider-local ID.
                         return
                     if matching_keys:
-                        key = matching_keys[0]
-                        identity = key[1:]
-                        tool_call = session.tool_calls.pop(key)
-                task.completed_tool_call_ids.update({
-                    identity,
-                    observed_identity,
-                })
+                        identity = matching_keys[0][1:]
+                        tool_call = session.tool_calls.pop(matching_keys[0])
+                task.completed_tool_call_ids.update({identity, observed_identity})
                 task.tool_call_ids.add(identity)
             else:
                 task.unidentified_tool_calls += 1
-                tool_call = None
             if tool_call is None:
                 tool_call = self._open_tool_call(task, event)
             self._finish_tool_call(task, tool_call, event)
 
     def record_skill_lifecycle(self, event: dict[str, Any]) -> None:
         """Emit one allowlisted skill fact without its local identity."""
-        action = str(event.get("action") or "").strip().lower()
-        if action == "loaded":
-            mark = SKILL_LOAD_MARK
-            fields = skill_load_fields(event)
+        if _text(event, "action").strip().lower() == "loaded":
+            mark, fields = SKILL_LOAD_MARK, skill_load_fields(event)
         else:
-            mark = SKILL_LIFECYCLE_MARK
-            fields = skill_lifecycle_fields(event)
+            mark, fields = SKILL_LIFECYCLE_MARK, skill_lifecycle_fields(event)
         if fields is None:
             return
 
-        session_id = str(event.get("session_id") or "")
-        task_id = str(event.get("task_id") or "")
-        session = self._task_session(
-            event,
-            allow_task_id_fallback=not session_id,
-        )
-        task = session.tasks.get(task_id) if session is not None else None
+        session_id, task_id = _text(event, "session_id"), _text(event, "task_id")
+        session, task = self._task_pair(event, allow_task_id_fallback=not session_id)
         if session is not None:
             if task is None:
                 return
             with session.lock:
-                if session.closing:
-                    return
                 if (
-                    session.tasks.get(task.task_id) is not task
+                    session.closing
+                    or session.tasks.get(task.task_id) is not task
                     or not self._event_matches_task_turn(task, event)
                 ):
                     return
-                self._run_in_task(
-                    task,
-                    self.relay.scope.event,
-                    mark,
-                    handle=task.handle,
-                    data=fields,
-                    metadata=self._event_metadata(),
-                )
+                self._mark(session, task, mark, fields)
             return
         if session_id and task_id:
             return
 
-        self.relay.get_scope_stack()
-        self.relay.scope.event(
-            mark,
-            data=fields,
-            metadata=self._event_metadata(),
+        self._with_scope_stack(
+            self.relay.scope.event, mark, data=fields, metadata=self._event_metadata()
         )
 
-    def end_model_call(self, event: dict[str, Any]) -> None:
-        session = self._any_session(event)
-        if session is None:
-            return
-        with session.lock:
-            if session.closing:
-                return
-            located = self._model_call_for(session, event)
-            if located is None:
-                return
-            model_call_key, model_call = located
-            model_call.fields = model_call_fields(event)
-            self._finish_model_call(session, model_call_key)
-
     def finish_task(self, event: dict[str, Any]) -> None:
         """Close one task scope exactly once with bounded terminal fields."""
-        task_id = str(event.get("task_id") or "")
         session = self._any_session(event)
         if session is None:
             return
         with session.lock:
             if session.closing:
                 return
-            finished = self._finish_task(session, task_id, event)
+            finished = self._finish_task(session, _text(event, "task_id"), event)
         if finished:
             self._flush_and_export("Hermes shared-metrics task flush failed")
 
@@ -559,36 +454,23 @@ class _Runtime:
         session = self._session(event)
         if session is None:
             return
-        failures: list[str] = []
-        with session.lock:
-            if session.closing:
-                return
-            session.closing = True
-            self._abort_tasks(
-                session,
-                {
-                    **event,
-                    "completed": False,
-                    "failed": True,
-                    "interrupted": False,
-                    "turn_exit_reason": "system_aborted",
-                },
-            )
+        if not self._abort_session(
+            session, {**event, **_ABORTED, "completed": False, "interrupted": False}
+        ):
+            return
         try:
             self.relay.subscribers.flush()
         except Exception as exc:
-            failures.append(f"subscriber flush failed: {exc}")
+            logger.warning(
+                "Hermes shared-metrics session %s closed with errors: subscriber flush failed: %s",
+                session.session_id,
+                exc,
+            )
         else:
             self._export()
         with self._sessions_lock:
             if self._sessions.get(session.session_id) is session:
                 self._sessions.pop(session.session_id, None)
-        if failures:
-            logger.warning(
-                "Hermes shared-metrics session %s closed with errors: %s",
-                session.session_id,
-                "; ".join(failures),
-            )
 
     def shutdown(self) -> None:
         with self._sessions_lock:
@@ -600,11 +482,8 @@ class _Runtime:
             return
         self._flush_and_export("Hermes shared-metrics shutdown flush failed")
         self._deregister()
-        # The final export above may have started a send. Give it the same
-        # bounded chance to finish that deactivate() gets — without this a
-        # short-lived CLI process exits immediately and kills the daemon
-        # thread mid-request, which is the common case for the one cadence
-        # this feature has.
+        # The final export may have started a send; give it the same bounded chance
+        # deactivate() gets, or a short-lived CLI exits and kills the daemon thread mid-request.
         self._join_send_thread()
         self._unregister_atexit()
 
@@ -623,18 +502,7 @@ class _Runtime:
         with self._sessions_lock:
             sessions = list(self._sessions.values())
         for session in sessions:
-            with session.lock:
-                if session.closing:
-                    continue
-                session.closing = True
-                self._abort_tasks(
-                    session,
-                    {
-                        "session_id": session.session_id,
-                        "failed": True,
-                        "turn_exit_reason": "system_aborted",
-                    },
-                )
+            self._abort_session(session, {"session_id": session.session_id, **_ABORTED})
         with self._sessions_lock:
             self._sessions.clear()
         with self._task_sessions_lock:
@@ -644,10 +512,10 @@ class _Runtime:
         self._unregister_atexit()
 
     def _join_send_thread(self, timeout: float = 2.0) -> None:
-        """Give an in-flight send a brief chance to finish at exit.
+        """Give an in-flight send a brief, bounded chance to finish at exit.
 
-        Bounded on purpose: pending packages stay in SQLite and go out next run, so blocking
-        shutdown on a slow network is the wrong trade. The daemon thread dies with the process.
+        Pending packages stay in SQLite and go out next run, so blocking on a slow network
+        is the wrong trade; the daemon thread dies with the process.
         """
         with self._send_lock:
             thread = self._send_thread
@@ -659,51 +527,33 @@ class _Runtime:
             logger.debug("Shared-metrics send thread join failed", exc_info=True)
 
     def _session(self, event: dict[str, Any]) -> _MetricsSession | None:
-        session_id = str(event.get("session_id") or "")
         with self._sessions_lock:
-            return self._sessions.get(session_id)
+            return self._sessions.get(_text(event, "session_id"))
 
     def _any_session(self, event: dict[str, Any]) -> _MetricsSession | None:
         """Owner session by task/turn correlation, else by session_id."""
-        return self._task_session(
-            event, allow_task_id_fallback=True
-        ) or self._session(event)
+        return self._task_session(event, allow_task_id_fallback=True) or self._session(event)
 
     def _task_for(
-        self,
-        event: dict[str, Any],
-        *,
-        start: bool,
+        self, event: dict[str, Any], *, start: bool
     ) -> tuple[_MetricsSession | None, _TaskRun | None]:
         """Resolve (session, task) for a task-scoped hook, optionally opening the task."""
-        task_id = str(event.get("task_id") or "")
-        session = self._task_session(event, allow_task_id_fallback=True)
-        task = session.tasks.get(task_id) if session is not None else None
+        session, task = self._task_pair(event, allow_task_id_fallback=True)
         if task is None and start:
             task = self.start_task(event)
             session = self._task_session(event) if task is not None else None
         return session, task
 
-    def _model_call_for(
-        self,
-        session: _MetricsSession,
-        event: dict[str, Any],
-    ) -> tuple[tuple[str, str], _ModelCall] | None:
-        model_call_key = self._existing_model_call_key(session, event)
-        if model_call_key is None:
-            return None
-        model_call = session.model_calls.get(model_call_key)
-        if model_call is None:
-            return None
-        return model_call_key, model_call
+    def _task_pair(
+        self, event: dict[str, Any], **lookup: Any
+    ) -> tuple[_MetricsSession | None, _TaskRun | None]:
+        session = self._task_session(event, **lookup)
+        task = session.tasks.get(_text(event, "task_id")) if session is not None else None
+        return session, task
 
     def _run_scoped(
-        self,
-        session: _MetricsSession,
-        task: _TaskRun | None,
-        callback: Callable[..., Any],
-        *args: Any,
-        **kwargs: Any,
+        self, session: _MetricsSession, task: _TaskRun | None, callback: Callable[..., Any],
+        *args: Any, **kwargs: Any,
     ) -> Any:
         """Run under the task context when the call belongs to a task, else the session."""
         if task is not None:
@@ -711,71 +561,54 @@ class _Runtime:
         return self._run_in_session(session, callback, *args, **kwargs)
 
     def _flush_and_export(self, failure_message: str) -> None:
-        try:
-            self.relay.subscribers.flush()
-        except Exception:
-            logger.warning(failure_message, exc_info=True)
-        else:
+        """Flush the Relay subscriber, then export; a failed flush skips the export."""
+        if self._guarded(failure_message, self._flush_ok):
             self._export()
 
-    def _abort_tasks(self, session: _MetricsSession, base_event: dict[str, Any]) -> None:
-        """Close every open task of a closing session as system-aborted (caller holds the lock)."""
-        for task_id in list(session.tasks):
-            self._finish_task(session, task_id, {**base_event, "task_id": task_id})
-        self._end_pending_model_calls(session, base_event)
+    def _flush_ok(self) -> bool:
+        self.relay.subscribers.flush()
+        return True
+
+    def _abort_session(self, session: _MetricsSession, base_event: dict[str, Any]) -> bool:
+        """Mark the session closing and system-abort its open tasks; False if already closing."""
+        with session.lock:
+            if session.closing:
+                return False
+            session.closing = True
+            for task_id in list(session.tasks):
+                self._finish_task(session, task_id, {**base_event, "task_id": task_id})
+            self._end_pending_model_calls(session, base_event)
+        return True
 
     def _unregister_atexit(self) -> None:
-        try:
+        with contextlib.suppress(Exception):
             atexit.unregister(self.shutdown)
-        except Exception:
-            pass
-
-    @staticmethod
-    def _task_key(event: dict[str, Any]) -> tuple[str, str] | None:
-        return _session_pair(event, "task_id")
 
     def _task_session(
-        self,
-        event: dict[str, Any],
-        *,
-        allow_task_id_fallback: bool = False,
+        self, event: dict[str, Any], *, allow_task_id_fallback: bool = False
     ) -> _MetricsSession | None:
-        session_id = str(event.get("session_id") or "")
-        task_id = str(event.get("task_id") or "")
+        session_id, task_id = _text(event, "session_id"), _text(event, "task_id")
         if not task_id:
             return None
-        task_key = (session_id, task_id) if session_id else None
-        turn_key = self._turn_key(event)
+        turn_key = _session_pair(event, "turn_id")
         with self._task_sessions_lock:
-            if turn_key is not None:
-                owner = self._turn_sessions.get(turn_key)
-                if owner is not None:
-                    return owner
-            if task_key is not None:
-                owner = self._task_sessions.get(task_key)
-                if owner is not None:
-                    return owner
+            owner = self._turn_sessions.get(turn_key) if turn_key is not None else None
+            if owner is None and session_id:
+                owner = self._task_sessions.get((session_id, task_id))
+            if owner is not None:
+                return owner
             if not allow_task_id_fallback:
                 return None
-            candidates: list[_MetricsSession] = []
-            for (_, candidate_task_id), session in self._task_sessions.items():
-                if candidate_task_id != task_id:
-                    continue
-                if not any(candidate is session for candidate in candidates):
-                    candidates.append(session)
-            return candidates[0] if len(candidates) == 1 else None
-
-    @staticmethod
-    def _turn_key(event: dict[str, Any]) -> tuple[str, str] | None:
-        return _session_pair(event, "turn_id")
+            return _sole(
+                session
+                for (_, candidate_task_id), session in self._task_sessions.items()
+                if candidate_task_id == task_id
+            )
 
     def _remember_turn(
-        self,
-        session: _MetricsSession,
-        task: _TaskRun,
-        event: dict[str, Any],
+        self, session: _MetricsSession, task: _TaskRun, event: dict[str, Any]
     ) -> None:
-        turn_id = str(event.get("turn_id") or "")
+        turn_id = _text(event, "turn_id")
         if not turn_id:
             return
         task.turn_ids.add(turn_id)
@@ -785,240 +618,156 @@ class _Runtime:
     @staticmethod
     def _tool_call_identity(event: dict[str, Any]) -> tuple[str, str, str]:
         """Identify one provider-local tool call without exporting its IDs."""
-        return (
-            str(event.get("api_request_id") or ""),
-            str(event.get("turn_id") or ""),
-            str(event.get("tool_call_id") or ""),
-        )
+        return _text(event, "api_request_id"), _text(event, "turn_id"), _text(event, "tool_call_id")
 
     @staticmethod
     def _tool_call_identities_are_compatible(
-        candidate: tuple[str, str, str],
-        observed: tuple[str, str, str],
+        candidate: tuple[str, str, str], observed: tuple[str, str, str]
     ) -> bool:
         """Match partial hook context without crossing known call boundaries."""
         if not observed[2] or candidate[2] != observed[2]:
             return False
         return all(
-            not candidate_value
-            or not observed_value
-            or candidate_value == observed_value
-            for candidate_value, observed_value in zip(
-                candidate[:2],
-                observed[:2],
-                strict=True,
-            )
+            not left or not right or left == right
+            for left, right in zip(candidate[:2], observed[:2], strict=True)
         )
 
     @classmethod
     def _compatible_tool_call_keys(
-        cls,
-        session: _MetricsSession,
-        task_id: str,
-        identity: tuple[str, str, str],
+        cls, session: _MetricsSession, task_id: str, identity: tuple[str, str, str]
     ) -> list[tuple[str, str, str, str]]:
         return [
             key
             for key in session.tool_calls
-            if key[0] == task_id
-            and cls._tool_call_identities_are_compatible(key[1:], identity)
+            if key[0] == task_id and cls._tool_call_identities_are_compatible(key[1:], identity)
         ]
 
     @staticmethod
-    def _event_matches_task_turn(
-        task: _TaskRun,
-        event: dict[str, Any],
-    ) -> bool:
+    def _event_matches_task_turn(task: _TaskRun, event: dict[str, Any]) -> bool:
         """Reject delayed hooks from a prior run that reused the task ID."""
-        turn_id = str(event.get("turn_id") or "")
+        turn_id = _text(event, "turn_id")
         if not turn_id:
             return True
         if turn_id in task.retired_turn_ids:
             return False
         return not task.turn_ids or turn_id in task.turn_ids
 
-    def _approval_task(
+    def _admits(
         self,
+        session: _MetricsSession,
+        task: _TaskRun,
         event: dict[str, Any],
+        *,
+        current: bool = False,
+    ) -> bool:
+        """Whether ``event`` may act on ``task`` (caller holds ``session.lock``).
+
+        Rejects closing sessions and stale turns; with ``current`` also requires ``task`` to
+        still be the session's live run for its ID. Admitted events have their turn remembered.
+        """
+        if session.closing or not self._event_matches_task_turn(task, event):
+            return False
+        if current and session.tasks.get(task.task_id) is not task:
+            return False
+        self._remember_turn(session, task, event)
+        return True
+
+    def _approval_task(
+        self, event: dict[str, Any]
     ) -> tuple[_MetricsSession | None, _TaskRun | None]:
         """Resolve approval correlation without guessing across ambiguous turns."""
         active = relay_runtime.active_turn()
         if active is not None:
-            correlated = {
-                **event,
-                "session_id": active.lease.session_id,
-                "task_id": active.task_id,
-            }
-            session = self._task_session(correlated)
-            task = session.tasks.get(active.task_id) if session is not None else None
+            session, task = self._task_pair(
+                {**event, "session_id": active.lease.session_id, "task_id": active.task_id}
+            )
             if task is not None:
                 return session, task
 
-        session = self._task_session(event)
-        task_id = str(event.get("task_id") or "")
-        task = session.tasks.get(task_id) if session is not None else None
+        session, task = self._task_pair(event)
         if task is not None:
             return session, task
 
-        turn_id = str(event.get("turn_id") or "")
+        turn_id = _text(event, "turn_id")
         if not turn_id:
             return None, None
         with self._task_sessions_lock:
-            candidates = [
+            session = _sole(
                 candidate
-                for (
-                    candidate_session_id,
-                    candidate_turn_id,
-                ), candidate in self._turn_sessions.items()
-                if candidate_turn_id == turn_id
-                and self._sessions.get(candidate_session_id) is candidate
-            ]
-        unique_sessions = {id(candidate): candidate for candidate in candidates}
-        if len(unique_sessions) != 1:
-            return None, None
-        session = next(iter(unique_sessions.values()))
-        matching_tasks = [
-            candidate
-            for candidate in session.tasks.values()
-            if turn_id in candidate.turn_ids
-        ]
-        if len(matching_tasks) != 1:
-            return None, None
-        return session, matching_tasks[0]
+                for (owner_id, candidate_turn_id), candidate in self._turn_sessions.items()
+                if candidate_turn_id == turn_id and self._sessions.get(owner_id) is candidate
+            )
+        task = (
+            _sole(task for task in session.tasks.values() if turn_id in task.turn_ids)
+            if session is not None
+            else None
+        )
+        return (None, None) if task is None else (session, task)
 
-    def _open_tool_call(
-        self,
-        task: _TaskRun,
-        event: dict[str, Any],
-    ) -> _ToolCall:
+    def _open_tool_call(self, task: _TaskRun, event: dict[str, Any]) -> _ToolCall:
         handle = self._run_in_task(
-            task,
-            self.relay.tools.call,
-            TOOL_CALL_SCOPE,
-            {},
-            handle=task.handle,
-            metadata=self._event_metadata(),
-        )
-        return _ToolCall(
-            handle=handle,
-            task_id=task.task_id,
-            category=tool_category(event),
-            started_ns=monotonic_ns(),
+            task, self.relay.tools.call, TOOL_CALL_SCOPE, {},
+            handle=task.handle, metadata=self._event_metadata(),
         )
+        return _ToolCall(handle, task.task_id, tool_category(event), monotonic_ns())
 
     def _finish_tool_call(
-        self,
-        task: _TaskRun,
-        tool_call: _ToolCall,
-        event: dict[str, Any],
+        self, task: _TaskRun, tool_call: _ToolCall, event: dict[str, Any]
     ) -> None:
         fields = tool_terminal_fields(
             event,
             category=tool_call.category,
             approval_outcome=tool_call.approval_outcome,
-            fallback_duration_ms=max(
-                0,
-                (monotonic_ns() - tool_call.started_ns) // 1_000_000,
-            ),
+            fallback_duration_ms=max(0, (monotonic_ns() - tool_call.started_ns) // 1_000_000),
+        )
+        self._guarded(
+            "Hermes shared-metrics tool call close failed",
+            self._run_in_task, task, self.relay.tools.call_end, tool_call.handle, fields,
+            metadata=self._event_metadata(),
         )
-        try:
-            self._run_in_task(
-                task,
-                self.relay.tools.call_end,
-                tool_call.handle,
-                fields,
-                metadata=self._event_metadata(),
-            )
-        except Exception:
-            logger.warning(
-                "Hermes shared-metrics tool call close failed",
-                exc_info=True,
-            )
 
     def _end_pending_tool_calls(
-        self,
-        session: _MetricsSession,
-        task: _TaskRun,
-        event: dict[str, Any],
+        self, session: _MetricsSession, task: _TaskRun, event: dict[str, Any]
     ) -> None:
-        pending_keys = [key for key in session.tool_calls if key[0] == task.task_id]
         task_outcome, _, _ = task_terminal_state(event)
-        status = {
-            "cancelled": "cancelled",
-            "timed_out": "timeout",
-        }.get(task_outcome, "error")
-        for key in pending_keys:
-            tool_call = session.tool_calls.pop(key, None)
-            if tool_call is not None:
-                self._finish_tool_call(task, tool_call, {**event, "status": status})
+        status = {"cancelled": "cancelled", "timed_out": "timeout"}.get(task_outcome, "error")
+        for key in [key for key in session.tool_calls if key[0] == task.task_id]:
+            self._finish_tool_call(task, session.tool_calls.pop(key), {**event, "status": status})
 
-    def _finish_model_call(
-        self,
-        session: _MetricsSession,
-        model_call_key: tuple[str, str],
-    ) -> None:
+    def _finish_model_call(self, session: _MetricsSession, model_call_key: tuple[str, str]) -> None:
         model_call = session.model_calls.pop(model_call_key, None)
         if model_call is None:
             return
-        try:
-            self._run_scoped(
-                session,
-                session.tasks.get(model_call.task_id),
-                self.relay.llm.call_end,
-                model_call.handle,
-                model_call.fields,
-                metadata=self._event_metadata(),
-            )
-        except Exception:
-            logger.warning(
-                "Hermes shared-metrics model call close failed", exc_info=True
-            )
+        self._guarded(
+            "Hermes shared-metrics model call close failed",
+            self._run_scoped, session, session.tasks.get(model_call.task_id),
+            self.relay.llm.call_end, model_call.handle, model_call.fields,
+            metadata=self._event_metadata(),
+        )
 
-    def _end_pending_model_calls(
-        self,
-        session: _MetricsSession,
-        event: dict[str, Any],
-    ) -> None:
-        task_id = str(event.get("task_id") or "")
-        model_call_keys = [
-            model_call_key
-            for model_call_key, model_call in session.model_calls.items()
-            if not task_id or model_call.task_id == task_id
-        ]
-        for model_call_key in model_call_keys:
-            self._finish_model_call(session, model_call_key)
+    def _end_pending_model_calls(self, session: _MetricsSession, event: dict[str, Any]) -> None:
+        task_id = _text(event, "task_id")
+        pending = [k for k, c in session.model_calls.items() if not task_id or c.task_id == task_id]
+        for key in pending:
+            self._finish_model_call(session, key)
 
     @staticmethod
-    def _new_model_call_key(event: dict[str, Any]) -> tuple[str, str] | None:
-        request_id = str(event.get("api_request_id") or "")
+    def _existing_model_call_key(
+        session: _MetricsSession, event: dict[str, Any]
+    ) -> tuple[str, str] | None:
+        """(task_id, request_id) of an open call; a task-less event may match by request alone."""
+        request_id = _text(event, "api_request_id")
         if not request_id:
             return None
-        return str(event.get("task_id") or ""), request_id
-
-    @classmethod
-    def _existing_model_call_key(
-        cls,
-        session: _MetricsSession,
-        event: dict[str, Any],
-    ) -> tuple[str, str] | None:
-        key = cls._new_model_call_key(event)
-        if key is None:
-            return None
+        key = (_text(event, "task_id"), request_id)
         if key in session.model_calls:
             return key
         if key[0]:
             return None
-        candidates = [
-            candidate for candidate in session.model_calls if candidate[1] == key[1]
-        ]
+        candidates = [candidate for candidate in session.model_calls if candidate[1] == request_id]
         return candidates[0] if len(candidates) == 1 else None
 
-    def _finish_task(
-        self,
-        session: _MetricsSession,
-        task_id: str,
-        event: dict[str, Any],
-    ) -> bool:
+    def _finish_task(self, session: _MetricsSession, task_id: str, event: dict[str, Any]) -> bool:
         task = session.tasks.get(task_id)
         if task is None:
             return False
@@ -1032,16 +781,11 @@ class _Runtime:
             retry_count=task.retry_count,
         )
         try:
-            self._run_in_task(
-                task,
-                relay_runtime.pop_relay_scope,
-                self.relay,
-                task.handle,
-                output=fields,
-                metadata=self._event_metadata(),
+            self._guarded(
+                "Hermes shared-metrics task close failed",
+                self._run_in_task, task, relay_runtime.pop_relay_scope, self.relay, task.handle,
+                output=fields, metadata=self._event_metadata(),
             )
-        except Exception:
-            logger.warning("Hermes shared-metrics task close failed", exc_info=True)
         finally:
             session.tasks.pop(task_id, None)
             session.retired_turn_ids.extend(task.turn_ids)
@@ -1057,27 +801,18 @@ class _Runtime:
 
     def _export(self) -> None:
         exported = self._safe(self.subscriber.store.create_and_export_package_if_due)
-        # Sending is opt-in and must never delay the caller: _export runs on
-        # finish_task, which is the user's interactive path. Errors inside the
-        # sender are already swallowed there; the thread is about latency, not
-        # correctness.
+        # Sending must never delay the caller: _export runs on finish_task, the user's
+        # interactive path. The thread is about latency, not correctness.
         if exported is not None:
             self._safe(self._send_exported_packages)
 
     def _observe_send_consent(self, send_enabled: bool) -> None:
-        """Reconcile consent windows with the observed config state.
-
-        Failures must never break the export hook, but they are logged at warning rather than debug:
-        silently failing to close a consent window is a privacy-relevant event, not routine
-        bookkeeping.
-        """
-        try:
-            _reconcile_store_consent(self.subscriber.store, send_enabled)
-        except Exception:
-            logger.warning(
-                "Unable to record a shared-metrics consent transition",
-                exc_info=True,
-            )
+        """Reconcile consent windows with observed config; failures never break the export
+        hook but log at warning (an unclosed consent window is privacy-relevant)."""
+        self._guarded(
+            "Unable to record a shared-metrics consent transition",
+            _reconcile_store_consent, self.subscriber.store, send_enabled,
+        )
 
     def _send_exported_packages(self) -> None:
         try:
@@ -1086,20 +821,14 @@ class _Runtime:
             logger.debug("Unable to read shared-metrics send policy", exc_info=True)
             return
 
-        # Observe the consent EDGE before deciding whether to send. Recording
-        # revocation inside the send loop (as an earlier fix did) can never
-        # work: the dominant case is the user turning sending off while no
-        # pass is running, and then this method returns below without ever
-        # constructing a sender. The window has to close on the transition,
-        # not on the next transmission that by definition will not happen.
+        # Observe the consent EDGE before deciding whether to send: the dominant revocation
+        # case is "sending turned off while no pass is running", invisible to the send loop.
         self._observe_send_consent(resolved.send)
-
         if not resolved.send:
             return
 
         with self._send_lock:
-            # One in-flight pass per process. A queued second pass would add
-            # nothing: the next hook fire picks up whatever is still pending.
+            # One in-flight pass per process; the next hook fire picks up what is pending.
             if self._send_thread is not None and self._send_thread.is_alive():
                 return
             thread = threading.Thread(
@@ -1112,23 +841,15 @@ class _Runtime:
             thread.start()
 
     def _run_send_pass(self, endpoint: str) -> None:
-        from hermes_cli.observability.shared_metrics_sender import (
-            SharedMetricsSender,
-        )
+        from hermes_cli.observability.shared_metrics_sender import SharedMetricsSender
 
         def still_consented() -> bool:
             """Re-read consent so revoking `send` stops an in-flight pass."""
             resolved = _resolved_send_config()
             return resolved.send and resolved.endpoint == endpoint
 
-        try:
-            SharedMetricsSender(
-                self.subscriber.store,
-                endpoint,
-                consent_check=still_consented,
-            ).send_pending()
-        except Exception:
-            logger.warning("Shared-metrics send pass failed", exc_info=True)
+        sender = SharedMetricsSender(self.subscriber.store, endpoint, consent_check=still_consented)
+        self._guarded("Shared-metrics send pass failed", sender.send_pending)
 
     def _event_metadata(self) -> dict[str, str]:
         return {
@@ -1137,28 +858,29 @@ class _Runtime:
         }
 
     @staticmethod
-    def _safe(callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+    def _guarded(message: str, callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+        """Run ``callback``; log-and-swallow any exception, returning None."""
         try:
             return callback(*args, **kwargs)
         except Exception:
-            logger.warning("Hermes shared metrics operation failed", exc_info=True)
+            logger.warning(message, exc_info=True)
             return None
 
+    @classmethod
+    def _safe(cls, callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+        return cls._guarded("Hermes shared metrics operation failed", callback, *args, **kwargs)
+
 
 def _resolved_send_config():
     """Resolve the opt-in send policy from the read-only config snapshot."""
     from hermes_cli.config import read_raw_config_readonly
-    from hermes_cli.observability.shared_metrics_send_config import (
-        resolve_send_config,
-    )
+    from hermes_cli.observability.shared_metrics_send_config import resolve_send_config
 
     return resolve_send_config(read_raw_config_readonly() or {})
 
 
 def _reconcile_store_consent(store: SharedMetricsStore, send_enabled: bool) -> None:
-    from hermes_cli.observability.shared_metrics_sender import (
-        reconcile_send_consent,
-    )
+    from hermes_cli.observability.shared_metrics_sender import reconcile_send_consent
     from hermes_cli.sqlite_util import write_txn
 
     with store._connection() as connection:
@@ -1172,22 +894,16 @@ def enabled() -> bool:
     try:
         from hermes_cli.config import read_raw_config_readonly
 
-        # Collection consent is profile-owned. Managed config overlays may
-        # control runtime policy, but cannot opt a profile into or out of
-        # shared metrics. Read-only fast path: this gate runs 2-3x per agent
-        # turn, and the mutable read_raw_config() paid a full config deepcopy
-        # on every call.
+        # Collection consent is profile-owned: managed overlays cannot opt a profile in or
+        # out. Read-only fast path — this gate runs 2-3x per agent turn and the mutable
+        # read_raw_config() paid a full config deepcopy on every call.
         config = read_raw_config_readonly() or {}
     except Exception:
         logger.debug("Unable to read Hermes shared-metrics policy", exc_info=True)
-        value = False
-    else:
-        telemetry = config.get("telemetry") if isinstance(config, dict) else None
-        shared_metrics = (
-            telemetry.get("shared_metrics") if isinstance(telemetry, dict) else None
-        )
-        value = isinstance(shared_metrics, dict) and shared_metrics.get("enabled") is True
-    if value:
+        config = None
+    telemetry = config.get("telemetry") if isinstance(config, dict) else None
+    shared_metrics = telemetry.get("shared_metrics") if isinstance(telemetry, dict) else None
+    if isinstance(shared_metrics, dict) and shared_metrics.get("enabled") is True:
         return True
     with _RUNTIME_LOCK:
         runtime = _RUNTIMES.pop(profile_key, None)
@@ -1206,14 +922,10 @@ _consent_reconcile_done = False
 def _reconcile_send_consent_once() -> None:
     """Reconcile consent windows with config, once per process.
 
-    Runs BEFORE and INDEPENDENT of the collection gate — that placement is the fix for the round-5
-    D1 leak, where the only idle-path consent observer sat behind ``handles_hook()`` and became dead
-    code the moment ``enabled: false`` was set. A user with collection off still gets their send-
-    consent windows reconciled here.
-
-    Skipped only when there is no store on disk AND consent is off: with no store there are no
-    packages, so there is nothing a window could protect, and creating ``~/.hermes/telemetry`` for
-    every fully-disabled user would be a behaviour change in the wrong direction.
+    Runs BEFORE and INDEPENDENT of the collection gate, so a user with ``enabled: false``
+    still gets send-consent windows reconciled. Skipped only when there is no store on disk
+    AND consent is off: nothing to protect, and creating ``~/.hermes/telemetry`` for every
+    fully-disabled user would be the wrong behaviour change.
     """
     global _consent_reconcile_done
     if _consent_reconcile_done:
@@ -1224,28 +936,20 @@ def _reconcile_send_consent_once() -> None:
         from hermes_constants import get_hermes_home
 
         resolved = _resolved_send_config()
-        # Probe for an existing store WITHOUT constructing one: the
-        # constructor creates the directory and schema as a side effect,
-        # which round 6 caught making this skip dead code — every
-        # fully-disabled user was getting a ~/.hermes/telemetry directory.
-        default_path = (
-            get_hermes_home() / "telemetry" / "shared_metrics" / "metrics.sqlite3"
-        )
+        # Probe WITHOUT constructing a store: the constructor creates the directory and
+        # schema as a side effect, which would make the skip below dead code.
+        default_path = get_hermes_home() / "telemetry" / "shared_metrics" / "metrics.sqlite3"
         if not resolved.send and not default_path.exists():
             return
         _reconcile_store_consent(SharedMetricsStore(), resolved.send)
     except Exception:
-        logger.warning(
-            "Unable to reconcile shared-metrics send consent", exc_info=True
-        )
+        logger.warning("Unable to reconcile shared-metrics send consent", exc_info=True)
 
 
 def observe_lifecycle(hook_name: str, **kwargs: Any) -> None:
     """Project one Hermes lifecycle event into the core Relay integration."""
     _reconcile_send_consent_once()
-    if not handles_hook(hook_name):
-        return
-    if not relay_runtime.relay_instrumentation_enabled():
+    if not handles_hook(hook_name) or not relay_runtime.relay_instrumentation_enabled():
         return
     runtime = _get_runtime()
     if runtime is None:
@@ -1253,17 +957,13 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None:
     try:
         _HOOK_HANDLERS[hook_name](runtime, kwargs)
     except Exception:
-        logger.warning(
-            "Hermes shared metrics hook failed: %s", hook_name, exc_info=True
-        )
+        logger.warning("Hermes shared metrics hook failed: %s", hook_name, exc_info=True)
 
 
 def _with_runtime_toolset(event: dict[str, Any]) -> dict[str, Any]:
     """Attach the toolset already declared by Hermes's runtime registry."""
-    if event.get("toolset"):
-        return event
-    tool_name = str(event.get("tool_name") or "")
-    if not tool_name:
+    tool_name = _text(event, "tool_name")
+    if event.get("toolset") or not tool_name:
         return event
     try:
         from model_tools import get_toolset_for_tool
@@ -1275,7 +975,7 @@ def _with_runtime_toolset(event: dict[str, Any]) -> dict[str, Any]:
 
 
 def _close_child_session(runtime: _Runtime, kwargs: dict[str, Any]) -> None:
-    child_session_id = str(kwargs.get("child_session_id") or "")
+    child_session_id = _text(kwargs, "child_session_id")
     if child_session_id:
         runtime.close_session({"session_id": child_session_id})
 
@@ -1295,13 +995,10 @@ _HOOK_HANDLERS: dict[str, Callable[[_Runtime, dict[str, Any]], Any]] = {
     "on_session_finalize": lambda rt, kw: rt.close_session(kw),
     "on_session_reset": lambda rt, kw: rt.close_session(kw),
 }
-assert frozenset(_HOOK_HANDLERS) == HANDLED_HOOKS
+HANDLED_HOOKS = frozenset(_HOOK_HANDLERS)
 
 
-def _prepare_core_session(
-    host: relay_runtime.RelayRuntime,
-    context: dict[str, Any],
-) -> None:
+def _prepare_core_session(host: relay_runtime.RelayRuntime, context: dict[str, Any]) -> None:
     """Prepare the profile subscriber before the coordinator opens a scope."""
     del context
     if host.profile_key == relay_runtime.current_profile_key() and enabled():
@@ -1309,27 +1006,20 @@ def _prepare_core_session(
 
 
 def start_task_run(
-    *,
-    session_id: str,
-    task_id: str,
-    platform: str,
-    parent_session_id: str = "",
+    *, session_id: str, task_id: str, platform: str, parent_session_id: str = ""
 ) -> None:
     """Start task metrics at the outer Hermes execution boundary."""
     if not enabled():
         return
     runtime = _get_runtime(retry_failed=True)
-    if runtime is None:
-        return
-    runtime._safe(
-        runtime.start_task,
-        {
-            "session_id": session_id,
-            "task_id": task_id,
-            "platform": platform,
-            "parent_session_id": parent_session_id,
-        },
-    )
+    if runtime is not None:
+        runtime._safe(
+            runtime.start_task,
+            {
+                "session_id": session_id, "task_id": task_id, "platform": platform,
+                "parent_session_id": parent_session_id,
+            },
+        )
 
 
 def finish_task_run(
@@ -1351,43 +1041,33 @@ def finish_task_run(
     interrupted = terminal.get("interrupted") is True
     completed = terminal.get("completed") is True
     failed = terminal.get("failed") is True
-    reason = str(
-        terminal.get("turn_exit_reason") or terminal.get("failure_reason") or ""
-    )
+    reason = str(terminal.get("turn_exit_reason") or terminal.get("failure_reason") or "")
     if error is not None:
-        interrupted = isinstance(error, (KeyboardInterrupt, InterruptedError)) or (
-            type(error).__name__ == "CancelledError"
+        interrupted = (
+            isinstance(error, (KeyboardInterrupt, InterruptedError))
+            or type(error).__name__ == "CancelledError"
         )
-        timed_out = isinstance(error, TimeoutError)
         completed = False
         failed = not interrupted
         if interrupted:
             reason = "interrupted_by_user"
-        elif timed_out:
-            reason = "timed_out"
         else:
-            reason = "system_aborted"
+            reason = "timed_out" if isinstance(error, TimeoutError) else "system_aborted"
     elif not reason:
         reason = "failed" if failed else "unknown"
 
     runtime._safe(
         runtime.finish_task,
         {
-            "session_id": session_id,
-            "task_id": task_id,
-            "platform": platform,
-            "completed": completed,
-            "failed": failed,
-            "interrupted": interrupted,
+            "session_id": session_id, "task_id": task_id, "platform": platform,
+            "completed": completed, "failed": failed, "interrupted": interrupted,
             "turn_exit_reason": reason,
         },
     )
 
 
 def _get_runtime(
-    *,
-    retry_failed: bool = False,
-    host: relay_runtime.RelayRuntime | None = None,
+    *, retry_failed: bool = False, host: relay_runtime.RelayRuntime | None = None
 ) -> _Runtime | None:
     profile_key = relay_runtime.current_profile_key()
     with _RUNTIME_LOCK:
@@ -1397,9 +1077,9 @@ def _get_runtime(
                 return runtime
             runtime.deactivate()
             _RUNTIMES.pop(profile_key, None)
-        if runtime is _RUNTIME_FAILED and not retry_failed:
-            return None
-        if runtime is _RUNTIME_FAILED:
+        elif runtime is _RUNTIME_FAILED:
+            if not retry_failed:
+                return None
             _RUNTIMES.pop(profile_key, None)
         try:
             runtime = _Runtime(host=host)
@@ -1412,8 +1092,7 @@ def _get_runtime(
 
 
 relay_runtime.SESSION_COORDINATOR.register_session_initializer(
-    SUBSCRIBER_NAME,
-    _prepare_core_session,
+    SUBSCRIBER_NAME, _prepare_core_session
 )
 
 
diff --git a/hermes_cli/observability/shared_metrics.py b/hermes_cli/observability/shared_metrics.py
index 33e6552be2..6953701cc2 100644
--- a/hermes_cli/observability/shared_metrics.py
+++ b/hermes_cli/observability/shared_metrics.py
@@ -32,6 +32,8 @@ _SCHEMA_BUSY_TIMEOUT_MS = 5_000
 _LOCAL_HISTORY_RETENTION_DAYS = 30
 _ACTIVE_INSTALL_STATE_KEY = "client_active_recorded_at"
 _ACTIVE_INSTALL_INTERVAL = timedelta(hours=24)
+# Column order of the client resource in every counter_aggregates statement.
+_RESOURCE_COLUMNS = ("hermes_version", "os_family", "architecture", "install_method")
 
 logger = logging.getLogger(__name__)
 
@@ -44,6 +46,25 @@ def _isoformat(value: datetime) -> str:
     return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
 
 
+def _compact_json(value: Any) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"))
+
+
+def _row_resource(row: sqlite3.Row) -> dict[str, str]:
+    return {column: row[column] for column in _RESOURCE_COLUMNS}
+
+
+def _ensure_private(path: Path, mode: int) -> None:
+    if mode == 0o700:
+        path.mkdir(parents=True, exist_ok=True, mode=mode)
+    else:
+        path.touch(mode=mode, exist_ok=True)
+    try:
+        path.chmod(mode)
+    except OSError:
+        pass
+
+
 class SharedMetricsStore:
     """Persist allowlisted counters and export immutable delta packages."""
 
@@ -55,16 +76,12 @@ class SharedMetricsStore:
         root = get_hermes_home() / "telemetry" / "shared_metrics"
         self.database_path = database_path or root / "metrics.sqlite3"
         self.outbox_directory = outbox_directory or root / "outbox"
-        self._ensure_private_directory(self.database_path.parent)
-        self._ensure_private_directory(self.outbox_directory)
-        self._ensure_private_file(self.database_path)
+        _ensure_private(self.database_path.parent, 0o700)
+        _ensure_private(self.outbox_directory, 0o700)
+        _ensure_private(self.database_path, 0o600)
         self._ensure_schema()
 
-    def record_model_call(
-        self,
-        dimensions: dict[str, str],
-        resource: dict[str, str],
-    ) -> None:
+    def record_model_call(self, dimensions: dict[str, str], resource: dict[str, str]) -> None:
         """Increment the terminal model-call counter for the current UTC day."""
         self.record_counter(MODEL_ROUTE_METRIC, dimensions, resource)
 
@@ -73,72 +90,50 @@ class SharedMetricsStore:
         dimensions: dict[str, str] = {}
         self._validate_counter(CLIENT_ACTIVE_METRIC, dimensions, resource)
         now = _utc_now()
-        with self._connection() as connection:
-            with write_txn(connection):
-                row = connection.execute(
-                    "SELECT value FROM telemetry_state WHERE key = ?",
-                    (_ACTIVE_INSTALL_STATE_KEY,),
-                ).fetchone()
-                if row is not None:
-                    last_recorded = self._parse_state_timestamp(row["value"])
-                    if last_recorded is not None and last_recorded > now:
-                        # A wall-clock correction must not suppress activity until
-                        # the stale future timestamp plus another full interval.
-                        connection.execute(
-                            """
-                            UPDATE telemetry_state
-                            SET value = ?
-                            WHERE key = ?
-                            """,
-                            (_isoformat(now), _ACTIVE_INSTALL_STATE_KEY),
-                        )
-                        return False
-                    if (
-                        last_recorded is not None
-                        and now < last_recorded + _ACTIVE_INSTALL_INTERVAL
-                    ):
-                        return False
+        with self._write() as connection:
+            row = connection.execute(
+                "SELECT value FROM telemetry_state WHERE key = ?",
+                (_ACTIVE_INSTALL_STATE_KEY,),
+            ).fetchone()
+            last_recorded = self._parse_state_timestamp(row["value"]) if row is not None else None
+            if last_recorded is not None:
+                if last_recorded > now:
+                    # A wall-clock correction must not suppress activity until the
+                    # stale future timestamp plus another full interval.
+                    connection.execute(
+                        "UPDATE telemetry_state SET value = ? WHERE key = ?",
+                        (_isoformat(now), _ACTIVE_INSTALL_STATE_KEY),
+                    )
+                    return False
+                if now < last_recorded + _ACTIVE_INSTALL_INTERVAL:
+                    return False
 
-                self._install_id(connection)
-                self._record_counter_in_transaction(
-                    connection,
-                    CLIENT_ACTIVE_METRIC,
-                    dimensions,
-                    resource,
-                    period_start=now.date().isoformat(),
-                )
-                connection.execute(
-                    """
-                    INSERT INTO telemetry_state(key, value)
-                    VALUES (?, ?)
-                    ON CONFLICT(key) DO UPDATE SET value = excluded.value
-                    """,
-                    (_ACTIVE_INSTALL_STATE_KEY, _isoformat(now)),
-                )
+            self._install_id(connection)
+            self._record_counter_in_transaction(
+                connection, CLIENT_ACTIVE_METRIC, dimensions, resource,
+                period_start=now.date().isoformat(),
+            )
+            connection.execute(
+                "INSERT INTO telemetry_state(key, value) VALUES (?, ?)"
+                " ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+                (_ACTIVE_INSTALL_STATE_KEY, _isoformat(now)),
+            )
         return True
 
     def record_counter(
-        self,
-        metric_name: str,
-        dimensions: dict[str, str],
-        resource: dict[str, str],
+        self, metric_name: str, dimensions: dict[str, str], resource: dict[str, str]
     ) -> None:
         """Increment one allowlisted counter for the current UTC day."""
         self._validate_counter(metric_name, dimensions, resource)
         with self._connection() as connection:
             self._record_counter_in_transaction(
-                connection,
-                metric_name,
-                dimensions,
-                resource,
+                connection, metric_name, dimensions, resource,
                 period_start=_utc_now().date().isoformat(),
             )
 
     @staticmethod
     def _validate_counter(
-        metric_name: str,
-        dimensions: dict[str, str],
-        resource: dict[str, str],
+        metric_name: str, dimensions: dict[str, str], resource: dict[str, str]
     ) -> None:
         if metric_name not in COUNTER_METRICS:
             raise ValueError(f"Unsupported shared metric: {metric_name}")
@@ -156,50 +151,29 @@ class SharedMetricsStore:
         *,
         period_start: str,
     ) -> None:
-        dimensions_json = json.dumps(
-            dimensions,
-            sort_keys=True,
-            separators=(",", ":"),
-        )
         connection.execute(
             """
             INSERT INTO counter_aggregates(
-                period_start,
-                metric_name,
-                hermes_version,
-                os_family,
-                architecture,
-                install_method,
-                dimensions_json,
-                value,
-                packaged_value
+                period_start, metric_name, hermes_version, os_family, architecture,
+                install_method, dimensions_json, value, packaged_value
             ) VALUES (?, ?, ?, ?, ?, ?, ?, 1, 0)
             ON CONFLICT(
-                period_start,
-                metric_name,
-                hermes_version,
-                os_family,
-                architecture,
-                install_method,
-                dimensions_json
+                period_start, metric_name, hermes_version, os_family, architecture,
+                install_method, dimensions_json
             )
             DO UPDATE SET value = value + 1
             """,
             (
                 period_start,
                 metric_name,
-                resource["hermes_version"],
-                resource["os_family"],
-                resource["architecture"],
-                resource["install_method"],
-                dimensions_json,
+                *(resource[column] for column in _RESOURCE_COLUMNS),
+                _compact_json(dimensions),
             ),
         )
 
     def create_and_export_package(self) -> list[Path]:
         """Commit one pending delta package, then atomically export the outbox."""
-        pending_periods = self._pending_period_count()
-        for _ in range(pending_periods):
+        for _ in range(self._pending_period_count()):
             if self._create_package() is None:
                 break
         return self._export_and_prune()
@@ -214,10 +188,7 @@ class SharedMetricsStore:
         try:
             self._prune_expired_history()
         except Exception:
-            logger.warning(
-                "Unable to prune expired shared-metrics history",
-                exc_info=True,
-            )
+            logger.warning("Unable to prune expired shared-metrics history", exc_info=True)
         return exported
 
     def counter_snapshot(self) -> list[dict[str, Any]]:
@@ -226,36 +197,19 @@ class SharedMetricsStore:
             rows = connection.execute(
                 """
                 SELECT
-                    period_start,
-                    metric_name,
-                    hermes_version,
-                    os_family,
-                    architecture,
-                    install_method,
-                    dimensions_json,
-                    value,
-                    packaged_value
+                    period_start, metric_name, hermes_version, os_family, architecture,
+                    install_method, dimensions_json, value, packaged_value
                 FROM counter_aggregates
                 ORDER BY
-                    period_start,
-                    hermes_version,
-                    os_family,
-                    architecture,
-                    install_method,
-                    metric_name,
-                    dimensions_json
+                    period_start, hermes_version, os_family, architecture, install_method,
+                    metric_name, dimensions_json
                 """
             ).fetchall()
         return [
             {
                 "period_start": row["period_start"],
                 "metric_name": row["metric_name"],
-                "resource": {
-                    "hermes_version": row["hermes_version"],
-                    "os_family": row["os_family"],
-                    "architecture": row["architecture"],
-                    "install_method": row["install_method"],
-                },
+                "resource": _row_resource(row),
                 "dimensions": json.loads(row["dimensions_json"]),
                 "value": row["value"],
                 "packaged_value": row["packaged_value"],
@@ -265,14 +219,9 @@ class SharedMetricsStore:
 
     @contextmanager
     def _connection(
-        self,
-        *,
-        busy_timeout_ms: int = _BUSY_TIMEOUT_MS,
+        self, *, busy_timeout_ms: int = _BUSY_TIMEOUT_MS
     ) -> Iterator[sqlite3.Connection]:
-        connection = sqlite3.connect(
-            self.database_path,
-            timeout=busy_timeout_ms / 1000,
-        )
+        connection = sqlite3.connect(self.database_path, timeout=busy_timeout_ms / 1000)
         try:
             connection.row_factory = sqlite3.Row
             connection.execute(f"PRAGMA busy_timeout={busy_timeout_ms}")
@@ -281,27 +230,17 @@ class SharedMetricsStore:
         finally:
             connection.close()
 
-    @staticmethod
-    def _ensure_private_directory(path: Path) -> None:
-        path.mkdir(parents=True, exist_ok=True, mode=0o700)
-        try:
-            path.chmod(0o700)
-        except OSError:
-            pass
-
-    @staticmethod
-    def _ensure_private_file(path: Path) -> None:
-        path.touch(mode=0o600, exist_ok=True)
-        try:
-            path.chmod(0o600)
-        except OSError:
-            pass
+    @contextmanager
+    def _write(self, *, busy_timeout_ms: int = _BUSY_TIMEOUT_MS) -> Iterator[sqlite3.Connection]:
+        """A connection inside one write transaction."""
+        with self._connection(busy_timeout_ms=busy_timeout_ms) as connection:
+            with write_txn(connection):
+                yield connection
 
     def _ensure_schema(self) -> None:
-        with self._connection(busy_timeout_ms=_SCHEMA_BUSY_TIMEOUT_MS) as connection:
-            # Serialize first-run creation and upgrades across Hermes processes.
-            with write_txn(connection):
-                self._ensure_schema_in_transaction(connection)
+        # Serialize first-run creation and upgrades across Hermes processes.
+        with self._write(busy_timeout_ms=_SCHEMA_BUSY_TIMEOUT_MS) as connection:
+            self._ensure_schema_in_transaction(connection)
 
     @staticmethod
     def _ensure_schema_in_transaction(connection: sqlite3.Connection) -> None:
@@ -340,11 +279,8 @@ class SharedMetricsStore:
         SharedMetricsStore._add_send_columns(connection)
         SharedMetricsStore._add_consent_tables(connection)
         connection.execute(
-            """
-            INSERT INTO telemetry_state(key, value)
-            VALUES ('schema_version', ?)
-            ON CONFLICT(key) DO UPDATE SET value = excluded.value
-            """,
+            "INSERT INTO telemetry_state(key, value) VALUES ('schema_version', ?)"
+            " ON CONFLICT(key) DO UPDATE SET value = excluded.value",
             (_STORE_SCHEMA_VERSION,),
         )
 
@@ -352,11 +288,10 @@ class SharedMetricsStore:
     def _add_send_columns(connection: sqlite3.Connection) -> None:
         """Add transmission bookkeeping to ``package_outbox``, idempotently.
 
-        The columns are ADDITIVE and nullable, and the schema version is deliberately NOT bumped:
-        ``_ensure_schema_in_transaction`` raises on unknown versions with no forward-compat branch,
-        so a bump would hard-fail an older Hermes (second profile, rollback) against the same
-        database. Old readers select named columns, never ``SELECT *``, so extra columns are
-        invisible to them.
+        Additive and nullable; the schema version is deliberately NOT bumped because
+        ``_ensure_schema_in_transaction`` raises on unknown versions, so a bump would
+        hard-fail an older Hermes (second profile, rollback) sharing the database. Old
+        readers select named columns, never ``SELECT *``.
         """
         existing = {
             str(row["name"])
@@ -371,17 +306,12 @@ class SharedMetricsStore:
             # Earliest next attempt; enforces backoff across process restarts.
             ("next_attempt_at", "TEXT"),
             ("last_error", "TEXT"),
-            # The identifier actually transmitted, frozen on the first
-            # attempt so retries stay byte-identical. Since the 2026-08-27
-            # product decision this is the stable install_id itself.
-            # Only the ~36-byte id is stored: the body is recomputed from
-            # payload_json, whose serialisation is deterministic.
+            # The install_id actually transmitted, frozen on the first attempt so
+            # retries stay byte-identical; the body is recomputed from payload_json.
             ("sent_install_id", "TEXT"),
-            # NULL until first claimed; rewritten on every claim. Settlement
-            # and the pre-POST revalidation are compare-and-set on this, so a
-            # claimant whose lease lapsed loses authority the moment another
-            # process reclaims (PR-review finding: without it, a suspended
-            # sender resuming after a reclaim double-POSTs the package).
+            # Rewritten on every claim. Settlement and the pre-POST revalidation are
+            # compare-and-set on it, so a claimant whose lease lapsed loses authority
+            # the moment another process reclaims (else a suspended sender double-POSTs).
             ("claim_token", "TEXT"),
         ):
             if column not in existing:
@@ -391,15 +321,11 @@ class SharedMetricsStore:
 
     @staticmethod
     def _add_consent_tables(connection: sqlite3.Connection) -> None:
-        """Create the consent-window tables, idempotently.
+        """Create the consent-window tables, idempotently (additive; version not bumped).
 
-        Additive like ``_add_send_columns`` — the schema version is deliberately NOT bumped, and old
-        readers never touch these tables.
-
-        ``send_consent_windows`` records consent as explicit intervals rather than a moving day-
-        stamp: a window is opened when send consent is observed, heartbeat-confirmed on every later
-        observation, and closed at the LAST CONFIRMED moment (never "now") when consent is observed
-        withdrawn.
+        ``send_consent_windows`` records consent as explicit intervals rather than a moving
+        day-stamp: opened when send consent is observed, heartbeat-confirmed on every later
+        observation, closed at the LAST CONFIRMED moment (never "now") on withdrawal.
         """
         connection.execute(
             """
@@ -434,13 +360,8 @@ class SharedMetricsStore:
                 value INTEGER NOT NULL,
                 packaged_value INTEGER NOT NULL DEFAULT 0,
                 PRIMARY KEY (
-                    period_start,
-                    metric_name,
-                    hermes_version,
-                    os_family,
-                    architecture,
-                    install_method,
-                    dimensions_json
+                    period_start, metric_name, hermes_version, os_family, architecture,
+                    install_method, dimensions_json
                 )
             )
             """
@@ -455,45 +376,26 @@ class SharedMetricsStore:
         connection.execute(
             """
             INSERT INTO counter_aggregates(
-                period_start,
-                metric_name,
-                hermes_version,
-                os_family,
-                architecture,
-                install_method,
-                dimensions_json,
-                value,
-                packaged_value
+                period_start, metric_name, hermes_version, os_family, architecture,
+                install_method, dimensions_json, value, packaged_value
             )
             SELECT
-                period_start,
-                metric_name,
-                hermes_version,
-                'unknown',
-                'unknown',
-                'unknown',
-                dimensions_json,
-                value,
-                packaged_value
+                period_start, metric_name, hermes_version, 'unknown', 'unknown', 'unknown',
+                dimensions_json, value, packaged_value
             FROM counter_aggregates_v1
             """
         )
         connection.execute("DROP TABLE counter_aggregates_v1")
 
     def _install_id(self, connection: sqlite3.Connection) -> str:
-        row = connection.execute(
-            "SELECT value FROM telemetry_state WHERE key = 'install_id'"
-        ).fetchone()
-        if row is not None:
-            return str(row["value"])
-        candidate = str(uuid.uuid4())
-        connection.execute(
-            "INSERT OR IGNORE INTO telemetry_state(key, value) VALUES ('install_id', ?)",
-            (candidate,),
-        )
-        row = connection.execute(
-            "SELECT value FROM telemetry_state WHERE key = 'install_id'"
-        ).fetchone()
+        query = "SELECT value FROM telemetry_state WHERE key = 'install_id'"
+        row = connection.execute(query).fetchone()
+        if row is None:
+            connection.execute(
+                "INSERT OR IGNORE INTO telemetry_state(key, value) VALUES ('install_id', ?)",
+                (str(uuid.uuid4()),),
+            )
+            row = connection.execute(query).fetchone()
         if row is None:
             raise RuntimeError("Unable to create the shared-metrics install identity")
         return str(row["value"])
@@ -504,9 +406,7 @@ class SharedMetricsStore:
             parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
         except (TypeError, ValueError):
             return None
-        if parsed.tzinfo is None:
-            return None
-        return parsed.astimezone(timezone.utc)
+        return None if parsed.tzinfo is None else parsed.astimezone(timezone.utc)
 
     def _pending_period_count(self) -> int:
         with self._connection() as connection:
@@ -514,20 +414,10 @@ class SharedMetricsStore:
                 """
                 SELECT COUNT(*) AS period_count
                 FROM (
-                    SELECT
-                        period_start,
-                        hermes_version,
-                        os_family,
-                        architecture,
-                        install_method
+                    SELECT period_start, hermes_version, os_family, architecture, install_method
                     FROM counter_aggregates
                     WHERE value > packaged_value
-                    GROUP BY
-                        period_start,
-                        hermes_version,
-                        os_family,
-                        architecture,
-                        install_method
+                    GROUP BY period_start, hermes_version, os_family, architecture, install_method
                 )
                 """
             ).fetchone()
@@ -535,127 +425,80 @@ class SharedMetricsStore:
 
     def _create_pending_packages_if_due(self) -> None:
         now = _utc_now()
-        with self._connection() as connection:
-            with write_txn(connection):
-                # Gate on the committed package, not its file write, so a failed
-                # outbox export can be retried without packaging deltas twice.
-                package_created_today = connection.execute(
-                    """
-                    SELECT 1
-                    FROM package_outbox
-                    WHERE substr(created_at, 1, 10) >= ?
-                    LIMIT 1
-                    """,
-                    (now.date().isoformat(),),
-                ).fetchone()
-                if package_created_today is not None:
-                    return
-                while self._create_package_in_transaction(connection, now) is not None:
-                    pass
+        with self._write() as connection:
+            # Gate on the committed package, not its file write, so a failed
+            # outbox export can be retried without packaging deltas twice.
+            package_created_today = connection.execute(
+                "SELECT 1 FROM package_outbox WHERE substr(created_at, 1, 10) >= ? LIMIT 1",
+                (now.date().isoformat(),),
+            ).fetchone()
+            if package_created_today is not None:
+                return
+            while self._create_package_in_transaction(connection, now) is not None:
+                pass
 
     def _create_package(self) -> dict[str, Any] | None:
-        now = _utc_now()
-        with self._connection() as connection:
-            with write_txn(connection):
-                return self._create_package_in_transaction(connection, now)
+        with self._write() as connection:
+            return self._create_package_in_transaction(connection, _utc_now())
 
     def _create_package_in_transaction(
-        self,
-        connection: sqlite3.Connection,
-        now: datetime,
+        self, connection: sqlite3.Connection, now: datetime
     ) -> dict[str, Any] | None:
         period_row = connection.execute(
             """
-                SELECT
-                    period_start,
-                    hermes_version,
-                    os_family,
-                    architecture,
-                    install_method
+                SELECT period_start, hermes_version, os_family, architecture, install_method
                 FROM counter_aggregates
                 WHERE value > packaged_value
-                ORDER BY
-                    period_start,
-                    hermes_version,
-                    os_family,
-                    architecture,
-                    install_method
+                ORDER BY period_start, hermes_version, os_family, architecture, install_method
                 LIMIT 1
                 """
         ).fetchone()
-        period_value = period_row["period_start"] if period_row is not None else None
-        if not period_value:
+        if period_row is None or not period_row["period_start"]:
             return None
+        period_value = period_row["period_start"]
 
+        resource = _row_resource(period_row)
+        resource_values = tuple(resource[column] for column in _RESOURCE_COLUMNS)
         rows = connection.execute(
             """
                 SELECT metric_name, dimensions_json, value, packaged_value
                 FROM counter_aggregates
-                WHERE period_start = ?
-                  AND hermes_version = ?
-                  AND os_family = ?
-                  AND architecture = ?
-                  AND install_method = ?
-                  AND value > packaged_value
+                WHERE period_start = ? AND hermes_version = ? AND os_family = ?
+                  AND architecture = ? AND install_method = ? AND value > packaged_value
                 ORDER BY metric_name, dimensions_json
                 """,
-            (
-                period_value,
-                period_row["hermes_version"],
-                period_row["os_family"],
-                period_row["architecture"],
-                period_row["install_method"],
-            ),
+            (period_value, *resource_values),
         ).fetchall()
-        period_start = datetime.fromisoformat(str(period_value)).replace(
-            tzinfo=timezone.utc
-        )
-        period_end = period_start + timedelta(days=1)
-        package_id = str(uuid.uuid4())
-        resource = {
-            "hermes_version": period_row["hermes_version"],
-            "os_family": period_row["os_family"],
-            "architecture": period_row["architecture"],
-            "install_method": period_row["install_method"],
-        }
+        period_start = datetime.fromisoformat(str(period_value)).replace(tzinfo=timezone.utc)
         if not client_resource_is_valid(resource):
             raise ValueError("Unsupported shared-metrics client resource")
         payload = {
             "schema_version": _PACKAGE_SCHEMA_VERSION,
-            "package_id": package_id,
+            "package_id": str(uuid.uuid4()),
             "install_id": self._install_id(connection),
             "period_start": _isoformat(period_start),
-            "period_end": _isoformat(period_end),
+            "period_end": _isoformat(period_start + timedelta(days=1)),
             "generated_at": _isoformat(now),
             "resource": resource,
             "metrics": [self._package_metric(row) for row in rows],
         }
-        payload_json = json.dumps(
-            payload,
-            sort_keys=True,
-            separators=(",", ":"),
-        )
         connection.execute(
             """
                 INSERT INTO package_outbox(
-                    package_id,
-                    period_start,
-                    period_end,
-                    payload_json,
-                    created_at
+                    package_id, period_start, period_end, payload_json, created_at
                 ) VALUES (?, ?, ?, ?, ?)
                 """,
             (
-                package_id,
+                payload["package_id"],
                 payload["period_start"],
                 payload["period_end"],
-                payload_json,
+                _compact_json(payload),
                 payload["generated_at"],
             ),
         )
-        # Advance the data high-water mark. This is the ONLY writer of the
-        # 'data' mark: it clamps consent-window opens so a rolled-back clock
-        # can never open a window underneath packages that already exist.
+        # Advance the data high-water mark. This is the ONLY writer of the 'data' mark:
+        # it clamps consent-window opens so a rolled-back clock can never open a window
+        # underneath packages that already exist.
         connection.execute(
             """
             INSERT INTO consent_marks(name, stamp) VALUES ('data', ?)
@@ -668,23 +511,11 @@ class SharedMetricsStore:
                 """
                     UPDATE counter_aggregates
                     SET packaged_value = value
-                    WHERE period_start = ?
-                      AND metric_name = ?
-                      AND hermes_version = ?
-                      AND os_family = ?
-                      AND architecture = ?
-                      AND install_method = ?
+                    WHERE period_start = ? AND metric_name = ? AND hermes_version = ?
+                      AND os_family = ? AND architecture = ? AND install_method = ?
                       AND dimensions_json = ?
                     """,
-                (
-                    period_value,
-                    row["metric_name"],
-                    period_row["hermes_version"],
-                    period_row["os_family"],
-                    period_row["architecture"],
-                    period_row["install_method"],
-                    row["dimensions_json"],
-                ),
+                (period_value, row["metric_name"], *resource_values, row["dimensions_json"]),
             )
         return payload
 
@@ -706,12 +537,8 @@ class SharedMetricsStore:
     def _export_pending_packages(self) -> list[Path]:
         with self._connection() as connection:
             rows = connection.execute(
-                """
-                SELECT package_id, payload_json
-                FROM package_outbox
-                WHERE exported_at IS NULL
-                ORDER BY created_at, package_id
-                """
+                "SELECT package_id, payload_json FROM package_outbox"
+                " WHERE exported_at IS NULL ORDER BY created_at, package_id"
             ).fetchall()
 
         exported: list[Path] = []
@@ -719,19 +546,12 @@ class SharedMetricsStore:
             package_id = str(row["package_id"])
             path = self.outbox_directory / f"{package_id}.json"
             atomic_json_write(
-                path,
-                json.loads(row["payload_json"]),
-                indent=2,
-                sort_keys=True,
-                mode=0o600,
+                path, json.loads(row["payload_json"]), indent=2, sort_keys=True, mode=0o600
             )
             with self._connection() as connection:
                 connection.execute(
-                    """
-                    UPDATE package_outbox
-                    SET exported_at = ?
-                    WHERE package_id = ? AND exported_at IS NULL
-                    """,
+                    "UPDATE package_outbox SET exported_at = ?"
+                    " WHERE package_id = ? AND exported_at IS NULL",
                     (_isoformat(_utc_now()), package_id),
                 )
             exported.append(path)
@@ -739,20 +559,13 @@ class SharedMetricsStore:
 
     def _prune_expired_history(self, *, now: datetime | None = None) -> None:
         """Remove exported local history after the bounded retention window."""
-        cutoff = (now or _utc_now()) - timedelta(
-            days=_LOCAL_HISTORY_RETENTION_DAYS
-        )
+        cutoff = (now or _utc_now()) - timedelta(days=_LOCAL_HISTORY_RETENTION_DAYS)
         cutoff_timestamp = _isoformat(cutoff)
-        cutoff_period = cutoff.date().isoformat()
         with self._connection() as connection:
             rows = connection.execute(
-                """
-                SELECT package_id
-                FROM package_outbox
-                WHERE exported_at IS NOT NULL
-                  AND exported_at < ?
-                ORDER BY exported_at, package_id
-                """,
+                "SELECT package_id FROM package_outbox"
+                " WHERE exported_at IS NOT NULL AND exported_at < ?"
+                " ORDER BY exported_at, package_id",
                 (cutoff_timestamp,),
             ).fetchall()
 
@@ -760,32 +573,23 @@ class SharedMetricsStore:
         for row in rows:
             package_id = str(row["package_id"])
             try:
-                (self.outbox_directory / f"{package_id}.json").unlink(
-                    missing_ok=True
-                )
+                (self.outbox_directory / f"{package_id}.json").unlink(missing_ok=True)
             except OSError:
                 logger.warning(
-                    "Unable to prune expired shared-metrics package %s",
-                    package_id,
-                    exc_info=True,
+                    "Unable to prune expired shared-metrics package %s", package_id, exc_info=True
                 )
                 continue
             removable_package_ids.append(package_id)
 
-        with self._connection() as connection:
-            with write_txn(connection):
-                for package_id in removable_package_ids:
-                    connection.execute(
-                        """
-                        DELETE FROM package_outbox
-                        WHERE package_id = ?
-                          AND exported_at IS NOT NULL
-                          AND exported_at < ?
-                        """,
-                        (package_id, cutoff_timestamp),
-                    )
+        with self._write() as connection:
+            for package_id in removable_package_ids:
                 connection.execute(
-                    """
+                    "DELETE FROM package_outbox"
+                    " WHERE package_id = ? AND exported_at IS NOT NULL AND exported_at < ?",
+                    (package_id, cutoff_timestamp),
+                )
+            connection.execute(
+                """
                     DELETE FROM counter_aggregates
                     WHERE period_start < ?
                       AND value = packaged_value
@@ -797,5 +601,5 @@ class SharedMetricsStore:
                                 = counter_aggregates.period_start
                       )
                     """,
-                    (cutoff_period,),
-                )
+                (cutoff.date().isoformat(),),
+            )
diff --git a/hermes_cli/observability/shared_metrics_contract.py b/hermes_cli/observability/shared_metrics_contract.py
index 50ac76e69c..cefab49d6c 100644
--- a/hermes_cli/observability/shared_metrics_contract.py
+++ b/hermes_cli/observability/shared_metrics_contract.py
@@ -100,6 +100,7 @@ _ARCHITECTURE_ALIASES = {
     "aarch64": "arm64", "arm64": "arm64",
     "i386": "x86", "i486": "x86", "i586": "x86", "i686": "x86", "x86": "x86",
 }
+_OS_FAMILIES = {"darwin": "macos", "linux": "linux", "macos": "macos", "windows": "windows"}
 
 
 def _norm(value: Any) -> str:
@@ -112,9 +113,7 @@ def _allowlisted(normalized: str, allowed: frozenset[str]) -> str:
 
 def client_os_family(value: Any) -> str:
     """Map a platform system name to the shared-metrics OS taxonomy."""
-    return {"darwin": "macos", "linux": "linux", "macos": "macos", "windows": "windows"}.get(
-        _norm(value), "unknown"
-    )
+    return _OS_FAMILIES.get(_norm(value), "unknown")
 
 
 def client_architecture(value: Any) -> str:
@@ -279,24 +278,25 @@ def _event_shape_matches(
     """Match the coarse Relay event shape.
 
     A ``str`` expectation compares against the stringified attribute; ``None`` requires the
-    attribute itself to be ``None``; ``_ANY`` skips the check. ``category_profile`` compares
-    with plain equality (marks carry ``None``, tool scopes ``{}``).
+    attribute itself to be ``None``; ``_ANY`` skips the check. Anything else (the
+    ``category_profile`` dict) compares with plain equality.
     """
     if _event_text(event, "kind") != kind:
         return False
-    for attr, expected, as_text in (
-        ("name", name, True),
-        ("category", category, True),
-        ("scope_category", scope_category, True),
-        ("category_profile", category_profile, False),
-    ):
+    expectations = {
+        "name": name, "category": category, "scope_category": scope_category,
+        "category_profile": category_profile,
+    }
+    for attr, expected in expectations.items():
         if expected is _ANY:
             continue
-        actual = getattr(event, attr, None)
-        if expected is None:
-            if actual is not None:
-                return False
-        elif (_event_text(event, attr) if as_text else actual) != expected:
+        if isinstance(expected, str):
+            matched = _event_text(event, attr) == expected
+        elif expected is None:
+            matched = getattr(event, attr, None) is None
+        else:
+            matched = getattr(event, attr, None) == expected
+        if not matched:
             return False
     return True
 
@@ -453,6 +453,7 @@ _SURFACE_ALIASES = {
     "api_server": "api",
     **dict.fromkeys(("cron", "scheduler", "scheduled"), "scheduled_task"),
 }
+_KNOWN_GATEWAY_PLATFORMS = frozenset({"discord", "email", "slack", "telegram", "teams", "whatsapp"})
 
 
 def execution_surface(kwargs: dict[str, Any]) -> str:
@@ -469,7 +470,7 @@ def execution_surface(kwargs: dict[str, Any]) -> str:
             return "gateway"
     except Exception:
         pass
-    if value in {"discord", "email", "slack", "telegram", "teams", "whatsapp"}:
+    if value in _KNOWN_GATEWAY_PLATFORMS:
         return "gateway"
     return "unknown" if value == "unknown" else "other"
 
@@ -477,10 +478,7 @@ def execution_surface(kwargs: dict[str, Any]) -> str:
 def task_start_fields(kwargs: dict[str, Any]) -> dict[str, str]:
     """Build the bounded fields recorded on a task scope start event."""
     surface = execution_surface(kwargs)
-    return {
-        "entrypoint": task_entrypoint(kwargs, surface),
-        "execution_surface": surface,
-    }
+    return {"entrypoint": task_entrypoint(kwargs, surface), "execution_surface": surface}
 
 
 _SURFACE_ENTRYPOINTS = {
@@ -495,10 +493,9 @@ def task_entrypoint(kwargs: dict[str, Any], surface: str | None = None) -> str:
     declared = _norm(kwargs.get("entrypoint"))
     if declared in TASK_ENTRYPOINTS:
         return declared
-    resolved_surface = surface or execution_surface(kwargs)
     if kwargs.get("parent_task_id") or kwargs.get("parent_session_id"):
         return "delegated"
-    return _SURFACE_ENTRYPOINTS.get(resolved_surface, "other")
+    return _SURFACE_ENTRYPOINTS.get(surface or execution_surface(kwargs), "other")
 
 
 def task_terminal_fields(
@@ -524,7 +521,7 @@ def task_terminal_fields(
 
 
 def task_terminal_state(kwargs: dict[str, Any]) -> tuple[str, str, str]:
-    """Map Hermes terminal state to bounded task outcome dimensions."""
+    """Map Hermes terminal state to bounded (outcome, end_reason, termination)."""
     reason = _norm(kwargs.get("turn_exit_reason"))
     if kwargs.get("interrupted") or "interrupt" in reason or "cancel" in reason:
         return "cancelled", "user_cancelled", "user_cancelled"
@@ -558,10 +555,7 @@ _LATENCY_THRESHOLDS = (
 
 
 def _bucket(value: float, thresholds: tuple[tuple[float, str], ...], last: str) -> str:
-    for upper, label in thresholds:
-        if value < upper:
-            return label
-    return last
+    return next((label for upper, label in thresholds if value < upper), last)
 
 
 def duration_bucket(duration_ms: int) -> str:
@@ -636,9 +630,7 @@ def tool_terminal_fields(
 ) -> dict[str, str]:
     """Build one bounded tool-call terminal payload."""
     return {
-        "approval_outcome": (
-            approval_outcome if approval_outcome in TOOL_APPROVAL_OUTCOMES else "unknown"
-        ),
+        "approval_outcome": _allowlisted(approval_outcome, TOOL_APPROVAL_OUTCOMES),
         "latency_bucket": tool_latency_bucket(
             kwargs.get("duration_ms"), fallback_duration_ms=fallback_duration_ms
         ),
diff --git a/hermes_cli/observability/shared_metrics_send_config.py b/hermes_cli/observability/shared_metrics_send_config.py
index 0d44906b19..de8ac454ad 100644
--- a/hermes_cli/observability/shared_metrics_send_config.py
+++ b/hermes_cli/observability/shared_metrics_send_config.py
@@ -8,14 +8,10 @@ from urllib.parse import urlparse
 
 logger = logging.getLogger(__name__)
 
-#: Production ingest endpoint. Overridable through config only.
-#:
-#: Deliberately NOT overridable by an environment variable: AGENTS.md reserves
-#: HERMES_* env vars for secrets, and a behavioural override here would be a
-#: consent hazard — a user who agreed to send metrics to Nous could have them
-#: silently redirected to any host by an inherited variable, with nothing
-#: visible in their config to show it. Tests and the staging E2E write this
-#: key into a throwaway profile instead.
+#: Production ingest endpoint. Overridable through config only — deliberately NOT by an
+#: environment variable: AGENTS.md reserves HERMES_* for secrets, and an inherited variable
+#: could silently redirect consented metrics to any host with nothing visible in config.
+#: Tests and the staging E2E write this key into a throwaway profile instead.
 DEFAULT_ENDPOINT = "https://telemetry.nousresearch.com/v1/telemetry"
 
 _LOCAL_HOSTS = frozenset({"localhost", "127.0.0.1", "::1", "[::1]"})
@@ -38,28 +34,20 @@ class SendConfig:
 
 
 def _endpoint_is_safe(endpoint: str) -> bool:
-    """Reject plaintext destinations unless they are loopback.
-
-    Telemetry must not leave a machine in clear text because of a typo in a config file. Loopback
-    stays allowed so tests can use a local HTTP server.
-    """
+    """Reject plaintext destinations unless they are loopback (so tests can use local HTTP)."""
     try:
         parsed = urlparse(endpoint)
     except ValueError:
         return False
     if parsed.scheme == "https":
         return True
-    if parsed.scheme == "http":
-        return (parsed.hostname or "") in _LOCAL_HOSTS
-    return False
+    return parsed.scheme == "http" and (parsed.hostname or "") in _LOCAL_HOSTS
 
 
 def resolve_send_config(config: dict | None) -> SendConfig:
-    """Resolve transmission settings from config plus the environment.
+    """Resolve transmission settings from config (endpoint: config > production default).
 
-    Endpoint precedence: config > production default.
-
-    ``send`` is returned as False whenever transmission cannot legitimately happen, so callers never
+    ``send`` is False whenever transmission cannot legitimately happen, so callers never
     have to re-check the combination.
     """
     global _warned_send_without_collection
@@ -74,8 +62,7 @@ def resolve_send_config(config: dict | None) -> SendConfig:
     send_requested = shared.get("send") is True
 
     if send_requested and not enabled:
-        # Loud, not silent: the user believes telemetry is being sent, and it
-        # never will be. Error level, once per process.
+        # Loud, not silent: the user believes telemetry is being sent, and it never will be.
         if not _warned_send_without_collection:
             _warned_send_without_collection = True
             logger.error(
diff --git a/hermes_cli/observability/shared_metrics_sender.py b/hermes_cli/observability/shared_metrics_sender.py
index 5bcedba764..dab55fdbd1 100644
--- a/hermes_cli/observability/shared_metrics_sender.py
+++ b/hermes_cli/observability/shared_metrics_sender.py
@@ -1,10 +1,8 @@
 """Transmit exported shared-metrics packages to the Nous telemetry service.
 
-Implements the sender side of the ingest contract (see the telemetry repo's ``CONTRACT.md``):
-
-* ``202`` — durably stored. Mark sent. * ``400`` — permanently malformed. Never retry. * ``429`` —
-keep, retry after ``Retry-After``. * ``5xx`` / timeout / connection error — keep, retry with
-backoff.
+Sender side of the ingest contract (telemetry repo ``CONTRACT.md``): ``202`` durably stored,
+mark sent; ``400`` permanently malformed, never retry; ``429`` keep, retry after
+``Retry-After``; ``5xx`` / timeout / connection error keep, retry with backoff.
 """
 
 from __future__ import annotations
@@ -18,11 +16,14 @@ import time
 import urllib.error
 import urllib.request
 import uuid
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 
 from hermes_cli.sqlite_util import write_txn
 
+from .shared_metrics import _isoformat, _utc_now
+
 logger = logging.getLogger(__name__)
 
 #: Contract recommends timing out at 30s and treating a timeout as retryable.
@@ -41,49 +42,40 @@ GZIP_THRESHOLD_BYTES = 4096
 MAX_PACKAGES_PER_PASS = 20
 
 #: How long a claimed row is held. The claim writes a LEASE INTO THE FUTURE:
-#: selection requires `next_attempt_at <= now`, so for the length of the lease
-#: no other process can take the package.
-#:
-#: This must exceed the worst case for ONE package — three 30s request
-#: timeouts plus 1s+5s of backoff, about 96s — which is why packages are
-#: claimed one at a time, immediately before being sent. An earlier revision
-#: claimed up to 20 rows under a single shared lease; a full batch can legally
-#: run ~1900s, so the later rows' leases expired while the pass still held
-#: them in memory and another process re-sent them.
+#: selection requires `next_attempt_at <= now`, so no other process can take the
+#: package for the length of the lease. Must exceed the worst case for ONE package
+#: (three 30s timeouts plus 1s+5s backoff, ~96s) — which is why packages are claimed
+#: one at a time right before sending: a batch of 20 under one lease can legally run
+#: ~1900s, so later rows' leases expired while still held and got re-sent.
 _CLAIM_LEASE_SECONDS = 300
 
 #: Floor applied after a pass fails to deliver, so a hard-down service is not
 #: retried on every task completion.
 _FAILURE_BACKOFF_SECONDS = 15 * 60
 
-#: Statuses that are permanent per the ingest contract. Deliberately narrow:
-#: 400 means the envelope is malformed and will never validate. 413 is added
-#: because a package over the service's 1 MiB cap cannot shrink on retry.
-#: Everything else — including 403 from the origin guard and 404 from a bad
-#: path — is retried, because those are usually deployment or edge
+#: Permanent statuses per the ingest contract. Deliberately narrow: 400 means the
+#: envelope will never validate; 413 because a package over the 1 MiB cap cannot
+#: shrink on retry. Everything else — including 403 from the origin guard and 404
+#: from a bad path — is retried, since those are usually deployment/edge
 #: misconfiguration that resolves without the package changing.
 _PERMANENT_STATUSES = frozenset({400, 413})
 
-#: Attempts after which a package is abandoned. Without a ceiling a
-#: permanently-poisoned row is retried until 30-day retention deletes it —
-#: measured at ~160 requests — which wastes the user's bandwidth and keeps a
-#: doomed package at the head of the queue.
+#: Attempts after which a package is abandoned. Without a ceiling a poisoned row
+#: is retried until 30-day retention deletes it (~160 requests), wasting the user's
+#: bandwidth and pinning a doomed package at the head of the queue.
 MAX_SEND_ATTEMPTS = 25
 
-
-def _utc_now() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-def _isoformat(value: datetime) -> str:
-    return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+#: Maximum distance one reconcile call can advance the 'obs' mark. Honest heartbeats
+#: arrive hours apart so the cap never binds normally; a machine off for months catches
+#: up in a few hook fires (fail-closed latency only). It bounds FORWARD clock poison:
+#: without it one glitched sample (NTP flap reading 2099) drags the mark — and every
+#: window open and confirmation horizon — decades ahead, a refused-data leak.
+MAX_OBS_ADVANCE_SECONDS = 30 * 24 * 3600
 
 
 def _parse_stamp(value: str) -> datetime:
     """Parse a stamp this module itself wrote (Z-suffixed ISO-8601, UTC)."""
-    return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(
-        timezone.utc
-    )
+    return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
 
 
 @dataclass
@@ -112,17 +104,11 @@ def _post(endpoint: str, payload: bytes, *, timeout: int) -> _Response:
     }
     body = payload
     if len(payload) > GZIP_THRESHOLD_BYTES:
-        # mtime=0: gzip embeds a timestamp by default, which would make two
-        # sends of one package differ on the wire. The service decompresses
-        # before storing so it would not change what lands in S3, but a
-        # deterministic body keeps "a resend is byte-identical" true at the
-        # transport layer too, and makes the property testable.
+        # mtime=0 keeps two sends of one package byte-identical on the wire too.
         body = gzip.compress(payload, mtime=0)
         headers["Content-Encoding"] = "gzip"
 
-    request = urllib.request.Request(
-        endpoint, data=body, headers=headers, method="POST"
-    )
+    request = urllib.request.Request(endpoint, data=body, headers=headers, method="POST")
     try:
         with urllib.request.urlopen(request, timeout=timeout) as response:
             return _Response(
@@ -143,25 +129,13 @@ def _retry_after_seconds(value: str | None, default: int) -> int:
     if not value:
         return default
     try:
-        # Contract sends seconds. Clamp so a hostile or bogus value cannot
-        # park a package for years, and never go below one second.
+        # Contract sends seconds. Clamp so a bogus value cannot park a package for
+        # years, and never go below one second.
         return max(1, min(int(float(value)), 86_400))
     except (TypeError, ValueError):
         return default
 
 
-#: Maximum distance one reconcile call can advance the 'obs' mark. Honest
-#: heartbeats arrive hours apart at most, so the cap never binds in normal
-#: operation; a machine legitimately off for months catches up in a few
-#: hook fires (fail-closed latency only). What it bounds is FORWARD clock
-#: poison: without it, a single glitched sample (NTP flap reading 2099)
-#: permanently drags the mark — and with it every window open and every
-#: confirmation horizon — decades ahead, which round 6 reproduced as a
-#: refused-data leak. Capped, one insane sample moves the mark at most
-#: this far, and real time overtakes it again.
-MAX_OBS_ADVANCE_SECONDS = 30 * 24 * 3600
-
-
 def reconcile_send_consent(
     connection: sqlite3.Connection,
     send_enabled: bool,
@@ -170,9 +144,8 @@ def reconcile_send_consent(
 ) -> None:
     """Reconcile the consent-window table with the observed config state.
 
-    THE ONLY writer of consent state. Must run inside a write transaction. A pure function of
-    (config, now, store): call it from anywhere, any number of times, in any order — the resulting
-    windows are the same.
+    THE ONLY writer of consent state. Must run inside a write transaction. A pure function
+    of (config, now, store): call it from anywhere, any number of times, in any order.
     """
     stamp = _isoformat(now or _utc_now())
     raw_stamp = stamp  # pre-cap observation time, used to clamp closes
@@ -181,8 +154,7 @@ def reconcile_send_consent(
     ).fetchone()
     if previous_obs is not None:
         ceiling = _isoformat(
-            _parse_stamp(str(previous_obs[0]))
-            + timedelta(seconds=MAX_OBS_ADVANCE_SECONDS)
+            _parse_stamp(str(previous_obs[0])) + timedelta(seconds=MAX_OBS_ADVANCE_SECONDS)
         )
         stamp = min(stamp, ceiling)
     connection.execute(
@@ -192,9 +164,7 @@ def reconcile_send_consent(
         """,
         (stamp,),
     )
-    marks = dict(
-        connection.execute("SELECT name, stamp FROM consent_marks").fetchall()
-    )
+    marks = dict(connection.execute("SELECT name, stamp FROM consent_marks").fetchall())
     obs = marks["obs"]  # >= stamp; immune to clock rollback
     data = marks.get("data")
 
@@ -218,17 +188,13 @@ def reconcile_send_consent(
                 (obs, open_row[0]),
             )
     elif open_row is not None:
-        # Close at the last CONFIRMED moment, but never after the closing
-        # observation's own raw stamp. The two clamps serve different
-        # adversaries and both are load-bearing:
-        # - min with last_confirmed_at: an unobserved gap (machine off,
-        #   hand-edited config) is never asserted as consented (v1's leak).
-        # - min with the RAW stamp (pre-cap, pre-MAX): if last_confirmed_at
-        #   was poisoned by a glitched-forward sample, an honest clock at
-        #   revoke time pulls the close back to the true revoke moment, so
-        #   the refused era that follows falls OUTSIDE the closed window
-        #   (round 6's D1 leak). A rolled-back clock at close time only
-        #   closes EARLIER — fail-closed.
+        # Close at the last CONFIRMED moment, but never after the closing observation's
+        # own raw stamp. Both clamps are load-bearing: min with last_confirmed_at means an
+        # unobserved gap (machine off, hand-edited config) is never asserted as consented;
+        # min with the RAW (pre-cap, pre-MAX) stamp means a last_confirmed_at poisoned by a
+        # glitched-forward sample is pulled back to the true revoke moment by an honest
+        # clock, so the refused era falls OUTSIDE the window. A rolled-back clock only
+        # closes EARLIER — fail-closed.
         connection.execute(
             "UPDATE send_consent_windows"
             " SET closed_at = MIN(last_confirmed_at, ?)"
@@ -237,10 +203,10 @@ def reconcile_send_consent(
         )
 
 
-#: Claim-time consent predicate: the package's period must fall entirely
-#: inside SOME recorded consent window. An open window vouches only up to its
-#: last confirmed moment, so a package whose period runs past it waits for
-#: the next reconcile heartbeat (fail-closed; released within one hook fire).
+#: Claim-time consent predicate: the package's period must fall entirely inside SOME
+#: recorded consent window. An open window vouches only up to its last confirmed moment,
+#: so a package whose period runs past it waits for the next reconcile heartbeat
+#: (fail-closed; released within one hook fire).
 CONSENT_GATE_SQL = """EXISTS (
     SELECT 1 FROM send_consent_windows w
     WHERE package_outbox.period_start >= w.opened_at
@@ -270,36 +236,38 @@ class SharedMetricsSender:
         self._sleep = sleep
         self._now = now
         self._max_attempts = max_attempts
-        # Called before every package. None disables the check for callers
-        # that have already established consent out of band (tests, E2E).
+        # Called before every package. None disables the check for callers that have
+        # already established consent out of band (tests, E2E).
         self._consent_check = consent_check
 
+    @contextmanager
+    def _write(self):
+        """One store connection inside a write transaction."""
+        with self._store._connection() as connection:
+            with write_txn(connection):
+                yield connection
+
     # -- selection ---------------------------------------------------------
 
     def _claim_next(self, now: datetime, seen: set[str]) -> dict | None:
         """Claim exactly ONE package, immediately before it is sent.
 
-        Claiming a batch up front fails: one shared lease would have to cover the whole pass, and 20
-        retrying packages can outlive any sane lease, so later rows expire and another process
-        re-sends them. ``seen`` (packages this pass finished with) is excluded IN SQL: with LIMIT 1,
-        returning None for a seen row would look like an empty queue and abandon everything behind
-        it, and rows can legitimately become eligible again mid-pass.
+        Claiming a batch up front fails: one shared lease would have to cover the whole
+        pass, and 20 retrying packages can outlive any sane lease. ``seen`` (packages this
+        pass finished with) is excluded IN SQL: with LIMIT 1, returning None for a seen row
+        would look like an empty queue and abandon everything behind it, and rows can
+        legitimately become eligible again mid-pass.
         """
-        with self._store._connection() as connection:
-            with write_txn(connection):
-                stamp = _isoformat(now)
-                lease_until = now + timedelta(seconds=_CLAIM_LEASE_SECONDS)
+        with self._write() as connection:
+            stamp = _isoformat(now)
+            lease_until = now + timedelta(seconds=_CLAIM_LEASE_SECONDS)
 
-                placeholders = ",".join("?" for _ in seen)
-                exclusion = (
-                    f" AND package_id NOT IN ({placeholders})" if seen else ""
-                )
-                # Consent is a READ here — the claim must never mutate the
-                # window table. The old design's opt_in_period() call at this
-                # exact spot meant selecting a row could rewrite what was
-                # permitted to be sent (and did, under a rolled-back clock).
-                row = connection.execute(
-                    f"""
+            placeholders = ",".join("?" for _ in seen)
+            exclusion = f" AND package_id NOT IN ({placeholders})" if seen else ""
+            # Consent is a READ here — the claim must never mutate the window table, or
+            # selecting a row could rewrite what is permitted to be sent.
+            row = connection.execute(
+                f"""
                     SELECT package_id, payload_json, sent_install_id
                     FROM package_outbox
                     WHERE exported_at IS NOT NULL
@@ -311,25 +279,20 @@ class SharedMetricsSender:
                     ORDER BY created_at, package_id
                     LIMIT 1
                     """,
-                    (stamp, MAX_SEND_ATTEMPTS, *sorted(seen)),
-                ).fetchone()
-                if row is None:
-                    return None
+                (stamp, MAX_SEND_ATTEMPTS, *sorted(seen)),
+            ).fetchone()
+            if row is None:
+                return None
 
-                package_id = str(row[0])
-                derived = row[2]
-                if not derived:
-                    derived = self._freeze_identity(
-                        connection, package_id, row[1], now
-                    )
-                    if derived is None:
-                        # Unusable row, already marked rejected. Signal the
-                        # caller to continue rather than stop.
-                        return {"package_id": package_id, "skip": True}
+            package_id = str(row[0])
+            derived = row[2] or self._freeze_identity(connection, package_id, row[1])
+            if derived is None:
+                # Unusable row, already marked rejected. Tell the caller to continue.
+                return {"package_id": package_id, "skip": True}
 
-                token = str(uuid.uuid4())
-                connection.execute(
-                    """
+            token = str(uuid.uuid4())
+            connection.execute(
+                """
                     UPDATE package_outbox
                     SET send_state = 'pending',
                         send_attempts = send_attempts + 1,
@@ -337,58 +300,49 @@ class SharedMetricsSender:
                         claim_token = ?
                     WHERE package_id = ?
                     """,
-                    # Lease INTO THE FUTURE: selection requires
-                    # next_attempt_at <= now, so no other process can take
-                    # this row while it is in flight. Success or a real
-                    # backoff overwrites it; if this process dies, it expires.
-                    # The token is this claim's identity: a reclaim after
-                    # expiry mints a new one, and every later write by THIS
-                    # claimant is compare-and-set against it, so a lapsed
-                    # claimant that resumes cannot settle or transmit.
-                    (_isoformat(lease_until), token, package_id),
-                )
-                return {
-                    "package_id": package_id,
-                    "payload_json": str(row[1]),
-                    "derived": str(derived),
-                    "claim_token": token,
-                    "skip": False,
-                }
+                # Lease INTO THE FUTURE (see _CLAIM_LEASE_SECONDS). The token is this
+                # claim's identity: a reclaim after expiry mints a new one, and every later
+                # write by THIS claimant is compare-and-set against it, so a lapsed claimant
+                # that resumes cannot settle or transmit.
+                (_isoformat(lease_until), token, package_id),
+            )
+            return {
+                "package_id": package_id,
+                "payload_json": str(row[1]),
+                "derived": str(derived),
+                "claim_token": token,
+                "skip": False,
+            }
 
+    @staticmethod
     def _freeze_identity(
-        self,
-        connection: sqlite3.Connection,
-        package_id: str,
-        payload_json,
-        now: datetime,
+        connection: sqlite3.Connection, package_id: str, payload_json
     ) -> str | None:
-        """Record the transmitted id on the row, or reject an unusable one.
+        """Record the transmitted install_id on the row, or reject an unusable one.
 
-        What remains of "freezing" is the validation and the audit column: ``sent_install_id``
-        records exactly what the wire will carry, and rejecting unusable rows here rather than
-        raising matters because an exception rolls back the claim transaction and blocks every
+        ``sent_install_id`` records exactly what the wire will carry. Rejecting here rather
+        than raising matters: an exception rolls back the claim transaction and blocks every
         healthy package behind this one.
         """
-        reason = None
         install_id = None
         try:
             payload = json.loads(payload_json)
         except (TypeError, ValueError):
             reason = "unreadable payload"
         else:
-            # Valid JSON is not enough: a top-level array, string, number or
-            # null parses cleanly and then has no .get().
+            # Valid JSON is not enough: a top-level array/string/number parses cleanly.
             if not isinstance(payload, dict):
                 reason = f"payload is {type(payload).__name__}, expected object"
             else:
                 install_id = payload.get("install_id")
-                if not isinstance(install_id, str) or not install_id.strip():
-                    reason = "payload has no usable install_id"
+                reason = (
+                    None
+                    if isinstance(install_id, str) and install_id.strip()
+                    else "payload has no usable install_id"
+                )
 
         if reason is not None:
-            logger.warning(
-                "Shared-metrics package %s cannot be sent (%s)", package_id, reason
-            )
+            logger.warning("Shared-metrics package %s cannot be sent (%s)", package_id, reason)
             connection.execute(
                 """
                 UPDATE package_outbox
@@ -407,16 +361,15 @@ class SharedMetricsSender:
 
     # -- transmission ------------------------------------------------------
 
-    def _body(self, payload_json: str, transmitted_id: str) -> bytes:
+    @staticmethod
+    def _body(payload_json: str, transmitted_id: str) -> bytes:
         """Rebuild the exact bytes to send.
 
-        The payload is recomputed from the stored package rather than kept as a second copy:
-        json.dumps with these options is deterministic. The install_id is written from the frozen
-        ``sent_install_id`` column rather than trusted implicitly, keeping "a resend is byte-
-        identical" anchored to one recorded value.
+        Recomputed from the stored package (json.dumps with these options is deterministic)
+        with install_id taken from the frozen ``sent_install_id`` column, keeping "a resend
+        is byte-identical" anchored to one recorded value.
         """
-        payload = json.loads(payload_json)
-        payload = dict(payload)
+        payload = dict(json.loads(payload_json))
         payload["install_id"] = transmitted_id
         return json.dumps(payload, indent=2, sort_keys=True).encode("utf-8")
 
@@ -430,54 +383,40 @@ class SharedMetricsSender:
     ) -> None:
         """Write send state for one package.
 
-        Guarded on send_state so a pass whose lease lapsed cannot resurrect a row another process
-        has already finished: without this, a slow sender could overwrite 'sent' back to 'pending'
-        and cause a re-send.
-
-        When ``token`` is given, the write is additionally compare-and-set on claim_token: it lands
-        only if THIS claim is still the current one. A claimant that lapsed and was superseded
-        writes zero rows — its settlement, backoff, and error strings all silently lose to the newer
-        claim's, which is the correct outcome.
+        Guarded on send_state so a pass whose lease lapsed cannot resurrect a row another
+        process already finished (overwriting 'sent' back to 'pending' would re-send). With
+        ``token`` the write is also compare-and-set on claim_token: a superseded claimant
+        writes zero rows and its settlement/backoff/error silently lose to the newer claim.
         """
         assignments = ", ".join(f"{name} = ?" for name in columns)
-        predicate = (
-            " AND (send_state IS NULL OR send_state = 'pending')"
-            if only_if_pending
-            else ""
-        )
+        predicate = " AND (send_state IS NULL OR send_state = 'pending')" if only_if_pending else ""
         params: list = [*columns.values(), package_id]
         if token is not None:
             predicate += " AND claim_token = ?"
             params.append(token)
-        with self._store._connection() as connection:
-            with write_txn(connection):
-                connection.execute(
-                    f"UPDATE package_outbox SET {assignments} "
-                    f"WHERE package_id = ?{predicate}",
-                    params,
-                )
+        with self._write() as connection:
+            connection.execute(
+                f"UPDATE package_outbox SET {assignments} WHERE package_id = ?{predicate}",
+                params,
+            )
 
     def _renew_claim(self, package_id: str, token: str | None) -> bool:
         """Atomically re-assert ownership and extend the lease. CAS, one row.
 
-        A read-only ownership check is not enough: a claimant whose lease expired while suspended
-        can pass the check (its token is still in the row if no one reclaimed yet) and then POST
-        while another process legitimately reclaims — the check-to-POST expiry race a seventh review
-        reproduced.
-
-        and only then pushing next_attempt_at a fresh lease into the future, so the upcoming POST
-        (30s timeout, well under the 300s lease) runs entirely inside renewed authority. rowcount ==
-        1 is the only grant.
+        A read-only ownership check is not enough: a claimant whose lease expired while
+        suspended still has its token in the row if nobody reclaimed yet, and would POST
+        while another process legitimately reclaims (check-to-POST expiry race). Pushing a
+        fresh lease in the same CAS keeps the upcoming POST (30s, well under the 300s lease)
+        entirely inside renewed authority. rowcount == 1 is the only grant.
         """
         if token is None:
             return False
         try:
             now = self._now()
             lease_until = now + timedelta(seconds=_CLAIM_LEASE_SECONDS)
-            with self._store._connection() as connection:
-                with write_txn(connection):
-                    cursor = connection.execute(
-                        """
+            with self._write() as connection:
+                cursor = connection.execute(
+                    """
                         UPDATE package_outbox
                         SET next_attempt_at = ?
                         WHERE package_id = ?
@@ -485,176 +424,120 @@ class SharedMetricsSender:
                           AND (send_state IS NULL OR send_state = 'pending')
                           AND next_attempt_at > ?
                         """,
-                        (
-                            _isoformat(lease_until),
-                            package_id,
-                            token,
-                            _isoformat(now),
-                        ),
-                    )
-                    return cursor.rowcount == 1
+                    (_isoformat(lease_until), package_id, token, _isoformat(now)),
+                )
+                return cursor.rowcount == 1
         except Exception:
             # If renewal itself fails, do not transmit on unproven authority.
-            logger.warning(
-                "Unable to renew shared-metrics claim", exc_info=True
-            )
+            logger.warning("Unable to renew shared-metrics claim", exc_info=True)
             return False
 
     def _defer(
-        self,
-        package_id: str,
-        delay_seconds: int,
-        reason: str,
-        *,
-        token: str | None = None,
+        self, package_id: str, delay_seconds: int, reason: str, *, token: str | None = None
     ) -> None:
-        # Defence in depth: no current caller can pass a non-positive delay
-        # (Retry-After is already clamped to [1, 86400] when parsed, and every
-        # other call site passes a positive constant), so this clamp is
-        # deliberately unreachable today and no test can distinguish it. It
-        # stays because a past deadline would make the row instantly
-        # re-eligible and let a pass spin on it — a cheap guard against a
-        # future caller that forgets.
-        delay = max(1, int(delay_seconds))
-        retry_at = self._now().timestamp() + delay
+        # Clamp to >= 1s: a past deadline would make the row instantly re-eligible and let
+        # a pass spin on it. Unreachable today (Retry-After is pre-clamped, other callers
+        # pass positive constants) — kept as a guard against a future caller.
+        retry_at = self._now().timestamp() + max(1, int(delay_seconds))
         self._mark(
             package_id,
             token=token,
             send_state="pending",
-            next_attempt_at=_isoformat(
-                datetime.fromtimestamp(retry_at, tz=timezone.utc)
-            ),
+            next_attempt_at=_isoformat(datetime.fromtimestamp(retry_at, tz=timezone.utc)),
             last_error=reason[:500],
         )
 
     def _send_one(self, package: dict) -> str:
         """Try one package. Returns 'sent', 'rejected', or 'deferred'.
 
-        Delivery is at-least-once. The pre-POST ownership check plus the token-fenced writes close
-        the claim->POST and settle-after-reclaim gaps, but a suspension landing MID-POST (bytes
-        already on the wire when the machine sleeps) can still duplicate: no client-side check can
-        revoke a request in flight.
+        Delivery is at-least-once. The pre-POST renewal plus token-fenced writes close the
+        claim->POST and settle-after-reclaim gaps, but a suspension landing MID-POST can
+        still duplicate: no client-side check can revoke a request in flight.
         """
         package_id = package["package_id"]
         token = package.get("claim_token")
         body = self._body(package["payload_json"], package["derived"])
 
+        def defer(delay: int, reason: str) -> str:
+            self._defer(package_id, delay, reason, token=token)
+            return "deferred"
+
         for attempt in range(1, self._max_attempts + 1):
-            # Atomically renew the claim before EVERY external POST. The
-            # renewal is compare-and-set on (token, pending, lease unexpired)
-            # and extends the lease past the request, so a suspended-then-
-            # resumed claimant whose lease lapsed yields here even if nobody
-            # has reclaimed yet — a read-only ownership check passed in that
-            # state and still double-sent (check-to-POST expiry race). The
-            # ingest key is minute-prefixed, so duplicates become distinct
+            # Renew before EVERY external POST (CAS on token, pending, lease unexpired), so a
+            # suspended-then-resumed claimant whose lease lapsed yields even before anyone
+            # reclaims. The ingest key is minute-prefixed, so duplicates become distinct
             # stored objects, not overwrites.
             if not self._renew_claim(package_id, token):
                 logger.info(
-                    "Shared-metrics claim on %s superseded or expired; yielding",
-                    package_id,
+                    "Shared-metrics claim on %s superseded or expired; yielding", package_id
                 )
                 return "deferred"
             try:
-                response = self._post(
-                    self._endpoint, body, timeout=REQUEST_TIMEOUT_SECONDS
-                )
+                response = self._post(self._endpoint, body, timeout=REQUEST_TIMEOUT_SECONDS)
             except Exception as exc:  # transport failure: offline, DNS, TLS
                 reason = f"{type(exc).__name__}: {exc}"
-                if attempt >= self._max_attempts:
-                    self._defer(
-                        package_id, _FAILURE_BACKOFF_SECONDS, reason, token=token
+            else:
+                if response.status == 202:
+                    self._mark(
+                        package_id, token=token, send_state="sent",
+                        sent_at=_isoformat(self._now()), last_error=None,
                     )
-                    return "deferred"
-                self._sleep(self._backoff(attempt))
-                continue
-
-            if response.status == 202:
-                self._mark(
-                    package_id,
-                    token=token,
-                    send_state="sent",
-                    sent_at=_isoformat(self._now()),
-                    last_error=None,
-                )
-                return "sent"
-
-            if response.status in _PERMANENT_STATUSES:
-                # Only statuses the contract (or the envelope schema) makes
-                # terminal. Everything else retries: 403 in particular is the
-                # ingest service's origin guard, which returns 403 during an
-                # edge/Transform-Rule misconfiguration — treating that as
-                # permanent would discard every package sent during the
-                # incident instead of retrying after recovery.
-                logger.warning(
-                    "Telemetry package %s rejected with HTTP %s; not retrying",
-                    package_id,
-                    response.status,
-                )
-                self._mark(
-                    package_id,
-                    token=token,
-                    send_state="rejected",
-                    last_error=f"HTTP {response.status}: {response.body[:400]}",
-                )
-                return "rejected"
-
-            if response.status == 429:
-                self._defer(
-                    package_id,
-                    _retry_after_seconds(response.retry_after, _FAILURE_BACKOFF_SECONDS),
-                    "rate limited",
-                    token=token,
-                )
-                return "deferred"
-
-            # 5xx and anything unexpected: retryable.
-            reason = f"HTTP {response.status}"
+                    return "sent"
+                if response.status in _PERMANENT_STATUSES:
+                    # Only contract-terminal statuses. 403 in particular is the origin guard
+                    # during an edge misconfiguration — treating it as permanent would
+                    # discard every package sent during the incident.
+                    logger.warning(
+                        "Telemetry package %s rejected with HTTP %s; not retrying",
+                        package_id,
+                        response.status,
+                    )
+                    self._mark(
+                        package_id, token=token, send_state="rejected",
+                        last_error=f"HTTP {response.status}: {response.body[:400]}",
+                    )
+                    return "rejected"
+                if response.status == 429:
+                    return defer(
+                        _retry_after_seconds(response.retry_after, _FAILURE_BACKOFF_SECONDS),
+                        "rate limited",
+                    )
+                # 5xx and anything unexpected: retryable.
+                reason = f"HTTP {response.status}"
             if attempt >= self._max_attempts:
-                self._defer(
-                    package_id, _FAILURE_BACKOFF_SECONDS, reason, token=token
-                )
-                return "deferred"
+                return defer(_FAILURE_BACKOFF_SECONDS, reason)
             self._sleep(self._backoff(attempt))
 
-        self._defer(
-            package_id, _FAILURE_BACKOFF_SECONDS, "attempts exhausted", token=token
-        )
-        return "deferred"
+        return defer(_FAILURE_BACKOFF_SECONDS, "attempts exhausted")
 
     @staticmethod
     def _backoff(attempt: int) -> float:
         """1s, 5s, 25s with full jitter."""
-        ceiling = _BACKOFF_BASE_SECONDS * (_BACKOFF_FACTOR ** (attempt - 1))
-        return random.uniform(0, ceiling)
+        return random.uniform(0, _BACKOFF_BASE_SECONDS * (_BACKOFF_FACTOR ** (attempt - 1)))
 
     # -- entry point -------------------------------------------------------
 
     def send_pending(self) -> SendOutcome:
         """Run one bounded pass. Never raises.
 
-        Claims and sends ONE package at a time so each row's lease only has to cover its own
-        transmission, and re-checks consent before every send so revoking `send` mid-pass stops the
-        remaining packages.
+        Claims and sends ONE package at a time so each row's lease only covers its own
+        transmission, and re-checks consent before every send so revoking `send` mid-pass
+        stops the remaining packages.
         """
         outcome = SendOutcome()
         seen: set[str] = set()
 
         for _ in range(MAX_PACKAGES_PER_PASS):
             if not self._still_consented():
-                # The user turned sending off while this pass was running.
-                # Stop without transmitting anything further, and reconcile
-                # so the window closes at its last confirmed moment. This is
-                # the same single writer every other observation point uses —
-                # not a separate recording mechanism.
+                # Stop transmitting and reconcile through the single consent writer so
+                # the window closes at its last confirmed moment.
                 logger.info("Shared-metrics sending disabled mid-pass; stopping")
                 self._reconcile(send_enabled=False)
                 break
             try:
                 package = self._claim_next(self._now(), seen)
             except Exception:
-                logger.warning(
-                    "Unable to select shared-metrics packages", exc_info=True
-                )
+                logger.warning("Unable to select shared-metrics packages", exc_info=True)
                 break
             if package is None:
                 break
@@ -662,42 +545,29 @@ class SharedMetricsSender:
             seen.add(package["package_id"])
             if package.get("skip"):
                 # Unusable row already marked rejected during the claim.
-                outcome.rejected += 1
-                continue
-
-            try:
-                result = self._send_one(package)
-            except Exception:
-                logger.warning("Unable to send shared-metrics package", exc_info=True)
-                outcome.deferred += 1
-                continue
-            if result == "sent":
-                outcome.sent += 1
-            elif result == "rejected":
-                outcome.rejected += 1
+                result = "rejected"
             else:
-                outcome.deferred += 1
+                try:
+                    result = self._send_one(package)
+                except Exception:
+                    logger.warning("Unable to send shared-metrics package", exc_info=True)
+                    result = "deferred"
+            setattr(outcome, result, getattr(outcome, result) + 1)
         return outcome
 
     def _reconcile(self, *, send_enabled: bool) -> None:
         """Run the single consent writer from within a pass."""
         try:
-            with self._store._connection() as connection:
-                with write_txn(connection):
-                    reconcile_send_consent(
-                        connection, send_enabled, now=self._now()
-                    )
+            with self._write() as connection:
+                reconcile_send_consent(connection, send_enabled, now=self._now())
         except Exception:
-            logger.warning(
-                "Unable to reconcile shared-metrics consent", exc_info=True
-            )
+            logger.warning("Unable to reconcile shared-metrics consent", exc_info=True)
 
     def _still_consented(self) -> bool:
         """Re-read profile-owned send consent.
 
-        Consent is a boundary, not cached config: docs promise ``send: false`` stops transmission
-        immediately, and a pass can run for minutes. Injected senders opt out via
-        consent_check=None.
+        Consent is a boundary, not cached config: docs promise ``send: false`` stops
+        transmission immediately, and a pass can run for minutes.
         """
         if self._consent_check is None:
             return True
@@ -705,8 +575,5 @@ class SharedMetricsSender:
             return bool(self._consent_check())
         except Exception:
             # Fail CLOSED: if consent cannot be established, do not transmit.
-            logger.warning(
-                "Unable to confirm shared-metrics send consent; stopping",
-                exc_info=True,
-            )
+            logger.warning("Unable to confirm shared-metrics send consent; stopping", exc_info=True)
             return False
diff --git a/hermes_cli/observability/shared_metrics_subscriber.py b/hermes_cli/observability/shared_metrics_subscriber.py
index 1104a8a0d3..48e96733b9 100644
--- a/hermes_cli/observability/shared_metrics_subscriber.py
+++ b/hermes_cli/observability/shared_metrics_subscriber.py
@@ -26,6 +26,20 @@ from .shared_metrics_contract import (
 
 logger = logging.getLogger(__name__)
 
+# Contract projections in match order; each yields (metric_name, dimensions) or None.
+_COUNTERS = (
+    client_active_counter,
+    lambda event: _named(MODEL_ROUTE_METRIC, model_call_dimensions(event)),
+    lambda event: _named(TOOL_CALL_METRIC, tool_call_dimensions(event)),
+    task_counter,
+    tool_approval_counter,
+    skill_counter,
+)
+
+
+def _named(metric_name: str, dimensions: dict | None) -> tuple[str, dict] | None:
+    return None if dimensions is None else (metric_name, dimensions)
+
 
 class SharedMetricsSubscriber:
     """Persist validated Hermes counters from Relay lifecycle events."""
@@ -56,17 +70,7 @@ class SharedMetricsSubscriber:
     @staticmethod
     def _classify(event: Any) -> tuple[str, dict] | None:
         """Return ``(metric_name, dimensions)`` for the first matching contract, else None."""
-        metric = client_active_counter(event)
-        if metric is not None:
-            return metric
-        for metric_name, project in (
-            (MODEL_ROUTE_METRIC, model_call_dimensions),
-            (TOOL_CALL_METRIC, tool_call_dimensions),
-        ):
-            dimensions = project(event)
-            if dimensions is not None:
-                return metric_name, dimensions
-        return task_counter(event) or tool_approval_counter(event) or skill_counter(event)
+        return next((m for m in (project(event) for project in _COUNTERS) if m is not None), None)
 
     def __call__(self, event: Any) -> None:
         if self._runtime_id is not None:
@@ -87,14 +91,8 @@ class SharedMetricsSubscriber:
                 if metric_name == CLIENT_ACTIVE_METRIC:
                     self.store.record_client_active(self._client_resource)
                 else:
-                    self.store.record_counter(
-                        metric_name,
-                        dimensions,
-                        self._client_resource,
-                    )
+                    self.store.record_counter(metric_name, dimensions, self._client_resource)
             except Exception:
                 logger.warning(
-                    "Unable to persist the Hermes shared metric: %s",
-                    metric_name,
-                    exc_info=True,
+                    "Unable to persist the Hermes shared metric: %s", metric_name, exc_info=True
                 )