""" signalmesh_api.py — SignalMesh Sovereign Liquid Matrix Gateway ======================================================= Public API: /api/* — requires X-SignalMesh-Key header Dashboard: /ui/* — server-side key, no browser exposure Landing: / — interactive HTML dashboard """ from __future__ import annotations import asyncio, hashlib, html, json, os, sys, time, re import urllib.parse from contextlib import asynccontextmanager from pathlib import Path from typing import Any, Dict, List, Optional _HERE = Path(__file__).parent sys.path.insert(0, str(_HERE)) import types _app_pkg = types.ModuleType("app") _log_pkg = types.ModuleType("app.logger") class _L: def info(self,*a,**k): pass def warning(self,*a,**k): pass def error(self,*a,**k): pass def debug(self,*a,**k): pass _log_pkg.logger = _L() sys.modules.setdefault("app", _app_pkg) sys.modules.setdefault("app.logger", _log_pkg) _app_pkg.logger = _log_pkg.logger import feedparser, httpx from fastapi import FastAPI, Header, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from pydantic import BaseModel from core.signal_registry import SignalRegistry, signal_registry from core.spatial_grid import SpatialGridManager from core.feed_hydrator import feed_hydrator from core.source_hydrator import source_hydrator from core import mport_articles from core import binder from core import world_state from core import broadcast_journal from core.antenna import AntennaForge, TARGETS as ANTENNA_TARGETS from core.manifest_parser import manifest_registry, ManifestParser from core.sec_omega import guarded_broadcast, sec_omega from core.coordinate_resolver import coordinate_resolver, NAMESPACES, SLOTS from core.relationship_graph import relationship_graph from core import submesh as mport from tools.autonomous_pumps import adys_pump grid = SpatialGridManager(index_path=str(_HERE / "workspace" / "spatial_index.json")) _ROWS = SpatialGridManager.GRID_ROWS _COLS = SpatialGridManager.GRID_COLS _PROT = ("security_","sql_","auth_","key_","secret_","credential_") _SEC_Q: List[Dict[str,Any]] = [] antenna_forge = AntennaForge(signal_registry) # ── Key tier system ─────────────────────────────────────────────────────────── # Tiers: admin (unlimited) | paid (unlimited) | update (update-channel only). # No public free tier. The legacy demo key lives on permanently as the OCODX # update-channel key: shipped builds poll release/update signals with it # (submesh reads + ocodx.* tune_in) and nothing else. _UPDATE_KEY = os.environ.get("SIGNALMESH_UPDATE_KEY", "smesh-free-demo") _ADMIN_KEY = os.environ.get("SIGNALMESH_API_KEY", "") _PAID_KEYS = set(filter(None, os.environ.get("SIGNALMESH_PAID_KEYS", "").split(","))) def _key_values(raw: str) -> set[str]: keys = set() for part in raw.split(","): part = part.strip() if not part: continue label, sep, value = part.partition(":") keys.add((value if sep else label).strip()) return {k for k in keys if k} def _owner_keys() -> set[str]: return ( _key_values(os.environ.get("SIGNALMESH_OWNER_KEYS", "")) | _key_values(os.environ.get("SIGNALMESH_OWNER_KEY", "")) | _key_values(os.environ.get("OWNER_KEY", "")) ) # Rate limit: the update key is shared by every shipped build _UPDATE_RATE_LIMIT = 5000 # calls per rolling 24h window _update_usage: Dict[str, List[float]] = {} # key → list of call timestamps def _tier(k: Optional[str]) -> str: if k and (k == _ADMIN_KEY or k in _owner_keys()): return "admin" if k and k in _PAID_KEYS: return "paid" if k and k == _UPDATE_KEY: return "update" return "none" def _check_key(k: Optional[str], require_paid: bool = False, allow_update: bool = False): t = _tier(k) if t == "none": if k and _db_get_by_key(k): return # valid DB paid user raise HTTPException(401, detail={ "error": "Invalid or missing X-SignalMesh-Key", "get_paid_key": "https://kyklos.io/apps/signalmesh/", }) if t == "update": if not allow_update or require_paid: raise HTTPException(403, detail={ "error": "This key is restricted to the OCODX update channel", "get_paid_key": "https://kyklos.io/apps/signalmesh/", }) _rate_check(k) def _rate_check(k: str): now = time.time() window = now - 86400 # 24h hits = [ts for ts in _update_usage.get(k, []) if ts > window] if len(hits) >= _UPDATE_RATE_LIMIT: raise HTTPException(429, detail={ "error": f"Update-channel limit ({_UPDATE_RATE_LIMIT} calls/24h) reached", }) hits.append(now) _update_usage[k] = hits # ── User database (SQLite + HF dataset backup) ──────────────────────────────── import sqlite3, secrets, threading # Persist outside /tmp where possible: HF Spaces persistent storage is /data. # Falls back to /tmp (still backed up to the HF dataset) if /data isn't writable. _DB_DIR = os.environ.get("SIGNALMESH_DB_DIR") or ( "/data" if os.path.isdir("/data") and os.access("/data", os.W_OK) else "/tmp") _DB_PATH = os.path.join(_DB_DIR, "signalmesh_users.db") _DB_LOCK = threading.Lock() def _db_conn(): conn = sqlite3.connect(_DB_PATH) conn.row_factory = sqlite3.Row return conn def _db_init(): try: from huggingface_hub import hf_hub_download import shutil p = hf_hub_download("acecalisto3/signalmesh-users", "users.db", repo_type="dataset", local_dir="/tmp") shutil.copy(p, _DB_PATH) print("[UserDB] Loaded from HF dataset") except Exception: print("[UserDB] Starting fresh") with _DB_LOCK: conn = _db_conn() conn.execute("""CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, email TEXT NOT NULL, api_key TEXT UNIQUE NOT NULL, tier TEXT DEFAULT 'paid', stripe_customer_id TEXT, stripe_session_id TEXT, active INTEGER DEFAULT 1, created_at REAL, last_seen REAL, call_count INTEGER DEFAULT 0 )""") conn.commit(); conn.close() def _db_backup(): try: from huggingface_hub import HfApi _hf = HfApi() try: _hf.create_repo("signalmesh-users", repo_type="dataset", private=True) except Exception: pass _hf.upload_file(path_or_fileobj=_DB_PATH, path_in_repo="users.db", repo_id="acecalisto3/signalmesh-users", repo_type="dataset", commit_message="auto-backup users.db") except Exception as e: print(f"[UserDB] Backup failed: {e}") def _db_create_user( email: str, stripe_customer_id: str, stripe_session_id: str, tier: str = "managed_monthly", ) -> str: key = "smesh-" + secrets.token_urlsafe(24) uid = secrets.token_hex(8) with _DB_LOCK: conn = _db_conn() conn.execute("INSERT OR IGNORE INTO users VALUES (?,?,?,?,?,?,1,?,?,0)", (uid, email, key, tier, stripe_customer_id, stripe_session_id, time.time(), time.time())) conn.commit(); conn.close() threading.Thread(target=_db_backup, daemon=True).start() return key def _db_get_by_key(key: str): with _DB_LOCK: conn = _db_conn() row = conn.execute("SELECT * FROM users WHERE api_key=? AND active=1", (key,)).fetchone() if row: conn.execute("UPDATE users SET last_seen=?,call_count=call_count+1 WHERE api_key=?", (time.time(), key)) conn.commit() conn.close() return dict(row) if row else None def _dev_from_key(key: str) -> int: """Read the device quota out of a signed OCODX1.. token.""" try: import base64 as _b64 pb = key.split(".")[1] data = json.loads(_b64.urlsafe_b64decode(pb + "=" * (-len(pb) % 4))) return int(data.get("dev", 0)) except Exception: return 0 def _db_get_by_email(email: str): with _DB_LOCK: conn = _db_conn() row = conn.execute("SELECT * FROM users WHERE email=? AND active=1 " "ORDER BY created_at DESC LIMIT 1", (email,)).fetchone() conn.close() return dict(row) if row else None def _db_add_seats(email: str, add: int, cid: str, sid: str) -> str: """Upgrade an existing license in place: re-mint the same email's key with dev = current + add. If they have no base license yet, treat the add as solo+add.""" cur = _db_get_by_email(email) if cur: tier = cur.get("tier", "solo") base = _dev_from_key(cur.get("api_key", "")) or TIER_DEVICES.get(tier, 1) else: tier, base = "solo", TIER_DEVICES.get("solo", 1) new_dev = base + max(0, add) key = _mint_ocodx_key(email, tier, new_dev) uid = (cur or {}).get("id") or secrets.token_hex(8) with _DB_LOCK: conn = _db_conn() if cur: conn.execute("UPDATE users SET api_key=?,stripe_session_id=?,last_seen=? WHERE id=?", (key, sid, time.time(), uid)) else: conn.execute("INSERT OR IGNORE INTO users VALUES (?,?,?,?,?,?,1,?,?,0)", (uid, email, key, tier, cid, sid, time.time(), time.time())) conn.commit(); conn.close() threading.Thread(target=_db_backup, daemon=True).start() return key def _db_get_by_session(session_id: str): with _DB_LOCK: conn = _db_conn() row = conn.execute("SELECT * FROM users WHERE stripe_session_id=?", (session_id,)).fetchone() conn.close() return dict(row) if row else None def _db_revoke(stripe_customer_id: str): with _DB_LOCK: conn = _db_conn() conn.execute("UPDATE users SET active=0 WHERE stripe_customer_id=?", (stripe_customer_id,)) conn.commit(); conn.close() threading.Thread(target=_db_backup, daemon=True).start() def _db_all() -> list: with _DB_LOCK: conn = _db_conn() rows = conn.execute("SELECT * FROM users ORDER BY created_at DESC").fetchall() conn.close() return [dict(r) for r in rows] def _socket_limit(key: Optional[str]) -> int: """Resolve the number of public MPort edge sockets a key may own.""" if key and (key == _ADMIN_KEY or key in _owner_keys()): return int(os.environ.get("SIGNALMESH_ADMIN_SOCKET_LIMIT", "1000")) if key and key in _PAID_KEYS: return int(os.environ.get("SIGNALMESH_STATIC_KEY_SOCKET_LIMIT", "1")) if key: user = _db_get_by_key(key) if user: # Current managed plans include one public relay. Additional relay # products can raise this without changing the MPort protocol. return max(1, int(user.get("socket_limit", 1) or 1)) return 0 _db_init() # ── OpenAPI tag groups ──────────────────────────────────────────────────────── _TAGS = [ {"name": "Core", "description": "Broadcast signals, tune in, check mesh status."}, {"name": "Feeds", "description": "RSS/Atom hydration — auto-broadcast feed items into the mesh."}, {"name": "Manifest", "description": "AGENTS.md ingestion — register external nodes by URL."}, {"name": "Coordinates", "description": "Addressable memory grid — store and resolve AA01-style coordinates."}, {"name": "Trails", "description": "Fuzzy keyword trails — self-healing synonym paths the mesh learns from misses."}, {"name": "Grid", "description": "72-node spatial grid state and SEC-Ω quarantine queue."}, {"name": "MPort", "description": "Discover, simulate, approve, and execute mesh-native capabilities."}, ] SITE_INGEST_DEFAULT = ",".join([ "nexus.overview=https://www.nexusaifirst.com/", "nexus.modules.roadmap=https://www.nexusaifirst.com/modules", "nexus.tech.architecture=https://www.nexusaifirst.com/docs/technical", "nexus.industries=https://www.nexusaifirst.com/industries", ]) def _page_text(html_raw: str, base: str) -> str: """Strip to prose but keep the hrefs — a signal with no link can't be cited.""" # Chrome first: nav/header/footer repeat on every page and otherwise dominate chunk 1. body = re.sub(r"<(script|style|nav|header|footer|aside|form)\b[\s\S]*?", " ", html_raw, flags=re.I) body = re.sub(r'<[a-z]+[^>]*\b(?:class|id)="[^"]*\b(?:nav|navbar|menu|footer|header|cookie|banner)\b[^"]*"[^>]*>', " ", body, flags=re.I) links = [] for href, label in re.findall(r']+href="([^"]+)"[^>]*>([\s\S]{0,80}?)', body, re.I): label = " ".join(re.sub(r"<[^>]+>", " ", label).split()) if label and href.startswith(("http", "/")) and not href.startswith("#"): links.append(f"{label} <{urllib.parse.urljoin(base, href)}>") text = " ".join(html.unescape(re.sub(r"<[^>]+>", " ", body)).split()) seen, uniq = set(), [] for l in links: if l not in seen: seen.add(l); uniq.append(l) return text + (" Links: " + " · ".join(uniq[:25]) if uniq else "") def _chunks(text: str, size: int = 360) -> List[str]: """_do_tune_in renders each signal at [:400]; anything past that is unreachable.""" out, cur = [], "" for w in text.split(): if len(cur) + len(w) + 1 > size: out.append(cur); cur = w else: cur = f"{cur} {w}".strip() if cur: out.append(cur) return out async def _ingest_sites() -> None: """Re-broadcast configured doc sites on boot — signals live in memory and a Space restart wipes them, so a one-shot broadcast does not survive a rebuild.""" spec = os.environ.get("SITE_INGEST", SITE_INGEST_DEFAULT).strip() if not spec: return async with httpx.AsyncClient(timeout=25.0, follow_redirects=True) as client: for pair in spec.split(","): freq, _, url = pair.strip().partition("=") if not (freq and url): continue try: r = await client.get(url, headers={"User-Agent": "SignalMesh/1.0"}) if r.status_code != 200: print(f"[site_ingest] {freq}: HTTP {r.status_code}"); continue parts = _chunks(_page_text(r.text, url)[:12000]) for p in parts: signal_registry.broadcast(name=freq, source_type="site_docs", data=f"{p} (source: {url})", metadata={"source_url": url}) print(f"[site_ingest] {freq}: {len(parts)} signals from {url}") except Exception as e: print(f"[site_ingest] {freq}: {str(e)[:120]}") @asynccontextmanager async def _lifespan(_app: FastAPI): binder.restore() # identity survives a rebuild broadcast_journal.replay(signal_registry) # ad-hoc signals survive a restart # Re-register runtime-added sources BEFORE the hydrators start, so they are # polled exactly like the curated ones instead of returning as dead history. broadcast_journal.replay_sources(feed_hydrator, source_hydrator) feed_hydrator.start() source_hydrator.start() antenna_forge.registry = signal_registry adys_pump.start() mport.load_state( try_hf=os.environ.get("SIGNALMESH_MPORT_RESTORE", "1") == "1" ) mport.broadcast_nicoli_catalog_now(signal_registry) persistence_task = asyncio.create_task(mport.persistence_loop()) ingest_task = asyncio.create_task(_ingest_sites()) try: yield finally: persistence_task.cancel(); ingest_task.cancel() app = FastAPI( title="SignalMesh — Sovereign Liquid Matrix Gateway", description=""" Ambient context bus for AI agent fleets. Agents broadcast to named frequencies; other agents tune in by keyword — no config, no service registry, no tool calls. ## Authentication All `/api/*` endpoints require an `X-SignalMesh-Key` header. | Tier | Key | Limits | |------|-----|--------| | **Managed** | personal key | Hosted API + one public MPort relay | [Get managed access →](https://kyklos.io/apps/signalmesh/#pricing) ## Quick start ```bash # Broadcast a signal curl -X POST https://acecalisto3-signalmesh.hf.space/api/broadcast \\ -H "X-SignalMesh-Key: YOUR_KEY" \\ -H "Content-Type: application/json" \\ -d '{"frequency":"my_agent","content":"hello mesh","source_type":"demo"}' # Tune in curl -X POST https://acecalisto3-signalmesh.hf.space/api/tune_in \\ -H "X-SignalMesh-Key: YOUR_KEY" \\ -H "Content-Type: application/json" \\ -d '{"keywords":["my_agent"]}' ``` """, version="2.0.0", docs_url="/docs", redoc_url=None, openapi_tags=_TAGS, lifespan=_lifespan, ) @app.get("/redoc", include_in_schema=False) async def redoc_html(): from fastapi.responses import HTMLResponse return HTMLResponse(""" SignalMesh API Docs """) _cors_origins = [ origin.strip() for origin in os.environ.get("SIGNALMESH_CORS_ORIGINS", "").split(",") if origin.strip() ] app.add_middleware( CORSMiddleware, allow_origins=_cors_origins or ["*"], allow_credentials=bool(_cors_origins), allow_methods=["*"], allow_headers=["*"], ) app.include_router( mport.make_router( check_key=_check_key, signal_registry=signal_registry, sec_omega=sec_omega, socket_limit=_socket_limit, ) ) # ── the editor ──────────────────────────────────────────────────────────────── # Mounted, not merged. It is a complete app — its own sessions, its own ACL read # from the client's Joomla, its own write adapter — and mounting keeps a single # implementation: the same code runs standalone on the machine that edits one # site, and here as a Studio tab. # # The prefix also settles the route collisions for free: both apps define # /healthz and /api/media, and neither has to give up its name. # The import is guarded: a build that ships without the package loses the tab, # not the Studio. try: import editor as _editor # noqa: E402 _editor_app = _editor.build() except ImportError as e: _editor_app = None print(f"[editor] not present, tab disabled: {e}", flush=True) if _editor_app is not None: app.mount("/api/editor", _editor_app) print("[editor] mounted at /api/editor", flush=True) # ── Models ──────────────────────────────────────────────────────────────────── class BroadcastReq(BaseModel): frequency: str; content: str bypass_gate: bool = False; source_type: str = "external" # rolling diff / provenance (all optional) change_id: int = 0; parent_change: int = 0 origin_agent: str = ""; change_type: str = "" files: List[str] = []; summary: str = "" ttl_ms: int = 0 class AntennaReq(BaseModel): frequency: str target: str name: Optional[str] = None class TuneInReq(BaseModel): keywords: List[str] role: Optional[str] = None # role-scoped delivery limit: int = 0 # 0 = every match; else newest N class CoordinateStoreReq(BaseModel): coordinate: str; data: Any symbolic: str = ""; change_id: int = 0; ttl_ms: int = 0 class IdentitySeedReq(BaseModel): canonical_locator: str revision: str = "" deployed_as: str = "" lineage_fingerprint: str = "" class RelationshipRoleReq(BaseModel): role: str; patterns: List[str] class RssSyncReq(BaseModel): url: str class ContextDiscoveryReq(BaseModel): text: str class HfInjectReq(BaseModel): repo_id: str repo_type: str = "model" # model | dataset | space class ArxivInjectReq(BaseModel): arxiv_id: str # e.g. "2303.08774" class FeedAddReq(BaseModel): url: str frequency: str interval: int = 3600 label: str = "" class ArticleFetchReq(BaseModel): url: str with_images: bool = False class ArticleSnapshotReq(BaseModel): frequency: str limit: int = 5 with_images: bool = False class FeedDiscoverReq(BaseModel): url: str class ManifestIngestReq(BaseModel): url: str # raw URL to an AGENTS.md file # ── Internal helpers ────────────────────────────────────────────────────────── def _node_idx(freq: str) -> int: return int(hashlib.sha256(freq.encode()).hexdigest(), 16) % (_ROWS * _COLS) def _agent_for(freq: str): idx = _node_idx(freq); r,c = divmod(idx, _COLS) return idx, grid.coords_to_agent.get((r,c), f"node-{idx}") def _mesh_size() -> int: return sum(len(v) for v in signal_registry.streams.values()) def _is_prot(freq: str) -> bool: return any(freq.lower().startswith(p) for p in _PROT) # ── Core logic (no auth — called by both /api/ and /ui/ routes) ─────────────── def _do_status(): return {"status":"live","signals_in_mesh":_mesh_size(), "active_frequencies":len(signal_registry.streams), "grid_nodes":_ROWS*_COLS,"quarantined":len(_SEC_Q),"version":"2.0.0", "mesh":"Sovereign Liquid Matrix"} def _do_broadcast(req: BroadcastReq): idx, agent = _agent_for(req.frequency) if _is_prot(req.frequency) and not req.bypass_gate: _SEC_Q.append({"frequency":req.frequency,"content":req.content, "source_type":req.source_type,"node":idx,"agent":agent, "queued_at":time.time()}) return {"status":"quarantined","node":idx,"agent":agent, "frequency":req.frequency, "reason":"Protected frequency — SEC-Ω staged. Use bypass_gate=true to commit."} result = guarded_broadcast(signal_registry, req.frequency, req.content, source_type=req.source_type, metadata={"node":idx,"agent":agent,"source":"external_api"}, ttl_ms=req.ttl_ms, change_id=req.change_id, parent_change=req.parent_change, origin_agent=req.origin_agent, change_type=req.change_type, files=req.files, summary=req.summary) if not result["ok"]: raise HTTPException(400, f"SEC-Ω blocked: {result['reason']}") return {"status":"live","node":idx,"agent":agent,"frequency":req.frequency, "fingerprint":result["fingerprint"],"signals_in_mesh":_mesh_size()} def _do_tune_in(keywords: List[str], role: Optional[str] = None, limit: int = 0): t0 = time.perf_counter() signals = signal_registry.tune_in(keywords) if role: signals = relationship_graph.filter_signals(role, signals) # ponytail: cap after role filtering — signals arrive newest-first, so the # slice keeps the freshest N rather than whatever survived the filter. if limit: signals = signals[:limit] us = (time.perf_counter()-t0)*1e6 # Collect any fuzzy trail bridges that fired during this call trails_fired = {} for s in signals: t = s.get("_trail") if t: via = next((k for k in keywords if k in signal_registry.fuzzy_trails), None) if via: trails_fired[via] = t if signals: lines = ["[SignalMesh — Live Context]"] for s in signals: age = round(s.get("age", 0), 1) prov = "" if s.get("change_id"): prov = f" | Δ{s['change_id']} by {s.get('origin_agent','?')} ({s.get('change_type','')})" trail_note = "" if s.get("_trail"): tr = s["_trail"] trail_note = f" [⟿ trail: {tr.get('bridged_to','?')} @ {tr.get('confidence','?')}]" lines.append(f"• [{s['name']}] ({s['type']}, {age}s ago{prov}{trail_note}): {str(s['content'])[:400]}") ctx = "\n".join(lines) else: ctx = "" # Check if any keyword was a pure miss (trail written but no signals yet on that frequency) misses = [] for kw in keywords: if kw in signal_registry.fuzzy_trails and not any( s["name"] == signal_registry.fuzzy_trails[kw][0] for s in signals ): best = signal_registry.fuzzy_trails[kw][0] conf = signal_registry.trail_gaps.get(kw, {}).get(best, 0) misses.append({"keyword": kw, "bridged_to": best, "confidence": conf, "branches": signal_registry.fuzzy_trails[kw][1:]}) return { "context": ctx, "signals_matched": len(signals), "keywords": keywords, "role": role, "latency_us": round(us, 2), "trails_fired": trails_fired, "misses": misses, } async def _do_rss_sync(url: str): try: async with httpx.AsyncClient(timeout=30.0) as c: r = await c.get(url, follow_redirects=True, headers={"User-Agent":"SignalMesh/2.0"}) r.raise_for_status() raw_bytes = r.content raw = r.text except Exception as e: raise HTTPException(502, f"Fetch failed: {e}") feed = feedparser.parse(raw) if feed.bozo and not feed.entries: feed = feedparser.parse(raw_bytes) if feed.bozo and not feed.entries: import re as _re sanitized = _re.sub(r'&(?!amp;|lt;|gt;|quot;|apos;|#\d+;|#x[\da-fA-F]+;)', '&', raw) sanitized = _re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', '', sanitized) feed = feedparser.parse(sanitized) if not feed.entries: exc = getattr(feed, 'bozo_exception', None) raise HTTPException(422, f"Feed parse error: {exc or 'no entries found'}") synced = [] for entry in feed.entries: title = entry.get("title","Untitled") link = entry.get("link","") or entry.get("id","") summary = entry.get("summary", entry.get("description",""))[:400] if not link: continue idx, agent = _agent_for(link) signal_registry.broadcast(name=f"rss_{agent}", source_type="rss_cp", data={"title":title,"link":link,"summary":summary}, metadata={"node":idx,"agent":agent,"feed_url":url}) synced.append({"title":title,"node":idx,"agent":agent,"link":link}) return {"synced":len(synced),"feed_url":url, "feed_title":feed.feed.get("title",""), "items":synced[:20],"signals_in_mesh":_mesh_size()} async def _do_context_discovery(text: str): url_pat = re.compile(r'https?://[^\s<>"\']+') all_urls = list(set(url_pat.findall(text))) rss_urls = [u for u in all_urls if any(k in u.lower() for k in ("rss","atom","xml","feed"))] other_urls = [u for u in all_urls if u not in rss_urls] for u in other_urls[:20]: idx, agent = _agent_for(u) signal_registry.broadcast(name=f"discovered_{agent}", source_type="context_discovery", data=u, metadata={"node":idx,"agent":agent}) rss_results = [] for rss_url in rss_urls[:5]: try: d = await _do_rss_sync(rss_url) rss_results.append({"url":rss_url,"synced":d["synced"],"feed_title":d.get("feed_title","")}) except Exception as e: rss_results.append({"url":rss_url,"error":str(e)}) return {"urls_found":len(all_urls),"rss_feeds_synced":rss_results, "signals_broadcast":len(other_urls),"signals_in_mesh":_mesh_size()} async def _do_hf_inject(repo_id: str, repo_type: str = "model"): hf_url = f"https://huggingface.co/api/{repo_type}s/{repo_id}" try: async with httpx.AsyncClient(timeout=20.0, follow_redirects=True) as c: r = await c.get(hf_url, headers={"User-Agent":"SignalMesh/2.0"}) r.raise_for_status(); data = r.json() except Exception as e: raise HTTPException(502, f"HuggingFace API error: {e}") content = { "id": data.get("id", repo_id), "type": repo_type, "downloads": data.get("downloads", 0), "likes": data.get("likes", 0), "tags": (data.get("tags") or [])[:12], "pipeline_tag": data.get("pipeline_tag",""), "private": data.get("private", False), } freq = f"hf_{repo_type}_{repo_id.replace('/','__').replace('-','_')}" idx, agent = _agent_for(freq) signal_registry.broadcast(name=freq, source_type="hf_inject", data=content, metadata={"node":idx,"agent":agent,"hf_url":f"https://huggingface.co/{repo_id}"}) return {"injected":content,"frequency":freq,"node":idx,"agent":agent, "hf_url":f"https://huggingface.co/{repo_id}","signals_in_mesh":_mesh_size()} async def _do_arxiv_inject(arxiv_id: str): clean_id = arxiv_id.strip().replace("https://arxiv.org/abs/","").replace("arxiv:","") url = f"https://export.arxiv.org/api/query?id_list={clean_id}&max_results=1" try: async with httpx.AsyncClient(timeout=20.0) as c: r = await c.get(url, headers={"User-Agent":"SignalMesh/2.0"}) r.raise_for_status(); raw = r.text except Exception as e: raise HTTPException(502, f"arxiv fetch error: {e}") feed = feedparser.parse(raw) if not feed.entries: raise HTTPException(404, f"Paper {clean_id} not found on arxiv") entry = feed.entries[0] authors = [a.get("name","") for a in entry.get("authors",[])][:5] content = { "arxiv_id": clean_id, "title": re.sub(r'\s+',' ', entry.get("title","")), "authors": authors, "summary": (entry.get("summary",""))[:600], "published": entry.get("published",""), "link": entry.get("link",""), } freq = f"arxiv_{clean_id.replace('.','_').replace('-','_')}" idx, agent = _agent_for(freq) signal_registry.broadcast(name=freq, source_type="arxiv_inject", data=content, metadata={"node":idx,"agent":agent}) return {"injected":content,"frequency":freq,"node":idx,"agent":agent, "signals_in_mesh":_mesh_size()} def _do_grid_state(): active_freqs = set(signal_registry.streams.keys()) nodes = [] for (r,c), kw in grid.coords_to_agent.items(): idx = r*_COLS+c sigs = [s for fn in active_freqs if kw.lower() in fn.lower() or fn.lower() in kw.lower() for s in signal_registry.streams.get(fn,[])] nodes.append({"node":idx,"row":r,"col":c,"agent":kw, "active":len(sigs)>0,"signal_count":len(sigs)}) occupied = {n["node"] for n in nodes} for i in range(_ROWS*_COLS): if i not in occupied: rr,cc = divmod(i,_COLS) nodes.append({"node":i,"row":rr,"col":cc,"agent":None,"active":False,"signal_count":0}) nodes.sort(key=lambda n:n["node"]) return {"nodes":nodes,"total_nodes":len(nodes), "active_count":sum(1 for n in nodes if n["active"]), "occupied_count":len(occupied)} def _do_frequencies(): res = [] for name, buf in signal_registry.streams.items(): if buf: lat = buf[-1] res.append({"frequency":name,"signal_count":len(buf), "latest_type":lat.source_type, "latest_age_s":round(time.time()-lat.timestamp,1), "node":_node_idx(name),"protected":_is_prot(name)}) res.sort(key=lambda f:f["signal_count"],reverse=True) return {"frequencies":res,"count":len(res)} # ── Public API (requires X-SignalMesh-Key header) ───────────────────────────── @app.get("/api/status", tags=["Core"], summary="Mesh health and signal count") def api_status(x_signalmesh_key: Optional[str]=Header(default=None)): _check_key(x_signalmesh_key); return _do_status() @app.post("/api/broadcast", tags=["Core"], summary="Broadcast a signal to a named frequency") def api_broadcast(req: BroadcastReq, x_signalmesh_key: Optional[str]=Header(default=None)): """ Broadcast content to a frequency. All agents tuned to matching keywords will receive it on their next `tune_in` call. SEC-Ω gates protected prefixes (security_, auth_, key_, etc.) — use `bypass_gate=true` to commit after review. """ _check_key(x_signalmesh_key); return _do_broadcast(req) @app.post("/api/tune_in", tags=["Core"], summary="Tune in — retrieve signals matching keywords") def api_tune_in(req: TuneInReq, x_signalmesh_key: Optional[str]=Header(default=None)): """ Retrieve all signals whose frequency matches any of the given keywords. Fuzzy matching via learned trails handles near-misses automatically. Pass `role` to filter by relationship graph (e.g. only signals relevant to CodeAgent). """ _check_key(x_signalmesh_key, allow_update=True) if _tier(x_signalmesh_key) == "update" and not all(str(kw).startswith("ocodx") for kw in req.keywords): raise HTTPException(403, detail="Update key may only tune ocodx.* frequencies") return _do_tune_in(req.keywords, req.role, req.limit) @app.get("/api/tune/{frequency}", tags=["Core"], summary="Tune to a single named frequency") def api_tune_freq(frequency: str, limit: int = 0, x_signalmesh_key: Optional[str]=Header(default=None)): """Quick single-frequency lookup with latency measurement.""" _check_key(x_signalmesh_key, allow_update=frequency.startswith("ocodx")) t0=time.perf_counter(); sigs=signal_registry.tune_in([frequency], limit); us=(time.perf_counter()-t0)*1e6 return {"frequency":frequency,"signals":sigs,"count":len(sigs),"latency_us":round(us,2)} @app.get("/api/grid", tags=["Grid"], summary="72-node spatial grid state") def api_grid(x_signalmesh_key: Optional[str]=Header(default=None)): """Returns the full 9×8 coordinate grid — each node's assigned agent and live signal count.""" _check_key(x_signalmesh_key); return _do_grid_state() @app.get("/api/frequencies", tags=["Grid"], summary="List all active broadcast frequencies") def api_frequencies(x_signalmesh_key: Optional[str]=Header(default=None)): """Returns every named frequency that has at least one signal in the mesh.""" _check_key(x_signalmesh_key); return _do_frequencies() @app.get("/api/quarantine", tags=["Grid"], summary="SEC-Ω quarantine queue") def api_quarantine(x_signalmesh_key: Optional[str]=Header(default=None)): """Signals staged by the SEC-Ω gate pending review. Use bypass_gate=true to commit.""" _check_key(x_signalmesh_key); return {"quarantined":_SEC_Q,"count":len(_SEC_Q)} @app.post("/api/tools/signal_broadcast", tags=["Core"], summary="Tool-compatible broadcast (MCP/LangChain)") def api_tool_broadcast(req: BroadcastReq, x_signalmesh_key: Optional[str]=Header(default=None)): """Identical to /api/broadcast — exposed under /tools/ for MCP and LangChain tool registries.""" _check_key(x_signalmesh_key); return _do_broadcast(req) @app.post("/api/tools/rss_sync", tags=["Feeds"], summary="One-shot RSS → mesh broadcast") async def api_tool_rss(req: RssSyncReq, x_signalmesh_key: Optional[str]=Header(default=None)): """Fetch an RSS/Atom feed and broadcast each item as a signal.""" _check_key(x_signalmesh_key); return await _do_rss_sync(req.url) @app.post("/api/tools/context_discovery", tags=["Core"], summary="Extract and broadcast context from free text") async def api_tool_ctx(req: ContextDiscoveryReq, x_signalmesh_key: Optional[str]=Header(default=None)): """NLP extraction — finds named entities and keywords, broadcasts each as a signal.""" _check_key(x_signalmesh_key); return await _do_context_discovery(req.text) @app.post("/api/tools/hf_inject", tags=["Core"], summary="Inject a HuggingFace repo as signals") async def api_hf_inject(req: HfInjectReq, x_signalmesh_key: Optional[str]=Header(default=None)): """Fetches repo metadata from HuggingFace Hub and broadcasts it as signals.""" _check_key(x_signalmesh_key); return await _do_hf_inject(req.repo_id, req.repo_type) @app.post("/api/tools/arxiv_inject", tags=["Core"], summary="Inject an arXiv paper as signals") async def api_arxiv_inject(req: ArxivInjectReq, x_signalmesh_key: Optional[str]=Header(default=None)): """Fetches abstract and metadata from arXiv and broadcasts as signals.""" _check_key(x_signalmesh_key); return await _do_arxiv_inject(req.arxiv_id) # ── UI proxy routes (no key exposure — dashboard uses these) ────────────────── @app.get("/ui/status") def ui_status(): return _do_status() @app.get("/health", include_in_schema=False) @app.get("/healthz", include_in_schema=False) def public_health(): """Keyless liveness probe with no user, key, or payload data.""" status = _do_status() return { "ok": True, "service": "signalmesh", "version": app.version, "signals_in_mesh": status.get("signals_in_mesh", 0), "active_frequencies": status.get("active_frequencies", 0), } def _do_world_state(): return world_state.build( signal_registry=signal_registry, grid=grid, feed_hydrator=feed_hydrator, source_hydrator=source_hydrator, coordinate_resolver=coordinate_resolver, binder=binder, mport=mport, node_idx=_node_idx, rows=_ROWS, cols=_COLS) @app.get("/ui/world", tags=["Core"], summary="Canonical read model for skins") def ui_world_state(): """One semantic projection every antenna and skin renders from. Server-side on purpose: if each client assembled its own, one would derive provenance edges from real evidence and another would draw /ui/relationship — an agent keyword table — as causality. Disposable; nothing here is stored. Roles with no producer report not_modeled and carry no values. """ return _do_world_state() @app.get("/identity", tags=["Core"], summary="Which copy of SignalMesh is canonical") def public_identity(): """Keyless. The centrally adjudicated record for SignalMesh itself. Exists so a cold agent in any directory can settle "which copy is real" in one request instead of inferring it from whichever clone looks newest — the inference that sent a full session's work into a stale fork on 2026-08-03. Returns authority_unresolved rather than guessing. """ return binder.self_identity() @app.post("/api/identity/seed", tags=["Core"], summary="Deliberately seed SignalMesh identity") def api_identity_seed(req: IdentitySeedReq, x_signalmesh_key: Optional[str]=Header(default=None)): """Owner/admin bootstrap for the canonical SignalMesh identity. This is deliberately not automatic. A missing record must stay unresolved until an owner chooses the canonical locator and revision. A second call is passed through to binder.register_self(), which refuses to overwrite the existing identity; corrections belong in binder.adjudicate(). """ if _tier(x_signalmesh_key) != "admin": raise HTTPException(403, "Admin only") canonical = req.canonical_locator.strip() if not canonical: raise HTTPException(400, "canonical_locator is required") return binder.register_self( canonical, revision=req.revision.strip() or None, deployed_as=req.deployed_as.strip(), lineage_fingerprint=req.lineage_fingerprint.strip() or None, ) @app.get("/mport", include_in_schema=False) def mport_console(): """Stable product URL; the Submesh path remains the protocol/API name. Lands on Articles, not Studio. Both links used to resolve to the same Studio page, so MPort and Studio were indistinguishable in the UI even though they are different planes — Studio ports a site, MPort executes capabilities. """ return RedirectResponse("/api/submesh/ui?tab=articles", status_code=307) @app.get("/studio", include_in_schema=False) def studio_console(): """URL → pixel-identical mesh → platform-specific bundle. Opens the MPort face on its Studio tab so Scan stays one click away.""" return RedirectResponse("/api/submesh/ui?tab=studio", status_code=307) @app.post("/ui/broadcast") def ui_broadcast(req: BroadcastReq): return _do_broadcast(req) @app.post("/ui/tune_in") def ui_tune_in(req: TuneInReq): return _do_tune_in(req.keywords, req.role, req.limit) @app.get("/ui/tune/{frequency}", tags=["Core"], summary="Tune to one frequency — no key") def ui_tune_freq(frequency: str, limit: int = 0): """Keyless read for antennas. Any receiver tunes and renders in its own wrapper.""" t0=time.perf_counter(); sigs=signal_registry.tune_in([frequency], limit); us=(time.perf_counter()-t0)*1e6 return {"frequency":frequency,"signals":sigs,"count":len(sigs),"latency_us":round(us,2)} @app.get("/ui/sources", tags=["Core"], summary="Pointer frequencies being hydrated") def ui_sources(): return {"sources": source_hydrator.list_sources()} @app.post("/ui/sources/{frequency}/refresh", tags=["Core"], summary="Re-read a frequency's source now") async def ui_source_refresh(frequency: str): """Pull from the source immediately instead of waiting for the poll.""" return await source_hydrator.refresh(frequency) def _public_base(request: Request) -> str: """The URL an antenna out in the world must call — not the in-container address.""" host = os.environ.get("SPACE_HOST") or os.environ.get("SIGNALMESH_PUBLIC_URL") if host: return host if host.startswith("http") else f"https://{host}" fwd_host = request.headers.get("x-forwarded-host") or request.headers.get("host") if fwd_host and not fwd_host.startswith(("127.", "localhost", "0.0.0.0")): proto = request.headers.get("x-forwarded-proto", "https") return f"{proto}://{fwd_host}" return str(request.base_url).rstrip("/") @app.get("/ui/antennas/{frequency}", tags=["Core"], summary="What shape is this frequency carrying?") def ui_antenna_inspect(frequency: str): return antenna_forge.inspect(frequency) @app.post("/ui/antennas", tags=["Core"], summary="Build a receiver for a frequency") def ui_antenna_build(req: AntennaReq, request: Request): """Emit a wrapper shaped to what the frequency actually carries.""" try: return antenna_forge.generate(req.frequency, req.target, req.name, mesh_url=_public_base(request)) except ValueError as e: raise HTTPException(400, detail={"error": str(e), "targets": list(ANTENNA_TARGETS)}) @app.get("/ui/grid") def ui_grid(): return _do_grid_state() @app.get("/ui/frequencies") def ui_frequencies(): return _do_frequencies() @app.get("/ui/quarantine") def ui_quarantine(): return {"quarantined":_SEC_Q,"count":len(_SEC_Q)} @app.post("/ui/rss_sync") async def ui_rss(req: RssSyncReq): return await _do_rss_sync(req.url) @app.post("/ui/context_discovery") async def ui_ctx(req: ContextDiscoveryReq): return await _do_context_discovery(req.text) @app.post("/ui/hf_inject") async def ui_hf(req: HfInjectReq): return await _do_hf_inject(req.repo_id, req.repo_type) @app.post("/ui/arxiv_inject") async def ui_arxiv(req: ArxivInjectReq): return await _do_arxiv_inject(req.arxiv_id) @app.get("/ui/feeds") def ui_feeds(): """List all active hydrator feeds — label, URL, freshness, item count.""" return {"feeds": feed_hydrator.list_feeds(), "count": len(feed_hydrator._feeds)} @app.post("/ui/feeds/add") async def ui_feeds_add(req: FeedAddReq): cfg = feed_hydrator.add_feed(req.url, req.frequency, req.interval, req.label) return {"added": True, "frequency": cfg.frequency, "url": cfg.url, "interval_s": cfg.interval, "persisted": True, "env_var": "SIGNALMESH_FEEDS", "env_value": feed_hydrator.env_value()} @app.post("/ui/feeds/discover") async def ui_feeds_discover(req: FeedDiscoverReq): """Probe any URL for RSS/Atom feeds — RSSHub Radar built in.""" feeds = await feed_hydrator.discover(req.url) return {"url": req.url, "found": feeds, "count": len(feeds)} @app.get("/api/feeds", tags=["Feeds"], summary="List active hydrator feeds") def api_feeds(x_signalmesh_key: Optional[str] = Header(default=None)): """Returns all registered RSS/Atom feeds with freshness and item counts.""" _check_key(x_signalmesh_key) return {"feeds": feed_hydrator.list_feeds(), "count": len(feed_hydrator._feeds)} @app.post("/api/feeds/add", tags=["Feeds"], summary="Register a new RSS/Atom feed") async def api_feeds_add(req: FeedAddReq, x_signalmesh_key: Optional[str] = Header(default=None)): """Add a feed — items auto-broadcast to that frequency on each poll. The registration is persisted (local + HF dataset). `env_value` is the same set as a SIGNALMESH_FEEDS variable, for anyone who wants it as config rather than state — we hand it back instead of writing the Space variable ourselves, because that write restarts the Space. """ _check_key(x_signalmesh_key) cfg = feed_hydrator.add_feed(req.url, req.frequency, req.interval, req.label) return {"added": True, "frequency": cfg.frequency, "url": cfg.url, "persisted": True, "env_var": "SIGNALMESH_FEEDS", "env_value": feed_hydrator.env_value()} @app.post("/api/feeds/persist", tags=["Feeds"], summary="Flush dynamic feeds to durable storage now") async def api_feeds_persist(x_signalmesh_key: Optional[str] = Header(default=None)): """Force the HF-dataset upload instead of waiting for the background loop.""" _check_key(x_signalmesh_key) ok = await asyncio.to_thread(feed_hydrator.persist_hf) return {"persisted": ok, "feeds": len(feed_hydrator.dynamic()), "dataset": None if not ok else "feeds.json", "env_var": "SIGNALMESH_FEEDS", "env_value": feed_hydrator.env_value()} @app.get("/api/mport/articles/{frequency}", tags=["MPort"], summary="Article identity held on a frequency — no fetching") def api_mport_articles(frequency: str, limit: int = 0, x_signalmesh_key: Optional[str] = Header(default=None)): """Title, link, summary, published, feed. The mesh's half of the contract.""" _check_key(x_signalmesh_key) return mport_articles.export_feed_json(frequency, limit) @app.post("/api/mport/article", tags=["MPort"], summary="Fetch one article body — MPort does the work, not the mesh") async def api_mport_article(req: ArticleFetchReq, x_signalmesh_key: Optional[str] = Header(default=None)): """Full body and image URLs. Deliberately never broadcast into the mesh.""" _check_key(x_signalmesh_key) return await mport_articles.fetch_article(req.url, req.with_images) @app.post("/api/mport/snapshot", tags=["MPort"], summary="Bundle the newest N articles on a frequency into an archive") async def api_mport_snapshot(req: ArticleSnapshotReq, x_signalmesh_key: Optional[str] = Header(default=None)): """Returns a download token, not bytes — "download Robert's latest five typewriter posts with their images" lands here after semantic resolution.""" _check_key(x_signalmesh_key) return await mport_articles.snapshot_articles(req.frequency, req.limit, req.with_images) @app.get("/ui/mport/articles/{frequency}", include_in_schema=False) def ui_mport_articles(frequency: str, limit: int = 0): """Keyless read — identity only, same as every other /ui/ receiver route.""" return mport_articles.export_feed_json(frequency, limit) @app.get("/ui/aliases", tags=["Trails"], summary="Names each frequency answers to") def ui_aliases(frequency: str = ""): """The alias index: feed titles, authors, hostnames and article titles that resolve to a frequency. Makes wording-based resolution inspectable.""" return signal_registry.get_aliases(frequency) @app.post("/api/feeds/discover", tags=["Feeds"], summary="Discover RSS feeds on any URL") async def api_feeds_discover(req: FeedDiscoverReq, x_signalmesh_key: Optional[str] = Header(default=None)): """Probe any URL — finds RSS/Atom feeds via link headers and common paths.""" _check_key(x_signalmesh_key) return {"url": req.url, "found": await feed_hydrator.discover(req.url)} @app.get("/ui/trails", tags=["Trails"], summary="All learned fuzzy keyword trails") def ui_trails(): """ Returns every keyword trail the mesh has self-healed — the fuzzy paths it builds when a keyword misses and a close-enough frequency is found via SequenceMatcher + token overlap. Empty dict on a fresh instance; grows as agents use the mesh. """ return signal_registry.get_trails() @app.delete("/ui/trails/{keyword}", tags=["Trails"], summary="Clear a learned trail") def ui_clear_trail(keyword: str): """Remove a specific keyword trail — the mesh will re-learn it from scratch on next miss.""" removed = signal_registry.clear_trail(keyword) return {"removed": removed, "keyword": keyword} @app.get("/api/trails", tags=["Trails"], summary="All learned fuzzy keyword trails (API)") def api_trails(x_signalmesh_key: Optional[str] = Header(default=None)): """Same as /ui/trails — requires API key.""" _check_key(x_signalmesh_key) return signal_registry.get_trails() # ── AGENTS.md Manifest endpoints ───────────────────────────────────────────── @app.get("/agents.md", response_class=HTMLResponse, include_in_schema=False) async def serve_agents_md(): """Serve this node's AGENTS.md manifest so any external agent can discover it.""" here = Path(__file__).parent / "AGENTS.md" if not here.exists(): raise HTTPException(404, "AGENTS.md not found on this node") return HTMLResponse(content=here.read_text(), media_type="text/markdown") @app.post("/api/manifest/ingest", tags=["Manifest"], summary="Ingest a remote AGENTS.md node") async def api_manifest_ingest(req: ManifestIngestReq, x_signalmesh_key: Optional[str] = Header(default=None)): """ Fetch a remote AGENTS.md and register its node in the mesh. Parses declared frequencies and auto-broadcasts them. SEC-Ω validated before ingestion. """ _check_key(x_signalmesh_key) try: async with httpx.AsyncClient(follow_redirects=True, timeout=10.0) as client: r = await client.get(req.url) r.raise_for_status() raw = r.text except Exception as e: raise HTTPException(502, f"Fetch failed: {e}") safe, reason = sec_omega.validate("manifest", raw[:2000], "external_manifest") if not safe: raise HTTPException(400, f"SEC-Ω blocked manifest: {reason}") parsed = manifest_registry.ingest(raw, source_uri=req.url) # Register all declared frequencies into the mesh for freq in parsed.get("frequencies", []): signal_registry.broadcast( name=freq["frequency"], source_type="manifest_ingestion", data={"node_uri": parsed["node_uri"], "mode": freq["mode"]}, metadata={"grid_hash": parsed["grid_hash"]}, ) return { "ingested": True, "node_uri": parsed["node_uri"], "grid_hash": parsed["grid_hash"], "frequencies_registered": [f["frequency"] for f in parsed["frequencies"]], "gc_strategy": parsed["gc_rules"].get("strategy", "auto"), "strip_tool_schemas": parsed["strip_tool_schemas"], } @app.get("/api/manifest/nodes", tags=["Manifest"], summary="List all registered mesh nodes") def api_manifest_nodes(x_signalmesh_key: Optional[str] = Header(default=None)): """All external nodes registered via AGENTS.md ingestion — URI, grid hash, declared frequencies.""" _check_key(x_signalmesh_key) return {"nodes": manifest_registry.all_nodes(), "count": len(manifest_registry._manifests)} @app.post("/ui/manifest/ingest") async def ui_manifest_ingest(req: ManifestIngestReq): """Dashboard-accessible manifest ingestion (no key — uses server-side validation).""" try: async with httpx.AsyncClient(follow_redirects=True, timeout=10.0) as client: r = await client.get(req.url); r.raise_for_status(); raw = r.text except Exception as e: raise HTTPException(502, f"Fetch failed: {e}") safe, reason = sec_omega.validate("manifest", raw[:2000], "ui_manifest") if not safe: raise HTTPException(400, f"SEC-Ω blocked: {reason}") parsed = manifest_registry.ingest(raw, source_uri=req.url) return {"ingested": True, "node_uri": parsed["node_uri"], "frequencies": len(parsed["frequencies"])} # ── Coordinate addressing ──────────────────────────────────────────────────── @app.post("/ui/coordinate", tags=["Coordinates"], summary="Store or update a coordinate") def ui_store_coordinate(req: CoordinateStoreReq): """Store content at a grid address (e.g. AA03). Supports symbolic names, TTL, and change_id for provenance.""" coord = coordinate_resolver.store( req.coordinate, req.data, symbolic=req.symbolic, change_id=req.change_id, ttl_ms=req.ttl_ms ) return {"coordinate": coord, "symbolic": req.symbolic, "stored": True} @app.get("/ui/resolve/{address:path}", tags=["Coordinates"], summary="Resolve a coordinate or symbolic name") def ui_resolve(address: str): """ Resolve any address form to stored content: - `AA03` — grid coordinate - `ARCH.DECISIONS.RETRY` — dot-path symbolic name - `AA03:D2` — namespaced slot - `AA03@371` — specific change_id version Returns 404 if not found or TTL expired. """ result = coordinate_resolver.resolve(address) if result is None: raise HTTPException(404, f"Coordinate '{address}' not found or expired") return {"address": address, "resolved": result} @app.get("/ui/coordinates", tags=["Coordinates"], summary="List all active coordinates") def ui_list_coordinates(): """All live (non-TTL-expired) coordinates with namespaces and slot definitions.""" return { "coordinates": coordinate_resolver.all_coordinates(), "namespaces": NAMESPACES, "slots": SLOTS, } # ── Relationship graph / scoped delivery ───────────────────────────────────── @app.get("/ui/relationship", tags=["Core"], summary="Role → frequency pattern map") def ui_get_relationship(): """The relationship graph — maps agent roles (CodeAgent, QAAgent, etc.) to frequency patterns for scoped delivery.""" return {"roles": relationship_graph.all_roles()} @app.post("/ui/relationship", tags=["Core"], summary="Set frequency patterns for a role") def ui_set_relationship(req: RelationshipRoleReq): """Register or update which frequencies a given role should receive on tune_in.""" relationship_graph.set_role(req.role, req.patterns) return {"role": req.role, "patterns": req.patterns, "updated": True} # ── Dashboard (v3 — clean bright build) ────────────────────────────────────── _HTML = r""" SignalMesh
Signals in Mesh
Frequencies
72
Grid Nodes
SEC-Ω Queue

