Skip to content

forge.health

Kubernetes-compatible /health and /ready endpoints with custom check registration.

forge.health

Health check module — standardized /health and /ready endpoints.

Provides a health check registry, built-in checks for module dependencies, and FastAPI routers for /health (liveness) and /ready (readiness) endpoints compatible with Kubernetes probes.

Classes

HealthModule

Bases: ForgeModule

Manages health and readiness checks for the forge runtime.

Integrates with K8s probes and coordinates module-level checks.

Source code in src/forge/health/module.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
class HealthModule(ForgeModule):
    """
    Manages health and readiness checks for the forge runtime.

    Integrates with K8s probes and coordinates module-level checks.
    """

    name = "health"
    dependencies: ClassVar[list[str]] = ["config"]

    def __init__(self) -> None:
        super().__init__()
        self._registry = HealthRegistry()
        self._runtime: Runtime | None = None

    @property
    def registry(self) -> HealthRegistry:
        """Get the health registry instance."""
        return self._registry

    @property
    def _config(self) -> Any:
        """Dynamically fetch the health configuration section."""
        if self._runtime:
            from forge.config.module import ConfigModule

            try:
                config_module = cast("ConfigModule", self._runtime.get(ConfigModule))
                return getattr(config_module.config, "health", None)
            except Exception:
                return None
        return None

    @property
    def include_details(self) -> bool:
        """Dynamically determine if check details should be included in responses."""
        config = self._config
        if config is None:
            return True
        if isinstance(config, dict):
            return bool(config.get("include_details", True))
        return bool(getattr(config, "include_details", True))

    @property
    def health_path(self) -> str:
        """Get the configured liveness check path."""
        config = self._config
        if config is None:
            return "/health"
        if isinstance(config, dict):
            return str(config.get("health_path", "/health"))
        return str(getattr(config, "health_path", "/health"))

    @property
    def ready_path(self) -> str:
        """Get the configured readiness check path."""
        config = self._config
        if config is None:
            return "/ready"
        if isinstance(config, dict):
            return str(config.get("ready_path", "/ready"))
        return str(getattr(config, "ready_path", "/ready"))

    @property
    def check_timeout(self) -> float:
        """Get the health check execution timeout."""
        config = self._config
        if config is None:
            return 5.0
        if isinstance(config, dict):
            return float(config.get("check_timeout", 5.0))
        return float(getattr(config, "check_timeout", 5.0))

    async def setup(self, runtime: Runtime) -> None:
        """Initialize the health module and configure settings."""
        self._runtime = runtime
        set_health_module(self)

        # OTel instruments
        meter = get_meter()
        self._otel_health_gauge: Any = None
        self._otel_health_latency: Any = None
        if meter is not None:
            self._otel_health_gauge = meter.create_up_down_counter(
                "health.check.status",
                description="Health check status (0=ok, 1=degraded, 2=error)",
                unit="1",
            )
            self._otel_health_latency = meter.create_histogram(
                "health.check.latency",
                description="Health check latency in milliseconds",
                unit="ms",
            )

        # Register default critical "runtime" check
        async def check_runtime() -> HealthResult:
            if runtime.is_initialized:
                return HealthResult.ok()
            return HealthResult.error("Runtime not initialized")

        self._registry.register("runtime", check_runtime, critical=True)

        # Register any pending decorator-registered checks
        for name, fn, critical in _decorator_registered_checks:
            self._registry.register(name, fn, critical=critical)
        _decorator_registered_checks.clear()

        # Configurable paths: update router paths dynamically
        from forge.health.router import health_router, liveness, readiness

        health_path = self.health_path
        ready_path = self.ready_path

        from fastapi.routing import APIRoute

        for route in health_router.routes:
            endpoint = getattr(route, "endpoint", None)
            if isinstance(route, APIRoute):
                if endpoint == liveness:
                    route.path = health_path
                elif endpoint == readiness:
                    route.path = ready_path

    async def teardown(self) -> None:
        """Teardown the health module."""
        set_health_module(None)
        self._runtime = None

    def register(self, name: str, check: HealthCheckFn, critical: bool = False) -> None:
        """Register a custom health check."""
        self._registry.register(name, check, critical=critical)

    def check(self, name: str, critical: bool = False) -> Callable[[HealthCheckFn], HealthCheckFn]:
        """Decorator to register a custom health check function."""

        def decorator(fn: HealthCheckFn) -> HealthCheckFn:
            self._registry.register(name, fn, critical=critical)
            return fn

        return decorator

    async def check_all(self) -> dict[str, dict[str, Any]]:
        """
        Run all registered checks and return results.

        Returns dict of check_name -> {status, message, latency_ms}
        """
        timeout = self.check_timeout

        # 1. Run custom registered checks
        custom_results = await self._registry.run_all(timeout=timeout)

        results: dict[str, dict[str, Any]] = {}
        for name, res in custom_results.items():
            results[name] = {
                "status": res.status,
                "message": res.message,
                "latency_ms": res.latency_ms,
            }

        # 2. Collect module health checks (synchronous)
        if self._runtime:
            import time

            for module in self._runtime._container.initialization_order():
                if module.name == self.name:
                    continue
                start = time.monotonic()
                try:
                    core_res = module.health_check()
                    latency_ms = (time.monotonic() - start) * 1000.0
                    status: str = getattr(core_res, "status", "ok")
                    if hasattr(status, "value"):
                        status = status.value

                    message: str | None = None
                    if status not in ("ok", "degraded", "error"):
                        status = "error"
                        message = f"Invalid health status returned: {status}"
                    else:
                        message = getattr(core_res, "message", None)
                except Exception as exc:
                    latency_ms = (time.monotonic() - start) * 1000.0
                    status = "error"
                    message = f"Check raised: {exc}"

                results[module.name] = {
                    "status": status,
                    "message": message,
                    "latency_ms": round(latency_ms, 2),
                }

        # Record OTel metrics for health checks
        self._record_otel_health_metrics(results)

        return results

    def _record_otel_health_metrics(self, results: dict[str, dict[str, Any]]) -> None:
        if self._otel_health_gauge is None or self._otel_health_latency is None:
            return

        for name, res in results.items():
            status_value: int = {"ok": 0, "degraded": 1, "error": 2}.get(
                res.get("status", "error"), 2
            )
            self._otel_health_gauge.add(
                status_value,
                {"check": name, "status": res.get("status", "error")},
            )
            latency = res.get("latency_ms", 0.0)
            if isinstance(latency, (int, float)):
                self._otel_health_latency.record(latency, {"check": name})

    def is_ready(self, check_results: dict[str, dict[str, Any]]) -> bool:
        """
        Determine overall readiness.

        Returns False if any critical check fails (status is error).
        """
        # Check custom registry checks
        for name, (_, critical) in self._registry._checks.items():
            if critical:
                res = check_results.get(name)
                if res and res.get("status") == "error":
                    return False

        # Check module checks (considered critical if error status returned)
        if self._runtime:
            for module in self._runtime._container.initialization_order():
                if module.name == self.name:
                    continue
                res = check_results.get(module.name)
                if res and res.get("status") == "error":
                    return False

        return True
