flayer.integrations

Module source

flayer.integrations

Consumer-neutral observation contracts without transport, dispatch, or authorization.

CheckEvidence dataclass

CheckEvidence(status: DiagnosticStatus, metrics: tuple[Metric, ...] = ())

A diagnostic outcome and measurements without private names, prose, or details.

status instance-attribute

status: DiagnosticStatus

metrics class-attribute instance-attribute

metrics: tuple[Metric, ...] = ()

DiagnosticEvidence dataclass

DiagnosticEvidence(checks: tuple[CheckEvidence, ...])

Ordered diagnostic checks retain truthful aggregation without external text.

checks instance-attribute

checks: tuple[CheckEvidence, ...]

Status property

Status: DiagnosticStatus

Use the established severity contract, including unsupported empty evidence.

IntegrationError

Bases: ContractError

An envelope violates the bounded evidence contract without exposing its input.

IntegrationOperation

Bases: StrEnum

Supported observations convey no infrastructure execution authorization.

STATUS class-attribute instance-attribute

STATUS = 'status'

HEALTH class-attribute instance-attribute

HEALTH = 'health'

BENCHMARK class-attribute instance-attribute

BENCHMARK = 'benchmark'

STATE class-attribute instance-attribute

STATE = 'state'

LIFECYCLE_REPORT class-attribute instance-attribute

LIFECYCLE_REPORT = 'lifecycle-report'

IntegrationRequest dataclass

IntegrationRequest(identity: StackIdentity, request_id: str, operation: IntegrationOperation, schema_version: int = 1)

Non-authorizing observation intent bound to exact scope, owner, and correlation.

identity instance-attribute

identity: StackIdentity

request_id instance-attribute

request_id: str

operation instance-attribute

operation: IntegrationOperation

schema_version class-attribute instance-attribute

schema_version: int = 1

IntegrationResult dataclass

IntegrationResult(request: IntegrationRequest, evidence: Evidence)

Allowlisted evidence bound to the exact original validated observation request.

request instance-attribute

request: IntegrationRequest

evidence instance-attribute

evidence: Evidence

LifecycleEvidence dataclass

LifecycleEvidence(action: str, status: str, recovery_required: bool, resources: tuple[ResourceCount, ...])

Historical lifecycle outcome and recovery requirement convey no mutation permission.

action instance-attribute

action: str

status instance-attribute

status: str

recovery_required instance-attribute

recovery_required: bool

resources instance-attribute

resources: tuple[ResourceCount, ...]

Metric dataclass

Metric(name: str, value: int | float | None)

One allowlisted bounded measurement that cannot contain strings or raw payloads.

name instance-attribute

name: str

value instance-attribute

value: int | float | None

ResourceCount dataclass

ResourceCount(status: str, count: int)

One stable lifecycle observation category without resource locators.

status instance-attribute

status: str

count instance-attribute

count: int

StateEvidence dataclass

StateEvidence(resource_count: int)

A local owned-state count makes no remote existence or health assertion.

resource_count instance-attribute

resource_count: int

AdaptDiagnostics

AdaptDiagnostics(request: IntegrationRequest, report: DiagnosticReport, observed_identity: StackIdentity) -> IntegrationResult

Bind selected status, health, or benchmark observations and omit arbitrary source text.

Source code in installed/flayer/integrations/adapters.py

def AdaptDiagnostics(
    request: IntegrationRequest, report: DiagnosticReport, observed_identity: StackIdentity,
) -> IntegrationResult:
    """Bind selected status, health, or benchmark observations and omit arbitrary source text."""

    RequireIdentity(observed_identity, request.identity)

    if (
        request.operation not in {
            IntegrationOperation.STATUS, IntegrationOperation.HEALTH, IntegrationOperation.BENCHMARK,
        }
        or not isinstance(report, DiagnosticReport)
        or report.command != request.operation.value
        or len(report.checks) > MAX_CHECKS
    ):
        raise IntegrationError("Diagnostic source does not match the integration request")

    checks = tuple(CheckEvidence(check.status, tuple(
        Metric(name, cast(int | float | None, check.details[name])) for name in sorted(METRIC_BOUNDS)
        if name in check.details
    )) for check in report.checks)

    return IntegrationResult(request, DiagnosticEvidence(checks))

AdaptLifecycle

AdaptLifecycle(request: IntegrationRequest, report: LifecycleReport, observed_identity: StackIdentity) -> IntegrationResult

Summarize supplied lifecycle evidence with caller-bound scope, never remote attestation.

Source code in installed/flayer/integrations/adapters.py

