diff --git a/package-lock.json b/package-lock.json index 23159024..3ff6dab6 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10,7 +10,9 @@ "dependencies": { "@agent-relay/cloud": "^9.1.6", "@agent-relay/factory": "^0.1.13", + "@agent-relay/fleet": "^9.1.6", "@agent-relay/harness-driver": "^9.1.6", + "@agent-relay/harnesses": "^9.1.6", "@agent-relay/integration-prompts": "^9.1.6", "@agent-relay/sdk": "^9.1.6", "@agentworkforce/deploy": "^4.0.2", @@ -34,6 +36,7 @@ "shiki": "^4.0.2", "unidiff": "^1.0.4", "use-stick-to-bottom": "^1.1.6", + "ws": "^8.21.0", "zod": "^3.25.76", "zustand": "^5.0.0" }, @@ -47,6 +50,7 @@ "@testing-library/react": "^16.3.2", "@types/react": "^19.0.0", "@types/react-dom": "^19.0.0", + "@types/ws": "^8.18.1", "@typescript-eslint/eslint-plugin": "^8.61.0", "@typescript-eslint/parser": "^8.61.0", "@vitejs/plugin-react": "^4.3.0", diff --git a/package.json b/package.json index 53b2d0ae..9d1ab4a0 100644 --- a/package.json +++ b/package.json @@ -37,7 +37,9 @@ "dependencies": { "@agent-relay/cloud": "^9.1.6", "@agent-relay/factory": "^0.1.13", + "@agent-relay/fleet": "^9.1.6", "@agent-relay/harness-driver": "^9.1.6", + "@agent-relay/harnesses": "^9.1.6", "@agent-relay/integration-prompts": "^9.1.6", "@agent-relay/sdk": "^9.1.6", "@agentworkforce/deploy": "^4.0.2", @@ -61,6 +63,7 @@ "shiki": "^4.0.2", "unidiff": "^1.0.4", "use-stick-to-bottom": "^1.1.6", + "ws": "^8.21.0", "zod": "^3.25.76", "zustand": "^5.0.0" }, @@ -74,6 +77,7 @@ "@testing-library/react": "^16.3.2", "@types/react": "^19.0.0", "@types/react-dom": "^19.0.0", + "@types/ws": "^8.18.1", "@typescript-eslint/eslint-plugin": "^8.61.0", "@typescript-eslint/parser": "^8.61.0", "@vitejs/plugin-react": "^4.3.0", diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index f4802376..147e8545 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -101,7 +101,8 @@ const mock = vi.hoisted(() => { } return { event_id: eventId, targets: target && !target.startsWith('#') ? [target] : [] } }), - brokerPid: 4242 + brokerPid: 4242, + baseUrl: 'http://127.0.0.1:4242' } return client } @@ -154,6 +155,35 @@ const mock = vi.hoisted(() => { return { state, createMockClient, HarnessDriverClient } }) +const fleetNodeMock = vi.hoisted(() => { + type SidecarMock = { registered: Promise; done: Promise; stop: ReturnType } + const sidecars: SidecarMock[] = [] + const startPearFleetSidecar = vi.fn((_options: unknown): SidecarMock => { + const sidecar: SidecarMock = { + registered: Promise.resolve({ + name: 'pear-project-1-local-fleet', + capabilities: [ + 'claude', + 'codex', + 'gemini', + 'opencode', + 'grok', + 'aider', + 'goose', + 'cursor', + 'droid' + ].map((cli) => ({ name: `spawn:${cli}`, kind: 'action' })) + }), + done: new Promise(() => undefined), + stop: vi.fn(async () => undefined) + } + sidecars.push(sidecar) + return sidecar + }) + + return { sidecars, startPearFleetSidecar } +}) + const electronMock = vi.hoisted(() => ({ app: { getAppPath: () => '/tmp/pear-app', @@ -171,6 +201,10 @@ vi.mock('@agent-relay/harness-driver', () => ({ HarnessDriverClient: mock.HarnessDriverClient })) +vi.mock('./pear-fleet-node', () => ({ + startPearFleetSidecar: fleetNodeMock.startPearFleetSidecar +})) + vi.mock('./auth', () => ({ getAccessToken: vi.fn(async () => 'token'), getApiUrl: vi.fn(() => 'https://cloud.example') @@ -581,6 +615,8 @@ describe('BrokerManager local + cloud coexistence', () => { mock.state.nextConnectedSessionErrors = [] mock.HarnessDriverClient.spawn.mockClear() mock.HarnessDriverClient.connect.mockClear() + fleetNodeMock.sidecars.length = 0 + fleetNodeMock.startPearFleetSidecar.mockClear() }) afterEach(async () => { @@ -667,6 +703,88 @@ describe('BrokerManager local + cloud coexistence', () => { await manager.shutdown() }) + it('registers a local fleet node for spawned brokers', async () => { + const manager = new BrokerManager() + + await manager.start(PROJECT_ID, '/tmp/project-1', 'pear-project-1', undefined as never, []) + + expect(fleetNodeMock.startPearFleetSidecar).toHaveBeenCalledTimes(1) + expect(fleetNodeMock.startPearFleetSidecar).toHaveBeenCalledWith(expect.objectContaining({ + projectId: PROJECT_ID, + cwd: '/tmp/project-1', + brokerName: 'pear-project-1', + connection: { + url: 'http://127.0.0.1:4242' + } + })) + + await manager.shutdown() + expect(fleetNodeMock.sidecars[0]?.stop).toHaveBeenCalledTimes(1) + }) + + it('keeps one local fleet node when reusing the same broker session', async () => { + const manager = new BrokerManager() + + await manager.start(PROJECT_ID, '/tmp/project-1', 'pear-project-1', undefined as never, []) + await manager.start(PROJECT_ID, '/tmp/project-1', 'pear-project-1', undefined as never, []) + + expect(fleetNodeMock.startPearFleetSidecar).toHaveBeenCalledTimes(1) + + await manager.shutdown() + }) + + it('does not register a local fleet node for cloud sandbox sessions', async () => { + const manager = new BrokerManager() + + await attachCloud(manager, ['cloud-agent']) + + expect(fleetNodeMock.startPearFleetSidecar).not.toHaveBeenCalled() + + await manager.shutdown() + }) + + it('does not block local broker start when fleet registration hangs', async () => { + const sidecar = { + registered: new Promise(() => undefined), + done: Promise.resolve(), + stop: vi.fn(async () => undefined) + } + fleetNodeMock.startPearFleetSidecar.mockImplementationOnce(() => { + fleetNodeMock.sidecars.push(sidecar) + return sidecar + }) + const manager = new BrokerManager() + + try { + const started = manager.start(PROJECT_ID, '/tmp/project-1', 'pear-project-1', undefined as never, []) + + await expect(started).resolves.toBe(true) + expect(mock.HarnessDriverClient.spawn).toHaveBeenCalledTimes(1) + } finally { + await manager.shutdown() + } + }, 5_000) + + it('does not block local broker shutdown when fleet stop hangs', async () => { + const sidecar = { + registered: Promise.resolve({ name: 'pear-project-1-local-fleet', capabilities: [] }), + done: new Promise(() => undefined), + stop: vi.fn(() => new Promise(() => undefined)) + } + fleetNodeMock.startPearFleetSidecar.mockImplementationOnce(() => { + fleetNodeMock.sidecars.push(sidecar) + return sidecar + }) + const manager = new BrokerManager() + + await manager.start(PROJECT_ID, '/tmp/project-1', 'pear-project-1', undefined as never, []) + const local = lastSpawned() + const shutdown = manager.shutdown() + + await expect(shutdown).resolves.toBeUndefined() + expect(local.shutdown).toHaveBeenCalled() + }, 5_000) + it('passes an explicit workspace key env pin to local broker spawn options', async () => { const previousWorkspaceKey = process.env.AGENT_RELAY_WORKSPACE_KEY process.env.AGENT_RELAY_WORKSPACE_KEY = 'rk_live_pinned' @@ -801,6 +919,11 @@ describe('BrokerManager local + cloud coexistence', () => { expect(mock.HarnessDriverClient.connect).toHaveBeenCalledWith({ cwd: tempDir, connectionPath }) expect(mock.HarnessDriverClient.spawn).not.toHaveBeenCalled() expect(mock.state.connectedClients).toHaveLength(1) + expect(fleetNodeMock.startPearFleetSidecar).toHaveBeenCalledWith(expect.objectContaining({ + projectId: PROJECT_ID, + cwd: tempDir, + brokerName: 'pear-project-1' + })) await manager.shutdown() } finally { diff --git a/src/main/broker.ts b/src/main/broker.ts index 51db7ebe..9184f7b8 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -71,6 +71,7 @@ import { resolveCommandOnPath, resolvePackageBin } from './mcp-command' +import { startPearFleetSidecar, type RunningPearFleetSidecar } from './pear-fleet-node' function isShellLikeCommand(cli: string): boolean { const normalized = basename(cli).toLowerCase() @@ -393,6 +394,8 @@ const BROKER_RELEASE_RETRY_BASE_MS = parsePositiveIntegerEnv('PEAR_BROKER_RELEAS // operation resets the streak. const MAX_BROKER_TIMEOUTS_BEFORE_REVIVE = 2 const BROKER_REVIVE_TERM_GRACE_MS = 1_500 +const LOCAL_FLEET_REGISTRATION_TIMEOUT_MS = 2_000 +const LOCAL_FLEET_STOP_TIMEOUT_MS = 1_000 const PERSONA_REGISTRATION_TIMEOUT_MS = 5_000 const PERSONA_REGISTRATION_STABILITY_MS = 1_000 const PERSONA_HARNESS_READY_TIMEOUT_MS = 120_000 @@ -415,6 +418,16 @@ function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)) } +function withTimeout(promise: Promise, timeoutMs: number, message: string): Promise { + let timer: ReturnType | undefined + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(message)), timeoutMs) + }) + return Promise.race([promise, timeout]).finally(() => { + if (timer) clearTimeout(timer) + }) +} + async function mapWithConcurrency( items: T[], concurrency: number, @@ -1059,6 +1072,8 @@ interface BrokerSession { // auto-shuts-down after 120s without a lease renewal, so we own the timer // here and clear it on shutdown. leaseTimer?: ReturnType + fleetSidecar?: RunningPearFleetSidecar + fleetSidecarCwd?: string operationQueue: BrokerOperationQueue } @@ -1355,6 +1370,7 @@ export class BrokerManager { existing.name = name await this.syncChannels(normalizedProjectId, nextChannels) await this.refreshEventStream(normalizedProjectId, 'existing-session-start', win) + await this.ensureLocalFleetSidecar(existing) this.sendStatus(normalizedProjectId, 'connected') return false } @@ -1371,6 +1387,7 @@ export class BrokerManager { started.cwd = cwd started.name = name await this.syncChannels(normalizedProjectId, nextChannels) + await this.ensureLocalFleetSidecar(started) this.sendStatus(normalizedProjectId, 'connected') return false } catch (err) { @@ -1385,7 +1402,7 @@ export class BrokerManager { if (existingClient) { const eventStreamGeneration = this.nextEventStreamGeneration() const unsubEvent = this.attachClient(normalizedProjectId, existingClient, win, eventStreamGeneration) - this.sessions.set(normalizedProjectId, { + const session: BrokerSession = { projectId: normalizedProjectId, client: existingClient, window: win, @@ -1397,10 +1414,12 @@ export class BrokerManager { pearLineage: new Map(), eventStreamGeneration, operationQueue: new BrokerOperationQueue(BROKER_OPERATION_CONCURRENCY) - }) + } + this.sessions.set(normalizedProjectId, session) existingClient.connectEvents() await this.syncChannels(normalizedProjectId, nextChannels) + await this.ensureLocalFleetSidecar(session) this.publishBrokerEvent(normalizedProjectId, normalizedProjectId, win, { kind: 'broker_initialized', name, @@ -1456,7 +1475,7 @@ export class BrokerManager { console.log('[broker] Started successfully for project:', normalizedProjectId) const eventStreamGeneration = this.nextEventStreamGeneration() const unsubEvent = this.attachClient(normalizedProjectId, client, win, eventStreamGeneration) - this.sessions.set(normalizedProjectId, { + const session: BrokerSession = { projectId: normalizedProjectId, client, window: win, @@ -1468,7 +1487,9 @@ export class BrokerManager { pearLineage: new Map(), eventStreamGeneration, operationQueue: new BrokerOperationQueue(BROKER_OPERATION_CONCURRENCY) - }) + } + this.sessions.set(normalizedProjectId, session) + await this.ensureLocalFleetSidecar(session) this.publishBrokerEvent(normalizedProjectId, normalizedProjectId, win, { kind: 'broker_initialized', @@ -1567,6 +1588,7 @@ export class BrokerManager { // eslint-disable-next-line prefer-const promise = (async () => { console.warn(`[broker] Broker for project ${projectId} is unreachable; restarting on a fresh port`) + await this.stopSessionFleetSidecar(session) this.dropSession(projectId, { disconnectOnly: true }) await terminateOwnedBrokerProcess(brokerPid) if (this.revivePromises.get(projectId) !== promise) return false @@ -1640,6 +1662,89 @@ export class BrokerManager { } } + private async ensureLocalFleetSidecar(session: BrokerSession): Promise { + if (session.cloudSandboxId) return + if (session.fleetSidecar && session.fleetSidecarCwd === session.cwd) return + if (session.fleetSidecar) { + await this.stopSessionFleetSidecar(session) + } + + const url = getClientBaseUrl(session.client) + if (!url) { + console.warn(`[broker] Local fleet node skipped for project ${session.projectId}: broker URL unavailable`) + return + } + + const sidecar = startPearFleetSidecar({ + projectId: session.projectId, + cwd: session.cwd, + brokerName: session.name, + connection: { + url, + ...(getClientApiKey(session.client) ? { apiKey: getClientApiKey(session.client) } : {}) + }, + log: (message) => console.log(`[broker] ${message}`), + warn: (message) => console.warn(`[broker] ${message}`) + }) + session.fleetSidecar = sidecar + session.fleetSidecarCwd = session.cwd + void sidecar.done.then( + () => { + if (session.fleetSidecar === sidecar) { + session.fleetSidecar = undefined + session.fleetSidecarCwd = undefined + } + }, + (err) => { + if (session.fleetSidecar === sidecar) { + session.fleetSidecar = undefined + session.fleetSidecarCwd = undefined + console.warn(`[broker] Local fleet node exited for project ${session.projectId}:`, err) + } + } + ) + + try { + await withTimeout( + sidecar.registered, + LOCAL_FLEET_REGISTRATION_TIMEOUT_MS, + `Local fleet node registration timed out after ${LOCAL_FLEET_REGISTRATION_TIMEOUT_MS}ms` + ) + } catch (err) { + const message = toErrorMessage(err) + if (/timed out/i.test(message)) { + console.warn(`[broker] Local fleet node registration still pending for project ${session.projectId}:`, message) + return + } + if (session.fleetSidecar === sidecar) { + await this.stopSessionFleetSidecar(session) + } else { + await withTimeout( + sidecar.stop(), + LOCAL_FLEET_STOP_TIMEOUT_MS, + `Local fleet node stop timed out after ${LOCAL_FLEET_STOP_TIMEOUT_MS}ms` + ).catch(() => undefined) + } + console.warn(`[broker] Local fleet node registration failed for project ${session.projectId}:`, err) + } + } + + private async stopSessionFleetSidecar(session: BrokerSession): Promise { + const sidecar = session.fleetSidecar + session.fleetSidecar = undefined + session.fleetSidecarCwd = undefined + if (!sidecar) return + try { + await withTimeout( + sidecar.stop(), + LOCAL_FLEET_STOP_TIMEOUT_MS, + `Local fleet node stop timed out after ${LOCAL_FLEET_STOP_TIMEOUT_MS}ms` + ) + } catch (err) { + console.warn(`[broker] Failed to stop local fleet node for project ${session.projectId}:`, err) + } + } + /** * Attach to an already-provisioned cloud sandbox (used by CloudAgentManager * which warms the box via the cloud-agents/{id}/box endpoint). connectCloud @@ -1686,6 +1791,7 @@ export class BrokerManager { const previous = this.sessions.get(sessionKey) if (previous) { try { + await this.stopSessionFleetSidecar(previous) await previous.client.shutdown() } catch (err) { // Non-fatal: we replace the session regardless, but a failed shutdown @@ -3926,6 +4032,7 @@ export class BrokerManager { const sessions = this.sessionsForProject(targetProjectId) for (const session of sessions) { try { + await this.stopSessionFleetSidecar(session) await session.client.shutdown() } catch (err) { // Non-fatal during teardown, but a failed cloud-session shutdown can @@ -3958,6 +4065,7 @@ export class BrokerManager { session.unsubEvent() if (session.leaseTimer) clearInterval(session.leaseTimer) + void this.stopSessionFleetSidecar(session) if (options.disconnectOnly) { const disconnect = (session.client as { disconnect?: () => void }).disconnect if (typeof disconnect === 'function') { @@ -3980,6 +4088,7 @@ export class BrokerManager { const session = this.sessions.get(sessionKey) if (!session) return try { + await this.stopSessionFleetSidecar(session) await session.client.shutdown() } catch (err) { // Non-fatal: we drop the session anyway, but a failed shutdown can leak diff --git a/src/main/pear-fleet-node.test.ts b/src/main/pear-fleet-node.test.ts new file mode 100644 index 00000000..a8e0889e --- /dev/null +++ b/src/main/pear-fleet-node.test.ts @@ -0,0 +1,111 @@ +import { describe, expect, it, vi } from 'vitest' +import { invokeNodeHandler, nodeInfo, nodeManifest } from '@agent-relay/fleet' +import { createPearFleetNodeDefinition, PEAR_LOCAL_SPAWN_HARNESSES } from './pear-fleet-node' + +const expectedCapabilities = Object.keys(PEAR_LOCAL_SPAWN_HARNESSES).map((cli) => `spawn:${cli}`) + +describe('Pear local fleet node', () => { + it('advertises local spawn capabilities and the project checkout', async () => { + const definition = createPearFleetNodeDefinition({ + projectId: 'project-1', + cwd: '/tmp/project-1', + brokerName: 'pear-project-1' + }) + + const manifest = nodeManifest(definition) + expect(manifest.name).toBe('pear-project-1-local-fleet') + expect(manifest.capabilities.map((capability) => capability.name)).toEqual(expectedCapabilities) + expect(manifest.capabilities).toEqual(expectedCapabilities.map((name) => expect.objectContaining({ + name, + metadata: expect.objectContaining({ + pearLocalNode: true, + clonePaths: { 'project-1': '/tmp/project-1' } + }) + }))) + }) + + it('spawns non-Claude/Codex harnesses through the broker', async () => { + const definition = createPearFleetNodeDefinition({ + projectId: 'project-1', + cwd: '/tmp/project-1', + brokerName: 'pear-project-1' + }) + const spawnAgent = vi.fn(async () => ({ name: 'goose-1', runtime: 'pty' })) + + await invokeNodeHandler(definition, 'spawn:goose', { + name: 'goose-1', + cwd: '/tmp/project-1', + args: ['--debug'], + channels: ['general'] + }, { + node: nodeInfo(definition), + relay: { + sendMessage: vi.fn(async () => undefined) + }, + invocationId: 'invoke-2', + spawnAgent + }) + + expect(spawnAgent).toHaveBeenCalledWith(expect.objectContaining({ + agent: expect.objectContaining({ + name: 'goose-1', + cli: 'goose', + args: ['--debug'], + channels: ['general'], + cwd: '/tmp/project-1' + }), + invocationId: 'invoke-2' + })) + }) + + it('spawns through the broker for the advertised checkout path', async () => { + const definition = createPearFleetNodeDefinition({ + projectId: 'project-1', + cwd: '/tmp/project-1', + brokerName: 'pear-project-1' + }) + const spawnAgent = vi.fn(async () => ({ name: 'worker-1', runtime: 'pty' })) + + await invokeNodeHandler(definition, 'spawn:claude', { + name: 'worker-1', + cwd: '/tmp/project-1', + task: 'check relay registration' + }, { + node: nodeInfo(definition), + relay: { + sendMessage: vi.fn(async () => undefined) + }, + invocationId: 'invoke-1', + spawnAgent + }) + + expect(spawnAgent).toHaveBeenCalledWith(expect.objectContaining({ + agent: expect.objectContaining({ + name: 'worker-1', + cli: 'claude', + cwd: '/tmp/project-1' + }), + initialTask: 'check relay registration', + invocationId: 'invoke-1' + })) + }) + + it('rejects unadvertised checkout paths', async () => { + const definition = createPearFleetNodeDefinition({ + projectId: 'project-1', + cwd: '/tmp/project-1', + brokerName: 'pear-project-1' + }) + + await expect(invokeNodeHandler(definition, 'spawn:claude', { + name: 'worker-1', + cwd: '/tmp/other-project' + }, { + node: nodeInfo(definition), + relay: { + sendMessage: vi.fn(async () => undefined) + }, + spawnAgent: vi.fn(async () => undefined) + })).rejects.toThrow('checkout path is not advertised by this node') + }) +}) diff --git a/src/main/pear-fleet-node.ts b/src/main/pear-fleet-node.ts new file mode 100644 index 00000000..322efa04 --- /dev/null +++ b/src/main/pear-fleet-node.ts @@ -0,0 +1,506 @@ +import { basename, resolve } from 'node:path' +import { + action, + defineNode, + invokeNodeHandler, + nodeInfo, + nodeManifest, + type FleetCapabilityValue, + type FleetNodeDefinition +} from '@agent-relay/fleet' +import { + aider, + claude, + codex, + cursor, + droid, + gemini, + goose, + grok, + opencode, + type PtyHarness +} from '@agent-relay/harnesses' +import { resolveStaticHarnessConfig, type RestartPolicy } from '@agent-relay/harness-driver' +import { PROTOCOL_VERSION, type NodeManifest } from '@agent-relay/harness-driver/protocol' +import WebSocket from 'ws' +import { z } from 'zod' + +export const PEAR_LOCAL_SPAWN_HARNESSES = { + claude, + codex, + gemini, + opencode, + grok, + aider, + goose, + cursor, + droid +} satisfies Record + +const RECONNECT_BASE_DELAY_MS = 500 +const RECONNECT_MAX_DELAY_MS = 5_000 +const REQUEST_TIMEOUT_MS = 5_000 +const DEREGISTER_TIMEOUT_MS = 1_000 + +const restartPolicySchema = z.object({ + enabled: z.boolean().optional(), + max_restarts: z.number().int().min(0).optional(), + cooldown_ms: z.number().int().min(0).optional(), + max_consecutive_failures: z.number().int().min(0).optional() +}).passthrough() + +const spawnCapabilityInputSchema = z.object({ + name: z.string().min(1).optional(), + agent: z.string().min(1).optional(), + clone_path: z.string().min(1).optional(), + clonePath: z.string().min(1).optional(), + cwd: z.string().min(1).optional(), + task: z.string().optional(), + model: z.string().min(1).optional(), + session_ref: z.string().min(1).optional(), + sessionRef: z.string().min(1).optional(), + channels: z.array(z.string().min(1)).optional(), + channel: z.string().min(1).optional(), + args: z.array(z.string()).optional(), + restart_policy: restartPolicySchema.optional(), + restartPolicy: restartPolicySchema.optional(), + skip_relay_prompt: z.boolean().optional() +}).passthrough() + +type RawSpawnCapabilityInput = z.infer +type SpawnCapabilityInput = RawSpawnCapabilityInput & { + name: string | undefined + clonePath: string | undefined + sessionRef: string | undefined + channels: string[] | undefined + restartPolicy: RestartPolicy | undefined +} + +export interface PearFleetNodeOptions { + projectId: string + cwd: string + brokerName: string +} + +export interface PearFleetSidecarOptions extends PearFleetNodeOptions { + connection: { + url: string + apiKey?: string + } + log?: (message: string) => void + warn?: (message: string) => void +} + +export interface RunningPearFleetSidecar { + readonly registered: Promise + readonly done: Promise + stop(): Promise +} + +type BrokerFrame = { + type?: string + request_id?: string + payload?: unknown +} + +type PendingRequest = { + resolve: (value: unknown) => void + reject: (error: Error) => void + timer: ReturnType +} + +export function createPearFleetNodeDefinition(options: PearFleetNodeOptions): FleetNodeDefinition { + const nodeName = pearFleetNodeName(options) + const clonePathKey = basename(options.cwd) || options.projectId || 'project' + const clonePaths = { [clonePathKey]: options.cwd } + const capabilities: Record = {} + for (const [capabilityCli, harness] of Object.entries(PEAR_LOCAL_SPAWN_HARNESSES)) { + capabilities[`spawn:${capabilityCli}`] = action({ + metadata: { + pearLocalNode: true, + cli: capabilityCli, + command: harness.command, + clonePaths + } + }, async (rawInput, ctx) => { + const input = parseSpawnCapabilityInput(rawInput) + if (!input.name) { + throw new Error(`spawn:${capabilityCli} requires name or agent`) + } + const cwd = resolvePearSpawnCwd(options.cwd, input) + const harnessConfig = resolveStaticHarnessConfig({ + name: input.name, + cli: harness.command, + definition: harness, + args: input.args, + task: input.task, + model: input.model, + cwd + }) + + const output = await ctx.spawnAgent({ + agent: { + name: input.name, + runtime: 'pty', + cli: harness.command, + ...(input.model ? { model: input.model } : {}), + ...(input.args && input.args.length > 0 ? { args: input.args } : {}), + ...(input.channels ? { channels: input.channels } : {}), + cwd, + ...(input.sessionRef ? { session_id: input.sessionRef } : {}), + ...(input.restartPolicy ? { restart_policy: input.restartPolicy as RestartPolicy } : {}), + harness_config: harnessConfig + }, + ...(input.task !== undefined ? { initialTask: input.task } : {}), + skipRelayPrompt: input.skip_relay_prompt ?? false, + invocationId: ctx.invocationId + }) + + return { + name: input.name, + capability: `spawn:${capabilityCli}`, + cwd, + invocationId: ctx.invocationId, + agent: output + } + }) as FleetCapabilityValue + } + + return defineNode({ + name: nodeName, + capabilities, + tags: ['pear', 'local', `project:${options.projectId}`], + version: 'pear-local-fleet-v1' + }) +} + +function parseSpawnCapabilityInput(input: unknown): SpawnCapabilityInput { + const parsed = spawnCapabilityInputSchema.parse(input) + return { + ...parsed, + name: parsed.name ?? parsed.agent, + clonePath: parsed.clonePath ?? parsed.clone_path, + sessionRef: parsed.sessionRef ?? parsed.session_ref, + channels: parsed.channels ?? (parsed.channel ? [parsed.channel] : undefined), + restartPolicy: parsed.restartPolicy ?? parsed.restart_policy + } +} + +function resolvePearSpawnCwd(projectCwd: string, input: SpawnCapabilityInput): string { + const explicitPath = input.clonePath ?? input.cwd + if (!explicitPath) return resolve(projectCwd) + + const requested = resolve(explicitPath) + const advertised = resolve(projectCwd) + if (requested !== advertised) { + throw new Error(`checkout path is not advertised by this node: ${explicitPath}`) + } + return requested +} + +export function startPearFleetSidecar(options: PearFleetSidecarOptions): RunningPearFleetSidecar { + const controller = new AbortController() + let resolveRegistered: (manifest: NodeManifest) => void + let rejectRegistered: (error: Error) => void + let registeredSettled = false + const registered = new Promise((resolve, reject) => { + resolveRegistered = (manifest) => { + registeredSettled = true + resolve(manifest) + } + rejectRegistered = (error) => { + registeredSettled = true + reject(error) + } + }) + + const definition = createPearFleetNodeDefinition(options) + const done = runPearFleetSidecarLoop({ + ...options, + definition, + signal: controller.signal, + onRegistered: (manifest) => { + if (!registeredSettled) resolveRegistered(manifest) + }, + onInitialFailure: (error) => { + if (!registeredSettled) rejectRegistered(error) + } + }) + + return { + registered, + done, + stop: async () => { + controller.abort() + await done + } + } +} + +function pearFleetNodeName(options: PearFleetNodeOptions): string { + const rawName = `${options.brokerName || options.projectId || 'pear'}-local-fleet` + return rawName.replace(/[^\w.-]+/gu, '-').replace(/^-+|-+$/gu, '') || 'pear-local-fleet' +} + +async function runPearFleetSidecarLoop(options: PearFleetSidecarOptions & { + definition: FleetNodeDefinition + signal: AbortSignal + onRegistered: (manifest: NodeManifest) => void + onInitialFailure: (error: Error) => void +}): Promise { + let attempt = 0 + let registeredOnce = false + + while (!options.signal.aborted) { + try { + await runPearFleetSidecarConnection({ + ...options, + onRegistered: (manifest) => { + registeredOnce = true + options.onRegistered(manifest) + } + }) + attempt = 0 + } catch (error) { + const err = toError(error) + if (!registeredOnce) { + options.onInitialFailure(err) + } + if (options.signal.aborted) return + options.warn?.(`Pear fleet sidecar disconnected: ${err.message}; reconnecting`) + } + + if (options.signal.aborted) return + attempt += 1 + await delay(Math.min(RECONNECT_MAX_DELAY_MS, RECONNECT_BASE_DELAY_MS * 2 ** (attempt - 1)), options.signal) + } +} + +function runPearFleetSidecarConnection(options: PearFleetSidecarOptions & { + definition: FleetNodeDefinition + signal: AbortSignal + onRegistered: (manifest: NodeManifest) => void +}): Promise { + return new Promise((resolve, reject) => { + const ws = new WebSocket(fleetWsUrl(options.connection.url), { + headers: options.connection.apiKey ? { 'X-API-Key': options.connection.apiKey } : undefined + }) + const pending = new Map() + let requestSeq = 0 + let settled = false + let nodeRegistered = false + + const settle = (fn: () => void): void => { + if (settled) return + settled = true + for (const pendingRequest of pending.values()) { + pendingRequest.reject(new Error('Pear fleet sidecar connection closed')) + } + pending.clear() + options.signal.removeEventListener('abort', abort) + fn() + } + + const sendRequest = (type: string, payload: unknown): Promise => { + if (ws.readyState !== WebSocket.OPEN) { + return Promise.reject(new Error('Pear fleet sidecar websocket is not open')) + } + const requestId = `pear_fleet_${Date.now()}_${++requestSeq}` + const frame = { + v: PROTOCOL_VERSION, + type, + request_id: requestId, + payload + } + + return new Promise((requestResolve, requestReject) => { + const timer = setTimeout(() => { + pending.delete(requestId) + requestReject(new Error(`Pear fleet sidecar ${type} request timed out after ${REQUEST_TIMEOUT_MS}ms`)) + }, REQUEST_TIMEOUT_MS) + const pendingRequest: PendingRequest = { + timer, + resolve: (value) => { + clearTimeout(timer) + requestResolve(value) + }, + reject: (error) => { + clearTimeout(timer) + requestReject(error) + } + } + pending.set(requestId, pendingRequest) + ws.send(JSON.stringify(frame), (error) => { + if (!error) return + if (!pending.has(requestId)) return + pending.delete(requestId) + pendingRequest.reject(error) + }) + }) + } + + const close = async (): Promise => { + if (ws.readyState === WebSocket.OPEN && nodeRegistered) { + await withTimeout(sendRequest('deregister_node', {}), DEREGISTER_TIMEOUT_MS).catch(() => undefined) + } + if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) { + ws.close() + } + } + + const ignoreCloseError = (error: unknown): void => { + options.warn?.(`Pear fleet sidecar close skipped: ${toError(error).message}`) + } + + const abort = (): void => { + void close().catch(ignoreCloseError).finally(() => settle(resolve)) + } + + const sendHandlerResult = async (invocationId: string, output: unknown, error?: unknown): Promise => { + const payload = error + ? { invocation_id: invocationId, error: toError(error).message } + : { invocation_id: invocationId, output: output ?? null } + await sendRequest('handler_result', payload) + } + + const handleInvoke = async (payload: unknown): Promise => { + const record = asRecord(payload) + const invocationId = typeof record.invocation_id === 'string' ? record.invocation_id : undefined + const name = typeof record.name === 'string' ? record.name : undefined + if (!invocationId || !name) return + + try { + const output = await invokeNodeHandler(options.definition, name, record.input, { + node: nodeInfo(options.definition), + invocationId, + relay: { + sendMessage: (input) => sendRequest('send_message', { + to: input.to, + text: input.text, + from: input.from ?? options.definition.name, + ...(input.threadId ? { thread_id: input.threadId } : {}), + ...(input.workspaceId ? { workspace_id: input.workspaceId } : {}), + ...(input.workspaceAlias ? { workspace_alias: input.workspaceAlias } : {}), + ...(input.mode ? { mode: input.mode } : {}), + ...(input.data ? { data: input.data } : {}) + }) + }, + spawnAgent: (input) => sendRequest('spawn_agent', { + agent: input.agent, + ...(input.initialTask !== undefined ? { initial_task: input.initialTask } : {}), + skip_relay_prompt: input.skipRelayPrompt ?? false, + ...((input.invocationId ?? invocationId) ? { invocation_id: input.invocationId ?? invocationId } : {}) + }) + }) + await sendHandlerResult(invocationId, output) + } catch (error) { + await sendHandlerResult(invocationId, undefined, error) + } + } + + options.signal.addEventListener('abort', abort, { once: true }) + + ws.on('open', () => { + void (async () => { + const manifest = nodeManifest(options.definition) + await sendRequest('hello', { + client_name: 'pear-local-fleet', + client_version: '1.0.0' + }) + await sendRequest('register_node', { manifest }) + nodeRegistered = true + await sendRequest('register_handlers', { names: Object.keys(options.definition.capabilities) }) + options.log?.(`Pear fleet node "${manifest.name}" registered with ${manifest.capabilities.length} capabilities.`) + options.onRegistered(manifest) + })().catch((error) => { + settle(() => reject(toError(error))) + void close().catch(ignoreCloseError) + }) + }) + + ws.on('message', (data) => { + const frame = parseBrokerFrame(data) + if (!frame) return + if (frame.request_id && pending.has(frame.request_id)) { + const pendingRequest = pending.get(frame.request_id) + pending.delete(frame.request_id) + if (!pendingRequest) return + if (frame.type === 'error') { + pendingRequest.reject(frameError(frame.payload)) + } else { + pendingRequest.resolve(readOkResult(frame.payload)) + } + return + } + if (frame.type === 'invoke_handler') { + void handleInvoke(frame.payload).catch((error) => options.warn?.(toError(error).message)) + } + }) + + ws.on('close', () => settle(resolve)) + ws.on('error', (error) => settle(() => reject(toError(error)))) + }) +} + +function parseBrokerFrame(data: WebSocket.RawData): BrokerFrame | null { + try { + const text = Array.isArray(data) ? Buffer.concat(data).toString('utf8') : data.toString() + return JSON.parse(text) as BrokerFrame + } catch { + return null + } +} + +function readOkResult(payload: unknown): unknown { + return payload && typeof payload === 'object' && 'result' in payload + ? (payload as { result: unknown }).result + : payload +} + +function frameError(payload: unknown): Error { + if (payload && typeof payload === 'object') { + const record = payload as Record + const error = new Error(typeof record.message === 'string' ? record.message : 'Pear fleet sidecar request failed') + error.name = typeof record.code === 'string' ? record.code : 'PearFleetSidecarError' + return error + } + return new Error('Pear fleet sidecar request failed') +} + +function fleetWsUrl(baseUrl: string): string { + return `${baseUrl.replace(/\/+$/u, '').replace(/^http/u, 'ws')}/api/fleet/ws` +} + +function asRecord(value: unknown): Record { + return value && typeof value === 'object' ? value as Record : {} +} + +function toError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)) +} + +function delay(ms: number, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.resolve() + return new Promise((resolve) => { + const onAbort = (): void => { + clearTimeout(timer) + signal.removeEventListener('abort', onAbort) + resolve() + } + const timer = setTimeout(() => { + signal.removeEventListener('abort', onAbort) + resolve() + }, ms) + signal.addEventListener('abort', onAbort, { once: true }) + }) +} + +function withTimeout(promise: Promise, timeoutMs: number): Promise { + let timer: ReturnType | undefined + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`Timed out after ${timeoutMs}ms`)), timeoutMs) + }) + return Promise.race([promise, timeout]).finally(() => { + if (timer) clearTimeout(timer) + }) +}