feat: dispatch + routing — the conductor directs the fleet (P4)

docs/conductor-design.md

@@ -116,11 +116,23 @@ is widened with `mcp__codeoid_fleet__*` when a fleet server is present, and the
 system-prompt append path no longer gates on memory so the conductor contract
 rides the `claude_code` preset.*)*
 
-**Read-only over targets, by construction (P3 scope).** The P3 fleet surface is
-`fleet_list` / `fleet_find` / `fleet_summary` / `fleet_recall` / `machine_map` —
-observation only. No send/spawn/interrupt tool exists yet (those are P4), and the
-conductor identity (P2) carries only `session:read`/`session:dispatch`, never
-`tools:write`/`tools:execute`, so nothing it delegates can mutate a target.
+**Read-only over targets, by construction (P3), dispatch behind approval (P4).**
+The read surface (`fleet_list` / `fleet_find` / `fleet_summary` / `fleet_recall` /
+`fleet_tasks` / `machine_map`) runs silently. The send surface
+(`fleet_send` / `fleet_spawn` / `fleet_interrupt`, P4) is kept OUT of the
+provider's `allowedTools` AND hard-blocked from auto-approval in every session
+mode — each dispatch rides the existing `approvalId` flow with the full tool
+input shown to the owner (R3 as an invariant, not a mode default). Approved
+dispatches execute through a durable SQLite work queue (`dispatch.ts`): atomic
+claims, boot-id stale reclaim, exponential retry backoff, failure-limit
+auto-block (the stuck-loop guard), and a per-tenant worker cap. Spawned workers
+are disposable `role:"worker"` sessions with shape-capped LEAF identities
+(scouts hold no `tools:write`; no worker ever holds `session:*`), an autonomous
+tool budget, and completion digests that flow back as batched, daemon-injected
+`<fleet_events>` turns — never raw transcripts. The conductor identity (P2)
+still carries only `session:read`/`session:dispatch`, never
+`tools:write`/`tools:execute`, so nothing it delegates can mutate a target;
+a worker's tool capability is sanctioned by the owner's fleet_spawn approval.
 
 **Provider-agnostic conductor.** Which backend drives the conductor is
 `config.conductor.provider` — any registered provider id, so an open-weight

packages/protocol/src/types.ts

@@ -122,9 +122,10 @@ export interface SessionInfo {
   attachedClients: number;
   /**
    * Session role. "conductor" marks the per-tenant conductor session (the
-   * fleet supervisor — one per account/project). Absent = normal session.
+   * fleet supervisor — one per account/project); "worker" marks a disposable
+   * dispatch-spawned worker. Absent = normal session.
    */
-  role?: "conductor";
+  role?: "conductor" | "worker";
   /** Id of the provider backing this session (e.g. "claude", "gemini"). */
   providerId?: string;
   /** Current execution mode (default "interactive"). */

src/config.ts

@@ -335,6 +335,38 @@ const OAuthSchemaFields = z
   })
   .default({});
 
