fix(relay): stream large git responses on the bulk lane

Large git diff/exec responses (up to MAX_GIT_BUFFER=10MB) were sent as one
JSON-RPC frame on the shared SSH channel. encodeJsonRpcFrame's synchronous
JSON.stringify + Buffer.from of the whole payload, plus the multi-MB write,
head-of-line-blocked interactive pty.data echo: opening a big diff (e.g. a
lockfile) over SSH while typing in another pane stalled the echo for seconds.

The 2026-07 fs.readFileStream fix added a per-client serialized bulk lane
(RelayDispatcher.notifyBulk) honoring write()===false + waitWriteDrain, plus an
ack credit window — but only fs streams used it; git responses never did.

Stream the diff family (git.diff/branchDiff/commitDiff) and git.exec through the
bulk lane in chunks with the same ack-window backpressure. The relay returns a
small sentinel result and pumps git.responseChunk frames; the client reassembles
and JSON.parses. Small responses stay single-frame.

Cross-version safe via an opt-in capability gate (no handshake change):
- client sends __streamResponse:true; old relays ignore it → single-frame.
- new relays only stream when the flag is present → old clients get single-frame.
- new client checks for the stream sentinel; a plain result (old relay) is
  returned directly.

STREAM_CHUNK_SIZE is untouched; git chunking uses its own GIT_RESPONSE_CHUNK_SIZE
and the client reassembles by concatenation, so chunk size is not baked into any
cross-version offset math.

Co-authored-by: Orca <help@stably.ai>

src/main/providers/ssh-git-provider.test.ts

@@ -7,6 +7,7 @@ type MockMultiplexer = {
   notify: ReturnType<typeof vi.fn>
   onNotification: ReturnType<typeof vi.fn>
   onNotificationByMethod: ReturnType<typeof vi.fn>
+  onDispose: ReturnType<typeof vi.fn>
   dispose: ReturnType<typeof vi.fn>
   isDisposed: ReturnType<typeof vi.fn>
 }
@@ -17,6 +18,9 @@ function createMockMux(): MockMultiplexer {
     notify: vi.fn(),
     onNotification: vi.fn(),
     onNotificationByMethod: vi.fn().mockReturnValue(vi.fn()),
+    // Why: requestGitStreamable subscribes to onDispose before awaiting the
+    // response so it can reject in-flight reassembly if the link drops.
+    onDispose: vi.fn().mockReturnValue(vi.fn()),
     dispose: vi.fn(),
     isDisposed: vi.fn().mockReturnValue(false)
   }
