"""Richard-only loopback console for evidence, approval, and sending. This module intentionally uses only :mod:`http.server`. It is not a general web framework or an authentication layer: the server binds to IPv4 loopback, and every request must carry the one locally rotated operator token. Draft lifecycle work is injected by the existing canonical event/coaching layer so this surface cannot create a second event schema or delivery path. """ from __future__ import annotations import hmac import html import ipaddress import inspect import json import re import asyncio from dataclasses import asdict, dataclass, is_dataclass from collections.abc import Mapping as MappingABC from datetime import date, datetime, time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Callable, Mapping, Protocol, TypeAlias from urllib.parse import parse_qs, quote, unquote, urlsplit from checkin_cli.models import ContractStatus, Event, EventType, validate_event TELEGRAM_SINGLE_MESSAGE_LIMIT_UTF16 = 4096 def telegram_utf16_length(text: str) -> int: """Return Telegram's single-message length in UTF-16 code units.""" return len(text.encode("utf-16-le")) // 2 class OperatorConsoleError(ValueError): """Raised when the console would violate its local security boundary.""" EvidenceLoader: TypeAlias = Callable[[str, str], Any] DraftEdit: TypeAlias = Callable[..., Any] DraftAction: TypeAlias = Callable[..., Any] @dataclass(frozen=True, slots=True) class OperatorConsoleConfig: """The deliberately small runtime configuration for the local console.""" token: str | Path bind_host: str = "127.0.0.1" port: int = 0 max_body_bytes: int = 64 * 1024 customer_key: str | None = None @dataclass(frozen=True, slots=True) class ConsoleOperations: """Canonical lifecycle callbacks supplied by the event/coaching layer.""" evidence_loader: EvidenceLoader | None = None edit_draft: DraftEdit | None = None approve_draft: DraftAction | None = None send_draft: DraftAction | None = None class CustomerTransport(Protocol): def send_customer(self, customer_key: str, destination: Any, text: str) -> Any: ... CanonicalEventSource: TypeAlias = Any class CoordinatorLifecycleAdapter: """Bind the console to the canonical coordinator and customer transport. The adapter deliberately owns no draft state. Evidence is reconstructed from the supplied canonical event source on every request, while lifecycle mutations are delegated to the coordinator/service that owns the profile EventStore. Customer delivery is an explicit three-argument boundary: ``(customer_key, destination, text)`` and must return a message receipt. """ def __init__( self, lifecycle_service: Any, canonical_event_source: CanonicalEventSource, customer_transport: CustomerTransport, ) -> None: if lifecycle_service is None: raise OperatorConsoleError("canonical lifecycle service is required") if not _canonical_event_source_usable(canonical_event_source): raise OperatorConsoleError("canonical event source must be an EventStore resolver") if not _transport_callable(customer_transport): raise OperatorConsoleError("customer transport boundary must return receipts") if not callable(getattr(lifecycle_service, "validate_delivery_transport", None)): raise OperatorConsoleError( "canonical lifecycle service must expose validate_delivery_transport" ) if not _owner_identity_usable(getattr(lifecycle_service, "owner", None)): raise OperatorConsoleError("canonical lifecycle service must expose owner") self._service = lifecycle_service self._event_source = canonical_event_source self._transport = customer_transport def evidence(self, customer_key: str, draft_id: str) -> Any: value = _resolve_event_source(self._event_source, customer_key) if isinstance(value, Path): return _read_draft_evidence(value, customer_key, draft_id) if hasattr(value, "_read_events") and callable(value._read_events): value = value._read_events() return _draft_evidence_from_value(value, customer_key=customer_key, draft_id=draft_id) def edit(self, customer_key: str, draft_id: str, text: str) -> Any: return self._lifecycle_action("edit_draft", customer_key, draft_id, self._owner(), text) def approve(self, customer_key: str, draft_id: str) -> Any: return self._lifecycle_action("approve_draft", customer_key, draft_id, self._owner()) def send(self, customer_key: str, draft_id: str) -> Any: owner = self._owner() prepare = getattr(self._service, "prepare_delivery", None) if not callable(prepare): prepare = getattr(self._service, "prepare_send_draft", None) if not callable(prepare): raise _UnavailableOperation("prepare_send") prepared = self._require_action(prepare(draft_id, owner), "prepare_send") self._ensure_action_customer(prepared, customer_key) text = _action_text(prepared) if not text: raise RuntimeError("approved draft has no text") if telegram_utf16_length(text) > TELEGRAM_SINGLE_MESSAGE_LIMIT_UTF16: raise RuntimeError("draft text is too long") status = _action_status(prepared) if status in {"delivered", "sent_audited"}: audited = getattr(self._service, "mark_sent_audited", None) if not callable(audited): raise RuntimeError("delivery receipt requires reconciliation") final = self._require_action( audited(draft_id, owner), "mark_sent_audited", ) return final transport_required = getattr(prepared, "transport_required", None) if isinstance(prepared, MappingABC): transport_required = prepared.get("transport_required") if transport_required is False: raise RuntimeError("delivery receipt requires reconciliation before retry") if not hasattr(prepared, "transport_required") and not isinstance(prepared, MappingABC): pending = getattr(self._service, "mark_delivery_pending", None) if callable(pending): self._require_action(pending(draft_id, owner), "mark_delivery_pending") destination = self._destination(customer_key, prepared) validator = getattr(self._service, "validate_delivery_transport", None) if not callable(validator) or _invoke_delivery_validation( validator, draft_id, owner, destination, text ) is not True: raise RuntimeError("delivery transport validation rejected") receipt = _invoke_customer_transport(self._transport, customer_key, destination, text) delivered = getattr(self._service, "mark_delivered", None) if callable(delivered): delivered_action = self._require_action( delivered(draft_id, owner, receipt), "mark_delivered", ) audited = getattr(self._service, "mark_sent_audited", None) if callable(audited): final = self._require_action( audited(draft_id, owner), "mark_sent_audited", ) else: final = delivered_action return {"action": final, "message_id": receipt} sender = getattr(self._service, "send_draft", None) if not callable(sender): raise _UnavailableOperation("send") marked = self._require_action(sender(draft_id, owner), "send") return {"action": marked, "message_id": receipt} def _owner(self) -> Any: owner = getattr(self._service, "owner", None) if owner is None: raise OperatorConsoleError("canonical lifecycle service owner is unavailable") return owner def _lifecycle_action(self, operation: str, customer_key: str, *args: Any) -> Any: callback = getattr(self._service, operation, None) if not callable(callback): raise _UnavailableOperation(operation) action = self._require_action(callback(*args), operation) self._ensure_action_customer(action, customer_key) return action @staticmethod def _require_action(action: Any, operation: str) -> Any: if not _action_accepted(action): detail = _action_error(action) or f"{operation} was rejected" raise RuntimeError(detail) return action @staticmethod def _ensure_action_customer(action: Any, customer_key: str) -> None: selection = _action_selection(action) if selection is None: return customer = getattr(selection, "customer", None) spec = getattr(customer, "spec", None) selected_key = getattr(spec, "customer_key", None) if selected_key is not None and selected_key != customer_key: raise ValueError("lifecycle action is scoped to another customer") def _destination(self, customer_key: str, action: Any) -> Any: selection = _action_selection(action) if selection is not None: customer = getattr(selection, "customer", None) spec = getattr(customer, "spec", None) selected_key = getattr(spec, "customer_key", None) destination = getattr(spec, "telegram", None) if selected_key == customer_key and _destination_usable(destination): return destination if selected_key is not None and selected_key != customer_key: raise ValueError("customer transport destination is scoped to another customer") customer_lookup = getattr(self._service, "customer", None) if callable(customer_lookup): customer = customer_lookup(customer_key) spec = getattr(customer, "spec", None) destination = getattr(spec, "telegram", None) if getattr(spec, "customer_key", customer_key) == customer_key and _destination_usable(destination): return destination raise ValueError("customer transport destination is unavailable") _CUSTOMER_KEY = re.compile(r"^[a-z0-9][a-z0-9_-]{2,63}$") _DRAFT_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") _DEFAULT_MAX_BODY = 64 * 1024 class OperatorConsole: """Own token checking and dispatch for one Richard/operator console.""" def __init__( self, token: str | Path | None = None, *, config: OperatorConsoleConfig | None = None, operations: ConsoleOperations | None = None, evidence_loader: EvidenceLoader | None = None, edit_draft: DraftEdit | None = None, approve_draft: DraftAction | None = None, send_draft: DraftAction | None = None, events_path: Path | None = None, customer_key: str | None = None, ) -> None: if config is not None: if token is not None: raise OperatorConsoleError("token must be supplied by config or argument, not both") if customer_key is not None: raise OperatorConsoleError("customer key must be supplied by config or argument, not both") token = config.token customer_key = config.customer_key self.bind_host = config.bind_host self.port = config.port self.max_body_bytes = config.max_body_bytes else: self.bind_host = "127.0.0.1" self.port = 0 self.max_body_bytes = _DEFAULT_MAX_BODY if token is None: raise OperatorConsoleError("operator token is required") self.token = _load_token(token) if not self.token: raise OperatorConsoleError("operator token must not be empty") if self.max_body_bytes < 1 or self.max_body_bytes > 1024 * 1024: raise OperatorConsoleError("console body limit is outside the bounded range") self.events_path = events_path if customer_key is not None and _CUSTOMER_KEY.fullmatch(customer_key) is None: raise OperatorConsoleError("customer key is invalid") self.customer_key = customer_key if operations is not None and any( callback is not None for callback in (evidence_loader, edit_draft, approve_draft, send_draft) ): raise OperatorConsoleError("operations and callback arguments cannot be mixed") self.operations = operations or ConsoleOperations( evidence_loader=evidence_loader, edit_draft=edit_draft, approve_draft=approve_draft, send_draft=send_draft, ) def create_server( self, *, bind_host: str | None = None, port: int | None = None, ) -> ThreadingHTTPServer: host = self.bind_host if bind_host is None else bind_host selected_port = self.port if port is None else port _require_loopback_bind(host) if not isinstance(selected_port, int) or not 0 <= selected_port <= 65535: raise OperatorConsoleError("console port is invalid") return _ConsoleHTTPServer((host, selected_port), _ConsoleRequestHandler, self) def evidence(self, customer_key: str, draft_id: str) -> Any: self._validate_identity(customer_key, draft_id) loader = self.operations.evidence_loader if loader is not None: return loader(customer_key, draft_id) if self.events_path is not None: return _read_draft_evidence(self.events_path, customer_key, draft_id) raise _UnavailableOperation("evidence") def edit(self, customer_key: str, draft_id: str, text: str) -> Any: self._validate_identity(customer_key, draft_id) _validate_draft_text(text) callback = self.operations.edit_draft if callback is None: raise _UnavailableOperation("edit") return _invoke_operator_callback(callback, customer_key, draft_id, text) def approve(self, customer_key: str, draft_id: str) -> Any: self._validate_identity(customer_key, draft_id) callback = self.operations.approve_draft if callback is None: raise _UnavailableOperation("approve") return _invoke_operator_callback(callback, customer_key, draft_id) def send(self, customer_key: str, draft_id: str) -> Any: self._validate_identity(customer_key, draft_id) state = self._approval_state(customer_key, draft_id) if state == "sent": raise RuntimeError("draft has already been sent") if state != "approved": raise RuntimeError("draft must be approved before send") callback = self.operations.send_draft if callback is None: raise _UnavailableOperation("send") return _invoke_operator_callback(callback, customer_key, draft_id) def _validate_identity(self, customer_key: str, draft_id: str) -> None: _validate_route_identity(customer_key, draft_id) if self.customer_key is not None and customer_key != self.customer_key: raise ValueError("customer is not bound to this console") def _approval_state(self, customer_key: str, draft_id: str) -> str | None: try: evidence = self.evidence(customer_key, draft_id) except (_UnavailableOperation, OSError, ValueError): return None try: return _draft_lifecycle_state(evidence, customer_key=customer_key, draft_id=draft_id) except (OSError, ValueError, TypeError): return None def _has_approval(self, customer_key: str, draft_id: str) -> bool: return self._approval_state(customer_key, draft_id) == "approved" class _UnavailableOperation(RuntimeError): def __init__(self, operation: str) -> None: super().__init__(f"{operation} operation is not configured") self.operation = operation def _transport_callable(transport: Any) -> bool: return callable(getattr(transport, "send_customer", None)) def _owner_identity_usable(owner: Any) -> bool: key = getattr(owner, "key", None) return ( isinstance(key, tuple) and len(key) == 3 and all(isinstance(item, str) and item for item in key) ) def _destination_usable(destination: Any) -> bool: return destination is not None and all( isinstance(getattr(destination, field, None), str) and bool(getattr(destination, field, "").strip()) for field in ("user_id", "chat_id", "topic_id") ) def _canonical_event_source_usable(source: Any) -> bool: if isinstance(source, MappingABC): return bool(source) and all( callable(getattr(value, "_read_events", None)) for value in source.values() ) for name in ("events_for", "for_customer", "store_for"): if callable(getattr(source, name, None)): return True return callable(getattr(source, "_read_events", None)) def _resolve_event_source(source: Any, customer_key: str) -> Any: if isinstance(source, MappingABC): if customer_key not in source: raise ValueError("canonical event source has no customer") return source[customer_key] for name in ("events_for", "for_customer", "store_for"): callback = getattr(source, name, None) if callable(callback): return callback(customer_key) if callable(getattr(source, "_read_events", None)): return source raise ValueError("canonical event source has no customer resolver") def _action_accepted(action: Any) -> bool: if isinstance(action, MappingABC): return action.get("accepted") is True return getattr(action, "accepted", False) is True def _action_status(action: Any) -> str: if isinstance(action, MappingABC): return str(action.get("status", "") or "") return str(getattr(action, "status", "") or "") def _action_error(action: Any) -> str: if isinstance(action, MappingABC): return str(action.get("error", "") or "") return str(getattr(action, "error", "") or "") def _action_text(action: Any) -> str: if isinstance(action, MappingABC): return str(action.get("text", "") or "").strip() return str(getattr(action, "text", "") or "").strip() def _action_selection(action: Any) -> Any: if isinstance(action, MappingABC): return action.get("selection") return getattr(action, "selection", None) def _invoke_delivery_validation( validator: Callable[..., Any], draft_id: str, owner: Any, destination: Any, text: str, ) -> Any: try: parameters = inspect.signature(validator).parameters except (TypeError, ValueError): return validator(draft_id, owner) kwargs: dict[str, Any] = {} if "destination" in parameters: kwargs["destination"] = destination if "text" in parameters: kwargs["text"] = text return validator(draft_id, owner, **kwargs) def _invoke_customer_transport( transport: CustomerTransport, customer_key: str, destination: Any, text: str, ) -> str: sender = getattr(transport, "send_customer", None) if not callable(sender): raise OperatorConsoleError("customer transport boundary is unavailable") result = sender(customer_key, destination, text) if inspect.isawaitable(result): try: result = asyncio.run(result) except RuntimeError as exc: raise RuntimeError("customer transport could not be awaited") from exc if result is False or result is None: raise RuntimeError("customer transport rejected delivery") if isinstance(result, MappingABC): if result.get("success") is False or result.get("ok") is False: raise RuntimeError("customer transport rejected delivery") message_id = result.get("message_id") else: if getattr(result, "success", True) is False: raise RuntimeError("customer transport rejected delivery") message_id = getattr(result, "message_id", result if isinstance(result, (str, int)) else None) if message_id is None or not str(message_id).strip(): raise RuntimeError("customer transport must return a message receipt") return str(message_id) def _draft_evidence_from_value(value: Any, *, customer_key: str, draft_id: str) -> dict[str, Any]: events = _canonical_event_candidates(value) matching = _matching_draft_events(events, customer_key=customer_key, draft_id=draft_id) state = _draft_lifecycle_state(matching, customer_key=customer_key, draft_id=draft_id) serialized = [_json_safe(event.model_dump(mode="json", exclude_none=True)) for event in matching] latest = matching[-1] if matching else None return { "customer_key": customer_key, "draft_id": draft_id, "events": serialized, "draft": ( _json_safe(latest.draft.model_dump(mode="json", exclude_none=True)) if latest is not None and latest.draft is not None else None ), "state": state, "approved": state == "approved", "sent": state == "sent", } class _ConsoleHTTPServer(ThreadingHTTPServer): allow_reuse_address = False daemon_threads = True def __init__( self, server_address: tuple[str, int], handler: type[BaseHTTPRequestHandler], console: OperatorConsole, ) -> None: self.console = console super().__init__(server_address, handler) class _ConsoleRequestHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "DualCoachPilotConsole/1" @property def console(self) -> OperatorConsole: server = self.server if not isinstance(server, _ConsoleHTTPServer): raise RuntimeError("invalid operator console server") return server.console def do_GET(self) -> None: # noqa: N802 - stdlib HTTP handler API if not self._authorize(): return path, parts = self._route() if path == "/": self._send_root_screen() return if len(parts) == 2 and parts[0] == "evidence" and self.console.customer_key is not None: parts = ("evidence", self.console.customer_key, parts[1]) if len(parts) == 3 and parts[0] == "evidence": try: result = self.console.evidence(parts[1], parts[2]) except _UnavailableOperation as exc: self._send_json(501, {"error": "operation_unavailable", "operation": exc.operation}) except ValueError as exc: self._send_json(400, {"error": "invalid_request", "detail": str(exc)}) except Exception: self._send_json(500, {"error": "operation_failed"}) else: self._send_json(200, {"ok": True, "evidence": result}) return self._send_json(404, {"error": "not_found"}) def do_POST(self) -> None: # noqa: N802 - stdlib HTTP handler API if not self._authorize(): return path, parts = self._route() if parts and parts[0] == "api": parts = parts[1:] if len(parts) == 3 and parts[0] == "draft" and self.console.customer_key is not None: parts = ("draft", self.console.customer_key, parts[1], parts[2]) if len(parts) != 4 or parts[0] != "draft" or parts[3] not in {"edit", "approve", "send"}: self._send_json(404, {"error": "not_found"}) return try: body = self._read_json_body() customer_key, draft_id, action = parts[1], parts[2], parts[3] if action == "edit": text = body.get("text", body.get("draft_text")) if isinstance(body, Mapping) else None if not isinstance(text, str): raise ValueError("edit requires text") result = self.console.edit(customer_key, draft_id, text) elif action == "approve": result = self.console.approve(customer_key, draft_id) else: result = self.console.send(customer_key, draft_id) except _UnavailableOperation as exc: self._send_json(501, {"error": "operation_unavailable", "operation": exc.operation}) except ValueError as exc: self._send_json(400, {"error": "invalid_request", "detail": str(exc)}) except RuntimeError as exc: self._send_json(409, {"error": "operation_rejected", "detail": str(exc)}) except Exception: self._send_json(500, {"error": "operation_failed"}) else: self._send_json(200, {"ok": True, "result": result}) def log_message(self, format: str, *args: Any) -> None: # Do not write request headers or bodies to the default process log. return def _route(self) -> tuple[str, tuple[str, ...]]: parsed = urlsplit(self.path) if parsed.query or parsed.fragment: return parsed.path, () path = parsed.path if not path.startswith("/") or "\\" in path: return path, () raw_parts = tuple(path.strip("/").split("/")) if path != "/" else () try: parts = tuple(unquote(part) for part in raw_parts) except ValueError: return path, () if any(not part or "/" in part or "\\" in part for part in parts): return path, () return path, parts def _send_root_screen(self) -> None: try: customer_key, draft_id = self._root_selection() except ValueError as exc: self._send_html( _render_console_screen( bound_customer=self.console.customer_key, error=str(exc), ), status=400, ) return if customer_key is None or draft_id is None: self._send_html( _render_console_screen( customer_key=customer_key, draft_id=draft_id, bound_customer=self.console.customer_key, ) ) return try: evidence = self.console.evidence(customer_key, draft_id) except _UnavailableOperation as exc: self._send_html( _render_console_screen( customer_key=customer_key, draft_id=draft_id, bound_customer=self.console.customer_key, error=f"{exc.operation} operation is not configured", ), status=501, ) except (OSError, ValueError): self._send_html( _render_console_screen( customer_key=customer_key, draft_id=draft_id, bound_customer=self.console.customer_key, error="evidence could not be loaded", ), status=400, ) except Exception: self._send_html( _render_console_screen( customer_key=customer_key, draft_id=draft_id, bound_customer=self.console.customer_key, error="evidence could not be loaded", ), status=500, ) else: self._send_html( _render_console_screen( customer_key=customer_key, draft_id=draft_id, bound_customer=self.console.customer_key, evidence=evidence, ) ) def _root_selection(self) -> tuple[str | None, str | None]: parsed = urlsplit(self.path) if parsed.fragment: raise ValueError("fragments are not accepted") try: query = parse_qs( parsed.query, keep_blank_values=True, max_num_fields=2, ) except ValueError as exc: raise ValueError("root query is invalid") from exc if any(name not in {"customer", "draft"} for name in query): raise ValueError("root query accepts only customer and draft") for name, values in query.items(): if len(values) != 1: raise ValueError(f"root query parameter {name} must appear once") customer_key = query.get("customer", [None])[0] draft_id = query.get("draft", [None])[0] if customer_key is not None and not isinstance(customer_key, str): raise ValueError("customer query is invalid") if draft_id is not None and not isinstance(draft_id, str): raise ValueError("draft query is invalid") if customer_key is not None: _validate_route_identity(customer_key, draft_id or "draft") if draft_id is not None: _validate_route_identity(customer_key or "customer", draft_id) return customer_key, draft_id def _authorize(self) -> bool: if not _request_is_loopback(self.client_address[0], self.headers.get("Host")): self._send_json(403, {"error": "loopback_required"}) return False supplied = self.headers.get("X-Operator-Token") if supplied is None: authorization = self.headers.get("Authorization", "") if authorization.startswith("Bearer "): supplied = authorization[7:] if supplied is None or not hmac.compare_digest(supplied, self.console.token): self.send_response(401) self.send_header("WWW-Authenticate", "Bearer") self._security_headers() self.send_header("Content-Length", "0") self.end_headers() return False return True def _read_json_body(self) -> Any: raw_length = self.headers.get("Content-Length") if raw_length is None: raise ValueError("content length is required") try: length = int(raw_length) except ValueError as exc: raise ValueError("content length is invalid") from exc if length < 0 or length > self.console.max_body_bytes: raise ValueError("request body is too large") raw = self.rfile.read(length) if len(raw) != length: raise ValueError("request body is incomplete") try: return json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ValueError("request body must be JSON") from exc def _send_json(self, status: int, payload: Any) -> None: body = json.dumps(_json_safe(payload), ensure_ascii=False, separators=(",", ":")).encode("utf-8") self.send_response(status) self._security_headers() self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _send_html(self, body: str, *, status: int = 200) -> None: encoded = body.encode("utf-8") self.send_response(status) self._security_headers() self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(encoded))) self.end_headers() self.wfile.write(encoded) def _security_headers(self) -> None: self.send_header("Cache-Control", "no-store") self.send_header("Content-Security-Policy", "default-src 'none'; script-src 'unsafe-inline'; connect-src 'self'; form-action 'self'") self.send_header("Referrer-Policy", "no-referrer") self.send_header("X-Content-Type-Options", "nosniff") # Short public factory names keep the integration surface explicit and testable. def create_operator_console_server( token: str | Path, *, port: int = 0, bind_host: str = "127.0.0.1", max_body_bytes: int = _DEFAULT_MAX_BODY, operations: ConsoleOperations | None = None, evidence_loader: EvidenceLoader | None = None, edit_draft: DraftEdit | None = None, approve_draft: DraftAction | None = None, send_draft: DraftAction | None = None, events_path: Path | None = None, customer_key: str | None = None, ) -> ThreadingHTTPServer: """Create, but do not start, the token-protected loopback server.""" return OperatorConsole( config=OperatorConsoleConfig( token=token, bind_host=bind_host, port=port, max_body_bytes=max_body_bytes, customer_key=customer_key, ), operations=operations, evidence_loader=evidence_loader, edit_draft=edit_draft, approve_draft=approve_draft, send_draft=send_draft, events_path=events_path, ).create_server() def create_production_operator_console_server( *, token_path: str | Path, canonical_event_store: CanonicalEventSource | None = None, canonical_event_source: CanonicalEventSource | None = None, lifecycle_service: Any | None = None, coordinator: Any | None = None, customer_transport: CustomerTransport | None = None, port: int = 0, bind_host: str = "127.0.0.1", max_body_bytes: int = _DEFAULT_MAX_BODY, customer_key: str | None = None, ) -> ThreadingHTTPServer: """Build the production console from canonical profile-owned services. Unlike the convenience factory above, this entrypoint refuses callback-only wiring. A rotated token file, canonical event source, lifecycle service, and customer transport boundary must all be supplied explicitly. """ if canonical_event_store is not None and canonical_event_source is not None: raise OperatorConsoleError("supply one canonical event source, not both") source = canonical_event_store if canonical_event_store is not None else canonical_event_source if source is None: raise OperatorConsoleError("canonical event source is required") if lifecycle_service is not None and coordinator is not None: raise OperatorConsoleError("supply lifecycle service or coordinator, not both") service = lifecycle_service if lifecycle_service is not None else coordinator if service is None: raise OperatorConsoleError("canonical lifecycle service is required") if customer_transport is None: raise OperatorConsoleError("customer transport boundary is required") if not isinstance(token_path, (str, Path)): raise OperatorConsoleError("production console requires a rotated token path") adapter = CoordinatorLifecycleAdapter(service, source, customer_transport) return create_operator_console_server( Path(token_path), port=port, bind_host=bind_host, max_body_bytes=max_body_bytes, operations=ConsoleOperations( evidence_loader=adapter.evidence, edit_draft=adapter.edit, approve_draft=adapter.approve, send_draft=adapter.send, ), customer_key=customer_key, ) build_production_operator_console_server = create_production_operator_console_server build_operator_console_server = create_operator_console_server create_server = create_operator_console_server ConsoleConfig = OperatorConsoleConfig OperatorConsoleServer = _ConsoleHTTPServer OperatorConsoleHandler = _ConsoleRequestHandler def serve_operator_console( token: str | Path, *, port: int = 0, bind_host: str = "127.0.0.1", max_body_bytes: int = _DEFAULT_MAX_BODY, operations: ConsoleOperations | None = None, evidence_loader: EvidenceLoader | None = None, edit_draft: DraftEdit | None = None, approve_draft: DraftAction | None = None, send_draft: DraftAction | None = None, events_path: Path | None = None, customer_key: str | None = None, ) -> None: """Run the local console until its caller shuts the process down.""" server = create_operator_console_server( token, port=port, bind_host=bind_host, max_body_bytes=max_body_bytes, operations=operations, evidence_loader=evidence_loader, edit_draft=edit_draft, approve_draft=approve_draft, send_draft=send_draft, events_path=events_path, customer_key=customer_key, ) try: server.serve_forever() finally: server.server_close() run_operator_console = serve_operator_console def serve_production_operator_console(**kwargs: Any) -> None: """Run the explicitly wired production console until shutdown.""" server = create_production_operator_console_server(**kwargs) try: server.serve_forever() finally: server.server_close() run_production_operator_console = serve_production_operator_console def load_operator_token(path: Path) -> str: """Read one local token file without following a symlink.""" return _load_token(path) def _load_token(value: str | Path) -> str: if isinstance(value, Path): if value.is_symlink() or not value.is_file(): raise OperatorConsoleError("operator token path must be a regular file") try: mode = value.stat().st_mode except OSError as exc: raise OperatorConsoleError("operator token metadata could not be read") from exc if mode & 0o077: raise OperatorConsoleError("operator token file must not be group or world accessible") try: content = value.read_text(encoding="utf-8") except OSError as exc: raise OperatorConsoleError("operator token could not be read") from exc if len(content) > 4096: raise OperatorConsoleError("operator token is too large") return content.strip() if not isinstance(value, str): raise OperatorConsoleError("operator token must be text or a path") return value def _require_loopback_bind(host: str) -> None: # Binding 0.0.0.0, ::, localhost, or an arbitrary interface would violate # the pilot boundary; SSH forwarding can still target 127.0.0.1. if host != "127.0.0.1": raise OperatorConsoleError("operator console must bind to 127.0.0.1") def _request_is_loopback(remote: str, host_header: str | None) -> bool: try: remote_ip = ipaddress.ip_address(remote) except ValueError: return False if not remote_ip.is_loopback: return False if not host_header: return True host = host_header.rsplit(":", 1)[0].strip("[]").casefold() return host in {"127.0.0.1", "localhost", "::1"} def _validate_route_identity(customer_key: str, draft_id: str) -> None: if _CUSTOMER_KEY.fullmatch(customer_key) is None: raise ValueError("customer key is invalid") if _DRAFT_ID.fullmatch(draft_id) is None: raise ValueError("draft id is invalid") def _validate_draft_text(text: str) -> None: if not text.strip(): raise ValueError("draft text must not be empty") if telegram_utf16_length(text) > TELEGRAM_SINGLE_MESSAGE_LIMIT_UTF16: raise ValueError("draft text is too long") def _invoke_operator_callback(callback: Callable[..., Any], customer_key: str, draft_id: str, *args: Any) -> Any: """Pass the fixed actor only when the callback declares that parameter.""" try: parameters = inspect.signature(callback).parameters except (TypeError, ValueError): parameters = {} if "actor" in parameters: return callback(customer_key, draft_id, *args, actor="richard") return callback(customer_key, draft_id, *args) _LIFECYCLE_EVENT_TYPES = frozenset( { EventType.DRAFT_CREATED, EventType.DRAFT_EDITED, EventType.DRAFT_APPROVED, EventType.DRAFT_SENT, } ) def _evidence_contains_approval(value: Any, draft_id: str) -> bool: """Return approval only when canonical draft events prove it.""" return _draft_lifecycle_state(value, draft_id=draft_id) == "approved" def _draft_lifecycle_state( value: Any, *, customer_key: str | None = None, draft_id: str, ) -> str | None: candidates = tuple(_canonical_event_candidates(value)) events = _matching_draft_events(candidates, customer_key=customer_key, draft_id=draft_id) state: str | None = None seen_types: set[EventType] = set() for event in events: event_type = event.event_type if event_type in seen_types: raise ValueError("draft lifecycle contains duplicate event types") seen_types.add(event_type) if event_type is EventType.DRAFT_CREATED: if state is not None: raise ValueError("draft lifecycle must start with draft_created") state = "created" elif event_type is EventType.DRAFT_EDITED: if state not in {"created", "edited", "approved"}: raise ValueError("draft_edited requires a created draft") state = "edited" elif event_type is EventType.DRAFT_APPROVED: if state not in {"created", "edited"}: raise ValueError("draft_approved requires an unapproved draft") state = "approved" elif event_type is EventType.DRAFT_SENT: if state != "approved": raise ValueError("draft_sent requires an approved draft") state = "sent" return state def _canonical_event_candidates(value: Any) -> list[Event]: """Extract canonical event objects from a loader result. Only ``events`` containers and mappings with an ``event_type`` discriminator are accepted. In particular, legacy ``payload``/``status`` dictionaries cannot establish approval. """ if value is None: return [] if isinstance(value, Event): try: return [validate_event(value)] except Exception as exc: raise ValueError("evidence contains an invalid canonical event") from exc if isinstance(value, Mapping): if "event_type" in value: try: return [validate_event(value)] except Exception as exc: raise ValueError("evidence contains an invalid canonical event") from exc if "events" in value: nested = value["events"] if not isinstance(nested, (tuple, list)): raise ValueError("evidence events must be a list") return _canonical_event_candidates(nested) if "evidence" in value: return _canonical_event_candidates(value["evidence"]) return [] if isinstance(value, (tuple, list)): events: list[Event] = [] for item in value: events.extend(_canonical_event_candidates(item)) return events raise ValueError("evidence contains a non-canonical event value") def _matching_draft_events( events: tuple[Event, ...] | list[Event], *, customer_key: str | None, draft_id: str, ) -> tuple[Event, ...]: matching: list[Event] = [] seen_event_ids: set[str] = set() seen_dedupe_keys: set[str] = set() for event in events: if event.event_id in seen_event_ids or event.dedupe_key in seen_dedupe_keys: raise ValueError("event stream contains duplicate canonical identities") seen_event_ids.add(event.event_id) seen_dedupe_keys.add(event.dedupe_key) if event.event_type not in _LIFECYCLE_EVENT_TYPES: continue payload = event.draft if payload is None: raise ValueError("draft lifecycle event is missing its draft payload") if payload.draft_id != draft_id: continue if customer_key is not None: expected_source = f"pilot:{customer_key}:{event.event_type.value}" if event.provenance.source_ref != expected_source: raise ValueError("draft evidence is not scoped to the requested customer") if event.status is not ContractStatus.ACCEPTED: raise ValueError("draft lifecycle evidence is not accepted") if event.event_type in {EventType.DRAFT_APPROVED, EventType.DRAFT_SENT} and payload.actor.value != "richard": raise ValueError("draft approval and send evidence must be owned by Richard") matching.append(event) return tuple(matching) def _read_draft_evidence(path: Path, customer_key: str, draft_id: str) -> dict[str, Any]: """Read canonical draft events and reconstruct the durable lifecycle.""" _validate_route_identity(customer_key, draft_id) if path.is_symlink() or not path.is_file(): raise ValueError("events path must be a regular file") try: lines = path.read_text(encoding="utf-8").splitlines() except (OSError, UnicodeDecodeError) as exc: raise ValueError("events path could not be read") from exc events: list[Event] = [] seen_event_ids: set[str] = set() seen_dedupe_keys: set[str] = set() for line in lines: if not line.strip(): continue try: event = validate_event(Event.model_validate_json(line)) except Exception as exc: raise ValueError("events stream contains invalid canonical evidence") from exc if event.event_id in seen_event_ids or event.dedupe_key in seen_dedupe_keys: raise ValueError("events stream contains duplicate canonical identities") seen_event_ids.add(event.event_id) seen_dedupe_keys.add(event.dedupe_key) events.append(event) return _draft_evidence_from_value(events, customer_key=customer_key, draft_id=draft_id) def _json_safe(value: Any) -> Any: if value is None or isinstance(value, (str, int, float, bool)): return value if isinstance(value, (date, datetime, time)): return value.isoformat() if is_dataclass(value): return _json_safe(asdict(value)) model_dump = getattr(value, "model_dump", None) if callable(model_dump): return _json_safe(model_dump(mode="json")) if isinstance(value, Mapping): return {str(key): _json_safe(item) for key, item in value.items()} if isinstance(value, (tuple, list, set, frozenset)): return [_json_safe(item) for item in value] return str(value) _SCREEN_MISSING = object() def _html_escape(value: Any) -> str: return html.escape(str(value), quote=True) def _screen_draft_text(evidence: Any) -> str: if not isinstance(evidence, MappingABC): return "" draft = evidence.get("draft") if not isinstance(draft, MappingABC): return "" text = draft.get("text") return text if isinstance(text, str) else "" def _screen_state(evidence: Any) -> str: if not isinstance(evidence, MappingABC): return "unknown" state = evidence.get("state") return state if isinstance(state, str) and state else "unknown" def _render_console_screen( *, customer_key: str | None = None, draft_id: str | None = None, bound_customer: str | None = None, evidence: Any = _SCREEN_MISSING, error: str | None = None, ) -> str: """Render the small Richard-only screen without embedding credentials. The token field is intentionally unnamed and has no server-provided value. Inline JavaScript reads it only when a user explicitly submits one of the forms, then sends the existing authentication header to the canonical JSON routes. All response text is assigned through ``textContent``. """ selected_customer = customer_key or bound_customer or "" selected_draft = draft_id or "" form_customer = _html_escape(selected_customer) form_draft = _html_escape(selected_draft) readonly = " readonly" if bound_customer is not None else "" bound_note = ( f"

