Update app.py

app.py

@@ -1,161 +1,63 @@
-import os, json, time, asyncio
-from typing import Dict, Any
-import httpx, websockets
-from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Body
+import os, time, json, asyncio
+from typing import Any, Dict
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
+from fastapi.responses import JSONResponse
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse, StreamingResponse
 
-FILES_DIR = os.environ.get("FILES_DIR", "/tmp/brain_app/files")
-TTS_WS_TARGET = os.environ.get("TTS_WS_TARGET", "")
-TTS_HTTP_TARGET = os.environ.get("TTS_HTTP_TARGET", "")
-STT_WS_TARGET = os.environ.get("STT_WS_TARGET", "")
-AGENT_WS_TARGET = os.environ.get("AGENT_WS_TARGET", "")
+BASE_DIR = "/tmp/brain_app"
+FILES_DIR = os.path.join(BASE_DIR, "files")
+LOGS_DIR = os.path.join(FILES_DIR, "logs")
+for d in (BASE_DIR, FILES_DIR, LOGS_DIR):
+    os.makedirs(d, exist_ok=True)
 
-os.makedirs(FILES_DIR, exist_ok=True)
+def log_event(kind: str, data: Dict[str, Any]):
+    rec = {"ts": time.time(), "kind": kind, "data": data}
+    with open(os.path.join(LOGS_DIR, "events.jsonl"), "a", encoding="utf-8") as f:
+        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
 
-app = FastAPI(title="Brain (Proxy/Orchestrator)")
+app = FastAPI(title="Brain (Skeleton)")
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=[""], allow_methods=[""], allow_headers=["*"], allow_credentials=True,
+    allow_origins=["*"], allow_credentials=True,
+    allow_methods=[""], allow_headers=[""],
 )
 
-telemetry_subs = set()     # Server-Sent Events subscribers
-logs_subs = set()          # SSE for generic logs/events
-
-def _sse_pack(d: Dict[str, Any]) -> bytes:
-    return f"data: {json.dumps(d, ensure_ascii=False)}\n\n".encode("utf-8")
-
-async def _sse_broadcast(pool, payload):
-    dead = []
-    for q in list(pool):
-        try:
-            await q.put(payload)
-        except Exception:
-            dead.append(q)
-    for q in dead:
-        pool.discard(q)
-
 @app.get("/health")
 def health():
     return {
-        "ok": True, "service": "brain",
-        "tts_ws_target": TTS_WS_TARGET, "tts_http_target": TTS_HTTP_TARGET,
-        "stt_ws_target": STT_WS_TARGET, "agent_ws_target": AGENT_WS_TARGET,
-        "files_dir": FILES_DIR, "time": time.time()
+        "ok": True,
+        "service": "brain-skeleton",
+        "time": time.time(),
+        "files_dir": FILES_DIR,
+        "logs_dir": LOGS_DIR,
     }
 
