diff --git a/engine/dispatcher.py b/engine/dispatcher.py
index f0fc55a..9013581 100644
--- a/engine/dispatcher.py
+++ b/engine/dispatcher.py
@@ -18,7 +18,7 @@ from datetime import datetime, timezone
from engine import db
from engine.hermes_client import HermesClient, HermesError
-from engine.protocol import extract_agent_message
+from engine.protocol import ceo_update, extract_agent_message
log = logging.getLogger("farmhouse.dispatcher")
@@ -81,7 +81,8 @@ class Dispatcher:
async def _roll_up_objectives(self) -> None:
rows = await db.fetch_all(
"""
- SELECT o.id, o.company_id FROM work_item o
+ SELECT o.id, o.company_id, o.title, o.owner_id, c.name AS company_name
+ FROM work_item o JOIN company c ON c.id = o.company_id
WHERE o.type = 'objective' AND o.status = 'in_progress'
AND EXISTS (SELECT 1 FROM work_item c WHERE c.parent_id = o.id)
AND NOT EXISTS (SELECT 1 FROM work_item c
@@ -89,6 +90,16 @@ class Dispatcher:
"""
)
for row in rows:
+ children = await db.fetch_all(
+ "SELECT title, status FROM work_item WHERE parent_id=%s ORDER BY created_at",
+ row["id"])
+ task_lines = [f"{c['title']} ({c['status']})" for c in children]
+ try:
+ update = await ceo_update(row["company_name"], row["title"], task_lines)
+ except Exception:
+ log.exception("ceo_update failed; falling back to plain summary")
+ update = (f"Milestone reached: {row['title']} — "
+ f"all {len(children)} tasks closed.")
async with db.pool.connection() as conn:
async with conn.transaction():
await conn.execute(
@@ -97,6 +108,11 @@ class Dispatcher:
await db.append_event(conn, actor="system", event_type="work_item.updated",
company_id=row["company_id"], work_item_id=row["id"],
payload={"status": "done", "note": "all child tasks done"})
+ await db.append_event(
+ conn, actor=f"employee:{row['owner_id']}",
+ event_type="milestone.update",
+ company_id=row["company_id"], work_item_id=row["id"],
+ payload={"objective_title": row["title"], "summary": update})
log.info("objective %s rolled up to done", row["id"])
async def _dispatch_ready(self) -> None:
diff --git a/engine/main.py b/engine/main.py
index 7f3ca19..919585c 100644
--- a/engine/main.py
+++ b/engine/main.py
@@ -4,6 +4,7 @@ Run: uvicorn engine.main:app --reload --port 8100
"""
from contextlib import asynccontextmanager
from pathlib import Path
+from uuid import UUID
import redis.asyncio as aioredis
from fastapi import FastAPI, HTTPException
@@ -275,6 +276,25 @@ async def create_objective(slug: str, body: ObjectiveCreate) -> dict:
"clarification_needed": plan.get("clarification_needed") or None}
+class DirectiveCreate(BaseModel):
+ message: str = Field(min_length=1, max_length=4000)
+
+
+@app.post("/companies/{slug}/directives", status_code=201)
+async def shareholder_directive(slug: str, body: DirectiveCreate) -> dict:
+ """Shareholder line: free-form message becomes an objective the manager
+ decomposes; completion comes back as a milestone.update event."""
+ title = body.message.strip().splitlines()[0][:80]
+ result = await create_objective(slug, ObjectiveCreate(title=title, brief=body.message))
+ company = await db.fetch_one("SELECT id FROM company WHERE slug = %s", slug)
+ async with db.pool.connection() as conn:
+ await db.append_event(
+ conn, actor="shareholder:philip", event_type="directive.received",
+ company_id=company["id"], work_item_id=UUID(result["objective_id"]),
+ source="api", payload={"message": body.message[:500]})
+ return result
+
+
@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)
diff --git a/engine/protocol.py b/engine/protocol.py
index 8815e53..65bf204 100644
--- a/engine/protocol.py
+++ b/engine/protocol.py
@@ -54,21 +54,27 @@ DECOMPOSE_FORMAT: dict[str, Any] = {
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"])
+ for attempt in (1, 2): # pool occasionally returns empty content; one retry
+ 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()
+ try:
+ return json.loads(r.json()["message"]["content"])
+ except json.JSONDecodeError:
+ if attempt == 2:
+ raise
+ raise RuntimeError("unreachable")
async def extract_agent_message(work_item_id: str, raw_report: str) -> AgentMessage:
@@ -81,6 +87,26 @@ async def extract_agent_message(work_item_id: str, raw_report: str) -> AgentMess
return AgentMessage.model_validate(data)
+CEO_UPDATE_FORMAT: dict[str, Any] = {
+ "type": "object",
+ "properties": {"update": {"type": "string"}},
+ "required": ["update"],
+}
+
+
+async def ceo_update(company_name: str, objective_title: str, task_lines: list[str]) -> str:
+ """CEO-voiced 2-4 sentence shareholder update for a completed objective."""
+ system = (
+ f"You are the CEO of {company_name} writing a short shareholder update "
+ "(2-4 sentences). Report the milestone plainly: what shipped and what "
+ "comes next. No hype. Do not invent facts not in the task list."
+ )
+ user = (f"Milestone reached: {objective_title}\nCompleted tasks:\n"
+ + "\n".join(f"- {t}" for t in task_lines))
+ data = await _chat_structured(system, user, CEO_UPDATE_FORMAT)
+ return data["update"]
+
+
async def decompose_objective(brief: str, company_name: str) -> dict:
"""Manager-style decomposition into 2-5 small tasks. Returns validated dict."""
system = (
diff --git a/engine/static/index.html b/engine/static/index.html
index 3476d93..4944067 100644
--- a/engine/static/index.html
+++ b/engine/static/index.html
@@ -29,6 +29,8 @@
button { background:var(--card); color:var(--text); border:1px solid var(--line); border-radius:6px;
padding:.4rem .9rem; cursor:pointer; font-size:.85rem; }
button:hover { border-color:var(--dim); }
+ textarea { width:100%; background:var(--bg); color:var(--text); border:1px solid var(--line);
+ border-radius:6px; padding:.5rem; font:inherit; resize:vertical; margin-bottom:.4rem; }
#pausebtn.paused { color:var(--bad); border-color:var(--bad); }
.muted { color:var(--dim); }
@@ -51,7 +53,21 @@ async function togglePause() {
refresh();
}
+async function sendDirective(ev, slug) {
+ ev.preventDefault();
+ const el = document.getElementById('dir-' + slug);
+ const message = el.value.trim();
+ if (!message) return false;
+ await j(`/companies/${slug}/directives`, {method:'POST',
+ headers:{'Content-Type':'application/json'}, body: JSON.stringify({message})});
+ el.value = '';
+ refresh();
+ return false;
+}
+
async function refresh() {
+ // don't clobber a directive mid-typing
+ if (document.activeElement && document.activeElement.tagName === 'TEXTAREA') return;
try {
const h = await j('/health');
paused = h.paused;
@@ -80,6 +96,14 @@ async function refresh() {
const txt = p.summary ?? p.reason ?? p.title ?? p.question ?? p.name ?? '';
return `${t}${e.event_type}${String(txt).slice(0,120)}
`;
}).join('');
+ const CEO_EVENTS = ['milestone.update', 'directive.received', 'clarification.requested'];
+ const ceoFeed = events.filter(e => CEO_EVENTS.includes(e.event_type)).map(e => {
+ const t = new Date(e.created_at).toLocaleTimeString();
+ const p = e.payload ?? {};
+ const txt = p.summary ?? p.message ?? p.question ?? '';
+ const who = e.event_type === 'directive.received' ? '🧑💼 you' : '👔 CEO';
+ return `${t}${who}${String(txt).slice(0,300)}
`;
+ }).join('');
html += `${c.name} ${c.lifecycle_state}
@@ -87,7 +111,15 @@ async function refresh() {
Objectives
Work items
-
+
+
Shareholder line
+
+
CEO updates
${ceoFeed || '
no updates yet
'}
+
Event feed
${feed}
+
No companies yet.
';