This console is bound to customer " f"{_html_escape(bound_customer)}.

" if bound_customer is not None else "

Enter the single pilot customer and draft identifiers to load evidence.

" ) error_block = ( f'' if error is not None else "" ) encoded_customer = quote(selected_customer, safe="") if selected_customer else "" encoded_draft = quote(selected_draft, safe="") if selected_draft else "" evidence_endpoint = ( f"/evidence/{encoded_customer}/{encoded_draft}" if encoded_customer and encoded_draft else "" ) action_root = ( f"/draft/{encoded_customer}/{encoded_draft}" if encoded_customer and encoded_draft else "#" ) has_evidence = ( customer_key is not None and draft_id is not None and error is None and evidence is not _SCREEN_MISSING ) safe_evidence = _json_safe(evidence) if has_evidence else {} evidence_json = ( json.dumps( safe_evidence, ensure_ascii=True, indent=2, sort_keys=True, ) if has_evidence else "" ) draft_text = _html_escape(_screen_draft_text(safe_evidence)) if has_evidence else "" safe_customer = _html_escape(selected_customer) safe_draft = _html_escape(selected_draft) panel_visibility = "" if has_evidence else " hidden" evidence_state = _html_escape(_screen_state(safe_evidence)) if has_evidence else "unknown" safe_evidence_endpoint = _html_escape(evidence_endpoint) script = """ """ return f""" Richard draft console

Richard-only draft console

{bound_note}

Required for each request; kept only in browser memory and never put in a URL.

{error_block}

Draft evidence

Customer
{safe_customer}
Draft
{safe_draft}
Lifecycle state
{evidence_state}
{_html_escape(evidence_json)}

Edit draft


  
Approve and send are separate explicit actions; sending never happens automatically.

  
Each action uses the canonical lifecycle endpoint and requires the existing operator authentication header.





{script}


"""


if __name__ == "__main__":
    raise SystemExit("operator_console is an integration module; configure a rotated local token first")