Build a Companion Dashboard

def post(self, endpoint, payload):
    async def _send():
        try:
            await asyncio.to_thread(
                requests.post,
                f"{DASHBOARD_URL}/{endpoint}",
                json=payload,
                timeout=5,
            )
        except Exception:
            pass  # Dashboard failures must never break the ability
    self.worker.session_tasks.create(_send())

{
    "session_id": "session-1772046000",    // Unique per ability invocation
    "timestamp": 1772046010.5,              // Unix timestamp, for ordering
    "elapsed": "00:10",                     // MM:SS, for human display
    "cycle_id": 1,                          // Monotonic counter, for dedup

    // Everything below is event-specific
    "tone": { "...": "..." },
    "scene": { "...": "..." },
    "speaker_registry": { "...": "..." },
    "running_log": ["..."]                  // Truncate on the way out
}

# =============================================================================
# DASHBOARD CONFIG
# =============================================================================

DASHBOARD_URL = "https://your-repl-name.replit.app/api/<ability>"
SESSION_ID_PREFIX = "session"
HEARTBEAT_INTERVAL = 10     # seconds, when idle
POST_TIMEOUT_FAST = 5       # seconds, for tight-loop posts
POST_TIMEOUT_SLOW = 15      # seconds, for session summaries

def post(self, endpoint: str, payload: dict, timeout: int = POST_TIMEOUT_FAST):
    """Fire-and-forget POST to the dashboard. Never blocks the main loop."""
    async def _send():
        try:
            resp = await asyncio.to_thread(
                requests.post,
                f"{DASHBOARD_URL}/{endpoint}",
                json=payload,
                timeout=timeout,
            )
            if resp.status_code != 200:
                self.log_error(f"[DASH] {endpoint}: {resp.status_code}")
        except Exception as e:
            self.log_error(f"[DASH] {endpoint} failed: {e}")
    self.worker.session_tasks.create(_send())

# At the start of run():
self.session_id = f"{SESSION_ID_PREFIX}-{int(time.time())}"
self.session_start = time.time()
self.post("session_start", {
    "session_id": self.session_id,
    "timestamp": time.time(),
    "ability_version": "1.0.0",
})

# At the end of run() (in the finally block):
self.post("session_end", {
    "session_id": self.session_id,
    "timestamp": time.time(),
    "duration_seconds": time.time() - self.session_start,
    "final_state": self.serialize_state(),
}, timeout=POST_TIMEOUT_SLOW)

async def heartbeat_loop(self):
    """Ping the dashboard every 10s so the frontend knows we're alive."""
    while self.is_running:
        self.post("heartbeat", {
            "session_id": self.session_id,
            "timestamp": time.time(),
            "uptime_seconds": time.time() - self.session_start,
            "cycle_id": self.cycle_id,
        })
        await self.worker.session_tasks.sleep(HEARTBEAT_INTERVAL)

# In run(), launch it alongside your main loop:
self.worker.session_tasks.create(self.heartbeat_loop())

from flask import Flask, request, jsonify, send_from_directory
from flask_cors import CORS
from collections import deque
import time

app = Flask(__name__)
CORS(app)  # Accept from anywhere — the Ability sandbox has no fixed IP

STATE = {
    "session_id": None,
    "last_update": None,
    "last_event": None,
    "tone": {},
    "scene": {},
    "speaker_registry": {},
    "running_log": [],
    "deep_insight": {},
}
HISTORY = deque(maxlen=100)  # Last 100 cycle snapshots

@app.route("/api/<ability>/<event>", methods=["POST"])
def ingest(ability, event):
    payload = request.get_json(silent=True) or {}
    print(f"[{ability}/{event}] {len(str(payload))} bytes from {payload.get('session_id','?')}")

    STATE["last_update"] = time.time()
    STATE["last_event"] = event
    STATE["session_id"] = payload.get("session_id", STATE["session_id"])

    # Merge nested objects; append to lists
    for key, val in payload.items():
        if isinstance(val, dict) and key in STATE and isinstance(STATE[key], dict):
            STATE[key].update(val)
        elif isinstance(val, list) and key in STATE and isinstance(STATE[key], list):
            STATE[key] = val[-50:]  # Keep last 50
        else:
            STATE[key] = val

    if event == "update":
        HISTORY.append(payload)

    return jsonify({"ok": True})

@app.route("/api/state")
def get_state():
    return jsonify(STATE)

@app.route("/api/history")
def get_history():
    return jsonify(list(HISTORY))

@app.route("/")
def index():
    return send_from_directory(".", "index.html")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)