⚡ Live Frequencies

No active frequencies yet — broadcast a signal or sync a feed.

📥 Ingest — RSS · HuggingFace · arXiv

RSS Feed Sync
🤗 HF Dataset Flash
📄 arxiv Paper Inject

📡 Antenna Grid — 9×8 Spatial Matrix (72 nodes)

Transmitting Silent

🛠 Build Antenna — materialize a frequency somewhere

Transmit puts a signal on a frequency. This builds the receiving end: it reads what the frequency is actually carrying right now and emits a wrapper shaped to it. Same signal, any number of antennas.
""" # ── OCODX license minting (Ed25519, offline-verifiable by the binary) ────────── # Only this host holds the private signing key. The OCODX binary holds the matching # PUBLIC key and verifies the minted token offline — no validation endpoint needed. TIER_DEVICES = {"solo": 1, "pro": 3, "team": 10} def _tier_from_session(sess: dict) -> tuple: """Resolve (tier, devices) from Checkout Session metadata set at checkout time.""" md = sess.get("metadata") or {} tier = md.get("tier", "pro") if tier not in TIER_DEVICES: tier = "pro" try: devices = int(md.get("devices", 0)) or TIER_DEVICES[tier] except (TypeError, ValueError): devices = TIER_DEVICES[tier] return tier, devices def _mint_ocodx_key(email: str, tier: str, devices: int, days: int = 0) -> str: import base64 as _b64 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey raw = os.environ.get("OCODX_SIGNING_KEY", "") if not raw: raise RuntimeError("OCODX_SIGNING_KEY not set — cannot mint license") priv = Ed25519PrivateKey.from_private_bytes( _b64.urlsafe_b64decode(raw + "=" * (-len(raw) % 4))) now = int(time.time()) payload = {"email": email, "tier": tier, "iat": now, "exp": now + days * 86400 if days > 0 else 0, "dev": devices} pb = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode() enc = lambda x: _b64.urlsafe_b64encode(x).decode().rstrip("=") return f"OCODX1.{enc(pb)}.{enc(priv.sign(pb))}" def _db_create_licensed_user(email, cid, sid, tier, devices) -> str: """Mint an OCODX license token and store it as the user's key.""" key = _mint_ocodx_key(email, tier, devices) uid = secrets.token_hex(8) with _DB_LOCK: conn = _db_conn() conn.execute("INSERT OR IGNORE INTO users VALUES (?,?,?,?,?,?,1,?,?,0)", (uid, email, key, tier, cid, sid, time.time(), time.time())) conn.commit(); conn.close() threading.Thread(target=_db_backup, daemon=True).start() return key def _provision_checkout_session(sess: dict) -> dict: """Mint the product the Stripe session actually purchased. SignalMesh/MPort purchases receive a managed ``smesh-`` API key. OCODX purchases continue to receive an offline-verifiable ``OCODX1`` license. Keeping these paths explicit prevents one product's checkout metadata from silently minting the other product's credential. """ if sess.get("payment_status") != "paid": raise ValueError("Payment not completed") md = sess.get("metadata") or {} product = str(md.get("product") or "signalmesh").lower() email = (sess.get("customer_details") or {}).get("email", "") cid = sess.get("customer", "") sid = sess.get("id", "") if not sid: raise ValueError("Checkout session has no id") existing = _db_get_by_session(sid) if existing: return { "api_key": existing["api_key"], "email": existing["email"], "tier": existing["tier"], "product": product, } if product == "signalmesh": checkout_tier = str(md.get("tier") or "monthly").lower() tier = ( "managed_lifetime" if checkout_tier in {"lifetime", "founding_lifetime"} else "managed_monthly" ) key = _db_create_user(email, cid, sid, tier=tier) return {"api_key": key, "email": email, "tier": tier, "product": product} if product == "ocodx": add_seats = int(md.get("add_seats", 0) or 0) if add_seats > 0: key = _db_add_seats(email, add_seats, cid, sid) row = _db_get_by_key(key) or {} tier = row.get("tier", "solo") else: tier, devices = _tier_from_session(sess) key = _db_create_licensed_user(email, cid, sid, tier, devices) return {"api_key": key, "email": email, "tier": tier, "product": product} raise ValueError(f"Unsupported checkout product: {product}") # ── Stripe webhook ──────────────────────────────────────────────────────────── @app.post("/api/webhook/stripe", include_in_schema=False) async def stripe_webhook(request: Request): payload = await request.body() sig = request.headers.get("stripe-signature", "") secret = os.environ.get("STRIPE_WEBHOOK_SECRET", "") try: _stripe_verify(payload, sig, secret) except Exception as e: raise HTTPException(400, f"Webhook signature invalid: {e}") event = json.loads(payload) etype = event.get("type", "") if etype == "checkout.session.completed": sess = event["data"]["object"] if sess.get("payment_status") == "paid": _provision_checkout_session(sess) elif etype == "customer.subscription.deleted": cid = event["data"]["object"].get("customer", "") if cid: _db_revoke(cid) return {"received": True} def _stripe_verify(payload: bytes, sig_header: str, secret: str): import hmac, hashlib if not secret: raise ValueError("STRIPE_WEBHOOK_SECRET is not configured") parts = {k: v for k, v in (p.split("=", 1) for p in sig_header.split(",") if "=" in p)} ts = parts.get("t", "") v1 = parts.get("v1", "") if not ts or not v1: raise ValueError("Stripe signature header is incomplete") try: if abs(time.time() - int(ts)) > 300: raise ValueError("Stripe signature timestamp is outside the 5-minute tolerance") except ValueError: raise except Exception as exc: raise ValueError("Stripe signature timestamp is invalid") from exc signed = f"{ts}.".encode() + payload expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, v1): raise ValueError("Signature mismatch") # ── Key claim ───────────────────────────────────────────────────────────────── @app.get("/api/keys/claim/{session_id}", tags=["Core"], summary="Claim API key after Stripe checkout") async def claim_key(session_id: str): """Verify Stripe checkout session and return API key. Safe to call multiple times.""" existing = _db_get_by_session(session_id) if existing: return {"api_key": existing["api_key"], "email": existing["email"], "tier": existing["tier"], "status": "active"} sk = os.environ.get("STRIPE_SECRET_KEY", "") if not sk: raise HTTPException(503, "Stripe not configured") try: import base64 as _b64, urllib.request as _ur req = _ur.Request(f"https://api.stripe.com/v1/checkout/sessions/{session_id}") req.add_header("Authorization", "Basic " + _b64.b64encode(f"{sk}:".encode()).decode()) with _ur.urlopen(req, timeout=10) as r: sess = json.loads(r.read()) except Exception as e: raise HTTPException(502, f"Stripe lookup failed: {e}") if sess.get("payment_status") != "paid": raise HTTPException(402, "Payment not completed") provisioned = _provision_checkout_session(sess) return {**provisioned, "status": "active"} # ── Key info ────────────────────────────────────────────────────────────────── @app.get("/api/keys/me", tags=["Core"], summary="Get info about your API key") def key_info(x_signalmesh_key: Optional[str] = Header(default=None)): """Returns tier, usage, and account info for the authenticated key.""" if not x_signalmesh_key: raise HTTPException(401, "No key provided") t = _tier(x_signalmesh_key) if t == "update": hits = [ts for ts in _update_usage.get(x_signalmesh_key, []) if ts > time.time() - 86400] return {"tier": "update", "calls_used_24h": len(hits), "limit_24h": _UPDATE_RATE_LIMIT, "note": "update channel only — submesh reads + ocodx.* tune_in", "upgrade": "https://kyklos.io/apps/signalmesh/"} if t in ("admin", "paid"): return {"tier": t, "public_socket_limit": _socket_limit(x_signalmesh_key)} db_user = _db_get_by_key(x_signalmesh_key) if db_user: return {"tier": db_user["tier"], "email": db_user["email"], "call_count": db_user["call_count"], "created_at": db_user["created_at"], "active": bool(db_user["active"]), "public_socket_limit": 1} raise HTTPException(401, "Invalid key") # ── Admin: list users ───────────────────────────────────────────────────────── @app.get("/api/admin/users", include_in_schema=False) def admin_users(x_signalmesh_key: Optional[str] = Header(default=None)): if _tier(x_signalmesh_key) != "admin": raise HTTPException(403, "Admin only") users = _db_all() for u in users: k = u.get("api_key", "") u["api_key"] = k[:12] + "..." if k else "" return {"users": users, "count": len(users)} @app.get("/admin", response_class=HTMLResponse, include_in_schema=False) def admin_dashboard(key: Optional[str] = None): """Sales + validation dashboard. Auth via ?key= (browser-friendly).""" if _tier(key) != "admin": return HTMLResponse("

