feat: async httpx inter-space calls, Karpathy project_state.md, vault push, idle worker protocol

backend.py

@@ -1748,14 +1748,20 @@ class ForgeExecuteRequest(BaseModel):
     action: str
     prompt: str
     context_rules: Optional[str] = ""
+    project_state_md: Optional[str] = ""  # Karpathy CLAUDE.md protocol — shared project context
 
 @app.post("/api/forge/execute")
 async def forge_execute(req: ForgeExecuteRequest, authorization: str = Header(None)):
     auth(authorization)
-    log_activity(f"[Forge] Received execution request for task {req.task_id}: '{req.prompt}'")
+    log_activity(f"[Forge] Received execution request for task {req.task_id}: '{req.prompt[:80]}'")
     try:
+        # Inject project_state.md into system context if provided (Karpathy protocol)
+        system_content = AGENTIC_SYSTEM_PROMPT + f"\nContext rules: {req.context_rules}"
+        if req.project_state_md:
+            system_content += f"\n\n---\n## Project State Document (read this first)\n{req.project_state_md}"
+
         messages = [
-            {"role": "system", "content": AGENTIC_SYSTEM_PROMPT + f"\nContext rules: {req.context_rules}"},
+            {"role": "system", "content": system_content},
             {"role": "user", "content": req.prompt}
         ]
         
@@ -1861,9 +1867,20 @@ async def db_heartbeat_loop():
         await asyncio.sleep(240) # Every 4 minutes
 
 
+import httpx
+
+# ---------------------------------------------------------------------------
+# Inter-Space URLs — The 6-Space Topology
+# Space 2 (self)  = shyota/claude-code-backend   — Cerebrum (Orchestrator)
+# Space 3         = augment17/claude-code-backend — Forge (Coder)
+# Space 4         = shyota/mcp-cloud-host         — Library (Research + MCP)
+# Space 5 (Vault) = Space 3 also handles vault push since agent-worker-loop
+#                   HF slot is not provisioned. SPACE5_URL = SPACE3_URL.
+# Space 6         = augment17/mcp-cloud-host      — Sandbox (UI Tester)
+# ---------------------------------------------------------------------------
 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", "https://augment17-better-chatbot.hf.space")
+SPACE5_URL = os.environ.get("SPACE5_URL", SPACE3_URL)  # Vault is co-hosted on Space 3
 SPACE6_URL = os.environ.get("SPACE6_URL", "https://augment17-mcp-cloud-host.hf.space")
 
 async def get_active_projects():
@@ -1882,117 +1899,232 @@ async def update_db_roadmap(project_name: str, roadmap: list):
     async with db_pool.acquire() as conn:
         await conn.execute("UPDATE eternity_projects SET roadmap = $1 WHERE project_name = $2", json.dumps(roadmap), project_name)
 
-def post_json(url: str, payload: dict) -> dict:
+# Async HTTP POST — replaces blocking urllib.request so the event loop never freezes
+async def apost_json(url: str, payload: dict, timeout: float = 120.0) -> dict:
     headers = {
         "Authorization": f"Bearer {BACKEND_API_KEY}",
         "Content-Type": "application/json"
     }
-    req = urllib.request.Request(
-        url,
-        data=json.dumps(payload).encode('utf-8'),
-        headers=headers,
-        method="POST"
-    )
     try:
-        with urllib.request.urlopen(req, timeout=120) as r:
-            return json.loads(r.read().decode('utf-8'))
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            r = await client.post(url, json=payload, headers=headers)
+            r.raise_for_status()
+            return r.json()
     except Exception as e:
-        print(f"[HTTP Error] POST to {url} failed: {e}")
+        log_activity(f"[HTTP Error] POST to {url} failed: {e}")
         return {"status": "error", "error": str(e)}
 
+# ---------------------------------------------------------------------------
+# Karpathy project_state.md — Shared Context Document
+# Each space receives this instead of bare prompts so agents always know
+# what phase they are in, what was built before, and what the research says.
+# ---------------------------------------------------------------------------
+def build_project_state_md(project_name: str, goal: str, mode: str,
+                            task_prompt: str, brief: str = "",
+                            prior_summary: str = "") -> str:
+    ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%MZ")
+    return f"""# Project State: {project_name}
+_Generated by Space 2 (Cerebrum) at {ts}_
+
+## Current Objective
+**Mode:** {mode.upper()}
+**Goal:** {goal}
+
+## Instructions for Space 3 (The Forge)
+{task_prompt}
+
+## Research Brief (from Space 4 — The Library)
+{brief if brief else '_No research brief yet._'}
+
+## Prior Work Summary
+{prior_summary if prior_summary else '_First cycle._'}
+
+## Context Rules
+- Write deterministic, testable code.
+- All files go in /tmp/workspace/.
+- Return a 3-line summary: status, file changed, test result.
+"""
+
 async def execute_build_cycle(project_name: str, goal: str):
     log_activity(f"[Build Mode] Initiating build cycle for project '{project_name}' (goal: '{goal}')")
     await rate_limiter.wait_for_nim()
