fix(terminal): attach never-activated daemon sessions on remote subscribe and provider-read fallback (#12589)

* fix(terminal): attach never-activated daemon sessions on remote subscribe and provider-read fallback

A daemon-backed terminal whose tab was never activated in the host UI was
never attached, so the daemon emitted no bytes: paired clients rendered
blank/frozen panes and `terminal read` returned an empty tail while the PTY
was alive.

- Runtime: first remote view subscriber of a known-but-unattached local
  daemon session triggers an attach through the pty controller — attach-only,
  no resize, no renderer mount/focus, headless-safe, deduped across
  concurrent subscribers, and never detached on release. Excludes SSH-scoped
  ids and sessions a local spawn already published this generation.
- Read path: withVisibleSnapshotFallback now falls back to the provider tail
  for an empty-tail never-attached live local session; unprovable state stays
  empty, never an error.
- pty controller: expose attach with getProviderForPty-style routing,
  answering false on doubt; local daemon provider only.
- Daemon adapter: attach rides the session's applied size instead of a
  hardcoded 80x24, sends attachOnly, and retires a pre-v31 daemon's
  accidental spawn instead of publishing it.

Deterministic harness drives the real terminal.multiplex handler against a
real OrcaRuntimeService with an injected daemon-model controller whose data
events are gated on attach; covers snapshot-capable and snapshot-null
daemons, concurrency, release, replacement-spawn exclusion, and negative
safety. Red on base, green with the fix, red again with the fix reverted.

* fix(terminal): refuse degraded-provider attach fallback and surface failed legacy-spawn retire

Verifier follow-ups on subscriber-driven daemon attach:

- DegradedDaemonPtyProvider.attach routed unknown ids to the in-process
  fallback, whose no-op attach resolves — the runtime then pinned a
  subscriber-driven attach as succeeded while the stream stayed blank.
  Attach now refuses any route that resolves to the fallback (a fallback pty
  cannot own a daemon-surviving session), so the controller answers false,
  no sticky success is recorded, and a later subscriber attaches once a
  daemon adapter proves the id. Session-probe adoption moved to
  degraded-daemon-session-routing alongside the new refusal.
- The pre-v31 attach-only TOCTOU retire (accidental legacy spawn kill) now
  logs a warning with the sessionId on kill failure instead of swallowing
  it, so an orphaned replacement shell is diagnosable.

Regressions: degraded provider refuses unowned/fallback-owned attach and
routes to a daemon once it proves the id (red on previous commit); runtime
harness pins refused-attach retry for a later subscriber; adapter test pins
the surfaced kill failure.

---------

Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>

src/main/daemon/daemon-pty-adapter.test.ts

@@ -1062,12 +1062,13 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => {
       const ensureConnectedSpy = vi
         .spyOn(DaemonClient.prototype, 'ensureConnected')
         .mockResolvedValue()
-      const requestSpy = vi.spyOn(DaemonClient.prototype, 'request').mockResolvedValue({
-        isNew: false,
-        pid: 4242,
-        shellState: 'unsupported',
-        snapshot: null
-      } as never)
+      const requestSpy = vi
+        .spyOn(DaemonClient.prototype, 'request')
+        .mockImplementation(async (type: string) =>
+          type === 'getSize'
+            ? ({ size: { cols: 80, rows: 24 } } as never)
+            : ({ isNew: false, pid: 4242, shellState: 'unsupported', snapshot: null } as never)
+        )
       const notifySpy = vi.spyOn(DaemonClient.prototype, 'notify')
       const legacy = new DaemonPtyAdapter({ socketPath, tokenPath, protocolVersion: 28 })
       try {
@@ -1092,12 +1093,13 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => {
       const ensureConnectedSpy = vi
         .spyOn(DaemonClient.prototype, 'ensureConnected')
         .mockResolvedValue()
-      const requestSpy = vi.spyOn(DaemonClient.prototype, 'request').mockResolvedValue({
-        isNew: false,
-        pid: 4242,
-        shellState: 'unsupported',
-        snapshot: null
-      } as never)
+      const requestSpy = vi
+        .spyOn(DaemonClient.prototype, 'request')
+        .mockImplementation(async (type: string) =>
+          type === 'getSize'
+            ? ({ size: { cols: 80, rows: 24 } } as never)
+            : ({ isNew: false, pid: 4242, shellState: 'unsupported', snapshot: null } as never)
+        )
       const notifySpy = vi.spyOn(DaemonClient.prototype, 'notify')
       const current = new DaemonPtyAdapter({ socketPath, tokenPath, protocolVersion: 32 })
       try {
@@ -1654,6 +1656,99 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => {
 
       adapter2.dispose()
     })
+
+    it('preserves the live session dimensions instead of forcing 80×24', async () => {
+      const { id } = await adapter.spawn({ cols: 137, rows: 41 })
+
+      const adapter2 = new DaemonPtyAdapter({ socketPath, tokenPath })
+      await adapter2.attach(id)
+
+      // The running TUI keeps its geometry: no resize, size unchanged.
+      expect(lastSubprocess.resize).not.toHaveBeenCalled()
+      expect(await adapter2.getAppliedSize(id)).toEqual({ cols: 137, rows: 41 })
+      adapter2.dispose()
+    })
+
+    it('refuses to create when the session is absent (attach-only)', async () => {
+      const adapter2 = new DaemonPtyAdapter({ socketPath, tokenPath })
+
+      await expect(adapter2.attach('wt-x@@deadbeef')).rejects.toThrow(
+        'Session not found: wt-x@@deadbeef'
+      )
+
+      // No shell was spawned as a side effect of the refused attach.
+      expect(lastSpawnOpts).toBeNull()
+      adapter2.dispose()
+    })
+
+    it('retires the accidental spawn of a pre-v31 daemon that ignores attachOnly', async () => {
+      const ensureConnectedSpy = vi
+        .spyOn(DaemonClient.prototype, 'ensureConnected')
+        .mockResolvedValue()
+      const requestSpy = vi
+        .spyOn(DaemonClient.prototype, 'request')
+        .mockImplementation(async (type: string) =>
+          type === 'getSize'
+            ? ({ size: { cols: 100, rows: 30 } } as never)
+            : type === 'createOrAttach'
+              ? ({ isNew: true, pid: 77, shellState: 'unsupported', snapshot: null } as never)
+              : ({} as never)
+        )
+      const legacy = new DaemonPtyAdapter({ socketPath, tokenPath, protocolVersion: 30 })
+      try {
+        await expect(legacy.attach('raced-legacy-session')).rejects.toThrow(
+          'Session not found: raced-legacy-session'
+        )
+
+        expect(requestSpy).toHaveBeenCalledWith(
+          'createOrAttach',
+          expect.objectContaining({ cols: 100, rows: 30, attachOnly: true })
+        )
+        expect(requestSpy).toHaveBeenCalledWith('kill', {
+          sessionId: 'raced-legacy-session',
+          immediate: true
+        })
+      } finally {
+        legacy.dispose()
+        requestSpy.mockRestore()
+        ensureConnectedSpy.mockRestore()
+      }
+    })
+
+    it('surfaces a failed retire of the accidental legacy spawn instead of swallowing it', async () => {
+      const ensureConnectedSpy = vi
+        .spyOn(DaemonClient.prototype, 'ensureConnected')
+        .mockResolvedValue()
+      const requestSpy = vi
+        .spyOn(DaemonClient.prototype, 'request')
+        .mockImplementation(async (type: string) => {
+          if (type === 'getSize') {
+            return { size: { cols: 100, rows: 30 } } as never
+          }
+          if (type === 'createOrAttach') {
+            return { isNew: true, pid: 77, shellState: 'unsupported', snapshot: null } as never
+          }
+          throw new Error('kill transport lost')
+        })
+      const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
+      const legacy = new DaemonPtyAdapter({ socketPath, tokenPath, protocolVersion: 30 })
+      try {
+        await expect(legacy.attach('orphaned-legacy-session')).rejects.toThrow(
+          'Session not found: orphaned-legacy-session'
+        )
+
+        // The orphaned replacement is at least diagnosable.
+        expect(warnSpy).toHaveBeenCalledWith(
+          '[daemon] attach-only retire of accidental legacy spawn failed',
+          expect.objectContaining({ sessionId: 'orphaned-legacy-session' })
+        )
+      } finally {
+        legacy.dispose()
+        warnSpy.mockRestore()
+        requestSpy.mockRestore()
+        ensureConnectedSpy.mockRestore()
+      }
+    })
   })
 
   describe('listProcesses', () => {

src/main/daemon/daemon-pty-adapter.ts

@@ -897,11 +897,31 @@ export class DaemonPtyAdapter implements IPtyProvider {
       this.setPtyBackgrounded(id, false)
     }
 
-    await this.client.request<CreateOrAttachResult>('createOrAttach', {
+    // Why size-first: attach must ride the session's own geometry — a fixed
+    // 80×24 here could resize a live agent's TUI — and a null size means the
+    // daemon cannot prove the session, so refuse rather than risk a create.
+    const size = await this.getAppliedSize(id)
+    if (!size) {
+      throw new SessionNotFoundError(id)
+    }
+    const result = await this.client.request<CreateOrAttachResult>('createOrAttach', {
       sessionId: id,
-      cols: 80,
-      rows: 24
+      cols: size.cols,
+      rows: size.rows,
+      attachOnly: true
     })
+    if (result.isNew) {
+      // Why: a pre-v31 daemon ignores attachOnly; retire its accidental spawn
+      // instead of publishing a fresh shell as an attach.
+      await this.client.request('kill', { sessionId: id, immediate: true }).catch((error) => {
+        // Why surface, not swallow: a failed retire leaves an untracked orphan shell.
+        console.warn('[daemon] attach-only retire of accidental legacy spawn failed', {
+          sessionId: id,
+          error
+        })
+      })
+      throw new SessionNotFoundError(id)
+    }
     this.clearSessionAwaitingDaemonRecovery(id)
   }
 

src/main/daemon/degraded-daemon-pty-provider.test.ts

@@ -190,6 +190,31 @@ it('preserves unavailable inspection from an owning daemon', async () => {
 })
 
 describe('DegradedDaemonPtyProvider', () => {
+  it('refuses attach for ids no daemon adapter owns instead of the fallback no-op', async () => {
+    const daemonSessions: string[] = []
+    const current = createDaemonAdapter('daemon', daemonSessions)
+    const fallback = createProvider('fallback', ['fresh-fallback-session'])
+    const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
+
+    // Unknown id: the fallback's resolving no-op attach must never read as
+    // success — the runtime would pin a blank stream as attached.
+    await expect(provider.attach('wt-1@@unknown')).rejects.toThrow(
+      'Session not found: wt-1@@unknown'
+    )
+    // Fallback-owned fresh sessions stream in-process; attach is refused too.
+    await expect(provider.attach('fresh-fallback-session')).rejects.toThrow(
+      'Session not found: fresh-fallback-session'
+    )
+    expect(fallback.attach).not.toHaveBeenCalled()
+    expect(current.attach).not.toHaveBeenCalled()
+
+    // Once a daemon adapter owns the id, attach routes to that adapter.
+    daemonSessions.push('wt-1@@learned')
+    await expect(provider.attach('wt-1@@learned')).resolves.toBeUndefined()
+    expect(current.attach).toHaveBeenCalledWith('wt-1@@learned')
+    expect(fallback.attach).not.toHaveBeenCalled()
+  })
+
   it('only delegates owner-listing authority to the provider that owns the id', async () => {
     const current = createDaemonAdapter('daemon', ['daemon-session'])
     const fallback = createProvider('fallback', [], true)

src/main/daemon/degraded-daemon-pty-provider.ts

@@ -12,6 +12,8 @@ import type {
   PtySpawnResult
 } from '../providers/types'
 import {
+  adoptOwningProvider,
+  attachDaemonOwnedSession,
   discoverDegradedDaemonSessions,
   findDaemonAdapter,
   listProviderSessionIds
@@ -82,7 +84,9 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
 
   spawn = (opts: PtySpawnOptions): Promise<PtySpawnResult> => this.freshSpawns.spawn(opts)
 
-  attach = (id: string): Promise<void> => this.providerFor(id).attach(id)
+  // Why refuse the fallback route (unknown ids resolve to it): see attachDaemonOwnedSession.
+  attach = (id: string): Promise<void> =>
+    attachDaemonOwnedSession(this.providerFor(id), this.fallback, id)
 
   hasPty(id: string): boolean {
     const mapped = this.sessionProviders.get(id)
@@ -343,13 +347,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
   }
 
   private findProviderForExistingSession(sessionId: string): IPtyProvider | null {
-    for (const provider of this.allProviders()) {
-      if (provider.hasPty?.(sessionId) === true) {
-        this.sessionProviders.set(sessionId, provider)
-        return provider
-      }
-    }
-    return null
+    return adoptOwningProvider(this.sessionProviders, this.allProviders(), sessionId)
   }
 
   private allProviders(): IPtyProvider[] {

src/main/daemon/degraded-daemon-session-routing.ts

@@ -1,5 +1,6 @@
 import type { IPtyProvider } from '../providers/types'
 import type { DaemonPtyAdapter } from './daemon-pty-adapter'
+import { SessionNotFoundError } from './daemon-errors'
 
 export async function discoverDegradedDaemonSessions(
   adapters: readonly DaemonPtyAdapter[],
@@ -25,6 +26,37 @@ export function listProviderSessionIds(
     .map(([id]) => id)
 }
 
+/** Attach-only session adoption: refuses the in-process fallback route. A
+ *  fallback pty cannot own a daemon-surviving session by definition, and its
+ *  no-op attach resolving would pin a subscriber-driven attach as succeeded
+ *  while the stream stays blank. */
+export async function attachDaemonOwnedSession(
+  owner: IPtyProvider,
+  fallback: IPtyProvider,
+  sessionId: string
+): Promise<void> {
+  if (owner === fallback) {
+    throw new SessionNotFoundError(sessionId)
+  }
+  await owner.attach(sessionId)
+}
+
+/** Probes providers for an id absent from the routing map and adopts the
+ *  first proven owner into the map. */
+export function adoptOwningProvider(
+  sessionProviders: Map<string, IPtyProvider>,
+  providers: readonly IPtyProvider[],
+  sessionId: string
+): IPtyProvider | null {
+  for (const provider of providers) {
+    if (provider.hasPty?.(sessionId) === true) {
+      sessionProviders.set(sessionId, provider)
+      return provider
+    }
+  }
+  return null
+}
+
 export function findDaemonAdapter(
   sessionProviders: ReadonlyMap<string, IPtyProvider>,
   daemonAdapters: readonly DaemonPtyAdapter[],

src/main/ipc/pty.test.ts

@@ -753,6 +753,7 @@ describe('registerPtyHandlers', () => {
     write: (ptyId: string, data: string) => boolean
     resize: (ptyId: string, cols: number, rows: number) => boolean
     probePtyLiveness: (ptyId: string) => Promise<boolean | null>
+    attach: (ptyId: string) => Promise<boolean>
   } {
     let controller:
       | {
@@ -760,6 +761,7 @@ describe('registerPtyHandlers', () => {
           write: (ptyId: string, data: string) => boolean
           resize: (ptyId: string, cols: number, rows: number) => boolean
           probePtyLiveness: (ptyId: string) => Promise<boolean | null>
+          attach: (ptyId: string) => Promise<boolean>
         }
       | undefined
     const runtime = {
@@ -877,6 +879,42 @@ describe('registerPtyHandlers', () => {
     })
   })
 
+  it('routes controller attach to the local daemon provider only, false on doubt', async () => {
+    const localProvider = createAgentClaimProvider({})
+    const sshProvider = createAgentClaimProvider({})
+    setLocalPtyProvider(localProvider as never)
+    registerSshPtyProvider('ssh-attach', sshProvider as never)
+    const controller = registerAgentClaimController()
+    const daemonPtyId = 'repo-1::/tmp/wt@@1a2b3c4d'
+    const ownedSshPtyId = 'owned-remote-pty'
+    setPtyOwnership(ownedSshPtyId, 'ssh-attach')
+    try {
+      // Local daemon session: attach flows to the provider.
+      await expect(controller.attach(daemonPtyId)).resolves.toBe(true)
+      expect(localProvider.attach).toHaveBeenCalledWith(daemonPtyId)
+
+      // SSH-scoped sessions are excluded — leases handle their reattach.
+      await expect(controller.attach(ownedSshPtyId)).resolves.toBe(false)
+      await expect(controller.attach('ssh:ssh-attach@@relay-pty')).resolves.toBe(false)
+      expect(sshProvider.attach).not.toHaveBeenCalled()
+
+      // Provider refusal (absent/unprovable session) answers false, not throw.
+      localProvider.attach.mockRejectedValueOnce(new Error('Session not found'))
+      await expect(controller.attach(daemonPtyId)).resolves.toBe(false)
+
+      // The in-process local provider streams without attach; never called.
+      const inProcess = new LocalPtyProvider()
+      const inProcessAttach = vi.spyOn(inProcess, 'attach')
+      setLocalPtyProvider(inProcess)
+      await expect(controller.attach(daemonPtyId)).resolves.toBe(false)
+      expect(inProcessAttach).not.toHaveBeenCalled()
+    } finally {
+      unregisterSshPtyProvider('ssh-attach')
+      clearPtyOwnershipForConnection('ssh-attach')
+      clearProviderPtyState(ownedSshPtyId)
+    }
+  })
+
   it('does not dispatch a runtime PTY spawn after its client disconnects', async () => {
     const provider = createAgentClaimProvider({})
     setLocalPtyProvider(provider as never)

src/main/ipc/pty.ts

@@ -5130,6 +5130,30 @@ export function registerPtyHandlers(
         return null
       }
     },
+    // Why: subscriber-driven ingestion for daemon sessions no renderer pane
+    // ever attached. Local daemon sessions only — SSH panes have their own
+    // lease machinery, and the in-process local provider streams without
+    // attach. Attach-only and false-on-doubt: never creates or resizes.
+    attach: async (ptyId) => {
+      if (ptyOwnership.get(ptyId) != null || parseAppSshPtyId(ptyId)) {
+        return false
+      }
+      let provider: IPtyProvider
+      try {
+        provider = getProviderForPty(ptyId)
+      } catch {
+        return false
+      }
+      if (provider !== localProvider || provider instanceof LocalPtyProvider) {
+        return false
+      }
+      try {
+        await provider.attach(ptyId)
+        return true
+      } catch {
+        return false
+      }
+    },
     kill: (ptyId) => {
       let connectionId: string | null | undefined = ptyOwnership.get(ptyId)
       const parsedSshId = connectionId === undefined ? parseAppSshPtyId(ptyId) : null

src/main/runtime/orca-runtime.ts

@@ -1621,6 +1621,10 @@ type RuntimePtyController = {
     agentSessionEnsure?: AgentSessionClaimedSpawnResult
   }>
   write(ptyId: string, data: string): boolean
+  /** Attach-only adoption of a live local daemon session so its output streams
+   *  to main without a renderer pane; never creates, resizes, or focuses.
+   *  False on doubt (absent session, SSH-scoped id, non-daemon provider). */
+  attach?(ptyId: string): Promise<boolean>
   kill(ptyId: string): boolean
   stopAndWait?(
     ptyId: string,
@@ -3056,6 +3060,17 @@ export class OrcaRuntimeService {
   // Preview windows consume the raw stream but deliberately leave terminal
   // query replies to main's headless emulator.
   private rawTerminalViewSubscriberCounts = new Map<string, number>()
+  // Why a sticky promise per PTY: the daemon only emits data for sessions this
+  // app has attached, so the first remote view subscriber of a never-attached
+  // local daemon session triggers a main-side attach. The map dedupes
+  // concurrent first-subscribes and keeps later subscribes no-ops; it is
+  // cleared per lifecycle generation (exit/respawn) and on failed attempts.
+  private subscriberDrivenProviderAttachesByPtyId = new Map<string, Promise<boolean>>()
+  // Why: a spawn through this app already attaches its provider stream, so
+  // subscriber-driven attach and the never-attached read fallback must target
+  // only inventory-discovered sessions no local spawn published this
+  // generation (a replacement spawn under a reused id starts clean).
+  private spawnPublishedPtys = new Set<string>()
 
   // Why: per-PTY driver state. The "driver" is whoever currently owns the
   // input/resize floor. While `kind === 'mobile'` the desktop renderer drops
@@ -9160,6 +9175,7 @@ export class OrcaRuntimeService {
       // Why: surface absence cannot distinguish an in-flight admission from a completed headless lifecycle.
       this.pendingPtyRegistrationIncarnations.set(ptyId, incarnationId ?? null)
     }
+    this.spawnPublishedPtys.add(ptyId)
     const pty = this.getOrCreatePtyWorktreeRecord(ptyId)
     if (pty) {
       if (incarnationId) {
@@ -9183,6 +9199,7 @@ export class OrcaRuntimeService {
     isWsl?: boolean
   ): void {
     this.assertPtyDidNotExitBeforeRegistration(ptyId, binding?.incarnationId)
+    this.spawnPublishedPtys.add(ptyId)
     // Why: record the renderer pane identity at spawn time so a stalled graph
     // sync can't hide that a live PTY already backs a pending mobile create.
     const paneKey =
@@ -10434,6 +10451,9 @@ export class OrcaRuntimeService {
   private advancePtyLifecycleGeneration(ptyId: string): void {
     this.ptyLifecycleGenerationById.set(ptyId, this.nextPtyLifecycleGeneration++)
     this.legacyWorkerRecoveredPtys.delete(ptyId)
+    // Why: a respawn under the same session id needs its own subscriber-driven attach.
+    this.subscriberDrivenProviderAttachesByPtyId.delete(ptyId)
+    this.spawnPublishedPtys.delete(ptyId)
     // Why: a provider response belongs to the process generation that issued
     // it; a respawn must neither reuse its frame nor join its in-flight call.
     this.providerBufferAcquisitionsByPtyId.delete(ptyId)
@@ -10613,6 +10633,7 @@ export class OrcaRuntimeService {
       ptyId,
       (this.remoteTerminalViewSubscriberCounts.get(ptyId) ?? 0) + 1
     )
+    this.ensureSubscriberDrivenProviderAttach(ptyId)
     this.notifyRemoteTerminalViewPresenceChanged(ptyId)
     let released = false
     return () => {
@@ -10630,6 +10651,57 @@ export class OrcaRuntimeService {
     }
   }
 
+  /** A local daemon session main knows is live but has never ingested a byte
+   *  from — i.e. no pane ever attached it, so the daemon is not emitting.
+   *  Headless state exists only after the first ingested byte; a snapshot
+   *  reconcile in flight implies a spawn-path attach already happened. */
+  private isKnownUnattachedLocalDaemonPty(ptyId: string): boolean {
+    if (this.headlessTerminals.has(ptyId) || this.providerSnapshotPreferredPtys.has(ptyId)) {
+      return false
+    }
+    // A spawn published (or admission pending) this generation already
+    // attaches the provider stream; a replacement under a reused id must not
+    // read as the discovered never-attached session it replaced.
+    if (
+      this.spawnPublishedPtys.has(ptyId) ||
+      this.pendingPtyRegistrationIncarnations.has(ptyId)
+    ) {
+      return false
+    }
+    // SSH panes have their own lease/reattach machinery.
+    if (parseAppSshPtyId(ptyId)) {
+      return false
+    }
+    const pty = this.ptysById.get(ptyId)
+    return pty !== undefined && pty.connectionId === null && pty.connected
+  }
+
+  /** First remote view subscriber of a never-attached local daemon session:
+   *  have main attach so output starts flowing. Attach-only (never spawns),
+   *  no resize, no renderer mount/focus — works headless. Releases never
+   *  detach: continued ingestion is the point, and daemon detach is stubbed. */
+  private ensureSubscriberDrivenProviderAttach(ptyId: string): void {
+    const controller = this.ptyController
+    if (
+      !controller?.attach ||
+      this.subscriberDrivenProviderAttachesByPtyId.has(ptyId) ||
+      !this.isKnownUnattachedLocalDaemonPty(ptyId)
+    ) {
+      return
+    }
+    const attach = controller.attach
+    // Async wrapper: a synchronous controller throw must not break subscribe.
+    const attempt = (async () => attach(ptyId))().catch(() => false)
+    this.subscriberDrivenProviderAttachesByPtyId.set(ptyId, attempt)
+    void attempt.then((attached) => {
+      // Why: an unprovable session must not be pinned as attached; a later
+      // subscriber may retry once the daemon can prove it.
+      if (!attached && this.subscriberDrivenProviderAttachesByPtyId.get(ptyId) === attempt) {
+        this.subscriberDrivenProviderAttachesByPtyId.delete(ptyId)
+      }
+    })
+  }
+
   /** Mark a raw-output viewer without transferring terminal query authority. */
   registerRawTerminalViewSubscriber(ptyId: string): () => void {
     this.rawTerminalViewSubscriberCounts.set(
@@ -11446,7 +11518,13 @@ export class OrcaRuntimeService {
     const blankFallback = shouldFallbackToVisibleTerminalSnapshot(read, opts)
     const recoveredWorkerFallback =
       read.tail.length === 0 && this.legacyWorkerRecoveredPtys.has(ptyId)
-    if (recoveredWorkerFallback) {
+    // Why: a live daemon session no pane ever attached has ingested zero bytes,
+    // so only the provider holds its screen. Unprovable state stays empty.
+    const neverAttachedProviderFallback =
+      read.tail.length === 0 &&
+      !recoveredWorkerFallback &&
+      this.isKnownUnattachedLocalDaemonPty(ptyId)
+    if (recoveredWorkerFallback || neverAttachedProviderFallback) {
       const providerLines = await this.readProviderTerminalTailLines(ptyId, opts.limit)
       if (providerLines.length > 0) {
         return buildVisibleSnapshotReadFallback(read, providerLines, opts.limit)

src/main/runtime/terminal-subscriber-driven-daemon-attach.test.ts

@@ -0,0 +1,430 @@
+/**
+ * Subscriber-driven daemon attach (never-activated tab ingestion).
+ *
+ * Invariant: on a remote-server host, a daemon-backed terminal whose tab was
+ * never activated in the host UI must still stream to paired clients and be
+ * readable via `terminal read`. The daemon only emits data for sessions this
+ * app has ATTACHED, and attach historically happened only via renderer pane
+ * mount — so a never-activated tab produced blank paired panes and empty CLI
+ * tails while the PTY was alive.
+ *
+ * Harness: real OrcaRuntimeService + real TERMINAL_METHODS multiplex handler,
+ * with an injected pty controller modeling a daemon provider whose data events
+ * are deliverable ONLY after attach(id). No window is ever attached — every
+ * assertion also holds for headless `orca serve`.
+ */
+import { describe, expect, it, vi } from 'vitest'
+import { OrcaRuntimeService } from './orca-runtime'
+import { RpcDispatcher } from './rpc/dispatcher'
+import type { RpcRequest } from './rpc/core'
+import { TERMINAL_METHODS } from './rpc/methods/terminal'
+import {
+  TerminalStreamOpcode,
+  decodeTerminalStreamFrame,
+  decodeTerminalStreamText,
+  encodeTerminalStreamFrame,
+  encodeTerminalStreamJson
+} from '../../shared/terminal-stream-protocol'
+
+const WORKTREE_ID = 'repo-1::/tmp/wt'
+const PTY_ID = `${WORKTREE_ID}@@1a2b3c4d`
+
+type RuntimeInternals = {
+  recordPtyWorktree: (
+    ptyId: string,
+    worktreeId: string,
+    state?: { connected?: boolean; connectionId?: string | null }
+  ) => unknown
+  issuePtyHandle: (pty: unknown) => string
+  headlessTerminals: Map<string, unknown>
+}
+
+function internals(runtime: OrcaRuntimeService): RuntimeInternals {
+  return runtime as unknown as RuntimeInternals
+}
+
+type DaemonSessionModel = {
+  cols: number
+  rows: number
+  attached: boolean
+  screen: string
+}
+
+/** Models the daemon provider boundary: inventory of live sessions, data
+ *  events gated on attach, and a configurable authoritative snapshot. */
+function createDaemonProviderModel(opts: { snapshotCapable: boolean }) {
+  const sessions = new Map<string, DaemonSessionModel>()
+  const attachCalls: string[] = []
+  const resizeCalls: [string, number, number][] = []
+  let runtime: OrcaRuntimeService | null = null
+  const controller = {
+    write: () => true,
+    kill: () => true,
+    getForegroundProcess: async () => null,
+    hasRendererSerializer: () => false,
+    getSize: (ptyId: string) => {
+      const session = sessions.get(ptyId)
+      return session ? { cols: session.cols, rows: session.rows } : null
+    },
+    resize: (ptyId: string, cols: number, rows: number) => {
+      resizeCalls.push([ptyId, cols, rows])
+      return true
+    },
+    attach: async (ptyId: string) => {
+      attachCalls.push(ptyId)
+      const session = sessions.get(ptyId)
+      // Attach-only: an absent session is refused, never created.
+      if (!session) {
+        return false
+      }
+      session.attached = true
+      return true
+    },
+    serializeProviderBuffer: async (ptyId: string) => {
+      const session = sessions.get(ptyId)
+      if (!session || !opts.snapshotCapable) {
+        // v19-style daemon: no authoritative snapshot (missing outputSequence).
+        return null
+      }
+      return {
+        data: session.screen,
+        cols: session.cols,
+        rows: session.rows,
+        seq: 0,
+        source: 'headless' as const
+      }
+    }
+  }
+  return {
+    controller,
+    sessions,
+    attachCalls,
+    resizeCalls,
+    bind(target: OrcaRuntimeService) {
+      runtime = target
+    },
+    /** Daemon stream event → main ingestion, exactly like ipc/pty.ts wiring —
+     *  but the daemon only emits for sessions this app has attached. */
+    emitData(ptyId: string, data: string): boolean {
+      const session = sessions.get(ptyId)
+      if (!session?.attached) {
+        return false
+      }
+      runtime?.onPtyData(ptyId, data, Date.now())
+      return true
+    }
+  }
+}
+
+function setupNeverAttachedDaemonSession(opts: { snapshotCapable: boolean; screen?: string }): {
+  runtime: OrcaRuntimeService
+  model: ReturnType<typeof createDaemonProviderModel>
+  handle: string
+  mountSpy: ReturnType<typeof vi.spyOn>
+} {
+  const runtime = new OrcaRuntimeService()
+  const model = createDaemonProviderModel(opts)
+  model.bind(runtime)
+  runtime.setPtyController(model.controller as never)
+  model.sessions.set(PTY_ID, {
+    cols: 137,
+    rows: 41,
+    attached: false,
+    screen: opts.screen ?? 'agent frozen screen\r\n'
+  })
+  // Same record + synthetic pty-form handle minting session.tabs.list uses
+  // (controller inventory → recordPtyWorktree → issuePtyHandle).
+  const record = internals(runtime).recordPtyWorktree(PTY_ID, WORKTREE_ID, { connected: true })
+  const handle = internals(runtime).issuePtyHandle(record)
+  const mountSpy = vi.spyOn(runtime, 'requestRendererTerminalTabMount')
+  return { runtime, model, handle, mountSpy }
+}
+
+function startMultiplex(runtime: OrcaRuntimeService, connectionId = 'conn-desktop') {
+  const messages: { result?: { type?: string; streamId?: number | null } }[] = []
+  const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
+  const handlers = new Map<
+    number,
+    (frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
+  >()
+  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(JSON.parse(msg))
+    },
+    {
+      connectionId,
+      sendBinary: (bytes: Uint8Array<ArrayBufferLike>) => {
+        binaryFrames.push(bytes)
+        return true
+      },
+      registerBinaryStreamHandler: (
+        streamId: number,
+        handler: (frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
+      ) => {
+        handlers.set(streamId, handler)
+        return () => {
+          if (handlers.get(streamId) === handler) {
+            handlers.delete(streamId)
+          }
+        }
+      }
+    }
+  )
+  return { messages, binaryFrames, handlers, dispatchPromise }
+}
+
+function sendSubscribe(
+  harness: ReturnType<typeof startMultiplex>,
+  streamId: number,
+  terminal: string,
+  clientId = 'client-desktop'
+): void {
+  harness.handlers.get(0)?.(
+    decodeTerminalStreamFrame(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Subscribe,
+        streamId: 0,
+        seq: 1,
+        payload: encodeTerminalStreamJson({
+          streamId,
+          terminal,
+          client: { id: clientId, type: 'desktop' }
+        })
+      })
+    )!
+  )
+}
+
+function sendUnsubscribe(harness: ReturnType<typeof startMultiplex>, streamId: number): void {
+  harness.handlers.get(streamId)?.(
+    decodeTerminalStreamFrame(
+      encodeTerminalStreamFrame({
+        opcode: TerminalStreamOpcode.Unsubscribe,
+        streamId,
+        seq: 2,
+        payload: encodeTerminalStreamJson({})
+      })
+    )!
+  )
+}
+
+async function waitForSubscribed(
+  harness: ReturnType<typeof startMultiplex>,
+  streamId: number
+): Promise<void> {
+  await vi.waitFor(() =>
+    expect(
+      harness.messages.some(
+        (message) => message.result?.type === 'subscribed' && message.result?.streamId === streamId
+      )
+    ).toBe(true)
+  )
+}
+
+function outputText(harness: ReturnType<typeof startMultiplex>): string {
+  return harness.binaryFrames
+    .map(decodeTerminalStreamFrame)
+    .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
+    .map((frame) => decodeTerminalStreamText(frame!.payload))
+    .join('')
+}
+
+function snapshotText(harness: ReturnType<typeof startMultiplex>): string {
+  return harness.binaryFrames
+    .map(decodeTerminalStreamFrame)
+    .filter((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotChunk)
+    .map((frame) => decodeTerminalStreamText(frame!.payload))
+    .join('')
+}
+
+describe('subscriber-driven daemon attach (never-activated tab)', () => {
+  it('attaches on first desktop multiplex subscribe and streams bytes (snapshot-null daemon)', async () => {
+    const { runtime, model, handle, mountSpy } = setupNeverAttachedDaemonSession({
+      snapshotCapable: false
+    })
+    const harness = startMultiplex(runtime)
+    await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
+
+    // Baseline: nothing has attached and the daemon refuses to emit.
+    expect(model.emitData(PTY_ID, 'pre-subscribe bytes')).toBe(false)
+
+    sendSubscribe(harness, 1, handle)
+    await waitForSubscribed(harness, 1)
+
+    // Main attached the session — once — without any renderer involvement.
+    await vi.waitFor(() => expect(model.attachCalls).toEqual([PTY_ID]))
+    // Capability matrix: snapshot-null daemon paints blank before live bytes.
+    expect(snapshotText(harness)).toBe('')
+
+    // Now daemon bytes flow: into the stream and into the host model.
+    expect(model.emitData(PTY_ID, 'hello from daemon\r\n')).toBe(true)
+    await vi.waitFor(() => expect(outputText(harness)).toContain('hello from daemon'))
+    expect(internals(runtime).headlessTerminals.has(PTY_ID)).toBe(true)
+    const read = await runtime.readTerminal(handle)
+    expect(read.tail.join('\n')).toContain('hello from daemon')
+
+    // No resize, no renderer mount/focus/navigation, no window required.
+    expect(model.resizeCalls).toEqual([])
+    expect(mountSpy).not.toHaveBeenCalled()
+  })
+
+  it('paints the provider snapshot first for a snapshot-capable daemon, then goes live', async () => {
+    const { runtime, model, handle } = setupNeverAttachedDaemonSession({
+      snapshotCapable: true,
+      screen: 'agent frozen screen\r\n'
+    })
+    const harness = startMultiplex(runtime)
+    await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
+
+    sendSubscribe(harness, 1, handle)
+    await waitForSubscribed(harness, 1)
+
+    await vi.waitFor(() => expect(model.attachCalls).toEqual([PTY_ID]))
+    expect(snapshotText(harness)).toContain('agent frozen screen')
+
+    expect(model.emitData(PTY_ID, 'now live\r\n')).toBe(true)
+    await vi.waitFor(() => expect(outputText(harness)).toContain('now live'))
+  })
+
+  it('attaches once across concurrent subscribers and keeps ingesting after one releases', async () => {
+    const { runtime, model, handle } = setupNeverAttachedDaemonSession({
+      snapshotCapable: false
+    })
+    const harness = startMultiplex(runtime)
+    await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
+
+    sendSubscribe(harness, 1, handle, 'client-a')
+    sendSubscribe(harness, 2, handle, 'client-b')
+    await waitForSubscribed(harness, 1)
+    await waitForSubscribed(harness, 2)
+
+    await vi.waitFor(() => expect(model.attachCalls).toEqual([PTY_ID]))
+
+    sendUnsubscribe(harness, 1)
+    // Ingestion continues after a release: no detach, model still fed.
+    expect(model.emitData(PTY_ID, 'still ingesting\r\n')).toBe(true)
+    const read = await runtime.readTerminal(handle)
+    expect(read.tail.join('\n')).toContain('still ingesting')
+
+    // A late re-subscribe is a no-op re-entry, not a second attach.
+    sendSubscribe(harness, 3, handle, 'client-c')
+    await waitForSubscribed(harness, 3)
+    expect(model.attachCalls).toEqual([PTY_ID])
+  })
+
+  it('never attaches SSH-scoped sessions or sessions absent from the daemon inventory', async () => {
+    const { runtime, model } = setupNeverAttachedDaemonSession({ snapshotCapable: false })
+    const sshPtyId = 'ssh:conn-9@@relay-pty-4'
+    const sshRecord = internals(runtime).recordPtyWorktree(sshPtyId, WORKTREE_ID, {
+      connected: true
+    })
+    const sshHandle = internals(runtime).issuePtyHandle(sshRecord)
+    const absentPtyId = `${WORKTREE_ID}@@99999999`
+    const absentRecord = internals(runtime).recordPtyWorktree(absentPtyId, WORKTREE_ID, {
+      connected: true
+    })
+    const absentHandle = internals(runtime).issuePtyHandle(absentRecord)
+
+    const harness = startMultiplex(runtime)
+    await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
+
+    sendSubscribe(harness, 1, sshHandle, 'client-ssh')
+    await waitForSubscribed(harness, 1)
+    expect(model.attachCalls).toEqual([])
+
+    // Absent session: subscribing must not create it or surface a new error.
+    sendSubscribe(harness, 2, absentHandle, 'client-absent')
+    await waitForSubscribed(harness, 2)
+    await Promise.resolve()
+    expect(model.sessions.has(absentPtyId)).toBe(false)
+    expect(model.emitData(absentPtyId, 'ghost')).toBe(false)
+    expect(harness.messages.some((message) => message.result?.type === 'error')).toBe(false)
+
+    // Unrelated live sessions are untouched by another pty's subscribers.
+    expect(model.sessions.get(PTY_ID)?.attached).toBe(false)
+    expect(model.attachCalls).not.toContain(PTY_ID)
+    expect(model.resizeCalls).toEqual([])
+  })
+
+  it('retries a refused attach for a later subscriber once the daemon learns the session', async () => {
+    const { runtime, model, handle } = setupNeverAttachedDaemonSession({ snapshotCapable: false })
+    // Degraded-daemon shape: main knows the record, the daemon does not own the
+    // id yet, and the controller answers false rather than a no-op success.
+    model.sessions.delete(PTY_ID)
+
+    const harness = startMultiplex(runtime)
+    await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
+    sendSubscribe(harness, 1, handle, 'client-a')
+    await waitForSubscribed(harness, 1)
+    await vi.waitFor(() => expect(model.attachCalls).toEqual([PTY_ID]))
+    // A refused attach must not pin sticky success while the stream is blank.
+    expect(model.emitData(PTY_ID, 'ghost')).toBe(false)
+
+    model.sessions.set(PTY_ID, { cols: 100, rows: 30, attached: false, screen: '' })
+    sendSubscribe(harness, 2, handle, 'client-b')
+    await waitForSubscribed(harness, 2)
+    await vi.waitFor(() => expect(model.attachCalls).toEqual([PTY_ID, PTY_ID]))
+    expect(model.emitData(PTY_ID, 'late daemon hello\r\n')).toBe(true)
+    await vi.waitFor(() => expect(outputText(harness)).toContain('late daemon hello'))
+  })
+
+  it('does not subscriber-attach or provider-read a session this app already spawned', async () => {
+    const { runtime, model, handle } = setupNeverAttachedDaemonSession({
+      snapshotCapable: true,
+      screen: 'predecessor frame\r\n'
+    })
+    // A spawn through this app attaches its stream at spawn time; a
+    // replacement under a reused id must not adopt the discovered-session path.
+    runtime.onPtySpawned(PTY_ID, undefined, { awaitsRegistration: false })
+
+    const harness = startMultiplex(runtime)
+    await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
+    sendSubscribe(harness, 1, handle)
+    await waitForSubscribed(harness, 1)
+
+    expect(model.attachCalls).toEqual([])
+    const read = await runtime.readTerminal(handle)
+    expect(read.tail).toEqual([])
+  })
+
+  it('terminal read falls back to the provider tail for a never-attached live pty', async () => {
+    const { runtime, handle } = setupNeverAttachedDaemonSession({
+      snapshotCapable: true,
+      screen: 'retained daemon line\r\n'
+    })
+
+    const read = await runtime.readTerminal(handle)
+
+    expect(read.tail.join('\n')).toContain('retained daemon line')
+  })
+
+  it('terminal read stays empty (not an error) when provider state is unprovable', async () => {
+    const { runtime, handle } = setupNeverAttachedDaemonSession({ snapshotCapable: false })
+
+    const read = await runtime.readTerminal(handle)
+
+    expect(read.tail).toEqual([])
+  })
+
+  it('terminal read keeps preferring the ingested model over provider state once attached', async () => {
+    const { runtime, model, handle } = setupNeverAttachedDaemonSession({
+      snapshotCapable: true,
+      screen: 'stale provider frame\r\n'
+    })
+    model.sessions.get(PTY_ID)!.attached = true
+    model.emitData(PTY_ID, 'live model line\r\n')
+
+    const read = await runtime.readTerminal(handle)
+
+    expect(read.tail.join('\n')).toContain('live model line')
+    expect(read.tail.join('\n')).not.toContain('stale provider frame')
+  })
+})