Skip to content

Deduplicate no-identity PTY chunks #91

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
khaliqgant merged 2 commits into from
Jun 5, 2026
Merged

Deduplicate no-identity PTY chunks #91

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
634e520
Deduplicate no-identity PTY chunks
khaliqgant Jun 5, 2026
f27fc13
chore: apply pr-reviewer fixes for #91
agent-relay-code[bot] Jun 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 64 additions & 2 deletions src/main/broker.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -492,7 +492,7 @@ describe('BrokerManager local + cloud coexistence', () => {
await manager.shutdown()
})

it('keeps repeated PTY chunks when broker events have no identity', async () => {
it('dedupes repeated PTY chunks when broker events have no identity', async () => {
const manager = new BrokerManager()
const win = createMockWindow()
const local = await startLocalWithWindow(manager, win)
Expand All @@ -512,7 +512,35 @@ describe('BrokerManager local + cloud coexistence', () => {

const ptyCalls = (win.webContents.send as ReturnType<typeof vi.fn>).mock.calls
.filter(([channel]) => channel === 'broker:pty-chunk')
expect(ptyCalls).toHaveLength(2)
expect(ptyCalls).toEqual([['broker:pty-chunk', PROJECT_ID, 'claude-1', 'pong\n']])

await manager.shutdown()
})

it('keeps distinct 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: 'po'
})
listener?.({
kind: 'worker_stream',
name: 'claude-1',
chunk: 'ng\n'
})

const ptyCalls = (win.webContents.send as ReturnType<typeof vi.fn>).mock.calls
.filter(([channel]) => channel === 'broker:pty-chunk')
expect(ptyCalls).toEqual([
['broker:pty-chunk', PROJECT_ID, 'claude-1', 'po'],
['broker:pty-chunk', PROJECT_ID, 'claude-1', 'ng\n']
])

await manager.shutdown()
})
Expand Down Expand Up @@ -625,6 +653,40 @@ describe('BrokerManager local + cloud coexistence', () => {

await manager.shutdown()
})

it('keeps repeated no-identity PTY chunks after intervening output', 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: '>'
})
listener?.({
kind: 'worker_stream',
name: 'claude-1',
chunk: ' '
})
listener?.({
kind: 'worker_stream',
name: 'claude-1',
chunk: '>'
})

const ptyCalls = (win.webContents.send as ReturnType<typeof vi.fn>).mock.calls
.filter(([channel]) => channel === 'broker:pty-chunk')
expect(ptyCalls).toEqual([
['broker:pty-chunk', PROJECT_ID, 'claude-1', '>'],
['broker:pty-chunk', PROJECT_ID, 'claude-1', ' '],
['broker:pty-chunk', PROJECT_ID, 'claude-1', '>']
])

await manager.shutdown()
})
})

