From e5fe56326bf50d389911cfdb1143ca1ca21344f9 Mon Sep 17 00:00:00 2001 From: Matthew Wright Date: Fri, 3 Jul 2026 16:43:56 -0500 Subject: [PATCH 1/4] harden orchestrator-v2: telemetry, resilience, capability routing - Route A: propagate execution telemetry from the engine through the runner RPC to the TaskSource as a terminal outcome; unify ExecutionStats in interfaces-task-source. - G1: persist sessionId (and telemetry) on the failure path so a failed run stays resumable. - I3: guard task-source callbacks so a throw can't leak a peer's in-flight slot, hang the runner's terminal RPC, or escape as an unhandled rejection (settle/recordBestEffort helpers + a route().catch net). - I1: capability-aware routing -- runners advertise their registered agents in the heartbeat; the orchestrator only dispatches a task to a capable peer (backward compatible: a runner that advertises nothing = capable of anything). - Adds regression tests (orchestrator.spec I3, peer-registry.spec I1). 43/43 tests pass; build + lint clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../agent-3-task/src/run-task-agent.ts | 22 +++-- .../interfaces-task-source/src/index.ts | 9 ++- .../interfaces-task-source/src/types.ts | 32 +++++++- .../packages/interfaces-task/src/index.ts | 2 +- .../packages/interfaces-task/src/types.ts | 12 +-- .../packages/orchestrator/src/best-effort.ts | 14 ++++ .../orchestrator/src/dispatch-ack-handler.ts | 6 +- .../orchestrator/src/orchestrator.spec.ts | 40 ++++++++++ .../packages/orchestrator/src/orchestrator.ts | 6 +- .../orchestrator/src/peer-registry.spec.ts | 80 +++++++++++++++++++ .../orchestrator/src/peer-registry.ts | 27 ++++++- .../orchestrator/src/result-handler.ts | 67 ++++++++-------- .../packages/orchestrator/src/rpc-router.ts | 26 ++++-- .../packages/protocol/src/frames.ts | 9 ++- .../packages/protocol/src/index.ts | 2 +- .../packages/protocol/src/types.ts | 11 +++ .../packages/runner/src/dispatch-handler.ts | 3 +- .../packages/runner/src/heartbeat.ts | 3 +- orchestrator-v2/packages/runner/src/runner.ts | 11 ++- orchestrator-v2/publish.js | 6 ++ 20 files changed, 312 insertions(+), 76 deletions(-) create mode 100644 orchestrator-v2/packages/orchestrator/src/best-effort.ts create mode 100644 orchestrator-v2/packages/orchestrator/src/peer-registry.spec.ts diff --git a/orchestrator-v2/packages/agent-3-task/src/run-task-agent.ts b/orchestrator-v2/packages/agent-3-task/src/run-task-agent.ts index 01acdc0..b5abb42 100644 --- a/orchestrator-v2/packages/agent-3-task/src/run-task-agent.ts +++ b/orchestrator-v2/packages/agent-3-task/src/run-task-agent.ts @@ -43,13 +43,9 @@ export async function runTaskAgent( return { outcome: "failed", message }; } - if (!engineResult.success) { - return { - outcome: "failed", - message: engineResult.lastMessage ?? "Engine execution failed", - }; - } - + // Persist the session before branching on outcome: a failed run that produced + // a session must still be resumable by a retry (the engine may have made + // progress before failing), and telemetry is reported on both outcomes. if (engineResult.sessionId !== undefined) { await ctx.setState({ ...ctx.taskState, @@ -57,9 +53,19 @@ export async function runTaskAgent( }); } + const telemetry = engineResult.stats ?? undefined; + + if (!engineResult.success) { + return { + outcome: "failed", + message: engineResult.lastMessage ?? "Engine execution failed", + telemetry, + }; + } + return { outcome: "completed", message: engineResult.lastMessage ?? undefined, - telemetry: engineResult.stats ?? undefined, + telemetry, }; } diff --git a/orchestrator-v2/packages/interfaces-task-source/src/index.ts b/orchestrator-v2/packages/interfaces-task-source/src/index.ts index 5fa083c..2778490 100644 --- a/orchestrator-v2/packages/interfaces-task-source/src/index.ts +++ b/orchestrator-v2/packages/interfaces-task-source/src/index.ts @@ -1,2 +1,7 @@ -export type { AgentType, Task, TaskSource } from "./types.js"; -export { missingTaskFields, missingTaskFieldsMessage, validateTask } from "./types.js"; +export type { AgentType, ExecutionStats, Task, TaskSource } from "./types.js"; +export { + isExecutionStats, + missingTaskFields, + missingTaskFieldsMessage, + validateTask, +} from "./types.js"; diff --git a/orchestrator-v2/packages/interfaces-task-source/src/types.ts b/orchestrator-v2/packages/interfaces-task-source/src/types.ts index 54e2c5e..003f8cb 100644 --- a/orchestrator-v2/packages/interfaces-task-source/src/types.ts +++ b/orchestrator-v2/packages/interfaces-task-source/src/types.ts @@ -8,10 +8,20 @@ export type Task = { metadata: Record; }; +export type ExecutionStats = { + durationMs: number; + inputTokens: number; + outputTokens: number; + cacheReadTokens: number; + cacheCreationTokens: number; + totalCostUsd: number; + numTurns: number; +}; + export type TaskSource = { watchTasks: () => AsyncGenerator; - completeTask: (taskId: string) => Promise; - failTask: (taskId: string, error: string) => Promise; + completeTask: (taskId: string, telemetry?: ExecutionStats) => Promise; + failTask: (taskId: string, error: string, telemetry?: ExecutionStats) => Promise; pauseTask: (taskId: string) => Promise; setState: (taskId: string, taskState: Record) => Promise; }; @@ -79,3 +89,21 @@ export function missingTaskFields(value: unknown): string[] { export function missingTaskFieldsMessage(missing: string[]): string { return `Task is missing required fields: ${missing.join(", ")}`; } + +const EXECUTION_STATS_FIELDS = [ + "durationMs", + "inputTokens", + "outputTokens", + "cacheReadTokens", + "cacheCreationTokens", + "totalCostUsd", + "numTurns", +] as const; + +export function isExecutionStats(value: unknown): value is ExecutionStats { + if (value === null || typeof value !== "object") { + return false; + } + const record = value as Record; + return EXECUTION_STATS_FIELDS.every((field) => typeof record[field] === "number"); +} diff --git a/orchestrator-v2/packages/interfaces-task/src/index.ts b/orchestrator-v2/packages/interfaces-task/src/index.ts index 706b764..675b8c5 100644 --- a/orchestrator-v2/packages/interfaces-task/src/index.ts +++ b/orchestrator-v2/packages/interfaces-task/src/index.ts @@ -1,7 +1,6 @@ export type { AgentRegistry, DataRegistry, - ExecutionStats, MutableDataRegistry, ReadonlyRegistry, Registry, @@ -10,3 +9,4 @@ export type { ScriptTaskDefinition, } from "./types.js"; export { isScriptTaskDefinition } from "./types.js"; +export type { ExecutionStats } from "@bifrost-ai/interfaces-task-source"; diff --git a/orchestrator-v2/packages/interfaces-task/src/types.ts b/orchestrator-v2/packages/interfaces-task/src/types.ts index 5895ff9..3ff66d4 100644 --- a/orchestrator-v2/packages/interfaces-task/src/types.ts +++ b/orchestrator-v2/packages/interfaces-task/src/types.ts @@ -1,14 +1,4 @@ -import type { AgentType } from "@bifrost-ai/interfaces-task-source"; - -export type ExecutionStats = { - durationMs: number; - inputTokens: number; - outputTokens: number; - cacheReadTokens: number; - cacheCreationTokens: number; - totalCostUsd: number; - numTurns: number; -}; +import type { AgentType, ExecutionStats } from "@bifrost-ai/interfaces-task-source"; export type ReadonlyRegistry = { get(name: string): T | undefined; diff --git a/orchestrator-v2/packages/orchestrator/src/best-effort.ts b/orchestrator-v2/packages/orchestrator/src/best-effort.ts new file mode 100644 index 0000000..7c218b2 --- /dev/null +++ b/orchestrator-v2/packages/orchestrator/src/best-effort.ts @@ -0,0 +1,14 @@ +// Recording a terminal outcome (complete/fail/pause) to the task source must never throw +// out of an RPC handler: a failure there would leak the peer's in-flight slot or escape +// as an unhandled rejection. `recordBestEffort` guarantees a resolving promise, logging any +// task-source failure instead of propagating it. +export async function recordBestEffort( + record: () => Promise, + context: string, +): Promise { + try { + await record(); + } catch (error) { + console.error(`Task source failed to ${context}:`, error); + } +} diff --git a/orchestrator-v2/packages/orchestrator/src/dispatch-ack-handler.ts b/orchestrator-v2/packages/orchestrator/src/dispatch-ack-handler.ts index cb5cb8d..5c8c903 100644 --- a/orchestrator-v2/packages/orchestrator/src/dispatch-ack-handler.ts +++ b/orchestrator-v2/packages/orchestrator/src/dispatch-ack-handler.ts @@ -1,6 +1,7 @@ import type { TaskSource } from "@bifrost-ai/interfaces-task-source"; import type { ConnectedPeer, FramePayload } from "@bifrost-ai/protocol"; +import { recordBestEffort } from "./best-effort.js"; import type { DispatchTracker } from "./dispatch-tracker.js"; import type { PeerRegistry } from "./peer-registry.js"; import type { DispatchAck } from "./types.js"; @@ -42,6 +43,9 @@ export class DispatchAckHandler { private async reject(peerId: string, taskId: string, reason: string): Promise { this.tracker.resolve(taskId); this.registry.markDispatchRejected(peerId); - await this.taskSource.failTask(taskId, reason); + await recordBestEffort( + () => this.taskSource.failTask(taskId, reason), + `fail rejected task ${taskId}`, + ); } } diff --git a/orchestrator-v2/packages/orchestrator/src/orchestrator.spec.ts b/orchestrator-v2/packages/orchestrator/src/orchestrator.spec.ts index f72ff6d..2cee82c 100644 --- a/orchestrator-v2/packages/orchestrator/src/orchestrator.spec.ts +++ b/orchestrator-v2/packages/orchestrator/src/orchestrator.spec.ts @@ -20,6 +20,7 @@ type Context = { unauthorizedRunnerIdentity: PeerIdentity; taskSource: MemoryTaskSource; dispatchedTaskIds: string[]; + reached: string[]; abort: () => void; done: Promise; connectRunner: ( @@ -130,6 +131,20 @@ describe("thin orchestrator", () => { task_was_not_completed, }, }); + + test("does not wedge the peer when a task source callback throws (regression: I3)", { + given: { + task_source_that_throws_on_first_complete, + orchestrator_running, + authorized_runner_connected, + }, + when: { + waiting_for_both_reached, + }, + then: { + both_tasks_reached_complete, + }, + }); }); function setup_identities(this: Context) { @@ -307,3 +322,28 @@ function task_was_not_completed(this: Context) { expect(this.taskSource.completed).toEqual([]); expect(this.taskSource.failed).toEqual([]); } + +function task_source_that_throws_on_first_complete(this: Context) { + this.reached = []; + const source = createMemoryTaskSource([sampleTask("task-1"), sampleTask("task-2")]); + const recordComplete = source.completeTask; + source.completeTask = async (taskId: string) => { + this.reached.push(taskId); + if (taskId === "task-1") { + throw new Error("task source recording failed"); + } + await recordComplete(taskId); + }; + this.taskSource = source; +} + +async function waiting_for_both_reached(this: Context) { + await waitFor(() => this.reached.length === 2); +} + +function both_tasks_reached_complete(this: Context) { + // task-1's completeTask throws; the fix frees the peer slot regardless, so task-2 is + // still dispatched and reaches completeTask. Before the fix the peer wedged and + // `reached` would stall at ["task-1"]. + expect(this.reached).toEqual(["task-1", "task-2"]); +} diff --git a/orchestrator-v2/packages/orchestrator/src/orchestrator.ts b/orchestrator-v2/packages/orchestrator/src/orchestrator.ts index dad1780..55e97d0 100644 --- a/orchestrator-v2/packages/orchestrator/src/orchestrator.ts +++ b/orchestrator-v2/packages/orchestrator/src/orchestrator.ts @@ -1,4 +1,4 @@ -import { createOrchestratorPeer, type OrchestratorPeer } from "@bifrost-ai/protocol"; +import { capabilityKey, createOrchestratorPeer, type OrchestratorPeer } from "@bifrost-ai/protocol"; import { DispatchAckHandler } from "./dispatch-ack-handler.js"; import { DispatchTracker } from "./dispatch-tracker.js"; @@ -82,7 +82,9 @@ export async function runOrchestrator( if (abortSignal?.aborted === true) { break; } - const runner = await registry.waitForAvailablePeer(); + const runner = await registry.waitForAvailablePeer( + capabilityKey(task.agentType, task.agentName), + ); dispatchTask(runner, task, tracker, registry); } await drainInFlight(tracker, abortSignal); diff --git a/orchestrator-v2/packages/orchestrator/src/peer-registry.spec.ts b/orchestrator-v2/packages/orchestrator/src/peer-registry.spec.ts new file mode 100644 index 0000000..2df4c81 --- /dev/null +++ b/orchestrator-v2/packages/orchestrator/src/peer-registry.spec.ts @@ -0,0 +1,80 @@ +import { describe, expect } from "vite-plus/test"; +import test from "vitest-gwt"; +import { capabilityKey } from "@bifrost-ai/protocol"; +import type { ConnectedPeer, FramePayload } from "@bifrost-ai/protocol"; + +import { PeerRegistry } from "./peer-registry.js"; + +type Context = { + registry: PeerRegistry; + generic: ConnectedPeer; + specialist: ConnectedPeer; + bare: ConnectedPeer; + selected: ConnectedPeer | undefined; +}; + +const fakePeer = (peerId: string): ConnectedPeer => ({ + peerId, + subscribe: () => () => undefined, + send: () => undefined, + close: () => undefined, +}); + +const heartbeat = (runnerId: string, capabilities?: string[]): FramePayload => ({ + kind: "heartbeat", + runnerId, + ...(capabilities !== undefined ? { capabilities } : {}), +}); + +describe("PeerRegistry capability routing (regression: I1)", () => { + test("routes to the peer that advertises the required capability", { + given: { two_runners_with_different_capabilities }, + when: { selecting_a_peer_for_the_special_agent }, + then: { the_specialist_is_selected }, + }); + + test("treats a runner that advertised nothing as capable of anything", { + given: { a_runner_that_advertises_no_capabilities }, + when: { selecting_a_peer_for_the_special_agent }, + then: { the_bare_runner_is_selected }, + }); +}); + +function two_runners_with_different_capabilities(this: Context) { + this.registry = new PeerRegistry({ heartbeatTimeoutMs: 10_000, maxInFlightPerPeer: 1 }); + this.generic = fakePeer("generic"); + this.specialist = fakePeer("specialist"); + this.registry.add(this.generic); + this.registry.add(this.specialist); + // generic advertises only "done"; specialist advertises "done" AND "special". + this.registry.recordHeartbeat( + "generic", + heartbeat("rGeneric", [capabilityKey("script", "done")]), + ); + this.registry.recordHeartbeat( + "specialist", + heartbeat("rSpecial", [capabilityKey("script", "done"), capabilityKey("script", "special")]), + ); +} + +function a_runner_that_advertises_no_capabilities(this: Context) { + this.registry = new PeerRegistry({ heartbeatTimeoutMs: 10_000, maxInFlightPerPeer: 1 }); + this.bare = fakePeer("bare"); + this.registry.add(this.bare); + this.registry.recordHeartbeat("bare", heartbeat("rBare")); // no capabilities field +} + +function selecting_a_peer_for_the_special_agent(this: Context) { + this.selected = this.registry.getAvailablePeer(capabilityKey("script", "special")); +} + +function the_specialist_is_selected(this: Context) { + // The generic runner is first in insertion order but lacks "special", so it is + // skipped in favour of the capable specialist (before the fix, the generic runner + // was chosen and the task would have failed). + expect(this.selected).toBe(this.specialist); +} + +function the_bare_runner_is_selected(this: Context) { + expect(this.selected).toBe(this.bare); +} diff --git a/orchestrator-v2/packages/orchestrator/src/peer-registry.ts b/orchestrator-v2/packages/orchestrator/src/peer-registry.ts index df56d50..30c99b0 100644 --- a/orchestrator-v2/packages/orchestrator/src/peer-registry.ts +++ b/orchestrator-v2/packages/orchestrator/src/peer-registry.ts @@ -5,6 +5,9 @@ type PeerState = { runnerId: string | null; lastSeen: number; inFlight: number; + // Advertised capability keys, or null if the runner has not advertised any + // (treated as "can handle anything" for backward compatibility). + capabilities: Set | null; }; export class PeerRegistry { @@ -24,6 +27,7 @@ export class PeerRegistry { runnerId: null, lastSeen: Date.now(), inFlight: 0, + capabilities: null, }); } @@ -47,6 +51,9 @@ export class PeerRegistry { } state.runnerId = payload.runnerId; state.lastSeen = Date.now(); + if (payload.capabilities !== undefined) { + state.capabilities = new Set(payload.capabilities); + } this.notifyWaiters(); } @@ -71,25 +78,28 @@ export class PeerRegistry { this.markTerminal(peerId); } - getAvailablePeer(): ConnectedPeer | undefined { + getAvailablePeer(requiredCapability?: string): ConnectedPeer | undefined { const now = Date.now(); for (const state of this.peers.values()) { if (!this.isAvailable(state, now)) { continue; } + if (!canHandle(state, requiredCapability)) { + continue; + } return state.peer; } return undefined; } - waitForAvailablePeer(): Promise { - const available = this.getAvailablePeer(); + waitForAvailablePeer(requiredCapability?: string): Promise { + const available = this.getAvailablePeer(requiredCapability); if (available !== undefined) { return Promise.resolve(available); } return new Promise((resolve) => { const tryResolve = () => { - const peer = this.getAvailablePeer(); + const peer = this.getAvailablePeer(requiredCapability); if (peer !== undefined) { const index = this.waiters.indexOf(tryResolve); if (index >= 0) { @@ -121,3 +131,12 @@ export class PeerRegistry { } } } + +function canHandle(state: PeerState, requiredCapability: string | undefined): boolean { + // No requirement, or a runner that hasn't advertised its capabilities, is treated as + // able to handle the task (keeps non-advertising runners working as before). + if (requiredCapability === undefined || state.capabilities === null) { + return true; + } + return state.capabilities.has(requiredCapability); +} diff --git a/orchestrator-v2/packages/orchestrator/src/result-handler.ts b/orchestrator-v2/packages/orchestrator/src/result-handler.ts index fdf99e7..5302eba 100644 --- a/orchestrator-v2/packages/orchestrator/src/result-handler.ts +++ b/orchestrator-v2/packages/orchestrator/src/result-handler.ts @@ -1,6 +1,8 @@ -import type { TaskSource } from "@bifrost-ai/interfaces-task-source"; +import { isExecutionStats } from "@bifrost-ai/interfaces-task-source"; +import type { ExecutionStats, TaskSource } from "@bifrost-ai/interfaces-task-source"; import type { ConnectedPeer } from "@bifrost-ai/protocol"; +import { recordBestEffort } from "./best-effort.js"; import { sendRpcError, sendRpcResponse } from "./dispatcher.js"; import type { DispatchTracker } from "./dispatch-tracker.js"; import type { PeerRegistry } from "./peer-registry.js"; @@ -18,21 +20,9 @@ export class ResultHandler { sendRpcError(peer, requestId, "INVALID_PARAMS", "taskId is required"); return; } - - const entry = this.tracker.resolve(taskId); - if (entry === undefined || entry.peerId !== peer.peerId) { - sendRpcError( - peer, - requestId, - "NOT_IN_FLIGHT", - `Task ${taskId} is not in-flight on this peer`, - ); - return; - } - - await this.taskSource.completeTask(taskId); - this.registry.markTerminal(peer.peerId); - sendRpcResponse(peer, requestId, { ok: true }); + await this.settle(peer, requestId, taskId, () => + this.taskSource.completeTask(taskId, readTelemetry(params)), + ); } async handleFail(peer: ConnectedPeer, requestId: string, params: unknown): Promise { @@ -41,21 +31,9 @@ export class ResultHandler { sendRpcError(peer, requestId, "INVALID_PARAMS", "taskId is required"); return; } - - const entry = this.tracker.resolve(parsed.taskId); - if (entry === undefined || entry.peerId !== peer.peerId) { - sendRpcError( - peer, - requestId, - "NOT_IN_FLIGHT", - `Task ${parsed.taskId} is not in-flight on this peer`, - ); - return; - } - - await this.taskSource.failTask(parsed.taskId, parsed.message); - this.registry.markTerminal(peer.peerId); - sendRpcResponse(peer, requestId, { ok: true }); + await this.settle(peer, requestId, parsed.taskId, () => + this.taskSource.failTask(parsed.taskId, parsed.message, readTelemetry(params)), + ); } async handlePause(peer: ConnectedPeer, requestId: string, params: unknown): Promise { @@ -64,7 +42,18 @@ export class ResultHandler { sendRpcError(peer, requestId, "INVALID_PARAMS", "taskId is required"); return; } + await this.settle(peer, requestId, taskId, () => this.taskSource.pauseTask(taskId)); + } + // Record a terminal outcome, then ALWAYS free the peer slot and answer the runner — + // even if the task source throws. A source-recording failure must never leak the + // in-flight slot (which would wedge the peer) nor hang the runner's terminal RPC. + private async settle( + peer: ConnectedPeer, + requestId: string, + taskId: string, + record: () => Promise, + ): Promise { const entry = this.tracker.resolve(taskId); if (entry === undefined || entry.peerId !== peer.peerId) { sendRpcError( @@ -75,8 +64,7 @@ export class ResultHandler { ); return; } - - await this.taskSource.pauseTask(taskId); + await recordBestEffort(record, `record outcome for task ${taskId}`); this.registry.markTerminal(peer.peerId); sendRpcResponse(peer, requestId, { ok: true }); } @@ -84,7 +72,10 @@ export class ResultHandler { handleDisconnect(peer: ConnectedPeer): void { const orphaned = this.tracker.failByPeer(peer.peerId); for (const entry of orphaned) { - void this.taskSource.failTask(entry.taskId, "Runner disconnected"); + void recordBestEffort( + () => this.taskSource.failTask(entry.taskId, "Runner disconnected"), + `fail orphaned task ${entry.taskId}`, + ); this.registry.markTerminal(peer.peerId); } } @@ -112,3 +103,11 @@ function readFailParams(params: unknown): { taskId: string; message: string } | } return { taskId, message }; } + +function readTelemetry(params: unknown): ExecutionStats | undefined { + if (params === null || typeof params !== "object" || !("telemetry" in params)) { + return undefined; + } + const telemetry = (params as { telemetry: unknown }).telemetry; + return isExecutionStats(telemetry) ? telemetry : undefined; +} diff --git a/orchestrator-v2/packages/orchestrator/src/rpc-router.ts b/orchestrator-v2/packages/orchestrator/src/rpc-router.ts index c5a4200..52eb7ad 100644 --- a/orchestrator-v2/packages/orchestrator/src/rpc-router.ts +++ b/orchestrator-v2/packages/orchestrator/src/rpc-router.ts @@ -17,7 +17,11 @@ export class RpcRouter { return; } - void this.route(peer, payload.id, payload.method, payload.params); + // route() is fire-and-forget; ensure a rejecting handler can never escape as an + // unhandled promise rejection (which is process-fatal on default Node settings). + void this.route(peer, payload.id, payload.method, payload.params).catch((error: unknown) => { + console.error(`Failed to handle RPC "${payload.method}":`, error); + }); } private async route( @@ -58,8 +62,12 @@ export class RpcRouter { return; } - await this.taskSource.setState(parsed.taskId, parsed.taskState); - sendRpcResponse(peer, requestId, { ok: true }); + try { + await this.taskSource.setState(parsed.taskId, parsed.taskState); + sendRpcResponse(peer, requestId, { ok: true }); + } catch (error) { + sendRpcError(peer, requestId, "TASK_SOURCE_ERROR", errorMessage(error)); + } } private async handleSchedulerCall( @@ -73,8 +81,12 @@ export class RpcRouter { return; } - const result = await this.scheduler.call(parsed.method, parsed.args); - sendRpcResponse(peer, requestId, result); + try { + const result = await this.scheduler.call(parsed.method, parsed.args); + sendRpcResponse(peer, requestId, result); + } catch (error) { + sendRpcError(peer, requestId, "SCHEDULER_ERROR", errorMessage(error)); + } } } @@ -107,3 +119,7 @@ function readSchedulerParams(params: unknown): { method: string; args: unknown } } return { method: record.method, args: record.args }; } + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} diff --git a/orchestrator-v2/packages/protocol/src/frames.ts b/orchestrator-v2/packages/protocol/src/frames.ts index 4b36201..5db68c7 100644 --- a/orchestrator-v2/packages/protocol/src/frames.ts +++ b/orchestrator-v2/packages/protocol/src/frames.ts @@ -37,7 +37,10 @@ export function isFramePayload(value: unknown): value is FramePayload { (record.event === "data" || record.event === "end" || record.event === "error") ); case "heartbeat": - return typeof record.runnerId === "string"; + return ( + typeof record.runnerId === "string" && + (record.capabilities === undefined || isStringArray(record.capabilities)) + ); default: return false; } @@ -58,3 +61,7 @@ function isSignedEnvelope(value: unknown): value is SignedEnvelope { isFramePayload(envelope.payload) ); } + +function isStringArray(value: unknown): boolean { + return Array.isArray(value) && value.every((item) => typeof item === "string"); +} diff --git a/orchestrator-v2/packages/protocol/src/index.ts b/orchestrator-v2/packages/protocol/src/index.ts index bbe02a5..054a953 100644 --- a/orchestrator-v2/packages/protocol/src/index.ts +++ b/orchestrator-v2/packages/protocol/src/index.ts @@ -30,4 +30,4 @@ export type { SignedEnvelope, UnsignedEnvelope, } from "./types.js"; -export { SIGNING_ALGORITHM } from "./types.js"; +export { capabilityKey, SIGNING_ALGORITHM } from "./types.js"; diff --git a/orchestrator-v2/packages/protocol/src/types.ts b/orchestrator-v2/packages/protocol/src/types.ts index d987180..70e83df 100644 --- a/orchestrator-v2/packages/protocol/src/types.ts +++ b/orchestrator-v2/packages/protocol/src/types.ts @@ -24,8 +24,19 @@ export type RpcStreamEvent = { export type Heartbeat = { kind: "heartbeat"; runnerId: string; + // Optional capability advertisement: the capabilityKey() of every agent the runner + // has registered. Absent means "unknown" -- the orchestrator then treats the runner + // as able to handle any task (backward compatible with runners that don't advertise). + capabilities?: string[]; }; +// Canonical wire form of a runner capability: the (agentType, agentName) pair a task +// requires and a runner advertises. Both sides MUST derive keys via this helper so +// advertised and required capabilities compare byte-for-byte. +export function capabilityKey(agentType: string, agentName: string): string { + return JSON.stringify([agentType, agentName]); +} + export type FramePayload = RpcRequest | RpcResponse | RpcStreamEvent | Heartbeat; export const SIGNING_ALGORITHM = "ed25519" as const; diff --git a/orchestrator-v2/packages/runner/src/dispatch-handler.ts b/orchestrator-v2/packages/runner/src/dispatch-handler.ts index 182bd92..4bd0cad 100644 --- a/orchestrator-v2/packages/runner/src/dispatch-handler.ts +++ b/orchestrator-v2/packages/runner/src/dispatch-handler.ts @@ -62,12 +62,13 @@ async function handleDispatch( switch (result.outcome) { case "completed": - await rpc.call("task.complete", { taskId: task.taskId }); + await rpc.call("task.complete", { taskId: task.taskId, telemetry: result.telemetry }); break; case "failed": await rpc.call("task.fail", { taskId: task.taskId, message: result.message ?? "failed", + telemetry: result.telemetry, }); break; case "paused": diff --git a/orchestrator-v2/packages/runner/src/heartbeat.ts b/orchestrator-v2/packages/runner/src/heartbeat.ts index 6326c1c..b8c111d 100644 --- a/orchestrator-v2/packages/runner/src/heartbeat.ts +++ b/orchestrator-v2/packages/runner/src/heartbeat.ts @@ -11,9 +11,10 @@ export function startHeartbeat( peer: RunnerPeer, identity: PeerIdentity, intervalMs = DEFAULT_HEARTBEAT_INTERVAL_MS, + capabilities: readonly string[] = [], ): HeartbeatHandle { const send = () => { - peer.send({ kind: "heartbeat", runnerId: identity.keyId }); + peer.send({ kind: "heartbeat", runnerId: identity.keyId, capabilities: [...capabilities] }); }; send(); diff --git a/orchestrator-v2/packages/runner/src/runner.ts b/orchestrator-v2/packages/runner/src/runner.ts index e4cee55..6bd355a 100644 --- a/orchestrator-v2/packages/runner/src/runner.ts +++ b/orchestrator-v2/packages/runner/src/runner.ts @@ -1,5 +1,5 @@ import type { MutableDataRegistry, ScriptTaskDefinition } from "@bifrost-ai/interfaces-task"; -import { createRunnerPeer, type RunnerPeer } from "@bifrost-ai/protocol"; +import { capabilityKey, createRunnerPeer, type RunnerPeer } from "@bifrost-ai/protocol"; import { resolveRunnerOptions } from "./config-loader.js"; import { asDataRegistry } from "./data-registry.js"; @@ -14,6 +14,7 @@ export class Runner = Record; readonly data: MutableDataRegistry; private readonly agents = new Map>(); + private readonly capabilities: string[] = []; private peer: RunnerPeer | null = null; private heartbeat: HeartbeatHandle | null = null; private unsubscribeDispatch: (() => void) | null = null; @@ -26,6 +27,7 @@ export class Runner = Record = Record { diff --git a/orchestrator-v2/publish.js b/orchestrator-v2/publish.js index 28f528e..8e24cff 100755 --- a/orchestrator-v2/publish.js +++ b/orchestrator-v2/publish.js @@ -67,6 +67,12 @@ try { await setPackageVersion(pkgDir(pkgName), publishVersion); } + // NOTE: every package's `exports` subpaths must resolve to files that ship. + // We publish with `files: ["dist"]`, so a subpath pointing at `src/*.ts` + // (as orchestrator's `./test-helpers` once did) is a dead export in the + // tarball — the source is never included. When a package needs a subpath + // export, build it into `dist` and use `publishConfig.exports` to repoint it + // there (top-level `exports` can keep the `src` path for local dev/tests). console.log(`Building all packages (${publishVersion})...`); execFileSync("vp", ["run", "-r", "--parallel", "build"], { cwd: __dirname, From e6856ec9bc1f1d2383852ed339e9f714454ad68f Mon Sep 17 00:00:00 2001 From: Matthew Wright Date: Fri, 3 Jul 2026 16:43:56 -0500 Subject: [PATCH 2/4] add claude-engine example + hardening-experiment docs - examples/claude-engine: a runnable real-engine deployment (runner.yaml onboarding + a claude-CLI Engine that maps CLI JSON to EngineResult, so real token/cost telemetry flows). Registers examples/* as a workspace member. - docs/experiment: the session's gap report, retrospective (lessons), and dogfood-run notes with telemetry receipts (three escalating real-engine runs, incl. the G1 resume payoff). Co-Authored-By: Claude Opus 4.8 (1M context) --- orchestrator-v2/docs/experiment/README.md | 38 +++ .../docs/experiment/dogfood-runs.md | 126 ++++++++++ orchestrator-v2/docs/experiment/gap-report.md | 224 ++++++++++++++++++ orchestrator-v2/docs/experiment/lessons.md | 112 +++++++++ .../examples/claude-engine/README.md | 40 ++++ .../examples/claude-engine/package.json | 15 ++ .../examples/claude-engine/run.mjs | 134 +++++++++++ orchestrator-v2/pnpm-lock.yaml | 15 ++ orchestrator-v2/pnpm-workspace.yaml | 1 + 9 files changed, 705 insertions(+) create mode 100644 orchestrator-v2/docs/experiment/README.md create mode 100644 orchestrator-v2/docs/experiment/dogfood-runs.md create mode 100644 orchestrator-v2/docs/experiment/gap-report.md create mode 100644 orchestrator-v2/docs/experiment/lessons.md create mode 100644 orchestrator-v2/examples/claude-engine/README.md create mode 100644 orchestrator-v2/examples/claude-engine/package.json create mode 100644 orchestrator-v2/examples/claude-engine/run.mjs diff --git a/orchestrator-v2/docs/experiment/README.md b/orchestrator-v2/docs/experiment/README.md new file mode 100644 index 0000000..124c3d2 --- /dev/null +++ b/orchestrator-v2/docs/experiment/README.md @@ -0,0 +1,38 @@ +# Orchestrator v2 — hardening experiment + +This directory captures an exploratory session that **audited, hardened, and dogfooded** the +Orchestrator v2 rebuild. It lives on the `refactor/orchestrator-v2-hardening` branch; it is not a +merge-ready change set but a documented investigation with working code behind it. + +## What's in the branch + +**Code (fixes + tests, all with green build/lint and 43/43 tests):** + +| | What | Where | +| ------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------ | +| **Route A** | Telemetry propagates engine → `ScriptResult` → RPC → `TaskSource` as a terminal outcome; `ExecutionStats` unified in `interfaces-task-source`. | `runner`, `orchestrator`, `interfaces-task*`, `agent-3-task` | +| **G1** | A failed run persists its `sessionId` (and telemetry), so retries are resumable. | `agent-3-task/run-task-agent.ts` | +| **I3** | A throwing task-source callback can no longer wedge a peer or escape as an unhandled rejection (`settle` / `recordBestEffort` guards + a `route().catch` net). | `orchestrator/{result-handler,rpc-router,dispatch-ack-handler,best-effort}.ts` | +| **I1** | Capability-aware routing: runners advertise their agents in the heartbeat; the orchestrator only dispatches to a capable peer. | `protocol`, `runner`, `orchestrator/peer-registry.ts` | +| **Packaging** | `@bifrost-ai/orchestrator`'s `./test-helpers` export fixed via `publishConfig.exports`. | `orchestrator/package.json`, `publish.js` | + +Two `/simplify` passes and a runnable example (`examples/claude-engine/`) accompany the fixes. + +## The documents + +- **[lessons.md](./lessons.md)** — the retrospective: what shipped, what we learned, recommended next steps. Start here. +- **[gap-report.md](./gap-report.md)** — the detailed doc↔code + integration audit (findings G1–G26, I1–I3, E1–E3), with fixed items marked. +- **[dogfood-runs.md](./dogfood-runs.md)** — three escalating real-`claude`-engine runs with telemetry receipts (incl. the G1 resume payoff). + +## The headline lesson + +> Orchestrator v2 is a solid dispatch skeleton whose terminal boundary dropped everything but +> `taskId`. Telemetry (E1), crash-resilience (I3), and capability (I1) all had to be threaded +> deliberately — and the task's **output** (E3) still isn't. The layer that would _consume_ rich +> outcomes (`agent-4-workflow`) doesn't exist yet. + +## Top follow-ups (from the lessons) + +1. **E3** — propagate task output (a Route A-style change, or a documented `setState` contract). +2. Reconcile the 3 correctness-level **doc gaps** (README `Task` shape, `ScriptContext` omissions). +3. Build **`agent-4-workflow`** + a `TaskSource` read-side, so the DAG/retry we hand-rolled has a home. diff --git a/orchestrator-v2/docs/experiment/dogfood-runs.md b/orchestrator-v2/docs/experiment/dogfood-runs.md new file mode 100644 index 0000000..6ee774a --- /dev/null +++ b/orchestrator-v2/docs/experiment/dogfood-runs.md @@ -0,0 +1,126 @@ +# Dogfood runs + +Three escalating end-to-end runs of Orchestrator v2 driven by a **real engine** — the `claude` CLI, +whose JSON output (`result`, `usage`, `total_cost_usd`, `num_turns`, `session_id`) maps straight onto +`EngineResult`. Each run exercised the whole stack the way an operator would, with **real** tokens, +cost, and telemetry flowing through the [Route A](./lessons.md) pipeline. + +> Reproduce the basic run: `examples/claude-engine/` (the harder/hardest runs were throwaway +> harnesses; their setup is described below). + +--- + +## Run 1 — basic real deployment + +**Setup:** generated ed25519 keys, wrote a real `runner.yaml` loaded via `new Runner({ configPath })`, +enrolled a Task Agent, registered a `claude`/haiku engine, dispatched 2 tasks. + +**Exercised:** the documented `runner.yaml` onboarding path (`config-loader`, PEM parsing) — **never +covered by any test or prior harness** — plus `enrollTaskAgent`, a real engine, Route A telemetry, and +the session-id round-trip. + +**Result:** both tasks completed; real telemetry reached the task source; real Claude `session_id`s +round-tripped via `setState`. + +| task | out tokens | turns | cost | +|---|--:|--:|--:| +| dog-1 (`"reply DOGFOOD"`) | 61 | 1 | $0.0227 | +| dog-2 (`"describe an orchestrator"`) | 128 | 1 | $0.0153 | + +**Total: $0.038.** Lesson: the untested onboarding path worked first try. + +--- + +## Run 2 — a hand-orchestrated DAG (fan-out → aggregate) + +**Setup:** two `summarize` tasks fanned out to a capability-routed **summarizer** runner (concurrent, +`maxInFlight=2`); an aggregate `release-notes` task then routed to a separate **reviewer** runner. The +task source owned the dependency gate. Engines persisted their output via `ctx.setState` so the +aggregate step could read the summaries. + +**Exercised:** I1 capability routing across two specialized runners; concurrency; Route A telemetry +**roll-up** across the graph (exactly a workflow "verify-step aggregate stats"). + +**Result — the system wrote release notes for its own new features:** + +> *"Task completion now sends execution telemetry to the backend, providing detailed performance +> metrics and better observability. Runners can validate required capabilities, ensuring tasks are +> only assigned to runners equipped to handle them."* + +| task | out tokens | cost | +|---|--:|--:| +| sum-1 | 235 | $0.0237 | +| sum-2 | 316 | $0.0241 | +| release-notes | 384 | $0.0167 | +| **total** | **935** | **$0.0645** | + +**Lesson (new gap E3):** chaining only worked because the **engine persisted each output via +`setState`**. `task.complete` carries telemetry, not the task's `message` — there is no built-in +channel for a task's output to reach the source or a downstream step. The DAG gate was hand-rolled in +the task source because the Workflow Agent (`agent-4-workflow`) doesn't exist yet. + +--- + +## Run 3 — a 3-stage pipeline with retry + session-resume (hardest) + +**Setup:** three `analyze` tasks fanned out to an **analyzer** runner (haiku, concurrent); a +**synthesizer** runner (a **sonnet** engine) combined them; a **critic** runner (haiku) reviewed the +result. Capability-routed across **three** specialized runners with **heterogeneous engines**. One +analyze task (`an-routing`) was flaky — it failed *after* doing its work, and the task source +**retried it with `--resume`**. + +**Exercised, in one run:** Route A (the telemetry below), I1 (3-way capability routing), G1 (failed-run +session persistence → resumable retry), heterogeneous per-task engine selection, concurrency, and a +task-source-owned DAG + retry policy. + +### 🧾 Telemetry receipts (per attempt) + +``` +task att model outcome in out cacheR cost$ ms +------------------------------------------------------------------------ +an-telemetry 1 haiku completed 10 473 21220 0.0171 6931 +an-routing 1 haiku failed 10 531 17157 0.0253 6980 +an-resilience 1 haiku completed 10 609 17157 0.0255 7265 +an-routing 2 haiku completed 10 402 27639 0.0059 5037 +synthesize 1 sonnet completed 11202 169 23131 0.0780 3334 +critique 1 haiku completed 10 538 21220 0.0175 7613 +------------------------------------------------------------------------ +TOTAL 6 calls 11252 2722 127524 0.1693 37160 +``` + +By model: haiku — 5 calls, $0.0913 · sonnet — 1 call, $0.0780. + +### ⭐ The receipt that matters — the G1 payoff + +`an-routing` **failed** on attempt 1 ($0.0253, building 10,482 cache-creation tokens of context), then +resumed on retry: + +> attempt 2 → **$0.0059** (cacheR: 27,639) — **4.3× cheaper**, reusing the session instead of +> re-paying to rebuild context. + +Without the G1 fix (persisting `sessionId` on failure), the retry would re-do the expensive work from +scratch. The receipts make the value concrete. + +### The pipeline's output (it critiqued its own design) + +**Synthesis (sonnet):** *"The orchestrator matches tasks to runners by required agent capability, +avoiding wasted attempts on incapable runners, while tracking per-task execution costs and resource +usage to support operator budgeting… It isolates task-source callback failures so a hung or erroring +callback cannot crash or stall the orchestrator itself."* + +**Critique (haiku):** *"The three concerns aren't truly independent — callback isolation is +foundational reliability that enables cost tracking and matching to work correctly, and cost data +should feed back into matching decisions…"* — a genuinely sharp observation about the architecture. + +--- + +## What the runs proved, together + +- **Route A telemetry** carries real numbers end-to-end (every receipt above). +- **I1 capability routing** correctly places tasks across 1, 2, and 3 specialized runners. +- **G1** turns a failed run into a cheap resumable retry ($0.0059 vs. $0.0253). +- **The onboarding path** (`runner.yaml`) and a **real engine** both work first try. +- **E3** (task output isn't propagated) is the next thing to fix — and the hand-rolled DAG/retry is the + argument for building `agent-4-workflow`. + +Total spend across all three runs: **~$0.27**. diff --git a/orchestrator-v2/docs/experiment/gap-report.md b/orchestrator-v2/docs/experiment/gap-report.md new file mode 100644 index 0000000..74db0e8 --- /dev/null +++ b/orchestrator-v2/docs/experiment/gap-report.md @@ -0,0 +1,224 @@ +# Orchestrator v2 — Documentation ↔ Implementation Gap Report + +**Purpose:** Points where the design docs describe behavior the current code does not implement (or implements differently), for the author to reconcile. **No fixes applied** — observations, not prescriptions. Pure dead-code / unused-export items are kept out of the doc-gap sections and collected in §8 so the documentation audit stays focused. + +**Scope:** `orchestrator-v2/docs/*` and package READMEs vs. `orchestrator-v2/packages/*/src/`, at the current `main` checkout. Each finding was read in the code; load-bearing ones were verified directly. + +## Design grounding (the 8-stage maturity model) + +Two design records ground this codebase (`devbox-8-stages-of-ai-maturity.pdf`, `bifrost-orchestrator-design.pdf`). They define the model `orchestrator-v2` implements: the **Trust Evolution — 8 stages** in three waves, where **layers alternate procedural / non-procedural** because *"prompts are guidance; the guarantee comes from the harness — wherever a guarantee is required, the layer must be procedural."* + +| Stage | Name | Nature | In this repo | +|---|---|---|---| +| 3 | Task | graded | `agent-3-task` — one autonomous session, a single unit of work | +| 4 | Workflow | procedural | `agent-4-workflow` (**planned**) — sequences Stage-3s with gates + adversarial review; *the* hard layer | +| 5 | Delegate | non-procedural | LLM picks which Stage-3/4 to run; no tools, may skip, framework-enforced kill limits | +| 6 | Coordinate | procedural | parallel Stage-5s (worktrees) | +| 7 | Supervise | non-procedural | consumes **telemetry from every level**, intervenes | +| 8 | Orchestrate | procedural | structural changes to the harness itself | + +**`orchestrator-v2` is the thin rebuild of Bifrost, the Stage-4 harness**, resting on two explicit premises: (1) **the iteration count moves into the task definition (default 1)**; (2) **agents may invoke a Stage-3 directly, without the grader loop.** Hence the engine is "dumb dispatch," agents are scripts, and the `TaskSource` owns readiness/dependencies/priority ("Bifrost is one task source; a Jira-JSON-blob backend is another"). Corollary: the `darker` tooling in this repo (`darker-challenger`/`-prober`/`-eye`, complexity/tier routing, TDD red/green) is a DevBox-lineage implementation of the same model — chiefly the Stage-4 adversarial-review layer. + +**How this reframes findings below:** +- **G2** is not a doc/code drift but design **premise #1** — the single `engine.execute` call is intentional; the doc describes the **v1** model that looped. Re-annotated in §1. +- **Telemetry (Route A / E1, implemented)** is a **Stage-7 prerequisite** — but it lacks the branch/session attribution Stage 7 needs (new gap **T1**, below). +- **I1's blocking limitation** is the design's own open question (*"a Stage-4 must-finish blocks the framework; can-pause risks never resuming, pushing prioritization onto the higher levels"*) — a known tension, not an oversight. +- **`agent-4-workflow` unbuilt** (§6) is the hard procedural layer; the design holds **adversarial review mandatory** for it (the canonical DevBox failure: all tests green, DB fully mocked, migration missing, system dead). + +**T1 (NEW, design-grounded) — telemetry has no branch/session attribution.** `ExecutionStats` carries `durationMs`/tokens/`totalCostUsd`/`numTurns` but nothing tying a run to a **work branch** or session. The design requires Stage-7 supervision telemetry to be *"session-aware and tied to a work branch — otherwise seven parallel red tasks all write to one log and you can't tell which maps to which branch."* Route A now propagates telemetry to the task source, but without attribution it can't feed supervision. · design/telemetry · grounded-in-design + +## What's solid (lead with this) +- **`docs/orchestrator.md`** and the **workspace/`docs` READMEs** are highly accurate — component names, option shapes, defaults, the RPC method table, `runner.yaml` keys, and every code example resolve against the code. +- **`docs/protocol.md`** has **no correctness gaps** (only the two minor completeness items and one clarity note in §3); the crypto / canonicalization / trust-model claims all hold, and it agrees with the protocol README. +- Config discovery, override precedence, heartbeat defaults, and error strings in the runner docs all match. + +## Summary of gaps + +| # | Area | Correctness | Clarity | Minor | +|---|------|:---:|:---:|:---:| +| 1 | Task Agent (`agent-3-task`) | 0 | 2 | 0 | +| 2 | Orchestrator | 0 | 1 | 1 | +| 3 | Protocol | 0 | 1 | 1 | +| 4 | Runner | 0 | 2 | 3 | +| 5 | Script tasks (`interfaces-task*`) | 2 | 1 | 2 | +| 6 | Workflow Agent (`agent-4-workflow`) | — planned (readiness checklist) — | | | +| 7 | READMEs | 0 | 1 | 1 | + +**2 correctness-level gaps, both in §5:** G12 (README `Task` type has the wrong field names) and G13 (`ScriptContext` doc omits the registries scripts actually use). §6 is a **planned, unbuilt** agent — a readiness checklist, not bugs. §8 lists implementation cleanup that is *not* a documentation issue. + +> **Update — implemented (Route A):** **E1** (telemetry propagation) and **G1** (failed-run session persistence) have been fixed end-to-end. Telemetry now travels as a terminal outcome to the task source, and failed runs persist their `sessionId`. Build clean, all **40 tests pass**, `vp lint` clean. Details in the Empirical validation section and §1. Changes are uncommitted. + +**Legend** — *Type:* `unimplemented` · `behavior-diverges` · `doc-stale` · `naming-mismatch` · `missing-from-doc`. *Confidence:* `confirmed` (verified in code) · `needs-check` (interpretation-dependent). + +--- + +## Empirical validation (live run) + +Beyond static reading, I stood up a **real orchestrator + real runner + `TestEngine`** and dispatched two Task Agent tasks through the actual signed-WebSocket path — a path the test suite does **not** cover (`runner.spec` only exercises stub `"script"` agents, never `agent-3-task`). Findings: + +- **The happy path works end-to-end** ✅ — signed handshake, heartbeat-gated availability, dispatch, task-agent execution against the engine, `setState`, `task.complete`. +- **G1 confirmed live → ✅ FIXED.** *Originally:* the successful task persisted its `sessionId`; the task whose engine returned `success:false` was failed **without** persisting the returned `sessionId`. *Fix:* `run-task-agent.ts` now persists `sessionId` **before** branching on outcome, so a failed run is resumable — re-verified live (the failed task now persists its session). +- **E1 (NEW) — telemetry is computed but never recorded → ✅ IMPLEMENTED (Route A).** *Originally:* `runTaskAgent` returned `telemetry` in its `ScriptResult`, but the runner's dispatch handler forwarded only `{ taskId }` on `task.complete` (`dispatch-handler.ts:65`); tokens/cost/`numTurns` were dropped before reaching the task source. *Fix:* telemetry now propagates as a **terminal outcome** — both `task.complete` and `task.fail` carry it (`dispatch-handler.ts`), the orchestrator's `ResultHandler` validates + forwards it (`readTelemetry`), and the `TaskSource` contract is widened to `completeTask(taskId, telemetry?)` / `failTask(taskId, error, telemetry?)` (`interfaces-task-source/src/types.ts`), with `ExecutionStats` unified as one canonical type in `interfaces-task-source`. Re-verified live: the source now receives telemetry on **both** success (`$0.005`) and failure (`$0.02`). · behavior-diverges · **resolved** +- **E2 (NEW) — `vp run ready` fails on a clean checkout.** From a fresh install (no `dist`), `vp run -r test` fails because cross-package `@bifrost-ai/*` imports resolve to unbuilt `dist/index.mjs` (e.g. `orchestrator.spec` can't resolve `@bifrost-ai/protocol`). The root `ready` script runs `test` **before** `build`, so `vp run ready` on a fresh clone fails at the test step. Running `vp run -r build` first makes all **40 tests pass**. · build/workflow · confirmed +- **E3 (NEW) — a task's output/message is not propagated; only telemetry is.** `task.complete` carries `{ taskId, telemetry }` and `TaskSource.completeTask(taskId, telemetry?)` records telemetry, but the task's `message` (the actual work output) is dropped — exactly as telemetry was before Route A. Surfaced by the **DAG dogfood**: chaining one task's result into a downstream task only worked because the **engine persisted its output via `ctx.setState`**; there is no built-in channel for a task's output to reach the orchestrator/source or a downstream step. This blocks output-chaining workflows and is the same shape as E1 (now fixed) but for the message. It reinforces the §6 workflow-readiness gaps: a real workflow must read child *outcomes*, not just telemetry. *(Candidate follow-up: a Route A-style extension carrying `message` on the terminal outcome, or a documented "engines persist output via setState" contract.)* · behavior · confirmed + +--- + +## Integration testing (architecture gaps) + +A live multi-runner harness (real orchestrator + real runners + script agents) exercising paths the unit specs don't cover. + +**Works correctly (no gap):** multi-runner **concurrency + back-pressure** (4 tasks / 2 runners / `maxInFlight=1` → 2 concurrent batches, ~624ms); **`maxInFlightPerPeer > 1`** (3 tasks concurrent on one runner, ~313ms); the **`paused`** terminal path; **disconnect cleanup** (orphaned task failed + slot freed); and **recovery by reconnection** — a fresh `Runner` (same identity) re-dials and resumes taking work. *(There is no **auto**-reconnect: a dropped `Runner` instance stays dropped; the app must construct a new one.)* + +**I1 — No capability-aware routing; a runner's rejection is terminal.** The orchestrator dispatches each task to the *first available* peer regardless of which agents that peer has registered (`peer-registry.ts` `getAvailablePeer`). If the chosen runner hasn't registered the task's agent, its dispatch handler replies `accepted:false` ("Unknown agent"), and `DispatchAckHandler.reject` **fails the task** (`dispatch-ack-handler.ts`) instead of re-dispatching to a capable peer. Demonstrated live: with **both** runners connected + available and only runner B registering the `special` agent, a task needing `special` was dispatched to runner A and **permanently failed** — `failTask(s1, "Unknown agent: special")` — runner B never got it. The design implicitly assumes **homogeneous runners** (every runner registers the same agents), but the per-runner `registerAgent` API permits heterogeneity and nothing documents or enforces the assumption. · behavior/architecture · confirmed + +> **✅ Fixed (this session) — capability-aware routing.** Runners now advertise the `capabilityKey(agentType, agentName)` of every registered agent in their heartbeat (optional field; `protocol` owns the canonical key format). The orchestrator's `PeerRegistry` records each peer's advertised capability set, and `getAvailablePeer`/`waitForAvailablePeer` filter by the task's required capability — so a task only dispatches to a peer that can run it. **Backward compatible:** a runner that advertises nothing (e.g. the stub runners used in tests) is treated as capable-of-anything. Verified live (a task needing `special` now routes to the runner that has it instead of failing on an incapable one), guarded by a new `peer-registry.spec.ts` (2 tests). 43/43 tests pass. *Known limitation: the dispatch loop is still sequential, so a task requiring a capability that **no** connected runner advertises now **waits** (blocking later tasks) rather than fail-fast — a concurrent dispatch loop or a route-timeout is a separate follow-up.* + +**I2 — A mid-task runner disconnect is a terminal task failure, with no retry.** When a runner drops mid-task, the orchestrator fails the orphaned task (`failTask(taskId, "Runner disconnected")`, `result-handler.ts` `handleDisconnect`) and does not re-dispatch it. Confirmed live. Consistent with the "thin orchestrator" design (retries are meant to live at the workflow/child level per `agent-4-workflow.md`), **but that layer doesn't exist yet** (§6) — so today a transient runner blip permanently fails a bare task with no recovery anywhere in the system. · behavior/resilience · confirmed + +**I3 — Task-source callbacks have no error handling; a single throw wedges the peer, leaks its slot, and escapes as an unhandled rejection.** The orchestrator's RPC handlers `await` the task source and only *then* release the slot / respond — `handleComplete` runs `await taskSource.completeTask(...)` **before** `registry.markTerminal(...)` and `sendRpcResponse(...)` (`result-handler.ts`). A real source does DB/network I/O, so a rejecting callback is expected eventually — and when it happens: (a) `markTerminal` is skipped → the peer's in-flight slot is **leaked**, so that peer is permanently unavailable and every later task routed to it hangs on `waitForAvailablePeer`; (b) `sendRpcResponse` is skipped → the runner's terminal RPC never resolves; (c) the rejection is **unhandled** — `RpcRouter.handle` invokes `route(...)` with no `await`/`.catch` (`rpc-router.ts`), so it escapes as an unhandled promise rejection (process-fatal on default Node). Demonstrated live: a `completeTask` that threw on `f1` left the peer wedged, `f2` never dispatched, the orchestrator's `done` never resolved (timed out), and exactly one unhandled rejection surfaced. Same shape for `failTask`/`pauseTask` (slot leak) and `taskSource.setState` (hung RPC + unhandled rejection; no slot leak since it isn't terminal). · behavior/robustness · confirmed + +> **✅ Fixed (this session).** The orchestrator now guards every task-source/scheduler callback. A `settle()` helper in `result-handler.ts` runs each terminal source call in `try/catch/finally`, so slot release (`markTerminal`) and the runner response **always** happen even if the source throws; `handleDisconnect`, `dispatch-ack-handler.reject`, and `rpc-router`'s `setState`/`scheduler.call` are likewise guarded; and `RpcRouter.handle` adds a `.catch` net so a rejecting handler can never escape as an unhandled rejection. Source failures are surfaced via `console.error` (not silently swallowed). Verified live — a throwing `completeTask` no longer wedges the peer (`f2` still dispatches, `done` resolves, **0** unhandled rejections) — and guarded by a new regression test (`orchestrator.spec.ts` → "does not wedge the peer when a task source callback throws"). 41/41 tests pass. *Follow-up: a real `onError` option would beat `console.error` for a library.* + +--- + +## 1. Task Agent — `docs/agent-3-task.md` ↔ `packages/agent-3-task` + +The implementation is 4 small files. The core (`run-task-agent.ts`) makes a **single `engine.execute()` call** — the behaviors the doc attributes to "the agent" mostly live in the `Engine`. + +### Clarity + +**G1 — Session/telemetry checkpointing is engine-dependent, not agent-provided; the agent also drops a failed-result `sessionId`.** +- **Doc says:** §Running — the agent "may checkpoint progress (session ID, partial telemetry) so that if something goes wrong mid-flight, a retry can pick up where it left off"; §3 — a parent retries "with the same session id." +- **Code does:** `runTaskAgent` persists `sessionId` **only after success** (`run-task-agent.ts:53-58`); both failure exits (`:41-44` catch, `:46-51` `!success`) return without `setState` and ignore any `sessionId`/`stats` on the returned `EngineResult`. So resume after a failure is possible **only if the engine checkpoints itself** via the `setState` it's handed in `EngineContext` (`engine/src/types.ts:28`, wired at `run-task-agent.ts:37`). Accurate framing: *the doc attributes checkpointing to the agent, but the agent contributes none beyond post-success persistence and discards a `sessionId` returned on a failed result — mid-run/after-failure resume depends entirely on engine-side checkpointing.* +- **Type:** behavior-diverges · **Confidence:** confirmed (**demonstrated in the live run — see Empirical validation**) +- **Status: ✅ Implemented (Route A).** `run-task-agent.ts` now persists `sessionId` *before* the success/failure branch (failed runs are resumable) and returns telemetry on the failure path; paired with E1's propagation, a failed run's partial telemetry reaches the task source. 40/40 tests pass. *(The broader §Running "checkpointing is the agent's job" framing remains a doc-wording question; the concrete failed-run data loss is fixed.)* + +**G2 — The doc's "turn loop / maximum turn limit inside the agent" describes the pre-rebuild (v1) model.** §Running describes a 5-step turn loop "inside the agent" with a "configured maximum turn limit"; `runTaskAgent` calls `engine.execute()` exactly **once** (`run-task-agent.ts:29`) — no loop or turn-limit anywhere in `agent-3-task`. This is **intended**, not a divergence: the thin-rebuild premise moves the iteration count into the task definition (**default 1**) and keeps the engine "dumb" (see *Design grounding*). The v1 orchestrator *did* loop (`orchestrator/…/core/orchestrator.ts` `runEngineLoop`, `maxFollowUps=10`); the doc still describes that older shape. The fix is to the **doc**, not the code — and, if per-task iteration is ever wanted, it belongs in the task definition/`TaskSource`, not the agent. · doc-stale (describes v1) · confirmed + +> **Undocumented contract (worth adding):** task agents register under agentType `"task"`, keyed by `agent.name` (`runner.ts:28`); dispatch resolves `agents.get(agentType).get(agentName)` (`dispatch-handler.ts:49`). A Task Agent task must be created with `agentType: "task"`, `agentName: ""`. + +_(Two unused-artifact items from this package — `skipFulfill`, the `agentDefinition` registry registration — are in §8.)_ + +--- + +## 2. Orchestrator — `docs/orchestrator.md` ↔ `packages/orchestrator` + +The most accurate doc in the set — no correctness or behavior gaps. Only completeness omissions. + +### Clarity + +**G3 — `orchestrator.md` doesn't enumerate the `taskSource.setState` RPC.** The doc lists `dispatch` + `task.complete`/`fail`/`pause` + `scheduler.call`, but omits `taskSource.setState`, which `RpcRouter` handles and proxies (`rpc-router.ts:308-310, 317-325`). **Note:** this is an *orchestrator-doc* completeness issue — `docs/protocol.md:95` already documents the method in its RPC table, so it is not a globally undocumented RPC. · missing-from-doc · confirmed + +### Minor + +**G4 — `runOrchestrator`'s `abortSignal` option isn't in the Configuration block.** The run entry point is `OrchestratorOptions & { abortSignal?: AbortSignal }` (`orchestrator.ts:381`) and the signal drives graceful shutdown/drain-abort (`:414-436`); the doc's options block never mentions it. · missing-from-doc · confirmed + +_(Dead code `isRpcResponse` is in §8.)_ + +--- + +## 3. Protocol — `docs/protocol.md` + `packages/protocol/README.md` ↔ `packages/protocol` + +**No correctness gaps** — the two docs agree with each other and with code. The items below are a clarity note and a completeness omission. + +### Clarity + +**G5 — README's "bad base64 → false" case relies on dead error-handling.** README says `verifyEnvelope` returns `false` on "bad base64." `sign.ts:63-65` wraps `Buffer.from(sig, "base64")` in a `try/catch`, but `Buffer.from` decodes leniently and never throws, so the `catch` is unreachable. The observable result still holds (`crypto.verify` returns `false` at `sign.ts:69`) — only the implied explicit base64 handling is dead. · doc-stale · confirmed + +### Minor + +**G6 — Ephemeral-port discovery is undocumented.** README says `port: 0` binds an ephemeral port but never says how to learn it: `OrchestratorPeer.address = { host, port }` (`types.ts:68`, `orchestrator.ts`). The returned peer shape is never spelled out. · missing-from-doc · confirmed + +> **Checked — not a gap:** the `heartbeat` frame's "keep peer alive" purpose *is* realized — the runner sends heartbeats on a timer (`runner/src/heartbeat.ts:16`) and the orchestrator uses them for liveness (`peer-registry.ts:49,109`). Distributed across packages, not the protocol package. + +_(Unused/test-only exports — `signRawMaterial`, `loadTrustedPublicKey`, `SIGNING_ALGORITHM` — are in §8.)_ + +--- + +## 4. Runner — `docs/runner.md` + `packages/runner/README.md` ↔ `packages/runner` + +Docs largely accurate; gaps are documentation-completeness, no correctness. + +### Clarity + +**G7 — Doc says a "scheduler" is reachable over RPC, but no scheduler surface exists at the script layer.** "Task source and scheduler are reached over RPC." (`docs/runner.md:128`). The `ScriptContext` built for scripts exposes only `data`, `agents`, `setState` (`script-context.ts:24-38`); no scheduler handle, and the interface has no scheduler field. `scheduler.call` exists on the orchestrator side (see §6/G18) but nothing in the runner reaches it. Reads as a forward-reference that isn't wired. · doc-stale / unimplemented · confirmed + +**G8 — The "RPC-backed ScriptContext" field table omits `agents`.** The table lists `taskId`, `agentType`, `agentName`, `taskState`, `metadata`, `data`, `setState` (`docs/runner.md:118-127`), but the context also populates `agents` (`script-context.ts:24-31`), a required interface field. · missing-from-doc · confirmed + +### Minor + +**G9 — Dispatch diagram omits the malformed-task rejection path.** The diagram shows only unknown-agent vs. known; code rejects a `validateTask` failure first, with a `missingTaskFields` reason (`dispatch-handler.ts:40-47`). · missing-from-doc · confirmed + +**G10 — Dispatch diagram attributes signature verification to the runner; it happens in the protocol layer** (`verifyEnvelope` in the connection layer, before any subscriber). The prose trust-model section is correct, so this is diagram imprecision. · doc-stale · confirmed + +**G11 — Undocumented (but live) public surface:** `RunnerOptions.abortSignal` (wired to `close()`, `runner.ts:60-64`), `Runner.getAgent`/`hasAgent` (`runner.ts:31-37`), and the `discoverConfigPath` export. · missing-from-doc · confirmed + +--- + +## 5. Script tasks — `docs/script-tasks.md` + `packages/interfaces-task-source/README.md` ↔ `interfaces-task*` + +The execution-primitive docs have drifted from the current `Task`/`ScriptContext` shapes — **both correctness gaps in this report are here.** + +### Correctness + +**G12 — The task-source README's `Task` type has the wrong field names.** +- **Doc says:** `Task = { id; scriptName; taskState; metadata }`, with a field table and prose "the runner uses `scriptName` to look up the script" (`interfaces-task-source/README.md:27-28, 36-37, 41, 100`). +- **Code does:** `Task = { taskId; agentType; agentName; taskState; metadata }` (`interfaces-task-source/src/types.ts:3-9`). No `id`, no `scriptName`; validation requires `["taskId","agentType","agentName"]` (`types.ts:19`), and the runner resolves by `agentType`+`agentName` (`dispatch-handler.ts:49`). Code written from this README fails `validateTask`. (The `TaskSource` *method* signatures in the same README are correct — only the `Task` type and the `scriptName` prose are stale.) +- **Type:** naming-mismatch + behavior-diverges · **Confidence:** confirmed + +**G13 — `script-tasks.md`'s `ScriptContext` omits the registries scripts actually use.** +- **Doc says:** `ScriptContext = { taskState; readonly metadata; setState }`, presented as the contract (`docs/script-tasks.md:21-24`). +- **Code does:** `ScriptContext` also has `taskId`, `agentType`, `agentName`, `readonly data: DataRegistry`, and `readonly agents: AgentRegistry` (`interfaces-task/src/types.ts:35-44`). `data` is load-bearing — the Task Agent reads `ctx.data.get(ENGINE_DATA_TYPE)` (`run-task-agent.ts:22`). The documented contract is a strict subset that would leave a reader unable to write an engine-backed agent. +- **Type:** doc-stale / missing-from-doc · **Confidence:** confirmed + +### Clarity + +**G14 — "`taskState` mutations persist" overstates behavior.** `docs/script-tasks.md:73` says mutations persist, but `taskState` is a getter over local state; persistence happens only inside `setState` → `Object.assign` + `rpc.call("taskSource.setState")` (`script-context.ts:32-42`). A direct `ctx.taskState` mutation without a following `setState` never goes over the wire. (The doc's own behavior table at `:42` is precise — internal tension.) · behavior-diverges · needs-check + +### Minor + +**G15 — "types only" claims are inaccurate — both packages ship runtime functions.** `script-tasks.md:53` calls `interfaces-task` "types only," but it exports the runtime guard `isScriptTaskDefinition` (`interfaces-task/src/types.ts:58-65`). The task-source README says "pure interface package — types only, no runtime logic" (`:150`), but it exports `validateTask`, `missingTaskFields`, `missingTaskFieldsMessage`, used by the runner (`dispatch-handler.ts:40-46`). · doc-stale · confirmed + +**G16 — Modeling omissions:** `ScriptTaskDefinition`/`ScriptContext` are generic over `TData` in code but shown non-generic; `setState` semantics (it **merges** via `Object.assign` and transmits the full accumulated state) are undocumented. · missing-from-doc · needs-check + +--- + +## 6. Workflow Agent — `docs/agent-4-workflow.md` (planned, no package) + +**Status: ~0% implemented, honestly labeled "Planned (#39)"** — no `packages/agent-4-workflow`, and a repo-wide grep for `workflow`/`agent-4` finds only `docs/`. Its unbuilt state is *acknowledged*, not hidden. The value is a **readiness checklist**: the enabling mechanisms it leans on are missing or half-built. + +**G17 — TaskSource lacks every capability the "schedule" step needs.** The doc has the workflow "create all child tasks, wire dependency edges, promote the ready ones, register every child as a blocker" (`agent-4-workflow.md:41-47, 82-99, 180-199`). But `TaskSource` has only `watchTasks`/`completeTask`/`failTask`/`pauseTask`/`setState` (`interfaces-task-source/src/types.ts:11-17`) — no create-child / dependency / promote / blocker method anywhere. · unimplemented · confirmed (whether these belong on the interface or inside a concrete source is `needs-check`) + +**G18 — `scheduler.call` is server-side only, unconsumed, and unreachable from scripts.** The natural channel a workflow script would use. Receiving side exists (`Scheduler` type `orchestrator/src/types.ts:5`; routed at `rpc-router.ts:42,76`; required option `types.ts:13`), **but** nothing sends `scheduler.call` (grep: only `rpc-router.ts`), the only implementation is `createNoopScheduler` (`test-helpers.ts:54`), and `ScriptContext` exposes no scheduler handle. Directly connects to runner-doc gap **G7**. · unused-artifact / unimplemented · confirmed + +**G19 — The doc's own "permanent child failure" trigger is marked "TO BE RESOLVED"** (`agent-4-workflow.md:201-207`) and is unimplemented — consistent with the doc. · unimplemented · confirmed + +> **Correctly builds on existing primitives:** `ScriptTaskDefinition`, the child Task Agent, and the `TaskSource` streaming/outcome interface all exist; the doc's cross-links resolve. + +--- + +## 7. READMEs — `docs/README.md` + workspace `README.md` + +Highly accurate — package table, code examples, the 5-method RPC table, `runner.yaml` keys, and prerequisites all check out. Two small gaps: + +### Clarity + +**G20 — `ctx.data.get(type)` returns a *read-only* registry, but the README calls it `Registry`.** `README.md:58` says `get(type)` returns `Registry`; the actual return is `ReadonlyRegistry` (`get`/`has` only, `interfaces-task/src/types.ts:23`). `Registry` is a distinct mutable type with `register()` — the runner hands scripts a read-only view on purpose. Could mislead a reader into thinking a script can `.register()` into `ctx.data`. · naming-mismatch · confirmed + +### Minor + +**G21 — Runner dependency sentence omits `interfaces-task-source`.** `docs/README.md:78` says the runner "consumes `protocol` and `interfaces-task`," but `runner/package.json` declares `interfaces-task-source` as a third workspace dep. Not wrong, just incomplete. · doc-stale · needs-check + +--- + +## 8. Implementation cleanup — NOT documentation gaps + +Dead code / unused exports / a latent code smell surfaced during the audit. Independent of the docs; listed separately so the gap report above stays about documentation. + +- **C1 — `EngineResult.skipFulfill` is never consumed.** Defined and returned by engines (`engine/src/types.ts:34`) but `runTaskAgent` never reads it. +- **C2 — The `agentDefinition` registry registration is unused on the run path.** `enrollTaskAgent` registers the `AgentDefinition` into `data[agentDefinition]` (`enroll-task-agent.ts:14`), but `runTaskAgent` uses the closured `agent`, never `ctx.data.get(AGENT_DEFINITION_DATA_TYPE)`. That registration and the `isAgentDefinition` guard are unconsumed within the package. +- **C3 — `isRpcResponse` is dead code.** Exported from `orchestrator/src/types.ts:356`, never imported/called, not re-exported from `index.ts`, and byte-identical to `isDispatchAck` — the predicate actually used (`orchestrator.ts:410`). +- **C4 — Unused / test-only protocol exports.** `signRawMaterial` (`sign.ts`, used only by a spec), `loadTrustedPublicKey` (`keys.ts`, unreferenced anywhere), and `SIGNING_ALGORITHM` — either prune or, if intended as public API, document them. +- **C5 — `protocol/src/runner.ts` references `connection` before its `const` declaration** inside the socket `error` handler. Works only because the handler fires asynchronously after initialization; a synchronous error would hit the temporal dead zone. Worth a defensive reorder. +- **C6 — Half-wired `scheduler.call` (see G18)** is dead in the current system, not just undocumented — no sender, noop-only implementation. diff --git a/orchestrator-v2/docs/experiment/lessons.md b/orchestrator-v2/docs/experiment/lessons.md new file mode 100644 index 0000000..4d37d21 --- /dev/null +++ b/orchestrator-v2/docs/experiment/lessons.md @@ -0,0 +1,112 @@ +# Orchestrator v2 — Session Retrospective & Lessons + +A working session that audited, hardened, and dogfooded the Orchestrator v2 rebuild. This +captures what changed, what we learned, and where it should go next. Companion to the +detailed **[gap report](./gap-report.md)**. + +--- + +## 1. What we shipped (all uncommitted, on the working tree) + +| Area | Change | Proof | +| ---------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------- | +| **Route A — telemetry propagation** | Telemetry now flows engine → `ScriptResult` → RPC → `TaskSource` as a terminal outcome: `task.complete`/`task.fail` carry it, `ResultHandler` reads+forwards it, `TaskSource.completeTask/failTask` widened with `telemetry?`, `ExecutionStats` unified in `interfaces-task-source`. | Live demo + 40→43 tests | +| **G1 — failed-run session persistence** | `run-task-agent.ts` persists `sessionId` _before_ the success/failure branch, so a failed run is resumable; returns telemetry on failure too. | Live demo | +| **I3 — task-source callback resilience** | A throwing `completeTask`/`failTask`/`pauseTask`/`setState` no longer wedges the peer (leaked slot) or escapes as an unhandled rejection. New `settle()` + `recordBestEffort()` guards; `.catch` net on `route()`. | New `orchestrator.spec` regression test + live | +| **I1 — capability-aware routing** | Runners advertise `capabilityKey(agentType, agentName)` in the heartbeat; the orchestrator only dispatches a task to a peer that can run it. Backward compatible (unadvertised = capable-of-anything). | New `peer-registry.spec` + live | +| **Packaging fix** | `@bifrost-ai/orchestrator` `./test-helpers` export pointed at `src` (dead in the tarball); fixed via `publishConfig.exports` + a build entry. | `pnpm pack` verified | +| **Quality** | Two `/simplify` passes: co-located `isExecutionStats`, `errorMessage`/`isStringArray`/`recordBestEffort` helpers, DRY'd terminal handlers. | build + lint clean | + +Two `/simplify` rounds and a fresh **example** (`examples/claude-engine/`) were added along the way. + +--- + +## 2. What we learned about the architecture + +**The skeleton is solid.** Confirmed working under load: multi-runner concurrency + back-pressure, +`maxInFlightPerPeer > 1`, the pause path, disconnect cleanup, recovery-by-reconnection, signed +handshake, and heartbeat liveness. The protocol and the thin dispatch loop are well-built. + +**The recurring lesson — the terminal boundary drops everything but `taskId`.** The orchestrator is +genuinely "thin": at task completion it moved a `taskId` and did slot accounting, and _everything +richer had to be threaded deliberately_: + +- **Telemetry** was computed then dropped at `dispatch-handler` → fixed (Route A / E1). +- **A throwing callback** leaked the slot + crashed the process → fixed (I3). +- **Capability** wasn't considered → tasks failed on incapable runners → fixed (I1). +- **The task's output/message** is _still_ dropped — only telemetry rides the terminal RPC (E3, below). + +So the through-line: v2 is a correct dispatch skeleton whose **outcome plumbing** needed fleshing out, +and the layer that would _consume_ rich outcomes (the Workflow Agent) **does not exist yet**. + +**Open gaps (see gap report for detail):** + +- **I2** — a mid-task disconnect is a terminal failure with no retry. By design for a thin + orchestrator, but the retry layer (agent-4-workflow) isn't built, so today a transient blip + permanently kills a task. +- **E2** — `vp run ready` fails on a clean checkout (it runs `test` before `build`; cross-package + imports resolve to unbuilt `dist`). +- **E3** — task output isn't propagated (surfaced by the DAG dogfood; see §3). +- **Docs** — the design docs drifted from code (the `Task`/`ScriptContext` shapes in the README are + wrong; the workflow agent is ~0% built with `scheduler.call` half-wired). 3 correctness-level doc gaps. +- A pre-existing root-`tsconfig` missing-`@types/node` lint error (unrelated to our work). + +--- + +## 3. Dogfooding + +**Basic (`examples/claude-engine`) — a real deployment.** Generated keys, a real `runner.yaml` +loaded via `Runner({ configPath })`, `enrollTaskAgent`, and a **real `claude`-CLI engine**. Two tasks +completed; real token/cost/turn telemetry reached the task source; real Claude `session_id`s +round-tripped. **The never-tested `runner.yaml` onboarding path worked first try.** + +**Harder — a hand-orchestrated DAG.** Two summary tasks fanned out to a capability-routed +_summarizer_ runner (concurrently, `maxInFlight=2`), then an aggregate _release-notes_ task routed to +a separate _reviewer_ runner. Real Claude did the work; telemetry rolled up across all three tasks +($0.0645). The system wrote accurate release notes for **its own** new features. + +What the harder run validated _and_ exposed: + +- ✅ I1 capability routing with two real specialized runners; concurrency; Route A telemetry roll-up + (exactly a workflow "verify-step aggregate stats"). +- ⚠️ **E3 — no output propagation.** Chaining `sum-1`/`sum-2` results into `release-notes` only worked + because the **engine persisted each output via `ctx.setState`**. `task.complete` carries telemetry, + not the `message`. There is no built-in way for a task's output to reach the source or a downstream + step — the same shape as the telemetry gap we fixed, but for the result. +- ⚠️ The DAG gate was **hand-rolled in the task source** because the Workflow Agent doesn't exist. + This _is_ what agent-4-workflow is meant to own (schedule the graph, gate by deps, aggregate). + Doing it by hand is the concrete argument for building that layer — and for a `TaskSource` that can + read child _outcomes_ (today it's write-only: complete/fail/pause/setState). + +**Hardest — a 3-stage pipeline with retry + session-resume.** Three analyses fanned out to an +_analyzer_ runner (concurrent), a _synthesizer_ runner (a **sonnet** engine) combined them, then a +_critic_ runner (haiku) reviewed the result — capability-routed across three specialized runners with +**heterogeneous engines**. One analyze task was flaky: it failed after doing its work, and the task +source **retried it with `--resume`** — attempt 2 cost **$0.0059 vs. $0.0253** (4.3× cheaper) by +reusing the session instead of rebuilding context. That is the concrete payoff of the G1 fix (failed +runs stay resumable). Per-attempt telemetry receipts and the pipeline's (self-critiquing) output are +in **[dogfood-runs.md](./dogfood-runs.md)**. + +--- + +## 4. Recommended next steps (in priority order) + +1. **E3 — propagate task output.** Either extend the terminal outcome to carry `message` (a Route + A-style change), or make "engines persist output via `setState`" an explicit, documented contract. + Prerequisite for any output-chaining workflow. +2. **Reconcile the docs (§5 of the gap report).** The 3 correctness-level doc gaps (README `Task` + shape, `ScriptContext` omissions) will actively mislead new contributors. +3. **Build agent-4-workflow** — the DAG dogfood shows the hand-rolled gate is load-bearing and + painful. It needs a read-side on `TaskSource` (read child outcomes) to aggregate. +4. **Fix E2** (build-before-test ordering / a `source` export condition) so `vp run ready` works on a + clean clone. +5. **I2 retry policy** — decide where task-level retry lives (workflow layer vs. orchestrator). + +--- + +## 5. Artifacts + +- **Gap report:** `gap-report.md` — full doc↔code + integration findings (I1–I3, E1–E3, G1–G26). +- **Dogfood runs:** `dogfood-runs.md` — the three real-engine runs with telemetry receipts. +- **Example:** `orchestrator-v2/examples/claude-engine/` — a runnable real-engine deployment. +- **Branch:** `refactor/orchestrator-v2-hardening` — Route A + I3 + I1 + `/simplify` + the example. 43/43 tests, build + lint clean. diff --git a/orchestrator-v2/examples/claude-engine/README.md b/orchestrator-v2/examples/claude-engine/README.md new file mode 100644 index 0000000..71e7306 --- /dev/null +++ b/orchestrator-v2/examples/claude-engine/README.md @@ -0,0 +1,40 @@ +# Example: a real `claude`-backed engine + +Runs Orchestrator v2 as a real deployment and drives it with a **real engine** — the +[`claude` CLI](https://docs.claude.com/en/docs/claude-code). It demonstrates the pieces a +real operator wires together: + +- **Key generation** and a **real `runner.yaml`** loaded via `new Runner({ configPath })` + (the documented onboarding path — key/PEM parsing in `config-loader`). +- A **Task Agent** (`@bifrost-ai/agent-3-task`) enrolled with `enrollTaskAgent`. +- A **real `Engine`** — the whole engine contract is `execute(context, sessionId?) => Promise`. + Here it shells out to `claude -p … --output-format json` and maps the CLI's JSON + (`result`, `usage`, `total_cost_usd`, `num_turns`, `session_id`) straight into an + `EngineResult`, so **real token/cost/turn telemetry flows to the task source**. + +The `claudeEngine` object at the top of [`run.mjs`](./run.mjs) is the reusable part — drop +your own engine in its place (a different model, a local LLM, a stub) and the rest is unchanged. + +## Run it + +```bash +# from the orchestrator-v2 root +vp install +vp run -r build # the example imports the packages from dist +claude --version # the `claude` CLI must be on PATH and authenticated + +node examples/claude-engine/run.mjs +``` + +Expected output (two small `haiku` calls, a few cents total): + +``` +orchestrator listening on ws://127.0.0.1:xxxxx + ✓ t1 completed telemetry={in:10, out:61, turns:1, $0.02..} + ✓ t2 completed telemetry={in:10, out:128, turns:1, $0.01..} +done. +``` + +> Note: `input_tokens` looks small because Claude reports the system-prompt tokens +> separately as `cache_read_input_tokens` / `cache_creation_input_tokens` (both captured in +> `ExecutionStats`), which is why the reported cost is realistic. diff --git a/orchestrator-v2/examples/claude-engine/package.json b/orchestrator-v2/examples/claude-engine/package.json new file mode 100644 index 0000000..514fb1c --- /dev/null +++ b/orchestrator-v2/examples/claude-engine/package.json @@ -0,0 +1,15 @@ +{ + "name": "@bifrost-ai/example-claude-engine", + "version": "0.0.0", + "private": true, + "type": "module", + "scripts": { + "start": "node run.mjs" + }, + "dependencies": { + "@bifrost-ai/orchestrator": "workspace:*", + "@bifrost-ai/runner": "workspace:*", + "@bifrost-ai/protocol": "workspace:*", + "@bifrost-ai/agent-3-task": "workspace:*" + } +} diff --git a/orchestrator-v2/examples/claude-engine/run.mjs b/orchestrator-v2/examples/claude-engine/run.mjs new file mode 100644 index 0000000..3ba94b6 --- /dev/null +++ b/orchestrator-v2/examples/claude-engine/run.mjs @@ -0,0 +1,134 @@ +// A real Orchestrator v2 deployment, end to end: +// - ed25519 keys generated in-process +// - a real runner.yaml written and loaded via `new Runner({ configPath })` +// - a Task Agent (agent-3-task) enrolled on the runner +// - a REAL engine backed by the `claude` CLI: its JSON output maps straight to +// EngineResult, so real token/cost/turn telemetry flows to the task source. +// +// Prereqs: `vp install && vp run -r build` (the packages are imported from dist), +// and the `claude` CLI on PATH. Run: `node run.mjs` (makes 2 small haiku calls). + +import { execFile } from "node:child_process"; +import { promisify } from "node:util"; +import { mkdtemp, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { runOrchestrator, loadAuthorizedRunners } from "@bifrost-ai/orchestrator"; +import { generateKeyPair, exportPublicKeyPem, exportPrivateKeyPem } from "@bifrost-ai/protocol"; +import { Runner, createDataRegistry } from "@bifrost-ai/runner"; +import { enrollTaskAgent, taskAgentDataGuards } from "@bifrost-ai/agent-3-task"; + +const execFileAsync = promisify(execFile); +const log = (...a) => console.log(...a); + +// ── The reusable bit: an Engine backed by the `claude` CLI ────────────────── +// An Engine is just `{ execute(context, sessionId?) => Promise }`. +const claudeEngine = { + async execute(ctx, sessionId) { + const args = ["-p", ctx.instructions, "--output-format", "json", "--model", "haiku"]; + if (sessionId !== undefined) args.push("--resume", sessionId); + try { + const { stdout } = await execFileAsync("claude", args, { + cwd: ctx.workingDir, + maxBuffer: 10 * 1024 * 1024, + timeout: 90_000, + }); + const r = JSON.parse(stdout); + return { + success: !r.is_error, + skipFulfill: false, + lastMessage: r.result ?? null, + sessionId: r.session_id, + stats: { + durationMs: r.duration_ms ?? 0, + inputTokens: r.usage?.input_tokens ?? 0, + outputTokens: r.usage?.output_tokens ?? 0, + cacheReadTokens: r.usage?.cache_read_input_tokens ?? 0, + cacheCreationTokens: r.usage?.cache_creation_input_tokens ?? 0, + totalCostUsd: r.total_cost_usd ?? 0, + numTurns: r.num_turns ?? 0, + }, + }; + } catch (error) { + return { success: false, skipFulfill: false, lastMessage: String(error?.message ?? error), stats: null }; + } + }, +}; + +// ── A minimal in-memory TaskSource that records telemetry ─────────────────── +const fmt = (t) => (t ? `{in:${t.inputTokens}, out:${t.outputTokens}, turns:${t.numTurns}, $${t.totalCostUsd.toFixed(4)}}` : "none"); +const workDir = await mkdtemp(join(tmpdir(), "claude-engine-example-")); +const tasks = [ + { taskId: "t1", agentType: "task", agentName: "assistant", metadata: {}, taskState: { workingDir: workDir, engineName: "claude", instructions: "Reply with exactly one word, no punctuation: HELLO" } }, + { taskId: "t2", agentType: "task", agentName: "assistant", metadata: {}, taskState: { workingDir: workDir, engineName: "claude", instructions: "In one short sentence, describe what a task orchestrator does." } }, +]; +const source = { + async *watchTasks() { + for (const t of tasks) yield t; + }, + async completeTask(taskId, telemetry) { + log(` ✓ ${taskId} completed telemetry=${fmt(telemetry)}`); + }, + async failTask(taskId, error) { + log(` ✗ ${taskId} failed: ${error}`); + }, + async pauseTask() {}, + async setState() {}, +}; + +// ── Wire it up ────────────────────────────────────────────────────────────── +const orchestratorIdentity = generateKeyPair("orchestrator"); +const runnerIdentity = generateKeyPair("runner-1"); + +const handle = await runOrchestrator({ + identity: orchestratorIdentity, + authorizedRunners: loadAuthorizedRunners([ + { keyId: runnerIdentity.keyId, publicKeyPem: exportPublicKeyPem(runnerIdentity.publicKey) }, + ]), + taskSource: source, + scheduler: { async call() { return { ok: true }; } }, + port: 0, +}); +const { host, port } = handle.peer.address; +log(`orchestrator listening on ws://${host}:${port}`); + +// Write a real runner.yaml and load the runner's keys/URL from it. +const indentPem = (pem) => pem.trimEnd().split("\n").map((l) => ` ${l}`).join("\n"); +const cfgDir = await mkdtemp(join(tmpdir(), "claude-engine-cfg-")); +const configPath = join(cfgDir, "runner.yaml"); +await writeFile( + configPath, + [ + "orchestrator:", + ` url: ws://${host}:${port}`, + ` keyId: ${orchestratorIdentity.keyId}`, + " publicKeyPem: |", + indentPem(exportPublicKeyPem(orchestratorIdentity.publicKey)), + "identity:", + ` keyId: ${runnerIdentity.keyId}`, + " privateKeyPem: |", + indentPem(exportPrivateKeyPem(runnerIdentity.privateKey)), + " publicKeyPem: |", + indentPem(exportPublicKeyPem(runnerIdentity.publicKey)), + ].join("\n"), + "utf-8", +); + +const data = createDataRegistry(taskAgentDataGuards); +data.get("engine").register("claude", claudeEngine); +const runner = new Runner({ data, configPath }); +enrollTaskAgent(runner, { + name: "assistant", + description: "Answers small prompts via the Claude engine", + tools: [], + template: { parameters: {} }, + promptBody: "", +}); +await runner.start(); + +await handle.done; +runner.close(); +handle.peer.close(); +log("done."); +process.exit(0); diff --git a/orchestrator-v2/pnpm-lock.yaml b/orchestrator-v2/pnpm-lock.yaml index dfc19d4..206a8c8 100644 --- a/orchestrator-v2/pnpm-lock.yaml +++ b/orchestrator-v2/pnpm-lock.yaml @@ -43,6 +43,21 @@ importers: specifier: 'catalog:' version: 0.2.1(@types/node@25.9.4)(jiti@2.7.0)(typescript@6.0.3)(vite@8.1.2(@types/node@25.9.4)(jiti@2.7.0)(yaml@2.9.0))(yaml@2.9.0) + examples/claude-engine: + dependencies: + '@bifrost-ai/agent-3-task': + specifier: workspace:* + version: link:../../packages/agent-3-task + '@bifrost-ai/orchestrator': + specifier: workspace:* + version: link:../../packages/orchestrator + '@bifrost-ai/protocol': + specifier: workspace:* + version: link:../../packages/protocol + '@bifrost-ai/runner': + specifier: workspace:* + version: link:../../packages/runner + packages/agent-3-task: dependencies: '@bifrost-ai/engine': diff --git a/orchestrator-v2/pnpm-workspace.yaml b/orchestrator-v2/pnpm-workspace.yaml index a16068a..e84f25c 100644 --- a/orchestrator-v2/pnpm-workspace.yaml +++ b/orchestrator-v2/pnpm-workspace.yaml @@ -1,5 +1,6 @@ packages: - "packages/*" + - "examples/*" overrides: vite: npm:@voidzero-dev/vite-plus-core@latest From 9020d17a2877de147df826f4072cfe5561de9ff8 Mon Sep 17 00:00:00 2001 From: Matthew Wright Date: Sat, 4 Jul 2026 11:09:54 -0500 Subject: [PATCH 3/4] address PR review: fail-closed capabilities, sendRpcError unwrap, test layout - I1: treat null (unadvertised) capabilities as "no capabilities" rather than "handles anything" (fail-closed) per review; stub runners now advertise, and peer-registry.spec asserts an unadvertised runner is not selected. - rpc-router: fold the error-message unwrap into sendRpcError and drop the standalone errorMessage helper. - heartbeat: pass the capabilities array directly instead of cloning it. - peer-registry.spec: move the Context type and helpers below the describe block. - lessons: note the fail-closed capability behaviour. Co-Authored-By: Claude Opus 4.8 (1M context) --- orchestrator-v2/docs/experiment/lessons.md | 2 +- .../packages/orchestrator/src/dispatcher.ts | 3 +- .../orchestrator/src/peer-registry.spec.ts | 39 ++++++++++--------- .../orchestrator/src/peer-registry.ts | 13 ++++--- .../packages/orchestrator/src/rpc-router.ts | 8 +--- .../packages/orchestrator/src/test-helpers.ts | 9 ++++- .../packages/runner/src/heartbeat.ts | 4 +- 7 files changed, 42 insertions(+), 36 deletions(-) diff --git a/orchestrator-v2/docs/experiment/lessons.md b/orchestrator-v2/docs/experiment/lessons.md index 4d37d21..04ccff7 100644 --- a/orchestrator-v2/docs/experiment/lessons.md +++ b/orchestrator-v2/docs/experiment/lessons.md @@ -13,7 +13,7 @@ detailed **[gap report](./gap-report.md)**. | **Route A — telemetry propagation** | Telemetry now flows engine → `ScriptResult` → RPC → `TaskSource` as a terminal outcome: `task.complete`/`task.fail` carry it, `ResultHandler` reads+forwards it, `TaskSource.completeTask/failTask` widened with `telemetry?`, `ExecutionStats` unified in `interfaces-task-source`. | Live demo + 40→43 tests | | **G1 — failed-run session persistence** | `run-task-agent.ts` persists `sessionId` _before_ the success/failure branch, so a failed run is resumable; returns telemetry on failure too. | Live demo | | **I3 — task-source callback resilience** | A throwing `completeTask`/`failTask`/`pauseTask`/`setState` no longer wedges the peer (leaked slot) or escapes as an unhandled rejection. New `settle()` + `recordBestEffort()` guards; `.catch` net on `route()`. | New `orchestrator.spec` regression test + live | -| **I1 — capability-aware routing** | Runners advertise `capabilityKey(agentType, agentName)` in the heartbeat; the orchestrator only dispatches a task to a peer that can run it. Backward compatible (unadvertised = capable-of-anything). | New `peer-registry.spec` + live | +| **I1 — capability-aware routing** | Runners advertise `capabilityKey(agentType, agentName)` in the heartbeat; the orchestrator only dispatches a task to a peer that can run it. All runners advertise; an unadvertised runner is treated as having no capabilities (fail-closed). | New `peer-registry.spec` + live | | **Packaging fix** | `@bifrost-ai/orchestrator` `./test-helpers` export pointed at `src` (dead in the tarball); fixed via `publishConfig.exports` + a build entry. | `pnpm pack` verified | | **Quality** | Two `/simplify` passes: co-located `isExecutionStats`, `errorMessage`/`isStringArray`/`recordBestEffort` helpers, DRY'd terminal handlers. | build + lint clean | diff --git a/orchestrator-v2/packages/orchestrator/src/dispatcher.ts b/orchestrator-v2/packages/orchestrator/src/dispatcher.ts index 7cfc757..4c71168 100644 --- a/orchestrator-v2/packages/orchestrator/src/dispatcher.ts +++ b/orchestrator-v2/packages/orchestrator/src/dispatcher.ts @@ -31,7 +31,8 @@ export function sendRpcResponse(peer: ConnectedPeer, id: string, result: unknown peer.send(payload); } -export function sendRpcError(peer: ConnectedPeer, id: string, code: string, message: string): void { +export function sendRpcError(peer: ConnectedPeer, id: string, code: string, error: unknown): void { + const message = error instanceof Error ? error.message : String(error); const payload: FramePayload = { kind: "rpc.response", id, diff --git a/orchestrator-v2/packages/orchestrator/src/peer-registry.spec.ts b/orchestrator-v2/packages/orchestrator/src/peer-registry.spec.ts index 2df4c81..0bf3cea 100644 --- a/orchestrator-v2/packages/orchestrator/src/peer-registry.spec.ts +++ b/orchestrator-v2/packages/orchestrator/src/peer-registry.spec.ts @@ -5,6 +5,20 @@ import type { ConnectedPeer, FramePayload } from "@bifrost-ai/protocol"; import { PeerRegistry } from "./peer-registry.js"; +describe("PeerRegistry capability routing (regression: I1)", () => { + test("routes to the peer that advertises the required capability", { + given: { two_runners_with_different_capabilities }, + when: { selecting_a_peer_for_the_special_agent }, + then: { the_specialist_is_selected }, + }); + + test("does not select a runner that has not advertised the capability", { + given: { a_runner_that_advertises_no_capabilities }, + when: { selecting_a_peer_for_the_special_agent }, + then: { no_peer_is_selected }, + }); +}); + type Context = { registry: PeerRegistry; generic: ConnectedPeer; @@ -26,20 +40,6 @@ const heartbeat = (runnerId: string, capabilities?: string[]): FramePayload => ( ...(capabilities !== undefined ? { capabilities } : {}), }); -describe("PeerRegistry capability routing (regression: I1)", () => { - test("routes to the peer that advertises the required capability", { - given: { two_runners_with_different_capabilities }, - when: { selecting_a_peer_for_the_special_agent }, - then: { the_specialist_is_selected }, - }); - - test("treats a runner that advertised nothing as capable of anything", { - given: { a_runner_that_advertises_no_capabilities }, - when: { selecting_a_peer_for_the_special_agent }, - then: { the_bare_runner_is_selected }, - }); -}); - function two_runners_with_different_capabilities(this: Context) { this.registry = new PeerRegistry({ heartbeatTimeoutMs: 10_000, maxInFlightPerPeer: 1 }); this.generic = fakePeer("generic"); @@ -69,12 +69,13 @@ function selecting_a_peer_for_the_special_agent(this: Context) { } function the_specialist_is_selected(this: Context) { - // The generic runner is first in insertion order but lacks "special", so it is - // skipped in favour of the capable specialist (before the fix, the generic runner - // was chosen and the task would have failed). + // generic is first in insertion order but lacks "special", so it is skipped in + // favour of the capable specialist. expect(this.selected).toBe(this.specialist); } -function the_bare_runner_is_selected(this: Context) { - expect(this.selected).toBe(this.bare); +function no_peer_is_selected(this: Context) { + // A runner that has not advertised is treated as having no capabilities (fail-closed), + // so it is not selected for a required capability. + expect(this.selected).toBeUndefined(); } diff --git a/orchestrator-v2/packages/orchestrator/src/peer-registry.ts b/orchestrator-v2/packages/orchestrator/src/peer-registry.ts index 30c99b0..6a43a73 100644 --- a/orchestrator-v2/packages/orchestrator/src/peer-registry.ts +++ b/orchestrator-v2/packages/orchestrator/src/peer-registry.ts @@ -5,8 +5,8 @@ type PeerState = { runnerId: string | null; lastSeen: number; inFlight: number; - // Advertised capability keys, or null if the runner has not advertised any - // (treated as "can handle anything" for backward compatibility). + // Advertised capability keys, or null if the runner has not advertised yet + // (treated as having no capabilities — all runners are expected to advertise). capabilities: Set | null; }; @@ -133,10 +133,13 @@ export class PeerRegistry { } function canHandle(state: PeerState, requiredCapability: string | undefined): boolean { - // No requirement, or a runner that hasn't advertised its capabilities, is treated as - // able to handle the task (keeps non-advertising runners working as before). - if (requiredCapability === undefined || state.capabilities === null) { + if (requiredCapability === undefined) { return true; } + // A runner that has not advertised its capabilities is treated as having none — all + // runners are expected to advertise. + if (state.capabilities === null) { + return false; + } return state.capabilities.has(requiredCapability); } diff --git a/orchestrator-v2/packages/orchestrator/src/rpc-router.ts b/orchestrator-v2/packages/orchestrator/src/rpc-router.ts index 52eb7ad..8398c82 100644 --- a/orchestrator-v2/packages/orchestrator/src/rpc-router.ts +++ b/orchestrator-v2/packages/orchestrator/src/rpc-router.ts @@ -66,7 +66,7 @@ export class RpcRouter { await this.taskSource.setState(parsed.taskId, parsed.taskState); sendRpcResponse(peer, requestId, { ok: true }); } catch (error) { - sendRpcError(peer, requestId, "TASK_SOURCE_ERROR", errorMessage(error)); + sendRpcError(peer, requestId, "TASK_SOURCE_ERROR", error); } } @@ -85,7 +85,7 @@ export class RpcRouter { const result = await this.scheduler.call(parsed.method, parsed.args); sendRpcResponse(peer, requestId, result); } catch (error) { - sendRpcError(peer, requestId, "SCHEDULER_ERROR", errorMessage(error)); + sendRpcError(peer, requestId, "SCHEDULER_ERROR", error); } } } @@ -119,7 +119,3 @@ function readSchedulerParams(params: unknown): { method: string; args: unknown } } return { method: record.method, args: record.args }; } - -function errorMessage(error: unknown): string { - return error instanceof Error ? error.message : String(error); -} diff --git a/orchestrator-v2/packages/orchestrator/src/test-helpers.ts b/orchestrator-v2/packages/orchestrator/src/test-helpers.ts index 65a0efc..52bf92c 100644 --- a/orchestrator-v2/packages/orchestrator/src/test-helpers.ts +++ b/orchestrator-v2/packages/orchestrator/src/test-helpers.ts @@ -1,6 +1,6 @@ import type { Task, TaskSource } from "@bifrost-ai/interfaces-task-source"; import type { FramePayload, PeerIdentity, RunnerPeer } from "@bifrost-ai/protocol"; -import { createRunnerPeer, generateKeyPair } from "@bifrost-ai/protocol"; +import { capabilityKey, createRunnerPeer, generateKeyPair } from "@bifrost-ai/protocol"; import { runOrchestrator } from "./orchestrator.js"; import type { Scheduler } from "./types.js"; @@ -159,7 +159,12 @@ export async function connectStubRunner(options: { }, ); - runner.send({ kind: "heartbeat", runnerId: options.runnerIdentity.keyId }); + // Advertise the sampleTask capability so the (fail-closed) router will dispatch to this stub. + runner.send({ + kind: "heartbeat", + runnerId: options.runnerIdentity.keyId, + capabilities: [capabilityKey("script", "echo")], + }); return runner; } diff --git a/orchestrator-v2/packages/runner/src/heartbeat.ts b/orchestrator-v2/packages/runner/src/heartbeat.ts index b8c111d..26aff99 100644 --- a/orchestrator-v2/packages/runner/src/heartbeat.ts +++ b/orchestrator-v2/packages/runner/src/heartbeat.ts @@ -11,10 +11,10 @@ export function startHeartbeat( peer: RunnerPeer, identity: PeerIdentity, intervalMs = DEFAULT_HEARTBEAT_INTERVAL_MS, - capabilities: readonly string[] = [], + capabilities: string[] = [], ): HeartbeatHandle { const send = () => { - peer.send({ kind: "heartbeat", runnerId: identity.keyId, capabilities: [...capabilities] }); + peer.send({ kind: "heartbeat", runnerId: identity.keyId, capabilities }); }; send(); From 72fac71be0053406d353ac9ef689ee315156f53e Mon Sep 17 00:00:00 2001 From: Matthew Wright Date: Sat, 4 Jul 2026 11:09:54 -0500 Subject: [PATCH 4/4] format docs and example oxfmt pass over pre-existing files flagged by vp check. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../docs/experiment/dogfood-runs.md | 34 ++++---- orchestrator-v2/docs/experiment/gap-report.md | 81 ++++++++++--------- .../examples/claude-engine/package.json | 4 +- .../examples/claude-engine/run.mjs | 49 +++++++++-- 4 files changed, 105 insertions(+), 63 deletions(-) diff --git a/orchestrator-v2/docs/experiment/dogfood-runs.md b/orchestrator-v2/docs/experiment/dogfood-runs.md index 6ee774a..a1144a2 100644 --- a/orchestrator-v2/docs/experiment/dogfood-runs.md +++ b/orchestrator-v2/docs/experiment/dogfood-runs.md @@ -22,10 +22,10 @@ the session-id round-trip. **Result:** both tasks completed; real telemetry reached the task source; real Claude `session_id`s round-tripped via `setState`. -| task | out tokens | turns | cost | -|---|--:|--:|--:| -| dog-1 (`"reply DOGFOOD"`) | 61 | 1 | $0.0227 | -| dog-2 (`"describe an orchestrator"`) | 128 | 1 | $0.0153 | +| task | out tokens | turns | cost | +| ------------------------------------ | ---------: | ----: | ------: | +| dog-1 (`"reply DOGFOOD"`) | 61 | 1 | $0.0227 | +| dog-2 (`"describe an orchestrator"`) | 128 | 1 | $0.0153 | **Total: $0.038.** Lesson: the untested onboarding path worked first try. @@ -43,16 +43,16 @@ aggregate step could read the summaries. **Result — the system wrote release notes for its own new features:** -> *"Task completion now sends execution telemetry to the backend, providing detailed performance +> _"Task completion now sends execution telemetry to the backend, providing detailed performance > metrics and better observability. Runners can validate required capabilities, ensuring tasks are -> only assigned to runners equipped to handle them."* +> only assigned to runners equipped to handle them."_ -| task | out tokens | cost | -|---|--:|--:| -| sum-1 | 235 | $0.0237 | -| sum-2 | 316 | $0.0241 | -| release-notes | 384 | $0.0167 | -| **total** | **935** | **$0.0645** | +| task | out tokens | cost | +| ------------- | ---------: | ----------: | +| sum-1 | 235 | $0.0237 | +| sum-2 | 316 | $0.0241 | +| release-notes | 384 | $0.0167 | +| **total** | **935** | **$0.0645** | **Lesson (new gap E3):** chaining only worked because the **engine persisted each output via `setState`**. `task.complete` carries telemetry, not the task's `message` — there is no built-in @@ -66,7 +66,7 @@ the task source because the Workflow Agent (`agent-4-workflow`) doesn't exist ye **Setup:** three `analyze` tasks fanned out to an **analyzer** runner (haiku, concurrent); a **synthesizer** runner (a **sonnet** engine) combined them; a **critic** runner (haiku) reviewed the result. Capability-routed across **three** specialized runners with **heterogeneous engines**. One -analyze task (`an-routing`) was flaky — it failed *after* doing its work, and the task source +analyze task (`an-routing`) was flaky — it failed _after_ doing its work, and the task source **retried it with `--resume`**. **Exercised, in one run:** Route A (the telemetry below), I1 (3-way capability routing), G1 (failed-run @@ -103,14 +103,14 @@ scratch. The receipts make the value concrete. ### The pipeline's output (it critiqued its own design) -**Synthesis (sonnet):** *"The orchestrator matches tasks to runners by required agent capability, +**Synthesis (sonnet):** _"The orchestrator matches tasks to runners by required agent capability, avoiding wasted attempts on incapable runners, while tracking per-task execution costs and resource usage to support operator budgeting… It isolates task-source callback failures so a hung or erroring -callback cannot crash or stall the orchestrator itself."* +callback cannot crash or stall the orchestrator itself."_ -**Critique (haiku):** *"The three concerns aren't truly independent — callback isolation is +**Critique (haiku):** _"The three concerns aren't truly independent — callback isolation is foundational reliability that enables cost tracking and matching to work correctly, and cost data -should feed back into matching decisions…"* — a genuinely sharp observation about the architecture. +should feed back into matching decisions…"_ — a genuinely sharp observation about the architecture. --- diff --git a/orchestrator-v2/docs/experiment/gap-report.md b/orchestrator-v2/docs/experiment/gap-report.md index 74db0e8..3c93fca 100644 --- a/orchestrator-v2/docs/experiment/gap-report.md +++ b/orchestrator-v2/docs/experiment/gap-report.md @@ -6,49 +6,51 @@ ## Design grounding (the 8-stage maturity model) -Two design records ground this codebase (`devbox-8-stages-of-ai-maturity.pdf`, `bifrost-orchestrator-design.pdf`). They define the model `orchestrator-v2` implements: the **Trust Evolution — 8 stages** in three waves, where **layers alternate procedural / non-procedural** because *"prompts are guidance; the guarantee comes from the harness — wherever a guarantee is required, the layer must be procedural."* +Two design records ground this codebase (`devbox-8-stages-of-ai-maturity.pdf`, `bifrost-orchestrator-design.pdf`). They define the model `orchestrator-v2` implements: the **Trust Evolution — 8 stages** in three waves, where **layers alternate procedural / non-procedural** because _"prompts are guidance; the guarantee comes from the harness — wherever a guarantee is required, the layer must be procedural."_ -| Stage | Name | Nature | In this repo | -|---|---|---|---| -| 3 | Task | graded | `agent-3-task` — one autonomous session, a single unit of work | -| 4 | Workflow | procedural | `agent-4-workflow` (**planned**) — sequences Stage-3s with gates + adversarial review; *the* hard layer | -| 5 | Delegate | non-procedural | LLM picks which Stage-3/4 to run; no tools, may skip, framework-enforced kill limits | -| 6 | Coordinate | procedural | parallel Stage-5s (worktrees) | -| 7 | Supervise | non-procedural | consumes **telemetry from every level**, intervenes | -| 8 | Orchestrate | procedural | structural changes to the harness itself | +| Stage | Name | Nature | In this repo | +| ----- | ----------- | -------------- | ------------------------------------------------------------------------------------------------------- | +| 3 | Task | graded | `agent-3-task` — one autonomous session, a single unit of work | +| 4 | Workflow | procedural | `agent-4-workflow` (**planned**) — sequences Stage-3s with gates + adversarial review; _the_ hard layer | +| 5 | Delegate | non-procedural | LLM picks which Stage-3/4 to run; no tools, may skip, framework-enforced kill limits | +| 6 | Coordinate | procedural | parallel Stage-5s (worktrees) | +| 7 | Supervise | non-procedural | consumes **telemetry from every level**, intervenes | +| 8 | Orchestrate | procedural | structural changes to the harness itself | **`orchestrator-v2` is the thin rebuild of Bifrost, the Stage-4 harness**, resting on two explicit premises: (1) **the iteration count moves into the task definition (default 1)**; (2) **agents may invoke a Stage-3 directly, without the grader loop.** Hence the engine is "dumb dispatch," agents are scripts, and the `TaskSource` owns readiness/dependencies/priority ("Bifrost is one task source; a Jira-JSON-blob backend is another"). Corollary: the `darker` tooling in this repo (`darker-challenger`/`-prober`/`-eye`, complexity/tier routing, TDD red/green) is a DevBox-lineage implementation of the same model — chiefly the Stage-4 adversarial-review layer. **How this reframes findings below:** + - **G2** is not a doc/code drift but design **premise #1** — the single `engine.execute` call is intentional; the doc describes the **v1** model that looped. Re-annotated in §1. - **Telemetry (Route A / E1, implemented)** is a **Stage-7 prerequisite** — but it lacks the branch/session attribution Stage 7 needs (new gap **T1**, below). -- **I1's blocking limitation** is the design's own open question (*"a Stage-4 must-finish blocks the framework; can-pause risks never resuming, pushing prioritization onto the higher levels"*) — a known tension, not an oversight. +- **I1's blocking limitation** is the design's own open question (_"a Stage-4 must-finish blocks the framework; can-pause risks never resuming, pushing prioritization onto the higher levels"_) — a known tension, not an oversight. - **`agent-4-workflow` unbuilt** (§6) is the hard procedural layer; the design holds **adversarial review mandatory** for it (the canonical DevBox failure: all tests green, DB fully mocked, migration missing, system dead). -**T1 (NEW, design-grounded) — telemetry has no branch/session attribution.** `ExecutionStats` carries `durationMs`/tokens/`totalCostUsd`/`numTurns` but nothing tying a run to a **work branch** or session. The design requires Stage-7 supervision telemetry to be *"session-aware and tied to a work branch — otherwise seven parallel red tasks all write to one log and you can't tell which maps to which branch."* Route A now propagates telemetry to the task source, but without attribution it can't feed supervision. · design/telemetry · grounded-in-design +**T1 (NEW, design-grounded) — telemetry has no branch/session attribution.** `ExecutionStats` carries `durationMs`/tokens/`totalCostUsd`/`numTurns` but nothing tying a run to a **work branch** or session. The design requires Stage-7 supervision telemetry to be _"session-aware and tied to a work branch — otherwise seven parallel red tasks all write to one log and you can't tell which maps to which branch."_ Route A now propagates telemetry to the task source, but without attribution it can't feed supervision. · design/telemetry · grounded-in-design ## What's solid (lead with this) + - **`docs/orchestrator.md`** and the **workspace/`docs` READMEs** are highly accurate — component names, option shapes, defaults, the RPC method table, `runner.yaml` keys, and every code example resolve against the code. - **`docs/protocol.md`** has **no correctness gaps** (only the two minor completeness items and one clarity note in §3); the crypto / canonicalization / trust-model claims all hold, and it agrees with the protocol README. - Config discovery, override precedence, heartbeat defaults, and error strings in the runner docs all match. ## Summary of gaps -| # | Area | Correctness | Clarity | Minor | -|---|------|:---:|:---:|:---:| -| 1 | Task Agent (`agent-3-task`) | 0 | 2 | 0 | -| 2 | Orchestrator | 0 | 1 | 1 | -| 3 | Protocol | 0 | 1 | 1 | -| 4 | Runner | 0 | 2 | 3 | -| 5 | Script tasks (`interfaces-task*`) | 2 | 1 | 2 | -| 6 | Workflow Agent (`agent-4-workflow`) | — planned (readiness checklist) — | | | -| 7 | READMEs | 0 | 1 | 1 | +| # | Area | Correctness | Clarity | Minor | +| --- | ----------------------------------- | :-------------------------------: | :-----: | :---: | +| 1 | Task Agent (`agent-3-task`) | 0 | 2 | 0 | +| 2 | Orchestrator | 0 | 1 | 1 | +| 3 | Protocol | 0 | 1 | 1 | +| 4 | Runner | 0 | 2 | 3 | +| 5 | Script tasks (`interfaces-task*`) | 2 | 1 | 2 | +| 6 | Workflow Agent (`agent-4-workflow`) | — planned (readiness checklist) — | | | +| 7 | READMEs | 0 | 1 | 1 | -**2 correctness-level gaps, both in §5:** G12 (README `Task` type has the wrong field names) and G13 (`ScriptContext` doc omits the registries scripts actually use). §6 is a **planned, unbuilt** agent — a readiness checklist, not bugs. §8 lists implementation cleanup that is *not* a documentation issue. +**2 correctness-level gaps, both in §5:** G12 (README `Task` type has the wrong field names) and G13 (`ScriptContext` doc omits the registries scripts actually use). §6 is a **planned, unbuilt** agent — a readiness checklist, not bugs. §8 lists implementation cleanup that is _not_ a documentation issue. > **Update — implemented (Route A):** **E1** (telemetry propagation) and **G1** (failed-run session persistence) have been fixed end-to-end. Telemetry now travels as a terminal outcome to the task source, and failed runs persist their `sessionId`. Build clean, all **40 tests pass**, `vp lint` clean. Details in the Empirical validation section and §1. Changes are uncommitted. -**Legend** — *Type:* `unimplemented` · `behavior-diverges` · `doc-stale` · `naming-mismatch` · `missing-from-doc`. *Confidence:* `confirmed` (verified in code) · `needs-check` (interpretation-dependent). +**Legend** — _Type:_ `unimplemented` · `behavior-diverges` · `doc-stale` · `naming-mismatch` · `missing-from-doc`. _Confidence:_ `confirmed` (verified in code) · `needs-check` (interpretation-dependent). --- @@ -57,10 +59,10 @@ Two design records ground this codebase (`devbox-8-stages-of-ai-maturity.pdf`, ` Beyond static reading, I stood up a **real orchestrator + real runner + `TestEngine`** and dispatched two Task Agent tasks through the actual signed-WebSocket path — a path the test suite does **not** cover (`runner.spec` only exercises stub `"script"` agents, never `agent-3-task`). Findings: - **The happy path works end-to-end** ✅ — signed handshake, heartbeat-gated availability, dispatch, task-agent execution against the engine, `setState`, `task.complete`. -- **G1 confirmed live → ✅ FIXED.** *Originally:* the successful task persisted its `sessionId`; the task whose engine returned `success:false` was failed **without** persisting the returned `sessionId`. *Fix:* `run-task-agent.ts` now persists `sessionId` **before** branching on outcome, so a failed run is resumable — re-verified live (the failed task now persists its session). -- **E1 (NEW) — telemetry is computed but never recorded → ✅ IMPLEMENTED (Route A).** *Originally:* `runTaskAgent` returned `telemetry` in its `ScriptResult`, but the runner's dispatch handler forwarded only `{ taskId }` on `task.complete` (`dispatch-handler.ts:65`); tokens/cost/`numTurns` were dropped before reaching the task source. *Fix:* telemetry now propagates as a **terminal outcome** — both `task.complete` and `task.fail` carry it (`dispatch-handler.ts`), the orchestrator's `ResultHandler` validates + forwards it (`readTelemetry`), and the `TaskSource` contract is widened to `completeTask(taskId, telemetry?)` / `failTask(taskId, error, telemetry?)` (`interfaces-task-source/src/types.ts`), with `ExecutionStats` unified as one canonical type in `interfaces-task-source`. Re-verified live: the source now receives telemetry on **both** success (`$0.005`) and failure (`$0.02`). · behavior-diverges · **resolved** +- **G1 confirmed live → ✅ FIXED.** _Originally:_ the successful task persisted its `sessionId`; the task whose engine returned `success:false` was failed **without** persisting the returned `sessionId`. _Fix:_ `run-task-agent.ts` now persists `sessionId` **before** branching on outcome, so a failed run is resumable — re-verified live (the failed task now persists its session). +- **E1 (NEW) — telemetry is computed but never recorded → ✅ IMPLEMENTED (Route A).** _Originally:_ `runTaskAgent` returned `telemetry` in its `ScriptResult`, but the runner's dispatch handler forwarded only `{ taskId }` on `task.complete` (`dispatch-handler.ts:65`); tokens/cost/`numTurns` were dropped before reaching the task source. _Fix:_ telemetry now propagates as a **terminal outcome** — both `task.complete` and `task.fail` carry it (`dispatch-handler.ts`), the orchestrator's `ResultHandler` validates + forwards it (`readTelemetry`), and the `TaskSource` contract is widened to `completeTask(taskId, telemetry?)` / `failTask(taskId, error, telemetry?)` (`interfaces-task-source/src/types.ts`), with `ExecutionStats` unified as one canonical type in `interfaces-task-source`. Re-verified live: the source now receives telemetry on **both** success (`$0.005`) and failure (`$0.02`). · behavior-diverges · **resolved** - **E2 (NEW) — `vp run ready` fails on a clean checkout.** From a fresh install (no `dist`), `vp run -r test` fails because cross-package `@bifrost-ai/*` imports resolve to unbuilt `dist/index.mjs` (e.g. `orchestrator.spec` can't resolve `@bifrost-ai/protocol`). The root `ready` script runs `test` **before** `build`, so `vp run ready` on a fresh clone fails at the test step. Running `vp run -r build` first makes all **40 tests pass**. · build/workflow · confirmed -- **E3 (NEW) — a task's output/message is not propagated; only telemetry is.** `task.complete` carries `{ taskId, telemetry }` and `TaskSource.completeTask(taskId, telemetry?)` records telemetry, but the task's `message` (the actual work output) is dropped — exactly as telemetry was before Route A. Surfaced by the **DAG dogfood**: chaining one task's result into a downstream task only worked because the **engine persisted its output via `ctx.setState`**; there is no built-in channel for a task's output to reach the orchestrator/source or a downstream step. This blocks output-chaining workflows and is the same shape as E1 (now fixed) but for the message. It reinforces the §6 workflow-readiness gaps: a real workflow must read child *outcomes*, not just telemetry. *(Candidate follow-up: a Route A-style extension carrying `message` on the terminal outcome, or a documented "engines persist output via setState" contract.)* · behavior · confirmed +- **E3 (NEW) — a task's output/message is not propagated; only telemetry is.** `task.complete` carries `{ taskId, telemetry }` and `TaskSource.completeTask(taskId, telemetry?)` records telemetry, but the task's `message` (the actual work output) is dropped — exactly as telemetry was before Route A. Surfaced by the **DAG dogfood**: chaining one task's result into a downstream task only worked because the **engine persisted its output via `ctx.setState`**; there is no built-in channel for a task's output to reach the orchestrator/source or a downstream step. This blocks output-chaining workflows and is the same shape as E1 (now fixed) but for the message. It reinforces the §6 workflow-readiness gaps: a real workflow must read child _outcomes_, not just telemetry. _(Candidate follow-up: a Route A-style extension carrying `message` on the terminal outcome, or a documented "engines persist output via setState" contract.)_ · behavior · confirmed --- @@ -68,17 +70,17 @@ Beyond static reading, I stood up a **real orchestrator + real runner + `TestEng A live multi-runner harness (real orchestrator + real runners + script agents) exercising paths the unit specs don't cover. -**Works correctly (no gap):** multi-runner **concurrency + back-pressure** (4 tasks / 2 runners / `maxInFlight=1` → 2 concurrent batches, ~624ms); **`maxInFlightPerPeer > 1`** (3 tasks concurrent on one runner, ~313ms); the **`paused`** terminal path; **disconnect cleanup** (orphaned task failed + slot freed); and **recovery by reconnection** — a fresh `Runner` (same identity) re-dials and resumes taking work. *(There is no **auto**-reconnect: a dropped `Runner` instance stays dropped; the app must construct a new one.)* +**Works correctly (no gap):** multi-runner **concurrency + back-pressure** (4 tasks / 2 runners / `maxInFlight=1` → 2 concurrent batches, ~624ms); **`maxInFlightPerPeer > 1`** (3 tasks concurrent on one runner, ~313ms); the **`paused`** terminal path; **disconnect cleanup** (orphaned task failed + slot freed); and **recovery by reconnection** — a fresh `Runner` (same identity) re-dials and resumes taking work. _(There is no **auto**-reconnect: a dropped `Runner` instance stays dropped; the app must construct a new one.)_ -**I1 — No capability-aware routing; a runner's rejection is terminal.** The orchestrator dispatches each task to the *first available* peer regardless of which agents that peer has registered (`peer-registry.ts` `getAvailablePeer`). If the chosen runner hasn't registered the task's agent, its dispatch handler replies `accepted:false` ("Unknown agent"), and `DispatchAckHandler.reject` **fails the task** (`dispatch-ack-handler.ts`) instead of re-dispatching to a capable peer. Demonstrated live: with **both** runners connected + available and only runner B registering the `special` agent, a task needing `special` was dispatched to runner A and **permanently failed** — `failTask(s1, "Unknown agent: special")` — runner B never got it. The design implicitly assumes **homogeneous runners** (every runner registers the same agents), but the per-runner `registerAgent` API permits heterogeneity and nothing documents or enforces the assumption. · behavior/architecture · confirmed +**I1 — No capability-aware routing; a runner's rejection is terminal.** The orchestrator dispatches each task to the _first available_ peer regardless of which agents that peer has registered (`peer-registry.ts` `getAvailablePeer`). If the chosen runner hasn't registered the task's agent, its dispatch handler replies `accepted:false` ("Unknown agent"), and `DispatchAckHandler.reject` **fails the task** (`dispatch-ack-handler.ts`) instead of re-dispatching to a capable peer. Demonstrated live: with **both** runners connected + available and only runner B registering the `special` agent, a task needing `special` was dispatched to runner A and **permanently failed** — `failTask(s1, "Unknown agent: special")` — runner B never got it. The design implicitly assumes **homogeneous runners** (every runner registers the same agents), but the per-runner `registerAgent` API permits heterogeneity and nothing documents or enforces the assumption. · behavior/architecture · confirmed -> **✅ Fixed (this session) — capability-aware routing.** Runners now advertise the `capabilityKey(agentType, agentName)` of every registered agent in their heartbeat (optional field; `protocol` owns the canonical key format). The orchestrator's `PeerRegistry` records each peer's advertised capability set, and `getAvailablePeer`/`waitForAvailablePeer` filter by the task's required capability — so a task only dispatches to a peer that can run it. **Backward compatible:** a runner that advertises nothing (e.g. the stub runners used in tests) is treated as capable-of-anything. Verified live (a task needing `special` now routes to the runner that has it instead of failing on an incapable one), guarded by a new `peer-registry.spec.ts` (2 tests). 43/43 tests pass. *Known limitation: the dispatch loop is still sequential, so a task requiring a capability that **no** connected runner advertises now **waits** (blocking later tasks) rather than fail-fast — a concurrent dispatch loop or a route-timeout is a separate follow-up.* +> **✅ Fixed (this session) — capability-aware routing.** Runners now advertise the `capabilityKey(agentType, agentName)` of every registered agent in their heartbeat (optional field; `protocol` owns the canonical key format). The orchestrator's `PeerRegistry` records each peer's advertised capability set, and `getAvailablePeer`/`waitForAvailablePeer` filter by the task's required capability — so a task only dispatches to a peer that can run it. **Backward compatible:** a runner that advertises nothing (e.g. the stub runners used in tests) is treated as capable-of-anything. Verified live (a task needing `special` now routes to the runner that has it instead of failing on an incapable one), guarded by a new `peer-registry.spec.ts` (2 tests). 43/43 tests pass. _Known limitation: the dispatch loop is still sequential, so a task requiring a capability that **no** connected runner advertises now **waits** (blocking later tasks) rather than fail-fast — a concurrent dispatch loop or a route-timeout is a separate follow-up._ **I2 — A mid-task runner disconnect is a terminal task failure, with no retry.** When a runner drops mid-task, the orchestrator fails the orphaned task (`failTask(taskId, "Runner disconnected")`, `result-handler.ts` `handleDisconnect`) and does not re-dispatch it. Confirmed live. Consistent with the "thin orchestrator" design (retries are meant to live at the workflow/child level per `agent-4-workflow.md`), **but that layer doesn't exist yet** (§6) — so today a transient runner blip permanently fails a bare task with no recovery anywhere in the system. · behavior/resilience · confirmed -**I3 — Task-source callbacks have no error handling; a single throw wedges the peer, leaks its slot, and escapes as an unhandled rejection.** The orchestrator's RPC handlers `await` the task source and only *then* release the slot / respond — `handleComplete` runs `await taskSource.completeTask(...)` **before** `registry.markTerminal(...)` and `sendRpcResponse(...)` (`result-handler.ts`). A real source does DB/network I/O, so a rejecting callback is expected eventually — and when it happens: (a) `markTerminal` is skipped → the peer's in-flight slot is **leaked**, so that peer is permanently unavailable and every later task routed to it hangs on `waitForAvailablePeer`; (b) `sendRpcResponse` is skipped → the runner's terminal RPC never resolves; (c) the rejection is **unhandled** — `RpcRouter.handle` invokes `route(...)` with no `await`/`.catch` (`rpc-router.ts`), so it escapes as an unhandled promise rejection (process-fatal on default Node). Demonstrated live: a `completeTask` that threw on `f1` left the peer wedged, `f2` never dispatched, the orchestrator's `done` never resolved (timed out), and exactly one unhandled rejection surfaced. Same shape for `failTask`/`pauseTask` (slot leak) and `taskSource.setState` (hung RPC + unhandled rejection; no slot leak since it isn't terminal). · behavior/robustness · confirmed +**I3 — Task-source callbacks have no error handling; a single throw wedges the peer, leaks its slot, and escapes as an unhandled rejection.** The orchestrator's RPC handlers `await` the task source and only _then_ release the slot / respond — `handleComplete` runs `await taskSource.completeTask(...)` **before** `registry.markTerminal(...)` and `sendRpcResponse(...)` (`result-handler.ts`). A real source does DB/network I/O, so a rejecting callback is expected eventually — and when it happens: (a) `markTerminal` is skipped → the peer's in-flight slot is **leaked**, so that peer is permanently unavailable and every later task routed to it hangs on `waitForAvailablePeer`; (b) `sendRpcResponse` is skipped → the runner's terminal RPC never resolves; (c) the rejection is **unhandled** — `RpcRouter.handle` invokes `route(...)` with no `await`/`.catch` (`rpc-router.ts`), so it escapes as an unhandled promise rejection (process-fatal on default Node). Demonstrated live: a `completeTask` that threw on `f1` left the peer wedged, `f2` never dispatched, the orchestrator's `done` never resolved (timed out), and exactly one unhandled rejection surfaced. Same shape for `failTask`/`pauseTask` (slot leak) and `taskSource.setState` (hung RPC + unhandled rejection; no slot leak since it isn't terminal). · behavior/robustness · confirmed -> **✅ Fixed (this session).** The orchestrator now guards every task-source/scheduler callback. A `settle()` helper in `result-handler.ts` runs each terminal source call in `try/catch/finally`, so slot release (`markTerminal`) and the runner response **always** happen even if the source throws; `handleDisconnect`, `dispatch-ack-handler.reject`, and `rpc-router`'s `setState`/`scheduler.call` are likewise guarded; and `RpcRouter.handle` adds a `.catch` net so a rejecting handler can never escape as an unhandled rejection. Source failures are surfaced via `console.error` (not silently swallowed). Verified live — a throwing `completeTask` no longer wedges the peer (`f2` still dispatches, `done` resolves, **0** unhandled rejections) — and guarded by a new regression test (`orchestrator.spec.ts` → "does not wedge the peer when a task source callback throws"). 41/41 tests pass. *Follow-up: a real `onError` option would beat `console.error` for a library.* +> **✅ Fixed (this session).** The orchestrator now guards every task-source/scheduler callback. A `settle()` helper in `result-handler.ts` runs each terminal source call in `try/catch/finally`, so slot release (`markTerminal`) and the runner response **always** happen even if the source throws; `handleDisconnect`, `dispatch-ack-handler.reject`, and `rpc-router`'s `setState`/`scheduler.call` are likewise guarded; and `RpcRouter.handle` adds a `.catch` net so a rejecting handler can never escape as an unhandled rejection. Source failures are surfaced via `console.error` (not silently swallowed). Verified live — a throwing `completeTask` no longer wedges the peer (`f2` still dispatches, `done` resolves, **0** unhandled rejections) — and guarded by a new regression test (`orchestrator.spec.ts` → "does not wedge the peer when a task source callback throws"). 41/41 tests pass. _Follow-up: a real `onError` option would beat `console.error` for a library._ --- @@ -89,12 +91,13 @@ The implementation is 4 small files. The core (`run-task-agent.ts`) makes a **si ### Clarity **G1 — Session/telemetry checkpointing is engine-dependent, not agent-provided; the agent also drops a failed-result `sessionId`.** + - **Doc says:** §Running — the agent "may checkpoint progress (session ID, partial telemetry) so that if something goes wrong mid-flight, a retry can pick up where it left off"; §3 — a parent retries "with the same session id." -- **Code does:** `runTaskAgent` persists `sessionId` **only after success** (`run-task-agent.ts:53-58`); both failure exits (`:41-44` catch, `:46-51` `!success`) return without `setState` and ignore any `sessionId`/`stats` on the returned `EngineResult`. So resume after a failure is possible **only if the engine checkpoints itself** via the `setState` it's handed in `EngineContext` (`engine/src/types.ts:28`, wired at `run-task-agent.ts:37`). Accurate framing: *the doc attributes checkpointing to the agent, but the agent contributes none beyond post-success persistence and discards a `sessionId` returned on a failed result — mid-run/after-failure resume depends entirely on engine-side checkpointing.* +- **Code does:** `runTaskAgent` persists `sessionId` **only after success** (`run-task-agent.ts:53-58`); both failure exits (`:41-44` catch, `:46-51` `!success`) return without `setState` and ignore any `sessionId`/`stats` on the returned `EngineResult`. So resume after a failure is possible **only if the engine checkpoints itself** via the `setState` it's handed in `EngineContext` (`engine/src/types.ts:28`, wired at `run-task-agent.ts:37`). Accurate framing: _the doc attributes checkpointing to the agent, but the agent contributes none beyond post-success persistence and discards a `sessionId` returned on a failed result — mid-run/after-failure resume depends entirely on engine-side checkpointing._ - **Type:** behavior-diverges · **Confidence:** confirmed (**demonstrated in the live run — see Empirical validation**) -- **Status: ✅ Implemented (Route A).** `run-task-agent.ts` now persists `sessionId` *before* the success/failure branch (failed runs are resumable) and returns telemetry on the failure path; paired with E1's propagation, a failed run's partial telemetry reaches the task source. 40/40 tests pass. *(The broader §Running "checkpointing is the agent's job" framing remains a doc-wording question; the concrete failed-run data loss is fixed.)* +- **Status: ✅ Implemented (Route A).** `run-task-agent.ts` now persists `sessionId` _before_ the success/failure branch (failed runs are resumable) and returns telemetry on the failure path; paired with E1's propagation, a failed run's partial telemetry reaches the task source. 40/40 tests pass. _(The broader §Running "checkpointing is the agent's job" framing remains a doc-wording question; the concrete failed-run data loss is fixed.)_ -**G2 — The doc's "turn loop / maximum turn limit inside the agent" describes the pre-rebuild (v1) model.** §Running describes a 5-step turn loop "inside the agent" with a "configured maximum turn limit"; `runTaskAgent` calls `engine.execute()` exactly **once** (`run-task-agent.ts:29`) — no loop or turn-limit anywhere in `agent-3-task`. This is **intended**, not a divergence: the thin-rebuild premise moves the iteration count into the task definition (**default 1**) and keeps the engine "dumb" (see *Design grounding*). The v1 orchestrator *did* loop (`orchestrator/…/core/orchestrator.ts` `runEngineLoop`, `maxFollowUps=10`); the doc still describes that older shape. The fix is to the **doc**, not the code — and, if per-task iteration is ever wanted, it belongs in the task definition/`TaskSource`, not the agent. · doc-stale (describes v1) · confirmed +**G2 — The doc's "turn loop / maximum turn limit inside the agent" describes the pre-rebuild (v1) model.** §Running describes a 5-step turn loop "inside the agent" with a "configured maximum turn limit"; `runTaskAgent` calls `engine.execute()` exactly **once** (`run-task-agent.ts:29`) — no loop or turn-limit anywhere in `agent-3-task`. This is **intended**, not a divergence: the thin-rebuild premise moves the iteration count into the task definition (**default 1**) and keeps the engine "dumb" (see _Design grounding_). The v1 orchestrator _did_ loop (`orchestrator/…/core/orchestrator.ts` `runEngineLoop`, `maxFollowUps=10`); the doc still describes that older shape. The fix is to the **doc**, not the code — and, if per-task iteration is ever wanted, it belongs in the task definition/`TaskSource`, not the agent. · doc-stale (describes v1) · confirmed > **Undocumented contract (worth adding):** task agents register under agentType `"task"`, keyed by `agent.name` (`runner.ts:28`); dispatch resolves `agents.get(agentType).get(agentName)` (`dispatch-handler.ts:49`). A Task Agent task must be created with `agentType: "task"`, `agentName: ""`. @@ -108,7 +111,7 @@ The most accurate doc in the set — no correctness or behavior gaps. Only compl ### Clarity -**G3 — `orchestrator.md` doesn't enumerate the `taskSource.setState` RPC.** The doc lists `dispatch` + `task.complete`/`fail`/`pause` + `scheduler.call`, but omits `taskSource.setState`, which `RpcRouter` handles and proxies (`rpc-router.ts:308-310, 317-325`). **Note:** this is an *orchestrator-doc* completeness issue — `docs/protocol.md:95` already documents the method in its RPC table, so it is not a globally undocumented RPC. · missing-from-doc · confirmed +**G3 — `orchestrator.md` doesn't enumerate the `taskSource.setState` RPC.** The doc lists `dispatch` + `task.complete`/`fail`/`pause` + `scheduler.call`, but omits `taskSource.setState`, which `RpcRouter` handles and proxies (`rpc-router.ts:308-310, 317-325`). **Note:** this is an _orchestrator-doc_ completeness issue — `docs/protocol.md:95` already documents the method in its RPC table, so it is not a globally undocumented RPC. · missing-from-doc · confirmed ### Minor @@ -130,7 +133,7 @@ _(Dead code `isRpcResponse` is in §8.)_ **G6 — Ephemeral-port discovery is undocumented.** README says `port: 0` binds an ephemeral port but never says how to learn it: `OrchestratorPeer.address = { host, port }` (`types.ts:68`, `orchestrator.ts`). The returned peer shape is never spelled out. · missing-from-doc · confirmed -> **Checked — not a gap:** the `heartbeat` frame's "keep peer alive" purpose *is* realized — the runner sends heartbeats on a timer (`runner/src/heartbeat.ts:16`) and the orchestrator uses them for liveness (`peer-registry.ts:49,109`). Distributed across packages, not the protocol package. +> **Checked — not a gap:** the `heartbeat` frame's "keep peer alive" purpose _is_ realized — the runner sends heartbeats on a timer (`runner/src/heartbeat.ts:16`) and the orchestrator uses them for liveness (`peer-registry.ts:49,109`). Distributed across packages, not the protocol package. _(Unused/test-only exports — `signRawMaterial`, `loadTrustedPublicKey`, `SIGNING_ALGORITHM` — are in §8.)_ @@ -163,11 +166,13 @@ The execution-primitive docs have drifted from the current `Task`/`ScriptContext ### Correctness **G12 — The task-source README's `Task` type has the wrong field names.** + - **Doc says:** `Task = { id; scriptName; taskState; metadata }`, with a field table and prose "the runner uses `scriptName` to look up the script" (`interfaces-task-source/README.md:27-28, 36-37, 41, 100`). -- **Code does:** `Task = { taskId; agentType; agentName; taskState; metadata }` (`interfaces-task-source/src/types.ts:3-9`). No `id`, no `scriptName`; validation requires `["taskId","agentType","agentName"]` (`types.ts:19`), and the runner resolves by `agentType`+`agentName` (`dispatch-handler.ts:49`). Code written from this README fails `validateTask`. (The `TaskSource` *method* signatures in the same README are correct — only the `Task` type and the `scriptName` prose are stale.) +- **Code does:** `Task = { taskId; agentType; agentName; taskState; metadata }` (`interfaces-task-source/src/types.ts:3-9`). No `id`, no `scriptName`; validation requires `["taskId","agentType","agentName"]` (`types.ts:19`), and the runner resolves by `agentType`+`agentName` (`dispatch-handler.ts:49`). Code written from this README fails `validateTask`. (The `TaskSource` _method_ signatures in the same README are correct — only the `Task` type and the `scriptName` prose are stale.) - **Type:** naming-mismatch + behavior-diverges · **Confidence:** confirmed **G13 — `script-tasks.md`'s `ScriptContext` omits the registries scripts actually use.** + - **Doc says:** `ScriptContext = { taskState; readonly metadata; setState }`, presented as the contract (`docs/script-tasks.md:21-24`). - **Code does:** `ScriptContext` also has `taskId`, `agentType`, `agentName`, `readonly data: DataRegistry`, and `readonly agents: AgentRegistry` (`interfaces-task/src/types.ts:35-44`). `data` is load-bearing — the Task Agent reads `ctx.data.get(ENGINE_DATA_TYPE)` (`run-task-agent.ts:22`). The documented contract is a strict subset that would leave a reader unable to write an engine-backed agent. - **Type:** doc-stale / missing-from-doc · **Confidence:** confirmed @@ -186,7 +191,7 @@ The execution-primitive docs have drifted from the current `Task`/`ScriptContext ## 6. Workflow Agent — `docs/agent-4-workflow.md` (planned, no package) -**Status: ~0% implemented, honestly labeled "Planned (#39)"** — no `packages/agent-4-workflow`, and a repo-wide grep for `workflow`/`agent-4` finds only `docs/`. Its unbuilt state is *acknowledged*, not hidden. The value is a **readiness checklist**: the enabling mechanisms it leans on are missing or half-built. +**Status: ~0% implemented, honestly labeled "Planned (#39)"** — no `packages/agent-4-workflow`, and a repo-wide grep for `workflow`/`agent-4` finds only `docs/`. Its unbuilt state is _acknowledged_, not hidden. The value is a **readiness checklist**: the enabling mechanisms it leans on are missing or half-built. **G17 — TaskSource lacks every capability the "schedule" step needs.** The doc has the workflow "create all child tasks, wire dependency edges, promote the ready ones, register every child as a blocker" (`agent-4-workflow.md:41-47, 82-99, 180-199`). But `TaskSource` has only `watchTasks`/`completeTask`/`failTask`/`pauseTask`/`setState` (`interfaces-task-source/src/types.ts:11-17`) — no create-child / dependency / promote / blocker method anywhere. · unimplemented · confirmed (whether these belong on the interface or inside a concrete source is `needs-check`) @@ -204,7 +209,7 @@ Highly accurate — package table, code examples, the 5-method RPC table, `runne ### Clarity -**G20 — `ctx.data.get(type)` returns a *read-only* registry, but the README calls it `Registry`.** `README.md:58` says `get(type)` returns `Registry`; the actual return is `ReadonlyRegistry` (`get`/`has` only, `interfaces-task/src/types.ts:23`). `Registry` is a distinct mutable type with `register()` — the runner hands scripts a read-only view on purpose. Could mislead a reader into thinking a script can `.register()` into `ctx.data`. · naming-mismatch · confirmed +**G20 — `ctx.data.get(type)` returns a _read-only_ registry, but the README calls it `Registry`.** `README.md:58` says `get(type)` returns `Registry`; the actual return is `ReadonlyRegistry` (`get`/`has` only, `interfaces-task/src/types.ts:23`). `Registry` is a distinct mutable type with `register()` — the runner hands scripts a read-only view on purpose. Could mislead a reader into thinking a script can `.register()` into `ctx.data`. · naming-mismatch · confirmed ### Minor diff --git a/orchestrator-v2/examples/claude-engine/package.json b/orchestrator-v2/examples/claude-engine/package.json index 514fb1c..a4c8f05 100644 --- a/orchestrator-v2/examples/claude-engine/package.json +++ b/orchestrator-v2/examples/claude-engine/package.json @@ -7,9 +7,9 @@ "start": "node run.mjs" }, "dependencies": { + "@bifrost-ai/agent-3-task": "workspace:*", "@bifrost-ai/orchestrator": "workspace:*", - "@bifrost-ai/runner": "workspace:*", "@bifrost-ai/protocol": "workspace:*", - "@bifrost-ai/agent-3-task": "workspace:*" + "@bifrost-ai/runner": "workspace:*" } } diff --git a/orchestrator-v2/examples/claude-engine/run.mjs b/orchestrator-v2/examples/claude-engine/run.mjs index 3ba94b6..a9e2b5e 100644 --- a/orchestrator-v2/examples/claude-engine/run.mjs +++ b/orchestrator-v2/examples/claude-engine/run.mjs @@ -51,17 +51,45 @@ const claudeEngine = { }, }; } catch (error) { - return { success: false, skipFulfill: false, lastMessage: String(error?.message ?? error), stats: null }; + return { + success: false, + skipFulfill: false, + lastMessage: String(error?.message ?? error), + stats: null, + }; } }, }; // ── A minimal in-memory TaskSource that records telemetry ─────────────────── -const fmt = (t) => (t ? `{in:${t.inputTokens}, out:${t.outputTokens}, turns:${t.numTurns}, $${t.totalCostUsd.toFixed(4)}}` : "none"); +const fmt = (t) => + t + ? `{in:${t.inputTokens}, out:${t.outputTokens}, turns:${t.numTurns}, $${t.totalCostUsd.toFixed(4)}}` + : "none"; const workDir = await mkdtemp(join(tmpdir(), "claude-engine-example-")); const tasks = [ - { taskId: "t1", agentType: "task", agentName: "assistant", metadata: {}, taskState: { workingDir: workDir, engineName: "claude", instructions: "Reply with exactly one word, no punctuation: HELLO" } }, - { taskId: "t2", agentType: "task", agentName: "assistant", metadata: {}, taskState: { workingDir: workDir, engineName: "claude", instructions: "In one short sentence, describe what a task orchestrator does." } }, + { + taskId: "t1", + agentType: "task", + agentName: "assistant", + metadata: {}, + taskState: { + workingDir: workDir, + engineName: "claude", + instructions: "Reply with exactly one word, no punctuation: HELLO", + }, + }, + { + taskId: "t2", + agentType: "task", + agentName: "assistant", + metadata: {}, + taskState: { + workingDir: workDir, + engineName: "claude", + instructions: "In one short sentence, describe what a task orchestrator does.", + }, + }, ]; const source = { async *watchTasks() { @@ -87,14 +115,23 @@ const handle = await runOrchestrator({ { keyId: runnerIdentity.keyId, publicKeyPem: exportPublicKeyPem(runnerIdentity.publicKey) }, ]), taskSource: source, - scheduler: { async call() { return { ok: true }; } }, + scheduler: { + async call() { + return { ok: true }; + }, + }, port: 0, }); const { host, port } = handle.peer.address; log(`orchestrator listening on ws://${host}:${port}`); // Write a real runner.yaml and load the runner's keys/URL from it. -const indentPem = (pem) => pem.trimEnd().split("\n").map((l) => ` ${l}`).join("\n"); +const indentPem = (pem) => + pem + .trimEnd() + .split("\n") + .map((l) => ` ${l}`) + .join("\n"); const cfgDir = await mkdtemp(join(tmpdir(), "claude-engine-cfg-")); const configPath = join(cfgDir, "runner.yaml"); await writeFile(