Implement MultiProviderRateLimiter, eternity state machines and Forge execution agent endpoints

backend.py

@@ -95,6 +95,44 @@ nim_client = AsyncOpenAI(
     api_key=NIM_API_KEY,
 )
 
+# ---------------------------------------------------------------------------
+# Rate Limiting & Multi-Provider Setup
+# ---------------------------------------------------------------------------
+
+MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY", "")
+mistral_client = AsyncOpenAI(
+    base_url="https://api.mistral.ai/v1",
+    api_key=MISTRAL_API_KEY if MISTRAL_API_KEY else "dummy_key",
+) if MISTRAL_API_KEY else None
+
+class MultiProviderRateLimiter:
+    def __init__(self):
+        self.nim_limit = 40
+        self.nim_window = 60
+        self.nim_calls = []
+        self.mistral_last_call = 0.0
+        self.lock = asyncio.Lock()
+
+    async def wait_for_mistral(self):
+        async with self.lock:
+            now = time.time()
+            elapsed = now - self.mistral_last_call
+            if elapsed < 1.0:
+                await asyncio.sleep(1.0 - elapsed)
+            self.mistral_last_call = time.time()
+
+    async def wait_for_nim(self):
+        async with self.lock:
+            now = time.time()
+            self.nim_calls = [t for t in self.nim_calls if now - t < self.nim_window]
+            if len(self.nim_calls) >= self.nim_limit - 2:
+                sleep_time = self.nim_window - (now - self.nim_calls[0])
+                print(f"[RateLimiter] Approaching NIM rate limit (40 RPM). Sleeping {sleep_time:.2f}s...")
+                await asyncio.sleep(sleep_time)
+            self.nim_calls.append(time.time())
+
+rate_limiter = MultiProviderRateLimiter()
+
 # ---------------------------------------------------------------------------
 # Tool Definitions (OpenAI function calling format)
 # ---------------------------------------------------------------------------
@@ -397,6 +435,16 @@ async def init_db():
                 );
                 CREATE INDEX IF NOT EXISTS idx_session_key ON agent_session_entries (project_key, session_id, subpath, id);
                 CREATE INDEX IF NOT EXISTS idx_project_session ON agent_session_entries (project_key, session_id);
