Merge simp/r3-06-p2-IP into simp/r3-06

2026-09-02 23:19:15 -07:00
parent 812820f1f5 7309c3aef3
commit bcc3153e09
7 changed files with 402 additions and 717 deletions
@@ -35,12 +35,8 @@ class GatewayHealthSnapshot:

 _RUNNING_PLATFORM_STATES = {"running", "connected", "ok", "ready"}
 _FATAL_PLATFORM_STATES = {"fatal", "degraded", "error", "failed"}
-_KNOWN_GATEWAY_STATES = {
-    "starting", "draining", "stopping", "stopped", "startup_failed", "unknown"
-} | _RUNNING_PLATFORM_STATES | _FATAL_PLATFORM_STATES
-_KNOWN_PLATFORM_STATES = _RUNNING_PLATFORM_STATES | _FATAL_PLATFORM_STATES | {
-    "connecting", "disconnected", "disabled", "paused", "retrying", "unknown"
-}
+_KNOWN_GATEWAY_STATES = _RUNNING_PLATFORM_STATES | _FATAL_PLATFORM_STATES | {"starting", "draining", "stopping", "stopped", "startup_failed", "unknown"}
+_KNOWN_PLATFORM_STATES = _RUNNING_PLATFORM_STATES | _FATAL_PLATFORM_STATES | {"connecting", "disconnected", "disabled", "paused", "retrying", "unknown"}
 _SUPERVISION_MODES = {"systemd", "s6", "container", "launchd", "manual", "unknown"}
 _SOURCE_LOGGER_RE = re.compile(r"^gateway(?:\.[A-Za-z_][A-Za-z0-9_]*)*$")

@@ -124,6 +120,14 @@ def platform_for_subsystem(subsystem: str) -> Optional[str]:
    return (subsystem.split(".", 1)[1] or None) if subsystem.startswith("platform.") else None


+def _coerce_pid(raw: Any) -> Optional[int]:
+    try:
+        pid = int(raw)
+    except (TypeError, ValueError):
+        return None
+    return pid if pid > 0 else None
+
+
 def _gateway_status(name: str, fallback: Callable[[], Any], /, **kwargs: Any) -> Any:
    """Prefer ``gateway.status.<name>`` (the runtime-status contract); fall back to the local approximation."""
    try:
@@ -133,52 +137,22 @@ def _gateway_status(name: str, fallback: Callable[[], Any], /, **kwargs: Any) ->
        return fallback()


-def _int_or_zero(raw: Any) -> int:
-    try:
-        return max(0, int(raw))
-    except (TypeError, ValueError):
-        return 0
-
-
 def _parse_active_agents(raw: Any) -> int:
-    return _gateway_status("parse_active_agents", lambda: _int_or_zero(raw), raw=raw)
+    def fallback() -> int:
+        try:
+            return max(0, int(raw))
+        except (TypeError, ValueError):
+            return 0
+
+    return _gateway_status("parse_active_agents", fallback, raw=raw)


-def _derive_busy(gateway_running: bool, gateway_state: Any, active_agents: Any) -> bool:
-    fallback = lambda: bool(gateway_running and gateway_state == "running" and _parse_active_agents(active_agents) > 0)  # noqa: E731
-    return _gateway_status("derive_gateway_busy", fallback, gateway_running=gateway_running, gateway_state=gateway_state, active_agents=active_agents)
-
-
-def _derive_drainable(gateway_running: bool, gateway_state: Any) -> bool:
-    fallback = lambda: bool(gateway_running and gateway_state == "running")  # noqa: E731
-    return _gateway_status("derive_gateway_drainable", fallback, gateway_running=gateway_running, gateway_state=gateway_state)
-
-
-def _base_attrs(*, install_id: str, version: str, supervision_mode: str) -> Dict[str, str]:
-    mode = str(supervision_mode or "unknown").lower()
-    return {
-        "service.instance.id": _safe_instance_id(install_id),
-        "service.version": _safe_metric_value(version, limit=64),
-        "hermes.supervision_mode": mode if mode in _SUPERVISION_MODES else "unknown",
-    }
-
-
-def _metric(name: str, value: int | float, attrs: Dict[str, str], **extra: str) -> GatewayMetric:
-    out = dict(attrs)
-    for key, val in extra.items():
-        if val is not None:
-            out[key] = _safe_metric_value(val)
-    return GatewayMetric(name=name, value=value, attributes=out)
-
-
-def _platforms_of(runtime: Optional[dict[str, Any]]) -> dict[str, Any]:
-    raw = (runtime or {}).get("platforms")
+def _dict_or_empty(raw: Any) -> dict[str, Any]:
    return raw if isinstance(raw, dict) else {}