Attributes
check_timeout property 
check_timeout: float

Get the health check execution timeout.

health_path property 
health_path: str

Get the configured liveness check path.

include_details property 
include_details: bool

Dynamically determine if check details should be included in responses.

ready_path property 
ready_path: str

Get the configured readiness check path.

registry property 
registry: HealthRegistry

Get the health registry instance.

Methods:
check 
check(name: str, critical: bool = False) -> Callable[[HealthCheckFn], HealthCheckFn]

Decorator to register a custom health check function.

Source code in src/forge/health/module.py 
152
153
154
155
156
157
158
159
def check(self, name: str, critical: bool = False) -> Callable[[HealthCheckFn], HealthCheckFn]:
    """Decorator to register a custom health check function."""

    def decorator(fn: HealthCheckFn) -> HealthCheckFn:
        self._registry.register(name, fn, critical=critical)
        return fn

    return decorator
check_all async 
check_all() -> dict[str, dict[str, Any]]

Run all registered checks and return results.

Returns dict of check_name -> {status, message, latency_ms}

Source code in src/forge/health/module.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
async def check_all(self) -> dict[str, dict[str, Any]]:
    """
    Run all registered checks and return results.

    Returns dict of check_name -> {status, message, latency_ms}
    """
    timeout = self.check_timeout

    # 1. Run custom registered checks
    custom_results = await self._registry.run_all(timeout=timeout)

    results: dict[str, dict[str, Any]] = {}
    for name, res in custom_results.items():
        results[name] = {
            "status": res.status,
            "message": res.message,
            "latency_ms": res.latency_ms,
        }

    # 2. Collect module health checks (synchronous)
    if self._runtime:
        import time

        for module in self._runtime._container.initialization_order():
            if module.name == self.name:
                continue
            start = time.monotonic()
            try:
                core_res = module.health_check()
                latency_ms = (time.monotonic() - start) * 1000.0
                status: str = getattr(core_res, "status", "ok")
                if hasattr(status, "value"):
                    status = status.value

                message: str | None = None
                if status not in ("ok", "degraded", "error"):
                    status = "error"
                    message = f"Invalid health status returned: {status}"
                else:
                    message = getattr(core_res, "message", None)
            except Exception as exc:
                latency_ms = (time.monotonic() - start) * 1000.0
                status = "error"
                message = f"Check raised: {exc}"

            results[module.name] = {
                "status": status,
                "message": message,
                "latency_ms": round(latency_ms, 2),
            }

    # Record OTel metrics for health checks
    self._record_otel_health_metrics(results)

    return results
