fix(ssh): prioritize mux control writes after drain

config/reliability-gates.jsonc

@@ -3104,7 +3104,7 @@
         "https://github.com/stablyai/orca/pull/7054"
       ],
       "invariant": "High-volume terminal output must stay bounded across daemon socket writes, SSH relay writer/decoder/source retention/reconnect, main model and projection admission, runtime metadata, detectors, and tail checks, main-to-renderer in-flight bytes, renderer scheduler queues, and hidden-output restore without starving focused input. Negotiated SSH spans remain exact and contiguous through activation, exit, cancellation, recovery, desktop projection, and required remote replacement; additional subscribers cannot stall the owner, while the required legacy primary retains backpressure.",
-      "oracle": "The current executable slice injects main-process renderer backlog pressure, then asserts unchanged path-provenance history reuse for pathless output, ordinary terminal-wait detection without a joined-tail allocation and with full prompt-family continuity, per-PTY and total pending-output caps, preserved sequenced-tail metadata, active-pending protection ahead of background trimming, and ACK-gated in-flight bounds. The joined source-intake/multiplex oracle blocks renderer or headless serialization, admits one snapshot-covered span and one trailing span, requires replacement reservation only after the authoritative sequence is known, advances upstream ACK eligibility exactly to the covered boundary after SnapshotEnd, and delivers then ACKs the trailing span through the ordinary live path without cancellation. Source-range ledger and multiplex seams additionally require cumulative byte ACK 40 then 100 for one 100-byte frame to release exact byte credit without early source settlement, admit a contiguous higher-generation recovered token while the prior token remains unsettled, and reject stale stream or source generations. Direct-SSH contracts additionally prove activation-response settlement before the first source frame, exact early-ACK reservation through send settlement or same-token retry, recovery-completion settlement before live admission, isolated saturated-subscriber eviction with healthy owner liveness, preserved gate-off primary backpressure, one-write ownership across write(false)/drain, liveness rebasing, failed-exit retention through late ACK or exact recovery, response-settlement cancellation authority, token-scoped exit cancellation, exact checkpoint-to-recoveryEnd continuity, stale-owner one-shot retry, no physical PTY teardown on recovery failure, exact provider-generation closure and once-only cleanup when recovery cancellation publication or proof rejects, 16 MiB-plus-1 MiB decoder caps, charged per-PTY/session retention budgets, and eight-wide isolated reattach. The live Electron perf oracle adds hidden-output floods, renderer scheduler queue depth, dropped-output-zero normal scenarios, and active key latency budgets before promotion.",
+      "oracle": "The current executable slice injects main-process renderer backlog pressure, then asserts unchanged path-provenance history reuse for pathless output, ordinary terminal-wait detection without a joined-tail allocation and with full prompt-family continuity, per-PTY and total pending-output caps, preserved sequenced-tail metadata, active-pending protection ahead of background trimming, and ACK-gated in-flight bounds. The joined source-intake/multiplex oracle blocks renderer or headless serialization, admits one snapshot-covered span and one trailing span, requires replacement reservation only after the authoritative sequence is known, advances upstream ACK eligibility exactly to the covered boundary after SnapshotEnd, and delivers then ACKs the trailing span through the ordinary live path without cancellation. Source-range ledger and multiplex seams additionally require cumulative byte ACK 40 then 100 for one 100-byte frame to release exact byte credit without early source settlement, admit a contiguous higher-generation recovered token while the prior token remains unsettled, and reject stale stream or source generations. Direct-SSH contracts additionally prove activation-response settlement before the first source frame, exact early-ACK reservation through send settlement or same-token retry, recovery-completion settlement before live admission, isolated saturated-subscriber eviction with healthy owner liveness, preserved gate-off primary backpressure, one-write ownership across write(false)/drain, control selection ahead of queued ordinary PTY input with FIFO lanes and bounded fairness, liveness rebasing, failed-exit retention through late ACK or exact recovery, response-settlement cancellation authority, token-scoped exit cancellation, exact checkpoint-to-recoveryEnd continuity, stale-owner one-shot retry, no physical PTY teardown on recovery failure, exact provider-generation closure and once-only cleanup when recovery cancellation publication or proof rejects, 16 MiB-plus-1 MiB decoder caps, charged per-PTY/session retention budgets, and eight-wide isolated reattach. The live Electron perf oracle adds hidden-output floods, renderer scheduler queue depth, dropped-output-zero normal scenarios, and active key latency budgets before promotion.",
       "commands": [
         "pnpm exec vitest run --config config/vitest.config.ts src/main/ipc/pty.test.ts",
         "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime-path-candidate-history.test.ts",
@@ -3114,7 +3114,7 @@
         "pnpm exec vitest run --config config/vitest.config.ts src/relay/dispatcher.test.ts src/relay/pty-handler-source-publication.test.ts --reporter=dot",
         "pnpm exec vitest run --config config/vitest.config.ts src/main/ssh/ssh-relay-session-recovery-races.test.ts --reporter=dot",
         "pnpm exec vitest run --config config/vitest.config.ts src/main/ipc/ssh-pty-remote-source-range-consumers.test.ts src/main/runtime/rpc/terminal-source-range-ledger.test.ts src/main/runtime/rpc/terminal-multiplex.test.ts --reporter=dot",
-        "pnpm exec vitest run --config config/vitest.config.ts src/shared/pty-consumer-session.test.ts src/main/ipc/ssh-pty-output-intake.test.ts src/main/ipc/ssh-pty-remote-source-range-consumers.test.ts src/main/runtime/rpc/terminal-multiplex.test.ts src/main/ssh/ssh-multiplexer-transport-writer.test.ts src/main/ssh/ssh-channel-multiplexer.test.ts src/main/ssh/ssh-pty-source-credit-rollout.test.ts src/main/ssh/ssh-relay-deploy.test.ts src/relay/dispatcher-client-writer.test.ts --reporter=dot",
+        "pnpm exec vitest run --config config/vitest.config.ts src/shared/pty-consumer-session.test.ts src/main/ipc/ssh-pty-output-intake.test.ts src/main/ipc/ssh-pty-remote-source-range-consumers.test.ts src/main/runtime/rpc/terminal-multiplex.test.ts src/main/ssh/ssh-multiplexer-transport-writer.test.ts src/main/ssh/ssh-channel-multiplexer.test.ts src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts src/main/ssh/ssh-pty-source-credit-rollout.test.ts src/main/ssh/ssh-relay-deploy.test.ts src/relay/dispatcher-client-writer.test.ts --reporter=dot",
         "pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/wsl-hook-relay-sentinel.test.ts --reporter=dot",
         "ORCA_E2E_SSH_DOCKER=1 ORCA_SSH_PTY_SOURCE_CREDIT_V1=1 pnpm exec playwright test tests/e2e/ssh-docker-relay-perf.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1"
       ],
