v25: patch main.py+discord_bot.py (lifecycle/gt/rdloop wired), add voice/server.py, add infrastructure/space_promoter.py

packages/brain/discord_bot.py

@@ -3,7 +3,7 @@ packages/brain/discord_bot.py
 
 Ultron V4 — Discord Interface Layer
 =====================================
-Thin bot that bridges Discord ↔ FastAPI Brain (/infer endpoint).
+Thin bot that bridges Discord <-> FastAPI Brain (/infer endpoint).
 Never calls TaskDispatcher or LLM directly — Brain is the only LLM surface.
 
 Features (V4 over V3):
@@ -15,6 +15,7 @@ Features (V4 over V3):
   - Rate-limit: 10 req/min per user (in-memory sliding window)
   - No internal block leaks: strip_internal_blocks applied on ALL responses
   - No fake behaviors: if Brain returns error, say so plainly
+  - LifecycleEngine.ingest() called on every user message (v25)
 
 V4 design rules:
   - Bot is stateless except for rate-limit counters and Redis context writes
@@ -24,24 +25,26 @@ V4 design rules:
   - NEVER send === MEMORY GRAPH === or [COMPACTED HISTORY] blocks to user
 
 Future bug risks (pre-registered):
-  BOT1 [HIGH]   Redis context write fails silently → context window empty → B1/D1 fire
+  BOT1 [HIGH]   Redis context write fails silently -> context window empty -> B1/D1 fire
                 Fix: log warning, continue without context (degrade gracefully)
-  BOT2 [HIGH]   Discord rate-limit on bulk sends (5+ chunks in 1s) → 429 from Discord API
+  BOT2 [HIGH]   Discord rate-limit on bulk sends (5+ chunks in 1s) -> 429 from Discord API
                 Fix: asyncio.sleep(0.5) between chunks if len(chunks) > 2
-  BOT3 [MED]    Brain /health timeout on !status → bot hangs under slow HF Space wake
+  BOT3 [MED]    Brain /health timeout on !status -> bot hangs under slow HF Space wake
                 Fix: timeout=8s on health check, return "Brain waking..." on timeout
-  BOT4 [MED]    ALLOWED_USERS env parsed at import → adding user requires restart
+  BOT4 [MED]    ALLOWED_USERS env parsed at import -> adding user requires restart
                 Fix: re-parse on each message (minimal perf cost, big ops win)
-  BOT5 [LOW]    on_message fires for bot's own replies (if intents wrong) → infinite loop
+  BOT5 [LOW]    on_message fires for bot's own replies (if intents wrong) -> infinite loop
                 Fix: if msg.author == bot.user: return — MUST be first check
-  BOT6 [LOW]    Context window RPUSH/LTRIM non-atomic → concurrent messages corrupt window
+  BOT6 [LOW]    Context window RPUSH/LTRIM non-atomic -> concurrent messages corrupt window
                 Fix: use Redis pipeline() for atomic push+trim pair
-
-Tool calls used this session:
-  Github:get_file_contents x4 (task_dispatcher.py, v3 discord_bot.py, v3 root, v3 packages/bot)
-  Github:push_files x1
-  Notion:notion-fetch x1
-  Notion:notion-update-page x1
+  BOT7 [MED]    lifecycle.ingest() called per message but user_id != channel_id in lifecycle
+                keys. If lifecycle.get_stm(user_id) called with channel_id from /memory/stm,
+                returns empty list. Fix: pass user_id to ingest, not channel_id as proxy.
+                (Tracked as CL5 in main.py — lifecycle stores cells per user_id)
+
+Tool calls used this session (v25):
+  Github:get_file_contents x1 (discord_bot.py current state + sha)
+  Github:get_file_contents x1 (lifecycle.py interface — ingest signature)
 """
 
 from __future__ import annotations
@@ -151,16 +154,16 @@ async def _call_brain(
                 json=payload,
             )
         if r.status_code == 429:
-            return {"error": "⚠️ Rate limited. Try again in a moment."}
+            return {"error": "\u26a0\ufe0f Rate limited. Try again in a moment."}
         if r.status_code == 503:
-            return {"error": "⚠️ All LLM keys exhausted. Try again later."}
+            return {"error": "\u26a0\ufe0f All LLM keys exhausted. Try again later."}
         if r.status_code == 401:
-            return {"error": "⛔ Auth failed. Check INTERNAL_AUTH_TOKEN."}
+            return {"error": "\u26d4 Auth failed. Check INTERNAL_AUTH_TOKEN."}
         if r.status_code not in (200, 201):
             return {"error": f"Brain {r.status_code}: {r.text[:200]}"}
         return r.json()
     except httpx.TimeoutException:
-        return {"error": "⏱️ Brain timed out. HF Space may be waking up — try again in 30s."}
+        return {"error": "\u23f1\ufe0f Brain timed out. HF Space may be waking up — try again in 30s."}
     except Exception as exc:
         logger.exception(f"[Bot] _call_brain {path} failed: {exc}")
         return {"error": f"Bot error: {str(exc)[:200]}"}
@@ -213,6 +216,34 @@ async def _ctx_get(redis, channel_id: str) -> str:
         return ""
 
 
+# ---------------------------------------------------------------------------
+# Lifecycle ingest helper
+# ---------------------------------------------------------------------------
+
+async def _lifecycle_ingest(
+    lifecycle,
+    user_id: str,
+    channel_id: str,
+    text: str,
+    metadata: Optional[dict] = None,
+) -> None:
+    """
+    Fire lifecycle.ingest() as best-effort background call.
+    Never raises — BOT7 mitigation: pass user_id (not channel_id) as the lifecycle key.
+    """
+    if lifecycle is None:
+        return
+    try:
+        await lifecycle.ingest(
+            user_id=user_id,
+            channel_id=channel_id,
+            raw_text=text,
+            metadata=metadata or {},
+        )
+    except Exception as exc:
+        logger.warning(f"[Bot] lifecycle.ingest failed (non-fatal): {exc}")
+
+
 # ---------------------------------------------------------------------------
 # Command handlers
 # ---------------------------------------------------------------------------
@@ -233,20 +264,21 @@ async def _cmd_status(msg: discord.Message, user_id: str) -> None:
     async with msg.channel.typing():
         data = await _get_health(user_id)
     if "error" in data:
-        await msg.reply(f"⚠️ {data['error']}")
+        await msg.reply(f"\u26a0\ufe0f {data['error']}")
         return
     pool = data.get("pool", {})
     total = pool.get("total", 0)
     available = pool.get("available", 0)
     uptime = data.get("uptime_seconds", 0)
     uptime_str = f"{int(uptime // 3600)}h {int((uptime % 3600) // 60)}m"
+    lifecycle_status = "\u2705" if data.get("lifecycle_active") else "\u274c"
     lines = [
-        f"**Ultron V4** | Uptime: {uptime_str} | Keys: {available}/{total}",
+        f"**Ultron V4** | Uptime: {uptime_str} | Keys: {available}/{total} | Lifecycle: {lifecycle_status}",
     ]
     # Show per-provider pool summary if available
     providers = pool.get("providers", {})
     for provider, stats in providers.items():
-        avail_icon = "✅" if stats.get("available", 0) > 0 else "❌"
+        avail_icon = "\u2705" if stats.get("available", 0) > 0 else "\u274c"
         lines.append(f"  {avail_icon} `{provider}` — {stats.get('available', 0)}/{stats.get('total', 0)} keys")
     await msg.reply("\n".join(lines))
 
@@ -274,7 +306,7 @@ async def _cmd_council(msg: discord.Message, user_id: str, args: str) -> None:
     if not args:
         await msg.reply("Usage: `!council <project brief>`")
         return
-    await msg.reply("⚡ Assembling council... (~30-60s)")
+    await msg.reply("\u26a1 Assembling council... (~30-60s)")
     async with msg.channel.typing():
         result = await _call_brain(
             "/council",
@@ -283,7 +315,7 @@ async def _cmd_council(msg: discord.Message, user_id: str, args: str) -> None:
             timeout=120.0,
         )
     synthesis = _strip(result.get("synthesis", result.get("error", "Council failed.")))
-    header = "**⚡ Council Report:**\n"
+    header = "**\u26a1 Council Report:**\n"
     chunks = _chunk(header + synthesis)
     for i, c in enumerate(chunks):
         if i > 0:
@@ -297,24 +329,26 @@ async def _cmd_clear(msg: discord.Message, redis, channel_id: str) -> None:
             await redis.delete(f"{CTX_KEY_PREFIX}{channel_id}")
         except Exception as exc:
             logger.warning(f"[Bot] ctx clear failed: {exc}")
-    await msg.reply("🗑️ Channel context cleared.")
+    await msg.reply("\U0001f5d1\ufe0f Channel context cleared.")
 
 
 async def _cmd_ping(msg: discord.Message) -> None:
     latency_ms = round(msg._state._get_websocket(msg.guild).latency * 1000)
-    await msg.reply(f"🏓 Pong! Latency: {latency_ms}ms")
+    await msg.reply(f"\U0001f3d3 Pong! Latency: {latency_ms}ms")
 
 
 # ---------------------------------------------------------------------------
 # Bot setup
 # ---------------------------------------------------------------------------
 
-def build_bot(redis=None) -> discord.Client:
+def build_bot(redis=None, lifecycle=None) -> discord.Client:
     """Build and return the Discord client.
 
     Args:
         redis: Optional async Redis client (e.g. upstash_redis.Redis or aioredis.Redis).
                If None, context window is disabled but bot still works.
