Shareholder line: directives endpoint + CEO milestone updates

- POST /companies/{slug}/directives: free-form shareholder message becomes
  an objective (reuses manager decomposition), logs directive.received
- Objective rollup now emits a CEO-voiced milestone.update event via one
  pool call (plain-text fallback if the LLM call fails)
- _chat_structured retries once on empty/invalid JSON from the pool
  (observed empty-content responses breaking protocol extraction)
- Dashboard: per-company shareholder textarea + CEO updates feed; refresh
  skips DOM rewrite while a textarea is focused

Verified end-to-end: directive -> 3 tasks -> rework after handoff block ->
rollup -> CEO milestone update; gaswatch index.html now on-brand.

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

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:

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)

engine/protocol.py

@@ -54,6 +54,7 @@ DECOMPOSE_FORMAT: dict[str, Any] = {
 
 async def _chat_structured(system: str, user: str, fmt: dict) -> dict:
     async with httpx.AsyncClient(timeout=180) as client:
+        for attempt in (1, 2):  # pool occasionally returns empty content; one retry
             r = await client.post(
                 f"{settings.ollama_pool_url}/api/chat",
                 json={
@@ -68,7 +69,12 @@ async def _chat_structured(system: str, user: str, fmt: dict) -> dict:
                 },
             )
             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 = (

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); }
 </style>
@@ -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 `<div class="ev"><span class="t">${t}</span><span class="type">${e.event_type}</span>${String(txt).slice(0,120)}</div>`;
       }).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 `<div class="ev"><span class="t">${t}</span><span class="type">${who}</span>${String(txt).slice(0,300)}</div>`;
+      }).join('');
       html += `<div class="card"><h1>${c.name} <span class="pill ${c.lifecycle_state}">${c.lifecycle_state}</span></h1>
         <div class="grid">
           <div>
@@ -87,7 +111,15 @@ async function refresh() {
             <h2>Objectives</h2><table>${objs || '<tr><td class="muted">none</td></tr>'}</table>
             <h2>Work items</h2><table>${rows || '<tr><td class="muted">none</td></tr>'}</table>
           </div>
-          <div><h2>Event feed</h2><div class="feed">${feed}</div></div>
+          <div>
+            <h2>Shareholder line</h2>
+            <form onsubmit="return sendDirective(event, '${c.slug}')">
+              <textarea id="dir-${c.slug}" rows="2" placeholder="Directive to ${c.name} leadership…"></textarea>
+              <button type="submit">Send directive</button>
+            </form>
+            <h2>CEO updates</h2><div class="feed">${ceoFeed || '<div class="muted">no updates yet</div>'}</div>
+            <h2>Event feed</h2><div class="feed">${feed}</div>
+          </div>
         </div></div>`;
     }
     document.getElementById('companies').innerHTML = html || '<p class="muted">No companies yet.</p>';