-    prompt = f"We are building: '{goal}'. Write a JSON instruction for Space 3 (The Forge) to code the next milestone. Respond ONLY with JSON matching the contract: " + '{"prompt": "task description", "context_rules": "rules"}'
+    plan_prompt = (
+        f"We are building: '{goal}'. "
+        "Write ONE JSON instruction for Space 3 (The Forge) to code the next milestone. "
+        "Respond ONLY with JSON: {\"prompt\": \"task description\", \"context_rules\": \"rules\"}"
+    )
     try:
         res = await nim_client.chat.completions.create(
             model="nvidia/llama-3.1-nemotron-70b-instruct",
-            messages=[{"role": "user", "content": prompt}],
+            messages=[{"role": "user", "content": plan_prompt}],
             max_tokens=300
         )
-        task = json.loads(res.choices[0].message.content.strip())
-        task_prompt = task.get("prompt")
+        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)
+        task_prompt = task.get("prompt", "")
         context_rules = task.get("context_rules", "")
     except Exception as e:
         log_activity(f"[Build Mode Error] NIM planning failed for project '{project_name}': {e}")
         return
 
-    log_activity(f"[Build Mode] Dispatching project '{project_name}' task to Space 3: '{task_prompt}'")
-    forge_res = post_json(f"{SPACE3_URL}/api/forge/execute", {
+    # 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
+    )
+
+    log_activity(f"[Build Mode] Dispatching to Space 3 (Forge) — project '{project_name}': '{task_prompt[:80]}'")
+    forge_res = await apost_json(f"{SPACE3_URL}/api/forge/execute", {
         "task_id": f"build_{project_name}_{int(time.time())}",
         "action": "execute_code",
         "prompt": task_prompt,
-        "context_rules": context_rules
+        "context_rules": context_rules,
+        "project_state_md": state_md  # Karpathy CLAUDE.md protocol
     })
-    
+
     if forge_res.get("status") == "success":
-        log_activity(f"[Build Mode] Space 3 success for project '{project_name}': {forge_res.get('summary')}")
-        log_activity(f"[Build Mode] Triggering Space 6 (The Sandbox) UI verification for project '{project_name}'...")
-        test_res = post_json(f"{SPACE6_URL}/api/sandbox/test", {
+        summary = forge_res.get("summary", "")
+        log_activity(f"[Build Mode] Space 3 (Forge) SUCCESS for '{project_name}': {summary}")
+
+        log_activity(f"[Build Mode] Triggering Space 6 (Sandbox) UI verification for '{project_name}'...")
+        test_res = await apost_json(f"{SPACE6_URL}/api/sandbox/test", {
             "test_cmd": "verify_ui",
-            "url": f"{SPACE3_URL}"
-        })
-        log_activity(f"[Build Mode] Space 6 Test Verdict: {test_res.get('verdict')} | Reason: {test_res.get('reason')}")
-        
-        log_activity("[Build Mode] Triggering Space 5 (The Vault) backup commit...")
-        post_json(f"{SPACE5_URL}/api/vault/push", {})
+            "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}")
+
+        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')}")
     else:
-        log_activity(f"[Build Mode Warning] Space 3 reported failure: {forge_res.get('error')}")
+        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] Initiating autonomous R&D cycle for project '{project_name}' (goal: '{goal}')")
-    log_activity(f"[Eternity Mode] Querying Space 4 (The Library) for research on project '{project_name}'...")
-    research_res = post_json(f"{SPACE4_URL}/api/research", {"query": f"novel algorithms and components for {goal}"})
+    log_activity(f"[Eternity Mode] R&D cycle starting for project '{project_name}' (goal: '{goal}')")
+
+    # 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", {
+        "query": f"novel algorithms, optimizations, and next features for: {goal}"
+    })
     brief = research_res.get("brief", "No new features found.")
     await update_db_brief(project_name, brief)
-    log_activity(f"[Eternity Mode] Received research brief for '{project_name}': {brief[:100]}...")
+    log_activity(f"[Eternity Mode] Space 4 (Library) brief received ({len(brief)} chars)")
 
+    # Step 2: Ask NIM to plan next feature based on research
     await rate_limiter.wait_for_nim()
