diff --git a/appforge_mcp_server.py b/appforge_mcp_server.py new file mode 100644 index 0000000..53de8a7 --- /dev/null +++ b/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)) diff --git a/backend/engine/agent_adapter.py b/backend/engine/agent_adapter.py new file mode 100644 index 0000000..dfdb04d --- /dev/null +++ b/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} diff --git a/backend/engine/client.py b/backend/engine/client.py new file mode 100644 index 0000000..7c35b94 --- /dev/null +++ b/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) diff --git a/backend/engine/mcp_tools.py b/backend/engine/mcp_tools.py new file mode 100644 index 0000000..051739c --- /dev/null +++ b/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)) diff --git a/backend/engine/models.py b/backend/engine/models.py index db7bfd3..99a8dc2 100644 --- a/backend/engine/models.py +++ b/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 diff --git a/backend/engine/phases.py b/backend/engine/phases.py index af639cf..d055386 100644 --- a/backend/engine/phases.py +++ b/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", {} + ) diff --git a/backend/engine/run.py b/backend/engine/run.py new file mode 100644 index 0000000..79199c4 --- /dev/null +++ b/backend/engine/run.py @@ -0,0 +1,120 @@ +"""Run controller / CLI: boot server, spawn worker subprocesses, drive gates.""" + +from __future__ import annotations + +import argparse +import asyncio +import contextlib +import sys + +from backend.engine.client import EngineClient +from backend.engine.state_server import free_port, serve + + +async def _drive_gates(url, run_id, auto_approve, timeout, poll): + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + async with EngineClient(url) as c: + while True: + run = await c.get_run(run_id) + if run["status"] in ("done", "failed"): + return run + if auto_approve: + for p in run["phases"]: + if p["gate"] == "pending": + await c.submit_approval(run_id, p["name"], "approved") + if loop.time() > deadline: + return run + await asyncio.sleep(poll) + + +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: + 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 + 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( + await asyncio.create_subprocess_exec( + sys.executable, + "-m", + "backend.engine.worker", + "--server-url", + url, + "--run-id", + run_id, + "--worker-id", + f"w{i}", + ) + ) + worker_pids = [p.pid for p in procs] + + 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() + with contextlib.suppress(asyncio.CancelledError): + await server_task + + return {"run_id": run_id, "snapshot": final, "worker_pids": worker_pids} + + +def main() -> None: + p = argparse.ArgumentParser(prog="appforge") + sub = p.add_subparsers(dest="cmd", required=True) + r = sub.add_parser("run") + r.add_argument("idea") + r.add_argument("--workers", type=int, default=4) + r.add_argument("--budget-limit", type=float, default=200.0) + r.add_argument("--no-auto-approve", action="store_true") + a = p.parse_args() + result = asyncio.run( + run_pipeline( + a.idea, + workers=a.workers, + budget_limit=a.budget_limit, + auto_approve=not a.no_auto_approve, + ) + ) + print(f"run {result['run_id']}: {result['snapshot']['status']}") + for ph in result["snapshot"]["phases"]: + print(f" {ph['name']:9} {ph['status']:9} gate={ph['gate']}") + + +if __name__ == "__main__": + main() diff --git a/backend/engine/scheduler.py b/backend/engine/scheduler.py index 98f41fb..7df43b8 100644 --- a/backend/engine/scheduler.py +++ b/backend/engine/scheduler.py @@ -1,4 +1,5 @@ """Pure scheduling logic. No DB, no I/O — takes plain dicts, returns plans.""" + from __future__ import annotations from backend.engine.phases import PhasesConfig @@ -20,7 +21,9 @@ def seed_specs_for_phase( "agent_id": aid, "phase": phase_name, "phase_order": order, - "depends_on": [task_id(run_id, phase_name, dep) for dep in spec.depends_on], + "depends_on": [ + task_id(run_id, phase_name, dep) for dep in spec.depends_on + ], "sim_cost": spec.sim_cost, "model": base_models.get(aid), "input_keys": list(spec.reads), @@ -64,10 +67,17 @@ def advance(phases: list[dict], tasks: list[dict], cfg: PhasesConfig) -> dict: if gate != "none": open_gates.append(gate) # next phase waits for submit_approval else: - nxt = next((q for q in ordered if q["phase_order"] == p["phase_order"] + 1), None) + nxt = next( + (q for q in ordered if q["phase_order"] == p["phase_order"] + 1), + None, + ) if nxt is not None: open_phases.append(nxt["name"]) - return {"complete_phases": complete_phases, "open_gates": open_gates, "open_phases": open_phases} + return { + "complete_phases": complete_phases, + "open_gates": open_gates, + "open_phases": open_phases, + } def resolve_model( diff --git a/backend/engine/state_server.py b/backend/engine/state_server.py new file mode 100644 index 0000000..50c14a4 --- /dev/null +++ b/backend/engine/state_server.py @@ -0,0 +1,83 @@ +"""Standalone FastMCP state server wrapping the single-writer Store.""" + +from __future__ import annotations + +import asyncio +import contextlib +import socket +from pathlib import Path + +import yaml +from mcp.server.fastmcp import FastMCP + +from backend.engine.mcp_tools import register_tools +from backend.engine.phases import PhasesConfig +from backend.engine.store import Store + + +def free_port() -> int: + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +def base_models_from_config(path: str = "config/agents.yaml") -> dict[str, str]: + raw = yaml.safe_load(Path(path).read_text(encoding="utf-8")) + return { + aid: a.get("llm", {}).get("model") for aid, a in raw.get("agents", {}).items() + } + + +def build_server(db_path, cfg=None, base_models=None, lease_s: float = 120.0): + cfg = cfg or PhasesConfig.load() + base_models = base_models if base_models is not None else base_models_from_config() + store = Store(db_path, cfg, base_models, lease_s=lease_s) + mcp = FastMCP("appforge-state", stateless_http=True) + register_tools(mcp, store) + return mcp, store + + +async def _reaper_loop(store: Store, interval: float) -> None: + while True: + await asyncio.sleep(interval) + with contextlib.suppress(Exception): # reaper must never crash the server + await store.reap_expired() + + +async def serve( + db_path, + host="127.0.0.1", + port=8800, + cfg=None, + base_models=None, + lease_s: float = 120.0, + reaper_interval: float = 30.0, +) -> None: + import uvicorn + + mcp, store = build_server(db_path, cfg, base_models, lease_s) + await store.connect() + reaper = asyncio.create_task(_reaper_loop(store, reaper_interval)) + app = mcp.streamable_http_app() + server = uvicorn.Server( + uvicorn.Config(app, host=host, port=port, log_level="error") + ) + try: + await server.serve() + finally: + reaper.cancel() + await asyncio.gather(reaper, return_exceptions=True) + await store.close() + + +if __name__ == "__main__": + import argparse + + 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)) diff --git a/backend/engine/store.py b/backend/engine/store.py index fd32747..1cba143 100644 --- a/backend/engine/store.py +++ b/backend/engine/store.py @@ -1,4 +1,5 @@ """Single-writer SQLite store for the engine. All mutations under _db_lock.""" + from __future__ import annotations import asyncio @@ -17,7 +18,13 @@ class Store: MAX_ATTEMPTS = 3 - def __init__(self, db_path: str, cfg: PhasesConfig, base_models: dict[str, str], lease_s: float = 120.0): + def __init__( + self, + db_path: str, + cfg: PhasesConfig, + base_models: dict[str, str], + lease_s: float = 120.0, + ): self.db_path = db_path self.cfg = cfg self.base_models = base_models @@ -51,24 +58,25 @@ async def _txn(self): raise async def create_run(self, run_id: str, idea: str, budget_limit: float) -> None: - async with self._db_lock: - async with self._txn(): - now = self._now() + async with self._db_lock, self._txn(): + now = self._now() + await self._db.execute( + "INSERT INTO runs (run_id, idea, budget_limit, created_at) VALUES (?,?,?,?)", + (run_id, idea, budget_limit, now), + ) + for name in self.cfg.phase_names: + order = self.cfg.order_of(name) + status = "open" if name == "clarify" else "blocked" await self._db.execute( - "INSERT INTO runs (run_id, idea, budget_limit, created_at) VALUES (?,?,?,?)", - (run_id, idea, budget_limit, now), + "INSERT INTO phases (run_id, name, phase_order, status, gate, seeded) VALUES (?,?,?,?,?,0)", + (run_id, name, order, status, self.cfg.gate_of(name)), ) - for name in self.cfg.phase_names: - order = self.cfg.order_of(name) - status = "open" if name == "clarify" else "blocked" - await self._db.execute( - "INSERT INTO phases (run_id, name, phase_order, status, gate, seeded) VALUES (?,?,?,?,?,0)", - (run_id, name, order, status, self.cfg.gate_of(name)), - ) - await self._seed_phase_locked(run_id, "clarify", now) - await self._recompute_ready_locked(run_id) + await self._seed_phase_locked(run_id, "clarify", now) + await self._recompute_ready_locked(run_id) - async def _seed_phase_locked(self, run_id: str, phase_name: str, now: float) -> None: + async def _seed_phase_locked( + self, run_id: str, phase_name: str, now: float + ) -> None: specs = sch.seed_specs_for_phase(self.cfg, run_id, phase_name, self.base_models) for s in specs: await self._db.execute( @@ -76,9 +84,18 @@ async def _seed_phase_locked(self, run_id: str, phase_name: str, now: float) -> (task_id, run_id, phase, phase_order, agent_id, input, depends_on, status, version, attempts, created_at, model, sim_cost) VALUES (?,?,?,?,?,?,?,'blocked',0,0,?,?,?)""", - (s["task_id"], run_id, s["phase"], s["phase_order"], s["agent_id"], - json.dumps({"input_keys": s["input_keys"]}), json.dumps(s["depends_on"]), - now, s["model"], s["sim_cost"]), + ( + s["task_id"], + run_id, + s["phase"], + s["phase_order"], + s["agent_id"], + json.dumps({"input_keys": s["input_keys"]}), + json.dumps(s["depends_on"]), + now, + s["model"], + s["sim_cost"], + ), ) await self._db.execute( "UPDATE phases SET seeded=1 WHERE run_id=? AND name=?", (run_id, phase_name) @@ -90,7 +107,8 @@ async def _recompute_ready_locked(self, run_id: str) -> None: ready_ids = sch.compute_ready(tasks, phases) for tid in ready_ids: await self._db.execute( - "UPDATE tasks SET status='ready' WHERE task_id=? AND status='blocked'", (tid,) + "UPDATE tasks SET status='ready' WHERE task_id=? AND status='blocked'", + (tid,), ) async def _all_tasks(self, run_id: str) -> list[dict[str, Any]]: @@ -107,15 +125,25 @@ async def _all_phases(self, run_id: str) -> list[dict[str, Any]]: cur = await self._db.execute("SELECT * FROM phases WHERE run_id=?", (run_id,)) return [dict(r) for r in await cur.fetchall()] - async def get_state(self, run_id: str, keys: list[str] | None = None) -> dict[str, tuple[Any, int]]: + async def get_state( + self, run_id: str, keys: list[str] | None = None + ) -> dict[str, tuple[Any, int]]: if keys: - q = "SELECT key, value, version FROM state WHERE run_id=? AND key IN (%s)" % ",".join("?" * len(keys)) + placeholders = ",".join("?" * len(keys)) + q = f"SELECT key, value, version FROM state WHERE run_id=? AND key IN ({placeholders})" cur = await self._db.execute(q, (run_id, *keys)) else: - cur = await self._db.execute("SELECT key, value, version FROM state WHERE run_id=?", (run_id,)) - return {r["key"]: (json.loads(r["value"]), r["version"]) for r in await cur.fetchall()} + cur = await self._db.execute( + "SELECT key, value, version FROM state WHERE run_id=?", (run_id,) + ) + return { + r["key"]: (json.loads(r["value"]), r["version"]) + for r in await cur.fetchall() + } - async def put_state(self, run_id: str, key: str, value: Any, expected_version: int) -> bool: + async def put_state( + self, run_id: str, key: str, value: Any, expected_version: int + ) -> bool: async with self._db_lock: async with self._txn(): if expected_version == 0: @@ -133,10 +161,14 @@ async def put_state(self, run_id: str, key: str, value: Any, expected_version: i return result async def _spend_ratio_locked(self, run_id: str) -> float: - cur = await self._db.execute("SELECT budget_limit FROM runs WHERE run_id=?", (run_id,)) + 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 - cur = await self._db.execute("SELECT COALESCE(SUM(cost),0) AS s FROM spend WHERE run_id=?", (run_id,)) + cur = await self._db.execute( + "SELECT COALESCE(SUM(cost),0) AS s FROM spend WHERE run_id=?", (run_id,) + ) spent = (await cur.fetchone())["s"] return (spent / limit) if limit > 0 else 1.0 @@ -165,18 +197,28 @@ async def claim_next_task(self, run_id: str, worker_id: str) -> ClaimResult | No else: ratio = await self._spend_ratio_locked(run_id) paths, skip = await self._downgrade_config() - model = sch.resolve_model(row["agent_id"], row["model"], ratio, paths, skip) + model = sch.resolve_model( + row["agent_id"], row["model"], ratio, paths, skip + ) if model != row["model"]: await self._db.execute( - "UPDATE tasks SET model=? WHERE task_id=?", (model, row["task_id"]) + "UPDATE tasks SET model=? WHERE task_id=?", + (model, row["task_id"]), ) input_keys = json.loads(row["input"]).get("input_keys", []) - state = await self.get_state(run_id, input_keys) if input_keys else {} + state = ( + await self.get_state(run_id, input_keys) if input_keys else {} + ) resolved_input = {k: v[0] for k, v in state.items()} result = ClaimResult( - task_id=row["task_id"], run_id=row["run_id"], phase=row["phase"], - phase_order=row["phase_order"], agent_id=row["agent_id"], - input=resolved_input, model=model, version=row["version"], + task_id=row["task_id"], + run_id=row["run_id"], + phase=row["phase"], + phase_order=row["phase_order"], + agent_id=row["agent_id"], + input=resolved_input, + model=model, + version=row["version"], ) return result @@ -204,72 +246,171 @@ async def reap_expired(self) -> int: return rowcount async def spend_total(self, run_id: str) -> float: - cur = await self._db.execute("SELECT COALESCE(SUM(cost),0) AS s FROM spend WHERE run_id=?", (run_id,)) + cur = await self._db.execute( + "SELECT COALESCE(SUM(cost),0) AS s FROM spend WHERE run_id=?", (run_id,) + ) return (await cur.fetchone())["s"] async def _advance_locked(self, run_id: str) -> None: now = self._now() - plan = sch.advance(await self._all_phases(run_id), await self._all_tasks(run_id), self.cfg) + plan = sch.advance( + await self._all_phases(run_id), await self._all_tasks(run_id), self.cfg + ) # gate-pending is derived below via cfg.gate_of; plan["open_gates"] is intentionally unused here. for name in plan["complete_phases"]: - await self._db.execute("UPDATE phases SET status='complete' WHERE run_id=? AND name=?", (run_id, name)) + await self._db.execute( + "UPDATE phases SET status='complete' WHERE run_id=? AND name=?", + (run_id, name), + ) if self.cfg.gate_of(name) != "none": - await self._db.execute("UPDATE phases SET gate='pending' WHERE run_id=? AND name=?", (run_id, name)) + await self._db.execute( + "UPDATE phases SET gate='pending' WHERE run_id=? AND name=?", + (run_id, name), + ) for name in plan["open_phases"]: - await self._db.execute("UPDATE phases SET status='open' WHERE run_id=? AND name=?", (run_id, name)) + await self._db.execute( + "UPDATE phases SET status='open' WHERE run_id=? AND name=?", + (run_id, name), + ) await self._seed_phase_locked(run_id, name, now) await self._recompute_ready_locked(run_id) - async def complete_task(self, task_id, worker_id, version, result, state_writes=None, spawn_tasks=None) -> bool: + async def complete_task( + self, + task_id, + worker_id, + version, + result, + state_writes=None, + spawn_tasks=None, # noqa: ARG002 + ) -> bool: async with self._db_lock: async with self._txn(): cur = await self._db.execute( "UPDATE tasks SET status='done', result=? WHERE task_id=? AND owner=? AND version=? AND status IN ('claimed','running')", - (json.dumps(result), task_id, worker_id, version)) + (json.dumps(result), task_id, worker_id, version), + ) if cur.rowcount != 1: ok = False else: - trow = await (await self._db.execute( - "SELECT run_id, agent_id, model, sim_cost FROM tasks WHERE task_id=?", (task_id,))).fetchone() + trow = await ( + await self._db.execute( + "SELECT run_id, agent_id, model, sim_cost FROM tasks WHERE task_id=?", + (task_id,), + ) + ).fetchone() run_id = trow["run_id"] for k, v in (state_writes or {}).items(): await self._db.execute( "INSERT INTO state (run_id, key, value, version) VALUES (?,?,?,1) " "ON CONFLICT(run_id, key) DO UPDATE SET value=excluded.value, version=state.version+1", - (run_id, k, json.dumps(v))) + (run_id, k, json.dumps(v)), + ) await self._db.execute( "INSERT INTO spend (run_id, task_id, agent_id, cost, model, ts) VALUES (?,?,?,?,?,?)", - (run_id, task_id, trow["agent_id"], trow["sim_cost"], trow["model"], self._now())) + ( + run_id, + task_id, + trow["agent_id"], + trow["sim_cost"], + trow["model"], + self._now(), + ), + ) await self._advance_locked(run_id) ok = True return ok - async def fail_task(self, task_id, worker_id, version, error) -> None: - async with self._db_lock: - async with self._txn(): - trow = await (await self._db.execute( + async def fail_task( + self, task_id, worker_id, version, error # noqa: ARG002 + ) -> None: + async with self._db_lock, self._txn(): + trow = await ( + await self._db.execute( "SELECT run_id, attempts FROM tasks WHERE task_id=? AND owner=? AND version=? AND status IN ('claimed','running')", - (task_id, worker_id, version))).fetchone() - if trow is not None: - attempts = trow["attempts"] + 1 - if attempts >= self.MAX_ATTEMPTS: - await self._db.execute("UPDATE tasks SET status='failed', attempts=? WHERE task_id=?", (attempts, task_id)) - await self._db.execute("UPDATE runs SET status='failed' WHERE run_id=?", (trow["run_id"],)) - else: - await self._db.execute("UPDATE tasks SET status='ready', owner=NULL, version=version+1, attempts=? WHERE task_id=?", (attempts, task_id)) + (task_id, worker_id, version), + ) + ).fetchone() + if trow is not None: + attempts = trow["attempts"] + 1 + if attempts >= self.MAX_ATTEMPTS: + await self._db.execute( + "UPDATE tasks SET status='failed', attempts=? WHERE task_id=?", + (attempts, task_id), + ) + await self._db.execute( + "UPDATE runs SET status='failed' WHERE run_id=?", + (trow["run_id"],), + ) + else: + await self._db.execute( + "UPDATE tasks SET status='ready', owner=NULL, version=version+1, attempts=? WHERE task_id=?", + (attempts, task_id), + ) async def submit_approval(self, run_id: str, phase: str, decision: str) -> None: - async with self._db_lock: - async with self._txn(): - now = self._now() - if decision == "approved": - await self._db.execute("UPDATE phases SET gate='approved' WHERE run_id=? AND name=?", (run_id, phase)) - order = self.cfg.order_of(phase) - nxt = next((n for n in self.cfg.phase_names if self.cfg.order_of(n) == order + 1), None) - if nxt is not None: - await self._db.execute("UPDATE phases SET status='open' WHERE run_id=? AND name=?", (run_id, nxt)) - await self._seed_phase_locked(run_id, nxt, now) - elif decision == "rejected": - await self._db.execute("UPDATE phases SET gate='rejected', status='open' WHERE run_id=? AND name=?", (run_id, phase)) - await self._db.execute("UPDATE tasks SET status='blocked', owner=NULL WHERE run_id=? AND phase=?", (run_id, phase)) - await self._recompute_ready_locked(run_id) + async with self._db_lock, self._txn(): + now = self._now() + if decision == "approved": + await self._db.execute( + "UPDATE phases SET gate='approved' WHERE run_id=? AND name=?", + (run_id, phase), + ) + order = self.cfg.order_of(phase) + nxt = next( + ( + n + for n in self.cfg.phase_names + if self.cfg.order_of(n) == order + 1 + ), + None, + ) + if nxt is not None: + await self._db.execute( + "UPDATE phases SET status='open' WHERE run_id=? AND name=?", + (run_id, nxt), + ) + await self._seed_phase_locked(run_id, nxt, now) + elif decision == "rejected": + await self._db.execute( + "UPDATE phases SET gate='rejected', status='open' WHERE run_id=? AND name=?", + (run_id, phase), + ) + await self._db.execute( + "UPDATE tasks SET status='blocked', owner=NULL WHERE run_id=? AND phase=?", + (run_id, phase), + ) + await self._recompute_ready_locked(run_id) + + async def snapshot(self, run_id: str) -> dict: + cur = await self._db.execute( + "SELECT status FROM runs WHERE run_id=?", (run_id,) + ) + run_row = await cur.fetchone() + run_status = run_row["status"] if run_row else "unknown" + phases = await self._all_phases(run_id) + tasks = await self._all_tasks(run_id) + terminal = max(phases, key=lambda p: p["phase_order"]) if phases else None + if run_status == "failed": + status = "failed" + elif terminal is not None and terminal["status"] == "complete": + status = "done" + else: + status = "running" + return { + "run_id": run_id, + "status": status, + "phases": [ + {"name": p["name"], "status": p["status"], "gate": p["gate"]} + for p in phases + ], + "tasks": [ + { + "agent_id": t["agent_id"], + "phase": t["phase"], + "status": t["status"], + "owner": t["owner"], + } + for t in tasks + ], + } diff --git a/backend/engine/worker.py b/backend/engine/worker.py new file mode 100644 index 0000000..e3ba990 --- /dev/null +++ b/backend/engine/worker.py @@ -0,0 +1,83 @@ +"""Independent worker: claim -> execute (+heartbeat) -> complete/fail loop.""" + +from __future__ import annotations + +import argparse +import asyncio +import contextlib + +from backend.agents.registry import get_registry +from backend.engine.agent_adapter import run_agent_task +from backend.engine.client import EngineClient +from backend.engine.phases import PhasesConfig + + +async def _heartbeat_loop(client, task_id, worker_id, interval): + while True: + await asyncio.sleep(interval) + if not await client.heartbeat(task_id, worker_id): + return # lost the lease + + +async def run_worker( + url, + run_id, + worker_id, + cfg=None, + registry=None, + poll_interval=0.05, + max_poll=2.0, + heartbeat_interval=20.0, +) -> int: + cfg = cfg or PhasesConfig.load() + registry = registry or get_registry() + completed = 0 + backoff = poll_interval + async with EngineClient(url) as client: + while True: + claim = await client.claim_next_task(run_id, worker_id) + if claim is None: + run = await client.get_run(run_id) + if run["status"] in ("done", "failed"): + return completed + await asyncio.sleep(backoff) + backoff = min(backoff * 2, max_poll) + continue + backoff = poll_interval + hb = asyncio.create_task( + _heartbeat_loop(client, claim["task_id"], worker_id, heartbeat_interval) + ) + try: + result, state_writes = await run_agent_task( + claim["agent_id"], + claim["phase"], + claim["input"], + claim["model"], + registry, + cfg, + ) + if await client.complete_task( + claim["task_id"], worker_id, claim["version"], result, state_writes + ): + completed += 1 + except Exception as e: # noqa: BLE001 + await client.fail_task( + claim["task_id"], worker_id, claim["version"], str(e) + ) + finally: + hb.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await hb + + +def main() -> None: + p = argparse.ArgumentParser() + p.add_argument("--server-url", required=True) + p.add_argument("--run-id", required=True) + p.add_argument("--worker-id", required=True) + a = p.parse_args() + asyncio.run(run_worker(a.server_url, a.run_id, a.worker_id)) + + +if __name__ == "__main__": + main() diff --git a/docs/superpowers/plans/2026-07-24-mcp-orchestration-engine-plan-b-server-workers.md b/docs/superpowers/plans/2026-07-24-mcp-orchestration-engine-plan-b-server-workers.md new file mode 100644 index 0000000..32b591d --- /dev/null +++ b/docs/superpowers/plans/2026-07-24-mcp-orchestration-engine-plan-b-server-workers.md @@ -0,0 +1,1041 @@ +# MCP Orchestration Engine — Plan B: MCP Server + Worker Processes Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Expose the Plan A coordination core as a genuine MCP state server (FastMCP, streamable-HTTP), and drive it from independent OS worker processes plus a run controller/CLI — so a product idea traverses all six phases end-to-end in mock mode across real processes. + +**Architecture:** A standalone FastMCP server wraps the single-writer `Store` and exposes its operations as MCP tools (JSON-string in/out). An `EngineClient` MCP client wraps those tools. `worker.py` is a separate OS process running a claim→execute(+heartbeat)→complete loop via `EngineClient`, executing agents through the existing registry. `run.py` boots the server, spawns N worker subprocesses, seeds the run, and drives approval gates. Full design: [`docs/superpowers/specs/2026-07-23-parallel-mcp-orchestration-engine-design.md`](../specs/2026-07-23-parallel-mcp-orchestration-engine-design.md). Builds on Plan A (`backend/engine/` store/scheduler/models/phases). + +**Tech Stack:** Python 3.11+, UV, `mcp==1.28.1` (FastMCP + streamable-HTTP client), `uvicorn`, `aiosqlite`, pytest + pytest-asyncio. Windows/win32. + +## Global Constraints + +- **Python** `>=3.11`; deps via **UV** (`uv add`, `uv run`). `mcp` is already pinned (`>=1.16,<2`, resolved 1.28.1) from Plan A. +- **Platform is Windows (win32).** Worker subprocesses launch as `python -m backend.engine.worker ...` (spawn — never rely on `fork`/POSIX signals). Close the server's `Store` aiosqlite connection before any `tmp_path` teardown (WAL files raise `PermissionError` otherwise). Subprocesses are terminated with `Process.terminate()` / `proc.kill()`, not POSIX signals. +- **Builds on Plan A** (`feat/parallel-mcp-orchestration-engine`, PR #5). Do NOT modify `store.py` write-path methods; the only Plan-A file changed is an ADDITIVE read-only `Store.snapshot` (Task 3). Do NOT touch `backend/graph.py`/`orchestrator.py` (LangGraph retirement is Plan C). +- **MCP tool convention (verified against mcp 1.28.1 — use everywhere):** + - Server: `@mcp.tool()` `async def name(, _json: str = "null") -> str:` — parse each `*_json` param with `json.loads`, call the store, `return json.dumps()`. Never return a bare dict/None; always a JSON string. + - Client: `res = await session.call_tool(name, args); if res.isError: raise RuntimeError(res.content[0].text); return json.loads(res.content[0].text)`. + - Server object: `FastMCP("appforge-state", stateless_http=True)`; ASGI app via `mcp.streamable_http_app()` under uvicorn; client connects to `http://host:port/mcp`. +- **`appforge_mcp_server.py`** entry alias at repo root (matches the resume artifact name), delegating to `backend.engine.state_server`. +- **Mock mode:** tests + the documented run use `MOCK_AGENTS=true` (default). No real Anthropic calls. +- **Embargo:** the string "OpenBarclay" must not appear anywhere. +- **Commits:** conventional style, no `Co-Authored-By`/attribution footer. +- Six phases + agents are fixed by `config/phases.yaml` (Plan A). The terminal phase is `iterate`. + +--- + +## File Structure + +| File | Responsibility | +|---|---| +| `backend/engine/state_server.py` | build + serve the FastMCP server; own the `Store`; run the reaper; base-model loading | +| `backend/engine/mcp_tools.py` | `register_tools(mcp, store)` — all MCP tool definitions (JSON in/out) | +| `backend/engine/client.py` | `EngineClient` async MCP-client wrapper (used by worker + controller + tests) | +| `backend/engine/agent_adapter.py` | `run_agent_task(...)` — task→agent bridge + the Clarify Q&A loop | +| `backend/engine/worker.py` | worker process loop (claim/execute/heartbeat/complete/fail) + `__main__` | +| `backend/engine/run.py` | controller/CLI: boot server, spawn workers, seed, drive gates + `__main__` | +| `appforge_mcp_server.py` | repo-root entry alias → `backend.engine.state_server` | +| `tests/engine/server_harness.py` | `running_server(db_path)` async CM + `free_port()` (test helper, not a test) | +| `backend/engine/store.py` | Task 3 only: ADD read-only `snapshot(run_id)` | +| `tests/engine/test_server_*.py`, `test_agent_adapter.py`, `test_worker*.py`, `test_run_e2e.py` | tests | +| `pyproject.toml` | Task 6: add `[project.scripts] appforge = "backend.engine.run:main"` | + +--- + +## Task 1: Server foundation + client + state-sharing tools (MCP state server DoD) + +**Files:** +- Create: `backend/engine/state_server.py`, `backend/engine/mcp_tools.py`, `backend/engine/client.py`, `tests/engine/server_harness.py` +- Test: `tests/engine/test_server_state.py` + +**Interfaces:** +- Consumes: Plan A `Store`, `PhasesConfig` (`backend/engine/store.py`, `phases.py`). +- Produces: + - `state_server.base_models_from_config(path="config/agents.yaml") -> dict[str,str]` + - `state_server.build_server(db_path, cfg=None, base_models=None, lease_s=120.0) -> tuple[FastMCP, Store]` + - `state_server.serve(db_path, host, port, cfg=None, base_models=None, lease_s=120.0, reaper_interval=30.0)` (async; connects store, starts reaper, runs uvicorn) + - `mcp_tools.register_tools(mcp, store)` registering tools `create_run`, `get_state`, `put_state` + - `client.EngineClient(url)` async CM with `create_run(idea, budget_limit=200.0) -> str`, `get_state(run_id, keys=None) -> dict[str, {"value","version"}]`, `put_state(run_id, key, value, expected_version) -> bool` + - `server_harness.free_port() -> int`, `server_harness.running_server(db_path, **kw)` async CM yielding the base URL + +- [ ] **Step 1: Write the failing test (two independent clients share state through the server)** + +`tests/engine/test_server_state.py`: +```python +import pytest + +from backend.engine.client import EngineClient +from tests.engine.server_harness import running_server + + +async def test_two_clients_share_state_through_server(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url: + # Client A creates a run and writes state + async with EngineClient(url) as a: + run_id = await a.create_run("Build a todo app", 5.0) + assert await a.put_state(run_id, "prd", {"text": "v1"}, expected_version=0) is True + # A SEPARATE client B reads it back through the server + async with EngineClient(url) as b: + state = await b.get_state(run_id, ["prd"]) + assert state["prd"]["value"] == {"text": "v1"} + assert state["prd"]["version"] == 1 + # CAS conflict path is observable across clients + assert await b.put_state(run_id, "prd", {"text": "stale"}, expected_version=0) is False + + +async def test_create_run_seeds_clarify(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url: + async with EngineClient(url) as c: + run_id = await c.create_run("idea", 5.0) + assert isinstance(run_id, str) and run_id +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/engine/test_server_state.py -v` +Expected: FAIL — `ModuleNotFoundError: backend.engine.client` / `tests.engine.server_harness`. + +- [ ] **Step 3: Write `backend/engine/mcp_tools.py`** + +```python +"""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}) +``` + +- [ ] **Step 4: Write `backend/engine/state_server.py`** + +```python +"""Standalone FastMCP state server wrapping the single-writer Store.""" +from __future__ import annotations + +import asyncio +from pathlib import Path + +import yaml +from mcp.server.fastmcp import FastMCP + +from backend.engine.mcp_tools import register_tools +from backend.engine.phases import PhasesConfig +from backend.engine.store import Store + + +def base_models_from_config(path: str = "config/agents.yaml") -> dict[str, str]: + raw = yaml.safe_load(Path(path).read_text(encoding="utf-8")) + return {aid: a.get("llm", {}).get("model") for aid, a in raw.get("agents", {}).items()} + + +def build_server(db_path, cfg=None, base_models=None, lease_s: float = 120.0): + cfg = cfg or PhasesConfig.load() + base_models = base_models if base_models is not None else base_models_from_config() + store = Store(db_path, cfg, base_models, lease_s=lease_s) + mcp = FastMCP("appforge-state", stateless_http=True) + register_tools(mcp, store) + return mcp, store + + +async def _reaper_loop(store: Store, interval: float) -> None: + while True: + await asyncio.sleep(interval) + try: + await store.reap_expired() + except Exception: # noqa: BLE001 - reaper must never crash the server + pass + + +async def serve(db_path, host="127.0.0.1", port=8800, cfg=None, base_models=None, + lease_s: float = 120.0, reaper_interval: float = 30.0) -> None: + import uvicorn + + mcp, store = build_server(db_path, cfg, base_models, lease_s) + await store.connect() + reaper = asyncio.create_task(_reaper_loop(store, reaper_interval)) + app = mcp.streamable_http_app() + server = uvicorn.Server(uvicorn.Config(app, host=host, port=port, log_level="error")) + try: + await server.serve() + finally: + reaper.cancel() + await store.close() + + +if __name__ == "__main__": + import argparse + + 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)) +``` + +- [ ] **Step 5: Write `backend/engine/client.py`** + +```python +"""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 streamablehttp_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() + r, w, _ = await self._stack.enter_async_context(streamablehttp_client(self.url)) + self._session = await self._stack.enter_async_context(ClientSession(r, w)) + await self._session.initialize() + 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: + raise RuntimeError(f"{name} failed: {res.content[0].text}") + 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"] +``` + +- [ ] **Step 6: Write `tests/engine/server_harness.py`** + +```python +"""Test helper: run the state server in-process on an ephemeral port.""" +from __future__ import annotations + +import asyncio +import socket +from contextlib import asynccontextmanager + +import httpx + +from backend.engine.state_server import build_server + + +def free_port() -> int: + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +@asynccontextmanager +async def running_server(db_path: str, **kw): + import uvicorn + + port = free_port() + mcp, store = build_server(db_path, **kw) + await store.connect() + app = mcp.streamable_http_app() + server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error")) + task = asyncio.create_task(server.serve()) + url = f"http://127.0.0.1:{port}/mcp" + # wait until the port accepts connections + for _ in range(100): + try: + async with httpx.AsyncClient() as h: + await h.get(f"http://127.0.0.1:{port}/mcp", timeout=0.2) + break + except Exception: # noqa: BLE001 + await asyncio.sleep(0.05) + try: + yield url + finally: + server.should_exit = True + await task + await store.close() # close before tmp_path teardown (win32 WAL) +``` +(`httpx` is already a dev dependency. A non-200 from the GET is fine — it proves the port is bound.) + +- [ ] **Step 7: Run tests to verify they pass** + +Run: `uv run pytest tests/engine/test_server_state.py -v` +Expected: 2 passed. + +- [ ] **Step 8: Commit** + +```bash +git add backend/engine/state_server.py backend/engine/mcp_tools.py backend/engine/client.py tests/engine/server_harness.py tests/engine/test_server_state.py +git commit -m "feat(engine): FastMCP state server + client + state-sharing tools" +``` + +--- + +## Task 2: Task-lifecycle tools (claim / complete / heartbeat / fail) + +**Files:** +- Modify: `backend/engine/mcp_tools.py`, `backend/engine/client.py` +- Test: `tests/engine/test_server_lifecycle.py` + +**Interfaces:** +- Consumes: Task 1 server/client; Plan A `Store.claim_next_task/complete_task/heartbeat/fail_task`, `ClaimResult`. +- Produces (on `EngineClient`): + - `claim_next_task(run_id, worker_id) -> dict | None` (keys: `task_id, phase, agent_id, input, model, version`) + - `complete_task(task_id, worker_id, version, result, state_writes=None) -> bool` + - `heartbeat(task_id, worker_id) -> bool` + - `fail_task(task_id, worker_id, version, error) -> None` + +- [ ] **Step 1: Write the failing test** + +`tests/engine/test_server_lifecycle.py`: +```python +from backend.engine.client import EngineClient +from tests.engine.server_harness import running_server + + +async def test_claim_complete_advances_to_prd_gate(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url: + async with EngineClient(url) as c: + run_id = await c.create_run("idea", 200.0) + claim = await c.claim_next_task(run_id, "w1") + assert claim is not None and claim["agent_id"] == "clarifying_pm" + ok = await c.complete_task(claim["task_id"], "w1", claim["version"], + result={"prd": "PRD"}, state_writes={"prd": "PRD"}) + assert ok is True + # behind the pending PRD gate nothing is claimable + assert await c.claim_next_task(run_id, "w2") is None + st = await c.get_state(run_id, ["prd"]) + assert st["prd"]["value"] == "PRD" + + +async def test_complete_wrong_version_rejected(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url: + async with EngineClient(url) as c: + run_id = await c.create_run("idea", 200.0) + claim = await c.claim_next_task(run_id, "w1") + assert await c.complete_task(claim["task_id"], "w1", 999, result={}, state_writes=None) is False + + +async def test_heartbeat_owner_guarded(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url: + async with EngineClient(url) as c: + run_id = await c.create_run("idea", 200.0) + claim = await c.claim_next_task(run_id, "w1") + assert await c.heartbeat(claim["task_id"], "w1") is True + assert await c.heartbeat(claim["task_id"], "w2") is False +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/engine/test_server_lifecycle.py -v` +Expected: FAIL — `AttributeError: 'EngineClient' object has no attribute 'claim_next_task'`. + +- [ ] **Step 3: Add the tools to `mcp_tools.py`** (inside `register_tools`, after the Task 1 tools) + +```python + @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}) +``` + +- [ ] **Step 4: Add the methods to `EngineClient`** + +```python + 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) +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `uv run pytest tests/engine/test_server_lifecycle.py -v` +Expected: 3 passed. + +- [ ] **Step 6: Commit** + +```bash +git add backend/engine/mcp_tools.py backend/engine/client.py tests/engine/test_server_lifecycle.py +git commit -m "feat(engine): claim/complete/heartbeat/fail MCP tools + client methods" +``` + +--- + +## Task 3: Gate + snapshot tools (submit_approval, get_run) + `Store.snapshot` + +**Files:** +- Modify: `backend/engine/store.py` (ADD read-only `snapshot`), `backend/engine/mcp_tools.py`, `backend/engine/client.py` +- Test: `tests/engine/test_server_gate.py` + +**Interfaces:** +- Consumes: Plan A `Store.submit_approval`, `_all_phases`, `_all_tasks`. +- Produces: + - `Store.snapshot(run_id) -> dict` with keys `run_id`, `status` (`"running"|"done"|"failed"`), `phases` (list of `{name, status, gate}`), `tasks` (list of `{agent_id, phase, status, owner}`). `status` is derived: `"failed"` if the run row is failed; `"done"` if the terminal (`iterate`) phase is `complete`; else `"running"`. (Read-only; does NOT change the write path.) + - `EngineClient.submit_approval(run_id, phase, decision) -> None`, `EngineClient.get_run(run_id) -> dict` + +- [ ] **Step 1: Write the failing test** + +`tests/engine/test_server_gate.py`: +```python +from backend.engine.client import EngineClient +from tests.engine.server_harness import running_server + + +async def _drive_clarify(c, run_id): + claim = await c.claim_next_task(run_id, "w1") + await c.complete_task(claim["task_id"], "w1", claim["version"], {"prd": "PRD"}, {"prd": "PRD"}) + + +async def test_approval_opens_design(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url: + async with EngineClient(url) as c: + run_id = await c.create_run("idea", 200.0) + await _drive_clarify(c, run_id) + run = await c.get_run(run_id) + assert run["status"] == "running" + clarify = next(p for p in run["phases"] if p["name"] == "clarify") + assert clarify["gate"] == "pending" + await c.submit_approval(run_id, "clarify", "approved") + # design now has 3 claimable tasks + got = set() + for w in ("w1", "w2", "w3"): + claim = await c.claim_next_task(run_id, w) + assert claim is not None + got.add(claim["agent_id"]) + assert got == {"solution_architect", "tech_lead", "uiux_designer"} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/engine/test_server_gate.py -v` +Expected: FAIL — `AttributeError: 'EngineClient' object has no attribute 'submit_approval'`. + +- [ ] **Step 3: Add `snapshot` to `Store` (`backend/engine/store.py`)** + +```python + async def snapshot(self, run_id: str) -> dict: + cur = await self._db.execute("SELECT status FROM runs WHERE run_id=?", (run_id,)) + run_row = await cur.fetchone() + run_status = run_row["status"] if run_row else "unknown" + phases = await self._all_phases(run_id) + tasks = await self._all_tasks(run_id) + terminal = max(phases, key=lambda p: p["phase_order"]) if phases else None + if run_status == "failed": + status = "failed" + elif terminal is not None and terminal["status"] == "complete": + status = "done" + else: + status = "running" + return { + "run_id": run_id, + "status": status, + "phases": [{"name": p["name"], "status": p["status"], "gate": p["gate"]} for p in phases], + "tasks": [{"agent_id": t["agent_id"], "phase": t["phase"], + "status": t["status"], "owner": t["owner"]} for t in tasks], + } +``` + +- [ ] **Step 4: Add the tools to `mcp_tools.py`** + +```python + @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)) +``` + +- [ ] **Step 5: Add the methods to `EngineClient`** + +```python + 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) +``` + +- [ ] **Step 6: Run tests + full engine suite** + +Run: `uv run pytest tests/engine/test_server_gate.py -v` (expect 1 passed), then `uv run pytest tests/engine -q` (all green). + +- [ ] **Step 7: Commit** + +```bash +git add backend/engine/store.py backend/engine/mcp_tools.py backend/engine/client.py tests/engine/test_server_gate.py +git commit -m "feat(engine): approval + snapshot MCP tools; read-only Store.snapshot" +``` + +--- + +## Task 4: Agent adapter + Clarify Q&A loop + +**Files:** +- Create: `backend/engine/agent_adapter.py` +- Test: `tests/engine/test_agent_adapter.py` + +**Interfaces:** +- Consumes: existing `backend/agents/registry.py` (`get_registry().get(agent_id, mock=...)`), `PhasesConfig` (`agents_of(phase).writes`), `backend/config.py` (`MAX_CLARIFYING_QUESTIONS`). +- Produces: + - `run_agent_task(agent_id, phase, task_input: dict, model, registry, cfg, max_questions=6) -> tuple[dict, dict]` returning `(result, state_writes)`. `result` = `{"agent_id", "output"}`; `state_writes` = `{writes_key: output}` where `writes_key = cfg.agents_of(phase)[agent_id].writes`. + - The Clarify agent (`agent_id == "clarifying_pm"`) runs an internal Q&A loop (`clarifying_pm` ↔ `product_owner`) terminating in a PRD. + +**Grounding (verified against `backend/agents/mock_agent.py`, mock mode):** +- Base `MockAgent.execute(self, task)` **ignores `task` entirely** (`# noqa: ARG002 (mock ignores task)`) and returns an `AgentResult` with `.artifact = "Mock output from {name}"` (a string). So passing a plain `dict` is safe for all 9 bare mocks (frontend/backend/database/ai_ml/security/devops/qa_test/technical_writer/delivery_summarizer/product_owner). **It also sleeps `config.get("delay", 1.0)` = ~1s per call** (see timing note in Tasks 5-6). +- `ClarifyingPmAgent.execute(dict)` returns `{"artifact": {"question": "..."}}` while `len(task["answers"]) < 3`, else `{"artifact": {"prd": "..."}}`. `ProductOwnerAgent` is a bare mock → `AgentResult(.artifact="Mock output from product_owner")`. So the loop below appends 3 answers over 3 questions, and the 4th clarifier call returns the PRD — **it terminates without a human**. The extractors handle both `AgentResult` (attribute `.artifact`) and dict (`["artifact"]`) shapes. +- Specialized mocks return their writes-key inside `artifact`: `solution_architect→{"adr":...}`, `tech_lead→{"tasks":[...]}`, `uiux_designer→{"design_spec":{...}}`. `_writes_value` reaches inside the dict; bare-string artifacts pass through unchanged. +- Registry resolution in mock mode: `registry.get(agent_id)` (env `MOCK_AGENTS=true`) resolves each id to its `mock_agent.` (specialized where defined, base behavior otherwise). Pass agents a plain `dict` (never a bare `AgentTask`). + +- [ ] **Step 1: Write the failing test** (uses the real mock agents via the registry) + +`tests/engine/test_agent_adapter.py`: +```python +import os + +import pytest + +from backend.agents.registry import get_registry, reset_registry +from backend.engine.agent_adapter import run_agent_task +from backend.engine.phases import PhasesConfig + +CFG = PhasesConfig.load("config/phases.yaml") + + +@pytest.fixture(autouse=True) +def mock_mode(): + os.environ["MOCK_AGENTS"] = "true" + reset_registry() + yield + reset_registry() + + +async def test_clarify_loop_yields_prd(tmp_path): + reg = get_registry() + result, writes = await run_agent_task("clarifying_pm", "clarify", + {"idea": "todo app"}, "m", reg, CFG, max_questions=6) + assert "prd" in writes and writes["prd"] # PRD produced without a human + assert result["agent_id"] == "clarifying_pm" + + +async def test_generic_agent_writes_its_key(tmp_path): + reg = get_registry() + # solution_architect writes 'adr' + result, writes = await run_agent_task("solution_architect", "design", + {"prd": "PRD"}, "m", reg, CFG) + assert "adr" in writes and writes["adr"] +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/engine/test_agent_adapter.py -v` +Expected: FAIL — `ModuleNotFoundError: backend.engine.agent_adapter`. + +- [ ] **Step 3: Implement `backend/engine/agent_adapter.py`** + +```python +"""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} +``` +Note: if the real mock `ClarifyingPmAgent` field names differ from `question`/`prd`/`final_prd`, adjust `_run_clarify_loop`'s reads so the test passes — the test (PRD produced) is the contract. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/engine/test_agent_adapter.py -v` +Expected: 2 passed. + +- [ ] **Step 5: Commit** + +```bash +git add backend/engine/agent_adapter.py tests/engine/test_agent_adapter.py +git commit -m "feat(engine): agent adapter + Clarify Q&A loop" +``` + +--- + +## Task 5: Worker process loop + +**Files:** +- Create: `backend/engine/worker.py` +- Test: `tests/engine/test_worker.py` + +**Interfaces:** +- Consumes: `EngineClient` (Tasks 1-3), `run_agent_task` (Task 4), the registry, `PhasesConfig`. +- Produces: + - `run_worker(url, run_id, worker_id, cfg=None, registry=None, poll_interval=0.05, max_poll=2.0, heartbeat_interval=20.0)` — async; loops claim→execute(+heartbeat)→complete/fail until `get_run().status in {done, failed}`; returns the count of tasks it completed. + - `worker.main()` + `__main__` for `python -m backend.engine.worker --server-url URL --run-id RID --worker-id WID`. + +- [ ] **Step 1: Write the failing test** (one in-process worker drives a full mock run; the test auto-approves gates concurrently) + +`tests/engine/test_worker.py`: +```python +import asyncio +import os + +import pytest + +from backend.agents.registry import reset_registry +from backend.engine.client import EngineClient +from backend.engine.worker import run_worker +from tests.engine.server_harness import running_server + + +@pytest.fixture(autouse=True) +def mock_mode(): + os.environ["MOCK_AGENTS"] = "true" + reset_registry() + yield + reset_registry() + + +async def _auto_approver(url, run_id, stop): + async with EngineClient(url) as c: + while not stop.is_set(): + run = await c.get_run(run_id) + for p in run["phases"]: + if p["gate"] == "pending": + await c.submit_approval(run_id, p["name"], "approved") + if run["status"] in ("done", "failed"): + return + await asyncio.sleep(0.05) + + +async def test_single_worker_completes_all_phases(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url: + async with EngineClient(url) as c: + run_id = await c.create_run("todo app", 200.0) + stop = asyncio.Event() + approver = asyncio.create_task(_auto_approver(url, run_id, stop)) + completed = await run_worker(url, run_id, "w1") + stop.set() + await approver + async with EngineClient(url) as c: + run = await c.get_run(run_id) + assert run["status"] == "done" + phases_done = {p["name"] for p in run["phases"] if p["status"] == "complete"} + assert phases_done == {"clarify", "design", "code", "test", "deploy", "iterate"} + assert completed >= 13 # all phase-worker tasks ran +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/engine/test_worker.py -v` +Expected: FAIL — `ModuleNotFoundError: backend.engine.worker`. + +- [ ] **Step 3: Implement `backend/engine/worker.py`** + +```python +"""Independent worker: claim -> execute (+heartbeat) -> complete/fail loop.""" +from __future__ import annotations + +import argparse +import asyncio + +from backend.agents.registry import get_registry +from backend.engine.agent_adapter import run_agent_task +from backend.engine.client import EngineClient +from backend.engine.phases import PhasesConfig + + +async def _heartbeat_loop(client, task_id, worker_id, interval): + while True: + await asyncio.sleep(interval) + if not await client.heartbeat(task_id, worker_id): + return # lost the lease + + +async def run_worker(url, run_id, worker_id, cfg=None, registry=None, + poll_interval=0.05, max_poll=2.0, heartbeat_interval=20.0) -> int: + cfg = cfg or PhasesConfig.load() + registry = registry or get_registry() + completed = 0 + backoff = poll_interval + async with EngineClient(url) as client: + while True: + claim = await client.claim_next_task(run_id, worker_id) + if claim is None: + run = await client.get_run(run_id) + if run["status"] in ("done", "failed"): + return completed + await asyncio.sleep(backoff) + backoff = min(backoff * 2, max_poll) + continue + backoff = poll_interval + hb = asyncio.create_task( + _heartbeat_loop(client, claim["task_id"], worker_id, heartbeat_interval) + ) + try: + result, state_writes = await run_agent_task( + claim["agent_id"], claim["phase"], claim["input"], claim["model"], registry, cfg + ) + await client.complete_task( + claim["task_id"], worker_id, claim["version"], result, state_writes + ) + completed += 1 + except Exception as e: # noqa: BLE001 + await client.fail_task(claim["task_id"], worker_id, claim["version"], str(e)) + finally: + hb.cancel() + + +def main() -> None: + p = argparse.ArgumentParser() + p.add_argument("--server-url", required=True) + p.add_argument("--run-id", required=True) + p.add_argument("--worker-id", required=True) + a = p.parse_args() + asyncio.run(run_worker(a.server_url, a.run_id, a.worker_id)) + + +if __name__ == "__main__": + main() +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/engine/test_worker.py -v` +Expected: 1 passed. (If it hangs, the Clarify loop from Task 4 isn't producing a `prd` — fix there.) + +- [ ] **Step 5: Commit** + +```bash +git add backend/engine/worker.py tests/engine/test_worker.py +git commit -m "feat(engine): worker process loop (claim/execute/heartbeat/complete)" +``` + +--- + +## Task 6: Run controller / CLI (multi-process end-to-end) + +**Files:** +- Create: `backend/engine/run.py`, `appforge_mcp_server.py` (repo-root alias) +- Modify: `pyproject.toml` (`[project.scripts] appforge = "backend.engine.run:main"`) +- Test: `tests/engine/test_run_e2e.py` + +**Interfaces:** +- Consumes: `state_server.serve`, `EngineClient`, `worker` module (as a subprocess). +- Produces: + - `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` — boots the server in-process, spawns N `python -m backend.engine.worker` SUBPROCESSES, seeds the run, drives gates (auto-approve), waits for `done`/`failed`/timeout, tears everything down, and returns the final snapshot plus `{"worker_pids": [...]}`. + - `run.main()` + `__main__` for `appforge run "" --workers N [--budget-limit X] [--no-auto-approve]`. + +- [ ] **Step 1: Write the failing test** (real worker SUBPROCESSES → genuine multi-process) + +`tests/engine/test_run_e2e.py`: +```python +import os + +import pytest + +from backend.engine.run import run_pipeline + + +@pytest.fixture(autouse=True) +def mock_mode(): + os.environ["MOCK_AGENTS"] = "true" + yield + + +async def test_end_to_end_multiprocess_run(tmp_path): + result = await run_pipeline( + "Build a todo app", workers=3, budget_limit=200.0, + db_path=str(tmp_path / "run.db"), timeout=90.0, + ) + assert result["snapshot"]["status"] == "done" + done = {p["name"] for p in result["snapshot"]["phases"] if p["status"] == "complete"} + assert done == {"clarify", "design", "code", "test", "deploy", "iterate"} + # genuine multi-process execution: >1 distinct worker PID actually ran + assert len(set(result["worker_pids"])) >= 1 # subprocesses spawned +``` +(The assertion is `>=1` because with mock timing one fast worker can drain the DAG; the point is the workers are real OS subprocesses. The multi-worker *contention* proof is the Plan C stress test.) + +- [ ] **Step 2: Run test to verify it fails** + +Run: `uv run pytest tests/engine/test_run_e2e.py -v` +Expected: FAIL — `ModuleNotFoundError: backend.engine.run`. + +- [ ] **Step 3: Implement `backend/engine/run.py`** + +```python +"""Run controller / CLI: boot server, spawn worker subprocesses, drive gates.""" +from __future__ import annotations + +import argparse +import asyncio +import sys + +from backend.engine.client import EngineClient +from backend.engine.state_server import serve +from tests.engine.server_harness import free_port # reused; pure helper + + +async def _drive_gates(url, run_id, auto_approve, timeout, poll): + loop = asyncio.get_event_loop() + deadline = loop.time() + timeout + async with EngineClient(url) as c: + while True: + run = await c.get_run(run_id) + if run["status"] in ("done", "failed"): + return run + if auto_approve: + for p in run["phases"]: + if p["gate"] == "pending": + await c.submit_approval(run_id, p["name"], "approved") + if loop.time() > deadline: + return run + await asyncio.sleep(poll) + + +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: + 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)) + # wait for readiness by creating the run (retries until the server answers) + run_id = None + 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 + await asyncio.sleep(0.05) + if run_id is None: + server_task.cancel() + raise RuntimeError("state server failed to start") + + procs = [] + for i in range(workers): + procs.append(await asyncio.create_subprocess_exec( + sys.executable, "-m", "backend.engine.worker", + "--server-url", url, "--run-id", run_id, "--worker-id", f"w{i}", + )) + worker_pids = [p.pid for p in procs] + + try: + 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 asyncio.TimeoutError: + p.kill() + server_task.cancel() + try: + await server_task + except asyncio.CancelledError: + pass + + return {"run_id": run_id, "snapshot": final, "worker_pids": worker_pids} + + +def main() -> None: + p = argparse.ArgumentParser(prog="appforge") + sub = p.add_subparsers(dest="cmd", required=True) + r = sub.add_parser("run") + r.add_argument("idea") + r.add_argument("--workers", type=int, default=4) + r.add_argument("--budget-limit", type=float, default=200.0) + r.add_argument("--no-auto-approve", action="store_true") + a = p.parse_args() + result = asyncio.run(run_pipeline( + a.idea, workers=a.workers, budget_limit=a.budget_limit, auto_approve=not a.no_auto_approve + )) + print(f"run {result['run_id']}: {result['snapshot']['status']}") + for ph in result["snapshot"]["phases"]: + print(f" {ph['name']:9} {ph['status']:9} gate={ph['gate']}") + + +if __name__ == "__main__": + main() +``` +Note: importing `free_port` from the test harness keeps one implementation; if a reviewer objects to importing a test helper into production code, move `free_port` into `state_server.py` and import it from there in both places (do this if flagged — it's a 3-line move). + +- [ ] **Step 4: Write `appforge_mcp_server.py` (repo-root alias)** + +```python +"""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)) +``` + +- [ ] **Step 5: Add the console script to `pyproject.toml`** + +Under `[project]` add (create the table if absent): +```toml +[project.scripts] +appforge = "backend.engine.run:main" +``` +Then `uv sync` so the entry point is installed. + +- [ ] **Step 6: Run the e2e test + full engine suite** + +Run: `uv run pytest tests/engine/test_run_e2e.py -v` (expect 1 passed — real subprocesses), then `uv run pytest tests/engine -q` (all green). Note: this test spawns processes and may take ~10-30s. + +- [ ] **Step 7: Commit** + +```bash +git add backend/engine/run.py appforge_mcp_server.py pyproject.toml uv.lock tests/engine/test_run_e2e.py +git commit -m "feat(engine): run controller/CLI + multi-process e2e + appforge_mcp_server alias" +``` + +--- + +## Self-Review (against the spec) + +**Spec coverage (Plan B scope = spec §4 components, §7 tool surface, §12 steps 4-6):** +- §4/§7 MCP state server + tools → Tasks 1-3 (`state_server`, `mcp_tools`, all tools: create_run, get_state, put_state, claim_next_task, complete_task, heartbeat, fail_task, submit_approval, get_run). §7 "two-process sharing (DoD)" → Task 1 `test_two_clients_share_state_through_server`. +- §4 worker + §3 Clarify loop + §7 adapter dict-contract → Tasks 4-5 (`agent_adapter` with defensive extraction + Clarify loop; `worker` with background heartbeat). +- §4 run controller + §10 documented-run scaffolding → Task 6 (`run.py`, `appforge` CLI, `appforge_mcp_server.py`), real worker subprocesses. +- **Deferred to Plan C (noted):** live claim-time budget downgrade tuning + `test_budget_downgrade_live` (§8), the N≫workers concurrency stress test `test_concurrency_no_collision` (§9), the preserved documented-run artifacts under `docs/runs/` (§10), reaper `MAX_ATTEMPTS` cap, LangGraph retirement + 25-test migration (§11), events Socket.IO bridge (§11). + +**Placeholder scan:** no TBD/TODO; every code step is complete. Two prose notes (Task 4 mock-field-name verification; Task 6 `free_port` location) flag review-time adjustments, not missing code. + +**Type consistency:** `EngineClient` method signatures match their tool params across Tasks 1-3; `claim` dict keys (`task_id, phase, agent_id, input, model, version`) are produced by `claim_next_task` (Task 2, from `ClaimResult.model_dump()`) and consumed identically by `worker.run_worker` (Task 5) and the adapter (`run_agent_task(agent_id, phase, task_input, model, ...)`, Task 4). `snapshot` shape (Task 3) is consumed by `_auto_approver`/`_drive_gates` (Tasks 5-6) via `run["phases"][*]["gate"]` and `run["status"]`. `run_agent_task` returns `(result, state_writes)` consumed by `worker`. + +**Clarify loop — verified, not assumed:** the mock field names (`question`/`prd`) and base-mock task-ignoring behavior were confirmed by reading `mock_agent.py`; the loop terminates with a real PRD after 3 mock Q&A rounds. The `_run_clarify_loop` fallback is a belt-and-suspenders guard so Task 5's full-run test can never deadlock on Clarify. + +**Timing:** base `MockAgent.execute` sleeps ~1s/call (no `delay` in `AgentConfig`). The Clarify task alone is ~7s (4 clarifier + 3 PO calls); a full single-worker run is ~20s and the multi-worker e2e ~10-25s. The `test_worker` and `test_run_e2e` integration tests are therefore slow-but-bounded — expect tens of seconds, not a hang. If a run exceeds its timeout, suspect the Clarify loop or a worker crash, not the delay. diff --git a/pyproject.toml b/pyproject.toml index fe43088..3e4faa3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,6 +58,9 @@ dev = [ "httpx>=0.27.0", ] +[project.scripts] +appforge = "backend.engine.run:main" + [project.urls] Homepage = "https://github.com/your-repo/devteam-ai-2025" Documentation = "https://github.com/your-repo/devteam-ai-2025#readme" diff --git a/tests/engine/server_harness.py b/tests/engine/server_harness.py new file mode 100644 index 0000000..0f14c7a --- /dev/null +++ b/tests/engine/server_harness.py @@ -0,0 +1,87 @@ +"""Test helper: run the state server in a dedicated background thread. + +Running uvicorn as an asyncio task inside the test's own event loop +contaminates resources across tests in the same process (the FastMCP +streamable-HTTP session manager / loop is not fully torn down between tests, +producing "ASGI callable returned without completing response" and hangs on +the second server). Running the server in its OWN thread + event loop fully +isolates it: the test's client talks to it over HTTP, and stopping joins the +thread (which closes the Store connection before tmp_path teardown — required +on win32 for the WAL files). +""" + +from __future__ import annotations + +import asyncio +import threading +from contextlib import asynccontextmanager + +from backend.engine.state_server import build_server, free_port + + +class _ServerThread: + def __init__(self, db_path: str, **kw): + self.db_path = db_path + self.kw = kw + self.port = free_port() + self._server = None + self._ready = threading.Event() + self._thread = threading.Thread(target=self._run, daemon=True) + + def _run(self) -> None: + import uvicorn + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + async def _serve() -> None: + mcp, store = build_server(self.db_path, **self.kw) + await store.connect() + app = mcp.streamable_http_app() + config = uvicorn.Config( + app, host="127.0.0.1", port=self.port, log_level="error" + ) + self._server = uvicorn.Server(config) + self._server.install_signal_handlers = lambda: None + + async def _watch_ready() -> None: + while not self._server.started: + await asyncio.sleep(0.01) + self._ready.set() + + watcher = asyncio.create_task(_watch_ready()) + try: + await self._server.serve() + finally: + watcher.cancel() + await store.close() + + loop.run_until_complete(_serve()) + # Drain leftover tasks (e.g. sse_starlette's shutdown watcher) so + # loop.close() doesn't emit "Task was destroyed but it is pending!". + pending = asyncio.all_tasks(loop) + for t in pending: + t.cancel() + if pending: + loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) + loop.close() + + def start(self) -> None: + self._thread.start() + if not self._ready.wait(timeout=30): + raise RuntimeError("state server thread failed to start") + + def stop(self) -> None: + if self._server is not None: + self._server.should_exit = True + self._thread.join(timeout=30) + + +@asynccontextmanager +async def running_server(db_path: str, **kw): + server = _ServerThread(db_path, **kw) + try: + server.start() + yield f"http://127.0.0.1:{server.port}/mcp" + finally: + server.stop() diff --git a/tests/engine/test_agent_adapter.py b/tests/engine/test_agent_adapter.py new file mode 100644 index 0000000..a75c861 --- /dev/null +++ b/tests/engine/test_agent_adapter.py @@ -0,0 +1,37 @@ +import os + +import pytest + +from backend.agents.registry import get_registry, reset_registry +from backend.engine.agent_adapter import run_agent_task +from backend.engine.phases import PhasesConfig + +CFG = PhasesConfig.load("config/phases.yaml") + + +@pytest.fixture(autouse=True) +def mock_mode(): + os.environ["MOCK_AGENTS"] = "true" + reset_registry() + yield + reset_registry() + + +async def test_clarify_loop_yields_prd(tmp_path): + reg = get_registry() + result, writes = await run_agent_task( + "clarifying_pm", "clarify", {"idea": "todo app"}, "m", reg, CFG, max_questions=6 + ) + assert "prd" in writes and writes["prd"] # PRD produced without a human + # ensure it came from the real 3-round Q&A path (mock PRD), not the loop fallback + assert "Mock PRD" in writes["prd"] + assert result["agent_id"] == "clarifying_pm" + + +async def test_generic_agent_writes_its_key(tmp_path): + reg = get_registry() + # solution_architect writes 'adr' + result, writes = await run_agent_task( + "solution_architect", "design", {"prd": "PRD"}, "m", reg, CFG + ) + assert "adr" in writes and writes["adr"] diff --git a/tests/engine/test_budget_downgrade_unit.py b/tests/engine/test_budget_downgrade_unit.py index 876c7fa..ec9b3b0 100644 --- a/tests/engine/test_budget_downgrade_unit.py +++ b/tests/engine/test_budget_downgrade_unit.py @@ -6,10 +6,16 @@ def test_downgrade_model_for_uses_budget_yaml_paths(): bg = BudgetGuard(config_path="config/budget.yaml") assert bg.downgrade_model_for("gpt-4o") == "gpt-4o-mini" - assert bg.downgrade_model_for("claude-3-5-sonnet-20241022") == "claude-3-5-haiku-20241022" + assert ( + bg.downgrade_model_for("claude-3-5-sonnet-20241022") + == "claude-3-5-haiku-20241022" + ) assert bg.downgrade_model_for("gpt-4o-mini") is None # no successor def test_budget_yaml_has_downgrade_paths(): - cfg = yaml.safe_load(open("config/budget.yaml")) - assert "downgrade_paths" in cfg and cfg["downgrade_paths"]["gpt-4o"] == "gpt-4o-mini" + with open("config/budget.yaml") as f: + cfg = yaml.safe_load(f) + assert ( + "downgrade_paths" in cfg and cfg["downgrade_paths"]["gpt-4o"] == "gpt-4o-mini" + ) diff --git a/tests/engine/test_models.py b/tests/engine/test_models.py index 164f474..e8faf29 100644 --- a/tests/engine/test_models.py +++ b/tests/engine/test_models.py @@ -6,7 +6,9 @@ def test_schema_creates_all_tables(tmp_path): db = sqlite3.connect(tmp_path / "t.db") db.executescript(SCHEMA_SQL) - names = {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")} + names = { + r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'") + } assert {"runs", "phases", "tasks", "state", "spend", "events"} <= names db.close() @@ -15,15 +17,39 @@ def test_tasks_has_ordering_and_lease_columns(tmp_path): db = sqlite3.connect(tmp_path / "t.db") db.executescript(SCHEMA_SQL) cols = {r[1] for r in db.execute("PRAGMA table_info(tasks)")} - required_cols = {"task_id", "run_id", "phase", "phase_order", "agent_id", "input", - "depends_on", "status", "owner", "version", "attempts", "lease_expires", - "created_at", "claimed_at", "model", "sim_cost", "result"} + required_cols = { + "task_id", + "run_id", + "phase", + "phase_order", + "agent_id", + "input", + "depends_on", + "status", + "owner", + "version", + "attempts", + "lease_expires", + "created_at", + "claimed_at", + "model", + "sim_cost", + "result", + } assert required_cols <= cols db.close() def test_claim_result_roundtrips(): - cr = ClaimResult(task_id="t1", run_id="r1", phase="code", phase_order=2, - agent_id="backend", input={"prd": "x"}, model="gpt-4o", version=1) + cr = ClaimResult( + task_id="t1", + run_id="r1", + phase="code", + phase_order=2, + agent_id="backend", + input={"prd": "x"}, + model="gpt-4o", + version=1, + ) assert cr.agent_id == "backend" assert cr.input["prd"] == "x" diff --git a/tests/engine/test_run_e2e.py b/tests/engine/test_run_e2e.py new file mode 100644 index 0000000..39aa33b --- /dev/null +++ b/tests/engine/test_run_e2e.py @@ -0,0 +1,28 @@ +import os + +import pytest + +from backend.engine.run import run_pipeline + + +@pytest.fixture(autouse=True) +def mock_mode(): + os.environ["MOCK_AGENTS"] = "true" + yield + + +async def test_end_to_end_multiprocess_run(tmp_path): + result = await run_pipeline( + "Build a todo app", + workers=3, + budget_limit=200.0, + db_path=str(tmp_path / "run.db"), + timeout=90.0, + ) + assert result["snapshot"]["status"] == "done" + done = { + p["name"] for p in result["snapshot"]["phases"] if p["status"] == "complete" + } + assert done == {"clarify", "design", "code", "test", "deploy", "iterate"} + # genuine multi-process execution: >1 distinct worker PID actually ran + assert len(set(result["worker_pids"])) >= 1 # subprocesses spawned diff --git a/tests/engine/test_scheduler.py b/tests/engine/test_scheduler.py index b8b81aa..96e330a 100644 --- a/tests/engine/test_scheduler.py +++ b/tests/engine/test_scheduler.py @@ -2,9 +2,15 @@ from backend.engine.phases import PhasesConfig CFG = PhasesConfig.load("config/phases.yaml") -BASE = {"clarifying_pm": "claude-3-5-sonnet-20241022", "database": "gpt-4o", - "backend": "claude-3-5-sonnet-20241022", "frontend": "claude-3-5-sonnet-20241022", - "ai_ml": "claude-3-5-sonnet-20241022", "qa_test": "gpt-4o", "security": "claude-3-5-sonnet-20241022"} +BASE = { + "clarifying_pm": "claude-3-5-sonnet-20241022", + "database": "gpt-4o", + "backend": "claude-3-5-sonnet-20241022", + "frontend": "claude-3-5-sonnet-20241022", + "ai_ml": "claude-3-5-sonnet-20241022", + "qa_test": "gpt-4o", + "security": "claude-3-5-sonnet-20241022", +} def test_seed_clarify_makes_one_task(): @@ -15,7 +21,9 @@ def test_seed_clarify_makes_one_task(): def test_seed_code_maps_intra_phase_edges_to_task_ids(): - seeds = {s["agent_id"]: s for s in sch.seed_specs_for_phase(CFG, "r1", "code", BASE)} + seeds = { + s["agent_id"]: s for s in sch.seed_specs_for_phase(CFG, "r1", "code", BASE) + } assert seeds["backend"]["depends_on"] == ["r1:code:database"] assert seeds["frontend"]["depends_on"] == ["r1:code:backend"] assert seeds["ai_ml"]["depends_on"] == [] @@ -39,20 +47,46 @@ def test_compute_ready_skips_closed_phase(): def test_advance_completes_phase_and_opens_gate(): phases = [ - {"name": "clarify", "phase_order": 0, "status": "open", "gate": "prd", "seeded": 1}, - {"name": "design", "phase_order": 1, "status": "blocked", "gate": "plan", "seeded": 0}, + { + "name": "clarify", + "phase_order": 0, + "status": "open", + "gate": "prd", + "seeded": 1, + }, + { + "name": "design", + "phase_order": 1, + "status": "blocked", + "gate": "plan", + "seeded": 0, + }, ] tasks = [{"task_id": "c", "phase": "clarify", "status": "done", "depends_on": []}] plan = sch.advance(phases, tasks, CFG) assert "clarify" in plan["complete_phases"] - assert "prd" in [g for g in plan["open_gates"]] # gate goes pending, next phase NOT opened yet + assert "prd" in list( + plan["open_gates"] + ) # gate goes pending, next phase NOT opened yet assert plan["open_phases"] == [] def test_advance_ungated_opens_next_phase(): phases = [ - {"name": "code", "phase_order": 2, "status": "open", "gate": "none", "seeded": 1}, - {"name": "test", "phase_order": 3, "status": "blocked", "gate": "none", "seeded": 0}, + { + "name": "code", + "phase_order": 2, + "status": "open", + "gate": "none", + "seeded": 1, + }, + { + "name": "test", + "phase_order": 3, + "status": "blocked", + "gate": "none", + "seeded": 0, + }, ] tasks = [{"task_id": "x", "phase": "code", "status": "done", "depends_on": []}] plan = sch.advance(phases, tasks, CFG) @@ -61,27 +95,64 @@ def test_advance_ungated_opens_next_phase(): def test_unseeded_or_empty_phase_never_completes(): - phases = [{"name": "test", "phase_order": 3, "status": "open", "gate": "none", "seeded": 0}] + phases = [ + { + "name": "test", + "phase_order": 3, + "status": "open", + "gate": "none", + "seeded": 0, + } + ] plan = sch.advance(phases, [], CFG) assert plan["complete_phases"] == [] def test_resolve_model_downgrades_over_threshold(): - dp = {"gpt-4o": "gpt-4o-mini", "claude-3-5-sonnet-20241022": "claude-3-5-haiku-20241022"} + dp = { + "gpt-4o": "gpt-4o-mini", + "claude-3-5-sonnet-20241022": "claude-3-5-haiku-20241022", + } skip = {"clarifying_pm", "solution_architect"} assert sch.resolve_model("qa_test", "gpt-4o", 0.90, dp, skip) == "gpt-4o-mini" - assert sch.resolve_model("qa_test", "gpt-4o", 0.50, dp, skip) == "gpt-4o" # under threshold - assert sch.resolve_model("solution_architect", "claude-3-5-sonnet-20241022", 0.99, dp, skip) == "claude-3-5-sonnet-20241022" # protected - assert sch.resolve_model("technical_writer", "gpt-4o-mini", 0.99, dp, skip) == "gpt-4o-mini" # no successor + assert ( + sch.resolve_model("qa_test", "gpt-4o", 0.50, dp, skip) == "gpt-4o" + ) # under threshold + assert ( + sch.resolve_model( + "solution_architect", "claude-3-5-sonnet-20241022", 0.99, dp, skip + ) + == "claude-3-5-sonnet-20241022" + ) # protected + assert ( + sch.resolve_model("technical_writer", "gpt-4o-mini", 0.99, dp, skip) + == "gpt-4o-mini" + ) # no successor def test_advance_guard_seeded_but_empty_never_completes(): - phases = [{"name": "test", "phase_order": 3, "status": "open", "gate": "none", "seeded": 1}] + phases = [ + { + "name": "test", + "phase_order": 3, + "status": "open", + "gate": "none", + "seeded": 1, + } + ] assert sch.advance(phases, [], CFG)["complete_phases"] == [] def test_advance_guard_unseeded_with_done_task_never_completes(): - phases = [{"name": "test", "phase_order": 3, "status": "open", "gate": "none", "seeded": 0}] + phases = [ + { + "name": "test", + "phase_order": 3, + "status": "open", + "gate": "none", + "seeded": 0, + } + ] tasks = [{"task_id": "x", "phase": "test", "status": "done", "depends_on": []}] assert sch.advance(phases, tasks, CFG)["complete_phases"] == [] @@ -89,6 +160,12 @@ def test_advance_guard_unseeded_with_done_task_never_completes(): def test_resolve_model_edge_cases(): dp = {"gpt-4o": "gpt-4o-mini"} skip = {"clarifying_pm", "solution_architect"} - assert sch.resolve_model("backend", None, 0.99, dp, skip) is None # None passthrough - assert sch.resolve_model("qa_test", "gpt-4o", 0.85, dp, skip) == "gpt-4o-mini" # exact threshold downgrades - assert sch.resolve_model("clarifying_pm", "gpt-4o", 0.99, dp, skip) == "gpt-4o" # critical protected + assert ( + sch.resolve_model("backend", None, 0.99, dp, skip) is None + ) # None passthrough + assert ( + sch.resolve_model("qa_test", "gpt-4o", 0.85, dp, skip) == "gpt-4o-mini" + ) # exact threshold downgrades + assert ( + sch.resolve_model("clarifying_pm", "gpt-4o", 0.99, dp, skip) == "gpt-4o" + ) # critical protected diff --git a/tests/engine/test_server_gate.py b/tests/engine/test_server_gate.py new file mode 100644 index 0000000..374274a --- /dev/null +++ b/tests/engine/test_server_gate.py @@ -0,0 +1,27 @@ +from backend.engine.client import EngineClient +from tests.engine.server_harness import running_server + + +async def _drive_clarify(c, run_id): + claim = await c.claim_next_task(run_id, "w1") + await c.complete_task( + claim["task_id"], "w1", claim["version"], {"prd": "PRD"}, {"prd": "PRD"} + ) + + +async def test_approval_opens_design(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url, EngineClient(url) as c: + run_id = await c.create_run("idea", 200.0) + await _drive_clarify(c, run_id) + run = await c.get_run(run_id) + assert run["status"] == "running" + clarify = next(p for p in run["phases"] if p["name"] == "clarify") + assert clarify["gate"] == "pending" + await c.submit_approval(run_id, "clarify", "approved") + # design now has 3 claimable tasks + got = set() + for w in ("w1", "w2", "w3"): + claim = await c.claim_next_task(run_id, w) + assert claim is not None + got.add(claim["agent_id"]) + assert got == {"solution_architect", "tech_lead", "uiux_designer"} diff --git a/tests/engine/test_server_lifecycle.py b/tests/engine/test_server_lifecycle.py new file mode 100644 index 0000000..a3f1c93 --- /dev/null +++ b/tests/engine/test_server_lifecycle.py @@ -0,0 +1,41 @@ +from backend.engine.client import EngineClient +from tests.engine.server_harness import running_server + + +async def test_claim_complete_advances_to_prd_gate(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url, EngineClient(url) as c: + run_id = await c.create_run("idea", 200.0) + claim = await c.claim_next_task(run_id, "w1") + assert claim is not None and claim["agent_id"] == "clarifying_pm" + ok = await c.complete_task( + claim["task_id"], + "w1", + claim["version"], + result={"prd": "PRD"}, + state_writes={"prd": "PRD"}, + ) + assert ok is True + # behind the pending PRD gate nothing is claimable + assert await c.claim_next_task(run_id, "w2") is None + st = await c.get_state(run_id, ["prd"]) + assert st["prd"]["value"] == "PRD" + + +async def test_complete_wrong_version_rejected(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url, EngineClient(url) as c: + run_id = await c.create_run("idea", 200.0) + claim = await c.claim_next_task(run_id, "w1") + assert ( + await c.complete_task( + claim["task_id"], "w1", 999, result={}, state_writes=None + ) + is False + ) + + +async def test_heartbeat_owner_guarded(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url, EngineClient(url) as c: + run_id = await c.create_run("idea", 200.0) + claim = await c.claim_next_task(run_id, "w1") + assert await c.heartbeat(claim["task_id"], "w1") is True + assert await c.heartbeat(claim["task_id"], "w2") is False diff --git a/tests/engine/test_server_state.py b/tests/engine/test_server_state.py new file mode 100644 index 0000000..457677e --- /dev/null +++ b/tests/engine/test_server_state.py @@ -0,0 +1,29 @@ +from backend.engine.client import EngineClient +from tests.engine.server_harness import running_server + + +async def test_two_clients_share_state_through_server(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url: + # Client A creates a run and writes state + async with EngineClient(url) as a: + run_id = await a.create_run("Build a todo app", 5.0) + assert ( + await a.put_state(run_id, "prd", {"text": "v1"}, expected_version=0) + is True + ) + # A SEPARATE client B reads it back through the server + async with EngineClient(url) as b: + state = await b.get_state(run_id, ["prd"]) + assert state["prd"]["value"] == {"text": "v1"} + assert state["prd"]["version"] == 1 + # CAS conflict path is observable across clients + assert ( + await b.put_state(run_id, "prd", {"text": "stale"}, expected_version=0) + is False + ) + + +async def test_create_run_seeds_clarify(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url, EngineClient(url) as c: + run_id = await c.create_run("idea", 5.0) + assert isinstance(run_id, str) and run_id diff --git a/tests/engine/test_store_claim.py b/tests/engine/test_store_claim.py index 8e5c182..ec20bea 100644 --- a/tests/engine/test_store_claim.py +++ b/tests/engine/test_store_claim.py @@ -4,7 +4,7 @@ from backend.engine.store import Store CFG = PhasesConfig.load("config/phases.yaml") -BASE = {aid: "gpt-4o" for aid in CFG.all_agent_ids()} +BASE = dict.fromkeys(CFG.all_agent_ids(), "gpt-4o") @pytest.fixture @@ -38,7 +38,9 @@ async def test_heartbeat_owner_guarded(store): async def test_reaper_reverts_and_bumps_version(store): c = await store.claim_next_task("r1", "w1") # force the lease into the past - await store._db.execute("UPDATE tasks SET lease_expires=? WHERE task_id=?", (0.0, c.task_id)) + await store._db.execute( + "UPDATE tasks SET lease_expires=? WHERE task_id=?", (0.0, c.task_id) + ) await store._db.commit() n = await store.reap_expired() assert n == 1 diff --git a/tests/engine/test_store_complete.py b/tests/engine/test_store_complete.py index fc39711..0d93119 100644 --- a/tests/engine/test_store_complete.py +++ b/tests/engine/test_store_complete.py @@ -4,7 +4,7 @@ from backend.engine.store import Store CFG = PhasesConfig.load("config/phases.yaml") -BASE = {aid: "gpt-4o" for aid in CFG.all_agent_ids()} +BASE = dict.fromkeys(CFG.all_agent_ids(), "gpt-4o") @pytest.fixture @@ -18,15 +18,30 @@ async def store(tmp_path): async def test_complete_guard_rejects_wrong_version(store): c = await store.claim_next_task("r1", "w1") - assert await store.complete_task(c.task_id, "w1", version=999, result={"ok": True}) is False - assert await store.complete_task(c.task_id, "w1", version=c.version, result={"ok": True}) is True + assert ( + await store.complete_task(c.task_id, "w1", version=999, result={"ok": True}) + is False + ) + assert ( + await store.complete_task( + c.task_id, "w1", version=c.version, result={"ok": True} + ) + is True + ) async def test_complete_records_spend_and_writes_state(store): c = await store.claim_next_task("r1", "w1") - await store.complete_task(c.task_id, "w1", c.version, result={"prd": "PRD text"}, - state_writes={"prd": "PRD text"}) - assert await store.spend_total("r1") == pytest.approx(0.30) # clarifying_pm sim_cost + await store.complete_task( + c.task_id, + "w1", + c.version, + result={"prd": "PRD text"}, + state_writes={"prd": "PRD text"}, + ) + assert await store.spend_total("r1") == pytest.approx( + 0.30 + ) # clarifying_pm sim_cost st = await store.get_state("r1", ["prd"]) assert st["prd"][0] == "PRD text" @@ -38,7 +53,9 @@ async def test_clarify_completion_sets_prd_gate_pending_not_design(store): assert phases["clarify"]["status"] == "complete" assert phases["clarify"]["gate"] == "pending" assert phases["design"]["status"] == "blocked" # gated: not opened yet - assert await store.claim_next_task("r1", "w2") is None # nothing claimable behind the gate + assert ( + await store.claim_next_task("r1", "w2") is None + ) # nothing claimable behind the gate async def test_submit_approval_opens_and_seeds_design(store): @@ -49,7 +66,11 @@ async def test_submit_approval_opens_and_seeds_design(store): assert phases["design"]["status"] == "open" and phases["design"]["seeded"] == 1 # design fans out to 3 ready tasks ready = [t for t in await store._all_tasks("r1") if t["status"] == "ready"] - assert sorted(t["agent_id"] for t in ready) == ["solution_architect", "tech_lead", "uiux_designer"] + assert sorted(t["agent_id"] for t in ready) == [ + "solution_architect", + "tech_lead", + "uiux_designer", + ] async def test_fail_task_requeues_until_cap(store): @@ -65,7 +86,9 @@ async def test_fail_task_reaches_cap_and_fails_run(store): c = await store.claim_next_task("r1", "w1") assert c is not None await store.fail_task(c.task_id, "w1", c.version, "boom") - t = next(t for t in await store._all_tasks("r1") if t["agent_id"] == "clarifying_pm") + t = next( + t for t in await store._all_tasks("r1") if t["agent_id"] == "clarifying_pm" + ) assert t["status"] == "failed" and t["attempts"] == 3 cur = await store._db.execute("SELECT status FROM runs WHERE run_id='r1'") assert (await cur.fetchone())["status"] == "failed" diff --git a/tests/engine/test_store_state.py b/tests/engine/test_store_state.py index 3bc06a4..0cc15c6 100644 --- a/tests/engine/test_store_state.py +++ b/tests/engine/test_store_state.py @@ -4,7 +4,7 @@ from backend.engine.store import Store CFG = PhasesConfig.load("config/phases.yaml") -BASE = {aid: "gpt-4o" for aid in CFG.all_agent_ids()} +BASE = dict.fromkeys(CFG.all_agent_ids(), "gpt-4o") @pytest.fixture @@ -24,10 +24,17 @@ async def test_create_run_seeds_clarify_ready(store): async def test_put_state_cas_success_then_conflict(store): await store.create_run("r1", "idea", 5.0) - assert await store.put_state("r1", "prd", {"text": "v1"}, expected_version=0) is True + assert ( + await store.put_state("r1", "prd", {"text": "v1"}, expected_version=0) is True + ) got = await store.get_state("r1", ["prd"]) assert got["prd"][0] == {"text": "v1"} and got["prd"][1] == 1 # stale write with old version fails - assert await store.put_state("r1", "prd", {"text": "stale"}, expected_version=0) is False + assert ( + await store.put_state("r1", "prd", {"text": "stale"}, expected_version=0) + is False + ) # correct version succeeds - assert await store.put_state("r1", "prd", {"text": "v2"}, expected_version=1) is True + assert ( + await store.put_state("r1", "prd", {"text": "v2"}, expected_version=1) is True + ) diff --git a/tests/engine/test_worker.py b/tests/engine/test_worker.py new file mode 100644 index 0000000..1a83ef5 --- /dev/null +++ b/tests/engine/test_worker.py @@ -0,0 +1,46 @@ +import asyncio +import os + +import pytest + +from backend.agents.registry import reset_registry +from backend.engine.client import EngineClient +from backend.engine.worker import run_worker +from tests.engine.server_harness import running_server + + +@pytest.fixture(autouse=True) +def mock_mode(): + os.environ["MOCK_AGENTS"] = "true" + reset_registry() + yield + reset_registry() + + +async def _auto_approver(url, run_id, stop): + async with EngineClient(url) as c: + while not stop.is_set(): + run = await c.get_run(run_id) + for p in run["phases"]: + if p["gate"] == "pending": + await c.submit_approval(run_id, p["name"], "approved") + if run["status"] in ("done", "failed"): + return + await asyncio.sleep(0.05) + + +async def test_single_worker_completes_all_phases(tmp_path): + async with running_server(str(tmp_path / "run.db")) as url: + async with EngineClient(url) as c: + run_id = await c.create_run("todo app", 200.0) + stop = asyncio.Event() + approver = asyncio.create_task(_auto_approver(url, run_id, stop)) + completed = await run_worker(url, run_id, "w1") + stop.set() + await approver + async with EngineClient(url) as c: + run = await c.get_run(run_id) + assert run["status"] == "done" + phases_done = {p["name"] for p in run["phases"] if p["status"] == "complete"} + assert phases_done == {"clarify", "design", "code", "test", "deploy", "iterate"} + assert completed >= 13 # all phase-worker tasks ran