Plan B: MCP state server + independent worker processes

appforge_mcp_server.py

@@ -0,0 +1,13 @@
+"""Entry alias: `python appforge_mcp_server.py [--db ...] [--port ...]` runs the state server."""
+from backend.engine.state_server import serve
+
+if __name__ == "__main__":
+    import argparse
+    import asyncio
+
+    p = argparse.ArgumentParser()
+    p.add_argument("--db", default="data/engine.db")
+    p.add_argument("--host", default="127.0.0.1")
+    p.add_argument("--port", type=int, default=8800)
+    a = p.parse_args()
+    asyncio.run(serve(a.db, a.host, a.port))

backend/engine/agent_adapter.py

@@ -0,0 +1,78 @@
+"""Bridge a claimed task to an agent via the registry (mock/real).
+
+No single execute() signature exists across agents: base MockAgent takes an
+AgentTask and returns AgentResult; specialized/real agents take a dict and
+return a dict. We always PASS a dict and read results defensively.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def _field(res: Any, name: str):
+    """Read `name` from a dict-or-attribute result, else None."""
+    if isinstance(res, dict):
+        return res.get(name)
+    return getattr(res, name, None)
+
+
+def _artifact(res: Any):
+    art = _field(res, "artifact")
+    return art if art is not None else res
+
+
+def _writes_value(res: Any, writes_key: str):
+    art = _artifact(res)
+    if isinstance(art, dict):
+        return art.get(writes_key, art)
+    return art
+
+
+async def _run_clarify_loop(task_input, registry, max_questions):
+    clarifier = registry.get("clarifying_pm")
+    po = registry.get("product_owner")
+    idea = task_input.get("idea", "")
+    questions: list[str] = []
+    answers: list[str] = []
+    for _ in range(max_questions + 1):
+        res = await clarifier.execute(
+            {
+                "idea": idea,
+                "questions": list(questions),
+                "answers": list(answers),
+                "mode": "autonomous",
+            }
+        )
+        art = _artifact(res)
+        prd = None
+        if isinstance(art, dict):
+            prd = art.get("prd") or art.get("final_prd")
+            question = art.get("question")
+        else:
+            question, prd = None, None
+        if prd:
+            return {"agent_id": "clarifying_pm", "output": prd}, {"prd": prd}
+        if not question:
+            break
+        questions.append(question)
+        ans = await po.execute({"question": question})
+        ans_art = _artifact(ans)
+        answers.append(ans_art if isinstance(ans_art, str) else str(ans_art))
+    # Fallback: synthesize a minimal PRD so the pipeline always advances in mock mode.
+    prd = f"PRD for: {idea}"
+    return {"agent_id": "clarifying_pm", "output": prd}, {"prd": prd}
+
+
+async def run_agent_task(
+    agent_id, phase, task_input, model, registry, cfg, max_questions=6
+):
+    if agent_id == "clarifying_pm":
+        return await _run_clarify_loop(task_input, registry, max_questions)
+    writes_key = cfg.agents_of(phase)[agent_id].writes
+    agent = registry.get(agent_id)
+    res = await agent.execute(
+        dict(task_input, agent_id=agent_id, model=model, mode="autonomous")
+    )
+    value = _writes_value(res, writes_key)
+    return {"agent_id": agent_id, "output": value}, {writes_key: value}

backend/engine/client.py

@@ -0,0 +1,100 @@
+"""Async MCP client wrapper for the AppForge state server."""
+
+from __future__ import annotations
+
+import json
+from contextlib import AsyncExitStack
+from typing import Any
+
+from mcp import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+
+
+class EngineClient:
+    def __init__(self, url: str):
+        self.url = url
+        self._stack: AsyncExitStack | None = None
+        self._session: ClientSession | None = None
+
+    async def __aenter__(self) -> EngineClient:
+        self._stack = AsyncExitStack()
+        try:
+            r, w, _ = await self._stack.enter_async_context(
+                streamable_http_client(self.url)
+            )
+            self._session = await self._stack.enter_async_context(ClientSession(r, w))
+            await self._session.initialize()
+        except BaseException:
+            await self._stack.aclose()  # don't leak the transport if setup fails
+            raise
+        return self
+
+    async def __aexit__(self, *exc) -> None:
+        await self._stack.aclose()
+
+    async def _call(self, name: str, **args: Any) -> Any:
+        res = await self._session.call_tool(name, args)
+        if res.isError:
+            msg = res.content[0].text if res.content else str(res)
+            raise RuntimeError(f"{name} failed: {msg}")
+        return json.loads(res.content[0].text)
+
+    async def create_run(self, idea: str, budget_limit: float = 200.0) -> str:
+        return (await self._call("create_run", idea=idea, budget_limit=budget_limit))[
+            "run_id"
+        ]
+
+    async def get_state(self, run_id: str, keys: list[str] | None = None) -> dict:
+        return await self._call("get_state", run_id=run_id, keys_json=json.dumps(keys))
+
+    async def put_state(
+        self, run_id: str, key: str, value: Any, expected_version: int
+    ) -> bool:
+        return (
+            await self._call(
+                "put_state",
+                run_id=run_id,
+                key=key,
+                value_json=json.dumps(value),
+                expected_version=expected_version,
+            )
+        )["ok"]
+
+    async def claim_next_task(self, run_id: str, worker_id: str) -> dict | None:
+        return await self._call("claim_next_task", run_id=run_id, worker_id=worker_id)
+
+    async def complete_task(
+        self, task_id, worker_id, version, result, state_writes=None
+    ) -> bool:
+        return (
+            await self._call(
+                "complete_task",
+                task_id=task_id,
+                worker_id=worker_id,
+                version=version,
+                result_json=json.dumps(result),
+                state_writes_json=json.dumps(state_writes),
+            )
+        )["ok"]
+
+    async def heartbeat(self, task_id: str, worker_id: str) -> bool:
+        return (await self._call("heartbeat", task_id=task_id, worker_id=worker_id))[
+            "ok"
+        ]
+
+    async def fail_task(self, task_id, worker_id, version, error: str) -> None:
+        await self._call(
+            "fail_task",
+            task_id=task_id,
+            worker_id=worker_id,
+            version=version,
+            error=error,
+        )
+
+    async def submit_approval(self, run_id: str, phase: str, decision: str) -> None:
+        await self._call(
+            "submit_approval", run_id=run_id, phase=phase, decision=decision
+        )
+
+    async def get_run(self, run_id: str) -> dict:
+        return await self._call("get_run", run_id=run_id)

backend/engine/mcp_tools.py

@@ -0,0 +1,73 @@
+"""MCP tool definitions for the AppForge state server. JSON string in/out."""
+
+from __future__ import annotations
+
+import json
+import uuid
+
+from backend.engine.store import Store
+
+
+def register_tools(mcp, store: Store) -> None:
+    @mcp.tool()
+    async def create_run(idea: str, budget_limit: float = 200.0) -> str:
+        run_id = uuid.uuid4().hex
+        await store.create_run(run_id, idea, budget_limit)
+        return json.dumps({"run_id": run_id})
+
+    @mcp.tool()
+    async def get_state(run_id: str, keys_json: str = "null") -> str:
+        keys = json.loads(keys_json)
+        state = await store.get_state(run_id, keys)
+        return json.dumps(
+            {k: {"value": v[0], "version": v[1]} for k, v in state.items()}
+        )
+
+    @mcp.tool()
+    async def put_state(
+        run_id: str, key: str, value_json: str, expected_version: int
+    ) -> str:
+        ok = await store.put_state(
+            run_id, key, json.loads(value_json), expected_version
+        )
+        return json.dumps({"ok": ok})
+
+    @mcp.tool()
+    async def claim_next_task(run_id: str, worker_id: str) -> str:
+        cr = await store.claim_next_task(run_id, worker_id)
+        return json.dumps(cr.model_dump() if cr is not None else None)
+
+    @mcp.tool()
+    async def complete_task(
+        task_id: str,
+        worker_id: str,
+        version: int,
+        result_json: str,
+        state_writes_json: str = "null",
+    ) -> str:
+        ok = await store.complete_task(
+            task_id,
+            worker_id,
+            version,
+            json.loads(result_json),
+            json.loads(state_writes_json),
+        )
+        return json.dumps({"ok": ok})
+
+    @mcp.tool()
+    async def heartbeat(task_id: str, worker_id: str) -> str:
+        return json.dumps({"ok": await store.heartbeat(task_id, worker_id)})
+
+    @mcp.tool()
+    async def fail_task(task_id: str, worker_id: str, version: int, error: str) -> str:
+        await store.fail_task(task_id, worker_id, version, error)
+        return json.dumps({"ok": True})
+
+    @mcp.tool()
+    async def submit_approval(run_id: str, phase: str, decision: str) -> str:
+        await store.submit_approval(run_id, phase, decision)
+        return json.dumps({"ok": True})
+
+    @mcp.tool()
+    async def get_run(run_id: str) -> str:
+        return json.dumps(await store.snapshot(run_id))

backend/engine/models.py

@@ -1,4 +1,5 @@
 """SQL schema + typed result models for the engine store."""
+
 from __future__ import annotations
 
 from pydantic import BaseModel

backend/engine/phases.py

@@ -1,4 +1,5 @@
 """Loader/validator for config/phases.yaml — the six-phase source of truth."""
+
 from __future__ import annotations
 
 from dataclasses import dataclass
@@ -30,7 +31,7 @@ def __init__(self, phases: list[PhaseSpec]):
         self._by_name = {p.name: p for p in self._phases}
 
     @classmethod
-    def load(cls, path: str = "config/phases.yaml") -> "PhasesConfig":
+    def load(cls, path: str = "config/phases.yaml") -> PhasesConfig:
         raw = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
         phases: list[PhaseSpec] = []
         for p in raw["phases"]:
@@ -45,7 +46,12 @@ def load(cls, path: str = "config/phases.yaml") -> "PhasesConfig":
                 for aid, a in p["agents"].items()
             }
             phases.append(
-                PhaseSpec(name=p["name"], order=int(p["order"]), gate=p.get("gate", "none"), agents=agents)
+                PhaseSpec(
+                    name=p["name"],
+                    order=int(p["order"]),
+                    gate=p.get("gate", "none"),
+                    agents=agents,
+                )
             )
         cfg = cls(phases)
         cfg._validate()
@@ -83,4 +89,6 @@ def all_agent_ids(self) -> list[str]:
 
 
 def load_downgrade_paths(path: str = "config/budget.yaml") -> dict[str, str]:
-    return yaml.safe_load(Path(path).read_text(encoding="utf-8")).get("downgrade_paths", {})
+    return yaml.safe_load(Path(path).read_text(encoding="utf-8")).get(
+        "downgrade_paths", {}
+    )