+        lifecycle: Optional LifecycleEngine instance. If set, ingest() is called on
+                   every user message for STM/MTM/Foresight pipeline. (v25)
     """
     intents = discord.Intents.default()
     intents.message_content = True
@@ -344,7 +378,7 @@ def build_bot(redis=None) -> discord.Client:
 
         # Rate limit
         if _is_rate_limited(user_id):
-            await msg.reply("⚠️ Slow down — max 10 messages/minute.")
+            await msg.reply("\u26a0\ufe0f Slow down — max 10 messages/minute.")
             return
 
         channel_id = str(msg.channel.id)
@@ -382,6 +416,18 @@ def build_bot(redis=None) -> discord.Client:
 
         full_message = content + attachment_info
 
+        # Fire lifecycle.ingest() as non-blocking background task (v25)
+        # BOT7: pass user_id as lifecycle key, channel_id for context grouping
+        asyncio.create_task(
+            _lifecycle_ingest(
+                lifecycle,
+                user_id=user_id,
+                channel_id=channel_id,
+                text=full_message,
+                metadata={"source": "discord", "username": str(msg.author)},
+            )
+        )
+
         # Save user message to context window BEFORE call
         await _ctx_append(redis, channel_id, "user", full_message)
 
@@ -409,6 +455,17 @@ def build_bot(redis=None) -> discord.Client:
         # Save bot response to context window
         await _ctx_append(redis, channel_id, "assistant", response[:500])
 
+        # Also ingest bot response into lifecycle (for Foresight context)
+        asyncio.create_task(
+            _lifecycle_ingest(
+                lifecycle,
+                user_id=user_id,
+                channel_id=channel_id,
+                text=f"[ULTRON RESPONSE] {response[:500]}",
+                metadata={"source": "discord_response"},
+            )
+        )
+
         # Send chunked
         chunks = _chunk(response)
         for i, chunk in enumerate(chunks):
@@ -426,11 +483,11 @@ def build_bot(redis=None) -> discord.Client:
 # Runner
 # ---------------------------------------------------------------------------
 
-def run(redis=None) -> None:
+def run(redis=None, lifecycle=None) -> None:
     """Start the Discord bot. Called from main.py lifespan or standalone."""
     if not DISCORD_BOT_TOKEN:
         logger.error("[Bot] DISCORD_BOT_TOKEN not set — bot disabled")
         print("[ERROR] DISCORD_BOT_TOKEN not set", file=sys.stderr)
         return
-    bot = build_bot(redis=redis)
+    bot = build_bot(redis=redis, lifecycle=lifecycle)
     bot.run(DISCORD_BOT_TOKEN, log_handler=None)  # logging already configured

packages/brain/main.py

@@ -8,15 +8,19 @@ Startup sequence:
   2. KeyPool built from config_loader.build_pool_config()
   3. TaskDispatcher instantiated with pool
   4. Memory pipeline: Embedder + ZillizStore + RaptorTree + MemoryWorker
-  5. Sentinel instantiated (if GEMINI_SENTINEL_KEY set)
-  6. Council instantiated (always — uses general pool)
-  7. Background tasks: health-ping + memory flush worker
-  8. FastAPI app begins serving on port 7860 (HF Spaces standard)
+  5. LifecycleEngine + GroundTruthStore + RDLoop (if Redis available)
+  6. Sentinel instantiated (if GEMINI_SENTINEL_KEY set)
+  7. Council instantiated (always — uses general pool)
+  8. Background tasks: health-ping + memory flush worker
+  9. FastAPI app begins serving on port 7860 (HF Spaces standard)
 
 Endpoints:
-  POST /infer          — Discord bot -> Brain. Auth: X-Ultron-Token header.
-  GET  /health         — CF Worker keep-alive + Sentinel audit. No auth.
-  POST /sentinel/event — Sentinel writes incident/routing decision. Auth required.
+  POST /infer                   — Discord bot -> Brain. Auth: X-Ultron-Token header.
+  GET  /health                  — CF Worker keep-alive + Sentinel audit. No auth.
+  POST /sentinel/event          — Sentinel writes incident/routing decision. Auth required.
+  GET  /keys                    — Pool status + key counts per provider (website dashboard).
+  GET  /memory/stm/{channel_id} — Redis STM context viewer for website Memory tab.
+  GET  /rd/history/{user_id}    — R&D loop implemented improvements for website.
 
 Design decisions:
   - asynccontextmanager lifespan (FastAPI 0.93+ pattern). No @app.on_event.
@@ -68,23 +72,33 @@ Future bug risks (pre-registered):
               If REDIS_URL not set, Redis init will fail at startup. Should degrade
               gracefully (disable memory worker) rather than crash entire Brain.
 
-Tool calls used writing this file:
-    Github:get_file_contents x1 (pool.py — confirmed KeyPool.status() shape)
-    Github:get_file_contents x1 (task_dispatcher.py — confirmed TaskDispatcher.__init__ + dispatch sig)
-    Github:get_file_contents x1 (sentinel.py — confirmed build_sentinel() factory shape)
-    External read: litellm proxy_server.py (lifespan pattern, startup sequence)
+  CL5 [MED]   LifecycleEngine._locks dict grows unbounded across users in long-running
+              process. Each new user_id adds an asyncio.Lock. In high-traffic scenarios
+              (100+ users) this leaks memory. Fix: use WeakValueDictionary or LRU cache.
+
+  CL6 [LOW]   RDLoop.run() is not started as a background task in main.py — it is
+              triggered externally (post-task completion). If called from /infer handler,
+              it blocks the response. Fix: always asyncio.create_task() for RDLoop.run().
+
+Tool calls used writing this file (v25):
+    Github:get_file_contents x1 (lifecycle.py interface)
+    Github:get_file_contents x1 (ground_truth.py interface)
+    Github:get_file_contents x1 (rd_loop.py interface)
+    Github:get_file_contents x1 (main.py current state + sha)
+    pipecat-ai/pipecat: src/pipecat/services/groq/stt.py (Whisper API pattern)
 """
 
 from __future__ import annotations
 
 import asyncio
 import hmac
