fix(terminal): stop gating OutputSpan on the desktop multiplex path

The OutputSpan gate is right for `terminal.subscribe` and wrong for
`terminal.multiplex`. Keying spans off a declared `outputSpan` capability treats
"peer did not declare it" as "peer cannot decode opcode 15", which is false on
multiplex: opcode 15 shipped in v1.4.147 (c0f0810dd9, 2026-07-19) and every
capability a peer could declare to prove it is younger — `ackOutput` by 9 days,
`ackOutputSourceRanges` by 10. So the gate read the entire installed base as
incapable and downgraded it.

The downgrade cannot be made safe there. A plain Output frame carries only
`(data, seq)`, so a seq-tracking peer reconstructs a run as `seq - data.length`,
which a transformed run (`rawLength !== data.length`) contradicts by definition.
Three failures reproduced through the real dispatcher against the shipped desktop
capability set:

- a zero-byte absorbed-query frame fails `TerminalSourceRangeLedger.canAccept`
  (`encodedBytes > 0`), parks in `ackPendingOutput`, and head-blocks the flush
  loop on the same head forever — all later output is lost;
- every transformed emission reads as a dropped-frame gap in `detectOutputGap`
  and triggers `requestResyncSnapshot`;
- a >48 KiB transformed run takes the `shouldDelayFinalSeq` arm, which omits
  `sourceRanges`, flipping the ledger's mapping mode and detaching the stream.

Multiplex now always frames transformed runs as OutputSpan; the vestigial
`outputSpan` capability is removed from the multiplex schema, stream state, and
the renderer's subscribe frame. Subscribe keeps the gate: mobile ignores Output
`seq` outright, which is what makes the downgrade lossless there — that property,
not the capability flag, is the licence, so the chunker option is now
`transformedRuns: 'span' | 'downgrade-to-text'` and the unsound call site cannot
be written by accident.

Coverage for the fallback arm was absent — every fixture was a 14-character
string, so the multi-chunk downgrade never ran and deleting a yield (48 KiB of
silently dropped output) passed everything. Adds over-cap, zero-byte and
per-frame-seq contracts plus a 600-case fuzz, transformed and multi-chunk
transformed steps to the cross-version journey (which pins current code against
the real v1.4.173 release client), and repairs the JS benchmark that the required
options argument broke.

Refs STA-3482

config/scripts/terminal-output-frame-chunk-benchmark.mjs

@@ -88,6 +88,13 @@ for (const [source, label, marker] of [
   ],
   [CHUNK_SOURCE, 'terminal-output-frame-chunks.ts', 'const text = data.slice(chunkStart, end)'],
   [CHUNK_SOURCE, 'terminal-output-frame-chunks.ts', 'data.charCodeAt(index + 1)'],
+  // This file is JavaScript, so tsc cannot flag a changed generator signature; assert the
+  // options contract textually instead of discovering it as a runtime TypeError.
+  [
+    CHUNK_SOURCE,
+    'terminal-output-frame-chunks.ts',
+    "transformedRuns: 'span' | 'downgrade-to-text'"
+  ],
   [CLIPBOARD_SOURCE, 'clipboard-text.ts', 'export function measureClipboardTextByteLength('],
   [CLIPBOARD_SOURCE, 'clipboard-text.ts', 'text.codePointAt(index)']
 ]) {
@@ -114,6 +121,13 @@ const { exceedsTerminalStreamChunkBytes, iterateTerminalOutputFrameChunks } = aw
   new URL('../../src/main/runtime/rpc/terminal-output-frame-chunks.ts', import.meta.url).href
 )
 
