From c358ef28511658bfed739b41cab5cb8d05b59ca2 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Tue, 7 Jul 2026 10:33:57 +0800 Subject: [PATCH 1/2] feat: conductor session + read-only fleet tools (P3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The conductor becomes a real, reachable session — not just P2 machinery. - role:"conductor" on session.create requests THE per-tenant conductor (singleton, idempotent): the daemon picks its name/workdir/provider, creates it on first request, and returns the existing one after. Persisted (Store + transcript meta) so it self-persists across restarts — one stable session. - codeoid_fleet in-process MCP server, injected only for the conductor: fleet_list / fleet_find (P1 cross-workspace resolution) / fleet_summary (episode digest, never raw scrollback) / fleet_recall / machine_map. Tools close over the manager's live tenant-scoped session view. Read-only by construction — no send/spawn (that's P4). Every call audits under the conductor's WIMSE URI. - Conductor identity wired (P2 -> live): creating the conductor calls registerConductor(ownerSub) and mints its working token by OWNER delegation (RFC 8693). The verified bearer token is retained per-connection (SocketData) as the delegation subject — never logged/persisted. - Per-session provider selection: each Session picks its backend from providerId (config.conductor.provider for the conductor), so any session — the conductor included — can run on claude/gemini/openai or a future open-weight provider. Stateless providers get a SessionProvider adapter. - Claude provider: allowedTools widened with mcp__codeoid_fleet__*; the system-prompt append no longer gates on memory (the conductor contract rides the claude_code preset). - Owner scopes: session:read / session:dispatch added to the CLI + web login scope sets so the owner->conductor delegation has a non-empty intersection. - UX: `codeoid attach conductor` create-or-gets the conductor from any client. - Unknown/future roles fail closed with a clear error (never a silent downgrade to a normal session); the wire frame still parses. Tests: fleet handlers (read surface, find resolution, conductor self-exclusion, memory-off fallback, audit) + conductor session lifecycle (singleton, role + provider persistence, tenancy isolation, resume, fail-closed roles). 1066 unit tests green; typecheck + lint clean. Co-Authored-By: Claude Fable 5 --- docs/conductor-design.md | 25 ++- packages/protocol/src/schemas.ts | 14 ++ packages/protocol/src/types.ts | 16 ++ src/cli.ts | 9 +- src/config.ts | 33 +++ src/daemon/fleet.ts | 299 +++++++++++++++++++++++++++ src/daemon/providers/claude/index.ts | 25 ++- src/daemon/providers/stateless.ts | 78 +++++++ src/daemon/server.ts | 12 +- src/daemon/session-manager.ts | 181 +++++++++++++++- src/daemon/session.ts | 131 +++++++++--- src/daemon/store.ts | 11 +- src/daemon/transcript.ts | 4 + src/terminal/client.ts | 23 ++- src/tests/conductor-session.test.ts | 178 ++++++++++++++++ src/tests/fleet.test.ts | 211 +++++++++++++++++++ web/src/lib/auth.ts | 4 + 17 files changed, 1214 insertions(+), 40 deletions(-) create mode 100644 src/daemon/fleet.ts create mode 100644 src/daemon/providers/stateless.ts create mode 100644 src/tests/conductor-session.test.ts create mode 100644 src/tests/fleet.test.ts diff --git a/docs/conductor-design.md b/docs/conductor-design.md index af8d620..dab54d0 100644 --- a/docs/conductor-design.md +++ b/docs/conductor-design.md @@ -107,13 +107,26 @@ Three additions, no architectural change: | **`codeoid_fleet` MCP server** | In-process Agent-SDK MCP server exposing fleet tools (list / spawn / send / watch / summarize / interrupt sessions, recall across threads). Bound to the conductor session only. | `buildMemoryMcpServer` at `session.ts:810` | | **Conductor identity grant** | The conductor's ZeroID agent identity additionally holds `session:*` scopes, so it can drive the fleet *as a first-class delegated authority* (see §4). | `AgentIdentityManager.registerSessionAgent` | -Injection point is already there: `session.ts:810` merges `codeoid_memory` into the +Injection point is already there: the Claude provider merges `codeoid_memory` into the `mcpServers` passed to `query()`. The conductor adds `codeoid_fleet` the same way, -gated on `role === "conductor"`. One P3 gotcha: the Claude provider's -`allowedTools` currently allowlists only `mcp__codeoid_memory__*` -(`providers/claude/index.ts`) — it must be widened to admit -`mcp__codeoid_fleet__*` for the conductor session, or the mounted server's tools -stay unreachable. +gated on `role === "conductor"`. *(Implemented in P3:* the manager builds the +fleet server — its tools close over the live, tenant-scoped session population — +and passes it to the conductor's `Session`; the Claude provider's `allowedTools` +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. + +**Provider-agnostic conductor.** Which backend drives the conductor is +`config.conductor.provider` — any registered provider id, so an open-weight +backend can run it once its provider exists. Caveat: MCP tools are only surfaced +by the Claude provider today, so a conductor on another provider chats but can't +see the fleet until that provider grows MCP support (the daemon logs this). --- diff --git a/packages/protocol/src/schemas.ts b/packages/protocol/src/schemas.ts index 4ced78c..2ef45d2 100644 --- a/packages/protocol/src/schemas.ts +++ b/packages/protocol/src/schemas.ts @@ -55,6 +55,20 @@ export const sessionCreateSchema = z.object({ type: z.literal("session.create"), name: nameField, workdir: pathField, + /** + * Session role. "conductor" requests THE per-tenant conductor session — + * the daemon chooses its name/workdir itself, creates it on first request, + * and returns the existing one afterwards (idempotent). Absent = a normal + * coding session. + * + * Validated as a bounded string, not a literal, on purpose: the frame must + * PARSE even for a role this daemon doesn't implement (a newer client, a + * future P4 worker role) — the daemon then fail-closes with a clear + * "unsupported role" error rather than the schema opaquely rejecting the + * whole create. Matches the "accept the frame, act on what you understand" + * wire contract. + */ + role: z.string().max(LIMITS.NAME_MAX).optional(), }); export const sessionListSchema = z.object({ ...base, type: z.literal("session.list") }); diff --git a/packages/protocol/src/types.ts b/packages/protocol/src/types.ts index b0c0cef..24cc13f 100644 --- a/packages/protocol/src/types.ts +++ b/packages/protocol/src/types.ts @@ -120,6 +120,13 @@ export interface SessionInfo { createdBy: string; createdAt: string; attachedClients: number; + /** + * Session role. "conductor" marks the per-tenant conductor session (the + * fleet supervisor — one per account/project). Absent = normal session. + */ + role?: "conductor"; + /** Id of the provider backing this session (e.g. "claude", "gemini"). */ + providerId?: string; /** Current execution mode (default "interactive"). */ mode?: SessionMode; /** Remaining turns budget for autonomous mode (undefined = unbounded, 0 = exhausted). */ @@ -690,6 +697,15 @@ export interface SessionCreateMsg extends BaseClientMsg { type: "session.create"; name: string; workdir: string; + /** + * Session role. "conductor" requests THE per-tenant conductor session — + * the daemon chooses its name/workdir itself, creates it on first request, + * and returns the existing one afterwards (idempotent). Absent = a normal + * coding session. Typed as an open string (not the `"conductor"` literal) + * so a future role from a newer client still type-checks on the wire; the + * daemon rejects roles it doesn't implement. + */ + role?: string; } /** diff --git a/src/cli.ts b/src/cli.ts index 56228c1..bf03489 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -116,6 +116,11 @@ const CODEOID_LOGIN_SCOPES = [ "session:interrupt", "session:approve", "session:destroy", + // Conductor scopes — the owner delegates these to its conductor identity + // (owner → conductor RFC 8693 exchange). Without them in the owner's token + // the delegation's scope intersection is empty and the conductor can't act. + "session:read", + "session:dispatch", "fs:read", "tools:read", "tools:write", @@ -315,7 +320,9 @@ program program .command("attach ") - .description("Attach to a session (interactive streaming)") + .description( + "Attach to a session by id or name (interactive streaming). Use 'conductor' to open the fleet supervisor (created on first use).", + ) .action(async (session: string) => { const config = loadConfig(); const client = new TerminalClient(config); diff --git a/src/config.ts b/src/config.ts index b16c049..e2c5907 100644 --- a/src/config.ts +++ b/src/config.ts @@ -302,6 +302,26 @@ const AgentIdentitySchema = z }) .default({ accountId: "personal", projectId: "dev" }); +/** + * Conductor session — the per-tenant fleet supervisor (docs/conductor-design.md). + * `provider` selects which backend drives it (any registered provider id, so an + * open-weight backend can run the conductor once its provider exists); `model` + * overrides the provider's default. Note: fleet MCP tools currently surface + * only under the "claude" provider (the one provider with MCP support) — a + * conductor on another provider still chats but cannot see the fleet yet. + */ +const ConductorSchema = z + .object({ + enabled: z.boolean().default(true), + /** Display name of the conductor session (also what `codeoid attach conductor` resolves). */ + name: z.string().default("conductor"), + /** Provider id driving the conductor ("claude" | "gemini" | "openai" | future). */ + provider: z.string().default("claude"), + /** Model override for the conductor (alias or full id). Empty = provider default. */ + model: z.string().optional(), + }) + .default({ enabled: true, name: "conductor", provider: "claude" }); + const AuthSchemaFields = z .object({ issuer: z.string().optional(), @@ -334,6 +354,7 @@ const RootSchema = z.object({ telemetry: TelemetrySchema, autoRotate: AutoRotateSchema, session: SessionSchema, + conductor: ConductorSchema, }); type ParsedConfig = z.infer; @@ -413,6 +434,17 @@ export interface CodeoidConfig { /** Per-call timeout (ms) for external MCP servers, surfaced as the SDK's per-server `timeout`. 0 = use SDK default. Defaults to 120000 when omitted. */ mcpToolTimeoutMs?: number; }; + /** + * The per-tenant conductor session (fleet supervisor). Optional in the + * type so hand-built test configs stay minimal; loadConfig always + * populates it (schema defaults). Absent = enabled with defaults. + */ + conductor?: { + enabled: boolean; + name: string; + provider: string; + model?: string; + }; } // ── Env-var override map ───────────────────────────────────────────────── @@ -636,6 +668,7 @@ export function loadConfig(opts: LoadOptions = {}): CodeoidConfig { telemetry: { osc8: osc8Mode }, autoRotate: parsed.autoRotate, session: parsed.session, + conductor: parsed.conductor, }; } diff --git a/src/daemon/fleet.ts b/src/daemon/fleet.ts new file mode 100644 index 0000000..96ca223 --- /dev/null +++ b/src/daemon/fleet.ts @@ -0,0 +1,299 @@ +/** + * Fleet MCP server — the conductor's read-only view of the session fleet + * (design §3, build plan P3). Injected ONLY into the `role:"conductor"` + * session; normal sessions never see these tools. + * + * Read-only by construction: every tool observes (list / find / summarize / + * recall / map) and none can act in a target session — dispatch (send-class) + * arrives in P4 behind the confirm flow. Summaries come from the memory + * engine's episode digests, never raw scrollback, so the conductor's context + * stays O(active threads) (design §2). + * + * Provider-agnostic core: this module only builds tool handlers + an SDK MCP + * server object. Which provider surfaces MCP tools is the provider's concern + * (only the Claude provider supports MCP today). + */ + +import { execFile } from "node:child_process"; +import { promisify } from "node:util"; +import { z } from "zod"; +import { + createSdkMcpServer, + tool, + type McpSdkServerConfigWithInstance, +} from "@anthropic-ai/claude-agent-sdk"; +import type { MemoryEngine } from "./memory/index.js"; + +const execFileAsync = promisify(execFile); + +/** What the conductor is allowed to know about a session — metadata only. */ +export interface FleetSessionView { + id: string; + name: string; + workdir: string; + workspaceId: string; + status: string; + role?: "conductor"; + providerId: string; + model?: string; + attachedClients: number; + createdAt: string; +} + +export interface FleetDeps { + /** Tenant-scoped session snapshot (the manager closes over auth). */ + listSessions(): FleetSessionView[]; + /** Memory engine — powers fleet_find / fleet_summary / fleet_recall. */ + memory?: MemoryEngine; + /** Audit sink — every fleet tool call lands in the audit log under the conductor's identity. */ + audit(action: string, detail: string): void; + /** The conductor's own session id (excluded from find results). */ + conductorSessionId(): string; +} + +/** Tool names as they appear to the provider allowlist (server key `codeoid_fleet`). */ +export const FLEET_TOOL_NAMES = [ + "fleet_list", + "fleet_find", + "fleet_summary", + "fleet_recall", + "machine_map", +] as const; + +/** + * System-prompt append for the conductor session. Kept beside the fleet + * tools because they define the conductor's whole contract. + */ +export const CONDUCTOR_SYSTEM_PROMPT_APPEND = `You are the codeoid CONDUCTOR — the owner's fleet supervisor, not a coding agent. + +Your job is to ROUTE and OBSERVE, never to do the work yourself: +- Use fleet_list / machine_map to see what sessions exist and where. +- Use fleet_find to resolve "which session was X?" questions across all workspaces. +- Use fleet_summary for a compressed digest of one session; use fleet_recall to pull specific past context. +- You have NO tools to edit files, run commands, or act inside any session. In this phase you are read-only over the fleet; directing sessions arrives later. +- Never dump raw transcripts or long tool output into your replies. Answer with compact, source-attributed summaries (session name + what/when). +- When the owner references past work ("the authz fix", "that session about X"), resolve it with fleet_find first and confirm which session you mean.`; + +const MAX_LIMIT = 20; +/** Per-repo git probe budget — machine_map must never hang the turn. */ +const GIT_PROBE_TIMEOUT_MS = 2_000; + +function ago(iso: string | number): string { + const t = typeof iso === "number" ? iso : Date.parse(iso); + if (!Number.isFinite(t)) return "unknown"; + const mins = Math.max(0, Math.round((Date.now() - t) / 60_000)); + if (mins < 60) return `${mins}m ago`; + const hours = Math.round(mins / 60); + if (hours < 48) return `${hours}h ago`; + return `${Math.round(hours / 24)}d ago`; +} + +function sessionLine(s: FleetSessionView): string { + const marker = s.role === "conductor" ? " [conductor — you]" : ""; + const model = s.model ? ` model=${s.model}` : ""; + return `- ${s.name} (${s.id.slice(0, 8)})${marker} — ${s.status}, ${s.attachedClients} client(s), provider=${s.providerId}${model}, workdir=${s.workdir}, created ${ago(s.createdAt)}`; +} + +/** Resolve a user-supplied session reference (name or id/prefix) to a view. */ +function resolveSession( + sessions: FleetSessionView[], + ref: string, +): FleetSessionView | undefined { + return ( + sessions.find((s) => s.id === ref) ?? + sessions.find((s) => s.name === ref) ?? + sessions.find((s) => s.id.startsWith(ref)) + ); +} + +/** + * Handler implementations, exposed separately from the SDK wiring so unit + * tests can call them without an MCP transport. Each returns the tool's + * text payload. + */ +export function createFleetHandlers(deps: FleetDeps) { + return { + async fleet_list(): Promise { + const sessions = deps.listSessions(); + deps.audit("fleet.list", `sessions=${sessions.length}`); + if (sessions.length === 0) return "No sessions in the fleet."; + // Group by workdir so the fleet reads as a machine map, not a flat list. + const byWorkdir = new Map(); + for (const s of sessions) { + const group = byWorkdir.get(s.workdir) ?? []; + group.push(s); + byWorkdir.set(s.workdir, group); + } + const blocks: string[] = []; + for (const [workdir, group] of byWorkdir) { + blocks.push(`${workdir}:\n${group.map(sessionLine).join("\n")}`); + } + return `${sessions.length} session(s) across ${byWorkdir.size} workspace(s):\n\n${blocks.join("\n\n")}`; + }, + + async fleet_find(args: { query: string; limit?: number }): Promise { + deps.audit("fleet.find", `query=${args.query.slice(0, 200)}`); + if (!deps.memory) { + return "Memory is disabled on this daemon — fleet_find needs the memory engine. Use fleet_list instead."; + } + const sessions = deps.listSessions(); + const sessionNames = new Map(sessions.map((s) => [s.id, s.name])); + const conductorId = deps.conductorSessionId(); + const hits = ( + await deps.memory.searchSessions({ + query: args.query, + // workspaceId absent = cross-workspace global resolution (P1). + limit: Math.min(args.limit ?? 5, MAX_LIMIT) + 1, + sessionNames, + }) + ).filter((h) => h.sessionId !== conductorId); + if (hits.length === 0) { + return `No session matched "${args.query}". It may predate memory, or try different terms.`; + } + const lines = hits.slice(0, Math.min(args.limit ?? 5, MAX_LIMIT)).map((h, i) => { + const name = sessionNames.get(h.sessionId) ?? "(no longer running)"; + const evidence = h.snippets + .slice(0, 2) + .map((sn) => ` · [${sn.kind}] ${sn.summary}`) + .join("\n"); + return `${i + 1}. ${name} (${h.sessionId.slice(0, 8)}) — ${h.matchCount} match(es), last activity ${ago(h.lastMatchAt)}\n${evidence}`; + }); + return `Top session(s) for "${args.query}":\n${lines.join("\n")}`; + }, + + async fleet_summary(args: { session: string }): Promise { + const sessions = deps.listSessions(); + const target = resolveSession(sessions, args.session); + deps.audit("fleet.summary", `session=${args.session.slice(0, 100)} resolved=${target?.id ?? "none"}`); + if (!target) { + return `No session matches "${args.session}". Use fleet_list to see the fleet.`; + } + const head = sessionLine(target); + if (!deps.memory) return `${head}\n(no memory engine — activity digest unavailable)`; + // Compressed digest: the session's recent episode SUMMARIES (one line + // each), never raw scrollback/transcript — the never-OOC guarantee. + const episodes = deps.memory + .timeline(target.workspaceId, 60) + .filter((e) => e.sessionId === target.id) + .slice(0, 12); + if (episodes.length === 0) return `${head}\n(no recorded activity yet)`; + const lines = episodes.map( + (e) => `- [${new Date(e.createdAt).toISOString()}] ${e.kind}${e.toolName ? `/${e.toolName}` : ""}: ${e.summary}`, + ); + return `${head}\n\nRecent activity (${episodes.length} episode(s), newest first):\n${lines.join("\n")}`; + }, + + async fleet_recall(args: { query: string; limit?: number }): Promise { + deps.audit("fleet.recall", `query=${args.query.slice(0, 200)}`); + if (!deps.memory) { + return "Memory is disabled on this daemon — fleet_recall needs the memory engine."; + } + const sessionNames = new Map(deps.listSessions().map((s) => [s.id, s.name])); + const hits = await deps.memory.recallGlobal({ + query: args.query, + limit: Math.min(args.limit ?? 6, MAX_LIMIT), + }); + if (hits.length === 0) return `Nothing recalled for "${args.query}".`; + const lines = hits.map((h) => { + const e = h.episode; + const name = sessionNames.get(e.sessionId) ?? e.sessionId.slice(0, 8); + return `- [${name}] ${e.kind}${e.toolName ? `/${e.toolName}` : ""}: ${e.summary}`; + }); + return `Recalled ${hits.length} episode(s) across the fleet:\n${lines.join("\n")}`; + }, + + async machine_map(): Promise { + const sessions = deps.listSessions(); + deps.audit("fleet.machine_map", `workspaces=${new Set(sessions.map((s) => s.workdir)).size}`); + if (sessions.length === 0) return "No sessions — the machine map is empty."; + const byWorkdir = new Map(); + for (const s of sessions) { + const group = byWorkdir.get(s.workdir) ?? []; + group.push(s); + byWorkdir.set(s.workdir, group); + } + const blocks = await Promise.all( + [...byWorkdir.entries()].map(async ([workdir, group]) => { + const git = await probeGit(workdir); + const members = group + .map((s) => `${s.name} (${s.status}${s.role === "conductor" ? ", conductor" : ""})`) + .join(", "); + return `${workdir}\n git: ${git}\n sessions: ${members}`; + }), + ); + return `Machine map — ${byWorkdir.size} workspace(s):\n\n${blocks.join("\n\n")}`; + }, + }; +} + +/** Branch + dirty state for a workdir; degrades to "not a git repo" fast. */ +async function probeGit(workdir: string): Promise { + try { + const { stdout: branch } = await execFileAsync( + "git", + ["-C", workdir, "rev-parse", "--abbrev-ref", "HEAD"], + { timeout: GIT_PROBE_TIMEOUT_MS }, + ); + const { stdout: status } = await execFileAsync( + "git", + ["-C", workdir, "status", "--porcelain"], + { timeout: GIT_PROBE_TIMEOUT_MS }, + ); + const dirty = status.trim().length > 0 ? "dirty" : "clean"; + return `${branch.trim()} (${dirty})`; + } catch { + return "not a git repo"; + } +} + +export function buildFleetMcpServer(deps: FleetDeps): McpSdkServerConfigWithInstance { + const handlers = createFleetHandlers(deps); + const text = (payload: string) => ({ + content: [{ type: "text" as const, text: payload }], + }); + + return createSdkMcpServer({ + name: "codeoid-fleet", + version: "0.1.0", + tools: [ + tool( + "fleet_list", + "List every session in the fleet, grouped by workspace — names, status, provider, attached clients. Your view of what exists right now.", + {}, + async () => text(await handlers.fleet_list()), + ), + tool( + "fleet_find", + "Resolve a natural-language reference to the right session(s) across ALL workspaces — 'the authz fix', 'that session about migrations'. Returns ranked sessions with evidence snippets. Use this FIRST whenever the owner references past work.", + { + query: z.string().describe("Natural-language description of the work/session to find"), + limit: z.number().int().min(1).max(MAX_LIMIT).optional().describe("Max sessions to return (default 5)"), + }, + async ({ query, limit }) => text(await handlers.fleet_find({ query, limit })), + ), + tool( + "fleet_summary", + "Compressed digest of ONE session: metadata plus its recent activity as one-line episode summaries. Never returns raw transcript.", + { + session: z.string().describe("Session name, id, or id prefix"), + }, + async ({ session }) => text(await handlers.fleet_summary({ session })), + ), + tool( + "fleet_recall", + "Recall specific past context across the WHOLE fleet (every workspace, every session) — returns the most relevant episode summaries.", + { + query: z.string().describe("What to recall"), + limit: z.number().int().min(1).max(MAX_LIMIT).optional().describe("Max episodes (default 6)"), + }, + async ({ query, limit }) => text(await handlers.fleet_recall({ query, limit })), + ), + tool( + "machine_map", + "Map of the machine: each workspace directory with its git branch/dirty state and which sessions live there.", + {}, + async () => text(await handlers.machine_map()), + ), + ], + }); +} diff --git a/src/daemon/providers/claude/index.ts b/src/daemon/providers/claude/index.ts index c5d9685..ff08e5c 100644 --- a/src/daemon/providers/claude/index.ts +++ b/src/daemon/providers/claude/index.ts @@ -19,6 +19,7 @@ import { type SubagentStartHookInput, type SubagentStopHookInput, type McpServerConfig, + type McpSdkServerConfigWithInstance, } from "@anthropic-ai/claude-agent-sdk"; import { randomUUID } from "node:crypto"; import { readFileSync } from "node:fs"; @@ -32,6 +33,7 @@ import { type MemoryEngine, } from "../../memory/index.js"; import type { CompressionRegistry } from "../../compress/index.js"; +import { FLEET_TOOL_NAMES } from "../../fleet.js"; import { rewriteBashToolInput } from "../../compress/index.js"; import type { CodeoidConfig } from "../../../config.js"; import type { AuthContext } from "../../../protocol/types.js"; @@ -53,6 +55,8 @@ export interface ClaudeProviderInit { identityManager?: AgentIdentityManager; /** Codeoid memory engine — injected as an MCP server. */ memory?: MemoryEngine; + /** codeoid_fleet MCP server — conductor sessions only (read-only fleet view). */ + fleet?: McpSdkServerConfigWithInstance; config?: CodeoidConfig; compressionRegistry?: CompressionRegistry; /** Called once per session with the live model catalog. */ @@ -248,6 +252,9 @@ export class ClaudeProvider implements SessionProvider { }), } : {}), + // Conductor sessions only — the read-only fleet view (P3). In-process, + // so the external-server timeout doesn't apply. + ...(init.fleet ? { codeoid_fleet: init.fleet } : {}), }; const mcpServers = Object.keys(merged).length > 0 ? merged : undefined; @@ -260,9 +267,16 @@ export class ClaudeProvider implements SessionProvider { // (which holds the root ZeroID key + control-channel secrets). See // buildAgentEnv (GHSA-38vh vector 3). env: buildAgentEnv(), - allowedTools: init.memory - ? ["mcp__codeoid_memory__recall", "mcp__codeoid_memory__recall_file", "mcp__codeoid_memory__timeline"] - : [], + allowedTools: [ + ...(init.memory + ? ["mcp__codeoid_memory__recall", "mcp__codeoid_memory__recall_file", "mcp__codeoid_memory__timeline"] + : []), + // Widened for the conductor's fleet server — without these entries + // the mounted server's tools stay unreachable (design §3 gotcha). + ...(init.fleet + ? FLEET_TOOL_NAMES.map((t) => `mcp__codeoid_fleet__${t}`) + : []), + ], permissionMode: "default", includePartialMessages: true, persistSession: true, @@ -273,7 +287,10 @@ export class ClaudeProvider implements SessionProvider { process.stderr.write(`[claude-subprocess ${sessionId.slice(0, 8)}] ${data}`); }, ...(mcpServers ? { mcpServers } : {}), - ...(init.memory + // Any non-empty append (memory guidance, conductor contract) rides on + // the claude_code preset — previously gated on memory alone, which + // silently dropped non-memory appends. + ...(init.memory || opts.systemPromptAppend ? { systemPrompt: { type: "preset" as const, diff --git a/src/daemon/providers/stateless.ts b/src/daemon/providers/stateless.ts new file mode 100644 index 0000000..2037392 --- /dev/null +++ b/src/daemon/providers/stateless.ts @@ -0,0 +1,78 @@ +/** + * StatelessSessionProvider — adapts a stateless AgentProvider (Gemini, + * OpenAI: rebuilds the full history from CanonicalTurn[] on every runTurn) + * to the SessionProvider surface Session requires. + * + * The extra SessionProvider members exist for ClaudeProvider's warm backing + * session (rotation, recovery, mid-turn queueing). A stateless provider has + * no backing session to lose or rotate, so the adapter's implementations are + * honest no-ops: the "backing id" is just a display/persistence label and + * recovery/rotation cannot apply. + */ + +import type { + AgentProvider, + ModelInfo, + SessionProvider, + TurnOpts, + TurnRun, +} from "./interface.js"; + +export class StatelessSessionProvider implements SessionProvider { + onRecoveryNeeded: ((content: string) => void) | undefined; + readonly #inner: AgentProvider; + #backingSessionId: string; + #hasQueried = false; + + constructor(inner: AgentProvider, backingSessionId: string) { + this.#inner = inner; + this.#backingSessionId = backingSessionId; + } + + get id(): string { + return this.#inner.id; + } + + get displayName(): string { + return this.#inner.displayName; + } + + get backingSessionId(): string { + return this.#backingSessionId; + } + + get hasQueried(): boolean { + return this.#hasQueried; + } + + /** Stateless providers consume the message synchronously per turn — nothing queues. */ + get queuedMessages(): number { + return 0; + } + + runTurn(opts: TurnOpts): TurnRun { + this.#hasQueried = true; + return this.#inner.runTurn(opts); + } + + listModels(): Promise { + return this.#inner.listModels(); + } + + resetToNewSession(newBackingId: string): void { + // No warm context to rotate away from — just adopt the new label. + this.#backingSessionId = newBackingId; + } + + setHasQueried(value: boolean): void { + this.#hasQueried = value; + } + + async teardown(): Promise { + // Nothing runs between turns; dispose() handles final cleanup. + } + + dispose(): Promise { + return this.#inner.dispose(); + } +} diff --git a/src/daemon/server.ts b/src/daemon/server.ts index 504f66a..9e3add7 100644 --- a/src/daemon/server.ts +++ b/src/daemon/server.ts @@ -51,6 +51,13 @@ type SocketData = { clientId: string; authenticated: boolean; auth: AuthContext | null; + /** + * The verified bearer token, kept for the connection's lifetime so flows + * that need the caller as an RFC 8693 delegation SUBJECT (owner → + * conductor token exchange) can present it. In-memory only — never + * logged, never persisted, dies with the socket. + */ + rawToken?: string; authTimer?: ReturnType; drainWaiters?: Array<() => void>; /** Protocol version the client declared on its auth frame (absent = legacy client). */ @@ -440,6 +447,7 @@ export class DaemonServer { } data.authenticated = true; + data.rawToken = authMsg.token; // Record what the client declared so capability-gated behaviour // (parts-only streaming, seq resume, …) can branch per connection. data.protocolVersion = authMsg.protocolVersion; @@ -522,7 +530,9 @@ export class DaemonServer { }; try { - const response = await self.#manager.handle(msg, data.auth!, client); + const response = await self.#manager.handle(msg, data.auth!, client, { + rawToken: data.rawToken, + }); ws.send(JSON.stringify(response)); } catch (err) { ws.send(JSON.stringify({ diff --git a/src/daemon/session-manager.ts b/src/daemon/session-manager.ts index c6b1b33..9a22641 100644 --- a/src/daemon/session-manager.ts +++ b/src/daemon/session-manager.ts @@ -8,9 +8,9 @@ * - Graceful drain on shutdown */ -import { existsSync, realpathSync, statSync } from "node:fs"; +import { existsSync, mkdirSync, realpathSync, statSync } from "node:fs"; import { homedir } from "node:os"; -import { resolve, sep } from "node:path"; +import { join, resolve, sep } from "node:path"; import { Session, type AttachedClient } from "./session.js"; import type { Store } from "./store.js"; import { hasScope, SCOPES } from "../protocol/scopes.js"; @@ -33,6 +33,7 @@ import { type ImportedSessionInit, } from "./share/index.js"; import type { AgentIdentityManager } from "./agent-identity.js"; +import { buildFleetMcpServer, type FleetSessionView } from "./fleet.js"; import { type MemoryEngine, workspaceIdFromPath } from "./memory/index.js"; import type { CodeoidConfig } from "../config.js"; import type { CompressionRegistry } from "./compress/index.js"; @@ -210,6 +211,16 @@ export class SessionManager { memory: this.#memory, config: this.#config, compressionRegistry: this.#compressionRegistry, + // The conductor self-persists (design R2): its role, provider + // selection, and fleet tools all come back across a restart. + role: meta.role, + providerId: meta.providerId, + defaultModel: + meta.role === "conductor" ? this.#config?.conductor?.model : undefined, + fleet: + meta.role === "conductor" + ? this.#buildFleetServer(meta.accountId, meta.projectId) + : undefined, onModels: (providerId, m) => this._cacheModels(providerId, m), }); @@ -283,6 +294,15 @@ export class SessionManager { msg: ClientMessage, auth: AuthContext, client: AttachedClient, + opts?: { + /** + * The caller's raw bearer token, retained by the transport for flows + * that need the owner as an RFC 8693 delegation SUBJECT — today only + * conductor creation (owner → conductor token exchange). Never logged, + * never persisted. + */ + rawToken?: string; + }, ): Promise { switch (msg.type) { case "ping": @@ -291,6 +311,21 @@ export class SessionManager { // event, by noticing the pong never arrives. return { type: "response.ok", requestId: msg.id, data: { pong: true } }; case "session.create": + if (msg.role === "conductor") { + return this.#createConductor(msg, auth, opts?.rawToken); + } + if (msg.role) { + // A role this daemon doesn't implement (newer client / future + // worker role). Fail closed rather than silently downgrading to a + // normal session — a caller asking for a constrained role must not + // get an unconstrained one. + return { + type: "response.error", + requestId: msg.id, + error: `Unsupported session role: "${msg.role}"`, + code: "invalid_request", + }; + } return this.#create(msg, auth); case "session.list": return this.#list(msg, auth); @@ -925,6 +960,148 @@ export class SessionManager { }; } + /** + * Create — or return — THE conductor session for the caller's tenant + * (design §3, build plan P3). Idempotent: one conductor per + * (account, project); a second create request answers with the existing + * one so `codeoid attach conductor` works from any client without + * coordination. The daemon chooses name/workdir/provider itself (from + * config.conductor) — the request's name/workdir are ignored. + */ + async #createConductor( + msg: Extract, + auth: AuthContext, + rawToken?: string, + ): Promise { + if (!hasScope(auth.scopes as string[], SCOPES.SESSION_CREATE)) { + return { type: "response.error", requestId: msg.id, error: "Missing scope: session:create", code: "forbidden" }; + } + if (this.#config?.conductor?.enabled === false) { + return { type: "response.error", requestId: msg.id, error: "Conductor is disabled (config.conductor.enabled)", code: "invalid_request" }; + } + + const existing = this.#conductorFor(auth.accountId, auth.projectId); + if (existing) { + return { type: "response.ok", requestId: msg.id, data: existing.toInfo() }; + } + + const rateCheck = this.#rateLimiter.check(auth.sub); + if (!rateCheck.allowed) { + return { type: "response.error", requestId: msg.id, error: rateCheck.reason, code: "rate_limited" }; + } + + // Durable identity (P2): reuse-or-register the conductor's ZeroID + // identity, then mint its working token by OWNER delegation — the + // caller's own bearer token is the RFC 8693 subject. Best-effort like + // the rest of the identity layer: the conductor still runs without it. + if (this.#identityManager) { + const identity = await this.#identityManager.registerConductor(auth.sub); + if (identity && rawToken) { + const token = await this.#identityManager.mintConductorToken(rawToken); + if (!token) { + console.error( + "[codeoid] owner->conductor delegation failed — conductor runs with metadata-only attribution (is session:read/session:dispatch in your token's scopes?)", + ); + } + } + } + + // The conductor is global (cross-workspace), so it gets a dedicated, + // daemon-owned empty workdir — NOT a repo, NOT ~ (protected-ancestor), + // and crucially not a directory with a user .mcp.json to auto-load. + const workdir = join(homedir(), ".codeoid-conductor"); + mkdirSync(workdir, { recursive: true }); + + const conductorConfig = this.#config?.conductor; + const providerId = conductorConfig?.provider ?? DEFAULT_PROVIDER_ID; + if (providerId !== "claude") { + console.warn( + `[codeoid] conductor provider is "${providerId}" — MCP fleet tools are only surfaced by the claude provider today; the conductor will chat but cannot see the fleet`, + ); + } + + const session = new Session({ + name: conductorConfig?.name ?? "conductor", + workdir, + role: "conductor", + providerId, + defaultModel: conductorConfig?.model, + fleet: this.#buildFleetServer(auth.accountId, auth.projectId), + auth, + store: this.#store, + transcriptStore: this.#transcriptStore, + identityManager: this.#identityManager, + memory: this.#memory, + config: this.#config, + compressionRegistry: this.#compressionRegistry, + onModels: (providerId, m) => this._cacheModels(providerId, m), + }); + + this.#sessions.set(session.id, session); + this.#rateLimiter.recordCreation(auth.sub); + this.#store.audit( + this.#identityManager?.conductorUri ?? auth.sub, + "conductor.session.created", + session.id, + `provider=${providerId}`, + ); + + return { type: "response.ok", requestId: msg.id, data: session.toInfo() }; + } + + /** The tenant's conductor session, if one is live. */ + #conductorFor(accountId: string, projectId: string): Session | undefined { + for (const session of this.#sessions.values()) { + if ( + session.role === "conductor" && + session.accountId === accountId && + session.projectId === projectId + ) { + return session; + } + } + return undefined; + } + + /** + * Build the codeoid_fleet MCP server for a tenant's conductor. Tools close + * over the manager, so the conductor always sees the LIVE session + * population — tenant-scoped exactly like session.list. + */ + #buildFleetServer(accountId: string, projectId: string) { + return buildFleetMcpServer({ + listSessions: (): FleetSessionView[] => { + const views: FleetSessionView[] = []; + for (const s of this.#sessions.values()) { + if (s.accountId !== accountId || s.projectId !== projectId) continue; + views.push({ + id: s.id, + name: s.name, + workdir: s.workdir, + workspaceId: s.workspaceId, + status: s.status, + role: s.role, + providerId: s.providerId, + model: s.toInfo().model, + attachedClients: s.attachedClientCount, + createdAt: s.createdAt, + }); + } + return views; + }, + memory: this.#memory, + audit: (action, detail) => + this.#store.audit( + this.#identityManager?.conductorUri ?? `conductor:${accountId}/${projectId}`, + action, + undefined, + detail, + ), + conductorSessionId: () => + this.#conductorFor(accountId, projectId)?.id ?? "", + }); + } + #list( msg: Extract, auth: AuthContext, diff --git a/src/daemon/session.ts b/src/daemon/session.ts index dc63952..a1ef7fd 100644 --- a/src/daemon/session.ts +++ b/src/daemon/session.ts @@ -10,7 +10,12 @@ */ import { ClaudeProvider } from "./providers/claude/index.js"; +import { GeminiProvider } from "./providers/gemini/index.js"; +import { OpenAIProvider } from "./providers/openai/index.js"; +import { StatelessSessionProvider } from "./providers/stateless.js"; import { CanonicalHistoryAccumulator } from "./providers/canonical.js"; +import { CONDUCTOR_SYSTEM_PROMPT_APPEND } from "./fleet.js"; +import type { McpSdkServerConfigWithInstance } from "@anthropic-ai/claude-agent-sdk"; import { type ProviderEvent, type NormalizedTurnResult, type TurnRun, type ToolApprovalFn, type SessionProvider, isSubagentEvent } from "./providers/interface.js"; import { randomUUID } from "node:crypto"; import type { @@ -130,6 +135,31 @@ export interface SessionCreateOptions { * rewrites Bash commands to route through the wrapper CLI when enabled. */ compressionRegistry?: CompressionRegistry; + /** + * Session role. "conductor" = the per-tenant fleet supervisor: it gets the + * conductor system prompt and the codeoid_fleet MCP server (when `fleet` + * is provided), and shows up in SessionInfo.role for clients. + */ + role?: "conductor"; + /** + * Provider id backing this session ("claude" | "gemini" | "openai"). + * Absent = claude. Every session carries its own selection so any session + * — the conductor included — can run on a different backend (e.g. an + * open-weight provider once one is registered). + */ + providerId?: string; + /** + * Pre-built codeoid_fleet MCP server (conductor sessions only). Built by + * the SessionManager because its tools close over the manager's tenant- + * scoped session view; the Session just hands it to the provider. + */ + fleet?: McpSdkServerConfigWithInstance; + /** + * Model default that outranks config.session.defaultModel for THIS + * session (still loses to a persisted per-session choice). Used by the + * conductor's config.conductor.model override. + */ + defaultModel?: string; /** * Provider override for testing. When present, replaces ClaudeProvider so * integration tests run without the Claude Agent SDK subprocess. @@ -147,6 +177,8 @@ export class Session { */ name: string; readonly workdir: string; + /** "conductor" marks the per-tenant fleet supervisor; undefined = normal. */ + readonly role?: "conductor"; readonly createdBy: string; readonly createdAt: string; /** @@ -369,6 +401,7 @@ export class Session { this.id = opts.existingId ?? randomUUID(); this.name = opts.name; this.workdir = opts.workdir; + this.role = opts.role; this.createdBy = opts.auth.sub; this.createdAt = new Date().toISOString(); this.accountId = opts.auth.accountId; @@ -386,36 +419,21 @@ export class Session { this.#rotationCount = stats.count; this.#lastRotatedAt = stats.lastRotatedAt; - // Model selection — prefer persisted session choice, fall back to - // config default, else leave null (SDK default). Always resolve to - // full id so downstream code doesn't see aliases. + // Model selection — prefer persisted session choice, then a per-session + // default (conductor's config.conductor.model), then the config default, + // else leave null (provider default). Always resolve to full id so + // downstream code doesn't see aliases. const persistedModel = this.#store.getSessionModel(this.id); this.#model = persistedModel.model ?? - resolveModelId(opts.config?.session.defaultModel ?? "") ?? + resolveModelId(opts.defaultModel ?? opts.config?.session.defaultModel ?? "") ?? null; this.#fallbackModel = persistedModel.fallbackModel ?? resolveModelId(opts.config?.session.fallbackModel ?? "") ?? null; - this.#provider = opts._testProvider ?? new ClaudeProvider({ - sessionId: this.id, - initialBackingId: this.#store.getClaudeCodeSessionId(this.id) ?? this.id, - // Pass the tenant-scoped workspace id in rather than have the provider - // re-derive it (which would drop the tenant and desync the memory MCP - // binding from where episodes are actually stored). - workspaceId: this.#workspaceId, - store: opts.store, - identityManager: opts.identityManager, - memory: opts.memory, - config: opts.config, - compressionRegistry: opts.compressionRegistry, - // Tag model reports with the provider's own id — the arrow runs only - // after construction (models arrive async on first query), so - // this.#provider is set by then. Works unchanged for any provider. - onModels: (m) => opts.onModels?.(this.#provider.id, m), - }); + this.#provider = opts._testProvider ?? this.#createProvider(opts); // Restore any pinned files the user had on this session before. try { @@ -489,13 +507,64 @@ export class Session { lastActivityAt: this.createdAt, accountId: opts.auth.accountId, projectId: opts.auth.projectId, + role: this.role, + providerId: this.#provider.id, }); } } + /** + * Construct the backing provider for this session from `opts.providerId`. + * Every session carries its own selection (the conductor takes its from + * config.conductor.provider), so any session can run on a different + * backend. Unknown ids warn and fall back to claude rather than throw — + * resume must survive a meta written by a newer codeoid. + */ + #createProvider(opts: SessionCreateOptions): SessionProvider { + const providerId = opts.providerId ?? "claude"; + switch (providerId) { + case "gemini": + return new StatelessSessionProvider( + new GeminiProvider({ defaultModel: this.#model ?? undefined }), + this.id, + ); + case "openai": + return new StatelessSessionProvider( + new OpenAIProvider({ defaultModel: this.#model ?? undefined }), + this.id, + ); + default: + if (providerId !== "claude") { + console.error( + `[codeoid/session ${this.id}] unknown provider "${providerId}" — falling back to claude`, + ); + } + return new ClaudeProvider({ + sessionId: this.id, + initialBackingId: this.#store.getClaudeCodeSessionId(this.id) ?? this.id, + // Pass the tenant-scoped workspace id in rather than have the provider + // re-derive it (which would drop the tenant and desync the memory MCP + // binding from where episodes are actually stored). + workspaceId: this.#workspaceId, + store: opts.store, + identityManager: opts.identityManager, + memory: opts.memory, + fleet: opts.fleet, + config: opts.config, + compressionRegistry: opts.compressionRegistry, + // Tag model reports with the provider's own id — the arrow runs only + // after construction (models arrive async on first query), so + // this.#provider is set by then. Works unchanged for any provider. + onModels: (m) => opts.onModels?.(this.#provider.id, m), + }); + } + } + get status(): SessionStatus { return this.#status; } /** Id of the provider backing this session (e.g. "claude"). */ get providerId(): string { return this.#provider.id; } + /** Tenant-scoped memory workspace id (for fleet views / cross-session search). */ + get workspaceId(): string { return this.#workspaceId; } get attachedClientCount(): number { return this.#clients.size; } /** @@ -918,7 +987,7 @@ export class Session { model: this.#model ?? undefined, fallbackModel: this.#fallbackModel ?? undefined, workdir: this.workdir, - systemPromptAppend: this.#memory ? this.#buildMemoryPromptAppend() : undefined, + systemPromptAppend: this.#buildPromptAppend(), canUseTool: this.#makeCanUseToolFn(recoverySender), sender: recoverySender, }); @@ -935,7 +1004,7 @@ export class Session { model: this.#model ?? undefined, fallbackModel: this.#fallbackModel ?? undefined, workdir: this.workdir, - systemPromptAppend: this.#memory ? this.#buildMemoryPromptAppend() : undefined, + systemPromptAppend: this.#buildPromptAppend(), canUseTool: this.#makeCanUseToolFn(sender), sender, }); @@ -1130,6 +1199,8 @@ export class Session { createdBy: this.createdBy, createdAt: this.createdAt, attachedClients: this.#clients.size, + role: this.role, + providerId: this.#provider.id, mode: this.#mode, turnsRemaining: this.#turnsRemaining, pinnedFiles: [...this.#pinnedFiles], @@ -1610,6 +1681,18 @@ export class Session { * omitted on cold sessions (no episodes yet) so the append stays identical * to the pre-index version — prompt cache stays warm for first turns. */ + /** + * Compose the per-turn system-prompt append: the conductor contract (for + * role:"conductor" sessions) plus the memory recall guidance (when memory + * is enabled). Stable per session so it stays in the cached prompt prefix. + */ + #buildPromptAppend(): string | undefined { + const parts: string[] = []; + if (this.role === "conductor") parts.push(CONDUCTOR_SYSTEM_PROMPT_APPEND); + if (this.#memory) parts.push(this.#buildMemoryPromptAppend()); + return parts.length > 0 ? parts.join("\n\n") : undefined; + } + #buildMemoryPromptAppend(): string { const index = this.#indexScheduler?.get() ?? ""; if (!index) return MEMORY_SYSTEM_PROMPT_APPEND; @@ -2605,6 +2688,8 @@ export class Session { lastActivityAt: new Date().toISOString(), accountId: this.accountId, projectId: this.projectId, + role: this.role, + providerId: this.#provider.id, }).catch(() => {}); } } diff --git a/src/daemon/store.ts b/src/daemon/store.ts index c1121ed..4d8faf5 100644 --- a/src/daemon/store.ts +++ b/src/daemon/store.ts @@ -48,6 +48,11 @@ export class Store { // Model selection — persisted so /model choice survives daemon restart. this.#addColumnIfMissing("sessions", "model", "TEXT"); this.#addColumnIfMissing("sessions", "fallback_model", "TEXT"); + // Session role ("conductor") + backing provider id — persisted so the + // conductor keeps its role and every session keeps its provider across + // daemon restarts. NULL = normal session / claude (pre-upgrade rows). + this.#addColumnIfMissing("sessions", "role", "TEXT"); + this.#addColumnIfMissing("sessions", "provider", "TEXT"); this.#db.exec(` @@ -149,8 +154,8 @@ export class Store { createSession(session: SessionInfo & { accountId: string; projectId: string }): void { this.#db .prepare( - `INSERT OR REPLACE INTO sessions (id, name, workdir, status, created_by, account_id, project_id, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + `INSERT OR REPLACE INTO sessions (id, name, workdir, status, created_by, account_id, project_id, created_at, role, provider) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ) .run( session.id, @@ -161,6 +166,8 @@ export class Store { session.accountId, session.projectId, session.createdAt, + session.role ?? null, + session.providerId ?? null, ); } diff --git a/src/daemon/transcript.ts b/src/daemon/transcript.ts index ec66892..aeba535 100644 --- a/src/daemon/transcript.ts +++ b/src/daemon/transcript.ts @@ -81,6 +81,10 @@ export interface TranscriptMeta { lastActivityAt: string; accountId: string; projectId: string; + /** "conductor" for the per-tenant conductor session; absent = normal. */ + role?: "conductor"; + /** Provider id backing the session; absent = claude (pre-upgrade metas). */ + providerId?: string; } /** Types we persist. Skip ephemeral events like heartbeats. */ diff --git a/src/terminal/client.ts b/src/terminal/client.ts index faa33ee..0614096 100644 --- a/src/terminal/client.ts +++ b/src/terminal/client.ts @@ -375,13 +375,34 @@ export class TerminalClient { // ── Internals ───────────────────────────────────────────────────────── async #resolveSession(nameOrId: string): Promise { + // `attach conductor` create-or-gets THE conductor session (idempotent on + // the daemon), so you can reach it from any client without knowing its id + // or creating it first. Match by role too — the conductor's display name + // is configurable. + if (nameOrId === "conductor") { + const created = await this.#request({ + type: "session.create", + id: randomUUID(), + name: "conductor", + workdir: ".", + role: "conductor", + }); + if (created.type === "response.ok") { + return (created.data as SessionInfo).id; + } + this.#printError(created); + return null; + } + if (nameOrId.includes("-") && nameOrId.length > 30) { return nameOrId; } const resp = await this.#request({ type: "session.list", id: randomUUID() }); if (resp.type === "session.list.result") { - const match = resp.sessions.find((s) => s.name === nameOrId); + const match = + resp.sessions.find((s) => s.name === nameOrId) ?? + resp.sessions.find((s) => s.role === nameOrId); if (match) return match.id; console.error(`Session not found: ${nameOrId}`); } else { diff --git a/src/tests/conductor-session.test.ts b/src/tests/conductor-session.test.ts new file mode 100644 index 0000000..b7aa698 --- /dev/null +++ b/src/tests/conductor-session.test.ts @@ -0,0 +1,178 @@ +/** + * Conductor session lifecycle (P3) — driven through SessionManager.handle so + * the singleton, role/provider persistence, tenancy, and resume paths are + * exercised end-to-end. No SDK subprocess runs: a Session only spawns its + * provider's backend on the first turn, and these tests never send one. + */ + +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Store } from "../daemon/store.js"; +import { TranscriptStore } from "../daemon/transcript.js"; +import { SessionManager } from "../daemon/session-manager.js"; +import type { CodeoidConfig } from "../config.js"; +import type { AuthContext, SessionInfo } from "../protocol/types.js"; +import { ALL_SCOPES } from "../protocol/scopes.js"; + +function auth(tenant: string): AuthContext { + return { + sub: `user:${tenant}`, + scopes: [...ALL_SCOPES] as AuthContext["scopes"], + delegationDepth: 0, + accountId: `acc-${tenant}`, + projectId: `proj-${tenant}`, + }; +} + +const AUTH_A = auth("a"); +const client = (a: AuthContext) => ({ id: `client-${a.accountId}`, auth: a, send: () => {} }); + +/** Minimal config with a conductor block — only the fields the manager reads. */ +function mkConfig(conductor?: Partial): CodeoidConfig { + return { + daemonUrl: "ws://127.0.0.1:7400", + dbPath: "/tmp/codeoid.db", + transcriptDir: "/tmp/transcripts", + auth: { baseUrl: "http://localhost:8899" }, + zeroidUrl: "http://localhost:8899", + workspaceIndex: { enabled: false, episodeThreshold: 5, timeThresholdMs: 60_000, debounceMs: 15_000 }, + compress: { enabled: false, excludeCommands: [], excludePatterns: [], compressPipes: false, minBytes: 1024 }, + labeling: {}, + telemetry: { osc8: "auto" }, + autoRotate: { enabled: false, warnPct: 0.6, rotatePct: 0.8, hardRotatePct: 0.9, minTurnsBeforeRotate: 3, strategy: "task-anchor" }, + session: {}, + conductor: { enabled: true, name: "conductor", provider: "claude", ...conductor }, + }; +} + +let tmp: string; +let store: Store; +let transcript: TranscriptStore; + +function newManager(config?: CodeoidConfig): SessionManager { + return new SessionManager(store, transcript, undefined, undefined, undefined, { config }); +} + +async function createConductor(mgr: SessionManager, a = AUTH_A): Promise { + const resp = await mgr.handle( + { type: "session.create", id: "req", name: "ignored", workdir: ".", role: "conductor" }, + a, + client(a), + ); + expect(resp.type).toBe("response.ok"); + return (resp as { data: SessionInfo }).data; +} + +beforeEach(() => { + tmp = mkdtempSync(join(tmpdir(), "codeoid-conductor-sess-")); + store = new Store(join(tmp, "codeoid.db")); + transcript = new TranscriptStore(join(tmp, "transcripts")); +}); + +afterEach(async () => { + try { await transcript.flush(); } catch {} + try { store.close(); } catch {} + try { rmSync(tmp, { recursive: true, force: true }); } catch {} +}); + +describe("conductor session", () => { + test("creating role:conductor yields a conductor session on the default provider", async () => { + const info = await createConductor(newManager(mkConfig())); + expect(info.role).toBe("conductor"); + expect(info.providerId).toBe("claude"); + // Daemon picks name/workdir — the request's name ("ignored") is dropped. + expect(info.name).toBe("conductor"); + }); + + test("a second conductor request returns the SAME session (idempotent singleton)", async () => { + const mgr = newManager(mkConfig()); + const first = await createConductor(mgr); + const second = await createConductor(mgr); + expect(second.id).toBe(first.id); + }); + + test("the conductor is discoverable via session.list with its role", async () => { + const mgr = newManager(mkConfig()); + const created = await createConductor(mgr); + const list = await mgr.handle({ type: "session.list", id: "req" }, AUTH_A, client(AUTH_A)); + expect(list.type).toBe("session.list.result"); + const found = (list as { sessions: SessionInfo[] }).sessions.find((s) => s.id === created.id); + expect(found?.role).toBe("conductor"); + }); + + test("config.conductor.provider selects the backend (provider-agnostic)", async () => { + const info = await createConductor(newManager(mkConfig({ provider: "gemini" }))); + expect(info.providerId).toBe("gemini"); + expect(info.role).toBe("conductor"); + }); + + test("a disabled conductor is refused", async () => { + const resp = await newManager(mkConfig({ enabled: false })).handle( + { type: "session.create", id: "req", name: "x", workdir: ".", role: "conductor" }, + AUTH_A, + client(AUTH_A), + ); + expect(resp.type).toBe("response.error"); + expect((resp as { error: string }).error).toContain("disabled"); + }); + + test("each tenant gets its own conductor", async () => { + const mgr = newManager(mkConfig()); + const a = await createConductor(mgr, auth("a")); + const b = await createConductor(mgr, auth("b")); + expect(a.id).not.toBe(b.id); + // Tenant A's list must not see tenant B's conductor. + const listA = await mgr.handle({ type: "session.list", id: "req" }, auth("a"), client(auth("a"))); + const ids = (listA as { sessions: SessionInfo[] }).sessions.map((s) => s.id); + expect(ids).toContain(a.id); + expect(ids).not.toContain(b.id); + }); + + test("an unknown role fails closed — never a silent downgrade to a normal session", async () => { + const mgr = newManager(mkConfig()); + const resp = await mgr.handle( + // A future/unimplemented role — the frame parses, the daemon refuses. + { type: "session.create", id: "req", name: "x", workdir: tmp, role: "leaf" }, + AUTH_A, + client(AUTH_A), + ); + expect(resp.type).toBe("response.error"); + expect((resp as { error: string }).error).toContain("Unsupported session role"); + // No session was created. + const list = await mgr.handle({ type: "session.list", id: "req" }, AUTH_A, client(AUTH_A)); + expect((list as { sessions: SessionInfo[] }).sessions).toHaveLength(0); + }); + + test("a normal session.create is unaffected — no role, default provider", async () => { + const resp = await newManager(mkConfig()).handle( + { type: "session.create", id: "req", name: "normal", workdir: tmp }, + AUTH_A, + client(AUTH_A), + ); + expect(resp.type).toBe("response.ok"); + const info = (resp as { data: SessionInfo }).data; + expect(info.role).toBeUndefined(); + expect(info.name).toBe("normal"); + }); + + test("the conductor self-persists: resume rebuilds it with role + provider", async () => { + const created = await createConductor(newManager(mkConfig({ provider: "gemini" }))); + await transcript.flush(); + + // Fresh manager over the same store/transcript = daemon restart. + const mgr2 = newManager(mkConfig({ provider: "gemini" })); + const resumed = await mgr2.resumeSessions(); + expect(resumed).toBeGreaterThanOrEqual(1); + + const list = await mgr2.handle({ type: "session.list", id: "req" }, AUTH_A, client(AUTH_A)); + const found = (list as { sessions: SessionInfo[] }).sessions.find((s) => s.id === created.id); + expect(found?.role).toBe("conductor"); + expect(found?.providerId).toBe("gemini"); + + // And it stays a singleton across the restart — no duplicate minted. + const again = await createConductor(mgr2); + expect(again.id).toBe(created.id); + }); +}); diff --git a/src/tests/fleet.test.ts b/src/tests/fleet.test.ts new file mode 100644 index 0000000..8191eb2 --- /dev/null +++ b/src/tests/fleet.test.ts @@ -0,0 +1,211 @@ +/** + * Fleet MCP handler tests (P3) — the conductor's read-only view of the + * fleet. Drives createFleetHandlers directly with a fake dependency set + * (plus a real StubEmbedder-backed MemoryEngine for the find/recall/summary + * paths), so behavior is exercised without an MCP transport or the SDK. + * + * The load-bearing properties: read-only surface, the conductor excludes + * itself from find, summaries are episode digests (not raw scrollback), and + * every tool audits. + */ + +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + createFleetHandlers, + FLEET_TOOL_NAMES, + type FleetDeps, + type FleetSessionView, +} from "../daemon/fleet.js"; +import { MemoryEngine, SqliteEpisodeStore } from "../daemon/memory/index.js"; +import type { Embedder } from "../daemon/memory/embedder.js"; + +class StubEmbedder implements Embedder { + readonly modelName = "stub"; + readonly dimensions = 8; + async init(): Promise {} + async embed(texts: string[]): Promise { + return texts.map((t) => { + const v = new Float32Array(this.dimensions); + for (let i = 0; i < t.length; i++) v[i % this.dimensions]! += t.charCodeAt(i) / 1000; + let norm = 0; + for (let i = 0; i < v.length; i++) norm += v[i]! * v[i]!; + norm = Math.sqrt(norm) || 1; + for (let i = 0; i < v.length; i++) v[i] = v[i]! / norm; + return v; + }); + } + async close(): Promise {} +} + +let tmp: string; +let episodeStore: SqliteEpisodeStore; +let memory: MemoryEngine; +let audits: Array<{ action: string; detail: string }>; + +/** A small fleet: two normal sessions + the conductor itself. */ +function fleet(): FleetSessionView[] { + return [ + { + id: "sess-authz-0001", + name: "authz-fix", + workdir: join(tmp, "highflame-authz"), + workspaceId: "ws-authz", + status: "idle", + providerId: "claude", + model: "claude-opus-4-8", + attachedClients: 1, + createdAt: new Date(Date.now() - 3_600_000).toISOString(), + }, + { + id: "sess-migr-0002", + name: "migration-work", + workdir: join(tmp, "highflame-admin"), + workspaceId: "ws-admin", + status: "thinking", + providerId: "gemini", + attachedClients: 0, + createdAt: new Date(Date.now() - 7_200_000).toISOString(), + }, + { + id: "sess-cond-9999", + name: "conductor", + workdir: join(tmp, ".codeoid-conductor"), + workspaceId: "ws-conductor", + status: "idle", + role: "conductor", + providerId: "claude", + attachedClients: 1, + createdAt: new Date().toISOString(), + }, + ]; +} + +function makeDeps(overrides?: Partial): FleetDeps { + return { + listSessions: () => fleet(), + memory, + audit: (action, detail) => audits.push({ action, detail }), + conductorSessionId: () => "sess-cond-9999", + ...overrides, + }; +} + +beforeEach(async () => { + tmp = mkdtempSync(join(tmpdir(), "codeoid-fleet-")); + episodeStore = new SqliteEpisodeStore(join(tmp, "memory.db")); + memory = new MemoryEngine({ store: episodeStore, embedder: new StubEmbedder() }); + await memory.init(); + audits = []; + // Distinct episodes per session so find/recall/summary have signal. + memory.ingest({ + workspaceId: "ws-authz", + sessionId: "sess-authz-0001", + kind: "user_turn", + summary: "fix the authz latest_only tenant scoping bug", + content: "the authz policy latest_only query dropped account and project scope", + filePaths: ["internal/authz/policy.go"], + tokenEstimate: 20, + createdAt: Date.now() - 3_600_000, + createdBy: "user:owner", + }); + memory.ingest({ + workspaceId: "ws-admin", + sessionId: "sess-migr-0002", + kind: "tool_call", + toolName: "Bash", + summary: "run the admin schema migration", + content: "golang-migrate up on the admin database, add quota columns", + filePaths: ["migrations/003_quota.sql"], + tokenEstimate: 20, + createdAt: Date.now() - 7_200_000, + createdBy: "user:owner", + }); +}); + +afterEach(async () => { + try { await memory.close(); } catch {} + try { rmSync(tmp, { recursive: true, force: true }); } catch {} +}); + +describe("fleet handlers — read surface", () => { + test("the fleet MCP tool set is exactly the read-only five", () => { + // A guardrail: no send-class tool leaks into P3. Dispatch arrives in P4. + expect([...FLEET_TOOL_NAMES]).toEqual([ + "fleet_list", + "fleet_find", + "fleet_summary", + "fleet_recall", + "machine_map", + ]); + }); + + test("fleet_list groups sessions by workspace and marks the conductor", async () => { + const out = await createFleetHandlers(makeDeps()).fleet_list(); + expect(out).toContain("authz-fix"); + expect(out).toContain("migration-work"); + expect(out).toContain("[conductor — you]"); + expect(out).toContain("provider=gemini"); + expect(audits.some((a) => a.action === "fleet.list")).toBe(true); + }); + + test("fleet_list on an empty fleet says so", async () => { + const out = await createFleetHandlers( + makeDeps({ listSessions: () => [] }), + ).fleet_list(); + expect(out).toBe("No sessions in the fleet."); + }); + + test("fleet_find resolves a natural-language reference and excludes the conductor", async () => { + const out = await createFleetHandlers(makeDeps()).fleet_find({ + query: "the authz latest_only fix", + }); + expect(out).toContain("authz-fix"); + expect(out).not.toContain("conductor"); + expect(audits.some((a) => a.action === "fleet.find")).toBe(true); + }); + + test("fleet_find degrades gracefully when memory is disabled", async () => { + const out = await createFleetHandlers( + makeDeps({ memory: undefined }), + ).fleet_find({ query: "anything" }); + expect(out).toContain("Memory is disabled"); + }); + + test("fleet_summary returns a compressed episode digest for one session", async () => { + const out = await createFleetHandlers(makeDeps()).fleet_summary({ + session: "authz-fix", + }); + expect(out).toContain("authz-fix"); + expect(out).toContain("latest_only"); // from the episode summary, not raw scrollback + expect(audits.some((a) => a.action === "fleet.summary")).toBe(true); + }); + + test("fleet_summary resolves by id prefix and reports unknown refs", async () => { + const handlers = createFleetHandlers(makeDeps()); + expect(await handlers.fleet_summary({ session: "sess-migr" })).toContain( + "migration-work", + ); + expect(await handlers.fleet_summary({ session: "nope" })).toContain( + "No session matches", + ); + }); + + test("fleet_recall pulls episode summaries across the whole fleet", async () => { + const out = await createFleetHandlers(makeDeps()).fleet_recall({ + query: "migration quota columns", + }); + expect(out).toContain("migration-work"); + expect(audits.some((a) => a.action === "fleet.recall")).toBe(true); + }); + + test("machine_map lists workspaces with git state (non-repo dirs → 'not a git repo')", async () => { + const out = await createFleetHandlers(makeDeps()).machine_map(); + expect(out).toContain("highflame-authz"); + expect(out).toContain("not a git repo"); // temp dirs aren't git repos + expect(out).toContain("authz-fix"); + expect(audits.some((a) => a.action === "fleet.machine_map")).toBe(true); + }); +}); diff --git a/web/src/lib/auth.ts b/web/src/lib/auth.ts index 293ef2a..2653208 100644 --- a/web/src/lib/auth.ts +++ b/web/src/lib/auth.ts @@ -43,6 +43,10 @@ export const DEFAULT_WEB_SCOPES = [ "session:interrupt", "session:approve", "session:destroy", + // Conductor scopes — delegated owner → conductor when the web UI opens the + // conductor session. Harmless on non-conductor use. + "session:read", + "session:dispatch", "fs:read", ].join(" "); From 6b72ca5b10b036617b2f2c997efb26fc31a022de Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Tue, 7 Jul 2026 10:49:27 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20address=20CodeRabbit=20review=20?= =?UTF-8?q?=E2=80=94=20conductor=20TOCTOU=20+=20reserved=20name?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Close a TOCTOU race in #createConductor: the singleton guard read #conductorFor before awaiting identity registration/token minting, so two concurrent conductor creates for one tenant could both pass and spawn two conductors. Re-check synchronously right before Session construction (the re-check → new Session → #sessions.set runs with no await between), so only one conductor is ever registered per (account, project). Test forces the race with a yielding identity stub; verified it fails without the re-check. - Reserve the conductor's configured display name from the normal session.create path — a regular session named "conductor" would shadow the singleton in session.list. Reject it and point the caller at role:"conductor". Co-Authored-By: Claude Fable 5 --- src/daemon/session-manager.ts | 30 ++++++++++++- src/tests/conductor-session.test.ts | 68 ++++++++++++++++++++++++++++- 2 files changed, 95 insertions(+), 3 deletions(-) diff --git a/src/daemon/session-manager.ts b/src/daemon/session-manager.ts index 9a22641..b093875 100644 --- a/src/daemon/session-manager.ts +++ b/src/daemon/session-manager.ts @@ -909,6 +909,11 @@ export class SessionManager { // ── Handlers ────────────────────────────────────────────────────────── + /** The configured display name of the conductor session (default "conductor"). */ + #conductorName(): string { + return this.#config?.conductor?.name ?? "conductor"; + } + #create( msg: Extract, auth: AuthContext, @@ -917,6 +922,18 @@ export class SessionManager { return { type: "response.error", requestId: msg.id, error: "Missing scope: session:create", code: "forbidden" }; } + // Reserve the conductor's display name for the singleton — a normal + // session named "conductor" would shadow it in session.list and confuse + // any name-based lookup. Point the caller at the role instead. + if (msg.name === this.#conductorName()) { + return { + type: "response.error", + requestId: msg.id, + error: `"${msg.name}" is reserved for the conductor session — create it with role:"conductor" instead`, + code: "invalid_request", + }; + } + // Rate limit check const rateCheck = this.#rateLimiter.check(auth.sub); if (!rateCheck.allowed) { @@ -1006,6 +1023,17 @@ export class SessionManager { } } + // Re-check the singleton after the awaits above: two near-simultaneous + // conductor creates for the same tenant both pass the first #conductorFor + // check (neither has a session registered yet), then both await identity + // work. Re-check now — the re-check → new Session → #sessions.set below + // runs with no `await` between, so this closes the TOCTOU window and only + // one conductor is ever registered per (account, project). + const raced = this.#conductorFor(auth.accountId, auth.projectId); + if (raced) { + return { type: "response.ok", requestId: msg.id, data: raced.toInfo() }; + } + // The conductor is global (cross-workspace), so it gets a dedicated, // daemon-owned empty workdir — NOT a repo, NOT ~ (protected-ancestor), // and crucially not a directory with a user .mcp.json to auto-load. @@ -1021,7 +1049,7 @@ export class SessionManager { } const session = new Session({ - name: conductorConfig?.name ?? "conductor", + name: this.#conductorName(), workdir, role: "conductor", providerId, diff --git a/src/tests/conductor-session.test.ts b/src/tests/conductor-session.test.ts index b7aa698..a23d90a 100644 --- a/src/tests/conductor-session.test.ts +++ b/src/tests/conductor-session.test.ts @@ -12,10 +12,39 @@ import { join } from "node:path"; import { Store } from "../daemon/store.js"; import { TranscriptStore } from "../daemon/transcript.js"; import { SessionManager } from "../daemon/session-manager.js"; +import type { AgentIdentityManager } from "../daemon/agent-identity.js"; import type { CodeoidConfig } from "../config.js"; import type { AuthContext, SessionInfo } from "../protocol/types.js"; import { ALL_SCOPES } from "../protocol/scopes.js"; +/** + * Identity manager stub whose async methods actually YIELD (await a + * microtask), so two concurrent conductor creates interleave at the await + * point — the exact window the TOCTOU re-check must close. Counts how many + * times a conductor was registered. + */ +function yieldingIdentityStub(): { mgr: AgentIdentityManager; registers: () => number } { + let registers = 0; + const stub = { + get conductorUri() { + return "wimse://test/conductor"; + }, + async registerConductor(sub: string) { + await Promise.resolve(); // force a yield so concurrent creates interleave + registers++; + return { identityId: "cond-id", wimseUri: "wimse://test/conductor", ownerSub: sub }; + }, + async mintConductorToken() { + await Promise.resolve(); + return "delegated-token"; + }, + async resumeConductor() { + return null; + }, + }; + return { mgr: stub as unknown as AgentIdentityManager, registers: () => registers }; +} + function auth(tenant: string): AuthContext { return { sub: `user:${tenant}`, @@ -51,8 +80,8 @@ let tmp: string; let store: Store; let transcript: TranscriptStore; -function newManager(config?: CodeoidConfig): SessionManager { - return new SessionManager(store, transcript, undefined, undefined, undefined, { config }); +function newManager(config?: CodeoidConfig, identity?: AgentIdentityManager): SessionManager { + return new SessionManager(store, transcript, identity, undefined, undefined, { config }); } async function createConductor(mgr: SessionManager, a = AUTH_A): Promise { @@ -157,6 +186,41 @@ describe("conductor session", () => { expect(info.name).toBe("normal"); }); + test("the conductor's display name is reserved from normal session creation", async () => { + // A normal session named "conductor" would shadow the singleton in + // session.list — refuse it and point at the role. + const resp = await newManager(mkConfig()).handle( + { type: "session.create", id: "req", name: "conductor", workdir: tmp }, + AUTH_A, + client(AUTH_A), + ); + expect(resp.type).toBe("response.error"); + expect((resp as { error: string }).error).toContain("reserved"); + }); + + test("concurrent conductor creates converge on ONE session (no TOCTOU dup)", async () => { + // The yielding identity stub forces an await between the first singleton + // check and Session construction — the exact race window. Both creates + // pass the first check; only the re-check must let one through. + const { mgr: identity, registers } = yieldingIdentityStub(); + const mgr = newManager(mkConfig(), identity); + const [a, b] = await Promise.all([ + createConductor(mgr), + createConductor(mgr), + ]); + expect(a.id).toBe(b.id); + const list = await mgr.handle({ type: "session.list", id: "req" }, AUTH_A, client(AUTH_A)); + const conductors = (list as { sessions: SessionInfo[] }).sessions.filter( + (s) => s.role === "conductor", + ); + expect(conductors).toHaveLength(1); + // The loser re-checks and returns the winner WITHOUT constructing a second + // Session — but both may have registered identity before the re-check + // (best-effort, idempotent on the ZeroID side), so we only assert the + // session singleton, which is the invariant attach relies on. + expect(registers()).toBeGreaterThanOrEqual(1); + }); + test("the conductor self-persists: resume rebuilds it with role + provider", async () => { const created = await createConductor(newManager(mkConfig({ provider: "gemini" }))); await transcript.flush();