flayer.integrations.contracts

Module source

flayer.integrations.contracts

Consumer-neutral immutable envelopes for strictly allowlisted integration evidence.

MAX_WIRE_BYTES module-attribute

MAX_WIRE_BYTES = 64 * 1024

MAX_CHECKS module-attribute

MAX_CHECKS = 128

MAX_RESOURCES module-attribute

MAX_RESOURCES = 4096

REQUEST_ID_PATTERN module-attribute

REQUEST_ID_PATTERN = re.compile('[A-Za-z0-9][A-Za-z0-9_.-]{0,63}\\Z')

METRIC_BOUNDS module-attribute

METRIC_BOUNDS = {'duration_seconds': (0, 30), 'received_bytes': (0, 1024 * 1024), 'byte_limit': (1, 1024 * 1024), 'http_status': (100, 599), 'throughput_mbps': (0, 1000000000000)}

INTEGER_METRICS module-attribute

INTEGER_METRICS = frozenset({'received_bytes', 'byte_limit', 'http_status'})

NULLABLE_METRICS module-attribute

NULLABLE_METRICS = frozenset({'http_status', 'throughput_mbps'})

LIFECYCLE_ACTIONS module-attribute

LIFECYCLE_ACTIONS = frozenset({'status', 'create', 'destroy', 'recover'})

LIFECYCLE_STATUSES module-attribute

LIFECYCLE_STATUSES = frozenset({'complete', 'incomplete', 'uncertain', 'rolled-back', 'rollback-incomplete'})

RESOURCE_STATUSES module-attribute

RESOURCE_STATUSES = frozenset({'recorded', 'present', 'missing', 'not-recorded'})

Evidence module-attribute

Evidence: TypeAlias = DiagnosticEvidence | StateEvidence | LifecycleEvidence

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

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

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.

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

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

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, ...]

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

RequireIdentity

RequireIdentity(actual: StackIdentity, expected: StackIdentity) -> None

Require all ownership and provider scope fields to match independently held context.

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

def RequireIdentity(actual: StackIdentity, expected: StackIdentity) -> None:
    """Require all ownership and provider scope fields to match independently held context."""

    if not isinstance(expected, StackIdentity) or actual != expected:
        raise IntegrationError("Integration identity does not match the expected ownership boundary")

ValidateCount

ValidateCount(value: object) -> int

Reject coercion and bound recorded-resource and observation counts.

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

def ValidateCount(value: object) -> int:
    """Reject coercion and bound recorded-resource and observation counts."""

    if type(value) is not int or not 0 <= value <= MAX_RESOURCES:
        raise IntegrationError("Integration resource count exceeds its supported bounds")

    return value