def AdaptLifecycle(
    request: IntegrationRequest, report: LifecycleReport, observed_identity: StackIdentity,
) -> IntegrationResult:
    """Summarize supplied lifecycle evidence with caller-bound scope, never remote attestation."""

    RequireIdentity(observed_identity, request.identity)

    if (
        request.operation is not IntegrationOperation.LIFECYCLE_REPORT
        or not isinstance(report, LifecycleReport)
        or not isinstance(report.resources, tuple)
        or len(report.resources) > MAX_RESOURCES
        or any(not isinstance(resource, LifecycleObservation) for resource in report.resources)
        or any(not isinstance(resource.status, str) or resource.status not in RESOURCE_STATUSES
               for resource in report.resources)
    ):
        raise IntegrationError("Lifecycle source does not match the integration request")

    counts = Counter(resource.status for resource in report.resources)
    evidence = LifecycleEvidence(
        report.action, report.status, report.recovery_required,
        tuple(ResourceCount(status, count) for status, count in sorted(counts.items())),
    )

    return IntegrationResult(request, evidence)

AdaptState

AdaptState(request: IntegrationRequest, state: StackState) -> IntegrationResult

Summarize recorded ownership without exporting locators or claiming remote readiness.

Source code in installed/flayer/integrations/adapters.py

def AdaptState(request: IntegrationRequest, state: StackState) -> IntegrationResult:
    """Summarize recorded ownership without exporting locators or claiming remote readiness."""

    if request.operation is not IntegrationOperation.STATE or not isinstance(state, StackState):
        raise IntegrationError("Owned state source does not match the integration request")

    RequireIdentity(state.identity, request.identity)

    return IntegrationResult(request, StateEvidence(len(state.resources)))

ParseRequest

ParseRequest(payload: str, expected_identity: StackIdentity) -> IntegrationRequest

Validate untrusted read-only intent without invoking any operation or permission check.

Source code in installed/flayer/integrations/wire.py

def ParseRequest(payload: str, expected_identity: StackIdentity) -> IntegrationRequest:
    """Validate untrusted read-only intent without invoking any operation or permission check."""

    try:
        return _Request(_Decode(payload), expected_identity, "request")

    except ValueError:
        raise IntegrationError("Unable to parse a valid owned integration request") from None

ParseResult

ParseResult(payload: str, expected_request: IntegrationRequest) -> IntegrationResult

Bind an untrusted representation to its request without asserting authenticity.

Source code in installed/flayer/integrations/wire.py

def ParseResult(payload: str, expected_request: IntegrationRequest) -> IntegrationResult:
    """Bind an untrusted representation to its request without asserting authenticity."""

    try:
        if not isinstance(expected_request, IntegrationRequest):
            raise IntegrationError("Integration result requires an originating request")

        value = _Decode(payload)
        request = _Request(value, expected_request.identity, "result")

        if request != expected_request:
            raise IntegrationError("Integration result does not match the originating request")

        table = RequireTable(value, "integration result")

        if request.operation is IntegrationOperation.STATE:
            state = RequireTable(table["evidence"], "state evidence")
            ValidateFields(state, frozenset({"resource_count"}), frozenset(), "state evidence")

            return IntegrationResult(request, StateEvidence(cast(int, state["resource_count"])))

        if request.operation is IntegrationOperation.LIFECYCLE_REPORT:
            return IntegrationResult(request, _ParseLifecycle(table["evidence"]))

        return IntegrationResult(request, _ParseDiagnostic(table["evidence"]))

    except (ValueError, TypeError):
        raise IntegrationError("Unable to parse a valid correlated integration result") from None

SerializeRequest

SerializeRequest(request: IntegrationRequest) -> str

Serialize validated observation intent without resolving targets or credentials.

Source code in installed/flayer/integrations/wire.py

def SerializeRequest(request: IntegrationRequest) -> str:
    """Serialize validated observation intent without resolving targets or credentials."""

    if not isinstance(request, IntegrationRequest):
        raise IntegrationError("Integration serialization requires a validated request")

    return _Encode(_Metadata(request, "request"))

SerializeResult

SerializeResult(result: IntegrationResult) -> str

Serialize only explicitly allowed evidence, never source objects or raw adapter data.

Source code in installed/flayer/integrations/wire.py

def SerializeResult(result: IntegrationResult) -> str:
    """Serialize only explicitly allowed evidence, never source objects or raw adapter data."""

    if not isinstance(result, IntegrationResult):
        raise IntegrationError("Integration serialization requires a validated result")

    value = _Metadata(result.request, "result")
    evidence = result.evidence

    if isinstance(evidence, DiagnosticEvidence):
        value["evidence"] = {
            "status": evidence.Status.value,
            "checks": [{"status": check.status.value,
                        "metrics": {metric.name: metric.value for metric in check.metrics}}
                       for check in evidence.checks],
        }

    elif isinstance(evidence, StateEvidence):
        value["evidence"] = {"resource_count": evidence.resource_count}

    else:
        value["evidence"] = {
            "action": evidence.action, "status": evidence.status,
            "recovery_required": evidence.recovery_required,
            "resources": [{"status": resource.status, "count": resource.count}
                          for resource in evidence.resources],
        }

    return _Encode(value)