+// The production generator takes a required options argument that this JS file cannot be
+// type-checked against; the marker list above fails loudly if its shape changes. `'span'`
+// is the mode `iterateBefore` models, so both arms emit identical frames.
+function iterateAfter(data, meta) {
+  return iterateTerminalOutputFrameChunks(data, meta, { transformedRuns: 'span' })
+}
+
 function previousGate(data) {
   return (
     data.length > TERMINAL_STREAM_CHUNK_BYTES ||
@@ -298,7 +312,7 @@ function median(samples) {
 function measureInterleaved(data, meta) {
   for (let index = 0; index < WARMUP; index += 1) {
     drain(iterateBefore, data, meta)
-    drain(iterateTerminalOutputFrameChunks, data, meta)
+    drain(iterateAfter, data, meta)
   }
   const beforeSamples = []
   const afterSamples = []
@@ -313,7 +327,7 @@ function measureInterleaved(data, meta) {
     const runAfter = () => {
       const start = performance.now()
       for (let index = 0; index < ITERATIONS; index += 1) {
-        drain(iterateTerminalOutputFrameChunks, data, meta)
+        drain(iterateAfter, data, meta)
       }
       afterSamples.push((performance.now() - start) / ITERATIONS)
     }
@@ -373,7 +387,7 @@ let comparedFixtures = 0
 for (const fixture of fixtures) {
   const meta = fixture.meta(fixture.data)
   const before = describeFrames(iterateBefore, fixture.data, meta)
-  const after = describeFrames(iterateTerminalOutputFrameChunks, fixture.data, meta)
+  const after = describeFrames(iterateAfter, fixture.data, meta)
   if (before !== after) {
     throw new Error(`frames differ for ${fixture.label}`)
   }

docs/reference/remote-wire-compatibility.md

@@ -51,27 +51,54 @@ peer confirms it. The existing pattern is `SetOutputPaused` (opcode 16):
 - the client sends opcode 16 only after that echo (`stream.supportsOutputPause`);
 - the host only acts on opcode 16 when it negotiated it (`stream.supportsOutputPause`).
 
-`OutputSpan` (opcode 15) is the worked example of getting this wrong. It shipped
-ungated on both the legacy `terminal.subscribe` stream and the multiplex stream, so
-transformed emissions (SSH relay, startup ingress, daemon keep-tail salvage) reached
-mobile as an opcode its vendored table does not contain — dropped with no error, no
-banner, no retry. It is now gated on `capabilities.outputSpan`, and a sender that has
-not been told the peer can decode a span **downgrades to plain `Output`** rather than
-sending nothing. That downgrade is the part worth copying: for an opcode that carries
-user-visible data, "don't send it" is still data loss, so a gate needs a fallback that
-preserves the payload, not just a suppressed frame.
+`OutputSpan` (opcode 15) is the worked example, and it is worth reading closely because
+the obvious fix was wrong on half the paths.
+
+It shipped ungated in v1.4.147 (2026-07-19) on both the legacy `terminal.subscribe`
+stream and the multiplex stream, so transformed emissions (SSH relay, startup ingress,
+daemon keep-tail salvage) reached mobile as an opcode its vendored table does not
+contain — dropped with no error, no banner, no retry. On `terminal.subscribe` it is now
+gated on `capabilities.outputSpan`, and a sender that has not been told the peer can
+decode a span **downgrades to plain `Output`** rather than sending nothing.
+
+On `terminal.multiplex` it is deliberately **not** gated. Two reasons, and both
+generalise:
+
+**A capability cannot retroactively prove incapability.** The gate keys off the peer
+declaring `outputSpan`, a flag no shipped build sends. Opcode 15 predates every
+capability a multiplex peer could declare — `ackOutput` by 9 days, `ackOutputSourceRanges`
+by 10 — so "did not declare it" is indistinguishable between a build from before the
+opcode and the entire installed base from after it. Since mixed versions are the normal
+state, a gate added after an opcode ships reads the whole fleet as incapable.
+
+**A fallback is only a fallback if it is lossless for that peer.** A plain `Output`
+frame carries `(data, seq)` and nothing else, so the receiver can only reconstruct a run
+as `seq - data.length`. A transformed run has `rawLength !== data.length` by definition.
+Downgrading it to `Output` therefore cannot be made correct for a peer that tracks
+`seq`, and the desktop multiplex client does: it head-blocks the ack queue on a
+zero-byte frame, reads a false dropped-frame gap on every transformed emission, and
+detaches the stream when a >48 KiB run flips the source-range ledger's mapping mode.
+The downgrade works on `terminal.subscribe` for exactly one reason — mobile ignores
+Output `seq` entirely — and that fact, not the capability flag, is what licenses it.
+
+So: for an opcode carrying user-visible data, "don't send it" is still data loss, and a
+gate needs a fallback that preserves the payload. But check what the fallback frame can
+actually express for the peer you are aiming it at before you reach for one, and prefer
+gating at the point where the new opcode was genuinely new rather than adding a gate to
+a path where it has already shipped.
 
 ### Mobile vendors its own opcode table
 
-`mobile/` is a separate pnpm workspace and cannot import `src/shared/`, so
-`mobile/src/transport/terminal-stream-protocol.ts` holds a **hand-copied** enum. It
-knows only 1-6 and 12. Adding an opcode to the shared file therefore does not add it
-to mobile, and nothing fails — not typecheck, not lint, not the cross-version harness,
-which pairs desktop builds only. The two tables drift silently by construction.
+`mobile/` is a separate pnpm workspace, so
+`mobile/src/transport/terminal-stream-protocol.ts` has historically held a **hand-copied**
+enum knowing only 1-6 and 12. Adding an opcode to the shared file did not add it to
+mobile, and nothing failed — not typecheck, not lint, not the cross-version harness,
+which pairs desktop builds only. The two tables drifted silently by construction.
 
-So "is this opcode safe to send?" cannot be answered by reading the shared enum. Until
-the vendored copy is generated from the shared one, treat every mobile-reachable
-opcode above 12 as unsupported unless the stream negotiated it.
+`terminal-subscribe-output-span-capability.test.ts` now parses mobile's real source and
+forces an explicit per-opcode decision, so a new opcode is neither "mobile decodes it"
+nor "gated" until someone says which. Until mobile shares the table outright, treat
+every mobile-reachable opcode above 12 as unsupported unless the stream negotiated it.
 
 Reuse an existing opcode with a new optional payload field (Rule 1) whenever that
 expresses the change; reach for a new opcode only when framing genuinely differs.

src/main/runtime/rpc/methods/terminal.ts

@@ -135,7 +135,6 @@ type TerminalMultiplexStream = {
   ackInFlightBytes: number
   ackWindowBytes: number
   supportsOutputPause: boolean
-  supportsOutputSpan: boolean
   supportsWriteUnavailable: boolean
   outputPaused: boolean
   supportsDesktopViewportClaims: boolean
@@ -571,10 +570,9 @@ function* iterateTerminalStreamTextPayloads(data: string): Generator<Uint8Array<
   if (!data) {
     return
   }
-  // Text-only payloads (no meta), so the span branch cannot fire; the caller
-  // discards opcodes anyway.
+  // Text-only payloads (no meta), so no run is transformed and neither mode can fire.
   for (const chunk of iterateTerminalOutputFrameChunks(data, undefined, {
-    supportsOutputSpan: false
+    transformedRuns: 'span'
   })) {
     yield chunk.bytes
   }
@@ -1057,7 +1055,10 @@ const TerminalMultiplexSubscribeFrame = TerminalHandle.extend({
       ackOutputSourceRanges: z.literal(1).optional(),
       desktopViewportClaims: z.literal(1).optional(),
       outputPause: z.literal(1).optional(),
-      outputSpan: z.literal(1).optional(),
+      // No `outputSpan` here on purpose: OutputSpan is unconditional on multiplex.
+      // Opcode 15 (v1.4.147, 2026-07-19) predates every capability a peer could declare
+      // to prove it, so absence proves nothing and would read the installed base as
+      // incapable. See docs/reference/remote-wire-compatibility.md.
       writeUnavailable: z.literal(1).optional()
     })
     .optional()
@@ -2511,7 +2512,6 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
           ackInFlightBytes: 0,
           ackWindowBytes: TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES,
           supportsOutputPause: request.capabilities?.outputPause === 1,
-          supportsOutputSpan: request.capabilities?.outputSpan === 1,
           supportsWriteUnavailable: request.capabilities?.writeUnavailable === 1,
           outputPaused: false,
           supportsDesktopViewportClaims: request.capabilities?.desktopViewportClaims === 1,
@@ -2539,8 +2539,13 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
                 meta.seq
               )
             }
+            // Why always spans: multiplex peers are desktops, which track output `seq`
+            // and run the ack/source-range ledger. Downgrading a transformed run to plain
+            // Output would strip the raw high-water mark the ledger and the client's
+            // gap detector both depend on. OutputSpan has also shipped unconditionally on
+            // this path since v1.4.147, so there is no peer a downgrade would help.
             for (const chunk of iterateTerminalOutputFrameChunks(data, meta, {
-              supportsOutputSpan: stream.supportsOutputSpan
+              transformedRuns: 'span'
             })) {
               queueOrSendOutput(stream, chunk)
             }
@@ -3160,7 +3165,10 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
           )
         }
         for (const chunk of iterateTerminalOutputFrameChunks(data, meta, {
-          supportsOutputSpan
+          // Mobile is this path's only client and it ignores Output `seq` entirely, so
+          // folding a transformed run into text frames loses nothing a phone can observe —
+          // whereas an undecodable OutputSpan loses the output itself (STA-3482).
+          transformedRuns: supportsOutputSpan ? 'span' : 'downgrade-to-text'
         })) {
           sendFrame(chunk.opcode ?? TerminalStreamOpcode.Output, chunk.bytes, chunk.seq)
         }

src/main/runtime/rpc/terminal-multiplex-transformed-output.test.ts

@@ -0,0 +1,292 @@
+import { describe, expect, it, vi } from 'vitest'
+import { RpcDispatcher } from './dispatcher'
+import type { RpcRequest } from './core'
+import type { OrcaRuntimeService, RuntimeTerminalDataMeta } from '../orca-runtime'
+import { TERMINAL_METHODS } from './methods/terminal'
+import type { RuntimeTerminalWait } from '../../../shared/runtime-types'
+import {
+  TerminalStreamOpcode,
+  decodeTerminalStreamFrame,
+  decodeTerminalStreamJson,
+  decodeTerminalStreamText,
+  encodeTerminalStreamFrame,
+  encodeTerminalStreamJson
+} from '../../../shared/terminal-stream-protocol'
+import { TERMINAL_STREAM_CHUNK_BYTES } from '../../../shared/terminal-multiplex-flow-control'
+
+const STREAM_ID = 7
+
+type DataListener = (data: string, meta?: RuntimeTerminalDataMeta) => void
+
+function startMultiplexHarness() {
+  const messages: string[] = []
+  const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
+  const handlers = new Map<
+    number,
+    (frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
+  >()
+  const cleanups = new Map<string, () => void>()
+  let dataListener: DataListener | undefined
+  const runtime = {
+    getRuntimeId: () => 'test-runtime',
+    registerRemoteTerminalViewSubscriber: () => () => {},
+    resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
+    requestRendererTerminalTabMount: vi.fn().mockReturnValue(true),
+    updateRemoteDesktopViewer: vi.fn().mockResolvedValue(true),
+    unregisterRemoteDesktopViewer: vi.fn().mockResolvedValue(true),
+    unregisterRemoteDesktopViewers: vi.fn().mockResolvedValue(true),
+    isPtyResizeDrivenRemotely: vi.fn().mockReturnValue(false),
+    getRemoteDesktopFitHold: vi.fn().mockReturnValue({ mode: 'desktop-fit', cols: 120, rows: 40 }),
+    isRemoteDesktopViewerOwner: vi.fn().mockReturnValue(false),
+    getPtyOutputSequence: vi.fn().mockReturnValue(0),
+    readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
+    serializeTerminalBuffer: vi
+      .fn()
+      .mockResolvedValue({ data: 'snapshot', cols: 120, rows: 40, source: 'headless', seq: 0 }),
+    serializeAuthoritativeTerminalBuffer: vi
+      .fn()
+      .mockResolvedValue({ data: 'snapshot', cols: 120, rows: 40, source: 'headless', seq: 0 }),
+    getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
+    getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
+    getLayout: vi.fn().mockReturnValue({ seq: 1 }),
+    subscribeToTerminalData: vi.fn((_ptyId: string, listener: DataListener) => {
+      dataListener = listener
+      return vi.fn()
+    }),
+    subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
+    subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
+    subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()),
+    getTerminalFitOverride: vi.fn().mockReturnValue(null),
+    getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
+    attachRemoteTerminalSourceRangeConsumer: vi.fn(() => true),
+    settleRemoteTerminalSourceRanges: vi.fn(),
+    cancelRemoteTerminalSourceRanges: vi.fn(),
+    reserveRemoteTerminalSourceRangeReplacement: vi.fn(() => null),
+    commitRemoteTerminalSourceRangeReplacement: vi.fn(() => true),
+    rollbackRemoteTerminalSourceRangeReplacement: vi.fn(() => true),
+    registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
+      cleanups.set(id, cleanup)
+    }),
+    cleanupSubscription: vi.fn((id: string) => {
+      cleanups.get(id)?.()
+    }),
+    waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
+  } as unknown as OrcaRuntimeService
+
+  const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
+  const request: RpcRequest = {
+    id: 'req-1',
+    authToken: 'tok',
+    method: 'terminal.multiplex',
+    params: {}
+  }
+  const dispatchPromise = dispatcher.dispatchStreaming(request, (msg) => messages.push(msg), {
+    connectionId: 'conn-transformed-output',
+    sendBinary: (bytes) => {
+      binaryFrames.push(bytes)
+      return true
+    },
+    registerBinaryStreamHandler: (streamId, handler) => {
+      handlers.set(streamId, handler)
+      return () => {
+        if (handlers.get(streamId) === handler) {
+          handlers.delete(streamId)
+        }
+      }
+    }
+  })
+
+  return {
+    messages,
+    binaryFrames,
+    handlers,
+    cleanups,
+    runtime,
+    dispatchPromise,
+    getDataListener: () => dataListener,
+    events: () => messages.map((message) => JSON.parse(message).result).filter(Boolean),
+    outputFrames: () =>
+      binaryFrames
+        .map(decodeTerminalStreamFrame)
+        .filter(
+          (frame) =>
+            frame?.opcode === TerminalStreamOpcode.Output ||
+            frame?.opcode === TerminalStreamOpcode.OutputSpan
+        )
+        .map((frame) => frame!)
+  }
+}
+
+async function subscribe(
+  harness: ReturnType<typeof startMultiplexHarness>,
+  capabilities: Record<string, 1>
+) {
+  await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
+  harness.handlers.get(0)?.(
+    decodeTerminalStreamFrame(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Subscribe,
+        streamId: 0,
+        seq: 1,
+        payload: encodeTerminalStreamJson({
+          streamId: STREAM_ID,
+          terminal: 'terminal-1',
+          client: { id: 'desktop-1', type: 'desktop' },
+          capabilities,
+          viewport: { cols: 120, rows: 40 }
+        })
+      })
+    )!
+  )
+  await vi.waitFor(() =>
+    expect(harness.events().some((event) => event?.type === 'subscribed')).toBe(true)
+  )
+  await vi.waitFor(() => expect(harness.getDataListener()).toBeDefined())
+  harness.binaryFrames.splice(0)
+  return harness.events().find((event) => event?.type === 'subscribed')
+}
+
+function sourceRange(displayStart: number, displayEnd: number, sourceStart = displayStart) {
+  const sourceEnd = sourceStart + (displayEnd - displayStart)
+  return {
+    id: 'pty-1',
+    spanId: `span-${displayStart}-${displayEnd}`,
+    providerGeneration: 5,
+    clientGeneration: 2,
+    ownerGeneration: 3,
+    ptyIncarnation: 'incarnation-1',
+    deliveryToken: 'token-1',
+    sourceStartSu: sourceStart,
+    sourceEndSu: sourceEnd,
+    displayStart,
+    displayEnd,
+    splittable: true,
+    transform: { transformed: false, rawLengthSu: sourceEnd - sourceStart, scalarSafe: true }
+  } as const
+}
+
+function renderedText(harness: ReturnType<typeof startMultiplexHarness>): string {
+  return harness
+    .outputFrames()
+    .map((frame) =>
+      frame.opcode === TerminalStreamOpcode.OutputSpan
+        ? ((decodeTerminalStreamJson<{ data?: string }>(frame.payload)?.data ?? '') as string)
+        : decodeTerminalStreamText(frame.payload)
+    )
+    .join('')
+}
+
+// Desktop's real multiplex capability set as shipped since 2026-07-29 (#11005).
+const SHIPPED_DESKTOP_CAPABILITIES = {
+  ackOutput: 1,
+  ackOutputSourceRanges: 1,
+  outputPause: 1,
+  writeUnavailable: 1,
+  desktopViewportClaims: 1
+} as const
+
+describe('terminal multiplex transformed output', () => {
+  // OutputSpan (opcode 15) shipped on this path in v1.4.147 (2026-07-19), before every
+  // capability a peer could declare to prove it decodes spans. Keying spans off any
+  // declared capability therefore reads the whole installed base as incapable.
+  it.each([
+    ['no capabilities at all', {}],
+    ['the pre-source-range capability set', { ackOutput: 1 } as Record<string, 1>],
+    ['the shipped desktop capability set', { ...SHIPPED_DESKTOP_CAPABILITIES }]
+  ])('sends OutputSpan for a transformed run with %s', async (_label, capabilities) => {
+    const harness = startMultiplexHarness()
+    await subscribe(harness, capabilities)
+    harness.getDataListener()!('xyz', { seq: 12, rawLength: 9, transformed: true })
+
+    await vi.waitFor(() =>
+      expect(harness.outputFrames().map((frame) => frame.opcode)).toContain(
+        TerminalStreamOpcode.OutputSpan
+      )
+    )
+    const span = harness
+      .outputFrames()
+      .find((frame) => frame.opcode === TerminalStreamOpcode.OutputSpan)!
+    expect(decodeTerminalStreamJson(span.payload)).toMatchObject({
+      data: 'xyz',
+      rawLength: 9,
+      transformed: true
+    })
+
+    harness.cleanups.get('terminal-multiplex:conn-transformed-output')?.()
+    await harness.dispatchPromise
+  })
+
+  it('does not head-block the ack queue behind an absorbed zero-byte emission', async () => {
+    const harness = startMultiplexHarness()
+    await subscribe(harness, { ...SHIPPED_DESKTOP_CAPABILITIES })
+    const emit = harness.getDataListener()!
+
+    // The absorbed-query shape: nine raw units consumed, nothing to display.
+    emit('', { seq: 9, rawLength: 9, transformed: true })
+    emit('after\r\n', { seq: 16, rawLength: 7 })
+
+    await vi.waitFor(() => expect(renderedText(harness)).toContain('after'))
+
+    harness.cleanups.get('terminal-multiplex:conn-transformed-output')?.()
+    await harness.dispatchPromise
+  })
+
+  it('keeps a transformed emission mappable so the client cannot see a false gap', async () => {
+    const harness = startMultiplexHarness()
+    await subscribe(harness, { ...SHIPPED_DESKTOP_CAPABILITIES })
+    const emit = harness.getDataListener()!
+
+    emit('abc', { seq: 3, rawLength: 3 })
+    await vi.waitFor(() => expect(renderedText(harness)).toContain('abc'))
+    // Three display units standing in for nine raw units.
+    emit('xyz', { seq: 12, rawLength: 9, transformed: true })
+    await vi.waitFor(() => expect(renderedText(harness)).toContain('xyz'))
+
+    // A client reconstructs the run start as `seq - rawLength`. On a plain Output
+    // frame it can only use `data.length`, so a downgraded transformed run reports
+    // a start past the previous high-water and reads as a dropped-frame gap.
+    const frames = harness.outputFrames()
+    let expectedSeq: number | undefined
+    for (const frame of frames) {
+      const seq = frame.seq > 0 ? frame.seq : undefined
+      const rawLength =
+        frame.opcode === TerminalStreamOpcode.OutputSpan
+          ? (decodeTerminalStreamJson<{ rawLength?: number }>(frame.payload)?.rawLength ?? 0)
+          : decodeTerminalStreamText(frame.payload).length
+      if (typeof seq === 'number' && typeof expectedSeq === 'number') {
+        expect(seq - rawLength).toBeLessThanOrEqual(expectedSeq)
+      }
+      if (typeof seq === 'number') {
+        expectedSeq = seq
+      }
+    }
+
+    harness.cleanups.get('terminal-multiplex:conn-transformed-output')?.()
+    await harness.dispatchPromise
+  })
+
+  it('keeps the stream attached for a transformed emission larger than one chunk', async () => {
+    const harness = startMultiplexHarness()
+    await subscribe(harness, { ...SHIPPED_DESKTOP_CAPABILITIES })
+    const emit = harness.getDataListener()!
+
+    emit('abc', { seq: 3, rawLength: 3, sourceRanges: [sourceRange(0, 3)] })
+    await vi.waitFor(() => expect(renderedText(harness)).toContain('abc'))
+
+    const large = 'L'.repeat(TERMINAL_STREAM_CHUNK_BYTES * 2 + 5)
+    emit(large, {
+      seq: 3 + large.length * 3,
+      rawLength: large.length * 3,
+      transformed: true,
+      sourceRanges: [sourceRange(3, 3 + large.length, 3)]
+    })
+
+    await vi.waitFor(() => expect(renderedText(harness).length).toBeGreaterThan(large.length))
+    expect(renderedText(harness)).toBe(`abc${large}`)
+    // A mapping-mode flip mid-stream detaches the stream and ends it.
+    expect(harness.events().some((event) => event?.type === 'end')).toBe(false)
+
+    harness.cleanups.get('terminal-multiplex:conn-transformed-output')?.()
+    await harness.dispatchPromise
+  })
+})

