Spaces:

GaryLeckey
/

aureon-os

Running

aureon-os / deployment /google-cloud /cloud_order_intent_bridge.py
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 7)
c626614 verified 2 days ago
Raw History Blame Contribute Delete
56.1 kB
"""Fail-closed owner order-intent bridge for the isolated cloud executor.

The bridge has one narrow authority: while the controller-signed release is
active, turn one fresh, strongly evidenced Unified Market Trader candidate into
the canonical ``aureon.cloud-trade-intent.v1`` message accepted by the private
Google Cloud authority chain.  It never loads exchange credentials and never
calls an exchange.  The Unified Market Trader's own publisher and executor must
remain held, which leaves the isolated cloud executor as the only mutation
boundary.
"""
from __future__ import annotations

import base64
import binascii
import hashlib
import json
import math
import os
import re
import sys
import tempfile
import time
from dataclasses import dataclass
from datetime import UTC, datetime
from decimal import Decimal, InvalidOperation, ROUND_DOWN
from pathlib import Path
from typing import Any, Callable, Mapping
from urllib.error import HTTPError, URLError
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey


ROOT = Path(__file__).resolve().parent
CONTRACT_ROOT = ROOT / "trading-control"
UPSTREAM_ROOT = ROOT.parents[1] / "aureon-os-upstream"
if str(UPSTREAM_ROOT) not in sys.path:
    sys.path.insert(0, str(UPSTREAM_ROOT))
if str(CONTRACT_ROOT) not in sys.path:
    sys.path.insert(0, str(CONTRACT_ROOT))

from contract import (  # noqa: E402
    ContractError,
    INTENT_SCHEMA,
    canonical_bytes,
    build_readiness_release_binding,
    controller_release_state_sha256,
    digest,
    unsigned_readiness_envelope,
    validate_execution_budget_readiness,
    validate_intent,
)
from cloud_state import (  # noqa: E402
    CloudStateError,
    validate_periodic_readiness_document,
)
from vm_workload_identity import (  # noqa: E402
    WorkloadIdentityError,
    broker_token,
)

from aureon.operator.authorization_lease import (  # noqa: E402
    MAX_LEASE_SECONDS,
    TRADING_CONTROL_SCOPE,
    build_operator_authorization_receipt,
)
from aureon.operator.cloud_trading_release import (  # noqa: E402
    CONTROLLER_KEY,
    LIVE_CONTROLS,
    read_controller_release,
)


PROJECT = "aureon-trading"
TOPIC = "aureon-trade-intents-v1"
TOPIC_RESOURCE = f"projects/{PROJECT}/topics/{TOPIC}"
PUBLISH_URL = f"https://pubsub.googleapis.com/v1/{TOPIC_RESOURCE}:publish"
METADATA_URL = (
    "http://metadata.google.internal/computeMetadata/v1/"
    "instance/service-accounts/default/token"
)
FIRESTORE_RELEASE_URL = (
    f"https://firestore.googleapis.com/v1/projects/{PROJECT}/databases/(default)/"
    "documents/aureon_trade_release/current"
)
FIRESTORE_READINESS_URL = (
    f"https://firestore.googleapis.com/v1/projects/{PROJECT}/databases/(default)/"
    "documents/aureon_trade_attestations/current_readiness"
)
KMS_PUBLIC_KEY_URL = "https://cloudkms.googleapis.com/v1/" + CONTROLLER_KEY + "/publicKey"
EXECUTOR_KEY = (
    "projects/aureon-trading/locations/europe-west1/keyRings/aureon-trading-prod/"
    "cryptoKeys/aureon-executor-signing/cryptoKeyVersions/1"
)
RECEIPT_KEY = (
    "projects/aureon-trading/locations/europe-west1/keyRings/aureon-trading-prod/"
    "cryptoKeys/aureon-receipt-signing/cryptoKeyVersions/1"
)
EXECUTOR_PUBLIC_KEY_URL = "https://cloudkms.googleapis.com/v1/" + EXECUTOR_KEY + "/publicKey"
RECEIPT_PUBLIC_KEY_URL = "https://cloudkms.googleapis.com/v1/" + RECEIPT_KEY + "/publicKey"

STATUS_SCHEMA = "aureon.cloud-order-intent-bridge-status.v1"
STATE_SCHEMA = "aureon.cloud-order-intent-bridge-state.v1"
EVIDENCE_SCHEMA = "aureon.owner-order-candidate-evidence.v2"
DEFAULT_RUNTIME = ROOT.parents[1] / "aureon-os-upstream" / "state" / "unified_runtime_status.json"
DEFAULT_STATE = Path("/var/lib/aureon-trade-bridge/order-intent-bridge-state.json")
DEFAULT_STATUS = Path(
    "/var/lib/aureon-control-exchange/trade-readback/latest-order-intent-bridge.json"
)

MAX_RUNTIME_BYTES = 16 * 1024 * 1024
MAX_RUNTIME_AGE_SECONDS = 30.0
MAX_OBSERVATION_AGE_SECONDS = 30.0
MAX_CLOCK_SKEW_SECONDS = 5.0
MIN_SIGNAL_CONFIDENCE = 0.70
MIN_CROSS_REFERENCES = 3
MAX_ORDER_USD = Decimal("5.00")
ROUTE_COOLDOWN_MS = 15 * 60 * 1000
STATE_RETENTION_MS = 24 * 60 * 60 * 1000
BRIDGE_INTERVAL_SECONDS = 10.0

LOCAL_EXECUTOR_HOLDS = {
    "AUREON_DEPLOYMENT_HOLD": "1",
    "AUREON_DISABLE_REAL_ORDERS": "1",
    "AUREON_DISABLE_EXCHANGE_MUTATIONS": "1",
    "AUREON_LIVE_TRADING": "0",
    "AUREON_ORDER_INTENT_PUBLISH": "0",
    "AUREON_UNIFIED_ORDER_EXECUTOR": "0",
}
LOCAL_PLAN_HOLDS = {
    "live_enabled": False,
    "real_orders_disabled": True,
    "exchange_mutations_disabled": True,
    "order_intent_publish_enabled": False,
    "executor_enabled": False,
}
SUPPORTED_ROUTE_TYPES = {
    "alpaca": "spot",
    "binance": "spot",
    "capital": "cfd",
    "kraken": "spot",
}
ROUTE_SYMBOL_FIELDS = {
    "alpaca": "alpaca_symbol",
    "binance": "binance_symbol",
    "capital": "capital_symbol",
    "kraken": "kraken_spot_symbol",
}
QUANTITY_DECIMALS = {"alpaca": 8, "binance": 8, "capital": 6, "kraken": 8}

