from __future__ import annotations import argparse import hmac import json import os import queue import re import secrets import threading import time from hashlib import sha256 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.parse import parse_qs, urlparse from continuum_daemon_registry import load_daemon_registry, match_daemons from continuum_whatsapp_adapter import ( normalize_webhook, verify_meta_signature, whatsapp_config_state, ) from continuum_provider_registry import provider_registry_state from continuum_translation_faculty import translation_faculty_state ROOT = Path(__file__).resolve().parent CHANNEL = Path(os.environ.get("CONTINUUM_BRAIN_CHANNEL", ROOT / "cli_brain_channel")) EVENT_LOG = CHANNEL / "continuum_link_events.jsonl" PEER_LINKS_FILE = CHANNEL / "continuum_peer_links.json" MAX_BODY_BYTES = 1_000_000 MAX_BACKLOG = 200 MAX_PEER_LINKS = 5 LOCAL_HOSTS = {"127.0.0.1", "localhost", "::1"} DEFAULT_ALLOWED_ORIGINS = ( "http://127.0.0.1:7870", "http://localhost:7870", "http://127.0.0.1:7871", "http://localhost:7871", ) DEFAULT_SLOTS = ( "personal", "voice", "whatsapp", "council", "business", "reputation", "ipfs", "marketplace", "facilities", "control", "provenance", ) ALL_SLOTS = "*" AUTH_TOKEN = os.environ.get("CONTINUUM_LINK_TOKEN") or secrets.token_urlsafe(32) AUTH_TOKEN_SOURCE = "env" if os.environ.get("CONTINUUM_LINK_TOKEN") else "ephemeral" RAW_TEXT_KEYS = {"raw", "raw_content", "raw_message", "body", "text", "message", "memo"} IDENTIFIER_KEYS = { "to", "from", "phone", "phone_number", "recipient", "recipient_id", "wa_id", "display_phone_number", "audio_media_id", "media_id", "audio_link", "media_link", } STARTED_AT = time.time() SUBSCRIBERS: dict[queue.Queue[dict[str, Any]], str] = {} SUBSCRIBERS_LOCK = threading.RLock() EVENT_LOCK = threading.RLock() PEER_LINKS_LOCK = threading.RLock() def _canonical_json(payload: Any) -> str: return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def _bool_env(name: str, default: bool = False) -> bool: value = os.environ.get(name) if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on"} def _hash_string(value: str) -> str: return sha256(value.encode("utf-8")).hexdigest() def _redacted_string(value: str, category: str) -> dict[str, Any]: value = str(value or "") return { "redacted": True, "category": category, "sha256": _hash_string(value), "length": len(value), } def _store_raw_content() -> bool: return _bool_env("CONTINUUM_LINK_STORE_RAW", False) def _store_identifiers() -> bool: return _bool_env("CONTINUUM_LINK_STORE_IDENTIFIERS", False) def _privacy_scrub(value: Any, key: str = "") -> Any: lowered = key.lower() if isinstance(value, dict): return {str(k): _privacy_scrub(v, str(k)) for k, v in value.items()} if isinstance(value, list): return [_privacy_scrub(item, key) for item in value] if isinstance(value, str): if lowered in RAW_TEXT_KEYS and not _store_raw_content(): return _redacted_string(value, "raw_text") if lowered in IDENTIFIER_KEYS and not _store_identifiers(): return _redacted_string(value, "identifier") return value def _allowed_origins() -> set[str]: configured = os.environ.get("CONTINUUM_LINK_ALLOWED_ORIGINS") if configured: return {origin.strip() for origin in configured.split(",") if origin.strip()} return set(DEFAULT_ALLOWED_ORIGINS) def _auth_digest() -> str: return _hash_string(AUTH_TOKEN)[:16] def _auth_state() -> dict[str, Any]: return { "required": True, "token_source": AUTH_TOKEN_SOURCE, "token_sha256_prefix": _auth_digest(), "accepted": ["Authorization: Bearer ", "X-Continuum-Token: ", "token query parameter for EventSource"], "secrets_exposed": False, } def _webhook_signature_required(host: str) -> bool: return _bool_env("CONTINUUM_REQUIRE_WHATSAPP_SIGNATURE", host not in LOCAL_HOSTS) def _now_ms() -> int: return int(time.time() * 1000) def _normalize_slot(value: Any) -> str: slot = str(value or "personal").strip().lower() slot = "".join(ch if ch.isalnum() or ch in {"-", "_"} else "-" for ch in slot) slot = slot.strip("-_") return slot or "personal" def _event_slot(event: dict[str, Any]) -> str: return _normalize_slot(event.get("slot") or "personal") def _make_event(payload: dict[str, Any], source: str = "local") -> dict[str, Any]: created_ms = int(payload.get("created_ms") or _now_ms()) kind = str(payload.get("kind") or "continuum.event") slot = _normalize_slot(payload.get("slot") or "personal") body = payload.get("payload") if body is None: body = { key: value for key, value in payload.items() if key not in {"schema", "event_id", "created_ms", "kind", "source", "slot"} } body = _privacy_scrub(body) seed = { "created_ms": created_ms, "kind": kind, "source": payload.get("source") or source, "slot": slot, "payload": body, } event_id = str(payload.get("event_id") or ("clink_" + sha256(_canonical_json(seed).encode("utf-8")).hexdigest()[:20])) return { "schema": "champion-continuum/link-event/v1", "event_id": event_id, "created_ms": created_ms, "kind": kind, "slot": slot, "source": str(payload.get("source") or source), "payload": body, } def _subscriber_allows(subscription_slot: str, event: dict[str, Any]) -> bool: return subscription_slot == ALL_SLOTS or subscription_slot == _event_slot(event) def _append_event(event: dict[str, Any]) -> str | None: try: CHANNEL.mkdir(parents=True, exist_ok=True) with EVENT_LOCK: with EVENT_LOG.open("a", encoding="utf-8") as handle: handle.write(json.dumps(event, ensure_ascii=False) + "\n") except OSError as exc: return f"{type(exc).__name__}: {exc}" with SUBSCRIBERS_LOCK: dead: list[queue.Queue[dict[str, Any]]] = [] for subscriber, subscription_slot in SUBSCRIBERS.items(): if not _subscriber_allows(subscription_slot, event): continue try: subscriber.put_nowait(event) except queue.Full: dead.append(subscriber) for subscriber in dead: SUBSCRIBERS.pop(subscriber, None) return None def _recent_events(limit: int = 50, slot: str = ALL_SLOTS) -> list[dict[str, Any]]: if not EVENT_LOG.exists(): return [] limit = max(1, min(int(limit or 50), MAX_BACKLOG)) subscription_slot = ALL_SLOTS if slot == ALL_SLOTS else _normalize_slot(slot) try: lines = EVENT_LOG.read_text(encoding="utf-8", errors="replace").splitlines() except OSError: return [] out: list[dict[str, Any]] = [] for line in reversed(lines): try: payload = json.loads(line) except json.JSONDecodeError: continue if isinstance(payload, dict) and _subscriber_allows(subscription_slot, payload): out.append(payload) if len(out) >= limit: break out.reverse() return out def _slot_counts() -> dict[str, int]: counts: dict[str, int] = {slot: 0 for slot in DEFAULT_SLOTS} if not EVENT_LOG.exists(): return counts try: lines = EVENT_LOG.read_text(encoding="utf-8", errors="replace").splitlines() except OSError: return counts for line in lines: try: payload = json.loads(line) except json.JSONDecodeError: continue if isinstance(payload, dict): slot = _event_slot(payload) counts[slot] = counts.get(slot, 0) + 1 return counts def _load_peer_links() -> list[dict[str, Any]]: with PEER_LINKS_LOCK: try: payload = json.loads(PEER_LINKS_FILE.read_text(encoding="utf-8")) except Exception: return [] links = payload.get("links") if isinstance(payload, dict) else payload if not isinstance(links, list): return [] out: list[dict[str, Any]] = [] for item in links[:MAX_PEER_LINKS]: if isinstance(item, dict): out.append(item) return out def _save_peer_links(links: list[dict[str, Any]]) -> None: CHANNEL.mkdir(parents=True, exist_ok=True) payload = { "schema": "champion-continuum/peer-links/v1", "updated_ms": _now_ms(), "max_links": MAX_PEER_LINKS, "links": links[:MAX_PEER_LINKS], } with PEER_LINKS_LOCK: PEER_LINKS_FILE.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") def _peer_link_state() -> dict[str, Any]: links = _load_peer_links() return { "schema": "champion-continuum/peer-links/v1", "mode": "metadata_registry", "max_links": MAX_PEER_LINKS, "count": len(links), "links": links, "note": "Peer SSE links are registered targets. This service does not auto-dial them yet.", "auth": { "local_link_token_required": True, "remote_token_storage": "hash_only_by_default", }, "channels": { "whatsapp": "conversation channel and webhook adapter", "google": "future OAuth/contact/calendar/identity lane; not required for local rooms", }, } def _build_peer_link(payload: dict[str, Any]) -> dict[str, Any]: label = str(payload.get("label") or payload.get("name") or "Peer Continuum").strip()[:80] url = str(payload.get("url") or payload.get("sse_url") or "").strip() slot = _normalize_slot(payload.get("slot") or payload.get("default_slot") or "personal") auth_hint = str(payload.get("auth_hint") or "remote_token_required").strip()[:80] token = str(payload.get("token") or payload.get("remote_token") or "").strip() token_sha = _hash_string(token) if token else "" link_id_seed = {"label": label, "url": url, "slot": slot} link_id = str(payload.get("link_id") or ("peer_" + sha256(_canonical_json(link_id_seed).encode("utf-8")).hexdigest()[:12])) return { "schema": "champion-continuum/peer-link/v1", "link_id": link_id, "label": label, "url": url, "default_slot": slot, "auth_hint": auth_hint, "token_sha256": token_sha, "token_present": bool(token), "enabled": bool(payload.get("enabled", True)), "created_ms": _now_ms(), "last_seen_ms": 0, "capabilities": [str(item) for item in payload.get("capabilities", [])[:32]] if isinstance(payload.get("capabilities"), list) else [], "external_connection_opened": False, } def _state() -> dict[str, Any]: with SUBSCRIBERS_LOCK: subscriber_count = len(SUBSCRIBERS) subscriber_slots = dict.fromkeys(DEFAULT_SLOTS, 0) subscriber_slots[ALL_SLOTS] = 0 for slot in SUBSCRIBERS.values(): subscriber_slots[slot] = int(subscriber_slots.get(slot, 0)) + 1 recent = _recent_events(limit=5) return { "status": "ok", "service": "champion-continuum-link", "schema": "champion-continuum/link-state/v1", "root": str(ROOT), "channel": str(CHANNEL), "event_log": str(EVENT_LOG), "uptime_seconds": round(time.time() - STARTED_AT, 3), "auth": _auth_state(), "privacy": { "store_raw_content": _store_raw_content(), "store_identifiers": _store_identifiers(), "default": "hash_and_length_only", }, "subscriber_count": subscriber_count, "subscriber_slots": subscriber_slots, "slots": list(DEFAULT_SLOTS), "slot_counts": _slot_counts(), "peer_links": _peer_link_state(), "utility_daemons": load_daemon_registry(CHANNEL), "recent_event_ids": [str(event.get("event_id") or "") for event in recent], "endpoints": { "health": "/health", "state": "/state", "settings": "/settings", "slots": "/slots", "faculties": "/faculties", "providers": "/providers", "heartbeat": "/heartbeat", "daemons": "/daemons", "daemon_match": "/daemons/match?capability=translation&output=text", "links": "/links", "link_register": "/link/register", "events": "/events?slot=personal&limit=50", "sse": "/sse?slot=personal", "sse_all": "/sse?slot=*", "post_event": "/event", "room_create": "/room/create", "whatsapp_config": "/whatsapp/config", "whatsapp_webhook": "/whatsapp/webhook", "council_intent": "/council/intent", "business_intent": "/business/intent", "ipfs_intent": "/ipfs/intent", }, "adapter_posture": { "translation_faculty": translation_faculty_state(), "whatsapp": whatsapp_config_state() | { "mode": "adapter_slot", "slot": "whatsapp", "official_path": "WhatsApp Business Cloud API", "secret_storage": "operator_supplied_environment_or_local_settings", "webhook_inbound_supported": True, "outbound_sender_implemented": False, "send_performed_by_adapter": False, }, }, "nostr_posture": { "mode": "continuum_native_local_first", "relay_published": False, "signing_required_for_local_drafts": False, "note": "Nostr relay/signing is a later layer; this service exposes the local Continuum event stream.", }, "ipfs_posture": { "mode": "optional_archive_intent", "slot": "ipfs", "local_pin_possible": True, "remote_pin_required": False, "cid_claimed": False, "paid_pinning_required": False, "note": "IPFS is free as a protocol. Persistence requires local pinning, peer pinning, or a pinning provider.", }, } def _settings_state() -> dict[str, Any]: state = _state() return { "status": "ok", "schema": "champion-continuum/link-settings/v1", "service": state["service"], "root": state["root"], "channel": state["channel"], "mode": { "server": "single_local_link_service", "slots": state["slots"], "peer_link_capacity": _peer_link_state()["max_links"], "peer_link_count": _peer_link_state()["count"], }, "auth": state["auth"], "privacy": state["privacy"], "providers": provider_registry_state(), "faculties": translation_faculty_state(), "utility_daemons": load_daemon_registry(CHANNEL), "peer_links": _peer_link_state(), "endpoints": state["endpoints"], "operator_notes": [ "One local link service can stream many slots.", "Up to five peer SSE targets can be registered as metadata.", "Remote peer tokens are represented by hashes by default.", "External transport, relay publish, and IPFS pinning remain approval-gated.", ], } def _new_room_code() -> str: return re.sub(r"[^A-Za-z0-9]", "", secrets.token_urlsafe(8)).upper()[:10] def _build_room_session(payload: dict[str, Any]) -> tuple[str, dict[str, Any]]: room_code = str(payload.get("room_code") or _new_room_code()).strip().upper() room_code = re.sub(r"[^A-Z0-9]", "", room_code)[:16] or _new_room_code() room_slot = _normalize_slot(payload.get("slot") or f"room-{room_code[:8].lower()}") languages = payload.get("languages") if isinstance(payload.get("languages"), dict) else {} requested_slots = payload.get("slots") if isinstance(payload.get("slots"), list) else ["personal", "voice", "whatsapp"] return room_code, { "kind": "continuum.room_session", "slot": room_slot, "payload": { "schema": "champion-continuum/room-session/v1", "room_code_sha256": _hash_string(room_code), "room_slot": room_slot, "room_label": _privacy_scrub(str(payload.get("room_label") or "Continuum conversation room"), "raw_message"), "speaker_label": _privacy_scrub(str(payload.get("speaker_label") or "Speaker A"), "recipient"), "listener_label": _privacy_scrub(str(payload.get("listener_label") or "Speaker B"), "recipient"), "source_lang": str(languages.get("source") or payload.get("source_lang") or "auto"), "target_lang": str(languages.get("target") or payload.get("target_lang") or "en-US"), "relationship_tone": str(payload.get("relationship_tone") or "warm and clear"), "allowed_slots": [_normalize_slot(slot) for slot in requested_slots], "join_paths": { "sse": f"/sse?slot={room_slot}", "events": f"/events?slot={room_slot}&limit=50", "auth": "Continuum link token required", }, "message_sent": False, "public_invite_published": False, "operator_approval_required": True, }, } def _build_heartbeat(payload: dict[str, Any]) -> dict[str, Any]: capabilities = payload.get("capabilities") if not isinstance(capabilities, list): capabilities = [] return { "kind": "continuum.heartbeat", "slot": _normalize_slot(payload.get("slot") or "control"), "payload": { "schema": "champion-continuum/heartbeat/v1", "component": str(payload.get("component") or "continuum-client"), "status": str(payload.get("status") or "ready"), "ttl_seconds": int(payload.get("ttl_seconds") or 30), "capabilities": [str(item) for item in capabilities[:32]], "note": _privacy_scrub(str(payload.get("note") or ""), "raw_message"), "external_effects_performed": False, }, } class ContinuumLinkHandler(BaseHTTPRequestHandler): server_version = "ChampionContinuumLink/0.1" def log_message(self, format: str, *args: Any) -> None: print(f"[continuum-link] {self.client_address[0]} - {format % args}") def _origin(self) -> str: return str(self.headers.get("Origin") or "").strip() def _origin_allowed(self) -> bool: origin = self._origin() return not origin or origin in _allowed_origins() def _set_cors_headers(self) -> None: origin = self._origin() if origin and origin in _allowed_origins(): self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Vary", "Origin") elif not origin: self.send_header("Access-Control-Allow-Origin", "http://127.0.0.1:7870") self.send_header("Vary", "Origin") def _auth_token_from_request(self, parsed: Any | None = None) -> str: auth = str(self.headers.get("Authorization") or "").strip() if auth.lower().startswith("bearer "): return auth[7:].strip() header_token = str(self.headers.get("X-Continuum-Token") or "").strip() if header_token: return header_token if parsed is not None: query_token = (parse_qs(parsed.query).get("token") or [""])[0] if query_token: return query_token return "" def _authorized(self, parsed: Any | None = None) -> bool: return hmac.compare_digest(self._auth_token_from_request(parsed), AUTH_TOKEN) def _send_auth_error(self) -> None: self._send_json({"status": "error", "error": "unauthorized", "auth": _auth_state()}, status=401) def _send_json(self, payload: Any, status: int = 200) -> None: raw = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self._set_cors_headers() self.send_header("Access-Control-Allow-Headers", "content-type, authorization, x-continuum-token, x-hub-signature-256") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.end_headers() self.wfile.write(raw) def _send_sse(self, event: dict[str, Any], event_name: str = "continuum") -> None: event_id = str(event.get("event_id") or "") data = json.dumps(event, ensure_ascii=False) self.wfile.write(f"id: {event_id}\n".encode("utf-8")) self.wfile.write(f"event: {event_name}\n".encode("utf-8")) for line in data.splitlines() or [""]: self.wfile.write(f"data: {line}\n".encode("utf-8")) self.wfile.write(b"\n") self.wfile.flush() def do_OPTIONS(self) -> None: if not self._origin_allowed(): self._send_json({"status": "error", "error": "origin_not_allowed"}, status=403) return self.send_response(204) self._set_cors_headers() self.send_header("Access-Control-Allow-Headers", "content-type, authorization, x-continuum-token, x-hub-signature-256") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.end_headers() def do_GET(self) -> None: parsed = urlparse(self.path) if not self._origin_allowed(): self._send_json({"status": "error", "error": "origin_not_allowed"}, status=403) return if parsed.path == "/health": self._send_json( { "status": "ok", "service": "champion-continuum-link", "uptime_seconds": time.time() - STARTED_AT, "auth_required": True, "token_sha256_prefix": _auth_digest(), } ) return if parsed.path != "/whatsapp/webhook" and not self._authorized(parsed): self._send_auth_error() return if parsed.path == "/state": self._send_json(_state()) return if parsed.path == "/settings": self._send_json(_settings_state()) return if parsed.path == "/slots": self._send_json({"status": "ok", "slots": list(DEFAULT_SLOTS), "slot_counts": _slot_counts()}) return if parsed.path == "/faculties": self._send_json({"status": "ok", "translation_faculty": translation_faculty_state()}) return if parsed.path == "/providers": self._send_json({"status": "ok", "provider_registry": provider_registry_state()}) return if parsed.path == "/daemons": self._send_json(load_daemon_registry(CHANNEL)) return if parsed.path == "/daemons/match": query = parse_qs(parsed.query) capability = (query.get("capability") or [""])[0] output = (query.get("output") or [""])[0] include_stale = (query.get("include_stale") or ["0"])[0].lower() in {"1", "true", "yes", "on"} self._send_json(match_daemons(capability=capability, output=output, include_stale=include_stale)) return if parsed.path == "/links": self._send_json({"status": "ok", "peer_links": _peer_link_state()}) return if parsed.path == "/whatsapp/config": self._send_json({"status": "ok", "config": whatsapp_config_state()}) return if parsed.path == "/whatsapp/webhook": query = parse_qs(parsed.query) mode = (query.get("hub.mode") or [""])[0] challenge = (query.get("hub.challenge") or [""])[0] token = (query.get("hub.verify_token") or [""])[0] expected = os.environ.get("WHATSAPP_VERIFY_TOKEN") or "" if mode == "subscribe" and expected and hmac.compare_digest(token, expected): raw = challenge.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/plain; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) return self._send_json({"status": "error", "error": "webhook_verification_failed"}, status=403) return if parsed.path == "/events": query = parse_qs(parsed.query) limit = int((query.get("limit") or ["50"])[0]) slot = (query.get("slot") or [ALL_SLOTS])[0] self._send_json({"status": "ok", "slot": slot, "events": _recent_events(limit=limit, slot=slot)}) return if parsed.path == "/sse": query = parse_qs(parsed.query) slot = (query.get("slot") or ["personal"])[0] subscription_slot = ALL_SLOTS if slot == ALL_SLOTS else _normalize_slot(slot) self.send_response(200) self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "keep-alive") self._set_cors_headers() self.end_headers() subscriber: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=MAX_BACKLOG) with SUBSCRIBERS_LOCK: SUBSCRIBERS[subscriber] = subscription_slot try: self.wfile.write(b": champion-continuum-link connected\n\n") self.wfile.flush() self._send_sse( _make_event( {"kind": "continuum.state", "slot": "control", "payload": _state()}, source="link-server", ), "state", ) for event in _recent_events(limit=25, slot=subscription_slot): self._send_sse(event) while True: try: event = subscriber.get(timeout=15) except queue.Empty: self.wfile.write(b": heartbeat\n\n") self.wfile.flush() continue self._send_sse(event) except (BrokenPipeError, ConnectionResetError, OSError): pass finally: with SUBSCRIBERS_LOCK: SUBSCRIBERS.pop(subscriber, None) return self._send_json({"status": "error", "error": "not_found", "path": parsed.path}, status=404) def do_POST(self) -> None: parsed = urlparse(self.path) if not self._origin_allowed(): self._send_json({"status": "error", "error": "origin_not_allowed"}, status=403) return if parsed.path not in { "/event", "/heartbeat", "/link/register", "/room/create", "/whatsapp/webhook", "/council/intent", "/business/intent", "/ipfs/intent", }: self._send_json({"status": "error", "error": "not_found", "path": parsed.path}, status=404) return if parsed.path != "/whatsapp/webhook" and not self._authorized(parsed): self._send_auth_error() return try: length = int(self.headers.get("Content-Length") or "0") except ValueError: length = 0 if length <= 0: self._send_json({"status": "error", "error": "empty_body"}, status=400) return if length > MAX_BODY_BYTES: self._send_json({"status": "error", "error": "body_too_large", "max_bytes": MAX_BODY_BYTES}, status=413) return raw = self.rfile.read(length) try: payload = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: self._send_json({"status": "error", "error": "invalid_json", "detail": str(exc)}, status=400) return if not isinstance(payload, dict): self._send_json({"status": "error", "error": "object_required"}, status=400) return if parsed.path == "/whatsapp/webhook": app_secret = os.environ.get("WHATSAPP_APP_SECRET") or "" signature_required = bool(getattr(self.server, "require_whatsapp_signature", False)) signature_checked = bool(app_secret) signature_valid = ( verify_meta_signature(raw, app_secret, self.headers.get("X-Hub-Signature-256")) if signature_checked else None ) if signature_required and not app_secret: self._send_json({"status": "error", "error": "whatsapp_app_secret_required"}, status=503) return if signature_checked and not signature_valid: self._send_json({"status": "error", "error": "invalid_meta_signature"}, status=403) return events = [ _make_event(item, source="whatsapp-webhook") for item in normalize_webhook( payload, store_raw=_store_raw_content(), store_identifiers=_store_identifiers(), ) ] append_errors: list[str] = [] for event in events: append_error = _append_event(event) if append_error: append_errors.append(append_error) if append_errors: self._send_json({"status": "error", "error": "event_log_write_failed", "details": append_errors}, status=500) return self._send_json( { "status": "ok", "received": len(events), "signature_required": signature_required, "signature_checked": signature_checked, "signature_valid": signature_valid, "events": events, }, status=202, ) return if parsed.path == "/heartbeat": event = _make_event(_build_heartbeat(payload), source="heartbeat") append_error = _append_event(event) if append_error: self._send_json({"status": "error", "error": "event_log_write_failed", "detail": append_error}, status=500) return self._send_json({"status": "ok", "event": event}, status=201) return if parsed.path == "/link/register": peer = _build_peer_link(payload) if not peer["url"].startswith(("http://", "https://")): self._send_json({"status": "error", "error": "valid_sse_url_required"}, status=400) return links = [item for item in _load_peer_links() if item.get("link_id") != peer["link_id"]] if len(links) >= MAX_PEER_LINKS: self._send_json({"status": "error", "error": "max_peer_links_reached", "max_links": MAX_PEER_LINKS}, status=409) return links.append(peer) _save_peer_links(links) event = _make_event( { "kind": "continuum.peer_link.registered", "slot": "control", "payload": { "link_id": peer["link_id"], "label": peer["label"], "url": peer["url"], "default_slot": peer["default_slot"], "token_present": peer["token_present"], "external_connection_opened": False, }, }, source="peer-link", ) append_error = _append_event(event) if append_error: self._send_json({"status": "error", "error": "event_log_write_failed", "detail": append_error}, status=500) return self._send_json({"status": "ok", "peer_link": peer, "event": event}, status=201) return if parsed.path == "/room/create": room_code, room_payload = _build_room_session(payload) event = _make_event(room_payload, source="room-session") append_error = _append_event(event) if append_error: self._send_json({"status": "error", "error": "event_log_write_failed", "detail": append_error}, status=500) return self._send_json( { "status": "ok", "room_code": room_code, "room_slot": event.get("slot"), "join_paths": event.get("payload", {}).get("join_paths", {}), "event": event, }, status=201, ) return if parsed.path == "/council/intent": event = _make_event( { "kind": str(payload.get("kind") or "council.agent_intent"), "slot": "council", "payload": { "agent": str(payload.get("agent") or "operator-selected-council"), "task": _privacy_scrub(str(payload.get("task") or ""), "raw_message"), "capability": str(payload.get("capability") or "observe_draft_review"), "requested_autonomy": str(payload.get("requested_autonomy") or "draft_only"), "allowed_external_effects": False, "operator_approval_required": True, "merkle_receipt_required": True, "cascade_lattice_receipt_required": True, "result": _privacy_scrub(payload.get("result") or {}, "result"), }, }, source="council-intent", ) append_error = _append_event(event) if append_error: self._send_json({"status": "error", "error": "event_log_write_failed", "detail": append_error}, status=500) return self._send_json({"status": "ok", "event": event}, status=201) return if parsed.path == "/business/intent": requested_lanes = payload.get("lanes") if not isinstance(requested_lanes, list): requested_lanes = ["bridge", "reputation", "ipfs"] event = _make_event( { "kind": "business.correspondence_directive_intent", "slot": "business", "payload": { "directive": _privacy_scrub(str(payload.get("directive") or ""), "raw_message"), "agent": str(payload.get("agent") or "operator-selected-council"), "lanes": [str(lane) for lane in requested_lanes], "bridge_role": "local_translation_and_correspondence_artifact", "reputation_role": "receipt_and_trust_weight", "ipfs_role": "optional_redacted_receipt_archive", "marketplace_role": "optional_listing_or_service_offer", "operator_approval_required": True, "external_effects_performed": False, "message_sent": False, "relay_published": False, "ipfs_pinned": False, "next_drafts": { "ipfs_intent": "/ipfs/intent", "council_intent": "/council/intent", }, }, }, source="business-intent", ) append_error = _append_event(event) if append_error: self._send_json({"status": "error", "error": "event_log_write_failed", "detail": append_error}, status=500) return self._send_json({"status": "ok", "event": event}, status=201) return if parsed.path == "/ipfs/intent": event = _make_event( { "kind": "ipfs.archive_intent", "slot": "ipfs", "payload": { "cid": str(payload.get("cid") or ""), "content_sha256": str(payload.get("content_sha256") or ""), "content_label": _privacy_scrub(str(payload.get("content_label") or ""), "raw_message"), "local_pin_requested": bool(payload.get("local_pin_requested", True)), "remote_pin_requested": bool(payload.get("remote_pin_requested", False)), "provider": str(payload.get("provider") or "local_ipfs_node"), "paid_provider_required": False, "operator_approval_required": True, "archive_performed": False, }, }, source="ipfs-intent", ) append_error = _append_event(event) if append_error: self._send_json({"status": "error", "error": "event_log_write_failed", "detail": append_error}, status=500) return self._send_json({"status": "ok", "event": event}, status=201) return event = _make_event(payload, source="http") append_error = _append_event(event) if append_error: self._send_json({"status": "error", "error": "event_log_write_failed", "detail": append_error}, status=500) return self._send_json({"status": "ok", "event": event}, status=201) def main() -> int: parser = argparse.ArgumentParser(description="Local Champion Continuum SSE/link service.") parser.add_argument("--host", default=os.environ.get("CONTINUUM_LINK_HOST", "127.0.0.1")) parser.add_argument("--port", type=int, default=int(os.environ.get("CONTINUUM_LINK_PORT", "7871"))) args = parser.parse_args() CHANNEL.mkdir(parents=True, exist_ok=True) server = ThreadingHTTPServer((args.host, args.port), ContinuumLinkHandler) server.require_whatsapp_signature = _webhook_signature_required(args.host) print(f"Champion Continuum link service listening on http://{args.host}:{args.port}/sse") print(f"Continuum link auth token source: {AUTH_TOKEN_SOURCE}; sha256 prefix: {_auth_digest()}") if AUTH_TOKEN_SOURCE == "ephemeral": print("Set CONTINUUM_LINK_TOKEN to a stable local secret before using external adapters.") try: server.serve_forever(poll_interval=0.5) except KeyboardInterrupt: print("Champion Continuum link service stopping") finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())