#!/usr/bin/env python3 """ Wizard-summon CLI — timmy-config#441 Usage: wizard-summon.py [--priority P0|P1|P2] "topic" [--deadline ISO8601] Creates/updates wizards/shared_context.json on Gitea, opens a PR, and posts structured notice to Telegram broadcast group. Exit codes: 0 = success (summon created/updated) 1 = error (network, auth, validation) 2 = blocked (already-open summon exists) """ import argparse import json import os import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Optional # ── Constants ── GITEA_TOKEN = Path.home() / ".config" / "gitea" / "token" GITEA_API = "https://forge.alexanderwhitestone.com/api/v1" REPO_OWNER = "Timmy_Foundation" REPO_NAME = "timmy-config" BRANCH = "step35/441-p1-wizard-to-wizard-communic" BASE_BRANCH = "main" SHARED_CONTEXT_PATH = "wizards/shared_context.json" TELEGRAM_TOKEN_FILE = Path.home() / ".config" / "telegram" / "special_bot" TELEGRAM_CHAT_ID = "-1003664764329" # Timmy Time group # Wizard identities — must match ansible/inventory and sidecar configs WIZARDS = ["timmy", "allegro", "bezalel", "ezra"] PRIORITIES = {"P0", "P1", "P2"} STATUSES = {"idle", "busy", "acked_summon", "error"} # ── Helpers ── def read_token() -> str: if not GITEA_TOKEN.exists(): raise FileNotFoundError(f"Gitea token not found at {GITEA_TOKEN}") return GITEA_TOKEN.read_text().strip() def read_telegram_token() -> Optional[str]: if not TELEGRAM_TOKEN_FILE.exists(): return None return TELEGRAM_TOKEN_FILE.read_text().strip() def gitea_request(method: str, path: str, body: Optional[dict] = None) -> dict: import urllib.request token = read_token() url = f"{GITEA_API}{path}" data = json.dumps(body).encode() if body else None req = urllib.request.Request( url, data=data, method=method, headers={ "Authorization": f"token {token}", "Content-Type": "application/json", "Accept": "application/json", } ) try: with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as e: err = e.read().decode() if e.read() else str(e) raise RuntimeError(f"Gitea API {method} {path} failed: HTTP {e.code} — {err}") from e def get_current_shared_context() -> dict: """Fetch current shared_context.yaml via raw GitHub-style content API.""" import urllib.request import base64 token = read_token() url = f"{GITEA_API}/repos/{REPO_OWNER}/{REPO_NAME}/contents/{SHARED_CONTEXT_PATH}?ref={BASE_BRANCH}" req = urllib.request.Request( url, headers={"Authorization": f"token {token}", "Accept": "application/json"} ) try: with urllib.request.urlopen(req, timeout=15) as resp: data = json.loads(resp.read()) content_b64 = data.get("content", "") content = base64.b64decode(content_b64).decode("utf-8") sha = data.get("sha") return {"content": content, "sha": sha} except urllib.error.HTTPError as e: if e.code == 404: return {"content": None, "sha": None} raise RuntimeError(f"Failed to fetch {SHARED_CONTEXT_PATH}: HTTP {e.code}") from e def parse_json(content: str) -> dict: import yaml return yaml.safe_load(content) or {} def dump_json(data: dict) -> str: import yaml return yaml.safe_dump(data, sort_keys=False, default_flow_style=False, allow_unicode=True) def now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def telegram_broadcast(message: str) -> bool: """Post structured message to Telegram group. Returns True if delivered.""" token = read_telegram_token() if not token: print("[warn] Telegram token not found — skipping broadcast", file=sys.stderr) return False import urllib.parse import urllib.request params = urllib.parse.urlencode({ "chat_id": TELEGRAM_CHAT_ID, "parse_mode": "MarkdownV2", "disable_web_page_preview": "true", }).encode() url = f"https://api.telegram.org/bot{token}/sendMessage" req = urllib.request.Request(url, data=params, method="POST") try: with urllib.request.urlopen(req, timeout=10) as resp: result = json.loads(resp.read()) return result.get("ok", False) except Exception as e: print(f"[error] Telegram send failed: {e}", file=sys.stderr) return False # ── Main ── def main(): ap = argparse.ArgumentParser(description="Summon all wizards — wizard-to-wizard comms (#441)") ap.add_argument("topic", help="Summon topic (concise, actionable)") ap.add_argument("--priority", choices=list(PRIORITIES), default="P1", help="Summon priority (default: P1)") ap.add_argument("--deadline", help="Deadline in ISO8601 (e.g., 2026-04-26T18:00:00Z)") ap.add_argument("--dry-run", action="store_true", help="Show what would happen, no API calls") args = ap.parse_args() # ── Validate ── if args.priority not in PRIORITIES: raise ValueError(f"Invalid priority: {args.priority} — must be one of {sorted(PRIORITIES)}") # Check for already-open active summon current = get_current_shared_context() if current["content"]: current_data = parse_json(current["content"]) active = current_data.get("active_summon", {}) if active and active.get("status") in ("open", "acknowledged"): print(f"[blocker] Active summon already open: {active.get('summon_id')} — {active.get('topic')}") print(f" Status: {active.get('status')}, priority: {active.get('priority')}") return 2 # blocked summon_id = f"SUM-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}" # ── Build new context ── if current["content"]: data = parse_json(current["content"]) else: data = { "version": "1.0", "updated_at": now_iso(), "wizard_status": {w: {"status": "idle", "last_seen": None} for w in WIZARDS}, "message_log": [], } # Replace active_summon data["active_summon"] = { "summon_id": summon_id, "priority": args.priority, "topic": args.topic, "summoner": "Alexander", "summoned_at": now_iso(), "deadline": args.deadline, "status": "open", "acknowledgements": {w: None for w in WIZARDS}, } data["updated_at"] = now_iso() new_content = dump_json(data) # ── Dry-run ── if args.dry_run: print("[dry-run] New shared_context.yaml:") print(new_content) print(f"\nSummon ID: {summon_id}, Priority: {args.priority}") return 0 # ── Commit to Gitea ── commit_msg = f"wizard-comm: create summon {summon_id} ({args.priority}) — {args.topic[:50]}" # Create/update file via Gitea API (PUT contents) path = f"/repos/{REPO_OWNER}/{REPO_NAME}/contents/{SHARED_CONTEXT_PATH}" # Encode to base64 for API import base64 content_b64 = base64.b64encode(new_content.encode("utf-8")).decode("ascii") body = { "message": commit_msg, "content": content_b64, "branch": BRANCH, "author": {"name": "step35-bot", "email": "step35-burn@alexanderwhitestone.com"}, } if current["sha"]: body["sha"] = current["sha"] result = gitea_request("PUT", path, body) commit_sha = result.get("commit", {}).get("sha", "?") print(f"Committed to branch '{BRANCH}': {commit_sha[:8]}") # ── Open PR ── pr_body = ( f"**Wizard Summon** — `{summon_id}`\n\n" f"**Priority:** {args.priority}\n" f"**Topic:** {args.topic}\n" f"**Summoner:** Alexander\n" f"**Deadline:** {args.deadline or 'not set'}\n\n" f"This PR creates the shared context update that implements wizard-to-wizard " f"communication via `wizards/shared_context.json`. Closes #441." ) pr = gitea_request( "POST", f"/repos/{REPO_OWNER}/{REPO_NAME}/pulls", body={ "title": f"feat(comm): wizard summon {summon_id} — {args.topic[:60]}", "body": pr_body, "head": BRANCH, "base": BASE_BRANCH, "draft": False, } ) pr_num = pr["number"] pr_url = pr["html_url"] print(f"Opened PR: #{pr_num} — {pr_url}") # ── Broadcast to Telegram ── priority_emoji = {"P0": "🚨", "P1": "⚠️", "P2": "📢"}.get(args.priority, "📣") telegram_msg = ( f"{priority_emoji} *Wizard Summon: {args.priority}*\n\n" f"*Topic:* {args.topic}\n" f"*Summoner:* Alexander\n" f"*Summon ID:* `{summon_id}`\n" f"*Gitea PR:* {pr_url}\n\n" f"_All wizards: acknowledge by updating `wizards/shared_context.json` with your status._" ) if telegram_broadcast(telegram_msg): print("Broadcast sent to Telegram (Timmy Time group)") else: print("[warn] Telegram broadcast skipped or failed", file=sys.stderr) print(f"\n✅ Summon {summon_id} created — wizard-to-wizard channel operational.") return 0 if __name__ == "__main__": sys.exit(main())