-# ---------- SSE streams ----------
-@app.get("/stream/telemetry")
-async def stream_telemetry():
-    q: asyncio.Queue = asyncio.Queue()
-    telemetry_subs.add(q)
-    async def gen():
-        try:
-            while True:
-                item = await q.get()
-                yield item
-        finally:
-            telemetry_subs.discard(q)
-    return StreamingResponse(gen(), media_type="text/event-stream")
-
-@app.get("/stream/logs")
-async def stream_logs():
-    q: asyncio.Queue = asyncio.Queue()
-    logs_subs.add(q)
-    async def gen():
-        try:
-            while True:
-                item = await q.get()
-                yield item
-        finally:
-            logs_subs.discard(q)
-    return StreamingResponse(gen(), media_type="text/event-stream")
-
-# ---------- Telemetry ingest ----------
-@app.post("/telemetry")
-async def telemetry(body: dict = Body(...)):
-    body["ts"] = time.time()
-    await _sse_broadcast(telemetry_subs, _sse_pack(body))
-    return {"ok": True}
-
-# ---------- Generic log/event ----------
-@app.post("/log")
-async def log(body: dict = Body(...)):
-    body["ts"] = time.time()
-    await _sse_broadcast(logs_subs, _sse_pack(body))
-    return {"ok": True}
-
-# ---------- HTTP proxy to TTS /speak ----------
-@app.post("/speak")
-async def speak_proxy(body: dict = Body(...)):
-    if not TTS_HTTP_TARGET:
-        return JSONResponse({"ok": False, "error": "TTS HTTP target not configured"}, 500)
-    try:
-        async with httpx.AsyncClient(timeout=120) as client:
-            r = await client.post(TTS_HTTP_TARGET, json=body)
-            r.raise_for_status()
-            out = r.json()
-            await _sse_broadcast(logs_subs, _sse_pack({"kind":"tts_http", "req": body, "resp": out}))
-            return JSONResponse(out, r.status_code)
-    except Exception as e:
-        await _sse_broadcast(logs_subs, _sse_pack({"kind":"error", "where":"speak_proxy", "error": repr(e)}))
-        return JSONResponse({"ok": False, "error": f"TTS proxy failed: {e}"}, 502)
+@app.post("/echo")
+async def echo(request: Request):
+    payload = await request.json()
+    log_event("echo_http_in", payload)
+    out = {"ok": True, "ts": time.time(), "echo": payload}
+    log_event("echo_http_out", out)
+    return JSONResponse(out)
 
-# ---------- bidirectional WS bridge ----------
-async def _bridge(client_ws: WebSocket, upstream_url: str):
+@app.websocket("/ws/echo")
+async def ws_echo(ws: WebSocket):
+    await ws.accept()
+    await ws.send_json({"event": "hello", "msg": "echo socket ready"})
     try:
-        upstream = await websockets.connect(upstream_url, ping_interval=None, max_size=8_000_000)
+        while True:
+            text = await ws.receive_text()
+            log_event("echo_ws_in", {"text": text})
+            await ws.send_json({"event": "echo", "text": text})
+    except WebSocketDisconnect:
+        pass
     except Exception as e:
-        await client_ws.send_json({"event":"error","error":f"Upstream connect failed: {repr(e)}"})
-        await client_ws.close(); return
-
-    async def c2u():
         try:
-            while True:
-                msg = await client_ws.receive()
-                if "bytes" in msg and msg["bytes"] is not None:
-                    await upstream.send(msg["bytes"])
-                elif "text" in msg and msg["text"] is not None:
-                    await upstream.send(msg["text"])
-        except WebSocketDisconnect:
+            await ws.send_json({"event": "error", "detail": str(e)})
+        except:
             pass
-        finally:
-            try: await upstream.close()
-            except: pass
-
-    async def u2c():
+    finally:
         try:
-            while True:
-                msg = await upstream.recv()
-                if isinstance(msg, (bytes, bytearray)):
-                    await client_ws.send_bytes(msg)
-                else:
-                    await client_ws.send_text(msg)
-        except Exception:
-            pass
-        finally:
-            try: await client_ws.close()
-            except: pass
-
-    await asyncio.gather(c2u(), u2c())
-
-@app.websocket("/ws/tts")
-async def ws_tts(ws: WebSocket):
-    await ws.accept()
-    if not TTS_WS_TARGET:
-        await ws.send_json({"event":"error","error":"TTS WS target not configured"}); await ws.close(); return
-    await _bridge(ws, TTS_WS_TARGET)
-
-@app.websocket("/ws/stt")
-async def ws_stt(ws: WebSocket):
-    await ws.accept()
-    if not STT_WS_TARGET:
-        await ws.send_json({"event":"error","error":"STT WS target not configured"}); await ws.close(); return
-    await _bridge(ws, STT_WS_TARGET)
-
-@app.websocket("/ws/agent")
-async def ws_agent(ws: WebSocket):
-    await ws.accept()
-    if not AGENT_WS_TARGET:
-        await ws.send_json({"event":"error","error":"Agent WS target not configured"}); await ws.close(); return
-    await _bridge(ws, AGENT_WS_TARGET)
+            await ws.close()
+        except:
+            pass