Slice 1: Hermes adapter with live-verified round-trips

- engine/hermes_client.py: cookie-auth async client — profiles (create/soul/
  model/delete), kanban tasks (create with tenant + runaway budget mapping,
  idempotency via work-item UUID, update/delete), dispatch, runs, workers.
- engine/config.py: env-driven settings (.env stays gitignored).
- tests/test_hermes_live.py: live smoke against Hermes 0.19.0 — all pass:
  profile round-trip, task create (t_*), idempotent re-create, status patch,
  cleanup. No dispatch triggered (no GPU burn).
- Discovery: kanban responses wrap in {task}; webhooks are inbound triggers,
  so engine polls run/task status (ADR-001 fallback is the primary channel).

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

engine/config.py

@@ -0,0 +1,20 @@
+"""Engine settings, loaded from environment / .env."""
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class Settings:
+    hermes_dashboard_url: str = os.getenv("HERMES_DASHBOARD_URL", "http://192.168.1.5:30433")
+    hermes_api_url: str = os.getenv("HERMES_API_URL", "http://192.168.1.5:30432")
+    hermes_api_key: str = os.getenv("HERMES_API_KEY", "")
+    hermes_username: str = os.getenv("HERMES_USERNAME", "")
+    hermes_password: str = os.getenv("HERMES_PASSWORD", "")
+    ollama_pool_url: str = os.getenv("OLLAMA_POOL_URL", "http://192.168.1.5:4000")
+    database_url: str = os.getenv("DATABASE_URL", "postgresql://farmhouse:farmhouse@localhost:5433/farmhouse")
+    redis_url: str = os.getenv("REDIS_URL", "redis://localhost:6380/0")
+
+
+settings = Settings()

engine/hermes_client.py

@@ -0,0 +1,152 @@
+"""Adapter for Hermes Agent 0.19.0 (ADR-001).
+
+The ONLY module allowed to talk to Hermes. Engine business logic imports this,
+never httpx-to-Hermes directly, so a Hermes upgrade is contained here.
+
+Two Hermes surfaces:
+  * Dashboard API (:30433) — cookie session auth (login endpoint), profiles,
+    kanban orchestration, runs. This is the integration surface.
+  * api_server (:30432) — OpenAI-compatible bearer-auth chat; one completion =
+    one full agent run. Used later for one-shot manager reasoning if needed.
+
+Event flow back to the engine is POLLING (kanban task status + runs): Hermes
+webhooks are inbound triggers, not outbound delivery.
+"""
+from typing import Any
+
+import httpx
+
+from engine.config import settings
+
+
+class HermesError(RuntimeError):
+    pass
+
+
+class HermesClient:
+    """Async client with lazy cookie login and one retry on session expiry."""
+
+    def __init__(self) -> None:
+        self._http = httpx.AsyncClient(base_url=settings.hermes_dashboard_url, timeout=30)
+        self._logged_in = False
+
+    async def aclose(self) -> None:
+        await self._http.aclose()
+
+    # ------------------------------------------------------------ auth
+
+    async def _login(self) -> None:
+        r = await self._http.post(
+            "/auth/password-login",
+            json={
+                "provider": "basic",
+                "username": settings.hermes_username,
+                "password": settings.hermes_password,
+            },
+        )
+        if r.status_code != 200 or not r.json().get("ok"):
+            raise HermesError(f"Hermes login failed: {r.status_code} {r.text[:200]}")
+        self._logged_in = True
+
+    async def _request(self, method: str, path: str, **kwargs) -> httpx.Response:
+        if not self._logged_in:
+            await self._login()
+        r = await self._http.request(method, path, **kwargs)
+        if r.status_code == 401:  # session expired mid-flight
+            await self._login()
+            r = await self._http.request(method, path, **kwargs)
+        if r.status_code >= 400:
+            raise HermesError(f"{method} {path} -> {r.status_code}: {r.text[:300]}")
+        return r
+
+    async def _json(self, method: str, path: str, **kwargs) -> Any:
+        r = await self._request(method, path, **kwargs)
+        return r.json() if r.content else None
+
+    # ------------------------------------------------------------ health
+
+    async def status(self) -> dict:
+        return await self._json("GET", "/api/status")
+
+    # ------------------------------------------------------------ profiles (= employees)
+
+    async def list_profiles(self) -> list[dict]:
+        data = await self._json("GET", "/api/profiles")
+        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."""
+        await self._json("POST", "/api/profiles", json={
+            "name": name,
+            "clone_from_default": 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})
+
+    async def get_soul(self, name: str) -> str:
+        data = await self._json("GET", f"/api/profiles/{name}/soul")
+        return data.get("content", "") if isinstance(data, dict) else str(data)
+
+    async def delete_profile(self, name: str) -> None:
+        await self._json("DELETE", f"/api/profiles/{name}")
+
+    # ------------------------------------------------------------ kanban (= work dispatch)
+
+    async def create_task(self, title: str, *, body: str, assignee: str, tenant: str,
+                          idempotency_key: str, max_runtime_seconds: int = 1800,
+                          goal_max_turns: int = 15, priority: int = 3,
+                          workspace_path: str | None = None) -> dict:
+        """One engine work-item dispatch = one kanban task.
+
+        tenant = FarmHouse company slug (board isolation).
+        goal_max_turns / max_runtime_seconds carry the work item's runaway budget.
+        idempotency_key = engine work_item UUID — safe re-dispatch after crashes.
+        """
+        payload = {
+            "title": title,
+            "body": body,
+            "assignee": assignee,
+            "tenant": tenant,
+            "priority": priority,
+            "idempotency_key": idempotency_key,
+            "max_runtime_seconds": max_runtime_seconds,
+            "goal_max_turns": goal_max_turns,
+            "triage": False,
+        }
+        if workspace_path:
+            payload["workspace_kind"] = "path"
+            payload["workspace_path"] = workspace_path
+        data = await self._json("POST", "/api/plugins/kanban/tasks", json=payload)
+        return data.get("task", data)
+
+    async def get_task(self, task_id: str) -> dict:
+        data = await self._json("GET", f"/api/plugins/kanban/tasks/{task_id}")
+        return data.get("task", data)
+
+    async def update_task(self, task_id: str, **fields) -> dict:
+        data = await self._json("PATCH", f"/api/plugins/kanban/tasks/{task_id}", json=fields)
+        return data.get("task", data)
+
+    async def delete_task(self, task_id: str) -> None:
+        await self._json("DELETE", f"/api/plugins/kanban/tasks/{task_id}")
+
+    async def board(self, tenant: str | None = None) -> dict:
+        params = {"tenant": tenant} if tenant else {}
+        return await self._json("GET", "/api/plugins/kanban/board", params=params)
+
+    async def dispatch_now(self) -> dict | None:
+        """Force a dispatch tick instead of waiting for Hermes' 60s interval."""
+        return await self._json("POST", "/api/plugins/kanban/dispatch")
+
+    async def get_run(self, run_id: str) -> dict:
+        return await self._json("GET", f"/api/plugins/kanban/runs/{run_id}")
+
+    async def active_workers(self) -> list[dict]:
+        data = await self._json("GET", "/api/plugins/kanban/workers/active")
+        return data if isinstance(data, list) else data.get("workers", [])
+
+    async def terminate_run(self, run_id: str) -> None:
+        await self._json("POST", f"/api/plugins/kanban/runs/{run_id}/terminate")