is_ready 
is_ready(check_results: dict[str, dict[str, Any]]) -> bool

Determine overall readiness.

Returns False if any critical check fails (status is error).

Source code in src/forge/health/module.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def is_ready(self, check_results: dict[str, dict[str, Any]]) -> bool:
    """
    Determine overall readiness.

    Returns False if any critical check fails (status is error).
    """
    # Check custom registry checks
    for name, (_, critical) in self._registry._checks.items():
        if critical:
            res = check_results.get(name)
            if res and res.get("status") == "error":
                return False

    # Check module checks (considered critical if error status returned)
    if self._runtime:
        for module in self._runtime._container.initialization_order():
            if module.name == self.name:
                continue
            res = check_results.get(module.name)
            if res and res.get("status") == "error":
                return False

    return True
register 
register(name: str, check: HealthCheckFn, critical: bool = False) -> None

Register a custom health check.

Source code in src/forge/health/module.py 
148
149
150
def register(self, name: str, check: HealthCheckFn, critical: bool = False) -> None:
    """Register a custom health check."""
    self._registry.register(name, check, critical=critical)
setup async 
setup(runtime: ForgeRuntime) -> None

Initialize the health module and configure settings.

Source code in src/forge/health/module.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
async def setup(self, runtime: Runtime) -> None:
    """Initialize the health module and configure settings."""
    self._runtime = runtime
    set_health_module(self)

    # OTel instruments
    meter = get_meter()
    self._otel_health_gauge: Any = None
    self._otel_health_latency: Any = None
    if meter is not None:
        self._otel_health_gauge = meter.create_up_down_counter(
            "health.check.status",
            description="Health check status (0=ok, 1=degraded, 2=error)",
            unit="1",
        )
        self._otel_health_latency = meter.create_histogram(
            "health.check.latency",
            description="Health check latency in milliseconds",
            unit="ms",
        )

    # Register default critical "runtime" check
    async def check_runtime() -> HealthResult:
        if runtime.is_initialized:
            return HealthResult.ok()
        return HealthResult.error("Runtime not initialized")

    self._registry.register("runtime", check_runtime, critical=True)

    # Register any pending decorator-registered checks
    for name, fn, critical in _decorator_registered_checks:
        self._registry.register(name, fn, critical=critical)
    _decorator_registered_checks.clear()

    # Configurable paths: update router paths dynamically
    from forge.health.router import health_router, liveness, readiness

    health_path = self.health_path
    ready_path = self.ready_path

    from fastapi.routing import APIRoute

    for route in health_router.routes:
        endpoint = getattr(route, "endpoint", None)
        if isinstance(route, APIRoute):
            if endpoint == liveness:
                route.path = health_path
            elif endpoint == readiness:
                route.path = ready_path
