Agent communication via redis (#141)

* support https in the agent

* send all commands via redis

Author: mariusandra

backend/app/fastapi.py

@@ -1,7 +1,7 @@
 import asyncio
 import json
 import os
-from contextlib import asynccontextmanager, suppress
+from contextlib import asynccontextmanager
 from httpx import AsyncClient, Limits
 from fastapi.staticfiles import StaticFiles
 from fastapi.responses import JSONResponse, HTMLResponse
@@ -13,8 +13,6 @@
 from fastapi.middleware.gzip import GZipMiddleware
 from app.middleware import GzipRequestMiddleware
 from app.ws.agent_ws import router as agent_ws_router
-from app.ws.agent_tasks import start_background_listener
-from app.redis import create_redis_connection
 from app.websockets import register_ws_routes, redis_listener
 from app.config import config
 from app.utils.sentry import initialize_sentry
@@ -24,14 +22,10 @@ async def lifespan(app: FastAPI):
     initialize_sentry()
     app.state.http_client = AsyncClient(limits=Limits(max_connections=20, max_keepalive_connections=10))
     app.state.http_semaphore = asyncio.Semaphore(10)
-    app.state.redis = create_redis_connection()
     task = asyncio.create_task(redis_listener())
-    start_background_listener(app)
     yield
     await app.state.http_client.aclose()
     task.cancel()
-    with suppress(Exception):
-        await app.state.redis.close()
 
 app = FastAPI(lifespan=lifespan)
 app.add_middleware(GZipMiddleware)

backend/app/tasks/deploy_agent.py

@@ -194,7 +194,7 @@ async def _create_local_build_archive(
         cmd = (
             f"cd {source_dir} && nimble setup && "
             f"{self.nim_path} compile --os:linux --cpu:{cpu} "
-            f"--compileOnly --genScript --nimcache:{build_dir} "
+            f"--compileOnly --genScript --nimcache:{build_dir} -d:ssl "
             f"{debug_opts} src/frameos_agent.nim 2>&1"
         )
 

backend/app/utils/remote_exec.py

@@ -68,7 +68,7 @@ async def _exec_via_agent(
         "timeout": timeout,
     }
 
-    await redis.rpush("agent:cmd:queue", json.dumps(message).encode())
+    await redis.rpush(f"agent:cmd:{frame.id}", json.dumps(message).encode())
 
     resp_key = f"agent:resp:{cmd_id}"
     res = await redis.blpop(resp_key, timeout=timeout)
@@ -108,7 +108,7 @@ async def _file_write_via_agent(
         "blob": base64.b64encode(zipped).decode(),
     }
 
-    await redis.rpush("agent:cmd:queue", json.dumps(message).encode())
+    await redis.rpush(f"agent:cmd:{frame.id}", json.dumps(message).encode())
 
     resp_key = f"agent:resp:{cmd_id}"
     res = await redis.blpop(resp_key, timeout=timeout)

backend/app/ws/agent_bridge.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import base64
+import json
+import uuid
+from arq import ArqRedis as Redis
+
+CMD_KEY  = "agent:cmd:{id}"     # per-frame inbound   queue
+RESP_KEY = "agent:resp:{id}"    # per-command outbound queue
+STREAM_KEY = "agent:cmd:stream:{id}"
+
+async def send_cmd(
+    redis: Redis,
+    frame_id: int,
+    payload: dict,
+    *,
+    blob: bytes | None = None,
+    timeout: int = 120,
+):
+    cmd_id = str(uuid.uuid4())
+    job = {
+        "id": cmd_id,
+        "frame_id": frame_id,
+        "payload": payload,
+        "timeout": timeout,
+    }
+    if blob is not None:
+        job["blob"] = base64.b64encode(blob).decode()
+
+    await redis.rpush(CMD_KEY.format(id=frame_id), json.dumps(job).encode())
+
+    res = await redis.blpop(RESP_KEY.format(id=cmd_id), timeout=timeout)
+    if res is None:                                    # ⏰ timed out
+        raise TimeoutError(f"agent timed-out after {timeout}s")
+
+    key, raw = res
+    reply = json.loads(raw)
+    if not reply.get("ok"):
+        raise RuntimeError(reply.get("result", {}).get("error") or reply.get("error", "agent error"))
+
+    if reply.get("binary"):
+        return base64.b64decode(reply["result"])
+    res = reply.get("result")
+
+    # --- nested binary inside an http dict ------------------------------
+    if isinstance(res, dict) and res.get("binary"):
+        body = res.get("body")
+        if isinstance(body, str):                  # base64 string → bytes
+            res["body"] = base64.b64decode(body)
+    return res