flayer.diagnostics

Module source

flayer.diagnostics

Safe diagnostic contracts with explicit provider and endpoint boundaries.

BenchmarkLimits dataclass

BenchmarkLimits(timeout_seconds: float = 3.0, count: int = 3, byte_limit: int = 64 * 1024)

Hard per-request deadline, request count, and response payload limits.

timeout_seconds class-attribute instance-attribute

timeout_seconds: float = 3.0

count class-attribute instance-attribute

count: int = 3

byte_limit class-attribute instance-attribute

byte_limit: int = 64 * 1024

EndpointMeasurement dataclass

EndpointMeasurement(status: DiagnosticStatus, duration_seconds: float, received_bytes: int = 0, http_status: int | None = None)

Safe metrics exclude response content, credentials, and target identifiers.

status instance-attribute

status: DiagnosticStatus

duration_seconds instance-attribute

duration_seconds: float

received_bytes class-attribute instance-attribute

received_bytes: int = 0

http_status class-attribute instance-attribute

http_status: int | None = None

EndpointTransport

Bases: Protocol

A measurement implementation enforces supplied limits without emitting raw data.

__call__

__call__(endpoint: str, timeout_seconds: float, byte_limit: int) -> EndpointMeasurement

Return one bounded HTTP measurement without exposing response payloads.

Source code in installed/flayer/diagnostics/benchmark.py

def __call__(
    self, endpoint: str, timeout_seconds: float, byte_limit: int
) -> EndpointMeasurement:
    """Return one bounded HTTP measurement without exposing response payloads."""

    ...

CheckResult dataclass

CheckResult(name: str, status: DiagnosticStatus, message: str, details: Mapping[str, object] = dict())

One diagnostic observation with details that are sanitized at serialization.

name instance-attribute

name: str

status instance-attribute

status: DiagnosticStatus

message instance-attribute

message: str

details class-attribute instance-attribute

details: Mapping[str, object] = field(default_factory=dict)

AsDict

AsDict() -> dict[str, object]

Return stable JSON-compatible fields without arbitrary raw provider output.

Source code in installed/flayer/diagnostics/contracts.py

def AsDict(self) -> dict[str, object]:
    """Return stable JSON-compatible fields without arbitrary raw provider output."""

    return {
        "name": Redact(self.name),
        "status": self.status.value,
        "message": Redact(self.message),
        "details": Redact(self.details),
    }

DiagnosticReport dataclass

DiagnosticReport(command: str, checks: tuple[CheckResult, ...])

Versioned report whose aggregate outcome never hides failed checks.

command instance-attribute

command: str

checks instance-attribute

checks: tuple[CheckResult, ...]

Status property

Status: DiagnosticStatus

Return the most severe result, treating an empty report as unsupported.

ExitCode

ExitCode() -> int

Return zero only when every requested observation succeeds.

Source code in installed/flayer/diagnostics/contracts.py

def ExitCode(self) -> int:
    """Return zero only when every requested observation succeeds."""

    return EXIT_CODES[self.Status]

AsDict

AsDict() -> dict[str, object]

Expose the report schema and sanitized observations in deterministic order.

Source code in installed/flayer/diagnostics/contracts.py

def AsDict(self) -> dict[str, object]:
    """Expose the report schema and sanitized observations in deterministic order."""

    return {
        "schema_version": 1,
        "command": Redact(self.command),
        "status": self.Status.value,
        "checks": [check.AsDict() for check in self.checks],
    }

DiagnosticStatus

Bases: str, Enum

Stable outcomes distinguish failure, expired evidence, and missing support.

OK class-attribute instance-attribute

OK = 'ok'

WARNING class-attribute instance-attribute

WARNING = 'warning'

FAILED class-attribute instance-attribute

FAILED = 'failed'

STALE class-attribute instance-attribute

STALE = 'stale'

UNSUPPORTED class-attribute instance-attribute

UNSUPPORTED = 'unsupported'

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

def __call__(self) -> CheckResult:
    """Perform an adapter-bounded probe and return one structured observation."""

    ...

MeasureEndpoint

MeasureEndpoint(endpoint: str, timeout_seconds: float, byte_limit: int) -> EndpointMeasurement

Terminate isolated network work at the deadline, including a stalled DNS resolver.

Source code in installed/flayer/diagnostics/benchmark.py

def MeasureEndpoint(endpoint: str, timeout_seconds: float, byte_limit: int) -> EndpointMeasurement:
    """Terminate isolated network work at the deadline, including a stalled DNS resolver."""

    ValidateEndpoint(endpoint)
    BenchmarkLimits(timeout_seconds=timeout_seconds, count=1, byte_limit=byte_limit)
    context = multiprocessing.get_context("spawn")
    receiver, sender = context.Pipe(duplex=False)
    process = context.Process(
        target=EndpointWorker, args=(sender, endpoint, timeout_seconds, byte_limit), daemon=True
    )
    started = time.monotonic()
    launched = False

    try:
        process.start()
        launched = True
        sender.close()
        remaining = max(0.0, timeout_seconds - (time.monotonic() - started))

        if receiver.poll(remaining):
            measurement: EndpointMeasurement = receiver.recv()

            return measurement

        return EndpointMeasurement(DiagnosticStatus.FAILED, timeout_seconds)

    except (OSError, EOFError, RuntimeError):
        return EndpointMeasurement(DiagnosticStatus.FAILED, time.monotonic() - started)

    finally:
        sender.close()
        receiver.close()

        if launched:
            if process.is_alive():
                process.terminate()

            process.join(timeout=1.0)

            if process.is_alive():
                process.kill()
                process.join(timeout=1.0)

        process.close()

