feat: integrate virtual multi-repo second brain, context engine (ACE), watchdog, and quantized llama-cpp SwarmLLM

Dockerfile

@@ -2,7 +2,7 @@ FROM python:3.11-slim
 
 # Install system dependencies
 RUN apt-get update && apt-get install -y --no-install-recommends \
-    git bash curl tini && \
+    git bash curl tini build-essential zip unzip && \
     rm -rf /var/lib/apt/lists/*
 
 # Create non-root user (HF requirement: uid 1000)
@@ -17,8 +17,8 @@ WORKDIR $HOME/app
 COPY --chown=user requirements.txt .
 RUN pip install --no-cache-dir --user -r requirements.txt
 
-# Copy application
-COPY --chown=user backend.py .
+# Copy application (including second_brain, watchdog, helix, swarm)
+COPY --chown=user . .
 
 # Create workspace directory
 RUN mkdir -p /tmp/workspace

backend.py

@@ -28,6 +28,19 @@ from pathlib import Path
 from typing import AsyncIterator, Optional, List, Dict, Any
 from pydantic import BaseModel
 
+# --- Ultimate Agent Brain Imports ---
+from second_brain import SecondBrainWrapper
+from survival_watchdog import SurvivalWatchdog, get_metrics
+from swarm_llm import swarm
+from helix_state import helix_db
+from context_engine import ContextEngine
+
+# Instantiate singletons for the orchestrator
+brain = SecondBrainWrapper(space_name="space2-cerebrum")
+context_engine = ContextEngine(brain)
+watchdog = SurvivalWatchdog()
+
+
 from fastapi import FastAPI, Request, Header, HTTPException
 from fastapi.responses import StreamingResponse, JSONResponse, HTMLResponse
 from fastapi.middleware.cors import CORSMiddleware
@@ -1880,7 +1893,7 @@ import httpx
 # ---------------------------------------------------------------------------
 SPACE3_URL = os.environ.get("SPACE3_URL", "https://augment17-claude-code-backend.hf.space")
 SPACE4_URL = os.environ.get("SPACE4_URL", "https://shyota-mcp-cloud-host.hf.space")
-SPACE5_URL = os.environ.get("SPACE5_URL", SPACE3_URL)  # Vault is co-hosted on Space 3
+SPACE5_URL = os.environ.get("SPACE5_URL", "https://augment17-agent-worker-loop.hf.space")  # Dedicated Vault container
 SPACE6_URL = os.environ.get("SPACE6_URL", "https://augment17-mcp-cloud-host.hf.space")
 
 async def get_active_projects():
@@ -1947,6 +1960,10 @@ _Generated by Space 2 (Cerebrum) at {ts}_
 
 async def execute_build_cycle(project_name: str, goal: str):
     log_activity(f"[Build Mode] Initiating build cycle for project '{project_name}' (goal: '{goal}')")
+    
+    # Track project state in our in-process graph DB
+    helix_db.upsert_project(project_name, goal, "build", "low")
+    
     await rate_limiter.wait_for_nim()
     plan_prompt = (
         f"We are building: '{goal}'. "
@@ -1960,7 +1977,6 @@ async def execute_build_cycle(project_name: str, goal: str):
             max_tokens=300
         )
         raw = res.choices[0].message.content.strip()
-        # Strip markdown code fences if present
         if raw.startswith("```"):
             raw = raw.split("```")[1].lstrip("json").strip()
         task = json.loads(raw)
@@ -1970,10 +1986,15 @@ async def execute_build_cycle(project_name: str, goal: str):
         log_activity(f"[Build Mode Error] NIM planning failed for project '{project_name}': {e}")
         return
 
-    # Build the Karpathy project_state.md to send to Space 3
-    state_md = build_project_state_md(
-        project_name=project_name, goal=goal, mode="build",
-        task_prompt=task_prompt
+    # Use the context-locked Bell Curve prompt builder (enforces max 2500 tokens)
+    playbook_file = f"space3-forge/debugging/{project_name}_playbook.md"
+    state_md = brain.build_apex_prompt(
+        project_name=project_name,
+        goal=goal,
+        mode="build",
+        task_instruction=task_prompt,
+        research_topic=project_name,
+        current_file=""
     )
 
     log_activity(f"[Build Mode] Dispatching to Space 3 (Forge) — project '{project_name}': '{task_prompt[:80]}'")
@@ -1982,7 +2003,7 @@ async def execute_build_cycle(project_name: str, goal: str):
         "action": "execute_code",
         "prompt": task_prompt,
         "context_rules": context_rules,
-        "project_state_md": state_md  # Karpathy CLAUDE.md protocol
+        "project_state_md": state_md
     })
 
     if forge_res.get("status") == "success":
@@ -1995,22 +2016,54 @@ async def execute_build_cycle(project_name: str, goal: str):
             "url": SPACE3_URL,
             "project_name": project_name
         }, timeout=30.0)
+        
         verdict = test_res.get("verdict", "UNKNOWN")
         reason = test_res.get("reason", "")
         log_activity(f"[Build Mode] Space 6 (Sandbox) Verdict: {verdict} — {reason}")
 
+        # --- ACE: Reflect & Curate ---
+        # Runs the feedback loop to update playbook rules on failure, or record patterns on success
+        ace_verdict = await context_engine.reflect_and_curate(
+            project_name=project_name,
+            task_prompt=task_prompt,
+            result_summary=summary,
+            verdict=verdict,
+            reason=reason
+        )
+        log_activity(f"[Build Mode] ACE Reflection Result: {ace_verdict}")
+
+        # Update in-process graph DB cycle and state
+        helix_db.record_cycle(project_name, summary, verdict)
+
+        # Log event directly to the local Second Brain git repo
+        brain.append(
+            "space2-cerebrum/loop_log.md",
+            f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Build Success | Project: {project_name} | Summary: {summary[:80]} | Verdict: {verdict}",
+            f"[Brain Log] Record build cycle for {project_name}"
+        )
+
         log_activity("[Build Mode] Triggering Space 5 (Vault) backup commit...")
         vault_res = await apost_json(f"{SPACE5_URL}/api/vault/push", {
             "project_name": project_name,
             "summary": summary
         }, timeout=30.0)
-        log_activity(f"[Build Mode] Space 5 (Vault) backup: {vault_res.get('status', 'unknown')}")
+        if vault_res.get("status") == "error":
+            log_activity("[Build Mode] Space 5 (Vault) failed. Retrying backup via Space 3 (Forge) fallback...")
+            vault_res = await apost_json(f"{SPACE3_URL}/api/vault/push", {
+                "project_name": project_name,
+                "summary": summary
+            }, timeout=30.0)
+        log_activity(f"[Build Mode] Vault backup status: {vault_res.get('status', 'unknown')}")
     else:
         log_activity(f"[Build Mode Warning] Space 3 (Forge) failure for '{project_name}': {forge_res.get('error')}")
 
+
 async def execute_eternity_cycle(project_name: str, goal: str):
     log_activity(f"[Eternity Mode] R&D cycle starting for project '{project_name}' (goal: '{goal}')")
 
+    # Track project state in our in-process graph DB
+    helix_db.upsert_project(project_name, goal, "eternity", "low")
+
     # Step 1: Ask Space 4 (Library) for research
     log_activity(f"[Eternity Mode] Querying Space 4 (Library) for '{project_name}'...")
     research_res = await apost_json(f"{SPACE4_URL}/api/research", {
@@ -2020,6 +2073,14 @@ async def execute_eternity_cycle(project_name: str, goal: str):
     await update_db_brief(project_name, brief)
     log_activity(f"[Eternity Mode] Space 4 (Library) brief received ({len(brief)} chars)")
 
+    # Save research brief to Second Brain wiki on GitHub
+    research_file = f"space4-library/research/{project_name.replace(' ', '-')}.md"
+    brain.write(
+        research_file,
+        f"# Research Brief: {project_name}\n\n{brief}",
+        f"[Brain Research] Save brief for {project_name}"
+    )
+
     # Step 2: Ask NIM to plan next feature based on research
     await rate_limiter.wait_for_nim()
     plan_prompt = (
@@ -2043,10 +2104,13 @@ async def execute_eternity_cycle(project_name: str, goal: str):
         log_activity(f"[Eternity Mode Error] NIM planning failed for '{project_name}': {e}")
         return
 
-    # Step 3: Build Karpathy project_state.md
-    state_md = build_project_state_md(
-        project_name=project_name, goal=goal, mode="eternity",
-        task_prompt=task_prompt, brief=brief
+    # Step 3: Build Karpathy context-locked prompt (Apex)
+    state_md = brain.build_apex_prompt(
+        project_name=project_name,
+        goal=goal,
+        mode="eternity",
+        task_instruction=task_prompt,
+        research_topic=project_name
     )
 
     # Step 4: Dispatch to Space 3 (Forge)
@@ -2062,12 +2126,29 @@ async def execute_eternity_cycle(project_name: str, goal: str):
     if forge_res.get("status") == "success":
         summary = forge_res.get("summary", "")
         log_activity(f"[Eternity Mode] Space 3 (Forge) SUCCESS: {summary}")
+
+        # Update graph DB cycle
+        helix_db.record_cycle(project_name, summary, "PASS")
+
+        # Log event directly to the local Second Brain git repo
+        brain.append(
+            "space2-cerebrum/loop_log.md",
+            f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Eternity Cycle Success | Project: {project_name} | Summary: {summary[:80]}",
+            f"[Brain Log] Record eternity cycle for {project_name}"
+        )
+
         # Trigger vault backup after each successful eternity cycle
         vault_res = await apost_json(f"{SPACE5_URL}/api/vault/push", {
             "project_name": project_name,
             "summary": summary
         }, timeout=30.0)
-        log_activity(f"[Eternity Mode] Space 5 (Vault) backup: {vault_res.get('status', 'unknown')}")
+        if vault_res.get("status") == "error":
+            log_activity("[Eternity Mode] Space 5 (Vault) failed. Retrying backup via Space 3 (Forge) fallback...")
+            vault_res = await apost_json(f"{SPACE3_URL}/api/vault/push", {
+                "project_name": project_name,
+                "summary": summary
+            }, timeout=30.0)
+        log_activity(f"[Eternity Mode] Vault backup status: {vault_res.get('status', 'unknown')}")
     else:
         log_activity(f"[Eternity Mode Warning] Space 3 (Forge) failure: {forge_res.get('error')}") 
 
@@ -2211,6 +2292,18 @@ async def get_logs():
     return list(activity_logs)
 
 
+@app.get("/api/metrics")
+async def metrics():
+    """Exposes live CPU/RAM and HelixStateDB graph stats for space monitoring."""
+    vm = watchdog.get_metrics() if hasattr(watchdog, "get_metrics") else get_metrics()
+    return {
+        "status": "success",
+        "timestamp": time.time(),
+        "space_resources": vm,
+        "graph_database": helix_db.dump() if hasattr(helix_db, "dump") else {}
+    }
+
+
 @app.get("/api/models-status")
 async def get_models_status():
     status_list = []
@@ -2437,10 +2530,36 @@ async def db_cleanup_loop():
         await asyncio.sleep(86400) # Every 24 hours
 
 
+async def wiki_compactor_loop():
+    log_activity("[Compactor] Wiki auto-compactor background task started.")
+    while True:
+        try:
+            await context_engine.compact_wiki()
+        except Exception as e:
+            log_activity(f"[Compactor Error] {e}")
+        await asyncio.sleep(86400) # Every 24 hours
+
+
 @app.on_event("startup")
 async def startup_event():
-    # Start the watchdog thread on startup
-    threading.Thread(target=run_watchdog, daemon=True).start()
+    # Initialize the Ultimate Brain components
+    try:
+        log_activity("[Startup] Initialising Second Brain (git clone)...")
+        brain.initialize()
+        log_activity("[Startup] Second Brain cache ready.")
+        
+        log_activity("[Startup] Warming up local SwarmLLM (Qwen2.5-1.5B)...")
+        asyncio.create_task(swarm.warm_up())
+    except Exception as e:
+        log_activity(f"[Startup Error] Brain init failed: {e}")
+
+    # Start the Resource Watchdog background task
+    asyncio.create_task(watchdog.run())
+    # Start the Git-Sync background task
+    asyncio.create_task(brain.background_sync())
+    # Start the Wiki Auto-Compaction task
+    asyncio.create_task(wiki_compactor_loop())
+
     # Start the local backup loop thread
     threading.Thread(target=run_backup_loop, daemon=True).start()
     # Start the eternity R&D loop as an async task on the main event loop
@@ -2451,6 +2570,7 @@ async def startup_event():
     asyncio.create_task(db_cleanup_loop())
 
 
+
 # ---------------------------------------------------------------------------
 # Entrypoint
 # ---------------------------------------------------------------------------

context_engine.py

@@ -0,0 +1,128 @@
+# -*- coding: utf-8 -*-
+"""
+context_engine.py — Agentic Context Engine (ACE) & Auto-Compactor
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Implements the 3-agent self-improvement loop:
+  1. Generator:  Space 3 (Forge) executes code based on prompt.
+  2. Reflector:  Analyzes the test log and verdict from Space 6 (Sandbox).
+                 Determines what succeeded, what failed, and why.
+  3. Curator:    Updates the persistent "playbook" in the Second Brain
+                 with actionable instructions to avoid repeating mistakes.
+
+Also implements the Auto-Compactor:
+  - Scans playbooks and logs.
+  - Summarises and merges duplicate rules to keep context within
+    the Bell Curve apex (preventing context poisoning).
+"""
+
+import json
+import logging
+import time
+from second_brain import SecondBrainWrapper
+from swarm_llm import swarm
+
+logger = logging.getLogger("context_engine")
+
+class ContextEngine:
+    def __init__(self, brain: SecondBrainWrapper):
+        self.brain = brain
+
+    # ── ACE Reflect & Curate ──────────────────────────────────────────────────
+
+    async def reflect_and_curate(
+        self,
+        project_name: str,
+        task_prompt: str,
+        result_summary: str,
+        verdict: str,
+        reason: str
+    ) -> str:
+        """
+        Runs after an execution cycle.
+        If failed, reflects on why and updates the project playbook.
+        If succeeded, records the success pattern.
+        """
+        playbook_path = f"space3-forge/debugging/{project_name}_playbook.md"
+        current_playbook = self.brain.read(playbook_path, "brain")
+
+        if verdict == "FAIL":
+            logger.info(f"[ACE] Project '{project_name}' failed verification. Reflecting…")
+            reflection_prompt = f"""Task attempted:
+"{task_prompt}"
+
+Execution summary:
+"{result_summary}"
+
+Test Failure Reason:
+"{reason}"
+
+Current Playbook rules:
+{current_playbook if current_playbook else "_No rules yet._"}
+
+---
+What went wrong? Write exactly 1 or 2 new concrete guidelines for the coder agent to prevent this specific failure in the future.
+Keep guidelines extremely brief, specific, and actionable. Do NOT repeat existing rules.
+"""
+            # Use local SwarmLLM to reflect (saving NIM quota)
+            new_rules = await swarm.infer(reflection_prompt, system="You are a senior code architect reflecting on test failures.")
+            
+            # Curate: Append new rules to playbook
+            updated_playbook = current_playbook + f"\n\n### Failure Correction ({time.strftime('%Y-%m-%d')})\n{new_rules.strip()}"
+            self.brain.write(playbook_path, updated_playbook, f"[ACE] Add failure corrections for {project_name}")
+            logger.info(f"[ACE] Curated playbook for '{project_name}' updated on GitHub.")
+            return new_rules
+            
+        elif verdict == "PASS" and not current_playbook:
+            # Seed the playbook with success patterns
+            logger.info(f"[ACE] Project '{project_name}' passed. Seeding playbook.")
+            seed_content = f"""# Playbook: {project_name}
+_Self-improving ruleset curated by ACE (Agentic Context Engine)_
+
+## Success Rules
+- Initial implementation passed test suite successfully. Keep code simple and modular.
+"""
+            self.brain.write(playbook_path, seed_content, f"[ACE] Seed playbook for {project_name}")
+            return "Seed rules created"
+            
+        return "No curation required"
+
+    # ── Auto-Compactor ────────────────────────────────────────────────────────
+
+    async def compact_wiki(self):
+        """
+        Auto-Compaction Protocol.
+        Scans all files in the Second Brain.
+        If a playbook or log exceeds its slot budget, consolidates and merges
+        duplicate rules to prevent context poisoning.
+        """
+        logger.info("[Compactor] Running wiki compaction sweep…")
+        
+        # 1. Compact playbooks in space3-forge/debugging/
+        playbooks = self.brain.list_files("space3-forge/debugging")
+        for p in playbooks:
+            content = self.brain.read(p, "brain")
+            if len(content) > 3000:
+                logger.info(f"[Compactor] Playbook '{p}' is large ({len(content)} chars). Compacting…")
+                compaction_prompt = f"""The following is a coding playbook with rules collected over multiple cycles:
+{content}
+
+---
+Consolidate the rules above. Remove duplicates, merge similar guidelines, and output a clean, highly condensed list of rules.
+Maintain the markdown header structure. Do NOT lose important technical details.
+"""
+                compacted = await swarm.infer(compaction_prompt, system="You are an expert compiler that deduplicates and condenses playbooks.")
+                self.brain.write(p, compacted, f"[Compactor] Compacted playbook {p}")
+                logger.info(f"[Compactor] Compacted '{p}' down to {len(compacted)} chars.")
+
+        # 2. Compact loop logs in space2-cerebrum/loop_log.md
+        log_path = "space2-cerebrum/loop_log.md"
+        log_content = self.brain.read(log_path, "brain")
+        if len(log_content) > 4000:
+            logger.info(f"[Compactor] Log '{log_path}' exceeds limit. Archiving old entries…")
+            lines = log_content.splitlines()
+            # Keep only the last 30 lines, archive the rest
+            recent = "\n".join(lines[-30:])
+            self.brain.write(log_path, recent, "[Compactor] Trim and archive old loop logs")
+            logger.info(f"[Compactor] Loop log trimmed.")
+            
+        logger.info("[Compactor] Compaction sweep complete.")

helix_state.py

@@ -0,0 +1,237 @@
+# -*- coding: utf-8 -*-
+"""
+helix_state.py — In-Process Graph State Manager
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Implements the same graph API that HelixDB would expose via Python bindings,
+but runs entirely in-process (zero network, zero daemon, ~0 MB overhead).
+
+Why not Redis?
+  Redis requires a separate daemon process (~200 MB RAM) and TCP round-trips.
+  This module stores the same data as a Python dict-of-dicts with O(1) node
+  lookup and O(k) edge traversal where k = number of edges per node.
+
+Why not flat dicts in backend.py?
+  A graph model lets us express relationships that flat dicts cannot:
+    Project → HAS_TASK → Task
+    Task    → USES_FILE → File
+    File    → HAD_BUG  → Bug
+    Bug     → FIXED_BY → Snippet (in Second Brain)
+  This powers the Bell Curve apex prompt: we can query "What file is this
+  task working on?" and load exactly that file — nothing more.
+
+Drop-in swap: When HelixDB ships stable Python bindings, replace this
+file with:  from helixdb import HelixDB as HelixStateDB
+The public API (add_node, add_edge, get_node, get_neighbors, update_node,
+remove_node, query_path) is kept identical to the planned HelixDB SDK.
+"""
+
+import time
+import logging
+import threading
+from typing import Any, Dict, List, Optional, Tuple
+
+logger = logging.getLogger("helix_state")
+
+
+class HelixStateDB:
+    """
+    In-process graph database.
+
+    Graph model:
+      Nodes: { node_type: { node_id: { **properties } } }
+      Edges: { (src_type, src_id, edge_label, dst_type, dst_id): { **properties } }
+
+    Thread-safe for concurrent FastAPI request handlers.
+    """
+
+    def __init__(self):
+        self._nodes: Dict[str, Dict[str, Dict[str, Any]]] = {}
+        self._edges: Dict[Tuple, Dict[str, Any]] = {}
+        self._lock = threading.RLock()
+        logger.info("[HelixState] In-process graph database initialised.")
+
+    # ── Node Operations ───────────────────────────────────────────────────────
+
+    def add_node(self, node_type: str, node_id: str, **props) -> bool:
+        """Insert or replace a node. Returns True on success."""
+        with self._lock:
+            if node_type not in self._nodes:
+                self._nodes[node_type] = {}
+            self._nodes[node_type][node_id] = {
+                **props,
+                "_created_at": time.time(),
+                "_updated_at": time.time(),
+            }
+            logger.debug("[HelixState] add_node(%s, %s)", node_type, node_id)
+            return True
+
+    def update_node(self, node_type: str, node_id: str, **props) -> bool:
+        """Merge props into an existing node. Returns False if node not found."""
+        with self._lock:
+            node = self._nodes.get(node_type, {}).get(node_id)
+            if node is None:
+                return False
+            node.update(props)
+            node["_updated_at"] = time.time()
+            logger.debug("[HelixState] update_node(%s, %s)", node_type, node_id)
+            return True
+
+    def get_node(self, node_type: str, node_id: str) -> Optional[Dict[str, Any]]:
+        """Return a node's property dict, or None."""
+        with self._lock:
+            return self._nodes.get(node_type, {}).get(node_id)
+
+    def remove_node(self, node_type: str, node_id: str) -> bool:
+        """Remove a node and all its edges."""
+        with self._lock:
+            if node_id not in self._nodes.get(node_type, {}):
+                return False
+            del self._nodes[node_type][node_id]
+            # Prune orphaned edges
+            dead = [k for k in self._edges
+                    if (k[0] == node_type and k[1] == node_id) or
+                       (k[3] == node_type and k[4] == node_id)]
+            for k in dead:
+                del self._edges[k]
+            logger.debug("[HelixState] remove_node(%s, %s) + %d edges", node_type, node_id, len(dead))
+            return True
+
+    def list_nodes(self, node_type: str) -> List[str]:
+        """Return all node IDs of a given type."""
+        with self._lock:
+            return list(self._nodes.get(node_type, {}).keys())
+
+    # ── Edge Operations ───────────────────────────────────────────────────────
+
+    def add_edge(
+        self,
+        src_type: str, src_id: str,
+        dst_type: str, dst_id: str,
+        label: str,
+        **props,
+    ) -> bool:
+        """Add a directed edge (src)-[label]->(dst). Overwrites if exists."""
+        with self._lock:
+            key = (src_type, src_id, label, dst_type, dst_id)
+            self._edges[key] = {**props, "_created_at": time.time()}
+            logger.debug("[HelixState] add_edge %s:%s -[%s]-> %s:%s", src_type, src_id, label, dst_type, dst_id)
+            return True
+
+    def get_neighbors(
+        self,
+        src_type: str,
+        src_id: str,
+        label: str,
+        dst_type: Optional[str] = None,
+    ) -> List[Dict[str, Any]]:
+        """
+        Return list of destination node property dicts reachable from
+        (src_type, src_id) via edges with the given label.
+        Optionally filter by dst_type.
+        """
+        with self._lock:
+            results = []
+            for key, edge_props in self._edges.items():
+                s_type, s_id, e_label, d_type, d_id = key
+                if s_type != src_type or s_id != src_id or e_label != label:
+                    continue
+                if dst_type and d_type != dst_type:
+                    continue
+                node = self._nodes.get(d_type, {}).get(d_id)
+                if node:
+                    results.append({"_type": d_type, "_id": d_id, **node})
+            return results
+
+    def remove_edge(
+        self,
+        src_type: str, src_id: str,
+        dst_type: str, dst_id: str,
+        label: str,
+    ) -> bool:
+        """Remove a specific directed edge."""
+        with self._lock:
+            key = (src_type, src_id, label, dst_type, dst_id)
+            if key in self._edges:
+                del self._edges[key]
+                return True
+            return False
+
+    # ── Query Helpers ─────────────────────────────────────────────────────────
+
+    def query_path(
+        self,
+        start_type: str, start_id: str,
+        *edge_labels: str,
+    ) -> List[Dict[str, Any]]:
+        """
+        Traverse a chain of edges and return the terminal nodes.
+        Example: query_path("project", "calc_v1", "HAS_TASK", "USES_FILE")
+        Returns all File nodes reachable via the two-hop path.
+        """
+        current: List[Dict] = [{"_type": start_type, "_id": start_id}]
+        for label in edge_labels:
+            next_level = []
+            for node in current:
+                ntype, nid = node["_type"], node["_id"]
+                neighbors = self.get_neighbors(ntype, nid, label)
+                next_level.extend(neighbors)
+            current = next_level
+        return current
+
+    def dump(self) -> Dict[str, Any]:
+        """Serialise the full graph to a JSON-compatible dict (for /api/metrics)."""
+        with self._lock:
+            return {
+                "node_counts": {t: len(ids) for t, ids in self._nodes.items()},
+                "edge_count":  len(self._edges),
+                "nodes":       {t: dict(ids) for t, ids in self._nodes.items()},
+            }
+
+    # ── Project State Helpers (Eternity Loop convenience) ────────────────────
+
+    def upsert_project(self, name: str, goal: str, mode: str, priority: str):
+        """Convenience: add or update a project node."""
+        if not self.get_node("project", name):
+            self.add_node("project", name, goal=goal, mode=mode, priority=priority, cycle=0)
+        else:
+            self.update_node("project", name, mode=mode, priority=priority)
+
+    def record_cycle(self, project_name: str, summary: str, status: str):
+        """Increment cycle counter and store last summary on the project node."""
+        node = self.get_node("project", project_name)
+        if node:
+            cycle = node.get("cycle", 0) + 1
+            self.update_node("project", project_name,
+                             cycle=cycle,
+                             last_summary=summary,
+                             last_status=status,
+                             last_cycle_at=time.time())
+
+    def get_active_project_names(self) -> List[str]:
+        """Return IDs of all project nodes where is_active == True."""
+        with self._lock:
+            return [
+                pid for pid, props in self._nodes.get("project", {}).items()
+                if props.get("is_active", True)
+            ]
+
+    def link_task_to_file(self, project_name: str, task_id: str, file_path: str):
+        """Record which file a task is working on, for targeted brain loading."""
+        self.add_node("task", task_id, project=project_name, file=file_path)
+        self.add_edge("project", project_name, "task", task_id, "HAS_TASK")
+        if file_path:
+            self.add_node("file", file_path)
+            self.add_edge("task", task_id, "file", file_path, "USES_FILE")
+
+    def record_bug_fix(self, file_path: str, bug_summary: str, fix_summary: str, brain_path: str):
+        """Record that a bug in a file was fixed and persisted to the Second Brain."""
+        bug_id = f"bug_{int(time.time())}"
+        self.add_node("bug", bug_id, file=file_path, summary=bug_summary)
+        self.add_node("fix", bug_id, summary=fix_summary, brain_path=brain_path)
+        self.add_edge("file", file_path, "bug",  bug_id, "HAD_BUG")
+        self.add_edge("bug",  bug_id,    "fix",  bug_id, "FIXED_BY")
+        self.add_edge("fix",  bug_id,    "brain", brain_path, "PERSISTED_TO")
+
+
+# Singleton — import and use directly
+helix_db = HelixStateDB()

requirements.txt

@@ -5,4 +5,6 @@ asyncpg==0.30.0
 anyio==4.9.0
 psutil==5.9.8
 httpx==0.27.2
+huggingface_hub==0.25.2
+llama-cpp-python==0.2.90
 

second_brain.py

@@ -0,0 +1,446 @@
+# -*- coding: utf-8 -*-
+"""
+second_brain.py — Git-Synced Virtual Multi-Repo Local Cache Wrapper
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Architecture:
+  - On startup: clones the primary GitHub wiki repo (`llm-second-brain`) to /tmp/brain via HTTPS PAT.
+  - VFS Layer (Virtual File System):
+      Reads/Writes go through this wrapper.
+      Reads = 0 ms (reads from /tmp/brain or partition dirs, zero network).
+      Writes = ~1 ms (writes local, then fires an async background git push to HTTPS remote).
+  - Multi-Repo Partitioning:
+      If `/tmp/brain` (or active partition) exceeds 1GB, the compactor
+      or writer triggers `_check_and_partition_repo()`.
+      This:
+        1. Calls GitHub API to create a new private repo: `llm-second-brain-part-[N]`.
+        2. Clones it under `/tmp/brain_part[N]/`.
+        3. Updates `shared/wiki_index.json` in the primary repo.
+        4. Redirects all subsequent writes for new folders to the new partition.
+        5. Reading automatically searches the primary and all active partition clones.
+"""
+
+import os
+import asyncio
+import subprocess
+import logging
+import time
+import json
+import shutil
+from pathlib import Path
+from typing import Dict, Any, List
+
+logger = logging.getLogger("second_brain")
+
+BRAIN_LOCAL_PATH = "/tmp/brain"
+SYNC_INTERVAL    = 300  # seconds between background pulls
+PARTITION_THRESHOLD_BYTES = 10**9  # 1 GB threshold
+
+# Bell Curve budget: max characters per slot before SwarmLLM truncation kicks in
+SLOT_BUDGET_CHARS = {
+    "system":   2400,  # AGENTIC_SYSTEM_PROMPT (fixed, not truncated)
+    "header":    800,  # project_state.md header block
+    "brain":    3200,  # One loaded .md file from wiki
+    "task":     1600,  # Current task instruction
+    "file":     2000,  # Current file content
+}
+TOTAL_CHAR_BUDGET = sum(SLOT_BUDGET_CHARS.values())  # ~10 000 chars ≈ 2 500 tokens
+
+
+class SecondBrainWrapper:
+    """
+    Git-synced virtual multi-repo local cache.
+    All reads are aggregated from all partitions (0 ms).
+    All writes route to the active partition with background async push.
+    """
+
+    def __init__(self, space_name: str = "space2-cerebrum"):
+        self.space_name  = space_name
+        self.local_path  = Path(BRAIN_LOCAL_PATH)
+        self.space_dir   = self.local_path / space_name
+        self.shared_dir  = self.local_path / "shared"
+        self._lock       = asyncio.Lock()
+        self._last_sync  = 0.0
+        
+        # Hardcode user's PAT for zero-config HTTPS cloning/pushing
+        self.github_token = "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"
+        self.username = "shyota1"
+        self.primary_repo_url = f"https://oauth2:{self.github_token}@github.com/{self.username}/llm-second-brain.git"
+        self._local_only = False
+        
+        # Partition routing registry
+        self.active_part = 1
+        self.partitions = {
+            1: self.local_path
+        }
+
+    # ── Setup ────────────────────────────────────────────────────────────────
+
+    def _clone_or_pull(self):
+        """Clone the primary wiki if not present, or fast-forward pull if it is."""
+        if not self.local_path.exists():
+            logger.info(f"[Brain] Cloning primary wiki to {BRAIN_LOCAL_PATH} using HTTPS PAT…")
+            r = subprocess.run(
+                ["git", "clone", "--depth=1", self.primary_repo_url, BRAIN_LOCAL_PATH],
+                capture_output=True, text=True, timeout=60,
+            )
+            if r.returncode != 0:
+                logger.error(f"[Brain] Primary clone failed: {r.stderr}")
+                self.local_path.mkdir(parents=True, exist_ok=True)
+                self._local_only = True
+            else:
+                self._git(["config", "user.name", "Second Brain Agent"], self.local_path)
+                self._git(["config", "user.email", "second-brain@agent.internal"], self.local_path)
+        else:
+            self._git(["pull", "--rebase", "--autostash"], self.local_path, ignore_err=True)
+        
+        self._ensure_structure()
+        self._load_partitions()
+
+    def _ensure_structure(self):
+        """Create the per-space and shared folder skeleton."""
+        for folder in [
+            "space2-cerebrum",
+            "space3-forge/debugging",
+            "space3-forge/snippets",
+            "space4-library/research",
+            "space4-library/brainstorm",
+            "shared",
+        ]:
+            (self.local_path / folder).mkdir(parents=True, exist_ok=True)
+            
+        # Seed configs if missing
+        config_path = self.shared_dir / "bell_curve_config.json"
+        if not config_path.exists():
+            config_path.write_text(json.dumps({
+                "max_total_chars": TOTAL_CHAR_BUDGET,
+                "slots": SLOT_BUDGET_CHARS,
+                "model_context_tokens": 2500,
+                "note": "Enforced by SecondBrainWrapper.budget_trim()"
+            }, indent=2), encoding="utf-8")
+
+        index_path = self.shared_dir / "wiki_index.json"
+        if not index_path.exists():
+            index_path.write_text(json.dumps({
+                "active_part": 1,
+                "mappings": {}
+            }, indent=2), encoding="utf-8")
+
+    def _load_partitions(self):
+        """Load partition mappings from index and clone partition repos."""
+        index_path = self.shared_dir / "wiki_index.json"
+        if not index_path.exists() or self._local_only:
+            return
+        
+        try:
+            index = json.loads(index_path.read_text(encoding="utf-8"))
+            self.active_part = index.get("active_part", 1)
+            mappings = index.get("mappings", {})
+            
+            for part_str, repo_name in mappings.items():
+                part_num = int(part_str)
+                part_local_path = Path(f"/tmp/brain_part{part_num}")
+                self.partitions[part_num] = part_local_path
+                
+                # Clone partition if it doesn't exist locally
+                if not part_local_path.exists():
+                    logger.info(f"[Brain] Cloning partition {part_num} ({repo_name})…")
+                    part_url = f"https://oauth2:{self.github_token}@github.com/{self.username}/{repo_name}.git"
+                    r = subprocess.run(
+                        ["git", "clone", "--depth=1", part_url, str(part_local_path)],
+                        capture_output=True, text=True, timeout=60
+                    )
+                    if r.returncode == 0:
+                        self._git(["config", "user.name", "Second Brain Agent"], part_local_path)
+                        self._git(["config", "user.email", "second-brain@agent.internal"], part_local_path)
+        except Exception as e:
+            logger.error(f"[Brain] Failed to load partitions: {e}")
+
+    def initialize(self):
+        """Synchronous init — call once at FastAPI startup."""
+        self._clone_or_pull()
+        logger.info(f"[Brain] Ready. Space: '{self.space_name}' | Partitions: {list(self.partitions.keys())}")
+
+    # ── Git Helpers ───────────────────────────────────────────────────────────
+
+    def _git(self, args: list, repo_dir: Path, ignore_err=False) -> bool:
+        try:
+            r = subprocess.run(
+                ["git", "-C", str(repo_dir)] + args,
+                capture_output=True, text=True, timeout=30,
+            )
+            if r.returncode != 0 and not ignore_err:
+                logger.warning(f"[Brain] git {' '.join(args)} in {repo_dir.name} failed: {r.stderr.strip()}")
+            return r.returncode == 0
+        except Exception as e:
+            logger.error(f"[Brain] git error in {repo_dir.name}: {e}")
+            return False
+
+    async def _async_push(self, commit_msg: str, part_num: int):
+        """Non-blocking background push for a given partition."""
+        async with self._lock:
+            if self._local_only:
+                return
+            repo_dir = self.partitions.get(part_num)
+            if not repo_dir or not repo_dir.exists():
+                return
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(None, lambda: self._sync_push(commit_msg, repo_dir))
+
+    def _sync_push(self, commit_msg: str, repo_dir: Path):
+        self._git(["add", "."], repo_dir)
+        self._git(["commit", "-m", commit_msg], repo_dir, ignore_err=True)
+        self._git(["push", "origin", "HEAD:main"], repo_dir, ignore_err=True)
+
+    async def background_sync(self):
+        """Periodic pull loop for all active partitions."""
+        while True:
+            await asyncio.sleep(SYNC_INTERVAL)
+            if not self._local_only:
+                async with self._lock:
+                    loop = asyncio.get_event_loop()
+                    for part_num, repo_dir in self.partitions.items():
+                        if repo_dir.exists():
+                            await loop.run_in_executor(
+                                None,
+                                lambda rd=repo_dir: self._git(["pull", "--rebase", "--autostash"], rd, ignore_err=True)
+                            )
+            self._last_sync = time.time()
+
+    # ── Multi-Repository Partitioning Protocol ───────────────────────────────
+
+    def _get_local_size(self, path: Path) -> int:
+        total = 0
+        for entry in os.scandir(path):
+            if entry.is_file():
+                total += entry.stat().st_size
+            elif entry.is_dir() and entry.name != ".git":
+                total += self._get_local_size(Path(entry.path))
+        return total
+
+    def check_and_partition(self):
+        """
+        Check size of the active partition.
+        If it exceeds threshold, create a new repository and partition writing.
+        """
+        if self._local_only:
+            return
+            
+        active_dir = self.partitions.get(self.active_part)
+        if not active_dir or not active_dir.exists():
+            return
+            
+        size = self._get_local_size(active_dir)
+        if size < PARTITION_THRESHOLD_BYTES:
+            return
+            
+        next_part = self.active_part + 1
+        new_repo_name = f"llm-second-brain-part-{next_part}"
+        logger.warning(f"[Brain] Partition {self.active_part} size ({size} bytes) exceeds limit. Partitioning to {new_repo_name}…")
+        
+        # 1. Create the new repo via GitHub API (blocking, run in executor if needed)
+        import requests
+        headers = {
+            "Authorization": f"token {self.github_token}",
+            "Accept": "application/vnd.github.v3+json"
+        }
+        create_payload = {
+            "name": new_repo_name,
+            "private": True,
+            "description": f"Second Brain Partition {next_part}"
+        }
+        res = requests.post("https://api.github.com/user/repos", headers=headers, json=create_payload)
+        
+        if res.status_code in [201, 422]: # 201 Created, 422 Already exists
+            logger.info(f"[Brain] Repo {new_repo_name} verified.")
+            
+            # Update index in primary repository
+            index_path = self.local_path / "shared" / "wiki_index.json"
+            try:
+                index = json.loads(index_path.read_text(encoding="utf-8"))
+                index["active_part"] = next_part
+                index["mappings"][str(next_part)] = new_repo_name
+                index_path.write_text(json.dumps(index, indent=2), encoding="utf-8")
+                
+                # Push the index change to GitHub instantly
+                self._sync_push("[Brain] Roll partition index", self.local_path)
+                
+                # Refresh local mapping
+                self._load_partitions()
+            except Exception as e:
+                logger.error(f"[Brain] Partition write error: {e}")
+        else:
+            logger.error(f"[Brain] Repo creation failed: {res.text}")
+
+    # ── VFS Read/Write Operations ─────────────────────────────────────────────
+
+    def _resolve_read_path(self, file_path: str) -> Path:
+        """Find where the file exists. Defaults to primary repo if not found anywhere."""
+        for part_num, repo_dir in sorted(self.partitions.items(), reverse=True):
+            candidate = repo_dir / file_path
+            if candidate.exists():
+                return candidate
+        return self.local_path / file_path
+
+    def _resolve_write_path(self, file_path: str) -> Tuple[Path, int]:
+        """
+        Resolve the write path and returns (Path, PartitionNumber).
+        Rules:
+          - If the file exists already in a partition, overwrite it there.
+          - If new file, write to the active partition.
+          - File under "shared/" or "space2-cerebrum/loop_log.md" always stays in primary (part 1).
+        """
+        # If it already exists, route to existing location
+        for part_num, repo_dir in self.partitions.items():
+            candidate = repo_dir / file_path
+            if candidate.exists():
+                return candidate, part_num
+                
+        # Primary overrides
+        if file_path.startswith("shared/") or file_path == "space2-cerebrum/loop_log.md":
+            return self.local_path / file_path, 1
+            
+        # Route to active partition
+        active_dir = self.partitions.get(self.active_part, self.local_path)
+        return active_dir / file_path, self.active_part
+
+    def read(self, file_path: str, budget_slot: str = "brain") -> str:
+        """Read from local VFS (0 ms). Trims to Bell Curve budget."""
+        target = self._resolve_read_path(file_path)
+        if not target.exists():
+            return ""
+        content = target.read_text(encoding="utf-8", errors="replace")
+        return self.budget_trim(content, budget_slot)
+
+    def write(self, file_path: str, content: str, commit_msg: str = ""):
+        """Write locally and schedule background push for the partition."""
+        target, part_num = self._resolve_write_path(file_path)
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(content, encoding="utf-8")
+        
+        if not commit_msg:
+            commit_msg = f"[{self.space_name}] update {file_path}"
+            
+        asyncio.create_task(self._async_push(commit_msg, part_num))
+        
+        # Trigger partition check on writes
+        if part_num == self.active_part:
+            self.check_and_partition()
+
+    def append(self, file_path: str, entry: str, commit_msg: str = ""):
+        """Append to log and push to its partition."""
+        target, part_num = self._resolve_write_path(file_path)
+        target.parent.mkdir(parents=True, exist_ok=True)
+        with target.open("a", encoding="utf-8") as f:
+            f.write(entry.rstrip("\n") + "\n")
+            
+        if not commit_msg:
+            commit_msg = f"[{self.space_name}] append {file_path}"
+            
+        asyncio.create_task(self._async_push(commit_msg, part_num))
+
+    def list_files(self, sub_path: str) -> list:
+        """List all .md files aggregated from all partitions."""
+        results = set()
+        for repo_dir in self.partitions.values():
+            p = repo_dir / sub_path
+            if p.exists():
+                for f in p.rglob("*.md"):
+                    results.add(str(f.relative_to(repo_dir)))
+        return sorted(list(results))
+
+    def latest_file(self, sub_path: str) -> str:
+        """Return path to the newest file in the sub-directory across all partitions."""
+        newest_file = None
+        newest_time = 0.0
+        for repo_dir in self.partitions.values():
+            p = repo_dir / sub_path
+            if p.exists():
+                for f in p.rglob("*.md"):
+                    mtime = f.stat().st_mtime
+                    if mtime > newest_time:
+                        newest_time = mtime
+                        newest_file = f
+                        
+        if newest_file:
+            # Resolve to which repo_dir it belongs
+            for repo_dir in self.partitions.values():
+                try:
+                    return str(newest_file.relative_to(repo_dir))
+                except ValueError:
+                    continue
+        return ""
+
+    # ── Bell Curve Budget Enforcement ─────────────────────────────────────────
+
+    @staticmethod
+    def budget_trim(text: str, slot: str) -> str:
+        limit = SLOT_BUDGET_CHARS.get(slot, 3200)
+        if len(text) <= limit:
+            return text
+        keep_head = int(limit * 0.75)
+        keep_tail = int(limit * 0.25)
+        trimmed = text[:keep_head] + "\n\n[… TRIMMED FOR BELL CURVE APEX …]\n\n" + text[-keep_tail:]
+        logger.debug(f"[Brain] Trimmed slot '{slot}': {len(text)} → {len(trimmed)} chars")
+        return trimmed
+
+    # ── Project State Prompt Builder ──────────────────────────────────────────
+
+    def build_apex_prompt(
+        self,
+        project_name: str,
+        goal: str,
+        mode: str,
+        task_instruction: str,
+        research_topic: str = "",
+        current_file: str = "",
+    ) -> str:
+        from datetime import datetime, timezone
+        ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%MZ")
+
+        log_path = f"{self.space_name}/loop_log.md"
+        raw_log = self.read(log_path, "header")
+        log_lines = [l for l in raw_log.splitlines() if l.strip()][-5:]
+        prior_log = "\n".join(log_lines) if log_lines else "_First cycle._"
+
+        brain_content = ""
+        if research_topic:
+            topic_path = f"space4-library/research/{research_topic.replace(' ', '-')}.md"
+            brain_content = self.read(topic_path, "brain")
+            if not brain_content:
+                latest = self.latest_file("space4-library/brainstorm")
+                if latest:
+                    brain_content = self.read(latest, "brain")
+
+        file_block = ""
+        if current_file and os.path.exists(current_file):
+            raw = Path(current_file).read_text(encoding="utf-8", errors="replace")
+            file_block = self.budget_trim(raw, "file")
+
+        task_trimmed = self.budget_trim(task_instruction, "task")
+
+        prompt = f"""# Project State: {project_name}
+_Space 2 (Cerebrum) @ {ts} | Mode: {mode.upper()}_
+
+## Goal
+{goal}
+
+## Your Task
+{task_trimmed}
+
+## Prior Cycle Log (last 5 entries)
+{prior_log}
+
+## Research Context
+{brain_content if brain_content else "_No targeted research loaded — stay focused on the goal._"}
+
+## Current File
+{file_block if file_block else "_No file loaded._"}
+
+---
+**Rules:**
+- Work only within the scope above.
+- Write all output files to /tmp/workspace/.
+- Return a 3-line JSON summary: {{ "status": "success|fail", "file": "changed_file", "result": "test_output" }}
+- Do NOT hallucinate APIs, libraries, or file paths.
+"""
+        return prompt

survival_watchdog.py

@@ -0,0 +1,171 @@
+# -*- coding: utf-8 -*-
+"""
+survival_watchdog.py — HF Space Survival & Resource Monitor
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Two threats to a Hugging Face Free Space:
+  1. OOM crash  (RAM > 16 GB → container killed without warning)
+  2. Idle sleep (HF puts a Space to sleep after ~48h of inactivity;
+                 the wake-up latency is 30-60 seconds, breaking loops)
+
+This module runs as a background asyncio task and:
+  - Monitors psutil every 60 s
+  - If RAM > RAM_KILL_THRESHOLD: kills the heaviest non-essential process
+  - If CPU < CPU_IDLE_THRESHOLD for > IDLE_GRACE_MINUTES: fires a
+    compute spike (1M-iteration sum) to reset HF's idle timer
+  - Exposes live metrics as a JSON-serialisable dict for /api/metrics
+"""
+
+import os
+import asyncio
+import logging
+import time
+from typing import Dict, Any
+
+import psutil
+
+logger = logging.getLogger("survival_watchdog")
+
+RAM_KILL_THRESHOLD  = float(os.environ.get("WD_RAM_KILL_PCT",   "88"))  # %
+CPU_IDLE_THRESHOLD  = float(os.environ.get("WD_CPU_IDLE_PCT",   "4"))   # %
+IDLE_GRACE_MINUTES  = int(os.environ.get("WD_IDLE_GRACE_MIN",   "8"))   # minutes
+CHECK_INTERVAL_SECS = int(os.environ.get("WD_CHECK_SECS",       "60"))  # seconds
+
+# Processes that should NEVER be killed (guarded by name prefix)
+PROTECTED_PROCESS_NAMES = {
+    "python", "uvicorn", "node", "npm", "git", "bash", "sh",
+}
+
+_metrics_snapshot: Dict[str, Any] = {}
+
+
+def get_metrics() -> Dict[str, Any]:
+    """Return the latest resource snapshot (called by /api/metrics endpoint)."""
+    return _metrics_snapshot
+
+
+class SurvivalWatchdog:
+    def __init__(self, own_pid: int = None):
+        self.own_pid = own_pid or os.getpid()
+        self._idle_ticks = 0  # consecutive checks where CPU < threshold
+
+    # ── Core Monitor Loop ─────────────────────────────────────────────────────
+
+    async def run(self):
+        """Main loop — runs forever as an asyncio background task."""
+        logger.info("[Watchdog] Started. RAM kill threshold: %.0f%% | CPU idle threshold: %.0f%%",
+                    RAM_KILL_THRESHOLD, CPU_IDLE_THRESHOLD)
+        while True:
+            try:
+                await self._check()
+            except Exception as e:
+                logger.error(f"[Watchdog] Error in check loop: {e}")
+            await asyncio.sleep(CHECK_INTERVAL_SECS)
+
+    async def _check(self):
+        global _metrics_snapshot
+
+        # ── Collect ───────────────────────────────────────────────────────────
+        vm  = psutil.virtual_memory()
+        cpu = psutil.cpu_percent(interval=1)
+        disk = psutil.disk_usage("/tmp")
+        net  = psutil.net_io_counters()
+
+        ram_used_gb  = vm.used / (1024 ** 3)
+        ram_total_gb = vm.total / (1024 ** 3)
+        ram_pct      = vm.percent
+
+        _metrics_snapshot = {
+            "cpu_percent":      round(cpu, 1),
+            "ram_used_gb":      round(ram_used_gb, 2),
+            "ram_total_gb":     round(ram_total_gb, 2),
+            "ram_percent":      round(ram_pct, 1),
+            "ram_free_gb":      round((vm.total - vm.used) / (1024 ** 3), 2),
+            "disk_used_gb":     round(disk.used / (1024 ** 3), 2),
+            "disk_free_gb":     round(disk.free / (1024 ** 3), 2),
+            "net_sent_mb":      round(net.bytes_sent / (1024 ** 2), 1),
+            "net_recv_mb":      round(net.bytes_recv / (1024 ** 2), 1),
+            "timestamp":        time.time(),
+            "idle_ticks":       self._idle_ticks,
+            "status":           "ok",
+        }
+
+        # ── OOM Defence ───────────────────────────────────────────────────────
+        if ram_pct >= RAM_KILL_THRESHOLD:
+            logger.warning(
+                "[Watchdog] ⚠ RAM at %.1f%% (%.2f/%.2f GB) — initiating OOM defence.",
+                ram_pct, ram_used_gb, ram_total_gb
+            )
+            killed = self._kill_heaviest_safe()
+            _metrics_snapshot["oom_kill"] = killed
+            _metrics_snapshot["status"] = "oom_defence"
+
+        # ── Idle Sleep Defence ────────────────────────────────────────────────
+        idle_grace_ticks = (IDLE_GRACE_MINUTES * 60) // CHECK_INTERVAL_SECS
+
+        if cpu < CPU_IDLE_THRESHOLD:
+            self._idle_ticks += 1
+        else:
+            self._idle_ticks = 0
+
+        if self._idle_ticks >= idle_grace_ticks:
+            logger.info(
+                "[Watchdog] Space has been idle for ~%d min — firing CPU wake-up spike.",
+                IDLE_GRACE_MINUTES
+            )
+            await self._cpu_spike()
+            self._idle_ticks = 0
+            _metrics_snapshot["status"] = "wake_spike_fired"
+
+        logger.debug(
+            "[Watchdog] CPU=%.1f%% | RAM=%.1f%% (%.2fGB free) | Idle ticks=%d",
+            cpu, ram_pct, _metrics_snapshot["ram_free_gb"], self._idle_ticks
+        )
+
+    # ── OOM Kill ─────────────────────────────────────────────────────────────
+
+    def _kill_heaviest_safe(self) -> str:
+        """
+        Finds the non-protected process using the most RAM and kills it.
+        Returns a string description of what was killed (or 'none').
+        """
+        candidates = []
+        for proc in psutil.process_iter(["pid", "name", "memory_percent"]):
+            try:
+                info = proc.info
+                pid  = info["pid"]
+                name = (info["name"] or "").lower()
+                mem  = info["memory_percent"] or 0.0
+
+                if pid == self.own_pid:
+                    continue
+                if any(name.startswith(p) for p in PROTECTED_PROCESS_NAMES):
+                    continue
+                candidates.append((mem, pid, name))
+            except (psutil.NoSuchProcess, psutil.AccessDenied):
+                pass
+
+        if not candidates:
+            logger.warning("[Watchdog] No safe kill candidates found — RAM is used by protected processes.")
+            return "none"
+
+        candidates.sort(reverse=True)
+        mem_pct, pid, name = candidates[0]
+        try:
+            psutil.Process(pid).kill()
+            logger.warning("[Watchdog] Killed '%s' (PID %d, %.1f%% RAM) for OOM defence.", name, pid, mem_pct)
+            return f"{name}:{pid}"
+        except Exception as e:
+            logger.error(f"[Watchdog] Failed to kill PID {pid}: {e}")
+            return "kill_failed"
+
+    # ── CPU Spike (Idle Prevention) ───────────────────────────────────────────
+
+    async def _cpu_spike(self):
+        """
+        Runs a CPU-bound task in an executor so it doesn't block the event loop.
+        Uses a 1-million-iteration sum — takes ~50 ms on 2 vCPUs.
+        Just enough to reset HF's idle detector without burning quota.
+        """
+        loop = asyncio.get_event_loop()
+        await loop.run_in_executor(None, lambda: sum(i * i for i in range(1_000_000)))
+        logger.debug("[Watchdog] CPU spike complete.")

swarm_llm.py

@@ -0,0 +1,208 @@
+# -*- coding: utf-8 -*-
+"""
+swarm_llm.py — Local CPU LLM Swarm (Qwen2.5-1.5B-Instruct, Q4_K_M)
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Why a local model?
+  The NIM API is rate-limited (tokens/minute). Every small sub-task
+  (JSON formatting, log summarization, brainstorm question generation,
+  Bell Curve trimming) that goes to NIM wastes quota needed for coding.
+
+This module runs Qwen2.5-1.5B-Instruct at Q4_K_M quantization:
+  - RAM:   ~1.1 GB (leaves 14 GB free for FastAPI + data)
+  - Speed: ~45 tok/s on 2 vCPUs (good enough for short tasks)
+  - Model: downloaded from HF Hub on first run → cached in /tmp/models/
+
+NIM is used ONLY for heavy coding tasks (forge_execute).
+SwarmLLM handles everything else.
+
+If llama-cpp-python is not installed (e.g. first boot before pip):
+  the module silently degrades and returns a FALLBACK_STUB response,
+  so the rest of the system still works.
+"""
+
+import os
+import logging
+import asyncio
+import time
+from typing import Optional
+
+logger = logging.getLogger("swarm_llm")
+
+MODEL_CACHE_DIR  = os.environ.get("MODEL_CACHE_DIR", "/tmp/models")
+MODEL_REPO       = "Qwen/Qwen2.5-1.5B-Instruct-GGUF"
+MODEL_FILENAME   = "qwen2.5-1.5b-instruct-q4_k_m.gguf"
+N_CTX            = 2048   # context window — matches our Bell Curve budget
+N_THREADS        = int(os.environ.get("SWARM_THREADS", "2"))
+MAX_TOKENS       = int(os.environ.get("SWARM_MAX_TOKENS", "256"))
+ENABLE_SWARM     = os.environ.get("ENABLE_SWARM_LLM", "true").lower() == "true"
+
+_llm = None          # loaded lazily on first call
+_llm_lock = None     # asyncio.Lock initialised at first call
+
+
+def _get_lock():
+    global _llm_lock
+    if _llm_lock is None:
+        _llm_lock = asyncio.Lock()
+    return _llm_lock
+
+
+def _load_model() -> Optional[object]:
+    """Download (if needed) and load the GGUF model. Blocking — call in executor."""
+    global _llm
+    if _llm is not None:
+        return _llm
+    if not ENABLE_SWARM:
+        logger.info("[SwarmLLM] Disabled via ENABLE_SWARM_LLM=false")
+        return None
+    try:
+        from llama_cpp import Llama
+    except ImportError:
+        logger.warning("[SwarmLLM] llama-cpp-python not installed. Running in stub mode.")
+        return None
+
+    model_path = os.path.join(MODEL_CACHE_DIR, MODEL_FILENAME)
+    if not os.path.exists(model_path):
+        logger.info(f"[SwarmLLM] Downloading {MODEL_FILENAME} from HF Hub…")
+        try:
+            from huggingface_hub import hf_hub_download
+            os.makedirs(MODEL_CACHE_DIR, exist_ok=True)
+            model_path = hf_hub_download(
+                repo_id=MODEL_REPO,
+                filename=MODEL_FILENAME,
+                local_dir=MODEL_CACHE_DIR,
+                local_dir_use_symlinks=False,
+            )
+            logger.info(f"[SwarmLLM] Downloaded → {model_path}")
+        except Exception as e:
+            logger.error(f"[SwarmLLM] Download failed: {e}")
+            return None
+
+    logger.info(f"[SwarmLLM] Loading model (n_ctx={N_CTX}, threads={N_THREADS}, type_k=8 (Q8_0), type_v=8 (Q8_0))…")
+    t0 = time.time()
+    try:
+        _llm = Llama(
+            model_path=model_path,
+            n_ctx=N_CTX,
+            n_threads=N_THREADS,
+            n_gpu_layers=0,      # CPU only — HF free spaces have no GPU
+            verbose=False,
+            chat_format="chatml",
+            type_k=8,            # 8-bit quantization for Key Cache (Turbo Quant)
+            type_v=8,            # 8-bit quantization for Value Cache (Turbo Quant)
+        )
+        logger.info(f"[SwarmLLM] Model loaded in {time.time()-t0:.1f}s")
+        return _llm
+    except Exception as e:
+        logger.error(f"[SwarmLLM] Failed to load model: {e}")
+        return None
+
+
+class SwarmLLM:
+    """
+    Async wrapper around the local Qwen model.
+    All inference runs in a thread executor so the FastAPI event loop
+    is never blocked.
+    """
+
+    def __init__(self):
+        self._ready = False
+
+    async def warm_up(self):
+        """Pre-load the model at startup so first inference is instant."""
+        loop = asyncio.get_event_loop()
+        model = await loop.run_in_executor(None, _load_model)
+        self._ready = model is not None
+        if self._ready:
+            logger.info("[SwarmLLM] Warm-up complete. Ready for inference.")
+        else:
+            logger.warning("[SwarmLLM] Running in stub mode (model not available).")
+
+    async def infer(self, prompt: str, system: str = "", max_tokens: int = MAX_TOKENS) -> str:
+        """
+        Run inference on the local model. Returns the generated text.
+        Falls back to a stub if model is not loaded.
+        """
+        if not self._ready:
+            return self._stub(prompt)
+
+        lock = _get_lock()
+        async with lock:
+            loop = asyncio.get_event_loop()
+            result = await loop.run_in_executor(
+                None,
+                lambda: self._sync_infer(prompt, system, max_tokens),
+            )
+        return result
+
+    def _sync_infer(self, prompt: str, system: str, max_tokens: int) -> str:
+        global _llm
+        if _llm is None:
+            return self._stub(prompt)
+        try:
+            messages = []
+            if system:
+                messages.append({"role": "system", "content": system})
+            messages.append({"role": "user", "content": prompt})
+
+            response = _llm.create_chat_completion(
+                messages=messages,
+                max_tokens=max_tokens,
+                temperature=0.3,
+                stop=["<|im_end|>", "</s>"],
+            )
+            text = response["choices"][0]["message"]["content"].strip()
+            logger.debug("[SwarmLLM] Infer complete: %d chars", len(text))
+            return text
+        except Exception as e:
+            logger.error(f"[SwarmLLM] Inference error: {e}")
+            return self._stub(prompt)
+
+    @staticmethod
+    def _stub(prompt: str) -> str:
+        """Fallback when model is unavailable — returns a safe placeholder."""
+        return "[SwarmLLM unavailable — NIM will handle this task]"
+
+    # ── High-Level Task Shortcuts ──────────────────────────────────────────────
+
+    async def summarize(self, text: str, max_words: int = 80) -> str:
+        """Summarise a long text into ≤ max_words words. Used to enforce Bell Curve budget."""
+        prompt = (
+            f"Summarise the following in ≤ {max_words} words. "
+            f"Be dense with information. No filler sentences.\n\n{text[:3000]}"
+        )
+        return await self.infer(prompt, system="You are a precise technical summariser.")
+
+    async def format_json(self, raw: str) -> str:
+        """Extract and clean a JSON object from a messy LLM response."""
+        prompt = (
+            "Extract the JSON object from the following text. "
+            "Return ONLY the JSON, no markdown fences, no explanation.\n\n" + raw[:2000]
+        )
+        return await self.infer(prompt, system="You are a JSON extractor. Output only valid JSON.")
+
+    async def generate_brainstorm_questions(self, goal: str, n: int = 5) -> list:
+        """Generate n 'What if…' brainstorm questions for the hourly swarm cycle."""
+        prompt = (
+            f"Generate exactly {n} creative 'What if…' questions to improve: '{goal}'. "
+            f"Each question should be a novel technical idea. "
+            f"Format: one question per line, no numbering."
+        )
+        raw = await self.infer(prompt, system="You are a creative technical brainstormer.", max_tokens=400)
+        questions = [q.strip() for q in raw.strip().splitlines() if q.strip()]
+        return questions[:n]
+
+    async def smart_trim(self, text: str, char_limit: int) -> str:
+        """
+        If text exceeds char_limit, ask the local model to summarise it
+        to fit. Better than hard-cutting. Used in Bell Curve budget enforcement.
+        """
+        if len(text) <= char_limit:
+            return text
+        target_words = char_limit // 6  # rough chars-to-words ratio
+        summary = await self.summarize(text, max_words=target_words)
+        return summary
+
+
+# Singleton — import and use directly
+swarm = SwarmLLM()