From cbc6f54e1cdf62b3fca443434829f19d5937b0f8 Mon Sep 17 00:00:00 2001 From: OrcaWin Date: Fri, 9 Oct 2026 11:40:51 -0700 Subject: [PATCH] refactor(runtime-core): clean up Phase 3-3.8 code (#26894) * refactor(runtime-core): clean up Phase 3-3.8 code - worktree.ps: summary lookup is a closure, not a 4-arg callback - relay: shared RelayStreamAckWindow for fs and git response streams - relay: terminal-frame slots live on RelayStreamRegistry - one isBinaryBuffer/BINARY_PROBE_BYTES in shared/binary-buffer - one automation owner fence on the controller, no casts - drop unchecked Store cast in remote repo registration - speech download reuses the shared abort-aware delay, no casts - RelayRuntimeServices: drop leftover handler list and param casts - use shared errorMessage() in audited files Co-Authored-By: Claude * test(cross-version-wire): pass worktree ps summary lookup in both builds' shapes --------- Co-authored-by: m4air Co-authored-by: Claude Co-authored-by: m4air --- src/main/automations/precheck-runner.ts | 7 +- .../filesystem-file-content-inspection.ts | 13 +-- src/main/memory/hydrate-local-pty-registry.ts | 10 +- src/main/providers/ssh-pty-errors.ts | 6 +- .../providers/ssh-sftp-filesystem-provider.ts | 8 +- .../service/service-inactive-accounts.ts | 7 +- src/main/rate-limits/service/service-types.ts | 17 ---- .../orca-runtime-automation-operations.ts | 55 +++++------ .../runtime/orca-runtime-get-worktree-ps.ts | 33 ++----- .../orca-runtime-preserved-branch-cleanup.ts | 3 +- ...runtime-stop-explicitly-closed-tab-ptys.ts | 6 +- ...runtime-write-orchestration-pointer-pty.ts | 37 +------ .../runtime/runtime-automation-controller.ts | 32 +++--- src/main/runtime/runtime-file-command-host.ts | 10 -- ...ile-commands-read-file-explorer-preview.ts | 2 +- ...ntime-file-commands-terminal-file-paths.ts | 2 +- ...e-commands-write-terminal-artifact-file.ts | 3 +- .../runtime-rpc/runtime-rpc-lifecycle.ts | 3 +- ...ime-worktree-agent-rows-structured.test.ts | 4 +- ...untime-worktree-agent-rows-verdict.test.ts | 8 +- .../runtime/runtime-worktree-agent-rows.ts | 18 +--- .../runtime/runtime-worktree-ps-activity.ts | 43 ++------ ...worktree-ps-unverifiable-terminals.test.ts | 8 +- ...time-worktree-ps-unverifiable-terminals.ts | 15 +-- ...ree-structured-agent-rows-liveness.test.ts | 7 +- .../runtime/runtime-worktree-summary-paths.ts | 35 +++++++ .../speech/speech-model-download-response.ts | 18 ---- .../speech/speech-model-download-transport.ts | 17 ++-- src/main/terminal-scrollback-snapshots.ts | 9 +- src/relay/agent-exec-handler.ts | 3 +- src/relay/ai-vault-transcript-stream.ts | 2 +- src/relay/fs-handler-file-read.ts | 13 +-- src/relay/fs-handler-terminal-artifact.ts | 2 +- src/relay/fs-handler-utils.ts | 14 +-- src/relay/fs-stream-registry.ts | 97 ++++++++----------- src/relay/fs-stream-terminal-frame-slots.ts | 51 ---------- src/relay/git-response-stream.ts | 86 ++++------------ src/relay/relay-daemon.ts | 9 +- src/relay/relay-grace-lifecycle.ts | 5 +- src/relay/relay-runtime-services.ts | 61 ++++-------- src/relay/relay-stream-ack-window.ts | 51 ++++++++++ src/relay/relay-watcher-setup-wait.ts | 3 +- src/shared/abort-aware-delay.ts | 16 +++ src/shared/binary-buffer.ts | 4 +- .../cross-version-stopping.unit.test.ts | 7 +- ...s-version-worktree-ps-verdict.unit.test.ts | 7 +- 46 files changed, 317 insertions(+), 550 deletions(-) delete mode 100644 src/relay/fs-stream-terminal-frame-slots.ts create mode 100644 src/relay/relay-stream-ack-window.ts create mode 100644 src/shared/abort-aware-delay.ts diff --git a/src/main/automations/precheck-runner.ts b/src/main/automations/precheck-runner.ts index d4ab63b4d75..bf777d02af1 100644 --- a/src/main/automations/precheck-runner.ts +++ b/src/main/automations/precheck-runner.ts @@ -5,6 +5,7 @@ import { MAX_AUTOMATION_PRECHECK_OUTPUT_CHARS } from '../../shared/automation-pr import { getSshConnectionManager } from '../ssh/ssh-target-registry' import { shellEscape } from '../ssh/ssh-connection-utils' import { admitSelfInitiatedTreeKill } from '../own-chromium-tree-kill-guard' +import { errorMessage } from '../../shared/error-message' type AutomationPrecheckExecutionTarget = | { @@ -266,11 +267,7 @@ async function runSshPrecheck( const channel = await connection.exec(remoteCommand) return await runSshChannelPrecheck({ precheck, channel, startedAt }) } catch (error) { - return failedPrecheckResult( - precheck, - startedAt, - error instanceof Error ? error.message : String(error) - ) + return failedPrecheckResult(precheck, startedAt, errorMessage(error)) } } diff --git a/src/main/ipc/filesystem/filesystem-file-content-inspection.ts b/src/main/ipc/filesystem/filesystem-file-content-inspection.ts index 6bd7fc754ed..4d386093e15 100644 --- a/src/main/ipc/filesystem/filesystem-file-content-inspection.ts +++ b/src/main/ipc/filesystem/filesystem-file-content-inspection.ts @@ -1,5 +1,6 @@ import { extname } from 'node:path' import { localLogFileIdentity } from '../../ai-vault/local-log-tail-reader' +import { BINARY_PROBE_BYTES, isBinaryBuffer } from '../../../shared/binary-buffer' import { fileTooLargeError, openLocalRegularFile, @@ -9,7 +10,6 @@ import { // Why: Monaco degrades features on large files like VS Code, so a 5MB block would needlessly lock out ordinary JSON/log files. export const MAX_TEXT_FILE_SIZE = 50 * 1024 * 1024 // 50MB -export const BINARY_PROBE_BYTES = 8192 // Why: previewable binaries are base64 blobs (not parsed as text), and local IPC has no frame limit (unlike the relay's 10MB), so 50MB is safe. export const MAX_PREVIEWABLE_BINARY_SIZE = 50 * 1024 * 1024 // 50MB export const PREVIEWABLE_BINARY_MIME_TYPES: Record = { @@ -90,17 +90,6 @@ export async function readLocalLogSnapshot(filePath: string): Promise { - console.warn( - '[memory] Boot-time pty-registry hydration failed:', - error instanceof Error ? error.message : String(error) - ) + console.warn('[memory] Boot-time pty-registry hydration failed:', errorMessage(error)) }) .finally(() => { clearTimeout(deadline) @@ -139,7 +137,7 @@ async function hydrateLocalPtyRegistry( complete = false console.warn( '[memory] Worktree enumeration failed during pty-registry hydration:', - result.reason instanceof Error ? result.reason.message : String(result.reason) + errorMessage(result.reason) ) continue } @@ -260,7 +258,7 @@ async function collectSessionInfos( throwIfSignalAborted(signal) console.warn( '[memory] listSessions failed for one adapter during hydration:', - error instanceof Error ? error.message : String(error) + errorMessage(error) ) return { complete: false, sessions: [] } } diff --git a/src/main/providers/ssh-pty-errors.ts b/src/main/providers/ssh-pty-errors.ts index 24f0467263e..bd25bff4f5c 100644 --- a/src/main/providers/ssh-pty-errors.ts +++ b/src/main/providers/ssh-pty-errors.ts @@ -1,3 +1,5 @@ +import { errorMessage } from '../../shared/error-message' + export const SSH_SESSION_EXPIRED_ERROR = 'SSH_SESSION_EXPIRED' export const SSH_PTY_IDENTITY_MISMATCH_ERROR = 'SSH_PTY_IDENTITY_MISMATCH' /** @@ -13,12 +15,12 @@ export const SSH_PTY_SOURCE_RESTORE_REQUIRED_ERROR = 'SSH_PTY_SOURCE_RESTORE_REQ export const SSH_PTY_HELD_BY_PREVIOUS_RELAY_ERROR = 'SSH_PTY_HELD_BY_PREVIOUS_RELAY' export function isSshPtyNotFoundError(error: unknown): boolean { - const message = error instanceof Error ? error.message : String(error) + const message = errorMessage(error) return /PTY ".+" not found/i.test(message) } export function isSshPtyIdentityMismatchError(error: unknown): boolean { - const message = error instanceof Error ? error.message : String(error) + const message = errorMessage(error) return message.includes(SSH_PTY_IDENTITY_MISMATCH_ERROR) || /identity mismatch/i.test(message) } diff --git a/src/main/providers/ssh-sftp-filesystem-provider.ts b/src/main/providers/ssh-sftp-filesystem-provider.ts index 016be885e57..ddbcabe951f 100644 --- a/src/main/providers/ssh-sftp-filesystem-provider.ts +++ b/src/main/providers/ssh-sftp-filesystem-provider.ts @@ -9,6 +9,7 @@ export { readSftpDirectory } from './ssh-sftp-directory-listing' import { extname } from 'node:path' import type { SFTPWrapper, Stats } from 'ssh2' import { IMAGE_FILE_MIME_TYPES } from '../../shared/image-file-extensions' +import { isBinaryBuffer } from '../../shared/binary-buffer' import { capturePathExistence, type PathExistenceResult } from '../../shared/path-existence-batch' import type { SearchResult } from '../../shared/code-search-types' import type { DirEntry } from '../../shared/filesystem-entry-types' @@ -23,10 +24,9 @@ import { import { fileStatFromSftpStats, lstatViaSftp, statViaSftp } from './ssh-filesystem-provider-sftp' import type { FileReadLimits, FileReadResult, FileStat, IFilesystemProvider } from './types' -// Why: same caps and probe window as the relay's fs.readFile so previews behave identically. +// Why: same caps as the relay's fs.readFile so previews behave identically. const MAX_TEXT_FILE_SIZE = 10 * 1024 * 1024 const MAX_PREVIEWABLE_BINARY_SIZE = 50 * 1024 * 1024 -const BINARY_PROBE_BYTES = 8192 const SFTP_NO_SUCH_FILE = 2 const PREVIEW_MIME_TYPES: Record = { ...IMAGE_FILE_MIME_TYPES, @@ -55,10 +55,6 @@ function normalizeSftpError(error: unknown): Error { return err } -function isBinaryBuffer(buffer: Buffer): boolean { - return buffer.subarray(0, BINARY_PROBE_BYTES).includes(0) -} - export class SshSftpFilesystemProvider implements IFilesystemProvider { private readonly channel: SftpFilesystemChannel diff --git a/src/main/rate-limits/service/service-inactive-accounts.ts b/src/main/rate-limits/service/service-inactive-accounts.ts index e59d1eeb3a9..e0d388639c8 100644 --- a/src/main/rate-limits/service/service-inactive-accounts.ts +++ b/src/main/rate-limits/service/service-inactive-accounts.ts @@ -1,11 +1,8 @@ import { fetchManagedAccountUsage } from '../claude-fetcher' import { fetchCodexRateLimits } from '../codex-fetcher' import { RateLimitServicePolling } from './service-polling' -import { - INACTIVE_CODEX_PROBE_STAGGER_MS, - INACTIVE_FETCH_DEBOUNCE_MS, - delayUnlessAborted -} from './service-types' +import { INACTIVE_CODEX_PROBE_STAGGER_MS, INACTIVE_FETCH_DEBOUNCE_MS } from './service-types' +import { delayUnlessAborted } from '../../../shared/abort-aware-delay' export abstract class RateLimitServiceInactiveAccounts extends RateLimitServicePolling { async fetchInactiveClaudeAccountsOnOpen(): Promise { diff --git a/src/main/rate-limits/service/service-types.ts b/src/main/rate-limits/service/service-types.ts index 5bae83b1449..eec4ffd50cb 100644 --- a/src/main/rate-limits/service/service-types.ts +++ b/src/main/rate-limits/service/service-types.ts @@ -162,23 +162,6 @@ export function normalizeClaudeConfigDir(dir: string | null | undefined): string return trimmed || null } -export function delayUnlessAborted(ms: number, signal: AbortSignal): Promise { - if (signal.aborted) { - return Promise.resolve() - } - return new Promise((resolve) => { - const onAbort = (): void => { - clearTimeout(timer) - resolve() - } - const timer = setTimeout(() => { - signal.removeEventListener('abort', onAbort) - resolve() - }, ms) - signal.addEventListener('abort', onAbort, { once: true }) - }) -} - export function isSameUsageWindow( a: ProviderRateLimits['session'], b: ProviderRateLimits['session'] diff --git a/src/main/runtime/orca-runtime-automation-operations.ts b/src/main/runtime/orca-runtime-automation-operations.ts index 32d47c59951..294d540c0ca 100644 --- a/src/main/runtime/orca-runtime-automation-operations.ts +++ b/src/main/runtime/orca-runtime-automation-operations.ts @@ -1,7 +1,6 @@ // @ts-nocheck -- mechanically split from OrcaRuntimeService; behavior is covered by AST equivalence and characterization tests. import { OrcaRuntimeWithPtyForegroundProcessReads } from './orca-runtime-pty-foreground-process-reads' import type { - AutomationOwnerFenceOperation, AutomationOwnerPrecondition, AutomationDestination } from '../../shared/automation-owner-precondition' @@ -25,33 +24,29 @@ import { makePaneKey } from '../../shared/stable-pane-id' import { runtimeWorktreeIdsEqual } from './runtime-worktree-path-identity' export class OrcaRuntimeWithAutomationOperations extends OrcaRuntimeWithPtyForegroundProcessReads { - protected fenceAutomationOwner( - id: string, + private readAutomationRuns( + automationId: string | undefined, expectedOwner: AutomationOwnerPrecondition | undefined, - operation: AutomationOwnerFenceOperation - ): void { - if (!this.store?.assertAutomationOwnerFence) { - if (expectedOwner) { - throw new Error('runtime_unavailable') - } - return + read: () => T + ): T { + if (expectedOwner && !automationId) { + throw new Error('An expected owner requires an automation id.') } - this.store.assertAutomationOwnerFence({ id, expectedOwner, operation }) + return this.automation.withExternalProbePriority(() => { + if (automationId) { + this.automation.assertOwner(automationId, expectedOwner, 'read') + } + return read() + }) } listAutomationRuns( automationId?: string, expectedOwner?: AutomationOwnerPrecondition ): AutomationRun[] { - if (expectedOwner && !automationId) { - throw new Error('An expected owner requires an automation id.') - } - return this.automation.withExternalProbePriority(() => { - if (automationId) { - this.fenceAutomationOwner(automationId, expectedOwner, 'read') - } - return this.automation.listRuns(automationId) - }) + return this.readAutomationRuns(automationId, expectedOwner, () => + this.automation.listRuns(automationId) + ) } listAutomationRunsPage( @@ -60,20 +55,14 @@ export class OrcaRuntimeWithAutomationOperations extends OrcaRuntimeWithPtyForeg limit?: number, cursor?: string ) { - if (expectedOwner && !automationId) { - throw new Error('An expected owner requires an automation id.') - } - return this.automation.withExternalProbePriority(() => { - if (automationId) { - this.fenceAutomationOwner(automationId, expectedOwner, 'read') - } - return this.automation.listRunsPage(automationId, limit, cursor) - }) + return this.readAutomationRuns(automationId, expectedOwner, () => + this.automation.listRunsPage(automationId, limit, cursor) + ) } showAutomation(id: string, expectedOwner?: AutomationOwnerPrecondition): Automation { const automation = this.automation.show(id) - this.fenceAutomationOwner(id, expectedOwner, 'read') + this.automation.assertOwner(id, expectedOwner, 'read') return automation } @@ -107,11 +96,11 @@ export class OrcaRuntimeWithAutomationOperations extends OrcaRuntimeWithPtyForeg updateAutomation( id: string, updates: RuntimeAutomationUpdateInput, - options?: unknown + options?: { expectedOwner?: AutomationOwnerPrecondition; destination?: AutomationDestination } ): Promise { return this.automation.withExternalProbePriority(() => { const source = this.automationChangeSelector(id) - return this.automation.update(id, updates, options as never).then((automation) => { + return this.automation.update(id, updates, options).then((automation) => { this.publishAutomationDefinitionChange(source, this.automationChangeSelector(automation.id)) return automation }) @@ -136,7 +125,7 @@ export class OrcaRuntimeWithAutomationOperations extends OrcaRuntimeWithPtyForeg expectedOwner?: AutomationOwnerPrecondition ): Promise { return this.automation.withExternalProbePriority(() => - this.automation.runNow(id, expectedOwner as never) + this.automation.runNow(id, expectedOwner) ) } diff --git a/src/main/runtime/orca-runtime-get-worktree-ps.ts b/src/main/runtime/orca-runtime-get-worktree-ps.ts index 1698676a3cb..2fe58e13de3 100644 --- a/src/main/runtime/orca-runtime-get-worktree-ps.ts +++ b/src/main/runtime/orca-runtime-get-worktree-ps.ts @@ -5,7 +5,10 @@ import { OrcaRuntimeWithStartTuiIdleVisibleReadProbe } from './orca-runtime-star import { DEFAULT_WORKTREE_PS_LIMIT } from './orca-runtime-postlude' import type { RuntimeWorktreePsResult } from '../../shared/runtime-types' import { buildRuntimeWorktreePsSummaries } from './runtime-worktree-ps-summaries' -import { buildRuntimeWorktreeSummaryPathIndex } from './runtime-worktree-summary-paths' +import { + buildRuntimeWorktreeSummaryPathIndex, + createRuntimeWorktreePsSummaryLookup +} from './runtime-worktree-summary-paths' import { applyRuntimeWorktreePsSessionActivity, applyRuntimeWorktreePsTerminalActivity @@ -85,18 +88,13 @@ export class OrcaRuntimeWithGetWorktreePs extends OrcaRuntimeWithStartTuiIdleVis platformByRepoId }) - const runtimeWorktreeSummaryPathIndex = buildRuntimeWorktreeSummaryPathIndex( + const getSummary = createRuntimeWorktreePsSummaryLookup( summaries, - resolvedWorktrees, - platformByRepoId + buildRuntimeWorktreeSummaryPathIndex(summaries, resolvedWorktrees, platformByRepoId) ) - const missingRuntimeWorktreeIds = new Set() const session = this.store?.getWorkspaceSession?.() const countedPtyIds = new Set() const workingTerminalEvidenceByWorktreeId = applyRuntimeWorktreePsTerminalActivity({ - summaries, - pathIndex: runtimeWorktreeSummaryPathIndex, - missingIds: missingRuntimeWorktreeIds, freshPtyLiveness, countedPtyIds, leaves: this.leaves.values(), @@ -105,37 +103,27 @@ export class OrcaRuntimeWithGetWorktreePs extends OrcaRuntimeWithStartTuiIdleVis session, getPaneKey: (leaf) => this.makeRuntimePaneKey(leaf), getTitleDisplayClear: (ptyId) => this.getPtyTitleDisplayClear(ptyId), - getSummary: (summaryMap, pathIndex, missingIds, worktreeId) => - this.getSummaryForRuntimeWorktreeId(summaryMap, pathIndex, missingIds, worktreeId) + getSummary }) applyRuntimeWorktreePsUnverifiableTerminals({ - summaries, - pathIndex: runtimeWorktreeSummaryPathIndex, - missingIds: missingRuntimeWorktreeIds, countedPtyIds, leaves: this.leaves.values(), ptysById: this.ptysById, getLivenessVerdict: (ptyId) => this.getPtyLivenessVerdict(ptyId), - getSummary: (summaryMap, pathIndex, missingIds, worktreeId) => - this.getSummaryForRuntimeWorktreeId(summaryMap, pathIndex, missingIds, worktreeId) + getSummary }) const { mirroredWorktreeIdByTabId, connectedPtyEvidence } = applyRuntimeWorktreePsSessionActivity({ store: this.store, summaries, repoById, - pathIndex: runtimeWorktreeSummaryPathIndex, - missingIds: missingRuntimeWorktreeIds, ptysById: this.ptysById, tabs: this.tabs, getTerminalHandlesForPty: (ptyId) => this.getExistingTerminalHandlesForPtyId(ptyId), - getSummary: (summaryMap, pathIndex, missingIds, worktreeId) => - this.getSummaryForRuntimeWorktreeId(summaryMap, pathIndex, missingIds, worktreeId) + getSummary }) attachRuntimeWorktreeAgentRows({ summaries, - pathIndex: runtimeWorktreeSummaryPathIndex, - missingWorktreeIds: missingRuntimeWorktreeIds, workingTerminalEvidenceByWorktreeId, rowSources: collectRuntimeWorktreeAgentSources({ mirroredWorktreeIdByTabId, @@ -144,8 +132,7 @@ export class OrcaRuntimeWithGetWorktreePs extends OrcaRuntimeWithStartTuiIdleVis hookSnapshots: this.getAgentStatusSnapshotFn?.() ?? [] }), orchestrationByPaneKey: this.agentOrchestrationProjection.buildByPaneKey(), - getSummary: (summaryMap, pathIndex, missingIds, worktreeId) => - this.getSummaryForRuntimeWorktreeId(summaryMap, pathIndex, missingIds, worktreeId) + getSummary }) const sorted = [...summaries.values()].sort(compareWorktreePs) diff --git a/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts b/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts index 992b371daf2..be727819a06 100644 --- a/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts +++ b/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts @@ -32,7 +32,6 @@ import { RuntimeMobileSpeechCatalog } from './runtime-mobile-speech-catalog' import { RuntimeMobileDictationController } from './runtime-mobile-dictation-controller' import { RuntimeProjectHostSetupController } from './runtime-project-host-setup-controller' import { addRemoteRepoFromPath } from '../ipc/repos/remote-repo-registration' -import type { Store } from '../persistence' import { RuntimeProjectGroupController } from './runtime-project-group-controller' import { RuntimeNestedRepoImport } from './runtime-nested-repo-import' import { RuntimeRepositoryRegistrationController } from './runtime-repository-registration-controller' @@ -253,7 +252,7 @@ export class OrcaRuntimeWithPreservedBranchCleanup extends OrcaRuntimeWithTermin (this as RuntimeCommandSurfaceHost).addRepo(path, kind, hostId), addRemoteRepo: async (remote) => { // The same registration the desktop IPC handler uses, so both surfaces agree on SSH hosts. - const result = await addRemoteRepoFromPath(this.requireStore() as unknown as Store, remote) + const result = await addRemoteRepoFromPath(this.requireStore(), remote) if ('error' in result) { throw new Error(result.error) } diff --git a/src/main/runtime/orca-runtime-stop-explicitly-closed-tab-ptys.ts b/src/main/runtime/orca-runtime-stop-explicitly-closed-tab-ptys.ts index ea954fbce8b..e87a6fd996d 100644 --- a/src/main/runtime/orca-runtime-stop-explicitly-closed-tab-ptys.ts +++ b/src/main/runtime/orca-runtime-stop-explicitly-closed-tab-ptys.ts @@ -5,6 +5,7 @@ import { SSH_PROVIDER_UNREGISTERED_REASON } from '../../shared/pty-liveness-verd import type { RuntimeTerminalClose } from '../../shared/runtime-types' import type { RuntimePtyTabCloseAuthority } from './runtime-terminal-state-records' import { parsePaneKey } from '../../shared/stable-pane-id' +import { errorMessage } from '../../shared/error-message' /** How an explicit close's stop of its addressed PTY ended. */ type ExplicitCloseStop = { stopped: boolean; pendingKillRecorded: boolean } @@ -27,10 +28,7 @@ export class OrcaRuntimeWithStopExplicitlyClosedTabPtys extends OrcaRuntimeWithF try { stopped = await this.ptyController.stopAndWait(ptyId, { deadlineMs }) } catch (error) { - this.markPtyLivenessUnverifiable( - ptyId, - error instanceof Error ? error.message : String(error) - ) + this.markPtyLivenessUnverifiable(ptyId, errorMessage(error)) } // Preserve an observed exit when a broader inventory check could not finish. if ( diff --git a/src/main/runtime/orca-runtime-write-orchestration-pointer-pty.ts b/src/main/runtime/orca-runtime-write-orchestration-pointer-pty.ts index 2d95f3bf3e5..244b88f52fd 100644 --- a/src/main/runtime/orca-runtime-write-orchestration-pointer-pty.ts +++ b/src/main/runtime/orca-runtime-write-orchestration-pointer-pty.ts @@ -10,10 +10,7 @@ import type { TerminalAgent } from '../../shared/terminal-agent' import { selectRuntimeHookAgentRowForPane } from './runtime-mobile-agent-status-projection' import { isTuiAgent } from '../../shared/tui-agent-config' import { resolvePublishedPaneAgentIdentity } from '../../shared/published-pane-agent-identity' -import type { RuntimeTerminalSummary, RuntimeWorktreePsSummary } from '../../shared/runtime-types' -import type { RuntimeWorktreeSummaryPathIndex } from './runtime-worktree-summary-paths' -import { parseRuntimeWorktreeId } from './runtime-worktree-path-identity' -import { findRuntimeWorktreeSummaryByPath } from './runtime-worktree-summary-paths' +import type { RuntimeTerminalSummary } from '../../shared/runtime-types' import type { ResolvedWorktree } from './runtime-worktree-path-identity' import { getLatestLeafTitle } from './runtime-worktree-status-projection' import { parseAppSshPtyId } from '../../shared/ssh-pty-id' @@ -108,38 +105,6 @@ export class OrcaRuntimeWithWriteOrchestrationPointerPty extends OrcaRuntimeWith return agentIdentity ? { agentIdentity } : {} } - protected getSummaryForRuntimeWorktreeId( - summaries: Map, - runtimeWorktreeSummaryPathIndex: RuntimeWorktreeSummaryPathIndex, - missingRuntimeWorktreeIds: Set, - runtimeWorktreeId: string - ): RuntimeWorktreePsSummary | null { - const exact = summaries.get(runtimeWorktreeId) - if (exact) { - return exact - } - if (missingRuntimeWorktreeIds.has(runtimeWorktreeId)) { - return null - } - const parsed = parseRuntimeWorktreeId(runtimeWorktreeId) - if (!parsed) { - return null - } - const comparisonPlatform = - runtimeWorktreeSummaryPathIndex.platformByRepoId.get(parsed.repoId) ?? process.platform - const indexed = findRuntimeWorktreeSummaryByPath( - runtimeWorktreeSummaryPathIndex, - parsed.repoId, - parsed.worktreePath, - comparisonPlatform - ) - if (indexed) { - return indexed - } - missingRuntimeWorktreeIds.add(runtimeWorktreeId) - return null - } - protected buildTerminalSummary( leaf: RuntimeLeafRecord, worktreesById: Map, diff --git a/src/main/runtime/runtime-automation-controller.ts b/src/main/runtime/runtime-automation-controller.ts index 173a6580964..2c6b95dfb95 100644 --- a/src/main/runtime/runtime-automation-controller.ts +++ b/src/main/runtime/runtime-automation-controller.ts @@ -11,6 +11,7 @@ import type { Worktree } from '../../shared/worktree/types' import type { RuntimeStore } from './runtime-store-contract' import type { AutomationListParams, AutomationListResult } from '../../shared/automation-list-scope' import type { + AutomationOwnerFenceOperation, AutomationOwnerPrecondition, AutomationDestination } from '../../shared/automation-owner-precondition' @@ -209,26 +210,27 @@ export class RuntimeAutomationController { if (!this.service) { throw new Error('runtime_unavailable') } - const service = this.service return await runAutomationNowFenced({ automationId: id, - service, - fence: () => { - if (!this.store?.assertAutomationOwnerFence) { - if (expectedOwner) { - throw new Error('runtime_unavailable') - } - return - } - this.store.assertAutomationOwnerFence({ - id, - expectedOwner, - operation: 'execute' - }) - } + service: this.service, + fence: () => this.assertOwner(id, expectedOwner, 'execute') }) } + assertOwner( + id: string, + expectedOwner: AutomationOwnerPrecondition | undefined, + operation: AutomationOwnerFenceOperation + ): void { + if (!this.store?.assertAutomationOwnerFence) { + if (expectedOwner) { + throw new Error('runtime_unavailable') + } + return + } + this.store.assertAutomationOwnerFence({ id, expectedOwner, operation }) + } + private copyPatchValues( updates: RuntimeAutomationUpdateInput, patch: AutomationUpdateInput diff --git a/src/main/runtime/runtime-file-command-host.ts b/src/main/runtime/runtime-file-command-host.ts index 3fb74c71632..0f9d684e3a1 100644 --- a/src/main/runtime/runtime-file-command-host.ts +++ b/src/main/runtime/runtime-file-command-host.ts @@ -187,13 +187,3 @@ export function isRuntimeDirectoryEntry(entry: { } return false } - -export function isBinaryBuffer(buffer: Buffer): boolean { - const len = Math.min(buffer.length, 8192) - for (let i = 0; i < len; i += 1) { - if (buffer[i] === 0) { - return true - } - } - return false -} diff --git a/src/main/runtime/runtime-file-commands-read-file-explorer-preview.ts b/src/main/runtime/runtime-file-commands-read-file-explorer-preview.ts index 25f7069e213..fc0a5a21917 100644 --- a/src/main/runtime/runtime-file-commands-read-file-explorer-preview.ts +++ b/src/main/runtime/runtime-file-commands-read-file-explorer-preview.ts @@ -20,7 +20,7 @@ import { NodeFileReadTooLargeError, readNodeFileWithinLimit } from '../../shared/node-bounded-file-reader' -import { isBinaryBuffer } from './runtime-file-command-host' +import { isBinaryBuffer } from '../../shared/binary-buffer' import type { DocPreviewFileAccessRequest, DocPreviewFileAccessResult diff --git a/src/main/runtime/runtime-file-commands-terminal-file-paths.ts b/src/main/runtime/runtime-file-commands-terminal-file-paths.ts index dca41999ad8..03ee354dcba 100644 --- a/src/main/runtime/runtime-file-commands-terminal-file-paths.ts +++ b/src/main/runtime/runtime-file-commands-terminal-file-paths.ts @@ -20,7 +20,7 @@ import { localTerminalArtifactRoots, readFileHandleBufferBounded } from './runtime-file-commands-terminal-artifact-access' -import { isBinaryBuffer } from './runtime-file-command-host' +import { isBinaryBuffer } from '../../shared/binary-buffer' import type { RuntimeFilePreviewResult } from '../../shared/runtime-types' import { isPathInsideOrEqual, diff --git a/src/main/runtime/runtime-file-commands-write-terminal-artifact-file.ts b/src/main/runtime/runtime-file-commands-write-terminal-artifact-file.ts index 26c41e66b5f..15a1a47a951 100644 --- a/src/main/runtime/runtime-file-commands-write-terminal-artifact-file.ts +++ b/src/main/runtime/runtime-file-commands-write-terminal-artifact-file.ts @@ -6,7 +6,8 @@ import { MOBILE_FILE_READ_MAX_BYTES, previewableBinaryByteLimit } from './runtime-file-commands-mobile-file-list-limit' -import { isBinaryBuffer, isMobileBinaryPath } from './runtime-file-command-host' +import { isBinaryBuffer } from '../../shared/binary-buffer' +import { isMobileBinaryPath } from './runtime-file-command-host' import { assertTerminalArtifactContentUnchanged, assertTerminalFileGrantFresh, diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts b/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts index efa05957f40..3a9ec672e3c 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-lifecycle.ts @@ -19,6 +19,7 @@ import { createRuntimeTransportMetadata, sweepOrphanedRuntimeSockets } from './runtime-rpc-socket-metadata' +import { errorMessage } from '../../../shared/error-message' export class RuntimeRpcLifecycle extends RuntimeRpcWebSocketDispatch { async start(): Promise { @@ -51,7 +52,7 @@ export class RuntimeRpcLifecycle extends RuntimeRpcWebSocketDispatch { reply(JSON.stringify(response)) }) .catch((error) => { - const message = error instanceof Error ? error.message : String(error) + const message = errorMessage(error) // Why: best-effort id recovery so the client can correlate the error frame to its pending request. let id = 'unknown' try { diff --git a/src/main/runtime/runtime-worktree-agent-rows-structured.test.ts b/src/main/runtime/runtime-worktree-agent-rows-structured.test.ts index 3b30f932a1e..7d6f94207be 100644 --- a/src/main/runtime/runtime-worktree-agent-rows-structured.test.ts +++ b/src/main/runtime/runtime-worktree-agent-rows-structured.test.ts @@ -59,8 +59,6 @@ function attach(summaries: AgentSessionStatusSummary[]): RuntimeWorktreePsSummar const summariesById = new Map([[WORKTREE_ID, row]]) attachRuntimeWorktreeAgentRows({ summaries: summariesById, - pathIndex: { byPath: new Map(), byRealPath: new Map() } as never, - missingWorktreeIds: new Set(), workingTerminalEvidenceByWorktreeId: new Map(), rowSources: collectRuntimeWorktreeAgentSources({ mirroredWorktreeIdByTabId: new Map(), @@ -72,7 +70,7 @@ function attach(summaries: AgentSessionStatusSummary[]): RuntimeWorktreePsSummar hookSnapshots: store.getStatusSnapshot() }), orchestrationByPaneKey: null, - getSummary: (map, _p, _m, id) => map.get(id) ?? null + getSummary: (id) => summariesById.get(id) ?? null }) return row } diff --git a/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts b/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts index ff82a61406e..ca955e241c0 100644 --- a/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts +++ b/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts @@ -109,11 +109,9 @@ function ingest(summary: AgentSessionStatusSummary, children: AgentChildWorkEvid hasHostSidebarActivity: false, agents: [] } as unknown as RuntimeWorktreePsSummary + const summaries = new Map([[WORKSPACE_ID, row]]) attachRuntimeWorktreeAgentRows({ - summaries: new Map([[WORKSPACE_ID, row]]), - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: `getSummary` below resolves every row by id, so the path index is never read. - pathIndex: { byPath: new Map(), byRealPath: new Map() } as never, - missingWorktreeIds: new Set(), + summaries, workingTerminalEvidenceByWorktreeId: new Map(), rowSources: collectRuntimeWorktreeAgentSources({ mirroredWorktreeIdByTabId: new Map(), @@ -125,7 +123,7 @@ function ingest(summary: AgentSessionStatusSummary, children: AgentChildWorkEvid hookSnapshots }), orchestrationByPaneKey: null, - getSummary: (map, _p, _m, id) => map.get(id) ?? null + getSummary: (id) => summaries.get(id) ?? null }) return { status: hookSnapshots[0], ps: row.agents[0] } } diff --git a/src/main/runtime/runtime-worktree-agent-rows.ts b/src/main/runtime/runtime-worktree-agent-rows.ts index 310b0f6bc52..3bad0943a10 100644 --- a/src/main/runtime/runtime-worktree-agent-rows.ts +++ b/src/main/runtime/runtime-worktree-agent-rows.ts @@ -1,7 +1,7 @@ import { isFreshNonDoneAgentStatus } from '../../shared/agent-status-types' import type { RuntimeWorktreeAgentRow, RuntimeWorktreePsSummary } from '../../shared/runtime-types' import { mergeWorktreeSummaryStatus } from './runtime-worktree-status-projection' -import type { RuntimeWorktreeSummaryPathIndex } from './runtime-worktree-summary-paths' +import type { RuntimeWorktreePsSummaryLookup } from './runtime-worktree-summary-paths' import type { RuntimeWorkingTerminalEvidence } from './runtime-worktree-ps-activity' import type { RuntimeWorktreeAgentSource } from './runtime-worktree-agent-source' export type { RuntimeAgentRowSnapshot } from './runtime-hook-agent-row-selection' @@ -14,20 +14,13 @@ type OrchestrationDisplay = { export function attachRuntimeWorktreeAgentRows(args: { summaries: Map - pathIndex: RuntimeWorktreeSummaryPathIndex - missingWorktreeIds: Set rowSources: ReadonlyMap workingTerminalEvidenceByWorktreeId: ReadonlyMap< string, readonly RuntimeWorkingTerminalEvidence[] > orchestrationByPaneKey: Record | null | undefined - getSummary: ( - summaries: Map, - pathIndex: RuntimeWorktreeSummaryPathIndex, - missingWorktreeIds: Set, - worktreeId: string - ) => RuntimeWorktreePsSummary | null + getSummary: RuntimeWorktreePsSummaryLookup }): void { const { rowSources } = args const now = Date.now() @@ -37,12 +30,7 @@ export function attachRuntimeWorktreeAgentRows(args: { if (!worktreeId) { continue } - const summary = args.getSummary( - args.summaries, - args.pathIndex, - args.missingWorktreeIds, - worktreeId - ) + const summary = args.getSummary(worktreeId) if (!summary) { continue } diff --git a/src/main/runtime/runtime-worktree-ps-activity.ts b/src/main/runtime/runtime-worktree-ps-activity.ts index 496acb46def..1b347c9080c 100644 --- a/src/main/runtime/runtime-worktree-ps-activity.ts +++ b/src/main/runtime/runtime-worktree-ps-activity.ts @@ -5,7 +5,7 @@ import { getRepoExecutionHostId, type ExecutionHostId } from '../../shared/execu import { parsePaneKey } from '../../shared/stable-pane-id' import type { RuntimeStore } from './runtime-store-contract' import type { RuntimeLeafRecord, RuntimePtyWorktreeRecord } from './runtime-terminal-state-records' -import type { RuntimeWorktreeSummaryPathIndex } from './runtime-worktree-summary-paths' +import type { RuntimeWorktreePsSummaryLookup } from './runtime-worktree-summary-paths' import { getLatestPtyTitle, getLeafDisplayRecord, @@ -18,13 +18,6 @@ import { } from './runtime-worktree-status-projection' import { runtimeWorktreeIdsEqual } from './runtime-worktree-path-identity' -type SummaryLookup = ( - summaries: Map, - pathIndex: RuntimeWorktreeSummaryPathIndex, - missingIds: Set, - worktreeId: string -) => RuntimeWorktreePsSummary | null - export type RuntimeWorkingTerminalEvidence = { paneKey: string | null ptyId: string | null @@ -32,9 +25,6 @@ export type RuntimeWorkingTerminalEvidence = { } export function applyRuntimeWorktreePsTerminalActivity(args: { - summaries: Map - pathIndex: RuntimeWorktreeSummaryPathIndex - missingIds: Set freshPtyLiveness: ReadonlySet | null /** Filled with every PTY this pass counts as live. */ countedPtyIds: Set @@ -44,7 +34,7 @@ export function applyRuntimeWorktreePsTerminalActivity(args: { session: WorkspaceSessionState | null | undefined getPaneKey: (leaf: RuntimeLeafRecord) => string getTitleDisplayClear: (ptyId: string) => TitleDisplayClear | null - getSummary: SummaryLookup + getSummary: RuntimeWorktreePsSummaryLookup }): Map { const workingEvidence = new Map() const savedTabOwnerById = new Map() @@ -78,12 +68,7 @@ export function applyRuntimeWorktreePsTerminalActivity(args: { ) { continue } - const summary = args.getSummary( - args.summaries, - args.pathIndex, - args.missingIds, - leaf.worktreeId - ) + const summary = args.getSummary(leaf.worktreeId) if (!summary) { continue } @@ -146,12 +131,7 @@ export function applyRuntimeWorktreePsTerminalActivity(args: { if (!owner) { continue } - const summary = args.getSummary( - args.summaries, - args.pathIndex, - args.missingIds, - owner.worktreeId - ) + const summary = args.getSummary(owner.worktreeId) if (!summary) { continue } @@ -197,13 +177,11 @@ export function applyRuntimeWorktreePsSessionActivity(args: { store: RuntimeStore | null summaries: Map repoById: ReadonlyMap - pathIndex: RuntimeWorktreeSummaryPathIndex - missingIds: Set ptysById: ReadonlyMap tabs: ReadonlyMap /** Non-minting: a listing must not issue handles, only recognise the ones already bound. */ getTerminalHandlesForPty: (ptyId: string) => readonly string[] - getSummary: SummaryLookup + getSummary: RuntimeWorktreePsSummaryLookup }): { mirroredWorktreeIdByTabId: Map connectedPtyEvidence: { @@ -229,7 +207,7 @@ export function applyRuntimeWorktreePsSessionActivity(args: { if (tabs.length === 0) { continue } - const summary = args.getSummary(args.summaries, args.pathIndex, args.missingIds, worktreeId) + const summary = args.getSummary(worktreeId) if (summary && tabs.some((tab) => tab.ptyId && args.ptysById.get(tab.ptyId)?.connected)) { summary.hasHostSidebarActivity = true } @@ -238,18 +216,13 @@ export function applyRuntimeWorktreePsSessionActivity(args: { if (tabs.length === 0) { continue } - const summary = args.getSummary(args.summaries, args.pathIndex, args.missingIds, worktreeId) + const summary = args.getSummary(worktreeId) if (summary) { summary.hasHostSidebarActivity = true } } if (session.activeWorktreeId) { - const summary = args.getSummary( - args.summaries, - args.pathIndex, - args.missingIds, - session.activeWorktreeId - ) + const summary = args.getSummary(session.activeWorktreeId) if (summary) { summary.isActive = true } diff --git a/src/main/runtime/runtime-worktree-ps-unverifiable-terminals.test.ts b/src/main/runtime/runtime-worktree-ps-unverifiable-terminals.test.ts index 8696b75ceba..b6c24cf3bba 100644 --- a/src/main/runtime/runtime-worktree-ps-unverifiable-terminals.test.ts +++ b/src/main/runtime/runtime-worktree-ps-unverifiable-terminals.test.ts @@ -9,18 +9,14 @@ describe('unverifiable terminal attribution', () => { ['wt-new', {}] ]) applyRuntimeWorktreePsUnverifiableTerminals({ - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: only the counters this pass writes are read back. - summaries: summaries as Map, - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: getSummary below ignores the index. - pathIndex: {} as never, - missingIds: new Set(), countedPtyIds: new Set(), // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the pass reads only ptyId and worktreeId. leaves: [{ ptyId: 'pty-1', worktreeId: 'wt-old' } as never], // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the pass reads only ptyId and worktreeId. ptysById: new Map([['pty-1', { ptyId: 'pty-1', worktreeId: 'wt-new' } as never]]), getLivenessVerdict: () => ({ status: 'unverifiable', reason: 'lost contact' }), - getSummary: (all, _index, _missing, worktreeId) => all.get(worktreeId) ?? null + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: only the counters this pass writes are read back. + getSummary: (id) => (summaries.get(id) as RuntimeWorktreePsSummary | undefined) ?? null }) expect(summaries.get('wt-new')?.unverifiableTerminalCount).toBe(1) expect(summaries.get('wt-old')?.unverifiableTerminalCount).toBeUndefined() diff --git a/src/main/runtime/runtime-worktree-ps-unverifiable-terminals.ts b/src/main/runtime/runtime-worktree-ps-unverifiable-terminals.ts index 23ee5dcab07..ec3f9f93d89 100644 --- a/src/main/runtime/runtime-worktree-ps-unverifiable-terminals.ts +++ b/src/main/runtime/runtime-worktree-ps-unverifiable-terminals.ts @@ -1,7 +1,6 @@ -import type { RuntimeWorktreePsSummary } from '../../shared/runtime-types' import type { PtyLivenessVerdict } from '../../shared/pty-liveness-verdict' import type { RuntimeLeafRecord, RuntimePtyWorktreeRecord } from './runtime-terminal-state-records' -import type { RuntimeWorktreeSummaryPathIndex } from './runtime-worktree-summary-paths' +import type { RuntimeWorktreePsSummaryLookup } from './runtime-worktree-summary-paths' /** * Counts each terminal that was not counted live but whose host only lost contact. @@ -9,19 +8,11 @@ import type { RuntimeWorktreeSummaryPathIndex } from './runtime-worktree-summary * the row read as no terminals at all (docs/reference/ssh-execution-boundary.md). */ export function applyRuntimeWorktreePsUnverifiableTerminals(args: { - summaries: Map - pathIndex: RuntimeWorktreeSummaryPathIndex - missingIds: Set countedPtyIds: ReadonlySet leaves: Iterable ptysById: ReadonlyMap getLivenessVerdict: (ptyId: string) => PtyLivenessVerdict | null - getSummary: ( - summaries: Map, - pathIndex: RuntimeWorktreeSummaryPathIndex, - missingIds: Set, - worktreeId: string - ) => RuntimeWorktreePsSummary | null + getSummary: RuntimeWorktreePsSummaryLookup }): void { // Like the live pass, the host's PTY record owns the worktree; a pane only fills in a PTY with no record. const ownerByPtyId = new Map() @@ -40,7 +31,7 @@ export function applyRuntimeWorktreePsUnverifiableTerminals(args: { ) { continue } - const summary = args.getSummary(args.summaries, args.pathIndex, args.missingIds, worktreeId) + const summary = args.getSummary(worktreeId) if (summary) { summary.unverifiableTerminalCount = (summary.unverifiableTerminalCount ?? 0) + 1 summary.hasHostSidebarActivity = true diff --git a/src/main/runtime/runtime-worktree-structured-agent-rows-liveness.test.ts b/src/main/runtime/runtime-worktree-structured-agent-rows-liveness.test.ts index 5f6020e8d0f..b37091cf17b 100644 --- a/src/main/runtime/runtime-worktree-structured-agent-rows-liveness.test.ts +++ b/src/main/runtime/runtime-worktree-structured-agent-rows-liveness.test.ts @@ -130,10 +130,9 @@ function worktreeFor(store: AgentHookServer): RuntimeWorktreePsSummary { status: 'inactive', agents: [] } as unknown as RuntimeWorktreePsSummary + const summaries = new Map([[WORKTREE_ID, row]]) attachRuntimeWorktreeAgentRows({ - summaries: new Map([[WORKTREE_ID, row]]), - pathIndex: { byPath: new Map(), byRealPath: new Map() } as never, - missingWorktreeIds: new Set(), + summaries, workingTerminalEvidenceByWorktreeId: new Map(), rowSources: collectRuntimeWorktreeAgentSources({ mirroredWorktreeIdByTabId: new Map(), @@ -145,7 +144,7 @@ function worktreeFor(store: AgentHookServer): RuntimeWorktreePsSummary { hookSnapshots: store.getStatusSnapshot() }), orchestrationByPaneKey: null, - getSummary: (map, _paths, _missing, id) => map.get(id) ?? null + getSummary: (id) => summaries.get(id) ?? null }) return row } diff --git a/src/main/runtime/runtime-worktree-summary-paths.ts b/src/main/runtime/runtime-worktree-summary-paths.ts index 77ecae27e5c..99a304f6789 100644 --- a/src/main/runtime/runtime-worktree-summary-paths.ts +++ b/src/main/runtime/runtime-worktree-summary-paths.ts @@ -1,6 +1,7 @@ import type { RuntimeWorktreePsSummary } from '../../shared/runtime-types' import { isWindowsAbsolutePathLike } from '../../shared/cross-platform-path' import { worktreePathComparisonKey } from '../ipc/worktree-path-comparison' +import { parseRuntimeWorktreeId } from './runtime-worktree-path-identity' type ResolvedWorktreePath = { id: string; repoId: string; path: string } type RuntimeWorktreeSummaryPathCandidate = { summary: RuntimeWorktreePsSummary; order: number } @@ -55,6 +56,40 @@ export function buildRuntimeWorktreeSummaryPathIndex( return index } +export type RuntimeWorktreePsSummaryLookup = (worktreeId: string) => RuntimeWorktreePsSummary | null + +export function createRuntimeWorktreePsSummaryLookup( + summaries: ReadonlyMap, + pathIndex: RuntimeWorktreeSummaryPathIndex +): RuntimeWorktreePsSummaryLookup { + const missingIds = new Set() + return (worktreeId) => { + const exact = summaries.get(worktreeId) + if (exact) { + return exact + } + if (missingIds.has(worktreeId)) { + return null + } + const parsed = parseRuntimeWorktreeId(worktreeId) + if (!parsed) { + return null + } + const platform = pathIndex.platformByRepoId.get(parsed.repoId) ?? process.platform + const indexed = findRuntimeWorktreeSummaryByPath( + pathIndex, + parsed.repoId, + parsed.worktreePath, + platform + ) + if (indexed) { + return indexed + } + missingIds.add(worktreeId) + return null + } +} + export function findRuntimeWorktreeSummaryByPath( index: RuntimeWorktreeSummaryPathIndex, repoId: string, diff --git a/src/main/speech/speech-model-download-response.ts b/src/main/speech/speech-model-download-response.ts index 226f0053b2c..64275b586e6 100644 --- a/src/main/speech/speech-model-download-response.ts +++ b/src/main/speech/speech-model-download-response.ts @@ -73,21 +73,3 @@ export function describeInterruptedDownload( `Model download interrupted at ${received} after ${attempts} attempts: ${causeMessage}` ) } - -export function sleepUnlessAborted(ms: number, signal: AbortSignal): Promise { - return new Promise((resolve) => { - if (signal.aborted) { - resolve() - return - } - const onAbort = (): void => { - clearTimeout(timer) - resolve() - } - const timer = setTimeout(() => { - signal.removeEventListener('abort', onAbort) - resolve() - }, ms) - signal.addEventListener('abort', onAbort, { once: true }) - }) -} diff --git a/src/main/speech/speech-model-download-transport.ts b/src/main/speech/speech-model-download-transport.ts index 929158a8ba6..1d887a1295e 100644 --- a/src/main/speech/speech-model-download-transport.ts +++ b/src/main/speech/speech-model-download-transport.ts @@ -6,11 +6,11 @@ import { MAX_RETRY_AFTER_MS, MAX_TOTAL_DOWNLOAD_REQUESTS, describeInterruptedDownload, - sleepUnlessAborted, type DownloadTotals } from './speech-model-download-response' import { isRetryableDownloadError, type HttpStatusError } from '../network/transient-download-error' import { SpeechModelHttpDownload } from './speech-model-http-download' +import { delayUnlessAborted } from '../../shared/abort-aware-delay' export abstract class SpeechModelDownloadTransport extends SpeechModelHttpDownload { protected getPartialDownloadBytes(filePath: string): number { @@ -72,7 +72,7 @@ export abstract class SpeechModelDownloadTransport extends SpeechModelHttpDownlo `Model download exceeded its expected size (${receivedBytes} of ${totals.totalBytes} bytes)` ) } - const incompleteResponse = new Error( + const incompleteResponse: HttpStatusError = new Error( `Model download response ended at ${receivedBytes} of ${totals.totalBytes} bytes` ) if (receivedBytes > offset) { @@ -80,9 +80,8 @@ export abstract class SpeechModelDownloadTransport extends SpeechModelHttpDownlo noProgressStreak = 0 continue } - const retryableIncompleteResponse = incompleteResponse as HttpStatusError - retryableIncompleteResponse.retryable = true - throw retryableIncompleteResponse + incompleteResponse.retryable = true + throw incompleteResponse } catch (err) { if (isAborted() || signal.aborted) { throw err @@ -99,11 +98,11 @@ export abstract class SpeechModelDownloadTransport extends SpeechModelHttpDownlo if (noProgressStreak >= MAX_NO_PROGRESS_ATTEMPTS) { throw describeInterruptedDownload(err, receivedBytes, totals.totalBytes, requestCount) } - const retryAfterMs = (err as HttpStatusError).retryAfterMs + const statusError: HttpStatusError | undefined = err instanceof Error ? err : undefined + const retryAfterMs = statusError?.retryAfterMs if (retryAfterMs !== undefined && retryAfterMs > MAX_RETRY_AFTER_MS) { - const statusCode = (err as HttpStatusError).httpStatusCode throw new Error( - `HTTP ${statusCode}; server requested retry after ${Math.ceil(retryAfterMs / 1_000)} seconds` + `HTTP ${statusError?.httpStatusCode}; server requested retry after ${Math.ceil(retryAfterMs / 1_000)} seconds` ) } console.warn( @@ -111,7 +110,7 @@ export abstract class SpeechModelDownloadTransport extends SpeechModelHttpDownlo modelId, err ) - await sleepUnlessAborted( + await delayUnlessAborted( retryAfterMs ?? DOWNLOAD_RETRY_DELAYS_MS[ Math.min(Math.max(0, noProgressStreak - 1), DOWNLOAD_RETRY_DELAYS_MS.length - 1) diff --git a/src/main/terminal-scrollback-snapshots.ts b/src/main/terminal-scrollback-snapshots.ts index 352edb5ebb4..e7d1561f61b 100644 --- a/src/main/terminal-scrollback-snapshots.ts +++ b/src/main/terminal-scrollback-snapshots.ts @@ -18,6 +18,7 @@ import { TERMINAL_SCROLLBACK_REPLAY_BYTE_LIMIT, TERMINAL_SCROLLBACK_STORE_BYTE_LIMIT } from '../shared/terminal-scrollback-limits' +import { errorMessage } from '../shared/error-message' const SNAPSHOT_DIR_NAME = 'terminal-scrollback' const REF_PREFIX = 'v1' @@ -156,9 +157,7 @@ export function writeTerminalScrollbackSnapshotSync(args: { } return ref } catch (err) { - console.warn( - `[terminal-scrollback] Failed to write snapshot: ${err instanceof Error ? err.message : String(err)}` - ) + console.warn(`[terminal-scrollback] Failed to write snapshot: ${errorMessage(err)}`) return null } } @@ -188,9 +187,7 @@ export async function writeTerminalScrollbackSnapshot(args: { renamed = true return ref } catch (err) { - console.warn( - `[terminal-scrollback] Failed to write snapshot: ${err instanceof Error ? err.message : String(err)}` - ) + console.warn(`[terminal-scrollback] Failed to write snapshot: ${errorMessage(err)}`) return null } finally { if (!renamed) { diff --git a/src/relay/agent-exec-handler.ts b/src/relay/agent-exec-handler.ts index f35302f7c31..d54c2098be1 100644 --- a/src/relay/agent-exec-handler.ts +++ b/src/relay/agent-exec-handler.ts @@ -9,6 +9,7 @@ import { mergeGitConfigEnvProtocol } from '../shared/git-credential-prompt-env' import { terminateRelaySubprocessTree } from './subprocess-tree-termination' import { RelayAgentProcessLifetime } from './relay-agent-process-lifetime' import { resolveLoginShellEnvironment } from '../main/startup/login-shell-environment' +import { errorMessage } from '../shared/error-message' const DEFAULT_TIMEOUT_MS = 60_000 const MAX_TIMEOUT_MS = 5 * 60 * 1000 @@ -234,7 +235,7 @@ export class AgentExecHandler { stderr: '', exitCode: null, timedOut: false, - spawnError: error instanceof Error ? error.message : String(error) + spawnError: errorMessage(error) }) return } diff --git a/src/relay/ai-vault-transcript-stream.ts b/src/relay/ai-vault-transcript-stream.ts index 42e24733551..7f88a6e4dbe 100644 --- a/src/relay/ai-vault-transcript-stream.ts +++ b/src/relay/ai-vault-transcript-stream.ts @@ -1,7 +1,7 @@ import { open } from 'node:fs/promises' import { throwIfAiVaultScanCancelled } from '../main/ai-vault/ai-vault-scan-cancellation' import { BinarySessionTranscriptError } from '../main/ai-vault/remote-session-content-lines' -import { BINARY_PROBE_BYTES, isBinaryBuffer } from './fs-handler-utils' +import { BINARY_PROBE_BYTES, isBinaryBuffer } from '../shared/binary-buffer' import { readNodeFileWithinLimit } from '../shared/node-bounded-file-reader' /** The same open handle supplies the probe and stream, including across renames. */ diff --git a/src/relay/fs-handler-file-read.ts b/src/relay/fs-handler-file-read.ts index c07c897fc8b..e0f7fa590a9 100644 --- a/src/relay/fs-handler-file-read.ts +++ b/src/relay/fs-handler-file-read.ts @@ -4,15 +4,14 @@ import { extname } from 'node:path' import type { RelayDispatcher, RequestContext } from './dispatcher' import { STREAM_ACK_WINDOW_CHUNKS, STREAM_CHUNK_SIZE } from './protocol' import type { RelayStreamRegistry } from './fs-stream-registry' -import { reserveTerminalFrameSlot } from './fs-stream-terminal-frame-slots' +import { BINARY_PROBE_BYTES, isBinaryBuffer } from '../shared/binary-buffer' import { - BINARY_PROBE_BYTES, IMAGE_MIME_TYPES, MAX_PREVIEWABLE_BINARY_SIZE, MAX_TEXT_FILE_SIZE, - isBinaryBuffer, isBinaryFilePrefix } from './fs-handler-utils' +import { errorMessage as describeError } from '../shared/error-message' export async function readRelayFileContent(filePath: string) { const stats = await stat(filePath) @@ -124,7 +123,7 @@ async function prepareRelayFileStream( // Why: reserved before the fd opens so a refusal costs nothing, and released only // once the terminal frame settles — see reserveTerminalFrameSlot. - const releaseTerminalFrameSlot = reserveTerminalFrameSlot(registry, context.clientId) + const releaseTerminalFrameSlot = registry.reserveTerminalFrameSlot(context.clientId) let handle: FileHandle | undefined let streamId: number try { @@ -267,7 +266,7 @@ async function pumpChunks( } else { endReason = 'error' errorCode = code ?? 'ESTREAMREAD' - errorMessage = err instanceof Error ? err.message : String(err) + errorMessage = describeError(err) } } @@ -310,9 +309,7 @@ async function pumpChunks( process.stderr.write(`[relay] stream stale id=${streamId}\n`) } } catch (err) { - process.stderr.write( - `[relay] stream notify failed id=${streamId}: ${err instanceof Error ? err.message : String(err)}\n` - ) + process.stderr.write(`[relay] stream notify failed id=${streamId}: ${describeError(err)}\n`) } } finally { // Why: the fd goes back first — a terminal frame that can never be delivered must not diff --git a/src/relay/fs-handler-terminal-artifact.ts b/src/relay/fs-handler-terminal-artifact.ts index 02af753f134..91f17d3174c 100644 --- a/src/relay/fs-handler-terminal-artifact.ts +++ b/src/relay/fs-handler-terminal-artifact.ts @@ -4,9 +4,9 @@ import { chmod, open, realpath, rename, rm, writeFile } from 'node:fs/promises' import type { FileHandle } from 'node:fs/promises' import { basename, dirname, extname, join } from 'node:path' import { randomUUID } from 'node:crypto' +import { isBinaryBuffer } from '../shared/binary-buffer' import { IMAGE_MIME_TYPES, - isBinaryBuffer, MAX_PREVIEWABLE_BINARY_SIZE, MAX_TEXT_FILE_SIZE } from './fs-handler-utils' diff --git a/src/relay/fs-handler-utils.ts b/src/relay/fs-handler-utils.ts index 4a951eac9f0..2af88784b9a 100644 --- a/src/relay/fs-handler-utils.ts +++ b/src/relay/fs-handler-utils.ts @@ -18,6 +18,7 @@ import { SEARCH_TIMEOUT_MS as SHARED_SEARCH_TIMEOUT_MS } from '../shared/text-search' import { IMAGE_FILE_MIME_TYPES } from '../shared/image-file-extensions' +import { BINARY_PROBE_BYTES, isBinaryBuffer } from '../shared/binary-buffer' import type { SearchResult as SharedSearchResult } from '../shared/code-search-types' import { absorbPendingRipgrepSpawnError, @@ -44,7 +45,6 @@ export const MAX_TEXT_FILE_SIZE = 10 * 1024 * 1024 // Reads above the legacy 16MB single-frame budget go through fs.readFileStream, // which chunks at STREAM_CHUNK_SIZE; see docs/relay-file-stream-design.md. export const MAX_PREVIEWABLE_BINARY_SIZE = 50 * 1024 * 1024 -export const BINARY_PROBE_BYTES = 8192 export const SEARCH_TIMEOUT_MS = SHARED_SEARCH_TIMEOUT_MS export const DEFAULT_MAX_RESULTS = 2000 @@ -53,18 +53,6 @@ export const IMAGE_MIME_TYPES: Record = { '.pdf': 'application/pdf' } -// ─── Binary detection ──────────────────────────────────────────────── - -export function isBinaryBuffer(buffer: Buffer): boolean { - const len = Math.min(buffer.length, 8192) - for (let i = 0; i < len; i++) { - if (buffer[i] === 0) { - return true - } - } - return false -} - export async function isBinaryFilePrefix( filePath: string, releaseHandle: (handle: Awaited>) => Promise = (handle) => diff --git a/src/relay/fs-stream-registry.ts b/src/relay/fs-stream-registry.ts index 59c0ad19032..e0cc395df2d 100644 --- a/src/relay/fs-stream-registry.ts +++ b/src/relay/fs-stream-registry.ts @@ -1,16 +1,9 @@ import type { FileHandle } from 'node:fs/promises' import { settlesWithin } from './settles-within' -import { MAX_CONCURRENT_STREAMS, RelayErrorCode, STREAM_ACK_STALL_RECHECK_MS } from './protocol' +import { MAX_CONCURRENT_STREAMS, RelayErrorCode } from './protocol' +import { RelayStreamAckWindow } from './relay-stream-ack-window' -type StreamEntry = { - handle: FileHandle - aborted: boolean - /** Highest chunk seq the client acknowledged (in-order; -1 = none yet). */ - ackedThroughSeq: number - /** Pumps parked on the ack credit window. Woken by acks, abort, release, - * and a periodic stall recheck so a vanished client cannot strand a pump. */ - ackWaiters: Set<() => void> -} +type StreamEntry = { handle: FileHandle; ack: RelayStreamAckWindow } // Why: a stalled but connected client can hold a terminal-frame slot open indefinitely. const OPERATION_SETTLE_DEADLINE_MS = 10_000 @@ -29,6 +22,11 @@ export class RelayStreamRegistry { private closing = new Map>() private closeFailures = new Map() private operations = new Set>() + // Why: terminal frames ride the non-dropping control lane, which destroys the link at 256 frames / 1 MB. + // Holding a per-client slot until the frame settles caps queued terminal frames at + // MAX_CONCURRENT_STREAMS per client (the control queue is per client); each slot is a + // registry operation, so shutdown waits for undelivered terminal frames too. + private pendingTerminalFramesByClient = new Map() constructor(private readonly operationDeadlineMs = OPERATION_SETTLE_DEADLINE_MS) {} @@ -49,6 +47,31 @@ export class RelayStreamRegistry { } } + /** Holds a per-client slot until a stream's terminal frame settles; see comment. */ + reserveTerminalFrameSlot(clientId: number): () => void { + const pending = this.pendingTerminalFramesByClient.get(clientId) ?? 0 + if (pending >= MAX_CONCURRENT_STREAMS) { + throw new TooManyStreamsError() + } + const finish = this.beginOperation() + this.pendingTerminalFramesByClient.set(clientId, pending + 1) + let released = false + return () => { + if (released) { + return + } + released = true + finish() + const remaining = (this.pendingTerminalFramesByClient.get(clientId) ?? 1) - 1 + // Drop the entry at zero so a long-lived registry cannot accumulate one per detached client. + if (remaining <= 0) { + this.pendingTerminalFramesByClient.delete(clientId) + return + } + this.pendingTerminalFramesByClient.set(clientId, remaining) + } + } + register(handle: FileHandle): number { if (this.disposed) { throw new Error('relay_file_stream_shutdown_fenced') @@ -66,25 +89,16 @@ export class RelayStreamRegistry { private retainHandle(handle: FileHandle): number { const streamId = this.nextId++ - this.streams.set(streamId, { - handle, - aborted: false, - ackedThroughSeq: -1, - ackWaiters: new Set() - }) + this.streams.set(streamId, { handle, ack: new RelayStreamAckWindow() }) return streamId } abort(streamId: number): void { - const entry = this.streams.get(streamId) - if (entry) { - entry.aborted = true - this.wakeAckWaiters(entry) - } + this.streams.get(streamId)?.ack.abort() } isAborted(streamId: number): boolean { - return this.streams.get(streamId)?.aborted ?? true + return this.streams.get(streamId)?.ack.aborted ?? true } get(streamId: number): StreamEntry | undefined { @@ -92,55 +106,24 @@ export class RelayStreamRegistry { } recordAck(streamId: number, seq: number): void { - const entry = this.streams.get(streamId) - if (!entry || typeof seq !== 'number' || !Number.isFinite(seq)) { - return - } - if (seq > entry.ackedThroughSeq) { - entry.ackedThroughSeq = seq - } - this.wakeAckWaiters(entry) + this.streams.get(streamId)?.ack.recordAck(seq) } ackedThroughSeq(streamId: number): number { - return this.streams.get(streamId)?.ackedThroughSeq ?? Number.MAX_SAFE_INTEGER + return this.streams.get(streamId)?.ack.ackedThroughSeq ?? Number.MAX_SAFE_INTEGER } /** Resolves on the next ack/abort/release for this stream, or after the * stall-recheck interval so callers can re-evaluate staleness. */ waitForAck(streamId: number): Promise { - const entry = this.streams.get(streamId) - if (!entry || entry.aborted) { - return Promise.resolve() - } - return new Promise((resolve) => { - let settled = false - const timer = setTimeout(() => finish(), STREAM_ACK_STALL_RECHECK_MS) - timer.unref?.() - const finish = (): void => { - if (settled) { - return - } - settled = true - clearTimeout(timer) - entry.ackWaiters.delete(finish) - resolve() - } - entry.ackWaiters.add(finish) - }) + return this.streams.get(streamId)?.ack.wait() ?? Promise.resolve() } /** Wake every parked pump (all streams) so it re-checks staleness — used * when a client detaches and its acks will never arrive. */ wakeAllAckWaiters(): void { for (const entry of this.streams.values()) { - this.wakeAckWaiters(entry) - } - } - - private wakeAckWaiters(entry: StreamEntry): void { - for (const waiter of Array.from(entry.ackWaiters)) { - waiter() + entry.ack.wake() } } diff --git a/src/relay/fs-stream-terminal-frame-slots.ts b/src/relay/fs-stream-terminal-frame-slots.ts deleted file mode 100644 index 5ede13b5d79..00000000000 --- a/src/relay/fs-stream-terminal-frame-slots.ts +++ /dev/null @@ -1,51 +0,0 @@ -import { MAX_CONCURRENT_STREAMS } from './protocol' -import { TooManyStreamsError, type RelayStreamRegistry } from './fs-stream-registry' - -/** - * Terminal frames (fs.streamEnd/fs.streamError) ride the control lane, which does not - * drop on overflow — it destroys the link at 256 queued frames / 1 MB. A stream's - * registry slot is gone the moment its last chunk is read, so on a socket that is not - * draining, back-to-back reads could stack one undelivered terminal frame each until - * that budget blew. Holding the slot until the frame settles keeps the number of queued - * terminal frames at MAX_CONCURRENT_STREAMS, far below the killing threshold; the - * overflow now costs one refused read (TooManyStreams, which clients already handle) - * instead of the whole connection. - * - * Counted per client, because the control queue this protects is per client: one peer whose - * socket stopped draining must not refuse reads for every other peer on the same relay. - * Counted independently of released file descriptors, and each slot is a registry operation - * so shutdown waits for undelivered terminal frames too. - */ -const pendingTerminalFramesByClient = new WeakMap>() - -export function reserveTerminalFrameSlot( - registry: RelayStreamRegistry, - clientId: number -): () => void { - let byClient = pendingTerminalFramesByClient.get(registry) - if (!byClient) { - byClient = new Map() - pendingTerminalFramesByClient.set(registry, byClient) - } - const pending = byClient.get(clientId) ?? 0 - if (pending >= MAX_CONCURRENT_STREAMS) { - throw new TooManyStreamsError() - } - const finish = registry.beginOperation() - byClient.set(clientId, pending + 1) - let released = false - return () => { - if (released) { - return - } - released = true - finish() - const remaining = (byClient.get(clientId) ?? 1) - 1 - // Drop the entry at zero so a long-lived registry cannot accumulate one per detached client. - if (remaining <= 0) { - byClient.delete(clientId) - return - } - byClient.set(clientId, remaining) - } -} diff --git a/src/relay/git-response-stream.ts b/src/relay/git-response-stream.ts index 7097b3b2d8d..254c140478a 100644 --- a/src/relay/git-response-stream.ts +++ b/src/relay/git-response-stream.ts @@ -1,7 +1,7 @@ // Streams large RPC responses onto the bulk lane in chunks instead of one // JSON-RPC frame, so a big reply cannot head-of-line-block interactive pty.data -// echo on the shared SSH channel. Mirrors the fs read-stream credit-window -// pattern (see fs-handler-file-read.ts) but the payload is an in-memory +// echo on the shared SSH channel. Shares RelayStreamAckWindow with the fs +// read stream (see fs-handler-file-read.ts) but the payload is an in-memory // serialized string rather than a file handle. // // ONE REGISTRY PER RELAY. The `git.*` method names below are the shipped wire @@ -18,18 +18,13 @@ import { GIT_RESPONSE_CHUNK_SIZE, GIT_RESPONSE_STREAM_THRESHOLD, STREAM_ACK_WINDOW_CHUNKS, - STREAM_ACK_STALL_RECHECK_MS, type GitResponseStreamMarker } from './protocol' import { settlesWithin } from './settles-within' +import { RelayStreamAckWindow } from './relay-stream-ack-window' +import { errorMessage } from '../shared/error-message' -type GitResponseStreamEntry = { - ownerClientId: number - aborted: boolean - /** Highest chunk seq the client acknowledged (in-order; -1 = none yet). */ - ackedThroughSeq: number - ackWaiters: Set<() => void> -} +type GitResponseStreamEntry = { ownerClientId: number; ack: RelayStreamAckWindow } /** Serialized git responses are chunked as base64 so multi-byte UTF-8 * sequences never split across a chunk boundary (the client concatenates the @@ -58,36 +53,22 @@ export class GitResponseStreamRegistry { throw new Error('relay_response_stream_shutdown_fenced') } const streamId = this.nextId++ - this.streams.set(streamId, { - ownerClientId, - aborted: false, - ackedThroughSeq: -1, - ackWaiters: new Set() - }) + this.streams.set(streamId, { ownerClientId, ack: new RelayStreamAckWindow() }) return streamId } recordAck(streamId: number, seq: number, clientId: number): void { const entry = this.streams.get(streamId) - if ( - !entry || - entry.ownerClientId !== clientId || - typeof seq !== 'number' || - !Number.isFinite(seq) - ) { + if (!entry || entry.ownerClientId !== clientId) { return } - if (seq > entry.ackedThroughSeq) { - entry.ackedThroughSeq = seq - } - this.wake(entry) + entry.ack.recordAck(seq) } abort(streamId: number, clientId: number): void { const entry = this.streams.get(streamId) if (entry?.ownerClientId === clientId) { - entry.aborted = true - this.wake(entry) + entry.ack.abort() } } @@ -95,38 +76,10 @@ export class GitResponseStreamRegistry { * detaches and its acks will never arrive. */ wakeAll(): void { for (const entry of this.streams.values()) { - this.wake(entry) + entry.ack.wake() } } - private wake(entry: GitResponseStreamEntry): void { - for (const waiter of Array.from(entry.ackWaiters)) { - waiter() - } - } - - private waitForAck(streamId: number): Promise { - const entry = this.streams.get(streamId) - if (!entry || entry.aborted) { - return Promise.resolve() - } - return new Promise((resolve) => { - let settled = false - const finish = (): void => { - if (settled) { - return - } - settled = true - clearTimeout(timer) - entry.ackWaiters.delete(finish) - resolve() - } - const timer = setTimeout(finish, STREAM_ACK_STALL_RECHECK_MS) - timer.unref?.() - entry.ackWaiters.add(finish) - }) - } - /** * Register a stream for `payload`, kick off the bulk-lane pump on a later * task (so the sentinel response reaches the client first), and return the @@ -188,24 +141,24 @@ export class GitResponseStreamRegistry { endReason = 'stale' break } - if (entry.aborted) { + if (entry.ack.aborted) { endReason = 'aborted' break } // Why: credit window — the client acks each chunk, bounding how many // bulk bytes a keystroke echo can queue behind on the shared channel. while ( - seq - entry.ackedThroughSeq > STREAM_ACK_WINDOW_CHUNKS && + seq - entry.ack.ackedThroughSeq > STREAM_ACK_WINDOW_CHUNKS && !context.isStale() && - !entry.aborted + !entry.ack.aborted ) { - await this.waitForAck(streamId) + await entry.ack.wait() } if (context.isStale()) { endReason = 'stale' break } - if (entry.aborted) { + if (entry.ack.aborted) { endReason = 'aborted' break } @@ -222,17 +175,17 @@ export class GitResponseStreamRegistry { chunks[seq] = '' } // Why: disposal may abort while the final chunk write is in flight. - if (endReason === 'end' && !entry.aborted && !context.isStale()) { + if (endReason === 'end' && !entry.ack.aborted && !context.isStale()) { await dispatcher.notifyBulk('git.responseEnd', { streamId }, { clientId }) } } catch (err) { - if (!context.isStale() && !entry.aborted) { + if (!context.isStale() && !entry.ack.aborted) { try { await dispatcher.notifyBulk( 'git.responseError', { streamId, - message: err instanceof Error ? err.message : String(err) + message: errorMessage(err) }, { clientId } ) @@ -250,8 +203,7 @@ export class GitResponseStreamRegistry { disposeAll(): void { this.disposed = true for (const entry of this.streams.values()) { - entry.aborted = true - this.wake(entry) + entry.ack.abort() } this.streams.clear() } diff --git a/src/relay/relay-daemon.ts b/src/relay/relay-daemon.ts index daca737a3df..31eb57b9f98 100644 --- a/src/relay/relay-daemon.ts +++ b/src/relay/relay-daemon.ts @@ -15,6 +15,7 @@ import { } from './relay-endpoint-credential-publication' import { SKILL_RELAY_CAPABILITIES } from './skill-install-handler' import { publishRelayPid } from './relay-pid-publication' +import { errorMessage } from '../shared/error-message' export async function runRelayDaemon(options: RelayLaunchOptions): Promise { if (options.detached && options.logFile) { @@ -30,9 +31,7 @@ export async function runRelayDaemon(options: RelayLaunchOptions): Promise } catch (reapError) { // Why log rather than swallow: exit must still win, but this line is the only // forensic trace a crashed remote daemon leaves behind for an orphaned shell. - relayLogLine( - `[relay] Fatal PTY reap failed: ${reapError instanceof Error ? reapError.message : String(reapError)}` - ) + relayLogLine(`[relay] Fatal PTY reap failed: ${errorMessage(reapError)}`) } socketOwnership.cleanup() process.exit(1) @@ -112,9 +111,7 @@ export async function runRelayDaemon(options: RelayLaunchOptions): Promise reconnectListener.setEndpointCredential(publishRelayEndpointCredential(options.credentialFile)) agentHooks.publishEndpointFile() } catch (error) { - relayLogLine( - `[relay] Startup failed: ${error instanceof Error ? error.message : String(error)}` - ) + relayLogLine(`[relay] Startup failed: ${errorMessage(error)}`) process.exit(1) return } diff --git a/src/relay/relay-grace-lifecycle.ts b/src/relay/relay-grace-lifecycle.ts index 653e2a44eaa..8415b71793c 100644 --- a/src/relay/relay-grace-lifecycle.ts +++ b/src/relay/relay-grace-lifecycle.ts @@ -7,6 +7,7 @@ import { import { relayLogLine } from './relay-diagnostic-log' import { SSH_RELAY_CONFIGURE_GRACE_TIME_METHOD } from '../shared/ssh-types' import type { RelayDispatcher } from './dispatcher' +import { errorMessage } from '../shared/error-message' type RelayGraceLifecycleOptions = { dispatcher: RelayDispatcher @@ -139,9 +140,7 @@ export class RelayGraceLifecycle { .catch((error) => { this.shutdownInFlight = false this.options.reopenOwnedProcesses() - relayLogLine( - `[relay] Shutdown deferred: ${error instanceof Error ? error.message : String(error)}` - ) + relayLogLine(`[relay] Shutdown deferred: ${errorMessage(error)}`) if (this.options.readSocketClientCount() === 0) { this.start('shutdown deferred', { retryDeferredShutdown: true }) } diff --git a/src/relay/relay-runtime-services.ts b/src/relay/relay-runtime-services.ts index 1325e792d07..e19e2e20973 100644 --- a/src/relay/relay-runtime-services.ts +++ b/src/relay/relay-runtime-services.ts @@ -24,6 +24,7 @@ import { RelayPtySourcePublication } from './relay-pty-source-publication' import { SkillInstallHandler } from './skill-install-handler' import { relayLogLine } from './relay-diagnostic-log' import { remoteCliRequestTimeoutMs } from './remote-cli-timeout' +import { errorMessage } from '../shared/error-message' export class RelayRuntimeServices { readonly ptyHandler: PtyHandler @@ -36,7 +37,6 @@ export class RelayRuntimeServices { private readonly responseStreams: GitResponseStreamRegistry private readonly aiVaultService: ReturnType | null private readonly sessionSearch: { dispose(): void } | null - private readonly registeredHandlers: readonly unknown[] constructor( readonly dispatcher: RelayDispatcher, @@ -76,12 +76,12 @@ export class RelayRuntimeServices { this.ptyHandler.shutdownForWorktreePath(rootPath) ) this.gitHandler = new GitHandler(dispatcher, context, watchRegistry, responseStreams) - const preflightHandler = new PreflightHandler(dispatcher) + new PreflightHandler(dispatcher) this.skillInstallHandler = new SkillInstallHandler(dispatcher) - const externalAutomationsHandler = new ExternalAutomationsHandler(dispatcher) - const portScanHandler = new PortScanHandler(dispatcher) + new ExternalAutomationsHandler(dispatcher) + new PortScanHandler(dispatcher) this.agentExecHandler = new AgentExecHandler(dispatcher) - const workspaceSessionHandler = new WorkspaceSessionHandler(dispatcher) + new WorkspaceSessionHandler(dispatcher) const relayPlatform = parseUnameToRelayPlatform(process.platform, process.arch) const hostPlatform = relayPlatform ? getRemoteHostPlatform(relayPlatform) : undefined this.aiVaultService = hostPlatform ? createRelayAiVaultService(homedir(), hostPlatform) : null @@ -96,23 +96,9 @@ export class RelayRuntimeServices { dataRoot: join(homedir(), RELAY_REMOTE_DIR), roots: { executionHostId: LOCAL_EXECUTION_HOST_ID }, settings: DEFAULT_AI_VAULT_SEARCH_SETTINGS, - onError: (error) => - relayLogLine( - `[relay] session search: ${error instanceof Error ? error.message : String(error)}` - ) + onError: (error) => relayLogLine(`[relay] session search: ${errorMessage(error)}`) }) - this.registeredHandlers = [ - preflightHandler, - this.skillInstallHandler, - externalAutomationsHandler, - portScanHandler, - this.agentExecHandler, - workspaceSessionHandler, - new AiVaultHandler(dispatcher, { - hostPlatform, - service: this.aiVaultService ?? undefined - }) - ] + new AiVaultHandler(dispatcher, { hostPlatform, service: this.aiVaultService ?? undefined }) registerRelayPluginHostCallHandlers( dispatcher, @@ -146,14 +132,10 @@ export class RelayRuntimeServices { */ async disposeExitOnlyServices(): Promise { await this.skillInstallHandler.dispose().catch((error) => { - relayLogLine( - `[relay] Skill upload cleanup failed: ${error instanceof Error ? error.message : String(error)}` - ) + relayLogLine(`[relay] Skill upload cleanup failed: ${errorMessage(error)}`) }) await this.aiVaultService?.dispose().catch((error) => { - relayLogLine( - `[relay] AI Vault sidecar shutdown failed: ${error instanceof Error ? error.message : String(error)}` - ) + relayLogLine(`[relay] AI Vault sidecar shutdown failed: ${errorMessage(error)}`) }) } @@ -167,26 +149,25 @@ export class RelayRuntimeServices { this.sessionSearch?.dispose() this.fsHandler.dispose() this.gitHandler.dispose() - void this.registeredHandlers } private registerSessionHandlers(context: RelayContext): void { - this.dispatcher.onNotification('session.registerRoot', (params) => { - const rootPath = params.rootPath as string - if (rootPath) { - context.registerRoot(rootPath) + const registerRoot = (params: Record): void => { + if (typeof params.rootPath === 'string' && params.rootPath) { + context.registerRoot(params.rootPath) } - }) + } + this.dispatcher.onNotification('session.registerRoot', (params) => registerRoot(params)) this.dispatcher.onRequest('session.registerRoot', async (params) => { - const rootPath = params.rootPath as string - if (rootPath) { - context.registerRoot(rootPath) - } + registerRoot(params) return { ok: true } }) - this.dispatcher.onRequest('session.resolveHome', async (params) => ({ - resolvedPath: expandTilde(params.path as string) - })) + this.dispatcher.onRequest('session.resolveHome', async (params) => { + if (typeof params.path !== 'string') { + throw new Error('session.resolveHome: path must be a string') + } + return { resolvedPath: expandTilde(params.path) } + }) } private registerRemoteCliRoutes(): void { diff --git a/src/relay/relay-stream-ack-window.ts b/src/relay/relay-stream-ack-window.ts new file mode 100644 index 00000000000..de4b1aecda4 --- /dev/null +++ b/src/relay/relay-stream-ack-window.ts @@ -0,0 +1,51 @@ +import { STREAM_ACK_STALL_RECHECK_MS } from './protocol' + +/** Waiters wake on ack, abort, or a stall recheck so a vanished client cannot strand a pump. */ +export class RelayStreamAckWindow { + aborted = false + /** Highest chunk seq the client acknowledged (in-order; -1 = none yet). */ + ackedThroughSeq = -1 + private readonly waiters = new Set<() => void>() + + recordAck(seq: number): void { + if (typeof seq !== 'number' || !Number.isFinite(seq)) { + return + } + if (seq > this.ackedThroughSeq) { + this.ackedThroughSeq = seq + } + this.wake() + } + + abort(): void { + this.aborted = true + this.wake() + } + + wake(): void { + for (const waiter of Array.from(this.waiters)) { + waiter() + } + } + + wait(): Promise { + if (this.aborted) { + return Promise.resolve() + } + return new Promise((resolve) => { + let settled = false + const finish = (): void => { + if (settled) { + return + } + settled = true + clearTimeout(timer) + this.waiters.delete(finish) + resolve() + } + const timer = setTimeout(finish, STREAM_ACK_STALL_RECHECK_MS) + timer.unref?.() + this.waiters.add(finish) + }) + } +} diff --git a/src/relay/relay-watcher-setup-wait.ts b/src/relay/relay-watcher-setup-wait.ts index f09dfb7470b..8502b228cbe 100644 --- a/src/relay/relay-watcher-setup-wait.ts +++ b/src/relay/relay-watcher-setup-wait.ts @@ -2,6 +2,7 @@ import { isWatcherProcessFailure } from '../main/ipc/parcel-watcher-process-fail import type { PromiseSettlementWaiters } from '../shared/promise-settlement-waiters' import type { RequestContext } from './dispatcher' import type { RelayWatcherTeardownState } from './relay-watcher-teardown-tracker' +import { errorMessage } from '../shared/error-message' export async function startInitialRelayWatch( state: RelayWatcherTeardownState, @@ -42,7 +43,7 @@ export async function awaitRelayWatcherSetupForClient( if (expectedAbort) { return } - const message = error instanceof Error ? error.message : String(error) + const message = errorMessage(error) process.stderr.write(`[relay] File watcher not available for ${state.rootPath}: ${message}\n`) throw error } diff --git a/src/shared/abort-aware-delay.ts b/src/shared/abort-aware-delay.ts new file mode 100644 index 00000000000..8bf55d597e3 --- /dev/null +++ b/src/shared/abort-aware-delay.ts @@ -0,0 +1,16 @@ +export function delayUnlessAborted(ms: number, signal: AbortSignal): Promise { + if (signal.aborted) { + return Promise.resolve() + } + return new Promise((resolve) => { + const onAbort = (): void => { + clearTimeout(timer) + resolve() + } + const timer = setTimeout(() => { + signal.removeEventListener('abort', onAbort) + resolve() + }, ms) + signal.addEventListener('abort', onAbort, { once: true }) + }) +} diff --git a/src/shared/binary-buffer.ts b/src/shared/binary-buffer.ts index f294d260367..efd410df3b4 100644 --- a/src/shared/binary-buffer.ts +++ b/src/shared/binary-buffer.ts @@ -1,8 +1,8 @@ // A NUL byte in the first chunk is git's own heuristic for "this is binary". -const BINARY_SNIFF_BYTES = 8192 +export const BINARY_PROBE_BYTES = 8192 export function isBinaryBuffer(buffer: Buffer): boolean { - const len = Math.min(buffer.length, BINARY_SNIFF_BYTES) + const len = Math.min(buffer.length, BINARY_PROBE_BYTES) for (let i = 0; i < len; i += 1) { if (buffer[i] === 0) { return true diff --git a/tests/e2e/cross-version-wire/cross-version-stopping.unit.test.ts b/tests/e2e/cross-version-wire/cross-version-stopping.unit.test.ts index 8fdfd9033de..855aa880695 100644 --- a/tests/e2e/cross-version-wire/cross-version-stopping.unit.test.ts +++ b/tests/e2e/cross-version-wire/cross-version-stopping.unit.test.ts @@ -79,8 +79,9 @@ function publishRow(host: HostRowModules, mainAgent: Record): R stateStartedAt: NOW - 60_000, mainAgent } + const summaries = new Map([[WORKTREE_ID, summary]]) host.attachRuntimeWorktreeAgentRows({ - summaries: new Map([[WORKTREE_ID, summary]]), + summaries, pathIndex: { byPath: new Map(), byRealPath: new Map() }, missingWorktreeIds: new Set(), workingTerminalEvidenceByWorktreeId: new Map(), @@ -94,8 +95,8 @@ function publishRow(host: HostRowModules, mainAgent: Record): R } }), orchestrationByPaneKey: null, - getSummary: (map: Map, _paths: unknown, _missing: unknown, id: string) => - map.get(id) ?? null + // Old builds pass (summaries, pathIndex, missingIds, id); newer ones pass (id). The id is last in both. + getSummary: (...args: unknown[]) => summaries.get(String(args.at(-1))) ?? null }) // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: JSON.parse of the rows the host just attached, which are plain objects. const [row] = JSON.parse(JSON.stringify(summary.agents)) as Row[] diff --git a/tests/e2e/cross-version-wire/cross-version-worktree-ps-verdict.unit.test.ts b/tests/e2e/cross-version-wire/cross-version-worktree-ps-verdict.unit.test.ts index ac53bc112b9..54f54051673 100644 --- a/tests/e2e/cross-version-wire/cross-version-worktree-ps-verdict.unit.test.ts +++ b/tests/e2e/cross-version-wire/cross-version-worktree-ps-verdict.unit.test.ts @@ -97,8 +97,9 @@ function publishRows(host: HostRowModules): Record { hasHostSidebarActivity: false, agents: [] } + const summaries = new Map([[WORKTREE_ID, summary]]) host.attachRuntimeWorktreeAgentRows({ - summaries: new Map([[WORKTREE_ID, summary]]), + summaries, pathIndex: { byPath: new Map(), byRealPath: new Map() }, missingWorktreeIds: new Set(), workingTerminalEvidenceByWorktreeId: new Map(), @@ -112,8 +113,8 @@ function publishRows(host: HostRowModules): Record { } }), orchestrationByPaneKey: null, - getSummary: (map: Map, _paths: unknown, _missing: unknown, id: string) => - map.get(id) ?? null + // Old builds pass (summaries, pathIndex, missingIds, id); newer ones pass (id). The id is last in both. + getSummary: (...args: unknown[]) => summaries.get(String(args.at(-1))) ?? null }) // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: JSON.parse of the rows the host just attached, which are plain objects keyed by paneKey. const rows = JSON.parse(JSON.stringify(summary.agents)) as AgentRow[]