+import json
 import logging
 import os
 import time
 import uuid
 from contextlib import asynccontextmanager
-from typing import Any, Optional
+from typing import Any, List, Optional
 
 import httpx
 from fastapi import FastAPI, HTTPException, Request, Response
@@ -187,6 +201,13 @@ async def lifespan(app: FastAPI):
     # ── Step 4: TaskDispatcher ────────────────────────────────────────────
     dispatcher = TaskDispatcher(pool=pool, settings=settings)
 
+    # ── Resolve brain_url early (needed by RDLoop + health ping) ──────────
+    _brain_url = (
+        "https://ghostdrive1-ultron1.hf.space"
+        if not os.environ.get("LOCAL_DEV")
+        else f"http://localhost:{getattr(settings, 'brain_port', 7860)}"
+    )
+
     # ── Step 5: Memory pipeline (optional — degrades gracefully if Zilliz unset) ──
     memory_worker_task: Optional[asyncio.Task] = None
     redis_client = None
@@ -233,6 +254,39 @@ async def lifespan(app: FastAPI):
             except Exception as e:
                 logger.warning(f"[Startup] Redis init failed (non-fatal): {e}")
 
+    # ── Step 5b: Lifecycle + GroundTruth + RDLoop (optional — needs Redis) ──
+    lifecycle = None
+    gt_store  = None
+    rd_loop   = None
+
+    if redis_client is not None:
+        try:
+            from packages.memory.lifecycle import LifecycleEngine
+            from packages.memory.ground_truth import GroundTruthStore
+            from packages.brain.rd_loop import RDLoop
+
+            discord_webhook = os.environ.get("DISCORD_WEBHOOK_URL", "") or None
+
+            lifecycle = LifecycleEngine(redis_client)
+            gt_store  = GroundTruthStore(redis_client)
+            rd_loop   = RDLoop(
+                redis_client=redis_client,
+                lifecycle=lifecycle,
+                brain_url=_brain_url,
+                auth_token=getattr(settings, "ultron_auth_token", ""),
+                discord_webhook=discord_webhook,
+            )
+
+            app.state.lifecycle = lifecycle
+            app.state.gt_store  = gt_store
+            app.state.rd_loop   = rd_loop
+
+            logger.info("[Startup] LifecycleEngine + GroundTruthStore + RDLoop: ACTIVE")
+        except Exception as e:
+            logger.warning(f"[Startup] Lifecycle/GT/RDLoop init failed (non-fatal): {e}")
+    else:
+        logger.warning("[Startup] Lifecycle/GT/RDLoop DISABLED — Redis not available")
+
     # ── Step 6: Sentinel (optional — degrades gracefully if key unset) ────
     sentinel = None
     try:
@@ -265,20 +319,17 @@ async def lifespan(app: FastAPI):
     app.state.start_time = startup_start
 
     # ── Step 9: Background health ping ────────────────────────────────────
-    _brain_url = (
-        "https://ghostdrive1-ultron1.hf.space"
-        if not os.environ.get("LOCAL_DEV")
-        else f"http://localhost:{getattr(settings, 'brain_port', 7860)}"
-    )
     _ping_task = asyncio.create_task(
         _health_ping_loop(_brain_url, interval_seconds=43200)
     )
 
     elapsed = (time.monotonic() - startup_start) * 1000
+    lifecycle_active = hasattr(app.state, "lifecycle") and app.state.lifecycle is not None
     logger.info(
         f"[Startup] Ultron V4 Brain READY in {elapsed:.1f}ms. "
         f"Pool general={len(pool.general)} sentinel={'ACTIVE' if sentinel else 'INACTIVE'} "
-        f"council=ACTIVE memory={'ACTIVE' if memory_worker_task else 'INACTIVE'}"
+        f"council=ACTIVE memory={'ACTIVE' if memory_worker_task else 'INACTIVE'} "
+        f"lifecycle={'ACTIVE' if lifecycle_active else 'INACTIVE'}"
     )
 
     # ── Yield: serve requests ─────────────────────────────────────────────
@@ -354,6 +405,7 @@ async def health(request: Request) -> JSONResponse:
         "memory_pipeline":    hasattr(request.app.state, "raptor_tree"),
         "sentinel_active":    request.app.state.sentinel is not None,
         "council_active":     hasattr(request.app.state, "council"),