-    prompt = f"Goal: '{goal}'. Research Brief: '{brief}'. Plan the next feature/optimization code. Respond ONLY with JSON: " + '{"prompt": "task description", "context_rules": "rules"}'
+    plan_prompt = (
+        f"Goal: '{goal}'.\nResearch Brief:\n{brief[:2000]}\n\n"
+        "Based on this research, plan the SINGLE most impactful next feature to implement. "
+        "Respond ONLY with JSON: {\"prompt\": \"task\", \"context_rules\": \"rules\"}"
+    )
     try:
         res = await nim_client.chat.completions.create(
             model="nvidia/llama-3.1-nemotron-70b-instruct",
-            messages=[{"role": "user", "content": prompt}],
+            messages=[{"role": "user", "content": plan_prompt}],
             max_tokens=300
         )
-        task = json.loads(res.choices[0].message.content.strip())
-        task_prompt = task.get("prompt")
+        raw = res.choices[0].message.content.strip()
+        if raw.startswith("```"):
+            raw = raw.split("```")[1].lstrip("json").strip()
+        task = json.loads(raw)
+        task_prompt = task.get("prompt", "")
         context_rules = task.get("context_rules", "")
     except Exception as e:
-        log_activity(f"[Eternity Mode Error] NIM planning failed for project '{project_name}': {e}")
+        log_activity(f"[Eternity Mode Error] NIM planning failed for '{project_name}': {e}")
         return
 
