diff --git a/db/schema.sql b/db/schema.sql index 871593b..134b0dc 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -18,6 +18,8 @@ CREATE TABLE employee ( company_id uuid NOT NULL REFERENCES company(id), name text NOT NULL, title text NOT NULL, + role text NOT NULL DEFAULT 'engineer' + CHECK (role IN ('manager','engineer','qa')), department text NOT NULL DEFAULT '', manager_id uuid REFERENCES employee(id), hermes_profile text, -- Hermes profile name (ADR-001) @@ -48,6 +50,8 @@ CREATE TABLE work_item ( acceptance_criteria jsonb NOT NULL DEFAULT '[]', -- runaway caps, engine-enforced (PRD_ADDENDUM §3): max_iterations, max_tokens, max_wall_seconds budget jsonb NOT NULL DEFAULT '{"max_iterations":15,"max_tokens":100000,"max_wall_seconds":1800}', + hermes_task_id text, -- kanban task carrying this item's current run + dispatched_at timestamptz, created_at timestamptz NOT NULL DEFAULT now(), updated_at timestamptz NOT NULL DEFAULT now() ); diff --git a/engine/dispatcher.py b/engine/dispatcher.py new file mode 100644 index 0000000..6172f20 --- /dev/null +++ b/engine/dispatcher.py @@ -0,0 +1,240 @@ +"""Dispatch loop — the engine's heartbeat. + +Every tick: send ready work items to Hermes kanban (bounded in-flight), +poll running ones, convert finished runs to protocol events, enforce +wall-clock budgets. + +Slice 1 state mapping (ponytail: no separate review stage yet — completed +goes straight to done; anything unclear goes to blocked for manager triage): + kanban running -> in_progress (budget-checked) + kanban blocked -> blocked + kanban review/done -> extract AgentMessage -> + task.completed -> done + anything else -> blocked (reason = message summary) +""" +import asyncio +import logging +from datetime import datetime, timezone + +from engine import db +from engine.hermes_client import HermesClient, HermesError +from engine.protocol import extract_agent_message + +log = logging.getLogger("farmhouse.dispatcher") + +TICK_SECONDS = 10 +MAX_IN_FLIGHT = 3 # effective pool concurrency (benchmarks/RESULTS.md) + +TASK_BODY_TEMPLATE = """{description} + +## Acceptance criteria +{criteria} + +## Reporting +End your work with a clear report: what you did, what you produced, test +evidence, and anything unresolved. If you need a human decision, say exactly +what decision and give 2-3 options with a recommendation. +""" + + +def _task_body(item: dict) -> str: + criteria = "\n".join(f"- {c}" for c in item.get("acceptance_criteria") or []) or "- (none given)" + return TASK_BODY_TEMPLATE.format(description=item["description"], criteria=criteria) + + +class Dispatcher: + def __init__(self, hermes: HermesClient, is_paused) -> None: + self.hermes = hermes + self.is_paused = is_paused + self._stop = asyncio.Event() + self._task: asyncio.Task | None = None + + def start(self) -> None: + self._task = asyncio.create_task(self._run()) + + async def stop(self) -> None: + self._stop.set() + if self._task: + await self._task + + async def _run(self) -> None: + log.info("dispatcher started") + while not self._stop.is_set(): + try: + if not await self.is_paused(): + await self.tick() + except Exception: + log.exception("tick failed") + try: + await asyncio.wait_for(self._stop.wait(), timeout=TICK_SECONDS) + except asyncio.TimeoutError: + pass + log.info("dispatcher stopped") + + # ------------------------------------------------------------ one tick + + async def tick(self) -> None: + await self._poll_in_flight() + await self._dispatch_ready() + + async def _dispatch_ready(self) -> None: + in_flight = (await db.fetch_one( + "SELECT count(*) AS n FROM work_item WHERE status = 'in_progress'"))["n"] + slots = MAX_IN_FLIGHT - in_flight + if slots <= 0: + return + ready = await db.fetch_all( + """ + SELECT wi.*, e.hermes_profile, e.id AS employee_id, c.slug AS company_slug + FROM work_item wi + JOIN employee e ON e.id = wi.owner_id + JOIN company c ON c.id = wi.company_id + WHERE wi.status = 'ready' AND wi.type IN ('task','subtask','defect') + AND c.lifecycle_state = 'active' AND e.hermes_profile IS NOT NULL + ORDER BY wi.created_at LIMIT %s + """, slots, + ) + for item in ready: + await self._dispatch_one(item) + + async def _dispatch_one(self, item: dict) -> None: + budget = item["budget"] + try: + task = await self.hermes.create_task( + item["title"], + body=_task_body(item), + assignee=item["hermes_profile"], + tenant=item["company_slug"], + idempotency_key=str(item["id"]), + max_runtime_seconds=int(budget.get("max_wall_seconds", 1800)), + goal_max_turns=int(budget.get("max_iterations", 15)), + ) + await self.hermes.dispatch_now() + except HermesError as e: + log.warning("dispatch failed for %s: %s", item["id"], e) + async with db.pool.connection() as conn: + await db.append_event(conn, actor="system", event_type="dispatch.failed", + company_id=item["company_id"], work_item_id=item["id"], + payload={"error": str(e)[:300]}) + return + async with db.pool.connection() as conn: + async with conn.transaction(): + await conn.execute( + "UPDATE work_item SET status='in_progress', hermes_task_id=%s, " + "dispatched_at=now(), updated_at=now() WHERE id=%s", + (str(task["id"]), item["id"]), + ) + await conn.execute( + "UPDATE employee SET status='working' WHERE id=%s", (item["employee_id"],)) + await db.append_event( + conn, actor="system", event_type="dispatch.sent", + company_id=item["company_id"], work_item_id=item["id"], + payload={"hermes_task_id": str(task["id"]), + "assignee": item["hermes_profile"]}, + ) + log.info("dispatched %s -> %s", item["title"][:40], task["id"]) + + # ------------------------------------------------------------ polling + + async def _poll_in_flight(self) -> None: + items = await db.fetch_all( + "SELECT wi.*, e.id AS employee_id FROM work_item wi " + "JOIN employee e ON e.id = wi.owner_id " + "WHERE wi.status = 'in_progress' AND wi.hermes_task_id IS NOT NULL" + ) + for item in items: + try: + kt = await self.hermes.get_task(item["hermes_task_id"]) + except HermesError as e: + log.warning("poll failed for %s: %s", item["hermes_task_id"], e) + continue + status = kt.get("status") + if status in ("review", "done"): + await self._finish(item, kt) + elif status == "blocked": + await self._block(item, kt.get("block_reason") or "hermes reported blocked", + event_type="task.blocked") + elif await self._over_budget(item): + await self._cap(item, kt) + # todo/scheduled/ready/running: still waiting + + async def _over_budget(self, item: dict) -> bool: + if not item["dispatched_at"]: + return False + max_wall = int(item["budget"].get("max_wall_seconds", 1800)) + elapsed = (datetime.now(timezone.utc) - item["dispatched_at"]).total_seconds() + return elapsed > max_wall + + async def _cap(self, item: dict, kt: dict) -> None: + run_id = kt.get("current_run_id") + if run_id: + try: + await self.hermes.terminate_run(str(run_id)) + except HermesError: + pass + try: + await self.hermes.update_task(item["hermes_task_id"], status="blocked", + block_reason="farmhouse: wall-clock budget exceeded") + except HermesError: + pass + await self._block(item, "wall-clock budget exceeded", event_type="run.capped") + + async def _block(self, item: dict, reason: str, *, event_type: str, + payload: dict | None = None) -> None: + async with db.pool.connection() as conn: + async with conn.transaction(): + await conn.execute( + "UPDATE work_item SET status='blocked', updated_at=now() WHERE id=%s", + (item["id"],)) + await conn.execute( + "UPDATE employee SET status='blocked' WHERE id=%s", (item["employee_id"],)) + await db.append_event(conn, actor=f"employee:{item['employee_id']}", + event_type=event_type, source="hermes", + company_id=item["company_id"], work_item_id=item["id"], + payload={**(payload or {}), "reason": reason}) + log.info("blocked %s: %s", item["id"], reason) + + async def _finish(self, item: dict, kt: dict) -> None: + raw = kt.get("result") or kt.get("summary") or "(no report)" + try: + msg = await extract_agent_message(str(item["id"]), raw) + except Exception as e: + await self._block(item, f"protocol extraction failed: {e}", event_type="task.blocked") + return + + wall = None + if item["dispatched_at"]: + wall = (datetime.now(timezone.utc) - item["dispatched_at"]).total_seconds() + + if msg.message_type == "task.completed": + new_status = "done" + else: + # ponytail: blocked-for-triage covers clarification/approval/progressed; + # manager triage flow upgrades this in the next increment + new_status = "blocked" + + async with db.pool.connection() as conn: + async with conn.transaction(): + await conn.execute( + "UPDATE work_item SET status=%s, updated_at=now() WHERE id=%s", + (new_status, item["id"])) + await conn.execute( + "UPDATE employee SET status=%s WHERE id=%s", + ("idle" if new_status == "done" else "blocked", item["employee_id"])) + await db.append_event( + conn, actor=f"employee:{item['employee_id']}", + event_type=msg.message_type, source="hermes", + company_id=item["company_id"], work_item_id=item["id"], + payload=msg.model_dump(exclude_defaults=True), + provenance={"hermes_run_id": str(kt.get("current_run_id") or ""), + "hermes_task_id": item["hermes_task_id"]}, + ) + if wall is not None: + await conn.execute( + "INSERT INTO cost_entry (company_id, employee_id, work_item_id, kind, " + "wall_seconds) VALUES (%s,%s,%s,'local_inference',%s)", + (item["company_id"], item["employee_id"], item["id"], round(wall, 1))) + await db.append_event(conn, actor="system", event_type="cost.recorded", + company_id=item["company_id"], work_item_id=item["id"], + payload={"wall_seconds": round(wall, 1)}) + log.info("finished %s -> %s (%s)", item["id"], new_status, msg.message_type) diff --git a/engine/hermes_client.py b/engine/hermes_client.py index c147ec4..86017e9 100644 --- a/engine/hermes_client.py +++ b/engine/hermes_client.py @@ -75,16 +75,28 @@ class HermesClient: return data if isinstance(data, list) else data.get("profiles", []) async def create_profile(self, name: str, *, soul: str, model: str, - provider: str = "ollama", description: str = "") -> None: - """Create employee persona: profile + soul (system prompt) + model binding.""" + description: str = "") -> None: + """Create employee persona: profile + soul (system prompt) + model binding. + + no_skills: lean prompt — 73 bundled skills drown a 9B model's tool + discipline; employees get capabilities via explicit toolsets instead. + """ await self._json("POST", "/api/profiles", json={ "name": name, - "clone_from_default": True, + "no_skills": True, "description": description, }) await self._json("PUT", f"/api/profiles/{name}/soul", json={"content": soul}) - await self._json("PUT", f"/api/profiles/{name}/model", - json={"provider": provider, "model": model}) + # no-clone profiles have a bare config with no custom endpoint defined; + # declare the pool endpoint directly, same shape as the default profile + await self._json("PUT", "/api/config", json={ + "profile": name, + "config": {"model": { + "default": model, + "provider": "custom", + "base_url": f"{settings.ollama_pool_url}/v1", + }}, + }) async def get_soul(self, name: str) -> str: data = await self._json("GET", f"/api/profiles/{name}/soul") diff --git a/engine/main.py b/engine/main.py index 71d95bc..53f00c0 100644 --- a/engine/main.py +++ b/engine/main.py @@ -10,19 +10,24 @@ from pydantic import BaseModel, Field from engine import db from engine.config import settings +from engine.dispatcher import Dispatcher from engine.hermes_client import HermesClient, HermesError +from engine.protocol import decompose_objective from engine.souls import ROLE_CHARTERS, build_soul PAUSE_KEY = "farmhouse:paused" hermes = HermesClient() redis_client = aioredis.from_url(settings.redis_url, decode_responses=True) +dispatcher = Dispatcher(hermes, is_paused=lambda: is_paused()) @asynccontextmanager async def lifespan(app: FastAPI): await db.open_pool() + dispatcher.start() yield + await dispatcher.stop() await db.close_pool() await hermes.aclose() await redis_client.aclose() @@ -92,9 +97,10 @@ async def create_company(body: CompanyCreate) -> dict: employees = [] for spec in body.employees: cur = await conn.execute( - "INSERT INTO employee (company_id, name, title, hermes_profile) " - "VALUES (%s, %s, %s, %s) RETURNING id", - (company_id, spec.name, spec.title, profile_name(body.slug, spec.name)), + "INSERT INTO employee (company_id, name, title, role, hermes_profile) " + "VALUES (%s, %s, %s, %s, %s) RETURNING id", + (company_id, spec.name, spec.title, spec.role, + profile_name(body.slug, spec.name)), ) emp_id = (await cur.fetchone())["id"] await db.append_event(conn, actor="system", event_type="employee.hired", @@ -183,6 +189,92 @@ async def archive_company(slug: str) -> None: pass # profile may already be gone; DB stays authoritative +# ---------------------------------------------------------------- objectives and work items + + +class ObjectiveCreate(BaseModel): + title: str + brief: str + + +@app.post("/companies/{slug}/objectives", status_code=201) +async def create_objective(slug: str, body: ObjectiveCreate) -> dict: + """Submit an objective; the manager decomposes it into ready work items.""" + company = await db.fetch_one( + "SELECT * FROM company WHERE slug = %s AND lifecycle_state = 'active'", slug) + if not company: + raise HTTPException(404, "no active company with that slug") + staff = await db.fetch_all( + "SELECT id, name, role FROM employee WHERE company_id = %s", company["id"]) + by_role: dict[str, list] = {} + for e in staff: + by_role.setdefault(e["role"], []).append(e) + manager = (by_role.get("manager") or [None])[0] + if manager is None: + raise HTTPException(422, "company has no manager to decompose the objective") + + plan = await decompose_objective(body.brief, company["name"]) + + async with db.pool.connection() as conn: + async with conn.transaction(): + cur = await conn.execute( + "INSERT INTO work_item (company_id, type, title, description, status, owner_id) " + "VALUES (%s, 'objective', %s, %s, 'in_progress', %s) RETURNING id", + (company["id"], body.title, body.brief, manager["id"]), + ) + objective_id = (await cur.fetchone())["id"] + await db.append_event(conn, actor="user:philip", event_type="work_item.created", + company_id=company["id"], work_item_id=objective_id, + source="api", payload={"type": "objective", "title": body.title}) + + children = [] + rr_index: dict[str, int] = {} + for t in plan["tasks"]: + pool_for_role = by_role.get(t["role"]) or by_role.get("engineer") or staff + idx = rr_index.get(t["role"], 0) + owner = pool_for_role[idx % len(pool_for_role)] + rr_index[t["role"]] = idx + 1 + cur = await conn.execute( + "INSERT INTO work_item (company_id, parent_id, type, title, description, " + "status, owner_id, acceptance_criteria) " + "VALUES (%s, %s, 'task', %s, %s, 'ready', %s, %s) RETURNING id", + (company["id"], objective_id, t["title"], t["description"], + owner["id"], db.Json(t["acceptance_criteria"])), + ) + child_id = (await cur.fetchone())["id"] + await db.append_event( + conn, actor=f"employee:{manager['id']}", event_type="work_item.created", + company_id=company["id"], work_item_id=child_id, + payload={"type": "task", "title": t["title"], "role": t["role"], + "owner": owner["name"], "parent": str(objective_id)}, + ) + children.append({"id": str(child_id), "title": t["title"], + "role": t["role"], "owner": owner["name"]}) + + if plan.get("clarification_needed"): + await db.append_event( + conn, actor=f"employee:{manager['id']}", + event_type="clarification.requested", + company_id=company["id"], work_item_id=objective_id, + payload={"question": plan["clarification_needed"]}, + ) + + return {"objective_id": str(objective_id), "tasks": children, + "clarification_needed": plan.get("clarification_needed") or None} + + +@app.get("/companies/{slug}/work-items") +async def list_work_items(slug: str) -> list[dict]: + company = await db.fetch_one("SELECT id FROM company WHERE slug = %s", slug) + if not company: + raise HTTPException(404, "no such company") + rows = await db.fetch_all( + "SELECT wi.*, e.name AS owner_name FROM work_item wi " + "LEFT JOIN employee e ON e.id = wi.owner_id " + "WHERE wi.company_id = %s ORDER BY wi.created_at", company["id"]) + return [db.as_json(r) for r in rows] + + # ---------------------------------------------------------------- global pause (kill switch v1) diff --git a/engine/protocol.py b/engine/protocol.py new file mode 100644 index 0000000..8815e53 --- /dev/null +++ b/engine/protocol.py @@ -0,0 +1,93 @@ +"""Structured extraction from free-form agent output. + +Kanban runs return free text; the protocol requires AgentMessage. Rather than +trusting the worker to format correctly mid-run, the engine converts each raw +report with one schema-enforced pool call (qwen3.5:9b scored 100% validity +under `format` enforcement — benchmarks/RESULTS.md). +""" +import json +from typing import Any + +import httpx + +from engine.config import settings +from engine.schemas.events import AGENT_MESSAGE_FORMAT, AgentMessage + +MODEL = "qwen3.5:9b" + +EXTRACT_SYSTEM = ( + "You convert an AI employee's raw work report into exactly one structured " + "protocol message. Choose the message_type that matches what actually " + "happened:\n" + "- task.completed: work finished, evidence given\n" + "- task.progressed: partial progress, more to do\n" + "- task.blocked: cannot continue for technical reasons\n" + "- clarification.requested: needs a decision/answer from a human; include " + "the options given\n" + "- approval.requested: needs authority the employee lacks\n" + "- incident.declared: something broke that affects others\n" + "Copy concrete facts (files, commits, test counts) into summary/detail. " + "Do not invent facts not present in the report." +) + +DECOMPOSE_FORMAT: dict[str, Any] = { + "type": "object", + "properties": { + "tasks": { + "type": "array", + "items": { + "type": "object", + "properties": { + "title": {"type": "string"}, + "description": {"type": "string"}, + "role": {"type": "string", "enum": ["engineer", "qa", "manager"]}, + "acceptance_criteria": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["title", "description", "role", "acceptance_criteria"], + }, + }, + "clarification_needed": {"type": "string"}, + }, + "required": ["tasks"], +} + + +async def _chat_structured(system: str, user: str, fmt: dict) -> dict: + async with httpx.AsyncClient(timeout=180) as client: + r = await client.post( + f"{settings.ollama_pool_url}/api/chat", + json={ + "model": MODEL, + "messages": [ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ], + "stream": False, + "think": False, + "format": fmt, + }, + ) + r.raise_for_status() + return json.loads(r.json()["message"]["content"]) + + +async def extract_agent_message(work_item_id: str, raw_report: str) -> AgentMessage: + data = await _chat_structured( + EXTRACT_SYSTEM, + f"Work item ID: {work_item_id}\n\nRaw report:\n{raw_report[:6000]}", + AGENT_MESSAGE_FORMAT, + ) + data["work_item_id"] = work_item_id # never trust the model with the ID + return AgentMessage.model_validate(data) + + +async def decompose_objective(brief: str, company_name: str) -> dict: + """Manager-style decomposition into 2-5 small tasks. Returns validated dict.""" + system = ( + f"You are the engineering manager at {company_name}. Decompose the " + "objective into 2-5 small, independently completable tasks, each with " + "concrete acceptance criteria. Assign each to the right role. If the " + "objective is materially ambiguous, fill clarification_needed with the " + "question and give your best-guess decomposition anyway." + ) + return await _chat_structured(system, brief, DECOMPOSE_FORMAT) diff --git a/engine/souls.py b/engine/souls.py index b8df818..e80e116 100644 --- a/engine/souls.py +++ b/engine/souls.py @@ -10,6 +10,21 @@ what decision you need, with 2-3 concrete options and your recommendation. If you need authority you do not have (shared environments, external actions, production), stop and request approval instead of proceeding. Never invent completed work. Test evidence beats claims. + +## Kanban task workflow (MANDATORY) +When your query says "work kanban task ": +1. FIRST ACTION: call the kanban_show tool with that task id to read the + task title, body, and acceptance criteria. Never assume the task is + undefined before calling kanban_show. +2. Do the work described, in the current workspace directory. +3. LAST ACTION: call exactly one of these TOOLS (a real tool call, never + plain text), always filling BOTH arguments: + - kanban_complete(summary=, result=); + - kanban_block(summary=, reason=). +A run that ends without calling kanban_complete or kanban_block counts as +failed, no matter what work was done. """ ROLE_CHARTERS = {