feat: the fleet board contract — clients can finally see the conductor (P5.0)

docs/conductor-design.md

@@ -397,8 +397,10 @@ resolves against the owner's tenancy, so no ownership hack is needed.
   embedded, direct `SessionManager` access). The owner DMs the bot; the message is
   routed to the conductor session. No new channel needed for v1. (Web UI cockpit
   remains the visual view.)
-  **Status: designed, not built.** [conductor-frontends-design.md](./conductor-frontends-design.md) specifies both; neither is implemented — Telegram has no conductor routing and the web UI has no conductor pane, so the only front door today is `codeoid attach conductor` in a terminal.
-  This is the single largest gap between "the feature is implemented" and "the feature is usable", and it is also what generates the usage the rest of the design assumes.
+  **Status: the contract is built; the surfaces are not.** [conductor-frontends-design.md](./conductor-frontends-design.md) specifies both.
+  **P5.0 has landed**: `fleet.subscribe` → `fleet.snapshot.result` + streamed `fleet.update`, gated on the new `fleet:read` scope, advertised as the `fleet.board` capability, and mirrored in the Rust `codeoid-protocol` crate (which also gained the `SessionInfo.role` field it was missing).
+  Clients can now read and follow the board — but **no client draws it yet** (P5.1–P5.4): Telegram has no conductor routing and the web UI has no conductor pane, so the only front door today is still `codeoid attach conductor` in a terminal.
+  This remains the single largest gap between "the feature is implemented" and "the feature is usable", and it is also what generates the usage the rest of the design assumes.
 - **Wake model:** the conductor is event-driven. Wake sources:
   1. owner message (Telegram/Web),
   2. child-session completion (daemon emits an event → conductor turn),

docs/conductor-frontends-design.md

@@ -325,11 +325,21 @@ Each slice is a shippable PR that leaves both clients in a working state.
 Per [conductor-build-plan.md](./conductor-build-plan.md), `main` is ruleset-protected, so each lands as its own reviewable PR.
 User decision (2026-07-17): **docked-first** — skip the drawer/modal stepping-stone and go straight to the docked conductor surface.
 
-**P5.0 — The contract.**
+**P5.0 — The contract. ✅ SHIPPED.**
 Add `fleet.subscribe` / `fleet.snapshot.result` / `fleet.update` + the `fleet:read` scope to `@highflame/codeoid-protocol`; mirror in the Rust crate *and* add the missing `role` field.
 Daemon exposes the read surface from `dispatch_tasks` / `dispatch_events` + session population, pushing the *whole subtree* (not just active-session children).
 Ship the capability matrix as data.
-Files: daemon `src/daemon/{fleet.ts,server.ts,store.ts}` · `packages/protocol` · `crates/codeoid-protocol`.
+Files: daemon `src/daemon/{session-manager.ts,dispatch.ts,server.ts,store.ts}` · `packages/protocol` · `crates/codeoid-protocol`.
+
+As built, with three decisions the sketch above did not settle:
+
+- **`fleet.unsubscribe` was added.** Without it a client that navigates away from the Conductor home could only stop the delta stream by dropping its socket.
+- **The board change signal is one hook, not fourteen.** `DispatcherHost.onBoardChange` fires once per entry path (enqueue, group enqueue, end of tick) rather than at each individual store mutation. Every mutation happens inside one of those paths, so coverage is complete by construction and a future mutation added inside the tick cannot be missed.
+- **Deltas are exactly-once via a compound watermark.** `updated_at` is millisecond-granular and one tick routinely settles several tasks in the same millisecond, so a single cursor either drops tasks (`>`) or repeats them (`>=`). The watermark carries `taskUpdatedAt` *plus the ids already sent at exactly that millisecond*; the query stays inclusive and the id set suppresses the repeats.
+
+Two things deliberately NOT on the wire: the dispatch `prompt` and the worker `workdir`. The board renders lifecycle, and the prompt is the one task field carrying arbitrary user text to every subscribed client. `dependsOn` is present on the wire type but never populated, exactly as §11 specifies.
+
+The daemon advertises `fleet.board` (`CAPABILITIES.FLEET_BOARD`) so a client feature-detects before offering a conductor surface and an older daemon degrades to chat-only rather than showing an empty board.
 
 **P5.1 — Chat + legible fleet (zero-graph).**
 The conductor chat works the moment you can attach to it — the `role:"conductor"` session renders its transcript + prompt like any agent for free.

