""" telegram_bot.py — JARVIS Telegram bot (Hugging Face Spaces primary, no Render needed). Polls @AyomideJarvis_bot via getUpdates (default) or serves a Telegram webhook, and replies via the Muse Spark brain (OpenCode Zen) with Firebase memory. ALL Telegram network I/O goes through telegram_transport.py: TELEGRAM_TRANSPORT=auto|direct|proxy, TELEGRAM_PROXY_URL, TELEGRAM_PROXY_AUTH See telegram_transport.py for the full variable list and log markers. Secrets (Space Secrets / .env — NEVER hardcoded): TELEGRAM_BOT_TOKEN (required), OPENCODE_API_KEY, OPENCODE_MODEL, FIREBASE_CREDENTIALS_JSON, JARVIS_TG_UID, N8N_REMINDER_WEBHOOK, TELEGRAM_MODE=polling|webhook, SPACE_PUBLIC_URL, TELEGRAM_WEBHOOK_SECRET Endpoints on :7860 (HF Docker requirement): / health (model, polls, offset, last error, transport) /diag HF-environment network test matrix (JSON, no secrets) /webhook/ Telegram webhook receiver (webhook mode only) JARVIS brain/skills below are unchanged behavior: Firebase memory/notes/ reminders, PC-status heartbeat, Firebase pc-command queue (executed on the PC by pc_heartbeat.py with full MCP), typo-tolerant commands, model-first 1.3 answers with hardcoded fallback only when the model fails. """ import os import re import json import time import signal import threading import requests # brain (Zen) + Firebase-adjacent posts; Telegram uses transport import telegram_transport as tg # ── .env (local dev only; HF uses Space Secrets) ──────────────────── def _load_dotenv(path=".env"): try: if not os.path.exists(path): return with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) k = k.strip(); v = v.strip().strip('"').strip("'") if k and v and k not in os.environ: os.environ[k] = v except Exception: pass _load_dotenv(".env") # ── Required secrets (fail fast, no hardcoded fallbacks) ──────────── BOT = (os.environ.get("TELEGRAM_BOT_TOKEN") or "").strip().strip('"').strip("'") if not BOT: print("TELEGRAM_NO_TOKEN TELEGRAM_BOT_TOKEN is not set. " "Add it to Hugging Face Space Secrets (or local .env).", flush=True) raise SystemExit(1) ZEN_URL = "https://opencode.ai/zen/v1/responses" MODEL = (os.environ.get("OPENCODE_MODEL") or "muse-spark-1.3-contributor-free").strip() KEY = (os.environ.get("OPENCODE_API_KEY") or "").strip() MODE = (os.environ.get("TELEGRAM_MODE") or "polling").strip().lower() SPACE_PUBLIC_URL = (os.environ.get("SPACE_PUBLIC_URL") or "").strip().rstrip("/") WEBHOOK_SECRET = (os.environ.get("TELEGRAM_WEBHOOK_SECRET") or "").strip() PC_BRIDGE_URL = (os.environ.get("PC_BRIDGE_URL") or "").strip() _poll_status = {"last_poll": "never", "last_err": "", "offset": 0, "polls": 0, "transport": tg.TRANSPORT, "replies": 0} _last_chat_id = None _shutdown = False def _on_signal(signum, frame): global _shutdown print(f"shutdown signal {signum} — finishing current poll, sir.", flush=True) _shutdown = True for _sig in ("SIGTERM", "SIGINT"): try: signal.signal(getattr(signal, _sig), _on_signal) except Exception: pass # ── Reminders support (Firebase or local fallback) ────────────────── try: import reminders as _rem except Exception: _rem = None def _init_firestore(): try: import firebase_admin from firebase_admin import credentials, firestore if not firebase_admin._apps: js = (os.environ.get("FIREBASE_CREDENTIALS_JSON") or "").strip() if js: firebase_admin.initialize_app(credentials.Certificate(json.loads(js))) else: p = os.path.join(os.path.dirname(__file__), "firebase_service_account.json") if os.path.exists(p): firebase_admin.initialize_app(credentials.Certificate(p)) return firestore.client() except Exception: return None def _pc_status(): try: db = _init_firestore() if db is None: return None, {"error": "no firebase"}, 9999 doc = db.collection("jarvis_status").document("pc").get() if doc.exists: d = doc.to_dict() age = time.time() - d.get("last_seen", 0) return age < 90, d, age except Exception as e: return None, {"error": str(e)}, 9999 return False, {}, 9999 def _chat_history_get(chat_id, limit=6): """Shared BAT<->bot conversation via memory_cloud (single doc, no index).""" try: import memory_cloud return [{"r": m.get("r"), "t": m.get("t", "")} for m in memory_cloud.chat_load(limit)] except Exception: pass return [] def _chat_history_put(chat_id, user_text, reply_text): try: import memory_cloud memory_cloud.chat_save("u", user_text or "", src=f"tg:{chat_id}") memory_cloud.chat_save("a", reply_text or "", src=f"tg:{chat_id}") except Exception: pass def _cloud_reminder_loop(): """Fire due Firebase reminders to Telegram even when the PC is off.""" db = _init_firestore() n8n_url = (os.environ.get("N8N_REMINDER_WEBHOOK") or "").strip() while not _shutdown: try: if db: now = time.time() for snap in db.collection("jarvis_reminders").where("due", "<=", now).stream(): task = (snap.to_dict() or {}).get("task") try: snap.reference.delete() except Exception: pass if task and _last_chat_id: try: tg.send_message(_last_chat_id, f"\u23f0 Reminder, sir: {task}") except Exception as e: print("reminder send err", tg._redact(str(e)), flush=True) print("cloud reminder fired:", task, flush=True) if task and n8n_url: try: requests.post(n8n_url, json={"text": f"Reminder: {task}"}, timeout=10) except Exception: pass except Exception: pass time.sleep(15) import datetime as _dt def _norm_triggers(text): """Fix jammed/misspelled action triggers before matching (speech-to-text garble).""" t = text.lower() t = re.sub(r"\bremind\s*me\b", "remind me", t) t = re.sub(r"\bremid\s*me\b", "remind me", t) # "remidmei want..." -> "remind me..." t = re.sub(r"\bremider\b", "reminder", t) t = re.sub(r"\bchage\b", "change", t) t = re.sub(r"\bchageto\b", "change to", t) # "chageto1121" t = re.sub(r"\bcnacel\b", "cancel", t) t = re.sub(r"\bdelet\b", "delete", t) return t def _latest_reminder(): """Newest pending reminder doc (or None). Streams all — count is tiny.""" try: db = _init_firestore() if db is None: return None, None docs = list(db.collection("jarvis_reminders").stream()) if not docs: return db, None docs.sort(key=lambda s: (s.to_dict() or {}).get("due", 0)) return db, docs[-1] except Exception: return None, None def _manage_reminder(text): """Edit/list/cancel the pending reminder. Returns reply or None if not applicable.""" low = _norm_triggers(text) if ("reminder" not in low and "remind" not in low and not any(k in low for k in ["change", "switch", "move", "update", "cancel", "delete", "remove", "forget"])): return None db, doc = _latest_reminder() if db is None: return "Couldn't reach the reminders database, sir." d = (doc.to_dict() or {}) if doc else {} if any(k in low for k in ["cancel", "delete", "remove", "forget"]): if not doc: return "No reminder to cancel, sir — none scheduled." doc.reference.delete() return f"Cancelled, sir — no more reminder to {d.get('task', '...')}." if "change" in low or "switch" in low or "move" in low or "update" in low: if not doc: return "No reminder to change, sir — tell me a new one instead." m = re.search(r"(\d{1,2})\s*[:.]?\s*(\d{2})\s*(am|pm|a\.m\.|p\.m\.)?\b", low) if not m: return "Change it to what time, sir?" hh, mm, ap = int(m.group(1)), int(m.group(2)), (m.group(3) or "") if "p" in ap and hh < 12: hh += 12 if "a" in ap and hh == 12: hh = 0 wat = _dt.datetime.now(_dt.timezone.utc) + _dt.timedelta(hours=1) target = wat.replace(hour=hh, minute=mm, second=0, microsecond=0) now = time.time() if target.timestamp() <= now + 3600: target = target + _dt.timedelta(days=1) doc.reference.update({"due": target.timestamp() - 3600}) when = (target).strftime("%I:%M %p").lstrip("0") return f"Done, sir — reminder moved to {when} WAT to {d.get('task', '...')}." if any(k in low for k in ["the reminder", "my reminder", "show reminder", "list reminder", "what.*remind", "pending reminder"]): if not doc: return "No reminders scheduled, sir." when = (_dt.datetime.fromtimestamp(d.get("due", 0), _dt.timezone.utc) + _dt.timedelta(hours=1)).strftime("%I:%M %p").lstrip("0") return f"Your reminder, sir: {d.get('task', '...')} at {when} WAT." return None def _parse_cloud_reminder(text): low = _norm_triggers(text) task, due_ts = "", None body = re.split(r"remind me (?:to )?", low, maxsplit=1)[-1] if "wake me up" in low: body = low.split("wake me up", 1)[-1] now = time.time() m = re.search(r"in (\d+)\s*(minute|min|hour|hr|second|sec)s?", body) if m: n = int(m.group(1)); unit = m.group(2) if unit.startswith("hour") or unit.startswith("hr"): delta = n * 3600 elif unit.startswith("minute") or unit.startswith("min"): delta = n * 60 else: delta = n due_ts = now + delta task = re.sub(r"in \d+\s*[\w]+s?", "", body).strip(" .") else: # Absolute time: "at" optional, colon optional ("at 10:48", "10 48pm", "10:48pm"). m = re.search(r"(?:\bat\s+)?(\d{1,2})\s*[:.]?\s*(\d{2})\s*(am|pm|a\.m\.|p\.m\.)?\b", body) if m: # "at HH:MM" means the user's wall clock (WAT = UTC+1); server runs UTC. hh, mm, ap = int(m.group(1)), int(m.group(2)), (m.group(3) or "") if 0 <= hh <= 23 and 0 <= mm <= 59: if "p" in ap and hh < 12: hh += 12 if "a" in ap and hh == 12: hh = 0 wat = _dt.datetime.now(_dt.timezone.utc) + _dt.timedelta(hours=1) target = wat.replace(hour=hh, minute=mm, second=0, microsecond=0) if now - 300 <= target.timestamp() - 3600 <= now: due_ts = now + 30 # just passed (<5 min) → nudge right now else: if target.timestamp() <= now + 3600: target = target + _dt.timedelta(days=1) due_ts = target.timestamp() - 3600 task = re.sub(r"(?:\bat\s+)?\d{1,2}\s*[:.]?\s*\d{2}\s*(?:am|pm|a\.m\.|p\.m\.)?\b", "", body).strip(" .") # Strip leftover fragments like "I." from speech-to-text garble. task = re.sub(r"\bif\s+something\b", "something", task).strip(" .") if due_ts is None and not task: return "What should I remind you about, and when, sir?" if due_ts is None: return f"Remind you {task} — when, sir? (e.g. in 30 minutes, or at 10:48 PM)" if not task: when_q = (_dt.datetime.fromtimestamp(due_ts, _dt.timezone.utc) + _dt.timedelta(hours=1)).strftime("%I:%M %p").lstrip("0") return f"At {when_q} WAT — remind you of what, sir?" if not task: task = "reminder" db = _init_firestore() if db is None: return "Couldn't save the reminder, sir — database unreachable." try: db.collection("jarvis_reminders").add({"task": task, "due": due_ts}) when = (_dt.datetime.fromtimestamp(due_ts, _dt.timezone.utc) + _dt.timedelta(hours=1)).strftime("%I:%M %p").lstrip("0") return f"Right away, sir. I'll remind you at {when} WAT to {task} — sent to Telegram even if my PC is off." except Exception as e: return f"Couldn't save reminder: {e}" def _enqueue_pc_command(text, chat_id): try: db = _init_firestore() if db is None: return False db.collection("jarvis_pc_commands").add({ "text": text, "chat_id": str(chat_id or ""), "created": time.time(), "status": "pending"}) return True except Exception: return False def _try_pc_bridge(text): low = text.lower() trigger = any(k in low for k in [" open ", "create a file", "screenshot", "play ", "pause", "youtube", "stop music", "close ", "search youtube", "open app", "control", "volume"]) if not trigger or not PC_BRIDGE_URL: return None try: r = requests.post(PC_BRIDGE_URL, json={"text": text}, timeout=25) if r.status_code == 200: return (r.json() or {}).get("reply") except Exception: pass return None def brain(text, _depth=0): # Approve/reject with auto-resume: approving a Telegram PC request # automatically continues the original command (no re-send needed). try: _m = re.search(r"\b(approve|reject)\s+(ap-[0-9a-f]{4,8})\b", (text or "").lower()) if _m and _depth < 2: from jarvis_core import approvals as _ap2, audit as _au2 _a2 = _ap2.decide(_m.group(2), _m.group(1) == "approve", by="boss") if _a2: try: _au2.record("boss", "telegram", f"{_m.group(1)} {_a2['id']}", tool="approvals", status="SUCCESS", result=(_a2.get("action") or "")[:150]) except Exception: pass _orig = ((_a2.get("meta") or {}).get("orig_text") or "").strip() if _m.group(1) == "approve" and _orig: _resume = brain(_orig, _depth + 1) return f"Approved, sir — continuing.\n{_resume}" except Exception: pass # Central router first: missions/jobs/approvals/health/routines/audit/memory. try: from jarvis_core.router import handle as core_handle routed = core_handle(text, channel="telegram") if routed: try: from jarvis_core import audit as _audit _audit.record("boss", "telegram", text[:120], tool="router", status="SUCCESS", result=routed[:150]) except Exception: pass return routed except Exception: pass # Full local brain first when running on the PC (MCP, Chrome, screenshots). try: import local_brain as lb ans = lb.handle(text) if ans: return ans except Exception: pass pcb = _try_pc_bridge(text) if pcb: return pcb low = _norm_triggers(text) nospace = low.replace(" ", "") # Reminder intents ALWAYS take the real handler — the model must never # roleplay setting one. ("remind", "remid", "reminder", "wake me up") _has_time = bool(re.search(r"(\d{1,2})\s*[:.]?\s*(\d{2})\s*(am|pm|a\.m\.|p\.m\.)?\b", low)) _edit_verb = any(k in low for k in ["change", "switch", "move", "update"]) _kill_verb = any(k in low for k in ["cancel", "delete", "remove", "forget"]) if ("wake me up" in low or "remind" in low or "remid" in low or (_edit_verb and (_has_time or " it" in low or " that" in low or "reminder" in low)) or (_kill_verb and (" it" in low or " that" in low or "reminder" in low))): managed = _manage_reminder(text) if managed: return managed return _parse_cloud_reminder(text) # Identity first — never dodge the model question. if any(k in low for k in ["which model", "what model", "what llm", "which llm", "brian", "brain llm", "what are you running on", "model are u", "model are you"]): return ("I run on Muse Spark, sir — model muse-spark-1.3-contributor-free, " "served via OpenCode Zen. Same brain on your BAT and here.") # Status questions get the detailed heartbeat reply — never the generic OFF text. if "is my pc on" in low or "is pc on" in low or "pc status" in low: on, d, age = _pc_status() if on is None: return f"I couldn't check PC status, sir: {d.get('error')}" if on: return (f"Yes, your PC is ON, sir — last heartbeat {int(age)}s ago " f"({d.get('last_seen_str')}). {d.get('details', '')}") return (f"Your PC is OFF, sir — no heartbeat for {int(age)}s " f"(last seen {d.get('last_seen_str', 'unknown')}). " "If the PC is actually on, open Jarvis_Local.bat so the heartbeat resumes.") has_access = any(k in low for k in ["access", "acess", "acces"]) or \ any(k in nospace for k in ["accessmypc", "acessmypc"]) has_pc = "pc" in low or "mypc" in nospace needs_pc = has_access and has_pc if not needs_pc: needs_pc = any(k in low for k in [ "downloads folder", "desktop folder", "documents folder", "what's in my", "what is in my", "list my files", "my files", "open my", "open the", "screenshot", "play video", "play music", "stop music", "volume", "search youtube", "open app", "open application", "create a file", "what is on my screen", "control my pc", "close the", "start the", "run program", "my pc's", "my pc ", "mypc", "navigate to"]) if needs_pc: try: from jarvis_core import approvals as _ap, audit as _au _risk = _ap.classify(text) except Exception: _ap, _au, _risk = None, None, "LOW_RISK" if _risk in ("MEDIUM_RISK", "HIGH_RISK") and _ap is not None: try: from jarvis_core import permissions as _pm _grant = _pm.recent_text_grant(text) except Exception: _grant = None if _grant is None: a = _ap.request(text, channel="telegram", actor="boss", meta={"orig_text": text}) try: _au.record("boss", "telegram", text[:120], tool="pc_queue", approval=_risk, approval_status=a["status"], status="SUCCESS", result=f"approval {a['id']}") except Exception: pass if a["status"] != "APPROVED": return ("That one needs your approval first, sir.\n" + _ap.format_request(a)) on, d, age = _pc_status() if on is True: if not PC_BRIDGE_URL: if _enqueue_pc_command(text, _last_chat_id): return ("Yes, sir — I have full access to your PC right now. " "I've queued your request; your PC will reply here in a few seconds.") return "Your PC is ON, sir, but I couldn't queue the task. Ask me directly on the PC." return "Your PC is ON, sir, but I couldn't reach it for that task right now." if on is False: return ("Your PC is OFF, sir, so I can't access your files or control apps. " "Turn it on (with Jarvis running) and ask again.") if _enqueue_pc_command(text, _last_chat_id): return "I've queued that for your PC, sir — it will run when your PC comes online." if "cold call" in low: return ("Cold calling is available on your PC BAT only, sir — open Jarvis_Local.bat " "and say 'cold call +234... say hello'.") if not KEY: return "Missing OPENCODE_API_KEY, sir." if "pending reminders" in low or "list reminders" in low or "what are my reminders" in low: db = _init_firestore() if db is None: return "Couldn't reach the reminders database, sir." try: items = [s.to_dict() for s in db.collection("jarvis_reminders").stream()] if not items: return "No reminders scheduled, sir." txt = " ".join( f"{i['task']} at {_dt.datetime.fromtimestamp(i['due']).strftime('%I:%M %p')}," for i in items) return f"Your reminders, sir: {txt}" except Exception as e: return f"Couldn't list reminders: {e}" if "save this to" in low and "database" in low: try: import memory_cloud ok = memory_cloud.remember(text) return ("Saved to your database, sir — Firebase \"jarvis_memory\" " "(project jarvisai-994bd).") if ok else "I couldn't save that, sir — database offline." except Exception as e: return f"Save failed: {e}" if "remember that" in low: try: import memory_cloud fact = text.lower().split("remember that", 1)[1].strip(" .") ok = memory_cloud.remember(fact) return "Right away, I'll remember that, sir — stored in Firebase \"jarvis_memory\"." \ if ok else "Couldn't save, sir." except Exception: pass if "is my pc on" in low or "is pc on" in low or "pc status" in low: on, d, age = _pc_status() if on is None: return f"I couldn't check PC status, sir: {d.get('error')}" if on: return (f"Yes, your PC is ON, sir — last heartbeat {int(age)}s ago " f"({d.get('last_seen_str')}). {d.get('details', '')}") return f"Your PC is OFF, sir — last seen {int(age)}s ago at {d.get('last_seen_str', 'unknown')}." # ── All remaining questions go to the 1.3 model (MCP-aware) ── _MCP_CAPS = ("You have full PC control via your local executor (71 MCP tools): open apps/files/sites, " "screenshots + screen vision, browser with 8 Chrome profiles (Default=haywhyeditz TikTok), " "YouTube/TikTok automation, volume, notes/memory, reminders. " "Data lives in Firebase project jarvisai-994bd (jarvis_memory, jarvis_notes, " "jarvis_reminders, jarvis_status/pc). " "Identity: you are Jarvis, model " + MODEL + " via " "OpenCode Zen — say so plainly when asked, never dodge. " "You CANNOT set, change, or cancel reminders yourself — if the user wants " "any of that, say you need the exact task and time, and never claim it is done.") try: _now_utc = _dt.datetime.now(_dt.timezone.utc) _now_wat = _now_utc + _dt.timedelta(hours=1) _clock = (f"Current time: {_now_wat.strftime('%I:%M:%S %p').lstrip('0')} WAT " f"({_now_utc.strftime('%H:%M:%S')} UTC) on {_now_wat.strftime('%A, %B %d, %Y')}. " "When asked for the time, always give the FULL time with seconds and timezone, sir.") except Exception: _clock = "" try: _msgs = [] for h in _chat_history_get(_last_chat_id): _msgs.append({"role": "user" if h.get("r") == "u" else "assistant", "content": h.get("t", "")}) _msgs.append({"role": "user", "content": text}) _mem = "" try: import memory_cloud _mem = (memory_cloud.recall(8) or "")[:800] except Exception: pass _ctx = ("Recent conversation: " + " | ".join( f"{'Boss' if m['role'] == 'user' else 'JARVIS'}: {m['content']}" for m in _msgs[:-1]) if len(_msgs) > 1 else "No prior conversation.") if _mem: _ctx += f" Remembered facts: {_mem}." payload = {"model": MODEL, "input": _msgs, "instructions": "You are J.A.R.V.I.S., witty loyal assistant. Under 2 sentences. " + _MCP_CAPS + " " + _clock + " " + _ctx} headers = {"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"} r = requests.post(ZEN_URL, json=payload, headers=headers, timeout=45) data = r.json() out = [] for item in data.get("output") or []: if item.get("type") == "message": for c in item.get("content") or []: if c.get("type") == "output_text": out.append(c.get("text", "")) ans = "".join(out).strip() if ans: return ans except Exception as e: print("model err", str(e)[:200], flush=True) # ── Offline fallback only (model failed) ── if any(k in low for k in ["where is your database", "where is your memory", "where do you store", "your database name", "what is your database", "where do you keep", "where is your data", "where do you save", "database name"]): return ("My data lives in Firebase, sir — project \"jarvisai-994bd\". " "Memory & conversation history: \"jarvis_memory\", notes: \"jarvis_notes\", " "reminders: \"jarvis_reminders\", and your PC status: \"jarvis_status/pc\". " "It's cloud-hosted, so I can recall it even when this PC is off.") if "where" in low and "conversation" in low and ("save" in low or "store" in low): return "Conversations are saved in Firebase jarvis_memory, sir." if "collect" in low and "details" in low and "pc" in low: on, d, age = _pc_status() if on: return f"Collected from your PC (ON): {d.get('details', '')} — heartbeat {int(age)}s ago, sir." return "Your PC is OFF, sir — I can't collect fresh details until it's on. Last details: " + \ d.get("details", "none") if "what time" in low or "time is it" in low or "whatisthetime" in nospace: _wat = _dt.datetime.now(_dt.timezone.utc) + _dt.timedelta(hours=1) return (f"It is {_wat.strftime('%I:%M:%S %p').lstrip('0')} WAT " f"({_wat.strftime('%A, %B %d, %Y')}), sir.") if "hello" in low and len(low) < 20: return "At your service, sir. What do you need?" return "(no reply)" def handle_update_text(text, chat_id): """Shared by polling loop and webhook handler. Returns reply string.""" global _last_chat_id _last_chat_id = chat_id print(f"\U0001f4e9 {text} from {chat_id}", flush=True) reply = brain(text) _chat_history_put(chat_id, text, reply) tg.send_message(chat_id, reply) _poll_status["replies"] += 1 print(f"\U0001f4e4 {reply[:80]}", flush=True) return reply # ── HTTP server: / health, /diag network matrix, /webhook receiver ── def _serve_http(): from http.server import BaseHTTPRequestHandler, HTTPServer class H(BaseHTTPRequestHandler): def do_GET(self): try: if self.path == "/dashboard" or self.path.startswith("/dashboard?"): from urllib.parse import urlparse, parse_qs try: from jarvis_core import missions, jobs, health as _h, audit as _a, events as _e from jarvis_core import approvals as _ap qs = parse_qs(urlparse(self.path).query) _key_ok = bool(WEBHOOK_SECRET) and (qs.get("key", [""])[0] == WEBHOOK_SECRET) _acted = "" _act, _aid = (qs.get("action", [""])[0] or "").lower(), qs.get("id", [""])[0] or "" if _act in ("approve", "reject") and _aid: if not _key_ok: _acted = "Missing or wrong ?key — approval NOT changed. Append ?key=YOUR_TELEGRAM_WEBHOOK_SECRET." else: _d = _ap.decide(_aid, _act == "approve", by="dashboard") if _d: _acted = f"{_aid} {_d.get('status')}." try: _a.record("boss", "dashboard", f"{_act} {_aid}", tool="approvals", status="SUCCESS", result=(_d.get("action") or "")[:150]) except Exception: pass else: _acted = f"{_aid} not found or already decided." st = _h.all_status() mrun = len([m for m in missions.list_all(active_only=True)]) jrun = len(jobs.list_all("RUNNING")) + len(jobs.list_all("QUEUED")) jfail = len(jobs.list_all("FAILED")) evts = _e.recent(5) acts = _a.recent(5) errs = _a.failures(3) L = ["JARVIS CONTROL DASHBOARD", "", "SYSTEM STATUS"] L += [f"{k}: {v['status']} ({v['detail']})" for k, v in st.items()] L += ["", f"MISSIONS active: {mrun}", f"JOBS running/queued: {jrun} failed: {jfail}", "", "PENDING APPROVALS"] _pend = _ap.pending()[:5] if _pend: for _p in _pend: _pid = _p.get("id") L.append(f"- {_pid} [{_p.get('risk')}] {_p.get('action')}") if _key_ok: L.append(f" approve: /dashboard?action=approve&id={_pid}&key={WEBHOOK_SECRET}") L.append(f" reject: /dashboard?action=reject&id={_pid}&key={WEBHOOK_SECRET}") else: L.append(" (add ?key=YOUR_TELEGRAM_WEBHOOK_SECRET to get one-tap links)") else: L.append("- none") if _acted: L += ["", f"ACTION: {_acted}"] L += ["", "RECENT EVENTS"] L += [f"- {e.get('type')} [{e.get('decision')}]" for e in evts] or ["- none"] L += ["", "RECENT ACTIONS"] L += [f"- {_a.format_row(a)}" for a in acts] or ["- none"] L += ["", "ERRORS"] L += [f"- {_a.format_row(a)}" for a in errs] or ["- none"] body = "\n".join(L).encode() except Exception as e: body = f"dashboard unavailable: {e}".encode() self.send_response(200) self.send_header("Content-type", "text/plain") self.end_headers() self.wfile.write(body) return if self.path == "/diag": body = json.dumps(tg.diag(), indent=1).encode() self.send_response(200) self.send_header("Content-type", "application/json") self.end_headers() self.wfile.write(body) return global _tg_ok_cache, _tg_ok_at try: now = time.time() except Exception: now = 0 if now - globals().get("_tg_ok_at", 0) > 60: try: me = tg.get_me() _tg_ok_cache = f"tg getMe ok bot={(me.get('result') or {}).get('username', '?')}" except tg.TelegramTransportError as e: _tg_ok_cache = f"{e.marker}: {tg._redact(str(e))[:100]}" except Exception as e: _tg_ok_cache = f"tg err {str(e)[:100]}" globals()["_tg_ok_at"] = now tg_ok = globals().get("_tg_ok_cache", "tg n/a") body = (f"JARVIS Telegram bot running\nModel {MODEL}\nMode {MODE}\n" f"Transport {tg.TRANSPORT}\nKey {'yes' if KEY else 'no'}\n" f"Polls {_poll_status['polls']} last:{_poll_status['last_poll']} " f"err:{_poll_status['last_err']}\nTG:{tg_ok}\n" f"Offset {_poll_status['offset']} Replies {_poll_status['replies']}").encode() self.send_response(200) self.send_header("Content-type", "text/plain") self.end_headers() self.wfile.write(body) except Exception as e: try: self.send_response(500) self.end_headers() except Exception: pass print("http err", str(e)[:150], flush=True) def do_POST(self): # Telegram webhook receiver: POST /webhook/ try: if MODE != "webhook" or not WEBHOOK_SECRET or \ self.path != f"/webhook/{WEBHOOK_SECRET}": self.send_response(404) self.end_headers() return length = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(length) if length else b"" self.send_response(200) self.end_headers() try: upd = json.loads(raw.decode("utf-8") or "{}") msg = upd.get("message") or {} text = msg.get("text") or "" cid = (msg.get("chat") or {}).get("id") if text and cid: threading.Thread(target=handle_update_text, args=(text, cid), daemon=True).start() except Exception as e: print("webhook update err", str(e)[:150], flush=True) except Exception as e: print("webhook err", str(e)[:150], flush=True) def log_message(self, *a): pass s = HTTPServer(("0.0.0.0", 7860), H) threading.Thread(target=s.serve_forever, daemon=True).start() print("HTTP server on :7860 (/ /diag /webhook)", flush=True) def _poll_loop(): global _last_chat_id offset = 0 try: tg.delete_webhook(drop_pending=True) print("deleteWebhook called (sole poller)", flush=True) except tg.TelegramTransportError as e: print(f"deleteWebhook {e.marker}: {tg._redact(str(e))[:150]}", flush=True) print(f"JARVIS Telegram poller live (model {MODEL}, transport {tg.TRANSPORT})", flush=True) while not _shutdown: try: _poll_status["polls"] += 1 _poll_status["last_poll"] = time.strftime("%H:%M:%S") data = tg.get_updates(offset) _poll_status["last_err"] = "" for upd in data.get("result") or []: offset = upd["update_id"] + 1 _poll_status["offset"] = offset msg = upd.get("message") or {} chat = msg.get("chat") or {} text = msg.get("text") or "" cid = chat.get("id") if not text or not cid: continue try: tg.send_chat_action(cid) except Exception: pass handle_update_text(text, cid) if not _shutdown: time.sleep(2) # never tight-loop the relay, even on empty results except tg.TelegramTransportError as e: _poll_status["last_err"] = f"{e.marker}: {tg._redact(str(e))[:100]}" print(f"poll {e.marker} {tg._redact(str(e))[:150]}", flush=True) # A rate-limited relay must be given a full minute to drain; # anything shorter re-trips the limiter and self-perpetuates. time.sleep(60 if "rate_limited" in str(e) else 5) except Exception as e: _poll_status["last_err"] = str(e)[:120] print("poll err", str(e)[:200], flush=True) time.sleep(3) def main(): _serve_http() threading.Thread(target=_cloud_reminder_loop, daemon=True).start() if MODE == "webhook": if not SPACE_PUBLIC_URL or not WEBHOOK_SECRET: print("TELEGRAM_WEBHOOK_INCOMPLETE set SPACE_PUBLIC_URL + TELEGRAM_WEBHOOK_SECRET, " "falling back to polling.", flush=True) _poll_loop() return url = f"{SPACE_PUBLIC_URL}/webhook/{WEBHOOK_SECRET}" try: tg.set_webhook(url) print(f"TELEGRAM_WEBHOOK_ACTIVE {SPACE_PUBLIC_URL}/webhook/", flush=True) except tg.TelegramTransportError as e: print(f"webhook setup {e.marker} — falling back to polling.", flush=True) _poll_loop() return while not _shutdown: time.sleep(10) return _poll_loop() if __name__ == "__main__": main()