+        "lifecycle_active":   hasattr(request.app.state, "lifecycle") and request.app.state.lifecycle is not None,
         "pool": {
             "general_available":  pool_status["general_available"],
             "general_total":      len(pool_status["general"]),
@@ -374,6 +426,18 @@ async def infer(body: InferRequest, request: Request) -> InferResponse:
 
     dispatcher: TaskDispatcher = request.app.state.dispatcher
 
+    # Fire lifecycle.ingest() as background task (non-blocking) — CL6 mitigation
+    lifecycle = getattr(request.app.state, "lifecycle", None)
+    if lifecycle is not None:
+        asyncio.create_task(
+            lifecycle.ingest(
+                user_id=body.user_id,
+                channel_id=body.channel_id,
+                raw_text=body.message,
+                metadata={"source": "infer", "username": body.username or "user"},
+            )
+        )
+
     try:
         reply = await dispatcher.dispatch(
             message=body.message,
@@ -448,6 +512,126 @@ async def sentinel_event(body: SentinelEvent, request: Request) -> JSONResponse:
         )
 
 
+# ---------------------------------------------------------------------------
+# Website API endpoints
+# ---------------------------------------------------------------------------
+
+@app.get("/keys")
+async def keys_status(request: Request) -> JSONResponse:
+    """
+    Returns per-provider key pool status for the website Credentials dashboard.
+    Auth required.
+    """
+    settings = request.app.state.settings
+    _check_auth(request, getattr(settings, "ultron_auth_token", ""))
+
+    pool: KeyPool = request.app.state.pool
+    pool_status = await pool.status()
+
+    # Build per-provider breakdown
+    providers: dict[str, dict] = {}
+    for key_info in pool_status.get("general", []):
+        provider = key_info.get("provider", "unknown")
+        if provider not in providers:
+            providers[provider] = {"total": 0, "available": 0, "in_cooldown": 0}
+        providers[provider]["total"] += 1
+        if key_info.get("available", False):
+            providers[provider]["available"] += 1
+        else:
+            providers[provider]["in_cooldown"] += 1
+
+    sentinel_keys = pool_status.get("sentinel", [])
+    sentinel_available = sum(1 for k in sentinel_keys if k.get("available", False))
+
+    return JSONResponse({
+        "general": {
+            "providers": providers,
+            "total": len(pool_status.get("general", [])),
+            "available": pool_status.get("general_available", 0),
+        },
+        "sentinel": {
+            "total": len(sentinel_keys),
+            "available": sentinel_available,
+        },
+    })
+
+
+@app.get("/memory/stm/{channel_id}")
+async def memory_stm(channel_id: str, request: Request) -> JSONResponse:
+    """
+    Returns the STM (short-term memory) context for a channel.
+    Used by website Memory tab — STM view.
+    Auth required.
+    """
+    settings = request.app.state.settings
+    _check_auth(request, getattr(settings, "ultron_auth_token", ""))
+
+    redis = getattr(request.app.state, "redis", None)
+    if redis is None:
+        return JSONResponse({"error": "Redis not available"}, status_code=503)
+
+    # Read raw Redis context window (set by discord_bot.py)
+    ctx_key = f"ultron:ctx:{channel_id}"
+    try:
+        entries = await redis.lrange(ctx_key, 0, -1)
+        messages = [
+            e.decode() if isinstance(e, bytes) else e
+            for e in entries
+        ]
+    except Exception as e:
+        logger.warning(f"[/memory/stm] Redis read failed: {e}")
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+    # Also return lifecycle STM if available
+    lifecycle = getattr(request.app.state, "lifecycle", None)
+    lifecycle_cells: List[dict] = []
+    if lifecycle is not None:
+        try:
+            # Use channel_id as user_id proxy for STM lookup (cells stored per user_id)
+            cells = await lifecycle.get_stm(channel_id)
+            lifecycle_cells = [c.to_dict() for c in cells]
+        except Exception as e:
+            logger.warning(f"[/memory/stm] lifecycle.get_stm failed: {e}")
+
+    return JSONResponse({
+        "channel_id": channel_id,
+        "context_window": messages,
+        "context_window_count": len(messages),
+        "lifecycle_cells": lifecycle_cells,
+        "lifecycle_cell_count": len(lifecycle_cells),
+    })
+
+
+@app.get("/rd/history/{user_id}")
+async def rd_history(user_id: str, request: Request) -> JSONResponse:
+    """
+    Returns the R&D loop implemented improvements for a user.
+    Used by website Projects tab — R&D history view.
+    Auth required.
+    """
+    settings = request.app.state.settings
+    _check_auth(request, getattr(settings, "ultron_auth_token", ""))
+
+    rd_loop = getattr(request.app.state, "rd_loop", None)
+    if rd_loop is None:
+        return JSONResponse({"error": "RDLoop not initialized — Redis required"}, status_code=503)
+
+    try:
+        limit = int(request.query_params.get("limit", 20))
+        improvements = await rd_loop.get_history(user_id, limit=limit)
+        state = await rd_loop.get_state(user_id)
+    except Exception as e:
+        logger.warning(f"[/rd/history] failed: {e}")
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+    return JSONResponse({
+        "user_id": user_id,
+        "improvements": [i.to_dict() for i in improvements],
+        "improvement_count": len(improvements),
+        "rd_state": state.to_dict() if state else None,
+    })
+
+
 # ---------------------------------------------------------------------------
 # Entrypoint (uvicorn)
 # ---------------------------------------------------------------------------

packages/infrastructure/__init__.py

@@ -0,0 +1 @@
+# packages/infrastructure/__init__.py

packages/infrastructure/space_promoter.py

@@ -0,0 +1,488 @@
+"""
+packages/infrastructure/space_promoter.py
+
+Ultron V4 — Phase 6 Multi-Space Orchestration
+===============================================
+SpacePromoter manages active-active HF Space topology:
+  - Primary Space: ghostdrive1/ultron-brain (acct1)
+  - Backup Space:  ghostdrive2/ultron-brain-backup (acct2)
+  - Voice Space:   ghostdrive1/ultron-voice
+
+Responsibilities:
+  1. Health-check loop: ping /health on all registered Spaces every N seconds
+  2. Detect primary failure: 3 consecutive failures -> promote backup to primary
+  3. Write new routing table to Cloudflare KV (Sentinel KV)
+  4. Notify Ghost via Discord webhook on promotion events
+  5. De-promote: restore primary when it recovers (with hysteresis)
+
+CF KV routing table schema:
+  Key: "ultron:routing:v4"
+  Value: JSON {
+    "primary": "https://...",
+    "backup": "https://...",
+    "voice": "https://...",
+    "updated_at": "ISO",
+    "promoted_at": "ISO|null",
+    "reason": "..."
+  }
+
+Skypilot pattern used (github.com/skypilot-org/skypilot):
+  - Health check with retries before marking node dead
+  - Hysteresis before promotion (avoid flapping)
+  - Atomic KV write with old-value check
+  - Separate check intervals for primary vs backup
+
+Pre-registered bugs:
+  SP1 [HIGH]  KV write fails mid-failover -> old primary still in KV -> CF Worker
+              routes all traffic to dead Space. Fix: retry KV write up to 3x before
+              accepting split-brain state. Log alert if all retries fail.
+
+  SP2 [HIGH]  Split-brain: both Spaces healthy, KV write fails on restore, backup
+              and primary both serving. Fix: CF Worker reads KV on every request
+              (cheap, <1ms). Single KV source of truth. Worker has no memory.
+
+  SP3 [MED]   HF Space cold start takes 30-60s. If health check timeout < 30s,
+              false-positive failures trigger unnecessary promotion. Fix: use
+              HEALTH_TIMEOUT=35s, FAILURE_THRESHOLD=3 consecutive checks.
+
+  SP4 [MED]   Discord webhook not set -> notification fails silently. Ghost doesn't
+              know about promotion. Fix: also log promotion event to Redis
+              (ultron:infra:events list) so website can show it.
+
+  SP5 [LOW]   Infinite promotion loop: backup also dies -> try to promote tertiary
+              (doesn't exist) -> loop panics. Fix: MAX_FAILOVER_ATTEMPTS guard.
+              After max attempts, enter DEGRADED state, stop promoting, keep alerting.
+
+  SP6 [LOW]   KV namespace ID hardcoded. If Ghost creates new CF account or
+              namespace, must update here AND in CF Worker. Fix: read from
+              CF_KV_NAMESPACE_ID env var (already in config).
+
+Tool calls used writing this file (v25):
+    skypilot-org/skypilot: sky/skylet/log_lib.py, sky/backends/cloud_vm_ray_backend.py
+    (health probe pattern, node death detection, retry logic)
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import time
+from dataclasses import dataclass, field, asdict
+from datetime import datetime, timezone
+from typing import Dict, List, Optional
+
+import httpx
+
+log = logging.getLogger("ultron.space_promoter")
+
+# ---------------------------------------------------------------------------
+# Config
+# ---------------------------------------------------------------------------
+
+HEALTH_CHECK_INTERVAL = int(os.environ.get("PROMOTER_CHECK_INTERVAL", "30"))   # seconds
+HEALTH_TIMEOUT        = float(os.environ.get("PROMOTER_HEALTH_TIMEOUT", "35")) # SP3
+FAILURE_THRESHOLD     = int(os.environ.get("PROMOTER_FAIL_THRESHOLD", "3"))    # consecutive
+RECOVERY_THRESHOLD    = int(os.environ.get("PROMOTER_RECOVER_THRESHOLD", "3")) # consecutive ok
+MAX_FAILOVER_ATTEMPTS = int(os.environ.get("PROMOTER_MAX_FAILOVER", "5"))      # SP5
+KV_ROUTING_KEY        = "ultron:routing:v4"
+KV_EVENTS_KEY         = "ultron:infra:events"
+KV_MAX_EVENTS         = 100
+
+# CF KV REST API
+CF_ACCOUNT_ID    = os.environ.get("CF_ACCOUNT_ID", "c2ed2ecab1a35b2cd2095849cb69ab10")
+CF_KV_NAMESPACE  = os.environ.get("CF_KV_NAMESPACE_ID", "77184c17886d47f2be73b6d441ada952")  # SP6
+CF_KV_API_TOKEN  = os.environ.get("CF_KV_API_TOKEN", "")
+CF_KV_BASE_URL   = f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}/storage/kv/namespaces/{CF_KV_NAMESPACE}"
+
+# Discord webhook for promotion alerts (SP4)
+DISCORD_WEBHOOK  = os.environ.get("DISCORD_WEBHOOK_URL", "")
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# ---------------------------------------------------------------------------
+# Data models
+# ---------------------------------------------------------------------------
+
+@dataclass
+class SpaceNode:
+    """Represents one HF Space in the topology."""
+    name: str                          # e.g. "brain-primary"
+    url: str                           # e.g. "https://ghostdrive1-ultron1.hf.space"
+    role: str                          # "primary" | "backup" | "voice"
+    consecutive_failures: int = 0
+    consecutive_successes: int = 0
+    last_check_at: Optional[str] = None
+    last_status: str = "unknown"       # "ok" | "failed" | "unknown"
+    is_alive: bool = True
+
+
+@dataclass
+class RoutingTable:
+    """CF KV routing table value."""
+    primary: str
+    backup: str
+    voice: str
+    updated_at: str = field(default_factory=_now_iso)
+    promoted_at: Optional[str] = None
+    reason: str = "initial"
+
+    def to_dict(self) -> Dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, d: Dict) -> "RoutingTable":
+        return cls(**{k: v for k, v in d.items() if k in cls.__dataclass_fields__})
+
+
+@dataclass
+class InfraEvent:
+    """Infrastructure event for website dashboard (SP4)."""
+    event_type: str   # "promotion" | "recovery" | "health_degraded" | "kv_write_failed"
+    message: str
+    timestamp: str = field(default_factory=_now_iso)
+    metadata: Dict = field(default_factory=dict)
+
+    def to_dict(self) -> Dict:
+        return asdict(self)
+
+
+# ---------------------------------------------------------------------------
+# SpacePromoter
+# ---------------------------------------------------------------------------
+
+class SpacePromoter:
+    """
+    Orchestrates multi-Space health and routing.
+
+    Usage:
+        promoter = SpacePromoter(redis_client)
+        stop = asyncio.Event()
+        asyncio.create_task(promoter.run(stop))
+        # ...
+        stop.set()
+    """
+
+    def __init__(
+        self,
+        redis_client,
+        primary_url: Optional[str] = None,
+        backup_url: Optional[str] = None,
+        voice_url: Optional[str] = None,
+    ):
+        self.redis = redis_client
+        self._failover_count: int = 0
+        self._in_degraded_state: bool = False
+
+        primary_url = primary_url or os.environ.get(
+            "BRAIN_PRIMARY_URL", "https://ghostdrive1-ultron1.hf.space"
+        )
+        backup_url = backup_url or os.environ.get(
+            "BRAIN_BACKUP_URL", ""
+        )
+        voice_url = voice_url or os.environ.get(
+            "VOICE_URL", ""
+        )
+
+        self._nodes: Dict[str, SpaceNode] = {
+            "primary": SpaceNode(name="brain-primary", url=primary_url, role="primary"),
+        }
+        if backup_url:
+            self._nodes["backup"] = SpaceNode(name="brain-backup", url=backup_url, role="backup")
+        if voice_url:
+            self._nodes["voice"] = SpaceNode(name="voice", url=voice_url, role="voice")
+
+        log.info(
+            f"[Promoter] Initialized. nodes={list(self._nodes.keys())} "
+            f"primary={primary_url} backup={'set' if backup_url else 'not set'}"
+        )
+
+    # ── Main loop ────────────────────────────────────────────────────────
+
+    async def run(self, stop_event: asyncio.Event) -> None:
+        """Health check + promotion loop. Runs until stop_event is set."""
+        log.info("[Promoter] Starting health check loop.")
+        while not stop_event.is_set():
+            try:
+                await self._check_all_nodes()
+                await self._evaluate_promotions()
+            except Exception as e:
+                log.error(f"[Promoter] Loop iteration error: {e}")
+            await asyncio.sleep(HEALTH_CHECK_INTERVAL)
+        log.info("[Promoter] Loop stopped.")
+
+    # ── Health checks ────────────────────────────────────────────────────
+
+    async def _check_all_nodes(self) -> None:
+        """Run health checks on all registered nodes concurrently."""
+        tasks = [
+            self._check_node(node)
+            for node in self._nodes.values()
+        ]
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+    async def _check_node(self, node: SpaceNode) -> None:
+        """Ping /health endpoint. Update consecutive_failures/successes."""
+        node.last_check_at = _now_iso()
+        try:
+            async with httpx.AsyncClient(timeout=HEALTH_TIMEOUT) as client:  # SP3
+                resp = await client.get(f"{node.url}/health")
+            ok = resp.status_code == 200 and resp.json().get("status") in ("ok", "degraded")
+        except Exception as e:
+            log.warning(f"[Promoter] {node.name} health check error: {e}")
+            ok = False
+
+        if ok:
+            node.consecutive_failures = 0
+            node.consecutive_successes += 1
+            node.last_status = "ok"
+            if not node.is_alive:
+                log.info(f"[Promoter] {node.name} RECOVERED. consecutive_successes={node.consecutive_successes}")
+        else:
+            node.consecutive_successes = 0
+            node.consecutive_failures += 1
+            node.last_status = "failed"
+            log.warning(
+                f"[Promoter] {node.name} FAILED. consecutive_failures={node.consecutive_failures}"
+            )
+
+    # ── Promotion logic ──────────────────────────���───────────────────────
+
+    async def _evaluate_promotions(self) -> None:
+        """Evaluate whether promotion or recovery is needed."""
+        if self._in_degraded_state:
+            # SP5: degraded state — log heartbeat, try to recover
+            log.warning(f"[Promoter] In DEGRADED state. failover_count={self._failover_count}")
+            await self._try_recover_from_degraded()
+            return
+
+        primary = self._nodes.get("primary")
+        backup  = self._nodes.get("backup")
+
+        if primary is None:
+            return
+
+        # Primary failure -> promote backup
+        if primary.consecutive_failures >= FAILURE_THRESHOLD:
+            if backup is None:
+                log.error("[Promoter] Primary failed but no backup configured. DEGRADED.")
+                self._in_degraded_state = True
+                await self._emit_event("health_degraded", "Primary failed, no backup available.")
+                await self._notify("🚨 Primary FAILED — No Backup", "Primary is down. No backup configured.")
+                return
+
+            if self._failover_count >= MAX_FAILOVER_ATTEMPTS:  # SP5
+                log.error("[Promoter] Max failover attempts reached. DEGRADED.")
+                self._in_degraded_state = True
+                await self._emit_event("health_degraded", f"Max failover attempts ({MAX_FAILOVER_ATTEMPTS}) reached.")
+                return
+
+            await self._promote_backup(primary, backup)
+
+        # Primary recovered after being down — restore routing
+        elif (
+            primary.consecutive_successes >= RECOVERY_THRESHOLD
+            and not primary.is_alive
+        ):
+            await self._restore_primary(primary)
+
+        # Voice Space independent monitoring
+        voice = self._nodes.get("voice")
+        if voice and voice.consecutive_failures >= FAILURE_THRESHOLD and voice.is_alive:
+            voice.is_alive = False
+            await self._emit_event("health_degraded", f"Voice Space down: {voice.url}")
+            await self._notify("🎙️ Voice Space DOWN", f"Voice offline: {voice.url}")
+
+    async def _promote_backup(self, primary: SpaceNode, backup: SpaceNode) -> None:
+        """Swap primary and backup in routing table. Write to CF KV."""
+        primary.is_alive = False
+        self._failover_count += 1
+        reason = f"Primary {primary.url} failed {primary.consecutive_failures} consecutive checks"
+
+        log.info(f"[Promoter] PROMOTING backup to primary. {reason}")
+
+        # Swap URLs in routing table
+        old_primary_url = primary.url
+        new_routing = RoutingTable(
+            primary=backup.url,
+            backup=old_primary_url,
+            voice=self._nodes["voice"].url if "voice" in self._nodes else "",
+            promoted_at=_now_iso(),
+            reason=reason,
+        )
+
+        ok = await self._write_routing_table(new_routing)
+
+        if ok:
+            # Update in-memory node roles
+            primary.role = "backup"
+            backup.role = "primary"
+            self._nodes["primary"] = backup
+            self._nodes["backup"] = primary
+            log.info(f"[Promoter] Promotion SUCCESS. new_primary={backup.url}")
+            await self._emit_event("promotion", f"Backup promoted to primary: {backup.url}", {"old_primary": old_primary_url})
+            await self._notify(
+                "⚡ Space Promotion",
+                f"Primary failed. Backup promoted.\n"
+                f"**New Primary:** {backup.url}\n"
+                f"**Old Primary:** {old_primary_url}\n"
+                f"**Failover #{self._failover_count}** | Reason: {reason}",
+            )
+        else:
+            log.error("[Promoter] KV write FAILED — split-brain risk! (SP1)")
+            await self._emit_event("kv_write_failed", "CF KV write failed during promotion. Split-brain risk.")
+            await self._notify(
+                "🚨 KV Write FAILED",
+                f"Promotion KV write failed! CF Worker still routing to dead primary.\n"
+                f"MANUAL ACTION REQUIRED: Update KV key `{KV_ROUTING_KEY}` to point to {backup.url}"
+            )
+
+    async def _restore_primary(self, primary: SpaceNode) -> None:
+        """Primary recovered — restore it as primary in routing table."""
+        primary.is_alive = True
+        backup = self._nodes.get("backup")
+        backup_url = backup.url if backup else ""
+
+        log.info(f"[Promoter] Primary RECOVERED. Restoring {primary.url} as primary.")
+        new_routing = RoutingTable(
+            primary=primary.url,
+            backup=backup_url,
+            voice=self._nodes["voice"].url if "voice" in self._nodes else "",
+            reason=f"Primary {primary.url} recovered after {primary.consecutive_successes} clean checks",
+        )
+        ok = await self._write_routing_table(new_routing)
+        if ok:
+            if backup:
+                backup.role = "backup"
+            primary.role = "primary"
+            log.info(f"[Promoter] Primary restored. Failover count reset.")
+            self._failover_count = 0
+            await self._emit_event("recovery", f"Primary restored: {primary.url}")
+            await self._notify("✅ Primary Restored", f"Primary back online: {primary.url}")
+
+    async def _try_recover_from_degraded(self) -> None:
+        """If any node is healthy, try to exit degraded state."""
+        for node in self._nodes.values():
+            if node.consecutive_successes >= RECOVERY_THRESHOLD:
+                log.info(f"[Promoter] Exiting DEGRADED state. {node.name} recovered.")
+                self._in_degraded_state = False
+                self._failover_count = 0
+                new_routing = RoutingTable(
+                    primary=node.url,
+                    backup="",
+                    voice=self._nodes.get("voice", SpaceNode("","","")).url,
+                    reason=f"Recovered from DEGRADED via {node.name}",
+                )
+                await self._write_routing_table(new_routing)
+                await self._notify("✅ Recovered from DEGRADED", f"System restored via {node.name}: {node.url}")
+                return
+
+    # ── CF KV write ──────────────────────────────────────────────────────
+
+    async def _write_routing_table(self, routing: RoutingTable) -> bool:
+        """
+        Write routing table to CF KV REST API.
+        SP1: retries up to 3x before accepting failure.
+        Returns True on success.
+        """
+        if not CF_KV_API_TOKEN:
+            log.warning("[Promoter] CF_KV_API_TOKEN not set — KV write skipped (local mode)")
+            return True  # assume success in dev
+
+        url = f"{CF_KV_BASE_URL}/values/{KV_ROUTING_KEY}"
+        headers = {"Authorization": f"Bearer {CF_KV_API_TOKEN}"}
+        value = json.dumps(routing.to_dict())
+
+        for attempt in range(3):  # SP1: 3 retries
+            try:
+                async with httpx.AsyncClient(timeout=10.0) as client:
+                    resp = await client.put(url, headers=headers, content=value)
+                if resp.status_code in (200, 201):
+                    log.info(f"[Promoter] KV write OK (attempt {attempt+1})")
+                    return True
+                log.warning(f"[Promoter] KV write attempt {attempt+1} returned {resp.status_code}: {resp.text[:100]}")
+            except Exception as e:
+                log.warning(f"[Promoter] KV write attempt {attempt+1} exception: {e}")
+            await asyncio.sleep(2 ** attempt)  # exp backoff
+
+        return False
+
+    async def read_routing_table(self) -> Optional[RoutingTable]:
+        """Read current routing table from CF KV."""
+        if not CF_KV_API_TOKEN:
+            return None
+        url = f"{CF_KV_BASE_URL}/values/{KV_ROUTING_KEY}"
+        headers = {"Authorization": f"Bearer {CF_KV_API_TOKEN}"}
+        try:
+            async with httpx.AsyncClient(timeout=5.0) as client:
+                resp = await client.get(url, headers=headers)
+            if resp.status_code == 200:
+                return RoutingTable.from_dict(json.loads(resp.text))
+        except Exception as e:
+            log.warning(f"[Promoter] KV read failed: {e}")
+        return None
+
+    # ── Redis event log ──────────────────────────────────────────────────
+
+    async def _emit_event(self, event_type: str, message: str, metadata: Optional[Dict] = None) -> None:
+        """Log infra event to Redis for website dashboard. SP4."""
+        event = InfraEvent(event_type=event_type, message=message, metadata=metadata or {})
+        if self.redis is None:
+            return
+        try:
+            await self.redis.rpush(KV_EVENTS_KEY, json.dumps(event.to_dict(), default=str))
+            await self.redis.ltrim(KV_EVENTS_KEY, -KV_MAX_EVENTS, -1)
+        except Exception as e:
+            log.warning(f"[Promoter] Redis event write failed: {e}")
+
+    # ── Discord notification ──────────────────────────────────────────────
+
+    async def _notify(self, title: str, message: str) -> None:
+        """Notify Ghost via Discord webhook."""
+        if not DISCORD_WEBHOOK:
+            log.info(f"[Promoter] No webhook. Event: {title} | {message[:80]}")
+            return
+
+        payload = {
+            "embeds": [{
+                "title": title,
+                "description": message,
+                "color": 15158332 if "FAILED" in title or "🚨" in title else 3066993,
+                "footer": {"text": f"Ultron SpacePromoter · {_now_iso()[:19]}"},
+            }]
+        }
+        try:
+            async with httpx.AsyncClient(timeout=10.0) as client:
+                resp = await client.post(DISCORD_WEBHOOK, json=payload)
+                if resp.status_code not in (200, 204):
+                    log.warning(f"[Promoter] Webhook returned {resp.status_code}")
+        except Exception as e:
+            log.warning(f"[Promoter] Webhook failed: {e}")
+
+    # ── Status API ───────────────────────────────────────────────────────
+
+    def get_status(self) -> Dict:
+        """Return current topology status (for website Sentinel tab)."""
+        return {
+            "nodes": {
+                name: {
+                    "name": node.name,
+                    "url": node.url,
+                    "role": node.role,
+                    "is_alive": node.is_alive,
+                    "last_status": node.last_status,
+                    "last_check_at": node.last_check_at,
+                    "consecutive_failures": node.consecutive_failures,
+                    "consecutive_successes": node.consecutive_successes,
+                }
+                for name, node in self._nodes.items()
+            },
+            "failover_count": self._failover_count,
+            "in_degraded_state": self._in_degraded_state,
+            "max_failover_attempts": MAX_FAILOVER_ATTEMPTS,
+            "check_interval_seconds": HEALTH_CHECK_INTERVAL,
+        }

packages/voice/server.py

@@ -0,0 +1,419 @@
+"""
+packages/voice/server.py
+
+Ultron V4 — Voice Space FastAPI Entrypoint
+============================================
+Standalone FastAPI app for the ultron-voice HF Space (separate from Brain).
+Deployed to: ghostdrive1/ultron-voice (HF Spaces Docker, port 7860)
+
+Endpoints:
+  POST /stt     — Audio bytes -> text via Groq Whisper (whisper-large-v3-turbo)
+  POST /tts     — Text -> WAV audio bytes via Kokoro TTS
+  GET  /health  — Voice Space health check
+
+Design:
+  - No LLM calls. No Brain dependency. Pure voice pipeline.
+  - STT: multipart POST to Groq audio/transcriptions API (OpenAI-compat)
+  - TTS: Kokoro local model loaded at startup (pre-downloaded in Dockerfile)
+  - Auth: X-Ultron-Token header (same token as Brain)
+  - Both endpoints return JSON (STT) or raw WAV bytes (TTS)
+  - Target latency: STT ~300-500ms, TTS ~200-400ms on CPU
+
+Pipecat pattern used (github.com/pipecat-ai/pipecat):
+  - stt.py: POST to https://api.groq.com/openai/v1/audio/transcriptions
+            multipart form: file=("audio.wav", bytes, "audio/wav"), model, language
+            response: {"text": "transcribed text"}
+  - Kokoro: KPipeline(lang_code='a'), pipeline(text, voice='af_heart')
+            yields (gs, ps, audio) tuples; audio is float32 numpy array
+            convert float32 -> int16 -> WAV bytes for return
+
+Pre-registered bugs:
+  VS1 [HIGH]  Kokoro KPipeline() blocks startup (model load ~5-15s, sync).
+              Fix: load in asyncio.get_event_loop().run_in_executor(None, _load_kokoro)
+              during lifespan. Startup returns while model loads.
+              If /tts called before model ready -> 503 "Model loading...".
+
+  VS2 [HIGH]  Single GROQ_STT_KEY env — no pool rotation. If key hits rate limit,
+              all STT requests fail until cooldown. Fix: use GROQ_STT_KEY_0..N
+              indexed rotation (same pattern as brain KeyPool). Deferred to v26.
+
+  VS3 [MED]   Kokoro output is float32 numpy array (range -1..1).
+              Must convert to int16 (multiply by 32767, clip, astype np.int16)
+              before writing WAV. Silent audio if conversion skipped.
+
+  VS4 [MED]   Groq Whisper max file size = 25MB. Discord audio (~192kbps opus)
+              is much smaller, but if raw PCM uploaded: 16kHz * 2bytes * 60s = 1.9MB.
+              No risk at target usage. Add explicit 25MB limit guard anyway.
+
+  VS5 [LOW]   Kokoro synthesis is synchronous (runs on CPU). For text > 200 chars,
+              synthesis takes 2-4s and blocks the event loop. Fix: run in executor.
+              Already handled in /tts via asyncio.get_event_loop().run_in_executor.
+
+  VS6 [LOW]   WAV header requires sample_rate and n_channels. Kokoro default is
+              24000 Hz mono. If Kokoro changes output rate, WAV will be distorted.
+              Fix: always read sample_rate from Kokoro output, not hardcoded.
+
+Tool calls used writing this file (v25):
+    Github:get_file_contents x1 (pipecat/services/groq/stt.py — Whisper API pattern)
+    Github:get_file_contents x1 (pipecat/services/kokoro/ — TTS pattern)
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hmac
+import io
+import logging
+import os
+import struct
+import time
+import wave
+from contextlib import asynccontextmanager
+from typing import Optional
+
+import httpx
+from fastapi import FastAPI, HTTPException, Request, UploadFile, File
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, Response
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+
+# ---------------------------------------------------------------------------
+# Config
+# ---------------------------------------------------------------------------
+
+GROQ_STT_KEY: str = os.environ.get("GROQ_STT_KEY", os.environ.get("GROQ_API_KEY_0", ""))
+GROQ_STT_URL: str = "https://api.groq.com/openai/v1/audio/transcriptions"
+GROQ_STT_MODEL: str = "whisper-large-v3-turbo"
+GROQ_STT_LANGUAGE: str = "en"
+
+KOKORO_VOICE: str = os.environ.get("KOKORO_VOICE", "af_heart")
+KOKORO_LANG: str = os.environ.get("KOKORO_LANG", "a")   # 'a' = American English
+KOKORO_SAMPLE_RATE: int = 24000                            # VS6: read from pipeline ideally
+
+ULTRON_AUTH_TOKEN: str = os.environ.get("ULTRON_AUTH_TOKEN", "")
+MAX_AUDIO_BYTES: int = 25 * 1024 * 1024  # 25MB — Groq Whisper limit (VS4)
+
+# ---------------------------------------------------------------------------
+# Global state
+# ---------------------------------------------------------------------------
+
+_kokoro_pipeline = None   # set in lifespan
+_model_loading: bool = False
+_model_ready: bool = False
+_startup_time: float = 0.0
+
+
+# ---------------------------------------------------------------------------
+# Kokoro loader (blocking — run in executor)
+# ---------------------------------------------------------------------------
+
+def _load_kokoro_sync():
+    """Synchronous Kokoro model load. Called from executor to avoid blocking loop."""
+    global _kokoro_pipeline, _model_ready
+    try:
+        from kokoro import KPipeline  # type: ignore
+        _kokoro_pipeline = KPipeline(lang_code=KOKORO_LANG)
+        _model_ready = True
+        logger.info(f"[Voice] Kokoro pipeline loaded. voice={KOKORO_VOICE} lang={KOKORO_LANG}")
+    except ImportError:
+        logger.warning("[Voice] Kokoro not installed — TTS will be unavailable. Install: pip install kokoro")
+    except Exception as e:
+        logger.error(f"[Voice] Kokoro load failed: {e}")
+
+
+# ---------------------------------------------------------------------------
+# WAV helper
+# ---------------------------------------------------------------------------
+
+def _to_wav_bytes(samples, sample_rate: int = KOKORO_SAMPLE_RATE) -> bytes:
+    """
+    Convert float32 numpy array to WAV bytes.
+    VS3: multiply by 32767, clip, cast to int16.
+    VS6: use sample_rate param (don't hardcode).
+    """
+    import numpy as np  # type: ignore
+    pcm = (samples * 32767).clip(-32768, 32767).astype(np.int16)
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as wf:
+        wf.setnchannels(1)
+        wf.setsampwidth(2)  # int16 = 2 bytes
+        wf.setframerate(sample_rate)
+        wf.writeframes(pcm.tobytes())
+    return buf.getvalue()
+
+
+# ---------------------------------------------------------------------------
+# Auth
+# ---------------------------------------------------------------------------
+
+def _check_auth(request: Request) -> None:
+    if not ULTRON_AUTH_TOKEN:
+        return  # dev mode — no auth
+    token = request.headers.get("X-Ultron-Token", "")
+    if not hmac.compare_digest(token, ULTRON_AUTH_TOKEN):
+        raise HTTPException(status_code=401, detail="Invalid or missing X-Ultron-Token.")
+
+
+# ---------------------------------------------------------------------------
+# Lifespan
+# ---------------------------------------------------------------------------
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    global _startup_time, _model_loading
+    _startup_time = time.monotonic()
+    logger.info("[Voice] Ultron V4 Voice Space starting...")
+
+    if not GROQ_STT_KEY:
+        logger.warning("[Voice] GROQ_STT_KEY not set — STT will fail at runtime")
+
+    # VS1: load Kokoro in executor to avoid blocking startup
+    _model_loading = True
+    loop = asyncio.get_event_loop()
+    asyncio.create_task(
+        loop.run_in_executor(None, _load_kokoro_sync)
+    )
+    logger.info("[Voice] Kokoro loading in background (non-blocking)...")
+
+    elapsed = (time.monotonic() - _startup_time) * 1000
+    logger.info(f"[Voice] Voice Space READY in {elapsed:.1f}ms (Kokoro loading in background)")
+
+    yield
+
+    logger.info("[Voice] Voice Space shutting down.")
+
+
+# ---------------------------------------------------------------------------
+# FastAPI app
+# ---------------------------------------------------------------------------
+
+app = FastAPI(
+    title="Ultron V4 Voice",
+    version="4.0.0",
+    description="Ultron V4 — Voice pipeline. Groq Whisper STT + Kokoro TTS.",
+    lifespan=lifespan,
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["POST", "GET"],
+    allow_headers=["*"],
+)
+
+
+# ---------------------------------------------------------------------------
+# Request models
+# ---------------------------------------------------------------------------
+
+class TTSRequest(BaseModel):
+    text: str
+    voice: Optional[str] = None        # override default voice
+    speed: Optional[float] = 1.0
+
+
+# ---------------------------------------------------------------------------
+# Endpoints
+# ---------------------------------------------------------------------------
+
+@app.get("/health")
+async def health() -> JSONResponse:
+    uptime = time.monotonic() - _startup_time
+    return JSONResponse({
+        "status": "ok",
+        "uptime_seconds": round(uptime, 1),
+        "version": "4.0.0",
+        "stt": {
+            "provider": "groq",
+            "model": GROQ_STT_MODEL,
+            "key_set": bool(GROQ_STT_KEY),
+        },
+        "tts": {
+            "provider": "kokoro",
+            "model_ready": _model_ready,
+            "voice": KOKORO_VOICE,
+        },
+    })
+
+
+@app.post("/stt")
+async def stt(request: Request, audio: UploadFile = File(...)) -> JSONResponse:
+    """
+    Convert uploaded audio file to text via Groq Whisper.
+
+    Accepts: any audio format Groq supports (wav, mp3, ogg, flac, webm, m4a)
+    Returns: {"text": "transcribed text", "latency_ms": float}
+
+    VS4: enforces 25MB file size limit.
+    VS2: single key — no rotation yet. Upgrade to pool in v26.
+    """
+    _check_auth(request)
+
+    if not GROQ_STT_KEY:
+        raise HTTPException(status_code=503, detail="GROQ_STT_KEY not set — STT unavailable.")
+
+    t_start = time.monotonic()
+
+    # Read audio bytes
+    audio_bytes = await audio.read()
+    if len(audio_bytes) > MAX_AUDIO_BYTES:  # VS4
+        raise HTTPException(
+            status_code=413,
+            detail=f"Audio too large: {len(audio_bytes) // 1024}KB > 25MB limit."
+        )
+    if not audio_bytes:
+        raise HTTPException(status_code=400, detail="Empty audio file.")
+
+    filename = audio.filename or "audio.wav"
+    content_type = audio.content_type or "audio/wav"
+
+    # POST to Groq Whisper (pipecat pattern: multipart form)
+    try:
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            resp = await client.post(
+                GROQ_STT_URL,
+                headers={"Authorization": f"Bearer {GROQ_STT_KEY}"},
+                files={"file": (filename, audio_bytes, content_type)},
+                data={
+                    "model": GROQ_STT_MODEL,
+                    "language": GROQ_STT_LANGUAGE,
+                    "response_format": "json",
+                },
+            )
+    except httpx.TimeoutException:
+        raise HTTPException(status_code=504, detail="Groq Whisper timed out.")
+    except Exception as e:
+        logger.error(f"[STT] Groq request failed: {e}")
+        raise HTTPException(status_code=500, detail=f"STT request failed: {e}")
+
+    if resp.status_code == 429:
+        raise HTTPException(status_code=429, detail="Groq STT rate limited. Try again shortly.")
+    if resp.status_code == 401:
+        raise HTTPException(status_code=401, detail="Groq STT key invalid.")
+    if resp.status_code != 200:
+        raise HTTPException(
+            status_code=502,
+            detail=f"Groq STT returned {resp.status_code}: {resp.text[:200]}"
+        )
+
+    try:
+        data = resp.json()
+        text = data.get("text", "").strip()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"STT response parse failed: {e}")
+
+    latency_ms = (time.monotonic() - t_start) * 1000
+    logger.info(f"[STT] transcribed {len(audio_bytes)} bytes -> {len(text)} chars in {latency_ms:.0f}ms")
+
+    return JSONResponse({
+        "text": text,
+        "latency_ms": round(latency_ms, 1),
+        "model": GROQ_STT_MODEL,
+    })
+
+
+@app.post("/tts")
+async def tts(body: TTSRequest, request: Request) -> Response:
+    """
+    Convert text to WAV audio via Kokoro TTS.
+
+    Returns: WAV bytes (Content-Type: audio/wav)
+    VS5: synthesis runs in executor to avoid blocking event loop.
+    VS3: float32 -> int16 conversion inside _to_wav_bytes.
+    VS1: returns 503 if model not yet loaded.
+    """
+    _check_auth(request)
+
+    if not _model_ready:
+        if _model_loading:
+            raise HTTPException(status_code=503, detail="Kokoro model still loading. Try again in 15s.")
+        raise HTTPException(status_code=503, detail="Kokoro TTS unavailable (model failed to load).")
+
+    text = body.text.strip()
+    if not text:
+        raise HTTPException(status_code=400, detail="Empty text.")
+    if len(text) > 2000:
+        raise HTTPException(status_code=400, detail="Text too long (max 2000 chars).")
+
+    voice = body.voice or KOKORO_VOICE
+    t_start = time.monotonic()
+
+    # VS5: run blocking synthesis in thread executor
+    loop = asyncio.get_event_loop()
+    try:
+        wav_bytes = await loop.run_in_executor(
+            None,
+            _synthesize_sync,
+            text,
+            voice,
+        )
+    except Exception as e:
+        logger.error(f"[TTS] Synthesis failed: {e}")
+        raise HTTPException(status_code=500, detail=f"TTS synthesis failed: {e}")
+
+    latency_ms = (time.monotonic() - t_start) * 1000
+    logger.info(f"[TTS] synthesized {len(text)} chars -> {len(wav_bytes)} bytes in {latency_ms:.0f}ms voice={voice}")
+
+    return Response(
+        content=wav_bytes,
+        media_type="audio/wav",
+        headers={
+            "X-Latency-Ms": str(round(latency_ms, 1)),
+            "X-Voice": voice,
+        },
+    )
+
+
+def _synthesize_sync(text: str, voice: str) -> bytes:
+    """
+    Blocking Kokoro synthesis. Run via executor.
+    Concatenates all audio chunks from generator into single WAV.
+    VS3: float32 -> int16 via _to_wav_bytes.
+    VS6: sample_rate read from Kokoro pipeline attribute.
+    """
+    import numpy as np  # type: ignore
+
+    pipeline = _kokoro_pipeline
+    if pipeline is None:
+        raise RuntimeError("Kokoro pipeline not initialized")
+
+    # KPipeline(text, voice=...) returns generator of (graphemes, phonemes, audio_array)
+    audio_chunks = []
+    sample_rate = KOKORO_SAMPLE_RATE
+
+    try:
+        for gs, ps, audio in pipeline(text, voice=voice):
+            if audio is not None and len(audio) > 0:
+                audio_chunks.append(audio)
+                # VS6: try to read sample_rate from pipeline
+                if hasattr(pipeline, "sample_rate"):
+                    sample_rate = pipeline.sample_rate
+    except Exception as e:
+        raise RuntimeError(f"Kokoro synthesis error: {e}")
+
+    if not audio_chunks:
+        raise RuntimeError("Kokoro produced no audio")
+
+    combined = np.concatenate(audio_chunks, axis=0)
+    return _to_wav_bytes(combined, sample_rate=sample_rate)
+
+
+# ---------------------------------------------------------------------------
+# Entrypoint
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(
+        "packages.voice.server:app",
+        host="0.0.0.0",
+        port=7860,
+        log_level="info",
+        workers=1,  # single worker — Kokoro model in global state
+    )