403 — append ?key=<admin key>

", status_code=403) users = _db_all() now = time.time() total = len(users) active = sum(1 for u in users if u.get("active")) revoked = total - active # "validated" = the license has actually been used at least once (device check-in) validated = sum(1 for u in users if (u.get("call_count") or 0) > 0 or u.get("last_seen")) active_30d = sum(1 for u in users if (u.get("last_seen") or 0) > now - 30 * 86400) by_tier: Dict[str, int] = {} for u in users: by_tier[u.get("tier", "?")] = by_tier.get(u.get("tier", "?"), 0) + 1 legit = sum(1 for u in users if str(u.get("api_key", "")).startswith("OCODX1.")) tier_rows = "".join(f"{t}{n}" for t, n in sorted(by_tier.items())) def row(u): k = u.get("api_key", "") kind = "OCODX1 ✓" if k.startswith("OCODX1.") else ("smesh (legacy)" if k else "—") seen = time.strftime("%Y-%m-%d", time.localtime(u["last_seen"])) if u.get("last_seen") else "never" made = time.strftime("%Y-%m-%d", time.localtime(u["created_at"])) if u.get("created_at") else "?" return (f"{u.get('email','')}{u.get('tier','')}{kind}" f"{'✓' if u.get('active') else '✗'}{u.get('call_count',0)}" f"{seen}{made}") recent = "".join(row(u) for u in users[:100]) html = f"""SignalMesh · Admin