src/main/runtime/rpc/terminal-output-frame-chunks-equivalence.test.ts

@@ -149,7 +149,7 @@ function describeFrames(frames: Iterable<TerminalOutputFrameChunk>): FrameShape[
 function expectEquivalent(data: string, meta: TerminalOutputMeta | undefined, label: string): void {
   const legacy = describeFrames(legacyIterateTerminalOutputFrameChunks(data, meta))
   const next = describeFrames(
-    iterateTerminalOutputFrameChunks(data, meta, { supportsOutputSpan: true })
+    iterateTerminalOutputFrameChunks(data, meta, { transformedRuns: 'span' })
   )
   expect(next, label).toEqual(legacy)
 }
@@ -323,7 +323,7 @@ describe('iterateTerminalOutputFrameChunks equivalence with the pre-optimization
     const meta = seqPreservingMeta(data, seq)
 
     expectEquivalent(data, meta, 'unsafe seq rounding')
-    const frames = [...iterateTerminalOutputFrameChunks(data, meta, { supportsOutputSpan: true })]
+    const frames = [...iterateTerminalOutputFrameChunks(data, meta, { transformedRuns: 'span' })]
     expect(frames.at(-1)?.seq).toBe(9_007_199_254_740_992)
   })
 
@@ -502,7 +502,7 @@ describe('iterateTerminalOutputFrameChunks equivalence with the pre-optimization
     const data = `${'a'.repeat(200 * 1024)}${SURROGATE_PAIR.repeat(4096)}${LONE_HIGH}`
     const frames = [
       ...iterateTerminalOutputFrameChunks(data, seqPreservingMeta(data, 999_999), {
-        supportsOutputSpan: true
+        transformedRuns: 'span'
       })
     ]
     expect(frames.length).toBeGreaterThan(4)
@@ -523,22 +523,23 @@ describe('iterateTerminalOutputFrameChunks equivalence with the pre-optimization
     const exact = '\u20ac'.repeat(TERMINAL_STREAM_CHUNK_BYTES / 3)
     expect(Buffer.byteLength(exact, 'utf8')).toBe(TERMINAL_STREAM_CHUNK_BYTES)
     expect([
-      ...iterateTerminalOutputFrameChunks(exact, undefined, { supportsOutputSpan: true })
+      ...iterateTerminalOutputFrameChunks(exact, undefined, { transformedRuns: 'span' })
     ]).toHaveLength(1)
     expect([...legacyIterateTerminalOutputFrameChunks(exact)]).toHaveLength(1)
     const overByOne = `${exact}a`
     expect(
-      [...iterateTerminalOutputFrameChunks(overByOne, undefined, { supportsOutputSpan: true })]
+      [...iterateTerminalOutputFrameChunks(overByOne, undefined, { transformedRuns: 'span' })]
         .length
     ).toBeGreaterThan(1)
     expect([...legacyIterateTerminalOutputFrameChunks(overByOne)].length).toBeGreaterThan(1)
   })
 