packages/protocol/src/schemas.test.ts

@@ -173,6 +173,8 @@ const samples: { [T in ClientTypes]: Extract<ClientMessage, { type: T }> } = {
   "pipeline.pack.select": { type: "pipeline.pack.select", id: "r43", packId: "aif-sdlc" },
   "push.register": { type: "push.register", id: "r44", token: "ExponentPushToken[abc]", platform: "ios" },
   "push.unregister": { type: "push.unregister", id: "r45", token: "ExponentPushToken[abc]" },
+  "fleet.subscribe": { type: "fleet.subscribe", id: "r46", scope: "tenant" },
+  "fleet.unsubscribe": { type: "fleet.unsubscribe", id: "r47" },
 };
 
 describe("fidelity — valid samples round-trip unchanged", () => {

packages/protocol/src/schemas.ts

@@ -362,6 +362,22 @@ export const collaborationPanelsSchema = z.object({
   sessionId: sessionIdField,
 });
 
+/**
+ * Fleet board subscribe/unsubscribe. `scope` is a closed literal rather than a
+ * free string so widening it later ("machine", "account") is an explicit,
+ * reviewable protocol change instead of something a client can just ask for.
+ */
+export const fleetSubscribeSchema = z.object({
+  ...base,
+  type: z.literal("fleet.subscribe"),
+  scope: z.literal("tenant"),
+});
+
+export const fleetUnsubscribeSchema = z.object({
+  ...base,
+  type: z.literal("fleet.unsubscribe"),
+});
+
 export const blackboardIndexSchema = z.object({
   ...base,
   type: z.literal("blackboard.index"),
@@ -662,6 +678,8 @@ export const clientMessageSchema = z.discriminatedUnion("type", [
   pipelinePackSelectSchema,
   pushRegisterSchema,
   pushUnregisterSchema,
+  fleetSubscribeSchema,
+  fleetUnsubscribeSchema,
 ]);
 
 /**

packages/protocol/src/scopes.ts

@@ -32,6 +32,18 @@ export const SCOPES = {
    * interrupt it, or spawn a disposable worker on the owner's behalf.
    */
   SESSION_DISPATCH: "session:dispatch",
+  /**
+   * Subscribe a CLIENT to the fleet board — the conductor's dispatch tasks,
+   * lifecycle events, and aggregate usage (`fleet.subscribe`).
+   *
+   * Distinct from `session:read`/`session:dispatch`, which are ZeroID scopes
+   * delegated to the conductor's own AGENT identity to gate its `fleet_*` MCP
+   * tools. This one gates a human's client reading the board over the wire.
+   * Separate from `session:list` because the board exposes orchestration
+   * internals — what was dispatched, what failed, what it cost — beyond the
+   * session enumeration a watcher already gets.
+   */
+  FLEET_READ: "fleet:read",
   /** Read files and list directories under a session's workdir */
   FS_READ: "fs:read",
   /** Read the settings manifest + current (non-secret) daemon configuration */
@@ -78,6 +90,10 @@ export const OPERATOR_SCOPES: readonly Scope[] = [
   SCOPES.SESSION_SEND,
   SCOPES.SESSION_INTERRUPT,
   SCOPES.SESSION_APPROVE,
+  // An operator drives the fleet, so the conductor board is part of their job.
+  // Deliberately NOT in WATCHER_SCOPES: a read-only watcher can already see
+  // sessions, but the board is orchestration state, not session output.
+  SCOPES.FLEET_READ,
   SCOPES.FS_READ,
   SCOPES.SETTINGS_READ,
   SCOPES.PIPELINE_CREATE,

packages/protocol/src/types.ts

@@ -98,6 +98,13 @@ export const CAPABILITIES = {
    * Expo token it sends for `PUSH`).
    */
   PUSH_NATIVE: "push.native",
+  /**
+   * The fleet board read+subscribe surface (`fleet.subscribe` →
+   * `fleet.snapshot.result` + streamed `fleet.update`). Advertised by the
+   * DAEMON; a client feature-detects before offering a conductor surface, so an
+   * older daemon degrades to chat-only rather than showing an empty board.
+   */
+  FLEET_BOARD: "fleet.board",
 } as const;
 
 export type Capability = (typeof CAPABILITIES)[keyof typeof CAPABILITIES];
@@ -901,7 +908,9 @@ export type ClientMessage =
   | PipelinePackTrustMsg
   | PipelinePackSelectMsg
   | PushRegisterMsg
-  | PushUnregisterMsg;
+  | PushUnregisterMsg
+  | FleetSubscribeMsg
+  | FleetUnsubscribeMsg;
 
 interface BaseClientMsg {
   /** Request ID for correlating responses */
@@ -1796,6 +1805,130 @@ export interface CollaborationPanelsResultMsg {
   panels: CollaborationPanel[];
 }
 
+// ── Fleet board (conductor front doors — docs/conductor-frontends-design.md §11) ─
+//
+// The conductor deliberately added ZERO client↔daemon wire types: it renders as
+// an ordinary session, its `fleet_*` calls as tool cards. That is enough to CHAT
+// with it; it is not enough to SEE the fleet, because the task board, worker
+// lifecycle and audit trail live only in daemon SQLite (`dispatch_tasks` /
+// `dispatch_events`).
+//
+// This is the one additive read+subscribe surface that closes that gap. It adds
+// no dispatch semantics — every field below is already held by the daemon.
+
+/**
+ * A dispatch task, projected for a client.
+ *
+ * Named `*Wire` per the `PipelineWire` / `PackWire` convention, and to stay
+ * distinct from the daemon-internal `FleetTaskView` in `src/daemon/fleet.ts`,
+ * which is what the conductor's own `fleet_tasks` TOOL sees. The two are
+ * deliberately different shapes: the tool view is what an LLM should read, this
+ * is what a UI needs to draw a node (worker join key, provenance, cost).
+ */
+export interface FleetTaskWire {
+  id: string;
+  kind: "send" | "spawn";
+  shape: "ship" | "scout";
+  status: "queued" | "claimed" | "running" | "done" | "failed" | "blocked";
+  attempts: number;
+  /** Epoch ms. */
+  createdAt: number;
+  /** spawn: the worker session this task created. Join key into `workers`. */
+  workerSessionId?: string;
+  /** send: the existing session this task was routed to. Join key into `workers`. */
+  targetSession?: string;
+  /** Compressed result — never a raw transcript (the never-OOC guarantee). */
+  resultDigest?: string;
+  error?: string;
+  /** Conductor WIMSE URI — who dispatched this. */
+  createdBy: string;
+  /** Dispatch group (fan-out barrier); absent = a standalone task. */
+  groupId?: string;
+  /**
+   * RESERVED and never populated in P5. The daemon has no dependency model —
+   * the conductor sequences in prose. Present so typed fan-in/join edges are a
+   * later non-breaking add rather than a wire break.
+   */
+  dependsOn?: string[];
+}
+
+/** A dispatch lifecycle event — the audit trail behind the board. */
+export interface FleetEventWire {
+  id: number;
+  taskId: string;
+  type: string;
+  /** Compressed digest of what happened. */
+  digest: string;
+  /** Epoch ms. */
+  createdAt: number;
+}
+
+/** Fleet-wide rollup, normalized across backends so one number spans vendors. */
+export interface FleetUsage {
+  /** Tasks not yet terminal (queued + claimed + running). */
+  activeTasks: number;
+  /** Tasks in `blocked` — the anti-spin failure cap tripped; needs a human. */
+  blockedTasks: number;
+  inputTokens: number;
+  outputTokens: number;
+  /** USD, summed across every backend that reports cost. */
+  totalCostUsd: number;
+}
+
+/** Everything a client needs to draw the fleet, in one payload. */
+export interface FleetSnapshot {
+  /** Absent when the tenant has no conductor session — a valid, common state. */
+  conductor?: SessionInfo;
+  /**
+   * Sessions the board references: spawned workers AND existing sessions that
+   * were dispatched to. Full `SessionInfo`, so a node renders with status,
+   * usage and backend without duplicating those fields onto the task.
+   */
+  workers: SessionInfo[];
+  /** Newest first. */
+  tasks: FleetTaskWire[];
+  /** Newest first. */
+  events: FleetEventWire[];
+  agg: FleetUsage;
+}
+
+/**
+ * An incremental board change. Carries the FULL task/event rather than a patch:
+ * a client that missed a delta still converges, and the payload is small.
+ */
+export type FleetDelta =
+  | { kind: "task"; task: FleetTaskWire; agg: FleetUsage }
+  | { kind: "event"; event: FleetEventWire; agg: FleetUsage };
+
+/** Subscribe to the fleet board: replies with a snapshot, then streams deltas. */
+export interface FleetSubscribeMsg extends BaseClientMsg {
+  type: "fleet.subscribe";
+  /** Only "tenant" today — the caller's own account+project board. */
+  scope: "tenant";
+}
+
+/**
+ * Stop the delta stream. Not in the original §11 sketch, but without it a client
+ * that navigates away from the Conductor home can only stop the stream by
+ * dropping its socket.
+ */
+export interface FleetUnsubscribeMsg extends BaseClientMsg {
+  type: "fleet.unsubscribe";
+}
+
+/** Reply to fleet.subscribe. */
+export interface FleetSnapshotResultMsg {
+  type: "fleet.snapshot.result";
+  requestId: string;
+  fleet: FleetSnapshot;
+}
+
+/** Broadcast to subscribed clients in the tenant (mirrors session.status_change). */
+export interface FleetUpdateMsg {
+  type: "fleet.update";
+  delta: FleetDelta;
+}
+
 /** One index row: what exists, at what version, by whom — never a body. */
 export interface BlackboardIndexEntry {
   /** A core kind (`spec`, `research`, …) or `extra/<key>`. */
@@ -2272,7 +2405,9 @@ export type DaemonMessage =
   | SettingsSetResultMsg
   | PipelineSnapshotMsg
   | PipelineListResultMsg
-  | PackListResultMsg;
+  | PackListResultMsg
+  | FleetSnapshotResultMsg
+  | FleetUpdateMsg;
 
 export interface AuthOkMsg {
   type: "auth.ok";

src/daemon/dispatch.ts

@@ -97,6 +97,20 @@ export interface DispatcherHost {
     events: DispatchEventRow[],
   ): Promise<readonly number[]>;
   audit(action: string, detail: string): void;
+  /**
+   * The task board changed — fleet subscribers need deltas
+   * (docs/conductor-frontends-design.md §11).
+   *
+   * Signalled once per entry path (enqueue, group enqueue, end of tick) rather
+   * than at each of the ~14 individual store mutations. Every mutation happens
+   * inside one of those paths, so this is complete by construction and cannot
+   * be missed by a future mutation added inside the tick. The host derives the
+   * precise deltas from its own watermark; this only says "something moved".
+   *
+   * Optional: a host with no connected clients (tests, a Telegram-only daemon)
+   * simply omits it.
+   */
+  onBoardChange?(): void;
 }
 
 /**
@@ -239,6 +253,7 @@ export class Dispatcher {
       `task=${id} kind=${input.kind} shape=${input.shape} target=${input.targetSession ?? input.workdir ?? "-"}` +
         `${input.provider ? ` provider=${input.provider}` : ""}${input.model ? ` model=${input.model}` : ""}`,
     );
+    this.#signalBoardChange();
     return id;
   }
 
@@ -306,6 +321,7 @@ export class Dispatcher {
         .join(",")
         .slice(0, 300)}`,
     );
+    this.#signalBoardChange();
     return { groupId, taskIds };
   }
 
@@ -328,6 +344,22 @@ export class Dispatcher {
       );
     } finally {
       this.#ticking = false;
+      // In `finally`, and after `#ticking` is cleared: a tick that threw
+      // part-way through has still usually moved some tasks, and a board that
+      // silently stopped updating after one bad tick is worse than a delta the
+      // client can reconcile. Never allowed to throw into the tick.
+      this.#signalBoardChange();
+    }
+  }
+
+  /** Tell the host the board moved. Failure here must never break dispatch. */
+  #signalBoardChange(): void {
+    try {
+      this.#host.onBoardChange?.();
+    } catch (err) {
+      console.error(
+        `[codeoid/dispatch] board-change notify failed: ${err instanceof Error ? err.message : String(err)}`,
+      );
     }
   }
 

src/daemon/server.ts

@@ -61,6 +61,7 @@ const SERVER_CAPABILITIES: string[] = [
   CAPABILITIES.UI_DIALOGS,
   CAPABILITIES.DYNAMIC_COMMANDS,
   CAPABILITIES.BLACKBOARD,
+  CAPABILITIES.FLEET_BOARD,
 ];
 
 /**