+/**
+ * Dispatch queue (P4) — send-class fleet actions run through a durable
+ * SQLite work queue with a dispatcher loop. Workers spawned by the queue run
+ * autonomously up to `workerToolBudget` tool calls, then wedge safely (the
+ * lease reclaims them).
+ */
+const DispatchSchema = z
+  .object({
+    enabled: z.boolean().default(true),
+    /** Dispatcher tick interval (ms): claim / reclaim / deliver cadence. */
+    tickMs: z.number().int().min(250).default(5_000),
+    /** Claim lease (ms) — an unrenewed claim past this is reclaimed (attempts++). */
+    leaseMs: z.number().int().min(10_000).default(10 * 60_000),
+    /** Consecutive failures (incl. reclaims) before a task auto-blocks. */
+    failureLimit: z.number().int().min(1).default(2),
+    /** Max concurrently running spawned workers per tenant. */
+    maxConcurrentWorkers: z.number().int().min(1).default(2),
+    /** Autonomous tool-call budget per spawned worker. */
+    workerToolBudget: z.number().int().min(1).default(50),
+    /** Base retry backoff (ms) for retryable failures — doubles per attempt, capped at leaseMs. */
+    retryBaseMs: z.number().int().min(0).default(15_000),
+  })
+  .default({
+    enabled: true,
+    tickMs: 5_000,
+    leaseMs: 10 * 60_000,
+    failureLimit: 2,
+    maxConcurrentWorkers: 2,
+    workerToolBudget: 50,
+    retryBaseMs: 15_000,
+  });
+
 const RootSchema = z.object({
   daemonUrl: z.string().default("ws://127.0.0.1:7400"),
   dbPath: z.string().default("codeoid.db"),
@@ -355,6 +387,7 @@ const RootSchema = z.object({
   autoRotate: AutoRotateSchema,
   session: SessionSchema,
   conductor: ConductorSchema,
+  dispatch: DispatchSchema,
 });
 
 type ParsedConfig = z.infer<typeof RootSchema>;
@@ -445,6 +478,20 @@ export interface CodeoidConfig {
     provider: string;
     model?: string;
   };
+  /**
+   * Send-class dispatch queue (P4). Optional in the type so hand-built test
+   * configs stay minimal; loadConfig always populates it. Absent = enabled
+   * with defaults.
+   */
+  dispatch?: {
+    enabled: boolean;
+    tickMs: number;
+    leaseMs: number;
+    failureLimit: number;
+    maxConcurrentWorkers: number;
+    workerToolBudget: number;
+    retryBaseMs: number;
+  };
 }
 
 // ── Env-var override map ─────────────────────────────────────────────────
@@ -495,6 +542,10 @@ const ENV_OVERRIDES: readonly EnvOverride[] = [
   { env: "CODEOID_AUTO_ROTATE_HARD_PCT", path: "autoRotate.hardRotatePct", kind: "float" },
   { env: "CODEOID_AUTO_ROTATE_MIN_TURNS", path: "autoRotate.minTurnsBeforeRotate", kind: "int" },
   { env: "CODEOID_DEFAULT_MODEL", path: "session.defaultModel", kind: "string" },
+  // Dispatch kill switch — disable send-class fleet dispatch per-invocation
+  // without touching config.json. Other dispatch knobs are file-config only,
+  // matching the conductor block's convention.
+  { env: "CODEOID_DISPATCH_ENABLED", path: "dispatch.enabled", kind: "boolean" },
   { env: "CODEOID_FALLBACK_MODEL", path: "session.fallbackModel", kind: "string" },
   { env: "CODEOID_TURN_STALL_TIMEOUT_MS", path: "session.turnStallTimeoutMs", kind: "int" },
   { env: "CODEOID_MCP_TOOL_TIMEOUT_MS", path: "session.mcpToolTimeoutMs", kind: "int" },
@@ -669,6 +720,7 @@ export function loadConfig(opts: LoadOptions = {}): CodeoidConfig {
     autoRotate: parsed.autoRotate,
     session: parsed.session,
     conductor: parsed.conductor,
+    dispatch: parsed.dispatch,
   };
 }
 

src/daemon/agent-identity.ts

@@ -69,6 +69,31 @@ export const CONDUCTOR_SCOPES = [
   SCOPES.SESSION_DISPATCH, // direct, interrupt, or spawn sessions
 ] as const;
 
+/**
+ * Worker scope profiles by dispatch shape (P4, hermes leaf/orchestrator).
+ *
+ * The LEAF property: no worker profile ever includes session:read /
+ * session:dispatch / tools:agent-spawning-fleet authority — a worker cannot
+ * see or direct the fleet even if fleet tools were somehow mounted on it.
+ * The SHAPE property: scouts investigate-and-report, so their identity holds
+ * no tools:write.
+ *
+ * Why the worker's token is a ROOT grant and not a conductor delegation:
+ * ZeroID grants the intersection (requested ∩ subject.granted ∩
+ * actor.allowed) on every RFC 8693 hop, and the conductor's own authority is
+ * deliberately session:read/session:dispatch only — so a chain rooted at the
+ * conductor can NEVER carry tools:write. That is R1 working as intended: the
+ * conductor cannot mint mutation authority. A worker's tool capability is
+ * instead sanctioned by the OWNER's explicit fleet_spawn approval (R3), and
+ * the identity records `created_by = conductor WIMSE URI` for the audit
+ * lineage. Revocation rides session teardown (deactivateSessionAgent), which
+ * cascade-revokes the worker's own delegation subtree.
+ */
+const WORKER_SCOPE_PROFILES: Record<"ship" | "scout", readonly string[]> = {
+  ship: ["tools:read", "tools:write", "tools:execute", "tools:agent"],
+  scout: ["tools:read", "tools:execute", "tools:agent"],
+};
+
 /** Sub-agents get read-only by default unless explicitly promoted. */
 const SUBAGENT_DEFAULT_SCOPES = ["tools:read"] as const;
 
@@ -209,6 +234,76 @@ export class AgentIdentityManager {
     }
   }
 