+                
+                CREATE TABLE IF NOT EXISTS eternity_system_state (
+                    id           SERIAL PRIMARY KEY,
+                    goal         TEXT NOT NULL,
+                    deadline     TIMESTAMPTZ NOT NULL,
+                    current_mode VARCHAR(20) NOT NULL DEFAULT 'build',
+                    roadmap      JSONB DEFAULT '[]',
+                    latest_brief TEXT,
+                    created_at   TIMESTAMPTZ NOT NULL DEFAULT now()
+                );
             """)
     except Exception as e:
         print(f"[DB] Warning: Could not initialize database: {e}")
@@ -1355,24 +1403,324 @@ async def delete_session(req: SessionDeleteRequest, authorization: str = Header(
         raise HTTPException(status_code=500, detail=str(e))
 
 
+from datetime import datetime, timedelta, timezone
+
+class EternityInitRequest(BaseModel):
+    goal: str
+    deadline_hours: int
+
+@app.post("/api/eternity/init")
+async def init_eternity_system(req: EternityInitRequest, authorization: str = Header(None)):
+    auth(authorization)
+    if not db_pool:
+        raise HTTPException(status_code=500, detail="Database not connected")
+    try:
+        deadline = datetime.now(timezone.utc) + timedelta(hours=req.deadline_hours)
+        async with db_pool.acquire() as conn:
+            # Delete any old state
+            await conn.execute("DELETE FROM eternity_system_state")
+            # Insert new state
+            await conn.execute(
+                "INSERT INTO eternity_system_state (goal, deadline, current_mode, roadmap) VALUES ($1, $2, 'build', $3)",
+                req.goal, deadline, "[]"
+            )
+        log_activity(f"[Eternity Loop] System initialized with goal: '{req.goal}' | Deadline: {deadline}")
+        return {"status": "success", "deadline": deadline.isoformat()}
+    except Exception as e:
+        log_activity(f"[Eternity Loop Error] Init failed: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/api/eternity/status")
+async def get_eternity_status(authorization: str = Header(None)):
+    auth(authorization)
+    if not db_pool:
+        raise HTTPException(status_code=500, detail="Database not connected")
+    try:
+        async with db_pool.acquire() as conn:
+            row = await conn.fetchrow("SELECT goal, deadline, current_mode, roadmap, latest_brief FROM eternity_system_state ORDER BY id DESC LIMIT 1")
+            if not row:
+                return {"active": False}
+                
+            deadline = row["deadline"]
+            now = datetime.now(timezone.utc)
+            remaining = max(0.0, (deadline - now).total_seconds())
+            
+            return {
+                "active": True,
+                "goal": row["goal"],
+                "deadline": deadline.isoformat(),
+                "current_mode": row["current_mode"],
+                "roadmap": json.loads(row["roadmap"]) if isinstance(row["roadmap"], str) else row["roadmap"],
+                "latest_brief": row["latest_brief"],
+                "time_remaining_seconds": remaining
+            }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+class ForgeExecuteRequest(BaseModel):
+    task_id: str
+    action: str
+    prompt: str
+    context_rules: Optional[str] = ""
+
+@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}'")
+    try:
+        messages = [
+            {"role": "system", "content": AGENTIC_SYSTEM_PROMPT + f"\nContext rules: {req.context_rules}"},
+            {"role": "user", "content": req.prompt}
+        ]
+        
+        final_messages = list(messages)
+        success = False
+        summary = ""
+        error = None
+        
+        for round_num in range(MAX_TOOL_ROUNDS + 1):
+            await rate_limiter.wait_for_nim()
+            response = await nim_client.chat.completions.create(
+                model=RECOMMENDED_MODEL,
+                messages=final_messages,
+                tools=TOOLS,
+                tool_choice="auto"
+            )
+            
+            msg = response.choices[0].message
+            tool_calls_payload = None
+            if msg.tool_calls:
+                tool_calls_payload = []
+                for tc in msg.tool_calls:
+                    tc_dict = {
+                        "id": tc.id,
+                        "type": tc.type,
+                        "function": {
+                            "name": tc.function.name,
+                            "arguments": tc.function.arguments
+                        }
+                    }
+                    tool_calls_payload.append(tc_dict)
+                    
+            final_messages.append({
+                "role": "assistant",
+                "content": msg.content,
+                "tool_calls": tool_calls_payload
+            })
+            
+            if not msg.tool_calls:
+                success = True
+                summary = msg.content or "Task completed."
+                break
+                
+            for tc in msg.tool_calls:
+                func_name = tc.function.name
+                func_args = json.loads(tc.function.arguments)
+                result = await execute_tool(func_name, func_args)
+                
+                final_messages.append({
+                    "role": "tool",
+                    "tool_call_id": tc.id,
+                    "content": result
+                })
+        else:
+            error = "Max tool call rounds exceeded."
+            
+        if success:
+            return {
+                "status": "success",
+                "summary": summary,
+                "error": None
+            }
+        else:
+            return {
+                "status": "error",
+                "error": error or "Task execution failed."
+            }
+            
+    except Exception as e:
+        log_activity(f"[Forge Error] Task execution failed: {e}")
+        return {
+            "status": "error",
+            "error": str(e)
+        }
+
+
 # ---------------------------------------------------------------------------
 # Database Retention & Heartbeat loops
 # ---------------------------------------------------------------------------
 
-async def db_cleanup_loop():
-    log_activity("Database Retention Cleanup task started")
+async def db_heartbeat_loop():
+    log_activity("Database Heartbeat task started")
     while True:
         try:
             if db_pool:
                 async with db_pool.acquire() as conn:
-                    # Purge session entries older than 30 days
-                    result = await conn.execute(
-                        "DELETE FROM agent_session_entries WHERE created_at < NOW() - INTERVAL '30 days'"
-                    )
-                    log_activity(f"[Cleanup] Nightly retention sweep complete. Status: {result}")
+                    await conn.execute("SELECT 1")
+                    log_activity("[Heartbeat] Pinged Aiven PostgreSQL successfully")
         except Exception as e:
-            log_activity(f"[Cleanup Warning] Failed to run retention cleanup: {e}")
-        await asyncio.sleep(86400) # Every 24 hours
+            log_activity(f"[Heartbeat Warning] Failed to ping database: {e}")
+        await asyncio.sleep(240) # Every 4 minutes
+
+
+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")
+SPACE6_URL = os.environ.get("SPACE6_URL", "https://augment17-mcp-cloud-host.hf.space")
+
+async def get_db_state():
+    async with db_pool.acquire() as conn:
+        return await conn.fetchrow("SELECT goal, deadline, current_mode FROM eternity_system_state ORDER BY id DESC LIMIT 1")
+
+async def update_db_mode(mode: str):
+    async with db_pool.acquire() as conn:
+        await conn.execute("UPDATE eternity_system_state SET current_mode = $1", mode)
+
+async def update_db_brief(brief: str):
+    async with db_pool.acquire() as conn:
+        await conn.execute("UPDATE eternity_system_state SET latest_brief = $1", brief)
+
+async def update_db_roadmap(roadmap: list):
+    async with db_pool.acquire() as conn:
+        await conn.execute("UPDATE eternity_system_state SET roadmap = $1", json.dumps(roadmap))
+
+def post_json(url: str, payload: dict) -> 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'))
+    except Exception as e:
+        print(f"[HTTP Error] POST to {url} failed: {e}")
+        return {"status": "error", "error": str(e)}
+
+async def execute_build_cycle(goal: str):
+    log_activity(f"[Build Mode] Initiating build cycle for 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"}'
+    try:
+        res = await nim_client.chat.completions.create(
+            model="nvidia/llama-3.1-nemotron-70b-instruct",
+            messages=[{"role": "user", "content": prompt}],
+            max_tokens=300
+        )
+        task = json.loads(res.choices[0].message.content.strip())
+        task_prompt = task.get("prompt")
+        context_rules = task.get("context_rules", "")
+    except Exception as e:
+        log_activity(f"[Build Mode Error] NIM planning failed: {e}")
+        return
+
+    log_activity(f"[Build Mode] Dispatching task to Space 3: '{task_prompt}'")
+    forge_res = post_json(f"{SPACE3_URL}/api/forge/execute", {
+        "task_id": f"build_{int(time.time())}",
+        "action": "execute_code",
+        "prompt": task_prompt,
+        "context_rules": context_rules
+    })
+    
+    if forge_res.get("status") == "success":
+        log_activity(f"[Build Mode] Space 3 success: {forge_res.get('summary')}")
+        log_activity("[Build Mode] Triggering Space 6 (The Sandbox) UI verification...")
+        test_res = post_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", {})
+    else:
+        log_activity(f"[Build Mode Warning] Space 3 reported failure: {forge_res.get('error')}")
+
+async def execute_eternity_cycle(goal: str):
+    log_activity(f"[Eternity Mode] Initiating autonomous R&D cycle for goal: '{goal}'")
+    log_activity("[Eternity Mode] Querying Space 4 (The Library) for new feature research...")
+    research_res = post_json(f"{SPACE4_URL}/api/research", {"query": f"novel scientific or chemical calculation features and algorithms for {goal}"})
+    brief = research_res.get("brief", "No new features found.")
+    await update_db_brief(brief)
+    log_activity(f"[Eternity Mode] Received research brief: {brief[:100]}...")
+
+    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"}'
+    try:
+        res = await nim_client.chat.completions.create(
+            model="nvidia/llama-3.1-nemotron-70b-instruct",
+            messages=[{"role": "user", "content": prompt}],
+            max_tokens=300
+        )
+        task = json.loads(res.choices[0].message.content.strip())
+        task_prompt = task.get("prompt")
+        context_rules = task.get("context_rules", "")
+    except Exception as e:
+        log_activity(f"[Eternity Mode Error] NIM planning failed: {e}")
+        return
+
+    log_activity(f"[Eternity Mode] Dispatching task to Space 3: '{task_prompt}'")
+    forge_res = post_json(f"{SPACE3_URL}/api/forge/execute", {
+        "task_id": f"eternity_{int(time.time())}",
+        "action": "execute_code",
+        "prompt": task_prompt,
+        "context_rules": context_rules
+    })
+
+    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", {})
+
+
+def run_eternity_loop():
+    log_activity("[Eternity Loop] Daemon thread started.")
+    while True:
+        try:
+            if not db_pool:
+                time.sleep(10)
+                continue
+                
+            loop = asyncio.new_event_loop()
+            state = loop.run_until_complete(get_db_state())
+            loop.close()
+            
+            if not state:
+                time.sleep(30)
+                continue
+                
+            goal = state["goal"]
+            deadline = state["deadline"]
+            current_mode = state["current_mode"]
+            
+            now = datetime.now(timezone.utc)
+            if current_mode == "build" and now >= deadline:
+                loop = asyncio.new_event_loop()
+                loop.run_until_complete(update_db_mode("eternity"))
+                loop.close()
+                current_mode = "eternity"
+                log_activity("[Eternity Loop] Deadline reached. Transitioned to Eternity R&D Mode.")
+            
+            if current_mode == "build":
+                loop = asyncio.new_event_loop()
+                loop.run_until_complete(execute_build_cycle(goal))
+                loop.close()
+                time.sleep(300)
+            else:
+                loop = asyncio.new_event_loop()
+                loop.run_until_complete(execute_eternity_cycle(goal))
+                loop.close()
+                time.sleep(43200) # 12 hours
+                
+        except Exception as e:
+            log_activity(f"[Eternity Loop Error] Loop crash: {e}")
+            time.sleep(60)
 
 
 @app.get("/", response_class=HTMLResponse)
@@ -1596,12 +1944,29 @@ def run_backup_loop():
             log_activity(f"[Backup Error] {e}")
 
 
+async def db_cleanup_loop():
+    log_activity("Database Retention Cleanup task started")
+    while True:
+        try:
+            if db_pool:
+                async with db_pool.acquire() as conn:
+                    result = await conn.execute(
+                        "DELETE FROM agent_session_entries WHERE created_at < NOW() - INTERVAL '30 days'"
+                    )
+                    log_activity(f"[Cleanup] Nightly retention sweep complete. Status: {result}")
+        except Exception as e:
+            log_activity(f"[Cleanup Warning] Failed to run retention cleanup: {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()
     # Start the local backup loop thread
     threading.Thread(target=run_backup_loop, daemon=True).start()
+    # Start the eternity R&D loop thread
+    threading.Thread(target=run_eternity_loop, daemon=True).start()
     # Start the db keep-alive loop on FastAPI event loop
     asyncio.create_task(db_heartbeat_loop())
     # Start the db nightly retention cleanup loop