From c8bfa0c0aba0de755d52acc72177157dfab2c1a2 Mon Sep 17 00:00:00 2001 From: Philip Date: Tue, 4 Aug 2026 00:36:36 -0500 Subject: [PATCH] =?UTF-8?q?Slice=201:=20engine=20core=20=E2=80=94=20compos?= =?UTF-8?q?e=20stack,=20db=20layer,=20company/employee=20API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- docker-compose.yml | 28 +++++++ engine/db.py | 74 ++++++++++++++++ engine/main.py | 204 +++++++++++++++++++++++++++++++++++++++++++++ engine/souls.py | 37 ++++++++ requirements.txt | 2 +- 5 files changed, 344 insertions(+), 1 deletion(-) create mode 100644 docker-compose.yml create mode 100644 engine/db.py create mode 100644 engine/main.py create mode 100644 engine/souls.py diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..6cf1750 --- /dev/null +++ b/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: diff --git a/engine/db.py b/engine/db.py new file mode 100644 index 0000000..e30e5c5 --- /dev/null +++ b/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)) diff --git a/engine/main.py b/engine/main.py new file mode 100644 index 0000000..71d95bc --- /dev/null +++ b/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} diff --git a/engine/souls.py b/engine/souls.py new file mode 100644 index 0000000..b8df818 --- /dev/null +++ b/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) diff --git a/requirements.txt b/requirements.txt index 69ba2f3..755aa28 100644 --- a/requirements.txt +++ b/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