From 33debace20d014a4a08be7aa2ef809541939c5ca Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 5 Jun 2026 11:42:54 +0200 Subject: [PATCH 1/8] Fix Electron main Zod bundling --- electron.vite.config.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/electron.vite.config.ts b/electron.vite.config.ts index e23bd4f0..d20b179b 100644 --- a/electron.vite.config.ts +++ b/electron.vite.config.ts @@ -6,7 +6,7 @@ import tailwindcss from '@tailwindcss/vite' export default defineConfig({ main: { plugins: [externalizeDepsPlugin({ - exclude: ['@agent-relay/sdk', '@agent-relay/cloud', '@agent-relay/harness-driver'] + exclude: ['@agent-relay/sdk', '@agent-relay/cloud', '@agent-relay/harness-driver', 'zod'] })], build: { rollupOptions: { From f8931bd4dd6da69daa2125623a44312c420da4e3 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 5 Jun 2026 11:46:36 +0200 Subject: [PATCH 2/8] Avoid duplicate integration updates on broker reuse --- src/main/broker.test.ts | 16 ++++++++++++++++ src/main/broker.ts | 13 +++++++------ src/main/ipc-handlers.ts | 10 ++++++---- 3 files changed, 29 insertions(+), 10 deletions(-) diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index eabcb8fc..cab9117d 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -270,6 +270,22 @@ describe('BrokerManager local + cloud coexistence', () => { expect(cloud.shutdown).toHaveBeenCalled() }) + it('reports an existing local broker start as reused', async () => { + const manager = new BrokerManager() + mock.state.nextLocalAgents = [] + + const firstStart = await manager.start(PROJECT_ID, '/tmp/project-1', 'pear-project-1', undefined as never, []) + const local = lastSpawned() + const secondStart = await manager.start(PROJECT_ID, '/tmp/project-1', 'pear-project-1', undefined as never, []) + + expect(firstStart).toBe(true) + expect(secondStart).toBe(false) + expect(mock.HarnessDriverClient.spawn).toHaveBeenCalledTimes(1) + expect(local.shutdown).not.toHaveBeenCalled() + + await manager.shutdown() + }) + 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 4e4029b9..6d3f5f23 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -1124,7 +1124,7 @@ export class BrokerManager { name: string, win: BrowserWindow, channels: string[] = [] - ): Promise { + ): Promise { const normalizedProjectId = projectId.trim() if (!normalizedProjectId) { throw new Error('Project id is required') @@ -1140,7 +1140,7 @@ export class BrokerManager { await this.syncChannels(normalizedProjectId, nextChannels) await this.refreshEventStream(normalizedProjectId, 'existing-session-start', win) this.sendStatus(normalizedProjectId, 'connected') - return + return false } const inFlight = this.startPromises.get(normalizedProjectId) @@ -1156,14 +1156,14 @@ export class BrokerManager { started.name = name await this.syncChannels(normalizedProjectId, nextChannels) this.sendStatus(normalizedProjectId, 'connected') - return + return false } catch (err) { this.sendStatusToWindow(win, normalizedProjectId, 'error', String(err)) throw err } } - const startBroker = async (): Promise => { + const startBroker = async (): Promise => { const existingClient = await this.connectExistingBroker(normalizedProjectId, cwd) if (existingClient) { const unsubEvent = this.attachClient(normalizedProjectId, existingClient, win) @@ -1190,7 +1190,7 @@ export class BrokerManager { source: 'local' }) this.sendStatus(normalizedProjectId, 'connected') - return + return true } const agentRelayMcpCommand = resolveAgentRelayMcpCommand() @@ -1241,12 +1241,13 @@ export class BrokerManager { source: 'local' }) this.sendStatus(normalizedProjectId, 'connected') + return true } const startPromise = startBroker() this.startPromises.set(normalizedProjectId, startPromise) try { - await startPromise + return await startPromise } catch (err) { console.error(`[broker] Failed to start for project ${normalizedProjectId}:`, err) this.sendStatusToWindow(win, normalizedProjectId, 'error', String(err)) diff --git a/src/main/ipc-handlers.ts b/src/main/ipc-handlers.ts index c1d1b7ce..41308f5c 100644 --- a/src/main/ipc-handlers.ts +++ b/src/main/ipc-handlers.ts @@ -179,10 +179,12 @@ export function registerIpcHandlers(): void { console.warn(`[broker] Project path no longer exists; skipping broker start: ${cwd}`) return false } - await brokerManager.start(projectId, cwd, name, win, channels) - void integrationsManager.notifyAgentState(projectId).catch((error) => { - console.warn('[integrations] Failed to notify agents after broker start:', error instanceof Error ? error.message : String(error)) - }) + const started = await brokerManager.start(projectId, cwd, name, win, channels) + if (started) { + void integrationsManager.notifyAgentState(projectId).catch((error) => { + console.warn('[integrations] Failed to notify agents after broker start:', error instanceof Error ? error.message : String(error)) + }) + } return true }) From 3069ea5cb7a585e003b5248f4efb0d326c7bbdf8 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 5 Jun 2026 11:50:19 +0200 Subject: [PATCH 3/8] Deduplicate repeated PTY stream chunks --- src/main/broker.test.ts | 50 +++++++++++++++++++++++++++++++++++++++++ src/main/broker.ts | 43 +++++++++++++++++++++++++++++++++++ 2 files changed, 93 insertions(+) diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index cab9117d..ee53a0ba 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -442,6 +442,56 @@ describe('BrokerManager local + cloud coexistence', () => { await manager.shutdown() }) + it('dedupes repeated PTY chunks from overlapping event streams', async () => { + const manager = new BrokerManager() + const win = createMockWindow() + const local = await startLocalWithWindow(manager, win) + const listener = local.onEvent.mock.calls.at(-1)?.[0] + expect(listener).toBeTypeOf('function') + + const chunkEvent = { + kind: 'worker_stream', + name: 'claude-1', + chunk: 'pong\n', + seq: 22 + } + listener?.(chunkEvent) + listener?.(chunkEvent) + + const ptyCalls = (win.webContents.send as ReturnType).mock.calls + .filter(([channel]) => channel === 'broker:pty-chunk') + expect(ptyCalls).toEqual([['broker:pty-chunk', PROJECT_ID, 'claude-1', 'pong\n']]) + + await manager.shutdown() + }) + + it('keeps legitimate repeated PTY chunks with different broker sequences', async () => { + const manager = new BrokerManager() + const win = createMockWindow() + const local = await startLocalWithWindow(manager, win) + const listener = local.onEvent.mock.calls.at(-1)?.[0] + expect(listener).toBeTypeOf('function') + + listener?.({ + kind: 'worker_stream', + name: 'claude-1', + chunk: 'pong\n', + seq: 23 + }) + listener?.({ + kind: 'worker_stream', + name: 'claude-1', + chunk: 'pong\n', + seq: 24 + }) + + const ptyCalls = (win.webContents.send as ReturnType).mock.calls + .filter(([channel]) => channel === 'broker:pty-chunk') + expect(ptyCalls).toHaveLength(2) + + await manager.shutdown() + }) + it('refreshEventStream rebinds the harness stream from the last seen sequence', async () => { const manager = new BrokerManager() const local = await startLocal(manager) diff --git a/src/main/broker.ts b/src/main/broker.ts index 6d3f5f23..26789a9d 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -343,6 +343,9 @@ const BROKER_EVENT_HISTORY_TTL_MS = 12 * 60 * 60 * 1_000 const DEFAULT_RECONCILE_MESSAGE_LIMIT = 50 const MAX_RECONCILE_MESSAGE_LIMIT = 100 const EVENT_STREAM_REBIND_COOLDOWN_MS = 5_000 +const PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS = 60_000 +const PTY_CHUNK_CONTENT_DEDUPE_TTL_MS = 1_000 +const MAX_PTY_CHUNK_DEDUPE_ENTRIES = 2_000 // After this many consecutive failures to open a PTY input stream, give up on // the WS fast path for that agent briefly and send over HTTP while it cools down. const MAX_INPUT_STREAM_OPEN_FAILURES = 3 @@ -1097,6 +1100,7 @@ export class BrokerManager { private brokerTimeoutCounts = new Map() private eventObservers = new Set() private eventHistory: BrokerEventRecord[] = [] + private recentPtyChunks = new Map() private eventSerial = 0 get cwd(): string | null { @@ -1881,6 +1885,9 @@ export class BrokerManager { 'name' in event && typeof event.name === 'string' && 'chunk' in event && typeof event.chunk === 'string' ) { + if (this.isDuplicatePtyChunk(sessionKey, event.name, event.chunk, event)) { + return + } const targetWindow = this.windowForSession(sessionKey, win) if (targetWindow && !targetWindow.isDestroyed()) { targetWindow.webContents.send('broker:pty-chunk', projectId, event.name, event.chunk) @@ -1933,6 +1940,42 @@ export class BrokerManager { } } + private isDuplicatePtyChunk(sessionKey: string, name: string, chunk: string, event: BrokerEvent): boolean { + const now = Date.now() + for (const [key, seenAt] of this.recentPtyChunks) { + const ttl = key.includes(':chunk:') ? PTY_CHUNK_CONTENT_DEDUPE_TTL_MS : PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS + if ( + now - seenAt > ttl || + this.recentPtyChunks.size > MAX_PTY_CHUNK_DEDUPE_ENTRIES + ) { + this.recentPtyChunks.delete(key) + } + } + + const eventRecord = event as Record + const seq = typeof eventRecord.seq === 'number' || typeof eventRecord.seq === 'string' + ? String(eventRecord.seq) + : '' + const eventId = typeof eventRecord.event_id === 'string' + ? eventRecord.event_id + : typeof eventRecord.id === 'string' + ? eventRecord.id + : '' + const identity = eventId || (seq ? `seq:${seq}` : '') + const key = identity + ? `${sessionKey}:${name}:${identity}` + : `${sessionKey}:${name}:chunk:${chunk}` + + const previous = this.recentPtyChunks.get(key) + const ttl = identity ? PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS : PTY_CHUNK_CONTENT_DEDUPE_TTL_MS + if (previous !== undefined && now - previous <= ttl) { + return true + } + + this.recentPtyChunks.set(key, now) + return false + } + private publishBrokerEvent( sessionKey: string, projectId: string, From a9a61ff3556dd99a5a944608611bb2d25457af38 Mon Sep 17 00:00:00 2001 From: "agent-relay-code[bot]" Date: Fri, 5 Jun 2026 09:51:16 +0000 Subject: [PATCH 4/8] chore: apply pr-reviewer fixes for #89 --- src/main/broker.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/broker.ts b/src/main/broker.ts index 26789a9d..01dc36c7 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -1078,7 +1078,7 @@ function sessionKeyFor(session: BrokerSession): string { export class BrokerManager { private sessions = new Map() - private startPromises = new Map>() + private startPromises = new Map>() private revivePromises = new Map>() // Which broker sessions (by session key) an agent name is registered on. // Both a project's local and cloud brokers join the same relay workspace, From 9479ae640c6e7edb44f71f584c6ab76098415f59 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 5 Jun 2026 11:52:58 +0200 Subject: [PATCH 5/8] Use precise broker start promise type --- src/main/broker.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/broker.ts b/src/main/broker.ts index 01dc36c7..9f67d26c 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -1078,7 +1078,7 @@ function sessionKeyFor(session: BrokerSession): string { export class BrokerManager { private sessions = new Map() - private startPromises = new Map>() + private startPromises = new Map>() private revivePromises = new Map>() // Which broker sessions (by session key) an agent name is registered on. // Both a project's local and cloud brokers join the same relay workspace, From 14664fcad1f35ece88f8af3a334d004f2b3339f1 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 5 Jun 2026 11:56:03 +0200 Subject: [PATCH 6/8] Document duplicate event hardening rules --- .../memory/MEMORY.md | 1 + .../memory/duplicate-event-hardening.md | 12 ++++++++++++ AGENTS.md | 14 ++++++++++++++ 3 files changed, 27 insertions(+) create mode 100644 .claude/projects/-Users-khaliqgant-Projects-AgentWorkforce-pear/memory/duplicate-event-hardening.md create mode 100644 AGENTS.md diff --git a/.claude/projects/-Users-khaliqgant-Projects-AgentWorkforce-pear/memory/MEMORY.md b/.claude/projects/-Users-khaliqgant-Projects-AgentWorkforce-pear/memory/MEMORY.md index 4cd3c0af..5b48dbff 100644 --- a/.claude/projects/-Users-khaliqgant-Projects-AgentWorkforce-pear/memory/MEMORY.md +++ b/.claude/projects/-Users-khaliqgant-Projects-AgentWorkforce-pear/memory/MEMORY.md @@ -1,3 +1,4 @@ - [Cloud credential sources](cloud-credential-sources.md) — Pear's two auth stores and the resolveCloudAuth() helper that unifies them with ../workforce and ../cloud - [Cloud agent semantics](cloud-agent-semantics.md) — a Pear "cloud agent" is a provider credential; delete fixed, attach/warm still unbuilt - [Relayfile workspace model](relayfile-workspace-model.md) — relayWorkspaceId is a fake random UUID; integrations Connect must use the SDK; account-wide workspace decided; catalog logos/filter done +- [Duplicate event hardening](duplicate-event-hardening.md) — broker starts, event streams, PTY chunks, spawned personas, and integration notifications must be idempotent and replay-tolerant diff --git a/.claude/projects/-Users-khaliqgant-Projects-AgentWorkforce-pear/memory/duplicate-event-hardening.md b/.claude/projects/-Users-khaliqgant-Projects-AgentWorkforce-pear/memory/duplicate-event-hardening.md new file mode 100644 index 00000000..cb794cf7 --- /dev/null +++ b/.claude/projects/-Users-khaliqgant-Projects-AgentWorkforce-pear/memory/duplicate-event-hardening.md @@ -0,0 +1,12 @@ +# Duplicate Event Hardening + +Treat duplicate broker/event delivery as expected behavior in Pear. Renderer startup flows, dashboard reconnects, broker stream refreshes, relay replay, and spawned persona sessions can all surface the same logical event more than once. + +When changing broker start, event streaming, PTY output, spawned personas, or integration notifications: + +- Make side effects idempotent and return whether state actually changed before notifying agents or publishing metadata. +- Coalesce same-project start/attach calls with keyed in-flight promises. +- Dedupe live events by stable identity (`event_id`, `id`, or `seq`) before falling back to short content-based suppression. +- Use generation tokens for refreshed listeners so stale event streams cannot publish IPC output. +- Preserve PTY-specific duplicate guards in main and tolerate repeated chunk metadata in renderer buffers. +- Add regression tests for replay, reconnect, repeated `ensureBroker()`, and repeated terminal output cases. diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 00000000..8787e255 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,14 @@ +# Agent Instructions + +## Duplicate Event Hardening + +Pear broker work must treat duplicate delivery as a normal failure mode. Renderer effects, dashboard reconnects, broker event-stream refreshes, and relay replay can all cause the same logical event to be observed more than once. + +- Make lifecycle operations idempotent. `broker:start`, agent registration, integration notifications, and mount/link setup should return or record whether they actually changed state before triggering side effects. +- Coalesce concurrent starts or attaches with keyed in-flight promises. Repeated UI calls for the same project/root/channels should wait on the existing operation instead of starting another broker or event stream. +- Prefer stable event identity over content matching. PTY and broker events should carry `event_id`, `id`, or `seq`; dedupe by that identity first and use short content-based windows only as a fallback. +- Scope live event listeners with a generation token when reconnecting or refreshing streams. Stale callbacks from an older listener must bail before publishing IPC events. +- Keep PTY stream delivery separately guarded in main and renderer code. Main should suppress duplicate `worker_stream` chunks before `broker:pty-chunk`; renderer buffers should tolerate repeated chunk metadata as a final guardrail. +- Do not post integration or launch metadata on reused broker sessions. Notify agents only after a real broker start, reconnect, or state transition, and make repeated payloads no-ops when possible. +- Add regression tests when touching broker start, event streaming, PTY buffering, spawned personas, or integration notifications. Include duplicate/replay cases, not just the happy path. +- Add low-noise telemetry for suppressed duplicates and missing event identity so replay issues are visible without flooding the terminal. From ff297c6b69ac540b4990c1778cfde3b54d1d658d Mon Sep 17 00:00:00 2001 From: "agent-relay-code[bot]" Date: Fri, 5 Jun 2026 09:56:50 +0000 Subject: [PATCH 7/8] chore: apply pr-reviewer fixes for #89 --- src/main/broker.test.ts | 25 +++++++++++++++++++++++++ src/main/broker.ts | 16 ++++++---------- 2 files changed, 31 insertions(+), 10 deletions(-) diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index ee53a0ba..3194ad00 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -492,6 +492,31 @@ describe('BrokerManager local + cloud coexistence', () => { await manager.shutdown() }) + it('keeps repeated PTY chunks when broker events have no identity', async () => { + const manager = new BrokerManager() + const win = createMockWindow() + const local = await startLocalWithWindow(manager, win) + const listener = local.onEvent.mock.calls.at(-1)?.[0] + expect(listener).toBeTypeOf('function') + + listener?.({ + kind: 'worker_stream', + name: 'claude-1', + chunk: 'pong\n' + }) + listener?.({ + kind: 'worker_stream', + name: 'claude-1', + chunk: 'pong\n' + }) + + const ptyCalls = (win.webContents.send as ReturnType).mock.calls + .filter(([channel]) => channel === 'broker:pty-chunk') + expect(ptyCalls).toHaveLength(2) + + await manager.shutdown() + }) + it('refreshEventStream rebinds the harness stream from the last seen sequence', async () => { const manager = new BrokerManager() const local = await startLocal(manager) diff --git a/src/main/broker.ts b/src/main/broker.ts index 9f67d26c..b10331e6 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -344,7 +344,6 @@ const DEFAULT_RECONCILE_MESSAGE_LIMIT = 50 const MAX_RECONCILE_MESSAGE_LIMIT = 100 const EVENT_STREAM_REBIND_COOLDOWN_MS = 5_000 const PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS = 60_000 -const PTY_CHUNK_CONTENT_DEDUPE_TTL_MS = 1_000 const MAX_PTY_CHUNK_DEDUPE_ENTRIES = 2_000 // After this many consecutive failures to open a PTY input stream, give up on // the WS fast path for that agent briefly and send over HTTP while it cools down. @@ -1885,7 +1884,7 @@ export class BrokerManager { 'name' in event && typeof event.name === 'string' && 'chunk' in event && typeof event.chunk === 'string' ) { - if (this.isDuplicatePtyChunk(sessionKey, event.name, event.chunk, event)) { + if (this.isDuplicatePtyChunk(sessionKey, event.name, event)) { return } const targetWindow = this.windowForSession(sessionKey, win) @@ -1940,12 +1939,11 @@ export class BrokerManager { } } - private isDuplicatePtyChunk(sessionKey: string, name: string, chunk: string, event: BrokerEvent): boolean { + private isDuplicatePtyChunk(sessionKey: string, name: string, event: BrokerEvent): boolean { const now = Date.now() for (const [key, seenAt] of this.recentPtyChunks) { - const ttl = key.includes(':chunk:') ? PTY_CHUNK_CONTENT_DEDUPE_TTL_MS : PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS if ( - now - seenAt > ttl || + now - seenAt > PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS || this.recentPtyChunks.size > MAX_PTY_CHUNK_DEDUPE_ENTRIES ) { this.recentPtyChunks.delete(key) @@ -1962,13 +1960,11 @@ export class BrokerManager { ? eventRecord.id : '' const identity = eventId || (seq ? `seq:${seq}` : '') - const key = identity - ? `${sessionKey}:${name}:${identity}` - : `${sessionKey}:${name}:chunk:${chunk}` + if (!identity) return false + const key = `${sessionKey}:${name}:${identity}` const previous = this.recentPtyChunks.get(key) - const ttl = identity ? PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS : PTY_CHUNK_CONTENT_DEDUPE_TTL_MS - if (previous !== undefined && now - previous <= ttl) { + if (previous !== undefined && now - previous <= PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS) { return true } From 47107100f49be55d32f4ee76a0f86c6f98f59f1c Mon Sep 17 00:00:00 2001 From: "agent-relay-code[bot]" Date: Fri, 5 Jun 2026 09:58:55 +0000 Subject: [PATCH 8/8] chore: apply pr-reviewer fixes for #89 --- src/main/broker.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/broker.ts b/src/main/broker.ts index b10331e6..8709a712 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -1077,7 +1077,7 @@ function sessionKeyFor(session: BrokerSession): string { export class BrokerManager { private sessions = new Map() - private startPromises = new Map>() + private startPromises = new Map>() private revivePromises = new Map>() // Which broker sessions (by session key) an agent name is registered on. // Both a project's local and cloud brokers join the same relay workspace,