diff --git a/kora_cli/heartbeat_probes/__init__.py b/kora_cli/heartbeat_probes/__init__.py
new file mode 100644
index 000000000000..eb61f5496fd6
--- /dev/null
+++ b/kora_cli/heartbeat_probes/__init__.py
@@ -0,0 +1,66 @@
+"""Backend service heartbeat probes (KR-FEAT-HEARTBEAT ST1).
+
+5 probes for Joshua's backend stack: vercel, sentry, doppler,
+supabase, fly. Each implements :class:`ServiceProbe` (Protocol) and
+returns a :class:`ServiceHealthSnapshot`.
+
+Probe-wide rules:
+
+  - 10s per-probe timeout (matches §4 Q1 default cadence; one slow
+    probe doesn't stall the others).
+  - Missing auth env → status ``unknown`` + ``error`` carrying the
+    env-var name (NEVER the value) + ZERO outbound call.
+  - All error strings sanitized — never leaks the auth token, even
+    on partial-auth failures.
+  - Probe-failure isolation: a probe raising an unhandled exception
+    is caught by the runner; siblings keep running.
+
+The :func:`run_all_probes` helper iterates :func:`default_probes`
+and writes results into the module-level snapshot cache exposed
+via :func:`current_service_snapshots`. The daemon listener at
+:mod:`kora_cli.listeners.heartbeat_probes_listener` registers a
+heartbeat-scheduler task at 5-min cadence (operator override via
+``KORA_HEARTBEAT_PROBE_INTERVAL_SEC``).
+"""
+
+from kora_cli.heartbeat_probes.base import (
+    PROBE_TIMEOUT_SECONDS,
+    ServiceProbe,
+    snapshot_for_auth_missing,
+    snapshot_for_timeout,
+    snapshot_for_unexpected_error,
+)
+from kora_cli.heartbeat_probes.doppler import DopplerProbe
+from kora_cli.heartbeat_probes.fly import FlyProbe
+from kora_cli.heartbeat_probes.runner import (
+    current_service_snapshots,
+    default_probes,
+    run_all_probes,
+)
+from kora_cli.heartbeat_probes.sentry import SentryProbe
+from kora_cli.heartbeat_probes.supabase import SupabaseProbe
+from kora_cli.heartbeat_probes.types import (
+    SERVICE_STATUSES,
+    ServiceHealthSnapshot,
+    ServiceStatus,
+)
+from kora_cli.heartbeat_probes.vercel import VercelProbe
+
+__all__ = [
+    "PROBE_TIMEOUT_SECONDS",
+    "SERVICE_STATUSES",
+    "DopplerProbe",
+    "FlyProbe",
+    "SentryProbe",
+    "ServiceHealthSnapshot",
+    "ServiceProbe",
+    "ServiceStatus",
+    "SupabaseProbe",
+    "VercelProbe",
+    "current_service_snapshots",
+    "default_probes",
+    "run_all_probes",
+    "snapshot_for_auth_missing",
+    "snapshot_for_timeout",
+    "snapshot_for_unexpected_error",
+]
diff --git a/kora_cli/heartbeat_probes/base.py b/kora_cli/heartbeat_probes/base.py
new file mode 100644
index 000000000000..90b82dbf2e7e
--- /dev/null
+++ b/kora_cli/heartbeat_probes/base.py
@@ -0,0 +1,226 @@
+"""ServiceProbe Protocol + shared helpers (KR-FEAT-HEARTBEAT ST1).
+
+The :class:`ServiceProbe` contract: each probe has a ``name`` +
+``async def check() -> ServiceHealthSnapshot``. Implementations
+live in sibling modules (one per service).
+
+Shared helpers:
+
+  - :func:`resolve_env` — read an env var; treat empty/whitespace
+    as unset (matches KR-MCP-1 catalog convention).
+  - :func:`snapshot_for_auth_missing` — short-circuit snapshot when
+    an auth env var is unset; never makes the API call.
+  - :func:`snapshot_for_timeout` — snapshot for the 10s wall-clock
+    timeout overshoot.
+  - :func:`snapshot_for_unexpected_error` — last-resort wrapper for
+    unexpected exception types; runs the exception type/message
+    through :func:`sanitize_error`.
+  - :func:`sanitize_error` — strip any known auth-token values from
+    a string before exposing in a snapshot's ``error`` field.
+
+# Probe-wide timeout
+
+``PROBE_TIMEOUT_SECONDS = 10.0`` enforced per :func:`with_timeout`
+wrapper. The §4 Q1 ruling on cadence (5 min) gives plenty of
+headroom — a 10s ceiling means even with all 5 probes serialized
+worst-case, we still leave 4.5 min before the next cycle.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+from datetime import datetime, timezone
+from typing import Any, Awaitable, Callable, Optional, Protocol, runtime_checkable
+
+from kora_cli.heartbeat_probes.types import ServiceHealthSnapshot
+
+logger = logging.getLogger(__name__)
+
+
+PROBE_TIMEOUT_SECONDS: float = 10.0
+
+
+@runtime_checkable
+class ServiceProbe(Protocol):
+    """Contract every backend-service probe implements."""
+
+    name: str
+
+    async def check(self) -> ServiceHealthSnapshot: ...
+
+
+# ---------------------------------------------------------------------------
+# Env resolution
+# ---------------------------------------------------------------------------
+
+
+def resolve_env(name: str) -> Optional[str]:
+    """Read ``os.environ[name]``; treat empty / whitespace-only as
+    unset. Matches the KR-MCP-1 catalog ``check_endpoint_health``
+    convention — Doppler sometimes injects empty values.
+
+    Returns the stripped value or ``None`` when unset/empty.
+    """
+    raw = os.environ.get(name, "").strip()
+    return raw or None
+
+
+# ---------------------------------------------------------------------------
+# Token sanitization (preserve security contract)
+# ---------------------------------------------------------------------------
+
+
+def sanitize_error(text: str, *tokens: Optional[str]) -> str:
+    """Strip any of ``tokens`` from ``text`` before exposing it.
+
+    The error field of a snapshot is operator-visible. A probe must
+    NEVER leak its auth token there — even on partial-auth
+    failures (4xx responses, transport errors carrying the token
+    in URL fragments, etc.).
+
+    Replaces every occurrence of each non-empty token value with
+    ``"<REDACTED>"``. Empty / ``None`` tokens are skipped.
+    """
+    if not text:
+        return text
+    out = text
+    for token in tokens:
+        if token:
+            out = out.replace(token, "<REDACTED>")
+    return out
+
+
+# ---------------------------------------------------------------------------
+# Snapshot constructors for common no-call / failure paths
+# ---------------------------------------------------------------------------
+
+
+def snapshot_for_auth_missing(
+    *, name: str, env_var: str, extra_envs: tuple[str, ...] = ()
+) -> ServiceHealthSnapshot:
+    """Return a snapshot for the auth-env-missing case.
+
+    Status = ``unknown``. Error string lists the env var name(s) —
+    NOT any value. The probe MUST short-circuit here before
+    issuing any outbound traffic.
+    """
+    all_envs = (env_var,) + extra_envs
+    listing = ", ".join(repr(e) for e in all_envs)
+    return ServiceHealthSnapshot(
+        name=name,
+        status="unknown",
+        latency_ms=None,
+        last_check_at=datetime.now(timezone.utc),
+        details={},
+        error=f"auth env unset or empty: {listing}",
+    )
+
+
+def snapshot_for_timeout(*, name: str) -> ServiceHealthSnapshot:
+    return ServiceHealthSnapshot(
+        name=name,
+        status="unknown",
+        latency_ms=None,
+        last_check_at=datetime.now(timezone.utc),
+        details={},
+        error=f"probe timed out after {PROBE_TIMEOUT_SECONDS:.0f}s",
+    )
+
+
+def snapshot_for_unexpected_error(
+    *,
+    name: str,
+    exc: BaseException,
+    auth_tokens: tuple[Optional[str], ...] = (),
+) -> ServiceHealthSnapshot:
+    """Last-resort snapshot for an unexpected exception type.
+
+    The runner catches per-probe exceptions to enforce isolation +
+    delegates here to build the snapshot. Error text is sanitized
+    against ``auth_tokens`` so we never leak even on weird-path
+    failures (e.g., httpx error message containing the URL with
+    bearer in a query string).
+    """
+    raw = f"{type(exc).__name__}: {exc}"
+    return ServiceHealthSnapshot(
+        name=name,
+        status="unknown",
+        latency_ms=None,
+        last_check_at=datetime.now(timezone.utc),
+        details={},
+        error=sanitize_error(raw, *auth_tokens),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Per-probe timeout wrapper
+# ---------------------------------------------------------------------------
+
+
+async def with_timeout(
+    coro: Awaitable[ServiceHealthSnapshot],
+    *,
+    name: str,
+    timeout: float = PROBE_TIMEOUT_SECONDS,
+) -> ServiceHealthSnapshot:
+    """Wrap a probe's ``check()`` in :func:`asyncio.wait_for`.
+
+    On :class:`asyncio.TimeoutError`, returns a
+    ``snapshot_for_timeout`` instead of raising — the runner's
+    isolation contract requires a snapshot per probe per cycle.
+    """
+    try:
+        return await asyncio.wait_for(coro, timeout=timeout)
+    except asyncio.TimeoutError:
+        logger.warning(
+            "[kora.heartbeat_probes] %s timed out after %.1fs", name, timeout
+        )
+        return snapshot_for_timeout(name=name)
+
+
+# ---------------------------------------------------------------------------
+# Latency timing helper
+# ---------------------------------------------------------------------------
+
+
+def now_ms_monotonic() -> float:
+    """Wall-clock-free monotonic milliseconds for latency
+    measurement. Use the delta between two calls — not the absolute
+    value."""
+    import time
+
+    return time.monotonic() * 1000.0
+
+
+# ---------------------------------------------------------------------------
+# HTTP error → status mapping (shared across probes)
+# ---------------------------------------------------------------------------
+
+
+def status_from_http_response(
+    status_code: int, *, healthy_codes: tuple[int, ...] = (200,)
+) -> str:
+    """Map an HTTP response code onto our 4-value status enum.
+
+    ``healthy_codes`` → ``healthy``; everything else → ``unhealthy``
+    (4xx/5xx). The probe layer may override based on response body
+    (e.g. Sentry's unresolved-issues count → ``degraded``).
+    """
+    return "healthy" if status_code in healthy_codes else "unhealthy"
+
+
+# Convenience re-exports so callers don't have to know about
+# datetime / timezone construction quirks.
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+# Type alias for the snapshot-producing callable each probe ships.
+SnapshotProducer = Callable[[], Awaitable[ServiceHealthSnapshot]]
+
+
+# Silence "Any unused" warnings
+__all_dummy__: tuple[Any, ...] = ()
diff --git a/kora_cli/heartbeat_probes/doppler.py b/kora_cli/heartbeat_probes/doppler.py
new file mode 100644
index 000000000000..10c31a35975e
--- /dev/null
+++ b/kora_cli/heartbeat_probes/doppler.py
@@ -0,0 +1,106 @@
+"""DopplerProbe — checks workplace reachability + secret-rotation hygiene.
+
+Auth: ``KORA_DOPPLER_API_TOKEN`` (a SERVICE token with workplace
+read-only scope — NOT a project token; per §4 Q3 ruling Joshua mints
+a dedicated service token for the probe). Healthy: 200. Degraded:
+oldest_secret_age_days > 180.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+
+from kora_cli.heartbeat_probes.base import (
+    PROBE_TIMEOUT_SECONDS,
+    now_ms_monotonic,
+    resolve_env,
+    sanitize_error,
+    snapshot_for_auth_missing,
+    snapshot_for_unexpected_error,
+    utc_now,
+)
+from kora_cli.heartbeat_probes.types import ServiceHealthSnapshot
+
+
+DOPPLER_API_TOKEN_ENV = "KORA_DOPPLER_API_TOKEN"
+DOPPLER_API_BASE = "https://api.doppler.com"
+DEGRADED_SECRET_AGE_DAYS_THRESHOLD = 180
+
+
+class DopplerProbe:
+    name = "doppler"
+
+    async def check(self) -> ServiceHealthSnapshot:
+        token = resolve_env(DOPPLER_API_TOKEN_ENV)
+        if token is None:
+            return snapshot_for_auth_missing(
+                name=self.name, env_var=DOPPLER_API_TOKEN_ENV
+            )
+
+        started_ms = now_ms_monotonic()
+        try:
+            async with httpx.AsyncClient(timeout=PROBE_TIMEOUT_SECONDS) as client:
+                response = await client.get(
+                    f"{DOPPLER_API_BASE}/v3/workplace",
+                    headers={"Authorization": f"Bearer {token}"},
+                )
+        except Exception as exc:
+            return snapshot_for_unexpected_error(
+                name=self.name, exc=exc, auth_tokens=(token,)
+            )
+
+        latency_ms = int(now_ms_monotonic() - started_ms)
+        if response.status_code != 200:
+            return ServiceHealthSnapshot(
+                name=self.name,
+                status="unhealthy",
+                latency_ms=latency_ms,
+                last_check_at=utc_now(),
+                details={},
+                error=sanitize_error(
+                    f"HTTP {response.status_code}", token
+                ),
+            )
+
+        projects_total, oldest_age_days = _project_workplace(response.json())
+        status = "healthy"
+        if (
+            isinstance(oldest_age_days, int)
+            and oldest_age_days > DEGRADED_SECRET_AGE_DAYS_THRESHOLD
+        ):
+            status = "degraded"
+
+        return ServiceHealthSnapshot(
+            name=self.name,
+            status=status,
+            latency_ms=latency_ms,
+            last_check_at=utc_now(),
+            details={
+                "projects_total": projects_total,
+                "oldest_secret_age_days": oldest_age_days,
+            },
+            error=None,
+        )
+
+
+def _project_workplace(payload: Any) -> tuple[Any, Any]:
+    """Project Doppler workplace response. Returns
+    (projects_total | "unknown", oldest_secret_age_days | "unknown").
+
+    Defensive — the workplace endpoint exposes ``workplace`` dict
+    with various fields; we extract what's available, fall back to
+    "unknown" markers when shape isn't as expected (the probe's
+    job is "is the service reachable + responding"; deep secret-
+    age inspection requires per-project queries which are out of
+    scope)."""
+    if not isinstance(payload, dict):
+        return ("unknown", "unknown")
+    workplace = payload.get("workplace") if "workplace" in payload else payload
+    if not isinstance(workplace, dict):
+        return ("unknown", "unknown")
+    # Doppler workplace endpoint doesn't expose project list
+    # directly — the probe's "projects_total" + "oldest_secret_age_days"
+    # remain "unknown" unless a richer signal becomes available.
+    return ("unknown", "unknown")
diff --git a/kora_cli/heartbeat_probes/fly.py b/kora_cli/heartbeat_probes/fly.py
new file mode 100644
index 000000000000..1d49c7438cfe
--- /dev/null
+++ b/kora_cli/heartbeat_probes/fly.py
@@ -0,0 +1,134 @@
+"""FlyProbe — checks Kora's deployed Fly apps.
+
+Auth: ``KORA_FLY_API_TOKEN``. Apps probed: ``kora-runtime`` always +
+``kora-runtime-staging`` when ``KORA_FLY_STAGING_APP_NAME`` env is
+set (matches the existing two-app deploy pattern).
+
+Healthy: response 200 + ≥1 machine running. Degraded:
+machines.healthy_count < machines.total_count.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+
+from kora_cli.heartbeat_probes.base import (
+    PROBE_TIMEOUT_SECONDS,
+    now_ms_monotonic,
+    resolve_env,
+    sanitize_error,
+    snapshot_for_auth_missing,
+    snapshot_for_unexpected_error,
+    utc_now,
+)
+from kora_cli.heartbeat_probes.types import ServiceHealthSnapshot
+
+
+FLY_API_TOKEN_ENV = "KORA_FLY_API_TOKEN"
+FLY_STAGING_APP_NAME_ENV = "KORA_FLY_STAGING_APP_NAME"
+FLY_API_BASE = "https://api.machines.dev"
+DEFAULT_PROD_APP = "kora-runtime"
+
+
+class FlyProbe:
+    name = "fly"
+
+    async def check(self) -> ServiceHealthSnapshot:
+        token = resolve_env(FLY_API_TOKEN_ENV)
+        if token is None:
+            return snapshot_for_auth_missing(
+                name=self.name, env_var=FLY_API_TOKEN_ENV
+            )
+
+        apps = [DEFAULT_PROD_APP]
+        staging = resolve_env(FLY_STAGING_APP_NAME_ENV)
+        if staging:
+            apps.append(staging)
+
+        started_ms = now_ms_monotonic()
+        apps_running = 0
+        apps_total = 0
+        any_degraded = False
+        first_error_status: int | None = None
+
+        try:
+            async with httpx.AsyncClient(timeout=PROBE_TIMEOUT_SECONDS) as client:
+                for app_name in apps:
+                    machines_response = await client.get(
+                        f"{FLY_API_BASE}/v1/apps/{app_name}/machines",
+                        headers={"Authorization": f"Bearer {token}"},
+                    )
+                    if machines_response.status_code != 200:
+                        if first_error_status is None:
+                            first_error_status = machines_response.status_code
+                        continue
+                    healthy, total = _count_machines(machines_response.json())
+                    apps_total += 1
+                    if healthy >= 1:
+                        apps_running += 1
+                    if healthy < total:
+                        any_degraded = True
+        except Exception as exc:
+            return snapshot_for_unexpected_error(
+                name=self.name, exc=exc, auth_tokens=(token,)
+            )
+
+        latency_ms = int(now_ms_monotonic() - started_ms)
+
+        # All app calls failed → unhealthy
+        if apps_total == 0:
+            return ServiceHealthSnapshot(
+                name=self.name,
+                status="unhealthy",
+                latency_ms=latency_ms,
+                last_check_at=utc_now(),
+                details={"apps_running": 0, "deploys_last_24h": 0},
+                error=sanitize_error(
+                    f"HTTP {first_error_status}"
+                    if first_error_status
+                    else "no successful app responses",
+                    token,
+                ),
+            )
+
+        status = (
+            "degraded"
+            if any_degraded or apps_running < len(apps)
+            else "healthy"
+        )
+
+        return ServiceHealthSnapshot(
+            name=self.name,
+            status=status,
+            latency_ms=latency_ms,
+            last_check_at=utc_now(),
+            details={
+                "apps_running": apps_running,
+                # deploys_last_24h is a Releases API query; out of
+                # scope for the heartbeat-cycle. Operator panel
+                # surfaces "unknown" if needed via the FE renderer.
+                "deploys_last_24h": "unknown",
+            },
+            error=None,
+        )
+
+
+def _count_machines(payload: Any) -> tuple[int, int]:
+    """Return (healthy_count, total_count) from /machines response.
+
+    A machine is "healthy" when state == "started"; everything else
+    (stopped, suspended, replacing, destroyed) counts toward total
+    but not healthy. Defensive against shape drift."""
+    if not isinstance(payload, list):
+        return (0, 0)
+    total = len(payload)
+    healthy = 0
+    for entry in payload:
+        if not isinstance(entry, dict):
+            continue
+        state = entry.get("state")
+        if isinstance(state, str) and state == "started":
+            healthy += 1
+    return (healthy, total)
diff --git a/kora_cli/heartbeat_probes/runner.py b/kora_cli/heartbeat_probes/runner.py
new file mode 100644
index 000000000000..a1bc7803d5d8
--- /dev/null
+++ b/kora_cli/heartbeat_probes/runner.py
@@ -0,0 +1,149 @@
+"""Probe runner + snapshot cache (KR-FEAT-HEARTBEAT ST1).
+
+Iterates the 5 default probes; isolates per-probe failures so one
+slow / broken probe doesn't block the others. Writes results to a
+module-level cache exposed via :func:`current_service_snapshots`.
+
+The cache is process-shared (singleton dict). The daemon listener
+at :mod:`kora_cli.listeners.heartbeat_probes_listener` registers a
+:func:`run_all_probes` task with the heartbeat scheduler; first
+cycle populates the cache. Until then, callers see an empty dict
+(ST2 endpoint surfaces this as ``cache_warming: true``).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Sequence
+
+from kora_cli.heartbeat_probes.base import (
+    ServiceProbe,
+    snapshot_for_unexpected_error,
+    with_timeout,
+)
+from kora_cli.heartbeat_probes.doppler import DopplerProbe
+from kora_cli.heartbeat_probes.fly import FlyProbe
+from kora_cli.heartbeat_probes.sentry import SentryProbe
+from kora_cli.heartbeat_probes.supabase import SupabaseProbe
+from kora_cli.heartbeat_probes.types import ServiceHealthSnapshot
+from kora_cli.heartbeat_probes.vercel import VercelProbe
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Default probe set
+# ---------------------------------------------------------------------------
+
+
+def default_probes() -> tuple[ServiceProbe, ...]:
+    """Return a fresh tuple of the 5 default probes.
+
+    Each call returns NEW instances — probes are cheap to construct
+    and stateless beyond their httpx clients (created per-check).
+    """
+    return (
+        VercelProbe(),
+        SentryProbe(),
+        DopplerProbe(),
+        SupabaseProbe(),
+        FlyProbe(),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Snapshot cache + accessor
+# ---------------------------------------------------------------------------
+
+
+_snapshot_cache: dict[str, ServiceHealthSnapshot] = {}
+
+
+def current_service_snapshots() -> dict[str, ServiceHealthSnapshot]:
+    """Return a defensive copy of the snapshot cache.
+
+    Read by ``/api/heartbeat/services`` (KR-FEAT-HEARTBEAT ST2).
+    Mutations on the returned dict don't leak into the cache.
+    """
+    return dict(_snapshot_cache)
+
+
+def _clear_snapshot_cache() -> None:
+    """Test hook + listener-shutdown helper."""
+    global _snapshot_cache
+    _snapshot_cache = {}
+
+
+# ---------------------------------------------------------------------------
+# Runner
+# ---------------------------------------------------------------------------
+
+
+async def run_all_probes(
+    probes: Sequence[ServiceProbe] | None = None,
+) -> dict[str, ServiceHealthSnapshot]:
+    """Run every probe + populate the snapshot cache.
+
+    Each probe runs under a 10s wall-clock timeout (per
+    :func:`with_timeout`). A probe that raises an unhandled
+    exception (anything not caught by its own ``check()``) is
+    caught here + recorded via :func:`snapshot_for_unexpected_error`
+    — guarantees one snapshot per probe per cycle.
+
+    Probes run SERIALLY: per-cycle latency stays bounded
+    (5 × 10s worst case = 50s, well under the 5-min cadence), and
+    keeping probes serial avoids accidental concurrent-token-use
+    pressure on rate-limited upstream APIs.
+
+    Returns the snapshot map for the cycle (also written to the
+    module cache).
+    """
+    if probes is None:
+        probes = default_probes()
+    results: dict[str, ServiceHealthSnapshot] = {}
+    for probe in probes:
+        try:
+            snapshot = await with_timeout(probe.check(), name=probe.name)
+        except Exception as exc:
+            # Defense in depth: probe's check() should never raise
+            # past with_timeout's catch — this guards against
+            # construction-time errors or unexpected exit paths.
+            logger.warning(
+                "[kora.heartbeat_probes] %s.check() raised unexpectedly: %r",
+                probe.name,
+                exc,
+            )
+            snapshot = snapshot_for_unexpected_error(
+                name=probe.name, exc=exc
+            )
+        results[probe.name] = snapshot
+        _snapshot_cache[probe.name] = snapshot
+    return results
+
+
+# ---------------------------------------------------------------------------
+# Cancellable runner for the scheduler
+# ---------------------------------------------------------------------------
+
+
+async def run_all_probes_scheduled() -> None:
+    """Scheduler-callable: zero-arg, returns None.
+
+    Wraps :func:`run_all_probes` so it matches
+    :data:`kora_cli.listeners.heartbeat.PeriodicCallable` type. A
+    scheduler-cancelled cycle still leaves the cache in whatever
+    state it was in (snapshots from completed probes); next cycle
+    overwrites."""
+    try:
+        await run_all_probes()
+    except asyncio.CancelledError:
+        # Scheduler shutdown — propagate so the wrapping task exits
+        raise
+    except Exception as exc:
+        # Unknown error in run_all_probes itself (not per-probe;
+        # those are caught above). Surface so the scheduler logs +
+        # keeps firing on cadence.
+        logger.exception(
+            "[kora.heartbeat_probes] run_all_probes raised: %r", exc
+        )
diff --git a/kora_cli/heartbeat_probes/sentry.py b/kora_cli/heartbeat_probes/sentry.py
new file mode 100644
index 000000000000..ef75f651b3b3
--- /dev/null
+++ b/kora_cli/heartbeat_probes/sentry.py
@@ -0,0 +1,91 @@
+"""SentryProbe — lists unresolved issues for the configured org.
+
+Auth: ``KORA_SENTRY_API_TOKEN`` + ``KORA_SENTRY_ORG`` env vars.
+Healthy: 200 + ≤10 unresolved issues. Degraded: 200 with >10.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+
+from kora_cli.heartbeat_probes.base import (
+    PROBE_TIMEOUT_SECONDS,
+    now_ms_monotonic,
+    resolve_env,
+    sanitize_error,
+    snapshot_for_auth_missing,
+    snapshot_for_unexpected_error,
+    utc_now,
+)
+from kora_cli.heartbeat_probes.types import ServiceHealthSnapshot
+
+
+SENTRY_API_TOKEN_ENV = "KORA_SENTRY_API_TOKEN"
+SENTRY_ORG_ENV = "KORA_SENTRY_ORG"
+SENTRY_API_BASE = "https://sentry.io"
+DEGRADED_UNRESOLVED_THRESHOLD = 10
+
+
+class SentryProbe:
+    name = "sentry"
+
+    async def check(self) -> ServiceHealthSnapshot:
+        token = resolve_env(SENTRY_API_TOKEN_ENV)
+        org = resolve_env(SENTRY_ORG_ENV)
+        if token is None or org is None:
+            return snapshot_for_auth_missing(
+                name=self.name,
+                env_var=SENTRY_API_TOKEN_ENV,
+                extra_envs=(SENTRY_ORG_ENV,),
+            )
+
+        started_ms = now_ms_monotonic()
+        try:
+            async with httpx.AsyncClient(timeout=PROBE_TIMEOUT_SECONDS) as client:
+                response = await client.get(
+                    f"{SENTRY_API_BASE}/api/0/organizations/{org}/issues/",
+                    headers={"Authorization": f"Bearer {token}"},
+                    params={"query": "is:unresolved", "limit": 100},
+                )
+        except Exception as exc:
+            return snapshot_for_unexpected_error(
+                name=self.name, exc=exc, auth_tokens=(token,)
+            )
+
+        latency_ms = int(now_ms_monotonic() - started_ms)
+        if response.status_code != 200:
+            return ServiceHealthSnapshot(
+                name=self.name,
+                status="unhealthy",
+                latency_ms=latency_ms,
+                last_check_at=utc_now(),
+                details={},
+                error=sanitize_error(
+                    f"HTTP {response.status_code}", token
+                ),
+            )
+
+        unresolved = _count_unresolved(response.json())
+        status = (
+            "degraded"
+            if unresolved > DEGRADED_UNRESOLVED_THRESHOLD
+            else "healthy"
+        )
+        return ServiceHealthSnapshot(
+            name=self.name,
+            status=status,
+            latency_ms=latency_ms,
+            last_check_at=utc_now(),
+            details={"unresolved_issues": unresolved},
+            error=None,
+        )
+
+
+def _count_unresolved(payload: Any) -> int:
+    """Count entries in the response list. Defensive against
+    shape drift — returns 0 if payload isn't a list."""
+    if not isinstance(payload, list):
+        return 0
+    return len(payload)
diff --git a/kora_cli/heartbeat_probes/supabase.py b/kora_cli/heartbeat_probes/supabase.py
new file mode 100644
index 000000000000..da2b57137e30
--- /dev/null
+++ b/kora_cli/heartbeat_probes/supabase.py
@@ -0,0 +1,89 @@
+"""SupabaseProbe — HEADs the PostgREST endpoint of IsoKron's DB.
+
+Auth: ``KORA_SUPABASE_ANON_KEY`` + ``KORA_SUPABASE_URL`` env vars.
+The anon key is the right surface for a heartbeat — it's expected
+in client-side SDKs + carries no privileged access.
+
+Healthy: HEAD returns 200/204. Degraded: connections_pct > 80
+(when surface available; "unknown" otherwise).
+"""
+
+from __future__ import annotations
+
+import httpx
+
+from kora_cli.heartbeat_probes.base import (
+    PROBE_TIMEOUT_SECONDS,
+    now_ms_monotonic,
+    resolve_env,
+    sanitize_error,
+    snapshot_for_auth_missing,
+    snapshot_for_unexpected_error,
+    utc_now,
+)
+from kora_cli.heartbeat_probes.types import ServiceHealthSnapshot
+
+
+SUPABASE_ANON_KEY_ENV = "KORA_SUPABASE_ANON_KEY"
+SUPABASE_URL_ENV = "KORA_SUPABASE_URL"
+DEGRADED_CONNECTIONS_PCT_THRESHOLD = 80.0
+
+
+class SupabaseProbe:
+    name = "supabase"
+
+    async def check(self) -> ServiceHealthSnapshot:
+        anon_key = resolve_env(SUPABASE_ANON_KEY_ENV)
+        url = resolve_env(SUPABASE_URL_ENV)
+        if anon_key is None or url is None:
+            return snapshot_for_auth_missing(
+                name=self.name,
+                env_var=SUPABASE_ANON_KEY_ENV,
+                extra_envs=(SUPABASE_URL_ENV,),
+            )
+
+        # PostgREST endpoint — the canonical reachability check
+        # against Supabase. HEAD on the root returns 200 when the
+        # API is up. Strip trailing slash to avoid double-slash.
+        rest_url = f"{url.rstrip('/')}/rest/v1/"
+
+        started_ms = now_ms_monotonic()
+        try:
+            async with httpx.AsyncClient(timeout=PROBE_TIMEOUT_SECONDS) as client:
+                response = await client.head(
+                    rest_url,
+                    headers={
+                        "apikey": anon_key,
+                        "Authorization": f"Bearer {anon_key}",
+                    },
+                )
+        except Exception as exc:
+            return snapshot_for_unexpected_error(
+                name=self.name, exc=exc, auth_tokens=(anon_key,)
+            )
+
+        latency_ms = int(now_ms_monotonic() - started_ms)
+        if response.status_code not in (200, 204):
+            return ServiceHealthSnapshot(
+                name=self.name,
+                status="unhealthy",
+                latency_ms=latency_ms,
+                last_check_at=utc_now(),
+                details={},
+                error=sanitize_error(
+                    f"HTTP {response.status_code}", anon_key
+                ),
+            )
+
+        # connections_pct is "unknown" — pulling it requires the
+        # Supabase Management API + a project ref, which is a
+        # different auth surface. Surfaced as a known-unknown so
+        # operator sees the gap; doesn't fail the probe.
+        return ServiceHealthSnapshot(
+            name=self.name,
+            status="healthy",
+            latency_ms=latency_ms,
+            last_check_at=utc_now(),
+            details={"connections_pct": "unknown"},
+            error=None,
+        )
diff --git a/kora_cli/heartbeat_probes/types.py b/kora_cli/heartbeat_probes/types.py
new file mode 100644
index 000000000000..6af57fcbabc8
--- /dev/null
+++ b/kora_cli/heartbeat_probes/types.py
@@ -0,0 +1,53 @@
+"""Pydantic types for KR-FEAT-HEARTBEAT.
+
+Internal probe shape — richer than the
+``HeartbeatService`` TS interface from KR-HB-PANEL (PR #103):
+
+  - Status enum extends with ``"unknown"`` (covers the "can't probe"
+    case: missing auth env, transport failure before the upstream
+    service answers, probe-loop crash).
+  - ``error: str | None`` carries the operator-readable failure
+    string (sanitized — never contains the auth token).
+  - ``latency_ms`` is nullable because an ``unknown`` probe never
+    completed a roundtrip.
+
+The KR-FEAT-HEARTBEAT ST2 endpoint projects this internal shape
+onto the TS contract — ``unknown`` status + ``error`` field are
+additive FE extensions (matches the KR-MCP-CONSUMPTION ST2
+additive-FE pattern shipped in PR #113).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Literal, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+ServiceStatus = Literal["healthy", "degraded", "unhealthy", "unknown"]
+
+SERVICE_STATUSES: tuple[ServiceStatus, ...] = (
+    "healthy",
+    "degraded",
+    "unhealthy",
+    "unknown",
+)
+
+
+class ServiceHealthSnapshot(BaseModel):
+    """One per-service health observation.
+
+    Per K-DG drift discipline: ``extra="forbid"`` rejects unknown
+    keys at construction so probe authors can't silently widen the
+    shape without coordinating with the FE.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    name: str = Field(min_length=1)
+    status: ServiceStatus
+    latency_ms: Optional[int] = None
+    last_check_at: datetime
+    details: dict[str, Any] = Field(default_factory=dict)
+    error: Optional[str] = None
diff --git a/kora_cli/heartbeat_probes/vercel.py b/kora_cli/heartbeat_probes/vercel.py
new file mode 100644
index 000000000000..b2b4d19ce67e
--- /dev/null
+++ b/kora_cli/heartbeat_probes/vercel.py
@@ -0,0 +1,129 @@
+"""VercelProbe — Phase 2 Feature 2 backend service probe.
+
+Lists recent deployments via the Vercel REST API; healthy if the
+endpoint returns 200 with at least one deployment. Degraded when
+the deploy error rate exceeds 10% over the last 24 hours.
+
+Auth: ``KORA_VERCEL_API_TOKEN`` Doppler-injected env var.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import httpx
+
+from kora_cli.heartbeat_probes.base import (
+    PROBE_TIMEOUT_SECONDS,
+    now_ms_monotonic,
+    resolve_env,
+    sanitize_error,
+    snapshot_for_auth_missing,
+    snapshot_for_unexpected_error,
+    utc_now,
+)
+from kora_cli.heartbeat_probes.types import ServiceHealthSnapshot
+
+
+VERCEL_API_TOKEN_ENV = "KORA_VERCEL_API_TOKEN"
+VERCEL_API_BASE = "https://api.vercel.com"
+DEGRADED_ERROR_RATE_THRESHOLD = 0.10
+
+
+class VercelProbe:
+    name = "vercel"
+
+    async def check(self) -> ServiceHealthSnapshot:
+        token = resolve_env(VERCEL_API_TOKEN_ENV)
+        if token is None:
+            return snapshot_for_auth_missing(
+                name=self.name, env_var=VERCEL_API_TOKEN_ENV
+            )
+
+        started_ms = now_ms_monotonic()
+        try:
+            async with httpx.AsyncClient(timeout=PROBE_TIMEOUT_SECONDS) as client:
+                response = await client.get(
+                    f"{VERCEL_API_BASE}/v6/deployments",
+                    headers={"Authorization": f"Bearer {token}"},
+                    params={"limit": 100},
+                )
+        except httpx.HTTPError as exc:
+            return snapshot_for_unexpected_error(
+                name=self.name, exc=exc, auth_tokens=(token,)
+            )
+        except Exception as exc:
+            return snapshot_for_unexpected_error(
+                name=self.name, exc=exc, auth_tokens=(token,)
+            )
+
+        latency_ms = int(now_ms_monotonic() - started_ms)
+        if response.status_code != 200:
+            return ServiceHealthSnapshot(
+                name=self.name,
+                status="unhealthy",
+                latency_ms=latency_ms,
+                last_check_at=utc_now(),
+                details={},
+                error=sanitize_error(
+                    f"HTTP {response.status_code}", token
+                ),
+            )
+
+        deployments_24h, error_rate = _summarize_deployments(response.json())
+        status: str = "healthy"
+        if deployments_24h == 0:
+            # No recent activity isn't an error — surface as healthy
+            # with 0 count. The dashboard reads details to decide if
+            # the "no activity" reading is noteworthy.
+            status = "healthy"
+        elif error_rate > DEGRADED_ERROR_RATE_THRESHOLD:
+            status = "degraded"
+
+        return ServiceHealthSnapshot(
+            name=self.name,
+            status=status,
+            latency_ms=latency_ms,
+            last_check_at=utc_now(),
+            details={
+                "deployments_last_24h": deployments_24h,
+                "error_rate_24h": round(error_rate, 4),
+            },
+            error=None,
+        )
+
+
+def _summarize_deployments(payload: Any) -> tuple[int, float]:
+    """Project Vercel's deployment list to (count_24h, error_rate).
+
+    Defensive against shape drift — unexpected payload returns
+    (0, 0.0) rather than raising.
+    """
+    if not isinstance(payload, dict):
+        return (0, 0.0)
+    deployments = payload.get("deployments")
+    if not isinstance(deployments, list):
+        return (0, 0.0)
+
+    cutoff_ms = int(
+        (datetime.now(timezone.utc) - timedelta(hours=24)).timestamp() * 1000
+    )
+    recent_total = 0
+    recent_errors = 0
+    for entry in deployments:
+        if not isinstance(entry, dict):
+            continue
+        created_at = entry.get("created")
+        if not isinstance(created_at, (int, float)):
+            continue
+        if created_at < cutoff_ms:
+            continue
+        recent_total += 1
+        state = entry.get("state") or entry.get("readyState")
+        if isinstance(state, str) and state.upper() in {"ERROR", "FAILED"}:
+            recent_errors += 1
+
+    if recent_total == 0:
+        return (0, 0.0)
+    return (recent_total, recent_errors / recent_total)
diff --git a/kora_cli/listeners/__init__.py b/kora_cli/listeners/__init__.py
index 20cef3797c21..f2051b89c35e 100644
--- a/kora_cli/listeners/__init__.py
+++ b/kora_cli/listeners/__init__.py
@@ -24,3 +24,8 @@
 # startup (no transport opens) means listener insertion here is
 # fast + can't fail on remote-MCP availability.
 from kora_cli.listeners import mcp_consumption  # noqa: F401
+# KR-FEAT-HEARTBEAT ST1 — service-probe listener. Registers
+# a heartbeat-scheduler task at module-import time. Probe-instance
+# construction happens per cycle (stateless across cycles), so
+# startup is a clean no-op + LOG line.
+from kora_cli.listeners import heartbeat_probes_listener  # noqa: F401
diff --git a/kora_cli/listeners/heartbeat_probes_listener.py b/kora_cli/listeners/heartbeat_probes_listener.py
new file mode 100644
index 000000000000..60a9a6c88a77
--- /dev/null
+++ b/kora_cli/listeners/heartbeat_probes_listener.py
@@ -0,0 +1,127 @@
+"""Heartbeat-probes daemon listener (KR-FEAT-HEARTBEAT ST1).
+
+Wires the 5 service probes into the daemon lifecycle:
+
+  - Daemon boot registers (via the daemon-listener side) — there's
+    no per-listener state to construct (probes are created per
+    cycle by the runner); startup is a clean no-op + LOG line.
+  - The heartbeat scheduler runs :func:`run_all_probes_scheduled`
+    every :func:`_read_probe_interval` seconds (default 300s,
+    operator override via ``KORA_HEARTBEAT_PROBE_INTERVAL_SEC``).
+  - Daemon shutdown clears the snapshot cache so a stale
+    pre-restart snapshot doesn't bleed into the post-restart
+    panel view.
+
+# Two distinct heartbeat tasks
+
+KR-MCP-CONSUMPTION ST2 registered ``mcp.health_check`` at 5min;
+this registers ``heartbeat.service_probes`` at 5min. Per §4 Q1
+ruling: keep them separate so one slow probe doesn't backpressure
+the other. Both share the cadence default but each can be tuned
+via its own env var.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+
+from kora_cli.daemon import DEFAULT_SHUTDOWN_TIMEOUT, register_daemon_listener
+from kora_cli.heartbeat_probes.runner import (
+    _clear_snapshot_cache,
+    run_all_probes_scheduled,
+)
+from kora_cli.listeners.heartbeat import register_periodic_task
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Cadence configuration
+# ---------------------------------------------------------------------------
+
+
+DEFAULT_PROBE_INTERVAL_SEC: float = 300.0  # 5 min per §4 Q1 ruling
+PROBE_INTERVAL_ENV: str = "KORA_HEARTBEAT_PROBE_INTERVAL_SEC"
+
+
+def _read_probe_interval() -> float:
+    """Read the probe-cycle interval from env or fall back to default.
+
+    Invalid (non-numeric, ≤0) values WARN-log + return the default.
+    Matches the KR-MCP-CONSUMPTION ST2 env-validation pattern.
+    """
+    raw = os.environ.get(PROBE_INTERVAL_ENV, "").strip()
+    if not raw:
+        return DEFAULT_PROBE_INTERVAL_SEC
+    try:
+        value = float(raw)
+    except ValueError:
+        logger.warning(
+            "[kora.heartbeat_probes] %s=%r not numeric; using default %ss",
+            PROBE_INTERVAL_ENV,
+            raw,
+            DEFAULT_PROBE_INTERVAL_SEC,
+        )
+        return DEFAULT_PROBE_INTERVAL_SEC
+    if value <= 0:
+        logger.warning(
+            "[kora.heartbeat_probes] %s=%s must be > 0; using default %ss",
+            PROBE_INTERVAL_ENV,
+            value,
+            DEFAULT_PROBE_INTERVAL_SEC,
+        )
+        return DEFAULT_PROBE_INTERVAL_SEC
+    return value
+
+
+# ---------------------------------------------------------------------------
+# Listener
+# ---------------------------------------------------------------------------
+
+
+class HeartbeatProbesListener:
+    """Lifecycle anchor for the probe scheduler.
+
+    Probes themselves are constructed per cycle by the runner —
+    listener has no per-instance state. Startup is a log line
+    confirming the listener is registered; shutdown clears the
+    snapshot cache so a stale snapshot doesn't survive daemon
+    restart.
+    """
+
+    async def startup(self) -> None:
+        interval = _read_probe_interval()
+        logger.info(
+            "[kora.heartbeat_probes] listener active; probe cycle every %ss "
+            "(KORA_HEARTBEAT_PROBE_INTERVAL_SEC override)",
+            interval,
+        )
+
+    async def shutdown(self) -> None:
+        _clear_snapshot_cache()
+        logger.info("[kora.heartbeat_probes] snapshot cache cleared")
+
+
+# ---------------------------------------------------------------------------
+# Factory + registration
+# ---------------------------------------------------------------------------
+
+
+def _factory():
+    listener = HeartbeatProbesListener()
+    return (listener.startup, listener.shutdown, DEFAULT_SHUTDOWN_TIMEOUT)
+
+
+register_daemon_listener("heartbeat_probes", _factory)
+
+
+# ---------------------------------------------------------------------------
+# Periodic task registration (import-time side effect)
+# ---------------------------------------------------------------------------
+
+register_periodic_task(
+    "heartbeat.service_probes",
+    interval_seconds=_read_probe_interval(),
+    callable=run_all_probes_scheduled,
+)
diff --git a/tests/kora_cli/test_heartbeat_probes/__init__.py b/tests/kora_cli/test_heartbeat_probes/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tests/kora_cli/test_heartbeat_probes/test_base.py b/tests/kora_cli/test_heartbeat_probes/test_base.py
new file mode 100644
index 000000000000..6bf77a6578e9
--- /dev/null
+++ b/tests/kora_cli/test_heartbeat_probes/test_base.py
@@ -0,0 +1,235 @@
+"""Base / helper tests (KR-FEAT-HEARTBEAT ST1)."""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timezone
+
+import pytest
+
+from kora_cli.heartbeat_probes.base import (
+    PROBE_TIMEOUT_SECONDS,
+    resolve_env,
+    sanitize_error,
+    snapshot_for_auth_missing,
+    snapshot_for_timeout,
+    snapshot_for_unexpected_error,
+    with_timeout,
+)
+from kora_cli.heartbeat_probes.types import (
+    SERVICE_STATUSES,
+    ServiceHealthSnapshot,
+)
+
+
+# ---------------------------------------------------------------------------
+# Status enum + snapshot Pydantic shape
+# ---------------------------------------------------------------------------
+
+
+def test_service_statuses_includes_unknown():
+    """KR-FEAT-HEARTBEAT extends the TS contract's 3-value enum with
+    "unknown" (covers auth-missing + timeout + probe-loop crash)."""
+    assert set(SERVICE_STATUSES) == {"healthy", "degraded", "unhealthy", "unknown"}
+
+
+def test_snapshot_rejects_unknown_keys():
+    """K-DG drift discipline — Pydantic ``extra="forbid"`` catches
+    probe authors silently widening the shape without coordinating
+    with the FE."""
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError):
+        ServiceHealthSnapshot(
+            name="x",
+            status="healthy",
+            last_check_at=datetime.now(timezone.utc),
+            wat_typo=True,  # type: ignore[call-arg]
+        )
+
+
+def test_snapshot_status_literal_enforced():
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError):
+        ServiceHealthSnapshot(
+            name="x",
+            status="not-a-status",  # type: ignore[arg-type]
+            last_check_at=datetime.now(timezone.utc),
+        )
+
+
+# ---------------------------------------------------------------------------
+# resolve_env — empty/whitespace treated as unset (Doppler corner case)
+# ---------------------------------------------------------------------------
+
+
+def test_resolve_env_returns_none_when_unset(monkeypatch):
+    monkeypatch.delenv("TEST_PROBE_VAR", raising=False)
+    assert resolve_env("TEST_PROBE_VAR") is None
+
+
+def test_resolve_env_returns_none_when_empty_string(monkeypatch):
+    monkeypatch.setenv("TEST_PROBE_VAR", "")
+    assert resolve_env("TEST_PROBE_VAR") is None
+
+
+def test_resolve_env_returns_none_when_whitespace_only(monkeypatch):
+    monkeypatch.setenv("TEST_PROBE_VAR", "   \n  ")
+    assert resolve_env("TEST_PROBE_VAR") is None
+
+
+def test_resolve_env_strips_and_returns_value(monkeypatch):
+    monkeypatch.setenv("TEST_PROBE_VAR", "  ghp_real  \n")
+    assert resolve_env("TEST_PROBE_VAR") == "ghp_real"
+
+
+# ---------------------------------------------------------------------------
+# sanitize_error — token redaction discipline
+# ---------------------------------------------------------------------------
+
+
+def test_sanitize_error_redacts_single_token():
+    text = "auth failed: Bearer ghp_realvalue123 rejected"
+    out = sanitize_error(text, "ghp_realvalue123")
+    assert "ghp_realvalue123" not in out
+    assert "<REDACTED>" in out
+
+
+def test_sanitize_error_redacts_multiple_tokens():
+    text = "auth1=token_a auth2=token_b"
+    out = sanitize_error(text, "token_a", "token_b")
+    assert "token_a" not in out
+    assert "token_b" not in out
+
+
+def test_sanitize_error_skips_none_and_empty_tokens():
+    text = "harmless"
+    out = sanitize_error(text, None, "", "  ")
+    # Empty tokens skipped; whitespace token would replace whitespace
+    # which is a bug — so only literally-empty + None are skipped.
+    # Whitespace token "  " would replace " " in text but our text
+    # has no whitespace to replace, so output equals input.
+    assert "harmless" in out
+
+
+def test_sanitize_error_preserves_unrelated_text():
+    text = "endpoint unreachable: HTTP 503"
+    out = sanitize_error(text, "secret_token")
+    assert out == text
+
+
+def test_sanitize_error_handles_empty_text():
+    assert sanitize_error("", "secret_token") == ""
+
+
+# ---------------------------------------------------------------------------
+# snapshot_for_auth_missing
+# ---------------------------------------------------------------------------
+
+
+def test_snapshot_for_auth_missing_carries_env_var_names_not_values():
+    """SECURITY: the error string lists ENV VAR NAMES never values.
+    Even on auth-missing path, no token-shaped string appears."""
+    snap = snapshot_for_auth_missing(name="vercel", env_var="KORA_VERCEL_API_TOKEN")
+    assert snap.status == "unknown"
+    assert snap.latency_ms is None
+    assert "KORA_VERCEL_API_TOKEN" in snap.error
+    # Sanity: no token-value-shaped substring
+    assert "Bearer" not in snap.error
+    assert "ghp_" not in snap.error
+
+
+def test_snapshot_for_auth_missing_with_extra_envs():
+    snap = snapshot_for_auth_missing(
+        name="supabase",
+        env_var="KORA_SUPABASE_ANON_KEY",
+        extra_envs=("KORA_SUPABASE_URL",),
+    )
+    assert "KORA_SUPABASE_ANON_KEY" in snap.error
+    assert "KORA_SUPABASE_URL" in snap.error
+
+
+# ---------------------------------------------------------------------------
+# snapshot_for_timeout + snapshot_for_unexpected_error
+# ---------------------------------------------------------------------------
+
+
+def test_snapshot_for_timeout_shape():
+    snap = snapshot_for_timeout(name="vercel")
+    assert snap.status == "unknown"
+    assert snap.latency_ms is None
+    assert "timed out" in snap.error.lower()
+
+
+def test_snapshot_for_unexpected_error_redacts_token():
+    """SECURITY: when an exception message happens to embed the
+    auth token (e.g. httpx error with URL containing the token in
+    a query string), the sanitize step strips it."""
+    snap = snapshot_for_unexpected_error(
+        name="fly",
+        exc=RuntimeError("upstream rejected ghp_realtoken123 with 401"),
+        auth_tokens=("ghp_realtoken123",),
+    )
+    assert snap.status == "unknown"
+    assert "ghp_realtoken123" not in snap.error
+    assert "<REDACTED>" in snap.error
+    # The exception type prefix is preserved
+    assert "RuntimeError" in snap.error
+
+
+def test_snapshot_for_unexpected_error_handles_no_tokens():
+    """When no tokens are passed (e.g. error is from probe internals
+    that never touched auth), no redaction needed."""
+    snap = snapshot_for_unexpected_error(
+        name="fly",
+        exc=ValueError("invalid response shape"),
+    )
+    assert snap.status == "unknown"
+    assert "ValueError" in snap.error
+    assert "invalid response shape" in snap.error
+
+
+# ---------------------------------------------------------------------------
+# with_timeout — 10s ceiling
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_with_timeout_returns_snapshot_for_timeout_on_overshoot(monkeypatch):
+    """A coroutine that takes longer than the timeout → returns a
+    snapshot_for_timeout instead of raising. The runner relies on
+    this contract."""
+
+    async def _slow():
+        await asyncio.sleep(0.5)
+        return ServiceHealthSnapshot(
+            name="x",
+            status="healthy",
+            last_check_at=datetime.now(timezone.utc),
+        )
+
+    snap = await with_timeout(_slow(), name="test", timeout=0.05)
+    assert snap.status == "unknown"
+    assert "timed out" in snap.error.lower()
+
+
+@pytest.mark.asyncio
+async def test_with_timeout_returns_snapshot_when_check_completes():
+    """Fast check returns its snapshot unchanged."""
+    expected = ServiceHealthSnapshot(
+        name="x",
+        status="healthy",
+        last_check_at=datetime.now(timezone.utc),
+    )
+
+    async def _fast():
+        return expected
+
+    snap = await with_timeout(_fast(), name="test", timeout=1.0)
+    assert snap is expected
+
+
+def test_probe_timeout_seconds_is_10():
+    """Spec pin: §4 Q1 implies 10s per-probe ceiling vs 5min cadence."""
+    assert PROBE_TIMEOUT_SECONDS == 10.0
diff --git a/tests/kora_cli/test_heartbeat_probes/test_probes.py b/tests/kora_cli/test_heartbeat_probes/test_probes.py
new file mode 100644
index 000000000000..d55b2c603b2d
--- /dev/null
+++ b/tests/kora_cli/test_heartbeat_probes/test_probes.py
@@ -0,0 +1,421 @@
+"""Per-probe tests (KR-FEAT-HEARTBEAT ST1).
+
+One section per probe. Each section covers:
+  - Missing auth env → status=unknown + ZERO httpx call
+  - 200 response with healthy shape → status=healthy + details
+  - 200 with degraded threshold tripped → status=degraded
+  - Non-200 response → status=unhealthy + sanitized error
+  - Transport error → status=unknown + sanitized error
+
+httpx is mocked at the AsyncClient class level so probes never
+hit the real network. The mock records call counts so we can
+verify the auth-missing path makes ZERO outbound calls.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import httpx
+import pytest
+
+from kora_cli.heartbeat_probes.doppler import (
+    DOPPLER_API_TOKEN_ENV,
+    DopplerProbe,
+)
+from kora_cli.heartbeat_probes.fly import (
+    FLY_API_TOKEN_ENV,
+    FlyProbe,
+)
+from kora_cli.heartbeat_probes.sentry import (
+    SENTRY_API_TOKEN_ENV,
+    SENTRY_ORG_ENV,
+    SentryProbe,
+)
+from kora_cli.heartbeat_probes.supabase import (
+    SUPABASE_ANON_KEY_ENV,
+    SUPABASE_URL_ENV,
+    SupabaseProbe,
+)
+from kora_cli.heartbeat_probes.vercel import (
+    VERCEL_API_TOKEN_ENV,
+    VercelProbe,
+)
+
+
+def _fake_response(*, status_code: int = 200, json_payload: Any = None) -> MagicMock:
+    """Build a MagicMock that quacks like httpx.Response."""
+    resp = MagicMock()
+    resp.status_code = status_code
+    resp.json = MagicMock(return_value=json_payload if json_payload is not None else {})
+    return resp
+
+
+def _patch_http_client(get_response=None, head_response=None):
+    """Patch httpx.AsyncClient so we control responses + can count
+    calls. Returns the AsyncMock instance so tests can introspect."""
+    fake_client = AsyncMock(spec=httpx.AsyncClient)
+    if get_response is not None:
+        fake_client.get = AsyncMock(return_value=get_response)
+    if head_response is not None:
+        fake_client.head = AsyncMock(return_value=head_response)
+    # AsyncClient is used as async context manager
+    fake_cm = MagicMock()
+    fake_cm.__aenter__ = AsyncMock(return_value=fake_client)
+    fake_cm.__aexit__ = AsyncMock(return_value=None)
+    return patch("httpx.AsyncClient", return_value=fake_cm), fake_client
+
+
+# =============================================================================
+# VercelProbe
+# =============================================================================
+
+
+@pytest.mark.asyncio
+async def test_vercel_missing_token_returns_unknown_zero_calls(monkeypatch):
+    monkeypatch.delenv(VERCEL_API_TOKEN_ENV, raising=False)
+    cm_patch, fake_client = _patch_http_client(get_response=_fake_response())
+    with cm_patch:
+        snap = await VercelProbe().check()
+    assert snap.status == "unknown"
+    assert snap.latency_ms is None
+    assert VERCEL_API_TOKEN_ENV in snap.error
+    # SECURITY: zero outbound calls when auth missing
+    fake_client.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_vercel_200_healthy(monkeypatch):
+    monkeypatch.setenv(VERCEL_API_TOKEN_ENV, "vercel_pat_test")
+    payload = {
+        "deployments": [
+            {"created": _ms_now(), "state": "READY"},
+            {"created": _ms_now(), "state": "READY"},
+        ]
+    }
+    cm_patch, fake_client = _patch_http_client(
+        get_response=_fake_response(status_code=200, json_payload=payload)
+    )
+    with cm_patch:
+        snap = await VercelProbe().check()
+    assert snap.status == "healthy"
+    assert snap.latency_ms is not None
+    assert snap.details["deployments_last_24h"] == 2
+    assert snap.details["error_rate_24h"] == 0.0
+
+
+@pytest.mark.asyncio
+async def test_vercel_200_degraded_when_error_rate_high(monkeypatch):
+    monkeypatch.setenv(VERCEL_API_TOKEN_ENV, "vercel_pat_test")
+    # 8 deploys, 2 errors → 25% > 10% threshold
+    payload = {
+        "deployments": [{"created": _ms_now(), "state": "READY"}] * 6
+        + [{"created": _ms_now(), "state": "ERROR"}] * 2
+    }
+    cm_patch, _ = _patch_http_client(
+        get_response=_fake_response(status_code=200, json_payload=payload)
+    )
+    with cm_patch:
+        snap = await VercelProbe().check()
+    assert snap.status == "degraded"
+    assert snap.details["error_rate_24h"] == 0.25
+
+
+@pytest.mark.asyncio
+async def test_vercel_non_200_unhealthy_with_sanitized_error(monkeypatch):
+    monkeypatch.setenv(VERCEL_API_TOKEN_ENV, "vercel_pat_test")
+    cm_patch, _ = _patch_http_client(get_response=_fake_response(status_code=503))
+    with cm_patch:
+        snap = await VercelProbe().check()
+    assert snap.status == "unhealthy"
+    assert "HTTP 503" in snap.error
+    assert "vercel_pat_test" not in snap.error
+
+
+@pytest.mark.asyncio
+async def test_vercel_transport_error_with_token_in_message_redacted(monkeypatch):
+    """SECURITY: if httpx error message contains the token (URL
+    fragment, etc.), the snapshot's error field must redact it."""
+    monkeypatch.setenv(VERCEL_API_TOKEN_ENV, "vercel_pat_test")
+    fake_client = AsyncMock()
+    fake_client.get = AsyncMock(
+        side_effect=httpx.ConnectError("upstream rejected token vercel_pat_test"),
+    )
+    fake_cm = MagicMock()
+    fake_cm.__aenter__ = AsyncMock(return_value=fake_client)
+    fake_cm.__aexit__ = AsyncMock(return_value=None)
+    with patch("httpx.AsyncClient", return_value=fake_cm):
+        snap = await VercelProbe().check()
+    assert snap.status == "unknown"
+    assert "vercel_pat_test" not in snap.error
+
+
+# =============================================================================
+# SentryProbe
+# =============================================================================
+
+
+@pytest.mark.asyncio
+async def test_sentry_missing_token_returns_unknown_zero_calls(monkeypatch):
+    monkeypatch.delenv(SENTRY_API_TOKEN_ENV, raising=False)
+    monkeypatch.setenv(SENTRY_ORG_ENV, "stormhaven")
+    cm_patch, fake_client = _patch_http_client(get_response=_fake_response())
+    with cm_patch:
+        snap = await SentryProbe().check()
+    assert snap.status == "unknown"
+    assert SENTRY_API_TOKEN_ENV in snap.error
+    fake_client.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_sentry_missing_org_returns_unknown(monkeypatch):
+    monkeypatch.setenv(SENTRY_API_TOKEN_ENV, "sentry_token_test")
+    monkeypatch.delenv(SENTRY_ORG_ENV, raising=False)
+    cm_patch, fake_client = _patch_http_client(get_response=_fake_response())
+    with cm_patch:
+        snap = await SentryProbe().check()
+    assert snap.status == "unknown"
+    assert SENTRY_ORG_ENV in snap.error
+    fake_client.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_sentry_200_healthy_below_threshold(monkeypatch):
+    monkeypatch.setenv(SENTRY_API_TOKEN_ENV, "sentry_token_test")
+    monkeypatch.setenv(SENTRY_ORG_ENV, "stormhaven")
+    cm_patch, _ = _patch_http_client(
+        get_response=_fake_response(status_code=200, json_payload=[{}, {}, {}])
+    )
+    with cm_patch:
+        snap = await SentryProbe().check()
+    assert snap.status == "healthy"
+    assert snap.details["unresolved_issues"] == 3
+
+
+@pytest.mark.asyncio
+async def test_sentry_degraded_above_threshold(monkeypatch):
+    monkeypatch.setenv(SENTRY_API_TOKEN_ENV, "sentry_token_test")
+    monkeypatch.setenv(SENTRY_ORG_ENV, "stormhaven")
+    cm_patch, _ = _patch_http_client(
+        get_response=_fake_response(status_code=200, json_payload=[{}] * 15)
+    )
+    with cm_patch:
+        snap = await SentryProbe().check()
+    assert snap.status == "degraded"
+    assert snap.details["unresolved_issues"] == 15
+
+
+@pytest.mark.asyncio
+async def test_sentry_non_200_unhealthy(monkeypatch):
+    monkeypatch.setenv(SENTRY_API_TOKEN_ENV, "sentry_token_test")
+    monkeypatch.setenv(SENTRY_ORG_ENV, "stormhaven")
+    cm_patch, _ = _patch_http_client(get_response=_fake_response(status_code=429))
+    with cm_patch:
+        snap = await SentryProbe().check()
+    assert snap.status == "unhealthy"
+    assert "HTTP 429" in snap.error
+    assert "sentry_token_test" not in snap.error
+
+
+# =============================================================================
+# DopplerProbe
+# =============================================================================
+
+
+@pytest.mark.asyncio
+async def test_doppler_missing_token_returns_unknown_zero_calls(monkeypatch):
+    monkeypatch.delenv(DOPPLER_API_TOKEN_ENV, raising=False)
+    cm_patch, fake_client = _patch_http_client(get_response=_fake_response())
+    with cm_patch:
+        snap = await DopplerProbe().check()
+    assert snap.status == "unknown"
+    fake_client.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_doppler_200_healthy(monkeypatch):
+    monkeypatch.setenv(DOPPLER_API_TOKEN_ENV, "doppler_token_test")
+    cm_patch, _ = _patch_http_client(
+        get_response=_fake_response(
+            status_code=200, json_payload={"workplace": {"name": "stormhaven"}}
+        )
+    )
+    with cm_patch:
+        snap = await DopplerProbe().check()
+    assert snap.status == "healthy"
+    # Workplace endpoint doesn't expose project list — known-unknown
+    assert snap.details["projects_total"] == "unknown"
+    assert snap.details["oldest_secret_age_days"] == "unknown"
+
+
+@pytest.mark.asyncio
+async def test_doppler_non_200_unhealthy(monkeypatch):
+    monkeypatch.setenv(DOPPLER_API_TOKEN_ENV, "doppler_token_test")
+    cm_patch, _ = _patch_http_client(get_response=_fake_response(status_code=401))
+    with cm_patch:
+        snap = await DopplerProbe().check()
+    assert snap.status == "unhealthy"
+    assert "HTTP 401" in snap.error
+    assert "doppler_token_test" not in snap.error
+
+
+# =============================================================================
+# SupabaseProbe
+# =============================================================================
+
+
+@pytest.mark.asyncio
+async def test_supabase_missing_envs_returns_unknown_zero_calls(monkeypatch):
+    monkeypatch.delenv(SUPABASE_ANON_KEY_ENV, raising=False)
+    monkeypatch.delenv(SUPABASE_URL_ENV, raising=False)
+    cm_patch, fake_client = _patch_http_client(head_response=_fake_response())
+    with cm_patch:
+        snap = await SupabaseProbe().check()
+    assert snap.status == "unknown"
+    assert SUPABASE_ANON_KEY_ENV in snap.error
+    assert SUPABASE_URL_ENV in snap.error
+    fake_client.head.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_supabase_200_healthy(monkeypatch):
+    monkeypatch.setenv(SUPABASE_ANON_KEY_ENV, "supabase_anon_test")
+    monkeypatch.setenv(SUPABASE_URL_ENV, "https://abc123.supabase.co")
+    cm_patch, _ = _patch_http_client(head_response=_fake_response(status_code=200))
+    with cm_patch:
+        snap = await SupabaseProbe().check()
+    assert snap.status == "healthy"
+    assert snap.details["connections_pct"] == "unknown"
+
+
+@pytest.mark.asyncio
+async def test_supabase_204_also_healthy(monkeypatch):
+    monkeypatch.setenv(SUPABASE_ANON_KEY_ENV, "supabase_anon_test")
+    monkeypatch.setenv(SUPABASE_URL_ENV, "https://abc123.supabase.co")
+    cm_patch, _ = _patch_http_client(head_response=_fake_response(status_code=204))
+    with cm_patch:
+        snap = await SupabaseProbe().check()
+    assert snap.status == "healthy"
+
+
+@pytest.mark.asyncio
+async def test_supabase_500_unhealthy_with_redacted_error(monkeypatch):
+    monkeypatch.setenv(SUPABASE_ANON_KEY_ENV, "supabase_anon_test")
+    monkeypatch.setenv(SUPABASE_URL_ENV, "https://abc123.supabase.co")
+    cm_patch, _ = _patch_http_client(head_response=_fake_response(status_code=500))
+    with cm_patch:
+        snap = await SupabaseProbe().check()
+    assert snap.status == "unhealthy"
+    assert "HTTP 500" in snap.error
+    assert "supabase_anon_test" not in snap.error
+
+
+# =============================================================================
+# FlyProbe
+# =============================================================================
+
+
+@pytest.mark.asyncio
+async def test_fly_missing_token_returns_unknown_zero_calls(monkeypatch):
+    monkeypatch.delenv(FLY_API_TOKEN_ENV, raising=False)
+    cm_patch, fake_client = _patch_http_client(get_response=_fake_response())
+    with cm_patch:
+        snap = await FlyProbe().check()
+    assert snap.status == "unknown"
+    fake_client.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_fly_one_app_healthy(monkeypatch):
+    monkeypatch.setenv(FLY_API_TOKEN_ENV, "fly_token_test")
+    cm_patch, _ = _patch_http_client(
+        get_response=_fake_response(
+            status_code=200,
+            json_payload=[{"state": "started"}, {"state": "started"}],
+        )
+    )
+    with cm_patch:
+        snap = await FlyProbe().check()
+    assert snap.status == "healthy"
+    assert snap.details["apps_running"] == 1
+
+
+@pytest.mark.asyncio
+async def test_fly_app_degraded_when_machines_partial(monkeypatch):
+    """Machine partially healthy (1 started + 1 stopped) → degraded."""
+    monkeypatch.setenv(FLY_API_TOKEN_ENV, "fly_token_test")
+    cm_patch, _ = _patch_http_client(
+        get_response=_fake_response(
+            status_code=200,
+            json_payload=[{"state": "started"}, {"state": "stopped"}],
+        )
+    )
+    with cm_patch:
+        snap = await FlyProbe().check()
+    assert snap.status == "degraded"
+
+
+@pytest.mark.asyncio
+async def test_fly_all_apps_fail_returns_unhealthy(monkeypatch):
+    monkeypatch.setenv(FLY_API_TOKEN_ENV, "fly_token_test")
+    cm_patch, _ = _patch_http_client(get_response=_fake_response(status_code=404))
+    with cm_patch:
+        snap = await FlyProbe().check()
+    assert snap.status == "unhealthy"
+    assert "404" in snap.error
+    assert "fly_token_test" not in snap.error
+
+
+# =============================================================================
+# Cross-probe: 0 calls when ALL auth envs missing
+# =============================================================================
+
+
+@pytest.mark.asyncio
+async def test_all_probes_make_zero_calls_when_all_envs_missing(monkeypatch):
+    """Defense-in-depth invariant: a fully-unconfigured kora install
+    makes ZERO outbound HTTP calls during a probe cycle."""
+    for env in (
+        VERCEL_API_TOKEN_ENV,
+        SENTRY_API_TOKEN_ENV,
+        SENTRY_ORG_ENV,
+        DOPPLER_API_TOKEN_ENV,
+        SUPABASE_ANON_KEY_ENV,
+        SUPABASE_URL_ENV,
+        FLY_API_TOKEN_ENV,
+    ):
+        monkeypatch.delenv(env, raising=False)
+
+    fake_client = AsyncMock()
+    fake_client.get = AsyncMock()
+    fake_client.head = AsyncMock()
+    fake_cm = MagicMock()
+    fake_cm.__aenter__ = AsyncMock(return_value=fake_client)
+    fake_cm.__aexit__ = AsyncMock(return_value=None)
+
+    with patch("httpx.AsyncClient", return_value=fake_cm):
+        for probe in (
+            VercelProbe(),
+            SentryProbe(),
+            DopplerProbe(),
+            SupabaseProbe(),
+            FlyProbe(),
+        ):
+            snap = await probe.check()
+            assert snap.status == "unknown"
+            assert snap.latency_ms is None
+
+    fake_client.get.assert_not_awaited()
+    fake_client.head.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+
+def _ms_now() -> int:
+    """Vercel uses ms-since-epoch for `created`. Recent → within 24h."""
+    return int(datetime.now(timezone.utc).timestamp() * 1000)
diff --git a/tests/kora_cli/test_heartbeat_probes/test_runner_and_listener.py b/tests/kora_cli/test_heartbeat_probes/test_runner_and_listener.py
new file mode 100644
index 000000000000..d8eb51f29e54
--- /dev/null
+++ b/tests/kora_cli/test_heartbeat_probes/test_runner_and_listener.py
@@ -0,0 +1,324 @@
+"""Runner + listener tests (KR-FEAT-HEARTBEAT ST1).
+
+Covers:
+  - run_all_probes populates the snapshot cache per probe
+  - Per-probe failure isolation — one probe raising doesn't block
+    the others
+  - current_service_snapshots returns a defensive copy
+  - Listener startup is a clean no-op + LOG; shutdown clears the
+    cache (no stale snapshots across daemon restart)
+  - _read_probe_interval: default / env override / invalid / ≤0
+    fallback
+  - Periodic task ``heartbeat.service_probes`` registered in
+    PERIODIC_TASK_REGISTRY at module-import time
+  - Daemon listener ``heartbeat_probes`` registered in
+    LISTENER_REGISTRY at module-import time
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from kora_cli import daemon as daemon_mod
+from kora_cli.heartbeat_probes.runner import (
+    _clear_snapshot_cache,
+    current_service_snapshots,
+    default_probes,
+    run_all_probes,
+    run_all_probes_scheduled,
+)
+from kora_cli.heartbeat_probes.types import ServiceHealthSnapshot
+from kora_cli.listeners import heartbeat as heartbeat_module
+from kora_cli.listeners.heartbeat_probes_listener import (
+    DEFAULT_PROBE_INTERVAL_SEC,
+    PROBE_INTERVAL_ENV,
+    HeartbeatProbesListener,
+    _read_probe_interval,
+)
+
+
+@pytest.fixture(autouse=True)
+def _reset_state():
+    _clear_snapshot_cache()
+    yield
+    _clear_snapshot_cache()
+
+
+# ---------------------------------------------------------------------------
+# default_probes — fresh instances per call
+# ---------------------------------------------------------------------------
+
+
+def test_default_probes_returns_five():
+    probes = default_probes()
+    assert len(probes) == 5
+    names = {p.name for p in probes}
+    assert names == {"vercel", "sentry", "doppler", "supabase", "fly"}
+
+
+def test_default_probes_returns_fresh_instances():
+    a = default_probes()
+    b = default_probes()
+    # Same shape, different objects (probes are constructed fresh)
+    assert {p.name for p in a} == {p.name for p in b}
+    assert a is not b
+
+
+# ---------------------------------------------------------------------------
+# run_all_probes — snapshot cache + isolation
+# ---------------------------------------------------------------------------
+
+
+class _FakeProbe:
+    """Test double mirroring the ServiceProbe Protocol shape."""
+
+    def __init__(self, name: str, snapshot=None, raise_exc=None):
+        self.name = name
+        self._snapshot = snapshot
+        self._raise = raise_exc
+        self.call_count = 0
+
+    async def check(self) -> ServiceHealthSnapshot:
+        self.call_count += 1
+        if self._raise is not None:
+            raise self._raise
+        return self._snapshot or _ok_snapshot(self.name)
+
+
+def _ok_snapshot(name: str) -> ServiceHealthSnapshot:
+    return ServiceHealthSnapshot(
+        name=name,
+        status="healthy",
+        latency_ms=50,
+        last_check_at=datetime.now(timezone.utc),
+        details={},
+    )
+
+
+@pytest.mark.asyncio
+async def test_run_all_probes_populates_cache_per_probe():
+    probes = [_FakeProbe("a"), _FakeProbe("b")]
+    result = await run_all_probes(probes=probes)
+    assert set(result.keys()) == {"a", "b"}
+    assert result["a"].status == "healthy"
+    cache = current_service_snapshots()
+    assert set(cache.keys()) == {"a", "b"}
+    assert cache["a"].status == "healthy"
+
+
+@pytest.mark.asyncio
+async def test_per_probe_failure_isolated_from_siblings():
+    """If probe 'a' raises an unhandled exception, probe 'b' still
+    runs + populates its snapshot."""
+    probes = [
+        _FakeProbe("a", raise_exc=RuntimeError("a went boom")),
+        _FakeProbe("b"),
+    ]
+    result = await run_all_probes(probes=probes)
+    assert set(result.keys()) == {"a", "b"}
+    # 'a' surfaces as unknown with the exception captured
+    assert result["a"].status == "unknown"
+    assert "RuntimeError" in result["a"].error
+    # 'b' completes normally
+    assert result["b"].status == "healthy"
+    assert probes[1].call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_current_service_snapshots_returns_defensive_copy():
+    probes = [_FakeProbe("a")]
+    await run_all_probes(probes=probes)
+    view = current_service_snapshots()
+    view.pop("a", None)  # mutate the returned dict
+    assert "a" not in view
+    # Internal cache unchanged
+    assert "a" in current_service_snapshots()
+
+
+@pytest.mark.asyncio
+async def test_run_all_probes_uses_default_probes_when_none():
+    """Calling without an explicit `probes` arg constructs the
+    default 5 probes. Tests against real httpx are mocked at
+    transport level; here we just verify the default set is
+    used + each name appears."""
+    import httpx
+
+    fake_client = AsyncMock()
+    fake_client.get = AsyncMock(side_effect=httpx.ConnectError("offline"))
+    fake_client.head = AsyncMock(side_effect=httpx.ConnectError("offline"))
+    from unittest.mock import MagicMock
+
+    fake_cm = MagicMock()
+    fake_cm.__aenter__ = AsyncMock(return_value=fake_client)
+    fake_cm.__aexit__ = AsyncMock(return_value=None)
+
+    # Set all envs so probes attempt the (mocked-failing) request
+    import os
+
+    for env, val in (
+        ("KORA_VERCEL_API_TOKEN", "x"),
+        ("KORA_SENTRY_API_TOKEN", "x"),
+        ("KORA_SENTRY_ORG", "test-org"),
+        ("KORA_DOPPLER_API_TOKEN", "x"),
+        ("KORA_SUPABASE_ANON_KEY", "x"),
+        ("KORA_SUPABASE_URL", "https://test.supabase.co"),
+        ("KORA_FLY_API_TOKEN", "x"),
+    ):
+        os.environ[env] = val
+
+    try:
+        with patch("httpx.AsyncClient", return_value=fake_cm):
+            result = await run_all_probes()
+        assert set(result.keys()) == {
+            "vercel",
+            "sentry",
+            "doppler",
+            "supabase",
+            "fly",
+        }
+        # All marked unknown because of the ConnectError
+        for snap in result.values():
+            assert snap.status == "unknown"
+    finally:
+        for env in (
+            "KORA_VERCEL_API_TOKEN",
+            "KORA_SENTRY_API_TOKEN",
+            "KORA_SENTRY_ORG",
+            "KORA_DOPPLER_API_TOKEN",
+            "KORA_SUPABASE_ANON_KEY",
+            "KORA_SUPABASE_URL",
+            "KORA_FLY_API_TOKEN",
+        ):
+            os.environ.pop(env, None)
+
+
+# ---------------------------------------------------------------------------
+# run_all_probes_scheduled — scheduler-facing wrapper
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_scheduled_wrapper_swallows_unexpected_errors_and_logs(caplog):
+    """The scheduler-facing callable must never propagate — a
+    non-cancellation exception from run_all_probes is logged and
+    the scheduler keeps firing."""
+    import logging
+
+    with patch(
+        "kora_cli.heartbeat_probes.runner.run_all_probes",
+        side_effect=RuntimeError("runner itself broke"),
+    ):
+        with caplog.at_level(
+            logging.ERROR, logger="kora_cli.heartbeat_probes.runner"
+        ):
+            await run_all_probes_scheduled()  # must not raise
+    assert any(
+        "run_all_probes raised" in r.message for r in caplog.records
+    )
+
+
+# ---------------------------------------------------------------------------
+# Listener: startup no-op + shutdown clears cache
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_listener_startup_is_clean_noop():
+    listener = HeartbeatProbesListener()
+    await listener.startup()  # no exception, no state
+
+
+@pytest.mark.asyncio
+async def test_listener_shutdown_clears_snapshot_cache():
+    # Seed a snapshot
+    await run_all_probes(probes=[_FakeProbe("a")])
+    assert "a" in current_service_snapshots()
+
+    listener = HeartbeatProbesListener()
+    await listener.shutdown()
+    assert current_service_snapshots() == {}
+
+
+# ---------------------------------------------------------------------------
+# _read_probe_interval
+# ---------------------------------------------------------------------------
+
+
+def test_interval_default_when_env_unset(monkeypatch):
+    monkeypatch.delenv(PROBE_INTERVAL_ENV, raising=False)
+    assert _read_probe_interval() == DEFAULT_PROBE_INTERVAL_SEC
+
+
+def test_interval_env_override(monkeypatch):
+    monkeypatch.setenv(PROBE_INTERVAL_ENV, "60")
+    assert _read_probe_interval() == 60.0
+
+
+def test_interval_invalid_value_falls_back(monkeypatch, caplog):
+    import logging
+
+    monkeypatch.setenv(PROBE_INTERVAL_ENV, "garbage")
+    with caplog.at_level(
+        logging.WARNING,
+        logger="kora_cli.listeners.heartbeat_probes_listener",
+    ):
+        result = _read_probe_interval()
+    assert result == DEFAULT_PROBE_INTERVAL_SEC
+    assert any("not numeric" in r.message for r in caplog.records)
+
+
+def test_interval_zero_falls_back(monkeypatch, caplog):
+    import logging
+
+    monkeypatch.setenv(PROBE_INTERVAL_ENV, "0")
+    with caplog.at_level(
+        logging.WARNING,
+        logger="kora_cli.listeners.heartbeat_probes_listener",
+    ):
+        result = _read_probe_interval()
+    assert result == DEFAULT_PROBE_INTERVAL_SEC
+
+
+def test_interval_negative_falls_back(monkeypatch):
+    monkeypatch.setenv(PROBE_INTERVAL_ENV, "-1")
+    assert _read_probe_interval() == DEFAULT_PROBE_INTERVAL_SEC
+
+
+# ---------------------------------------------------------------------------
+# Registry side: listener + periodic task wired at import time
+# ---------------------------------------------------------------------------
+
+
+def test_daemon_listener_registered_in_registry():
+    from kora_cli.listeners import heartbeat_probes_listener  # noqa: F401
+
+    names = {name for name, _factory in daemon_mod.LISTENER_REGISTRY}
+    assert "heartbeat_probes" in names
+
+
+def test_periodic_task_registered_in_heartbeat_registry():
+    from kora_cli.listeners import heartbeat_probes_listener  # noqa: F401
+
+    names = {t.name for t in heartbeat_module.PERIODIC_TASK_REGISTRY}
+    assert "heartbeat.service_probes" in names
+    task = next(
+        t
+        for t in heartbeat_module.PERIODIC_TASK_REGISTRY
+        if t.name == "heartbeat.service_probes"
+    )
+    assert task.interval_seconds > 0
+    assert task.callable is run_all_probes_scheduled
+
+
+def test_mcp_and_heartbeat_tasks_are_independent():
+    """Per §4 Q1 ruling: don't share scheduler tasks across MCP +
+    heartbeat probes. Both at 5min default, both via the same
+    scheduler, but registered as DISTINCT named tasks so one slow
+    cycle doesn't block the other."""
+    names = {t.name for t in heartbeat_module.PERIODIC_TASK_REGISTRY}
+    assert "heartbeat.service_probes" in names
+    assert "mcp.health_check" in names