+  /**
+   * Register a dispatch-spawned WORKER identity (P4). Shape-capped LEAF
+   * profile (see WORKER_SCOPE_PROFILES for why the token is a root grant
+   * sanctioned by the owner's fleet_spawn approval, not a conductor
+   * delegation), with `created_by` = the conductor's WIMSE URI so the audit
+   * lineage reads owner → conductor → worker. Stored under the session id
+   * like any session agent, so tool audit and teardown cascade unchanged.
+   */
+  async registerWorker(
+    sessionId: string,
+    sessionName: string,
+    shape: "ship" | "scout",
+  ): Promise<{ wimseUri: string; token: string }> {
+    const externalId = `codeoid-worker-${sessionId.slice(0, 8)}`;
+    const scopes = WORKER_SCOPE_PROFILES[shape];
+    const lineage = this.#conductor?.wimseUri ?? "codeoid:dispatch";
+
+    try {
+      const registerReq = {
+        name: `codeoid/worker/${shape}/${sessionName}`,
+        external_id: externalId,
+        sub_type: "tool_agent" as const,
+        trust_level: "first_party" as const,
+        framework: "claude-agent-sdk",
+        publisher: "codeoid",
+        created_by: lineage,
+        allowed_scopes: [...scopes],
+        metadata: JSON.stringify({
+          session_id: sessionId,
+          role: "worker",
+          shape,
+          spawned_by: lineage,
+        }),
+      };
+      const resp = await this.#client.agents.register(
+        registerReq as RegisterAgentRequest,
+      );
+
+      const tokenResp = await this.#client.tokens.issueApiKey(resp.api_key, {
+        scope: scopes.join(" "),
+      });
+
+      this.#agents.set(sessionId, {
+        identityId: resp.identity.id,
+        wimseUri: resp.identity.wimse_uri,
+        token: tokenResp.access_token,
+        apiKey: resp.api_key,
+        // Orchestrator client for the worker's OWN sub-agents (Explore etc.)
+        // — their delegated scopes intersect with the shape profile, so a
+        // scout's sub-agents can't hold tools:write either.
+        client: this.#clientForAgent(resp.api_key),
+      });
+
+      this.#store.audit(
+        resp.identity.wimse_uri,
+        "worker.identity.registered",
+        sessionId,
+        `shape=${shape} spawned_by=${lineage} scopes=${scopes.join(",")}`,
+      );
+
+      return { wimseUri: resp.identity.wimse_uri, token: tokenResp.access_token };
+    } catch (err) {
+      console.error(
+        `[codeoid] failed to register worker identity for ${sessionName}:`,
+        err instanceof Error ? err.message : err,
+      );
+      return { wimseUri: `anonymous:worker:${sessionId}`, token: "" };
+    }
+  }
+
   /**
    * Register a sub-agent identity when Claude spawns one (SubagentStart hook).
    * Token is delegated from the parent session agent with attenuated scopes.