@@ -320,7 +324,10 @@ describe('SshGitProvider', () => {
       'git.exec',
       {
         args: ['clone', '--progress', '--', 'git@example.com:repo.git', 'repo'],
-        cwd: '/home/user'
+        cwd: '/home/user',
+        // Why: exec opts into response streaming so a large stdout is chunked
+        // onto the bulk lane; old relays ignore the flag.
+        __streamResponse: true
       },
       {
         signal: controller.signal,
@@ -353,7 +360,8 @@ describe('SshGitProvider', () => {
     })
     expect(mux.request).toHaveBeenCalledWith('git.exec', {
       args: ['diff', '--cached', '--patch', '--minimal', '--no-color', '--no-ext-diff'],
-      cwd: '/home/user/repo'
+      cwd: '/home/user/repo',
+      __streamResponse: true
     })
   })
 
@@ -636,7 +644,10 @@ describe('SshGitProvider', () => {
     expect(mux.request).toHaveBeenCalledWith('git.diff', {
       worktreePath: '/home/user/repo',
       filePath: 'src/index.ts',
-      staged: true
+      staged: true,
+      // Why: opts into response streaming; a small result still comes back as a
+      // single frame (relay decides), and old relays ignore the flag.
+      __streamResponse: true
     })
     expect(result).toEqual(diffResult)
   })
@@ -880,7 +891,8 @@ describe('SshGitProvider', () => {
     const result = await provider.getBranchDiff('/home/user/repo', 'main')
     expect(mux.request).toHaveBeenCalledWith('git.branchDiff', {
       worktreePath: '/home/user/repo',
-      baseRef: 'main'
+      baseRef: 'main',
+      __streamResponse: true
     })
     expect(result).toEqual(diffs)
   })

src/main/providers/ssh-git-provider.ts

@@ -21,6 +21,7 @@ import type {
 import type { GitHistoryOptions, GitHistoryResult } from '../../shared/git-history'
 import { buildHostedRemoteCommitUrl, buildHostedRemoteFileUrl } from '../git/hosted-remote-url'
 import { JsonRpcErrorCode } from '../ssh/relay-protocol'
+import { requestGitStreamable } from '../ssh/ssh-git-response-stream-reader'
 import type { CommitMessageDraftContext } from '../../shared/commit-message-generation'
 import type { CommitMessagePlan } from '../../shared/commit-message-plan'
 import type { RemoteCommitMessageExecResult } from '../text-generation/commit-message-text-generation'
@@ -373,7 +374,7 @@ export class SshGitProvider implements IGitProvider {
     return this.gitDiffReadDedupe.run(
       stableInFlightKey(['diff', worktreePath, filePath, staged, compareAgainstHead]),
       async () =>
-        (await this.mux.request('git.diff', {
+        (await requestGitStreamable(this.mux, 'git.diff', {
           worktreePath,
           filePath,
           staged,
@@ -597,7 +598,7 @@ export class SshGitProvider implements IGitProvider {
         keyOptions.oldPath ?? null
       ]),
       async () =>
-        (await this.mux.request('git.branchDiff', {
+        (await requestGitStreamable(this.mux, 'git.branchDiff', {
           worktreePath,
           baseRef,
           ...options
@@ -619,7 +620,7 @@ export class SshGitProvider implements IGitProvider {
         args.oldPath ?? null
       ]),
       async () =>
-        (await this.mux.request('git.commitDiff', {
+        (await requestGitStreamable(this.mux, 'git.commitDiff', {
           worktreePath,
           ...args
         })) as GitDiffResult
@@ -761,8 +762,8 @@ export class SshGitProvider implements IGitProvider {
     options?: { signal?: AbortSignal; timeoutMs?: number }
   ): Promise<{ stdout: string; stderr: string }> {
     const result = options
-      ? await this.mux.request('git.exec', { args, cwd }, options)
-      : await this.mux.request('git.exec', { args, cwd })
+      ? await requestGitStreamable(this.mux, 'git.exec', { args, cwd }, options)
+      : await requestGitStreamable(this.mux, 'git.exec', { args, cwd })
     return result as {
       stdout: string
       stderr: string

src/main/ssh/relay-protocol.ts

@@ -55,6 +55,35 @@ export const STREAM_CHUNK_SIZE = 256 * 1024
  * client. */
 export const MAX_CONCURRENT_STREAMS = 16
 
+// ── Git response streaming (see docs/relay-git-response-stream-design.md) ──
+
+/** Serialized-JSON size above which the relay chunks a streamable git response
+ * (diff family + exec) onto the bulk lane instead of one JSON-RPC frame. Mirror
+ * of the relay-side constant; the client only opts in — the relay owns the
+ * decision — so this is documentation of the shared contract. */
+export const GIT_RESPONSE_STREAM_THRESHOLD = 256 * 1024
+
+/** Per-chunk size (serialized-result UTF-8 bytes) for git response streaming.
+ * The client reassembles by concatenation and does not depend on this value,
+ * so it stays cross-version safe. */
+export const GIT_RESPONSE_CHUNK_SIZE = 128 * 1024
+
+/** Sentinel the relay returns as the RPC result when the real payload streams
+ * as git.responseChunk frames. Absent from old relays, so a new client falls
+ * back to the plain result they return. */
+export type GitResponseStreamMarker = {
+  __orcaGitResponseStream: { streamId: number; totalBytes: number; chunkCount: number }
+}
+
+export function isGitResponseStreamMarker(value: unknown): value is GitResponseStreamMarker {
+  return (
+    typeof value === 'object' &&
+    value !== null &&
+    typeof (value as GitResponseStreamMarker).__orcaGitResponseStream === 'object' &&
+    (value as GitResponseStreamMarker).__orcaGitResponseStream !== null
+  )
+}
+
 // ── JSON-RPC types ──────────────────────────────────────────────────
 
 export type JsonRpcRequest = {

src/main/ssh/ssh-git-response-stream-reader.ts

@@ -0,0 +1,240 @@
+import type { SshChannelMultiplexer } from './ssh-channel-multiplexer'
+import { RelayErrorCode, isGitResponseStreamMarker } from './relay-protocol'
+
+const SENTINEL_STREAM_ID = -1
+
+export class GitResponseStreamError extends Error {
+  readonly code = RelayErrorCode.StreamProtocolError
+  constructor(message: string) {
+    super(message)
+  }
+}
+
+type PendingFrame =
+  | { kind: 'chunk'; params: Record<string, unknown> }
+  | { kind: 'end'; params: Record<string, unknown> }
+  | { kind: 'error'; params: Record<string, unknown> }
+
+/**
+ * Request a git method that may return a large payload, opting into response
+ * streaming so a big diff/exec response is chunked onto the relay's bulk lane
+ * instead of one JSON-RPC frame (which would head-of-line-block pty.data echo
+ * on the shared SSH channel).
+ *
+ * Cross-version behavior:
+ * - New relay + big result → returns the stream sentinel; we reassemble chunks.
+ * - New relay + small result, or old client → plain single-frame result.
+ * - Old relay (ignores `__streamResponse`) → returns the plain result; the
+ *   marker check fails and we return it directly, i.e. today's behavior.
+ */
+export function requestGitStreamable(
+  mux: SshChannelMultiplexer,
+  method: string,
+  params: Record<string, unknown>,
+  options?: { signal?: AbortSignal }
+): Promise<unknown> {
+  // Why: subscribe to chunk/end/error BEFORE awaiting the sentinel response so a
+  // chunk that lands in the same dispatch tick as the response is not dropped
+  // (mirrors readFileViaStream). streamIdRef stays SENTINEL until the sentinel
+  // resolves; frames are queued until then and drained.
+  const streamIdRef = { current: SENTINEL_STREAM_ID }
+  const unsubscribers: (() => void)[] = []
+  const cleanup = (): void => {
+    while (unsubscribers.length > 0) {
+      try {
+        unsubscribers.pop()?.()
+      } catch {
+        // best-effort
+      }
+    }
+  }
+
+  return new Promise<unknown>((resolve, reject) => {
+    const parts: Buffer[] = []
+    let expectedSeq = 0
+    let receivedBytes = 0
+    let totalBytes = 0
+    let chunkCount = 0
+    let settled = false
+    let metadataReady = false
+    const pending: PendingFrame[] = []
+
+    const cancel = (): void => {
+      if (streamIdRef.current !== SENTINEL_STREAM_ID && !mux.isDisposed()) {
+        try {
+          mux.notify('git.cancelResponseStream', { streamId: streamIdRef.current })
+        } catch {
+          // best-effort
+        }
+      }
+    }
+    const fail = (err: Error): void => {
+      if (settled) {
+        return
+      }
+      settled = true
+      cancel()
+      cleanup()
+      reject(err)
+    }
+    const succeed = (value: unknown): void => {
+      if (settled) {
+        return
+      }
+      settled = true
+      cleanup()
+      resolve(value)
+    }
+
+    const handleChunk = (p: Record<string, unknown>): void => {
+      if (settled || p.streamId !== streamIdRef.current) {
+        return
+      }
+      const seq = p.seq as number
+      const data = p.data as string
+      if (typeof seq !== 'number' || typeof data !== 'string') {
+        fail(new GitResponseStreamError(`Malformed chunk for git stream ${streamIdRef.current}`))
+        return
+      }
+      if (seq !== expectedSeq) {
+        fail(
+          new GitResponseStreamError(
+            `Out-of-order chunk for git stream ${streamIdRef.current}: expected ${expectedSeq}, got ${seq}`
+          )
+        )
+        return
+      }
+      const decoded = Buffer.from(data, 'base64')
+      parts.push(decoded)
+      receivedBytes += decoded.length
+      expectedSeq += 1
+      // Why: credit-based flow control — the relay caps unacked chunks so a big
+      // response cannot queue unbounded ahead of interactive pty.data frames.
+      mux.notify('git.responseAck', { streamId: streamIdRef.current, seq })
+    }
+
+    const handleEnd = (p: Record<string, unknown>): void => {
+      if (settled || p.streamId !== streamIdRef.current) {
+        return
+      }
+      if (expectedSeq !== chunkCount || receivedBytes !== totalBytes) {
+        fail(
+          new GitResponseStreamError(
+            `Git stream ${streamIdRef.current} incomplete: chunks ${expectedSeq}/${chunkCount}, bytes ${receivedBytes}/${totalBytes}`
+          )
+        )
+        return
+      }
+      try {
+        succeed(JSON.parse(Buffer.concat(parts).toString('utf-8')))
+      } catch (err) {
+        fail(
+          new GitResponseStreamError(
+            `Git stream ${streamIdRef.current} JSON parse failed: ${String(err)}`
+          )
+        )
+      }
+    }
+
+    const handleStreamError = (p: Record<string, unknown>): void => {
+      if (settled || p.streamId !== streamIdRef.current) {
+        return
+      }
+      fail(new Error((p.message as string | undefined) ?? 'git response stream error'))
+    }
+
+    const drainPending = (): void => {
+      while (!settled && pending.length > 0) {
+        const frame = pending.shift()!
+        if (frame.kind === 'chunk') {
+          handleChunk(frame.params)
+        } else if (frame.kind === 'end') {
+          handleEnd(frame.params)
+        } else {
+          handleStreamError(frame.params)
+        }
+      }
+    }
+
+    unsubscribers.push(
+      mux.onNotificationByMethod('git.responseChunk', (p) => {
+        if (!metadataReady) {
+          pending.push({ kind: 'chunk', params: p })
+          return
+        }
+        handleChunk(p)
+      })
+    )
+    unsubscribers.push(
+      mux.onNotificationByMethod('git.responseEnd', (p) => {
+        if (!metadataReady) {
+          pending.push({ kind: 'end', params: p })
+          return
+        }
+        handleEnd(p)
+      })
+    )
+    unsubscribers.push(
+      mux.onNotificationByMethod('git.responseError', (p) => {
+        if (!metadataReady) {
+          pending.push({ kind: 'error', params: p })
+          return
+        }
+        handleStreamError(p)
+      })
+    )
+    unsubscribers.push(
+      mux.onDispose((reason) => {
+        const err = new Error(
+          reason === 'connection_lost'
+            ? 'SSH connection lost, reconnecting...'
+            : 'Multiplexer disposed'
+        ) as Error & { code: string }
+        err.code = reason === 'connection_lost' ? 'CONNECTION_LOST' : 'DISPOSED'
+        fail(err)
+      })
+    )
+
+    if (options?.signal) {
+      const signal = options.signal
+      if (signal.aborted) {
+        const err = new Error('Request was cancelled') as Error & { name: string }
+        err.name = 'AbortError'
+        fail(err)
+        return
+      }
+      const onAbort = (): void => {
+        const err = new Error('Request was cancelled') as Error & { name: string }
+        err.name = 'AbortError'
+        fail(err)
+      }
+      signal.addEventListener('abort', onAbort, { once: true })
+      unsubscribers.push(() => signal.removeEventListener('abort', onAbort))
+    }
+
+    // Why: forward options only when present so callers that previously issued a
+    // 2-arg mux.request keep the same call shape (and their tests).
+    const streamParams = { ...params, __streamResponse: true }
+    const requestPromise = options
+      ? mux.request(method, streamParams, options)
+      : mux.request(method, streamParams)
+    void requestPromise
+      .then((result) => {
+        if (settled) {
+          return
+        }
+        // Old relay / small result: plain single-frame value, no stream follows.
+        if (!isGitResponseStreamMarker(result)) {
+          succeed(result)
+          return
+        }
+        const marker = result.__orcaGitResponseStream
+        totalBytes = marker.totalBytes
+        chunkCount = marker.chunkCount
+        streamIdRef.current = marker.streamId
+        metadataReady = true
+        drainPending()
+      })
+      .catch((err) => fail(err as Error))
+  })
+}

src/relay/git-handler.ts

@@ -61,6 +61,8 @@ import { getGitCloneFailureMessage } from '../shared/git-clone-failure-message'
 import { syncForkDefaultBranch, validateGitForkSyncExpectedUpstream } from '../shared/git-fork-sync'
 import { InFlightPromiseDedupe, stableInFlightKey } from '../shared/in-flight-promise-dedupe'
 import { GIT_FETCH_SKIP_AUTO_MAINTENANCE_CONFIG_ARGS } from '../shared/git-fetch-auto-maintenance'
+import { GitResponseStreamRegistry } from './git-response-stream'
+import { GIT_RESPONSE_STREAM_THRESHOLD } from './protocol'
 
 const execFileAsync = promisify(execFile)
 const MAX_GIT_BUFFER = 10 * 1024 * 1024
@@ -171,6 +173,9 @@ function execFileWithStdin(
 export class GitHandler {
   private dispatcher: RelayDispatcher
   private readonly gitDiffReadDedupe = new InFlightPromiseDedupe<unknown>()
+  // Why: large diff/exec responses are chunked onto the bulk lane so they do
+  // not head-of-line-block interactive pty.data echo on the shared SSH channel.
+  private readonly responseStreams = new GitResponseStreamRegistry()
 
   // Why: configured submodule paths change rarely; an instance-level TTL cache
   // avoids re-reading `.gitmodules` on every diff click over SSH, and being
@@ -182,6 +187,13 @@ export class GitHandler {
   constructor(dispatcher: RelayDispatcher, _context: RelayContext) {
     this.dispatcher = dispatcher
     this.registerHandlers()
+    // Why: a detached client's git.responseAck frames will never arrive; wake
+    // any pump parked on the ack window so it re-checks staleness and exits.
+    this.dispatcher.onClientDetached?.(() => this.responseStreams.wakeAll())
+  }
+
+  dispose(): void {
+    this.responseStreams.disposeAll()
   }
 
   private registerHandlers(): void {
@@ -190,7 +202,7 @@ export class GitHandler {
     this.dispatcher.onRequest('git.checkIgnored', (p) => this.checkIgnored(p))
     this.dispatcher.onRequest('git.history', (p) => this.history(p))
     this.dispatcher.onRequest('git.commit', (p) => this.commit(p))
-    this.dispatcher.onRequest('git.diff', (p) => this.getDiff(p))
+    this.dispatcher.onRequest('git.diff', (p, context) => this.getDiff(p, context))
     this.dispatcher.onRequest('git.stage', (p) => this.stage(p))
     this.dispatcher.onRequest('git.unstage', (p) => this.unstage(p))
     this.dispatcher.onRequest('git.bulkStage', (p) => this.bulkStage(p))
@@ -215,8 +227,8 @@ export class GitHandler {
     this.dispatcher.onRequest('git.pull', (p) => this.pull(p))
     this.dispatcher.onRequest('git.fastForward', (p) => this.fastForward(p))
     this.dispatcher.onRequest('git.rebaseFromBase', (p) => this.rebaseFromBase(p))
-    this.dispatcher.onRequest('git.branchDiff', (p) => this.branchDiff(p))
-    this.dispatcher.onRequest('git.commitDiff', (p) => this.commitDiff(p))
+    this.dispatcher.onRequest('git.branchDiff', (p, context) => this.branchDiff(p, context))
+    this.dispatcher.onRequest('git.commitDiff', (p, context) => this.commitDiff(p, context))
     this.dispatcher.onRequest('git.listWorktrees', (p, context) => this.listWorktrees(p, context))
     this.dispatcher.onRequest('git.addWorktree', (p) => this.addWorktree(p))
     this.dispatcher.onRequest('git.removeWorktree', (p) => this.removeWorktree(p))
@@ -231,6 +243,43 @@ export class GitHandler {
     this.dispatcher.onRequest('git.exec', (p, context) => this.exec(p, context))
     this.dispatcher.onRequest('git.clone', (p, context) => this.clone(p, context))
     this.dispatcher.onRequest('git.isGitRepo', (p) => this.isGitRepo(p))
+    this.dispatcher.onNotification('git.responseAck', (p) => this.responseAck(p))
+    this.dispatcher.onNotification('git.cancelResponseStream', (p) => this.cancelResponseStream(p))
+  }
+
+  private responseAck(params: Record<string, unknown>): void {
+    const streamId = params.streamId
+    const seq = params.seq
+    if (typeof streamId === 'number' && typeof seq === 'number') {
+      this.responseStreams.recordAck(streamId, seq)
+    }
+  }
+
+  private cancelResponseStream(params: Record<string, unknown>): void {
+    const streamId = params.streamId
+    if (typeof streamId === 'number') {
+      this.responseStreams.abort(streamId)
+    }
+  }
+
+  // Why: when the client opted into response streaming and the serialized result
+  // exceeds the threshold, chunk it onto the bulk lane and return a small
+  // sentinel as the RPC result. Old clients omit the flag (single-frame, as
+  // today); old relays never call this, so a new client falls back to the plain
+  // result they return.
+  private maybeStreamResponse(
+    result: unknown,
+    params: Record<string, unknown>,
+    context: RequestContext | undefined
+  ): unknown {
+    if (params.__streamResponse !== true || !context) {
+      return result
+    }
+    const payload = Buffer.from(JSON.stringify(result ?? null), 'utf-8')
+    if (payload.length <= GIT_RESPONSE_STREAM_THRESHOLD) {
+      return result
+    }
+    return this.responseStreams.startStream(payload, this.dispatcher, context)
   }
 
   private async runWithDiffDedupeClear<T>(run: () => Promise<T>): Promise<T> {
@@ -352,7 +401,7 @@ export class GitHandler {
     })
   }
 
-  private async getDiff(params: Record<string, unknown>) {
+  private async getDiff(params: Record<string, unknown>, context?: RequestContext) {
     const worktreePath = params.worktreePath as string
     const filePath = params.filePath as string
     // Why: filePath is relative to worktreePath and used in readWorkingFile via
@@ -366,7 +415,7 @@ export class GitHandler {
     const compareAgainstHead = params.compareAgainstHead as boolean | undefined
     // Why: register the in-flight dedupe synchronously (before any await) so
     // concurrent identical reads coalesce; submodule routing happens inside.
-    return this.gitDiffReadDedupe.run(
+    const result = await this.gitDiffReadDedupe.run(
       stableInFlightKey(['diff', worktreePath, filePath, staged, compareAgainstHead]),
       async () => {
         // Why: gitlink paths can't be read as blobs and submodule working dirs
@@ -431,6 +480,7 @@ export class GitHandler {
         )
       }
     )
+    return this.maybeStreamResponse(result, params, context)
   }
 
   private async stage(params: Record<string, unknown>) {
@@ -1027,7 +1077,7 @@ export class GitHandler {
     }
   }
 
-  private async branchDiff(params: Record<string, unknown>) {
+  private async branchDiff(params: Record<string, unknown>, context?: RequestContext) {
     const worktreePath = params.worktreePath as string
     const baseRef = params.baseRef as string
     if (baseRef.startsWith('-')) {
@@ -1038,7 +1088,7 @@ export class GitHandler {
       filePath: params.filePath as string | undefined,
       oldPath: params.oldPath as string | undefined
     }
-    return this.gitDiffReadDedupe.run(
+    const result = await this.gitDiffReadDedupe.run(
       stableInFlightKey([
         'branchDiff',
         worktreePath,
@@ -1056,9 +1106,10 @@ export class GitHandler {
           options
         )
     )
+    return this.maybeStreamResponse(result, params, context)
   }
 
-  private async commitDiff(params: Record<string, unknown>) {
+  private async commitDiff(params: Record<string, unknown>, context?: RequestContext) {
     const worktreePath = params.worktreePath as string
     const args = {
       commitOid: params.commitOid as string,
@@ -1066,7 +1117,7 @@ export class GitHandler {
       filePath: params.filePath as string,
       oldPath: params.oldPath as string | undefined
     }
-    return this.gitDiffReadDedupe.run(
+    const result = await this.gitDiffReadDedupe.run(
       stableInFlightKey([
         'commitDiff',
         worktreePath,
@@ -1077,6 +1128,7 @@ export class GitHandler {
       ]),
       () => commitDiffEntry(this.gitBuffer.bind(this), worktreePath, args)
     )
+    return this.maybeStreamResponse(result, params, context)
   }
 
   private async exec(params: Record<string, unknown>, context?: RequestContext) {
@@ -1085,7 +1137,7 @@ export class GitHandler {
 
     validateGitExecArgs(args)
     const { stdout, stderr } = await this.git(args, cwd, { signal: context?.signal })
-    return { stdout, stderr }
+    return this.maybeStreamResponse({ stdout, stderr }, params, context)
   }
 
   private async clone(params: Record<string, unknown>, context?: RequestContext) {

src/relay/git-response-pty-echo-backpressure.integration.test.ts

@@ -0,0 +1,289 @@
+/**
+ * Regression: SSH typing latency under large git responses.
+ *
+ * The relay and the client share ONE ordered SSH channel. A large git diff/exec
+ * response sent as a single JSON-RPC frame queues megabytes into the outbound
+ * pipe at once; an interactive pty.data echo emitted mid-response then sits
+ * behind all of it and typing feels seconds-slow.
+ *
+ * These tests model the SSH channel as a congestible in-memory pipe and assert
+ * deterministic byte bounds instead of wall-clock latency:
+ *  - WITHOUT the fix (client does not opt in), the whole response queues ahead
+ *    of a pty echo;
+ *  - WITH the fix (client opts in), the response streams on the bulk lane and
+ *    at most ~1 chunk frame sits ahead of the echo;
+ *  - both paths reassemble the identical git result.
+ */
+import { describe, expect, it, beforeEach, afterEach } from 'vitest'
+import { mkdtempSync, mkdirSync, writeFileSync } from 'node:fs'
+import { rm } from 'node:fs/promises'
+import { execFileSync } from 'node:child_process'
+import * as path from 'node:path'
+import { tmpdir } from 'node:os'
+
+import {
+  SshChannelMultiplexer,
+  type MultiplexerTransport
+} from '../main/ssh/ssh-channel-multiplexer'
+import { requestGitStreamable } from '../main/ssh/ssh-git-response-stream-reader'
+
+import { RelayDispatcher } from './dispatcher'
+import { RelayContext } from './context'
+import { GitHandler } from './git-handler'
+import { GIT_RESPONSE_CHUNK_SIZE } from './protocol'
+
+// One framed git.responseChunk: base64 (4/3) + JSON envelope + header slack.
+const FRAMED_CHUNK_BYTES = Math.ceil((GIT_RESPONSE_CHUNK_SIZE * 4) / 3) + 512
+const SINK_HIGH_WATER_MARK = 64 * 1024
+
+async function waitUntil(
+  predicate: () => boolean,
+  what: string,
+  timeoutMs = 10_000
+): Promise<void> {
+  const deadline = Date.now() + timeoutMs
+  while (!predicate()) {
+    if (Date.now() > deadline) {
+      throw new Error(`waitUntil timed out: ${what}`)
+    }
+    await new Promise((r) => setImmediate(r))
+  }
+}
+
+async function waitUntilSettled(read: () => number, stableTurns = 25): Promise<void> {
+  let last = read()
+  let stable = 0
+  while (stable < stableTurns) {
+    await new Promise((r) => setImmediate(r))
+    const current = read()
+    if (current === last) {
+      stable += 1
+    } else {
+      stable = 0
+      last = current
+    }
+  }
+}
+
+type Harness = {
+  mux: SshChannelMultiplexer
+  dispatcher: RelayDispatcher
+  gitHandler: GitHandler
+  queuedBytes: () => number
+  deliverAll: () => void
+  startAutoDeliver: () => void
+  dispose: () => void
+}
+
+function createHarness(opts: { congested: boolean }): Harness {
+  let relayFeed: ((data: Buffer) => void) | null = null
+  const clientDataCallbacks: ((data: Buffer) => void)[] = []
+
+  const clientTransport: MultiplexerTransport = {
+    write: (data: Buffer) => {
+      // Client → relay (keystrokes, acks) is the opposite duplex direction and
+      // is not blocked by relay→client congestion.
+      setImmediate(() => relayFeed?.(data))
+    },
+    onData: (cb) => {
+      clientDataCallbacks.push(cb)
+    },
+    onClose: () => {}
+  }
+
+  const outQueue: Buffer[] = []
+  let queuedBytes = 0
+  const drainWaiters = new Set<() => void>()
+  const fireDrainIfIdle = (): void => {
+    if (queuedBytes > 0) {
+      return
+    }
+    for (const cb of Array.from(drainWaiters)) {
+      drainWaiters.delete(cb)
+      cb()
+    }
+  }
+
+  const dispatcher = new RelayDispatcher(
+    (data: Buffer) => {
+      outQueue.push(data)
+      queuedBytes += data.length
+      if (!opts.congested) {
+        return true
+      }
+      return queuedBytes < SINK_HIGH_WATER_MARK
+    },
+    {
+      waitWriteDrain: (cb: () => void) => {
+        drainWaiters.add(cb)
+        fireDrainIfIdle()
+      }
+    }
+  )
+  relayFeed = (data: Buffer) => dispatcher.feed(data)
+
+  const deliverAll = (): void => {
+    while (outQueue.length > 0) {
+      const buf = outQueue.shift()!
+      queuedBytes -= buf.length
+      for (const cb of clientDataCallbacks) {
+        cb(buf)
+      }
+    }
+    fireDrainIfIdle()
+  }
+
+  let autoDeliverTimer: ReturnType<typeof setInterval> | null = null
+  const startAutoDeliver = (): void => {
+    if (autoDeliverTimer) {
+      return
+    }
+    autoDeliverTimer = setInterval(deliverAll, 1)
+  }
+
+  const context = new RelayContext()
+  const gitHandler = new GitHandler(dispatcher, context)
+  const mux = new SshChannelMultiplexer(clientTransport)
+
+  return {
+    mux,
+    dispatcher,
+    gitHandler,
+    queuedBytes: () => queuedBytes,
+    deliverAll,
+    startAutoDeliver,
+    dispose: () => {
+      if (autoDeliverTimer) {
+        clearInterval(autoDeliverTimer)
+      }
+      mux.dispose()
+      dispatcher.dispose()
+      gitHandler.dispose()
+    }
+  }
+}
+
+// Build a repo whose staged diff for `big.txt` is several MB so git.diff
+// returns a payload well over the stream threshold.
+function makeRepoWithLargeStagedDiff(dir: string): void {
+  const env = { ...process.env }
+  const run = (args: string[]): void => {
+    execFileSync('git', args, { cwd: dir, stdio: 'pipe', env })
+  }
+  run(['init'])
+  run(['config', 'user.email', 'test@test.com'])
+  run(['config', 'user.name', 'Test'])
+  // Distinct non-repeating lines keep the diff from compressing to nothing.
+  // Stay under the render limits (120k lines / 6M chars) so the diff result
+  // carries the full ~4MB content and exceeds the stream threshold.
+  const lines: string[] = []
+  for (let i = 0; i < 12_000; i += 1) {
+    lines.push(`line ${i} ${'x'.repeat(100)}`)
+  }
+  writeFileSync(path.join(dir, 'big.txt'), lines.join('\n'))
+  run(['add', 'big.txt'])
+}
+
+describe('large git response vs pty.data echo head-of-line blocking', () => {
+  let tmpDir: string
+  let repoDir: string
+
+  beforeEach(() => {
+    tmpDir = mkdtempSync(path.join(tmpdir(), 'relay-git-hol-'))
+    repoDir = path.join(tmpDir, 'repo')
+    mkdirSync(repoDir, { recursive: true })
+    makeRepoWithLargeStagedDiff(repoDir)
+  })
+
+  afterEach(async () => {
+    await rm(tmpDir, { recursive: true, force: true })
+  })
+
+  it('WITHOUT opt-in: the whole diff queues ahead of a pty echo (single-frame HOL)', async () => {
+    const harness = createHarness({ congested: true })
+    try {
+      let queuedBytesAheadOfEcho = -1
+      harness.dispatcher.onNotification('pty.data', (params) => {
+        queuedBytesAheadOfEcho = harness.queuedBytes()
+        harness.dispatcher.notify('pty.data', { id: params.id, data: params.data })
+      })
+
+      // Plain request WITHOUT __streamResponse: single-frame response, as today.
+      const diffPromise = harness.mux.request('git.diff', {
+        worktreePath: repoDir,
+        filePath: 'big.txt',
+        staged: true
+      })
+      // Let the whole single JSON-RPC frame land in the congested pipe.
+      await waitUntil(() => harness.queuedBytes() > SINK_HIGH_WATER_MARK, 'diff frame queued')
+      await waitUntilSettled(() => harness.queuedBytes())
+
+      harness.mux.notify('pty.data', { id: 'pty-1', data: 'x' })
+      await waitUntil(() => queuedBytesAheadOfEcho >= 0, 'echo emitted by relay')
+
+      // The echo sits behind the entire (multi-hundred-KB) diff frame — far
+      // more than a single bulk chunk would be.
+      expect(queuedBytesAheadOfEcho).toBeGreaterThan(512 * 1024)
+
+      harness.startAutoDeliver()
+      const result = (await diffPromise) as { diff?: string }
+      expect(typeof result).toBe('object')
+    } finally {
+      harness.dispose()
+    }
+  }, 30_000)
+
+  it('WITH opt-in: the diff streams on the bulk lane and the echo stays bounded', async () => {
+    const harness = createHarness({ congested: true })
+    try {
+      let queuedBytesAheadOfEcho = -1
+      harness.dispatcher.onNotification('pty.data', (params) => {
+        queuedBytesAheadOfEcho = harness.queuedBytes()
+        harness.dispatcher.notify('pty.data', { id: params.id, data: params.data })
+      })
+
+      const diffPromise = requestGitStreamable(harness.mux, 'git.diff', {
+        worktreePath: repoDir,
+        filePath: 'big.txt',
+        staged: true
+      })
+      // Deliver the sentinel response, then let the pump run into congestion.
+      await waitUntil(() => harness.queuedBytes() > 0, 'sentinel queued')
+      harness.deliverAll()
+      await waitUntil(() => harness.queuedBytes() > 0, 'first chunk queued')
+      await waitUntilSettled(() => harness.queuedBytes())
+
+      harness.mux.notify('pty.data', { id: 'pty-1', data: 'x' })
+      await waitUntil(() => queuedBytesAheadOfEcho >= 0, 'echo emitted by relay')
+
+      // At most one in-flight bulk frame (the write that saturated the sink)
+      // plus slack sits ahead of the echo.
+      expect(queuedBytesAheadOfEcho).toBeLessThan(2 * FRAMED_CHUNK_BYTES)
+
+      harness.startAutoDeliver()
+      const streamed = (await diffPromise) as Record<string, unknown>
+      expect(typeof streamed).toBe('object')
+      expect(streamed).not.toHaveProperty('__orcaGitResponseStream')
+    } finally {
+      harness.dispose()
+    }
+  }, 30_000)
+
+  it('streamed result equals the single-frame result', async () => {
+    const harness = createHarness({ congested: false })
+    try {
+      harness.startAutoDeliver()
+      const params = { worktreePath: repoDir, filePath: 'big.txt', staged: true }
+      // Why: request both concurrently so the shared 1ms delivery pump drives
+      // the plain frame and the streamed chunk/ack round-trips without either
+      // starving the other on a slow CI event loop.
+      const [plain, streamed] = await Promise.all([
+        harness.mux.request('git.diff', params),
+        requestGitStreamable(harness.mux, 'git.diff', params)
+      ])
+      expect(streamed).toEqual(plain)
+    } finally {
+      harness.dispose()
+    }
+  }, 30_000)
+})

src/relay/git-response-stream.ts

@@ -0,0 +1,190 @@
+// Streams large git RPC responses (diff family + exec) onto the bulk lane in
+// chunks instead of one JSON-RPC frame, so a big diff 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 serialized string rather than a file handle.
+import type { RelayDispatcher, RequestContext } from './dispatcher'
+import {
+  GIT_RESPONSE_CHUNK_SIZE,
+  STREAM_ACK_WINDOW_CHUNKS,
+  STREAM_ACK_STALL_RECHECK_MS,
+  type GitResponseStreamMarker
+} from './protocol'
+
+type GitResponseStreamEntry = {
+  aborted: boolean
+  /** Highest chunk seq the client acknowledged (in-order; -1 = none yet). */
+  ackedThroughSeq: number
+  ackWaiters: Set<() => void>
+}
+
+/** Serialized git responses are chunked as base64 so multi-byte UTF-8
+ * sequences never split across a chunk boundary (the client concatenates the
+ * decoded bytes and parses once). */
+function encodeChunks(payload: Buffer): string[] {
+  const chunks: string[] = []
+  for (let offset = 0; offset < payload.length; offset += GIT_RESPONSE_CHUNK_SIZE) {
+    chunks.push(payload.subarray(offset, offset + GIT_RESPONSE_CHUNK_SIZE).toString('base64'))
+  }
+  return chunks
+}
+
+export class GitResponseStreamRegistry {
+  private streams = new Map<number, GitResponseStreamEntry>()
+  private nextId = 1
+
+  private register(): number {
+    const streamId = this.nextId++
+    this.streams.set(streamId, { aborted: false, ackedThroughSeq: -1, ackWaiters: new Set() })
+    return streamId
+  }
+
+  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.wake(entry)
+  }
+
+  abort(streamId: number): void {
+    const entry = this.streams.get(streamId)
+    if (entry) {
+      entry.aborted = true
+      this.wake(entry)
+    }
+  }
+
+  /** Wake every parked pump so it re-checks staleness — used when a client
+   * detaches and its acks will never arrive. */
+  wakeAll(): void {
+    for (const entry of this.streams.values()) {
+      this.wake(entry)
+    }
+  }
+
+  private wake(entry: GitResponseStreamEntry): void {
+    for (const waiter of Array.from(entry.ackWaiters)) {
+      waiter()
+    }
+  }
+
+  private waitForAck(streamId: number): Promise<void> {
+    const entry = this.streams.get(streamId)
+    if (!entry || entry.aborted) {
+      return Promise.resolve()
+    }
+    return new Promise<void>((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
+   * sentinel marker to send as the RPC result.
+   */
+  startStream(
+    payload: Buffer,
+    dispatcher: RelayDispatcher,
+    context: RequestContext
+  ): GitResponseStreamMarker {
+    const streamId = this.register()
+    const chunks = encodeChunks(payload)
+    // Why: kick the pump off the response task so the client sees the sentinel
+    // (and can subscribe/reassemble) before the first chunk frame arrives.
+    setImmediate(() => {
+      void this.pump(streamId, chunks, dispatcher, context)
+    })
+    return {
+      __orcaGitResponseStream: { streamId, totalBytes: payload.length, chunkCount: chunks.length }
+    }
+  }
+
+  private async pump(
+    streamId: number,
+    chunks: string[],
+    dispatcher: RelayDispatcher,
+    context: RequestContext
+  ): Promise<void> {
+    const entry = this.streams.get(streamId)
+    if (!entry) {
+      return
+    }
+    const clientId = context.clientId
+    let seq = 0
+    let endReason: 'end' | 'aborted' | 'stale' = 'end'
+    try {
+      for (seq = 0; seq < chunks.length; seq += 1) {
+        if (context.isStale()) {
+          endReason = 'stale'
+          break
+        }
+        if (entry.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 &&
+          !context.isStale() &&
+          !entry.aborted
+        ) {
+          await this.waitForAck(streamId)
+        }
+        if (context.isStale()) {
+          endReason = 'stale'
+          break
+        }
+        if (entry.aborted) {
+          endReason = 'aborted'
+          break
+        }
+        // Why: notifyBulk waits out sink saturation so chunk frames never pile
+        // up in the outbound pipe ahead of interactive pty.data frames.
+        await dispatcher.notifyBulk(
+          'git.responseChunk',
+          { streamId, seq, data: chunks[seq] },
+          {
+            clientId
+          }
+        )
+      }
+      if (endReason === 'end') {
+        dispatcher.notify('git.responseEnd', { streamId })
+      }
+    } catch (err) {
+      if (!context.isStale() && !entry.aborted) {
+        dispatcher.notify('git.responseError', {
+          streamId,
+          message: err instanceof Error ? err.message : String(err)
+        })
+      }
+    } finally {
+      this.streams.delete(streamId)
+    }
+  }
+
+  disposeAll(): void {
+    for (const id of Array.from(this.streams.keys())) {
+      this.abort(id)
+    }
+    this.streams.clear()
+  }
+}

src/relay/protocol.ts

@@ -59,6 +59,27 @@ export const STREAM_ACK_WINDOW_CHUNKS = 4
  * (and its open file handle) forever. */
 export const STREAM_ACK_STALL_RECHECK_MS = 1_000
 
+// ── Git response streaming (see docs/relay-git-response-stream-design.md) ──
+
+/** Serialized-JSON size above which a streamable git response (diff family +
+ * exec) is chunked onto the bulk lane instead of one JSON-RPC frame, so a large
+ * diff cannot head-of-line-block interactive pty.data echo on the shared SSH
+ * channel. Below this, single-frame is cheaper and avoids stream overhead. */
+export const GIT_RESPONSE_STREAM_THRESHOLD = 256 * 1024
+
+/** Per-chunk size (UTF-8 bytes of the serialized result) for git response
+ * streaming. Independent from STREAM_CHUNK_SIZE — this offset math is not
+ * shared with fs streams, so tuning it here is cross-version safe as long as
+ * the client reassembles by concatenation (it does not depend on chunk size). */
+export const GIT_RESPONSE_CHUNK_SIZE = 128 * 1024
+
+/** Sentinel result returned in place of a large git response: the real payload
+ * follows as git.responseChunk frames on the bulk lane. Old relays never emit
+ * this, so a new client falls back to the plain result they return. */
+export type GitResponseStreamMarker = {
+  __orcaGitResponseStream: { streamId: number; totalBytes: number; chunkCount: number }
+}
+
 export const RelayErrorCode = {
   TooManyStreams: -33006,
   StreamProtocolError: -33007

src/relay/relay.ts

@@ -454,10 +454,7 @@ async function main(): Promise<void> {
 
   const ptyHandler = new PtyHandler(dispatcher, graceTimeMs)
   const fsHandler = new FsHandler(dispatcher, context)
-  // Why: GitHandler registers its own request handlers on construction,
-  // so we hold the reference only for potential future disposal.
-  const _gitHandler = new GitHandler(dispatcher, context)
-  void _gitHandler
+  const gitHandler = new GitHandler(dispatcher, context)
 
   const _preflightHandler = new PreflightHandler(dispatcher)
   const _externalAutomationsHandler = new ExternalAutomationsHandler(dispatcher)
@@ -1013,6 +1010,7 @@ async function main(): Promise<void> {
     dispatcher.dispose()
     ptyHandler.dispose()
     fsHandler.dispose()
+    gitHandler.dispose()
     hookServer.stop()
     // Why: Node's Unix server.close() can unlink the listen path. If the path
     // was externally removed and rebound by a newer relay, closing this older