📡 SignalMesh — License Admin

{total}keys sold
{validated}validated (used)
{active_30d}active · 30d
{active}active
{revoked}revoked
{legit}OCODX1 tokens

By tier

{tier_rows}
tiercount

Recent (latest 100)

{recent}
emailtierkey typeactiveuseslast seencreated
""" return HTMLResponse(html) @app.get("/", response_class=HTMLResponse) def root(): return _HTML _CHAT_SYSTEM = ( "You are SignalMesh, the Sovereign Liquid Matrix agent. Answer the message that was actually " "sent, in as few words as it takes.\n\n" "1. If the live context below already answers the question, answer FROM it — summarise the " "signals in plain prose. Never write code to go fetch what you were already handed.\n" "2. Write code only when asked to write code. Then it is complete and working, no placeholders.\n" "3. Never echo, quote, or paste the context block back to the user. It is your input, not your " "output. Cite a signal by name if you need to attribute something.\n" "4. If the request is to build or wire something and you are missing what you'd need to build " "it, say what you can do in one line, then ask only the questions whose answers change the " "build. No preamble, no options essay.\n" "5. You may open with a short signal tag for the dominant matched node (e.g. '▶ L1 · Python " "Pro') when a specialist lens actually applies. Otherwise skip it." ) # name -> (default_base_url, key_env, model_env, default_model) _CHAT_PROVIDERS = { "groq": ("https://api.groq.com/openai/v1", "GROQ_API_KEY", "GROQ_MODEL", "llama-3.1-8b-instant"), "gemini": ("https://generativelanguage.googleapis.com/v1beta/openai", "GEMINI_API_KEY", "GEMINI_MODEL", "gemini-2.0-flash"), "opencode_zen": ("https://opencode.ai/zen/v1", "OPENCODE_ZEN_API_KEY", "OPENCODE_ZEN_MODEL", "glm-4.6"), "huggingface": ("https://router.huggingface.co/v1", "HF_TOKEN", "HF_CHAT_MODEL", "meta-llama/Llama-3.1-8B-Instruct"), "xai": ("https://api.x.ai/v1", "XAI_API_KEY", "XAI_MODEL", "grok-2-latest"), } _CHAT_ORDER_DEFAULT = "huggingface,groq,gemini,opencode_zen,xai" def _chat_cfg(name: str): base, key_env, model_env, default_model = _CHAT_PROVIDERS[name] # .strip() everything: a secret pasted into the Space UI often carries a trailing # newline, and "moonshotai/Kimi-K3\n" is a 400 from the router, not a fallback. base = os.environ.get(f"{name.upper()}_BASE_URL", base).strip().rstrip("/") return base, os.environ.get(key_env, "").strip(), (os.environ.get(model_env) or default_model).strip() def _chat_order() -> List[str]: raw = os.environ.get("CHAT_PROVIDER_ORDER", _CHAT_ORDER_DEFAULT) return [p.strip() for p in raw.split(",") if p.strip() in _CHAT_PROVIDERS] _CHAT_SESSIONS: Dict[str, List[dict]] = {} _CHAT_TOPIC: Dict[str, List[str]] = {} # last keywords that actually tuned _CHAT_ACTIVE: Dict[str, List[str]] = {} _CHAT_LOCK = threading.Lock() _CHAT_MAX_HISTORY = 16 _CHAT_STOP = {"the","and","for","that","with","this","you","your","are","can","what","how","why", "who","when","where","a","an","to","of","in","on","is","it","me","my","do","does", "i","we","be","as","at","or","if","so","but","not","get","let","now","then","please", "help","need","want","make","using","use","about"} _ROW_LETTERS = "ALIODQBGU" def _chat_keywords(text: str, limit: int = 8) -> List[str]: out, seen = [], set() for t in re.split(r"[^a-zA-Z0-9_+#-]+", (text or "").lower()): if len(t) >= 3 and t not in _CHAT_STOP and t not in seen: seen.add(t); out.append(t) if len(out) >= limit: break return out def _freq_coord(freq: str) -> str: idx = _node_idx(freq); r, c = divmod(idx, _COLS) return f"{_ROW_LETTERS[r]}{c+1}" if 0 <= r < len(_ROW_LETTERS) else f"N{idx}" def _render_signal(content: Any) -> str: """Feed rows arrive as dicts; a raw repr is unreadable and the model echoes it back.""" if isinstance(content, dict): title = content.get("title") or content.get("name") or content.get("summary") or "" link = content.get("link") or content.get("url") or "" if title: return f"{str(title)[:200]}{' — ' + link if link else ''}" content = json.dumps(content, separators=(",", ":")) return str(content or "")[:300] def _mesh_hydrate(keywords: List[str], role: Optional[str] = None, max_signals: int = 12): """One tune_in pass -> (context_str, [grid_coords]). Capped to keep the prompt lean.""" try: signals = signal_registry.tune_in(keywords) if role: signals = relationship_graph.filter_signals(role, signals) except Exception: signals = [] signals = signals[:max_signals] lines, coords = [], [] if signals: lines.append("[SignalMesh — Live Context]") for s in signals: nm = s.get("name", "?") # the frequency name is what the user can act on; A1-style cells are for the canvas if nm not in coords: coords.append(nm) lines.append(f"• [{nm}] {_render_signal(s.get('content'))}") return "\n".join(lines)[:4000], coords async def _chat_one(client, base, key, model, messages, max_tokens, temperature): r = await client.post(base.rstrip("/") + "/chat/completions", headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, json={"model": model, "messages": messages, "max_tokens": max_tokens, "temperature": temperature}) if r.status_code != 200: raise RuntimeError(f"HTTP {r.status_code}: {r.text[:160]}") msg = ((r.json().get("choices") or [{}])[0].get("message") or {}) txt = (msg.get("content") or "").strip() if not txt: raise RuntimeError("empty completion") return txt async def _chat_complete(messages, max_tokens=1024, temperature=0.6): errors = [] async with httpx.AsyncClient(timeout=45.0) as client: for name in _chat_order(): base, key, model = _chat_cfg(name) if not key: errors.append(f"{name}: skipped, no key set") # never fail silently continue try: return (await _chat_one(client, base, key, model, messages, max_tokens, temperature), f"{name}:{model}", errors) except Exception as e: errors.append(f"{name}: {str(e)[:120]}") return None, "none", errors def _live_frequencies(term: str) -> List[str]: stem = re.sub(r"^(www\.|https?://)|\.(com|io|ai|org|net)$", "", term.lower()).split(".")[0] return [n for n, buf in signal_registry.streams.items() if buf and stem and stem in n.lower()] _ASKING_ABOUT = re.compile( r"^\s*(?:tell me about|what(?:'?s| is| are)|who(?:'?s| is)|explain|describe)\s+(.{2,60}?)\s*[?.!]*\s*$", re.I) _PRONOUNS = {"it", "that", "this", "they", "them", "he", "she", "these", "those", "one"} def _empty_mesh_reply(msg: str, context: str, history: list) -> Optional[str]: """A question about a named thing with nothing in the mesh is answered by the mesh, not by the model — that is the path that invented a Meta product called NexusAI. Build/code requests fall through untouched; they were never mesh lookups.""" m = _ASKING_ABOUT.match(msg) if context or not m or history: return None # mid-thread, the subject is the thread — never gate a follow-up subject = m.group(1).strip() head = re.split(r"[^a-z0-9]+", subject.lower())[0] if subject else "" if not head or head in _PRONOUNS: return None # "who is it for" names nothing to look up near = sorted({n for n in signal_registry.streams if signal_registry.streams[n] and n.split(".")[0].lower()[:4] in msg.lower()}) out = [f"The mesh is carrying nothing on {subject} — so I'm not answering from my own " f"training data and calling it live."] if near: out.append("\nClosest frequencies: " + ", ".join(near[:6])) host = os.environ.get("SIGNALMESH_PUBLIC_URL") or ( f"https://{os.environ['SPACE_HOST']}" if os.environ.get("SPACE_HOST") else "") out.append(f"\nTo make this answerable, broadcast it:\n" f' curl -X POST {host}/ui/broadcast \\\n' f' -H "Content-Type: application/json" \\\n' f' -d \'{{"frequency":"","content":"…"}}\'\n' f"or add it to SITE_INGEST as = and it re-hydrates on every boot.") return "\n".join(out) def _fast_reply(msg: str) -> Optional[str]: """Deterministic answers for requests whose shape we already know. An antenna is built from (frequency, target, name) against a frequency that already carries signals — so the missing pieces are knowable without a model. """ m = re.search(r"\bantenna\b[^.?!]{0,30}?\bfor\b\s+([^\s,.?!]+)", msg, re.I) if not m: return None term = m.group(1).strip() live = _live_frequencies(term) out = [f"On it — antenna for {term}."] if live: out.append("\nCarrying signals now: " + ", ".join(sorted(live)[:8])) out.append("\nTwo things I need:") out.append("1. Which frequency above should it receive?") else: out.append(f"\nNothing in the mesh is carrying {term} yet — an antenna is shaped against " "live signals, so that frequency has to be broadcast first.") out.append("\nTwo things I need:") out.append("1. What should be broadcast onto it — docs, API responses, a feed?") out.append("2. Where does it materialize? " + " · ".join(ANTENNA_TARGETS)) out.append("\n nextjs_route → an API route in their app · shell → a curl/CLI receiver") out.append(" system_prompt → context injected into an agent · html_embed → a drop-in widget") out.append("\nAnswer those two and I'll emit the wrapper.") return "\n".join(out) class ChatReq(BaseModel): session_id: str = "" message: str role: Optional[str] = None keywords: Optional[List[str]] = None max_tokens: int = 1024 temperature: float = 0.6 class ChatResetReq(BaseModel): session_id: str = "" @app.post("/api/chat", tags=["Core"], summary="Chat with the mesh — hydrated context + multi-provider failover") async def api_chat(req: ChatReq, x_signalmesh_key: Optional[str] = Header(default=None)): _check_key(x_signalmesh_key) return await _do_chat(req) @app.post("/ui/chat", tags=["Core"], summary="Chat with the mesh — no key") async def ui_chat(req: ChatReq): """Keyless, like the rest of /ui/*. The mesh is the demo; the model runs on our keys.""" return await _do_chat(req) @app.post("/ui/chat/reset", tags=["Core"], summary="Reset a chat session — no key") def ui_chat_reset(req: ChatResetReq): sid = (req.session_id or "anon").strip() with _CHAT_LOCK: _CHAT_SESSIONS.pop(sid, None); _CHAT_ACTIVE.pop(sid, None); _CHAT_TOPIC.pop(sid, None) return {"status": "reset", "session_id": sid} async def _do_chat(req: ChatReq): msg = (req.message or "").strip() if not msg: raise HTTPException(400, "message is required") sid = (req.session_id or "anon").strip() kws = req.keywords or _chat_keywords(msg) context, coords = _mesh_hydrate(kws, req.role) with _CHAT_LOCK: history = list(_CHAT_SESSIONS.get(sid, [])) prev = _CHAT_ACTIVE.get(sid, []) carried = list(_CHAT_TOPIC.get(sid, [])) # A follow-up ("who is it for", "what about pricing") carries no routable keyword of # its own. Stay tuned to whatever the thread is already about instead of going deaf. if not context and carried: context, coords = _mesh_hydrate(carried, req.role) kws = carried elif context: with _CHAT_LOCK: _CHAT_TOPIC[sid] = kws sys_msg = _CHAT_SYSTEM + ("\n\n" + context if context else "\n\n[SignalMesh — no live signals matched this query.]\n" "Say so in your first line — the mesh is carrying nothing on this. Anything you add " "after that is your own training data, not mesh knowledge: label it as unverified and " "keep it short. Never present it as what the mesh knows, and never invent specifics " "(dates, versions, features, pricing) about a named product you have no signals for.") messages = ([{"role": "system", "content": sys_msg}] + history[-_CHAT_MAX_HISTORY:] + [{"role": "user", "content": msg}]) fast = _fast_reply(msg) or _empty_mesh_reply(msg, context, history) if fast: reply, provider, errors = fast, "mesh", [] else: reply, provider, errors = await _chat_complete(messages, req.max_tokens, req.temperature) if reply is None: reply = ("⚠ No language provider answered. What each one said:\n\n" + "\n".join(f"· {e}" for e in errors) if errors else "⚠ No language provider is configured on the mesh. Set at least one of " "GROQ_API_KEY, GEMINI_API_KEY, OPENCODE_ZEN_API_KEY, HF_TOKEN, or XAI_API_KEY " "as a Space secret to enable chat.") provider = "none" with _CHAT_LOCK: h = _CHAT_SESSIONS.get(sid, []) h.append({"role": "user", "content": msg}) if provider != "none": h.append({"role": "assistant", "content": reply}) _CHAT_SESSIONS[sid] = h[-_CHAT_MAX_HISTORY:] _CHAT_ACTIVE[sid] = coords hist_len = sum(1 for m in _CHAT_SESSIONS[sid] if m["role"] == "user") new_sigs = [c for c in coords if c not in prev] removed = [c for c in prev if c not in coords] return { "reply": reply, "provider": provider, "active_signals": coords, "new_signals": new_sigs, "removed_signals": removed, "history_len": hist_len, "trail": [], "had_reask": False, "fuzzy_acquired": [], "seek_needed": False, "seek_gaps": [], "seek_prompt": "", "provider_errors": errors, } @app.post("/api/chat/reset", tags=["Core"], summary="Reset a chat session") def api_chat_reset(req: ChatResetReq, x_signalmesh_key: Optional[str] = Header(default=None)): _check_key(x_signalmesh_key) sid = (req.session_id or "anon").strip() with _CHAT_LOCK: _CHAT_SESSIONS.pop(sid, None); _CHAT_ACTIVE.pop(sid, None); _CHAT_TOPIC.pop(sid, None) return {"status": "reset", "session_id": sid} @app.get("/api/chat/session/{session_id}", tags=["Core"], summary="Chat session info") def api_chat_session(session_id: str, x_signalmesh_key: Optional[str] = Header(default=None)): _check_key(x_signalmesh_key) with _CHAT_LOCK: h = _CHAT_SESSIONS.get(session_id, []) hist_len = sum(1 for m in h if m["role"] == "user") return {"session_id": session_id, "history_len": hist_len, "active_signals": _CHAT_ACTIVE.get(session_id, []), "signal_detail": []} _CHAT_HTML = r""" SignalMesh · Live Chat
Sovereign Liquid Matrix live

Ask the mesh

Live ambient context from Reddit, Hacker News, arXiv, HuggingFace and more — answered in real time.

r/MachineLearning trends
Reddit trending today
Hacker News front page
Latest arXiv AI papers
SignalMesh answers from live mesh signals · responses may vary
""" @app.get("/chat", response_class=HTMLResponse, include_in_schema=False) def chat_ui(): return _CHAT_HTML if __name__ == "__main__": import uvicorn port = int(os.environ.get("PORT", os.environ.get("SIGNALMESH_PORT", 7475))) print(f"\n📡 SignalMesh — Sovereign Liquid Matrix Gateway v2.0") print(f" http://0.0.0.0:{port} → /docs → /redoc") print(f" UI routes: /ui/* (no key) | API routes: /api/* (X-SignalMesh-Key)") print(f" Set SIGNALMESH_API_KEY env var to change the key\n") uvicorn.run("signalmesh_api:app", host="0.0.0.0", port=port, reload=False)