Plan D: web Socket.IO bridge + LangGraph retirement

backend/config.py

@@ -54,7 +54,6 @@ class Config:
     log_level: str
     budget_limit: float
     anthropic_model: str
-    sqlite_path: str
     max_clarifying_questions: int
     enable_phase4: bool
     engine_lease_ttl: float = 120.0
@@ -68,15 +67,13 @@ class Config:
 
     @classmethod
     def load(cls) -> Config:
-        default_sqlite = str(REPO_ROOT / "data" / "checkpoints.db")
         return cls(
             anthropic_api_key=os.getenv("ANTHROPIC_API_KEY"),
             mock_agents=_env_bool("MOCK_AGENTS", True),
             debug=_env_bool("DEBUG", False),
             log_level=os.getenv("LOG_LEVEL", "INFO"),
             budget_limit=_env_float("BUDGET_LIMIT", 200.0),
             anthropic_model=os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-6"),
-            sqlite_path=os.getenv("SQLITE_PATH", default_sqlite),
             max_clarifying_questions=_env_int("MAX_CLARIFYING_QUESTIONS", 6),
             enable_phase4=_env_bool("ENABLE_PHASE4", True),
             engine_lease_ttl=_env_float("ENGINE_LEASE_TTL", 120.0),

backend/engine/run.py

@@ -6,11 +6,20 @@
 import asyncio
 import contextlib
 import sys
+from dataclasses import dataclass, field
 
 from backend.engine.client import EngineClient
 from backend.engine.state_server import free_port, serve
 
 
+@dataclass
+class RunHandle:
+    run_id: str
+    url: str
+    procs: list = field(default_factory=list)
+    server_task: object = None
+
+
 async def _drive_gates(url, run_id, auto_approve, timeout, poll):
     loop = asyncio.get_running_loop()
     deadline = loop.time() + timeout
@@ -28,41 +37,41 @@ async def _drive_gates(url, run_id, auto_approve, timeout, poll):
             await asyncio.sleep(poll)
 
 
-async def run_pipeline(
+async def start_run(
     idea,
     workers=4,
     budget_limit=200.0,
-    auto_approve=True,
     db_path=None,
     host="127.0.0.1",
     port=None,
-    poll=0.1,
-    timeout=60.0,
-) -> dict:
+) -> RunHandle:
+    """Boot the state server, wait for it to accept a create_run, spawn worker
+    subprocesses, and return the handle. Does NOT drive gates."""
     db_path = db_path or "data/engine.db"
     port = port or free_port()
     url = f"http://{host}:{port}/mcp"
 
     server_task = asyncio.create_task(serve(db_path, host, port))
-    procs = []
-    worker_pids = []
     run_id = None
-    final = None
+    # wait for the server to accept connections, then create the run
+    for _ in range(200):
+        try:
+            async with EngineClient(url) as c:
+                run_id = await c.create_run(idea, budget_limit)
+            break
+        except Exception:  # noqa: BLE001 - server may not be up yet
+            await asyncio.sleep(0.05)
+    if run_id is None:
+        server_task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await server_task
+        raise RuntimeError("state server failed to start")
+
+    # spawn worker subprocesses
+    handle = RunHandle(run_id=run_id, url=url, procs=[], server_task=server_task)
     try:
-        # wait for the server to accept connections, then create the run
-        for _ in range(200):
-            try:
-                async with EngineClient(url) as c:
-                    run_id = await c.create_run(idea, budget_limit)
-                break
-            except Exception:  # noqa: BLE001 - server may not be up yet
-                await asyncio.sleep(0.05)
-        if run_id is None:
-            raise RuntimeError("state server failed to start")
-
-        # spawn worker subprocesses (inside the try so a mid-spawn failure still tears down)
         for i in range(workers):
-            procs.append(
+            handle.procs.append(
                 await asyncio.create_subprocess_exec(
                     sys.executable,
                     "-m",
@@ -75,23 +84,50 @@ async def run_pipeline(
                     f"w{i}",
                 )
             )
-        worker_pids = [p.pid for p in procs]
+    except BaseException:
+        await stop_run(handle)  # terminate spawned procs + cancel/await server task
+        raise
+    return handle
 
-        final = await _drive_gates(url, run_id, auto_approve, timeout, poll)
-    finally:
-        for p in procs:
-            if p.returncode is None:
-                p.terminate()
-        for p in procs:
-            try:
-                await asyncio.wait_for(p.wait(), timeout=5.0)
-            except TimeoutError:
-                p.kill()
-        server_task.cancel()
+
+async def stop_run(handle: RunHandle) -> None:
+    """Terminate worker subprocesses and cancel/await the server task."""
+    for p in handle.procs:
+        if p.returncode is None:
+            p.terminate()
+    for p in handle.procs:
+        try:
+            await asyncio.wait_for(p.wait(), timeout=5.0)
+        except TimeoutError:
+            p.kill()
+    if handle.server_task is not None:
+        handle.server_task.cancel()
         with contextlib.suppress(asyncio.CancelledError):
-            await server_task
+            await handle.server_task
+
+
+async def run_pipeline(
+    idea,
+    workers=4,
+    budget_limit=200.0,
+    auto_approve=True,
+    db_path=None,
+    host="127.0.0.1",
+    port=None,
+    poll=0.1,
+    timeout=60.0,
+) -> dict:
+    handle = await start_run(idea, workers, budget_limit, db_path, host, port)
+    worker_pids = [p.pid for p in handle.procs]
+    final = None
+    try:
+        final = await _drive_gates(
+            handle.url, handle.run_id, auto_approve, timeout, poll
+        )
+    finally:
+        await stop_run(handle)
 
-    return {"run_id": run_id, "snapshot": final, "worker_pids": worker_pids}
+    return {"run_id": handle.run_id, "snapshot": final, "worker_pids": worker_pids}
 
 
 def main() -> None:

backend/engine/store.py

@@ -399,6 +399,15 @@ async def snapshot(self, run_id: str) -> dict:
             status = "done"
         else:
             status = "running"
+        cur = await self._db.execute(
+            "SELECT COALESCE(SUM(cost),0) AS s FROM spend WHERE run_id=?", (run_id,)
+        )
+        spent = (await cur.fetchone())["s"]
+        cur = await self._db.execute(
+            "SELECT budget_limit FROM runs WHERE run_id=?", (run_id,)
+        )
+        row = await cur.fetchone()
+        limit = row["budget_limit"] if row else 0.0
         return {
             "run_id": run_id,
             "status": status,
@@ -416,4 +425,5 @@ async def snapshot(self, run_id: str) -> dict:
                 }
                 for t in tasks
             ],
+            "budget": {"spent": spent, "limit": limit},
         }

backend/engine/webbridge.py

@@ -0,0 +1,152 @@
+"""Pure mappers: engine snapshot -> the React frontend's Socket.IO contract."""
+
+from __future__ import annotations
+
+_PHASE_NUMBER = {
+    "clarify": 3,
+    "design": 4,
+    "code": 6,
+    "test": 7,
+    "deploy": 8,
+    "iterate": 10,
+}
+PHASE_GATE_KIND = {"clarify": "prd", "design": "plan"}
+_WRITES_KEY = {"clarify": "prd", "design": "adr"}  # content shown on the approval card
+
+
+def phase_number(name: str) -> int:
+    return _PHASE_NUMBER.get(name, 0)
+
+
+def _threshold_bucket(spent: float, limit: float) -> int:
+    if limit <= 0:
+        return 0
+    ratio = spent / limit
+    for bucket in (100, 95, 85, 75, 50):
+        if ratio >= bucket / 100:
+            return bucket
+    return 0
+
+
+def _agent_status(task: dict, base_models: dict) -> str:
+    status = task["status"]
+    if status == "failed":
+        return "error"
+    if status == "done":
+        base = base_models.get(task["agent_id"])
+        if task.get("model") and base and task["model"] != base:
+            return "downgraded"
+        return "complete"
+    if status in ("claimed", "running") or task.get("owner"):
+        return "running"
+    return "pending"
+
+
+def _agents_map(snapshot: dict, base_models: dict) -> dict:
+    return {
+        t["agent_id"]: {
+            "id": t["agent_id"],
+            "name": t["agent_id"],
+            "status": _agent_status(t, base_models),
+        }
+        for t in snapshot["tasks"]
+    }
+
+
+def to_project_state(snapshot: dict, idea: str, state: dict) -> dict:
+    agents = _agents_map(snapshot, {})
+    pending = None
+    for p in snapshot["phases"]:
+        if p["gate"] == "pending":
+            kind = PHASE_GATE_KIND.get(p["name"], "prd")
+            content = state.get(_WRITES_KEY.get(p["name"], "prd")) or ""
+            pending = {
+                "agent": p["name"],
+                "phase": phase_number(p["name"]),
+                "content": content if isinstance(content, str) else str(content),
+                "kind": kind,
+            }
+            break
+    open_phase = next((p for p in snapshot["phases"] if p["status"] == "open"), None)
+    fe_status = {"done": "complete", "failed": "failed"}.get(
+        snapshot["status"], "running"
+    )
+    if pending is not None:
+        fe_status = "paused"
+    b = snapshot.get("budget", {"spent": 0.0, "limit": 0.0})
+    budget = {
+        "spent": b.get("spent", 0.0),
+        "limit": b.get("limit", 0.0),
+        "threshold": _threshold_bucket(b.get("spent", 0.0), b.get("limit", 0.0)),
+    }
+    return {
+        "project_id": snapshot["run_id"],
+        "idea": idea,
+        "messages": [],
+        "agents": agents,
+        "approval_pending": pending,
+        "budget": budget,
+        "phase": phase_number(open_phase["name"]) if open_phase else 3,
+        "prd": state.get("prd"),
+        "status": fe_status,
+        "adr": state.get("adr"),
+    }
+
+
+def diff_to_events(
+    prev: dict | None, new: dict, state: dict, base_models: dict
+) -> list[tuple[str, dict]]:
+    events: list[tuple[str, dict]] = []
+    prev_tasks = {t["agent_id"]: t for t in (prev["tasks"] if prev else [])}
+    for t in new["tasks"]:
+        old = prev_tasks.get(t["agent_id"])
+        new_s = _agent_status(t, base_models)
+        old_s = _agent_status(old, base_models) if old else "pending"
+        if new_s != old_s:
+            events.append(("agent_status", {"agent": t["agent_id"], "status": new_s}))
+    prev_phase = {p["name"]: p for p in (prev["phases"] if prev else [])}
+    for p in new["phases"]:
+        op = prev_phase.get(p["name"])
+        if p["status"] == "complete" and (op is None or op["status"] != "complete"):
+            events.append(
+                (
+                    "phase_complete",
+                    {
+                        "phase": phase_number(p["name"]),
+                        "summary": f"{p['name']} complete",
+                        "status": "success",
+                    },
+                )
+            )
+        if p["gate"] == "pending" and (op is None or op["gate"] != "pending"):
+            kind = PHASE_GATE_KIND.get(p["name"], "prd")
+            content = state.get(_WRITES_KEY.get(p["name"], "prd")) or ""
+            events.append(
+                (
+                    "approval_required",
+                    {
+                        "agent": p["name"],
+                        "phase": phase_number(p["name"]),
+                        "content": (
+                            content if isinstance(content, str) else str(content)
+                        ),
+                        "kind": kind,
+                    },
+                )
+            )
+    nb = new.get("budget", {})
+    ob = prev.get("budget", {}) if prev else {}
+    if nb and nb.get("spent") != ob.get("spent"):
+        events.append(
+            (
+                "budget_update",
+                {
+                    "spent": nb.get("spent", 0.0),
+                    "limit": nb.get("limit", 0.0),
+                    "threshold": _threshold_bucket(
+                        nb.get("spent", 0.0), nb.get("limit", 0.0)
+                    ),
+                },
+            )
+        )
+    return events