-def _platform_error_code(pdata: dict[str, Any]) -> str:
-    # classify_* is idempotent on its own labels, so error_class == error_code downstream.
-    return classify_gateway_error(pdata.get("error_code") or pdata.get("error_message"))
+def _platforms_of(runtime: Optional[dict[str, Any]]) -> dict[str, Any]:
+    return _dict_or_empty((runtime or {}).get("platforms"))


 def build_gateway_health_snapshot(
@@ -189,32 +163,49 @@ def build_gateway_health_snapshot(
    runtime = runtime or {}
    gateway_state = _bounded_state(runtime.get("gateway_state"), allowed=_KNOWN_GATEWAY_STATES)
    active_agents = _parse_active_agents(runtime.get("active_agents", 0))
-    busy = _derive_busy(gateway_running, gateway_state, active_agents)
-    drainable = _derive_drainable(gateway_running, gateway_state)
+    running = gateway_running and gateway_state == "running"
+    busy = _gateway_status(
+        "derive_gateway_busy", lambda: bool(running and _parse_active_agents(active_agents) > 0),
+        gateway_running=gateway_running, gateway_state=gateway_state, active_agents=active_agents,
+    )
+    drainable = _gateway_status(
+        "derive_gateway_drainable", lambda: bool(running), gateway_running=gateway_running, gateway_state=gateway_state
+    )
    platforms = _platforms_of(runtime)
-    base = _base_attrs(install_id=install_id, version=version, supervision_mode=supervision_mode)
+    mode = str(supervision_mode or "unknown").lower()
+    base = {
+        "service.instance.id": _safe_instance_id(install_id),
+        "service.version": _safe_metric_value(version, limit=64),
+        "hermes.supervision_mode": mode if mode in _SUPERVISION_MODES else "unknown",
+    }
+
+    def metric(name: str, value: int | float, **extra: str) -> GatewayMetric:
+        attrs = dict(base)
+        for key, val in extra.items():
+            if val is not None:
+                attrs[key] = _safe_metric_value(val)
+        return GatewayMetric(name=name, value=value, attributes=attrs)
+
    metrics: list[GatewayMetric] = [
-        _metric("hermes.gateway.up", 1 if gateway_running else 0, base),
-        _metric("hermes.gateway.active_agents", active_agents, base),
-        _metric("hermes.gateway.busy", 1 if busy else 0, base),
-        _metric("hermes.gateway.drainable", 1 if drainable else 0, base),
-        _metric("hermes.gateway.restart_requested", 1 if runtime.get("restart_requested") else 0, base),
-        _metric("hermes.gateway.state", 1, base, **{"hermes.gateway.state": gateway_state}),
+        metric("hermes.gateway.up", int(bool(gateway_running))),
+        metric("hermes.gateway.active_agents", active_agents),
+        metric("hermes.gateway.busy", int(bool(busy))),
+        metric("hermes.gateway.drainable", int(bool(drainable))),
+        metric("hermes.gateway.restart_requested", int(bool(runtime.get("restart_requested")))),
+        metric("hermes.gateway.state", 1, **{"hermes.gateway.state": gateway_state}),
    ]
    fatal_count = 0
    events: list[GatewayHealthEvent | GatewayDiagnosticEvent] = []
+    # classify_* is idempotent on its own labels, so error_class == error_code downstream.
    for platform, pdata in platforms.items():
-        pdata = pdata if isinstance(pdata, dict) else {}
+        pdata = _dict_or_empty(pdata)
        state = _bounded_state(pdata.get("state"), allowed=_KNOWN_PLATFORM_STATES)
-        error_code = _platform_error_code(pdata)
+        error_code = classify_gateway_error(pdata.get("error_code") or pdata.get("error_message"))
        is_degraded = state in _FATAL_PLATFORM_STATES
-        if is_degraded:
-            fatal_count += 1
+        fatal_count += is_degraded
        pattrs = {"hermes.platform": str(platform), "hermes.platform.state": state}
-        metrics.append(_metric("hermes.platform.up", 1 if state in _RUNNING_PLATFORM_STATES else 0, base, **pattrs))
-        metrics.append(_metric(
-            "hermes.platform.degraded", 1 if is_degraded else 0, base, **pattrs, **{"hermes.error_code": error_code}
-        ))
+        metrics.append(metric("hermes.platform.up", int(state in _RUNNING_PLATFORM_STATES), **pattrs))
+        metrics.append(metric("hermes.platform.degraded", int(is_degraded), **pattrs, **{"hermes.error_code": error_code}))
        if is_degraded:
            events.append(GatewayDiagnosticEvent(
                name="platform.fatal", subsystem=f"platform.{platform}", platform=str(platform),
@@ -282,14 +273,13 @@ def _platform_events(
    old_platforms = _platforms_of(previous)
    out: list[GatewayDiagnosticEvent] = []
    for platform, pdata in _platforms_of(current).items():
-        pdata = pdata if isinstance(pdata, dict) else {}
-        prev_raw = old_platforms.get(platform, {})
-        prev = prev_raw if isinstance(prev_raw, dict) else {}
+        pdata = _dict_or_empty(pdata)
+        prev = _dict_or_empty(old_platforms.get(platform, {}))
        old_state = _optional_state(prev.get("state"), allowed=_KNOWN_PLATFORM_STATES)
        new_state = _optional_state(pdata.get("state"), allowed=_KNOWN_PLATFORM_STATES)
        if old_state == new_state or not new_state:
            continue
-        error_code = _platform_error_code(pdata)
+        error_code = classify_gateway_error(pdata.get("error_code") or pdata.get("error_message"))
        common: dict[str, Any] = dict(
            subsystem=f"platform.{platform}", platform=str(platform), error_code=error_code, error_class=error_code,
            profile=profile, version=version, severity="error" if new_state in {"fatal", "failed", "error"} else "warning",
@@ -311,14 +301,6 @@ def emit_runtime_status_transition(previous: Optional[dict[str, Any]], current:
        logger.debug("gateway runtime status transition emit failed", exc_info=True)


-def _coerce_pid(raw: Any) -> Optional[int]:
-    try:
-        pid = int(raw)
-    except (TypeError, ValueError):
-        return None
-    return pid if pid > 0 else None
-
-
 class GatewayDiagnosticLogHandler(logging.Handler):
    """Allowlisted warning/error bridge for gateway-owned diagnostics."""

@@ -39,7 +39,6 @@ from agent.monitoring.redaction import redact_bounded

 logger = logging.getLogger(__name__)

-_DEFAULT_DIAGNOSTIC_SCOPE = "hermes.gateway.diagnostics"
 _METRICS_SDK = (
    "OTLPLogExporter", "OTLPMetricExporter", "Observation", "LogRecord", "LoggerProvider",
    "INVALID_SPAN_ID", "INVALID_TRACE_ID", "TraceFlags", "SeverityNumber",
@@ -90,7 +89,7 @@ class GatewayHealthExportRuntime:
                bus.unsubscribe(sub)
        # Network flush/close runs under one bounded daemon-thread deadline so it can
        # never delay gateway teardown indefinitely.
-        closeables = subscribers + ([self.metric_provider] if self.metric_provider is not None else [])
+        closeables = [item for item in (self.streamer, self.log_streamer, self.metric_provider) if item is not None]

        def _close() -> None:
            for item in closeables:
@@ -112,23 +111,9 @@ def _enabled(config: Dict[str, Any]) -> bool:
    return bool(_gateway_health_config(config).get("enabled") and otlp_exporter.is_enabled(config))


-def _require_metrics_sdk(*, auto_install: bool = True, prompt: bool = False) -> Dict[str, Any]:
-    try:
-        return otlp_exporter._require_sdk(_METRICS_SDK, auto_install=auto_install, prompt=prompt)
-    except Exception as exc:
-        raise RuntimeError(f"OTLP metrics SDK unavailable: {exc}") from exc
-
-
 def _exporter_kwargs(config: Dict[str, Any], signal: str) -> Dict[str, Any]:
    otlp = _otlp_config(config)
-    return {
-        "endpoint": _signal_endpoint(str(otlp.get("endpoint")), signal),
-        "headers": _resolve_headers(otlp.get("headers_env")) or None,
-    }
-
-
-def _resource(config: Dict[str, Any], sdk: Dict[str, Any], telemetry_scope: str) -> Any:
-    return sdk["Resource"].create(_runtime_resource_attributes(config, telemetry_scope=telemetry_scope))
+    return {"endpoint": _signal_endpoint(str(otlp.get("endpoint")), signal), "headers": _resolve_headers(otlp.get("headers_env")) or None}


 # Ordered detection: systemd > s6 > container > launchd > manual (first match wins).
@@ -175,10 +160,8 @@ def _read_background_work_count() -> int:
    runs) deliberately does NOT include: backgrounded ``delegate_task`` subagents,
    ``terminal(background=true)`` processes, kanban workers.  TASK-granular: a fan-out batch of N
    contributes N (real concurrent load), unlike the pool's one-slot-per-batch accounting."""
-    return (
-        _count("background-work async-delegation count failed", "tools.async_delegation", lambda m: m.active_task_count())
-        + _count("background-work process-registry count failed", "tools.process_registry",
-                 lambda m: m.process_registry.count_running())
+    return _count("background-work async-delegation count failed", "tools.async_delegation", lambda m: m.active_task_count()) + _count(
+        "background-work process-registry count failed", "tools.process_registry", lambda m: m.process_registry.count_running()
    )


@@ -203,14 +186,12 @@ def _read_runtime_snapshot(config: Dict[str, Any]):
        logger.warning("background-work snapshot unavailable; metric not exported (error_type=%s)", type(exc).__name__)
        logger.debug("background-work snapshot traceback", exc_info=True)
    try:
-        cron_snapshot = _read_cron_snapshot()
+        gateway_snapshot.metrics.extend(_read_cron_snapshot().metrics)
    except Exception as exc:
        # Cron telemetry silently dropping out is a release-relevant regression: WARN with only
        # the exception *type* (the message could carry paths); exc_info stays on DEBUG.
        logger.warning("cron health snapshot unavailable; cron telemetry not exported (error_type=%s)", type(exc).__name__)
        logger.debug("cron health snapshot traceback", exc_info=True)
-        return gateway_snapshot
-    gateway_snapshot.metrics.extend(cron_snapshot.metrics)
    return gateway_snapshot


@@ -225,11 +206,11 @@ def _emit_snapshot_events(config: Dict[str, Any]) -> None:


 def _start_metric_provider(config: Dict[str, Any], sdk: Dict[str, Any]) -> Any:
-    gh = _gateway_health_config(config)
    exporter = sdk["OTLPMetricExporter"](**_exporter_kwargs(config, "metrics"))
-    interval_ms = max(5, int(gh.get("export_interval_seconds", 60))) * 1000
+    interval_ms = max(5, int(_gateway_health_config(config).get("export_interval_seconds", 60))) * 1000
    reader = sdk["PeriodicExportingMetricReader"](exporter, export_interval_millis=interval_ms)
-    provider = sdk["MeterProvider"](metric_readers=[reader], resource=_resource(config, sdk, "gateway_health"))
+    resource = sdk["Resource"].create(_runtime_resource_attributes(config, telemetry_scope="gateway_health"))
+    provider = sdk["MeterProvider"](metric_readers=[reader], resource=resource)
    meter = provider.get_meter("hermes.gateway.health")
    Observation = sdk["Observation"]

@@ -248,24 +229,22 @@ def _start_metric_provider(config: Dict[str, Any], sdk: Dict[str, Any]) -> Any:
    return provider


-_SEVERITY_NAMES = {
-    "critical": "FATAL", "fatal": "FATAL", "error": "ERROR", "info": "INFO", "information": "INFO", "debug": "DEBUG",
-}
+_SEVERITY_NAMES = {"critical": "FATAL", "fatal": "FATAL", "error": "ERROR", "info": "INFO", "information": "INFO", "debug": "DEBUG"}


 def _severity_number(sdk: Dict[str, Any], severity: Any) -> Any:
-    sev = str(severity or "warning").lower()
-    return getattr(sdk["SeverityNumber"], _SEVERITY_NAMES.get(sev, "WARN"))
+    return getattr(sdk["SeverityNumber"], _SEVERITY_NAMES.get(str(severity or "warning").lower(), "WARN"))


 class GatewayDiagnosticLogStreamer(EmitterStreamer):
    """Emitter subscriber that sends gateway diagnostic events as OTLP logs."""

    def __init__(self, config: Dict[str, Any], sdk: Dict[str, Any]):
-        self._provider = sdk["LoggerProvider"](resource=_resource(config, sdk, "gateway_diagnostics"))
+        resource = sdk["Resource"].create(_runtime_resource_attributes(config, telemetry_scope="gateway_diagnostics"))
+        self._provider = sdk["LoggerProvider"](resource=resource)
        self._processor = sdk["BatchLogRecordProcessor"](sdk["OTLPLogExporter"](**_exporter_kwargs(config, "logs")))
        self._provider.add_log_record_processor(self._processor)
-        self._logger = self._provider.get_logger(_DEFAULT_DIAGNOSTIC_SCOPE)
+        self._logger = self._provider.get_logger("hermes.gateway.diagnostics")
        self._sdk = sdk
        self.exported = 0

@@ -288,33 +267,16 @@ class GatewayDiagnosticLogStreamer(EmitterStreamer):
            self.exported += 1


-def _start_snapshot_thread(config: Dict[str, Any], stop_event: threading.Event) -> threading.Thread:
-    interval = max(5, int(_gateway_health_config(config).get("logs_export_interval_seconds", 5)))
-
-    def _run() -> None:
-        while not stop_event.wait(interval):
-            _emit_snapshot_events(config)
-
-    thread = threading.Thread(target=_run, name="hermes-gateway-health-export", daemon=True)
-    thread.start()
-    return thread
-
-
-def _attach_log_handler(config: Dict[str, Any]) -> Any:
-    gh = _gateway_health_config(config)
-    if not gh.get("diagnostic_events_enabled", True) or not gh.get("warning_error_events_enabled", True):
-        return None
-    handler = GatewayDiagnosticLogHandler(profile=_profile(), version=_version())
-    root = logging.getLogger()
-    if handler not in root.handlers:
-        root.addHandler(handler)
-    return handler
-
-
 def _gateway_health_event(ev: Dict[str, Any]) -> bool:
    return ev.get("event") in {"gateway_health", "cron_execution"}


+def _fail(runtime: GatewayHealthExportRuntime, log: Callable[..., None], msg: str, reason: str) -> GatewayHealthExportRuntime:
+    log(msg, exc_info=True)
+    runtime.shutdown()
+    return GatewayHealthExportRuntime(enabled=False, reason=reason)
+
+
 def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRuntime:
    """Start P0 gateway health export if configured. Never raises."""
    if not _enabled(config):
@@ -326,7 +288,7 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu
    sdk: Optional[Dict[str, Any]] = None
    if metrics_on or diagnostics_on:
        try:
-            sdk = _require_metrics_sdk(prompt=False)
+            sdk = otlp_exporter._require_sdk(_METRICS_SDK, auto_install=True, prompt=False)
        except Exception:
            logger.warning("monitoring.gateway_health_export.enabled but OTLP SDK is unavailable; install 'hermes-agent[otlp]'", exc_info=True)
            return GatewayHealthExportRuntime(enabled=False, reason="otlp_unavailable")
@@ -334,9 +296,7 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu
        try:
            runtime.metric_provider = _start_metric_provider(config, sdk)
        except Exception:
-            logger.warning("gateway health OTLP metrics failed to start", exc_info=True)
-            runtime.shutdown()
-            return GatewayHealthExportRuntime(enabled=False, reason="metrics_start_failed")
+            return _fail(runtime, logger.warning, "gateway health OTLP metrics failed to start", "metrics_start_failed")
    if diagnostics_on and sdk is not None:
        try:
            runtime.streamer = otlp_exporter.start_streaming(config, event_filter=_gateway_health_event)
@@ -346,18 +306,29 @@ def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRu
            emitter.get_emitter().subscribe(log_streamer)
            runtime.log_streamer = log_streamer
        except Exception:
-            logger.debug("gateway diagnostic OTLP export failed to start", exc_info=True)
-            runtime.shutdown()
-            return GatewayHealthExportRuntime(enabled=False, reason="diagnostics_start_failed")
+            return _fail(runtime, logger.debug, "gateway diagnostic OTLP export failed to start", "diagnostics_start_failed")
    try:
-        runtime.log_handler = _attach_log_handler(config)
+        if diagnostics_on and gh.get("warning_error_events_enabled", True):
+            handler = GatewayDiagnosticLogHandler(profile=_profile(), version=_version())
+            root = logging.getLogger()
+            if handler not in root.handlers:
+                root.addHandler(handler)
+            runtime.log_handler = handler
    except Exception:
        logger.debug("gateway diagnostic log handler failed to attach", exc_info=True)
    if diagnostics_on:
        try:
            _emit_snapshot_events(config)
-            runtime.stop_event = threading.Event()
-            runtime.thread = _start_snapshot_thread(config, runtime.stop_event)
+            stop_event = runtime.stop_event = threading.Event()
+            interval = max(5, int(gh.get("logs_export_interval_seconds", 5)))
+
+            def _run() -> None:
+                while not stop_event.wait(interval):
+                    _emit_snapshot_events(config)
+
+            thread = threading.Thread(target=_run, name="hermes-gateway-health-export", daemon=True)
+            thread.start()
+            runtime.thread = thread
        except Exception:
            logger.debug("gateway health snapshot thread failed to start", exc_info=True)
    return runtime
@@ -69,8 +69,6 @@ class WebhookTarget(_ToolMatcherMixin):
        return self.name or self.url


-# --- Public API -----------------------------------------------------------------
-
 def register_from_config(cfg: Optional[Dict[str, Any]]) -> List[WebhookTarget]:
    """Register every configured outbound webhook on the plugin manager.  Malformed ``hooks.outbound``
    means zero targets — never raises.  Returns the targets that ended up wired (deduplicated)."""
@@ -117,8 +115,7 @@ def iter_configured_targets(cfg: Optional[Dict[str, Any]]) -> List[WebhookTarget
    if not isinstance(raw, list):
        logger.warning("hooks.outbound must be a list of webhook targets; got %s", type(raw).__name__)
        return []
-    targets = (_parse_single_target(i, entry) for i, entry in enumerate(raw))
-    return [t for t in targets if t is not None]
+    return [t for t in (_parse_single_target(i, entry) for i, entry in enumerate(raw)) if t is not None]


 def flush(timeout: float = 5.0) -> bool:
@@ -154,8 +151,6 @@ def reset_for_tests() -> None:
        pass


-# --- Config parsing -------------------------------------------------------------
-
 def _parse_single_target(index: int, raw: Any) -> Optional[WebhookTarget]:
    from hermes_cli.plugins import VALID_HOOKS

@@ -201,46 +196,34 @@ def _parse_single_target(index: int, raw: Any) -> Optional[WebhookTarget]:
        warn(".timeout must be an int (got %r); using default %ds", timeout_raw, DEFAULT_TIMEOUT_SECONDS)
        timeout = DEFAULT_TIMEOUT_SECONDS
    name = raw.get("name")
+    # ``secret_env`` (env var name, preferred) wins over inline ``secret``.
+    secret_env = raw.get("secret_env")
+    if isinstance(secret_env, str) and secret_env.strip():
+        secret = os.environ.get(secret_env.strip(), "") or None
+        if secret is None:
+            warn(".secret_env=%r is not set in the environment — deliveries will be UNSIGNED", secret_env.strip())
+    else:
+        secret = raw.get("secret")
+        secret = secret if isinstance(secret, str) and secret else None
    return WebhookTarget(
-        url=url, events=events, name=name.strip() if isinstance(name, str) else "", secret=_resolve_secret(index, raw),
+        url=url, events=events, name=name.strip() if isinstance(name, str) else "", secret=secret,
        matcher=matcher, timeout=max(1, min(timeout, MAX_TIMEOUT_SECONDS)),
    )


-def _resolve_secret(index: int, raw: Dict[str, Any]) -> Optional[str]:
-    """``secret_env`` (env var name, preferred) wins over inline ``secret``."""
-    secret_env = raw.get("secret_env")
-    if isinstance(secret_env, str) and secret_env.strip():
-        value = os.environ.get(secret_env.strip(), "")
-        if value:
-            return value
-        logger.warning(
-            "hooks.outbound[%d].secret_env=%r is not set in the environment — deliveries will be UNSIGNED",
-            index, secret_env.strip(),
-        )
-        return None
-    secret = raw.get("secret")
-    return secret if isinstance(secret, str) and secret else None
-
-
-# --- Callback + delivery --------------------------------------------------------
-
 def _make_callback(event: str, target: WebhookTarget):
    """Build the notify-only closure ``invoke_hook()`` calls per firing."""

    def _callback(**kwargs: Any) -> None:
        if event in _TOOL_SCOPED_EVENTS and not target.matches_tool(kwargs.get("tool_name")):
-            return None
+            return
        delivery_id = uuid.uuid4().hex
        try:
            body = _serialize_payload(event, kwargs, delivery_id)
        except Exception:  # a bad payload must not hurt the loop
-            logger.warning(
-                "outbound webhook payload serialization failed (event=%s target=%s)", event, target.label, exc_info=True,
-            )
-            return None
+            logger.warning("outbound webhook payload serialization failed (event=%s target=%s)", event, target.label, exc_info=True)
+            return
        _enqueue(_build_delivery(event, target, body, delivery_id))
-        return None

    _callback.__name__ = f"outbound_webhook[{event}:{target.label}]"
    _callback.__qualname__ = _callback.__name__
@@ -268,14 +251,19 @@ def _build_delivery(event: str, target: WebhookTarget, body: bytes, delivery_id:
    if target.secret:
        digest = hmac.new(target.secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
        headers["X-Hermes-Signature-256"] = f"sha256={digest}"
-    return {
-        "url": target.url, "label": target.label, "event": event,
-        "body": body, "headers": headers, "timeout": target.timeout,
-    }
+    return {"url": target.url, "label": target.label, "event": event, "body": body, "headers": headers, "timeout": target.timeout}


 def _enqueue(delivery: Dict[str, Any]) -> None:
-    _ensure_worker()
+    global _worker
+    if _worker is None or not _worker.is_alive():
+        with _worker_lock:
+            if _worker is None or not _worker.is_alive():
+                _worker = threading.Thread(target=_worker_loop, name="outbound-webhooks", daemon=True)
+                _worker.start()
+                # Daemon worker: a short-lived process could exit right after enqueuing on_session_end.
+                # Drain at interpreter shutdown, bounded so a dead endpoint can only delay exit, never hang it.
+                atexit.register(flush, timeout=5.0)
    try:
        _delivery_queue.put_nowait(delivery)
    except queue.Full:
@@ -285,20 +273,6 @@ def _enqueue(delivery: Dict[str, Any]) -> None:
        )


-def _ensure_worker() -> None:
-    global _worker
-    if _worker is not None and _worker.is_alive():
-        return
-    with _worker_lock:
-        if _worker is not None and _worker.is_alive():
-            return
-        _worker = threading.Thread(target=_worker_loop, name="outbound-webhooks", daemon=True)
-        _worker.start()
-        # Daemon worker: a short-lived process could exit right after enqueuing on_session_end.
-        # Drain at interpreter shutdown, bounded so a dead endpoint can only delay exit, never hang it.
-        atexit.register(flush, timeout=5.0)
-
-
 def _worker_loop() -> None:
    while True:
        delivery = _delivery_queue.get()
@@ -306,11 +280,8 @@ def _worker_loop() -> None:
            if delivery is not None:
                _deliver(delivery)
        except Exception:  # pragma: no cover — defensive
-            logger.warning(
-                "outbound webhook delivery crashed (target=%s)",
-                delivery.get("label") if isinstance(delivery, dict) else "?",
-                exc_info=True,
-            )
+            label = delivery.get("label") if isinstance(delivery, dict) else "?"
+            logger.warning("outbound webhook delivery crashed (target=%s)", label, exc_info=True)
        finally:
            _delivery_queue.task_done()

@@ -49,6 +49,11 @@ def _median(values) -> int:
    return ordered[len(ordered) // 2]


+def _median_box_size(boxes) -> tuple[int, int]:
+    """``(median width, median height)`` of ``(l, t, r, b)`` *boxes*, each at least 1."""
+    return max(1, _median(r - l for l, _t, r, _b in boxes)), max(1, _median(b - t for _l, t, _r, b in boxes))
+
+
 def _blank(size=(CELL_WIDTH, CELL_HEIGHT)):
    return Image.new("RGBA", size, (0, 0, 0, 0))

@@ -111,26 +116,10 @@ def _unvisited_components(w: int, h: int, visited: bytearray, accept):
            yield _flood(w, h, visited, [(x, y)], accept)


-def _has_transparency(image) -> bool:
-    """True if the strip already carries a real alpha background (>5% of pixels at/below the floor)."""
-    alpha = image.getchannel("A")
-    return alpha.getextrema()[0] <= _ALPHA_FLOOR and sum(alpha.histogram()[: _ALPHA_FLOOR + 1]) > image.width * image.height * 0.05
-
-
-def _dominant_corner_color(image) -> tuple[int, int, int]:
-    """Most common opaque color among the four corners."""
-    w, h = image.width, image.height
-    px = image.load()
-    corners = [px[x, y] for x, y in ((0, 0), (w - 1, 0), (0, h - 1), (w - 1, h - 1))]
-    counter = Counter(tuple(c[:3]) for c in corners if c[3] > _ALPHA_FLOOR)
-    return counter.most_common(1)[0][0] if counter else (0, 255, 0)
-
-
 def _near_key_mask(image, key: tuple[int, int, int], tol: int = 48):
    """``L`` mask, 255 where a pixel is within *tol* per-channel of *key*.

-    Tight on purpose: marks only near-pure backdrop so trapped chroma pockets
-    seed the flood while chroma-tinted character pixels stay outside it.
+    Tight on purpose: marks only near-pure backdrop so trapped chroma pockets seed the flood while chroma-tinted character pixels stay outside it.
    """
    r, g, b = (ch.point(lambda v, k=k: 255 if abs(v - k) <= tol else 0) for ch, k in zip(image.split()[:3], key))
    return ImageChops.darker(ImageChops.darker(r, g), b)
@@ -139,8 +128,7 @@ def _near_key_mask(image, key: tuple[int, int, int], tol: int = 48):
 def _remove_masked(rgba, mask):
    """Clear the pixels *mask* (``L``, 255 = remove) selects, then erode alpha by 1px (3x3 min).

-    The erosion drops the antialiased key/sprite blend ring (too far from the key to
-    match); the sprite's own thick outline keeps the silhouette intact.
+    The erosion drops the antialiased key/sprite blend ring (too far from the key to match); the sprite's thick outline keeps the silhouette.
    """
    out = Image.composite(_blank(rgba.size), rgba, mask)
    out.putalpha(out.getchannel("A").filter(ImageFilter.MinFilter(3)))
@@ -150,23 +138,26 @@ def _remove_masked(rgba, mask):
 def remove_background(image, *, chroma_key: tuple[int, int, int] | None = None, threshold: float = 90.0):
    """Return *image* (RGBA) with its flat background keyed out to transparent.

-    Already-transparent strips are left alone (holes repaired). Otherwise key out
-    *chroma_key* (or the dominant corner color) from the border inward: a global
-    color match punched holes wherever an interior highlight matched the backdrop.
+    Strips already carrying a real alpha background (>5% of pixels at/below the floor)
+    are left alone (holes repaired). Otherwise key out *chroma_key* (or the most common
+    opaque corner color) from the border inward: a global color match punched holes
+    wherever an interior highlight matched the backdrop.
    """
    rgba = image.convert("RGBA")
-    if _has_transparency(rgba):
+    w, h = rgba.size
+    alpha = rgba.getchannel("A")
+    if alpha.getextrema()[0] <= _ALPHA_FLOOR and sum(alpha.histogram()[: _ALPHA_FLOOR + 1]) > w * h * 0.05:
        return _repair_internal_alpha_holes(rgba)
-    key = chroma_key or _dominant_corner_color(rgba)
+    px = rgba.load()
+    if not (key := chroma_key):
+        corners = Counter(tuple(px[x, y][:3]) for x, y in ((0, 0), (w - 1, 0), (0, h - 1), (w - 1, h - 1)) if px[x, y][3] > _ALPHA_FLOOR)
+        key = corners.most_common(1)[0][0] if corners else (0, 255, 0)
    # Fast path for saturated chroma keys (our prompts use hot magenta): C-level
    # channel ops clear border backdrop and enclosed pockets alike, no Python flood.
    if max(key) - min(key) >= 120:
        opaque = rgba.getchannel("A").point(lambda a: 255 if a > _ALPHA_FLOOR else 0)
        return _remove_masked(rgba, ImageChops.darker(_near_key_mask(rgba, key), opaque))

-    w, h = rgba.size
-    px = rgba.load()
-
    def _is_bg(x: int, y: int) -> bool:
        r, g, b, a = px[x, y]
        return a > _ALPHA_FLOOR and math.sqrt((r - key[0]) ** 2 + (g - key[1]) ** 2 + (b - key[2]) ** 2) <= threshold
@@ -183,17 +174,13 @@ def remove_background(image, *, chroma_key: tuple[int, int, int] | None = None,
 def _repair_internal_alpha_holes(image):
    """Fill transparent islands fully enclosed by opaque sprite pixels.

-    Some providers return "transparent" PNGs with swiss-cheese alpha inside the
-    character; enclosed holes get the average color of their opaque neighbours.
+    Some providers return "transparent" PNGs with swiss-cheese alpha inside the character; enclosed holes take the average opaque-neighbour color.
    """
    rgba = image.convert("RGBA")
    w, h = rgba.size
    px = rgba.load()
    visited = bytearray(w * h)
-
-    def _is_transparent(x: int, y: int) -> bool:
-        return px[x, y][3] <= _ALPHA_FLOOR
-
+    _is_transparent = lambda x, y: px[x, y][3] <= _ALPHA_FLOOR  # noqa: E731
    _border_flood(w, h, visited, _is_transparent)  # edge-connected transparency = background
    for hole in _unvisited_components(w, h, visited, _is_transparent):
        seen = set(hole)
@@ -224,8 +211,7 @@ def _fit_to_cell(image):
 def _drop_side_bleed(image):
    """Remove tiny separated left/right lobes (neighbour-pose slivers) before fitting.

-    Component extraction may have grouped a near sliver with the subject; the
-    column profile still shows it as a low-mass lobe. Only those go, so wide poses survive.
+    Component extraction may group a near sliver with the subject; the column profile still shows it as a low-mass lobe. Only those go (wide poses survive).
    """
    rgba = image.convert("RGBA")
    w, h = rgba.size
@@ -270,11 +256,7 @@ def _component_boxes(image) -> list[tuple[tuple[int, int, int, int], int]]:
    w, h = r0 - l0, b0 - t0
    alpha = rgba.getchannel("A").load()
    components = _unvisited_components(w, h, bytearray(w * h), lambda x, y: alpha[l0 + x, t0 + y] > _ALPHA_FLOOR)
-    out: list[tuple[tuple[int, int, int, int], int]] = []
-    for pixels in components:
-        xs, ys = zip(*pixels)
-        out.append(((l0 + min(xs), t0 + min(ys), l0 + max(xs) + 1, t0 + max(ys) + 1), len(pixels)))
-    return out
+    return [((l0 + min(xs), t0 + min(ys), l0 + max(xs) + 1, t0 + max(ys) + 1), len(xs)) for xs, ys in map(lambda p: zip(*p), components)]


 def _isolate_slot_subject(image):
@@ -286,17 +268,13 @@ def _isolate_slot_subject(image):
    main_box, main_mass = max(comps, key=lambda item: item[1])
    ml, _mt, mr, _mb = main_box
    mw = max(1, mr - ml)
-
-    def _keep(box, mass) -> bool:
+    out = _blank(rgba.size)
+    for box, mass in comps:
        # Keep attached-looking accessories (halos); drop sparkles/tears/noise.
        left, _top, right, _bottom = box
        overlap = max(0, min(right, mr) - max(left, ml))
        near_main = (ml - mw * 0.25) <= (left + right) / 2 <= (mr + mw * 0.25)
-        return box == main_box or (mass >= max(24, main_mass * 0.035) and (overlap >= mw * 0.3 or near_main))
-
-    out = _blank(rgba.size)
-    for box, mass in comps:
-        if _keep(box, mass):
+        if box == main_box or (mass >= max(24, main_mass * 0.035) and (overlap >= mw * 0.3 or near_main)):
            out.alpha_composite(rgba.crop(box), (box[0], box[1]))
    return out

@@ -314,7 +292,7 @@ def _group_component_rows(boxes: list[tuple[int, int, int, int]]) -> list[list[t
    if not boxes:
        return []
    cy = lambda b: (b[1] + b[3]) / 2  # noqa: E731
-    row_tol = max(12, _median(max(1, b[3] - b[1]) for b in boxes) * 0.55)
+    row_tol = max(12, _median_box_size(boxes)[1] * 0.55)
    rows: list[list[tuple[int, int, int, int]]] = []
    centers: list[float] = []
    for box in sorted(boxes, key=cy):
@@ -334,10 +312,8 @@ def _group_component_rows(boxes: list[tuple[int, int, int, int]]) -> list[list[t
 def _merge_related_boxes(boxes: list[tuple[int, int, int, int]]) -> list[tuple[int, int, int, int]]:
    """Merge disconnected parts of one subject (capes, tails, props) on the same row.

-    Merges when vertical spans overlap and the horizontal gap is tiny relative to
-    the component size; never bridges the larger gaps between separate poses.
+    Merges when vertical spans overlap and the horizontal gap is tiny relative to component size; never bridges the larger gaps between separate poses.
    """
-
    def related(a, b) -> bool:
        (al, at, ar, ab), (bl, bt, br, bb) = a, b
        v_overlap, min_h = max(0, min(ab, bb) - max(at, bt)), max(1, min(ab - at, bb - bt))
@@ -354,9 +330,8 @@ def _merge_related_boxes(boxes: list[tuple[int, int, int, int]]) -> list[tuple[i
            if used[i]:
                continue
            used[i] = True
-            for j in range(i + 1, len(boxes)):
-                if not used[j] and related(a, boxes[j]):
-                    b = boxes[j]
+            for j, b in enumerate(boxes[i + 1 :], i + 1):
+                if not used[j] and related(a, b):
                    a = (min(a[0], b[0]), min(a[1], b[1]), max(a[2], b[2]), max(a[3], b[3]))
                    used[j] = changed = True
            merged.append(a)
@@ -366,7 +341,6 @@ def _merge_related_boxes(boxes: list[tuple[int, int, int, int]]) -> list[tuple[i

 def _component_crops(strip, frame_count: int, *, require_padding: bool = False) -> list | None:
    """Frames as connected subjects in reading order (robust to 2D grids); ``None`` when *frame_count* can't be met."""
-
    def attempt(source) -> list | None:
        subjects = _significant_subject_boxes(source, min_mass=64)
        if len(subjects) < frame_count:
@@ -434,8 +408,7 @@ def _content_runs(profile: list[int], *, threshold: int = 2) -> list[tuple[tuple
 def _frame_x_ranges(strip, frame_count: int) -> list[tuple[int, int]] | None:
    """Per-frame ``(left, right)`` column ranges from the row's empty gutters.

-    Extra spans merge across the smallest gaps (a detached halo sits closer to its
-    body than to the next pose); fewer spans than frames → ``None`` (poses touching).
+    Extra spans merge across the smallest gaps (a detached halo sits closer to its body than to the next pose); fewer spans than frames → ``None``.
    """
    runs = _content_runs(_column_profile(strip))
    groups = [[l, r] for (l, r), m in runs if m >= max(m for _run, m in runs) * 0.02] if runs else []
@@ -475,8 +448,7 @@ def _validate_extracted_frames(frames: list, frame_count: int) -> None:
        boxes.append(bbox)
    if frame_count <= 1:
        return
-    med_w = max(1, _median(b[2] - b[0] for b in boxes))
-    med_h = max(1, _median(b[3] - b[1] for b in boxes))
+    med_w, med_h = _median_box_size(boxes)
    for i, (left, top, right, bottom) in enumerate(boxes):
        if _is_multi_pose_outlier(right - left, bottom - top, med_w, med_h):
            raise ValueError(f"frame {i} is a multi-pose width outlier")
@@ -487,16 +459,15 @@ def extract_strip_frames(
 ) -> list:
    """Turn one generated row strip into *frame_count* frames.

-    Keys out the background, then isolates padded subjects (components, then equal
-    slots). When that fails ``components`` raises while ``auto`` salvages leniently.
-    *fit* centers each frame into a cell; hatching passes ``fit=False`` so
+    Keys out the background, then isolates padded subjects (components, then equal slots). When that fails ``components``
+    raises while ``auto`` salvages leniently. *fit* centers each frame into a cell; hatching passes ``fit=False`` so
    :func:`normalize_cells` can register the whole pet with one shared scale.
    """
    strip = remove_background(_load_rgba(strip), chroma_key=chroma_key)
    frames = _component_crops(strip, frame_count, require_padding=True) or _slot_crops(strip, frame_count, require_padding=True)
-    if frames is None and method == "components":
-        raise ValueError(f"could not segment {frame_count} padded sprites from strip")
    if frames is None:
+        if method == "components":
+            raise ValueError(f"could not segment {frame_count} padded sprites from strip")
        frames = _component_crops(strip, frame_count, require_padding=False)
    if frames is None:
        frames = _salvage_frames(strip, frame_count)
@@ -506,8 +477,7 @@ def extract_strip_frames(

 def _salvage_frames(strip, frame_count: int) -> list:
    """Lenient last resort: gutter ranges (severing expected gutters if needed), else raw slots."""
-    source = strip
-    ranges = _frame_x_ranges(source, frame_count)
+    source, ranges = strip, _frame_x_ranges(strip, frame_count)
    if ranges is None:
        source = _sever_expected_gutters(strip, frame_count)
        ranges = _frame_x_ranges(source, frame_count)
@@ -526,10 +496,7 @@ def _column_profile(image) -> list[int]:
 def _best_shift(ref: list[int], prof: list[int], window: int) -> int:
    """Integer dx that best aligns *prof* onto *ref* (1-D cross-correlation; the body dominates, limbs barely move it)."""
    n = len(ref)
-
-    def score(d: int) -> int:
-        return sum(ref[x] * prof[x - d] for x in range(max(0, d), min(n, n + d)))
-
+    score = lambda d: sum(ref[x] * prof[x - d] for x in range(max(0, d), min(n, n + d)))  # noqa: E731
    return max(range(-window, window + 1), key=score)  # ties → smallest dx


@@ -554,9 +521,7 @@ def normalize_cells(frames_by_state: dict[str, list], *, pad: int = _NORMALIZE_P
        profiles = [_column_profile(f) for f in canvas]
        ref = [_median(p[x] for p in profiles) for x in range(w0)]
        window = max(8, w0 // 5)
-        aligned = [
-            _place(f, (w0 + 2 * window, h0), (window + _best_shift(ref, prof, window), 0)) for f, prof in zip(canvas, profiles)
-        ]
+        aligned = [_place(f, (w0 + 2 * window, h0), (window + _best_shift(ref, prof, window), 0)) for f, prof in zip(canvas, profiles)]
        boxes = [b for b in (a.getbbox() for a in aligned) if b]
        union = (min(b[0] for b in boxes), min(b[1] for b in boxes), max(b[2] for b in boxes), max(b[3] for b in boxes))
        prepared[state] = (aligned, union, _median(b[3] - b[1] for b in boxes))
@@ -603,8 +568,7 @@ def compose_atlas(frames_by_state: dict[str, list]):
    atlas = _blank((ATLAS_WIDTH, ATLAS_HEIGHT))
    for state, row, count in ROW_SPECS:
        for col, frame in enumerate((frames_by_state.get(state) or [])[:count]):
-            cell = frame.convert("RGBA")
-            cell = cell if cell.size == (CELL_WIDTH, CELL_HEIGHT) else _fit_to_cell(cell)
+            cell = cell if (cell := frame.convert("RGBA")).size == (CELL_WIDTH, CELL_HEIGHT) else _fit_to_cell(cell)
            atlas.alpha_composite(cell, (col * CELL_WIDTH, row * CELL_HEIGHT))
    return _clear_transparent_rgb(atlas)

@@ -621,9 +585,7 @@ def validate_atlas(atlas) -> dict:

 def _check_atlas_cells(atlas) -> tuple[list[str], list[str], list[str]]:
    """Occupancy/collapse/residue checks for a correctly-sized atlas → ``(errors, warnings, filled_states)``."""
-    errors: list[str] = []
-    warnings: list[str] = []
-    filled_states: list[str] = []
+    errors, warnings, filled_states = [], [], []
    cell_boxes_by_state: dict[str, list[tuple[int, int, int, int]]] = {}
    for state, row, count in ROW_SPECS:
        cells = [atlas.crop((c * CELL_WIDTH, row * CELL_HEIGHT, (c + 1) * CELL_WIDTH, (row + 1) * CELL_HEIGHT)) for c in range(count)]
@@ -638,16 +600,14 @@ def _check_atlas_cells(atlas) -> tuple[list[str], list[str], list[str]]:
    # A valid pet must occupy the cell: one bad row can poison global
    # normalization and shrink every state while still passing "non-empty".
    all_boxes = [b for boxes in cell_boxes_by_state.values() for b in boxes]
-    global_med_w = _median(r - l for l, _t, r, _b in all_boxes) if all_boxes else 0
-    global_med_h = _median(b - t for _l, t, _r, b in all_boxes) if all_boxes else 0
+    global_med_w, global_med_h = _median_box_size(all_boxes) if all_boxes else (0, 0)
    if all_boxes and global_med_h < max(56, round(CELL_HEIGHT * 0.28)):
        errors.append(f"atlas sprites are too small after normalization (median frame height {global_med_h}px)")
    for state, boxes in cell_boxes_by_state.items():
        if len(boxes) <= 1:
            continue
-        widths, heights = [r - l for l, _t, r, _b in boxes], [b - t for _l, t, _r, b in boxes]
-        med_w, med_h = max(1, _median(widths)), max(1, _median(heights))
-        if _is_multi_pose_outlier(max(widths), max(heights), med_w, med_h):
+        med_w, med_h = _median_box_size(boxes)
+        if _is_multi_pose_outlier(max(r - l for l, _t, r, _b in boxes), max(b - t for _l, t, _r, b in boxes), med_w, med_h):
            errors.append(f"state '{state}' contains a multi-pose frame outlier")
        # Per-state collapse guard: one malformed row must not pass on the
        # strength of the healthy ones.
@@ -18,7 +18,7 @@ import sys
 import zlib
 from dataclasses import KW_ONLY, dataclass
 from functools import lru_cache
-from itertools import groupby
+from itertools import groupby, takewhile
 from pathlib import Path

 from agent.pet.constants import DEFAULT_SCALE, FRAME_H, FRAME_W, FRAMES_PER_STATE, PetState, state_row_index
@@ -43,10 +43,8 @@ def detect_terminal_graphics() -> str:
    # half-blocks; users who enabled them can pin display.pet.render_mode.
    if term_program == "vscode":
        return "unicode"
-    if os.environ.get("KITTY_WINDOW_ID") or "kitty" in term or "ghostty" in term or term_program == "ghostty":
-        return "kitty"
-    if _is_wezterm():  # speaks kitty and iterm; kitty has richer placement
-        return "kitty"
+    if os.environ.get("KITTY_WINDOW_ID") or "kitty" in term or "ghostty" in term or term_program == "ghostty" or _is_wezterm():
+        return "kitty"  # WezTerm speaks kitty and iterm; kitty has richer placement
    if term_program == "iterm.app" or os.environ.get("ITERM_SESSION_ID"):
        return "iterm"
    if term_program == "mintty" or "foot" in term or "mlterm" in term or "sixel" in term:
@@ -63,11 +61,9 @@ def resolve_mode(configured: str | None, *, stream=None) -> str:
    """Effective render mode from ``display.pet.render_mode`` + env; ``off`` when not a TTY."""
    mode = (configured or "auto").strip().lower()
    mode = mode if mode in RENDER_MODES else "auto"
-    if mode == "off":
-        return "off"
    stream = stream or sys.stdout
    try:
-        if not (hasattr(stream, "isatty") and stream.isatty()):
+        if mode == "off" or not (hasattr(stream, "isatty") and stream.isatty()):
            return "off"
    except (ValueError, OSError):
        return "off"
@@ -93,13 +89,8 @@ def _raw_frames(sheet_path: str, state_value: str, frame_w: int, frame_h: int, f
        cols, rows = max(1, sheet.width // frame_w), max(1, sheet.height // frame_h)
        # Clamp to the sheet: some pets ship fewer rows than the taxonomy reserves.
        top = min(state_row_index(state_value, rows) * frame_h, max(0, sheet.height - frame_h))
-        frames = []
-        for i in range(min(frames_per_state, cols)):
-            frame = sheet.crop((i * frame_w, top, (i + 1) * frame_w, top + frame_h))
-            if _frame_is_blank(frame):
-                break
-            frames.append(frame)
-        return tuple(frames)
+        crops = (sheet.crop((i * frame_w, top, (i + 1) * frame_w, top + frame_h)) for i in range(min(frames_per_state, cols)))
+        return tuple(takewhile(lambda f: not _frame_is_blank(f), crops))
    except Exception as exc:  # noqa: BLE001 - cosmetic feature, never fatal
        logger.debug("pet frame decode failed (%s, %s): %s", sheet_path, state_value, exc)
        return ()
@@ -129,27 +120,23 @@ def _png_b64(frame) -> str:
    return base64.standard_b64encode(buf.getvalue()).decode("ascii")


-def _crop_frames_to_alpha_union(frames):
-    """Crop every frame to the union opaque bbox: kitty paints transparent margins too, so an untrimmed pet looks small and adrift."""
-    boxes = [b for b in (f.getchannel("A").getbbox() for f in frames) if b]
-    if not boxes:
-        return frames
-    union = (min(b[0] for b in boxes), min(b[1] for b in boxes), max(b[2] for b in boxes), max(b[3] for b in boxes))
-    return [f.crop(union) for f in frames]
-
-
 # Nominal terminal cell size in pixels. kitty fits an image to its cell rectangle
 # preserving aspect, so a frame that isn't a whole cell multiple rounds up, clipping
 # the bottom row ("clipped feet"); snapping to an exact multiple avoids that.
 _CELL_W, _CELL_H = 8, 16


-def _snap_frames_to_cell_grid(frames):
-    """Resize frames so width/height are exact multiples of the cell box (all frames share the union-cropped size)."""
-    if not frames:
-        return frames
+def _fit_frames_to_cell_grid(frames):
+    """Crop *frames* (non-empty) to their union opaque bbox, then resize so width/height are exact cell-box multiples.
+
+    kitty paints transparent margins too, so an untrimmed pet looks small and adrift.
+    """
    from PIL import Image

+    boxes = [b for b in (f.getchannel("A").getbbox() for f in frames) if b]
+    if boxes:
+        union = (min(b[0] for b in boxes), min(b[1] for b in boxes), max(b[2] for b in boxes), max(b[3] for b in boxes))
+        frames = [f.crop(union) for f in frames]
    w, h = frames[0].size
    target = (max(1, round(w / _CELL_W)) * _CELL_W, max(1, round(h / _CELL_H)) * _CELL_H)
    return frames if (w, h) == target else [f.resize(target, Image.LANCZOS) for f in frames]
@@ -162,10 +149,10 @@ def _kitty_apc(ctrl: str, data: str) -> str:
    return "".join(f"\x1b_G{ctrl + ',' if i == 0 else ''}m={int(i != last)};{piece}\x1b\\" for i, piece in enumerate(pieces))


-def _encode_kitty(frame, *, cell_cols: int | None = None, cell_rows: int | None = None) -> str:
+def _encode_kitty(frame) -> str:
    """kitty transmit+display at the cursor; ``c``/``r`` pin the cell box so frames overwrite each other."""
-    ctrl = "f=100,a=T,q=2" + (f",c={cell_cols}" if cell_cols else "") + (f",r={cell_rows}" if cell_rows else "")
-    return _kitty_apc(ctrl, _png_b64(frame))
+    cols, rows = _cell_box(frame)
+    return _kitty_apc(f"f=100,a=T,q=2,c={cols},r={rows}", _png_b64(frame))


 # kitty Unicode placeholders: Ink owns the screen and measures every cell, so it
@@ -218,19 +205,11 @@ def _encode_kitty_virtual(frame, *, image_id: int, cols: int, rows: int) -> str:
    return _kitty_apc(f"a=T,U=1,i={image_id},c={cols},r={rows},f=100,q=2", _png_b64(frame))


-def _encode_iterm(frame, *, cell_cols: int | None = None, cell_rows: int | None = None) -> str:
-    """iTerm2 inline image (OSC 1337 File)."""
+def _encode_iterm(frame) -> str:
+    """iTerm2 inline image (OSC 1337 File) pinned to the frame's cell box."""
    payload = _png_b64(frame)
-    args = ["inline=1", f"size={len(payload)}", "preserveAspectRatio=1"]
-    args += [f"width={cell_cols}"] if cell_cols else []
-    args += [f"height={cell_rows}"] if cell_rows else []
-    return f"\x1b]1337;File={';'.join(args)}:{payload}\x07"
-
-
-def _sixel_runs(chars: list[str]) -> str:
-    """Run-length encode one sixel band line (``!<n><ch>`` for runs longer than 3)."""
-    runs = ((ch, len(list(group))) for ch, group in groupby(chars))
-    return "".join("!%d%s" % (n, ch) if n > 3 else ch * n for ch, n in runs)
+    cols, rows = _cell_box(frame)
+    return f"\x1b]1337;File=inline=1;size={len(payload)};preserveAspectRatio=1;width={cols};height={rows}:{payload}\x07"


 def _encode_sixel(frame) -> str:
@@ -252,12 +231,12 @@ def _encode_sixel(frame) -> str:
        ys = range(band, min(band + 6, h))
        for color_idx in used:
            chars = [chr(63 + sum(1 << (y - band) for y in ys if alpha[x, y] > 32 and px[x, y] == color_idx)) for x in range(w)]
-            out.append("#%d" % color_idx + _sixel_runs(chars) + "$")  # carriage return within band
+            runs = ((ch, len(list(group))) for ch, group in groupby(chars))  # run-length: ``!<n><ch>`` for runs longer than 3
+            out.append("#%d" % color_idx + "".join("!%d%s" % (n, ch) if n > 3 else ch * n for ch, n in runs) + "$")  # ``$`` = band CR
        out.append("-")  # next band
    return "".join(out) + "\x1b\\"


-_HALF_BLOCK = "▀"
 # A single half-block cell: top pixel + bottom pixel as (r, g, b, a) tuples.
 Cell = tuple[tuple[int, int, int, int], tuple[int, int, int, int]]

@@ -277,10 +256,9 @@ def _downscale_cells(frame, *, target_cols: int) -> list[list[Cell]]:

 def _encode_unicode(frame, *, target_cols: int) -> str:
    """Truecolor ANSI half-blocks (one char = 2 vertical pixels)."""
-
    def cell(top, bottom) -> str:
        (tr, tg, tb, ta), (br, bg, bb, ba) = top, bottom
-        return "\x1b[0m " if ta < 32 and ba < 32 else f"\x1b[38;2;{tr};{tg};{tb}m\x1b[48;2;{br};{bg};{bb}m{_HALF_BLOCK}"
+        return "\x1b[0m " if ta < 32 and ba < 32 else f"\x1b[38;2;{tr};{tg};{tb}m\x1b[48;2;{br};{bg};{bb}m▀"

    return "\n".join("".join(cell(t, b) for t, b in row) + "\x1b[0m" for row in _downscale_cells(frame, target_cols=target_cols))

@@ -288,12 +266,14 @@ def _encode_unicode(frame, *, target_cols: int) -> str:
 def _cell_box(frame) -> tuple[int, int]:
    """Terminal cell box (~8×16 px per cell) for a scaled frame.

-    kitty stretches the image to fill ``c``×``r`` cells, so this must track the
-    scaled pixel size, not a native-aspect column count (that upscales small pets).
+    kitty stretches the image to fill ``c``×``r`` cells, so track the scaled pixel size, not a native-aspect column count (that upscales small pets).
    """
    return max(1, frame.width // _CELL_W), max(1, frame.height // _CELL_H)


+_ENCODERS = {"kitty": _encode_kitty, "iterm": _encode_iterm, "sixel": _encode_sixel}
+
+
@dataclass(eq=False)
 class PetRenderer:
    """Holds a pet's spritesheet and yields encoded frames per (state, index); decoded frames are cached."""
@@ -309,8 +289,7 @@ class PetRenderer:

    def __post_init__(self) -> None:
        self.spritesheet = str(self.spritesheet)
-        if self.mode not in RENDER_MODES:
-            self.mode = "unicode"
+        self.mode = self.mode if self.mode in RENDER_MODES else "unicode"

    @property
    def available(self) -> bool:
@@ -327,35 +306,25 @@ class PetRenderer:
    def cells(self, state: PetState | str, index: int, *, cols: int | None = None) -> list[list[Cell]]:
        """One frame as a half-block cell grid for Ink's native color props; ``[]`` when unavailable."""
        frames = self._frames(state)
-        if not frames:
-            return []
-        return _downscale_cells(frames[index % len(frames)], target_cols=cols or self.unicode_cols)
+        return _downscale_cells(frames[index % len(frames)], target_cols=cols or self.unicode_cols) if frames else []

    def kitty_payload(self, state: PetState | str, *, image_id: int) -> dict | None:
        """kitty placeholder payload ``{cols, rows, placeholder, frames}`` (transmit escapes + static text grid); ``None`` if no frames."""
-        frames = self._frames(state)
-        if not frames:
+        if not (frames := self._frames(state)):
            return None
-        frames = _snap_frames_to_cell_grid(_crop_frames_to_alpha_union(frames))
+        frames = _fit_frames_to_cell_grid(frames)
        cols, rows = _cell_box(frames[0])
        encoded = [_encode_kitty_virtual(f, image_id=image_id, cols=cols, rows=rows) for f in frames]
        return {"cols": cols, "rows": rows, "placeholder": kitty_placeholder_rows(cols, rows), "frames": encoded}

    def frame(self, state: PetState | str, index: int) -> str:
        """Encoded escape string for one frame (``index`` taken modulo the frame count), or ``""``."""
-        if self.mode == "off":
-            return ""
-        frames = self._frames(state)
-        if not frames:
+        if self.mode == "off" or not (frames := self._frames(state)):
            return ""
        frame = frames[index % len(frames)]
        try:
-            if self.mode in ("kitty", "iterm"):
-                cell_cols, cell_rows = _cell_box(frame)
-                encode = _encode_kitty if self.mode == "kitty" else _encode_iterm
-                return encode(frame, cell_cols=cell_cols, cell_rows=cell_rows)
-            if self.mode == "sixel":
-                return _encode_sixel(frame)
+            if self.mode in _ENCODERS:
+                return _ENCODERS[self.mode](frame)
            return _encode_unicode(frame, target_cols=self.unicode_cols)
        except Exception as exc:  # noqa: BLE001 - degrade silently
            logger.debug("pet frame encode failed (mode=%s): %s", self.mode, exc)
@@ -55,18 +55,19 @@ class InstalledPet:
        return self.created_by == "generator"


-def pets_dir() -> Path:
-    """Return the profile-scoped pets directory (created on demand)."""
-    path = get_hermes_home() / "pets"
+def _ensure_dir(path: Path) -> Path:
    path.mkdir(parents=True, exist_ok=True)
    return path


+def pets_dir() -> Path:
+    """Return the profile-scoped pets directory (created on demand)."""
+    return _ensure_dir(get_hermes_home() / "pets")
+
+
 def _thumb_path(slug: str) -> Path:
    """Cached thumbnail for *slug* (lives OUTSIDE the pet dir, under ``pets/.thumbs/``)."""
-    path = pets_dir() / ".thumbs"
-    path.mkdir(parents=True, exist_ok=True)
-    return path / f"{slug}.png"
+    return _ensure_dir(pets_dir() / ".thumbs") / f"{slug}.png"


 def _read_pet_json(directory: Path) -> dict:
@@ -97,25 +98,16 @@ def _safe_slug(slug: str) -> str:

 def load_pet(slug: str) -> InstalledPet | None:
    """Return the :class:`InstalledPet` for *slug*, or ``None`` if absent."""
-    slug = _safe_slug(slug)
-    directory = pets_dir() / slug
-    if not slug or not directory.is_dir():
+    if not (slug := _safe_slug(slug)) or not (directory := pets_dir() / slug).is_dir():
        return None
    meta = _read_pet_json(directory)
-    return InstalledPet(
-        slug=slug,
-        display_name=str(meta.get("displayName", "") or slug),
-        description=str(meta.get("description", "") or ""),
-        directory=directory,
-        spritesheet=_resolve_spritesheet(directory, meta),
-        created_by=str(meta.get("createdBy", "") or ""),
-    )
+    name, desc, by = (str(meta.get(k, "") or d) for k, d in (("displayName", slug), ("description", ""), ("createdBy", "")))
+    return InstalledPet(slug, name, desc, directory, _resolve_spritesheet(directory, meta), by)


 def _usable_pet(slug: str, error: str = "") -> InstalledPet | None:
    """:func:`load_pet` result only when its spritesheet is on disk; raises :class:`PetStoreError` *error* if given."""
-    pet = load_pet(slug)
-    if pet and pet.exists:
+    if (pet := load_pet(slug)) and pet.exists:
        return pet
    if error:
        raise PetStoreError(error)
@@ -124,16 +116,14 @@ def _usable_pet(slug: str, error: str = "") -> InstalledPet | None:

 def installed_pets() -> list[InstalledPet]:
    """Return every installed pet (dirs containing a usable spritesheet)."""
-    pets = (_usable_pet(child.name) for child in sorted(pets_dir().iterdir()) if child.is_dir())
-    return [pet for pet in pets if pet]
+    return [pet for child in sorted(pets_dir().iterdir()) if child.is_dir() and (pet := _usable_pet(child.name))]


 def resolve_active_pet(configured_slug: str | None = None) -> InstalledPet | None:
    """The configured slug (``display.pet.slug``) if installed, else the first pet alphabetically."""
    if configured_slug and (pet := _usable_pet(configured_slug.strip())):
        return pet
-    pets = installed_pets()
-    return pets[0] if pets else None
+    return next(iter(installed_pets()), None)


 def install_pet(slug: str, *, force: bool = False, timeout: float = _DOWNLOAD_TIMEOUT) -> InstalledPet:
@@ -153,8 +143,7 @@ def install_pet(slug: str, *, force: bool = False, timeout: float = _DOWNLOAD_TI
    if not _is_petdex_host(entry.spritesheet_url):
        raise PetStoreError(f"refusing non-petdex spritesheet host for '{slug}'")

-    directory = pets_dir() / slug
-    directory.mkdir(parents=True, exist_ok=True)
+    directory = _ensure_dir(pets_dir() / slug)
    sprite_ext = ".png" if entry.spritesheet_url.lower().split("?")[0].endswith(".png") else ".webp"
    sprite_path = directory / f"spritesheet{sprite_ext}"
    _download(entry.spritesheet_url, sprite_path, timeout=timeout)
@@ -163,8 +152,7 @@ def install_pet(slug: str, *, force: bool = False, timeout: float = _DOWNLOAD_TI
    meta: dict = {}
    if entry.pet_json_url and _is_petdex_host(entry.pet_json_url):
        try:
-            data = _http_get(entry.pet_json_url, timeout).json()
-            meta = data if isinstance(data, dict) else {}
+            meta = data if isinstance(data := _http_get(entry.pet_json_url, timeout).json(), dict) else {}
        except Exception as exc:  # noqa: BLE001 - non-fatal, fall back below
            logger.debug("pet.json fetch failed for %s: %s", slug, exc)
    meta = meta or {"id": slug, "displayName": entry.display_name, "description": ""}
@@ -177,8 +165,7 @@ def install_pet(slug: str, *, force: bool = False, timeout: float = _DOWNLOAD_TI

 def slugify(name: str) -> str:
    """Lowercase, hyphenate, and strip a display name into a filesystem slug."""
-    slug = re.sub(r"[^a-z0-9]+", "-", (name or "").strip().lower()).strip("-")
-    return slug or "pet"
+    return re.sub(r"[^a-z0-9]+", "-", (name or "").strip().lower()).strip("-") or "pet"


 def unique_slug(name: str) -> str:
@@ -188,28 +175,21 @@ def unique_slug(name: str) -> str:
    return next(slug for slug in candidates if not (pets_dir() / slug).exists())


-def _write_spritesheet(source, dest: Path) -> None:
-    """Write *source* (PIL image, bytes, or path) as a lossless WebP at *dest*."""
-    if isinstance(source, (bytes, bytearray)):
-        dest.write_bytes(bytes(source))
-        return
-    from agent.pet.generate.atlas import _load_rgba
-
-    _load_rgba(source).save(dest, format="WEBP", lossless=True, quality=100, method=6, exact=True)
-
-
 def register_local_pet(spritesheet, *, slug: str, display_name: str = "", description: str = "") -> InstalledPet:
-    """Write a locally-generated pet (PIL image, WebP/PNG bytes, or path) into the store.
+    """Write a locally-generated pet (PIL image, WebP/PNG bytes, or path) into the store as lossless WebP.

-    Appears in :func:`installed_pets` immediately; :func:`install_pet` returns on-disk
-    pets before consulting the manifest, so no manifest entry is needed.
+    Appears in :func:`installed_pets` immediately (no manifest entry needed).
    """
    slug = slugify(slug)
-    directory = pets_dir() / slug
-    directory.mkdir(parents=True, exist_ok=True)
+    directory = _ensure_dir(pets_dir() / slug)
    sprite_path = directory / "spritesheet.webp"
    try:
-        _write_spritesheet(spritesheet, sprite_path)
+        if isinstance(spritesheet, (bytes, bytearray)):
+            sprite_path.write_bytes(bytes(spritesheet))
+        else:
+            from agent.pet.generate.atlas import _load_rgba
+
+            _load_rgba(spritesheet).save(sprite_path, format="WEBP", lossless=True, quality=100, method=6, exact=True)
    except Exception as exc:  # noqa: BLE001 - normalize to one error type
        raise PetStoreError(f"could not write spritesheet for '{slug}': {exc}") from exc
    meta = {"id": slug, "displayName": display_name or slug, "description": description or "", "spritesheetPath": sprite_path.name}
@@ -241,20 +221,6 @@ def _is_petdex_host(url: str) -> bool:
    return host == "petdex.dev" or host.endswith(".petdex.dev")


-def _thumb_source_bytes(slug: str, source_url: str, timeout: float) -> bytes | None:
-    """Spritesheet bytes for a thumbnail: the installed sheet, else a petdex-hosted *source_url*."""
-    pet = _usable_pet(slug)
-    if pet:
-        with contextlib.suppress(OSError):
-            return pet.spritesheet.read_bytes()
-    if source_url and _is_petdex_host(source_url):
-        try:
-            return _http_get(source_url, timeout).content
-        except Exception as exc:  # noqa: BLE001 - cosmetic, degrade to placeholder
-            logger.debug("thumb fetch failed for %s: %s", slug, exc)
-    return None
-
-
 def thumbnail_png(slug: str, *, source_url: str = "", timeout: float = 30.0) -> bytes | None:
    """Small idle-frame (top-left cell) PNG for *slug*, cached on disk; ``None`` on any failure.

@@ -262,14 +228,20 @@ def thumbnail_png(slug: str, *, source_url: str = "", timeout: float = 30.0) ->
    never fetches arbitrary client URLs). Server-side so it rides the authenticated
    gateway as a same-origin data URL, sidestepping CSP/hotlink limits.
    """
-    slug = slug.strip()
-    if not slug:
+    if not (slug := slug.strip()):
        return None
-    cache = _thumb_path(slug)
-    if cache.is_file():
+    if (cache := _thumb_path(slug)).is_file():
        with contextlib.suppress(OSError):
            return cache.read_bytes()
-    sheet_bytes = _thumb_source_bytes(slug, source_url, timeout)
+    sheet_bytes = None
+    if pet := _usable_pet(slug):
+        with contextlib.suppress(OSError):
+            sheet_bytes = pet.spritesheet.read_bytes()
+    if sheet_bytes is None and source_url and _is_petdex_host(source_url):
+        try:
+            sheet_bytes = _http_get(source_url, timeout).content
+        except Exception as exc:  # noqa: BLE001 - cosmetic, degrade to placeholder
+            logger.debug("thumb fetch failed for %s: %s", slug, exc)
    if not sheet_bytes:
        return None
    try:
@@ -296,19 +268,16 @@ def remove_pet(slug: str) -> bool:
    # Drop the cached thumb too or a later pet reusing this slug shows the stale one.
    with contextlib.suppress(OSError):
        _thumb_path(slug).unlink(missing_ok=True)
-    directory = pets_dir() / slug
-    if not directory.is_dir():
+    if not (directory := pets_dir() / slug).is_dir():
        return False
    shutil.rmtree(directory, ignore_errors=True)
    return not directory.exists()


 def rename_pet(slug: str, display_name: str) -> str | None:
-    """Rename a pet's ``displayName`` AND realign its slug/dir to match.
+    """Rename a pet's ``displayName`` AND move its dir/thumb to ``slugify(name)`` when that's a free, different slug.

-    Generated pets hatch under a provisional slug; naming makes that the identity.
-    The dir (and thumb) moves to ``slugify(name)`` when that's a free, different
-    slug. Returns the resulting slug, or ``None`` on failure.
+    Generated pets hatch under a provisional slug; naming makes that the identity. Returns the resulting slug, or ``None``.
    """
    slug = _safe_slug(slug)
    display_name = (display_name or "").strip()
@@ -321,16 +290,12 @@ def rename_pet(slug: str, display_name: str) -> str | None:
    new_slug = slug
    desired = slugify(display_name)
    if desired and desired != slug and not (pets_dir() / desired).exists():
-        try:
+        with contextlib.suppress(OSError):  # keep the provisional slug if the move fails
            directory.rename(pets_dir() / desired)
-        except OSError:
-            pass  # keep the provisional slug if the move fails
-        else:
            with contextlib.suppress(OSError):
                _thumb_path(slug).rename(_thumb_path(desired))
            directory = pets_dir() / desired
            new_slug = meta["id"] = desired
-
    try:
        _write_pet_json(directory, meta)
    except OSError:
@@ -54,9 +54,8 @@ _DEFAULT_TUNNEL_PORT = 9090

 # Hosts allowed by default for AI inference traffic.  Anything else is 403'd.
 _DEFAULT_ALLOWED_HOSTS: Tuple[str, ...] = (
-    "openrouter.ai", "*.openrouter.ai", "api.openai.com", "api.anthropic.com",
-    "generativelanguage.googleapis.com", "api.x.ai", "api.mistral.ai", "api.groq.com",
-    "api.together.xyz", "api.deepseek.com", "inference.nousresearch.com",
+    "openrouter.ai", "*.openrouter.ai", "api.openai.com", "api.anthropic.com", "generativelanguage.googleapis.com",
+    "api.x.ai", "api.mistral.ai", "api.groq.com", "api.together.xyz", "api.deepseek.com", "inference.nousresearch.com",
 )

 # Provider env-var name -> upstream hosts on which the Authorization Bearer token is swapped.
@@ -74,10 +73,8 @@ _BEARER_PROVIDERS: Dict[str, Tuple[str, ...]] = {
 # Gemini's ``?key=<token>`` style is covered by match_query.
 _HEADER_AUTH_PROVIDERS: Dict[str, Dict[str, Tuple[str, ...]]] = {
    "ANTHROPIC_API_KEY": {"hosts": ("api.anthropic.com",), "match_headers": ("x-api-key", "Authorization"), "aliases": ()},
-    "AZURE_OPENAI_API_KEY": {
-        "hosts": ("*.openai.azure.com", "*.cognitiveservices.azure.com", "*.services.ai.azure.com"),
-        "match_headers": ("api-key", "Authorization"), "aliases": (),
-    },
+    "AZURE_OPENAI_API_KEY": {"hosts": ("*.openai.azure.com", "*.cognitiveservices.azure.com", "*.services.ai.azure.com"),
+                             "match_headers": ("api-key", "Authorization"), "aliases": ()},
    "GEMINI_API_KEY": {"hosts": ("generativelanguage.googleapis.com",), "match_headers": ("x-goog-api-key",), "aliases": ("GOOGLE_API_KEY",)},
 }

@@ -95,16 +92,14 @@ _DEFAULT_UPSTREAM_DENY_CIDRS: Tuple[str, ...] = (
    "198.18.0.0/15",      # RFC2544 benchmark range
 )

-# Minimal daemon env; everything else is stripped so /proc/<pid>/environ never exposes operator secrets.
+# Minimal daemon env (SYSTEMROOT/USERPROFILE are Windows); everything else is stripped so
+# /proc/<pid>/environ never exposes operator secrets.
 _PROXY_SUBPROCESS_ENV_ALLOWLIST: Tuple[str, ...] = (
-    "PATH", "HOME", "TMPDIR", "TZ", "LANG", "LC_ALL", "LC_CTYPE", "NO_COLOR", "SSL_CERT_DIR", "SSL_CERT_FILE",
-    "SYSTEMROOT", "USERPROFILE",  # Windows
+    "PATH", "HOME", "TMPDIR", "TZ", "LANG", "LC_ALL", "LC_CTYPE", "NO_COLOR", "SSL_CERT_DIR", "SSL_CERT_FILE", "SYSTEMROOT", "USERPROFILE",
 )

 # Always stripped — these would recurse the proxy through itself or a corporate proxy.
-_PROXY_SUBPROCESS_ENV_STRIP: Tuple[str, ...] = (
-    "HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy", "ALL_PROXY", "all_proxy", "NO_PROXY", "no_proxy",
-)
+_PROXY_SUBPROCESS_ENV_STRIP: Tuple[str, ...] = ("HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy", "ALL_PROXY", "all_proxy", "NO_PROXY", "no_proxy")

 # SIGKILL doesn't exist on Windows; SIGTERM there is TerminateProcess() — same semantics.
 _KILL_SIGNAL = getattr(signal, "SIGKILL", signal.SIGTERM)
@@ -122,7 +117,6 @@ _proxy_nonce: Optional[str] = None

@dataclass
 class ProxyStatus:
-    """Snapshot of the iron-proxy installation + runtime state."""
    enabled: bool = False
    binary_path: Optional[Path] = None
    binary_version: Optional[str] = None
@@ -144,8 +138,8 @@ class ProxyStatus:

@dataclass
 class TokenMapping:
-    """Sandbox-visible proxy token -> upstream credential lookup.  ``real_env_name`` is read from
-    iron-proxy's OWN env at egress; ``alias_env_names`` are extra SANDBOX names for the same token."""
+    """Sandbox-visible proxy token -> upstream credential lookup.  ``real_env_name`` is read from iron-proxy's
+    OWN env at egress; ``alias_env_names`` are extra SANDBOX names for the same token."""
    proxy_token: str
    real_env_name: str
    upstream_hosts: Tuple[str, ...]
@@ -158,17 +152,14 @@ def _hermes_bin_dir() -> Path:
    return get_hermes_home() / "bin"


-def _proxy_state_dir_ro() -> Path:
-    """Proxy state dir without creating it (status probes, pidfile reads)."""
+def _proxy_state_dir_ro() -> Path:  # without creating it (status probes, pidfile reads)
    from hermes_constants import get_hermes_home
    return get_hermes_home() / "proxy"


 def _proxy_state_dir() -> Path:
-    """Proxy state dir (CA key, pidfile, logs), created 0o700; chmod is unconditional so a
-    pre-existing slack-umask dir gets tightened."""
-    d = _proxy_state_dir_ro()
-    d.mkdir(parents=True, exist_ok=True)
+    """Proxy state dir (CA key, pidfile, logs), created 0o700; unconditional chmod tightens a pre-existing slack-umask dir."""
+    (d := _proxy_state_dir_ro()).mkdir(parents=True, exist_ok=True)
    with suppress(OSError):  # Windows no-op / shared fs we don't own; files still get explicit perms
        d.chmod(0o700)
    return d
@@ -180,16 +171,12 @@ def _platform_binary_name() -> str:

 def _platform_asset_name() -> str:
    """Map (uname, arch) -> ``iron-proxy_<version>_<os>_<arch>.tar.gz``; no Windows builds upstream."""
-    system = platform.system()
-    machine = platform.machine().lower()
-    os_name = {"Linux": "linux", "Darwin": "darwin"}.get(system)
-    if os_name:
+    system, machine = platform.system(), platform.machine().lower()
+    if os_name := {"Linux": "linux", "Darwin": "darwin"}.get(system):
        arch = "arm64" if machine in ("arm64", "aarch64") else "amd64"
        return f"iron-proxy_{_IRON_PROXY_VERSION}_{os_name}_{arch}.tar.gz"
    if system == "Windows":
-        raise RuntimeError(
-            f"iron-proxy does not ship native Windows binaries as of v{_IRON_PROXY_VERSION}. Run the proxy on a Linux/macOS host, or inside WSL."
-        )
+        raise RuntimeError(f"iron-proxy does not ship native Windows binaries as of v{_IRON_PROXY_VERSION}. Run the proxy on a Linux/macOS host, or inside WSL.")
    raise RuntimeError(f"Unsupported platform for iron-proxy auto-install: {system} {machine}")


@@ -198,8 +185,7 @@ def find_iron_proxy(*, install_if_missing: bool = False) -> Optional[Path]:
    managed = _hermes_bin_dir() / _platform_binary_name()
    if managed.exists() and os.access(managed, os.X_OK):
        return managed
-    system = shutil.which("iron-proxy")
-    if system:
+    if system := shutil.which("iron-proxy"):
        return Path(system)
    if not install_if_missing:
        return None
@@ -212,23 +198,19 @@ def find_iron_proxy(*, install_if_missing: bool = False) -> Optional[Path]:

 def install_iron_proxy(*, force: bool = False) -> Path:
    """Download, verify, and install the pinned binary; raises on any failure."""
-    bin_dir = _hermes_bin_dir()
-    bin_dir.mkdir(parents=True, exist_ok=True)
+    (bin_dir := _hermes_bin_dir()).mkdir(parents=True, exist_ok=True)
    target = bin_dir / _platform_binary_name()
    if target.exists() and not force:
        return target
    asset_name = _platform_asset_name()
    with tempfile.TemporaryDirectory(prefix="hermes-iron-proxy-") as tmpdir:
-        tmp = Path(tmpdir)
-        archive_path = tmp / asset_name
-        checksum_path = tmp / _IRON_PROXY_CHECKSUM_NAME
+        archive_path, checksum_path = (tmp := Path(tmpdir)) / asset_name, tmp / _IRON_PROXY_CHECKSUM_NAME
        logger.info("Downloading %s", f"{_IRON_PROXY_RELEASE_BASE}/{asset_name}")
        _release_asset(asset_name, archive_path)
        _release_asset(_IRON_PROXY_CHECKSUM_NAME, checksum_path)
        # Best-effort GPG check of checksums.txt closes the release-channel tamper gap.
        _verify_checksums_signature(tmp, checksum_path)
-        expected = _expected_sha256(checksum_path, asset_name)
-        actual = _sha256_file(archive_path)
+        expected, actual = _expected_sha256(checksum_path, asset_name), _sha256_file(archive_path)
        if expected.lower() != actual.lower():
            raise RuntimeError(f"Checksum mismatch for {asset_name}: expected {expected}, got {actual}")
        with tarfile.open(archive_path, "r:gz") as tf:
@@ -254,8 +236,8 @@ def install_iron_proxy(*, force: bool = False) -> Path:
 def _release_asset(name: str, dest: Path) -> None:
    """Download one pinned-release asset to ``dest``; RuntimeError on any URL error."""
    url = f"{_IRON_PROXY_RELEASE_BASE}/{name}"
-    req = urllib.request.Request(url, headers={"User-Agent": "hermes-agent"})
    try:
+        req = urllib.request.Request(url, headers={"User-Agent": "hermes-agent"})
        with urllib.request.urlopen(req, timeout=_DOWNLOAD_TIMEOUT) as resp, open(dest, "wb") as f:  # noqa: S310
            shutil.copyfileobj(resp, f)
    except urllib.error.URLError as exc:
@@ -263,11 +245,9 @@ def _release_asset(name: str, dest: Path) -> None:


 def _verify_checksums_signature(tmp: Path, checksum_path: Path) -> bool:
-    """Best-effort GPG check of checksums.txt in an ephemeral keyring.  False (with a warning) when
-    gpg or the signature assets are unavailable — SHA-256 stays enforced, gpg is never a hard
-    dependency.  Raises ONLY on a present-but-bad signature (tamper signal)."""
-    gpg = shutil.which("gpg")
-    if not gpg:
+    """Best-effort GPG check of checksums.txt in an ephemeral keyring.  False (with a warning) when gpg or the signature
+    assets are unavailable — SHA-256 stays enforced, gpg is never a hard dependency.  Raises ONLY on a present-but-bad signature."""
+    if not (gpg := shutil.which("gpg")):
        logger.warning("gpg not found on PATH — skipping iron-proxy release-signature verification (SHA-256 checksum check still enforced).")
        return False
    sig_path, pubkey_path = tmp / _IRON_PROXY_CHECKSUM_SIG_NAME, tmp / _IRON_PROXY_PUBKEY_NAME
@@ -275,25 +255,16 @@ def _verify_checksums_signature(tmp: Path, checksum_path: Path) -> bool:
        _release_asset(_IRON_PROXY_CHECKSUM_SIG_NAME, sig_path)
        _release_asset(_IRON_PROXY_PUBKEY_NAME, pubkey_path)
    except RuntimeError as exc:
-        logger.warning(
-            "iron-proxy release signature assets unavailable (%s) — skipping GPG verification (SHA-256 checksum check still enforced).", exc,
-        )
+        logger.warning("iron-proxy release signature assets unavailable (%s) — skipping GPG verification (SHA-256 checksum check still enforced).", exc)
        return False
-    gnupg_home = tmp / "gnupg"
-    gnupg_home.mkdir(mode=0o700, exist_ok=True)
+    (gnupg_home := tmp / "gnupg").mkdir(mode=0o700, exist_ok=True)
    gpg_base = [gpg, "--homedir", str(gnupg_home), "--batch", "--no-tty"]
-    imp = _run([*gpg_base, "--import", str(pubkey_path)], timeout=60)
-    if imp.returncode != 0:
-        logger.warning(
-            "Could not import iron-proxy signing key — skipping GPG verification (SHA-256 still enforced): %s",
-            imp.stderr.decode("utf-8", "replace")[:200],
-        )
+    if (imp := _run([*gpg_base, "--import", str(pubkey_path)], timeout=60)).returncode != 0:
+        logger.warning("Could not import iron-proxy signing key — skipping GPG verification (SHA-256 still enforced): %s", imp.stderr.decode("utf-8", "replace")[:200])
        return False
-    verify = _run([*gpg_base, "--verify", str(sig_path), str(checksum_path)], timeout=60)
-    if verify.returncode != 0:
+    if (verify := _run([*gpg_base, "--verify", str(sig_path), str(checksum_path)], timeout=60)).returncode != 0:
        raise RuntimeError(
-            "iron-proxy checksums.txt failed GPG signature verification — refusing to install (possible release-channel "
-            f"tampering). gpg: {verify.stderr.decode('utf-8', 'replace')[:300]}"
+            f"iron-proxy checksums.txt failed GPG signature verification — refusing to install (possible release-channel tampering). gpg: {verify.stderr.decode('utf-8', 'replace')[:300]}"
        )
    logger.info("Verified iron-proxy checksums.txt GPG signature.")
    return True
@@ -309,18 +280,14 @@ def _expected_sha256(checksum_file: Path, asset_name: str) -> str:


 def _sha256_file(path: Path) -> str:
-    h = hashlib.sha256()
    with open(path, "rb") as f:
-        for chunk in iter(lambda: f.read(65536), b""):
-            h.update(chunk)
-    return h.hexdigest()
+        return hashlib.file_digest(f, "sha256").hexdigest()


 def _pick_tar_member(tf: tarfile.TarFile, binary_name: str) -> tarfile.TarInfo:
    """Find the binary in the archive (flat or one dir deep); reject abs paths and ``..``."""
    candidates = [
-        m for m in tf.getmembers()
-        if m.isfile() and not m.name.startswith("/") and ".." not in Path(m.name).parts and Path(m.name).name == binary_name
+        m for m in tf.getmembers() if m.isfile() and not m.name.startswith("/") and ".." not in Path(m.name).parts and Path(m.name).name == binary_name
    ]
    if not candidates:
        raise RuntimeError(f"Could not find {binary_name} inside downloaded archive (members: {[m.name for m in tf.getmembers()[:5]]}...)")
@@ -333,7 +300,6 @@ def _allowlisted_env() -> Dict[str, str]:


 def _run(argv: List[str], *, timeout: int, text: bool = False, **kwargs) -> "subprocess.CompletedProcess":
-    """``subprocess.run`` with captured output; argv[0] is always a trusted PATH/system binary."""
    if text:
        kwargs.update(text=True, encoding="utf-8", errors="replace")
    return subprocess.run(argv, capture_output=True, timeout=timeout, stdin=subprocess.DEVNULL, **kwargs)  # noqa: S603
@@ -341,23 +307,21 @@ def _run(argv: List[str], *, timeout: int, text: bool = False, **kwargs) -> "sub

 def iron_proxy_version(binary: Path) -> str:
    """``iron-proxy --version`` output, stripped and cached by path.  Empty on failure."""
-    key = str(binary)
-    if key in _VERSION_CACHE:
+    if (key := str(binary)) in _VERSION_CACHE:
        return _VERSION_CACHE[key]
    try:
        # Scrubbed env: a PATH-resolved binary must not see the host's API keys.
        res = _run([str(binary), "--version"], timeout=_RUN_TIMEOUT, text=True, env=_allowlisted_env())
    except (OSError, subprocess.TimeoutExpired):
        return ""
-    out = (res.stdout or res.stderr or "").strip()
-    if out:  # never cache empty output — it would poison status for the process lifetime
+    if out := (res.stdout or res.stderr or "").strip():  # never cache empty output — it would poison status for the process lifetime
        _VERSION_CACHE[key] = out
    return out


 def _write_private_file(path: Path, data: bytes) -> None:
-    """Create/truncate ``path`` 0o600 from the first byte (no chmod-after TOCTOU), O_NOFOLLOW
-    against a planted symlink, fchmod to tighten a pre-existing file."""
+    """Create/truncate ``path`` 0o600 from the first byte (no chmod-after TOCTOU), O_NOFOLLOW against a planted
+    symlink, fchmod to tighten a pre-existing file."""
    fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC | _O_NOFOLLOW, 0o600)
    try:
        with suppress(OSError, AttributeError):
@@ -387,11 +351,8 @@ def ensure_ca_cert(*, force: bool = False) -> Tuple[Path, Path]:
    with tempfile.TemporaryDirectory(prefix="hermes-proxy-ca-") as tmpdir:
        tmp_key, tmp_crt = Path(tmpdir) / "ca.key", Path(tmpdir) / "ca.crt"
        _run(["openssl", "genrsa", "-out", str(tmp_key), "4096"], timeout=60, check=True)
-        _run([
-            "openssl", "req", "-x509", "-new", "-nodes", "-key", str(tmp_key), "-sha256", "-days", "3650",
-            "-subj", "/CN=hermes iron-proxy CA", "-addext", "basicConstraints=critical,CA:TRUE",
-            "-addext", "keyUsage=critical,keyCertSign", "-out", str(tmp_crt),
-        ], timeout=60, check=True)
+        _run(["openssl", "req", "-x509", "-new", "-nodes", "-key", str(tmp_key), "-sha256", "-days", "3650", "-subj", "/CN=hermes iron-proxy CA",
+              "-addext", "basicConstraints=critical,CA:TRUE", "-addext", "keyUsage=critical,keyCertSign", "-out", str(tmp_crt)], timeout=60, check=True)
        # Key: stage 0o600 against a fresh inode, then atomically rename into place.
        key_staged = ca_key.with_suffix(ca_key.suffix + ".staged")
        key_staged.unlink(missing_ok=True)
@@ -420,8 +381,7 @@ def _read_text_or_none(p: Path) -> Optional[str]:
 def ensure_management_token(*, force: bool = False) -> str:
    """Return the management-API bearer key (0600 at <proxy>/management.token), minting on first call."""
    p = _proxy_state_dir() / "management.token"
-    existing = None if force else _read_text_or_none(p)
-    if existing:
+    if not force and (existing := _read_text_or_none(p)):
        return existing
    token = mint_proxy_token(prefix="hermes-mgmt")
    _write_private_file(p, token.encode("utf-8"))
@@ -437,17 +397,6 @@ def _yaml():
        return None


-def _load_proxy_yaml(cfg: Path):
-    """Parsed proxy.yaml, or ``{}`` when missing/unreadable/PyYAML absent."""
-    yaml = _yaml()
-    if yaml is None:
-        return {}
-    try:
-        return yaml.safe_load(cfg.read_text(encoding="utf-8")) or {}
-    except (OSError, yaml.YAMLError):
-        return {}
-
-
 def _parse_listen(listen) -> Optional[Tuple[str, int]]:
    """``"host:port"`` -> ``(host, port)``; empty host means loopback."""
    if not isinstance(listen, str) or ":" not in listen:
@@ -461,8 +410,12 @@ def _parse_listen(listen) -> Optional[Tuple[str, int]]:


 def _config_listen(section: str, *keys: str, config_path: Optional[Path] = None) -> Optional[Tuple[str, int]]:
-    """``(host, port)`` from the first truthy ``proxy.yaml[section][key]``, or None."""
-    block = _load_proxy_yaml(config_path or (_proxy_state_dir_ro() / "proxy.yaml")).get(section) or {}
+    """``(host, port)`` from the first truthy ``proxy.yaml[section][key]``, or None (also when file/PyYAML is missing)."""
+    yaml, data = _yaml(), {}
+    if yaml is not None:
+        with suppress(OSError, yaml.YAMLError):
+            data = yaml.safe_load((config_path or (_proxy_state_dir_ro() / "proxy.yaml")).read_text(encoding="utf-8")) or {}
+    block = data.get(section) or {}
    return _parse_listen(next((block[k] for k in keys if block.get(k)), ""))


@@ -470,41 +423,28 @@ def _read_management_listen_from_config(config_path: Optional[Path] = None) -> O
    return _config_listen("management", "listen", config_path=config_path)


-def _read_http_listen_from_config() -> Optional[Tuple[str, int]]:
-    """Sandbox-facing listener: ``tunnel_listen`` (CONNECT/MITM), falling back to ``http_listen``
-    for configs written before the listener-role split."""
-    return _config_listen("proxy", "tunnel_listen", "http_listen")
-
-
 def _probe_target() -> Tuple[str, int]:
-    """Configured bind host/port to probe — on Linux that's the docker bridge, where a loopback
-    connect would report a healthy daemon as down."""
-    return _read_http_listen_from_config() or ("127.0.0.1", _DEFAULT_TUNNEL_PORT)
+    """Configured bind host/port to probe — on Linux the docker bridge, where a loopback connect would report a healthy
+    daemon as down.  ``tunnel_listen`` (CONNECT/MITM) falls back to ``http_listen`` for pre-listener-role-split configs."""
+    return _config_listen("proxy", "tunnel_listen", "http_listen") or ("127.0.0.1", _DEFAULT_TUNNEL_PORT)


-# Management-API error status -> operator message (422 = validation rejected, running ruleset
-# unchanged; 401 = daemon started with a different management.token).
+# Management-API error status -> operator message (422 = validation rejected, ruleset unchanged; 401 = daemon started with another management.token).
 _RELOAD_HTTP_ERRORS = {
    422: "iron-proxy rejected the new config (validation failed; the running ruleset is unchanged): {body}",
-    401: "management API rejected our key (401).  The running daemon was started with a different management.token — "
-         "run `hermes egress restart`.",
+    401: "management API rejected our key (401).  The running daemon was started with a different management.token — run `hermes egress restart`.",
 }


 def reload_proxy() -> bool:
-    """``POST /v1/reload`` (validation failures leave the running config untouched); RuntimeError
-    with an actionable message on any failure."""
-    pid = _read_pid()
-    if not pid or not _pid_alive(pid):
+    """``POST /v1/reload`` (validation failures leave the running config untouched); actionable RuntimeError on any failure."""
+    if not (pid := _read_pid()) or not _pid_alive(pid):
        raise RuntimeError("iron-proxy is not running — nothing to reload.  Run `hermes egress start`.")
-    mgmt = _read_management_listen_from_config()
-    if mgmt is None:
+    if (mgmt := _read_management_listen_from_config()) is None:
        raise RuntimeError(
-            "The generated proxy.yaml has no management listener (written before reload support).  "
-            "Re-run `hermes egress setup` and use `hermes egress restart` this one time."
+            "The generated proxy.yaml has no management listener (written before reload support).  Re-run `hermes egress setup` and use `hermes egress restart` this one time."
        )
-    token = _read_text_or_none(_proxy_state_dir_ro() / "management.token")
-    if not token:
+    if not (token := _read_text_or_none(_proxy_state_dir_ro() / "management.token")):
        raise RuntimeError("management.token is missing — re-run `hermes egress setup`, then `hermes egress restart`.")
    host, port = mgmt
    req = urllib.request.Request(f"http://{host}:{port}/v1/reload", method="POST", headers={"Authorization": f"Bearer {token}"}, data=b"")
@@ -522,22 +462,19 @@ def reload_proxy() -> bool:
    except (urllib.error.URLError, OSError) as exc:
        # A daemon started from a pre-management config is alive but has no listener.
        raise RuntimeError(
-            f"could not reach the management API at {host}:{port} ({exc}).  "
-            "If the daemon was started before reload support, run `hermes egress restart` once."
+            f"could not reach the management API at {host}:{port} ({exc}).  If the daemon was started before reload support, run `hermes egress restart` once."
        ) from exc


 def _default_http_listen(tunnel_port: int) -> List[str]:
-    """Single bind (v0.39 allows one): docker bridge on Linux (what ``host.docker.internal``
-    resolves to; loopback is unreachable from containers), loopback on Docker Desktop (VPNkit).
-    NEVER 0.0.0.0: a LAN peer with a leaked sandbox token could spend the operator's API quota."""
+    """Single bind (v0.39 allows one): docker bridge on Linux (what ``host.docker.internal`` resolves to; loopback is
+    unreachable from containers), loopback on Docker Desktop (VPNkit).  NEVER 0.0.0.0: a LAN peer with a leaked
+    sandbox token could spend the operator's API quota."""
    if platform.system() == "Linux":
-        bridge_ip = _detect_docker_bridge_ip()
-        if bridge_ip and bridge_ip != "127.0.0.1":
+        if (bridge_ip := _detect_docker_bridge_ip()) and bridge_ip != "127.0.0.1":
            return [f"{bridge_ip}:{tunnel_port}"]
        logger.warning(
-            "No docker bridge (docker0) detected — binding iron-proxy to loopback only.  "
-            "Docker sandboxes will NOT be able to reach the proxy until it is restarted with docker running."
+            "No docker bridge (docker0) detected — binding iron-proxy to loopback only.  Docker sandboxes will NOT be able to reach the proxy until it is restarted with docker running."
        )
    return [f"127.0.0.1:{tunnel_port}"]

@@ -549,18 +486,11 @@ def _detect_docker_bridge_ip() -> Optional[str]:
        res = _run(["ip", "-4", "-o", "addr", "show", "docker0"], timeout=2, text=True)
    except (OSError, subprocess.TimeoutExpired):
        return None
-    if res.returncode != 0:
-        return None
    # Expected: "<n>: docker0  inet 172.17.0.1/16 ..." — first inet token (per line) wins.
-    candidate = None
-    for parts in (line.split() for line in res.stdout.splitlines()):
-        if "inet" in parts[:-1]:
-            candidate = parts[parts.index("inet") + 1].split("/")[0]
-            break
-    if not candidate:
-        return None
+    lines = (line.split() for line in res.stdout.splitlines()) if res.returncode == 0 else ()
+    candidate = next((parts[parts.index("inet") + 1].split("/")[0] for parts in lines if "inet" in parts[:-1]), None)
    try:
-        addr = ipaddress.IPv4Address(candidate)
+        addr = ipaddress.IPv4Address(candidate)  # None/"" raise too
    except (ipaddress.AddressValueError, ValueError):
        return None
    if addr.is_unspecified or addr.is_loopback or addr.is_multicast or addr.is_reserved or addr.is_link_local or addr.is_global:
@@ -570,38 +500,32 @@ def _detect_docker_bridge_ip() -> Optional[str]:


 def build_proxy_config(
-    *, mappings: List[TokenMapping], ca_cert: Path, ca_key: Path, tunnel_port: int = _DEFAULT_TUNNEL_PORT,
-    audit_log: Optional[Path] = None, allowed_hosts: Optional[List[str]] = None,
-    upstream_deny_cidrs: Optional[List[str]] = None, http_listen: Optional[List[str]] = None,
+    *, mappings: List[TokenMapping], ca_cert: Path, ca_key: Path, tunnel_port: int = _DEFAULT_TUNNEL_PORT, audit_log: Optional[Path] = None,
+    allowed_hosts: Optional[List[str]] = None, upstream_deny_cidrs: Optional[List[str]] = None, http_listen: Optional[List[str]] = None,
 ) -> Dict:
-    """iron-proxy YAML config dict (v0.39.0 schema).  Real secrets come from iron-proxy's OWN env
-    (``source: {type: env}``); the sandbox never sees them.  ``upstream_deny_cidrs=None`` = default
-    SSRF deny list, ``[]`` opts out.  ``audit_log`` is forward-compat only (v0.39 rejects ``audit_path``)."""
+    """iron-proxy YAML config dict (v0.39.0 schema).  Real secrets come from iron-proxy's OWN env (``source: {type: env}``);
+    the sandbox never sees them.  ``upstream_deny_cidrs=None`` = default SSRF deny list, ``[]`` opts out.
+    ``audit_log`` is forward-compat only (v0.39 rejects ``audit_path``)."""
    hosts: List[str] = list(allowed_hosts or _DEFAULT_ALLOWED_HOSTS)
-    for m in mappings:
-        for h in m.upstream_hosts:
-            if h not in hosts:
-                hosts.append(h)
+    for h in (h for m in mappings for h in m.upstream_hosts):
+        if h not in hosts:
+            hosts.append(h)
    deny_cidrs = list(_DEFAULT_UPSTREAM_DENY_CIDRS if upstream_deny_cidrs is None else upstream_deny_cidrs)
    # Query scan covers ``?key=<token>`` SDKs; body inspection deliberately off.  ``require`` fails
    # closed: an allowlisted-host request WITHOUT the proxy token is rejected, so a real key sent
    # directly can't cross the boundary.
-    secrets_rules = [
-        {
-            "source": {"type": "env", "var": m.real_env_name},
-            "replace": {
-                "proxy_value": m.proxy_token, "match_headers": list(m.match_headers or ("Authorization",)),
-                "match_query": True, "match_body": False, "require": True,
-            },
-            "rules": [{"host": h} for h in m.upstream_hosts],
-        }
-        for m in mappings
-    ]
+    secrets_rules = [{
+        "source": {"type": "env", "var": m.real_env_name},
+        "replace": {
+            "proxy_value": m.proxy_token, "match_headers": list(m.match_headers or ("Authorization",)),
+            "match_query": True, "match_body": False, "require": True,
+        },
+        "rules": [{"host": h} for h in m.upstream_hosts],
+    } for m in mappings]
    # ONE string per listener field.  tunnel_listen is the CONNECT+MITM listener sandboxes reach via
    # HTTPS_PROXY (a CONNECT to http_listen is forwarded upstream and 400s); http_listen is plain-HTTP
    # forward on tunnel_port+1.
-    listens = list(http_listen) if http_listen else _default_http_listen(tunnel_port)
-    primary_listen = listens[0] if listens else f"127.0.0.1:{tunnel_port}"
+    primary_listen = (list(http_listen) if http_listen else _default_http_listen(tunnel_port) or [f"127.0.0.1:{tunnel_port}"])[0]
    bind_host = primary_listen.rsplit(":", 1)[0] or "127.0.0.1"
    return {
        # Required by the parser; tunnel-only mode never binds an exposed DNS port.
@@ -618,17 +542,14 @@ def build_proxy_config(
        # Loopback only: sandboxes must never reach the management surface.
        "management": {"listen": f"127.0.0.1:{tunnel_port + _MGMT_PORT_OFFSET}", "api_key_env": _MGMT_API_KEY_ENV},
        "tls": {"ca_cert": str(ca_cert), "ca_key": str(ca_key), "cert_cache_size": 1000, "leaf_cert_expiry_hours": 168},
-        "transforms": [
-            {"name": "allowlist", "config": {"domains": hosts}},
-            {"name": "secrets", "config": {"secrets": secrets_rules}},
-        ],
+        "transforms": [{"name": "allowlist", "config": {"domains": hosts}}, {"name": "secrets", "config": {"secrets": secrets_rules}}],
        "log": {"level": "info"},
    }


 def _open_private_append(path: Path, *, strict_chmod: bool) -> int:
-    """Open ``path`` O_APPEND|O_CREAT 0o600 with O_NOFOLLOW (planted symlinks refused); fchmod
-    tightens a pre-existing file (failure fatal iff ``strict_chmod``).  Raises OSError; caller owns the fd."""
+    """O_APPEND|O_CREAT 0o600 + O_NOFOLLOW (planted symlinks refused); fchmod tightens a pre-existing
+    file (failure fatal iff ``strict_chmod``).  Raises OSError; caller owns the fd."""
    fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_APPEND | _O_NOFOLLOW, 0o600)
    try:
        os.fchmod(fd, 0o600)
@@ -640,20 +561,17 @@ def _open_private_append(path: Path, *, strict_chmod: bool) -> int:


 def ensure_audit_log(audit_path: Path) -> None:
-    """Pre-create the audit log 0o600 (forward-compat: v0.39 never writes it).  RuntimeError on any
-    OSError (planted symlink, immutable dir, full disk)."""
+    """Pre-create the audit log 0o600 (forward-compat: v0.39 never writes it); RuntimeError on any OSError."""
    try:
        os.close(_open_private_append(audit_path, strict_chmod=True))
    except OSError as exc:
        raise RuntimeError(
-            f"Refusing to start: could not pre-create audit log {audit_path} with restrictive permissions ({exc}).  "
-            f"Move or chmod any existing file at that path and retry."
+            f"Refusing to start: could not pre-create audit log {audit_path} with restrictive permissions ({exc}).  Move or chmod any existing file at that path and retry."
        ) from exc


 def _write_state_file_atomic(state: Path, name: str, dump) -> Path:
-    """0600 temp file + atomic replace: the file holds proxy tokens, and chmod-after-replace would
-    leave a world-readable TOCTOU window."""
+    """0600 temp file + atomic replace: the file holds proxy tokens; chmod-after-replace would be a world-readable TOCTOU window."""
    tmp_path = state / f".{name}.tmp"
    with open(tmp_path, "w", encoding="utf-8") as f:
        dump(f)
@@ -664,33 +582,23 @@ def _write_state_file_atomic(state: Path, name: str, dump) -> Path:

 def write_proxy_config(config: Dict) -> Path:
    """Serialize the config dict to ``<hermes_home>/proxy/proxy.yaml`` (safe_dump, no Python tags)."""
-    yaml = _yaml()
-    if yaml is None:
+    if (yaml := _yaml()) is None:
        raise RuntimeError("PyYAML is required to write the iron-proxy config but is not installed.")
-    return _write_state_file_atomic(
-        _proxy_state_dir(), "proxy.yaml", lambda f: yaml.safe_dump(config, f, default_flow_style=False, sort_keys=False),
-    )
+    return _write_state_file_atomic(_proxy_state_dir(), "proxy.yaml", lambda f: yaml.safe_dump(config, f, default_flow_style=False, sort_keys=False))


 def write_mappings(mappings: List[TokenMapping]) -> Path:
    """Persist sandbox-visible tokens to ``mappings.json`` (read by the Docker backend, not iron-proxy)."""
-    payload = {
-        "version": 1,
-        "tokens": [
-            {
-                "proxy_token": m.proxy_token, "env_name": m.real_env_name, "upstream_hosts": list(m.upstream_hosts),
-                "match_headers": list(m.match_headers), "alias_env_names": list(m.alias_env_names),
-            }
-            for m in mappings
-        ],
-    }
+    payload = {"version": 1, "tokens": [{
+        "proxy_token": m.proxy_token, "env_name": m.real_env_name, "upstream_hosts": list(m.upstream_hosts),
+        "match_headers": list(m.match_headers), "alias_env_names": list(m.alias_env_names),
+    } for m in mappings]}
    return _write_state_file_atomic(_proxy_state_dir(), "mappings.json", lambda f: json.dump(payload, f, indent=2))


 def load_mappings() -> List[TokenMapping]:
    """Read mappings.json, if it exists.  Empty list on any error."""
-    f = _proxy_state_dir() / "mappings.json"
-    if not f.exists():
+    if not (f := _proxy_state_dir() / "mappings.json").exists():
        return []
    try:
        payload = json.loads(f.read_text(encoding="utf-8"))
@@ -699,28 +607,19 @@ def load_mappings() -> List[TokenMapping]:
        return []
    out: List[TokenMapping] = []
    for item in payload.get("tokens", []):
-        try:  # pre-header-auth files load with the bearer defaults they were written under
-            out.append(TokenMapping(
-                item["proxy_token"], item["env_name"], tuple(item.get("upstream_hosts") or ()),
-                tuple(item.get("match_headers") or ("Authorization",)), tuple(item.get("alias_env_names") or ()),
-            ))
-        except (KeyError, TypeError):
-            continue
+        with suppress(KeyError, TypeError):  # pre-header-auth files load with the bearer defaults they were written under
+            out.append(TokenMapping(item["proxy_token"], item["env_name"], tuple(item.get("upstream_hosts") or ()),
+                                    tuple(item.get("match_headers") or ("Authorization",)), tuple(item.get("alias_env_names") or ())))
    return out


-def _env_names(available_env_names: Optional[List[str]]) -> set:
-    """Explicit override (Bitwarden adapter) or the non-empty names in the host env."""
-    return set(available_env_names) if available_env_names is not None else {k for k, v in os.environ.items() if v}
-
-
 def discover_provider_mappings(*, available_env_names: Optional[List[str]] = None) -> List[TokenMapping]:
-    """One TokenMapping per known provider whose env var is set (bearer providers first).  Canonical
-    OR any alias present -> ONE mapping on the canonical name (the subprocess-env builder mirrors aliases)."""
-    names = _env_names(available_env_names)
+    """One TokenMapping per known provider whose env var is set (bearer providers first).  Canonical OR any alias
+    present -> ONE mapping on the canonical name (the subprocess-env builder mirrors aliases).
+    ``available_env_names`` (Bitwarden adapter) overrides the non-empty names in the host env."""
+    names = set(available_env_names) if available_env_names is not None else {k for k, v in os.environ.items() if v}
    specs = [(n, h, ("Authorization",), ()) for n, h in _BEARER_PROVIDERS.items()] + [
-        (n, tuple(s["hosts"]), tuple(s["match_headers"]), tuple(s.get("aliases") or ()))
-        for n, s in _HEADER_AUTH_PROVIDERS.items()
+        (n, tuple(s["hosts"]), tuple(s["match_headers"]), tuple(s.get("aliases") or ())) for n, s in _HEADER_AUTH_PROVIDERS.items()
    ]
    return [
        TokenMapping(mint_proxy_token(prefix=env_name.lower().replace("_api_key", "")), env_name, hosts, headers, aliases)
@@ -731,18 +630,15 @@ def discover_provider_mappings(*, available_env_names: Optional[List[str]] = Non

 def discover_uncovered_providers(*, available_env_names: Optional[List[str]] = None) -> List[str]:
    """Env names of recognized providers the proxy can't swap (SigV4 / SDK-minted OAuth)."""
-    names = _env_names(available_env_names)
+    names = set(available_env_names) if available_env_names is not None else {k for k, v in os.environ.items() if v}
    return [n for n in _NON_BEARER_PROVIDERS if n in names]


 def merge_mappings(*, existing: List[TokenMapping], discovered: List[TokenMapping], rotate: bool = False) -> List[TokenMapping]:
-    """Tokens already in ``existing`` are preserved (containers baked with them keep working) while
-    hosts/headers/aliases refresh from ``discovered``; ``rotate=True`` re-mints; undiscovered providers drop."""
+    """Existing tokens are preserved (containers baked with them keep working), hosts/headers/aliases refresh
+    from ``discovered``; ``rotate=True`` re-mints; undiscovered providers drop."""
    by_name = {} if rotate else {m.real_env_name: m for m in existing}
-    return [
-        replace(d, proxy_token=by_name[d.real_env_name].proxy_token) if d.real_env_name in by_name else d
-        for d in discovered
-    ]
+    return [replace(d, proxy_token=by_name[d.real_env_name].proxy_token) if d.real_env_name in by_name else d for d in discovered]


 def _pidfile() -> Path:
@@ -751,8 +647,8 @@ def _pidfile() -> Path:

 def _read_pid() -> Optional[int]:
    try:
-        pid = int((_proxy_state_dir_ro() / "iron-proxy.pid").read_text(encoding="utf-8").strip())
-    except (OSError, ValueError):
+        pid = int(_read_text_or_none(_proxy_state_dir_ro() / "iron-proxy.pid") or "")
+    except ValueError:
        return None
    return pid if pid > 0 else None

@@ -764,20 +660,18 @@ def _pid_proc_starttime(pid: int) -> Optional[str]:
    except OSError:
        return None
    # comm may contain spaces/parens, so split after the LAST ")"; field 22 -> tail index 19.
-    rparen = text.rfind(")")
-    fields = text[rparen + 1:].split() if rparen >= 0 else []
+    fields = text[rparen + 1:].split() if (rparen := text.rfind(")")) >= 0 else []
    return fields[19] if len(fields) > 19 else None


 def _persisted_nonce_path() -> Path:
-    """On-disk nonce sibling of the pidfile, so stop/status in a later CLI process can still
-    defeat PID recycling."""
+    """On-disk nonce sibling of the pidfile, so stop/status in a later CLI process can still defeat PID recycling."""
    return _proxy_state_dir_ro() / "iron-proxy.nonce"


 def _read_persisted_nonce() -> Optional[str]:
-    """Nonce from disk, or None if missing/unreadable/empty/not owned by us (callers fall back to
-    argv0 matching).  O_NOFOLLOW: this read decides whether stop_proxy SIGKILLs a PID."""
+    """Nonce from disk, or None if missing/unreadable/empty/not owned by us (callers fall back to argv0
+    matching).  O_NOFOLLOW: this read decides whether stop_proxy SIGKILLs a PID."""
    try:
        fd = os.open(str(_persisted_nonce_path()), os.O_RDONLY | _O_NOFOLLOW)
    except OSError:
@@ -791,9 +685,9 @@ def _read_persisted_nonce() -> Optional[str]:


 def _pid_alive(pid: int) -> bool:
-    """True iff ``pid`` is alive AND an iron-proxy process.  PID-reuse defense, in priority order:
-    nonce in /proc/<pid>/environ, argv[0] basename in /proc/<pid>/cmdline, ``ps -o comm=`` basename
-    (a loose ``"iron-proxy" in cmdline`` would hit ``tail iron-proxy.log``)."""
+    """True iff ``pid`` is alive AND an iron-proxy process.  PID-reuse defense, in priority order: nonce in
+    /proc/<pid>/environ, argv[0] basename in /proc/<pid>/cmdline, ``ps -o comm=`` basename (a loose
+    ``"iron-proxy" in cmdline`` would hit ``tail iron-proxy.log``)."""
    if pid <= 0:
        return False
    try:
@@ -815,10 +709,8 @@ def _pid_alive(pid: int) -> bool:
            if any(f"{_HERMES_IRON_PROXY_NONCE_ENV}={n}".encode() in env_bytes for n in nonce_candidates):
                return True
    with suppress(OSError):
-        cmdline_path = Path(f"/proc/{pid}/cmdline")
-        if cmdline_path.exists():
-            argv0 = cmdline_path.read_bytes().split(b"\x00")[0].decode("utf-8", errors="ignore")
-            return os.path.basename(argv0).startswith("iron-proxy")
+        if (cmdline_path := Path(f"/proc/{pid}/cmdline")).exists():
+            return os.path.basename(cmdline_path.read_bytes().split(b"\x00")[0].decode("utf-8", errors="ignore")).startswith("iron-proxy")
    with suppress(OSError, subprocess.TimeoutExpired):  # macOS / non-Linux fallback
        res = _run(["ps", "-p", str(pid), "-o", "comm="], timeout=2, text=True)
        if res.returncode == 0:
@@ -831,22 +723,17 @@ def start_proxy(
    *, binary: Optional[Path] = None, config_path: Optional[Path] = None, extra_env: Optional[Dict[str, str]] = None,
    install_if_missing: bool = True, refresh_secrets_from_bitwarden: bool = False, bitwarden_config: Optional[Dict] = None,
 ) -> ProxyStatus:
-    """Spawn iron-proxy as a managed background subprocess (idempotent if already running).
-    ``refresh_secrets_from_bitwarden`` re-fetches secrets from BWS — the ``credential_source: bitwarden`` rotation promise."""
+    """Spawn iron-proxy as a managed background subprocess (idempotent if already running).  ``refresh_secrets_from_bitwarden``
+    re-fetches secrets from BWS — the ``credential_source: bitwarden`` rotation promise."""
    global _proxy_nonce
-    existing = _read_pid()
-    if existing and _pid_alive(existing):
+    if (existing := _read_pid()) and _pid_alive(existing):
        return get_status()
-    bin_path = binary or find_iron_proxy(install_if_missing=install_if_missing)
-    if bin_path is None:
+    if (bin_path := binary or find_iron_proxy(install_if_missing=install_if_missing)) is None:
        raise RuntimeError("iron-proxy binary not available — run `hermes egress install`.")
-    cfg = config_path or (_proxy_state_dir() / "proxy.yaml")
-    if not cfg.exists():
+    if not (cfg := config_path or (_proxy_state_dir() / "proxy.yaml")).exists():
        raise RuntimeError(f"iron-proxy config not found at {cfg}. Run `hermes egress setup` first.")
    # Minimal env: os.environ.copy() would expose every operator secret via /proc/<pid>/environ.
-    env = _build_proxy_subprocess_env(
-        extra_env=extra_env, refresh_from_bitwarden=refresh_secrets_from_bitwarden, bitwarden_config=bitwarden_config,
-    )
+    env = _build_proxy_subprocess_env(extra_env=extra_env, refresh_from_bitwarden=refresh_secrets_from_bitwarden, bitwarden_config=bitwarden_config)
    # v0.39 validates api_key_env is non-empty when management.listen is set.
    if _read_management_listen_from_config(cfg) is not None:
        env[_MGMT_API_KEY_ENV] = ensure_management_token()
@@ -870,14 +757,12 @@ def start_proxy(
        pidfile.unlink(missing_ok=True)
        return RuntimeError(f"{msg}Last log lines:\n{tail}")

-    def _exited_error() -> RuntimeError:
-        return _abort(f"iron-proxy exited immediately (code {proc.returncode}). ", kill=False)
-
    def _interrupt_handler(_signum, _frame):  # pragma: no cover - signal path
        _kill_and_wait(proc, grace_seconds=2)  # Ctrl-C while waiting must not leak an orphan holding the port
        pidfile.unlink(missing_ok=True)
        raise KeyboardInterrupt()

+    _exited_error = lambda: _abort(f"iron-proxy exited immediately (code {proc.returncode}). ", kill=False)  # noqa: E731
    # Probe the CONFIGURED bind host (on Linux the docker bridge, where loopback never connects).
    probe_host, tunnel_port = _probe_target()
    with _interrupt_guard(_interrupt_handler):
@@ -894,7 +779,7 @@ def start_proxy(

@contextmanager
 def _interrupt_guard(handler):
-    """Route SIGINT/SIGTERM to ``handler`` for the block (POSIX main thread only), restoring the previous handlers after."""
+    """Route SIGINT/SIGTERM to ``handler`` for the block (POSIX main thread only); previous handlers restored after."""
    if platform.system() == "Windows" or threading.current_thread() is not threading.main_thread():
        yield
        return
@@ -907,22 +792,21 @@ def _interrupt_guard(handler):


 def _spawn_daemon(bin_path: Path, cfg: Path, env: Dict[str, str], log_path: Path) -> "subprocess.Popen":
-    """Popen with stdout/stderr appended to ``log_path`` (0o600 from the first byte, O_NOFOLLOW so a
-    planted symlink e.g. to authorized_keys can't receive output, owner-checked).  Our log fd is
-    closed right after Popen — the child has its dup."""
+    """Popen with stdout/stderr appended to ``log_path`` (0o600 from the first byte, O_NOFOLLOW so a planted
+    symlink e.g. to authorized_keys can't receive output, owner-checked).  Our log fd closes after Popen — the child has its dup."""
    try:
        log_fd = _open_private_append(log_path, strict_chmod=False)
    except OSError as exc:
        raise RuntimeError(f"Refusing to write iron-proxy log {log_path}: {exc}.  Remove that path manually and retry.") from exc
    if not _fd_owned_by_us(log_fd):
-        st = os.fstat(log_fd)
+        uid = os.fstat(log_fd).st_uid
        os.close(log_fd)
-        raise RuntimeError(f"iron-proxy log {log_path} has unexpected owner uid={st.st_uid}; refusing to write.")
+        raise RuntimeError(f"iron-proxy log {log_path} has unexpected owner uid={uid}; refusing to write.")
    try:
        # start_new_session is POSIX-only (Windows isn't supported anyway — no upstream binary).
-        session_kw = {} if platform.system() == "Windows" else {"start_new_session": True}
        return subprocess.Popen(  # noqa: S603
-            [str(bin_path), "-config", str(cfg)], env=env, stdin=subprocess.DEVNULL, stdout=log_fd, stderr=subprocess.STDOUT, **session_kw,
+            [str(bin_path), "-config", str(cfg)], env=env, stdin=subprocess.DEVNULL, stdout=log_fd, stderr=subprocess.STDOUT,
+            **({} if platform.system() == "Windows" else {"start_new_session": True}),
        )
    except OSError as exc:
        raise RuntimeError(f"failed to spawn iron-proxy: {exc}") from exc
@@ -932,8 +816,7 @@ def _spawn_daemon(bin_path: Path, cfg: Path, env: Dict[str, str], log_path: Path


 def _await_listening(proc: "subprocess.Popen", host: str, port: int, *, on_exit) -> bool:
-    """Poll until ``host:port`` accepts or the grace window lapses (do-while: at least one check
-    even with a 0s window).  Raises ``on_exit()`` if the child dies meanwhile."""
+    """Poll until ``host:port`` accepts or the grace window lapses (do-while: >=1 check even at 0s); raises ``on_exit()`` if the child dies."""
    deadline = time.time() + _STARTUP_GRACE_SECONDS
    while True:
        if proc.poll() is not None:
@@ -946,17 +829,15 @@ def _await_listening(proc: "subprocess.Popen", host: str, port: int, *, on_exit)


 def _write_pidfile_safely(pidfile: Path, pid: int) -> None:
-    """O_EXCL + O_NOFOLLOW + ownership check, then persist the nonce.  An existing pidfile is either
-    a concurrent start (fail cleanly) or a stale crash leftover (unlink and retry once)."""
+    """O_EXCL + O_NOFOLLOW + ownership check, then persist the nonce.  An existing pidfile is either a concurrent
+    start (fail cleanly) or a stale crash leftover (unlink and retry once)."""
    open_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW
    try:
        fd = os.open(str(pidfile), open_flags, 0o600)
    except FileExistsError:
-        existing_pid = _read_pid()
-        if existing_pid and _pid_alive(existing_pid):
+        if (existing_pid := _read_pid()) and _pid_alive(existing_pid):
            raise RuntimeError(
-                f"Another iron-proxy start appears to be in progress (pidfile {pidfile} -> pid {existing_pid}).  "
-                f"Run `hermes egress stop` if that proxy is stuck."
+                f"Another iron-proxy start appears to be in progress (pidfile {pidfile} -> pid {existing_pid}).  Run `hermes egress stop` if that proxy is stuck."
            )
        pidfile.unlink(missing_ok=True)
        fd = os.open(str(pidfile), open_flags, 0o600)
@@ -976,7 +857,6 @@ def _write_pidfile_safely(pidfile: Path, pid: int) -> None:


 def _kill_and_wait(proc: "subprocess.Popen", *, grace_seconds: int = 2) -> None:
-    """Best-effort SIGTERM → wait → SIGKILL for a child we own."""
    try:
        proc.terminate()
    except OSError:
@@ -993,18 +873,16 @@ def _kill_and_wait(proc: "subprocess.Popen", *, grace_seconds: int = 2) -> None:
 def _build_proxy_subprocess_env(
    *, extra_env: Optional[Dict[str, str]] = None, refresh_from_bitwarden: bool = False, bitwarden_config: Optional[Dict] = None,
 ) -> Dict[str, str]:
-    """Allowlisted infra vars + the secrets named in mappings.  With ``refresh_from_bitwarden`` and a
-    populated ``bitwarden_config`` secrets come from BWS (the rotation guarantee); without
-    ``allow_env_fallback`` any BWS shortfall fails closed instead of keeping stale host-env values."""
-    env = _allowlisted_env()
-    parent = os.environ
+    """Allowlisted infra vars + the secrets named in mappings.  With ``refresh_from_bitwarden`` and a populated
+    ``bitwarden_config`` secrets come from BWS (the rotation guarantee); without ``allow_env_fallback`` any BWS
+    shortfall fails closed instead of keeping stale host-env values."""
+    env, parent = _allowlisted_env(), os.environ
    # Forward ONLY mapped secrets; the rule is keyed on the canonical name, so mirror an alias value into it.
    mappings = load_mappings()
    needed = {m.real_env_name for m in mappings}
    alias_sources = {m.real_env_name: m.alias_env_names for m in mappings if m.alias_env_names}
    for name in needed:
-        source = name if name in parent else next((a for a in alias_sources.get(name, ()) if parent.get(a)), None)
-        if source is not None:
+        if (source := name if name in parent else next((a for a in alias_sources.get(name, ()) if parent.get(a)), None)) is not None:
            env[name] = parent[source]
    if refresh_from_bitwarden and bitwarden_config:
        _refresh_secrets_from_bitwarden(env, needed, bitwarden_config, bool(bitwarden_config.get("allow_env_fallback")))
@@ -1017,16 +895,17 @@ def _build_proxy_subprocess_env(
    return env


-def _bitwarden_shortfall(allow_env_fallback: bool, error: str, warning: str, *args) -> None:
-    """Raise ``error`` unless the operator opted into the legacy host-env fallback (then log ``warning``)."""
+def _bitwarden_shortfall(allow_env_fallback: bool, error: str, warning: str, *args, cause: Optional[BaseException] = None) -> None:
+    """Raise ``error`` (chained to ``cause`` when given) unless the operator opted into the legacy host-env fallback (then log ``warning``)."""
    if not allow_env_fallback:
+        if cause is not None:
+            raise RuntimeError(error) from cause
        raise RuntimeError(error)
    logger.warning(warning, *args)


 def _refresh_secrets_from_bitwarden(env: Dict[str, str], needed: set, bitwarden_config: Dict, allow_env_fallback: bool) -> None:
-    """Overwrite ``env[needed]`` with fresh (uncached) BWS values; only mapped names are injected so
-    unrelated BWS secrets never leak into the daemon env."""
+    """Overwrite ``env[needed]`` with fresh (uncached) BWS values; only mapped names are injected so unrelated BWS secrets never leak."""
    try:
        # Lazy: the bitwarden module isn't importable in every install.
        from agent.secret_sources import bitwarden as bw
@@ -1038,40 +917,35 @@ def _refresh_secrets_from_bitwarden(env: Dict[str, str], needed: set, bitwarden_
                allow_env_fallback,
                "credential_source=bitwarden but the access-token env or project_id is empty.  Either set both, switch to "
                "credential_source: env, or set `proxy.allow_env_fallback: true` to opt into the legacy fallback behaviour.",
-                "credential_source=bitwarden but access-token env or project_id is empty — "
-                "proxy will fall back to parent env (allow_env_fallback=true).",
+                "credential_source=bitwarden but access-token env or project_id is empty — proxy will fall back to parent env (allow_env_fallback=true).",
            )
            return
        secrets, warnings = bw.fetch_bitwarden_secrets(access_token=access_token, project_id=project_id, cache_ttl_seconds=0, use_cache=False)
    except ImportError as exc:
        # A dependency vanishing between setup and restart must not silently degrade.
-        if not allow_env_fallback:
-            raise RuntimeError(
-                "Bitwarden refresh module unavailable at proxy start (credential_source=bitwarden with "
-                "proxy.allow_env_fallback: false).  Either fix the import, switch to credential_source: env, or set "
-                "`proxy.allow_env_fallback: true` to opt into the legacy fallback behaviour."
-            ) from exc
-        logger.warning("Bitwarden refresh module unavailable at proxy start, falling back to parent env (allow_env_fallback=true): %s", exc)
+        _bitwarden_shortfall(
+            allow_env_fallback,
+            "Bitwarden refresh module unavailable at proxy start (credential_source=bitwarden with "
+            "proxy.allow_env_fallback: false).  Either fix the import, switch to credential_source: env, or set "
+            "`proxy.allow_env_fallback: true` to opt into the legacy fallback behaviour.",
+            "Bitwarden refresh module unavailable at proxy start, falling back to parent env (allow_env_fallback=true): %s", exc, cause=exc,
+        )
        return
    missing = sorted(needed - set(secrets))
-    for n in needed:
-        if n in secrets:
-            env[n] = secrets[n]
+    env.update((n, secrets[n]) for n in needed if n in secrets)
    if missing:
        _bitwarden_shortfall(
            allow_env_fallback,
            f"Bitwarden refresh did not return secrets for {missing}.  Either add the secrets to your BWS project, switch to "
            f"credential_source: env via `hermes egress setup --no-bitwarden`, or set `proxy.allow_env_fallback: true` in "
            f"config.yaml to opt into the legacy host-env fallback.",
-            "Bitwarden refresh did not return secrets for %s — falling back to host env for those names (allow_env_fallback=true).",
-            missing,
+            "Bitwarden refresh did not return secrets for %s — falling back to host env for those names (allow_env_fallback=true).", missing,
        )
    if warnings:  # log only the count: the taint analyzer can't tell bws status text is non-secret
        logger.warning("Bitwarden refresh produced %d warning(s); run `hermes secrets bitwarden status` for detail.", len(warnings))


 def _forget_daemon() -> None:
-    """Best-effort removal of pidfile + persisted nonce, and drop the in-process nonce."""
    global _proxy_nonce
    _pidfile().unlink(missing_ok=True)
    with suppress(OSError):
@@ -1080,7 +954,7 @@ def _forget_daemon() -> None:


 def stop_proxy() -> bool:
-    """Stop the managed iron-proxy.  Returns True if it was running."""
+    """Returns True if it was running."""
    pid = _read_pid()
    if not pid or not _pid_alive(pid):
        _forget_daemon()
@@ -1100,8 +974,7 @@ def stop_proxy() -> bool:
        time.sleep(0.1)
    else:
        starttime_after = _pid_proc_starttime(pid)
-        recycled = (starttime_before is not None and starttime_after is not None and starttime_before != starttime_after) or not _pid_alive(pid)
-        if recycled:
+        if (starttime_before is not None and starttime_after is not None and starttime_before != starttime_after) or not _pid_alive(pid):
            logger.warning("iron-proxy pid=%s appears recycled before SIGKILL; not killing.", pid)
        else:
            with suppress(ProcessLookupError):
@@ -1115,22 +988,17 @@ def get_status() -> ProxyStatus:
    """Snapshot the proxy state without side effects (called per Docker container create)."""
    status = ProxyStatus()
    probe_host, status.tunnel_port = _probe_target()
-    binary = find_iron_proxy(install_if_missing=False)
-    if binary:
-        status.binary_path = binary
-        status.binary_version = iron_proxy_version(binary)
+    if binary := find_iron_proxy(install_if_missing=False):
+        status.binary_path, status.binary_version = binary, iron_proxy_version(binary)
    state = _proxy_state_dir_ro()
    cfg, ca = state / "proxy.yaml", state / "ca.crt"
-    status.config_path = cfg if cfg.exists() else None
-    status.ca_cert_path = ca if ca.exists() else None
-    pid = _read_pid()
-    if pid and _pid_alive(pid):
+    status.config_path, status.ca_cert_path = (cfg if cfg.exists() else None), (ca if ca.exists() else None)
+    if (pid := _read_pid()) and _pid_alive(pid):
        status.pid, status.listening = pid, _port_listening(probe_host, status.tunnel_port)
    return status


 def _port_listening(host: str, port: int) -> bool:
-    """Cheap TCP connect probe — True iff something accepts on host:port."""
    import socket
    try:
        with socket.create_connection((host, port), timeout=0.5):
@@ -1149,7 +1017,6 @@ def _tail_log(path: Path, *, lines: int = 20) -> str:


 def _reset_for_tests() -> None:
-    """Clear the module's mutable globals (``_VERSION_CACHE``, ``_proxy_nonce``)."""
    global _proxy_nonce
    _VERSION_CACHE.clear()
    _proxy_nonce = None