Slice 1: engine core — compose stack, db layer, company/employee API

- docker-compose.yml: Postgres 16 (schema auto-applied) + Redis, nonstandard
  host ports 5433/6380.
- engine/db.py: async psycopg pool, append_event in caller's transaction.
- engine/souls.py: versioned role charters (manager/engineer/qa) + reporting
  protocol appended to every soul.
- engine/main.py: POST/GET companies, employee hire with Hermes profile
  creation (outside tx, retryable; DB authoritative), event feed, archive
  with profile cleanup, global pause/resume in Redis.
- Verified live end-to-end: create -> 4 events + 3 real Hermes profiles;
  pause flips health; archive 204 removes profiles, keeps rows.

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

docker-compose.yml

@@ -0,0 +1,28 @@
+services:
+  postgres:
+    image: postgres:16-alpine
+    environment:
+      POSTGRES_USER: farmhouse
+      POSTGRES_PASSWORD: farmhouse
+      POSTGRES_DB: farmhouse
+    ports:
+      - "5433:5432"
+    volumes:
+      - pgdata:/var/lib/postgresql/data
+      - ./db/schema.sql:/docker-entrypoint-initdb.d/01-schema.sql:ro
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U farmhouse"]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+
+  redis:
+    image: redis:7-alpine
+    ports:
+      - "6380:6379"
+    volumes:
+      - redisdata:/data
+
+volumes:
+  pgdata:
+  redisdata:

engine/db.py

@@ -0,0 +1,74 @@
+"""Postgres access — one async pool, event append, thin query helpers."""
+import json
+from typing import Any
+from uuid import UUID
+
+from psycopg.rows import dict_row
+from psycopg.types.json import Json
+from psycopg_pool import AsyncConnectionPool
+
+from engine.config import settings
+
+pool = AsyncConnectionPool(settings.database_url, open=False, kwargs={"row_factory": dict_row})
+
+
+async def open_pool() -> None:
+    await pool.open()
+
+
+async def close_pool() -> None:
+    await pool.close()
+
+
+async def fetch_all(query: str, *args: Any) -> list[dict]:
+    async with pool.connection() as conn:
+        cur = await conn.execute(query, args)
+        return await cur.fetchall()
+
+
+async def fetch_one(query: str, *args: Any) -> dict | None:
+    async with pool.connection() as conn:
+        cur = await conn.execute(query, args)
+        return await cur.fetchone()
+
+
+async def execute(query: str, *args: Any) -> None:
+    async with pool.connection() as conn:
+        await conn.execute(query, args)
+
+
+async def append_event(
+    conn: Any,
+    *,
+    actor: str,
+    event_type: str,
+    company_id: UUID | None = None,
+    work_item_id: UUID | None = None,
+    correlation_id: UUID | None = None,
+    source: str = "engine",
+    payload: dict | None = None,
+    policy_decision: dict | None = None,
+    provenance: dict | None = None,
+) -> dict:
+    """Append to the immutable log. Caller passes its open connection so the
+    event lands in the same transaction as the projection update."""
+    cur = await conn.execute(
+        """
+        INSERT INTO event (company_id, actor, event_type, work_item_id,
+                           correlation_id, source, payload, policy_decision, provenance)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
+        RETURNING event_id, created_at
+        """,
+        (
+            company_id, actor, event_type, work_item_id, correlation_id, source,
+            Json(payload or {}),
+            Json(policy_decision) if policy_decision else None,
+            Json(provenance) if provenance else None,
+        ),
+    )
+    return await cur.fetchone()
+
+
+def as_json(row: dict) -> dict:
+    """Make a DB row JSON-serializable (UUID/datetime to str)."""
+    return json.loads(json.dumps(row, default=str))

engine/main.py

