From 8dff373a852fa49aa472d0aa44643f5b121634c2 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 5 Jun 2026 14:56:51 +0200 Subject: [PATCH 1/2] pathing --- src/main/broker.test.ts | 38 ++++++++++++++++++++++++++++-- src/main/broker.ts | 52 ++++++++++++++++++++++++++++------------- 2 files changed, 72 insertions(+), 18 deletions(-) diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index fd147437..0b50b30d 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -63,8 +63,10 @@ const mock = vi.hoisted(() => { const state = { spawnedClients: [] as MockClient[], constructedClients: [] as MockClient[], + connectedClients: [] as MockClient[], nextLocalAgents: [] as string[], - nextCloudAgents: [] as string[] + nextCloudAgents: [] as string[], + nextConnectedAgents: [] as string[] } class HarnessDriverClient { @@ -75,7 +77,9 @@ const mock = vi.hoisted(() => { }) static connect = vi.fn(() => { - throw new Error('not used in tests') + const client = createMockClient(state.nextConnectedAgents.splice(0)) + state.connectedClients.push(client) + return client }) constructor() { @@ -314,9 +318,12 @@ describe('BrokerManager local + cloud coexistence', () => { beforeEach(() => { mock.state.spawnedClients.length = 0 mock.state.constructedClients.length = 0 + mock.state.connectedClients.length = 0 mock.state.nextLocalAgents = [] mock.state.nextCloudAgents = [] + mock.state.nextConnectedAgents = [] mock.HarnessDriverClient.spawn.mockClear() + mock.HarnessDriverClient.connect.mockClear() }) it('keeps the local session alive when a cloud sandbox attaches', async () => { @@ -354,6 +361,33 @@ describe('BrokerManager local + cloud coexistence', () => { await manager.shutdown() }) + it('reuses current harness-driver connection files instead of spawning another broker', async () => { + const tempDir = await mkdtemp(join(tmpdir(), 'pear-current-connection-')) + const connectionPath = join(tempDir, '.agentworkforce', 'relay', 'connection.json') + await mkdir(dirname(connectionPath), { recursive: true }) + await writeFile(connectionPath, JSON.stringify({ + url: 'http://127.0.0.1:43210', + apiKey: 'test-key', + pid: 4242 + })) + + try { + const manager = new BrokerManager() + mock.state.nextConnectedAgents = ['codex-1'] + + const started = await manager.start(PROJECT_ID, tempDir, 'pear-project-1', undefined as never, []) + + expect(started).toBe(true) + expect(mock.HarnessDriverClient.connect).toHaveBeenCalledWith({ cwd: tempDir, connectionPath }) + expect(mock.HarnessDriverClient.spawn).not.toHaveBeenCalled() + expect(mock.state.connectedClients).toHaveLength(1) + + await manager.shutdown() + } finally { + await rm(tempDir, { recursive: true, force: true }) + } + }) + it('keeps the cloud session alive when a local broker starts afterwards', async () => { const manager = new BrokerManager() const cloud = await attachCloud(manager, ['cloud-agent']) diff --git a/src/main/broker.ts b/src/main/broker.ts index dcdad0af..9171c0ea 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -718,7 +718,7 @@ function getBrokerConnectionFileInfo( return { hasApiKey: false } } - const connectionPath = join(cwd, '.agent-relay', 'connection.json') + const connectionPath = resolveBrokerConnectionPath(cwd) if (!existsSync(connectionPath)) { return { path: connectionPath, status: 'missing', hasApiKey: false } } @@ -745,6 +745,18 @@ function getBrokerConnectionFileInfo( } } +function brokerConnectionPathCandidates(cwd: string): string[] { + return [ + join(cwd, '.agentworkforce', 'relay', 'connection.json'), + join(cwd, '.agent-relay', 'connection.json') + ] +} + +function resolveBrokerConnectionPath(cwd: string): string { + return brokerConnectionPathCandidates(cwd).find((candidate) => existsSync(candidate)) ?? + brokerConnectionPathCandidates(cwd)[0] +} + function toInboundDeliveryMode(mode?: TerminalAttachMode): InboundDeliveryMode { return mode === 'drive' ? 'manual_flush' : 'auto_inject' } @@ -845,14 +857,19 @@ function getBrokerRuntimeSafeName(name: string): string { } function getBrokerRuntimeCleanupPaths(cwd: string, brokerName: string): string[] { - const root = join(cwd, '.agent-relay') + const legacyRoot = join(cwd, '.agent-relay') + const currentRoot = join(cwd, '.agentworkforce', 'relay') const safeName = getBrokerRuntimeSafeName(brokerName) return [ - join(root, 'connection.json'), - join(root, `broker-${safeName}.lock`), - join(root, `state-${safeName}.json`), - join(root, `pending-${safeName}.json`) + join(currentRoot, 'connection.json'), + join(currentRoot, `broker-${safeName}.lock`), + join(currentRoot, `state-${safeName}.json`), + join(currentRoot, `pending-${safeName}.json`), + join(legacyRoot, 'connection.json'), + join(legacyRoot, `broker-${safeName}.lock`), + join(legacyRoot, `state-${safeName}.json`), + join(legacyRoot, `pending-${safeName}.json`) ] } @@ -1268,20 +1285,23 @@ export class BrokerManager { } private async connectExistingBroker(projectId: string, cwd: string): Promise { - const connectionPath = join(cwd, '.agent-relay', 'connection.json') - if (!existsSync(connectionPath)) { + const connectionPaths = brokerConnectionPathCandidates(cwd).filter((candidate) => existsSync(candidate)) + if (connectionPaths.length === 0) { return null } - try { - const client = AgentRelayClient.connect({ cwd }) - await client.getSession() - console.log(`[broker] Reusing existing broker for project ${projectId}: ${connectionPath}`) - return client - } catch (err) { - console.warn(`[broker] Existing broker connection is not reusable for project ${projectId}:`, err) - return null + for (const connectionPath of connectionPaths) { + try { + const client = AgentRelayClient.connect({ cwd, connectionPath }) + await client.getSession() + console.log(`[broker] Reusing existing broker for project ${projectId}: ${connectionPath}`) + return client + } catch (err) { + console.warn(`[broker] Existing broker connection is not reusable for project ${projectId}:`, err) + } } + + return null } // Re-establish a local broker whose process has died or wedged. When this From b218297edff3c7d243eba414e40c82e36d14454d Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 5 Jun 2026 15:07:35 +0200 Subject: [PATCH 2/2] chore: apply broker review fixes --- src/main/broker.test.ts | 88 ++++++++++++++++++++++++++++++++++++++++- src/main/broker.ts | 28 +++++++++++-- 2 files changed, 111 insertions(+), 5 deletions(-) diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index 0b50b30d..808b9d88 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -26,7 +26,9 @@ type MockClient = { release: ReturnType subscribeChannels: ReturnType unsubscribeChannels: ReturnType + getStatus: ReturnType brokerPid?: number + baseUrl?: string agentNames: string[] } @@ -45,6 +47,10 @@ const mock = vi.hoisted(() => { snapshot: vi.fn(async () => ({ rows: 24, cols: 80, cursor: { x: 0, y: 0 }, screen: 'aGVsbG8=' })), resizePty: vi.fn(async () => undefined), getPending: vi.fn(async () => []), + getStatus: vi.fn(async () => ({ + agents: client.agentNames.map((name) => ({ name, runtime: 'pty', channels: [] })), + pending_delivery_count: 0 + })), onEvent: vi.fn(() => () => undefined), addListener: vi.fn(() => () => undefined), connectEvents: vi.fn(), @@ -66,7 +72,8 @@ const mock = vi.hoisted(() => { connectedClients: [] as MockClient[], nextLocalAgents: [] as string[], nextCloudAgents: [] as string[], - nextConnectedAgents: [] as string[] + nextConnectedAgents: [] as string[], + nextConnectedSessionErrors: [] as Error[] } class HarnessDriverClient { @@ -78,6 +85,10 @@ const mock = vi.hoisted(() => { static connect = vi.fn(() => { const client = createMockClient(state.nextConnectedAgents.splice(0)) + const sessionError = state.nextConnectedSessionErrors.shift() + if (sessionError) { + client.getSession.mockRejectedValueOnce(sessionError) + } state.connectedClients.push(client) return client }) @@ -322,6 +333,7 @@ describe('BrokerManager local + cloud coexistence', () => { mock.state.nextLocalAgents = [] mock.state.nextCloudAgents = [] mock.state.nextConnectedAgents = [] + mock.state.nextConnectedSessionErrors = [] mock.HarnessDriverClient.spawn.mockClear() mock.HarnessDriverClient.connect.mockClear() }) @@ -388,6 +400,80 @@ describe('BrokerManager local + cloud coexistence', () => { } }) + it('reports the matching connection file when a stale current file and matching legacy file coexist', async () => { + const tempDir = await mkdtemp(join(tmpdir(), 'pear-connection-status-')) + const currentConnectionPath = join(tempDir, '.agentworkforce', 'relay', 'connection.json') + const legacyConnectionPath = join(tempDir, '.agent-relay', 'connection.json') + await mkdir(dirname(currentConnectionPath), { recursive: true }) + await mkdir(dirname(legacyConnectionPath), { recursive: true }) + + try { + const manager = new BrokerManager() + mock.state.nextLocalAgents = [] + await manager.start(PROJECT_ID, tempDir, 'pear-project-1', undefined as never, []) + const local = lastSpawned() + local.baseUrl = 'http://127.0.0.1:43210' + local.brokerPid = 4242 + + await writeFile(currentConnectionPath, JSON.stringify({ + url: 'http://127.0.0.1:1', + api_key: 'stale-key', + pid: 1 + })) + await writeFile(legacyConnectionPath, JSON.stringify({ + url: 'http://127.0.0.1:43210/', + api_key: 'legacy-key', + pid: 4242 + })) + + const [details] = await manager.listBrokerDetails() + + expect(details.connectionPath).toBe(legacyConnectionPath) + expect(details.connectionFileStatus).toBe('matches') + expect(details.apiKey).toBe('legacy-key') + + await manager.shutdown() + } finally { + await rm(tempDir, { recursive: true, force: true }) + } + }) + + it('disconnects failed broker connection probes before trying the next candidate', async () => { + const tempDir = await mkdtemp(join(tmpdir(), 'pear-connection-fallback-')) + const currentConnectionPath = join(tempDir, '.agentworkforce', 'relay', 'connection.json') + const legacyConnectionPath = join(tempDir, '.agent-relay', 'connection.json') + await mkdir(dirname(currentConnectionPath), { recursive: true }) + await mkdir(dirname(legacyConnectionPath), { recursive: true }) + await writeFile(currentConnectionPath, JSON.stringify({ + url: 'http://127.0.0.1:1', + api_key: 'stale-key', + pid: 1 + })) + await writeFile(legacyConnectionPath, JSON.stringify({ + url: 'http://127.0.0.1:43210', + api_key: 'legacy-key', + pid: 4242 + })) + + try { + const manager = new BrokerManager() + mock.state.nextConnectedSessionErrors = [new Error('stale broker')] + + const started = await manager.start(PROJECT_ID, tempDir, 'pear-project-1', undefined as never, []) + + expect(started).toBe(true) + expect(mock.HarnessDriverClient.connect).toHaveBeenCalledTimes(2) + expect(mock.HarnessDriverClient.connect).toHaveBeenNthCalledWith(1, { cwd: tempDir, connectionPath: currentConnectionPath }) + expect(mock.HarnessDriverClient.connect).toHaveBeenNthCalledWith(2, { cwd: tempDir, connectionPath: legacyConnectionPath }) + expect(mock.state.connectedClients[0].disconnect).toHaveBeenCalledTimes(1) + expect(mock.HarnessDriverClient.spawn).not.toHaveBeenCalled() + + await manager.shutdown() + } finally { + await rm(tempDir, { recursive: true, force: true }) + } + }) + it('keeps the cloud session alive when a local broker starts afterwards', async () => { const manager = new BrokerManager() const cloud = await attachCloud(manager, ['cloud-agent']) diff --git a/src/main/broker.ts b/src/main/broker.ts index 9171c0ea..9e791159 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -718,7 +718,25 @@ function getBrokerConnectionFileInfo( return { hasApiKey: false } } - const connectionPath = resolveBrokerConnectionPath(cwd) + const candidates = brokerConnectionPathCandidates(cwd) + let firstExisting: BrokerConnectionFileInfo | undefined + + for (const connectionPath of candidates) { + if (!existsSync(connectionPath)) continue + + const info = readBrokerConnectionFileInfo(connectionPath, baseUrl, brokerPid) + if (info.status === 'matches') return info + firstExisting ??= info + } + + return firstExisting ?? { path: candidates[0], status: 'missing', hasApiKey: false } +} + +function readBrokerConnectionFileInfo( + connectionPath: string, + baseUrl: string | undefined, + brokerPid: number | undefined +): BrokerConnectionFileInfo { if (!existsSync(connectionPath)) { return { path: connectionPath, status: 'missing', hasApiKey: false } } @@ -753,8 +771,8 @@ function brokerConnectionPathCandidates(cwd: string): string[] { } function resolveBrokerConnectionPath(cwd: string): string { - return brokerConnectionPathCandidates(cwd).find((candidate) => existsSync(candidate)) ?? - brokerConnectionPathCandidates(cwd)[0] + const candidates = brokerConnectionPathCandidates(cwd) + return candidates.find((candidate) => existsSync(candidate)) ?? candidates[0] } function toInboundDeliveryMode(mode?: TerminalAttachMode): InboundDeliveryMode { @@ -1291,12 +1309,14 @@ export class BrokerManager { } for (const connectionPath of connectionPaths) { + let client: AgentRelayClient | undefined try { - const client = AgentRelayClient.connect({ cwd, connectionPath }) + client = AgentRelayClient.connect({ cwd, connectionPath }) await client.getSession() console.log(`[broker] Reusing existing broker for project ${projectId}: ${connectionPath}`) return client } catch (err) { + client?.disconnect() console.warn(`[broker] Existing broker connection is not reusable for project ${projectId}:`, err) } }