-    log_activity(f"[Eternity Mode] Dispatching task to Space 3: '{task_prompt}'")
-    forge_res = post_json(f"{SPACE3_URL}/api/forge/execute", {
+    # 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 4: Dispatch to Space 3 (Forge)
+    log_activity(f"[Eternity Mode] Dispatching to Space 3 (Forge): '{task_prompt[:80]}'")
+    forge_res = await apost_json(f"{SPACE3_URL}/api/forge/execute", {
         "task_id": f"eternity_{project_name}_{int(time.time())}",
         "action": "execute_code",
         "prompt": task_prompt,
-        "context_rules": context_rules
+        "context_rules": context_rules,
+        "project_state_md": state_md
     })
 
     if forge_res.get("status") == "success":
-        log_activity(f"[Eternity Mode] Space 3 success: {forge_res.get('summary')}")
-        log_activity("[Eternity Mode] Triggering Space 5 (The Vault) backup commit...")
-        post_json(f"{SPACE5_URL}/api/vault/push", {})
-
-
-last_run_times = {} # project_name -> timestamp
+        summary = forge_res.get("summary", "")
+        log_activity(f"[Eternity Mode] Space 3 (Forge) SUCCESS: {summary}")
+        # 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')}")
+    else:
+        log_activity(f"[Eternity Mode Warning] Space 3 (Forge) failure: {forge_res.get('error')}") 
+
+
+last_run_times = {}  # project_name -> last execution timestamp
+IDLE_TASK_INTERVAL = 1800  # Run idle tasks every 30 min when no active projects
+_last_idle_run = 0.0
+
+async def execute_idle_worker_cycle():
+    """Idle Worker Protocol — runs when no active projects exist.
+    Dispatches self-improvement tasks to Space 3 (Forge) to keep
+    all containers warm and utilise free compute."""
+    log_activity("[Idle Worker] No active projects. Dispatching background optimisation task to Space 3...")
+    idle_tasks = [
+        "Scan /tmp/workspace for any Python or JS files. For each file found, "
+        "check if a corresponding test file exists. If not, write one basic unit test and run it. "
+        "Return a summary of files scanned and tests added.",
+        "Scan /tmp/workspace for duplicate function definitions across files. "
+        "Report any found and refactor the worst one to a shared utility module.",
+        "Run a syntax check on all files in /tmp/workspace and fix any issues found.",
+    ]
+    import random
+    chosen = random.choice(idle_tasks)
+    state_md = build_project_state_md(
+        project_name="idle-maintenance", goal="system self-improvement",
+        mode="idle", task_prompt=chosen
+    )
+    res = await apost_json(f"{SPACE3_URL}/api/forge/execute", {
+        "task_id": f"idle_{int(time.time())}",
+        "action": "execute_code",
+        "prompt": chosen,
+        "context_rules": "Be concise. Fix only what is necessary. Return a 3-line summary.",
+        "project_state_md": state_md
+    }, timeout=60.0)
+    log_activity(f"[Idle Worker] Space 3 response: {res.get('status')} — {str(res.get('summary', ''))[:120]}")
 
 async def run_eternity_loop():
-    log_activity("[Eternity Loop] Async task started on main loop.")
+    global _last_idle_run
+    log_activity("[Eternity Loop] Async orchestration task started on FastAPI main loop.")
     loop_counter = 0
     while True:
         try:
             if not db_pool:
                 await asyncio.sleep(10)
                 continue
-                
+
             projects = await get_active_projects()
-            
+
             if not projects:
-                await asyncio.sleep(30)
+                # --- Idle Worker Protocol ---
+                now_ts = time.time()
+                if now_ts - _last_idle_run >= IDLE_TASK_INTERVAL:
+                    _last_idle_run = now_ts
+                    await execute_idle_worker_cycle()
+                await asyncio.sleep(60)
                 continue
-                
+
             loop_counter += 1
             for p in projects:
                 name = p["project_name"]
@@ -2000,35 +2132,36 @@ async def run_eternity_loop():
                 deadline = p["deadline"]
                 current_mode = p["current_mode"]
                 priority = p["priority"]
-                
-                # Priority gating: Low priority skips 3 out of 4 cycles to save token rate limits
+
+                # Priority gating: "low" skips 3 of every 4 cycles to conserve RPM
                 if priority == "low" and (loop_counter % 4) != 0:
+                    log_activity(f"[Eternity Loop] Skipping low-priority '{name}' this cycle ({loop_counter})")
                     continue
-                    
+
                 now = datetime.now(timezone.utc)
                 if current_mode == "build" and now >= deadline:
                     await update_db_mode(name, "eternity")
                     current_mode = "eternity"
-                    log_activity(f"[Eternity Loop] Project '{name}' deadline reached. Transitioned to Eternity R&D Mode.")
-                
+                    log_activity(f"[Eternity Loop] Project '{name}' deadline passed — transitioned to Eternity R&D Mode.")
+
                 if current_mode == "build":
                     await execute_build_cycle(name, goal)
                 else:
-                    # Enforce the sleep interval for Eternity Mode independently per project
                     last_run = last_run_times.get(name, 0.0)
                     now_ts = time.time()
                     interval = int(os.environ.get("ETERNITY_LOOP_INTERVAL", "3600"))
                     if now_ts - last_run < interval:
+                        remaining_min = int((interval - (now_ts - last_run)) / 60)
+                        log_activity(f"[Eternity Loop] Project '{name}' sleeping — next R&D in {remaining_min}m")
                         continue
                     last_run_times[name] = now_ts
-                    
                     await execute_eternity_cycle(name, goal)
-            
-            # Base check interval (5 minutes)
+
+            # Base check interval: 5 minutes
             await asyncio.sleep(300)
-            
+
         except Exception as e:
-            log_activity(f"[Eternity Loop Error] Loop crash: {e}")
+            log_activity(f"[Eternity Loop Error] Unhandled exception: {e}")
             await asyncio.sleep(60)
 
 
@@ -2037,6 +2170,42 @@ async def dashboard():
     return HTMLResponse(content=DASHBOARD_HTML)
 
 
+# ---------------------------------------------------------------------------
+# Vault Push Endpoint (Space 5 co-hosted here on Space 3)
+# Space 2 (Cerebrum) calls this after every successful build/eternity cycle.
+# It zips /tmp/workspace and logs the backup event.
+# A real git push would require SSH credentials injected via HF secrets.
+# ---------------------------------------------------------------------------
+class VaultPushRequest(BaseModel):
+    project_name: Optional[str] = "unknown"
+    summary: Optional[str] = ""
+
+@app.post("/api/vault/push")
+async def vault_push(req: VaultPushRequest, authorization: str = Header(None)):
+    auth(authorization)
+    workspace = "/tmp/workspace"
+    archive = "/tmp/vault_snapshot.zip"
+    try:
+        if os.path.exists(workspace) and os.listdir(workspace):
+            proc = await asyncio.create_subprocess_exec(
+                "zip", "-r", archive, workspace,
+                stdout=asyncio.subprocess.DEVNULL,
+                stderr=asyncio.subprocess.DEVNULL
+            )
+            await asyncio.wait_for(proc.communicate(), timeout=30)
+            size_kb = int(os.path.getsize(archive) / 1024) if os.path.exists(archive) else 0
+            log_activity(f"[Vault] Snapshot created for '{req.project_name}': {size_kb}KB | {req.summary[:80]}")
+            return {"status": "success", "archive_kb": size_kb, "project_name": req.project_name}
+        else:
+            log_activity(f"[Vault] Workspace empty — no snapshot for '{req.project_name}'")
+            return {"status": "skipped", "reason": "workspace empty"}
+    except Exception as e:
+        log_activity(f"[Vault Error] Snapshot failed for '{req.project_name}': {e}")
+        return {"status": "error", "error": str(e)}
+
+
+
+
 @app.get("/api/logs")
 async def get_logs():
     return list(activity_logs)

requirements.txt

@@ -4,4 +4,5 @@ openai==1.86.0
 asyncpg==0.30.0
 anyio==4.9.0
 psutil==5.9.8
+httpx==0.27.2