Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 6 additions & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,3 +16,9 @@ Pear broker work must treat duplicate delivery as a normal failure mode. Rendere
## Terminal Screen Convergence

The broker daemon's PTY emulation (served as attach snapshots, observable via `agent-relay-broker dump-pty`) is the ground truth for what a worker terminal shows. The renderer's xterm grid can diverge from it (e.g. xterm reflow-scrolls on width resizes; PTY-side emulators don't), and diff-painting TUIs like Claude Code then *preserve* the divergence forever — they skip cells they believe unchanged, so stale glyphs bleed through new rows and stacked repaint frames accumulate. `src/renderer/src/lib/terminal-reconciler.ts` closes the loop: when a terminal is quiet it compares the viewport to the broker's `plain` snapshot and repaints from the self-framing `ansi` snapshot on confirmed divergence. Do not remove it after fixing any individual corruption vector — it is the convergence backstop for the whole class, and its gating invariants (quiet window, activity-serial recheck, dimension match, confirm-twice, rate limit) each guard a real re-corruption path documented in the module header. Repairs log `[terminal] viewport diverged from broker screen` — that line firing is the signal a new creation vector exists and is worth hunting.

The gates themselves must not be able to disable the backstop silently or permanently:

- **Persistent dims mismatch escalates, never just skips.** The dimension-match gate exists for resizes in flight, but a PTY that *stays* at a different size than the grid is the exact state that creates stacked-frame corruption (the TUI frames repaints for the wrong row count) — and it would also gate the reconciler off forever. After consecutive mismatched quiet checks the reconciler calls `onPersistentDimsMismatch`, which invalidates the size-sync ack and re-asserts the rendered grid as the PTY size (logs `[terminal] PTY RxC disagrees with grid RxC`).
- **Stranded predictions roll back instead of holding the quiet gate shut.** Optimistic-echo predictions are only confirmed or dropped by server output; a keystroke the TUI swallows (or one typed into a hung agent) leaves `hasPredictions` true with no echo ever coming, which would close the quiet gate permanently. After `STRANDED_PREDICTION_ROLLBACK_MS` of output silence the registry rolls the predictions back (erasing only optimistic glyphs — confirmed bytes are untouched) and lets reconciliation proceed.
- **A thrown check never kills the poll loop.** `snapshotTerminal` degrades to null, but the IPC bridge itself can reject; the check loop catches and rate-limit-logs instead of dying on an unhandled rejection.
1 change: 1 addition & 0 deletions src/main/__tests__/integration-event-bridge.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2344,6 +2344,7 @@
await harness.emit(changeEvent('/chat/channels/C123ABC/threads/1780607825_485189/replies/1780611452_510669.json', 'chat'))
await harness.emit(changeEvent('/chat/channels/C123ABC__proj-cloud/threads/1780607825_485189/replies/1780611452_510669.json', 'chat'))

await waitForSent(harness, 1)
assert.deepEqual(harness.sent.map((message) => message.input.to), ['alice'])
})

Expand Down Expand Up @@ -2417,7 +2418,7 @@
const workspaceRoot = join(homedir(), '.agentworkforce', 'pear', 'relayfile', 'workspaces', localMountWorkspaceId)
const localRoot = join(workspaceRoot, 'slack', 'channels', 'C123ABC', 'messages')
const localDraftPath = join(localRoot, 'codex-175-self-echo.json')
const remoteDraftPath = '/slack/channels/C123ABC/messages/codex-175-self-echo.json'

Check warning on line 2421 in src/main/__tests__/integration-event-bridge.test.ts

View workflow job for this annotation

GitHub Actions / checks

'remoteDraftPath' is assigned a value but never used
const outboundText = 'Pear posted this from an agent'
await mkdir(localRoot, { recursive: true })

Expand Down
12 changes: 12 additions & 0 deletions src/main/integration-remote-paths.ts
Original file line number Diff line number Diff line change
Expand Up @@ -136,3 +136,15 @@ export function isSlackDmListablePath(remotePath: string): boolean {
if (segments[1] === 'channels' && /^D/u.test(segments[2] ?? '')) return true
return false
}