@@ -3148,6 +3148,7 @@
         "src/main/runtime/rpc/terminal-multiplex.test.ts",
         "src/main/ssh/ssh-multiplexer-transport-writer.test.ts",
         "src/main/ssh/ssh-channel-multiplexer.test.ts",
+        "src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts",
         "src/main/ssh/ssh-pty-source-credit-rollout.test.ts",
         "src/main/ssh/ssh-relay-deploy.test.ts",
         "src/relay/dispatcher-client-writer.test.ts",
@@ -3309,10 +3310,19 @@
           "file": "src/main/ssh/ssh-multiplexer-transport-writer.test.ts",
           "assertions": [
             "write(false) owns one frame and later ordinary traffic waits for drain",
+            "queued control is selected before ordinary PTY backlog at drain while each lane remains FIFO",
+            "four-control fairness guarantees ordinary progress without starving control",
             "one coalesced liveness bypass is allowed per saturated epoch",
             "saturation and drain transitions are reported exactly once"
           ]
         },
+        {
+          "file": "src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts",
+          "assertions": [
+            "source ACK, cancellation, exit, request, and response control frames precede queued pasted PTY input after drain",
+            "control and ordinary frames preserve lane FIFO while ordinary input progresses after four control writes"
+          ]
+        },
         {
           "file": "src/main/ssh/ssh-channel-multiplexer.test.ts",
           "assertions": [
@@ -3408,7 +3418,7 @@
       },
       "redGreenEvidence": {
         "status": "partial",
-        "evidence": "Tests assert pathless runtime output reuses unchanged path-provenance history with zero old-candidate byte scans, ordinary saturated terminal tails retain no rebuilt wait text or repeated phrase scans while every blocked/ready prompt family remains live, main pending renderer output is capped per PTY and in total, direct-SSH source/recovery memory and both frame decoders stay within exact caps, and 50-PTY reconnect isolates slow/failing siblings. Before the snapshot-admission fix, the focused consumer oracle failed because the trailing span entered transferring instead of remaining open; after retaining model-sequence ends and reserving at the serialized fence, the joined main/runtime oracle passed for renderer and headless sources with upstream boundaries 4 then 8. Before cumulative partial ACK and source-rotation support, the ledger rejected ACK 40 of 100 and the multiplex seam detached instead of publishing recovered token B; both exact oracles pass while stale stream and source generations remain rejected. Other controls previously failed with serial reattach, UTF-8-only accounting, and roughly 32 MiB decoder retention. Needs daemon-stream contract coverage and broader hidden-output/input-latency perf artifacts before promotion."
+        "evidence": "Tests assert pathless runtime output reuses unchanged path-provenance history with zero old-candidate byte scans, ordinary saturated terminal tails retain no rebuilt wait text or repeated phrase scans while every blocked/ready prompt family remains live, main pending renderer output is capped per PTY and in total, direct-SSH source/recovery memory and both frame decoders stay within exact caps, and 50-PTY reconnect isolates slow/failing siblings. The deterministic main-to-relay drain oracle failed with ordinary-2 and ordinary-3 ahead of a later control frame under the single FIFO, then passed with control-first lane selection and four-write ordinary fairness. Before the snapshot-admission fix, the focused consumer oracle failed because the trailing span entered transferring instead of remaining open; after retaining model-sequence ends and reserving at the serialized fence, the joined main/runtime oracle passed for renderer and headless sources with upstream boundaries 4 then 8. Before cumulative partial ACK and source-rotation support, the ledger rejected ACK 40 of 100 and the multiplex seam detached instead of publishing recovered token B; both exact oracles pass while stale stream and source generations remain rejected. Other controls previously failed with serial reattach, UTF-8-only accounting, and roughly 32 MiB decoder retention. Needs daemon-stream contract coverage and broader hidden-output/input-latency perf artifacts before promotion."
       },
       "performanceBudget": {
         "required": true,

src/main/ssh/ssh-channel-multiplexer-backpressure.test.ts

@@ -136,7 +136,7 @@ describe('SshChannelMultiplexer backpressure hardening', () => {
     expect(seen).toEqual(Array.from({ length: 65 }, (_, index) => index))
   })
 
-  it('keeps acknowledgements, cancellation, exit, requests, and responses ordered after drain', async () => {
+  it('prioritizes source ACK and control while preserving FIFO and ordinary progress', async () => {
     mux.dispose()
     const written: Buffer[] = []
     let drain = (): void => {}
@@ -163,7 +163,9 @@ describe('SshChannelMultiplexer backpressure hardening', () => {
     mux.onRequest('client.control', () => ({ accepted: true }))
     const controller = new AbortController()
 
-    mux.notify('pty.data', { id: 'pty-1', data: 'ordinary' })
+    mux.notify('pty.data', { id: 'pty-1', data: 'ordinary-1' })
+    mux.notify('pty.data', { id: 'pty-1', data: 'ordinary-2' })
+    mux.notify('pty.data', { id: 'pty-1', data: 'ordinary-3' })
     mux.notify('pty.ackData', { acknowledgements: [] })
     const request = mux.request('fs.scan', {}, { signal: controller.signal })
     controller.abort()
@@ -180,13 +182,21 @@ describe('SshChannelMultiplexer backpressure hardening', () => {
     const payloads = written.map((frame) =>
       JSON.parse(frame.subarray(HEADER_LENGTH, HEADER_LENGTH + frame.readUInt32BE(9)).toString())
     )
-    expect(payloads.map((payload) => payload.method ?? `response:${payload.id}`)).toEqual([
-      'pty.data',
+    expect(
+      payloads.map((payload) =>
+        payload.method === 'pty.data'
+          ? `pty.data:${payload.params.data}`
+          : (payload.method ?? `response:${payload.id}`)
+      )
+    ).toEqual([
+      'pty.data:ordinary-1',
       'pty.ackData',
       'fs.scan',
       'rpc.cancel',
       'pty.exit',
-      'response:91'
+      'pty.data:ordinary-2',
+      'response:91',
+      'pty.data:ordinary-3'
     ])
     await expect(request).rejects.toMatchObject({ name: 'AbortError' })
   })

src/main/ssh/ssh-multiplexer-transport-writer.test.ts

@@ -47,7 +47,51 @@ function transportHarness(writeResults: (boolean | void)[]): WriterHarness {
 }
 
 describe('SshMultiplexerTransportWriter', () => {
-  it('preserves FIFO order and waits for drain after write(false)', () => {
+  it('selects queued control before ordinary backlog at the drain boundary', () => {
+    const harness = transportHarness([false, true, true, true])
+    const writer = new SshMultiplexerTransportWriter(harness.transport, vi.fn())
+
+    writer.enqueue(Buffer.from('ordinary-1'), 'ordinary')
+    writer.enqueue(Buffer.from('ordinary-2'), 'ordinary')
+    writer.enqueue(Buffer.from('ordinary-3'), 'ordinary')
+    writer.enqueue(Buffer.from('control'), 'control')
+
+    expect(harness.writes.map(String)).toEqual(['ordinary-1'])
+    harness.drain()
+    expect(harness.writes.map(String)).toEqual([
+      'ordinary-1',
+      'control',
+      'ordinary-2',
+      'ordinary-3'
+    ])
+    writer.dispose()
+  })
+
+  it('preserves FIFO within each lane and prevents ordinary starvation', () => {
+    const harness = transportHarness([false, ...Array<boolean>(8).fill(true)])
+    const writer = new SshMultiplexerTransportWriter(harness.transport, vi.fn())
+
+    writer.enqueue(Buffer.from('ordinary-1'), 'ordinary')
+    writer.enqueue(Buffer.from('ordinary-2'), 'ordinary')
+    for (let index = 1; index <= 6; index++) {
+      writer.enqueue(Buffer.from(`control-${index}`), 'control')
+    }
+
+    harness.drain()
+    expect(harness.writes.map(String)).toEqual([
+      'ordinary-1',
+      'control-1',
+      'control-2',
+      'control-3',
+      'control-4',
+      'ordinary-2',
+      'control-5',
+      'control-6'
+    ])
+    writer.dispose()
+  })
+
+  it('waits for drain and settles each write once', () => {
     const harness = transportHarness([false, true, true])
     const failed = vi.fn()
     const writer = new SshMultiplexerTransportWriter(harness.transport, failed)

src/main/ssh/ssh-multiplexer-transport-writer.ts

@@ -1,4 +1,5 @@
 import { HEADER_LENGTH, MAX_MESSAGE_SIZE } from './relay-protocol'
+import { SshMultiplexerWriterLaneScheduler } from './ssh-multiplexer-writer-lane-scheduler'
 
 export type MultiplexerWriteSettlement = { ok: true } | { ok: false; error: Error }
 
@@ -41,7 +42,7 @@ function onceSettlement(
 }
 
 export class SshMultiplexerTransportWriter {
-  private readonly queue: WriterEntry[] = []
+  private readonly scheduler = new SshMultiplexerWriterLaneScheduler<WriterEntry>()
   private readonly inFlight = new Set<WriterEntry>()
   private readonly settleOnDrain = new Set<WriterEntry>()
   private ordinaryBytes = 0
@@ -91,7 +92,7 @@ export class SshMultiplexerTransportWriter {
     if (lane === 'liveness' && this.saturated) {
       this.writeEntry(entry)
     } else {
-      this.queue.push(entry)
+      this.scheduler.enqueue(entry, lane)
       this.pump()
     }
     return true
@@ -105,7 +106,7 @@ export class SshMultiplexerTransportWriter {
     this.saturated = false
     this.removeDrainListener?.()
     this.removeDrainListener = null
-    for (const entry of this.queue.splice(0)) {
+    for (const entry of this.scheduler.clear()) {
       this.release(entry, { ok: false, error })
     }
     for (const entry of Array.from(this.inFlight)) {
@@ -133,7 +134,7 @@ export class SshMultiplexerTransportWriter {
     this.pumping = true
     try {
       while (!this.closed && !this.saturated) {
-        const entry = this.queue.shift()
+        const entry = this.scheduler.select()
         if (!entry) {
           return
         }

src/main/ssh/ssh-multiplexer-writer-lane-scheduler.ts

@@ -0,0 +1,73 @@
+import type { MultiplexerWriterLane } from './ssh-multiplexer-transport-writer'
+
+const CONTROL_WRITES_BEFORE_ORDINARY = 4
+
+type LaneQueue<T> = {
+  entries: T[]
+  head: number
+}
+
+function createLaneQueue<T>(): LaneQueue<T> {
+  return { entries: [], head: 0 }
+}
+
+function hasEntries<T>(queue: LaneQueue<T>): boolean {
+  return queue.head < queue.entries.length
+}
+
+function shift<T>(queue: LaneQueue<T>): T | undefined {
+  const entry = queue.entries[queue.head]
+  if (entry === undefined) {
+    return undefined
+  }
+  queue.head += 1
+  if (queue.head === queue.entries.length) {
+    queue.entries.length = 0
+    queue.head = 0
+  }
+  return entry
+}
+
+function clear<T>(queue: LaneQueue<T>): T[] {
+  const entries = queue.entries.slice(queue.head)
+  queue.entries.length = 0
+  queue.head = 0
+  return entries
+}
+
+export class SshMultiplexerWriterLaneScheduler<T extends object> {
+  private readonly ordinary = createLaneQueue<T>()
+  private readonly control = createLaneQueue<T>()
+  private readonly liveness = createLaneQueue<T>()
+  private controlWritesSinceOrdinary = 0
+
+  enqueue(entry: T, lane: MultiplexerWriterLane): void {
+    this[lane].entries.push(entry)
+  }
+
+  select(): T | undefined {
+    const liveness = shift(this.liveness)
+    if (liveness) {
+      return liveness
+    }
+    if (
+      hasEntries(this.control) &&
+      (!hasEntries(this.ordinary) ||
+        this.controlWritesSinceOrdinary < CONTROL_WRITES_BEFORE_ORDINARY)
+    ) {
+      this.controlWritesSinceOrdinary += 1
+      return shift(this.control)
+    }
+    const ordinary = shift(this.ordinary)
+    if (ordinary) {
+      this.controlWritesSinceOrdinary = 0
+      return ordinary
+    }
+    return shift(this.control)
+  }
+
+  clear(): T[] {
+    this.controlWritesSinceOrdinary = 0
+    return [...clear(this.liveness), ...clear(this.control), ...clear(this.ordinary)]
+  }
+}