ProbeEndpoint

ProbeEndpoint(endpoint: str, timeout_seconds: float = 3.0, transport: EndpointTransport = MeasureEndpoint) -> CheckResult

Check explicit HTTP reachability without pretending it validates guest hardening.

Source code in installed/flayer/diagnostics/benchmark.py

def ProbeEndpoint(
    endpoint: str, timeout_seconds: float = 3.0,
    transport: EndpointTransport = MeasureEndpoint,
) -> CheckResult:
    """Check explicit HTTP reachability without pretending it validates guest hardening."""

    ValidateEndpoint(endpoint)
    limits = BenchmarkLimits(timeout_seconds=timeout_seconds, count=1, byte_limit=1)
    measurement = SafeMeasurement(endpoint, limits, transport)

    return CheckResult(
        "endpoint", measurement.status,
        "Endpoint answered successfully" if measurement.status is DiagnosticStatus.OK
        else "Endpoint probe did not succeed",
        {"duration_seconds": measurement.duration_seconds, "http_status": measurement.http_status},
    )

RunBenchmark

RunBenchmark(endpoint: str, limits: BenchmarkLimits | None = None, transport: EndpointTransport = MeasureEndpoint) -> tuple[CheckResult, ...]

Measure a fixed count of bounded downloads without inferring upload or tunnel speed.

Source code in installed/flayer/diagnostics/benchmark.py

def RunBenchmark(
    endpoint: str, limits: BenchmarkLimits | None = None,
    transport: EndpointTransport = MeasureEndpoint,
) -> tuple[CheckResult, ...]:
    """Measure a fixed count of bounded downloads without inferring upload or tunnel speed."""

    ValidateEndpoint(endpoint)
    resolved_limits = limits or BenchmarkLimits()
    checks: list[CheckResult] = []

    for index in range(resolved_limits.count):
        measurement = SafeMeasurement(endpoint, resolved_limits, transport)
        status = measurement.status

        if status is DiagnosticStatus.OK and measurement.received_bytes == 0:
            status = DiagnosticStatus.FAILED

        throughput = (
            measurement.received_bytes * 8 / measurement.duration_seconds / 1_000_000
            if measurement.duration_seconds > 0 else None
        )
        checks.append(CheckResult(
            f"http-download-{index + 1}", status,
            "Bounded HTTP download measured" if status is DiagnosticStatus.OK
            else "Bounded HTTP download did not succeed",
            {
                "received_bytes": measurement.received_bytes,
                "duration_seconds": measurement.duration_seconds,
                "throughput_mbps": throughput,
                "http_status": measurement.http_status,
                "byte_limit": resolved_limits.byte_limit,
            },
        ))

    return tuple(checks)

Redact

Redact(value: object, _depth: int = 0) -> object

Recursively redact credential fields and common sensitive text before output.

Source code in installed/flayer/diagnostics/contracts.py

def Redact(value: object, _depth: int = 0) -> object:
    """Recursively redact credential fields and common sensitive text before output."""

    if _depth >= 16:
        return REDACTED

    if isinstance(value, Mapping):
        sanitized: dict[str, object] = {}

        for key, item in value.items():
            sensitive = not isinstance(key, str) or bool(SECRET_KEY.search(key))
            safe_key = key if isinstance(key, str) and SAFE_KEY.fullmatch(key) else REDACTED

            if sensitive:
                safe_key = REDACTED

            candidate = safe_key
            suffix = 2

            while candidate in sanitized:
                candidate = f"{safe_key}:{suffix}"
                suffix += 1

            sanitized[candidate] = REDACTED if sensitive else Redact(item, _depth + 1)

        return sanitized

    if isinstance(value, str):
        return SECRET_VALUE.sub(REDACTED, value)

    if isinstance(value, Sequence) and not isinstance(value, (bytes, bytearray)):
        return [Redact(item, _depth + 1) for item in value]

    if isinstance(value, bool) or value is None or isinstance(value, int):
        return value

    if isinstance(value, float):
        return value if math.isfinite(value) else None

    return REDACTED

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

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
    )

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

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})

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

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)

RuntimeChecks

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

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

Source code in installed/flayer/diagnostics/health.py

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])},
        ),
    )

RenderJson

RenderJson(report: DiagnosticReport) -> str

Serialize only the versioned public report with deterministic JSON keys.

Source code in installed/flayer/diagnostics/reporting.py

def RenderJson(report: DiagnosticReport) -> str:
    """Serialize only the versioned public report with deterministic JSON keys."""

    return json.dumps(report.AsDict(), sort_keys=True, allow_nan=False)

RenderText

RenderText(report: DiagnosticReport) -> str

Render one safe single-line observation per check without raw details.

Source code in installed/flayer/diagnostics/reporting.py

def RenderText(report: DiagnosticReport) -> str:
    """Render one safe single-line observation per check without raw details."""

    public = report.AsDict()
    lines = [f"{public['command']}: {public['status']}"]

    for check in report.checks:
        safe = check.AsDict()
        lines.append(f"[{safe['status']}] {safe['name']}: {safe['message']}")

    return "\n".join(" ".join(line.split()) for line in lines)