describe('BrokerManager spawnAgent CLI preflight', () => {
Expand Down
35 changes: 29 additions & 6 deletions src/main/broker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -344,6 +344,7 @@ 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.
Expand Down Expand Up @@ -621,8 +622,12 @@ function isBrokerUnreachableError(err: unknown): boolean {
return someInCauseChain(err, (node) => {
const code = node.code
if (code === 'ECONNREFUSED' || code === 'ECONNRESET') return true
if (code === 'UND_ERR_SOCKET') {
const message = node.message
return typeof message === 'string' && /other side closed|socket closed/i.test(message)
}
const message = node.message
return typeof message === 'string' && /ECONNREFUSED|ECONNRESET/.test(message)
return typeof message === 'string' && /ECONNREFUSED|ECONNRESET|UND_ERR_SOCKET/.test(message)
})
}

Expand Down Expand Up @@ -1952,8 +1957,11 @@ export class BrokerManager {
private isDuplicatePtyChunk(sessionKey: string, name: string, event: BrokerEvent): boolean {
const now = Date.now()
for (const [key, seenAt] of this.recentPtyChunks) {
const ttl = key.startsWith('chunk:')
? PTY_CHUNK_CONTENT_DEDUPE_TTL_MS
: PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS
if (
now - seenAt > PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS ||
now - seenAt > ttl ||
this.recentPtyChunks.size > MAX_PTY_CHUNK_DEDUPE_ENTRIES
) {
this.recentPtyChunks.delete(key)
Comment on lines 1959 to 1967

@gemini-code-assist gemini-code-assist Bot Jun 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Iterating over the entire recentPtyChunks map (which can contain up to 2,000 entries) on every single PTY chunk arrival introduces a significant performance bottleneck. Since PTY chunks can arrive at an extremely high frequency, this $O(N)$ loop can lead to high CPU usage and UI/typing latency.

By only running the pruning loop when the map size actually exceeds MAX_PTY_CHUNK_DEDUPE_ENTRIES, we can keep the common case $O(1)$ and completely avoid unnecessary iterations.

    const shouldPrune = this.recentPtyChunks.size > MAX_PTY_CHUNK_DEDUPE_ENTRIES
    for (const [key, seenAt] of shouldPrune ? 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)

Expand All @@ -1970,14 +1978,28 @@ export class BrokerManager {
? eventRecord.id
: ''
const identity = eventId || (seq ? `seq:${seq}` : '')
if (!identity) return false

const key = `${sessionKey}:${name}:${identity}`
const chunk = typeof eventRecord.chunk === 'string' ? eventRecord.chunk : ''
if (!identity && !chunk) return false
Comment on lines +1981 to +1982

@gemini-code-assist gemini-code-assist Bot Jun 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

If a PTY chunk is extremely large (e.g., several kilobytes or megabytes of command output), storing the entire chunk string as a key in recentPtyChunks can consume significant memory and slow down map operations.

Since deduplication of no-identity chunks is primarily intended for small repeated terminal sequences (like spinner frames, prompt redraws, or cursor movements), we should skip deduplication for chunks exceeding a reasonable size threshold (e.g., 1024 characters).

    const chunk = typeof eventRecord.chunk === 'string' ? eventRecord.chunk : ''
    if (!identity && (!chunk || chunk.length > 1024)) return false


const contentKeyPrefix = `chunk:${sessionKey}:${name}:`
const key = identity
? `identity:${sessionKey}:${name}:${identity}`
: `${contentKeyPrefix}${chunk}`
const ttl = identity
? PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS
: PTY_CHUNK_CONTENT_DEDUPE_TTL_MS
const previous = this.recentPtyChunks.get(key)
if (previous !== undefined && now - previous <= PTY_CHUNK_IDENTITY_DEDUPE_TTL_MS) {
if (previous !== undefined && now - previous <= ttl) {
return true
}

if (!identity) {
for (const previousKey of this.recentPtyChunks.keys()) {
if (previousKey.startsWith(contentKeyPrefix)) {
this.recentPtyChunks.delete(previousKey)
}
}
}
this.recentPtyChunks.set(key, now)
return false
}
Expand Down Expand Up @@ -2669,6 +2691,7 @@ export class BrokerManager {
// and the broker's transient "internal channel closed" while it's
// tearing down. Both surface as noise that's not actionable.
if (isMissingAgentError(err)) return
if (isBrokerUnreachableError(err)) return

@codeant-ai codeant-ai Bot Jun 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggestion: This unconditionally suppresses unreachable transport errors in fire-and-forget mode, which makes real broker outages (ECONNREFUSED/ECONNRESET) silently drop user input with no warning signal. The shutdown-race suppression should stay narrowly scoped to known teardown-only errors, otherwise actionable failures become invisible. [logic error]

Severity Level: Critical 🚨 
- ❌ Keystrokes dropped silently when local broker unreachable.
- ⚠️ Users lack visibility into broker input failures.
Steps of Reproduction ✅ 
1. Open a project and attach a terminal in the renderer so that `useTerminal` at
`src/renderer/src/hooks/use-terminal.ts:180-199` is active; this hook defines `sendInput`
to call `pear.broker.sendInputFast(projectId, agentName!, data)` whenever the user types
or pastes (see also the global handlers at
`src/renderer/src/hooks/use-terminal.ts:500-563`).

2. On the preload/main boundary, `sendInputFast` in `src/preload/index.ts:24-27` forwards
input via `ipcRenderer.send('broker:send-input-fast', projectId, name, data)`, and the IPC
handler in `src/main/ipc-handlers.ts:40-42` calls
`brokerManager.sendInputFireAndForget(projectId, name, data)`, entering
`BrokerManager.sendInputFireAndForget` at `src/main/broker.ts:57-71` (PR hunk lines
2676-2690).

3. Induce a real broker outage—for example by killing the local Agent Relay broker process
while the session is still open or otherwise making the broker port unreachable. The next
keystroke from the terminal causes `sendInputFireAndForget` to call
`this.sendInput(projectId, trimmedName, data)` (at `src/main/broker.ts:62` in the
snippet), which ultimately calls `session.client.sendInput(trimmedName, data)` at
`src/main/broker.ts:49`. With the broker gone, the SDK surfaces a network error whose
cause has `code` `ECONNREFUSED`, `ECONNRESET`, or `UND_ERR_SOCKET`, so
`isBrokerUnreachableError` at `src/main/broker.ts:621-631` returns `true`.

4. That error is caught by the `.catch((err) => { ... })` in `sendInputFireAndForget` at
`src/main/broker.ts:62-71`; `isMissingAgentError(err)` is `false`, but the new `if
(isBrokerUnreachableError(err)) return` at PR line 2686 short-circuits the handler, so the
function returns without logging anything or surfacing an error. The keystroke is silently
dropped (no PTY write, no warning); yet the comment above (`2620-56`) still states that
"Errors get logged and the broker's existing stream-fallback logic in sendInput takes care
of reopening a broken stream." Recovery for unreachable brokers is implemented separately
in `withWedgeRecovery` at `src/main/broker.ts:2888-2973`, but `sendInputFireAndForget`
does not use that helper, so in this path broker outages result in unlogged, silently
discarded user input.

Fix in Cursor | Fix in VSCode Claude

(Use Cmd/Ctrl + Click for best experience)

Prompt for AI Agent 🤖 
This is a comment left during a code review.

**Path:** src/main/broker.ts
**Line:** 2686:2686
**Comment:**
	*Logic Error: This unconditionally suppresses unreachable transport errors in fire-and-forget mode, which makes real broker outages (`ECONNREFUSED`/`ECONNRESET`) silently drop user input with no warning signal. The shutdown-race suppression should stay narrowly scoped to known teardown-only errors, otherwise actionable failures become invisible.

Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix
👍 | 👎

const message = toErrorMessage(err)
if (/internal channel closed|internal reply dropped/i.test(message)) return
console.warn(`[broker] sendInputFireAndForget failed for ${trimmedName}:`, err)
Expand Down
Loading