_HEX64 = re.compile(r"[0-9a-f]{64}")
_MESSAGE_ID = re.compile(r"[0-9]{1,64}")
_REVISION = re.compile(r"[0-9a-f]{40,64}")
_SAFE_TEXT = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:/-]{0,127}")
_SYMBOL = re.compile(r"[A-Z0-9][A-Z0-9./_-]{1,31}")
_STATE_FIELDS = {"schema", "entries"}
_ENTRY_FIELDS = {
    "intent_id", "client_order_id", "intent_sha256", "candidate_fingerprint",
    "route_key", "reserved_at_ms", "published_at_ms", "pubsub_message_id", "status",
}


class BridgeError(RuntimeError):
    """A stable fail-closed bridge failure."""


class CandidateUnavailable(BridgeError):
    """No currently safe candidate; the bridge itself can still be healthy."""


class NoRedirects(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise BridgeError("cloud_redirect_denied")


@dataclass(frozen=True)
class _Response:
    status_code: int
    content: bytes

    def json(self) -> Any:
        return json.loads(self.content)


def _fixed_http(
    url: str,
    *,
    method: str = "GET",
    headers: Mapping[str, str] | None = None,
    body: bytes | None = None,
    timeout: float = 8.0,
    accepted_statuses: tuple[int, ...] = (200,),
) -> _Response:
    """Use only the fixed Google endpoints, without proxies or redirects."""

    allowed = {
        ("GET", METADATA_URL),
        ("GET", FIRESTORE_RELEASE_URL),
        ("GET", FIRESTORE_READINESS_URL),
        ("GET", KMS_PUBLIC_KEY_URL),
        ("GET", EXECUTOR_PUBLIC_KEY_URL),
        ("GET", RECEIPT_PUBLIC_KEY_URL),
        ("POST", PUBLISH_URL),
    }
    verb = str(method or "").upper()
    if (verb, url) not in allowed:
        raise BridgeError("cloud_endpoint_not_allowlisted")
    if body is not None and len(body) > 131_072:
        raise BridgeError("cloud_request_too_large")
    opener = build_opener(ProxyHandler({}), NoRedirects())
    try:
        with opener.open(
            Request(url, data=body, headers=dict(headers or {}), method=verb),
            timeout=timeout,
        ) as response:
            status = int(response.status)
            raw = response.read(1_048_577)
    except HTTPError as exc:
        status = int(exc.code)
        try:
            raw = exc.read(1_048_577)
        except OSError:
            raw = b""
    except (URLError, TimeoutError, OSError) as exc:
        raise BridgeError("cloud_http_request_failed") from exc
    if len(raw) > 1_048_576:
        raise BridgeError("cloud_http_response_too_large")
    if status not in accepted_statuses:
        raise BridgeError("cloud_http_status_invalid")
    return _Response(status, raw)


def access_token(reader: Callable[..., _Response] = _fixed_http) -> str:
    try:
        token = broker_token()
    except WorkloadIdentityError as exc:
        raise BridgeError(str(exc)) from exc
    if token is not None:
        return token
    response = reader(
        METADATA_URL,
        headers={"Metadata-Flavor": "Google"},
        timeout=5.0,
        accepted_statuses=(200,),
    )
    try:
        value = response.json()
    except (ValueError, TypeError, UnicodeError) as exc:
        raise BridgeError("metadata_token_response_invalid") from exc
    token = value.get("access_token") if isinstance(value, Mapping) else None
    expires = value.get("expires_in") if isinstance(value, Mapping) else None
    if (
        not isinstance(token, str)
        or not 20 <= len(token) <= 8192
        or type(expires) is not int
        or expires < 30
    ):
        raise BridgeError("metadata_token_response_invalid")
    return token


def _controller_state(
    *,
    reader: Callable[..., _Response] = _fixed_http,
) -> tuple[dict[str, Any] | None, str | None]:
    """Call the existing signature verifier through a fixed-endpoint adapter."""

    def requester(method: str, url: str, *, headers=None, timeout=8, **_kwargs):
        statuses = (200, 404) if url == FIRESTORE_RELEASE_URL else (200,)
        return reader(
            url,
            method=method,
            headers=headers,
            timeout=float(timeout),
            accepted_statuses=statuses,
        )

    try:
        return read_controller_release(
            environ={"AUREON_CONTROLLER_KEY_VERSION": CONTROLLER_KEY},
            requester=requester,
            token_provider=lambda: access_token(reader),
        )
    except BridgeError as exc:
        return None, str(exc)


def _decode_firestore_scalar(value: Any) -> Any:
    if not isinstance(value, Mapping):
        raise BridgeError("execution_budget_attestation_invalid")
    if "nullValue" in value:
        return None
    if "booleanValue" in value and type(value["booleanValue"]) is bool:
        return value["booleanValue"]
    if "integerValue" in value:
        try:
            return int(value["integerValue"])
        except (TypeError, ValueError, OverflowError) as exc:
            raise BridgeError("execution_budget_attestation_invalid") from exc
    if "stringValue" in value and isinstance(value["stringValue"], str):
        return value["stringValue"]
    raise BridgeError("execution_budget_attestation_invalid")


def _probe_state(
    *,
    reader: Callable[..., _Response] = _fixed_http,
    now_ms: int,
) -> tuple[dict[str, Any] | None, str | None]:
    try:
        token = access_token(reader)
        response = reader(
            FIRESTORE_READINESS_URL,
            method="GET",
            headers={"Authorization": "Bearer " + token, "Accept": "application/json"},
            timeout=8.0,
            accepted_statuses=(200, 404),
        )
        if response.status_code == 404:
            return None, "periodic_readiness_attestation_missing"
        value = response.json()
        if (
            not isinstance(value, Mapping)
            or not isinstance(value.get("fields"), Mapping)
            or not isinstance(value.get("updateTime"), str)
        ):
            raise BridgeError("execution_budget_attestation_invalid")
        row = {
            key: _decode_firestore_scalar(item)
            for key, item in value["fields"].items()
        }
        row["update_time"] = value["updateTime"]
        document = validate_periodic_readiness_document(row, now_ms=now_ms)
        executor_attestation = document["executor_attestation"]
        receipt_endorsement = document["receipt_endorsement"]
        for envelope, expected_key, url in (
            (executor_attestation, EXECUTOR_KEY, EXECUTOR_PUBLIC_KEY_URL),
            (receipt_endorsement, RECEIPT_KEY, RECEIPT_PUBLIC_KEY_URL),
        ):
            if envelope.get("key_version") != expected_key:
                raise BridgeError("periodic_readiness_key_pin_mismatch")
            public_response = reader(
                url,
                method="GET",
                headers={"Authorization": "Bearer " + token, "Accept": "application/json"},
                timeout=8.0,
                accepted_statuses=(200,),
            )
            public_value = public_response.json()
            if (
                not isinstance(public_value, Mapping)
                or public_value.get("name") != expected_key
                or public_value.get("algorithm") != "EC_SIGN_ED25519"
                or not isinstance(public_value.get("pem"), str)
            ):
                raise BridgeError("periodic_readiness_public_key_invalid")
            try:
                public = serialization.load_pem_public_key(public_value["pem"].encode("ascii"))
                signature = base64.b64decode(envelope["signature"], validate=True)
            except (ValueError, TypeError, UnicodeError, binascii.Error) as exc:
                raise BridgeError("periodic_readiness_signature_invalid") from exc
            if not isinstance(public, Ed25519PublicKey) or len(signature) != 64:
                raise BridgeError("periodic_readiness_signature_invalid")
            try:
                public.verify(
                    signature,
                    canonical_bytes(unsigned_readiness_envelope(envelope)),
                )
            except InvalidSignature as exc:
                raise BridgeError("periodic_readiness_signature_invalid") from exc
        token = ""
        return {
            key: value
            for key, value in document.items()
            if key not in {"executor_attestation", "receipt_endorsement"}
        }, None
    except (BridgeError, CloudStateError, ValueError, TypeError, UnicodeError) as exc:
        return None, (
            str(exc) if isinstance(exc, (BridgeError, CloudStateError))
            else "execution_budget_attestation_invalid"
        )


def _sha_text(value: str) -> str:
    return hashlib.sha256(value.encode("ascii")).hexdigest()


def _finite_number(value: Any, *, positive: bool = False) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise CandidateUnavailable("candidate_number_invalid")
    number = float(value)
    if not math.isfinite(number) or (positive and number <= 0):
        raise CandidateUnavailable("candidate_number_invalid")
    return number


def _timestamp_ms(value: Any, *, code: str) -> int:
    if isinstance(value, bool):
        raise CandidateUnavailable(code)
    if isinstance(value, (int, float)):
        number = float(value)
    elif isinstance(value, str) and 1 <= len(value) <= 64:
        try:
            parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
        except ValueError as exc:
            raise CandidateUnavailable(code) from exc
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=UTC)
        number = parsed.timestamp()
    else:
        raise CandidateUnavailable(code)
    if not math.isfinite(number) or number <= 0:
        raise CandidateUnavailable(code)
    return int(number * 1000)