teardown async 
teardown() -> None

Teardown the health module.

Source code in src/forge/health/module.py 
143
144
145
146
async def teardown(self) -> None:
    """Teardown the health module."""
    set_health_module(None)
    self._runtime = None

HealthRegistry

Registry managing custom registered health check functions.

Source code in src/forge/health/checks.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class HealthRegistry:
    """Registry managing custom registered health check functions."""

    def __init__(self) -> None:
        self._checks: dict[str, tuple[HealthCheckFn, bool]] = {}

    def register(self, name: str, check_fn: HealthCheckFn, critical: bool = False) -> None:
        """Register a health check function."""
        self._checks[name] = (check_fn, critical)

    def unregister(self, name: str) -> None:
        """Unregister a health check function by name."""
        self._checks.pop(name, None)

    async def _run_single(self, name: str, check_fn: HealthCheckFn, timeout: float) -> HealthResult:  # noqa: ASYNC109
        """Run a single check function, enforcing a timeout and catching errors."""
        start_time = time.monotonic()
        try:
            # Run the check function and enforce timeout
            res = await asyncio.wait_for(check_fn(), timeout=timeout)
            latency_ms = (time.monotonic() - start_time) * 1000.0

            # Extract status and message from result (supports core or local HealthResult)
            status: Literal["ok", "degraded", "error"] = "ok"
            message: str | None = None

            if res is not None:
                raw_status = getattr(res, "status", "ok")
                if hasattr(raw_status, "value"):
                    raw_status = raw_status.value
                if raw_status in ("ok", "degraded", "error"):
                    status = cast("Literal['ok', 'degraded', 'error']", raw_status)
                else:
                    status = "error"
                    message = f"Invalid health status returned: {raw_status}"

                message = getattr(res, "message", message)

            return HealthResult(status=status, message=message, latency_ms=round(latency_ms, 2))
        except TimeoutError:
            latency_ms = (time.monotonic() - start_time) * 1000.0
            return HealthResult(
                status="error",
                message=f"Check timed out after {timeout}s",
                latency_ms=round(latency_ms, 2),
            )
        except Exception as exc:
            latency_ms = (time.monotonic() - start_time) * 1000.0
            return HealthResult(status="error", message=str(exc), latency_ms=round(latency_ms, 2))

    async def run_all(self, timeout: float = 5.0) -> dict[str, HealthResult]:  # noqa: ASYNC109
        """Run all registered checks concurrently."""
        if not self._checks:
            return {}

        names = list(self._checks.keys())
        tasks = [self._run_single(name, fn, timeout) for name, (fn, _) in self._checks.items()]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        dict_results: dict[str, HealthResult] = {}
        for name, res in zip(names, results, strict=True):
            if isinstance(res, HealthResult):
                dict_results[name] = res
            else:
                dict_results[name] = HealthResult(status="error", message=str(res))
        return dict_results

    async def run_critical(self, timeout: float = 5.0) -> dict[str, HealthResult]:  # noqa: ASYNC109
        """Run only critical registered checks concurrently."""
        critical_checks = {name: (fn, crit) for name, (fn, crit) in self._checks.items() if crit}
        if not critical_checks:
            return {}

        names = list(critical_checks.keys())
        tasks = [self._run_single(name, fn, timeout) for name, (fn, _) in critical_checks.items()]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        dict_results: dict[str, HealthResult] = {}
        for name, res in zip(names, results, strict=True):
            if isinstance(res, HealthResult):
                dict_results[name] = res
            else:
                dict_results[name] = HealthResult(status="error", message=str(res))
        return dict_results
Methods:
register 
register(name: str, check_fn: HealthCheckFn, critical: bool = False) -> None

Register a health check function.

Source code in src/forge/health/checks.py 
43
44
45
def register(self, name: str, check_fn: HealthCheckFn, critical: bool = False) -> None:
    """Register a health check function."""
    self._checks[name] = (check_fn, critical)