// The adapter-linear `states` resource materializes the team's workflow
// states at /linear/states, but Linear integrations configure mount scopes
// like /linear/issues — the states subtree is reference data alongside those
// mounts, not a configurable mount of its own, so the mount-path scope checks
// above reject it. Recognize exactly that subtree so list/read (never write)
// can be allowed when a Linear integration is visible, mirroring the Slack DM
// carve-out.
export function isLinearStatesListablePath(remotePath: string): boolean {
const segments = (remotePath || '').trim().replace(/\/+$/, '').split('/').filter(Boolean)
return segments[0] === 'linear' && segments[1] === 'states'
}
110 changes: 106 additions & 4 deletions src/main/integrations.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,11 @@
writeFile: vi.fn(async (input: { workspaceId: string; path: string; baseRevision: string; content: string }) => {
writeFileCalls.push(input)
return { opId: 'op-1', status: 'queued' }
})
}),
listTree: vi.fn(async (_workspaceId: string, _options: { path: string; depth?: number; cursor?: string }) => ({

Check warning on line 66 in src/main/integrations.test.ts

View workflow job for this annotation

GitHub Actions / checks

'_options' is defined but never used

Check warning on line 66 in src/main/integrations.test.ts

View workflow job for this annotation

GitHub Actions / checks

'_workspaceId' is defined but never used
entries: [] as Array<{ path: string; type: 'file' | 'dir' }>,
nextCursor: null as string | null
}))
}
const shellOpenExternal = vi.fn(async () => undefined)
const workspaceHandle = {
Expand Down Expand Up @@ -335,6 +339,7 @@
mock.joinWorkspaceCalls.splice(0)
mock.relayClient.readFile.mockClear()
mock.relayClient.writeFile.mockClear()
mock.relayClient.listTree.mockClear()
mock.shellOpenExternal.mockClear()
mock.workspaceHandle.requestJson.mockReset()
mock.workspaceHandle.requestJson.mockImplementation(async (_request: { path: string }) => {
Expand Down Expand Up @@ -386,17 +391,23 @@
)
})

it('writes remote files through Relayfile using the configured project scope', async () => {
it('writes remote files at the current revision (read-then-write)', async () => {
const manager = new IntegrationsManager()

await expect(manager.writeRemoteFile('project-1', '/slack/channels/C123/messages/draft.json', '{"text":"hello"}'))
.resolves.toBeUndefined()

// The file exists remotely (mock read returns rev-1), so the write must
// carry that revision — not '0', which the server rejects with a 409 for
// any existing file (e.g. Issues status changes updating the issue record).
expect(mock.readFileCalls).toEqual([
{ workspaceId: 'account-workspace-id', path: '/slack/channels/C123/messages/draft.json' }
])
expect(mock.writeFileCalls).toEqual([
{
workspaceId: 'account-workspace-id',
path: '/slack/channels/C123/messages/draft.json',
baseRevision: '0',
baseRevision: 'rev-1',
content: '{"text":"hello"}'
}
])
Expand All @@ -405,12 +416,103 @@
workspaceId: 'account-workspace-id',
options: {
agentName: 'pear-integrations-writer',
scopes: ['relayfile:fs:write:/**']
scopes: ['relayfile:fs:read:/**', 'relayfile:fs:write:/**']
}
}
])
})

it('creates a new remote file with baseRevision 0 when the pre-write read 404s', async () => {
mock.relayClient.readFile.mockRejectedValueOnce(
Object.assign(new Error('not found'), { status: 404 })
)
const manager = new IntegrationsManager()

await expect(manager.writeRemoteFile('project-1', '/slack/channels/C123/messages/draft.json', '{"text":"hi"}'))
.resolves.toBeUndefined()

expect(mock.writeFileCalls).toEqual([
expect.objectContaining({ baseRevision: '0', content: '{"text":"hi"}' })
])
})