-  // The chunking loop is only reached when rawLength === data.length and transformed
-  // is falsy, which makes canPreserveChunkSeq === (typeof meta.seq === 'number') and
-  // therefore shouldDelayFinalSeq unconditionally false. The branch survives here
-  // (it also survived in the pre-optimization code) purely as defence in depth.
-  it('never reaches the delayed-final-seq branch for any meta shape', () => {
+  // Under 'span' the chunking loop is only reached when rawLength === data.length and
+  // transformed is falsy, which makes canPreserveChunkSeq === (typeof meta.seq === 'number')
+  // and therefore shouldDelayFinalSeq unconditionally false. Under 'downgrade-to-text' the
+  // span branch is skipped, so the same meta shapes DO reach it — that is the whole
+  // fallback path, and the block below is its only coverage.
+  it('never reaches the delayed-final-seq branch under span framing', () => {
     for (const data of ['', 'a', 'abc', 'x'.repeat(200)]) {
       for (const seq of [undefined, 0, 5] as (number | undefined)[]) {
         for (const rawDelta of [undefined, 0, 1, -1] as (number | undefined)[]) {
@@ -564,3 +565,150 @@ describe('iterateTerminalOutputFrameChunks equivalence with the pre-optimization
     }
   })
 })
+
+/**
+ * The 'downgrade-to-text' contract, asserted directly because the legacy reference has no
+ * downgrade mode to compare against. Three properties are load-bearing for the mobile
+ * `terminal.subscribe` client:
+ *   1. every display code unit arrives, in order (a dropped yield is silent data loss);
+ *   2. no frame exceeds the wire cap;
+ *   3. no frame except the last claims a seq, and the last claims exactly `meta.seq`,
+ *      because a transformed run's seq cannot map back to UTF-16 chunk offsets.
+ */
+function expectDowngradeContract(
+  data: string,
+  meta: TerminalOutputMeta | undefined,
+  label: string
+): void {
+  const frames = [
+    ...iterateTerminalOutputFrameChunks(data, meta, { transformedRuns: 'downgrade-to-text' })
+  ]
+  for (const frame of frames) {
+    expect(frame.opcode, `${label}: downgraded frames are plain Output`).toBeUndefined()
+    expect(frame.bytes.byteLength, `${label}: frame within wire cap`).toBeLessThanOrEqual(
+      TERMINAL_STREAM_CHUNK_BYTES
+    )
+  }
+  const rejoined = Buffer.concat(frames.map((frame) => Buffer.from(frame.bytes))).toString('utf8')
+  expect(rejoined, `${label}: every display code unit survives`).toBe(
+    Buffer.from(new TextEncoder().encode(data)).toString('utf8')
+  )
+  expect(
+    frames.reduce((total, frame) => total + frame.displayLength, 0),
+    `${label}: displayLength totals the payload`
+  ).toBe(data.length)
+  const seqs = frames.map((frame) => frame.seq)
+  expect(
+    seqs.slice(0, -1).filter((seq) => seq !== undefined),
+    `${label}: no early seq`
+  ).toEqual([])
+  expect(seqs.at(-1), `${label}: final frame carries the raw high-water mark`).toBe(meta?.seq)
+}
+
+describe('iterateTerminalOutputFrameChunks downgrade-to-text framing', () => {
+  it('never emits OutputSpan, whatever the meta says', () => {
+    for (const meta of metaShapesFor('transformed payload')) {
+      const frames = [
+        ...iterateTerminalOutputFrameChunks('transformed payload', meta.meta, {
+          transformedRuns: 'downgrade-to-text'
+        })
+      ]
+      expect(
+        frames.map((frame) => frame.opcode),
+        meta.label
+      ).toEqual(frames.map(() => undefined))
+    }
+  })
+
+  it('emits a single empty frame for an absorbed zero-byte transformed run', () => {
+    const frames = [
+      ...iterateTerminalOutputFrameChunks(
+        '',
+        { seq: 9, rawLength: 9, transformed: true },
+        { transformedRuns: 'downgrade-to-text' }
+      )
+    ]
+    expect(frames).toHaveLength(1)
+    expect(frames[0]?.opcode).toBeUndefined()
+    expect(frames[0]?.bytes.byteLength).toBe(0)
+    expect(frames[0]?.displayLength).toBe(0)
+    expect(frames[0]?.seq).toBe(9)
+  })
+
+  it('splits a transformed run across chunks without losing or reordering text', () => {
+    for (const chunkCount of [1, 2, 3, 5, 9]) {
+      const data = `${'m'.repeat(TERMINAL_STREAM_CHUNK_BYTES * chunkCount)}${SURROGATE_PAIR}tail`
+      const meta: TerminalOutputMeta = {
+        seq: 3 * data.length,
+        rawLength: 3 * data.length,
+        transformed: true
+      }
+      expectDowngradeContract(data, meta, `downgrade chunks=${chunkCount}`)
+      // Anti-vacuous: the split arm ran, not the single-frame arm.
+      expect(
+        [...iterateTerminalOutputFrameChunks(data, meta, { transformedRuns: 'downgrade-to-text' })]
+          .length
+      ).toBeGreaterThan(chunkCount)
+    }
+  })
+
+  it('holds the contract across UTF-8 widths, lone surrogates and cap boundaries', () => {
+    const units = ['a', 'é', '€', SURROGATE_PAIR, LONE_HIGH, LONE_LOW]
+    for (const unit of units) {
+      for (const codeUnits of [1, 1024, TERMINAL_STREAM_CHUNK_BYTES, 90 * 1024]) {
+        const data = repeatToLength(unit, codeUnits)
+        for (const meta of [
+          { seq: 7 * data.length, rawLength: 7 * data.length, transformed: true },
+          { seq: 5, rawLength: data.length + 11 },
+          { rawLength: data.length + 3 },
+          undefined
+        ] as (TerminalOutputMeta | undefined)[]) {
+          expectDowngradeContract(data, meta, `unit=${escapeUnits(unit)} units=${codeUnits}`)
+        }
+      }
+    }
+    for (let offset = -4; offset <= 4; offset += 1) {
+      const data = `${'a'.repeat(TERMINAL_STREAM_CHUNK_BYTES + offset)}${SURROGATE_PAIR}${'b'.repeat(64)}`
+      expectDowngradeContract(
+        data,
+        { seq: 2 * data.length, rawLength: 2 * data.length, transformed: true },
+        `cap boundary offset=${offset}`
+      )
+    }
+  })
+
+  it('fuzzes 600 near-cap transformed payloads', () => {
+    const random = makeRandom(0x0d09_6ade)
+    for (let trial = 0; trial < 600; trial += 1) {
+      const fillerLength = TERMINAL_STREAM_CHUNK_BYTES - 6 + Math.floor(random() * 12)
+      const data = 'q'.repeat(fillerLength) + randomText(random, 1 + Math.floor(random() * 24))
+      expectDowngradeContract(
+        data,
+        { seq: 4 * data.length, rawLength: 4 * data.length, transformed: true },
+        `fuzz-downgrade trial=${trial}`
+      )
+    }
+  }, 30_000)
+
+  it('matches span framing byte for byte when no run is transformed', () => {
+    // Same input, both modes: only a transformed run may frame differently, so an
+    // accidental behaviour change in the shared chunk loop shows up here.
+    for (const data of [
+      '',
+      'a',
+      'x'.repeat(200 * 1024),
+      repeatToLength(SURROGATE_PAIR, 60 * 1024)
+    ]) {
+      for (const meta of [undefined, { seq: 42, rawLength: data.length }]) {
+        expect(
+          describeFrames(
+            iterateTerminalOutputFrameChunks(data, meta, { transformedRuns: 'downgrade-to-text' })
+          ),
+          `len=${data.length} seq=${meta?.seq}`
+        ).toEqual(
+          describeFrames(iterateTerminalOutputFrameChunks(data, meta, { transformedRuns: 'span' }))
+        )
+      }
+    }
+  })
+})

src/main/runtime/rpc/terminal-output-frame-chunks.ts

@@ -51,9 +51,18 @@ export function exceedsTerminalStreamChunkBytes(data: string): boolean {
 }
 
 export type TerminalOutputFrameChunkOptions = {
-  // Required, not defaulted: the caller owns the peer's negotiated capability, and a
-  // default would silently re-introduce ungated spans at the next call site.
-  supportsOutputSpan: boolean
+  /**
+   * How to frame a transformed run, where `rawLength !== data.length`.
+   *
+   * `'downgrade-to-text'` folds the run into plain `Output` frames for a peer that cannot
+   * decode `OutputSpan`. That is lossy by construction: an `Output` frame carries only
+   * `(data, seq)`, so the receiver can only derive the run start as `seq - data.length`.
+   * It is therefore legal ONLY for a peer that ignores `seq` outright — today just the
+   * mobile `terminal.subscribe` client. A seq-tracking peer must always get `'span'`.
+   *
+   * Required, not defaulted: the choice is a wire-compatibility decision per call site.
+   */
+  transformedRuns: 'span' | 'downgrade-to-text'
 }
 
 export function* iterateTerminalOutputFrameChunks(
@@ -62,7 +71,7 @@ export function* iterateTerminalOutputFrameChunks(
   options: TerminalOutputFrameChunkOptions
 ): Generator<TerminalOutputFrameChunk> {
   const rawLength = meta?.rawLength ?? data.length
-  if (options.supportsOutputSpan && (meta?.transformed || rawLength !== data.length)) {
+  if (options.transformedRuns === 'span' && (meta?.transformed || rawLength !== data.length)) {
     yield {
       opcode: TerminalStreamOpcode.OutputSpan,
       bytes: encodeTerminalStreamJson({ data, rawLength, transformed: true }),
@@ -82,9 +91,8 @@ export function* iterateTerminalOutputFrameChunks(
     return
   }
   const canPreserveChunkSeq = typeof meta?.seq === 'number' && rawLength === data.length
-  // This is the downgrade path for a peer without `outputSpan`: it carries the raw
-  // high-water mark on the final frame only, because a transformed run's seq cannot
-  // map back to UTF-16 chunk offsets. Matches what hosts sent before OutputSpan existed.
+  // Only reachable under `'downgrade-to-text'`: a transformed run's seq is a raw high-water
+  // mark that cannot map back to UTF-16 chunk offsets, so no intermediate frame may claim one.
   const shouldDelayFinalSeq = !canPreserveChunkSeq && typeof meta?.seq === 'number'
   const startSeq = canPreserveChunkSeq ? meta.seq! - rawLength : undefined
   let chunkStart = 0

src/main/runtime/rpc/terminal-output-frame-source-ranges.test.ts

@@ -55,7 +55,7 @@ describe('terminal output frame source ranges', () => {
             })
           ]
         },
-        { supportsOutputSpan: true }
+        { transformedRuns: 'span' }
       )
     )
 
@@ -96,7 +96,7 @@ describe('terminal output frame source ranges', () => {
         transformed: true,
         sourceRanges: [sourceRange]
       },
-      { supportsOutputSpan: true }
+      { transformedRuns: 'span' }
     )
 
     expect(frame?.sourceRanges).toEqual([sourceRange])

src/main/runtime/rpc/terminal-subscribe-output-span-capability.test.ts

@@ -5,10 +5,9 @@ import type { RuntimeTerminalWait } from '../../../shared/runtime-types'
 import {
   TerminalStreamOpcode,
   decodeTerminalStreamFrame,
-  decodeTerminalStreamText,
-  encodeTerminalStreamFrame,
-  encodeTerminalStreamJson
+  decodeTerminalStreamText
 } from '../../../shared/terminal-stream-protocol'
+import { TERMINAL_STREAM_CHUNK_BYTES } from '../../../shared/terminal-multiplex-flow-control'
 import type { OrcaRuntimeService } from '../orca-runtime'
 import type { RpcRequest } from './core'
 import { RpcDispatcher } from './dispatcher'
@@ -28,9 +27,18 @@ type LegacyStreamHarness = {
   cleanup: () => void
 }
 
-// Why parsed from mobile's source instead of hard-coded: mobile vendors its own
-// copy of the opcode table (separate pnpm workspace, so it cannot be imported).
-// Reading the real file is what makes this test fail when the two drift.
+function sharedOpcodes(): Set<number> {
+  return new Set(
+    Object.values(TerminalStreamOpcode).filter(
+      (value): value is number => typeof value === 'number'
+    )
+  )
+}
+
+// Why parsed from mobile's source instead of hard-coded: mobile is a separate pnpm
+// workspace, so its transport module cannot be imported from here. Reading the real
+// file is what makes this test fail when the two tables drift. A mobile build that
+// re-exports the shared table cannot drift at all, so it inherits every opcode.
 function mobileDecodableOpcodes(): Set<number> {
   const source = readFileSync(
     join(process.cwd(), 'mobile/src/transport/terminal-stream-protocol.ts'),
@@ -38,7 +46,10 @@ function mobileDecodableOpcodes(): Set<number> {
   )
   const body = /enum TerminalStreamOpcode \{([\s\S]*?)\}/.exec(source)?.[1]
   if (!body) {
-    throw new Error('mobile vendored TerminalStreamOpcode enum not found')
+    if (/TerminalStreamOpcode[\s\S]*?['"][^'"]*shared\/terminal-stream-protocol['"]/.test(source)) {
+      return sharedOpcodes()
+    }
+    throw new Error('mobile TerminalStreamOpcode is neither a vendored enum nor a shared re-export')
   }
   const opcodes = new Set<number>()
   for (const match of body.matchAll(/^\s*\w+\s*=\s*(\d+)/gm)) {
@@ -190,142 +201,59 @@ describe('terminal.subscribe OutputSpan capability gate', () => {
     capable.cleanup()
     await capable.dispatchPromise
   })
-})
 
-async function multiplexStream(capabilities: Record<string, 1>): Promise<{
-  binaryFrames: Uint8Array<ArrayBufferLike>[]
-  emit: (data: string, meta?: TerminalDataMeta) => void
-  cleanup: () => void
-  dispatchPromise: Promise<unknown>
-}> {
-  const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
-  const cleanups = new Map<string, () => void>()
-  const handlers = new Map<
-    number,
-    (frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
-  >()
-  let onData: ((data: string, meta?: TerminalDataMeta) => void) | undefined
-  const runtime = {
-    getRuntimeId: () => 'test-runtime',
-    // The multiplex subscribe path resolves handles via resolveLiveLeafForHandle.
-    resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-mux-span' }),
-    resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-mux-span' }),
-    requestRendererTerminalTabMount: vi.fn().mockReturnValue(true),
-    updateRemoteDesktopViewer: vi.fn().mockResolvedValue(true),
-    unregisterRemoteDesktopViewers: vi.fn().mockResolvedValue(true),
-    isPtyResizeDrivenRemotely: vi.fn().mockReturnValue(false),
-    getRemoteDesktopFitHold: vi.fn().mockReturnValue({ mode: 'desktop-fit', cols: 120, rows: 40 }),
-    isRemoteDesktopViewerOwner: vi.fn().mockReturnValue(false),
-    readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
-    serializeAuthoritativeTerminalBuffer: vi
-      .fn()
-      .mockResolvedValue({ data: 'snap', cols: 120, rows: 40 }),
-    serializeTerminalBuffer: vi.fn().mockResolvedValue({ data: 'snap', cols: 120, rows: 40 }),
-    getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
-    getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
-    getLayout: vi.fn().mockReturnValue({ seq: 1 }),
-    getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
-    getTerminalFitOverride: vi.fn().mockReturnValue(null),
-    getPtyOutputSequence: vi.fn(() => 0),
-    subscribeToTerminalData: vi.fn((_ptyId: string, listener: typeof onData) => {
-      onData = listener
-      return vi.fn()
-    }),
-    registerRemoteTerminalViewSubscriber: vi.fn(() => vi.fn()),
-    unregisterRemoteDesktopViewer: vi.fn(),
-    subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
-    subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
-    subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()),
-    updateDesktopViewport: vi.fn().mockResolvedValue(true),
-    registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
-      cleanups.set(id, cleanup)
-    }),
-    cleanupSubscription: vi.fn((id: string) => cleanups.get(id)?.()),
-    waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
-  } as unknown as OrcaRuntimeService
-  const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
-  const request: RpcRequest = {
-    id: 'req-mux-span',
-    authToken: 'tok',
-    method: 'terminal.multiplex',
-    params: {}
-  }
-  const dispatchPromise = dispatcher.dispatchStreaming(request, vi.fn(), {
-    connectionId: 'conn-mux-span',
-    sendBinary: (bytes) => {
-      binaryFrames.push(bytes)
-    },
-    registerBinaryStreamHandler: (streamId, handler) => {
-      handlers.set(streamId, handler)
-      return () => handlers.delete(streamId)
-    }
-  })
-  await vi.waitFor(() => expect(handlers.has(0)).toBe(true))
-  handlers.get(0)!(
-    decodeTerminalStreamFrame(
-      encodeTerminalStreamFrame({
-        opcode: TerminalStreamOpcode.Subscribe,
-        streamId: 0,
-        seq: 1,
-        payload: encodeTerminalStreamJson({
-          streamId: 7,
-          terminal: 'terminal-span',
-          client: { id: 'desktop-mux', type: 'desktop' },
-          capabilities,
-          viewport: { cols: 120, rows: 40 }
-        })
-      })
-    )!
-  )
-  await vi.waitFor(() => expect(onData).toBeTypeOf('function'))
-  binaryFrames.splice(0)
-  return {
-    binaryFrames,
-    emit: (data, meta) => onData?.(data, meta),
-    cleanup: () => runtime.cleanupSubscription('terminal-multiplex:conn-mux-span'),
-    dispatchPromise
-  }
-}
-
-describe('terminal.multiplex OutputSpan capability gate', () => {
-  it('downgrades to Output for a desktop build that predates the capability', async () => {
-    const old = await multiplexStream({ ackOutput: 1, desktopViewportClaims: 1 })
-
-    old.emit('visible output', TRANSFORMED_META)
-    await vi.waitFor(() => expect(outputTextIn(old.binaryFrames)).toBe('visible output'))
-
-    const opcodes = old.binaryFrames.map((bytes) => decodeTerminalStreamFrame(bytes)!.opcode)
-    expect(opcodes).not.toContain(TerminalStreamOpcode.OutputSpan)
-
-    old.cleanup()
-    await old.dispatchPromise
-  })
-
-  it('keeps sending OutputSpan to a desktop build that negotiates it', async () => {
-    const current = await multiplexStream({
-      ackOutput: 1,
-      desktopViewportClaims: 1,
-      outputSpan: 1
-    })
-
-    current.emit('visible output', TRANSFORMED_META)
-    await vi.waitFor(() =>
-      expect(
-        current.binaryFrames.some(
-          (bytes) => decodeTerminalStreamFrame(bytes)?.opcode === TerminalStreamOpcode.OutputSpan
-        )
-      ).toBe(true)
+  // The downgrade's multi-frame arm. Every fixture in this suite used to be a
+  // 14-character string, so the chunk loop never ran and a deleted yield here —
+  // 48 KiB of silently dropped user output — passed every test in the repo.
+  it('delivers a transformed emission larger than one chunk, whole and in order', async () => {
+    const legacy = await subscribeLegacyBinaryStream(
+      { id: 'mobile-large', type: 'mobile' },
+      { terminalBinaryStream: 1 }
     )
 
-    current.cleanup()
-    await current.dispatchPromise
+    const data = `${'a'.repeat(TERMINAL_STREAM_CHUNK_BYTES)}MIDDLE${'b'.repeat(TERMINAL_STREAM_CHUNK_BYTES)}TAIL`
+    legacy.emit(data, { seq: data.length * 3, rawLength: data.length * 3, transformed: true })
+    await vi.waitFor(() => expect(outputTextIn(legacy.binaryFrames)).toHaveLength(data.length))
+
+    // Anti-vacuous: prove the multi-chunk arm ran rather than the single-frame arm.
+    const outputs = legacy.binaryFrames
+      .map((bytes) => decodeTerminalStreamFrame(bytes))
+      .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
+    expect(outputs.length).toBeGreaterThan(2)
+    for (const frame of outputs) {
+      expect(frame!.payload.byteLength).toBeLessThanOrEqual(TERMINAL_STREAM_CHUNK_BYTES)
+    }
+    expect(outputTextIn(legacy.binaryFrames)).toBe(data)
+    // Only the last frame may carry the raw high-water mark; the others cannot map it.
+    expect(outputs.at(-1)!.seq).toBe(data.length * 3)
+
+    legacy.cleanup()
+    await legacy.dispatchPromise
+  })
+
+  it('keeps delivering after an absorbed zero-byte transformed emission', async () => {
+    const legacy = await subscribeLegacyBinaryStream(
+      { id: 'mobile-absorbed', type: 'mobile' },
+      { terminalBinaryStream: 1 }
+    )
+
+    // The absorbed-query shape: raw units consumed, nothing to display.
+    legacy.emit('', { seq: 9, rawLength: 9, transformed: true })
+    legacy.emit('after', { seq: 14, rawLength: 5 })
+    await vi.waitFor(() => expect(outputTextIn(legacy.binaryFrames)).toBe('after'))
+
+    const opcodes = legacy.binaryFrames.map((bytes) => decodeTerminalStreamFrame(bytes)!.opcode)
+    expect(opcodes).not.toContain(TerminalStreamOpcode.OutputSpan)
+
+    legacy.cleanup()
+    await legacy.dispatchPromise
   })
 
   it('forces a mobile decision for every opcode in the shared table', () => {
-    // The drift guard. mobile/ is a separate workspace with a hand-copied enum, so
-    // adding an opcode to the shared table changes nothing on the phone and nothing
-    // fails. Each opcode must be classified here on purpose; a new one is neither
-    // "decodable" nor "gated" until someone says which, and this test says so.
+    // The drift guard. While mobile hand-copies the enum, adding an opcode to the
+    // shared table changes nothing on the phone and nothing fails. Each opcode must be
+    // classified on purpose; a new one is neither "decodable" nor "gated" until someone
+    // says which, and this test is where they say it.
     const decodableByMobile = mobileDecodableOpcodes()
     // Opcodes a host may never send to a mobile stream unless it negotiated them,
     // or that only ever travel client -> host.
@@ -348,15 +276,4 @@ describe('terminal.multiplex OutputSpan capability gate', () => {
       .filter((opcode) => !notSentUnnegotiatedToMobile.has(opcode))
     expect(unclassified).toEqual([])
   })
-
-  it('has the shipping desktop multiplexer declare the capability', () => {
-    // Guards the other half of the gate: if the renderer stops advertising
-    // outputSpan, the host silently downgrades every desktop stream.
-    const source = readFileSync(
-      join(process.cwd(), 'src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts'),
-      'utf8'
-    )
-    const capabilities = /capabilities: \{([\s\S]*?)\}/.exec(source)?.[1]
-    expect(capabilities).toContain('outputSpan: 1')
-  })
 })

src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.test.ts

@@ -278,7 +278,6 @@ describe('createRemoteRuntimePtyTransport', () => {
         ackOutputSourceRanges: 1,
         desktopViewportClaims: 1,
         outputPause: 1,
-        outputSpan: 1,
         writeUnavailable: 1
       })
     )

src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts

@@ -507,7 +507,6 @@ class RemoteRuntimeTerminalMultiplexer {
             ackOutput: 1,
             ackOutputSourceRanges: 1,
             outputPause: 1,
-            outputSpan: 1,
             writeUnavailable: 1,
             ...(args.client.type === 'desktop' ? { desktopViewportClaims: 1 } : {})
           }

src/renderer/src/runtime/runtime-terminal-stream.test.ts

@@ -107,7 +107,6 @@ describe('remote runtime terminal data subscriptions', () => {
           ackOutputSourceRanges?: 1
           desktopViewportClaims?: 1
           outputPause?: 1
-          outputSpan?: 1
           writeUnavailable?: 1
         }
       }>(subscribeFrame.payload)
@@ -117,7 +116,6 @@ describe('remote runtime terminal data subscriptions', () => {
       ackOutputSourceRanges: 1,
       desktopViewportClaims: 1,
       outputPause: 1,
-      outputSpan: 1,
       writeUnavailable: 1
     })
 

src/shared/terminal-stream-protocol.ts

@@ -28,9 +28,9 @@ export enum TerminalStreamOpcode {
   // Why 14: Ack already occupies 13 on current clients; older runtimes ignore
   // this opcode and still receive the compatibility Resize frame behind it.
   ClaimViewport = 14,
-  // Negotiated per stream: mobile's vendored decoder stops at 12 and drops this
-  // frame silently, so an ungated span is lost output. Senders downgrade to
-  // Output when the peer did not confirm `outputSpan`.
+  // Unconditional on `terminal.multiplex`; negotiated on `terminal.subscribe`, whose
+  // mobile client vendors a decoder that stops at 12 and drops this frame silently.
+  // Only the subscribe path may downgrade to Output, because only mobile ignores `seq`.
   OutputSpan = 15,
   // Negotiated per stream; older hosts reject unknown opcodes, so clients send only after capability confirmation.
   SetOutputPaused = 16,

tests/e2e/cross-version-wire/cross-version-terminal-wire.unit.test.ts

@@ -27,18 +27,28 @@ const EXPECTED_JOURNEY_FRAMES = [
   'H>C SnapshotEnd',
   'C>H Input',
   'H>C Output',
+  // A transformed run (rawLength !== data.length) must travel as OutputSpan on multiplex.
+  // If this ever reads `H>C Output`, a host downgraded a run it cannot losslessly downgrade.
+  'H>C OutputSpan',
+  'H>C OutputSpan',
   'C>H SnapshotRequest',
+  // Two chunks from here on: the large transformed run pushed the scrollback past one.
   'H>C SnapshotStart',
   'H>C SnapshotChunk',
+  'H>C SnapshotChunk',
   'H>C SnapshotEnd',
   'C>H Subscribe',
   'H>C SnapshotStart',
   'H>C SnapshotChunk',
+  'H>C SnapshotChunk',
   'H>C SnapshotEnd',
   'C>H Input',
   'C>H Unsubscribe'
 ]
 
+// Stable across every build pairing in this suite; the codec never renumbers an opcode.
+const SNAPSHOT_REQUEST_OPCODE = 11
+
 let baselineRef: string
 let current: TerminalWireBuild
 let baseline: TerminalWireBuild
@@ -81,10 +91,36 @@ function expectWireCompatible(record: JourneyRecord): void {
 
   // Rule 3 — what the host publishes, as the client actually rendered it.
   expect(record.snapshotsRendered[0]).toBe(JOURNEY_INPUTS.initialBuffer)
-  expect(record.dataRendered.join('')).toBe(JOURNEY_INPUTS.output)
-  expect(record.revealSnapshot?.data).toBe(
-    `${JOURNEY_INPUTS.initialBuffer}${JOURNEY_INPUTS.output}`
+  expect(record.dataRendered.join('')).toBe(
+    `${JOURNEY_INPUTS.output}${JOURNEY_INPUTS.transformed}${JOURNEY_INPUTS.largeTransformed}`
   )
+  expect(record.revealSnapshot?.data).toBe(
+    `${JOURNEY_INPUTS.initialBuffer}${JOURNEY_INPUTS.output}${JOURNEY_INPUTS.transformed}${JOURNEY_INPUTS.largeTransformed}`
+  )
+
+  // The transformed runs arrived whole and carried their RAW length, not their display
+  // length. A downgraded run would report `rawLength === length`, which makes the client
+  // reconstruct a start past its high-water mark and read a dropped-frame gap.
+  const transformed = record.dataMeta.filter((entry) => entry.transformed)
+  expect(transformed).toEqual([
+    {
+      length: JOURNEY_INPUTS.transformed.length,
+      rawLength: JOURNEY_INPUTS.transformedRawLength,
+      transformed: true
+    },
+    {
+      length: JOURNEY_INPUTS.largeTransformed.length,
+      rawLength: JOURNEY_INPUTS.largeTransformedRawLength,
+      transformed: true
+    }
+  ])
+  // A false gap self-heals by pulling a fresh snapshot, which would show up as an extra
+  // client -> host SnapshotRequest. Only the scripted hide/reveal may issue one.
+  expect(
+    record.observed.filter(
+      (frame) => frame.direction === 'client-to-host' && frame.opcode === SNAPSHOT_REQUEST_OPCODE
+    )
+  ).toHaveLength(1)
   expect(record.revealSnapshot).toMatchObject({ cols: 120, rows: 40 })
   for (const start of record.snapshotStarts) {
     expect(start).toMatchObject({ kind: 'scrollback', cols: 120, rows: 40, source: 'headless' })

tests/e2e/cross-version-wire/host-terminal-runtime-stub.ts

@@ -1,6 +1,7 @@
 export type HostTerminalDataMeta = {
   seq?: number
   rawLength?: number
+  transformed?: boolean
   cwd?: string
 }
 
@@ -22,6 +23,14 @@ export type HostTerminalRuntimeStub = {
   serializeCount: number
   /** Push PTY output to every host-side data listener. */
   emitOutput: (data: string, meta?: HostTerminalDataMeta) => void
+  /**
+   * Push a transformed run: `display` is what the user should see, `rawLength` the
+   * larger raw code-unit run it stands for (an SSH relay stripping OSC sequences, or
+   * a query absorbed with nothing left to display). This is the only emission shape
+   * whose `seq` cannot be recovered from the display text, so it is the one that
+   * decides whether a peer may be sent OutputSpan or plain Output.
+   */
+  emitTransformedOutput: (display: string, rawLength: number) => void
   /** Names of runtime methods the host called that the stub does not implement. */
   missingRuntimeMethods: string[]
   /** Run the host's registered teardown for one connection, as a socket close does. */
@@ -51,6 +60,7 @@ export function createHostTerminalRuntimeStub(
     buffer: options.initialBuffer ?? '',
     serializeCount: 0,
     emitOutput: () => {},
+    emitTransformedOutput: () => {},
     missingRuntimeMethods: [],
     closeConnection: () => {}
   }
@@ -83,6 +93,16 @@ export function createHostTerminalRuntimeStub(
     }
   }
 
+  stub.emitTransformedOutput = (display, rawLength) => {
+    stub.buffer += display
+    // The high-water mark advances by the RAW run, which is exactly why a plain
+    // Output frame cannot carry it: the receiver only sees `display`.
+    outputSequence += rawLength
+    for (const listener of Array.from(dataListeners)) {
+      listener(display, { seq: outputSequence, rawLength, transformed: true })
+    }
+  }
+
   const serialize = async (): Promise<{
     data: string
     cols: number

tests/e2e/cross-version-wire/terminal-skew-journey.ts

@@ -15,6 +15,8 @@ export const JOURNEY_STEPS = [
   'first-snapshot',
   'input-reaches-process',
   'live-output',
+  'transformed-output',
+  'large-transformed-output',
   'reveal-snapshot',
   'transport-drop',
   'resubscribe',
@@ -28,6 +30,15 @@ const FIRST_INPUT = 'echo cross-version\r'
 const SECOND_INPUT = 'echo after-reconnect\r'
 const LIVE_OUTPUT = 'cross-version live output\r\n'
 const INITIAL_BUFFER = 'initial scrollback\r\n'
+// A transformed run: the host consumed TRANSFORMED_RAW_LENGTH raw code units and has
+// this much left to display. `rawLength !== display.length` is the shape that decides
+// OutputSpan vs plain Output, and it had no cross-version coverage before STA-3482.
+const TRANSFORMED_OUTPUT = 'transformed run\r\n'
+const TRANSFORMED_RAW_LENGTH = TRANSFORMED_OUTPUT.length + 31
+// Past TERMINAL_STREAM_CHUNK_BYTES (48 KiB) so the transformed run must be split; the
+// multi-frame arm is where a dropped yield loses a whole chunk of user output.
+const LARGE_TRANSFORMED_OUTPUT = `HEAD${'x'.repeat(64 * 1024)}TAIL\r\n`
+const LARGE_TRANSFORMED_RAW_LENGTH = LARGE_TRANSFORMED_OUTPUT.length + 97
 const BARRIER_TIMEOUT_MS = 10_000
 
 export type JourneyRecord = {
@@ -45,6 +56,8 @@ export type JourneyRecord = {
   snapshotsRendered: string[]
   /** Live output the client's pane received. */
   dataRendered: string[]
+  /** Per-delivery `(rawLength, transformed)` as the CLIENT reconstructed them. */
+  dataMeta: { length: number; rawLength?: number; transformed?: boolean }[]
   /** Exact texts the host wrote to the PTY. */
   inputAtProcess: string[]
   /** Snapshot the reveal step resolved with. */
@@ -104,6 +117,7 @@ export async function runTerminalSkewJourney(args: {
     snapshotStarts: [],
     snapshotsRendered: [],
     dataRendered: [],
+    dataMeta: [],
     inputAtProcess: hostStub.writtenInput,
     revealSnapshot: null,
     transportCloses: 0,
@@ -139,8 +153,13 @@ export async function runTerminalSkewJourney(args: {
 
   let subscribedCount = 0
   const callbacks = {
-    onData: (data: string) => {
+    onData: (data: string, meta?: { rawLength?: number; transformed?: boolean }) => {
       record.dataRendered.push(data)
+      record.dataMeta.push({
+        length: data.length,
+        rawLength: meta?.rawLength,
+        transformed: meta?.transformed
+      })
     },
     onSnapshot: (data: string) => {
       record.snapshotsRendered.push(data)
@@ -192,6 +211,19 @@ export async function runTerminalSkewJourney(args: {
     )
     record.completed.push('live-output')
 
+    hostStub.emitTransformedOutput(TRANSFORMED_OUTPUT, TRANSFORMED_RAW_LENGTH)
+    await barrier('transformed-output: client never rendered the transformed run', () =>
+      record.dataRendered.join('').includes(TRANSFORMED_OUTPUT.trim())
+    )
+    record.completed.push('transformed-output')
+
+    hostStub.emitTransformedOutput(LARGE_TRANSFORMED_OUTPUT, LARGE_TRANSFORMED_RAW_LENGTH)
+    await barrier(
+      'large-transformed-output: client never rendered the full multi-chunk transformed run',
+      () => record.dataRendered.join('').includes(LARGE_TRANSFORMED_OUTPUT)
+    )
+    record.completed.push('large-transformed-output')
+
     // Hide/reveal: the pane drops xterm and asks the host to re-publish the buffer.
     const revealed = await terminal.serializeBuffer({ scrollbackRows: 200 })
     if (!revealed) {
@@ -239,5 +271,9 @@ export const JOURNEY_INPUTS = {
   first: FIRST_INPUT,
   second: SECOND_INPUT,
   output: LIVE_OUTPUT,
+  transformed: TRANSFORMED_OUTPUT,
+  transformedRawLength: TRANSFORMED_RAW_LENGTH,
+  largeTransformed: LARGE_TRANSFORMED_OUTPUT,
+  largeTransformedRawLength: LARGE_TRANSFORMED_RAW_LENGTH,
   initialBuffer: INITIAL_BUFFER
 }