@@ -0,0 +1,204 @@
+"""FarmHouse engine API — Slice 1 core.
+
+Run: uvicorn engine.main:app --reload --port 8100
+"""
+from contextlib import asynccontextmanager
+
+import redis.asyncio as aioredis
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+
+from engine import db
+from engine.config import settings
+from engine.hermes_client import HermesClient, HermesError
+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)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    await db.open_pool()
+    yield
+    await db.close_pool()
+    await hermes.aclose()
+    await redis_client.aclose()
+
+
+app = FastAPI(title="FarmHouse Holdings Engine", version="0.1.0", lifespan=lifespan)
+
+# ---------------------------------------------------------------- request models
+
+
+class EmployeeSpec(BaseModel):
+    name: str
+    title: str
+    role: str  # manager | engineer | qa
+
+    def validate_role(self) -> None:
+        if self.role not in ROLE_CHARTERS:
+            raise HTTPException(422, f"unknown role {self.role!r}; valid: {sorted(ROLE_CHARTERS)}")
+
+
+class CompanyCreate(BaseModel):
+    slug: str = Field(pattern=r"^[a-z0-9][a-z0-9-]{1,40}$")
+    name: str
+    brief: str
+    employees: list[EmployeeSpec] = Field(min_length=1)
+
+
+# ---------------------------------------------------------------- helpers
+
+
+def profile_name(company_slug: str, employee_name: str) -> str:
+    return f"fh-{company_slug}-{employee_name.lower().replace(' ', '-')}"
+
+
+async def is_paused() -> bool:
+    return await redis_client.get(PAUSE_KEY) == "1"
+
+
+# ---------------------------------------------------------------- endpoints
+
+
+@app.get("/health")
+async def health() -> dict:
+    row = await db.fetch_one("SELECT count(*) AS companies FROM company")
+    return {"status": "ok", "companies": row["companies"], "paused": await is_paused()}
+
+
+@app.post("/companies", status_code=201)
+async def create_company(body: CompanyCreate) -> dict:
+    for spec in body.employees:
+        spec.validate_role()
+    if await db.fetch_one("SELECT id FROM company WHERE slug = %s", body.slug):
+        raise HTTPException(409, f"company {body.slug!r} already exists")
+
+    async with db.pool.connection() as conn:
+        async with conn.transaction():
+            cur = await conn.execute(
+                "INSERT INTO company (slug, name, description, lifecycle_state) "
+                "VALUES (%s, %s, %s, 'active') RETURNING id",
+                (body.slug, body.name, body.brief),
+            )
+            company_id = (await cur.fetchone())["id"]
+            await db.append_event(conn, actor="user:philip", event_type="company.created",
+                                  company_id=company_id, source="api",
+                                  payload={"slug": body.slug, "name": body.name})
+
+            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)),
+                )
+                emp_id = (await cur.fetchone())["id"]
+                await db.append_event(conn, actor="system", event_type="employee.hired",
+                                      company_id=company_id,
+                                      payload={"employee_id": str(emp_id), "name": spec.name,
+                                               "title": spec.title, "role": spec.role})
+                employees.append({"id": emp_id, "spec": spec})
+
+    # Hermes profiles outside the tx: DB is source of truth; profile creation is
+    # retryable. Failure leaves employees hired with a repairable profile gap.
+    profile_errors = []
+    for emp in employees:
+        spec = emp["spec"]
+        try:
+            await hermes.create_profile(
+                profile_name(body.slug, spec.name),
+                soul=build_soul(spec.role, name=spec.name, title=spec.title, company_name=body.name),
+                model="qwen3.5:9b",
+                description=f"{body.name}: {spec.title}",
+            )
+        except HermesError as e:
+            profile_errors.append({"employee": spec.name, "error": str(e)[:200]})
+
+    return {
+        "company_id": str(company_id),
+        "slug": body.slug,
+        "employees": [{"id": str(e["id"]), "name": e["spec"].name} for e in employees],
+        "profile_errors": profile_errors,
+    }
+
+
+@app.get("/companies")
+async def list_companies() -> list[dict]:
+    rows = await db.fetch_all(
+        "SELECT c.*, count(e.id) AS employee_count FROM company c "
+        "LEFT JOIN employee e ON e.company_id = c.id GROUP BY c.id ORDER BY c.created_at"
+    )
+    return [db.as_json(r) for r in rows]
+
+
+@app.get("/companies/{slug}")
+async def get_company(slug: str) -> dict:
+    company = await db.fetch_one("SELECT * FROM company WHERE slug = %s", slug)
+    if not company:
+        raise HTTPException(404, "no such company")
+    employees = await db.fetch_all(
+        "SELECT id, name, title, department, status, authority_level, hermes_profile "
+        "FROM employee WHERE company_id = %s ORDER BY created_at", company["id"]
+    )
+    return db.as_json({**company, "employees": employees})
+
+
+@app.get("/companies/{slug}/events")
+async def company_events(slug: str, limit: int = 50) -> 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 * FROM event WHERE company_id = %s ORDER BY created_at DESC LIMIT %s",
+        company["id"], limit,
+    )
+    return [db.as_json(r) for r in rows]
+
+
+@app.delete("/companies/{slug}", status_code=204)
+async def archive_company(slug: str) -> None:
+    """Archive (lifecycle change + Hermes profile removal). Rows are kept."""
+    company = await db.fetch_one("SELECT * FROM company WHERE slug = %s", slug)
+    if not company:
+        raise HTTPException(404, "no such company")
+    employees = await db.fetch_all(
+        "SELECT hermes_profile FROM employee WHERE company_id = %s", company["id"]
+    )
+    async with db.pool.connection() as conn:
+        async with conn.transaction():
+            await conn.execute(
+                "UPDATE company SET lifecycle_state = 'archived' WHERE id = %s", (company["id"],)
+            )
+            await db.append_event(conn, actor="user:philip", event_type="company.archived",
+                                  company_id=company["id"], source="api")
+    for row in employees:
+        if row["hermes_profile"]:
+            try:
+                await hermes.delete_profile(row["hermes_profile"])
+            except HermesError:
+                pass  # profile may already be gone; DB stays authoritative
+
+
+# ---------------------------------------------------------------- global pause (kill switch v1)
+
+
+@app.post("/pause")
+async def pause() -> dict:
+    await redis_client.set(PAUSE_KEY, "1")
+    async with db.pool.connection() as conn:
+        await db.append_event(conn, actor="user:philip", event_type="company.paused",
+                              source="api", payload={"scope": "global"})
+    return {"paused": True}
+
+
+@app.post("/resume")
+async def resume() -> dict:
+    await redis_client.delete(PAUSE_KEY)
+    async with db.pool.connection() as conn:
+        await db.append_event(conn, actor="user:philip", event_type="company.resumed",
+                              source="api", payload={"scope": "global"})
+    return {"paused": False}