run_all async 
run_all(timeout: float = 5.0) -> dict[str, HealthResult]

Run all registered checks concurrently.

Source code in src/forge/health/checks.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def run_all(self, timeout: float = 5.0) -> dict[str, HealthResult]:  # noqa: ASYNC109
    """Run all registered checks concurrently."""
    if not self._checks:
        return {}

    names = list(self._checks.keys())
    tasks = [self._run_single(name, fn, timeout) for name, (fn, _) in self._checks.items()]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    dict_results: dict[str, HealthResult] = {}
    for name, res in zip(names, results, strict=True):
        if isinstance(res, HealthResult):
            dict_results[name] = res
        else:
            dict_results[name] = HealthResult(status="error", message=str(res))
    return dict_results
run_critical async 
run_critical(timeout: float = 5.0) -> dict[str, HealthResult]

Run only critical registered checks concurrently.

Source code in src/forge/health/checks.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
async def run_critical(self, timeout: float = 5.0) -> dict[str, HealthResult]:  # noqa: ASYNC109
    """Run only critical registered checks concurrently."""
    critical_checks = {name: (fn, crit) for name, (fn, crit) in self._checks.items() if crit}
    if not critical_checks:
        return {}

    names = list(critical_checks.keys())
    tasks = [self._run_single(name, fn, timeout) for name, (fn, _) in critical_checks.items()]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    dict_results: dict[str, HealthResult] = {}
    for name, res in zip(names, results, strict=True):
        if isinstance(res, HealthResult):
            dict_results[name] = res
        else:
            dict_results[name] = HealthResult(status="error", message=str(res))
    return dict_results
unregister 
unregister(name: str) -> None

Unregister a health check function by name.

Source code in src/forge/health/checks.py 
47
48
49
def unregister(self, name: str) -> None:
    """Unregister a health check function by name."""
    self._checks.pop(name, None)

HealthResult dataclass

Result of a single health check.

Source code in src/forge/health/checks.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@dataclass
class HealthResult:
    """Result of a single health check."""

    status: Literal["ok", "degraded", "error"]
    message: str | None = None
    latency_ms: float = 0.0

    @classmethod
    def ok(cls, message: str | None = None) -> HealthResult:
        """Create a successful health check result."""
        return cls(status="ok", message=message)

    @classmethod
    def degraded(cls, message: str) -> HealthResult:
        """Create a degraded health check result."""
        return cls(status="degraded", message=message)

    @classmethod
    def error(cls, message: str) -> HealthResult:
        """Create a failing health check result."""
        return cls(status="error", message=message)
Methods:
degraded classmethod 
degraded(message: str) -> HealthResult

Create a degraded health check result.

Source code in src/forge/health/checks.py 
23
24
25
26
@classmethod
def degraded(cls, message: str) -> HealthResult:
    """Create a degraded health check result."""
    return cls(status="degraded", message=message)
error classmethod 
error(message: str) -> HealthResult

Create a failing health check result.

Source code in src/forge/health/checks.py 
28
29
30
31
@classmethod
def error(cls, message: str) -> HealthResult:
    """Create a failing health check result."""
    return cls(status="error", message=message)
ok classmethod 
ok(message: str | None = None) -> HealthResult

Create a successful health check result.

Source code in src/forge/health/checks.py 
18
19
20
21
@classmethod
def ok(cls, message: str | None = None) -> HealthResult:
    """Create a successful health check result."""
    return cls(status="ok", message=message)

Functions:

check 

check(name: str, critical: bool = False) -> Callable[[HealthCheckFn], HealthCheckFn]

Decorator to register a custom health check.

Source code in src/forge/health/checks.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def check(name: str, critical: bool = False) -> Callable[[HealthCheckFn], HealthCheckFn]:
    """Decorator to register a custom health check."""

    def decorator(fn: HealthCheckFn) -> HealthCheckFn:
        from forge.health._state import get_health_module

        hm = get_health_module()
        if hm is not None:
            hm.registry.register(name, fn, critical=critical)
        else:
            _decorator_registered_checks.append((name, fn, critical))
        return fn

    return decorator