Skip to content

flayer.diagnostics.health

Module source

flayer.diagnostics.health

Injected health probes and freshness validation without cloud access assumptions.

HealthProbe

Bases: Protocol

An adapter supplies read-only observations without transferring authorization.

__call__ 

__call__() -> CheckResult

Perform an adapter-bounded probe and return one structured observation.

Source code in installed/flayer/diagnostics/health.py 
17
18
19
20
def __call__(self) -> CheckResult:
    """Perform an adapter-bounded probe and return one structured observation."""

    ...

RuntimeChecks 

RuntimeChecks() -> tuple[CheckResult, ...]

Check only supported Python execution; never imply remote infrastructure health.

Source code in installed/flayer/diagnostics/health.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
def RuntimeChecks() -> tuple[CheckResult, ...]:
    """Check only supported Python execution; never imply remote infrastructure health."""

    supported = sys.version_info >= (3, 11)

    return (
        CheckResult(
            "python-runtime",
            DiagnosticStatus.OK if supported else DiagnosticStatus.FAILED,
            "Python runtime is supported" if supported else "Python 3.11 or newer is required",
            {"version": ".".join(str(part) for part in sys.version_info[:3])},
        ),
    )

RunChecks 

RunChecks(probes: Iterable[HealthProbe]) -> tuple[CheckResult, ...]

Collect injected observations and suppress raw exception text on probe failure.

Source code in installed/flayer/diagnostics/health.py 
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
def RunChecks(probes: Iterable[HealthProbe]) -> tuple[CheckResult, ...]:
    """Collect injected observations and suppress raw exception text on probe failure."""

    checks: list[CheckResult] = []

    for probe in probes:
        try:
            result = probe()

            if not isinstance(result, CheckResult):
                raise TypeError("Probe must return CheckResult")

            checks.append(result)

        except Exception:
            checks.append(CheckResult(
                "probe", DiagnosticStatus.FAILED, "Probe failed without safe structured evidence"
            ))

    if not checks:
        checks.append(CheckResult(
            "probe", DiagnosticStatus.UNSUPPORTED, "No health probe was supplied"
        ))

    return tuple(checks)

ProviderStatus 

ProviderStatus(observation: Mapping[str, object] | None = None) -> CheckResult

Interpret a supplied generic provider observation without discovering resources.

Source code in installed/flayer/diagnostics/health.py 
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
def ProviderStatus(observation: Mapping[str, object] | None = None) -> CheckResult:
    """Interpret a supplied generic provider observation without discovering resources."""

    if observation is None:
        return CheckResult(
            "provider-state", DiagnosticStatus.UNSUPPORTED, "No provider observation was supplied"
        )

    if not isinstance(observation, Mapping):
        return CheckResult(
            "provider-state", DiagnosticStatus.FAILED, "Provider observation must be a mapping"
        )

    state = observation.get("state")

    if state not in ("ready", "pending", "failed", "absent"):
        return CheckResult(
            "provider-state", DiagnosticStatus.FAILED, "Provider observation has an invalid state"
        )

    status = {
        "ready": DiagnosticStatus.OK,
        "pending": DiagnosticStatus.WARNING,
        "failed": DiagnosticStatus.FAILED,
        "absent": DiagnosticStatus.FAILED,
    }[str(state)]

    return CheckResult("provider-state", status, "Provider state observed", {"state": state})

EvaluateHealthReport 

EvaluateHealthReport(report: Mapping[str, object] | None, max_age_seconds: float = 300, current_time: float | None = None) -> CheckResult

Validate versioned health evidence and separate stale evidence from failed health.

Source code in installed/flayer/diagnostics/health.py 
 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
def EvaluateHealthReport(
    report: Mapping[str, object] | None,
    max_age_seconds: float = 300,
    current_time: float | None = None,
) -> CheckResult:
    """Validate versioned health evidence and separate stale evidence from failed health."""

    now = time.time() if current_time is None else current_time

    if (
        isinstance(max_age_seconds, bool)
        or not isinstance(max_age_seconds, (int, float))
        or not math.isfinite(max_age_seconds)
        or max_age_seconds <= 0
        or isinstance(now, bool)
        or not isinstance(now, (int, float))
        or not math.isfinite(now)
    ):
        raise ValueError("Health freshness bounds must be finite and positive")

    if report is None:
        return CheckResult(
            "health-report", DiagnosticStatus.UNSUPPORTED, "No health report was supplied"
        )

    if not isinstance(report, Mapping):
        return CheckResult(
            "health-report", DiagnosticStatus.FAILED, "Health report must be a mapping"
        )

    if type(report.get("schema_version")) is not int or report.get("schema_version") != 1:
        return CheckResult(
            "health-report", DiagnosticStatus.UNSUPPORTED, "Health report schema is unsupported"
        )

    timestamp = report.get("timestamp")
    outcome = report.get("status")

    if (
        isinstance(timestamp, bool)
        or not isinstance(timestamp, (int, float))
        or not math.isfinite(timestamp)
        or outcome not in ("ok", "warning", "failed")
    ):
        return CheckResult(
            "health-report", DiagnosticStatus.FAILED, "Health report fields are invalid"
        )

    age = now - timestamp
    details: dict[str, object] = {"age_seconds": age, "max_age_seconds": max_age_seconds}

    if age < 0 or age > max_age_seconds:
        return CheckResult(
            "health-report", DiagnosticStatus.STALE,
            "Health report is expired or dated in the future", details,
        )

    return CheckResult(
        "health-report", DiagnosticStatus(str(outcome)), "Health report is current", details
    )