engine/souls.py

@@ -0,0 +1,37 @@
+"""Employee soul (system prompt) generation. Versioned via employee.soul_version."""
+
+PROTOCOL_RULES = """
+## Reporting protocol
+You are an employee of {company_name}, a FarmHouse Holdings company. Work only
+within your assignment. When you finish, produce a final summary that states:
+what you did, what you produced (files, commits, test results), and anything
+still unresolved. If the task is materially ambiguous, stop and state exactly
+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.
+"""
+
+ROLE_CHARTERS = {
+    "manager": """You are {name}, {title} of {company_name}.
+You convert objectives into small, well-scoped tasks with acceptance criteria.
+You review employee reports, consolidate questions so the owner is not
+overwhelmed, and reclassify blocked work: decide whether it needs a decision
+from the owner (clarification), an authority grant (approval), or a retry with
+better instructions. You never implement code yourself.""",
+    "engineer": """You are {name}, {title} at {company_name}.
+You implement well-scoped tasks in an isolated git workspace. You write clean,
+minimal code matching the existing style, run the relevant checks before
+declaring anything done, and commit with clear messages. If acceptance
+criteria are ambiguous, you ask rather than guess.""",
+    "qa": """You are {name}, {title} at {company_name}.
+You verify work like a skeptical user: run the automated tests, then exercise
+the actual product (browser or API) the way a human would. You file precise
+defects with reproduction steps and evidence (output, screenshots, traces).
+You re-verify fixes before closing anything. A claim without evidence fails.""",
+}
+
+
+def build_soul(role: str, *, name: str, title: str, company_name: str) -> str:
+    charter = ROLE_CHARTERS[role].format(name=name, title=title, company_name=company_name)
+    return charter + "\n" + PROTOCOL_RULES.format(company_name=company_name)

requirements.txt

@@ -3,7 +3,7 @@ uvicorn[standard]>=0.30
 httpx>=0.27
 pydantic>=2.7
 python-dotenv>=1.0
-psycopg[binary]>=3.2
+psycopg[binary,pool]>=3.2
 redis>=5.0
 pytest>=8.0
 pytest-asyncio>=0.24