def _safe_text(value: Any, *, code: str) -> str:
    text = str(value or "")
    if not _SAFE_TEXT.fullmatch(text):
        raise CandidateUnavailable(code)
    return text


def _read_json_file(path: Path, *, limit: int, missing_code: str, invalid_code: str) -> tuple[dict[str, Any], bytes, os.stat_result]:
    if not path.is_absolute() or path.is_symlink():
        raise BridgeError(invalid_code)
    try:
        resolved = path.resolve(strict=True)
        stat = resolved.stat()
    except OSError as exc:
        raise BridgeError(missing_code) from exc
    if not resolved.is_file() or not 0 < stat.st_size <= limit or resolved != path.resolve():
        raise BridgeError(invalid_code)
    try:
        raw = resolved.read_bytes()
        value = json.loads(raw)
    except (OSError, ValueError, TypeError, UnicodeError) as exc:
        raise BridgeError(invalid_code) from exc
    if not isinstance(value, dict):
        raise BridgeError(invalid_code)
    return value, raw, stat


def _durable_write(path: Path, value: Mapping[str, Any], *, shared: bool = False) -> None:
    if not path.is_absolute() or path.is_symlink():
        raise BridgeError("bridge_output_path_invalid")
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        parent = path.parent.resolve(strict=True)
    except OSError as exc:
        raise BridgeError("bridge_output_path_invalid") from exc
    if path.resolve().parent != parent:
        raise BridgeError("bridge_output_path_invalid")
    try:
        raw = json.dumps(
            value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False,
        ).encode("ascii") + b"\n"
    except (TypeError, ValueError, UnicodeError) as exc:
        raise BridgeError("bridge_output_invalid") from exc
    handle, temporary = tempfile.mkstemp(prefix=".intent-bridge-", dir=parent)
    try:
        with os.fdopen(handle, "wb") as stream:
            stream.write(raw)
            stream.flush()
            os.fsync(stream.fileno())
        os.chmod(temporary, 0o640 if shared else 0o600)
        os.replace(temporary, path)
        if os.name == "posix":
            directory = os.open(parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
            try:
                os.fsync(directory)
            finally:
                os.close(directory)
    except OSError as exc:
        raise BridgeError("bridge_output_write_failed") from exc
    finally:
        try:
            os.unlink(temporary)
        except FileNotFoundError:
            pass


def _empty_state() -> dict[str, Any]:
    return {"schema": STATE_SCHEMA, "entries": []}


def _load_state(path: Path, *, now_ms: int) -> dict[str, Any]:
    if not path.exists():
        if path.is_symlink():
            raise BridgeError("bridge_state_invalid")
        return _empty_state()
    value, _raw, _stat = _read_json_file(
        path, limit=262_144, missing_code="bridge_state_missing", invalid_code="bridge_state_invalid",
    )
    entries = value.get("entries")
    if set(value) != _STATE_FIELDS or value.get("schema") != STATE_SCHEMA or not isinstance(entries, list) or len(entries) > 256:
        raise BridgeError("bridge_state_invalid")
    retained: list[dict[str, Any]] = []
    for item in entries:
        if not isinstance(item, Mapping) or set(item) != _ENTRY_FIELDS:
            raise BridgeError("bridge_state_invalid")
        row = dict(item)
        if (
            not _HEX64.fullmatch(str(row["intent_id"] or ""))
            or not _HEX64.fullmatch(str(row["client_order_id"] or ""))
            or not _HEX64.fullmatch(str(row["intent_sha256"] or ""))
            or not _HEX64.fullmatch(str(row["candidate_fingerprint"] or ""))
            or not _SAFE_TEXT.fullmatch(str(row["route_key"] or ""))
            or type(row["reserved_at_ms"]) is not int
            or row["reserved_at_ms"] <= 0
            or row["status"] not in {"pending", "published"}
            or (row["status"] == "pending" and (row["published_at_ms"] is not None or row["pubsub_message_id"] is not None))
            or (
                row["status"] == "published"
                and (
                    type(row["published_at_ms"]) is not int
                    or row["published_at_ms"] < row["reserved_at_ms"]
                    or not _MESSAGE_ID.fullmatch(str(row["pubsub_message_id"] or ""))
                )
            )
        ):
            raise BridgeError("bridge_state_invalid")
        if now_ms - row["reserved_at_ms"] <= STATE_RETENTION_MS:
            retained.append(row)
    return {"schema": STATE_SCHEMA, "entries": retained}


def _validate_local_holds(payload: Mapping[str, Any], environ: Mapping[str, str]) -> Mapping[str, Any]:
    if any(environ.get(key) != expected for key, expected in LOCAL_EXECUTOR_HOLDS.items()):
        raise BridgeError("bridge_service_local_holds_not_enforced")
    plan = payload.get("exchange_action_plan")
    if not isinstance(plan, Mapping) or any(plan.get(key) is not expected for key, expected in LOCAL_PLAN_HOLDS.items()):
        raise BridgeError("unified_market_trader_local_executor_not_held")
    blockers = plan.get("global_blockers")
    required = {
        "live_trading_not_enabled", "real_orders_disabled",
        "exchange_mutations_disabled", "order_intent_publish_disabled",
    }
    if not isinstance(blockers, list) or not required.issubset({str(item) for item in blockers}):
        raise BridgeError("unified_market_trader_local_executor_not_held")
    return plan


def _validate_authorization(value: Any, *, now_ms: int) -> dict[str, Any]:
    if not isinstance(value, Mapping):
        raise BridgeError("operator_authorization_invalid")
    event = value.get("ledger_event_sha256")
    authorized = value.get("authorized_at")
    expires = value.get("expires_at")
    if (
        value.get("active") is not True
        or value.get("state") != "active"
        or value.get("account_scope") != "owner_cloud_account"
        or tuple(value.get("scope") or ()) != TRADING_CONTROL_SCOPE
        or not _HEX64.fullmatch(str(event or ""))
        or isinstance(authorized, bool)
        or not isinstance(authorized, (int, float))
        or isinstance(expires, bool)
        or not isinstance(expires, (int, float))
        or not math.isfinite(float(authorized))
        or not math.isfinite(float(expires))
        or int(float(authorized) * 1000) > now_ms
        or int(float(expires) * 1000) <= now_ms
        or float(expires) - float(authorized) > MAX_LEASE_SECONDS
    ):
        raise BridgeError("operator_authorization_invalid")
    return {
        "event_sha256": str(event),
        "expires_at_ms": int(float(expires) * 1000),
        "scope_sha256": hashlib.sha256(
            json.dumps(list(TRADING_CONTROL_SCOPE), separators=(",", ":")).encode("ascii")
        ).hexdigest(),
    }


def _validate_release(
    value: Any,
    *,
    authorization: Mapping[str, Any],
    revision: str,
    now_ms: int,
) -> dict[str, Any]:
    if not isinstance(value, Mapping):
        raise BridgeError("controller_release_missing")
    row = dict(value)
    if (
        row.get("state") != "active"
        or row.get("requested_controls_sha256") != digest(LIVE_CONTROLS)
        or row.get("authorization_event_sha256") != authorization["event_sha256"]
        or row.get("authorization_expires_at_ms") != authorization["expires_at_ms"]
        or row.get("source_revision") != revision
        or row.get("controller_key_version") != CONTROLLER_KEY
        or type(row.get("signed_at_ms")) is not int
        or row["signed_at_ms"] > now_ms + int(MAX_CLOCK_SKEW_SECONDS * 1000)
        or row.get("authorization_expires_at_ms", 0) <= now_ms + 60_000
    ):
        raise BridgeError("controller_release_not_current_for_owner_runtime")
    return row


def _validate_execution_budget_attestation(
    value: Any,
    *,
    release: Mapping[str, Any],
    authorization: Mapping[str, Any],
    revision: str,
    now_ms: int,
) -> dict[str, Any]:
    try:
        row = validate_periodic_readiness_document(value, now_ms=now_ms)
        binding = build_readiness_release_binding(release)
    except CloudStateError as exc:
        raise BridgeError(str(exc)) from exc
    except ContractError as exc:
        raise BridgeError("execution_budget_attestation_release_join_invalid") from exc
    if (
        row["release_generation"] != release.get("generation")
        or row["release_request_id"] != release.get("request_id")
        or row["release_state_sha256"] != controller_release_state_sha256(release)
        or row["release_state_sha256"] != binding["release_state_sha256"]
        or row["authorization_event_sha256"] != authorization["event_sha256"]
        or row["authorization_expires_at_ms"] != authorization["expires_at_ms"]
        or row["source_revision"] != revision
        or row["source_revision"] != release.get("source_revision")
        or row["observed_at_ms"] < int(release.get("signed_at_ms") or 0) - 5_000
        or row["expires_at_ms"] > int(release.get("authorization_expires_at_ms") or 0)
    ):
        raise BridgeError("execution_budget_attestation_release_join_invalid")
    try:
        readiness = row["executor_attestation"]["statement"][
            "execution_budget_readiness"
        ]
        readiness = validate_execution_budget_readiness(
            readiness,
            source_revision=revision,
            now_ms=now_ms,
        )
    except (ContractError, ValueError, TypeError, UnicodeError) as exc:
        raise BridgeError("execution_budget_attestation_invalid") from exc
    if Decimal(readiness["maximum_order_usd"]) != MAX_ORDER_USD:
        raise BridgeError("execution_budget_cap_mismatch")
    return readiness


def _route_quote_asset(venue: str, symbol: str) -> str:
    if venue in {"alpaca", "capital"}:
        return "USD"
    normalized = symbol.upper().replace("-", "/").replace("_", "/")
    if "/" in normalized:
        quote = normalized.rsplit("/", 1)[-1]
    else:
        quote = next(
            (asset for asset in ("FDUSD", "USDT", "USDC", "USD") if normalized.endswith(asset)),
            "",
        )
    quote = {"ZUSD": "USD"}.get(quote, quote)
    if quote not in {"FDUSD", "USDT", "USDC", "USD"}:
        raise CandidateUnavailable("candidate_route_quote_asset_unsupported")
    return quote


def _runtime_payload(path: Path, *, now_ms: int) -> tuple[dict[str, Any], str, int]:
    payload, raw, stat = _read_json_file(
        path, limit=MAX_RUNTIME_BYTES,
        missing_code="unified_runtime_status_missing", invalid_code="unified_runtime_status_invalid",
    )
    generated_ms = _timestamp_ms(payload.get("generated_at"), code="unified_runtime_timestamp_invalid")
    file_age_ms = now_ms - int(stat.st_mtime * 1000)
    generated_age_ms = now_ms - generated_ms
    if (
        payload.get("ok") is not True
        or payload.get("source") != "unified-market-trader"
        or payload.get("partial") is True
        or file_age_ms < -5_000
        or file_age_ms > int(MAX_RUNTIME_AGE_SECONDS * 1000)
        or generated_age_ms < -5_000
        or generated_age_ms > int(MAX_RUNTIME_AGE_SECONDS * 1000)
    ):
        raise BridgeError("unified_runtime_status_not_fresh_complete")
    gate = payload.get("trading_gate_evidence")
    hnc = payload.get("hnc_cognitive_proof")
    if not isinstance(gate, Mapping) or gate.get("analysis_ready") is not True:
        raise CandidateUnavailable("trading_gate_analysis_not_ready")
    if (
        not isinstance(hnc, Mapping)
        or hnc.get("status") != "passing"
        or not isinstance(hnc.get("real_data"), Mapping)
        or hnc["real_data"].get("passed") is not True
        or type(hnc.get("passed_count")) is not int
        or type(hnc.get("step_count")) is not int
        or hnc["step_count"] < 1
        or hnc["passed_count"] != hnc["step_count"]
    ):
        raise CandidateUnavailable("hnc_cognitive_proof_not_passing")
    return payload, hashlib.sha256(raw).hexdigest(), generated_ms


def _candidate_evidence(
    candidate: Mapping[str, Any],
    route: Mapping[str, Any],
    *,
    runtime_generated_at_ms: int,
    now_ms: int,
    execution_budget_readiness: Mapping[str, Any],
) -> dict[str, Any]:
    symbol = str(candidate.get("symbol") or "").upper()
    side = str(candidate.get("side") or "").upper()
    venue = str(route.get("venue") or "").lower()
    route_symbol = str(route.get("symbol") or "").upper()
    market_type = str(route.get("market_type") or "").lower()
    if not _SYMBOL.fullmatch(symbol):
        raise CandidateUnavailable("candidate_symbol_invalid")
    if not _SYMBOL.fullmatch(route_symbol):
        raise CandidateUnavailable("candidate_route_symbol_invalid")
    if side not in {"BUY", "SELL"}:
        raise CandidateUnavailable("candidate_side_invalid")
    if SUPPORTED_ROUTE_TYPES.get(venue) != market_type:
        raise CandidateUnavailable("candidate_route_not_supported_by_cloud_adapter")
    binding_field = ROUTE_SYMBOL_FIELDS.get(venue)
    bound_route_symbol = (
        str(candidate.get(binding_field) or "").upper() if binding_field is not None else ""
    )
    if bound_route_symbol != route_symbol:
        raise CandidateUnavailable("candidate_route_symbol_binding_mismatch")
    # Spot sells need a fresh base-asset balance proof that the current runtime
    # candidate does not carry.  Capital CFDs can express either direction.
    if side == "SELL" and venue != "capital":
        raise CandidateUnavailable("spot_sell_asset_balance_evidence_missing")

    confidence = _finite_number(candidate.get("confidence"))
    support_count = candidate.get("support_count")
    scanner = candidate.get("scanner_fusion")
    if (
        confidence < MIN_SIGNAL_CONFIDENCE
        or type(support_count) is not int
        or support_count < 1
        or not isinstance(scanner, Mapping)
        or scanner.get("usable_for_decision") is not True
        or scanner.get("fed_to_decision_logic") is not True
        or type(scanner.get("cross_reference_count")) is not int
        or scanner["cross_reference_count"] < MIN_CROSS_REFERENCES
        or _finite_number(scanner.get("scanner_fusion_score")) <= 0
    ):
        raise CandidateUnavailable("candidate_signal_evidence_not_strong")
    if candidate.get("data_status") != "live" or candidate.get("actionable") is not True:
        raise CandidateUnavailable("candidate_not_live_actionable")
    if candidate.get("intent_publish_blockers") != []:
        raise CandidateUnavailable("candidate_has_publish_blockers")

    required_empty = ("blockers", "runtime_clearances", "guards", "clearance_required")
    if (
        route.get("ready") is not True
        or route.get("end_user_trade_available") is not True
        or route.get("trade_clearance_state") != "available"
        or route.get("guard_state") != "available"
        or any(route.get(key) != [] for key in required_empty)
        or route.get("model_coverage_ready") is not True
        or type(route.get("model_count")) is not int
        or route["model_count"] < 1
    ):
        raise CandidateUnavailable("candidate_route_evidence_not_ready")
    route_key = _safe_text(route.get("route_key"), code="candidate_route_key_invalid")
    expected_route_key = f"{venue}:{market_type}:{route_symbol}:{side}"
    if route_key != expected_route_key:
        raise CandidateUnavailable("candidate_route_key_binding_mismatch")

    venue_budget = execution_budget_readiness["venues"].get(venue)
    quote_asset = _route_quote_asset(venue, route_symbol)
    quote_budget = (
        venue_budget.get("quotes", {}).get(quote_asset)
        if isinstance(venue_budget, Mapping)
        else None
    )
    if (
        not isinstance(venue_budget, Mapping)
        or venue_budget.get("authenticated") is not True
        or not isinstance(quote_budget, Mapping)
        or quote_budget.get("status") != "sufficient"
        or Decimal(str(quote_budget.get("allowance_usd") or "0")) < MAX_ORDER_USD
    ):
        raise CandidateUnavailable("candidate_route_execution_budget_not_ready")

    reference_price = _finite_number(candidate.get("reference_price"), positive=True)
    source_prices = candidate.get("source_prices")
    observations = candidate.get("market_observations")
    if not isinstance(source_prices, Mapping) or not source_prices or len(source_prices) > 32:
        raise CandidateUnavailable("provider_price_evidence_missing")
    if not isinstance(observations, list) or not observations or len(observations) > 64:
        raise CandidateUnavailable("provider_timestamp_evidence_missing")
    prices: dict[str, float] = {}
    for key, value in source_prices.items():
        source_name = _safe_text(key, code="provider_price_source_invalid")
        prices[source_name] = _finite_number(value, positive=True)
    average = sum(prices.values()) / len(prices)
    if abs(reference_price - average) > max(1e-8, abs(average) * 1e-6):
        raise CandidateUnavailable("reference_price_not_bound_to_provider_prices")
    observation_rows: list[dict[str, Any]] = []
    feeds: set[str] = set()
    routed_provider_observations: list[dict[str, Any]] = []
    for value in observations:
        if not isinstance(value, Mapping):
            raise CandidateUnavailable("provider_timestamp_evidence_invalid")
        source = _safe_text(value.get("source"), code="provider_timestamp_source_invalid")
        feed = _safe_text(value.get("feed"), code="provider_timestamp_source_invalid")
        observation_symbol = str(value.get("symbol") or "").upper()
        if observation_symbol != symbol or not _SYMBOL.fullmatch(observation_symbol):
            raise CandidateUnavailable("provider_price_symbol_join_invalid")
        provider_symbol = str(value.get("provider_symbol") or "").upper()
        if not _SYMBOL.fullmatch(provider_symbol):
            raise CandidateUnavailable("provider_price_provider_symbol_invalid")
        observation_price = _finite_number(value.get("price"), positive=True)
        if feed not in prices or observation_price != prices[feed] or feed in feeds:
            raise CandidateUnavailable("provider_price_observation_join_invalid")
        source_ms = _timestamp_ms(value.get("source_timestamp"), code="provider_timestamp_invalid")
        max_age = _finite_number(value.get("max_age_sec"), positive=True)
        allowed_age = min(max_age, MAX_OBSERVATION_AGE_SECONDS)
        age_ms = now_ms - source_ms
        if age_ms < -int(MAX_CLOCK_SKEW_SECONDS * 1000) or age_ms > int(allowed_age * 1000):
            raise CandidateUnavailable("provider_timestamp_not_fresh")
        feeds.add(feed)
        row = {
            "feed": feed,
            "source": source,
            "symbol": observation_symbol,
            "provider_symbol": provider_symbol,
            "price": observation_price,
            "source_timestamp_ms": source_ms,
            "max_age_ms": int(allowed_age * 1000),
        }
        observation_rows.append(row)
        if source.lower() == venue or feed.lower() == venue:
            if provider_symbol != route_symbol:
                raise CandidateUnavailable("route_provider_price_symbol_join_invalid")
            routed_provider_observations.append(row)
    if set(prices) != feeds:
        raise CandidateUnavailable("provider_price_timestamp_join_missing")
    if not routed_provider_observations:
        raise CandidateUnavailable("route_provider_price_observation_missing")

    active_scanners = scanner.get("active_scanners")
    if not isinstance(active_scanners, list) or len(active_scanners) < MIN_CROSS_REFERENCES:
        raise CandidateUnavailable("candidate_scanner_evidence_missing")
    scanner_names = sorted({_safe_text(item, code="candidate_scanner_invalid") for item in active_scanners})
    evidence = {
        "schema": EVIDENCE_SCHEMA,
        "runtime_generated_at_ms": runtime_generated_at_ms,
        "candidate_id": _safe_text(candidate.get("candidate_id"), code="candidate_id_invalid"),
        "lifecycle_id": _safe_text(candidate.get("lifecycle_id"), code="candidate_lifecycle_id_invalid"),
        "symbol": symbol,
        "side": side,
        "confidence": confidence,
        "support_count": support_count,
        "reference_price": reference_price,
        "source_prices": dict(sorted(prices.items())),
        "market_observations": sorted(
            observation_rows,
            key=lambda row: (row["source_timestamp_ms"], row["feed"], row["source"]),
        ),
        "scanner_fusion": {
            "score": _finite_number(scanner["scanner_fusion_score"]),
            "cross_reference_count": scanner["cross_reference_count"],
            "active_scanners": scanner_names,
        },
        "route": {
            "venue": venue,
            "market_type": market_type,
            "symbol": route_symbol,
            "route_key": route_key,
            "model_count": route["model_count"],
            "cash_capability_state": _safe_text(
                route.get("cash_capability_state"), code="candidate_cash_capability_invalid",
            ),
            "execution_budget_quote_asset": quote_asset,
            "execution_budget_allowance_usd": float(quote_budget["allowance_usd"]),
            "execution_budget_readiness_sha256": digest(execution_budget_readiness),
            "candidate_symbol_binding_field": binding_field,
            "provider_observation_sha256": digest({
                "schema": "aureon.route-provider-price-observation-set.v1",
                "venue": venue,
                "candidate_symbol": symbol,
                "route_symbol": route_symbol,
                "observations": sorted(
                    routed_provider_observations,
                    key=lambda row: (
                        row["source_timestamp_ms"], row["feed"], row["source"],
                    ),
                ),
            }),
        },
    }
    canonical_bytes(evidence)
    return evidence


def select_candidate(
    payload: Mapping[str, Any],
    *,
    runtime_generated_at_ms: int,
    now_ms: int,
    execution_budget_readiness: Mapping[str, Any],
) -> dict[str, Any]:
    try:
        execution_budget_readiness = validate_execution_budget_readiness(
            execution_budget_readiness,
            source_revision=str(execution_budget_readiness.get("source_revision") or ""),
            now_ms=now_ms,
        )
    except (AttributeError, ContractError) as exc:
        raise CandidateUnavailable("execution_budget_attestation_invalid") from exc
    flow = payload.get("shared_order_flow")
    if not isinstance(flow, Mapping):
        raise CandidateUnavailable("shared_order_flow_missing")
    generated_ms = _timestamp_ms(flow.get("generated_at"), code="shared_order_flow_timestamp_invalid")
    if abs(generated_ms - runtime_generated_at_ms) > int(MAX_RUNTIME_AGE_SECONDS * 1000):
        raise CandidateUnavailable("shared_order_flow_not_current_tick")
    candidates = flow.get("active_order_flow")
    if not isinstance(candidates, list):
        raise CandidateUnavailable("qualifying_candidate_unavailable")

    ranked: list[tuple[tuple[Any, ...], dict[str, Any]]] = []
    reason_codes: set[str] = set()
    for candidate in candidates[:20]:
        if not isinstance(candidate, Mapping):
            continue
        routes = candidate.get("execution_routes")
        if not isinstance(routes, list):
            continue
        for route in routes[:12]:
            if not isinstance(route, Mapping):
                continue
            try:
                evidence = _candidate_evidence(
                    candidate, route,
                    runtime_generated_at_ms=runtime_generated_at_ms,
                    now_ms=now_ms,
                    execution_budget_readiness=execution_budget_readiness,
                )
            except CandidateUnavailable as exc:
                reason_codes.add(str(exc))
                continue
            selection_rank = candidate.get("selection_rank")
            rank = selection_rank if type(selection_rank) is int and selection_rank > 0 else 1_000_000
            score = evidence["scanner_fusion"]["score"]
            ranked.append((
                (
                    rank,
                    -float(evidence["confidence"]),
                    -float(score),
                    evidence["route"]["venue"],
                    evidence["route"]["route_key"],
                ),
                evidence,
            ))
    if not ranked:
        if len(reason_codes) == 1:
            raise CandidateUnavailable(next(iter(reason_codes)))
        raise CandidateUnavailable("qualifying_candidate_unavailable")
    ranked.sort(key=lambda item: item[0])
    return ranked[0][1]


def _quantity(price: float, venue: str) -> tuple[float, float]:
    places = QUANTITY_DECIMALS[venue]
    try:
        price_decimal = Decimal(str(price))
        step = Decimal(1).scaleb(-places)
        quantity = (MAX_ORDER_USD / price_decimal).quantize(step, rounding=ROUND_DOWN)
        sized_notional = quantity * price_decimal
        quote = MAX_ORDER_USD
    except (InvalidOperation, ZeroDivisionError) as exc:
        raise CandidateUnavailable("venue_safe_quantity_unavailable") from exc
    if (quantity <= 0 or sized_notional <= 0 or sized_notional > MAX_ORDER_USD
            or quote <= 0 or quote > MAX_ORDER_USD or quantity > Decimal("1000000000")):
        raise CandidateUnavailable("venue_safe_quantity_unavailable")
    return float(quantity), float(quote)


def build_intent(
    evidence: Mapping[str, Any],
    *,
    authorization: Mapping[str, Any],
    release: Mapping[str, Any],
    revision: str,
    runtime_sha256: str,
    now_ms: int,
) -> tuple[dict[str, Any], str]:
    created_at_ms = int(evidence["runtime_generated_at_ms"])
    expires_at_ms = created_at_ms + 60_000
    if created_at_ms > now_ms + 5_000 or expires_at_ms < now_ms + 1_000:
        raise CandidateUnavailable("candidate_intent_lifetime_insufficient")
    if authorization["expires_at_ms"] < expires_at_ms:
        raise CandidateUnavailable("operator_authorization_expires_before_intent")
    market_sha = digest(evidence)
    route = evidence["route"]
    quantity, quote = _quantity(float(evidence["reference_price"]), str(route["venue"]))
    seed = {
        "schema": "aureon.cloud-order-intent-id-seed.v1",
        "source_revision": revision,
        "authorization_event_sha256": authorization["event_sha256"],
        "authorization_expires_at_ms": authorization["expires_at_ms"],
        "market_evidence_sha256": market_sha,
        "venue": route["venue"],
        "symbol": route["symbol"],
        "side": evidence["side"],
        "quantity": quantity,
        "quote_value_usd": quote,
    }
    seed_sha = digest(seed)
    intent_id = _sha_text("intent:" + seed_sha)
    client_order_id = _sha_text("client:" + seed_sha)
    release_measurement = {
        key: release[key]
        for key in (
            "schema", "state", "generation", "request_id", "requested_controls_sha256",
            "authorization_event_sha256", "authorization_expires_at_ms", "source_revision",
            "signed_at_ms", "controller_key_version", "signature", "update_time",
        )
    }
    runtime_measurement = {
        "schema": "aureon.cloud-order-runtime-measurement.v1",
        "source_revision": revision,
        "runtime_status_sha256": runtime_sha256,
        "controller_release_sha256": digest(release_measurement),
        "local_executor_held": True,
    }
    intent = {
        "schema": INTENT_SCHEMA,
        "intent_id": intent_id,
        "kind": "order",
        "tenant_commitment": _sha_text("owner_cloud_account"),
        "account_scope": "owner",
        "created_at_ms": created_at_ms,
        "expires_at_ms": expires_at_ms,
        "authorization": {
            "active": True,
            "expires_at_ms": authorization["expires_at_ms"],
            "event_sha256": authorization["event_sha256"],
            "scope_sha256": authorization["scope_sha256"],
            "release_authority": False,
            "mutation_authority": False,
        },
        "requested_controls": dict(LIVE_CONTROLS),
        "market_evidence_sha256": market_sha,
        "runtime_measurement_sha256": digest(runtime_measurement),
        "source_revision": revision,
        "order": {
            "venue": route["venue"],
            "symbol": route["symbol"],
            "side": evidence["side"],
            "order_type": "MARKET",
            "quantity": quantity,
            "quote_value_usd": quote,
            "limit_price": None,
            "client_order_id": client_order_id,
        },
        "dry_run": False,
    }
    validated = validate_intent(
        intent, now_ms=now_ms, max_order_usd=float(MAX_ORDER_USD), live_release=True,
    )
    if canonical_bytes(validated) != canonical_bytes(intent):
        raise BridgeError("intent_contract_normalization_changed_payload")
    return intent, digest(evidence)


def publish_intent(
    intent: Mapping[str, Any],
    *,
    reader: Callable[..., _Response] = _fixed_http,
    current_time: Callable[[], float] = time.time,
) -> str:
    try:
        now_ms = int(float(current_time()) * 1000)
    except (TypeError, ValueError, OverflowError) as exc:
        raise BridgeError("bridge_clock_invalid") from exc
    try:
        validated = validate_intent(
            intent, now_ms=now_ms, max_order_usd=float(MAX_ORDER_USD), live_release=True,
        )
        wire = canonical_bytes(validated)
    except Exception as exc:
        if isinstance(exc, BridgeError):
            raise
        raise BridgeError("intent_contract_validation_failed") from exc
    if wire != canonical_bytes(intent):
        raise BridgeError("intent_contract_normalization_changed_payload")
    token = access_token(reader)
    envelope = {
        "messages": [{
            "data": base64.b64encode(wire).decode("ascii"),
            "attributes": {
                "schema": INTENT_SCHEMA,
                "kind": "order",
                "intent_id": intent["intent_id"],
            },
            "orderingKey": intent["intent_id"],
        }],
    }
    response = reader(
        PUBLISH_URL,
        method="POST",
        headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"},
        body=json.dumps(envelope, sort_keys=True, separators=(",", ":")).encode("ascii"),
        timeout=15.0,
        accepted_statuses=(200,),
    )
    token = ""
    try:
        value = response.json()
    except (ValueError, TypeError, UnicodeError) as exc:
        raise BridgeError("intent_publish_response_invalid") from exc
    ids = value.get("messageIds") if isinstance(value, Mapping) else None
    if (
        not isinstance(ids, list)
        or len(ids) != 1
        or not isinstance(ids[0], str)
        or not _MESSAGE_ID.fullmatch(ids[0])
    ):
        raise BridgeError("intent_publish_response_invalid")
    return ids[0]


def _status(
    *,
    now_ms: int,
    revision: str,
    state: str,
    operational: bool,
    candidate_ready: bool,
    blockers: list[str],
    source: Mapping[str, Any] | None = None,
    release: Mapping[str, Any] | None = None,
    last_publish: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    return {
        "schema": STATUS_SCHEMA,
        "observed_at": now_ms / 1000.0,
        "revision": revision,
        "state": state,
        "operational": operational,
        "candidate_ready": candidate_ready,
        "blockers": list(dict.fromkeys(blockers)),
        "source": dict(source) if isinstance(source, Mapping) else None,
        "controller_release": dict(release) if isinstance(release, Mapping) else None,
        "last_publish": dict(last_publish) if isinstance(last_publish, Mapping) else None,
        "release_authority": bool(release),
        "mutation_authority": False,
    }


@dataclass
class OrderIntentBridge:
    revision: str
    runtime_path: Path = DEFAULT_RUNTIME
    state_path: Path = DEFAULT_STATE
    status_path: Path = DEFAULT_STATUS
    environ: Mapping[str, str] | None = None
    current_time: Callable[[], float] = time.time
    authorization_builder: Callable[..., Mapping[str, Any]] = build_operator_authorization_receipt
    controller_reader: Callable[..., tuple[dict[str, Any] | None, str | None]] = _controller_state
    attestation_reader: Callable[..., tuple[dict[str, Any] | None, str | None]] = _probe_state
    publisher: Callable[..., str] = publish_intent
    http_reader: Callable[..., _Response] = _fixed_http

    def __post_init__(self) -> None:
        if not _REVISION.fullmatch(str(self.revision or "")):
            raise BridgeError("bridge_revision_invalid")

    def cycle(self) -> dict[str, Any]:
        try:
            now = float(self.current_time())
        except (TypeError, ValueError, OverflowError) as exc:
            raise BridgeError("bridge_clock_invalid") from exc
        if not math.isfinite(now) or now <= 0:
            raise BridgeError("bridge_clock_invalid")
        now_ms = int(now * 1000)
        env = os.environ if self.environ is None else self.environ
        source_summary: dict[str, Any] | None = None
        release_summary: dict[str, Any] | None = None
        operational = False
        try:
            payload, runtime_sha, generated_ms = _runtime_payload(self.runtime_path, now_ms=now_ms)
            _validate_local_holds(payload, env)
            source_summary = {
                "runtime_status_sha256": runtime_sha,
                "generated_at_ms": generated_ms,
                "age_ms": now_ms - generated_ms,
            }
            authorization = _validate_authorization(self.authorization_builder(now=now), now_ms=now_ms)
            release, release_error = self.controller_reader(reader=self.http_reader)
            if release is None:
                raise BridgeError(release_error or "controller_release_missing")
            release = _validate_release(
                release, authorization=authorization, revision=self.revision, now_ms=now_ms,
            )
            attestation, attestation_error = self.attestation_reader(
                reader=self.http_reader,
                now_ms=now_ms,
            )
            if attestation is None:
                raise BridgeError(attestation_error or "execution_budget_attestation_missing")
            execution_budget_readiness = _validate_execution_budget_attestation(
                attestation,
                release=release,
                authorization=authorization,
                revision=self.revision,
                now_ms=now_ms,
            )
            release_summary = {
                "state": "active",
                "generation": release["generation"],
                "request_id": release["request_id"],
                "source_revision": release["source_revision"],
                "authorization_event_sha256": release["authorization_event_sha256"],
                "authorization_expires_at_ms": release["authorization_expires_at_ms"],
            }
            operational = True
            evidence = select_candidate(
                payload,
                runtime_generated_at_ms=generated_ms,
                now_ms=now_ms,
                execution_budget_readiness=execution_budget_readiness,
            )
            intent, candidate_fingerprint = build_intent(
                evidence,
                authorization=authorization,
                release=release,
                revision=self.revision,
                runtime_sha256=runtime_sha,
                now_ms=now_ms,
            )
            route_key = evidence["route"]["route_key"]
            state = _load_state(self.state_path, now_ms=now_ms)
            if any(
                row["intent_id"] == intent["intent_id"]
                or row["candidate_fingerprint"] == candidate_fingerprint
                for row in state["entries"]
            ):
                raise CandidateUnavailable("intent_replay_blocked")
            if any(
                row["route_key"] == route_key
                and now_ms - row["reserved_at_ms"] < ROUTE_COOLDOWN_MS
                for row in state["entries"]
            ):
                raise CandidateUnavailable("route_cooldown_active")

            intent_sha = hashlib.sha256(canonical_bytes(intent)).hexdigest()
            pending = {
                "intent_id": intent["intent_id"],
                "client_order_id": intent["order"]["client_order_id"],
                "intent_sha256": intent_sha,
                "candidate_fingerprint": candidate_fingerprint,
                "route_key": route_key,
                "reserved_at_ms": now_ms,
                "published_at_ms": None,
                "pubsub_message_id": None,
                "status": "pending",
            }
            state["entries"].append(pending)
            _durable_write(self.state_path, state)
            try:
                message_id = self.publisher(intent, reader=self.http_reader)
            except Exception as exc:
                status = _status(
                    now_ms=now_ms, revision=self.revision, state="publish_outcome_unknown",
                    operational=False, candidate_ready=False,
                    blockers=[str(exc) if isinstance(exc, BridgeError) else "intent_publish_failed"],
                    source=source_summary, release=release_summary,
                )
                _durable_write(self.status_path, status, shared=self.status_path == DEFAULT_STATUS)
                return status
            pending.update(
                status="published", published_at_ms=now_ms, pubsub_message_id=message_id,
            )
            _durable_write(self.state_path, state)
            receipt = {
                "topic": TOPIC_RESOURCE,
                "intent_id": intent["intent_id"],
                "client_order_id": intent["order"]["client_order_id"],
                "intent_sha256": intent_sha,
                "market_evidence_sha256": intent["market_evidence_sha256"],
                "venue": intent["order"]["venue"],
                "symbol": intent["order"]["symbol"],
                "side": intent["order"]["side"],
                "quote_value_usd": intent["order"]["quote_value_usd"],
                "pubsub_message_id": message_id,
                "published_at_ms": now_ms,
            }
            status = _status(
                now_ms=now_ms, revision=self.revision, state="published",
                operational=True, candidate_ready=True, blockers=[],
                source=source_summary, release=release_summary, last_publish=receipt,
            )
            _durable_write(self.status_path, status, shared=self.status_path == DEFAULT_STATUS)
            return status
        except CandidateUnavailable as exc:
            status = _status(
                now_ms=now_ms, revision=self.revision, state="idle",
                operational=operational, candidate_ready=False, blockers=[str(exc)],
                source=source_summary, release=release_summary,
            )
        except BridgeError as exc:
            status = _status(
                now_ms=now_ms, revision=self.revision, state="blocked",
                operational=False, candidate_ready=False, blockers=[str(exc)],
                source=source_summary, release=release_summary,
            )
        _durable_write(self.status_path, status, shared=self.status_path == DEFAULT_STATUS)
        return status


__all__ = [
    "BRIDGE_INTERVAL_SECONDS", "BridgeError", "CandidateUnavailable",
    "DEFAULT_RUNTIME", "DEFAULT_STATE", "DEFAULT_STATUS", "LOCAL_EXECUTOR_HOLDS",
    "FIRESTORE_READINESS_URL", "OrderIntentBridge", "PUBLISH_URL", "PROJECT", "STATUS_SCHEMA", "TOPIC",
    "access_token", "build_intent", "publish_intent", "select_candidate",
]