pyrightconfig.json

@@ -0,0 +1,5 @@
+{
+  "venvPath": "/home/philip/.venvs",
+  "venv": "farmhouse",
+  "extraPaths": ["."]
+}

requirements.txt

@@ -0,0 +1,9 @@
+fastapi>=0.115
+uvicorn[standard]>=0.30
+httpx>=0.27
+pydantic>=2.7
+python-dotenv>=1.0
+psycopg[binary]>=3.2
+redis>=5.0
+pytest>=8.0
+pytest-asyncio>=0.24

tests/test_hermes_live.py

@@ -0,0 +1,93 @@
+"""Live integration smoke for HermesClient against the real Hermes instance.
+
+Creates a disposable profile + kanban task, verifies round-trips, cleans up.
+Does NOT trigger a dispatch — no agent run, no GPU burn.
+
+Run: .venv python -m pytest tests/test_hermes_live.py -v   (or python tests/test_hermes_live.py)
+"""
+import asyncio
+import sys
+import uuid
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from engine.hermes_client import HermesClient, HermesError  # noqa: E402
+
+TEST_PROFILE = "fh-smoke-test"
+TEST_TENANT = "fh-smoke"
+SOUL = "You are a FarmHouse smoke-test employee. You only ever say 'ok'."
+
+
+async def cleanup(h: HermesClient) -> None:
+    try:
+        await h.delete_profile(TEST_PROFILE)
+    except HermesError:
+        pass
+    board = await h.board(tenant=TEST_TENANT)
+    for col in board.get("columns", []):
+        for task in col.get("tasks", []):
+            try:
+                await h.delete_task(str(task["id"]))
+            except HermesError:
+                pass
+
+
+async def main() -> None:
+    h = HermesClient()
+    try:
+        # health
+        st = await h.status()
+        assert st["gateway_running"] is True, st
+        print(f"status ok: hermes {st['version']}, gateway {st['gateway_state']}")
+
+        await cleanup(h)  # in case a previous run died
+
+        # profile round-trip
+        await h.create_profile(TEST_PROFILE, soul=SOUL, model="qwen3.5:9b",
+                               description="disposable smoke-test profile")
+        soul = await h.get_soul(TEST_PROFILE)
+        assert SOUL in soul, f"soul mismatch: {soul[:100]!r}"
+        names = [p.get("name") for p in await h.list_profiles()]
+        assert TEST_PROFILE in names, names
+        print("profile round-trip ok")
+
+        # kanban round-trip (no dispatch)
+        wid = str(uuid.uuid4())
+        task = await h.create_task(
+            "Smoke: do nothing", body="Smoke-test task. Do not run.",
+            assignee=TEST_PROFILE, tenant=TEST_TENANT,
+            idempotency_key=wid, max_runtime_seconds=60, goal_max_turns=2,
+        )
+        task_id = str(task["id"])
+        got = await h.get_task(task_id)
+        assert got.get("idempotency_key") in (wid, None) or got.get("title") == "Smoke: do nothing", got
+        print(f"kanban task created: id={task_id}, status={got.get('status')}")
+
+        # idempotency: same key second time must not duplicate
+        dup = await h.create_task(
+            "Smoke: do nothing", body="dup", assignee=TEST_PROFILE, tenant=TEST_TENANT,
+            idempotency_key=wid, max_runtime_seconds=60, goal_max_turns=2,
+        )
+        assert str(dup["id"]) == task_id, f"idempotency broken: {dup['id']} != {task_id}"
+        print("idempotency ok")
+
+        # status transition via engine-side patch
+        await h.update_task(task_id, status="blocked", block_reason="smoke")
+        assert (await h.get_task(task_id)).get("status") == "blocked"
+        print("task update ok")
+
+        await cleanup(h)
+        # profile really gone
+        assert TEST_PROFILE not in [p.get("name") for p in await h.list_profiles()]
+        print("cleanup ok — ALL SMOKE CHECKS PASS")
+    finally:
+        await h.aclose()
+
+
+def test_hermes_live() -> None:
+    asyncio.run(main())
+
+
+if __name__ == "__main__":
+    asyncio.run(main())