diff --git a/src/main/daemon/daemon-pty-adapter.test.ts b/src/main/daemon/daemon-pty-adapter.test.ts index 4c5d39184ad..b769f1e88ef 100644 --- a/src/main/daemon/daemon-pty-adapter.test.ts +++ b/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', () => { diff --git a/src/main/daemon/daemon-pty-adapter.ts b/src/main/daemon/daemon-pty-adapter.ts index acbe37b76f3..92f8abe5895 100644 --- a/src/main/daemon/daemon-pty-adapter.ts +++ b/src/main/daemon/daemon-pty-adapter.ts @@ -897,11 +897,31 @@ export class DaemonPtyAdapter implements IPtyProvider { this.setPtyBackgrounded(id, false) } - await this.client.request('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('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) } diff --git a/src/main/daemon/degraded-daemon-pty-provider.test.ts b/src/main/daemon/degraded-daemon-pty-provider.test.ts index 1f1bac4529f..644f732be02 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.test.ts +++ b/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) diff --git a/src/main/daemon/degraded-daemon-pty-provider.ts b/src/main/daemon/degraded-daemon-pty-provider.ts index e8a584f8cbd..4fef029f03b 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.ts +++ b/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 => this.freshSpawns.spawn(opts) - attach = (id: string): Promise => this.providerFor(id).attach(id) + // Why refuse the fallback route (unknown ids resolve to it): see attachDaemonOwnedSession. + attach = (id: string): Promise => + 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[] { diff --git a/src/main/daemon/degraded-daemon-session-routing.ts b/src/main/daemon/degraded-daemon-session-routing.ts index 30d934f3834..8f0ef61d22b 100644 --- a/src/main/daemon/degraded-daemon-session-routing.ts +++ b/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 { + 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, + 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, daemonAdapters: readonly DaemonPtyAdapter[], diff --git a/src/main/ipc/pty.test.ts b/src/main/ipc/pty.test.ts index 59201709ce0..39bf34b7a82 100644 --- a/src/main/ipc/pty.test.ts +++ b/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 + attach: (ptyId: string) => Promise } { 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 + attach: (ptyId: string) => Promise } | 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) diff --git a/src/main/ipc/pty.ts b/src/main/ipc/pty.ts index 94538e71b79..d87c856afa4 100644 --- a/src/main/ipc/pty.ts +++ b/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 diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 88e59f89fb3..491de83f4d2 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/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 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() + // 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>() + // 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() // 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) diff --git a/src/main/runtime/terminal-subscriber-driven-daemon-attach.test.ts b/src/main/runtime/terminal-subscriber-driven-daemon-attach.test.ts new file mode 100644 index 00000000000..7fd6c42b5fb --- /dev/null +++ b/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 +} + +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() + 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 + handle: string + mountSpy: ReturnType +} { + 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[] = [] + const handlers = new Map< + number, + (frame: NonNullable>) => 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) => { + binaryFrames.push(bytes) + return true + }, + registerBinaryStreamHandler: ( + streamId: number, + handler: (frame: NonNullable>) => void + ) => { + handlers.set(streamId, handler) + return () => { + if (handlers.get(streamId) === handler) { + handlers.delete(streamId) + } + } + } + } + ) + return { messages, binaryFrames, handlers, dispatchPromise } +} + +function sendSubscribe( + harness: ReturnType, + 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, streamId: number): void { + harness.handlers.get(streamId)?.( + decodeTerminalStreamFrame( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Unsubscribe, + streamId, + seq: 2, + payload: encodeTerminalStreamJson({}) + }) + )! + ) +} + +async function waitForSubscribed( + harness: ReturnType, + streamId: number +): Promise { + await vi.waitFor(() => + expect( + harness.messages.some( + (message) => message.result?.type === 'subscribed' && message.result?.streamId === streamId + ) + ).toBe(true) + ) +} + +function outputText(harness: ReturnType): string { + return harness.binaryFrames + .map(decodeTerminalStreamFrame) + .filter((frame) => frame?.opcode === TerminalStreamOpcode.Output) + .map((frame) => decodeTerminalStreamText(frame!.payload)) + .join('') +} + +function snapshotText(harness: ReturnType): 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') + }) +})