"""The only V2R1 ingress adapter: verified context to routed ledger record.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime from threading import RLock from typing import Callable, Mapping, Any from .message_ledger import LedgerRecord, MessageLedger from .router import Route, TenantRouter from .session_pool import SessionLease, SessionPool from .tenant_context import CanonicalMessage, TenantContext, utc_now @dataclass(frozen=True) class IngressResult: record: LedgerRecord route: Route session: SessionLease class ChannelAdapter: """Fail-closed runtime ingress composition. The adapter intentionally accepts no tenant or Cell authority outside the signed context and TenantRouter. A process lock keeps rollback handling deterministic for a single runtime while SQLite protects cross-process ledger writes. """ def __init__( self, *, router: TenantRouter, ledger: MessageLedger, session_pool: SessionPool, signing_key: bytes, clock: Callable[[], datetime] = utc_now, ) -> None: self._router = router self._ledger = ledger self._session_pool = session_pool self._signing_key = signing_key self._clock = clock self._lock = RLock() def ingest(self, *, token: str, message: Mapping[str, Any]) -> IngressResult: with self._lock: context = TenantContext.verify(token, self._signing_key, now=self._clock()) canonical = CanonicalMessage.from_mapping(message) canonical.assert_context(context) route = self._router.assert_context_route(context) self._router.assert_message_route(canonical) self._session_pool.assert_available(context, route) existed = self._ledger.matches(canonical) record = self._ledger.append(canonical) try: lease = self._session_pool.acquire(context, route) except Exception: if not existed: self._ledger.remove(record) raise return IngressResult(record=record, route=route, session=lease)