aiq_bridge.py

"""
NVIDIA AI-Q Blueprint Embedded Bridge for Fiber-MoE Autonomous World Model
Integrates `/aiq-research` and `/aiq-deploy` directly into the agent substrate.
"""

import os
import sys
import subprocess
import json
from pathlib import Path

AIQ_SKILL_DIR = Path(r"C:\Users\gemin\.gemini\config\skills\aiq-research")
AIQ_SCRIPT = AIQ_SKILL_DIR / "scripts" / "aiq.py"
DEFAULT_AIQ_SERVER_URL = os.environ.get("AIQ_SERVER_URL", "http://localhost:8000")

class AIQBridge:
    def __init__(self, server_url: str = DEFAULT_AIQ_SERVER_URL):
        self.server_url = server_url
        self.script_path = str(AIQ_SCRIPT)

    def check_health(self) -> dict:
        """Ping AI-Q server /health endpoint."""
        cmd = [sys.executable, self.script_path, "health"]
        env = os.environ.copy()
        env["AIQ_SERVER_URL"] = self.server_url
        try:
            res = subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=10)
            if res.returncode == 0:
                try:
                    return {"status": "healthy", "details": json.loads(res.stdout.strip())}
                except Exception:
                    return {"status": "healthy", "raw": res.stdout.strip()}
            return {"status": "unreachable", "error": res.stderr.strip() or res.stdout.strip()}
        except Exception as e:
            return {"status": "error", "error": str(e)}

    def execute_chat(self, query: str) -> dict:
        """Route shallow query or dispatch async deep research job."""
        cmd = [sys.executable, self.script_path, "chat", query]
        env = os.environ.copy()
        env["AIQ_SERVER_URL"] = self.server_url
        res = subprocess.run(cmd, capture_output=True, text=True, env=env)
        try:
            return json.loads(res.stdout.strip())
        except Exception:
            return {"raw_output": res.stdout.strip(), "stderr": res.stderr.strip()}

    def poll_research(self, job_id: str) -> dict:
        """Poll deep research job until completion."""
        cmd = [sys.executable, self.script_path, "research_poll", job_id]
        env = os.environ.copy()
        env["AIQ_SERVER_URL"] = self.server_url
        res = subprocess.run(cmd, capture_output=True, text=True, env=env)
        try:
            return json.loads(res.stdout.strip())
        except Exception:
            return {"raw_output": res.stdout.strip()}

if __name__ == "__main__":
    bridge = AIQBridge()
    print("Testing AIQBridge health check:")
    health = bridge.check_health()
    print(json.dumps(health, indent=2))