it('re-reads and retries exactly once on a revision conflict', async () => {
const fileAt = (revision: string) => ({
path: '/slack/channels/C123/messages/draft.json',
revision,
contentType: 'application/json',
content: '{}',
encoding: 'utf-8' as const
})
mock.relayClient.readFile
.mockResolvedValueOnce(fileAt('rev-1'))
.mockResolvedValueOnce(fileAt('rev-2'))
mock.relayClient.writeFile.mockRejectedValueOnce(
Object.assign(new Error('revision conflict'), { status: 409, code: 'revision_conflict' })
)
const manager = new IntegrationsManager()

await expect(manager.writeRemoteFile('project-1', '/slack/channels/C123/messages/draft.json', '{"a":1}'))
.resolves.toBeUndefined()

expect(
mock.relayClient.writeFile.mock.calls.map(([input]) => (input as { baseRevision: string }).baseRevision)
).toEqual(['rev-1', 'rev-2'])
})

it('does not swallow a second consecutive revision conflict', async () => {
const conflict = () =>
Object.assign(new Error('revision conflict'), { status: 409, code: 'revision_conflict' })
mock.relayClient.writeFile.mockRejectedValueOnce(conflict()).mockRejectedValueOnce(conflict())
const manager = new IntegrationsManager()

await expect(manager.writeRemoteFile('project-1', '/slack/channels/C123/messages/draft.json', '{}'))
.rejects.toMatchObject({ status: 409 })
expect(mock.relayClient.writeFile).toHaveBeenCalledTimes(2)
})

it('allows /linear/states list + read when a Linear integration is visible, but never write', async () => {
mock.store.projects[0].integrations.push({
id: 'linear-integration-1',
name: 'Linear',
type: 'linear',
provider: 'linear',
integrationId: 'linear-integration-1',
scope: {},
mountPaths: ['/linear/issues'],
connectedAt: '2026-06-05T00:00:00.000Z',
notifyAgent: true,
subscribeAgent: false,
downloadHistoricalData: false,
visibleInProject: true
})
mock.relayClient.listTree.mockResolvedValueOnce({
entries: [
{ path: '/linear/states', type: 'dir' },
{ path: '/linear/states/state-1.json', type: 'file' }
],
nextCursor: null
})
const manager = new IntegrationsManager()

await expect(manager.listRemoteDirectory('project-1', '/linear/states')).resolves.toEqual([
{ name: 'state-1.json', path: '/linear/states/state-1.json', type: 'file' }
])
await expect(manager.readRemoteFile('project-1', '/linear/states/state-1.json'))
.resolves.toMatchObject({ kind: 'text' })
// The carve-out is read-only: workflow states are reference data.
await expect(manager.writeRemoteFile('project-1', '/linear/states/state-1.json', '{}'))
.rejects.toThrow('Integration remote file is outside this project integration scope')
})

it('rejects /linear/states listing without a visible Linear integration', async () => {
const manager = new IntegrationsManager()

await expect(manager.listRemoteDirectory('project-1', '/linear/states'))
.rejects.toThrow('Integration remote directory is outside this project integration scope')
expect(mock.relayClient.listTree).not.toHaveBeenCalled()
})

