Slice 1: dispatcher loop + objective decomposition — first live objective completed

- engine/dispatcher.py: 10s tick, bounded in-flight (3), ready->kanban dispatch
  with budget mapping (goal_max_turns/max_runtime_seconds), poll->AgentMessage
  extraction->events, wall-clock cap with run termination, blocked triage state.
- engine/protocol.py: schema-enforced pool calls — raw report -> AgentMessage,
  objective brief -> task decomposition (3.6s live).
- engine/main.py: POST /companies/{slug}/objectives (manager decompose ->
  child work items, role-matched round-robin), GET work-items.
- schema: work_item.hermes_task_id/dispatched_at, employee.role.
- Hermes integration fixes discovered live:
  * profiles must use the pool custom endpoint (provider config written into
    per-profile config; 'ollama' provider = Ollama Cloud, wants API key)
  * kanban toolset was globally disabled in deployed Hermes config — removed
    from agent.disabled_toolsets (backup in scratchpad)
  * no_skills profiles (73 bundled skills drowned 9B tool discipline)
  * souls spell out kanban_show first, kanban_complete/kanban_block with
    required args last — after this, tasks complete reliably.
- Live result: GasWatch objective decomposed into 3 tasks, all completed by
  qwen3.5:9b employees (mvp_features.md, onepager.md, report.md), 4 cost
  entries totaling ~24 min wall, full event trail.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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()
 );

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)

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:
+                             description: str = "") -> None:
-        """Create employee persona: profile + soul (system prompt) + model binding."""
+        """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",
+        # no-clone profiles have a bare config with no custom endpoint defined;
-                         json={"provider": provider, "model": model})
+        # 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")

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) "
+                    "INSERT INTO employee (company_id, name, title, role, hermes_profile) "
-                    "VALUES (%s, %s, %s, %s) RETURNING id",
+                    "VALUES (%s, %s, %s, %s, %s) RETURNING id",
-                    (company_id, spec.name, spec.title, profile_name(body.slug, spec.name)),
+                    (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)
 
 

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)

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 <id>":
+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=<one line>, result=<full report: what you did,
+     files produced, evidence, anything unresolved>);
+   - kanban_block(summary=<one line>, reason=<precise blocker; if a human
+     decision is needed, 2-3 options and your recommendation>).
+A run that ends without calling kanban_complete or kanban_block counts as
+failed, no matter what work was done.
 """
 
 ROLE_CHARTERS = {