fix(agents): exclude pre-paste snapshot from remote echo verification

On remote/SSH runtimes the echo watch subscribed with onSnapshot wired to
the same watcher, so the scanner's first input was the terminal buffer
snapshot from before the paste. A short prompt's head sample (or a stale
paste placeholder) could appear in that pre-paste screen and falsely verify
delivery onto a swallow screen — the local sidecar never sees pre-paste
output, so this was a remote-only reopening of #7466.

Add an includeSnapshot option to subscribeToRuntimeTerminalData (default
true, preserving the readiness/background-session consumers) and have the
echo watch opt out. The echo still arrives as live onData; a rare race that
places it in the snapshot instead degrades to a safe false-negative (the
recovery toast), never a false-positive submit.

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

src/renderer/src/lib/prompt-echo-wait.ts

@@ -52,11 +52,16 @@ export function watchForPromptEcho(
     }
 
     if (isRemoteRuntimePtyId(ptyId)) {
+      // Why: skip the initial buffer snapshot — it is the pre-paste screen and
+      // could match the sample (or a stale paste placeholder) before the echo,
+      // falsely verifying delivery onto a swallow screen (#7466). Only output
+      // rendered after we subscribe counts, matching the local sidecar.
       void subscribeToRuntimeTerminalData(
         settings,
         ptyId,
         `desktop:prompt-echo:${ptyId}`,
-        observeData
+        observeData,
+        { includeSnapshot: false }
       )
         .then((remoteUnsubscribe) => {
           if (settled) {

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

@@ -4,6 +4,7 @@ import {
   decodeTerminalStreamFrame,
   decodeTerminalStreamJson,
   encodeTerminalStreamFrame,
+  encodeTerminalStreamJson,
   encodeTerminalStreamText
 } from '../../../shared/terminal-stream-protocol'
 import {
@@ -117,6 +118,81 @@ describe('remote runtime terminal data subscriptions', () => {
     expect(_getRemoteRuntimeTerminalMultiplexerCountForTest()).toBe(0)
   })
 
+  async function currentStreamId(): Promise<number> {
+    await vi.waitFor(() => expect(sendBinary).toHaveBeenCalled())
+    const subscribeFrame = decodeTerminalStreamFrame(sendBinary.mock.calls[0][0])
+    return decodeTerminalStreamJson<{ streamId: number }>(subscribeFrame!.payload)!.streamId
+  }
+
+  function deliverInitialSnapshot(streamId: number, text: string): void {
+    callbacks?.onBinary?.(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.SnapshotStart,
+        streamId,
+        seq: 0,
+        payload: encodeTerminalStreamJson({ cols: 80, rows: 24 })
+      })
+    )
+    callbacks?.onBinary?.(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.SnapshotChunk,
+        streamId,
+        seq: 0,
+        payload: encodeTerminalStreamText(text)
+      })
+    )
+    callbacks?.onBinary?.(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.SnapshotEnd,
+        streamId,
+        seq: 0,
+        payload: new Uint8Array(0)
+      })
+    )
+  }
+
+  function deliverOutput(streamId: number, text: string): void {
+    callbacks?.onBinary?.(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Output,
+        streamId,
+        seq: 1,
+        payload: encodeTerminalStreamText(text)
+      })
+    )
+  }
+
+  it('replays the initial buffer snapshot to the watcher by default', async () => {
+    const watcher = vi.fn()
+    await subscribeToRuntimeTerminalData(
+      { activeRuntimeEnvironmentId: 'env-fallback' },
+      'remote:env-1@@terminal-1',
+      'watcher-1',
+      watcher
+    )
+    deliverInitialSnapshot(await currentStreamId(), 'PRE-PASTE SCREEN')
+
+    expect(watcher.mock.calls.map((call) => call[0])).toContain('PRE-PASTE SCREEN')
+  })
+
+  it('omits the initial snapshot from the watcher when includeSnapshot is false', async () => {
+    const watcher = vi.fn()
+    await subscribeToRuntimeTerminalData(
+      { activeRuntimeEnvironmentId: 'env-fallback' },
+      'remote:env-1@@terminal-1',
+      'watcher-1',
+      watcher,
+      { includeSnapshot: false }
+    )
+    const streamId = await currentStreamId()
+    deliverInitialSnapshot(streamId, 'PRE-PASTE SCREEN')
+    // Live output rendered after the subscription still reaches the watcher.
+    deliverOutput(streamId, 'live')
+
+    expect(watcher).not.toHaveBeenCalledWith('PRE-PASTE SCREEN')
+    expect(watcher).toHaveBeenCalledWith('live')
+  })
+
   it('keeps the shared terminal multiplexer until the last watcher closes', async () => {
     const firstDispose = await subscribeToRuntimeTerminalData(
       { activeRuntimeEnvironmentId: 'env-fallback' },

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

@@ -56,7 +56,8 @@ export async function subscribeToRuntimeTerminalData(
   settings: Pick<GlobalSettings, 'activeRuntimeEnvironmentId'> | null | undefined,
   ptyId: string,
   clientId: string,
-  watcher: (data: string) => void
+  watcher: (data: string) => void,
+  options?: { includeSnapshot?: boolean }
 ): Promise<() => void> {
   const terminal = getRemoteRuntimeTerminalHandle(ptyId)
   const ownerEnvironmentId = getRemoteRuntimePtyEnvironmentId(ptyId)
@@ -67,12 +68,17 @@ export async function subscribeToRuntimeTerminalData(
     return () => {}
   }
 
+  // Why: the initial buffer snapshot is the screen state from *before* this
+  // subscription attached. Consumers that must only judge freshly-rendered
+  // output (prompt-echo verification) opt out so a pre-paste screen can't be
+  // mistaken for the paste echoing back (#7466); the default replays it.
+  const includeSnapshot = options?.includeSnapshot ?? true
   const stream = await getRemoteRuntimeTerminalMultiplexer(target.environmentId).subscribeTerminal({
     terminal,
     client: { id: clientId, type: 'desktop' },
     callbacks: {
       onData: (data) => watcher(data),
-      onSnapshot: watcher
+      onSnapshot: includeSnapshot ? watcher : () => {}
     }
   })