it('rejects remote writes outside the configured project scope', async () => {
const manager = new IntegrationsManager()

Expand Down
50 changes: 45 additions & 5 deletions src/main/integrations.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import {
canListRemoteDirectoryForMountPaths,
canShowRemoteDirectoryEntryForMountPaths,
isRelayfilePathWithinRoot,
isLinearStatesListablePath,
isSlackDmListablePath,
normalizeRemoteDirectoryPath,
remotePathName
Expand Down Expand Up @@ -1111,7 +1112,8 @@ export class IntegrationsManager {
const allowSlackDms = this.slackDmListingEnabledForProject(projectId)
const withinScope =
mountPaths.some((mountPath) => isRelayfilePathWithinRoot(mountPath, path)) ||
(allowSlackDms && isSlackDmListablePath(path))
(allowSlackDms && isSlackDmListablePath(path)) ||
(this.linearStatesListingEnabledForProject(projectId) && isLinearStatesListablePath(path))
if (!withinScope) {
throw new Error('Integration remote file is outside this project integration scope')
}
Expand Down Expand Up @@ -1153,7 +1155,30 @@ export class IntegrationsManager {
}

await this.withIntegrationRemoteWriterHandle(async (handle) => {
await handle.client().writeFile({ workspaceId: handle.workspaceId, path, baseRevision: '0', content })
const client = handle.client()
// The server enforces optimistic concurrency: a write lands only when
// baseRevision matches the file's current revision, where '0' means
// "must not exist yet". Creates (e.g. Slack writeback message files)
// pass on '0', but updates to existing canonical files (e.g. an Issues
// status change writing stateId onto the issue record) need the live
// revision — read it first, and retry once if a concurrent sync bumps
// the file between our read and write.
const writeAtCurrentRevision = async (): Promise<void> => {
let baseRevision = '0'
try {
const existing = await client.readFile(handle.workspaceId, path)
baseRevision = existing.revision
Comment on lines +1169 to +1170

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

To ensure robust defensive programming, consider using optional chaining and a nullish coalescing operator when accessing existing.revision. This prevents potential runtime TypeError crashes if the file read response is unexpectedly empty or null.

Suggested change
const existing = await client.readFile(handle.workspaceId, path)
baseRevision = existing.revision
const existing = await client.readFile(handle.workspaceId, path)
baseRevision = existing?.revision ?? '0'

} catch (error) {
if (!isHttpStatus(error, 404)) throw error
}
await client.writeFile({ workspaceId: handle.workspaceId, path, baseRevision, content })
}
try {
await writeAtCurrentRevision()
} catch (error) {
if (!isHttpStatus(error, 409)) throw error
await writeAtCurrentRevision()
}
})
}

Expand All @@ -1163,7 +1188,12 @@ export class IntegrationsManager {
if (!path || path === '/') throw new Error('Integration remote directory path is required')
const mountPaths = this.listableRemoteMountPaths(projectId)
const allowSlackDms = this.slackDmListingEnabledForProject(projectId)
if (!canListRemoteDirectoryForMountPaths(path, mountPaths) && !(allowSlackDms && isSlackDmListablePath(path))) {
const allowLinearStates = this.linearStatesListingEnabledForProject(projectId)
if (
!canListRemoteDirectoryForMountPaths(path, mountPaths) &&
!(allowSlackDms && isSlackDmListablePath(path)) &&
!(allowLinearStates && isLinearStatesListablePath(path))
) {
throw new Error('Integration remote directory is outside this project integration scope')
}

Expand All @@ -1182,7 +1212,8 @@ export class IntegrationsManager {
if (entry.path === path) continue
if (
!canShowRemoteDirectoryEntryForMountPaths(entry.path, mountPaths) &&
!(allowSlackDms && isSlackDmListablePath(entry.path))
!(allowSlackDms && isSlackDmListablePath(entry.path)) &&
!(allowLinearStates && isLinearStatesListablePath(entry.path))
) {
continue
}
Expand Down Expand Up @@ -1599,7 +1630,7 @@ export class IntegrationsManager {
})
const handle = await setup.joinWorkspace(workspaceId, {
agentName: mode === 'read' ? 'pear-integrations-reader' : 'pear-integrations-writer',
scopes: mode === 'read' ? ['relayfile:fs:read:/**'] : ['relayfile:fs:write:/**']
scopes: mode === 'read' ? ['relayfile:fs:read:/**'] : ['relayfile:fs:read:/**', 'relayfile:fs:write:/**']
})
if (mode === 'read') this.integrationRemoteReaderHandle = handle
else this.integrationRemoteWriterHandle = handle
Expand Down Expand Up @@ -2301,6 +2332,15 @@ export class IntegrationsManager {
return this.visibleIntegrationsForProject(projectId).some((integration) => slackListenDms(integration))
}

// True when any visible Linear integration exists for the project. The
// /linear/states reference subtree is listable/readable (never writable)
// under exactly this condition — see isLinearStatesListablePath.
private linearStatesListingEnabledForProject(projectId: string): boolean {
return this.visibleIntegrationsForProject(projectId).some(
(integration) => toRelayfileProvider(integration.provider) === 'linear'
)
}

private async listSystemMessageAgents(
bridge: IntegrationSystemMessageBridge,
projectId: string,
Expand Down
88 changes: 88 additions & 0 deletions src/renderer/src/lib/terminal-reconciler.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@ import {
createTerminalReconciler,
screensMatch,
RECONCILE_CONFIRM_CHECKS,
RECONCILE_DIMS_KICK_GAP_MS,
RECONCILE_ERROR_LOG_GAP_MS,
RECONCILE_MIN_REPAIR_GAP_MS,
type ReconcileSnapshot,
type ReconcileViewport,
Expand Down Expand Up @@ -160,6 +162,92 @@ describe('createTerminalReconciler', () => {
reconciler.dispose()
})

it('requires fresh divergence confirmations after a dims-mismatch skip', async () => {
const { deps, repairs, state } = makeHarness()
const reconciler = createTerminalReconciler(deps)
diverge(state)
await reconciler.checkNow()
state.viewport = { rows: 3, cols: 10, lines: ['row one', 'GARBAGE tw', ''] }
await reconciler.checkNow()
state.viewport = { rows: 2, cols: 10, lines: ['row one', 'GARBAGE tw'] }
await reconciler.checkNow()
expect(repairs).toEqual([])
await reconciler.checkNow()
expect(repairs).toHaveLength(1)
reconciler.dispose()
})

it('escalates a persistent dims mismatch to onPersistentDimsMismatch, rate-limited', async () => {
const { deps, repairs, state } = makeHarness()
const kicks: Array<{ grid: { rows: number; cols: number }; snapshot: { rows: number; cols: number } }> = []
deps.onPersistentDimsMismatch = (grid, snapshot) => kicks.push({ grid, snapshot })
const reconciler = createTerminalReconciler(deps)
state.viewport = { rows: 3, cols: 10, lines: ['row one', 'row two', ''] }
await reconciler.checkNow()
// First sighting could be a resize still propagating — no kick yet.
expect(kicks).toEqual([])
await reconciler.checkNow()
expect(kicks).toEqual([
{ grid: { rows: 3, cols: 10 }, snapshot: { rows: 2, cols: 10 } }
])
// Still mismatched inside the kick gap: no re-kick.
await reconciler.checkNow()
expect(kicks).toHaveLength(1)
state.now += RECONCILE_DIMS_KICK_GAP_MS + 1
await reconciler.checkNow()
expect(kicks).toHaveLength(2)
// Dims mismatch never repaints.
expect(repairs).toEqual([])
reconciler.dispose()
})

it('resets the dims-mismatch streak once dimensions agree again', async () => {
const { deps, state } = makeHarness()
const kicks: number[] = []
deps.onPersistentDimsMismatch = () => kicks.push(1)
const reconciler = createTerminalReconciler(deps)
const matched = state.viewport!
state.viewport = { rows: 3, cols: 10, lines: ['row one', 'row two', ''] }
await reconciler.checkNow()
state.viewport = matched
await reconciler.checkNow()
state.viewport = { rows: 3, cols: 10, lines: ['row one', 'row two', ''] }
await reconciler.checkNow()
expect(kicks).toEqual([])
reconciler.dispose()
})

it('survives a throwing snapshot fetch and keeps checking (rate-limited log)', async () => {
const { deps, repairs, state } = makeHarness()
const logs: string[] = []
deps.log = (message) => logs.push(message)
let failNext = true
const baseFetch = deps.fetchSnapshot
deps.fetchSnapshot = async (format) => {
if (failNext) {
failNext = false
throw new Error('ipc bridge dead')
}
return baseFetch(format)
}
const reconciler = createTerminalReconciler(deps)
diverge(state)
await reconciler.checkNow() // throws internally; must not propagate
expect(logs.some((m) => m.includes('reconcile check failed'))).toBe(true)
await confirmCycles(reconciler)
expect(repairs).toHaveLength(1)
// Error log is rate-limited: a second failure inside the gap is silent.
const logCount = logs.filter((m) => m.includes('reconcile check failed')).length
failNext = true
await reconciler.checkNow()
expect(logs.filter((m) => m.includes('reconcile check failed'))).toHaveLength(logCount)
state.now += RECONCILE_ERROR_LOG_GAP_MS + 1
failNext = true
await reconciler.checkNow()
expect(logs.filter((m) => m.includes('reconcile check failed'))).toHaveLength(logCount + 1)
reconciler.dispose()
})

it('rate-limits repairs', async () => {
const { deps, repairs, state } = makeHarness()
const reconciler = createTerminalReconciler(deps)
Expand Down
Loading
Loading