fix: apply review feedback for the watcher, guarded-save and chat-start stacks (#31216)

Fixes bugs found reviewing #30270, #30605, #30885, #30417, #31113, #30415,
#30992 and #30997, and removes code those PRs made unnecessary.

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Neil
2026-10-20 14:07:27 -07:00
committed by GitHub
co-authored by Claude
parent e507a3342a
commit 5f40d0c88b
78 changed files with 802 additions and 512 deletions
+1
View File
@@ -181,6 +181,7 @@
"../src/main/codex/codex-managed-trust-reconciliation.ts",
"../src/main/provider-process/provider-process-exit-deadline.ts",
"../src/main/provider-process/managed-provider-process.ts",
"../src/main/provider-process/provider-error-chain.ts",
"../src/main/provider-process/provider-executable-missing.ts",
"../src/main/provider-process/provider-group-orphan-guard.ts",
"../src/main/provider-process/provider-process-row-read.ts",
+6 -1
View File
@@ -1,4 +1,5 @@
import type { AuthMethod } from './generated/acp-protocol.generated'
import { withProviderRequestTimeout } from '../provider-process/provider-request-timeout'
export class AcpRpcError extends Error {
constructor(
@@ -79,8 +80,12 @@ export class AcpConnectionClosedError extends Error {
}
export class AcpRequestTimeoutError extends Error {
constructor(readonly method: string) {
constructor(
readonly method: string,
timeoutMs: number
) {
super(`ACP request timed out: ${method}`)
this.name = 'AcpRequestTimeoutError'
withProviderRequestTimeout(this, { method, timeoutMs })
}
}
+12
View File
@@ -8,6 +8,7 @@ import {
AcpRpcError
} from './acp-errors'
import { AcpScriptedAgent, deferred, tick } from './acp-scripted-agent.test-support'
import { providerRequestTimeoutOf } from '../provider-process/provider-request-timeout'
const peers: AcpJsonRpcPeer[] = []
const agents: AcpScriptedAgent[] = []
@@ -218,6 +219,17 @@ describe('ACP JSON-RPC peer', () => {
expect(agent.frames.map((frame) => frame.method)).toEqual(['wait', 'next'])
})
it('names the request and its deadline on a timeout, so a failed start can say which stalled', async () => {
vi.useFakeTimers()
const { peer } = fixture()
const timedOut = peer.request('session/new', {}, { timeoutMs: 300 }).catch((error) => error)
await vi.advanceTimersByTimeAsync(300)
expect(providerRequestTimeoutOf(await timedOut)).toEqual({
method: 'session/new',
timeoutMs: 300
})
})
it('bounds incoming requests without expiring user decisions', async () => {
vi.useFakeTimers()
const signal = deferred<AbortSignal>()
+1 -1
View File
@@ -137,7 +137,7 @@ export class AcpJsonRpcPeer {
? undefined
: setTimeout(() => {
this.pending.delete(id)
const error = new AcpRequestTimeoutError(method)
const error = new AcpRequestTimeoutError(method, timeoutMs)
controller.abort(error)
reject(error)
}, timeoutMs)
@@ -8,6 +8,7 @@ import {
} from './claude-structured-session-test-support'
import type { ClaudeStructuredSessionEvent } from './claude-structured-session-state'
import { CLAUDE_INITIALIZE_TIMEOUT_MS } from './claude-structured-session-startup'
import { providerRequestTimeoutOf } from '../provider-process/provider-request-timeout'
const ACQUIRE = { identity: identityFor(), fence: 7, spawnToken: 'spawn-9' }
@@ -126,11 +127,15 @@ describe('Claude host-awaited open', () => {
const claude = fakeClaude({ initDelayMs: 10 * 60_000 })
const { adapter } = adapterFor(claude)
await adapter.acquire(ACQUIRE)
const open = expect(claudeOpened(adapter, 'session-1')).rejects.toThrow(
'did not answer initialize'
)
const open = claudeOpened(adapter, 'session-1').catch((error: unknown) => error)
await vi.advanceTimersByTimeAsync(CLAUDE_INITIALIZE_TIMEOUT_MS)
await open
const error = await open
expect(error).toBeInstanceOf(Error)
expect(String(error)).toContain('did not answer initialize')
expect(providerRequestTimeoutOf(error)).toEqual({
method: 'initialize',
timeoutMs: CLAUDE_INITIALIZE_TIMEOUT_MS
})
await adapter.closeSession('session-1')
expect(claude.connections[0].closeCount).toBe(1)
})
@@ -3,6 +3,7 @@
import type { AgentSessionAccountKind } from '../../shared/agent-session-availability'
import type { StructuredAgentSessionProviderOpenResult } from '../native-chat/agent-session-wire/structured-agent-session-provider-open-contract'
import { withProviderRequestTimeout } from '../provider-process/provider-request-timeout'
import type { ClaudeStreamJsonConnection } from './claude-stream-json-connection'
import { ClaudeSlashCommandCatalog } from './claude-slash-command-catalog'
import {
@@ -92,7 +93,12 @@ export async function readClaudeStartupFacts(input: {
timer = setTimeout(
() =>
reject(
new Error(`claude did not answer initialize within ${CLAUDE_INITIALIZE_TIMEOUT_MS}ms`)
withProviderRequestTimeout(
new Error(
`claude did not answer initialize within ${CLAUDE_INITIALIZE_TIMEOUT_MS}ms`
),
{ method: 'initialize', timeoutMs: CLAUDE_INITIALIZE_TIMEOUT_MS }
)
),
CLAUDE_INITIALIZE_TIMEOUT_MS
)
+5 -11
View File
@@ -7,6 +7,7 @@ import {
resolveProviderChildEnv,
type ProviderProcessLaunch
} from '../provider-process/provider-process-launch'
import { findInErrorChain } from '../provider-process/provider-error-chain'
export class CodexCliInstallationError extends AgentSessionPreSpawnError {
readonly codexInstallation: CodexInstallationProblem
@@ -26,17 +27,10 @@ export class CodexCliInstallationError extends AgentSessionPreSpawnError {
}
}
export function codexInstallationProblemOf(
error: unknown,
depth = 0
): CodexInstallationProblem | undefined {
if (depth >= 6 || !(error instanceof Error)) {
return undefined
}
if (error instanceof CodexCliInstallationError) {
return error.codexInstallation
}
return codexInstallationProblemOf(error.cause, depth + 1)
export function codexInstallationProblemOf(error: unknown): CodexInstallationProblem | undefined {
return findInErrorChain(error, (inner) =>
inner instanceof CodexCliInstallationError ? inner.codexInstallation : undefined
)
}
export async function requireSupportedCodexCli(
@@ -31,6 +31,7 @@ const ARGS = { worktreePath: WORKTREE_PATH, connectionId: 'conn-1' }
const OVERFLOW_PAYLOAD = {
worktreePath: WORKTREE_PATH,
events: [{ kind: 'overflow', absolutePath: WORKTREE_PATH }],
resync: true,
connectionId: 'conn-1'
}
const RETRY_GIVE_UP_MS = 61_000
@@ -48,7 +48,8 @@ export function scheduleLocalCapacityRetry(
if (root?.listeners.has(listener.id) && !listener.isDestroyed()) {
listener.send('fs:changed', {
worktreePath: root.rootPath,
events: [{ kind: 'overflow', absolutePath: root.rootPath }]
events: [{ kind: 'overflow', absolutePath: root.rootPath }],
resync: true
} satisfies FsChangedPayload)
}
})
@@ -156,7 +156,8 @@ export async function restoreLocalWatcherAfterFailedRemoval(worktreePath: string
}
sender.send('fs:changed', {
worktreePath: suspended.worktreePath,
events: [{ kind: 'overflow', absolutePath: suspended.worktreePath }]
events: [{ kind: 'overflow', absolutePath: suspended.worktreePath }],
resync: true
} satisfies FsChangedPayload)
} catch (error) {
if (!isCurrentWatcherSender(sender, signal)) {
@@ -678,7 +678,8 @@ describe('local filesystem watcher unsubscribe cleanup', () => {
expect(subscribeParcelWatcher).toHaveBeenCalledTimes(2)
expect(sender.send).toHaveBeenCalledWith('fs:changed', {
worktreePath: '/tmp/repo',
events: [{ kind: 'overflow', absolutePath: '/tmp/repo' }]
events: [{ kind: 'overflow', absolutePath: '/tmp/repo' }],
resync: true
})
})
@@ -85,7 +85,8 @@ describe('native filesystem watcher capacity recovery', () => {
await vi.waitFor(() =>
expect(sender.send).toHaveBeenCalledWith('fs:changed', {
worktreePath: '/tmp/native-capacity-root',
events: [{ kind: 'overflow', absolutePath: '/tmp/native-capacity-root' }]
events: [{ kind: 'overflow', absolutePath: '/tmp/native-capacity-root' }],
resync: true
})
)
@@ -1,10 +1,6 @@
import type { WebContents } from 'electron'
import type { FsChangedPayload } from '../../shared/filesystem-entry-types'
import { isWatchRootCapacityRefusal } from '../../shared/watch-root-capacity-refusal'
import {
WATCH_BATCH_INTERVAL_MS,
WATCH_BATCH_LEADING_MS
} from '../../shared/filesystem-watch-batch-window'
import { getSshFilesystemProvider } from '../providers/ssh-filesystem-dispatch'
import { MAX_BATCHED_WATCHER_EVENTS } from './filesystem-watcher-event-batch'
import { createRemoteWatcherEventBatch } from './remote-watcher-event-batch'
@@ -172,8 +168,6 @@ async function doInstallRemoteWatcher(
// still covers a flush that races a close.
const batch = createRemoteWatcherEventBatch({
rootPath: worktreePath,
leadingMs: WATCH_BATCH_LEADING_MS,
intervalMs: WATCH_BATCH_INTERVAL_MS,
maxEvents: MAX_BATCHED_WATCHER_EVENTS,
deliver: (events) => {
const state = watcherLifecycleState.remoteWatchers.get(key)
@@ -81,6 +81,7 @@ describe('remote filesystem watcher re-arm', () => {
expect(senderOne.send).toHaveBeenCalledWith('fs:changed', {
worktreePath: '/home/me/repo',
events: [{ kind: 'overflow', absolutePath: '/home/me/repo' }],
resync: true,
connectionId: 'conn-1'
})
@@ -119,7 +119,8 @@ export async function restoreRemoteWatcherAfterFailedRemoval(
sender.send('fs:changed', {
worktreePath,
events: [{ kind: 'overflow', absolutePath: worktreePath }],
connectionId
connectionId,
resync: true
} satisfies FsChangedPayload)
}
// Why: an overlapping retried removal may have re-suspended owners into this same entry.
@@ -32,7 +32,8 @@ function flushRemoteWatcherResync(key: string): void {
listener.send('fs:changed', {
worktreePath: state.worktreePath,
events: [{ kind: 'overflow', absolutePath: state.worktreePath }],
connectionId: getRemoteWatcherKeyConnectionId(key)
connectionId: getRemoteWatcherKeyConnectionId(key),
resync: true
} satisfies FsChangedPayload)
sent = true
} catch (error) {
@@ -83,7 +83,8 @@ export function scheduleRemoteWatcherRetryCore(
listener.send('fs:changed', {
worktreePath,
events: [{ kind: 'overflow', absolutePath: worktreePath }],
connectionId
connectionId,
resync: true
} satisfies FsChangedPayload)
}
// Why: overflow only refreshes once — without this the watch stays dead until the app restarts.
@@ -36,6 +36,7 @@ const ARGS = { worktreePath: WORKTREE_PATH, connectionId: 'conn-1' }
const OVERFLOW_PAYLOAD = {
worktreePath: WORKTREE_PATH,
events: [{ kind: 'overflow', absolutePath: WORKTREE_PATH }],
resync: true,
connectionId: 'conn-1'
}
+4
View File
@@ -316,6 +316,7 @@ describe('registerFilesystemWatcherHandlers', () => {
expect(sender.send).toHaveBeenCalledWith('fs:changed', {
worktreePath: '/home/me/repo',
events: [{ kind: 'overflow', absolutePath: '/home/me/repo' }],
resync: true,
connectionId: 'conn-1'
})
)
@@ -375,6 +376,7 @@ describe('registerFilesystemWatcherHandlers', () => {
expect(sender.send).toHaveBeenCalledWith('fs:changed', {
worktreePath: '/home/me/repo',
events: [{ kind: 'overflow', absolutePath: '/home/me/repo' }],
resync: true,
connectionId: 'conn-1'
})
@@ -447,6 +449,7 @@ describe('registerFilesystemWatcherHandlers', () => {
expect(sender.send).toHaveBeenCalledWith('fs:changed', {
worktreePath: '/home/me/repo',
events: [{ kind: 'overflow', absolutePath: '/home/me/repo' }],
resync: true,
connectionId: 'conn-1'
})
}
@@ -577,6 +580,7 @@ describe('registerFilesystemWatcherHandlers', () => {
expect(sender.send).toHaveBeenCalledWith('fs:changed', {
worktreePath: '/home/me/repo',
events: [{ kind: 'overflow', absolutePath: '/home/me/repo' }],
resync: true,
connectionId: 'conn-1'
})
const replacementEvents = watchMock.mock.calls[1][1] as (events: unknown[]) => void
@@ -1,6 +1,10 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import type { FsChangeEvent } from '../../shared/filesystem-entry-types'
import { createRemoteWatcherEventBatch } from './remote-watcher-event-batch'
import {
WATCH_BATCH_INTERVAL_MS,
WATCH_BATCH_LEADING_MS
} from '../../shared/filesystem-watch-batch-window'
const ROOT = '/home/u/repo'
@@ -8,8 +12,6 @@ function makeBatch(deliver: (events: FsChangeEvent[]) => void, maxEvents = 5_000
return createRemoteWatcherEventBatch({
rootPath: ROOT,
deliver,
leadingMs: 25,
intervalMs: 100,
maxEvents
})
}
@@ -33,7 +35,7 @@ describe('createRemoteWatcherEventBatch', () => {
}
expect(deliver).not.toHaveBeenCalled()
vi.advanceTimersByTime(5)
vi.advanceTimersByTime(WATCH_BATCH_LEADING_MS - 20)
expect(deliver).toHaveBeenCalledTimes(1)
expect(deliver.mock.calls[0][0]).toHaveLength(20)
})
@@ -43,10 +45,10 @@ describe('createRemoteWatcherEventBatch', () => {
const batch = makeBatch(deliver)
batch.push([{ kind: 'update', absolutePath: `${ROOT}/a.ts` }])
vi.advanceTimersByTime(25)
vi.advanceTimersByTime(WATCH_BATCH_LEADING_MS)
expect(deliver).toHaveBeenCalledTimes(1)
batch.push([{ kind: 'update', absolutePath: `${ROOT}/b.ts` }])
vi.advanceTimersByTime(99)
vi.advanceTimersByTime(WATCH_BATCH_INTERVAL_MS - 1)
expect(deliver).toHaveBeenCalledTimes(1)
vi.advanceTimersByTime(1)
expect(deliver).toHaveBeenCalledTimes(2)
@@ -301,8 +303,6 @@ describe('createRemoteWatcherEventBatch', () => {
const batch = createRemoteWatcherEventBatch({
rootPath: 'C:\\Repo',
deliver,
leadingMs: 25,
intervalMs: 100,
maxEvents: 5_000
})
+1 -8
View File
@@ -8,8 +8,6 @@ import {
export type RemoteWatcherEventBatchOptions = {
rootPath: string
deliver: (events: FsChangeEvent[]) => void
leadingMs: number
intervalMs: number
maxEvents: number
}
@@ -72,8 +70,6 @@ function coalesceRemoteEvents(raw: FsChangeEvent[], rootPath: string): FsChangeE
export function createRemoteWatcherEventBatch({
rootPath,
deliver,
leadingMs,
intervalMs,
maxEvents
}: RemoteWatcherEventBatchOptions): RemoteWatcherEventBatch {
let buffered: FsChangeEvent[] = []
@@ -109,10 +105,7 @@ export function createRemoteWatcherEventBatch({
if (timer) {
return
}
timer = setTimeout(
flush,
watchBatchFlushDelayMs(Date.now(), lastFlushAt, leadingMs, intervalMs)
)
timer = setTimeout(flush, watchBatchFlushDelayMs(Date.now(), lastFlushAt))
}
return {
@@ -234,7 +234,7 @@ export function createStructuredAgentSessionConversationLifetime(host: {
sessionId: string,
failed: StructuredAgentSessionProviderChildIdentity,
error: unknown,
openMs?: number
openMs: number
): Promise<void> =>
serialize(sessionId, async () => {
const session = sessions.get(sessionId)
@@ -249,7 +249,7 @@ export function createStructuredAgentSessionConversationLifetime(host: {
sessionId,
failure: failure.kind,
...failure.requestTimeout,
...(openMs === undefined ? {} : { openMs }),
openMs,
error
})
try {
@@ -44,7 +44,7 @@ export class StructuredAgentSessionEventRecovery {
sessionId: string,
child: StructuredAgentSessionProviderChildIdentity,
error: unknown,
openMs?: number
openMs: number
) => Promise<void>
trackOpen: <T>(open: Promise<T>) => Promise<T>
}
@@ -22,6 +22,7 @@ import {
import { AgentSessionAcquisitionRefusal } from './structured-agent-session-adapter'
import { providerExecutableMissing } from '../../provider-process/provider-executable-missing'
import { providerRequestTimeoutOf } from '../../provider-process/provider-request-timeout'
import { findInErrorChain } from '../../provider-process/provider-error-chain'
import { argumentProblemOf } from '../structured-agent-arguments-error'
import { codexInstallationProblemOf } from '../../codex/codex-cli-installation-error'
@@ -53,22 +54,13 @@ export function withObservedProviderExit<TError extends Error>(error: TError): T
return Object.assign(error, { providerExitObserved: true })
}
/** Whether the adapter saw the child exit on its own. Follows `cause` and the errors a cleanup
* aggregated, since the acquisition errors wrap what the adapter threw. */
export function providerExitObserved(error: unknown, depth = 0): boolean {
if (depth >= 6 || !(error instanceof Error)) {
return false
}
if ('providerExitObserved' in error && error.providerExitObserved === true) {
return true
}
if (
error instanceof AggregateError &&
error.errors.some((inner) => providerExitObserved(inner, depth + 1))
) {
return true
}
return providerExitObserved(error.cause, depth + 1)
/** Whether the error's chain marks an exit the adapter saw on its own. */
export function providerExitObserved(error: unknown): boolean {
return (
findInErrorChain(error, (inner) =>
'providerExitObserved' in inner && inner.providerExitObserved === true ? true : undefined
) ?? false
)
}
/** A start that did not land. A refusal the adapter typed keeps its situation, and an exit the
@@ -96,17 +88,11 @@ export function providerStartupFailureFact(cause?: unknown): SubmissionRejection
if (providerExecutableMissing(cause)) {
return agentSessionFailureFact('cliMissing')
}
const detail = providerDiagnosticOf(cause)
if (providerExitObserved(cause)) {
return agentSessionFailureFact('providerStartFailed', {
detail,
argumentProblem: argumentProblemOf(cause)
})
}
return agentSessionFailureFact('startFailed', {
detail,
const exited = providerExitObserved(cause)
return agentSessionFailureFact(exited ? 'providerStartFailed' : 'startFailed', {
detail: providerDiagnosticOf(cause),
argumentProblem: argumentProblemOf(cause),
requestTimeout: providerRequestTimeoutOf(cause)
...(exited ? {} : { requestTimeout: providerRequestTimeoutOf(cause) })
})
}
@@ -68,9 +68,6 @@ export function deferredStructuredAgentSessionLogger(
}
}
/** Per-attempt timings: written, but a repeat that differs only in them is still a repeat. */
const UNKEYED_TIMING_FIELDS: ReadonlySet<string> = new Set(['openMs'])
/** The production logger: a failed span in the local trace file, plus the console. A repeat of
* the same entry is written at most once per window, carrying how many were swallowed. */
export function createStructuredAgentSessionLogger(options?: {
@@ -81,14 +78,18 @@ export function createStructuredAgentSessionLogger(options?: {
(level: LogLevel) =>
(message: string, fields: StructuredAgentSessionLogFields): void => {
const { scope, error, ...rest } = fields
const attributes = { level, message, ...rest, ...errorAttributes(error) }
// Keyed on all the entry writes but its stack and timings: only the same failure repeats.
const attributes: Record<string, unknown> = {
level,
message,
...rest,
...errorAttributes(error)
}
// Keyed on all the entry writes but its stack and timing: only the same failure repeats.
const { openMs: _openMs, ...keyed } = attributes
const suppressed = repeats.admit(
stableJson([
scope,
Object.fromEntries(
Object.entries(attributes).filter(([name]) => !UNKEYED_TIMING_FIELDS.has(name))
),
keyed,
error instanceof Error ? `${error.name}: ${error.message}` : null
])
)
@@ -12,7 +12,7 @@ export async function openStructuredAgentSessionProviderChild(
sessionId: string,
child: StructuredAgentSessionProviderChildIdentity,
error: unknown,
openMs?: number
openMs: number
) => Promise<void>
},
sessionId: string,
@@ -1,4 +1,5 @@
import type { AgentSessionArgumentProblem } from '../../shared/agent-session-failure'
import { findInErrorChain } from '../provider-process/provider-error-chain'
/** A saved Arguments refusal with only the option name, never its value. */
export class StructuredAgentArgumentsError extends Error {
@@ -22,23 +23,8 @@ export class StructuredAgentArgumentsError extends Error {
}
/** Finds a typed Arguments refusal through acquisition wrappers. */
export function argumentProblemOf(
error: unknown,
depth = 0
): AgentSessionArgumentProblem | undefined {
if (depth >= 6 || !(error instanceof Error)) {
return undefined
}
if (error instanceof StructuredAgentArgumentsError) {
return error.argumentProblem
}
if (error instanceof AggregateError) {
for (const inner of error.errors) {
const problem = argumentProblemOf(inner, depth + 1)
if (problem) {
return problem
}
}
}
return argumentProblemOf(error.cause, depth + 1)
export function argumentProblemOf(error: unknown): AgentSessionArgumentProblem | undefined {
return findInErrorChain(error, (inner) =>
inner instanceof StructuredAgentArgumentsError ? inner.argumentProblem : undefined
)
}
@@ -0,0 +1,24 @@
/** The first value `pick` reads from `error`, its `cause`s or the errors a cleanup aggregated,
* since acquisition errors wrap what a connection or adapter threw. Bounded to six links. */
export function findInErrorChain<T>(
error: unknown,
pick: (error: Error) => T | undefined,
depth = 0
): T | undefined {
if (depth >= 6 || !(error instanceof Error)) {
return undefined
}
const found = pick(error)
if (found !== undefined) {
return found
}
if (error instanceof AggregateError) {
for (const inner of error.errors) {
const nested = findInErrorChain(inner, pick, depth + 1)
if (nested !== undefined) {
return nested
}
}
}
return findInErrorChain(error.cause, pick, depth + 1)
}
@@ -1,5 +1,6 @@
import { existsSync } from 'node:fs'
import { isAbsolute } from 'node:path'
import { findInErrorChain } from './provider-error-chain'
// A provider CLI that is not installed, told apart from every other failed start: its own spawn
// reported ENOENT. Typed where the spawn is observed, then read wherever the start is worded.
@@ -25,19 +26,13 @@ export function withMissingProviderExecutable<TError extends Error>(error: TErro
return Object.assign(error, { providerExecutableMissing: true })
}
/** Follows `cause` and aggregated errors, since the acquisition errors wrap what a connection threw. */
export function providerExecutableMissing(error: unknown, depth = 0): boolean {
if (depth >= 6 || !(error instanceof Error)) {
return false
}
if ('providerExecutableMissing' in error && error.providerExecutableMissing === true) {
return true
}
if (
error instanceof AggregateError &&
error.errors.some((inner) => providerExecutableMissing(inner, depth + 1))
) {
return true
}
return providerExecutableMissing(error.cause, depth + 1)
/** Whether the error's chain marks a missing executable. */
export function providerExecutableMissing(error: unknown): boolean {
return (
findInErrorChain(error, (inner) =>
'providerExecutableMissing' in inner && inner.providerExecutableMissing === true
? true
: undefined
) ?? false
)
}
@@ -3,9 +3,9 @@ import { describe, expect, it, vi } from 'vitest'
import {
guardSupervisedProviderGroup,
sweepOrphanedProviderGroup,
type ProcessRowRead,
type ProviderGroupSweepVerdict
} from './provider-group-orphan-guard'
import type { ProcessRowRead } from './provider-process-row-read'
const STARTED = 'Mon Oct 19 10:00:00 2026'
const identity = { pgid: 4242, startedAt: STARTED }
@@ -238,6 +238,27 @@ describe('guardSupervisedProviderGroup', () => {
expect(signalGroup).not.toHaveBeenCalled()
})
it('stops waiting out the start second as soon as the supervisor exits', async () => {
const child = new FakeSupervisor()
const readProcessRow = vi.fn(async () => row({ ppid: 777 }))
const swept = new Promise((resolve) =>
guardSupervisedProviderGroup(child, {
site: 'test',
supervised: true,
readProcessRow,
now: () => Date.parse(STARTED) + 100,
onSwept: resolve
})
)
child.report({ pgid: 4242 })
await vi.waitFor(() => expect(readProcessRow).toHaveBeenCalledTimes(1))
const exitedAt = performance.now()
child.emit('exit', null, 'SIGKILL')
await expect(swept).resolves.toBe('unproven')
expect(performance.now() - exitedAt).toBeLessThan(500)
expect(readProcessRow).toHaveBeenCalledTimes(1)
})
it('warns once and watches nothing for a supervised spawn without the report channel', () => {
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
const child = new FakeSupervisor()
@@ -1,9 +1,8 @@
import { hasUnambiguousStartTime } from '../pty-descendant-termination'
import { recordSelfInitiatedTreeKill } from '../crash-reporting/self-initiated-tree-kill-log'
import { setTimeout as delay } from 'node:timers/promises'
import { readPosixProcessRow, type ProcessRowRead } from './provider-process-row-read'
export type { ProcessRowRead } from './provider-process-row-read'
// Covers the one stop the supervisor cannot make itself: its own SIGKILL leaves the provider's
// detached group running, so Orca, which outlives it, finishes the stop. POSIX only: Windows
// providers have no supervisor.
@@ -25,7 +24,8 @@ export type ProviderGroupGuardDeps = {
readProcessRow?: (pid: number) => Promise<ProcessRowRead>
signalGroup?: (pgid: number, signal: NodeJS.Signals | 0) => void
now?: () => number
sleep?: (ms: number) => Promise<void>
/** May settle either way once `signal` aborts; the proof checks it after. */
sleep?: (ms: number, signal: AbortSignal) => Promise<void>
onProven?: (identity: ProviderGroupIdentity) => void
/** `unproven`: no group was reported and proven ours, so nothing was swept. */
onSwept?: (verdict: ProviderGroupSweepVerdict | 'unproven') => void
@@ -103,7 +103,7 @@ export function guardSupervisedProviderGroup(
}
let identity: Promise<ProviderGroupIdentity | null> = Promise.resolve(null)
// A gone supervisor parents nothing, so a re-read it would wait for can only fail; skip it.
const proof = createProofWait(deps.sleep)
const proof = new AbortController()
let channelClosed = false
let onChannelClosed = (): void => {}
child.once('disconnect', () => {
@@ -113,7 +113,7 @@ export function guardSupervisedProviderGroup(
child.once('message', (message: unknown) => {
const pgid = reportedGroup(message)
if (pgid !== null) {
identity = proveReportedGroup(pgid, supervisorPid, deps, proof)
identity = proveReportedGroup(pgid, supervisorPid, deps, proof.signal)
void identity.then((proven) => proven && deps.onProven?.(proven))
}
})
@@ -129,7 +129,7 @@ export function guardSupervisedProviderGroup(
}
})
child.once('exit', () => {
proof.stop()
proof.abort()
// Exit can be observed before the report is delivered.
void untilChannelClosed()
.then(() => identity)
@@ -143,16 +143,18 @@ export function guardSupervisedProviderGroup(
* The leader's identity, read while this supervisor still parents it (so its pid was not reused).
* lstart has one-second resolution, so a start in the read's own second is re-proven after that
* second ends; otherwise a pid reused within that second would carry the same start time. A
* supervisor gone by then no longer parents it, so nothing is armed.
* supervisor gone by then no longer parents it, so nothing is armed. Accepted: on macOS a
* supervisor SIGKILLed within its provider's first second leaves that group unswept.
*/
async function proveReportedGroup(
pgid: number,
supervisorPid: number,
deps: ProviderGroupGuardDeps,
proof: ProofWait
signal: AbortSignal
): Promise<ProviderGroupIdentity | null> {
const readRow = deps.readProcessRow ?? readPosixProcessRow
const now = deps.now ?? Date.now
const sleep = deps.sleep ?? sleepUnref
for (let attempt = 0; attempt < IDENTITY_READ_ATTEMPTS; attempt += 1) {
// The boundary before ps starts, as the descendant sweep takes it.
const capturedAtMs = now()
@@ -166,42 +168,16 @@ async function proveReportedGroup(
) {
return { pgid, startedAt: row.startedAt }
}
if (!(await proof.wait(1_000 - (now() % 1_000) + SECOND_BOUNDARY_MARGIN_MS))) {
await sleep(1_000 - (now() % 1_000) + SECOND_BOUNDARY_MARGIN_MS, signal).catch(() => {})
if (signal.aborted) {
return null
}
}
return null
}
type ProofWait = { wait(ms: number): Promise<boolean>; stop(): void }
/** Waits that end early, false, once stopped; their timers never outlive the stop. */
function createProofWait(sleep?: (ms: number) => Promise<void>): ProofWait {
let stopped = false
let cancel = (): void => {}
return {
async wait(ms) {
if (stopped) {
return false
}
if (sleep) {
await sleep(ms)
return !stopped
}
return new Promise<boolean>((resolve) => {
const timer = setTimeout(() => resolve(!stopped), ms)
timer.unref?.()
cancel = () => {
clearTimeout(timer)
resolve(false)
}
})
},
stop() {
stopped = true
cancel()
}
}
function sleepUnref(ms: number, signal: AbortSignal): Promise<void> {
return delay(ms, undefined, { signal, ref: false })
}
function hasReportChannel(child: GuardedChild): boolean {
@@ -5,6 +5,7 @@ import {
readAgentSessionRequestTimeout,
type AgentSessionRequestTimeout
} from '../../shared/agent-session-request-timeout'
import { findInErrorChain } from './provider-error-chain'
/** Marks the error a provider request's own deadline rejected with. */
export function withProviderRequestTimeout<TError extends Error>(
@@ -14,28 +15,11 @@ export function withProviderRequestTimeout<TError extends Error>(
return Object.assign(error, { providerRequestTimeout: timeout })
}
/** The timed-out request an error carries. Follows `cause` and aggregated errors, since the
* acquisition errors wrap what the adapter threw. */
export function providerRequestTimeoutOf(
error: unknown,
depth = 0
): AgentSessionRequestTimeout | undefined {
if (depth >= 6 || !(error instanceof Error)) {
return undefined
}
if ('providerRequestTimeout' in error) {
const timeout = readAgentSessionRequestTimeout(error.providerRequestTimeout)
if (timeout) {
return timeout
}
}
if (error instanceof AggregateError) {
for (const inner of error.errors) {
const found = providerRequestTimeoutOf(inner, depth + 1)
if (found) {
return found
}
}
}
return providerRequestTimeoutOf(error.cause, depth + 1)
/** The timed-out request an error's chain carries. */
export function providerRequestTimeoutOf(error: unknown): AgentSessionRequestTimeout | undefined {
return findInErrorChain(error, (inner) =>
'providerRequestTimeout' in inner
? readAgentSessionRequestTimeout(inner.providerRequestTimeout)
: undefined
)
}
@@ -13,6 +13,7 @@ vi.mock('node:fs', () => ({
}))
import {
WINDOWS_RUNTIME_FILE_WATCH_LEADING_MS,
WINDOWS_RUNTIME_FILE_WATCH_REFRESH_MS,
watchWindowsRuntimeFileExplorer
} from './windows-file-watch'
@@ -37,4 +38,19 @@ describe('watchWindowsRuntimeFileExplorer', () => {
expect(callback).toHaveBeenCalledTimes(2_000 / WINDOWS_RUNTIME_FILE_WATCH_REFRESH_MS)
expect(callback).toHaveBeenCalledWith([{ kind: 'overflow', absolutePath: 'C:\\repo' }])
})
it('refreshes a one-off save after a quiet period on the short leading delay', () => {
const callback = vi.fn()
watchWindowsRuntimeFileExplorer('C:\\repo', callback, vi.fn())
fsWatch.listener?.()
vi.advanceTimersByTime(WINDOWS_RUNTIME_FILE_WATCH_LEADING_MS)
expect(callback).toHaveBeenCalledTimes(1)
// A change right after that refresh waits out the steady interval.
fsWatch.listener?.()
vi.advanceTimersByTime(WINDOWS_RUNTIME_FILE_WATCH_REFRESH_MS - 1)
expect(callback).toHaveBeenCalledTimes(1)
vi.advanceTimersByTime(1)
expect(callback).toHaveBeenCalledTimes(2)
})
})
@@ -2,9 +2,12 @@ import type { FsChangeEvent } from '../../../../shared/filesystem-entry-types'
import { PhysicalExitTracker } from '../../../../shared/physical-exit-tracker'
import { watch as watchFs } from 'node:fs'
import { WatcherProcessFailure } from '../../../ipc/parcel-watcher-process-failure'
import { watchBatchFlushDelayMs } from '../../../../shared/filesystem-watch-batch-window'
// Why: each emit is a whole-tree overflow refresh, so pace it slower than precise watcher batches.
export const WINDOWS_RUNTIME_FILE_WATCH_REFRESH_MS = 500
// Why: a one-off save after a quiet period still refreshes promptly.
export const WINDOWS_RUNTIME_FILE_WATCH_LEADING_MS = 150
export const WINDOWS_RUNTIME_FILE_WATCH_CLOSE_DEADLINE_MS = 10_000
export function watchWindowsRuntimeFileExplorer(
@@ -14,6 +17,7 @@ export function watchWindowsRuntimeFileExplorer(
): () => Promise<void> {
let disposed = false
let timer: ReturnType<typeof setTimeout> | null = null
let lastEmitAt = 0
let closeStarted = false
const physicalClose = new PhysicalExitTracker()
@@ -22,6 +26,7 @@ export function watchWindowsRuntimeFileExplorer(
if (disposed) {
return
}
lastEmitAt = Date.now()
callback([{ kind: 'overflow', absolutePath: rootPath }])
}
@@ -33,7 +38,15 @@ export function watchWindowsRuntimeFileExplorer(
if (timer) {
return
}
timer = setTimeout(emitOverflow, WINDOWS_RUNTIME_FILE_WATCH_REFRESH_MS)
timer = setTimeout(
emitOverflow,
watchBatchFlushDelayMs(
Date.now(),
lastEmitAt,
WINDOWS_RUNTIME_FILE_WATCH_LEADING_MS,
WINDOWS_RUNTIME_FILE_WATCH_REFRESH_MS
)
)
}
// Why: Parcel's Watchman probe can crash the headless server on Windows; use a conservative overflow refresh instead.
@@ -59,6 +59,7 @@ import {
import { RuntimeFileCommands } from './orca-runtime-files'
import {
WINDOWS_RUNTIME_FILE_WATCH_CLOSE_DEADLINE_MS,
WINDOWS_RUNTIME_FILE_WATCH_LEADING_MS,
WINDOWS_RUNTIME_FILE_WATCH_REFRESH_MS
} from './files/watch/windows-file-watch'
@@ -149,13 +150,21 @@ describe('RuntimeFileCommands file watching', () => {
emit?.()
emit?.()
await vi.advanceTimersByTimeAsync(WINDOWS_RUNTIME_FILE_WATCH_REFRESH_MS - 1)
await vi.advanceTimersByTimeAsync(WINDOWS_RUNTIME_FILE_WATCH_LEADING_MS - 1)
expect(onEvents).not.toHaveBeenCalled()
await vi.advanceTimersByTimeAsync(1)
expect(onEvents).toHaveBeenCalledTimes(1)
expect(onEvents).toHaveBeenCalledWith([{ kind: 'overflow', absolutePath: 'C:\\repo' }])
// A sustained stream refreshes on the slower cadence, measured from the last refresh.
emit?.()
await vi.advanceTimersByTimeAsync(WINDOWS_RUNTIME_FILE_WATCH_REFRESH_MS - 1)
expect(onEvents).toHaveBeenCalledTimes(1)
await vi.advanceTimersByTimeAsync(1)
expect(onEvents).toHaveBeenCalledTimes(2)
await unsubscribe()
expect(watcher.close).toHaveBeenCalledTimes(1)
})
@@ -7,11 +7,14 @@ const readRuntimeFileContentMock = vi.hoisted(() => vi.fn())
vi.mock('sonner', () => ({
toast: toastMock
}))
vi.mock('@/store', () => ({
useAppStore: {
getState: vi.fn()
vi.mock('@/store', () => {
const getState = vi.fn()
return {
useAppStore: Object.assign((select: (state: unknown) => unknown) => select(getState()), {
getState
})
}
}))
})
vi.mock('@/runtime/runtime-file-read-client', () => ({
readRuntimeFileContent: readRuntimeFileContentMock
}))
@@ -14,13 +14,14 @@ import { getDiskBaselineSignature } from './diff-content-signature'
import { trackExternalChangeConflictAction } from './editor-external-change-telemetry'
import { attemptEditorFileSave } from './editor-file-save-attempt'
import {
editorModelDraftHolder,
editorModelSiblingTabs,
isEditorBaselineRestampSuperseded,
isEditorModelDirty,
isWholeEditorState,
setEditorDiskBaseline,
trackEditorBaselineRestamp
} from './editor-save-disk-precondition'
import { setModelExternalMutation } from './editor-save-rebaseline'
import { notifyEditorFileCacheStale } from './editor-autosave'
// Why: when an external process (usually an agent) rewrites a file while the
@@ -36,14 +37,14 @@ export function reloadTabContentFromDisk(
reloadContent: (file: OpenFile) => void
): void {
const state = useAppStore.getState()
const discardedDraft = state.editorDrafts[file.id]
const discardedDiskSignature = state.openFiles.find(
(openFile) => openFile.id === file.id
)?.lastKnownDiskSignature
const liveTab = state.openFiles.find((openFile) => openFile.id === file.id)
// Why holder: a clean tab's Reload discards the unsaved text a sibling of the model holds.
const holder = liveTab ? editorModelDraftHolder(state, liveTab) : undefined
const discardedDraft = state.editorDrafts[holder?.id ?? file.id]
const discardedDiskSignature = liveTab?.lastKnownDiskSignature
// Why: drop the draft before reloading — the buffer shadows loaded content
// (editBuffers ?? fileContents), so a reload alone would keep showing the
// stale unsaved text.
const liveTab = state.openFiles.find((openFile) => openFile.id === file.id)
// Why every sibling: they share the model a reload replaces, so their drafts go with it.
const siblings = liveTab ? editorModelSiblingTabs(state, liveTab) : [file]
for (const tab of siblings) {
@@ -105,7 +106,8 @@ export function reloadTabContentFromDisk(
export function keepTabEditsOverExternalChange(file: OpenFile): void {
const state = useAppStore.getState()
state.setExternalMutation(file.id, null)
// Why model-wide: a sibling left marked would keep its banner and suspended autosave if the read fails.
setModelExternalMutation(state, file.id, null)
// Why: the dismissal must survive restart — without advancing the baseline
// to the current disk content, the restored-tab conflict scan re-derives
// the dismissed conflict from the stale signature on every launch.
@@ -158,9 +160,8 @@ export function keepTabEditsOverExternalChange(file: OpenFile): void {
export function overwriteDiskWithTabEdits(file: OpenFile): Promise<boolean> {
const state = useAppStore.getState()
const live = state.openFiles.find((openFile) => openFile.id === file.id)
const tabs = live && isWholeEditorState(state) ? editorModelSiblingTabs(state, live) : [file]
// Why: the unsaved text may live in another tab of the same model; with none there is nothing to write.
const holder = tabs.find((tab) => tab.isDirty && state.editorDrafts[tab.id] !== undefined)
const holder = live ? editorModelDraftHolder(state, live) : undefined
return holder
? attemptEditorFileSave({ fileId: holder.id, overwrite: true })
: Promise.resolve(false)
@@ -180,14 +181,15 @@ export function ExternalFileChangeBanner({
}): React.JSX.Element {
const [compareOpen, setCompareOpen] = useState(false)
// Why model-level: another tab of the same code model may hold the unsaved text.
const storeState = useAppStore.getState()
const liveFile = isWholeEditorState(storeState)
? storeState.openFiles.find((openFile) => openFile.id === file.id)
: undefined
const isModelDirty =
liveFile && isWholeEditorState(storeState)
? isEditorModelDirty(storeState, liveFile)
: file.isDirty
const isModelDirty = useAppStore((state) => {
const live = state.openFiles.find((openFile) => openFile.id === file.id)
return live ? isEditorModelDirty(state, live) : file.isDirty
})
const modelDraft = useAppStore((state) => {
const live = state.openFiles.find((openFile) => openFile.id === file.id)
const holder = live ? editorModelDraftHolder(state, live) : undefined
return holder ? state.editorDrafts[holder.id] : undefined
})
const handleReload = (): void => {
trackExternalChangeConflictAction(file, 'reload')
@@ -257,7 +259,7 @@ export function ExternalFileChangeBanner({
{compareOpen && (
<ExternalFileChangeCompareDialog
file={file}
currentContent={currentContent}
currentContent={modelDraft ?? currentContent}
open={compareOpen}
onOpenChange={setCompareOpen}
onReload={handleReload}
@@ -53,17 +53,31 @@ afterEach(cleanup)
it('re-reads every loaded and in-flight section of a mounted review when its workspace catches up', () => {
const requestSectionReload = renderReview()
dispatch({ worktreePath: '/repo', events: [{ kind: 'overflow', absolutePath: '/repo' }] })
dispatch({
worktreePath: '/repo',
events: [{ kind: 'overflow', absolutePath: '/repo' }],
resync: true
})
expect(requestSectionReload.mock.calls).toEqual([[0], [2], [3]])
})
it('ignores other workspaces, other hosts and ordinary path events', () => {
it('ignores other workspaces, other hosts, path events and storm overflows', () => {
const requestSectionReload = renderReview()
dispatch({ worktreePath: '/other', events: [{ kind: 'overflow', absolutePath: '/other' }] })
dispatch(
{ worktreePath: '/repo', events: [{ kind: 'overflow', absolutePath: '/repo' }] },
'env-1'
)
dispatch({ worktreePath: '/repo', events: [{ kind: 'update', absolutePath: '/repo/a.ts' }] })
const resync = (worktreePath: string): FsChangedPayload => ({
worktreePath,
events: [{ kind: 'overflow', absolutePath: worktreePath }],
resync: true
})
dispatch(resync('/other'))
dispatch(resync('/repo'), 'env-1')
dispatch({
worktreePath: '/repo',
events: [{ kind: 'update', absolutePath: '/repo/a.ts' }]
})
// Why: a storm overflow only refreshes git status; re-reading every section per flush would fan out.
dispatch({
worktreePath: '/repo',
events: [{ kind: 'overflow', absolutePath: '/repo' }]
})
expect(requestSectionReload).not.toHaveBeenCalled()
})
@@ -111,15 +111,16 @@ export function useCombinedDiffSectionRevalidation({
if (treeMode !== 'all' && treeMode !== 'uncommitted') {
return
}
// Why: an overflow (or a switch-back catch-up) names no paths, and snapshot-backed reviews skip
// Why: a resync (the watch was down) names no paths, and snapshot-backed reviews skip
// status-driven reloads, so re-read every loaded working-tree section of this workspace.
// A storm overflow only refreshes git status, so it never re-reads every section.
const handler = (event: WindowEventMap[typeof ORCA_WORKTREE_FILE_CHANGE_EVENT]): void => {
const { payload, runtimeEnvironmentId } = event.detail
if (
(runtimeEnvironmentId?.trim() || null) !== (file.runtimeEnvironmentId?.trim() || null) ||
normalizeRuntimePathForComparison(payload.worktreePath) !==
normalizeRuntimePathForComparison(file.filePath) ||
!payload.events.some((change) => change.kind === 'overflow')
payload.resync !== true
) {
return
}
@@ -6,6 +6,7 @@ import { getDiskBaselineSignature } from '../../../../../../shared/disk-baseline
import type { DiffSection } from '../../diff-section-types'
import type { OpenFile } from '@/store/slices/editor'
import { useCombinedDiffSectionActions } from './use-combined-diff-section-actions'
import { getStoredTextDiffResult } from '../../large-diff-section-content'
const mocks = vi.hoisted(() => ({ writeRuntimeFile: vi.fn() }))
@@ -96,19 +97,16 @@ it('guards section saves, marks a refusal as a section conflict, and overwrites
expect(mocks.writeRuntimeFile.mock.calls[2]?.[4]).toBeUndefined()
})
it('treats an empty loaded file as a known version', async () => {
const empty: DiffSection = {
...section(),
diffResult: {
kind: 'text',
originalContent: 'old',
modifiedContent: '',
originalIsBinary: false,
modifiedIsBinary: false
}
}
const sections = [empty]
const { result } = renderHook(() =>
const LIMITED: DiffSection['largeDiffRenderLimit'] = {
limited: true,
reason: 'character-count',
lineCounts: null,
characterCount: 7_000_000,
limits: { maxLinesPerSide: 120_000, maxCombinedCharacters: 6_000_000 }
}
function renderSave(sections: DiffSection[], setSections = vi.fn()) {
return renderHook(() =>
useCombinedDiffSectionActions({
activeGroupId: undefined,
branchCompare: null,
@@ -121,9 +119,25 @@ it('treats an empty loaded file as a known version', async () => {
sections,
sectionsRef: { current: sections },
setSectionHeights: vi.fn(),
setSections: vi.fn()
setSections
})
)
).result
}
// Why limited: a huge paste into an empty file trips the live render limit without pruning the load.
it('treats an empty loaded file as a known version, even after a huge paste', async () => {
const empty: DiffSection = {
...section(),
diffResult: {
kind: 'text',
originalContent: 'old',
modifiedContent: '',
originalIsBinary: false,
modifiedIsBinary: false
},
largeDiffRenderLimit: LIMITED
}
const result = renderSave([empty])
mocks.writeRuntimeFile.mockResolvedValueOnce({ ok: true })
await act(() => result.current.handleSectionSaveRef.current(0))
expect(mocks.writeRuntimeFile.mock.calls[0]?.[4]).toEqual({
@@ -132,6 +146,20 @@ it('treats an empty loaded file as a known version', async () => {
})
})
it('refuses to guess a version for a load whose bodies were pruned', async () => {
const loaded = section().diffResult
const pruned: DiffSection = {
...section(),
diffResult: loaded && getStoredTextDiffResult(loaded, LIMITED),
largeDiffRenderLimit: LIMITED
}
const setSections = vi.fn()
const result = renderSave([pruned], setSections)
await act(() => result.current.handleSectionSaveRef.current(0))
expect(mocks.writeRuntimeFile).not.toHaveBeenCalled()
expect(setSections.mock.calls[0]?.[0]([pruned])).toEqual([{ ...pruned, saveConflict: true }])
})
it('serializes overlapping saves of one section and never clobbers newer edits', async () => {
let resolveFirst: (value: { ok: true }) => void = () => undefined
let resolveSecond: (value: { ok: true }) => void = () => undefined
@@ -11,7 +11,11 @@ import type { OpenFile } from '@/store/slices/editor'
import type { GitBranchChangeEntry } from '../../../../../../shared/git-diff-compare-types'
import { canOpenDiffSectionPreviewToSide } from '../../diff-section-preview'
import { getLargeDiffRenderLimit } from '../../large-diff-render-limit'
import { getStoredTextDiffContent, getStoredTextDiffResult } from '../../large-diff-section-content'
import {
getStoredTextDiffContent,
getStoredTextDiffResult,
isPrunedTextDiffResult
} from '../../large-diff-section-content'
import { removeDiffSectionMeasuredHeight } from '../../diff-section-height-cache'
import type { DiffSection } from '../../diff-section-types'
import type { DiffSectionItemProps } from '../../diff-section-item-props'
@@ -173,8 +177,7 @@ export function useCombinedDiffSectionActions({
// Why first: a save earlier in this section's chain wrote it, maybe before state caught up.
const loaded =
writtenBySectionRef.current.get(section.key) ??
(section.diffResult?.kind === 'text' &&
!(section.diffResult.modifiedContent === '' && section.largeDiffRenderLimit?.limited)
(section.diffResult?.kind === 'text' && !isPrunedTextDiffResult(section.diffResult)
? section.diffResult.modifiedContent
: undefined)
const markConflict = (): void =>
@@ -75,6 +75,15 @@ describe('applyEditorFileLoadFailure', () => {
expect(setExternalMutation).not.toHaveBeenCalled()
})
it('matches a path containing an apostrophe', () => {
expect(
isPathSpecificMissingFileError(
new Error("ENOENT: no such file or directory, open '/repo/it's.md'"),
"/repo/it's.md"
)
).toBe(true)
})
it('matches the path regardless of separator spelling', () => {
expect(
isPathSpecificMissingFileError(
@@ -5,7 +5,8 @@ import { normalizeRuntimePathForComparison } from '../../../../shared/cross-plat
import type { FileContent } from './editor-panel-content-types'
// Node's fs message names the syscall and the path; IPC and runtime RPC both carry it verbatim.
const MISSING_FILE_MESSAGE = /ENOENT: no such file or directory, \w+ '([^']+)'/
// Greedy to the last quote: file names may contain apostrophes.
const MISSING_FILE_MESSAGE = /ENOENT: no such file or directory, \w+ '(.+)'/
/** True only when the failure names this exact file as missing — never a worktree or host lookup. */
export function isPathSpecificMissingFileError(error: unknown, filePath: string): boolean {
@@ -78,6 +78,13 @@ export function isEditorModelDirty(state: AppState, file: OpenFile): boolean {
return editorModelSiblingTabs(state, file).some((tab) => tab.isDirty)
}
/** The tab holding the model's unsaved text; it may be a sibling of `file`, not `file` itself. */
export function editorModelDraftHolder(state: AppState, file: OpenFile): OpenFile | undefined {
return editorModelSiblingTabs(state, file).find(
(tab) => tab.isDirty && state.editorDrafts[tab.id] !== undefined
)
}
/** Whether a store snapshot is whole enough to resolve a tab's model siblings. */
export function isWholeEditorState(state: unknown): state is AppState {
return (
@@ -138,6 +138,14 @@ export function createEditorSaveQueue(store: AppStoreApi): EditorSaveQueue {
: null
const fileContext = getEditorFileOperationContext(state, liveFile, worktree?.path ?? null)
const connectionId = fileContext.connectionId
const precondition = editorSavePrecondition(liveFile, trigger)
if (precondition.kind === 'blocked') {
// Why: a mobile refusal is reported to mobile; the desktop tab may not even be dirty.
if (trigger !== 'mobile') {
setModelExternalMutation(store.getState(), file.id, 'changed')
}
return 'blocked-stale'
}
// Why: stamp before writing so useEditorExternalWatch ignores our own fs:changed echo (editor-self-write-registry).
recordSelfWrite(
liveFile.filePath,
@@ -147,15 +155,6 @@ export function createEditorSaveQueue(store: AppStoreApi): EditorSaveQueue {
? SELF_WRITE_REMOTE_TTL_MS
: undefined
)
const precondition = editorSavePrecondition(liveFile, trigger)
if (precondition.kind === 'blocked') {
clearSelfWrite(liveFile.filePath, liveFile.runtimeEnvironmentId)
// Why: a mobile refusal is reported to mobile; the desktop tab may not even be dirty.
if (trigger !== 'mobile') {
setModelExternalMutation(store.getState(), file.id, 'changed')
}
return 'blocked-stale'
}
let writeResult: FileWriteResult
try {
writeResult = await writeEditorSave(
@@ -319,6 +319,39 @@ describe('two tabs sharing one model', () => {
}
})
it("Reload from a clean tab offers Undo of the sibling's discarded edits", () => {
const { store, a, b, edit } = openSharedTabs('A')
edit(b.id, 'mine')
for (const id of [a.id, b.id]) {
store.getState().setExternalMutation(id, 'changed')
}
const toastMock = vi.mocked(toast)
toastMock.mockClear()
reloadTabContentFromDisk(store.getState().openFiles[0], vi.fn())
expect(store.getState().editorDrafts[b.id]).toBeUndefined()
const undo: unknown = toastMock.mock.calls[0]?.[1]?.action
expect(isClickAction(undo)).toBe(true)
if (isClickAction(undo)) {
undo.onClick()
}
expect(store.getState().editorDrafts[b.id]).toBe('mine')
})
it('Keep clears the conflict on every tab of the model before the disk read settles', () => {
const { store, a, b, edit, tab } = openSharedTabs('A')
edit(a.id, 'mine')
for (const id of [a.id, b.id]) {
store.getState().setExternalMutation(id, 'changed')
}
mocks.readRuntimeFileContent.mockReturnValue(new Promise(() => {}))
keepTabEditsOverExternalChange(store.getState().openFiles[0])
expect(tab(b.id)?.externalMutation).toBeUndefined()
})
it('Undo of Reload yields to a sibling that typed since, touching neither tab', () => {
const { store, a, b, edit, tab } = openSharedTabs('A')
edit(a.id, 'mine')
@@ -22,11 +22,18 @@ const requestedReloads = new Map<string, string>()
/** Drops a stale cache so it reloads from disk, once per cached text and baseline: a reload that lands the same text can't loop. */
export function requestStaleCacheReload(fileId: string, content: string): void {
const file = useAppStore.getState()?.openFiles?.find((openFile) => openFile.id === fileId)
const openFiles = useAppStore.getState()?.openFiles ?? []
const file = openFiles.find((openFile) => openFile.id === fileId)
const stamp = `${file?.lastKnownDiskSignature ?? ''}\n${getDiskBaselineSignature(content)}`
if (requestedReloads.get(fileId) === stamp) {
return
}
// Why: drop closed tabs' stamps, which no settling mount will ever clear.
for (const id of requestedReloads.keys()) {
if (!openFiles.some((openFile) => openFile.id === id)) {
requestedReloads.delete(id)
}
}
requestedReloads.set(fileId, stamp)
notifyEditorFileCacheStale(fileId)
}
@@ -7,6 +7,13 @@ export function shouldPruneLargeDiffContent(
return renderLimit?.limited === true
}
const prunedTextDiffResults = new WeakSet<GitDiffResult>()
/** Whether this stored result dropped its bodies, so its '' is not the file's text. */
export function isPrunedTextDiffResult(result: GitDiffResult): boolean {
return prunedTextDiffResults.has(result)
}
export function getStoredTextDiffResult(
result: GitDiffResult,
renderLimit: LargeDiffRenderLimit | null | undefined
@@ -17,11 +24,9 @@ export function getStoredTextDiffResult(
// Why: after the fallback has enough metadata, retaining multi-MB bodies in
// section/view caches recreates the memory pressure the fallback avoids.
return {
...result,
originalContent: '',
modifiedContent: ''
}
const pruned: GitDiffResult = { ...result, originalContent: '', modifiedContent: '' }
prunedTextDiffResults.add(pruned)
return pruned
}
export function getStoredTextDiffContent(
@@ -104,4 +104,33 @@ describe('useMarkdownDocumentWatchRefresh', () => {
}
expect(refresh.mock.calls.length).toBeGreaterThanOrEqual(2)
})
it('lets a slow scan publish while membership changes never stop', async () => {
const scans: { startedAt: number; settledAt?: number }[] = []
const refresh = vi.fn(() => {
const scan: { startedAt: number; settledAt?: number } = {
startedAt: Date.now()
}
scans.push(scan)
// Why: a 40 s listing outlives the 30 s watchdog.
return new Promise<void>((resolve) =>
setTimeout(() => {
scan.settledAt = Date.now()
resolve()
}, 40_000)
)
})
await act(async () => root.render(createElement(Harness, { refresh })))
for (let i = 0; i < 60; i++) {
change([{ kind: 'create', absolutePath: `/repo/s${i}.md` }])
await advance(5_000)
}
// A scan's result publishes only if no newer scan started before it settled.
const published = scans.filter(
(scan, index) =>
scan.settledAt !== undefined &&
scans.slice(index + 1).every((later) => later.startedAt >= scan.settledAt!)
)
expect(published.length).toBeGreaterThanOrEqual(2)
})
})
@@ -74,6 +74,7 @@ export function useMarkdownDocumentWatchRefresh({
let scanSeq = 0
let activeScan: number | null = null
let scanWatchdog: ReturnType<typeof setTimeout> | undefined
let watchdogScale = 1
let changedDuringScan = false
let disposed = false
const endScan = (scan: number): void => {
@@ -97,11 +98,18 @@ export function useMarkdownDocumentWatchRefresh({
const scan = ++scanSeq
activeScan = scan
changedDuringScan = false
// Why: local and UNC listings have no timeout; stop waiting for a hung scan after the
// shared-request join window so later changes can scan again.
scanWatchdog = setTimeout(() => endScan(scan), MARKDOWN_DOCUMENT_LIST_JOIN_WINDOW_MS)
// Why: listings have no timeout; a doubling watchdog frees a hung scan yet lets a slow one publish.
scanWatchdog = setTimeout(() => {
watchdogScale = Math.min(watchdogScale * 2, 8)
endScan(scan)
}, MARKDOWN_DOCUMENT_LIST_JOIN_WINDOW_MS * watchdogScale)
// Each pane can join a scan begun after the latest change, while bypassing an older snapshot.
void refresh(false, latestChangeAt).finally(() => endScan(scan))
void refresh(false, latestChangeAt).finally(() => {
if (activeScan === scan) {
watchdogScale = 1
}
endScan(scan)
})
}, DOCUMENT_WATCH_DEBOUNCE_MS)
}
const handleChange = (event: WindowEventMap[typeof ORCA_WORKTREE_FILE_CHANGE_EVENT]): void => {
@@ -124,9 +124,7 @@ describe('linked-directory watcher invalidation', () => {
refreshTree,
refreshDir: async () => undefined,
isCoveredByFullRefresh: () => true,
dirConcurrency: 4,
trailingMs: 0,
maxWaitMs: 0
dirConcurrency: 4
})
f.refreshTree.mockImplementation(scheduler.requestFullRefresh)
for (let i = 0; i < 100; i++) {
@@ -2,9 +2,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { createFileExplorerWatchRefreshScheduler } from './file-explorer-watch-refresh-scheduler'
import type { FileExplorerTreeRefreshOutcome } from './file-explorer-types'
const TRAILING_MS = 150
const MAX_WAIT_MS = 500
function createDeferred(): {
promise: Promise<FileExplorerTreeRefreshOutcome>
resolve: (outcome?: FileExplorerTreeRefreshOutcome) => void
@@ -29,8 +26,6 @@ function setup(
refreshDir,
isCoveredByFullRefresh: () => false,
dirConcurrency: 4,
trailingMs: TRAILING_MS,
maxWaitMs: MAX_WAIT_MS,
...rest
})
return { refreshTree, refreshDir, scheduler }
@@ -51,22 +46,7 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
for (let index = 0; index < 20; index++) {
scheduler.requestFullRefresh()
}
await vi.advanceTimersByTimeAsync(TRAILING_MS)
expect(refreshTree).toHaveBeenCalledTimes(1)
})
it('flushes at the max wait even when requests keep arriving inside the trailing window', async () => {
const { refreshTree, scheduler } = setup()
scheduler.requestFullRefresh()
for (let index = 0; index < 4; index++) {
await vi.advanceTimersByTimeAsync(100)
scheduler.requestFullRefresh()
}
expect(refreshTree).not.toHaveBeenCalled()
await vi.advanceTimersByTimeAsync(MAX_WAIT_MS - 400)
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(1)
})
@@ -81,7 +61,7 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
scheduler.requestDirRefresh('C:/repo/src')
scheduler.requestDirRefresh('/repo/docs/')
scheduler.requestDirRefresh('/repo/docs')
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshDir).toHaveBeenCalledTimes(2)
expect(refreshDir.mock.calls.map(([dirPath]) => dirPath)).toEqual(['C:/repo/src', '/repo/docs'])
@@ -94,7 +74,7 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
scheduler.requestDirRefresh('/repo/collapsed')
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(1)
expect(refreshDir).toHaveBeenCalledTimes(1)
@@ -108,7 +88,7 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
scheduler.requestDirRefresh('/repo/src')
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(1)
expect(refreshDir).not.toHaveBeenCalled()
@@ -126,10 +106,10 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
scheduler.requestDirRefresh('/repo/src')
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
gate.resolve('superseded')
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshDir).toHaveBeenCalledTimes(1)
expect(refreshDir).toHaveBeenCalledWith('/repo/src')
@@ -147,10 +127,10 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
scheduler.requestDirRefresh('/repo/src')
scheduler.requestDirRefresh('/repo/docs')
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
gate.resolve('root-unreadable')
await vi.advanceTimersByTimeAsync(MAX_WAIT_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshDir).not.toHaveBeenCalled()
})
@@ -161,7 +141,7 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
})
scheduler.requestDirRefresh('/repo/src')
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).not.toHaveBeenCalled()
expect(refreshDir).toHaveBeenCalledTimes(1)
@@ -172,15 +152,15 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
const { refreshTree, scheduler } = setup({ refreshTree: () => gate.promise })
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(1)
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(MAX_WAIT_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(1)
gate.resolve()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(2)
})
@@ -190,14 +170,14 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
const { refreshDir, scheduler } = setup({ refreshTree: () => gate.promise })
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
scheduler.requestDirRefresh('/repo/late')
await vi.advanceTimersByTimeAsync(MAX_WAIT_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshDir).not.toHaveBeenCalled()
gate.resolve()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshDir).toHaveBeenCalledTimes(1)
expect(refreshDir).toHaveBeenCalledWith('/repo/late')
@@ -217,7 +197,7 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
for (let index = 0; index < 20; index++) {
scheduler.requestDirRefresh(`/repo/d${index}`)
}
await vi.advanceTimersByTimeAsync(TRAILING_MS + 20 * 10)
await vi.advanceTimersByTimeAsync(20 * 10)
expect(refreshDir).toHaveBeenCalledTimes(20)
expect(peakInFlight).toBeLessThanOrEqual(4)
@@ -229,7 +209,7 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
scheduler.requestFullRefresh()
scheduler.requestDirRefresh('/repo/src')
expect(scheduler.cancel()).toBe(true)
await vi.advanceTimersByTimeAsync(MAX_WAIT_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).not.toHaveBeenCalled()
expect(refreshDir).not.toHaveBeenCalled()
@@ -246,14 +226,14 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
const { refreshTree, scheduler } = setup({ refreshTree: () => gate.promise })
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(1)
// Nothing pending and no timer armed: the in-flight run is the only discarded work.
expect(scheduler.cancel()).toBe(true)
gate.resolve()
await vi.advanceTimersByTimeAsync(MAX_WAIT_MS)
await vi.advanceTimersByTimeAsync(0)
})
it('stops refreshing queued dirs when cancelled mid-run', async () => {
@@ -271,12 +251,12 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
for (let index = 0; index < 4; index++) {
scheduler.requestDirRefresh(`/repo/d${index}`)
}
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshDir).toHaveBeenCalledTimes(1)
scheduler.cancel()
gate.resolve()
await vi.advanceTimersByTimeAsync(MAX_WAIT_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshDir).toHaveBeenCalledTimes(1)
})
@@ -291,13 +271,13 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
scheduler.requestDirRefresh('/repo/late')
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
// refreshTree already ran against the old expanded set, so this dir was
// never re-read and its pending refresh must survive.
expanded.add('/repo/late')
gate.resolve()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshDir).toHaveBeenCalledTimes(1)
expect(refreshDir).toHaveBeenCalledWith('/repo/late')
@@ -308,33 +288,17 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
const { refreshTree, refreshDir, scheduler } = setup({ refreshTree: () => gate.promise })
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
scheduler.requestDirRefresh('/repo/src')
scheduler.cancel()
gate.resolve()
await vi.advanceTimersByTimeAsync(MAX_WAIT_MS)
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(1)
expect(refreshDir).not.toHaveBeenCalled()
})
it('uses the injected timer functions', async () => {
const schedule = vi.fn((callback: () => void, delayMs: number) => setTimeout(callback, delayMs))
const clear = vi.fn((timer: ReturnType<typeof setTimeout>) => {
clearTimeout(timer)
})
const { refreshTree, scheduler } = setup({ schedule, clear })
scheduler.requestFullRefresh()
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
expect(schedule).toHaveBeenCalledTimes(2)
expect(clear).toHaveBeenCalledTimes(1)
expect(refreshTree).toHaveBeenCalledTimes(1)
})
it('keeps a full refresh ahead of the dir refreshes it does not cover', async () => {
const order: string[] = []
const { scheduler } = setup({
@@ -349,57 +313,22 @@ describe('createFileExplorerWatchRefreshScheduler', () => {
scheduler.requestDirRefresh('/repo/collapsed')
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(TRAILING_MS)
await vi.advanceTimersByTimeAsync(0)
expect(order).toEqual(['tree', '/repo/collapsed'])
})
// Why: local workspaces pass a zero window because the main process already coalesced the burst
// on the same 150/500 timings. Stacking a second debounce there only doubled paint latency.
describe('zero window (local transport)', () => {
const zeroWindow = { trailingMs: 0, maxWaitMs: 0 }
it('flushes on the next tick instead of waiting out a trailing window', async () => {
const { refreshTree, scheduler } = setup(zeroWindow)
it('flushes on the next tick and coalesces one payload burst into a single refresh', async () => {
const { refreshTree, refreshDir, scheduler } = setup()
for (let index = 0; index < 20; index++) {
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(0)
scheduler.requestDirRefresh('/repo/src')
}
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(1)
})
it('still coalesces one payload burst into a single refresh', async () => {
const { refreshTree, refreshDir, scheduler } = setup(zeroWindow)
for (let index = 0; index < 20; index++) {
scheduler.requestFullRefresh()
scheduler.requestDirRefresh('/repo/src')
}
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(1)
expect(refreshDir).toHaveBeenCalledTimes(1)
expect(refreshDir).toHaveBeenCalledWith('/repo/src')
})
it('does not start a second run while one is in flight', async () => {
const deferred = createDeferred()
const { refreshTree, scheduler } = setup({
...zeroWindow,
refreshTree: () => deferred.promise
})
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(0)
scheduler.requestFullRefresh()
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(1)
deferred.resolve()
await vi.advanceTimersByTimeAsync(0)
expect(refreshTree).toHaveBeenCalledTimes(2)
})
expect(refreshTree).toHaveBeenCalledTimes(1)
expect(refreshDir).toHaveBeenCalledTimes(1)
expect(refreshDir).toHaveBeenCalledWith('/repo/src')
})
})
@@ -2,8 +2,6 @@ import type { FileExplorerTreeRefreshOutcome } from './file-explorer-types'
import { normalizeRuntimePathForComparison } from '../../../../shared/cross-platform-path'
import { forEachWithConcurrency } from '../../../../shared/map-with-concurrency'
type SchedulerTimer = ReturnType<typeof setTimeout>
export type FileExplorerWatchRefreshScheduler = {
requestFullRefresh: () => void
requestDirRefresh: (dirPath: string) => void
@@ -17,59 +15,43 @@ export type CreateFileExplorerWatchRefreshSchedulerParams = {
/** Dirs a full tree refresh already re-reads; the scheduler has no view of `expanded`. */
isCoveredByFullRefresh: (dirPath: string) => boolean
dirConcurrency: number
trailingMs: number
maxWaitMs: number
schedule?: (callback: () => void, delayMs: number) => SchedulerTimer
clear?: (timer: SchedulerTimer) => void
}
/**
* Coalesces watcher-driven File Explorer refreshes into one debounced run.
* Coalesces watcher-driven File Explorer refreshes into one run on the next tick, one run at a time.
*
* Why: an overflow storm otherwise issues K × (1 + N_expanded) readDirectory
* round trips on the same SSH mux that carries PTY output. Transport-agnostic on
* purpose — it sits above both the Electron fs:changed bus and the runtime-RPC
* file-change subscription.
* Why no debounce window: every producer already flushes on the shared watch batch window
* (filesystem-watcher for local/SSH, createFileWatchEventBatcher for runtime files.watch), so a
* second window here only adds latency. A new transport must batch at the producer before joining.
*/
export function createFileExplorerWatchRefreshScheduler({
refreshTree,
refreshDir,
isCoveredByFullRefresh,
dirConcurrency,
trailingMs,
maxWaitMs,
schedule = setTimeout,
clear = clearTimeout
dirConcurrency
}: CreateFileExplorerWatchRefreshSchedulerParams): FileExplorerWatchRefreshScheduler {
let pendingFull = false
const pendingDirs = new Map<string, string>()
let firstRequestAt: number | null = null
let timer: SchedulerTimer | null = null
let timer: ReturnType<typeof setTimeout> | null = null
let inFlight: Promise<void> | null = null
let disposed = false
function clearTimer(): void {
function arm(): void {
if (timer !== null) {
clear(timer)
timer = null
return
}
}
function arm(delayMs: number): void {
clearTimer()
timer = schedule(() => {
timer = setTimeout(() => {
timer = null
// Why: a run started here would race the in-flight one; the settle
// handler re-arms instead, so nothing pending is dropped.
if (inFlight === null) {
run()
}
}, delayMs)
}
function armForRequest(): void {
firstRequestAt ??= Date.now()
arm(Math.max(0, Math.min(trailingMs, firstRequestAt + maxWaitMs - Date.now())))
}, 0)
}
function run(): void {
@@ -77,7 +59,6 @@ export function createFileExplorerWatchRefreshScheduler({
const dirs = Array.from(pendingDirs.values())
pendingFull = false
pendingDirs.clear()
firstRequestAt = null
const started = (async () => {
// Why: isCoveredByFullRefresh reads a live `expanded` ref, so resolve it
@@ -114,7 +95,7 @@ export function createFileExplorerWatchRefreshScheduler({
inFlight = null
}
if (!disposed && (pendingFull || pendingDirs.size > 0)) {
arm(trailingMs)
arm()
}
})
}
@@ -129,23 +110,25 @@ export function createFileExplorerWatchRefreshScheduler({
// superset of what refreshTree re-reads; subsuming them would strand a
// stale listing for a collapsed-but-cached dir.
pendingFull = true
armForRequest()
arm()
},
requestDirRefresh: (dirPath: string) => {
if (disposed) {
return
}
pendingDirs.set(normalizeRuntimePathForComparison(dirPath), dirPath)
armForRequest()
arm()
},
cancel: () => {
const discardedWork =
pendingFull || pendingDirs.size > 0 || timer !== null || inFlight !== null
disposed = true
clearTimer()
if (timer !== null) {
clearTimeout(timer)
timer = null
}
pendingFull = false
pendingDirs.clear()
firstRequestAt = null
return discardedWork
}
}
@@ -164,13 +164,7 @@ export function useFileExplorerWatch({
normalizeRuntimePathForComparison(dirPath) ===
normalizeRuntimePathForComparison(currentWorktreePath) ||
expandedRef.current.has(dirPath),
dirConcurrency: fileExplorerRefreshConcurrency(owner),
// Why 0ms for every transport: each producer already flushes on the shared
// watch batch window — local/SSH via filesystem-watcher, runtime files.watch via
// createFileWatchEventBatcher — so a second window here only adds latency, not coalescing.
// A new transport must batch at the producer before joining this path.
trailingMs: 0,
maxWaitMs: 0
dirConcurrency: fileExplorerRefreshConcurrency(owner)
})
activeResyncByWatchKey.set(currentWatchKey, scheduler.requestFullRefresh)
@@ -177,7 +177,7 @@ export function buildEditorExternalWatchEventHandler(
}
if (payload.events.some((event) => event.kind === 'overflow')) {
reconcileEditorTabsAfterOverflow(target)
reconcileEditorTabsAfterOverflow(target, payload.resync === true)
return
}
if (batchPaths.changes.length === 0) {
@@ -69,7 +69,7 @@ describe('reconcileEditorTabsAfterOverflow', () => {
tab('staged.ts', { mode: 'diff', diffSource: 'staged' })
)
harness.runtimePathsExist.mockResolvedValue([{ exists: true }])
reconcileEditorTabsAfterOverflow(target)
reconcileEditorTabsAfterOverflow(target, true)
await vi.waitFor(() => expect(harness.scheduleReload).toHaveBeenCalledTimes(2))
expect(harness.scheduleReload).toHaveBeenCalledWith(notification('staged.ts'))
@@ -85,7 +85,7 @@ describe('reconcileEditorTabsAfterOverflow', () => {
{ exists: false },
{ exists: true }
])
reconcileEditorTabsAfterOverflow(target)
reconcileEditorTabsAfterOverflow(target, true)
await vi.waitFor(() => expect(harness.scheduleReload).toHaveBeenCalledTimes(1))
expect(harness.scheduleReload).toHaveBeenCalledWith(notification('here.ts'))
@@ -103,7 +103,7 @@ describe('reconcileEditorTabsAfterOverflow', () => {
])('reloads as before when %s', async (_label, arrange) => {
arrange()
setTabs(tab('a.ts'))
reconcileEditorTabsAfterOverflow(target)
reconcileEditorTabsAfterOverflow(target, true)
await vi.waitFor(() =>
expect(harness.scheduleReload).toHaveBeenCalledWith(notification('a.ts'))
)
@@ -112,17 +112,46 @@ describe('reconcileEditorTabsAfterOverflow', () => {
it('hands dirty tabs to the disk check instead of reloading them', () => {
setTabs(tab('dirty.ts', { isDirty: true, lastKnownDiskSignature: 'sig' }))
reconcileEditorTabsAfterOverflow(target)
reconcileEditorTabsAfterOverflow(target, true)
expect(setPendingDiskBaselineVerification).toHaveBeenCalledWith('dirty.ts', true)
expect(harness.runtimePathsExist).not.toHaveBeenCalled()
expect(harness.scheduleReload).not.toHaveBeenCalled()
})
it('a storm overflow reloads clean tabs without probing, pausing autosave or dropping combined bodies', () => {
setTabs(
tab('a.md', { externalMutation: 'deleted' }),
tab('dirty.ts', { isDirty: true, lastKnownDiskSignature: 'sig' })
)
reconcileEditorTabsAfterOverflow(target, false)
expect(harness.scheduleReload.mock.calls).toEqual([[notification('a.md')]])
expect(setExternalMutation).toHaveBeenCalledWith('a.md', null)
expect(harness.runtimePathsExist).not.toHaveBeenCalled()
expect(setPendingDiskBaselineVerification).not.toHaveBeenCalled()
expect(harness.forgetCombinedBodies).not.toHaveBeenCalled()
})
it('a storm overflow supersedes a pending resync probe', async () => {
setTabs(tab('a.ts'))
let answer: (rows: unknown) => void = () => {}
harness.runtimePathsExist.mockImplementationOnce(
() => new Promise((resolve) => (answer = resolve))
)
reconcileEditorTabsAfterOverflow(target, true)
reconcileEditorTabsAfterOverflow(target, false)
expect(getPendingOverflowProbeCountForTest()).toBe(0)
answer([{ exists: false }])
await new Promise((resolve) => setTimeout(resolve, 0))
expect(setExternalMutation).not.toHaveBeenCalledWith('a.ts', 'deleted')
})
it('only touches tabs of the matching runtime owner', async () => {
setTabs(tab('local.md'), tab('runtime.md', { runtimeEnvironmentId: 'env-1' }))
harness.runtimePathsExist.mockResolvedValue([{ exists: true }])
reconcileEditorTabsAfterOverflow({ ...target, runtimeEnvironmentId: 'env-1' })
reconcileEditorTabsAfterOverflow({ ...target, runtimeEnvironmentId: 'env-1' }, true)
await vi.waitFor(() => expect(harness.scheduleReload).toHaveBeenCalledTimes(1))
expect(harness.scheduleReload).toHaveBeenCalledWith(notification('runtime.md', 'env-1'))
@@ -142,7 +171,7 @@ describe('reconcileEditorTabsAfterOverflow', () => {
// The last tab (the only one in the second chunk) is gone.
paths.map((path) => ({ exists: path !== '/repo/f128.ts' }))
)
reconcileEditorTabsAfterOverflow(target)
reconcileEditorTabsAfterOverflow(target, true)
await vi.waitFor(() => expect(harness.scheduleReload).toHaveBeenCalledTimes(128))
expect(harness.runtimePathsExist.mock.calls.map(([, paths]) => paths.length)).toEqual([128, 1])
@@ -156,8 +185,8 @@ describe('reconcileEditorTabsAfterOverflow', () => {
harness.runtimePathsExist
.mockImplementationOnce(() => new Promise((resolve) => (answerFirst = resolve)))
.mockResolvedValueOnce([{ exists: false }])
reconcileEditorTabsAfterOverflow(target)
reconcileEditorTabsAfterOverflow(target)
reconcileEditorTabsAfterOverflow(target, true)
reconcileEditorTabsAfterOverflow(target, true)
await vi.waitFor(() => expect(setExternalMutation).toHaveBeenCalledWith('a.ts', 'deleted'))
answerFirst([{ exists: true }])
@@ -181,7 +210,7 @@ describe('reconcileEditorTabsAfterOverflow', () => {
harness.runtimePathsExist.mockImplementationOnce(
() => new Promise((resolve) => (answer = resolve))
)
reconcileEditorTabsAfterOverflow(target)
reconcileEditorTabsAfterOverflow(target, true)
supersedeOverflowProbesForBatch(
{
createOrUpdatePaths: new Map(),
@@ -207,7 +236,7 @@ describe('reconcileEditorTabsAfterOverflow', () => {
harness.runtimePathsExist.mockImplementationOnce(
() => new Promise((resolve) => (answer = resolve))
)
reconcileEditorTabsAfterOverflow(target)
reconcileEditorTabsAfterOverflow(target, true)
expect(getPendingOverflowProbeCountForTest()).toBe(1)
setTabs()
@@ -79,12 +79,15 @@ type EditTabProbe = {
}
/**
* An overflow (or a switch-back catch-up) names no paths, so every tab of the workspace may be stale.
* Clean tabs reload unless disk proves the file gone; dirty tabs get the restored-tab disk check.
* An overflow names no paths, so every clean tab of the workspace reloads.
* A resync (the watch was down) also probes clean edit tabs for deletion and gives dirty tabs the restored-tab
* disk check; an event-storm overflow skips both so it never pauses autosave or fans out per-tab probes.
*/
export function reconcileEditorTabsAfterOverflow(target: OverflowTarget): void {
// Why: unmounted combined-diff tabs get no per-path reload, so drop their cached bodies before a remount replays them.
forgetCombinedDiffViewBodiesForWorktree(target.worktreeId)
export function reconcileEditorTabsAfterOverflow(target: OverflowTarget, resync: boolean): void {
if (resync) {
// Why: unmounted combined-diff tabs get no per-path reload, so drop their cached bodies before a remount replays them.
forgetCombinedDiffViewBodiesForWorktree(target.worktreeId)
}
const state = useAppStore.getState()
const probes: EditTabProbe[] = []
for (const file of state.openFiles) {
@@ -96,6 +99,9 @@ export function reconcileEditorTabsAfterOverflow(target: OverflowTarget): void {
continue
}
if (file.isDirty) {
if (!resync) {
continue
}
// Why: a dirty tab is never reloaded; the restored-tab scan shows the changed-on-disk banner if disk moved.
// Mirrors the scan's eligibility so it always resolves the flag.
if (
@@ -115,7 +121,9 @@ export function reconcileEditorTabsAfterOverflow(target: OverflowTarget): void {
runtimeEnvironmentId: target.runtimeEnvironmentId ?? null,
...getLocalWindowsWslAliasOption(target)
}
if (file.mode !== 'edit') {
if (file.mode !== 'edit' || !resync) {
// Why: this reload is newer than any pending resync probe for the tab.
probeGenerationByFileId.delete(file.id)
if (file.externalMutation) {
state.setExternalMutation(file.id, null)
}
@@ -111,7 +111,8 @@ describe('editor runtime file watch recovery', () => {
)
expect(onPayload).toHaveBeenCalledWith({
worktreePath: '/notes',
events: [{ kind: 'overflow', absolutePath: '/notes' }]
events: [{ kind: 'overflow', absolutePath: '/notes' }],
resync: true
})
stop()
})
@@ -134,7 +135,8 @@ describe('editor runtime file watch recovery', () => {
expect(onPayload).toHaveBeenCalledOnce()
expect(onPayload).toHaveBeenCalledWith({
worktreePath: '/repo',
events: [{ kind: 'overflow', absolutePath: '/repo' }]
events: [{ kind: 'overflow', absolutePath: '/repo' }],
resync: true
})
stop()
expect(stopFresh).toHaveBeenCalledOnce()
@@ -58,7 +58,8 @@ export function subscribeEditorRuntimeFileWatch(
if (resync) {
onPayload({
worktreePath: target.worktreePath,
events: [{ kind: 'overflow', absolutePath: target.worktreePath }]
events: [{ kind: 'overflow', absolutePath: target.worktreePath }],
resync: true
})
}
})
@@ -70,6 +70,7 @@ function resyncOf(target: TestWatchTarget): [unknown, string | null, TestWatchTa
{
worktreePath: target.worktreePath,
events: [{ kind: 'overflow', absolutePath: target.worktreePath }],
resync: true,
connectionId: target.connectionId
},
target.runtimeEnvironmentId,
@@ -215,7 +215,8 @@ describe('useEditorExternalWatch subscriptions', () => {
expect(subscriptionState.handleFsChanged).toHaveBeenCalledWith(
{
worktreePath: '/runtime/repo',
events: [{ kind: 'overflow', absolutePath: '/runtime/repo' }]
events: [{ kind: 'overflow', absolutePath: '/runtime/repo' }],
resync: true
},
'runtime-1'
)
@@ -220,7 +220,8 @@ describe('switching back to a workspace that was not watched', () => {
await act(async () =>
emitIpc({
worktreePath: '/background',
events: [{ kind: 'overflow', absolutePath: '/background' }]
events: [{ kind: 'overflow', absolutePath: '/background' }],
resync: true
})
)
await act(async () => vi.advanceTimersByTimeAsync(200))
@@ -69,7 +69,8 @@ export function useEditorExternalWatch(): void {
markWorktreeCatchUpPayload({
worktreePath: target.worktreePath,
events: [{ kind: 'overflow', absolutePath: target.worktreePath }],
connectionId: target.connectionId
connectionId: target.connectionId,
resync: true
}),
target.runtimeEnvironmentId,
target
@@ -312,6 +312,63 @@ describe('mobile markdown bridge', () => {
}
})
it.each([
['saves', 'D1'],
['types', null]
] as const)(
'refuses a phone save when the desktop %s during the phone-verifying read',
async (_label, desktopSave) => {
openMarkdownFile()
const tabId = useAppStore.getState().openFiles[0].id
useAppStore.getState().setLastKnownDiskSignature(tabId, getDiskBaselineSignature('D0'))
let diskContent = 'D0'
const readFile = vi.fn().mockImplementation(async () => {
const content = diskContent
if (desktopSave) {
diskContent = desktopSave
useAppStore
.getState()
.setLastKnownDiskSignature(tabId, getDiskBaselineSignature(desktopSave))
} else {
useAppStore.getState().setEditorDraft(tabId, 'desktop typing')
useAppStore.getState().markFileDirty(tabId, true)
}
return { content, isBinary: false }
})
const writeFile = vi.fn().mockImplementation(async ({ content, diskPrecondition }) => {
const refusal = guardedWriteRefusal(diskPrecondition, diskContent)
if (refusal) {
return refusal
}
diskContent = content
return { ok: true }
})
setupWindow({ readFile, writeFile })
const detachBridge = attachMobileMarkdownBridge()
const detachAutosave = attachEditorAutosaveController(useAppStore as never)
try {
const response = await sendRequest({
id: 'save-race',
operation: 'save',
worktreeId: 'wt-1',
tabId: 'tab-md',
baseVersion: hashMarkdownContent('D0'),
content: 'phone edit'
})
expect(response).toMatchObject({ id: 'save-race', ok: false })
expect(diskContent).toBe(desktopSave ?? 'D0')
if (!desktopSave) {
expect(useAppStore.getState().editorDrafts[tabId]).toBe('desktop typing')
}
} finally {
detachAutosave()
detachBridge()
}
}
)
it('lets the phone save after an external change to a clean desktop tab', async () => {
openMarkdownFile()
const tabId = useAppStore.getState().openFiles[0].id
@@ -121,11 +121,20 @@ async function saveMobileMarkdownTab(
}
const state = useAppStore.getState()
const stampedBaseline = stampMobileVerifiedBaseline(freshTarget.sourceFile, current)
const previousDraft = state.editorDrafts[freshTarget.sourceFile.id]
const previousDirty = freshTarget.sourceFile.isDirty
state.setEditorDraft(freshTarget.sourceFile.id, content)
state.markFileDirty(freshTarget.sourceFile.id, true)
// Why live: a desktop save during the read moved the baseline past the version the phone saw.
const sourceFile = state.openFiles.find((file) => file.id === freshTarget.sourceFile.id)
if (!sourceFile) {
throw new Error('tab_not_found')
}
// Why: desktop typing during the disk read is newer than the version the phone edited.
if (current.source === 'file' && state.editorDrafts[sourceFile.id] !== undefined) {
throw new Error('conflict')
}
const stampedBaseline = stampMobileVerifiedBaseline(sourceFile, current)
const previousDraft = state.editorDrafts[sourceFile.id]
const previousDirty = sourceFile.isDirty
state.setEditorDraft(sourceFile.id, content)
state.markFileDirty(sourceFile.id, true)
let verified: string
try {
await waitForPositiveSave(freshTarget.sourceFile, content)
@@ -28,7 +28,6 @@ export function buildWebSessionTabsFinalPatch(
removedTerminalResourceIds,
mirroredTerminalIds,
nextOpenFiles,
nextEditorBuffers,
nextBrowserTabsByWorktree,
nextUnifiedTabsByWorktree,
nextGroupsByWorktree,
@@ -96,9 +95,8 @@ export function buildWebSessionTabsFinalPatch(
...agentStatusPatch,
...retractedTabSweepPatch,
...remirroredClosedTabLiftPatch,
...(nextOpenFiles !== state.openFiles
? { openFiles: nextOpenFiles, editorBuffers: nextEditorBuffers }
: {}),
// Why no editorBuffers: the store's buffer middleware adopts the host's mirrored tab state.
...(nextOpenFiles !== state.openFiles ? { openFiles: nextOpenFiles } : {}),
...(nextTabsByWorktree !== state.tabsByWorktree ? { tabsByWorktree: nextTabsByWorktree } : {}),
...(nextBrowserTabsByWorktree !== state.browserTabsByWorktree
? { browserTabsByWorktree: nextBrowserTabsByWorktree }
@@ -13,7 +13,6 @@ import {
} from '../host-session-snapshot-authority'
import type { prepareWebSessionTabsSnapshotBase } from './apply-preparation-base'
import type { OpenFile } from '../../store/slices/editor'
import { createEditorBuffers } from '../../store/slices/editor/buffers/editor-buffer-record'
import {
advanceWebSessionOpenFilesIndex,
sameOpenFiles,
@@ -178,11 +177,6 @@ export function prepareWebSessionTabsSnapshotBrowser(
return sameOpenFiles(state.openFiles, next) ? state.openFiles : next
})()
advanceWebSessionOpenFilesIndex(batchContext, nextOpenFiles, worktreeId)
// Why: a mirrored tab's buffer state is the host's, so its record is (re)created with the tab.
const nextEditorBuffers =
nextOpenFiles === state.openFiles
? state.editorBuffers
: createEditorBuffers(state.editorBuffers ?? {}, mirroredOpenFiles)
const retainedUnifiedTabs = currentUnifiedTabs.filter((tab) => {
if (tab.contentType === 'agent-session') {
// A matching host row is authoritative; retaining the provisional tab beside its mirror
@@ -243,7 +237,6 @@ export function prepareWebSessionTabsSnapshotBrowser(
removedEditorFileIds,
nextWorktreeOpenFileIds,
nextOpenFiles,
nextEditorBuffers,
retainedUnifiedTabs,
existingViewModeByTabId
}
@@ -61,7 +61,6 @@ const LAYOUT_SLICES = [
'terminalLayoutsByTabId',
'ptyIdsByTabId',
'openFiles',
'editorBuffers',
'browserTabsByWorktree',
'browserPagesByWorkspace',
'remoteBrowserPageHandlesByPageId',
@@ -257,7 +257,6 @@ export type WebSessionTabsSyncState = Pick<
// Why: a client draft is the evidence that a mirrored file's dirty flag is the client's
// own and must survive a host republish (#21392); absent here, the host flag wins.
| 'editorDrafts'
| 'editorBuffers'
| 'localOnlyScrollbackByTabId'
| 'migrationUnsupportedByPtyId'
| 'manuallyUnreadTurnsByPaneKey'
@@ -206,6 +206,20 @@ describe('editor buffer records', () => {
})
})
it("adopts a host republish of an open tab's buffer state, as remote session sync writes it", () => {
const id = openEditTab()
useAppStore.setState((state) => ({
openFiles: state.openFiles.map((file) =>
file.id === id ? { ...file, isDirty: true, lastKnownDiskSignature: 'host' } : file
)
}))
expect(useAppStore.getState().editorBuffers[id]).toMatchObject({
isDirty: true,
lastKnownDiskSignature: 'host'
})
})
it('rejects a slice write that changes a mirror field without its record', () => {
type State = {
openFiles: OpenFile[]
+4 -3
View File
@@ -77,10 +77,11 @@ const withDevelopmentStoreProbes = (createState: StateCreator<AppState, [], []>)
? withStoreIdentityChurnProbe(createState)
: createState
// Why buffer records outermost: the probes must see each caller's own partial and frame.
export const useAppStore = create<AppState>()(
withDevelopmentStoreProbes(
withReactCommitCascadeWriteProbe(
withEditorBufferRecords((...a) => {
withEditorBufferRecords(
withDevelopmentStoreProbes(
withReactCommitCascadeWriteProbe((...a) => {
// Why first: settling inside the bump's own notify means hook subscribers never see it unsettled.
// Switch point: the stamp-sorted list reads no settled epoch.
if (!ORCA_SIDEBAR_INPUTS_IS_MAIN) {
@@ -1,6 +1,9 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import { create, type StoreApi, type UseBoundStore } from 'zustand'
import { withReactCommitCascadeWriteProbe } from './react-commit-cascade-write-probe'
import { withEditorBufferRecords } from './editor-buffer-record-lifecycle'
import type { EditorBufferRecords } from './slices/editor/buffers/editor-buffer-record'
import type { OpenFile } from './slices/editor/types/open-file'
import type { ReactCommitCascadeWriteBoundary } from '@/lib/react-commit-cascade-store-write-samples'
const { probe, noteWrite } = vi.hoisted(() => ({
@@ -71,4 +74,33 @@ describe('withReactCommitCascadeWriteProbe', () => {
expect(noteWrite).not.toHaveBeenCalled()
expect(useProbedStore.getState().ticks).toBe(4)
})
// Why: composed as in store/index.ts, the buffer middleware must not turn every write into one
// updater from one frame, or the crash report loses changed keys and distinct write sites.
it('sees each original partial and caller under the editor buffer middleware', () => {
type BufferedState = {
ticks: number
openFiles: OpenFile[]
editorBuffers: EditorBufferRecords
tick: () => void
}
const useBufferedStore = create<BufferedState>()(
withEditorBufferRecords(
withReactCommitCascadeWriteProbe<BufferedState>((set) => ({
ticks: 0,
openFiles: [],
editorBuffers: {},
tick: () => set({ ticks: 1 })
}))
)
)
probe.armed = true
useBufferedStore.getState().tick()
useBufferedStore.setState({ ticks: 2 })
expect(noteWrite.mock.calls.map(([, partial]) => partial)).toEqual([{ ticks: 1 }, { ticks: 2 }])
expect(noteWrite.mock.calls[1]?.[0]).toBe(useBufferedStore.setState)
expect(useBufferedStore.getState().ticks).toBe(2)
})
})
@@ -16,22 +16,38 @@ import {
} from '@/lib/react-commit-cascade-store-write-samples'
import type { StateCreator } from 'zustand'
/** zustand's two `setState` call shapes, as a union so a wrapper can forward them uncast. */
export type StoreSetArgs<TState> =
| [
partial: TState | Partial<TState> | ((state: TState) => TState | Partial<TState>),
replace?: false
]
| [state: TState | ((state: TState) => TState), replace: true]
export function withReactCommitCascadeWriteProbe<TState>(
createState: StateCreator<TState, [], []>
): StateCreator<TState, [], []> {
return (set, get, api) => {
const wrapped = ((partial: unknown, replace?: unknown): void => {
if (reactCommitCascadeWriteProbe.armed) {
try {
noteReactCommitCascadeStoreWrite(wrapped, partial)
} catch {
// This is the app's universal write path: a diagnostic that throws here
// would drop the write itself.
const probe = (write: typeof set): typeof set => {
const wrapped = (...args: StoreSetArgs<TState>): void => {
if (reactCommitCascadeWriteProbe.armed) {
try {
noteReactCommitCascadeStoreWrite(wrapped, args[0])
} catch {
// This is the app's universal write path: a diagnostic that throws here
// would drop the write itself.
}
}
if (args[1] === true) {
write(...args)
} else {
write(...args)
}
}
;(set as (nextPartial: unknown, nextReplace?: unknown) => void)(partial, replace)
}) as typeof set
api.setState = wrapped as typeof api.setState
return createState(wrapped, get, api)
return wrapped
}
// Why separately: an outer middleware may hand slices a different `set` than api.setState.
api.setState = probe(api.setState)
return createState(probe(set), get, api)
}
}
@@ -78,21 +78,6 @@ export function writeEditorBuffer(
return { openFiles, editorBuffers }
}
/** (Re)creates records for tabs whose buffer state comes from outside the store, e.g. a host snapshot. */
export function createEditorBuffers(
records: EditorBufferRecords,
files: readonly OpenFile[]
): EditorBufferRecords {
if (files.length === 0) {
return records
}
const next = { ...records }
for (const file of files) {
next[file.id] = editorBufferRecordFromFile(file)
}
return next
}
/** Moves each migrated tab's record to its new id in the same update that re-keys the tab. */
export function rekeyEditorBuffers(
records: EditorBufferRecords,
@@ -18,6 +18,7 @@
* double-invoke it or hand it a stale state.
*/
import type { StateCreator } from 'zustand'
import type { StoreSetArgs } from './react-commit-cascade-write-probe'
export const storeIdentityChurnProbe = { armed: false, captureSites: false }
@@ -177,30 +178,41 @@ export function withStoreIdentityChurnProbe<TState>(
createState: StateCreator<TState, [], []>
): StateCreator<TState, [], []> {
return (set, get, api) => {
const wrapped = ((partial: unknown, replace?: unknown): void => {
if (!storeIdentityChurnProbe.armed) {
;(set as (nextPartial: unknown, nextReplace?: unknown) => void)(partial, replace)
return
const probe = (write: typeof set): typeof set => {
const forward = (args: StoreSetArgs<TState>): void => {
if (args[1] === true) {
write(...args)
} else {
write(...args)
}
}
const previous = get() as Record<string, unknown>
// Why the write is passed through untouched: zustand owns when and how an
// updater runs. The probe only compares the states on either side of it.
;(set as (nextPartial: unknown, nextReplace?: unknown) => void)(partial, replace)
try {
const next = get() as Record<string, unknown>
if (next === previous) {
return (...args: StoreSetArgs<TState>): void => {
if (!storeIdentityChurnProbe.armed) {
forward(args)
return
}
const fields =
replace !== true && partial !== null && typeof partial === 'object'
? Object.keys(partial)
: Object.keys(next)
recordWrite(previous, next, fields)
} catch {
// A diagnostic on the app's universal write path must never break writes.
const previous = get()
// Why the write is passed through untouched: zustand owns when and how an
// updater runs. The probe only compares the states on either side of it.
forward(args)
try {
const next = get()
if (next === previous || !isPlainRecord(previous) || !isPlainRecord(next)) {
return
}
const [partial, replace] = args
const fields =
replace !== true && partial !== null && typeof partial === 'object'
? Object.keys(partial)
: Object.keys(next)
recordWrite(previous, next, fields)
} catch {
// A diagnostic on the app's universal write path must never break writes.
}
}
}) as typeof set
api.setState = wrapped as typeof api.setState
return createState(wrapped, get, api)
}
// Why separately: an outer middleware may hand slices a different `set` than api.setState.
api.setState = probe(api.setState)
return createState(probe(set), get, api)
}
}
@@ -37,19 +37,44 @@ describe('web runtime call pinned to the probed runtime', () => {
it('sends a runtime-pinned write only on the connection where that runtime answered', async () => {
await callEnvironmentEnvelope('env-1', 'status.get')
await callEnvironmentEnvelope('env-1', 'files.write', {}, 15_000, undefined, 'runtime-a')
expect(session.client.call).toHaveBeenCalledTimes(2)
expect(session.client.call).toHaveBeenLastCalledWith(
'files.write',
{},
{ timeoutMs: 15_000, expectedConnectionEpoch: 1 }
)
})
it('refuses a runtime-pinned write after the client reconnected to another runtime', async () => {
await callEnvironmentEnvelope('env-1', 'status.get')
// The same client reconnects to an older server before the write is sent.
session.client.connectionEpoch = 2
session.client.runtimeId = 'runtime-b'
await expect(
callEnvironmentEnvelope('env-1', 'files.write', {}, 15_000, undefined, 'runtime-a')
).rejects.toThrow(RUNTIME_CONNECTION_CHANGED_MESSAGE)
expect(session.client.call).not.toHaveBeenCalledWith(
'files.write',
expect.anything(),
expect.anything()
)
})
// Why: the status owner's reconnect probes bypass this wrapper, so cached callers would never recover.
it('re-probes a reconnect to the same runtime and pins the write to the new connection', async () => {
await callEnvironmentEnvelope('env-1', 'status.get')
session.client.connectionEpoch = 2
await callEnvironmentEnvelope('env-1', 'files.write', {}, 15_000, undefined, 'runtime-a')
expect(session.client.call).toHaveBeenLastCalledWith(
'files.write',
{},
{
timeoutMs: 15_000,
expectedConnectionEpoch: 1
}
{ timeoutMs: 15_000, expectedConnectionEpoch: 2 }
)
})
@@ -57,6 +82,10 @@ describe('web runtime call pinned to the probed runtime', () => {
await expect(
callEnvironmentEnvelope('env-1', 'files.write', {}, 15_000, undefined, 'runtime-never-probed')
).rejects.toThrow(RUNTIME_CONNECTION_CHANGED_MESSAGE)
expect(session.client.call).not.toHaveBeenCalled()
expect(session.client.call).not.toHaveBeenCalledWith(
'files.write',
expect.anything(),
expect.anything()
)
})
})
@@ -74,34 +74,46 @@ function responseRuntimeId(response: unknown): string | undefined {
: undefined
}
type RuntimeClient = ReturnType<typeof getClientForEnvironment>
async function probeRuntimeStatus(
client: RuntimeClient,
params: unknown,
timeoutMs: number | undefined
): Promise<RuntimeRpcResponse<unknown>> {
const epoch = client.connectionEpoch
const response = await client.call('status.get', params, { timeoutMs })
const runtimeId = responseRuntimeId(response)
if (response.ok && runtimeId && client.connectionEpoch === epoch) {
probedConnections.set(client, { epoch, runtimeId })
}
return response
}
/**
* Why: a request shaped for one runtime's capabilities must reach that runtime. The same client
* reconnects internally, so pin it to the connection the matching status probe ran on; with no
* such probe, refuse so the caller re-probes.
* reconnects internally, so pin it to the connection a matching status probe ran on. Probes that
* bypass this wrapper (the status owner's) leave no proof, so a missing or stale one is re-probed.
*/
function callPinnedToRuntime(
client: ReturnType<typeof getClientForEnvironment>,
async function callPinnedToRuntime(
client: RuntimeClient,
method: string,
params: unknown,
timeoutMs: number | undefined,
expectedEnvironmentRuntimeId: string | undefined
): Promise<RuntimeRpcResponse<unknown>> {
if (method === 'status.get') {
const epoch = client.connectionEpoch
return client.call(method, params, { timeoutMs }).then((response) => {
const runtimeId = responseRuntimeId(response)
if (response.ok && runtimeId && client.connectionEpoch === epoch) {
probedConnections.set(client, { epoch, runtimeId })
}
return response
})
return probeRuntimeStatus(client, params, timeoutMs)
}
if (expectedEnvironmentRuntimeId === undefined) {
return client.call(method, params, { timeoutMs })
}
if (probedConnections.get(client)?.epoch !== client.connectionEpoch) {
await probeRuntimeStatus(client, undefined, timeoutMs)
}
const probed = probedConnections.get(client)
if (probed?.runtimeId !== expectedEnvironmentRuntimeId) {
return Promise.reject(new Error(RUNTIME_CONNECTION_CHANGED_MESSAGE))
throw new Error(RUNTIME_CONNECTION_CHANGED_MESSAGE)
}
return client.call(method, params, { timeoutMs, expectedConnectionEpoch: probed.epoch })
}
+2
View File
@@ -29,4 +29,6 @@ export type FsChangedPayload = {
events: FsChangeEvent[]
/** SSH connection the root was watched on; absent for local roots. Lets the renderer tell a local and an SSH root at the same path apart. */
connectionId?: string
/** Set on an overflow sent because the watch was down (re-watch, late install, reconnect), not because events were too many. */
resync?: true
}