From fe28b2c18b5f59a6bfb7751eaac06181e1b4b8ba Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Fri, 21 Aug 2026 04:57:57 -0400 Subject: [PATCH] fix(native-chat): route SSH transcripts through owner host --- config/reliability-gates.jsonc | 143 ++++ .../mobile-native-chat-eligibility.test.ts | 12 +- .../session/mobile-native-chat-eligibility.ts | 5 +- .../use-mobile-native-chat-controller.ts | 3 +- .../use-mobile-native-chat-session.test.ts | 37 +- .../session/use-mobile-native-chat-session.ts | 35 +- .../server-hook-http-ingest.test.ts | 30 + .../server-ingest-terminal-status.test.ts | 21 + ...ver-last-status-hydrate-validation.test.ts | 44 ++ src/main/agent-hooks/server.ts | 30 +- src/main/ipc/native-chat.test.ts | 58 ++ src/main/ipc/native-chat.ts | 7 +- .../native-chat-transcript-owner.ts | 44 ++ src/main/native-chat/session-file-resolver.ts | 2 + .../ssh-transcript-host-routing.test.ts | 634 ++++++++++++++++++ .../native-chat/ssh-transcript-range-fs.ts | 93 +++ src/main/native-chat/ssh-transcript-read.ts | 91 +++ .../ssh-transcript-subscription.ts | 195 ++++++ .../native-chat/transcript-file-version.ts | 14 +- .../native-chat/transcript-host-verdict.ts | 46 ++ .../transcript-incremental-reader.ts | 78 ++- .../native-chat/transcript-native-watcher.ts | 9 + src/main/native-chat/transcript-range-fs.ts | 24 + .../native-chat/transcript-tail-boundary.ts | 34 +- .../native-chat/transcript-tail-reader.ts | 111 ++- .../native-chat/transcript-watch-contract.ts | 3 + .../native-chat/transcript-watch-engine.ts | 127 ++-- src/main/native-chat/transcript-watch-read.ts | 60 ++ .../transcript-watch-wsl-exact-path.test.ts | 12 + src/main/native-chat/transcript-watch.ts | 34 +- .../providers/ssh-filesystem-dispatch.test.ts | 31 + src/main/providers/ssh-filesystem-dispatch.ts | 32 + .../rpc/methods/native-chat-rpc-schema.ts | 40 ++ .../runtime/rpc/methods/native-chat.test.ts | 44 +- src/main/runtime/rpc/methods/native-chat.ts | 52 +- src/preload/api/native-chat-api.ts | 7 +- src/preload/index.ts | 12 +- .../native-chat/NativeChatSessionGate.tsx | 3 +- .../components/native-chat/NativeChatView.tsx | 3 + .../native-chat-live-session-types.ts | 28 + .../native-chat-live-source-key.ts | 16 + .../native-chat-pane-resolution.test.ts | 30 +- .../native-chat-pane-resolution.ts | 3 + .../native-chat-runtime-contract.ts | 3 +- .../native-chat-session-transport.test.ts | 57 ++ .../native-chat-session-transport.ts | 27 +- .../native-chat-subscription-identity.ts | 8 + .../native-chat/native-chat-view-types.ts | 2 + .../use-native-chat-live-session.ts | 112 ++-- ...se-native-chat-provider-session-locator.ts | 23 + .../use-native-chat-retained-session.ts | 3 +- ...-pty-transport-host-session-launch.test.ts | 6 +- .../remote-runtime-pty-transport.ts | 7 +- ...me-session-terminal-host-authority.test.ts | 6 +- .../src/runtime/web-runtime-session.ts | 9 +- src/shared/agent-session-resume.test.ts | 55 ++ src/shared/agent-session-resume.ts | 31 +- .../workspace-session-sleeping-agents.ts | 8 +- ...adless-native-chat-runtime-routing.spec.ts | 65 ++ 59 files changed, 2471 insertions(+), 288 deletions(-) create mode 100644 src/main/native-chat/native-chat-transcript-owner.ts create mode 100644 src/main/native-chat/ssh-transcript-host-routing.test.ts create mode 100644 src/main/native-chat/ssh-transcript-range-fs.ts create mode 100644 src/main/native-chat/ssh-transcript-read.ts create mode 100644 src/main/native-chat/ssh-transcript-subscription.ts create mode 100644 src/main/native-chat/transcript-host-verdict.ts create mode 100644 src/main/native-chat/transcript-range-fs.ts create mode 100644 src/main/native-chat/transcript-watch-read.ts create mode 100644 src/main/runtime/rpc/methods/native-chat-rpc-schema.ts create mode 100644 src/renderer/src/components/native-chat/native-chat-live-session-types.ts create mode 100644 src/renderer/src/components/native-chat/native-chat-live-source-key.ts create mode 100644 src/renderer/src/components/native-chat/native-chat-subscription-identity.ts create mode 100644 src/renderer/src/components/native-chat/use-native-chat-provider-session-locator.ts create mode 100644 tests/e2e/headless-native-chat-runtime-routing.spec.ts diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 12b937b187f..6877bc2c651 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -4640,6 +4640,149 @@ ], "demotionRule": "Demote if discovery and launch can resolve different authorities, a scoped Floating probe overwrites the active-project inventory, or local context selection adds recurring scans or retained in-flight entries." }, + { + "id": "native-chat.transcript-host-ownership", + "title": "Native chat reads and tails transcripts only on their owning execution host", + "maturity": "experimental", + "protection": "partial", + "owner": "native-chat", + "layer": "cross-boundary", + "surfaces": [ + "provider-session identity", + "agent hook ingress", + "transcript snapshot", + "transcript live tail", + "Electron IPC", + "paired runtime RPC", + "mobile native chat" + ], + "platforms": ["macos", "linux", "windows"], + "providers": ["local", "ssh", "wsl", "remote-runtime"], + "coveredPlatforms": ["macos", "linux"], + "coveredProviders": ["local", "ssh", "wsl", "remote-runtime"], + "coverageNotes": "A deterministic host-routing oracle covers local, exact SSH, unknown ownership, reconnect, provider replacement, bounded range reads, and WSL exact-path contracts. A Docker Linux SSH/relay target plus headed macOS Electron/CDP proves remote snapshot and tail rendering for SSH git and folder workspaces. A compiled windowless serve journey covers the shared runtime boundary.", + "motivatingLinks": [ + "https://github.com/stablyai/orca/issues/13663", + "https://github.com/stablyai/orca/pull/15515", + "https://github.com/stablyai/orca/pull/15517" + ], + "invariant": "A located provider session resolves and reads transcript bytes only through its stamped execution owner. Unknown or unavailable ownership is unverifiable and never falls back to the desktop filesystem. SSH snapshots and tails use bounded positional reads and reject stale provider generations; locator-free legacy requests retain host-local behavior.", + "oracle": "The focused oracle gives the desktop and SSH providers identical session ids and transcript paths but different bytes, then requires the exact SSH owner for snapshot and live append. It also proves unknown ownership cannot install a local watcher, provider loss and replacement rearm once, throwing advisory callbacks cannot kill retry, and large appends never call whole-file read or reread the prefix. The identical baseline/candidate/disabled/revert comparison and a real Docker SSH plus Electron/CDP journey supply independent red/green and rendered evidence.", + "commands": [ + "pnpm exec vitest run --config config/vitest.config.ts src/main/native-chat/ssh-transcript-host-routing.test.ts --reporter=dot", + "pnpm exec vitest run --config config/vitest.config.ts src/main/agent-hooks/server-hook-http-ingest.test.ts src/main/agent-hooks/server-ingest-terminal-status.test.ts src/main/agent-hooks/server-last-status-hydrate-validation.test.ts src/main/ipc/native-chat.test.ts src/main/providers/ssh-filesystem-dispatch.test.ts src/main/runtime/rpc/methods/native-chat.test.ts src/renderer/src/components/native-chat/native-chat-pane-resolution.test.ts src/renderer/src/components/native-chat/native-chat-session-transport.test.ts src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-host-session-launch.test.ts src/renderer/src/runtime/web-runtime-session-terminal-host-authority.test.ts src/shared/agent-session-resume.test.ts --reporter=dot", + "pnpm exec vitest run --config config/vitest.config.ts src/main/native-chat/transcript-watch-wsl-exact-path.test.ts --reporter=dot", + "pnpm --dir mobile exec vitest run --root .. mobile/src/session/mobile-native-chat-eligibility.test.ts mobile/src/session/use-mobile-native-chat-session.test.ts --reporter=dot", + "pnpm exec electron-vite build --mode e2e && SKIP_BUILD=1 pnpm exec playwright test tests/e2e/headless-native-chat-runtime-routing.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1", + "Isolated headed Orca Electron/CDP against a throwaway Docker Linux SSH/relay target, with same-path desktop poison and remote-only live append" + ], + "testFiles": [ + "src/main/native-chat/ssh-transcript-host-routing.test.ts", + "src/main/native-chat/transcript-watch-wsl-exact-path.test.ts", + "src/main/agent-hooks/server-hook-http-ingest.test.ts", + "src/main/agent-hooks/server-ingest-terminal-status.test.ts", + "src/main/agent-hooks/server-last-status-hydrate-validation.test.ts", + "src/main/ipc/native-chat.test.ts", + "src/main/providers/ssh-filesystem-dispatch.test.ts", + "src/main/runtime/rpc/methods/native-chat.test.ts", + "src/renderer/src/components/native-chat/native-chat-pane-resolution.test.ts", + "src/renderer/src/components/native-chat/native-chat-session-transport.test.ts", + "src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-host-session-launch.test.ts", + "src/renderer/src/runtime/web-runtime-session-terminal-host-authority.test.ts", + "src/shared/agent-session-resume.test.ts", + "mobile/src/session/mobile-native-chat-eligibility.test.ts", + "mobile/src/session/use-mobile-native-chat-session.test.ts", + "tests/e2e/headless-native-chat-runtime-routing.spec.ts" + ], + "assertionRefs": [ + { + "file": "src/main/native-chat/ssh-transcript-host-routing.test.ts", + "assertions": [ + "the exact SSH owner wins over desktop and other-SSH same-id/path lookalikes", + "unknown ownership never reads or watches locally", + "remote snapshot and append survive provider loss, replacement, and advisory callback failure", + "every remote request is at most 256 KiB and a large append does not reread the prefix or call readFile" + ] + }, + { + "file": "src/main/agent-hooks/server-hook-http-ingest.test.ts", + "assertions": [ + "local hook ingress stamps local ownership", + "authenticated remote hook ingress overwrites a spoofed owner with the connection's SSH owner" + ] + }, + { + "file": "src/main/runtime/rpc/methods/native-chat.test.ts", + "assertions": [ + "runtime read and subscription forward the normalized provider-session locator" + ] + }, + { + "file": "tests/e2e/headless-native-chat-runtime-routing.spec.ts", + "assertions": [ + "a compiled windowless runtime reads an explicit local owner", + "the same real file is unverifiable through a located session whose owner is absent" + ] + } + ], + "evidenceRuns": [ + { + "date": "2026-08-21", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/native-chat/ssh-transcript-host-routing.test.ts --reporter=dot", + "result": "passed", + "durationSeconds": 3, + "summary": "The finalized candidate passed all 16 deterministic ownership, reconnect, replacement-race, advisory, legacy, binding, and bounded-range cases; disabling exact SSH-owner recognition failed 11/16 and restoration returned to 16/16. The byte-identical earlier 13-case oracle failed 11/13 on latest origin/main and the advisory-only PR head, passed 13/13 on the candidate, failed 9/13 with the owner fix disabled, and returned to 13/13 after restoration." + }, + { + "date": "2026-08-21", + "runner": "manual", + "platform": "macos", + "command": "Isolated headed Orca Electron/CDP against a throwaway Docker Linux SSH/relay target, with same-path desktop poison and remote-only live append", + "result": "passed", + "durationSeconds": 180, + "summary": "SSH git-worktree and folder-workspace native chat rendered the real remote initial snapshot and later remote-only append; DOM assertions found neither desktop poison marker. The remote bytes were independently read inside Docker while the desktop poison file remained present." + }, + { + "date": "2026-08-21", + "runner": "local", + "platform": "macos", + "command": "pnpm exec electron-vite build --mode e2e && SKIP_BUILD=1 pnpm exec playwright test tests/e2e/headless-native-chat-runtime-routing.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1", + "result": "passed", + "durationSeconds": 9, + "summary": "The compiled headless runtime passed explicit-owner read and unknown-owner no-fallback assertions through the registered runtime RPC." + } + ], + "runtimeBudget": { + "p95Seconds": 15, + "scope": "16-case deterministic routing oracle plus one compiled headless runtime journey; Docker/headed Electron remains manual" + }, + "flakeHistory": { + "status": "soaking", + "evidence": "Repeated focused local oracles and the compiled headless journey pass; long-term CI history is not yet available." + }, + "redGreenEvidence": { + "status": "complete", + "evidence": "The byte-identical 13-case oracle failed 11/13 on latest origin/main 59892a2f13 and advisory-only PR head aba8aafafa, passed 13/13 on the candidate, failed 9/13 when only exact SSH-owner recognition was disabled, and passed 13/13 after restoration. The finalized 16-case candidate passed 16/16, failed 11/16 under the same mutation, and returned to 16/16 after restoration. Baseline headed Electron visibly selected desktop lookalike bytes in both SSH workspace kinds; the candidate rendered only remote snapshot and tail bytes." + }, + "performanceBudget": { + "required": true, + "evidence": "SSH reads require readFileRange support, cap each request at the shared 256 KiB maximum, and tail from the last byte offset. The large-append oracle requires total requested bytes no more than append size plus 256 bytes, rejects any whole-file read, and verifies the prefix is not polled again." + }, + "promotionCriteria": [ + "Collect 100 consecutive focused CI passes or 14 days of soak history.", + "Keep the Docker SSH git and folder workspace snapshot/tail journey green when transcript routing changes.", + "Add physical Windows SSH and WSL live evidence before claiming full platform coverage.", + "Exercise mixed-version live pairing against at least one older client and one older host." + ], + "knownGaps": [ + "Physical Windows SSH and live WSL transcript journeys were not run; WSL exact-path behavior is contract-tested.", + "The live Docker target used the current relay; older live relay and mixed-version pairing remain contract-only.", + "SSH provider sessions without an authoritative transcript path fail closed instead of running a broad remote session-id scan." + ], + "demotionRule": "Demote if a located session can read a non-owner filesystem, provider loss selects local bytes, reconnect leaves duplicate or stale watchers, any SSH poll uses whole-file reads, or the bounded-range oracle flakes without a product or harness defect." + }, { "id": "agent-session.provider-ownership", "title": "Provider sessions are resumed once per workspace ownership claim", diff --git a/mobile/src/session/mobile-native-chat-eligibility.test.ts b/mobile/src/session/mobile-native-chat-eligibility.test.ts index 7daa4babea6..bc0cceda048 100644 --- a/mobile/src/session/mobile-native-chat-eligibility.test.ts +++ b/mobile/src/session/mobile-native-chat-eligibility.test.ts @@ -59,7 +59,12 @@ describe('resolveMobileNativeChat', () => { ).toEqual({ agent: 'claude', sessionId: 'sess-1', - transcriptPath: '/tmp/claude-real-transcript.jsonl' + transcriptPath: '/tmp/claude-real-transcript.jsonl', + providerSession: { + key: 'session_id', + id: 'sess-1', + transcriptPath: '/tmp/claude-real-transcript.jsonl' + } }) }) @@ -69,14 +74,15 @@ describe('resolveMobileNativeChat', () => { type: 'terminal', agentStatus: status({ agentType: 'codex' }) }) - ).toEqual({ agent: 'codex', sessionId: null, transcriptPath: null }) + ).toEqual({ agent: 'codex', sessionId: null, transcriptPath: null, providerSession: null }) }) it('admits OpenClaude with its distinct agent identity', () => { expect(resolveMobileNativeChat({ type: 'terminal', launchAgent: 'openclaude' })).toEqual({ agent: 'openclaude', sessionId: null, - transcriptPath: null + transcriptPath: null, + providerSession: null }) }) diff --git a/mobile/src/session/mobile-native-chat-eligibility.ts b/mobile/src/session/mobile-native-chat-eligibility.ts index abda64b04ab..712e07e66c4 100644 --- a/mobile/src/session/mobile-native-chat-eligibility.ts +++ b/mobile/src/session/mobile-native-chat-eligibility.ts @@ -1,4 +1,5 @@ import type { AgentStatusEntry } from '../../../src/shared/agent-status-types' +import type { AgentProviderSessionMetadata } from '../../../src/shared/agent-session-resume' import { isRuntimeOwnedSshTargetId } from '../../../src/shared/execution-host' import { isNativeChatSupportedAgent, @@ -23,6 +24,7 @@ export type MobileNativeChatResolution = { /** Hook-reported transcript path. Recent Claude sessions cannot always be * resolved from the provider session id, so mobile forwards this to runtime. */ transcriptPath: string | null + providerSession: AgentProviderSessionMetadata | null } export type MobileNativeChatTab = { @@ -60,7 +62,8 @@ export function resolveMobileNativeChat( return { agent, sessionId: tab.agentStatus?.providerSession?.id ?? null, - transcriptPath: tab.agentStatus?.providerSession?.transcriptPath ?? null + transcriptPath: tab.agentStatus?.providerSession?.transcriptPath ?? null, + providerSession: tab.agentStatus?.providerSession ?? null } } diff --git a/mobile/src/session/use-mobile-native-chat-controller.ts b/mobile/src/session/use-mobile-native-chat-controller.ts index 27ad8049c1d..9ebba69a4e6 100644 --- a/mobile/src/session/use-mobile-native-chat-controller.ts +++ b/mobile/src/session/use-mobile-native-chat-controller.ts @@ -85,7 +85,8 @@ export function useMobileNativeChatController(args: { sourceIdentity: encodeNativeChatTranscriptIdentity([hostId, worktreeId]), agent: activeChatResolution?.agent ?? null, sessionId: activeChatSessionId, - transcriptPath: activeChatResolution?.transcriptPath ?? null + transcriptPath: activeChatResolution?.transcriptPath ?? null, + providerSession: activeChatResolution?.providerSession ?? null }) const { composerText: chatComposerText, diff --git a/mobile/src/session/use-mobile-native-chat-session.test.ts b/mobile/src/session/use-mobile-native-chat-session.test.ts index 2ff97a3370c..d3cc25019d9 100644 --- a/mobile/src/session/use-mobile-native-chat-session.test.ts +++ b/mobile/src/session/use-mobile-native-chat-session.test.ts @@ -2,6 +2,7 @@ import { createElement } from 'react' import { act, create, type ReactTestRenderer } from 'react-test-renderer' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { NativeChatMessage } from '../../../src/shared/native-chat-types' +import type { AgentProviderSessionMetadata } from '../../../src/shared/agent-session-resume' import type { RpcClient } from '../transport/rpc-client' import { useMobileNativeChatSession, @@ -32,23 +33,51 @@ describe('useMobileNativeChatSession', () => { renderer = null }) - function Harness({ client }: { client: RpcClient | null }): null { + function Harness({ + client, + providerSession + }: { + client: RpcClient | null + providerSession?: AgentProviderSessionMetadata + }): null { state = useMobileNativeChatSession({ client, sourceIdentity: 'host-a\0workspace-a', agent: 'claude', sessionId: 'session', - transcriptPath: null + transcriptPath: null, + providerSession }) return null } - async function mount(client: RpcClient): Promise { + async function mount( + client: RpcClient, + providerSession?: AgentProviderSessionMetadata + ): Promise { await act(async () => { - renderer = create(createElement(Harness, { client })) + renderer = create(createElement(Harness, { client, providerSession })) }) } + it('forwards SSH ownership through the mobile subscribe request', async () => { + const providerSession = { + key: 'session_id' as const, + id: 'session', + transcriptPath: '/tmp/session.jsonl', + executionHostId: 'ssh:builder' as const + } + const subscribe = vi.fn(() => () => {}) + + await mount({ subscribe } as unknown as RpcClient, providerSession) + + expect(subscribe).toHaveBeenCalledWith( + 'nativeChat.subscribe', + expect.objectContaining({ providerSession }), + expect.any(Function) + ) + }) + it('drops an older-page response captured before transcript replacement', async () => { let resolveEarlier: (response: unknown) => void = () => {} const sendRequest = vi.fn( diff --git a/mobile/src/session/use-mobile-native-chat-session.ts b/mobile/src/session/use-mobile-native-chat-session.ts index a79b5648dd7..a12093b0fcf 100644 --- a/mobile/src/session/use-mobile-native-chat-session.ts +++ b/mobile/src/session/use-mobile-native-chat-session.ts @@ -1,10 +1,11 @@ -import { useCallback, useEffect, useRef, useState } from 'react' +import { useCallback, useEffect, useMemo, useRef, useState } from 'react' import { createNativeChatTranscriptRetention, encodeNativeChatTranscriptIdentity } from '../../../src/shared/native-chat-transcript-retention' import { createNativeChatMerger, replaceList } from '../../../src/shared/native-chat-merge' import type { NativeChatMessage } from '../../../src/shared/native-chat-types' +import type { AgentProviderSessionMetadata } from '../../../src/shared/agent-session-resume' import { buildNativeChatSubscriptionId } from '../../../src/shared/native-chat-stream-unsubscribe' import type { RpcClient } from '../transport/rpc-client' import { @@ -51,14 +52,32 @@ export function useMobileNativeChatSession(args: { agent: string | null sessionId: string | null transcriptPath: string | null + providerSession?: AgentProviderSessionMetadata | null }): MobileNativeChatSession { - const { client, sourceIdentity, agent, sessionId, transcriptPath } = args + const { client, sourceIdentity, agent, sessionId, transcriptPath, providerSession } = args const [messages, setMessages] = useState([]) + const providerSessionKey = providerSession?.key + const providerSessionId = providerSession?.id + const providerTranscriptPath = providerSession?.transcriptPath + const providerExecutionHostId = providerSession?.executionHostId + const providerSessionLocator = useMemo( + () => + providerSessionKey && providerSessionId + ? { + key: providerSessionKey, + id: providerSessionId, + ...(providerTranscriptPath ? { transcriptPath: providerTranscriptPath } : {}), + ...(providerExecutionHostId ? { executionHostId: providerExecutionHostId } : {}) + } + : null, + [providerExecutionHostId, providerSessionId, providerSessionKey, providerTranscriptPath] + ) const identity = encodeNativeChatTranscriptIdentity([ sourceIdentity, agent, sessionId, - transcriptPath + transcriptPath, + providerSessionLocator?.executionHostId ?? null ]) // Pre-read status is a pure function of the props, so derive it rather than // letting the effect write it a commit later. @@ -145,7 +164,8 @@ export function useMobileNativeChatSession(args: { sessionId, limit: limitRef.current, subscriptionId: buildNativeChatSubscriptionId(agent, sessionId), - ...(transcriptPath ? { transcriptPath } : {}) + ...(transcriptPath ? { transcriptPath } : {}), + ...(providerSessionLocator ? { providerSession: providerSessionLocator } : {}) }, (raw) => { if (cancelled) { @@ -206,7 +226,7 @@ export function useMobileNativeChatSession(args: { cancelled = true unsubscribe() } - }, [client, agent, sessionId, transcriptPath, identity, setList]) + }, [client, agent, sessionId, transcriptPath, providerSessionLocator, identity, setList]) const loadEarlier = useCallback(() => { if (!client || !agent || !sessionId || loadingEarlierRef.current || !hasMore) { @@ -232,7 +252,8 @@ export function useMobileNativeChatSession(args: { sessionId, limit: beforeOffset === null ? nextLimit : pageLimit, ...(beforeOffset === null ? {} : { beforeOffset }), - ...(transcriptPath ? { transcriptPath } : {}) + ...(transcriptPath ? { transcriptPath } : {}), + ...(providerSessionLocator ? { providerSession: providerSessionLocator } : {}) }) if (!response.ok) { return @@ -271,7 +292,7 @@ export function useMobileNativeChatSession(args: { } } })() - }, [client, agent, sessionId, transcriptPath, hasMore, setList]) + }, [client, agent, sessionId, transcriptPath, providerSessionLocator, hasMore, setList]) // Held for any unsettled read, not just an in-flight one: a stream error or a // dropped client would otherwise trade the conversation for an error card. diff --git a/src/main/agent-hooks/server-hook-http-ingest.test.ts b/src/main/agent-hooks/server-hook-http-ingest.test.ts index 622fc7d428b..1040ad1a143 100644 --- a/src/main/agent-hooks/server-hook-http-ingest.test.ts +++ b/src/main/agent-hooks/server-hook-http-ingest.test.ts @@ -29,6 +29,36 @@ afterEach(() => { }) describe('AgentHookServer listener replay', () => { + it('stamps authenticated loopback provider sessions as local', async () => { + const server = new AgentHookServer() + await server.start({ env: 'production' }) + try { + const env = server.buildPtyEnv() + const response = await fetch(`http://127.0.0.1:${env.ORCA_AGENT_HOOK_PORT}/hook/claude`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Orca-Agent-Hook-Token': env.ORCA_AGENT_HOOK_TOKEN + }, + body: JSON.stringify( + buildBody({ + hook_event_name: 'UserPromptSubmit', + prompt: 'local owner', + session_id: 'local-session' + }) + ) + }) + + expect(response.status).toBe(204) + expect(server.getStatusSnapshot()[0]?.providerSession).toEqual({ + key: 'session_id', + id: 'local-session', + executionHostId: 'local' + }) + } finally { + server.stop() + } + }) it('ignores local nested Claude Stop while a parent Codex hook status is active', async () => { const server = new AgentHookServer() await server.start({ env: 'production' }) diff --git a/src/main/agent-hooks/server-ingest-terminal-status.test.ts b/src/main/agent-hooks/server-ingest-terminal-status.test.ts index 8661701e1dc..49d102416dd 100644 --- a/src/main/agent-hooks/server-ingest-terminal-status.test.ts +++ b/src/main/agent-hooks/server-ingest-terminal-status.test.ts @@ -27,6 +27,27 @@ afterEach(() => { }) describe('AgentHookServer ingestTerminalStatus', () => { + it('overwrites remote provider-session ownership with the authenticated connection', () => { + const server = new AgentHookServer() + server.ingestRemote( + { + paneKey: PANE, + providerSession: { + key: 'session_id', + id: 'session-1', + executionHostId: 'ssh:spoofed' + }, + payload: { state: 'working', prompt: 'route transcript', agentType: 'claude' } + }, + 'real-owner' + ) + + expect(server.getStatusSnapshot()[0]?.providerSession).toEqual({ + key: 'session_id', + id: 'session-1', + executionHostId: 'ssh:real-owner' + }) + }) it('preserves a hook turn stamp when an OSC repaint omits hook-only completion text', () => { const server = new AgentHookServer() const listener = vi.fn() diff --git a/src/main/agent-hooks/server-last-status-hydrate-validation.test.ts b/src/main/agent-hooks/server-last-status-hydrate-validation.test.ts index 3ad27b64a38..9e21f5ebef9 100644 --- a/src/main/agent-hooks/server-last-status-hydrate-validation.test.ts +++ b/src/main/agent-hooks/server-last-status-hydrate-validation.test.ts @@ -121,6 +121,50 @@ describe('Last-status persistence', () => { } }) + it('upgrades persisted provider sessions with their stored execution owner', async () => { + mkdirSync(join(userDataPath, 'agent-hooks'), { recursive: true }) + const receivedAt = recentTs() + writeFileSync( + lastStatusPath(), + JSON.stringify({ + version: 2, + entries: { + [PANE]: { + paneKey: PANE, + tabId: 'tab-1', + worktreeId: 'wt-1', + connectionId: 'persisted-ssh-owner', + receivedAt, + stateStartedAt: receivedAt, + providerSession: { + key: 'session_id', + id: 'persisted-session', + transcriptPath: '/remote/session.jsonl' + }, + payload: { state: 'done', prompt: '', agentType: 'claude' } + } + } + }), + 'utf8' + ) + const server = new AgentHookServer() + await server.start({ env: 'production', userDataPath }) + try { + const listener = vi.fn() + server.setListener(listener) + expect(listener).toHaveBeenCalledWith( + expect.objectContaining({ + providerSession: expect.objectContaining({ + id: 'persisted-session', + executionHostId: 'ssh:persisted-ssh-owner' + }) + }) + ) + } finally { + server.stop() + } + }) + it('rejects a stale version mismatch on hydrate', async () => { mkdirSync(join(userDataPath, 'agent-hooks'), { recursive: true }) writeFileSync( diff --git a/src/main/agent-hooks/server.ts b/src/main/agent-hooks/server.ts index 2b0457e064a..0b2ba5b3fe6 100644 --- a/src/main/agent-hooks/server.ts +++ b/src/main/agent-hooks/server.ts @@ -97,6 +97,7 @@ import { normalizeAgentProviderSession, type AgentProviderSessionMetadata } from '../../shared/agent-session-resume' +import { LOCAL_EXECUTION_HOST_ID, toSshExecutionHostId } from '../../shared/execution-host' import { isCommandCodeNewTurnWhileWorking } from '../../shared/command-code-turn-boundary' export type { AgentHookSource } @@ -329,7 +330,14 @@ function sanitizeHydratedEntry( if (!payload) { return null } - const providerSession = normalizeAgentProviderSession(record.providerSession) ?? undefined + const normalizedProviderSession = + normalizeAgentProviderSession(record.providerSession) ?? undefined + const providerSession: AgentProviderSessionMetadata | undefined = normalizedProviderSession + ? { + ...normalizedProviderSession, + executionHostId: connectionId ? toSshExecutionHostId(connectionId) : LOCAL_EXECUTION_HOST_ID + } + : undefined const providerSessionOnly = record.providerSessionOnly === true const retainedForLiveness = record.retainedForLiveness === true const validRetainedIdentity = Boolean( @@ -1340,6 +1348,17 @@ export class AgentHookServer { onAccepted?: () => void, origin: AgentStatusObservationOrigin = 'hook' ): EnrichedAgentHookEventPayload { + if (payload.providerSession) { + payload = { + ...payload, + providerSession: { + ...payload.providerSession, + executionHostId: payload.connectionId + ? toSshExecutionHostId(payload.connectionId) + : LOCAL_EXECUTION_HOST_ID + } + } + } if (payload.hookEventName === 'UserPromptSubmit') { // Why: the prompt boundary is authoritative even when text is unchanged; its next OSC working row must not inherit the prior cron/background turn stamp. this.activeHookTurnCompletedAtByPaneKey.delete(payload.paneKey) @@ -2310,7 +2329,14 @@ export class AgentHookServer { typeof envelope.toolAgentType === 'string' && envelope.toolAgentType.trim().length > 0 ? envelope.toolAgentType.trim() : undefined - const providerSession = normalizeAgentProviderSession(envelope.providerSession) ?? undefined + const normalizedProviderSession = + normalizeAgentProviderSession(envelope.providerSession) ?? undefined + const providerSession: AgentProviderSessionMetadata | undefined = normalizedProviderSession + ? { + ...normalizedProviderSession, + executionHostId: toSshExecutionHostId(trimmedConnectionId) + } + : undefined // Why: relay crosses a trust boundary — re-run the canonical normalizer to enforce caps/invariants (returns null on malformed). const validatedPayload = normalizeAgentStatusPayload(envelope.payload) if (!validatedPayload) { diff --git a/src/main/ipc/native-chat.test.ts b/src/main/ipc/native-chat.test.ts index 4d0e1765a44..4950b0dc1a3 100644 --- a/src/main/ipc/native-chat.test.ts +++ b/src/main/ipc/native-chat.test.ts @@ -2,6 +2,12 @@ import { appendFile, mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { AgentProviderSessionMetadata } from '../../shared/agent-session-resume' +import type { IFilesystemProvider } from '../providers/types' +import { + registerSshFilesystemProvider, + unregisterSshFilesystemProvider +} from '../providers/ssh-filesystem-dispatch' const { handlers, listeners } = vi.hoisted(() => ({ handlers: new Map unknown>(), @@ -59,6 +65,7 @@ async function invokeReadSession(args: { sessionId: string limit?: number transcriptPath?: string + providerSession?: AgentProviderSessionMetadata }): Promise { registerNativeChatHandlers() const handler = handlers.get('nativeChat:readSession') @@ -69,6 +76,57 @@ async function invokeReadSession(args: { } describe('nativeChat:readSession handler', () => { + it('routes a stamped SSH provider session through ranged desktop IPC reads', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-ipc-ssh-')) + tempRoots.push(root) + const transcriptPath = join(root, 'same-path.jsonl') + await writeFile( + transcriptPath, + `${JSON.stringify({ + type: 'assistant', + uuid: 'desktop-poison', + message: { role: 'assistant', content: [{ type: 'text', text: 'wrong host' }] } + })}\n` + ) + const remoteBytes = Buffer.from( + `${JSON.stringify({ + type: 'assistant', + uuid: 'remote-real', + message: { role: 'assistant', content: [{ type: 'text', text: 'right host' }] } + })}\n` + ) + const readFile = vi.fn(async () => { + throw new Error('whole-file read forbidden') + }) + const provider = { + stat: async () => ({ size: remoteBytes.length, type: 'file', mtime: 1, mtimeMs: 1 }), + readFile, + readFileRange: async (_path: string, position: number, length: number) => { + const bytes = remoteBytes.subarray(position, position + length) + return { bytes, bytesRead: bytes.length } + }, + supportsFileRangeRead: async () => true + } as unknown as IFilesystemProvider + registerSshFilesystemProvider('ipc-owner', provider) + try { + const result = await invokeReadSession({ + agent: 'claude', + sessionId: 'same-session', + transcriptPath, + providerSession: { + key: 'session_id', + id: 'same-session', + transcriptPath, + executionHostId: 'ssh:ipc-owner' + } + }) + expect(result).toMatchObject({ messages: [{ id: 'remote-real' }] }) + expect(result).not.toMatchObject({ messages: [{ id: 'desktop-poison' }] }) + expect(readFile).not.toHaveBeenCalled() + } finally { + unregisterSshFilesystemProvider('ipc-owner') + } + }) it('preserves notFound so a just-created session stays in retry/loading', async () => { const result = (await invokeReadSession({ agent: 'claude', diff --git a/src/main/ipc/native-chat.ts b/src/main/ipc/native-chat.ts index 4ef0f48b872..2cc2a637c51 100644 --- a/src/main/ipc/native-chat.ts +++ b/src/main/ipc/native-chat.ts @@ -4,6 +4,7 @@ import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types' +import type { AgentProviderSessionMetadata } from '../../shared/agent-session-resume' import { clearNativeChatTranscriptCache } from '../native-chat/transcript-read-cache' import type { ReadTranscriptResult } from '../native-chat/transcript-reader' import { @@ -26,6 +27,7 @@ export type NativeChatReadSessionArgs = { /** Authoritative transcript path from the agent hook (providerSession), used to * locate the file when the session id no longer names it (recent Claude Code). */ transcriptPath?: string + providerSession?: AgentProviderSessionMetadata } // Why: render and parse only the recent window so long transcripts do not stall @@ -40,6 +42,7 @@ async function readSession(args: NativeChatReadSessionArgs): Promise 0 ? Math.floor(args.limit) : DESKTOP_READ_WINDOW // Replace any prior subscription under the same id (session change/resubscribe). const pending = beginPendingSubscription(sender.id, subscriptionId) @@ -180,6 +184,7 @@ async function handleSubscribe(event: IpcMainEvent, args: NativeChatSubscribeArg agent, sessionId, transcriptPath, + providerSession, initialLimit: limit, onInitialSnapshot: (messages, hasMore, _beforeOffset, error, lifecycle) => { if (sender.isDestroyed()) { diff --git a/src/main/native-chat/native-chat-transcript-owner.ts b/src/main/native-chat/native-chat-transcript-owner.ts new file mode 100644 index 00000000000..9c99490185a --- /dev/null +++ b/src/main/native-chat/native-chat-transcript-owner.ts @@ -0,0 +1,44 @@ +import { extname } from 'node:path' +import { + normalizeAgentProviderSession, + type AgentProviderSessionMetadata +} from '../../shared/agent-session-resume' +import { parseExecutionHostId } from '../../shared/execution-host' + +export type NativeChatTranscriptOwner = + | { kind: 'legacy-local' } + | { kind: 'local'; providerSession: AgentProviderSessionMetadata } + | { + kind: 'ssh' + connectionId: string + providerSession: AgentProviderSessionMetadata + transcriptPath: string | null + } + | { kind: 'unknown' } + +export function resolveNativeChatTranscriptOwner(args: { + sessionId: string + providerSession?: AgentProviderSessionMetadata +}): NativeChatTranscriptOwner { + if (args.providerSession === undefined) { + return { kind: 'legacy-local' } + } + const providerSession = normalizeAgentProviderSession(args.providerSession) + if (!providerSession || providerSession.id !== args.sessionId) { + return { kind: 'unknown' } + } + const host = parseExecutionHostId(providerSession.executionHostId) + if (host?.kind === 'local') { + return { kind: 'local', providerSession } + } + if (host?.kind === 'ssh') { + const transcriptPath = providerSession.transcriptPath?.trim() + return { + kind: 'ssh', + connectionId: host.targetId, + providerSession, + transcriptPath: transcriptPath && extname(transcriptPath) === '.jsonl' ? transcriptPath : null + } + } + return { kind: 'unknown' } +} diff --git a/src/main/native-chat/session-file-resolver.ts b/src/main/native-chat/session-file-resolver.ts index a45850f2d46..dc138bffec5 100644 --- a/src/main/native-chat/session-file-resolver.ts +++ b/src/main/native-chat/session-file-resolver.ts @@ -1,6 +1,7 @@ import { homedir } from 'node:os' import { basename, extname, join } from 'node:path' import type { AgentType } from '../../shared/native-chat-types' +import type { AgentProviderSessionMetadata } from '../../shared/agent-session-resume' import { resolveNativeChatTranscriptAgent, type NativeChatTranscriptAgent @@ -72,6 +73,7 @@ export type ResolveSessionFileOptions = { * directly — recent Claude Code names the transcript with a UUID that differs * from the hook session_id, so the id-based glob below would miss it. */ transcriptPath?: string + providerSession?: AgentProviderSessionMetadata } /** diff --git a/src/main/native-chat/ssh-transcript-host-routing.test.ts b/src/main/native-chat/ssh-transcript-host-routing.test.ts new file mode 100644 index 00000000000..92fb56d5a45 --- /dev/null +++ b/src/main/native-chat/ssh-transcript-host-routing.test.ts @@ -0,0 +1,634 @@ +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { AgentProviderSessionMetadata } from '../../shared/agent-session-resume' +import type { FileRangeReadResult, FileStat, IFilesystemProvider } from '../providers/types' +import { MAX_FILE_RANGE_READ_BYTES } from '../../shared/file-range-read' + +const mocks = vi.hoisted(() => ({ + getProvider: vi.fn<(connectionId: string) => IFilesystemProvider | undefined>(), + generation: 1, + changeListeners: new Set<(connectionId: string, generation: number) => void>() +})) + +vi.mock('../providers/ssh-filesystem-dispatch', () => ({ + getSshFilesystemProvider: mocks.getProvider, + getSshFilesystemProviderSnapshot: (connectionId: string) => { + const provider = mocks.getProvider(connectionId) + return provider ? { provider, generation: mocks.generation } : null + }, + onSshFilesystemProviderRegistered: () => () => undefined, + onSshFilesystemProviderChanged: ( + listener: (connectionId: string, generation: number) => void + ) => { + mocks.changeListeners.add(listener) + return () => mocks.changeListeners.delete(listener) + }, + SSH_FILESYSTEM_PROVIDER_UNAVAILABLE_MESSAGE: 'Remote connection dropped.' +})) + +import { + nativeChatLineDecoderForAgent, + readNativeChatTranscriptTail +} from './transcript-tail-reader' +import { getActiveNativeChatWatcherCount, subscribeNativeChatTranscript } from './transcript-watch' +import { readIncrementalTranscriptMessages } from './transcript-incremental-reader' +import { TranscriptRangeReadInvalidatedError, type TranscriptRangeFs } from './transcript-range-fs' + +const SESSION_ID = 'session-ssh-owner' +const EXECUTION_HOST_ID = 'ssh:ssh-owner' + +function claudeLine(id: string, text: string): string { + return `${JSON.stringify({ + type: 'assistant', + uuid: id, + message: { role: 'assistant', content: [{ type: 'text', text }] } + })}\n` +} + +function memoryProvider(files: Map) { + const readFile = vi.fn(async () => { + throw new Error('whole-file reads are forbidden for transcript tailing') + }) + const readFileRange = vi.fn( + async (filePath: string, position: number, length: number): Promise => { + const bytes = files.get(filePath)?.subarray(position, position + length) + if (!bytes) { + throw Object.assign(new Error(`ENOENT: no such file or directory, open '${filePath}'`), { + code: 'ENOENT' + }) + } + return { bytes, bytesRead: bytes.length } + } + ) + const provider = { + async stat(filePath: string): Promise { + const bytes = files.get(filePath) + if (!bytes) { + throw Object.assign(new Error(`ENOENT: no such file or directory, stat '${filePath}'`), { + code: 'ENOENT' + }) + } + return { size: bytes.length, type: 'file', mtime: 1, mtimeMs: 1, dev: 7, ino: 11 } + }, + readFile, + readFileRange, + async supportsFileRangeRead() { + return true + } + } as unknown as IFilesystemProvider + return { provider, readFile, readFileRange } +} + +function routedArgs(transcriptPath: string): { + agent: 'claude' + sessionId: string + transcriptPath: string + providerSession: AgentProviderSessionMetadata +} { + return { + agent: 'claude' as const, + sessionId: SESSION_ID, + transcriptPath, + providerSession: { + key: 'session_id' as const, + id: SESSION_ID, + transcriptPath, + executionHostId: EXECUTION_HOST_ID + } + } +} + +function changeProvider(provider: IFilesystemProvider | undefined): void { + mocks.getProvider.mockReturnValue(provider) + mocks.generation++ + for (const listener of mocks.changeListeners) { + listener('ssh-owner', mocks.generation) + } +} + +let tempRoots: string[] = [] + +beforeEach(() => { + mocks.getProvider.mockReset() + mocks.generation = 1 + mocks.changeListeners.clear() + tempRoots = [] +}) + +afterEach(async () => { + await Promise.all(tempRoots.map((root) => rm(root, { recursive: true, force: true }))) +}) + +describe('native chat SSH transcript host routing (#13663)', () => { + it('reads the owning provider while rejecting a desktop same-path lookalike', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-host-route-')) + tempRoots.push(root) + const transcriptPath = join(root, 'session.jsonl') + await writeFile(transcriptPath, claudeLine('desktop-poison', 'wrong host')) + const files = new Map([[transcriptPath, Buffer.from(claudeLine('remote-real', 'right host'))]]) + const { provider, readFile, readFileRange } = memoryProvider(files) + mocks.getProvider.mockReturnValue(provider) + + const result = await readNativeChatTranscriptTail({ + ...routedArgs(transcriptPath), + limit: 40 + } as Parameters[0]) + + expect(result).toMatchObject({ messages: [{ id: 'remote-real' }], hasMore: false }) + expect(result).not.toMatchObject({ messages: [{ id: 'desktop-poison' }] }) + expect(mocks.getProvider).toHaveBeenCalledWith('ssh-owner') + expect(readFile).not.toHaveBeenCalled() + expect(readFileRange).toHaveBeenCalled() + }) + + it('retries a ranged snapshot when the remote file is replaced mid-read', async () => { + const transcriptPath = '/tmp/replaced-mid-read.jsonl' + const oldBytes = Buffer.from( + `${claudeLine('old-large', 'a'.repeat(80_000))}${claudeLine('old-tail', 'old')}` + ) + const newBytes = Buffer.from( + `${claudeLine('new-large', 'b'.repeat(80_000))}${claudeLine('new-tail', 'new')}` + ) + let current = oldBytes + let inode = 11 + let rangeReadCount = 0 + const readFileRange = vi.fn( + async (_filePath: string, position: number, length: number): Promise => { + const bytes = current.subarray(position, position + length) + rangeReadCount++ + if (rangeReadCount === 3) { + current = newBytes + inode = 12 + } + return { bytes, bytesRead: bytes.length } + } + ) + const provider = { + async stat(): Promise { + return { + size: current.length, + type: 'file', + mtime: inode, + mtimeMs: inode, + dev: 7, + ino: inode + } + }, + readFile: vi.fn(async () => { + throw new Error('whole-file reads are forbidden for transcript tailing') + }), + readFileRange, + async supportsFileRangeRead() { + return true + } + } as unknown as IFilesystemProvider + mocks.getProvider.mockReturnValue(provider) + + const result = await readNativeChatTranscriptTail({ + ...routedArgs(transcriptPath), + limit: 40 + }) + + expect(result).toMatchObject({ messages: [{ id: 'new-large' }, { id: 'new-tail' }] }) + expect(result).not.toMatchObject({ messages: [{ id: 'old-large' }, { id: 'old-tail' }] }) + expect(readFileRange.mock.calls.length).toBeGreaterThan(4) + }) + + it('withholds remote append batches and restores the cursor after invalidation', async () => { + const bytes = Buffer.from( + Array.from({ length: 50 }, (_, index) => claudeLine(`stale-${index}`, `stale ${index}`)).join( + '' + ) + ) + const onBatch = vi.fn() + const state = { + offset: 0, + pendingChunks: [] as Buffer[], + pendingStart: 0, + pendingBytes: 0, + droppingOversizedRecord: false + } + const rangeFs: TranscriptRangeFs = { + async stat() { + return { + size: bytes.length, + identity: '1:7:11', + identityReliable: true, + mtimeMs: 1, + ctimeMs: 1 + } + }, + async read(_filePath, position, length) { + return bytes.subarray(position, position + length) + }, + async assertStable() { + throw new TranscriptRangeReadInvalidatedError() + } + } + + await expect( + readIncrementalTranscriptMessages( + '/tmp/invalidated-append.jsonl', + state, + nativeChatLineDecoderForAgent('claude')!, + onBatch, + undefined, + undefined, + undefined, + rangeFs + ) + ).rejects.toBeInstanceOf(TranscriptRangeReadInvalidatedError) + expect(onBatch).not.toHaveBeenCalled() + expect(state).toMatchObject({ offset: 0, pendingStart: 0, pendingBytes: 0 }) + }) + + it('emits the remote snapshot and later tail without whole-file polling', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-host-tail-')) + tempRoots.push(root) + const transcriptPath = join(root, 'session.jsonl') + await writeFile(transcriptPath, claudeLine('desktop-poison', 'wrong host')) + const files = new Map([[transcriptPath, Buffer.from(claudeLine('remote-first', 'first'))]]) + const { provider, readFile, readFileRange } = memoryProvider(files) + mocks.getProvider.mockReturnValue(provider) + const snapshots: string[][] = [] + const appended: string[] = [] + + const subscription = await subscribeNativeChatTranscript({ + ...routedArgs(transcriptPath), + initialLimit: 40, + onInitialSnapshot: (messages) => snapshots.push(messages.map((message) => message.id)), + onAppend: (messages) => appended.push(...messages.map((message) => message.id)), + debounceMs: 0, + reconciliationIntervalMs: 20 + } as Parameters[0]) + + try { + await vi.waitFor(() => expect(snapshots).toContainEqual(['remote-first'])) + files.set( + transcriptPath, + Buffer.from(`${claudeLine('remote-first', 'first')}${claudeLine('remote-next', 'next')}`) + ) + await vi.waitFor(() => expect(appended).toContain('remote-next')) + + expect(snapshots.flat()).not.toContain('desktop-poison') + expect(readFile).not.toHaveBeenCalled() + expect(readFileRange).toHaveBeenCalled() + } finally { + subscription.unsubscribe() + } + }) + + it('never falls back locally when ownership is missing', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-host-unknown-')) + tempRoots.push(root) + const transcriptPath = join(root, 'session.jsonl') + await writeFile(transcriptPath, claudeLine('desktop-poison', 'wrong host')) + + const result = await readNativeChatTranscriptTail({ + agent: 'claude', + sessionId: SESSION_ID, + transcriptPath, + providerSession: { key: 'session_id', id: SESSION_ID, transcriptPath }, + limit: 40 + }) + + expect(result).toEqual({ + error: 'Transcript unverifiable on the remote host', + unverifiable: true + }) + expect(mocks.getProvider).not.toHaveBeenCalled() + }) + + it('never installs a local watcher when subscription ownership is missing', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-host-unknown-watch-')) + tempRoots.push(root) + const transcriptPath = join(root, 'session.jsonl') + await writeFile(transcriptPath, claudeLine('desktop-poison', 'wrong host')) + const before = getActiveNativeChatWatcherCount() + const snapshots: { ids: string[]; error?: string }[] = [] + + const subscription = await subscribeNativeChatTranscript({ + agent: 'claude', + sessionId: SESSION_ID, + transcriptPath, + providerSession: { key: 'session_id', id: SESSION_ID, transcriptPath }, + onInitialSnapshot: (messages, _hasMore, _beforeOffset, error) => + snapshots.push({ ids: messages.map((message) => message.id), error }), + onAppend: () => {} + }) + + expect(snapshots).toEqual([{ ids: [], error: 'Transcript unverifiable on the remote host' }]) + expect(getActiveNativeChatWatcherCount()).toBe(before) + expect(mocks.getProvider).not.toHaveBeenCalled() + subscription.unsubscribe() + }) + + it('keeps stamped local sessions on the existing local reader', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-host-local-')) + tempRoots.push(root) + const transcriptPath = join(root, 'session.jsonl') + await writeFile(transcriptPath, claudeLine('local-real', 'local host')) + + const result = await readNativeChatTranscriptTail({ + agent: 'claude', + sessionId: SESSION_ID, + transcriptPath, + providerSession: { + key: 'session_id', + id: SESSION_ID, + transcriptPath, + executionHostId: 'local' + }, + limit: 40 + }) + + expect(result).toMatchObject({ messages: [{ id: 'local-real' }] }) + expect(mocks.getProvider).not.toHaveBeenCalled() + }) + + it('keeps locator-free legacy requests host-local for older clients', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-host-legacy-local-')) + tempRoots.push(root) + const transcriptPath = join(root, 'session.jsonl') + await writeFile(transcriptPath, claudeLine('legacy-local', 'older client')) + + const result = await readNativeChatTranscriptTail({ + agent: 'claude', + sessionId: SESSION_ID, + transcriptPath, + limit: 40 + }) + + expect(result).toMatchObject({ messages: [{ id: 'legacy-local' }] }) + expect(mocks.getProvider).not.toHaveBeenCalled() + }) + + it('selects the exact owner when two hosts expose the same id and path', async () => { + const transcriptPath = '/tmp/same-session.jsonl' + const wrong = memoryProvider( + new Map([[transcriptPath, Buffer.from(claudeLine('wrong-remote', 'wrong host'))]]) + ) + const right = memoryProvider( + new Map([[transcriptPath, Buffer.from(claudeLine('right-remote', 'right host'))]]) + ) + mocks.getProvider.mockImplementation((connectionId) => + connectionId === 'ssh-owner' ? right.provider : wrong.provider + ) + + const result = await readNativeChatTranscriptTail({ + ...routedArgs(transcriptPath), + limit: 40 + }) + + expect(result).toMatchObject({ messages: [{ id: 'right-remote' }] }) + expect(right.readFileRange).toHaveBeenCalled() + expect(wrong.readFileRange).not.toHaveBeenCalled() + }) + + it('does not run desktop id discovery when an SSH locator has no transcript path', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-host-no-path-')) + tempRoots.push(root) + const projectDir = join(root, 'project') + await mkdir(projectDir, { recursive: true }) + await writeFile(join(projectDir, `${SESSION_ID}.jsonl`), claudeLine('desktop-id-hit', 'wrong')) + + const result = await readNativeChatTranscriptTail({ + agent: 'claude', + sessionId: SESSION_ID, + claudeProjectsDir: root, + providerSession: { + key: 'session_id', + id: SESSION_ID, + executionHostId: EXECUTION_HOST_ID + }, + limit: 40 + }) + + expect(result).toMatchObject({ notFound: true }) + expect(result).not.toHaveProperty('messages') + + const snapshots: { ids: string[]; error?: string }[] = [] + const before = getActiveNativeChatWatcherCount() + const subscription = await subscribeNativeChatTranscript({ + agent: 'claude', + sessionId: SESSION_ID, + claudeProjectsDir: root, + providerSession: { + key: 'session_id', + id: SESSION_ID, + executionHostId: EXECUTION_HOST_ID + }, + onInitialSnapshot: (messages, _hasMore, _beforeOffset, error) => + snapshots.push({ ids: messages.map((message) => message.id), error }), + onAppend: () => {} + }) + + expect(snapshots).toEqual([{ ids: [], error: 'Transcript unavailable' }]) + expect(getActiveNativeChatWatcherCount()).toBe(before) + subscription.unsubscribe() + }) + + it('rearms on provider replacement and emits the new remote snapshot', async () => { + const transcriptPath = '/tmp/reconnected-session.jsonl' + const first = memoryProvider( + new Map([[transcriptPath, Buffer.from(claudeLine('before-reconnect', 'first'))]]) + ) + const second = memoryProvider( + new Map([[transcriptPath, Buffer.from(claudeLine('after-reconnect', 'second'))]]) + ) + mocks.getProvider.mockReturnValue(first.provider) + const snapshots: string[][] = [] + const subscription = await subscribeNativeChatTranscript({ + ...routedArgs(transcriptPath), + initialLimit: 40, + onInitialSnapshot: (messages) => snapshots.push(messages.map((message) => message.id)), + onAppend: () => {}, + debounceMs: 0, + reconciliationIntervalMs: 20 + }) + + try { + await vi.waitFor(() => expect(snapshots).toContainEqual(['before-reconnect'])) + changeProvider(second.provider) + await vi.waitFor(() => expect(snapshots).toContainEqual(['after-reconnect'])) + expect(first.readFile).not.toHaveBeenCalled() + expect(second.readFile).not.toHaveBeenCalled() + } finally { + subscription.unsubscribe() + } + }) + + it('recovers from provider loss with one fresh watcher and no stale snapshot', async () => { + const transcriptPath = '/work/folder-workspace/session.jsonl' + const first = memoryProvider( + new Map([[transcriptPath, Buffer.from(claudeLine('before-loss', 'first'))]]) + ) + const restored = memoryProvider( + new Map([[transcriptPath, Buffer.from(claudeLine('after-restore', 'second'))]]) + ) + mocks.getProvider.mockReturnValue(first.provider) + const before = getActiveNativeChatWatcherCount() + const snapshots: { ids: string[]; error?: string }[] = [] + const subscription = await subscribeNativeChatTranscript({ + ...routedArgs(transcriptPath), + initialLimit: 40, + onInitialSnapshot: (messages, _hasMore, _beforeOffset, error) => + snapshots.push({ ids: messages.map((message) => message.id), error }), + onAppend: () => {}, + debounceMs: 0, + reconciliationIntervalMs: 20 + }) + + try { + await vi.waitFor(() => + expect(snapshots).toContainEqual({ ids: ['before-loss'], error: undefined }) + ) + expect(getActiveNativeChatWatcherCount()).toBe(before + 1) + + changeProvider(undefined) + await vi.waitFor(() => + expect(snapshots).toContainEqual({ + ids: [], + error: 'Transcript unverifiable on the remote host' + }) + ) + expect(getActiveNativeChatWatcherCount()).toBe(before) + + changeProvider(restored.provider) + await vi.waitFor(() => + expect(snapshots).toContainEqual({ ids: ['after-restore'], error: undefined }) + ) + expect(getActiveNativeChatWatcherCount()).toBe(before + 1) + expect(snapshots.some((snapshot) => snapshot.ids.includes('before-loss'))).toBe(true) + expect(snapshots.at(-1)?.ids).toEqual(['after-restore']) + } finally { + subscription.unsubscribe() + } + expect(getActiveNativeChatWatcherCount()).toBe(before) + }) + + it('keeps retrying when an unverifiable advisory subscriber throws', async () => { + const transcriptPath = '/tmp/throwing-subscriber-session.jsonl' + const restored = memoryProvider( + new Map([[transcriptPath, Buffer.from(claudeLine('recovered-after-throw', 'second'))]]) + ) + mocks.getProvider.mockReturnValue(undefined) + const snapshots: string[][] = [] + const subscription = await subscribeNativeChatTranscript({ + ...routedArgs(transcriptPath), + initialLimit: 40, + onInitialSnapshot: (messages, _hasMore, _beforeOffset, error) => { + if (error) { + throw new Error('subscriber is closing') + } + snapshots.push(messages.map((message) => message.id)) + }, + onAppend: () => {}, + debounceMs: 0, + reconciliationIntervalMs: 20 + }) + + try { + changeProvider(restored.provider) + await vi.waitFor(() => expect(snapshots).toContainEqual(['recovered-after-throw'])) + } finally { + subscription.unsubscribe() + } + }) + + it('emits one unresolved advisory while an exact remote path is still missing', async () => { + const transcriptPath = '/tmp/not-yet-written.jsonl' + const { provider } = memoryProvider(new Map()) + mocks.getProvider.mockReturnValue(provider) + const errors: (string | undefined)[] = [] + const subscription = await subscribeNativeChatTranscript({ + ...routedArgs(transcriptPath), + onInitialSnapshot: (_messages, _hasMore, _beforeOffset, error) => errors.push(error), + onAppend: () => {}, + resolvePollIntervalMs: 5, + unresolvedNoticeMs: 0 + }) + + try { + await vi.waitFor(() => + expect(errors).toEqual([ + 'No transcript found for this session on this machine. If the agent runs on a remote host its transcript lives there; otherwise it may not have been written yet.' + ]) + ) + } finally { + subscription.unsubscribe() + } + }) + + it('keeps provider methods bound to their owning instance', async () => { + const transcriptPath = '/tmp/bound-provider-session.jsonl' + const provider = { + files: new Map([[transcriptPath, Buffer.from(claudeLine('bound-remote', 'right host'))]]), + async stat(this: { files: Map }, filePath: string): Promise { + const bytes = this.files.get(filePath)! + return { size: bytes.length, type: 'file', mtime: 1, mtimeMs: 1 } + }, + async readFileRange( + this: { files: Map }, + filePath: string, + position: number, + length: number + ): Promise { + const bytes = this.files.get(filePath)!.subarray(position, position + length) + return { bytes, bytesRead: bytes.length } + }, + async supportsFileRangeRead() { + return true + } + } as unknown as IFilesystemProvider + mocks.getProvider.mockReturnValue(provider) + + const result = await readNativeChatTranscriptTail({ + ...routedArgs(transcriptPath), + limit: 40 + }) + + expect(result).toMatchObject({ messages: [{ id: 'bound-remote' }] }) + }) + + it('pages a large append in bounded ranges without rereading the prefix', async () => { + const transcriptPath = '/tmp/large-append-session.jsonl' + const initial = Buffer.from(claudeLine('initial', 'first')) + const files = new Map([[transcriptPath, initial]]) + const { provider, readFile, readFileRange } = memoryProvider(files) + mocks.getProvider.mockReturnValue(provider) + const appended: string[] = [] + const subscription = await subscribeNativeChatTranscript({ + ...routedArgs(transcriptPath), + initialLimit: 40, + onInitialSnapshot: () => {}, + onAppend: (messages) => appended.push(...messages.map((message) => message.id)), + debounceMs: 0, + reconciliationIntervalMs: 20 + }) + + try { + await vi.waitFor(() => expect(readFileRange).toHaveBeenCalled()) + readFileRange.mockClear() + const append = Buffer.from( + Array.from({ length: 3_000 }, (_, index) => + claudeLine(`append-${index}`, `bounded remote append ${index}`) + ).join('') + ) + expect(append.length).toBeGreaterThan(MAX_FILE_RANGE_READ_BYTES) + files.set(transcriptPath, Buffer.concat([initial, append])) + + await vi.waitFor(() => expect(appended).toContain('append-2999')) + const requestedLengths = readFileRange.mock.calls.map((call) => call[2]) + expect(Math.max(...requestedLengths)).toBeLessThanOrEqual(MAX_FILE_RANGE_READ_BYTES) + expect(requestedLengths.reduce((sum, length) => sum + length, 0)).toBeLessThanOrEqual( + append.length + 256 + ) + expect(readFile).not.toHaveBeenCalled() + } finally { + subscription.unsubscribe() + } + }) +}) diff --git a/src/main/native-chat/ssh-transcript-range-fs.ts b/src/main/native-chat/ssh-transcript-range-fs.ts new file mode 100644 index 00000000000..849966ec081 --- /dev/null +++ b/src/main/native-chat/ssh-transcript-range-fs.ts @@ -0,0 +1,93 @@ +import { MAX_FILE_RANGE_READ_BYTES } from '../../shared/file-range-read' +import { getSshFilesystemProviderSnapshot } from '../providers/ssh-filesystem-dispatch' +import type { IFilesystemProvider } from '../providers/types' +import { TranscriptHostUnverifiableError } from './transcript-host-verdict' +import { + TranscriptRangeReadInvalidatedError, + type TranscriptFileStamp, + type TranscriptRangeFs +} from './transcript-range-fs' + +type ProviderGeneration = { + provider: IFilesystemProvider + generation: number +} + +function requireSameProvider(connectionId: string, expected: ProviderGeneration): void { + const current = getSshFilesystemProviderSnapshot(connectionId) + if ( + !current || + current.provider !== expected.provider || + current.generation !== expected.generation + ) { + throw new TranscriptHostUnverifiableError() + } +} + +export async function createSshTranscriptRangeFs( + connectionId: string, + signal?: AbortSignal +): Promise { + const snapshot = getSshFilesystemProviderSnapshot(connectionId) + if (!snapshot?.provider.readFileRange) { + throw new TranscriptHostUnverifiableError() + } + if ( + snapshot.provider.supportsFileRangeRead && + !(await snapshot.provider.supportsFileRangeRead({ signal })) + ) { + throw new TranscriptHostUnverifiableError() + } + const providerSnapshot: ProviderGeneration = snapshot + requireSameProvider(connectionId, providerSnapshot) + const provider = providerSnapshot.provider + async function stat(filePath: string, statSignal?: AbortSignal): Promise { + statSignal?.throwIfAborted() + requireSameProvider(connectionId, providerSnapshot) + const stamp = await provider.stat(filePath) + requireSameProvider(connectionId, providerSnapshot) + const mtimeMs = stamp.mtimeMs ?? stamp.mtime + return { + size: stamp.size, + identity: `${providerSnapshot.generation}:${stamp.dev ?? 0}:${stamp.ino ?? 0}`, + identityReliable: stamp.dev !== undefined && stamp.ino !== undefined, + mtimeMs, + ctimeMs: mtimeMs + } + } + const rangeFs: TranscriptRangeFs = { + stat, + async read(filePath, position, length, readSignal) { + readSignal?.throwIfAborted() + const parts: Buffer[] = [] + let remaining = length + let cursor = position + while (remaining > 0) { + requireSameProvider(connectionId, providerSnapshot) + const window = Math.min(remaining, MAX_FILE_RANGE_READ_BYTES) + const result = await provider.readFileRange!(filePath, cursor, window, { + signal: readSignal + }) + requireSameProvider(connectionId, providerSnapshot) + const bytes = result.bytes.subarray(0, result.bytesRead) + parts.push(bytes) + cursor += bytes.length + remaining -= bytes.length + if (bytes.length < window) { + break + } + } + return parts.length === 1 ? parts[0]! : Buffer.concat(parts) + }, + async assertStable(filePath, openingStamp, stableSignal) { + const closingStamp = await stat(filePath, stableSignal) + const identityChanged = closingStamp.identity !== openingStamp.identity + const unpinnedVersionChanged = + !openingStamp.identityReliable && closingStamp.mtimeMs !== openingStamp.mtimeMs + if (identityChanged || unpinnedVersionChanged) { + throw new TranscriptRangeReadInvalidatedError() + } + } + } + return rangeFs +} diff --git a/src/main/native-chat/ssh-transcript-read.ts b/src/main/native-chat/ssh-transcript-read.ts new file mode 100644 index 00000000000..44ff58c1588 --- /dev/null +++ b/src/main/native-chat/ssh-transcript-read.ts @@ -0,0 +1,91 @@ +import type { AgentType } from '../../shared/native-chat-types' +import type { ResolveSessionFileOptions } from './session-file-resolver' +import type { NativeChatTranscriptOwner } from './native-chat-transcript-owner' +import { createSshTranscriptRangeFs } from './ssh-transcript-range-fs' +import type { + NativeChatLineDecoder, + NativeChatTranscriptTailResult, + readNativeChatTranscriptTailFile +} from './transcript-tail-reader' +import { TranscriptRangeReadInvalidatedError } from './transcript-range-fs' +import type { NativeChatTurnLifecycleDecoder } from './transcript-turn-lifecycle' +import { + isTranscriptHostUnverifiableError, + transcriptUnverifiableResult +} from './transcript-host-verdict' + +type NativeChatSshOwner = Extract +type ReadTranscriptTailFile = typeof readNativeChatTranscriptTailFile + +type OwnedSshTranscriptReadDependencies = { + decode: NativeChatLineDecoder | null + decodeLifecycle: NativeChatTurnLifecycleDecoder | null + readTailFile: ReadTranscriptTailFile +} + +export async function readOwnedSshNativeChatTranscriptTail( + owner: NativeChatSshOwner, + args: ResolveSessionFileOptions & { + agent: AgentType + sessionId: string + limit: number + beforeOffset?: number + }, + signal: AbortSignal | undefined, + dependencies: OwnedSshTranscriptReadDependencies +): Promise { + const { decode, decodeLifecycle, readTailFile } = dependencies + if (!decode) { + return { error: 'Transcript unavailable' } + } + if (!owner.transcriptPath) { + return { error: 'Transcript unavailable', notFound: true } + } + try { + const rangeFs = await createSshTranscriptRangeFs(owner.connectionId, signal) + let result: Awaited> + try { + result = await readTailFile( + owner.transcriptPath, + args.limit, + decode, + true, + args.beforeOffset, + decodeLifecycle, + signal, + rangeFs + ) + } catch (error) { + if (!(error instanceof TranscriptRangeReadInvalidatedError)) { + throw error + } + result = await readTailFile( + owner.transcriptPath, + args.limit, + decode, + true, + args.beforeOffset, + decodeLifecycle, + signal, + rangeFs + ) + } + return { + messages: result.messages, + ...(args.beforeOffset === undefined && result.lifecycle + ? { lifecycle: result.lifecycle } + : {}), + hasMore: result.hasMore, + beforeOffset: result.beforeOffset + } + } catch (error) { + signal?.throwIfAborted() + if ((error as NodeJS.ErrnoException | null)?.code === 'ENOENT') { + return { error: error instanceof Error ? error.message : String(error), notFound: true } + } + if (isTranscriptHostUnverifiableError(error)) { + return transcriptUnverifiableResult() + } + return { error: error instanceof Error ? error.message : String(error) } + } +} diff --git a/src/main/native-chat/ssh-transcript-subscription.ts b/src/main/native-chat/ssh-transcript-subscription.ts new file mode 100644 index 00000000000..5b749852bda --- /dev/null +++ b/src/main/native-chat/ssh-transcript-subscription.ts @@ -0,0 +1,195 @@ +import { onSshFilesystemProviderChanged } from '../providers/ssh-filesystem-dispatch' +import type { NativeChatTranscriptOwner } from './native-chat-transcript-owner' +import { createSshTranscriptRangeFs } from './ssh-transcript-range-fs' +import { + DEFAULT_TRANSCRIPT_UNRESOLVED_NOTICE_MS, + isTranscriptHostUnverifiableError, + TRANSCRIPT_UNVERIFIABLE_MESSAGE, + UNRESOLVED_TRANSCRIPT_MESSAGE +} from './transcript-host-verdict' +import { nativeChatLineDecoderForAgent } from './transcript-tail-reader' +import { installTranscriptWatcher } from './transcript-watch-engine' +import type { + NativeChatTranscriptSubscription, + SubscribeNativeChatTranscriptArgs +} from './transcript-watch-contract' + +const INITIAL_RETRY_MS = 500 +const MAX_RETRY_MS = 5_000 +type SshOwner = Extract + +export function subscribeSshNativeChatTranscript( + owner: SshOwner, + args: SubscribeNativeChatTranscriptArgs, + setupSignal?: AbortSignal +): NativeChatTranscriptSubscription { + const decode = nativeChatLineDecoderForAgent(args.agent) + if (!decode || setupSignal?.aborted) { + return { unsubscribe: () => {}, watching: false } + } + const lineDecoder = decode + let closed = false + let installed: NativeChatTranscriptSubscription | null = null + let retryTimer: ReturnType | null = null + let retryMs = args.resolvePollIntervalMs ?? INITIAL_RETRY_MS + let attemptVersion = 0 + let attemptController: AbortController | null = null + let attemptInFlight = false + let rerunRequested = false + let unverifiableEmitted = false + let unresolvedNoticeEmitted = false + let missingPathEmitted = false + const startedAt = Date.now() + const unresolvedNoticeMs = args.unresolvedNoticeMs ?? DEFAULT_TRANSCRIPT_UNRESOLVED_NOTICE_MS + + const stopProviderChanges = onSshFilesystemProviderChanged((connectionId) => { + if (connectionId !== owner.connectionId || closed) { + return + } + attemptVersion++ + attemptController?.abort() + installed?.unsubscribe() + installed = null + clearRetry() + void runAttempt() + }) + + const abortFromSetup = (): void => subscription.unsubscribe() + setupSignal?.addEventListener('abort', abortFromSetup, { once: true }) + + function clearRetry(): void { + if (retryTimer) { + clearTimeout(retryTimer) + retryTimer = null + } + } + + function emitUnverifiable(): void { + if (unverifiableEmitted || !args.onInitialSnapshot) { + return + } + unverifiableEmitted = true + try { + args.onInitialSnapshot([], false, 0, TRANSCRIPT_UNVERIFIABLE_MESSAGE) + } catch { + // The advisory cannot own retry liveness when its subscriber is closing. + } + } + + function emitUnresolvedNotice(): void { + if ( + unverifiableEmitted || + unresolvedNoticeEmitted || + !args.onInitialSnapshot || + Date.now() - startedAt < unresolvedNoticeMs + ) { + return + } + unresolvedNoticeEmitted = true + try { + args.onInitialSnapshot([], false, 0, UNRESOLVED_TRANSCRIPT_MESSAGE) + } catch { + // The advisory cannot own retry liveness when its subscriber is closing. + } + } + + function scheduleRetry(): void { + if (closed || installed || retryTimer) { + return + } + retryTimer = setTimeout(() => { + retryTimer = null + void runAttempt() + }, retryMs) + retryTimer.unref?.() + if (args.resolvePollIntervalMs === undefined) { + retryMs = Math.min(retryMs * 2, MAX_RETRY_MS) + } + } + + async function runAttempt(): Promise { + if (closed || installed) { + return + } + if (attemptInFlight) { + rerunRequested = true + return + } + attemptInFlight = true + const version = attemptVersion + const controller = new AbortController() + attemptController = controller + try { + if (!owner.transcriptPath) { + if (!missingPathEmitted) { + missingPathEmitted = true + try { + args.onInitialSnapshot?.([], false, 0, 'Transcript unavailable') + } catch { + // A terminal missing-path result must not escape a closing subscriber. + } + } + return + } + const rangeFs = await createSshTranscriptRangeFs(owner.connectionId, controller.signal) + const result = await installTranscriptWatcher( + owner.transcriptPath, + lineDecoder, + { + ...args, + filePath: owner.transcriptPath, + transcriptPath: owner.transcriptPath, + rangeFs + }, + controller.signal + ) + if (closed || controller.signal.aborted || version !== attemptVersion) { + result?.unsubscribe() + return + } + if (result) { + installed = result + retryMs = args.resolvePollIntervalMs ?? INITIAL_RETRY_MS + return + } + scheduleRetry() + emitUnresolvedNotice() + } catch (error) { + if (!closed && !controller.signal.aborted) { + if (isTranscriptHostUnverifiableError(error)) { + emitUnverifiable() + } + scheduleRetry() + emitUnresolvedNotice() + } + } finally { + if (attemptController === controller) { + attemptController = null + } + attemptInFlight = false + if (rerunRequested && !closed && !installed) { + rerunRequested = false + void runAttempt() + } + } + } + + const subscription: NativeChatTranscriptSubscription = { + watching: true, + unsubscribe: () => { + if (closed) { + return + } + closed = true + attemptVersion++ + attemptController?.abort() + setupSignal?.removeEventListener('abort', abortFromSetup) + stopProviderChanges() + clearRetry() + installed?.unsubscribe() + installed = null + } + } + void runAttempt() + return subscription +} diff --git a/src/main/native-chat/transcript-file-version.ts b/src/main/native-chat/transcript-file-version.ts index 2de3c76821f..6e7cc0c1d9b 100644 --- a/src/main/native-chat/transcript-file-version.ts +++ b/src/main/native-chat/transcript-file-version.ts @@ -1,4 +1,5 @@ import { readTranscriptSlice, wslGatedStat } from './wsl-transcript-fs-access' +import type { TranscriptRangeFs } from './transcript-range-fs' const BOUNDARY_FINGERPRINT_BYTES = 64 @@ -11,8 +12,12 @@ export type TranscriptFileVersion = { export async function readTranscriptFileVersion( filePath: string, - signal?: AbortSignal + signal?: AbortSignal, + rangeFs?: TranscriptRangeFs ): Promise { + if (rangeFs) { + return rangeFs.stat(filePath, signal) + } const value = await wslGatedStat(filePath, 'exact', signal) return { identity: `${value.dev}:${value.ino}`, @@ -25,13 +30,16 @@ export async function readTranscriptFileVersion( export async function boundaryFingerprint( filePath: string, offset: number, - signal?: AbortSignal + signal?: AbortSignal, + rangeFs?: TranscriptRangeFs ): Promise { if (offset <= 0) { return '' } const start = Math.max(0, offset - BOUNDARY_FINGERPRINT_BYTES) - const slice = await readTranscriptSlice(filePath, start, offset - start, 'exact', signal) + const slice = rangeFs + ? await rangeFs.read(filePath, start, offset - start, signal) + : await readTranscriptSlice(filePath, start, offset - start, 'exact', signal) return slice.toString('base64') } diff --git a/src/main/native-chat/transcript-host-verdict.ts b/src/main/native-chat/transcript-host-verdict.ts new file mode 100644 index 00000000000..efcb1a32f76 --- /dev/null +++ b/src/main/native-chat/transcript-host-verdict.ts @@ -0,0 +1,46 @@ +import { FileRangeReadUnsupportedError } from '../providers/types' +import { SSH_FILESYSTEM_PROVIDER_UNAVAILABLE_MESSAGE } from '../providers/ssh-filesystem-dispatch' +import { WslTranscriptFsError } from './wsl-transcript-fs-gate' + +export const TRANSCRIPT_UNVERIFIABLE_MESSAGE = 'Transcript unverifiable on the remote host' +export const DEFAULT_TRANSCRIPT_UNRESOLVED_NOTICE_MS = 60_000 +export const UNRESOLVED_TRANSCRIPT_MESSAGE = + 'No transcript found for this session on this machine. If the agent runs on a remote host its transcript lives there; otherwise it may not have been written yet.' + +export class TranscriptHostUnverifiableError extends Error { + readonly verdict = 'unverifiable' as const + + constructor(message = TRANSCRIPT_UNVERIFIABLE_MESSAGE) { + super(message) + this.name = 'TranscriptHostUnverifiableError' + } +} + +export function isTranscriptHostUnverifiableError(error: unknown): boolean { + if ( + error instanceof TranscriptHostUnverifiableError || + error instanceof FileRangeReadUnsupportedError + ) { + return true + } + const candidate = error as { code?: unknown; message?: unknown } | null + return ( + candidate?.code === 'CONNECTION_LOST' || + candidate?.code === 'DISPOSED' || + candidate?.code === 'SSH_SESSION_EXPIRED' || + candidate?.message === SSH_FILESYSTEM_PROVIDER_UNAVAILABLE_MESSAGE + ) +} + +export function transcriptUnverifiableResult(): { error: string; unverifiable: true } { + return { error: TRANSCRIPT_UNVERIFIABLE_MESSAGE, unverifiable: true } +} + +export function transcriptInitialReadErrorMessage(error: unknown): string { + if (error instanceof WslTranscriptFsError) { + return error.message + } + return isTranscriptHostUnverifiableError(error) + ? TRANSCRIPT_UNVERIFIABLE_MESSAGE + : 'Transcript unavailable' +} diff --git a/src/main/native-chat/transcript-incremental-reader.ts b/src/main/native-chat/transcript-incremental-reader.ts index 1dd60f58292..c6b625abf76 100644 --- a/src/main/native-chat/transcript-incremental-reader.ts +++ b/src/main/native-chat/transcript-incremental-reader.ts @@ -1,10 +1,16 @@ import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types' +import { Readable } from 'node:stream' import { transcriptFallbackId } from './transcript-fallback-id' import { MAX_NATIVE_CHAT_TRANSCRIPT_RECORD_BYTES, type NativeChatLineDecoder } from './transcript-tail-reader' -import { openTranscriptReadStream, wslGatedStat } from './wsl-transcript-fs-access' +import { + openTranscriptReadStream, + WSL_TRANSCRIPT_READ_CHUNK_BYTES, + wslGatedStat +} from './wsl-transcript-fs-access' +import type { TranscriptRangeFs } from './transcript-range-fs' const APPEND_BATCH_MESSAGE_LIMIT = 40 @@ -31,19 +37,28 @@ export async function readIncrementalTranscriptMessages( onBatch?: (messages: NativeChatMessage[]) => void, decodeLifecycle?: (line: string, fallbackId: string) => NativeChatTurnLifecycle | null, onLifecycle?: (lifecycle: NativeChatTurnLifecycle) => void, - signal?: AbortSignal + signal?: AbortSignal, + rangeFs?: TranscriptRangeFs ): Promise { - const end = (await wslGatedStat(filePath, 'exact', signal)).size + const openingStamp = rangeFs ? await rangeFs.stat(filePath, signal) : null + const end = openingStamp?.size ?? (await wslGatedStat(filePath, 'exact', signal)).size if (end <= state.offset) { return [] } const messages: NativeChatMessage[] = [] - const stream = openTranscriptReadStream( - filePath, - { start: state.offset, end: end - 1 }, - 'exact', - signal - ) + const openingState = rangeFs + ? { + offset: state.offset, + pendingChunks: [...state.pendingChunks], + pendingStart: state.pendingStart, + pendingBytes: state.pendingBytes, + droppingOversizedRecord: state.droppingOversizedRecord + } + : null + const stagedBatches: NativeChatMessage[][] = [] + const stream = rangeFs + ? Readable.from(rangeTranscriptChunks(rangeFs, filePath, state.offset, end - 1, signal)) + : openTranscriptReadStream(filePath, { start: state.offset, end: end - 1 }, 'exact', signal) try { let absoluteOffset = state.offset for await (const rawChunk of stream) { @@ -65,7 +80,22 @@ export async function readIncrementalTranscriptMessages( absoluteOffset += chunk.length state.offset = absoluteOffset } + if (rangeFs && openingStamp) { + await rangeFs.assertStable(filePath, openingStamp, signal) + for (const batch of stagedBatches) { + onBatch?.(batch) + } + } return messages + } catch (error) { + if (openingState) { + state.offset = openingState.offset + state.pendingChunks = openingState.pendingChunks + state.pendingStart = openingState.pendingStart + state.pendingBytes = openingState.pendingBytes + state.droppingOversizedRecord = openingState.droppingOversizedRecord + } + throw error } finally { // Early exits (throw/oversized-record bail) must not leak the fd or, on // UNC, the gated handle the generator's finally closes. @@ -111,7 +141,35 @@ export async function readIncrementalTranscriptMessages( } messages.push(message) if (onBatch && messages.length >= APPEND_BATCH_MESSAGE_LIMIT) { - onBatch(messages.splice(0)) + const batch = messages.splice(0) + if (rangeFs) { + stagedBatches.push(batch) + } else { + onBatch(batch) + } + } + } +} + +async function* rangeTranscriptChunks( + rangeFs: TranscriptRangeFs, + filePath: string, + start: number, + endInclusive: number, + signal?: AbortSignal +): AsyncGenerator { + let position = start + while (position <= endInclusive) { + signal?.throwIfAborted() + const length = Math.min(WSL_TRANSCRIPT_READ_CHUNK_BYTES, endInclusive - position + 1) + const chunk = await rangeFs.read(filePath, position, length, signal) + if (chunk.length === 0) { + return + } + yield chunk + position += chunk.length + if (chunk.length < length) { + return } } } diff --git a/src/main/native-chat/transcript-native-watcher.ts b/src/main/native-chat/transcript-native-watcher.ts index 8cccd022c57..21f97e9ee1b 100644 --- a/src/main/native-chat/transcript-native-watcher.ts +++ b/src/main/native-chat/transcript-native-watcher.ts @@ -10,6 +10,15 @@ export type TranscriptNativeWatcher = { dispose: () => void } +export function createIdleTranscriptNativeWatcher(): TranscriptNativeWatcher { + return { + bind: () => false, + invalidate: () => {}, + needsRebind: () => false, + dispose: () => {} + } +} + /** * Optional fs.watch acceleration for transcript reconciliation. Native watches * can fail on otherwise-readable remote filesystems, so binding is retryable diff --git a/src/main/native-chat/transcript-range-fs.ts b/src/main/native-chat/transcript-range-fs.ts new file mode 100644 index 00000000000..2cdd668874a --- /dev/null +++ b/src/main/native-chat/transcript-range-fs.ts @@ -0,0 +1,24 @@ +export type TranscriptFileStamp = { + size: number + identity: string + identityReliable: boolean + mtimeMs: number + ctimeMs: number +} + +export class TranscriptRangeReadInvalidatedError extends Error { + constructor() { + super('Transcript changed during remote read') + this.name = 'TranscriptRangeReadInvalidatedError' + } +} + +export type TranscriptRangeFs = { + stat(filePath: string, signal?: AbortSignal): Promise + read(filePath: string, position: number, length: number, signal?: AbortSignal): Promise + assertStable( + filePath: string, + openingStamp: TranscriptFileStamp, + signal?: AbortSignal + ): Promise +} diff --git a/src/main/native-chat/transcript-tail-boundary.ts b/src/main/native-chat/transcript-tail-boundary.ts index b4479ac2bd0..081a537e24e 100644 --- a/src/main/native-chat/transcript-tail-boundary.ts +++ b/src/main/native-chat/transcript-tail-boundary.ts @@ -1,33 +1,29 @@ -import { wslGatedRead } from './wsl-transcript-fs-access' -import type { TranscriptFileHandle } from './wsl-transcript-fs-access' - export const TAIL_CHUNK_BYTES = 64 * 1024 +export type TranscriptTailChunkRead = (position: number, length: number) => Promise /** * The byte at `position`, or null when the file shrank below it between the * caller's stat and this read (allocUnsafe would otherwise hand back garbage). */ export async function readTranscriptByteAt( - handle: TranscriptFileHandle, - filePath: string, + read: TranscriptTailChunkRead, position: number, signal?: AbortSignal ): Promise { - const byte = Buffer.allocUnsafe(1) - const { bytesRead } = await wslGatedRead(handle, filePath, byte, 0, 1, position, 'exact', signal) signal?.throwIfAborted() - return bytesRead === 1 ? byte[0] : null + const bytes = await read(position, 1) + signal?.throwIfAborted() + return bytes.length === 1 ? (bytes[0] ?? null) : null } /** End offset (exclusive) of the last newline-terminated line at or before `end`. */ export async function findLastCompleteLineEnd( - handle: TranscriptFileHandle, - filePath: string, + read: TranscriptTailChunkRead, end: number, signal?: AbortSignal ): Promise { signal?.throwIfAborted() - const lastByte = await readTranscriptByteAt(handle, filePath, end - 1, signal) + const lastByte = await readTranscriptByteAt(read, end - 1, signal) if (lastByte === null) { // File shrank between stat and probe. return 0 @@ -39,23 +35,13 @@ export async function findLastCompleteLineEnd( while (cursor > 0) { signal?.throwIfAborted() const start = Math.max(0, cursor - TAIL_CHUNK_BYTES) - const buffer = Buffer.allocUnsafe(cursor - start) - const { bytesRead } = await wslGatedRead( - handle, - filePath, - buffer, - 0, - buffer.length, - start, - 'exact', - signal - ) + const buffer = await read(start, cursor - start) signal?.throwIfAborted() - if (bytesRead < buffer.length) { + if (buffer.length < cursor - start) { // File shrank mid-walk: any boundary computed from stale offsets is wrong. return 0 } - const newline = buffer.subarray(0, bytesRead).lastIndexOf(0x0a) + const newline = buffer.lastIndexOf(0x0a) if (newline !== -1) { return start + newline + 1 } diff --git a/src/main/native-chat/transcript-tail-reader.ts b/src/main/native-chat/transcript-tail-reader.ts index b6fa9bafdb0..82fe2c25417 100644 --- a/src/main/native-chat/transcript-tail-reader.ts +++ b/src/main/native-chat/transcript-tail-reader.ts @@ -3,6 +3,10 @@ import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types' +import type { TranscriptRangeFs } from './transcript-range-fs' +import { readOwnedSshNativeChatTranscriptTail } from './ssh-transcript-read' +import { resolveNativeChatTranscriptOwner } from './native-chat-transcript-owner' +import { transcriptUnverifiableResult } from './transcript-host-verdict' import { resolveNativeChatTranscriptAgent } from '../../shared/native-chat-agent-support' import { resolveSessionFilePath, type ResolveSessionFileOptions } from './session-file-resolver' import { @@ -57,7 +61,8 @@ export async function readNativeChatTranscriptTailFile( includeTrailingLine = false, endOffset?: number, decodeLifecycle?: NativeChatTurnLifecycleDecoder | null, - signal?: AbortSignal + signal?: AbortSignal, + rangeFs?: TranscriptRangeFs ): Promise<{ messages: NativeChatMessage[] lifecycle?: NativeChatTurnLifecycle @@ -68,15 +73,36 @@ export async function readNativeChatTranscriptTailFile( oversizedRecordCount?: number }> { signal?.throwIfAborted() + const openingStamp = rangeFs ? await rangeFs.stat(filePath, signal) : null const end = Math.min( - (await wslGatedStat(filePath, 'exact', signal)).size, + openingStamp?.size ?? (await wslGatedStat(filePath, 'exact', signal)).size, endOffset ?? Number.MAX_SAFE_INTEGER ) signal?.throwIfAborted() if (end === 0) { return { messages: [], consumedTo: 0, hasMore: false, beforeOffset: 0 } } - const handle = await wslGatedOpen(filePath, 'exact', signal) + const handle = rangeFs ? undefined : await wslGatedOpen(filePath, 'exact', signal) + const readChunk = async (position: number, length: number): Promise => { + if (rangeFs) { + return rangeFs.read(filePath, position, length, signal) + } + if (!handle) { + return Buffer.alloc(0) + } + const buffer = Buffer.allocUnsafe(length) + const { bytesRead } = await wslGatedRead( + handle, + filePath, + buffer, + 0, + length, + position, + 'exact', + signal + ) + return buffer.subarray(0, bytesRead) + } const lineParts: Buffer[] = [] let lineBytes = 0 let lineOversized = false @@ -88,12 +114,12 @@ export async function readNativeChatTranscriptTailFile( signal?.throwIfAborted() const consumedTo = includeTrailingLine ? end - : await findLastCompleteLineEnd(handle, filePath, end, signal) + : await findLastCompleteLineEnd(readChunk, end, signal) if (consumedTo === 0) { return { messages: [], consumedTo: 0, hasMore: false, beforeOffset: 0 } } const newestFirst: { message: NativeChatMessage; offset: number }[] = [] - const finalByte = await readTranscriptByteAt(handle, filePath, consumedTo - 1, signal) + const finalByte = await readTranscriptByteAt(readChunk, consumedTo - 1, signal) if (finalByte === null) { // File shrank between stat and probe: report empty, the next poll re-stats. return { messages: [], consumedTo: 0, hasMore: false, beforeOffset: 0 } @@ -103,25 +129,16 @@ export async function readNativeChatTranscriptTailFile( while (cursor > 0 && newestFirst.length <= limit) { signal?.throwIfAborted() const start = Math.max(0, cursor - TAIL_CHUNK_BYTES) - const buffer = Buffer.allocUnsafe(cursor - start) - const { bytesRead } = await wslGatedRead( - handle, - filePath, - buffer, - 0, - buffer.length, - start, - 'exact', - signal - ) + const requestedLength = cursor - start + const buffer = await readChunk(start, requestedLength) signal?.throwIfAborted() // A short read means the file shrank mid-walk: stop paging back rather // than stitch non-adjacent bytes into records. - if (bytesRead < buffer.length) { + if (buffer.length < requestedLength) { break } - let segmentEnd = bytesRead - for (let index = bytesRead - 1; index >= 0 && newestFirst.length <= limit; index--) { + let segmentEnd = buffer.length + for (let index = buffer.length - 1; index >= 0 && newestFirst.length <= limit; index--) { if (buffer[index] !== 0x0a) { continue } @@ -144,7 +161,7 @@ export async function readNativeChatTranscriptTailFile( // Why: slice(-0) returns the whole array, so a non-positive limit must // window to nothing explicitly rather than leak every buffered record. const selected = limit > 0 ? chronological.slice(Math.max(0, chronological.length - limit)) : [] - return { + const result = { messages: selected.map((entry) => entry.message), ...(lifecycle ? { lifecycle } : {}), consumedTo, @@ -153,8 +170,14 @@ export async function readNativeChatTranscriptTailFile( ...(malformedRecordCount > 0 ? { malformedRecordCount } : {}), ...(oversizedRecordCount > 0 ? { oversizedRecordCount } : {}) } + if (rangeFs && openingStamp) { + await rangeFs.assertStable(filePath, openingStamp, signal) + } + return result } finally { - await closeTranscriptHandle(handle, filePath) + if (handle) { + await closeTranscriptHandle(handle, filePath) + } } function retainPart(part: Buffer): void { @@ -211,6 +234,15 @@ export async function readNativeChatTranscriptTailFile( } } +export type NativeChatTranscriptTailResult = + | { + messages: NativeChatMessage[] + lifecycle?: NativeChatTurnLifecycle + hasMore: boolean + beforeOffset: number + } + | { error: string; notFound?: true; unverifiable?: true } + export async function readNativeChatTranscriptTail( args: ResolveSessionFileOptions & { agent: AgentType @@ -221,15 +253,24 @@ export async function readNativeChatTranscriptTail( beforeOffset?: number }, signal?: AbortSignal -): Promise< - | { - messages: NativeChatMessage[] - lifecycle?: NativeChatTurnLifecycle - hasMore: boolean - beforeOffset: number - } - | { error: string; notFound?: true } -> { +): Promise { + const owner = args.filePath + ? ({ kind: 'legacy-local' } as const) + : resolveNativeChatTranscriptOwner(args) + if (owner.kind === 'unknown') { + return transcriptUnverifiableResult() + } + if (owner.kind === 'ssh') { + return readOwnedSshNativeChatTranscriptTail(owner, args, signal, { + decode: nativeChatLineDecoderForAgent(args.agent), + decodeLifecycle: nativeChatTurnLifecycleDecoderForAgent(args.agent), + readTailFile: readNativeChatTranscriptTailFile + }) + } + const localArgs = + owner.kind === 'local' + ? { ...args, transcriptPath: owner.providerSession.transcriptPath } + : args const decode = nativeChatLineDecoderForAgent(args.agent) const decodeLifecycle = nativeChatTurnLifecycleDecoderForAgent(args.agent) if (!decode) { @@ -238,7 +279,8 @@ export async function readNativeChatTranscriptTail( let filePath: string | null try { filePath = - args.filePath ?? (await resolveSessionFilePath(args.agent, args.sessionId, args, signal)) + localArgs.filePath ?? + (await resolveSessionFilePath(localArgs.agent, localArgs.sessionId, localArgs, signal)) } catch (error) { signal?.throwIfAborted() // Why: gate refusal is transient unavailability with retry guidance — @@ -254,19 +296,18 @@ export async function readNativeChatTranscriptTail( try { const result = await readNativeChatTranscriptTailFile( filePath, - args.limit, + localArgs.limit, decode, true, - args.beforeOffset, + localArgs.beforeOffset, decodeLifecycle, signal ) - signal?.throwIfAborted() return { messages: result.messages, // Why: an older pagination page must not rewind the live lifecycle; only // the current transcript tail can authoritatively describe turn state. - ...(args.beforeOffset === undefined && result.lifecycle + ...(localArgs.beforeOffset === undefined && result.lifecycle ? { lifecycle: result.lifecycle } : {}), hasMore: result.hasMore, diff --git a/src/main/native-chat/transcript-watch-contract.ts b/src/main/native-chat/transcript-watch-contract.ts index 760e07fc967..0236facd88d 100644 --- a/src/main/native-chat/transcript-watch-contract.ts +++ b/src/main/native-chat/transcript-watch-contract.ts @@ -4,6 +4,7 @@ import type { NativeChatTurnLifecycle } from '../../shared/native-chat-types' import type { ResolveSessionFileOptions } from './session-file-resolver' +import type { TranscriptRangeFs } from './transcript-range-fs' export type SubscribeNativeChatTranscriptArgs = ResolveSessionFileOptions & { agent: AgentType @@ -33,6 +34,8 @@ export type SubscribeNativeChatTranscriptArgs = ResolveSessionFileOptions & { unresolvedNoticeMs?: number /** Test-only override for the host-side watcher reconciliation interval. */ reconciliationIntervalMs?: number + /** Positional filesystem for a transcript owned by another execution host. */ + rangeFs?: TranscriptRangeFs } export type NativeChatTranscriptSubscription = { diff --git a/src/main/native-chat/transcript-watch-engine.ts b/src/main/native-chat/transcript-watch-engine.ts index bbc8a52635c..85c4415c5fa 100644 --- a/src/main/native-chat/transcript-watch-engine.ts +++ b/src/main/native-chat/transcript-watch-engine.ts @@ -10,9 +10,12 @@ import { resetIncrementalTranscriptState, type IncrementalTranscriptState } from './transcript-incremental-reader' -import { createTranscriptNativeWatcher } from './transcript-native-watcher' -import { readNativeChatTranscriptTailFile } from './transcript-tail-reader' +import { + createIdleTranscriptNativeWatcher, + createTranscriptNativeWatcher +} from './transcript-native-watcher' import { nativeChatTurnLifecycleDecoderForAgent } from './transcript-turn-lifecycle' +import { emitTranscriptWatchAppends, readTranscriptWatchSnapshot } from './transcript-watch-read' import type { NativeChatTranscriptSubscription, SubscribeNativeChatTranscriptArgs @@ -20,6 +23,11 @@ import type { import { createTranscriptWatchScheduler } from './transcript-watch-scheduler' import { wslGatedStat } from './wsl-transcript-fs-access' import { WslTranscriptFsError } from './wsl-transcript-fs-gate' +import { + isTranscriptHostUnverifiableError, + transcriptInitialReadErrorMessage +} from './transcript-host-verdict' +import { TranscriptRangeReadInvalidatedError } from './transcript-range-fs' const ROTATION_RETRY_MS = 25 const MAX_ROTATION_RETRY_MS = 2_000 @@ -43,13 +51,14 @@ export async function installTranscriptWatcher( * waiter immediately instead of at the 30s deadline. */ signal?: AbortSignal ): Promise { + const rangeFs = args.rangeFs try { - await wslGatedStat(filePath, 'exact', signal) + await (rangeFs ? rangeFs.stat(filePath, signal) : wslGatedStat(filePath, 'exact', signal)) } catch (error) { // Why: "not flushed yet" degrades to resolve-polling, but a stalled distro // must reach the caller so it can surface a retryable message instead of // stranding the client at `loading`. - if (error instanceof WslTranscriptFsError) { + if (error instanceof WslTranscriptFsError || isTranscriptHostUnverifiableError(error)) { throw error } return null @@ -93,31 +102,42 @@ export async function installTranscriptWatcher( } } - async function readAndEmitAppends(): Promise { + const watchReadContext = { + filePath, + state, + decode, + decodeLifecycle, + signal: gateAbort.signal, + rangeFs + } + + function readAndEmitAppends(): Promise { + return emitTranscriptWatchAppends(watchReadContext, onAppend, () => closed) + } + + async function readInitialTranscript(): Promise<{ + messages: NativeChatMessage[] + lifecycle?: NativeChatTurnLifecycle + }> { let lifecycle: NativeChatTurnLifecycle | undefined - const remaining = await readIncrementalTranscriptMessages( + const messages = await readIncrementalTranscriptMessages( filePath, state, decode, - (messages) => { - if (!closed) { - onAppend(messages) - } - }, + undefined, decodeLifecycle ?? undefined, (nextLifecycle) => { lifecycle = nextLifecycle }, - gateAbort.signal + gateAbort.signal, + rangeFs ) - if (!closed && (remaining.length > 0 || lifecycle)) { - onAppend(remaining, lifecycle) - } + return { messages, ...(lifecycle ? { lifecycle } : {}) } } async function finishSuccessfulDrain(startVersion: TranscriptFileVersion): Promise { - watchedBoundary = await boundaryFingerprint(filePath, state.offset, gateAbort.signal) - const completedVersion = await readTranscriptFileVersion(filePath, gateAbort.signal) + watchedBoundary = await boundaryFingerprint(filePath, state.offset, gateAbort.signal, rangeFs) + const completedVersion = await readTranscriptFileVersion(filePath, gateAbort.signal, rangeFs) if (transcriptFileVersionChanged(completedVersion, startVersion)) { // Why: a write racing this drain needs another pass even when the reader // happened to reach its new EOF; timestamp-only rewrites may need replace. @@ -137,8 +157,13 @@ export async function installTranscriptWatcher( } async function drainOnce(): Promise { - const current = await readTranscriptFileVersion(filePath, gateAbort.signal) - const currentBoundary = await boundaryFingerprint(filePath, state.offset, gateAbort.signal) + const current = await readTranscriptFileVersion(filePath, gateAbort.signal, rangeFs) + const currentBoundary = await boundaryFingerprint( + filePath, + state.offset, + gateAbort.signal, + rangeFs + ) if (closed) { return } @@ -166,15 +191,7 @@ export async function installTranscriptWatcher( // Why: 0 is a valid window — an explicit undefined check keeps an empty // snapshot empty instead of falling back to an unbounded incremental read. contentReplaced && !initialDrain && onReplace && initialLimit !== undefined - ? await readNativeChatTranscriptTailFile( - filePath, - initialLimit, - decode, - false, - undefined, - decodeLifecycle, - gateAbort.signal - ) + ? await readTranscriptWatchSnapshot(watchReadContext, initialLimit) : null if (closed) { return @@ -195,15 +212,7 @@ export async function installTranscriptWatcher( const initialSnapshot = initialDrain && onInitialSnapshot && initialLimit !== undefined - ? await readNativeChatTranscriptTailFile( - filePath, - initialLimit, - decode, - false, - undefined, - decodeLifecycle, - gateAbort.signal - ) + ? await readTranscriptWatchSnapshot(watchReadContext, initialLimit) : null if (closed) { return @@ -222,22 +231,11 @@ export async function installTranscriptWatcher( ) await readAndEmitAppends() } else { - let lifecycle: NativeChatTurnLifecycle | undefined - const messages = await readIncrementalTranscriptMessages( - filePath, - state, - decode, - undefined, - decodeLifecycle ?? undefined, - (nextLifecycle) => { - lifecycle = nextLifecycle - }, - gateAbort.signal - ) + const initial = await readInitialTranscript() if (closed) { return } - onInitialSnapshot(messages, false, 0, undefined, lifecycle) + onInitialSnapshot(initial.messages, false, 0, undefined, initial.lifecycle) } } else { initialDrain = false @@ -266,14 +264,15 @@ export async function installTranscriptWatcher( // A still-pending initial drain also surfaces one error snapshot so a // watching client isn't stranded at 'loading' when the read keeps // throwing; initialDrain stays true so a recovered read can still win. - if (!closed && initialDrain && onInitialSnapshot && !initialErrorEmitted) { + if ( + !closed && + initialDrain && + onInitialSnapshot && + !initialErrorEmitted && + !(error instanceof TranscriptRangeReadInvalidatedError) + ) { initialErrorEmitted = true - onInitialSnapshot( - [], - false, - 0, - error instanceof WslTranscriptFsError ? error.message : 'Transcript unavailable' - ) + onInitialSnapshot([], false, 0, transcriptInitialReadErrorMessage(error)) } scheduleRotationRetry() break @@ -289,7 +288,7 @@ export async function installTranscriptWatcher( return } try { - const current = await readTranscriptFileVersion(filePath, gateAbort.signal) + const current = await readTranscriptFileVersion(filePath, gateAbort.signal, rangeFs) if (closed) { return } @@ -311,11 +310,13 @@ export async function installTranscriptWatcher( drain: () => void drain(), reconcile }) - const nativeWatcher = createTranscriptNativeWatcher( - filePath, - () => scheduler.scheduleEventDrain(), - scheduleRotationRetry - ) + const nativeWatcher = rangeFs + ? createIdleTranscriptNativeWatcher() + : createTranscriptNativeWatcher( + filePath, + () => scheduler.scheduleEventDrain(), + scheduleRotationRetry + ) nativeWatcher.bind() activeWatcherCount++ diff --git a/src/main/native-chat/transcript-watch-read.ts b/src/main/native-chat/transcript-watch-read.ts new file mode 100644 index 00000000000..9a99b481cc3 --- /dev/null +++ b/src/main/native-chat/transcript-watch-read.ts @@ -0,0 +1,60 @@ +import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types' +import { + readIncrementalTranscriptMessages, + type IncrementalTranscriptState +} from './transcript-incremental-reader' +import type { TranscriptRangeFs } from './transcript-range-fs' +import { + readNativeChatTranscriptTailFile, + type NativeChatLineDecoder +} from './transcript-tail-reader' +import type { NativeChatTurnLifecycleDecoder } from './transcript-turn-lifecycle' + +type TranscriptWatchReadContext = { + filePath: string + state: IncrementalTranscriptState + decode: NativeChatLineDecoder + decodeLifecycle: NativeChatTurnLifecycleDecoder | null + signal: AbortSignal + rangeFs?: TranscriptRangeFs +} + +export async function emitTranscriptWatchAppends( + context: TranscriptWatchReadContext, + onAppend: (messages: NativeChatMessage[], lifecycle?: NativeChatTurnLifecycle) => void, + isClosed: () => boolean +): Promise { + let lifecycle: NativeChatTurnLifecycle | undefined + const remaining = await readIncrementalTranscriptMessages( + context.filePath, + context.state, + context.decode, + (messages) => { + if (!isClosed()) { + onAppend(messages) + } + }, + context.decodeLifecycle ?? undefined, + (nextLifecycle) => { + lifecycle = nextLifecycle + }, + context.signal, + context.rangeFs + ) + if (!isClosed() && (remaining.length > 0 || lifecycle)) { + onAppend(remaining, lifecycle) + } +} + +export function readTranscriptWatchSnapshot(context: TranscriptWatchReadContext, limit: number) { + return readNativeChatTranscriptTailFile( + context.filePath, + limit, + context.decode, + false, + undefined, + context.decodeLifecycle, + context.signal, + context.rangeFs + ) +} diff --git a/src/main/native-chat/transcript-watch-wsl-exact-path.test.ts b/src/main/native-chat/transcript-watch-wsl-exact-path.test.ts index 814addff83c..f061dcbaae1 100644 --- a/src/main/native-chat/transcript-watch-wsl-exact-path.test.ts +++ b/src/main/native-chat/transcript-watch-wsl-exact-path.test.ts @@ -63,6 +63,12 @@ describe('exact hook path install on a Windows host with WSL (#10326)', () => { agent: 'codex', sessionId: 'wsl-sess', transcriptPath: ROLLOUT_LINUX, + providerSession: { + key: 'session_id', + id: 'wsl-sess', + transcriptPath: ROLLOUT_LINUX, + executionHostId: 'local' + }, resolvePollIntervalMs: 10, onAppend: () => {} }) @@ -84,6 +90,12 @@ describe('exact hook path install on a Windows host with WSL (#10326)', () => { agent: 'codex', sessionId: 'wsl-sess', transcriptPath: ROLLOUT_LINUX, + providerSession: { + key: 'session_id', + id: 'wsl-sess', + transcriptPath: ROLLOUT_LINUX, + executionHostId: 'local' + }, resolvePollIntervalMs: 10, onAppend: () => {} }) diff --git a/src/main/native-chat/transcript-watch.ts b/src/main/native-chat/transcript-watch.ts index 36bd853e40f..eb5d3c3acad 100644 --- a/src/main/native-chat/transcript-watch.ts +++ b/src/main/native-chat/transcript-watch.ts @@ -12,6 +12,13 @@ import type { } from './transcript-watch-contract' import { nativeChatLineDecoderForAgent } from './transcript-tail-reader' import { WslTranscriptFsError, wslTranscriptFsRefusal } from './wsl-transcript-fs-gate' +import { resolveNativeChatTranscriptOwner } from './native-chat-transcript-owner' +import { subscribeSshNativeChatTranscript } from './ssh-transcript-subscription' +import { + DEFAULT_TRANSCRIPT_UNRESOLVED_NOTICE_MS, + TRANSCRIPT_UNVERIFIABLE_MESSAGE, + UNRESOLVED_TRANSCRIPT_MESSAGE +} from './transcript-host-verdict' export { readNativeChatTranscriptTail } from './transcript-tail-reader' export { getActiveNativeChatWatcherCount } from './transcript-watch-engine' @@ -54,11 +61,6 @@ const FALLBACK_RESOLVE_POLL_MS = 5_000 // poll forever. Past this deadline say so instead of spinning silently. A slow // first flush can legitimately exceed it (#8401), so this is advisory only: the // poll keeps running and the real transcript replaces the notice. -const UNRESOLVED_NOTICE_MS = 60_000 -// Claims only what the deadline establishes — nothing resolved here — because a -// remote host, a cold WSL distro and a slow first flush all land on it. -const UNRESOLVED_TRANSCRIPT_MESSAGE = - 'No transcript found for this session on this machine. If the agent runs on a remote host its transcript lives there; otherwise it may not have been written yet.' function exactTranscriptPath(args: SubscribeNativeChatTranscriptArgs): string | null { const path = args.transcriptPath?.trim() @@ -72,7 +74,7 @@ function exactTranscriptPath(args: SubscribeNativeChatTranscriptArgs): string | * unsubscribe() cancels it. Reports watching:true — the engine's first drain * delivers the initial snapshot once the file appears, so subscribers must not * settle a merely not-yet-flushed transcript into a permanent error (#8401). - * Past UNRESOLVED_NOTICE_MS it emits one advisory error snapshot (#13663) and + * Past the unresolved deadline it emits one advisory error snapshot (#13663) and * keeps polling, so a late transcript still replaces it. */ function subscribeViaResolvePoll( @@ -97,7 +99,7 @@ function subscribeViaResolvePoll( let gateErrorEmitted = false let unresolvedNoticeEmitted = false const startedAt = Date.now() - const unresolvedNoticeMs = args.unresolvedNoticeMs ?? UNRESOLVED_NOTICE_MS + const unresolvedNoticeMs = args.unresolvedNoticeMs ?? DEFAULT_TRANSCRIPT_UNRESOLVED_NOTICE_MS const resolveController = new AbortController() /** Both advisories run under `void runAttempt()`, so a subscriber that throws @@ -263,6 +265,24 @@ export async function subscribeNativeChatTranscript( return { unsubscribe: () => {}, watching: false } } + if (!args.filePath) { + const owner = resolveNativeChatTranscriptOwner(args) + if (owner.kind === 'unknown') { + try { + args.onInitialSnapshot?.([], false, 0, TRANSCRIPT_UNVERIFIABLE_MESSAGE) + } catch { + // The caller owns teardown; unknown ownership must still avoid local IO. + } + return { unsubscribe: () => {}, watching: true } + } + if (owner.kind === 'ssh') { + return subscribeSshNativeChatTranscript(owner, args, setupSignal) + } + if (owner.kind === 'local') { + args = { ...args, transcriptPath: owner.providerSession.transcriptPath } + } + } + let installed: NativeChatTranscriptSubscription | null try { installed = await attemptInstall(args, decode, setupSignal) diff --git a/src/main/providers/ssh-filesystem-dispatch.test.ts b/src/main/providers/ssh-filesystem-dispatch.test.ts index 9b932be8c38..1c055a56bf0 100644 --- a/src/main/providers/ssh-filesystem-dispatch.test.ts +++ b/src/main/providers/ssh-filesystem-dispatch.test.ts @@ -2,6 +2,8 @@ import { describe, expect, it, vi } from 'vitest' import { getSshFilesystemProvider, + getSshFilesystemProviderSnapshot, + onSshFilesystemProviderChanged, onSshFilesystemProviderRegistered, registerSshFilesystemProvider, unregisterSshFilesystemProvider @@ -67,3 +69,32 @@ describe('onSshFilesystemProviderRegistered', () => { warnSpy.mockRestore() }) }) + +describe('SSH filesystem provider generations', () => { + it('fences register, unregister, and replacement with monotonic generations', () => { + const states: { generation: number; provider: IFilesystemProvider | undefined }[] = [] + const unsubscribe = onSshFilesystemProviderChanged((connectionId, generation) => { + if (connectionId === 'generation-conn') { + states.push({ generation, provider: getSshFilesystemProvider(connectionId) }) + } + }) + const replacement = {} as IFilesystemProvider + + registerSshFilesystemProvider('generation-conn', provider) + const first = getSshFilesystemProviderSnapshot('generation-conn') + unregisterSshFilesystemProvider('generation-conn') + registerSshFilesystemProvider('generation-conn', replacement) + const third = getSshFilesystemProviderSnapshot('generation-conn') + + expect(states.map((state) => state.generation)).toEqual([ + first!.generation, + first!.generation + 1, + first!.generation + 2 + ]) + expect(states.map((state) => state.provider)).toEqual([provider, undefined, replacement]) + expect(third).toEqual({ provider: replacement, generation: first!.generation + 2 }) + + unsubscribe() + unregisterSshFilesystemProvider('generation-conn') + }) +}) diff --git a/src/main/providers/ssh-filesystem-dispatch.ts b/src/main/providers/ssh-filesystem-dispatch.ts index 04ef209bef1..7606437d77f 100644 --- a/src/main/providers/ssh-filesystem-dispatch.ts +++ b/src/main/providers/ssh-filesystem-dispatch.ts @@ -8,6 +8,28 @@ export const SSH_FILESYSTEM_PROVIDER_UNAVAILABLE_MESSAGE = // Why: a reconnect builds a fresh provider, so anything holding remote state tied to the old // transport (file watches) needs a signal to rebuild it — nothing else marks that boundary. const registrationListeners = new Set<(connectionId: string) => void>() +const providerChangeListeners = new Set<(connectionId: string, generation: number) => void>() +const providerGenerations = new Map() + +function bumpProviderGeneration(connectionId: string): number { + const generation = (providerGenerations.get(connectionId) ?? 0) + 1 + providerGenerations.set(connectionId, generation) + for (const listener of providerChangeListeners) { + try { + listener(connectionId, generation) + } catch (error) { + console.warn('[ssh-filesystem] provider change listener failed:', error) + } + } + return generation +} + +export function onSshFilesystemProviderChanged( + listener: (connectionId: string, generation: number) => void +): () => void { + providerChangeListeners.add(listener) + return () => providerChangeListeners.delete(listener) +} export function onSshFilesystemProviderRegistered( listener: (connectionId: string) => void @@ -23,6 +45,7 @@ export function registerSshFilesystemProvider( provider: IFilesystemProvider ): void { sshProviders.set(connectionId, provider) + bumpProviderGeneration(connectionId) for (const listener of registrationListeners) { try { listener(connectionId) @@ -35,12 +58,21 @@ export function registerSshFilesystemProvider( export function unregisterSshFilesystemProvider(connectionId: string): void { sshProviders.delete(connectionId) + bumpProviderGeneration(connectionId) } export function getSshFilesystemProvider(connectionId: string): IFilesystemProvider | undefined { return sshProviders.get(connectionId) } +export function getSshFilesystemProviderSnapshot(connectionId: string): { + provider: IFilesystemProvider + generation: number +} | null { + const provider = sshProviders.get(connectionId) + return provider ? { provider, generation: providerGenerations.get(connectionId) ?? 0 } : null +} + export function requireSshFilesystemProvider(connectionId: string): IFilesystemProvider { const provider = getSshFilesystemProvider(connectionId) if (!provider) { diff --git a/src/main/runtime/rpc/methods/native-chat-rpc-schema.ts b/src/main/runtime/rpc/methods/native-chat-rpc-schema.ts new file mode 100644 index 00000000000..b3aa651edde --- /dev/null +++ b/src/main/runtime/rpc/methods/native-chat-rpc-schema.ts @@ -0,0 +1,40 @@ +import { z } from 'zod' +import { normalizeAgentProviderSession } from '../../../../shared/agent-session-resume' +import type { AgentType } from '../../../../shared/native-chat-types' + +export const MOBILE_NATIVE_CHAT_MAX_WINDOW = 2000 + +const NativeChatProviderSession = z.unknown().transform((value, context) => { + const normalized = normalizeAgentProviderSession(value) + if (!normalized) { + context.addIssue({ code: 'custom', message: 'Invalid provider session' }) + return z.NEVER + } + return normalized +}) + +export const NativeChatSession = z.object({ + agent: z + .unknown() + .transform((value) => (typeof value === 'string' ? value : '')) + .pipe(z.string().min(1, 'Missing agent')) + .transform((value) => value as AgentType), + sessionId: z + .unknown() + .transform((value) => (typeof value === 'string' ? value : '')) + .pipe(z.string().min(1, 'Missing session id')), + limit: z + .number() + .int() + .positive() + .transform((value) => Math.min(value, MOBILE_NATIVE_CHAT_MAX_WINDOW)) + .optional(), + subscriptionId: z.string().min(1).optional(), + transcriptPath: z.string().min(1).optional(), + providerSession: NativeChatProviderSession.optional(), + beforeOffset: z.number().int().nonnegative().optional() +}) + +export const NativeChatUnsubscribe = z.object({ + subscriptionId: z.string().min(1).optional() +}) diff --git a/src/main/runtime/rpc/methods/native-chat.test.ts b/src/main/runtime/rpc/methods/native-chat.test.ts index af885e35de3..9b50a38171d 100644 --- a/src/main/runtime/rpc/methods/native-chat.test.ts +++ b/src/main/runtime/rpc/methods/native-chat.test.ts @@ -21,7 +21,10 @@ const cachedResult = vi.hoisted(() => ({ } } })) -const tailRead = vi.hoisted(() => ({ signal: undefined as AbortSignal | undefined })) +const tailRead = vi.hoisted(() => ({ + signal: undefined as AbortSignal | undefined, + args: null as Record | null +})) const watcher = vi.hoisted(() => ({ args: null as null | { onInitialSnapshot?: ( @@ -53,6 +56,7 @@ const watcher = vi.hoisted(() => ({ timestamp: number | null } ) => void + providerSession?: unknown }, watching: true, setupSignal: undefined as AbortSignal | undefined, @@ -63,8 +67,10 @@ const watcher = vi.hoisted(() => ({ unsubscribe: vi.fn() })) vi.mock('../../../native-chat/transcript-watch', () => ({ - readNativeChatTranscriptTail: ({ limit }: { limit: number }, signal?: AbortSignal) => { + readNativeChatTranscriptTail: (args: { limit: number }, signal?: AbortSignal) => { tailRead.signal = signal + tailRead.args = args + const { limit } = args const messages = cachedResult.value.messages return Promise.resolve({ messages: messages.slice(-limit), @@ -157,6 +163,21 @@ function activeWatcherArgs(): NonNullable { } describe('nativeChat.readSession clientKind truncation gating', () => { + it('forwards the provider-session locator through runtime RPC reads', async () => { + const providerSession = { + key: 'session_id', + id: 's', + transcriptPath: '/tmp/s.jsonl', + executionHostId: 'ssh:builder' + } + + await readSessionHandler()( + { agent: 'claude', sessionId: 's', providerSession }, + ctxWith('mobile') + ) + + expect(tailRead.args).toMatchObject({ providerSession }) + }) it('passes request cancellation to transcript resolution', async () => { const controller = new AbortController() const context = { ...ctxWith('runtime'), signal: controller.signal } @@ -416,6 +437,25 @@ describe('nativeChat.readSession clientKind truncation gating', () => { }) describe('nativeChat.subscribe initial snapshot', () => { + it('forwards the provider-session locator through runtime RPC subscriptions', async () => { + watcher.watching = true + watcher.args = null + const providerSession = { + key: 'session_id', + id: 's', + transcriptPath: '/tmp/s.jsonl', + executionHostId: 'ssh:builder' + } + + await subscribeHandler()( + { agent: 'claude', sessionId: 's', providerSession }, + streamingContext('mobile'), + vi.fn() + ) + + expect(activeWatcherArgs()).toMatchObject({ providerSession }) + }) + it('preserves long assistant text across mobile stream frame types', async () => { watcher.watching = true watcher.args = null diff --git a/src/main/runtime/rpc/methods/native-chat.ts b/src/main/runtime/rpc/methods/native-chat.ts index 3b8d42287de..9c0dd4d7341 100644 --- a/src/main/runtime/rpc/methods/native-chat.ts +++ b/src/main/runtime/rpc/methods/native-chat.ts @@ -1,9 +1,4 @@ -import { z } from 'zod' -import type { - NativeChatBlock, - NativeChatMessage, - AgentType -} from '../../../../shared/native-chat-types' +import type { NativeChatBlock, NativeChatMessage } from '../../../../shared/native-chat-types' import { readNativeChatTranscriptTail, subscribeNativeChatTranscript, @@ -12,49 +7,17 @@ import { } from '../../../native-chat/transcript-watch' import { defineMethod, defineStreamingMethod, type RpcAnyMethod, type RpcContext } from '../core' import { sanitizeNativeChatRpcImageBlock } from './native-chat-rpc-image-block' +import { + MOBILE_NATIVE_CHAT_MAX_WINDOW, + NativeChatSession, + NativeChatUnsubscribe +} from './native-chat-rpc-schema' // Why: native chat renders an agent's own transcript (Claude/Codex JSONL). The // desktop reaches the readers via Electron IPC; mobile/web clients reach the // same pure readers through these runtime RPC methods so the native chat view // works over the paired connection, not just in the desktop renderer. -const NativeChatSession = z.object({ - agent: z - .unknown() - .transform((v) => (typeof v === 'string' ? v : '')) - .pipe(z.string().min(1, 'Missing agent')) - .transform((v) => v as AgentType), - sessionId: z - .unknown() - .transform((v) => (typeof v === 'string' ? v : '')) - .pipe(z.string().min(1, 'Missing session id')), - // How many of the most-recent messages to return. Clients start small for a - // fast first paint and raise it to page older history in as the user scrolls. - // Clamp (don't reject) a limit past the max window so a client paging beyond it - // gets the capped tail and pagination stops cleanly — a hard `.max` rejection - // would fail the read and stall "load earlier" at the boundary. - limit: z - .number() - .int() - .positive() - .transform((value) => Math.min(value, MOBILE_NATIVE_CHAT_MAX_WINDOW)) - .optional(), - // Optional client-supplied cleanup token. When present, the subscribe handler - // keys the fs-watcher cleanup under it so registration and unsubscribe derive - // from the SAME token (back-compat: falls back to `agent:sessionId` when absent, - // which is exactly what existing mobile clients rely on). - subscriptionId: z.string().min(1).optional(), - // Authoritative transcript path from the agent hook (providerSession), used to - // locate the file directly when the session id no longer names it (recent - // Claude Code). Optional for back-compat with older clients. - transcriptPath: z.string().min(1).optional(), - beforeOffset: z.number().int().nonnegative().optional() -}) - -const NativeChatUnsubscribe = z.object({ - subscriptionId: z.string().min(1).optional() -}) - // Why: a long agent session can hold thousands of turns (with full tool I/O). // Shipping all of them over the paired connection and rendering them at once // freezes the mobile app, so the runtime RPC windows to the most recent slice — @@ -63,7 +26,6 @@ const NativeChatUnsubscribe = z.object({ // Small first page for a fast initial paint; the client raises `limit` to load // older history as the user scrolls back. const MOBILE_NATIVE_CHAT_DEFAULT_WINDOW = 40 -const MOBILE_NATIVE_CHAT_MAX_WINDOW = 2000 // Why: a single tool result (a big file read, a long diff) can be hundreds of KB. // The mobile view only previews tool block bodies, so truncate them on the wire // to keep the payload small; the marker tells the user content was clipped. @@ -211,6 +173,7 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [ agent: params.agent, sessionId: params.sessionId, transcriptPath: params.transcriptPath, + providerSession: params.providerSession, limit, beforeOffset: params.beforeOffset }, @@ -272,6 +235,7 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [ agent: params.agent, sessionId: params.sessionId, transcriptPath: params.transcriptPath, + providerSession: params.providerSession, initialLimit: limit, onInitialSnapshot: (messages, hasMore, beforeOffset, error, lifecycle) => { if (closed) { diff --git a/src/preload/api/native-chat-api.ts b/src/preload/api/native-chat-api.ts index 01eac44bde6..cd5f881a9cf 100644 --- a/src/preload/api/native-chat-api.ts +++ b/src/preload/api/native-chat-api.ts @@ -3,6 +3,7 @@ import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types' +import type { AgentProviderSessionMetadata } from '../../shared/agent-session-resume' // notFound marks a not-yet-on-disk miss (retry-worthy) vs a real read/parse error (#8401). export type NativeChatReadSessionResult = @@ -10,7 +11,7 @@ export type NativeChatReadSessionResult = messages: NativeChatMessage[] lifecycle?: NativeChatTurnLifecycle } - | { error: string; notFound?: true } + | { error: string; notFound?: true; unverifiable?: true } /** Messages appended to a live-tailed transcript since the previous emit. */ export type NativeChatAppendedMessages = NativeChatMessage[] @@ -49,6 +50,7 @@ export type NativeChatSubscribeArgs = { sessionId: string /** Authoritative transcript path from the agent hook (providerSession). */ transcriptPath?: string + providerSession?: AgentProviderSessionMetadata /** First snapshot size; later readSession calls grow this for pagination. */ limit?: number } @@ -60,7 +62,8 @@ export type NativeChatApi = { agent: AgentType, sessionId: string, limit?: number, - transcriptPath?: string + transcriptPath?: string, + providerSession?: AgentProviderSessionMetadata ) => Promise /** Live-tail a transcript. The first frame is a bounded race-safe snapshot; * later frames contain only newly appended messages. */ diff --git a/src/preload/index.ts b/src/preload/index.ts index b8a00c6d647..c4c95e473e8 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -4466,9 +4466,16 @@ const api = { agent: AgentType, sessionId: string, limit?: number, - transcriptPath?: string + transcriptPath?: string, + providerSession?: AgentProviderSessionMetadata ): Promise => - ipcRenderer.invoke('nativeChat:readSession', { agent, sessionId, limit, transcriptPath }), + ipcRenderer.invoke('nativeChat:readSession', { + agent, + sessionId, + limit, + transcriptPath, + providerSession + }), /** Start live tailing; onAppended fires with only newly-appended messages. Returns an unsubscribe fn that closes the watcher. */ subscribe: ( args: { @@ -4476,6 +4483,7 @@ const api = { agent: AgentType sessionId: string transcriptPath?: string + providerSession?: AgentProviderSessionMetadata limit?: number }, onFrame: (frame: NativeChatSubscriptionFrame) => void diff --git a/src/renderer/src/components/native-chat/NativeChatSessionGate.tsx b/src/renderer/src/components/native-chat/NativeChatSessionGate.tsx index 0d7b6615b53..21a95591a4a 100644 --- a/src/renderer/src/components/native-chat/NativeChatSessionGate.tsx +++ b/src/renderer/src/components/native-chat/NativeChatSessionGate.tsx @@ -35,7 +35,8 @@ export function NativeChatSessionGate({ return { ...currentResolution, sessionId: previousResolution.sessionId, - transcriptPath: previousResolution.transcriptPath + transcriptPath: previousResolution.transcriptPath, + providerSession: previousResolution.providerSession } } return currentResolution diff --git a/src/renderer/src/components/native-chat/NativeChatView.tsx b/src/renderer/src/components/native-chat/NativeChatView.tsx index 4fc6df0f13b..d8b117d083c 100644 --- a/src/renderer/src/components/native-chat/NativeChatView.tsx +++ b/src/renderer/src/components/native-chat/NativeChatView.tsx @@ -95,6 +95,7 @@ export default function NativeChatView({ agent={resolution.agent} sessionId={resolution.sessionId} transcriptPath={resolution.transcriptPath} + providerSession={resolution.providerSession} isVisible={isVisible} targetPtyId={targetPtyId} terminalTabId={terminalTabId} @@ -112,6 +113,7 @@ function NativeChatResolvedView({ agent, sessionId, transcriptPath, + providerSession, isVisible, targetPtyId, terminalTabId, @@ -129,6 +131,7 @@ function NativeChatResolvedView({ agent, sessionId, transcriptPath, + providerSession, runtimeEnvironmentId, enabled: isVisible }) diff --git a/src/renderer/src/components/native-chat/native-chat-live-session-types.ts b/src/renderer/src/components/native-chat/native-chat-live-session-types.ts new file mode 100644 index 00000000000..50911ba6e6b --- /dev/null +++ b/src/renderer/src/components/native-chat/native-chat-live-session-types.ts @@ -0,0 +1,28 @@ +import type { AgentProviderSessionMetadata } from '../../../../shared/agent-session-resume' +import type { + AgentType, + NativeChatMessage, + NativeChatSession +} from '../../../../shared/native-chat-types' + +export type UseNativeChatLiveSessionArgs = { + paneKey: string + agent: AgentType + sessionId: string | null + transcriptPath?: string | null + providerSession?: AgentProviderSessionMetadata | null + runtimeEnvironmentId?: string | null + enabled?: boolean +} + +export type ReadState = + | { phase: 'loading' } + | { phase: 'ready'; messages: NativeChatMessage[] } + | { phase: 'error'; error: string } + +export type NativeChatLiveSession = NativeChatSession & { + hasMore: boolean + loadingEarlier: boolean + loadEarlier: () => void + readPhase: ReadState['phase'] +} diff --git a/src/renderer/src/components/native-chat/native-chat-live-source-key.ts b/src/renderer/src/components/native-chat/native-chat-live-source-key.ts new file mode 100644 index 00000000000..08e4856d43f --- /dev/null +++ b/src/renderer/src/components/native-chat/native-chat-live-source-key.ts @@ -0,0 +1,16 @@ +import type { AgentProviderSessionMetadata } from '../../../../shared/agent-session-resume' +import type { UseNativeChatLiveSessionArgs } from './native-chat-live-session-types' + +export function nativeChatLiveSourceKey( + args: UseNativeChatLiveSessionArgs, + providerSession: AgentProviderSessionMetadata | null +): string { + return JSON.stringify([ + args.paneKey, + args.runtimeEnvironmentId ?? null, + args.agent, + args.sessionId, + args.transcriptPath ?? null, + providerSession?.executionHostId ?? null + ]) +} diff --git a/src/renderer/src/components/native-chat/native-chat-pane-resolution.test.ts b/src/renderer/src/components/native-chat/native-chat-pane-resolution.test.ts index 42f9605fdea..a97268bf47d 100644 --- a/src/renderer/src/components/native-chat/native-chat-pane-resolution.test.ts +++ b/src/renderer/src/components/native-chat/native-chat-pane-resolution.test.ts @@ -26,7 +26,11 @@ describe('resolveNativeChatSession', () => { agentStatusEntry: entry({ paneKey, agentType: 'claude', - providerSession: { key: 'session_id', id: 'sess-abc' } + providerSession: { + key: 'session_id', + id: 'sess-abc', + executionHostId: 'ssh:builder' + } }), ptyId: 'pty-1' }) @@ -34,6 +38,11 @@ describe('resolveNativeChatSession', () => { agent: 'claude', sessionId: 'sess-abc', transcriptPath: null, + providerSession: { + key: 'session_id', + id: 'sess-abc', + executionHostId: 'ssh:builder' + }, ptyId: 'pty-1', paneKey }) @@ -60,6 +69,11 @@ describe('resolveNativeChatSession', () => { agent: 'claude', sessionId: 'sess-abc', transcriptPath: '/home/u/.claude/projects/slug/real-uuid.jsonl', + providerSession: { + key: 'session_id', + id: 'sess-abc', + transcriptPath: '/home/u/.claude/projects/slug/real-uuid.jsonl' + }, ptyId: 'pty-1', paneKey }) @@ -75,7 +89,14 @@ describe('resolveNativeChatSession', () => { agentStatusEntry: entry({ paneKey, agentType: 'claude' }), ptyId: 'pty-1' }) - ).toEqual({ agent: 'claude', sessionId: null, transcriptPath: null, ptyId: 'pty-1', paneKey }) + ).toEqual({ + agent: 'claude', + sessionId: null, + transcriptPath: null, + providerSession: null, + ptyId: 'pty-1', + paneKey + }) }) it('resolves two split leaves independently to their own values', () => { @@ -105,6 +126,7 @@ describe('resolveNativeChatSession', () => { agent: 'claude', sessionId: 'left-sess', transcriptPath: null, + providerSession: { key: 'session_id', id: 'left-sess' }, ptyId: 'pty-left', paneKey: leftKey }) @@ -112,6 +134,7 @@ describe('resolveNativeChatSession', () => { agent: 'codex', sessionId: 'right-sess', transcriptPath: null, + providerSession: { key: 'session_id', id: 'right-sess' }, ptyId: 'pty-right', paneKey: rightKey }) @@ -134,6 +157,7 @@ describe('resolveNativeChatSession', () => { agent: 'codex', sessionId: 'codex-1', transcriptPath: null, + providerSession: { key: 'session_id', id: 'codex-1' }, ptyId: 'pty-1', paneKey }) @@ -154,6 +178,7 @@ describe('resolveNativeChatSession', () => { agent: resolvedAgent, sessionId: null, transcriptPath: null, + providerSession: null, ptyId: 'pty-1', paneKey }) @@ -295,6 +320,7 @@ describe('resolveNativeChatSession', () => { agent: 'codex', sessionId: 'codex-live', transcriptPath: null, + providerSession: { key: 'session_id', id: 'codex-live' }, ptyId: 'pty-1', paneKey }) diff --git a/src/renderer/src/components/native-chat/native-chat-pane-resolution.ts b/src/renderer/src/components/native-chat/native-chat-pane-resolution.ts index a363b977ff1..8d2513c4ec9 100644 --- a/src/renderer/src/components/native-chat/native-chat-pane-resolution.ts +++ b/src/renderer/src/components/native-chat/native-chat-pane-resolution.ts @@ -1,4 +1,5 @@ import type { AgentStatusEntry, AgentType } from '../../../../shared/agent-status-types' +import type { AgentProviderSessionMetadata } from '../../../../shared/agent-session-resume' import type { TuiAgent } from '../../../../shared/tui-agent' import { isNativeChatSupportedAgent } from './native-chat-availability' @@ -33,6 +34,7 @@ export type NativeChatPaneResolution = { /** Authoritative transcript path from the hook, when reported. Preferred over * reconstructing the path from sessionId (recent Claude Code diverges them). */ transcriptPath: string | null + providerSession: AgentProviderSessionMetadata | null ptyId: string | null paneKey: string } @@ -54,6 +56,7 @@ export function resolveNativeChatSession( agent, sessionId: input.agentStatusEntry?.providerSession?.id ?? null, transcriptPath: input.agentStatusEntry?.providerSession?.transcriptPath ?? null, + providerSession: input.agentStatusEntry?.providerSession ?? null, ptyId: input.ptyId, paneKey: input.paneKey } diff --git a/src/renderer/src/components/native-chat/native-chat-runtime-contract.ts b/src/renderer/src/components/native-chat/native-chat-runtime-contract.ts index 329c93f852f..32d1b31638f 100644 --- a/src/renderer/src/components/native-chat/native-chat-runtime-contract.ts +++ b/src/renderer/src/components/native-chat/native-chat-runtime-contract.ts @@ -53,7 +53,8 @@ export function parseRuntimeNativeChatReadSessionResult( if (typeof record.error === 'string') { return { error: record.error, - ...(record.notFound === true ? { notFound: true } : {}) + ...(record.notFound === true ? { notFound: true } : {}), + ...(record.unverifiable === true ? { unverifiable: true } : {}) } } return { error: RUNTIME_NATIVE_CHAT_READ_ERROR } diff --git a/src/renderer/src/components/native-chat/native-chat-session-transport.test.ts b/src/renderer/src/components/native-chat/native-chat-session-transport.test.ts index 47832b10897..54000e98579 100644 --- a/src/renderer/src/components/native-chat/native-chat-session-transport.test.ts +++ b/src/renderer/src/components/native-chat/native-chat-session-transport.test.ts @@ -82,6 +82,32 @@ describe('getNativeChatSessionTransport — selection', () => { expect(nativeChatReadSession).not.toHaveBeenCalled() }) + it('forwards the whole provider-session locator through desktop IPC', async () => { + nativeChatReadSession.mockResolvedValue({ messages: [] }) + const providerSession = { + key: 'session_id' as const, + id: 'sess-1', + transcriptPath: '/t/path', + executionHostId: 'ssh:builder' as const + } + + await getNativeChatSessionTransport(null).readSession( + 'claude', + 'sess-1', + 40, + '/t/path', + providerSession + ) + + expect(nativeChatReadSession).toHaveBeenCalledWith( + 'claude', + 'sess-1', + 40, + '/t/path', + providerSession + ) + }) + it('validates lifecycle metadata on runtime read responses', async () => { markRuntimeEnvironmentCompatible(ENV) const lifecycle = { state: 'completed', turnId: 'turn-1', timestamp: 42 } as const @@ -206,6 +232,37 @@ describe('runtime subscribe', () => { }) }) + it('forwards the whole provider-session locator through runtime subscriptions', async () => { + markRuntimeEnvironmentCompatible(ENV) + stubSubscribe() + const providerSession = { + key: 'session_id' as const, + id: 'sess-1', + transcriptPath: '/t/path', + executionHostId: 'ssh:builder' as const + } + + getNativeChatSessionTransport(ENV).subscribe( + { + subscriptionId: 's-1', + agent: 'claude', + sessionId: 'sess-1', + transcriptPath: '/t/path', + providerSession + }, + vi.fn() + ) + await Promise.resolve() + + expect(runtimeEnvironmentsSubscribe).toHaveBeenCalledWith( + expect.objectContaining({ + method: 'nativeChat.subscribe', + params: expect.objectContaining({ providerSession }) + }), + expect.any(Object) + ) + }) + it('validates lifecycle metadata on runtime stream frames', async () => { markRuntimeEnvironmentCompatible(ENV) const { deliver } = stubSubscribe() diff --git a/src/renderer/src/components/native-chat/native-chat-session-transport.ts b/src/renderer/src/components/native-chat/native-chat-session-transport.ts index 0fe22211db0..f13427135b9 100644 --- a/src/renderer/src/components/native-chat/native-chat-session-transport.ts +++ b/src/renderer/src/components/native-chat/native-chat-session-transport.ts @@ -43,8 +43,10 @@ export function toRuntimeNativeChatErrorMessage(err: unknown): string { * using this adapter (R3). Preserves whatever `subscribe` returns (sync fn on * desktop, promise on the web bridge) — the hook's teardown handles both (R6). */ const localNativeChatTransport: NativeChatSessionTransport = { - readSession: (agent, sessionId, limit, transcriptPath) => - window.api.nativeChat.readSession(agent, sessionId, limit, transcriptPath), + readSession: (agent, sessionId, limit, transcriptPath, providerSession) => + providerSession + ? window.api.nativeChat.readSession(agent, sessionId, limit, transcriptPath, providerSession) + : window.api.nativeChat.readSession(agent, sessionId, limit, transcriptPath), subscribe: (args, onFrame) => window.api.nativeChat.subscribe(args, onFrame) } @@ -52,12 +54,18 @@ function createRuntimeNativeChatTransport(environmentId: string): NativeChatSess const target: RuntimeClientTarget = { kind: 'environment', environmentId } return { - readSession: async (agent, sessionId, limit, transcriptPath) => { + readSession: async (agent, sessionId, limit, transcriptPath, providerSession) => { try { const result = await callRuntimeRpc( target, 'nativeChat.readSession', - { agent, sessionId, limit, transcriptPath }, + { + agent, + sessionId, + limit, + transcriptPath, + ...(providerSession ? { providerSession } : {}) + }, { timeoutMs: 15_000 } ) return parseRuntimeNativeChatReadSessionResult(result) @@ -66,7 +74,7 @@ function createRuntimeNativeChatTransport(environmentId: string): NativeChatSess } }, subscribe: (args, onFrame) => { - const { subscriptionId, agent, sessionId, transcriptPath, limit } = args + const { subscriptionId, agent, sessionId, transcriptPath, providerSession, limit } = args let cancelled = false let receivedInitial = false let handleUnsubscribe: (() => void) | null = null @@ -105,7 +113,14 @@ function createRuntimeNativeChatTransport(environmentId: string): NativeChatSess { selector: environmentId, method: 'nativeChat.subscribe', - params: { subscriptionId, agent, sessionId, transcriptPath, limit }, + params: { + subscriptionId, + agent, + sessionId, + transcriptPath, + providerSession, + limit + }, timeoutMs: 15_000 }, { diff --git a/src/renderer/src/components/native-chat/native-chat-subscription-identity.ts b/src/renderer/src/components/native-chat/native-chat-subscription-identity.ts new file mode 100644 index 00000000000..bad47ee68dd --- /dev/null +++ b/src/renderer/src/components/native-chat/native-chat-subscription-identity.ts @@ -0,0 +1,8 @@ +let subscriptionCounter = 0 + +export const NOTFOUND_RETRY_WINDOW_MS = 60_000 + +export function nextNativeChatSubscriptionId(): string { + subscriptionCounter += 1 + return `native-chat-${subscriptionCounter}-${Date.now()}` +} diff --git a/src/renderer/src/components/native-chat/native-chat-view-types.ts b/src/renderer/src/components/native-chat/native-chat-view-types.ts index de095f9b232..8020841a501 100644 --- a/src/renderer/src/components/native-chat/native-chat-view-types.ts +++ b/src/renderer/src/components/native-chat/native-chat-view-types.ts @@ -1,5 +1,6 @@ import type { TuiAgent } from '../../../../shared/tui-agent' import type { NativeChatSession } from '../../../../shared/native-chat-types' +import type { AgentProviderSessionMetadata } from '../../../../shared/agent-session-resume' import type { NativeChatContextMenuActions } from './use-native-chat-context-menu' export type NativeChatViewProps = { @@ -27,6 +28,7 @@ export type NativeChatResolvedViewProps = { agent: NativeChatSession['agent'] sessionId: string | null transcriptPath: string | null + providerSession: AgentProviderSessionMetadata | null isVisible: boolean targetPtyId: string | null terminalTabId: string diff --git a/src/renderer/src/components/native-chat/use-native-chat-live-session.ts b/src/renderer/src/components/native-chat/use-native-chat-live-session.ts index 141a65ed501..078cf17a9f4 100644 --- a/src/renderer/src/components/native-chat/use-native-chat-live-session.ts +++ b/src/renderer/src/components/native-chat/use-native-chat-live-session.ts @@ -1,9 +1,7 @@ import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from 'react' import { NATIVE_CHAT_SOURCE_PRIORITY, - type AgentType, - type NativeChatMessage, - type NativeChatSession + type NativeChatMessage } from '../../../../shared/native-chat-types' import { applyAppend, @@ -22,53 +20,27 @@ import { useNativeChatHookStatus } from './use-native-chat-hook-status' import { useNativeChatAssembledMessages } from './use-native-chat-assembled-messages' import { createNativeChatReadRetryTimer } from './native-chat-read-retry-timer' import { openNativeChatTranscriptStream } from './native-chat-stream-teardown' +import { useNativeChatProviderSessionLocator } from './use-native-chat-provider-session-locator' +import { + nextNativeChatSubscriptionId, + NOTFOUND_RETRY_WINDOW_MS +} from './native-chat-subscription-identity' +import type { + NativeChatLiveSession, + ReadState, + UseNativeChatLiveSessionArgs +} from './native-chat-live-session-types' +import { nativeChatLiveSourceKey } from './native-chat-live-source-key' -export type UseNativeChatLiveSessionArgs = { - /** Composite `${tabId}:${leafId}` key — selects the live hook entry. */ - paneKey: string - agent: AgentType - /** The agent's own session id, or null before it reports one — nothing to read/tail, so the view shows live hook state. */ - sessionId: string | null - /** Authoritative transcript path from the hook, preferred over reconstructing it from sessionId. Null when not reported. */ - transcriptPath?: string | null - /** Runtime owner (Model B): non-null routes read/subscribe to the remote host; null keeps the local IPC path. */ - runtimeEnvironmentId?: string | null - /** False suspends transcript IO while retaining the last committed session. */ - enabled?: boolean -} - -/** A live session plus the older-history pagination controls the view needs. */ -export type NativeChatLiveSession = NativeChatSession & { - /** True when an older page may still exist (the last read filled the window). */ - hasMore: boolean - /** Whether an older-history page is currently loading. */ - loadingEarlier: boolean - /** Grow the read window to page in older history (scrolled-to-top trigger). */ - loadEarlier: () => void - /** Raw initial-read phase. `status` is not a substitute: a live 'working' hook - * outranks (and so hides) 'loading', which would let a consumer deciding from - * an empty list treat an in-flight transcript as real history. */ - readPhase: ReadState['phase'] -} +export type { + NativeChatLiveSession, + ReadState, + UseNativeChatLiveSessionArgs +} from './native-chat-live-session-types' // Stable empty-base reference so a non-ready read doesn't churn the base axis. const EMPTY_MESSAGES: readonly NativeChatMessage[] = [] -let subscriptionCounter = 0 - -function nextSubscriptionId(): string { - subscriptionCounter += 1 - return `native-chat-${subscriptionCounter}-${Date.now()}` -} - -// Why: a new session's transcript can take minutes to appear on disk (#8401). -const NOTFOUND_RETRY_WINDOW_MS = 60_000 - -export type ReadState = - | { phase: 'loading' } - | { phase: 'ready'; messages: NativeChatMessage[] } - | { phase: 'error'; error: string } - /** * Renderer hook that streams a NativeChatSession for a pane: windowed * `readSession` + live `subscribe` tail, merged with live hook turn-state. @@ -86,12 +58,21 @@ export type ReadState = export function useNativeChatLiveSession( args: UseNativeChatLiveSessionArgs ): NativeChatLiveSession { - const { paneKey, agent, sessionId, transcriptPath, runtimeEnvironmentId, enabled = true } = args + const { + paneKey, + agent, + sessionId, + transcriptPath, + providerSession, + runtimeEnvironmentId, + enabled = true + } = args // Stable per owner id so a re-render without an owner flip keeps the same transport and doesn't re-subscribe. const transport = useMemo( () => getNativeChatSessionTransport(runtimeEnvironmentId ?? null), [runtimeEnvironmentId] ) + const providerSessionLocator = useNativeChatProviderSessionLocator(providerSession) const [read, setRead] = useState({ phase: 'loading' }) const [hasMore, setHasMore] = useState(false) const [loadingEarlier, setLoadingEarlier] = useState(false) @@ -117,13 +98,7 @@ export function useNativeChatLiveSession( const latestTransport = useRef(transport) latestTransport.current = transport const transcriptEpochRef = useRef(0) - const sourceKey = JSON.stringify([ - paneKey, - runtimeEnvironmentId ?? null, - agent, - sessionId, - transcriptPath ?? null - ]) + const sourceKey = nativeChatLiveSourceKey(args, providerSessionLocator) const retainedSourceKeyRef = useRef(sourceKey) useEffect(() => { @@ -179,7 +154,13 @@ export function useNativeChatLiveSession( return } void transport - .readSession(agent, activeSessionId, limitRef.current, transcriptPath ?? undefined) + .readSession( + agent, + activeSessionId, + limitRef.current, + transcriptPath ?? undefined, + providerSessionLocator ?? undefined + ) .then((result) => { if (cancelled || !latestEnabled.current || frameArrived) { return @@ -210,7 +191,7 @@ export function useNativeChatLiveSession( loadSession(0) - const subscriptionId = nextSubscriptionId() + const subscriptionId = nextNativeChatSubscriptionId() const closeStream = openNativeChatTranscriptStream( transport, { @@ -218,6 +199,7 @@ export function useNativeChatLiveSession( agent, sessionId, transcriptPath: transcriptPath ?? undefined, + providerSession: providerSessionLocator ?? undefined, limit: limitRef.current }, (frame) => { @@ -259,7 +241,16 @@ export function useNativeChatLiveSession( closeStream() } // `transport` identity changes on an owner flip, re-running this effect to re-subscribe against the new host. - }, [agent, enabled, sessionId, sourceKey, transcriptPath, transport, transcriptLifecycleControl]) + }, [ + agent, + enabled, + providerSessionLocator, + sessionId, + sourceKey, + transcriptPath, + transport, + transcriptLifecycleControl + ]) const loadEarlier = useCallback(() => { if ( @@ -276,7 +267,13 @@ export function useNativeChatLiveSession( const lifecycleRevision = transcriptLifecycleControl.revision() setLoadingEarlier(true) void transport - .readSession(agent, sessionId, nextLimit, transcriptPath ?? undefined) + .readSession( + agent, + sessionId, + nextLimit, + transcriptPath ?? undefined, + providerSessionLocator ?? undefined + ) .then((result) => { // Ignore a stale resolve from a swapped session or flipped owner — either would paint the wrong host's history. if ( @@ -307,6 +304,7 @@ export function useNativeChatLiveSession( }) }, [ agent, + providerSessionLocator, sessionId, transcriptPath, transport, diff --git a/src/renderer/src/components/native-chat/use-native-chat-provider-session-locator.ts b/src/renderer/src/components/native-chat/use-native-chat-provider-session-locator.ts new file mode 100644 index 00000000000..24b3bc47618 --- /dev/null +++ b/src/renderer/src/components/native-chat/use-native-chat-provider-session-locator.ts @@ -0,0 +1,23 @@ +import { useMemo } from 'react' +import type { AgentProviderSessionMetadata } from '../../../../shared/agent-session-resume' + +export function useNativeChatProviderSessionLocator( + providerSession: AgentProviderSessionMetadata | null | undefined +): AgentProviderSessionMetadata | null { + const key = providerSession?.key + const id = providerSession?.id + const transcriptPath = providerSession?.transcriptPath + const executionHostId = providerSession?.executionHostId + return useMemo( + () => + key && id + ? { + key, + id, + ...(transcriptPath ? { transcriptPath } : {}), + ...(executionHostId ? { executionHostId } : {}) + } + : null, + [executionHostId, id, key, transcriptPath] + ) +} diff --git a/src/renderer/src/components/native-chat/use-native-chat-retained-session.ts b/src/renderer/src/components/native-chat/use-native-chat-retained-session.ts index 6d176f2ac55..0cc6812a2e7 100644 --- a/src/renderer/src/components/native-chat/use-native-chat-retained-session.ts +++ b/src/renderer/src/components/native-chat/use-native-chat-retained-session.ts @@ -19,7 +19,8 @@ export function useNativeChatRetainedSession( args.runtimeEnvironmentId ?? null, args.agent, args.sessionId, - args.transcriptPath ?? null + args.transcriptPath ?? null, + args.providerSession?.executionHostId ?? null ]) const activeIdentityRef = useRef(identity) const retentionRef = useRef(createNativeChatTranscriptRetention()) diff --git a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-host-session-launch.test.ts b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-host-session-launch.test.ts index 6185198d59e..7073737ff07 100644 --- a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-host-session-launch.test.ts +++ b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-host-session-launch.test.ts @@ -316,7 +316,11 @@ describe('createRemoteRuntimePtyTransport', () => { env: { CLIENT_ONLY: 'must-not-cross' }, launchAgent: 'claude', agentArgsOverride: '--permission-mode plan', - resumeProviderSession: { key: 'session_id', id: 'provider-session' }, + resumeProviderSession: { + key: 'session_id', + id: 'provider-session', + executionHostId: 'ssh:builder' + }, tabId: 'tab-1', leafId: '11111111-1111-4111-8111-111111111111' }) diff --git a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts index 7d50b9f48de..fe3a8a6a536 100644 --- a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts +++ b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts @@ -9,6 +9,7 @@ import type { RuntimeCreateAgentSessionResult, RuntimeEnsureAgentSessionResult } from '../../../../shared/agent-session-host-authority' +import { agentProviderSessionForResume } from '../../../../shared/agent-session-resume' import type { ExecutionHostId } from '../../../../shared/execution-host' import type { TabActivationIntent } from '../../../../shared/tab-activation-intent' import type { @@ -2010,7 +2011,11 @@ export function createRemoteRuntimePtyTransport( const envToSend = options.env ?? env const envToDeleteToSend = options.envToDelete ?? envToDelete const launchConfigToSend = options.launchConfig ?? launchConfig - const resumeProviderSessionToSend = options.resumeProviderSession ?? resumeProviderSession + const resumeProviderSessionCandidate = + options.resumeProviderSession ?? resumeProviderSession + const resumeProviderSessionToSend = resumeProviderSessionCandidate + ? agentProviderSessionForResume(resumeProviderSessionCandidate) + : undefined const launchTokenToSend = options.launchToken ?? launchToken const launchAgentToSend = options.launchAgent ?? launchAgent const legacyCreateParams = { diff --git a/src/renderer/src/runtime/web-runtime-session-terminal-host-authority.test.ts b/src/renderer/src/runtime/web-runtime-session-terminal-host-authority.test.ts index 209cd182da9..5e7d6e8303c 100644 --- a/src/renderer/src/runtime/web-runtime-session-terminal-host-authority.test.ts +++ b/src/renderer/src/runtime/web-runtime-session-terminal-host-authority.test.ts @@ -167,7 +167,11 @@ describe('createWebRuntimeSessionTerminal', () => { ...(sessionKind === 'resume' ? { command: "codex resume 'session-1'", - providerSession: { key: 'session_id' as const, id: 'session-1' } + providerSession: { + key: 'session_id' as const, + id: 'session-1', + executionHostId: 'ssh:builder' as const + } } : {}), activate diff --git a/src/renderer/src/runtime/web-runtime-session.ts b/src/renderer/src/runtime/web-runtime-session.ts index f8ceac02da2..1df30e87994 100644 --- a/src/renderer/src/runtime/web-runtime-session.ts +++ b/src/renderer/src/runtime/web-runtime-session.ts @@ -15,9 +15,10 @@ import type { } from '../../../shared/runtime-types' import type { TerminalPaneSplitSource } from '../../../shared/feature-education-telemetry' import type { StartupCommandDelivery } from '../../../shared/codex-startup-delivery' -import type { - SleepingAgentLaunchConfig, - AgentProviderSessionMetadata +import { + agentProviderSessionForResume, + type AgentProviderSessionMetadata, + type SleepingAgentLaunchConfig } from '../../../shared/agent-session-resume' import { BROWSER_TAB_CREATE_KNOWN_ID_RUNTIME_CAPABILITY } from '../../../shared/protocol-version' import { agentResumeHostAuthorityCapability } from './agent-resume-host-authority-capability' @@ -294,7 +295,7 @@ async function createWebRuntimeSessionTerminalResult( kind: 'explicit', worktree: toRuntimeWorktreeSelector(args.worktreeId), agent, - providerSession: args.providerSession!, + providerSession: agentProviderSessionForResume(args.providerSession!), ...(args.launchConfig?.ompResumeFilePath ? { ompResumeFilePath: args.launchConfig.ompResumeFilePath } : {}), diff --git a/src/shared/agent-session-resume.test.ts b/src/shared/agent-session-resume.test.ts index c6025936431..a58979de7ec 100644 --- a/src/shared/agent-session-resume.test.ts +++ b/src/shared/agent-session-resume.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from 'vitest' import { + agentProviderSessionForResume, agentProviderSessionsEqual, extractAgentProviderSession, getAgentResumeArgv, @@ -166,4 +167,58 @@ describe('agent session resume metadata', () => { }) ).toEqual({ key: 'session_id', id: 'ok' }) }) + + it('normalizes host ownership and keeps it in identity comparisons', () => { + const local = normalizeAgentProviderSession({ + key: 'session_id', + id: 'same', + executionHostId: 'local' + }) + const remote = normalizeAgentProviderSession({ + key: 'session_id', + id: 'same', + executionHostId: 'ssh:builder' + }) + + expect(local).toMatchObject({ executionHostId: 'local' }) + expect(remote).toMatchObject({ executionHostId: 'ssh:builder' }) + expect(agentProviderSessionsEqual('claude', local ?? undefined, remote ?? undefined)).toBe( + false + ) + expect( + normalizeAgentProviderSession({ + key: 'session_id', + id: 'same', + executionHostId: 'invalid-host' + }) + ).toEqual({ key: 'session_id', id: 'same' }) + }) + + it('treats omitted legacy ownership as local during identity comparison', () => { + expect( + agentProviderSessionsEqual( + 'claude', + { key: 'session_id', id: 'same' }, + { key: 'session_id', id: 'same', executionHostId: 'local' } + ) + ).toBe(true) + expect( + agentProviderSessionsEqual( + 'claude', + { key: 'session_id', id: 'same' }, + { key: 'session_id', id: 'same', executionHostId: 'ssh:builder' } + ) + ).toBe(false) + }) + + it('strips host ownership at strict legacy resume seams', () => { + expect( + agentProviderSessionForResume({ + key: 'session_id', + id: 'same', + transcriptPath: '/tmp/same.jsonl', + executionHostId: 'ssh:builder' + }) + ).toEqual({ key: 'session_id', id: 'same', transcriptPath: '/tmp/same.jsonl' }) + }) }) diff --git a/src/shared/agent-session-resume.ts b/src/shared/agent-session-resume.ts index 3e763fc20a6..ddfd8cb463d 100644 --- a/src/shared/agent-session-resume.ts +++ b/src/shared/agent-session-resume.ts @@ -1,5 +1,10 @@ import type { AgentHookSource } from './agent-hook-relay' import type { AgentStatusState } from './agent-status-types' +import { + LOCAL_EXECUTION_HOST_ID, + normalizeExecutionHostId, + type ExecutionHostId +} from './execution-host' import type { TuiAgent } from './tui-agent' export const RESUMABLE_TUI_AGENTS = [ @@ -26,6 +31,8 @@ export type AgentProviderSessionKey = 'session_id' | 'conversation_id' export type AgentProviderSessionMetadata = { key: AgentProviderSessionKey id: string + /** Execution host that owns the provider session and its transcript bytes. */ + executionHostId?: ExecutionHostId /** Authoritative on-disk transcript/rollout path reported by the agent's hook * (Claude/Codex `transcript_path`), when available. Native chat reads this * directly because recent Claude Code versions name the transcript file with a @@ -161,7 +168,16 @@ export function normalizeAgentProviderSession(raw: unknown): AgentProviderSessio // Why: persisted/relay metadata crosses a trust boundary too; apply the same // control-character rejection used for hook-reported transcript paths. const transcriptPath = readTranscriptPathFromKeys(record, ['transcriptPath']) - return transcriptPath ? { key, id, transcriptPath } : { key, id } + const executionHostId = + typeof record.executionHostId === 'string' + ? (normalizeExecutionHostId(record.executionHostId) ?? undefined) + : undefined + return { + key, + id, + ...(transcriptPath ? { transcriptPath } : {}), + ...(executionHostId ? { executionHostId } : {}) + } } /** Compare the provider-owned values that identify the CLI resume target. @@ -177,10 +193,23 @@ export function agentProviderSessionsEqual( return ( left.key === right.key && left.id === right.id && + (left.executionHostId ?? LOCAL_EXECUTION_HOST_ID) === + (right.executionHostId ?? LOCAL_EXECUTION_HOST_ID) && ((agent !== 'pi' && agent !== 'prime-agent') || left.transcriptPath === right.transcriptPath) ) } +/** Older runtime resume schemas do not know transcript ownership metadata. */ +export function agentProviderSessionForResume( + providerSession: AgentProviderSessionMetadata +): Omit { + return { + key: providerSession.key, + id: providerSession.id, + ...(providerSession.transcriptPath ? { transcriptPath: providerSession.transcriptPath } : {}) + } +} + export function extractAgentProviderSession( source: AgentHookSource, payload: Record diff --git a/src/shared/workspace-session-sleeping-agents.ts b/src/shared/workspace-session-sleeping-agents.ts index 4ee620efa82..f8730591646 100644 --- a/src/shared/workspace-session-sleeping-agents.ts +++ b/src/shared/workspace-session-sleeping-agents.ts @@ -1,4 +1,5 @@ import { z } from 'zod' +import { normalizeExecutionHostId, type ExecutionHostId } from './execution-host' import { getAgentResumeArgv, normalizeAgentProviderSession, @@ -19,7 +20,12 @@ const agentProviderSessionSchema = z.preprocess( id: z.string().min(1).max(512), // Why: Pi resumes by its authoritative session file, so dropping this // field during hydration makes an otherwise valid record unusable. - transcriptPath: z.string().min(1).optional() + transcriptPath: z.string().min(1).optional(), + executionHostId: z + .custom( + (value) => typeof value === 'string' && normalizeExecutionHostId(value) === value + ) + .optional() }) ) diff --git a/tests/e2e/headless-native-chat-runtime-routing.spec.ts b/tests/e2e/headless-native-chat-runtime-routing.spec.ts new file mode 100644 index 00000000000..cc53b07c977 --- /dev/null +++ b/tests/e2e/headless-native-chat-runtime-routing.spec.ts @@ -0,0 +1,65 @@ +import { rmSync, writeFileSync } from 'node:fs' +import path from 'node:path' +import { expect, test } from './helpers/orca-app' +import { launchHeadlessPairedRuntimeHost } from './helpers/headless-paired-runtime-host' + +function claudeLine(id: string, text: string): string { + return `${JSON.stringify({ + type: 'assistant', + uuid: id, + message: { role: 'assistant', content: [{ type: 'text', text }] } + })}\n` +} + +test('routes native-chat ownership through a headless runtime', async () => { + test.setTimeout(180_000) + const host = await launchHeadlessPairedRuntimeHost() + const homePath = await host.app.evaluate(({ app }) => app.getPath('home')) + const transcriptPath = path.join(homePath, `native-chat-headless-${Date.now()}.jsonl`) + writeFileSync(transcriptPath, claudeLine('headless-real', 'headless runtime bytes')) + + try { + const owned = await host.client.call<{ + messages: { id: string }[] + hasMore: boolean + }>('nativeChat.readSession', { + agent: 'claude', + sessionId: 'headless-session', + transcriptPath, + providerSession: { + key: 'session_id', + id: 'headless-session', + transcriptPath, + executionHostId: 'local' + }, + limit: 40 + }) + expect(owned.result).toMatchObject({ + messages: [{ id: 'headless-real' }], + hasMore: false + }) + + // A located session with unknown ownership must not inherit the server's filesystem. + const unowned = await host.client.call<{ + error: string + unverifiable: boolean + }>('nativeChat.readSession', { + agent: 'claude', + sessionId: 'headless-session', + transcriptPath, + providerSession: { + key: 'session_id', + id: 'headless-session', + transcriptPath + }, + limit: 40 + }) + expect(unowned.result).toEqual({ + error: 'Transcript unverifiable on the remote host', + unverifiable: true + }) + } finally { + rmSync(transcriptPath, { force: true }) + await host.dispose() + } +})