From 2990d274c9054060711addca4e3dd860ab269758 Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Sun, 25 Oct 2026 12:08:50 -0700 Subject: [PATCH] fix(agent-hooks): follow Claude sessions continued into a daemon job (adopts #26569) (#33605) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Panes follow a Claude Code conversation that moves into a background job (agent view ← then Enter, and /clear inside the job): on the presence recheck Orca reads the pane's own transcript `continued-in` record and the job's state.json (UUID-validated, sibling-folder only, overtaken pointers ignored, cwd kept) instead of trusting job hooks, so mobile and native chat stop freezing on the old session. Refs #24117 Supersedes #26569 Co-authored-by: mtsr404 <10844682+mtsr404@users.noreply.github.com> Co-authored-by: dimakrest <12894495+dimakrest@users.noreply.github.com> --- ...server-claude-session-continuation.test.ts | 162 +++++++++ .../server/server-agent-presence.ts | 35 ++ .../claude-session-continuation.test.ts | 333 ++++++++++++++++++ .../claude-session-continuation.ts | 312 ++++++++++++++++ ...server-claude-session-continuation.test.ts | 105 ++++++ src/wsl-guest/agent-hook-server.ts | 4 +- src/wsl-guest/relay-agent-presence.ts | 54 ++- 7 files changed, 994 insertions(+), 11 deletions(-) create mode 100644 src/main/agent-hooks/server-claude-session-continuation.test.ts create mode 100644 src/shared/agent-hook-listener/claude-session-continuation.test.ts create mode 100644 src/shared/agent-hook-listener/claude-session-continuation.ts create mode 100644 src/wsl-guest/agent-hook-server-claude-session-continuation.test.ts diff --git a/src/main/agent-hooks/server-claude-session-continuation.test.ts b/src/main/agent-hooks/server-claude-session-continuation.test.ts new file mode 100644 index 00000000000..89c6246be67 --- /dev/null +++ b/src/main/agent-hooks/server-claude-session-continuation.test.ts @@ -0,0 +1,162 @@ +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { AgentHookServer } from './server' +import type { probeAgentProcessPresence } from '../../shared/agent-process-presence-probe' +import { buildBody, GOOD_PANE, PANE, postHookEvent } from './server.test-fixtures' + +const probe = vi.hoisted(() => vi.fn(async () => 'live')) +vi.mock('../telemetry/client', () => ({ track: vi.fn() })) +vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: () => ({}) })) +vi.mock('../../shared/agent-process-presence-probe', () => ({ probeAgentProcessPresence: probe })) + +const OLD = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' +const FORK = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb' + +let root: string +let projectDir: string +const servers: AgentHookServer[] = [] + +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), 'orca-claude-continuation-')) + projectDir = join(root, 'claude', 'projects', '-work-repo') + mkdirSync(projectDir, { recursive: true }) +}) + +afterEach(() => { + for (const server of servers.splice(0)) { + server.stop() + } + rmSync(root, { recursive: true, force: true }) + probe.mockReset().mockResolvedValue('live') + vi.useRealTimers() +}) + +function transcript(sessionId: string, extra = ''): string { + const transcriptPath = join(projectDir, `${sessionId}.jsonl`) + writeFileSync( + transcriptPath, + `${JSON.stringify({ type: 'user', sessionId, uuid: 'u1', cwd: '/work/repo' })}\n${extra}` + ) + return transcriptPath +} + +function forkOldIntoJob(): { oldPath: string; forkPath: string } { + const pointer = { type: 'continued-in', sessionId: OLD, continuedInSessionId: FORK } + return { + oldPath: transcript(OLD, `${JSON.stringify(pointer)}\n`), + forkPath: transcript(FORK) + } +} + +async function createServer(): Promise { + const server = new AgentHookServer() + servers.push(server) + await server.start({ env: 'production', userDataPath: join(root, 'orca') }) + return server +} + +async function prompt(server: AgentHookServer, paneKey: string, sessionId: string, path: string) { + const [tabId] = paneKey.split(':') + const response = await postHookEvent( + server, + buildBody( + { + hook_event_name: 'UserPromptSubmit', + session_id: sessionId, + transcript_path: path, + prompt: 'keep going' + }, + { + paneKey, + tabId, + agentProcess: JSON.stringify({ pid: 4001, platform: process.platform, startTime: tabId }) + } + ) + ) + expect(response.status).toBe(204) +} + +function row(server: AgentHookServer, paneKey = PANE) { + return server.getStatusSnapshot().find((entry) => entry.paneKey === paneKey) +} + +describe('Claude session continued into a daemon job', () => { + it("rebinds the pane to the fork named by its own transcript, keeping the row's state", async () => { + const server = await createServer() + const { oldPath, forkPath } = forkOldIntoJob() + await prompt(server, PANE, OLD, oldPath) + const before = row(server) + expect(before?.providerSession).toMatchObject({ id: OLD, transcriptPath: oldPath }) + + await server.checkAgentPresence(PANE) + + const after = row(server) + expect(after?.providerSession).toEqual({ + key: 'session_id', + id: FORK, + transcriptPath: forkPath + }) + expect(after?.state).toBe(before?.state) + expect(after?.prompt).toBe(before?.prompt) + server.flushStatusPersistSync() + const persisted = JSON.parse( + readFileSync(join(root, 'orca', 'agent-hooks', 'last-status.json'), 'utf8') + ) + expect(persisted.entries[PANE].providerSession.id).toBe(FORK) + }) + + it('restamps the evidence clock for a suspension seen after the rebinding', async () => { + const t0 = 1_800_000_000_000 + vi.useFakeTimers({ toFake: ['Date'] }) + vi.setSystemTime(t0) + const server = await createServer() + const { oldPath } = forkOldIntoJob() + await prompt(server, PANE, OLD, oldPath) + const emitted = vi.fn() + server.setListener(emitted) + emitted.mockClear() + vi.setSystemTime(t0 + 60_000) + probe.mockImplementation(async (_identity, _read, _platform, observe) => { + observe?.(true) + return 'unverifiable' + }) + + await server.checkAgentPresence(PANE) + + expect(emitted.mock.calls[0]?.[0]).toMatchObject({ + providerSession: { id: FORK }, + evidenceObservedAt: t0 + }) + expect(emitted.mock.lastCall?.[0]).toMatchObject({ + providerSession: { id: FORK }, + agentPresence: { suspended: true }, + evidenceObservedAt: t0 + 60_000 + }) + }) + + it('leaves the pane alone while its transcript has no continuation', async () => { + const server = await createServer() + const oldPath = transcript(OLD) + transcript(FORK) + await prompt(server, PANE, OLD, oldPath) + + await server.checkAgentPresence(PANE) + + expect(row(server)?.providerSession?.id).toBe(OLD) + }) + + it('does not claim a fork that another pane already reports', async () => { + const server = await createServer() + const { oldPath, forkPath } = forkOldIntoJob() + await prompt(server, PANE, OLD, oldPath) + await prompt(server, GOOD_PANE, FORK, forkPath) + expect(row(server, GOOD_PANE)?.providerSession?.id).toBe(FORK) + + await server.checkAgentPresence(PANE) + + expect(row(server)?.providerSession?.id).toBe(OLD) + expect(row(server, GOOD_PANE)?.providerSession?.id).toBe(FORK) + }) +}) diff --git a/src/main/agent-hooks/server/server-agent-presence.ts b/src/main/agent-hooks/server/server-agent-presence.ts index 81655a793ce..5a277f76fa8 100644 --- a/src/main/agent-hooks/server/server-agent-presence.ts +++ b/src/main/agent-hooks/server/server-agent-presence.ts @@ -1,5 +1,9 @@ import { cliWriterTopology, type CliWriterObservation } from './cli-writer-observation' import type { AgentStatusIpcPayload } from '../../../shared/agent-status-types' +import { + ClaudeSessionContinuationTracker, + resolvePaneClaudeContinuation +} from '../../../shared/agent-hook-listener/claude-session-continuation' import type { AgentHookEventPayload } from '../../../shared/agent-hook-listener/listener-event' import { isSameAgentProcess, @@ -17,6 +21,7 @@ export abstract class AgentHookServerAgentPresence extends AgentHookServerLifecy promise: Promise } >() + private readonly claudeContinuations = new ClaudeSessionContinuationTracker() /** A hook proves its sender alive; only another process can challenge the owner. */ checkAgentPresenceAfterHook(event: AgentHookEventPayload, row: AgentHookEventPayload): void { @@ -96,6 +101,7 @@ export abstract class AgentHookServerAgentPresence extends AgentHookServerLifecy checkAgentPresence(paneKey: string): Promise { const resolved = this.resolvePaneKeyAlias(paneKey) + this.followClaudeSessionContinuation(resolved) const row = this.state.lastStatusByPaneKey.get(resolved) if (!row?.agentPresence?.process || row.agentPresence.ended) { return Promise.resolve(null) @@ -136,6 +142,7 @@ export abstract class AgentHookServerAgentPresence extends AgentHookServerLifecy const { suspended: _suspended, ...unsuspended } = presence this.applyNormalizedStatus({ ...row, + isReplay: undefined, hookEventName: undefined, hasExplicitPrompt: undefined, agentPresence: suspended ? { ...unsuspended, suspended: true } : unsuspended @@ -151,4 +158,32 @@ export abstract class AgentHookServerAgentPresence extends AgentHookServerLifecy this.presenceChecks.set(row, { token, statusHookGeneration, promise: check }) return check } + + /** A Claude session forked into a daemon job keeps running in this pane, but #15304 declines + * the job's hooks; the pane's own transcript names the fork, so rebind the row to it. */ + private followClaudeSessionContinuation(paneKey: string): void { + const row = this.state.lastStatusByPaneKey.get(paneKey) + if (!row || row.connectionId !== null) { + return + } + const next = resolvePaneClaudeContinuation( + this.claudeContinuations, + this.state.lastStatusByPaneKey, + paneKey + ) + if (next) { + this.applyNormalizedStatus( + { + ...row, + providerSession: next, + // Why: only the identity moved; the state is unchanged, so keep its evidence clock. + isReplay: true, + hookEventName: undefined, + hasExplicitPrompt: undefined + }, + undefined, + 'process' + ) + } + } } diff --git a/src/shared/agent-hook-listener/claude-session-continuation.test.ts b/src/shared/agent-hook-listener/claude-session-continuation.test.ts new file mode 100644 index 00000000000..81f27cf75f5 --- /dev/null +++ b/src/shared/agent-hook-listener/claude-session-continuation.test.ts @@ -0,0 +1,333 @@ +import { appendFileSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { ClaudeSessionContinuationTracker } from './claude-session-continuation' +import { scanFileRegionsBackward } from './reverse-file-region-scan' +import type * as ReverseFileRegionScan from './reverse-file-region-scan' + +vi.mock('./reverse-file-region-scan', async (importOriginal) => { + const actual = await importOriginal() + return { scanFileRegionsBackward: vi.fn(actual.scanFileRegionsBackward) } +}) + +const OLD = '11111111-1111-4111-8111-111111111111' +const FORK = '22222222-2222-4222-8222-222222222222' +const SECOND_FORK = '33333333-3333-4333-8333-333333333333' +const CLEARED = '44444444-4444-4444-8444-444444444444' +const FILE_ID = '55555555-5555-4555-8555-555555555555' + +const createdDirs: string[] = [] + +afterEach(() => { + vi.mocked(scanFileRegionsBackward).mockClear() + for (const dir of createdDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }) + } +}) + +function makeProject(): { configDir: string; projectDir: string } { + const configDir = mkdtempSync(join(tmpdir(), 'claude-continuation-')) + createdDirs.push(configDir) + const projectDir = join(configDir, 'projects', '-work-repo') + mkdirSync(projectDir, { recursive: true }) + return { configDir, projectDir } +} + +function line(record: Record): string { + return `${JSON.stringify(record)}\n` +} + +function turn(sessionId: string, uuid: string): string { + return line({ type: 'user', sessionId, uuid, cwd: '/work/repo', message: { content: 'hi' } }) +} + +function pointer(sessionId: string, continuedInSessionId: string): string { + return line({ type: 'continued-in', sessionId, continuedInSessionId }) +} + +function writeTranscript(projectDir: string, name: string, body: string): string { + const transcriptPath = join(projectDir, `${name}.jsonl`) + writeFileSync(transcriptPath, body) + return transcriptPath +} + +function writeJobState(configDir: string, state: Record): void { + const jobDir = join(configDir, 'jobs', String(state.sessionId).slice(0, 8)) + mkdirSync(jobDir, { recursive: true }) + writeFileSync(join(jobDir, 'state.json'), JSON.stringify(state)) +} + +function session(id: string, transcriptPath: string, cwd?: string) { + return { key: 'session_id' as const, id, transcriptPath, ...(cwd ? { cwd } : {}) } +} + +describe('ClaudeSessionContinuationTracker', () => { + it('follows a continued-in pointer that is not the last line', () => { + const { projectDir } = makeProject() + const oldPath = writeTranscript( + projectDir, + OLD, + turn(OLD, 'u1') + pointer(OLD, FORK) + line({ type: 'cost-state', sessionId: OLD }) + ) + const forkPath = writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toEqual( + session(FORK, forkPath) + ) + }) + + it('returns null when the transcript has no pointer out of this session', () => { + const { projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(SECOND_FORK, FORK)) + writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toBeNull() + }) + + it('does not rebind to a file that never names the forked session', () => { + const { projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(OLD, FORK)) + writeTranscript(projectDir, FORK, turn(SECOND_FORK, 'u1')) + + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toBeNull() + }) + + it('retries a pointer whose fork reaches disk after the first check, without rescanning', () => { + const { projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(OLD, FORK)) + const tracker = new ClaudeSessionContinuationTracker() + expect(tracker.resolve(session(OLD, oldPath))).toBeNull() + expect(tracker.resolve(session(OLD, oldPath))).toBeNull() + + const forkPath = writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + expect(tracker.resolve(session(OLD, oldPath))).toEqual(session(FORK, forkPath)) + const oldScans = vi.mocked(scanFileRegionsBackward).mock.calls.filter(([p]) => p === oldPath) + expect(oldScans).toHaveLength(1) + }) + + it('keeps the folder the session ran in when following a fork or a /clear', () => { + const { configDir, projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(OLD, FORK)) + const forkPath = writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + const tracker = new ClaudeSessionContinuationTracker() + expect(tracker.resolve(session(OLD, oldPath, '/work/repo/pkg'))).toEqual( + session(FORK, forkPath, '/work/repo/pkg') + ) + + const clearedPath = writeTranscript(projectDir, CLEARED, turn(CLEARED, 'u2')) + writeJobState(configDir, { + sessionId: FORK, + resumeSessionId: CLEARED, + linkScanPath: clearedPath + }) + expect(tracker.resolve(session(FORK, forkPath, '/work/repo/pkg'))).toEqual( + session(CLEARED, clearedPath, '/work/repo/pkg') + ) + }) + + it.each([ + ['a later prompt', turn(OLD, 'u2')], + [ + 'a later finished reply', + line({ type: 'assistant', sessionId: OLD, message: { stop_reason: 'end_turn' } }) + ] + ])('ignores a pointer the old session overtook with %s', (_label, later) => { + const { projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(OLD, FORK) + later) + writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toBeNull() + }) + + it('still follows a pointer after a tool result, an errored reply, a command or an interrupt', () => { + const { projectDir } = makeProject() + const userText = (content: unknown) => + line({ type: 'user', sessionId: OLD, message: { content } }) + const oldPath = writeTranscript( + projectDir, + OLD, + turn(OLD, 'u1') + + pointer(OLD, FORK) + + userText('/clear') + + userText([{ type: 'text', text: '[Request interrupted by user]' }]) + + userText([{ type: 'text', text: 'x' }]) + + line({ type: 'user', sessionId: OLD, message: { content: [{ type: 'tool_result' }] } }) + + line({ + type: 'assistant', + sessionId: OLD, + isApiErrorMessage: true, + message: { stop_reason: 'stop_sequence' } + }) + ) + const forkPath = writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toEqual( + session(FORK, forkPath) + ) + }) + + it('follows a fork whose first record is larger than the head it reads', () => { + const { projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(OLD, FORK)) + const huge = line({ type: 'user', sessionId: FORK, message: { content: 'x'.repeat(300_000) } }) + const forkPath = writeTranscript(projectDir, FORK, huge) + + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toEqual( + session(FORK, forkPath) + ) + }) + + it('rejects ids and job paths that are not Claude session ids', () => { + const { configDir, projectDir } = makeProject() + const escaped = '../../escaped' + writeTranscript(projectDir, escaped, turn(escaped, 'u1')) + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(OLD, escaped)) + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toBeNull() + + const forkPath = writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + writeJobState(configDir, { + sessionId: FORK, + resumeSessionId: escaped, + linkScanPath: join(projectDir, `${escaped}.jsonl`) + }) + expect(new ClaudeSessionContinuationTracker().resolve(session(FORK, forkPath))).toBeNull() + + writeJobState(configDir, { + sessionId: FORK, + resumeSessionId: CLEARED, + linkScanPath: writeTranscript(projectDir, 'not-a-session', turn(CLEARED, 'u2')) + }) + expect(new ClaudeSessionContinuationTracker().resolve(session(FORK, forkPath))).toBeNull() + }) + + it('finds a pointer appended after a scan that found none', () => { + const { projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1')) + const forkPath = writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + const tracker = new ClaudeSessionContinuationTracker() + expect(tracker.resolve(session(OLD, oldPath))).toBeNull() + + appendFileSync(oldPath, pointer(OLD, FORK)) + expect(tracker.resolve(session(OLD, oldPath))).toEqual(session(FORK, forkPath)) + }) + + it('reads only the tail of a transcript larger than the first scan', () => { + const { projectDir } = makeProject() + const filler = turn(OLD, 'filler').repeat(Math.ceil((3 * 1024 * 1024) / 90)) + const oldPath = writeTranscript(projectDir, OLD, filler + pointer(OLD, FORK)) + const forkPath = writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toEqual( + session(FORK, forkPath) + ) + }) + + it("uses the job's state when the fork is not named after its session", () => { + const { configDir, projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(OLD, FORK)) + const forkPath = writeTranscript(projectDir, FILE_ID, turn(FORK, 'u1')) + const jobDir = join(configDir, 'jobs', FORK.slice(0, 8)) + mkdirSync(jobDir, { recursive: true }) + writeFileSync( + join(jobDir, 'state.json'), + JSON.stringify({ sessionId: FORK, linkScanPath: forkPath }) + ) + + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toEqual( + session(FORK, forkPath) + ) + }) + + it("ignores a job state that points outside the old transcript's project", () => { + const { configDir, projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(OLD, FORK)) + const elsewhere = join(configDir, 'projects', '-other-repo') + mkdirSync(elsewhere, { recursive: true }) + const forkPath = writeTranscript(elsewhere, 'transcript-file-uuid', turn(FORK, 'u1')) + const jobDir = join(configDir, 'jobs', FORK.slice(0, 8)) + mkdirSync(jobDir, { recursive: true }) + writeFileSync( + join(jobDir, 'state.json'), + JSON.stringify({ sessionId: FORK, linkScanPath: forkPath }) + ) + + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toBeNull() + }) + + it('follows a chain of forks and stops on a cycle', () => { + const { projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(OLD, FORK)) + writeTranscript(projectDir, FORK, turn(FORK, 'u1') + pointer(FORK, SECOND_FORK)) + const secondPath = writeTranscript( + projectDir, + SECOND_FORK, + turn(SECOND_FORK, 'u1') + pointer(SECOND_FORK, OLD) + ) + + expect(new ClaudeSessionContinuationTracker().resolve(session(OLD, oldPath))).toEqual( + session(SECOND_FORK, secondPath) + ) + }) + + it('follows /clear inside a job through the job state, after the transcripts stop growing', () => { + const { configDir, projectDir } = makeProject() + const oldPath = writeTranscript(projectDir, OLD, turn(OLD, 'u1') + pointer(OLD, FORK)) + const forkPath = writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + writeJobState(configDir, { sessionId: FORK, resumeSessionId: FORK, linkScanPath: forkPath }) + const tracker = new ClaudeSessionContinuationTracker() + expect(tracker.resolve(session(OLD, oldPath))).toEqual(session(FORK, forkPath)) + expect(tracker.resolve(session(FORK, forkPath))).toBeNull() + + const clearedPath = writeTranscript(projectDir, CLEARED, turn(CLEARED, 'u2')) + writeJobState(configDir, { + sessionId: FORK, + resumeSessionId: CLEARED, + linkScanPath: clearedPath + }) + expect(tracker.resolve(session(FORK, forkPath))).toEqual(session(CLEARED, clearedPath)) + expect(tracker.resolve(session(OLD, oldPath))).toEqual(session(CLEARED, clearedPath)) + + // A second `/clear`: the job state stays under FORK, so the row on CLEARED must still find it. + const reclearedPath = writeTranscript(projectDir, SECOND_FORK, turn(SECOND_FORK, 'u3')) + writeJobState(configDir, { + sessionId: FORK, + resumeSessionId: SECOND_FORK, + linkScanPath: reclearedPath + }) + expect(tracker.resolve(session(CLEARED, clearedPath))).toEqual( + session(SECOND_FORK, reclearedPath) + ) + }) + + it.each([ + [ + 'names the same session', + (_dir: string, forkPath: string) => ({ resumeSessionId: FORK, linkScanPath: forkPath }) + ], + [ + 'points outside the project', + (configDir: string) => { + const elsewhere = join(configDir, 'projects', '-other-repo') + mkdirSync(elsewhere, { recursive: true }) + return { + resumeSessionId: CLEARED, + linkScanPath: writeTranscript(elsewhere, CLEARED, turn(CLEARED, 'u2')) + } + } + ], + [ + 'points at a file that never names the session', + (_dir: string, forkPath: string) => ({ + resumeSessionId: CLEARED, + linkScanPath: writeTranscript(join(forkPath, '..'), CLEARED, turn(SECOND_FORK, 'u2')) + }) + ] + ])('ignores a job state that %s', (_label, makeState) => { + const { configDir, projectDir } = makeProject() + const forkPath = writeTranscript(projectDir, FORK, turn(FORK, 'u1')) + writeJobState(configDir, { sessionId: FORK, ...makeState(configDir, forkPath) }) + + expect(new ClaudeSessionContinuationTracker().resolve(session(FORK, forkPath))).toBeNull() + }) +}) diff --git a/src/shared/agent-hook-listener/claude-session-continuation.ts b/src/shared/agent-hook-listener/claude-session-continuation.ts new file mode 100644 index 00000000000..ad97417730f --- /dev/null +++ b/src/shared/agent-hook-listener/claude-session-continuation.ts @@ -0,0 +1,312 @@ +import { closeSync, openSync, readFileSync, readSync, statSync } from 'node:fs' +import path from 'node:path' +import { + normalizeAgentProviderSession, + type AgentProviderSessionMetadata +} from '../agent-session-resume' +import type { AgentHookEventPayload } from './listener-event' +import { scanFileRegionsBackward } from './reverse-file-region-scan' +import { TRANSCRIPT_CHUNK_BYTES } from './transcript-reader' + +const FIRST_SCAN_BYTES = 1024 * 1024 +// Why: a rescan reads only what was appended, plus enough to re-read a line split at the old end. +const RESCAN_OVERLAP_BYTES = 64 * 1024 +const FORK_HEAD_SCAN_BYTES = 256 * 1024 +const MAX_CONTINUATION_HOPS = 8 +const MAX_REMEMBERED_ENTRIES = 256 +// Claude Code accepts only this shape for continuedInSessionId; it also keeps ids out of path joins. +// Claude Code skips tagged text (commands, reminders) and interrupt markers when finding prompts. +const NOT_A_PROMPT = /^\s*<[a-z][\w-]*[\s>]|^\[Request interrupted by user/ +const CLAUDE_SESSION_ID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i + +/** + * Follows the `continued-in` record Claude Code writes when it forks a pane's session into a + * daemon job. That job's hooks are declined (#15304), so the pane's own transcript is the only + * evidence of where the conversation went. + */ +export class ClaudeSessionContinuationTracker { + /** Last scan per transcript, so an unchanged file is not re-read while its fork is pending. */ + private readonly scanByTranscript = new Map() + /** A session `/clear` started inside a job → that job's first session, which names its state dir. */ + private readonly jobSessionBySession = new Map() + + resolve(session: AgentProviderSessionMetadata): AgentProviderSessionMetadata | null { + const visited = new Set([session.id]) + let current = session + for (let hop = 0; hop < MAX_CONTINUATION_HOPS; hop++) { + const next = this.findNextSession(current) + if (!next || visited.has(next.id)) { + break + } + visited.add(next.id) + current = next + } + return current === session ? null : current + } + + private findNextSession( + session: AgentProviderSessionMetadata + ): AgentProviderSessionMetadata | null { + const transcriptPath = session.transcriptPath + if (session.key !== 'session_id' || !transcriptPath?.endsWith('.jsonl')) { + return null + } + // Why before the size memo: after `/clear` the old transcript stops growing, so only the job's + // state says where the conversation went. + const jobSessionId = this.jobSessionBySession.get(session.id) ?? session.id + const cleared = resolveClearedJobSession(transcriptPath, jobSessionId, session) + if (cleared) { + remember(this.jobSessionBySession, cleared.id, jobSessionId) + return cleared + } + let size: number + try { + size = statSync(transcriptPath).size + } catch { + return null + } + const scan = this.scanByTranscript.get(transcriptPath) + let pointer = scan?.pointer + if (scan?.size !== size) { + const isRescan = scan !== undefined && scan.size < size + const found = readContinuationPointer( + transcriptPath, + session.id, + isRescan ? size - scan.size + RESCAN_OVERLAP_BYTES : FIRST_SCAN_BYTES + ) + // Why keep the old pointer on an empty rescan: only the appended bytes were read. + pointer = found === undefined && isRescan ? scan.pointer : (found ?? undefined) + remember(this.scanByTranscript, transcriptPath, { size, pointer }) + } + // Why retry a remembered pointer: its fork may reach disk after the old file stops growing. + return pointer ? resolveForkedSession(transcriptPath, pointer, session.cwd) : null + } +} + +function remember(map: Map, key: string, value: V): void { + map.delete(key) + map.set(key, value) + while (map.size > MAX_REMEMBERED_ENTRIES) { + const oldest = map.keys().next().value + if (oldest === undefined) { + break + } + map.delete(oldest) + } +} + +/** The fork a pane's Claude row should now point at. A fork another pane reports belongs to that + * pane, whose own hooks keep it current. */ +export function resolvePaneClaudeContinuation( + tracker: ClaudeSessionContinuationTracker, + rows: ReadonlyMap, + paneKey: string +): AgentProviderSessionMetadata | null { + const row = rows.get(paneKey) + if ( + !row?.providerSession || + row.payload.agentType !== 'claude' || + row.providerSessionOnly || + row.structuredHost + ) { + return null + } + const next = tracker.resolve(row.providerSession) + if (!next) { + return null + } + for (const [otherPaneKey, other] of rows) { + if (otherPaneKey !== paneKey && other.providerSession?.id === next.id) { + return null + } + } + return next +} + +function parseRecord(line: string): Record | undefined { + try { + const parsed: unknown = JSON.parse(line) + return typeof parsed === 'object' && parsed !== null && !Array.isArray(parsed) + ? Object.fromEntries(Object.entries(parsed)) + : undefined + } catch { + return undefined + } +} + +/** The newest pointer out of `sessionId`, or null when a later turn continued this session + * instead (close to Claude Code's own reader); undefined when the scan saw neither. */ +function readContinuationPointer( + transcriptPath: string, + sessionId: string, + maxScanBytes: number +): string | null | undefined { + return scanFileRegionsBackward<{ pointer: string | null }>( + transcriptPath, + { chunkBytes: TRANSCRIPT_CHUNK_BYTES, maxScanBytes }, + (region) => { + const lines = region.toString('utf8').split('\n') + for (let index = lines.length - 1; index >= 0; index--) { + const text = lines[index] + if ( + !text.includes('"continued-in"') && + !text.includes('"type":"user"') && + !text.includes('"type":"assistant"') + ) { + continue + } + const record = parseRecord(text.trim()) + if ( + record?.type === 'continued-in' && + record.sessionId === sessionId && + isClaudeSessionId(record.continuedInSessionId) + ) { + return { pointer: record.continuedInSessionId } + } + if (record && isCompletedTurn(record)) { + return { pointer: null } + } + } + return undefined + } + )?.pointer +} + +/** A real prompt or a finished, non-error reply: the old session went on after any older pointer. */ +function isCompletedTurn(record: Record): boolean { + const message = record.message + const fields = + typeof message === 'object' && message !== null && !Array.isArray(message) + ? Object.fromEntries(Object.entries(message)) + : {} + if (record.type === 'assistant') { + return record.isApiErrorMessage !== true && typeof fields.stop_reason === 'string' + } + if (record.type !== 'user' || record.isMeta === true || record.isCompactSummary === true) { + return false + } + const content: unknown = fields.content + const texts: unknown[] = typeof content === 'string' ? [content] : [] + if (Array.isArray(content)) { + if (content.some((block) => block?.type === 'tool_result')) { + return false + } + texts.push(...content.filter((block) => block?.type === 'text').map((block) => block.text)) + } + return texts.some( + (text) => typeof text === 'string' && text.trim() !== '' && !NOT_A_PROMPT.test(text) + ) +} + +function isClaudeSessionId(value: unknown): value is string { + return typeof value === 'string' && CLAUDE_SESSION_ID.test(value) +} + +function resolveForkedSession( + transcriptPath: string, + forkedSessionId: string, + cwd: string | undefined +): AgentProviderSessionMetadata | null { + const forkedPath = findForkedTranscript(transcriptPath, forkedSessionId) + if (!forkedPath || !forkHeadNamesSession(forkedPath, forkedSessionId)) { + return null + } + return normalizeAgentProviderSession({ + key: 'session_id', + id: forkedSessionId, + transcriptPath: forkedPath, + cwd + }) +} + +/** The fork lives beside the old transcript; the job's state names it when the file is not `.jsonl`. */ +function findForkedTranscript(transcriptPath: string, sessionId: string): string | undefined { + const byId = path.join(path.dirname(transcriptPath), `${sessionId}.jsonl`) + if (isFile(byId)) { + return byId + } + const state = readJobState(transcriptPath, sessionId) + return state && isSiblingFile(transcriptPath, state.linkScanPath) ? state.linkScanPath : undefined +} + +/** `/clear` inside a job starts a new transcript without writing `continued-in`; the job's state + * keeps its first session id and names the current one as `resumeSessionId`. */ +function resolveClearedJobSession( + transcriptPath: string, + jobSessionId: string, + session: AgentProviderSessionMetadata +): AgentProviderSessionMetadata | null { + const state = readJobState(transcriptPath, jobSessionId) + if ( + !state || + !isClaudeSessionId(state.resumeSessionId) || + state.resumeSessionId === session.id || + !isSiblingFile(transcriptPath, state.linkScanPath) || + !forkHeadNamesSession(state.linkScanPath, state.resumeSessionId) + ) { + return null + } + return normalizeAgentProviderSession({ + key: 'session_id', + id: state.resumeSessionId, + transcriptPath: state.linkScanPath, + cwd: session.cwd + }) +} + +type JobState = { linkScanPath: unknown; resumeSessionId: unknown } + +/** /projects//.jsonl → /jobs//state.json, when it is this session's job. */ +function readJobState(transcriptPath: string, sessionId: string): JobState | undefined { + if (!isClaudeSessionId(sessionId)) { + return undefined + } + const configDir = path.dirname(path.dirname(path.dirname(transcriptPath))) + const statePath = path.join(configDir, 'jobs', sessionId.slice(0, 8), 'state.json') + let state: Record | undefined + try { + state = parseRecord(readFileSync(statePath, 'utf8')) + } catch { + return undefined + } + return state?.sessionId === sessionId + ? { linkScanPath: state.linkScanPath, resumeSessionId: state.resumeSessionId } + : undefined +} + +function isSiblingFile(transcriptPath: string, candidate: unknown): candidate is string { + return ( + typeof candidate === 'string' && + path.dirname(candidate) === path.dirname(transcriptPath) && + isClaudeSessionId(path.basename(candidate, '.jsonl')) && + path.extname(candidate) === '.jsonl' && + isFile(candidate) + ) +} + +function isFile(filePath: string): boolean { + try { + return statSync(filePath).isFile() + } catch { + return false + } +} + +/** The fork's own records must carry the new id; a stray file under that name proves nothing. */ +function forkHeadNamesSession(forkedPath: string, sessionId: string): boolean { + let head: string + try { + const fd = openSync(forkedPath, 'r') + try { + const buffer = Buffer.alloc(FORK_HEAD_SCAN_BYTES) + const read = readSync(fd, buffer, 0, FORK_HEAD_SCAN_BYTES, 0) + head = buffer.subarray(0, read).toString('utf8') + } finally { + closeSync(fd) + } + } catch { + return false + } + // Why a substring: a huge first record would not parse inside the head window. + return head.includes(`"sessionId":"${sessionId}"`) +} diff --git a/src/wsl-guest/agent-hook-server-claude-session-continuation.test.ts b/src/wsl-guest/agent-hook-server-claude-session-continuation.test.ts new file mode 100644 index 00000000000..eee53db57bc --- /dev/null +++ b/src/wsl-guest/agent-hook-server-claude-session-continuation.test.ts @@ -0,0 +1,105 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { RelayAgentHookServer } from './agent-hook-server' +import type { AgentHookRelayEnvelope } from '../shared/agent-hook-relay' +import type { probeAgentProcessPresence } from '../shared/agent-process-presence-probe' +import { makePaneKey } from '../shared/stable-pane-id' + +const probe = vi.hoisted(() => vi.fn(async () => 'live')) +vi.mock('../shared/agent-process-presence-probe', () => ({ probeAgentProcessPresence: probe })) + +const PANE_KEY = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111') +const OLD = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' +const FORK = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb' + +function line(record: unknown): string { + return `${JSON.stringify(record)}\n` +} + +describe('RelayAgentHookServer Claude session continuation', () => { + const dirs: string[] = [] + const servers: RelayAgentHookServer[] = [] + + afterEach(() => { + for (const server of servers.splice(0)) { + server.stop() + } + for (const dir of dirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }) + } + probe.mockReset().mockResolvedValue('live') + }) + + async function promptForkedPane() { + const dir = mkdtempSync(join(tmpdir(), 'relay-hook-claude-continuation-')) + dirs.push(dir) + const projectDir = join(dir, 'claude', 'projects', '-work-repo') + mkdirSync(projectDir, { recursive: true }) + const oldPath = join(projectDir, `${OLD}.jsonl`) + const forkPath = join(projectDir, `${FORK}.jsonl`) + writeFileSync( + oldPath, + line({ type: 'user', sessionId: OLD, uuid: 'u1' }) + + line({ type: 'continued-in', sessionId: OLD, continuedInSessionId: FORK }) + ) + writeFileSync(forkPath, line({ type: 'user', sessionId: FORK, uuid: 'u1' })) + const forward = vi.fn<(envelope: AgentHookRelayEnvelope) => void>() + const server = new RelayAgentHookServer({ endpointDir: dir, forward }) + servers.push(server) + await server.start() + const { port, token } = server.getCoordinates() + const response = await fetch(`http://127.0.0.1:${port}/hook/claude`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Orca-Agent-Hook-Token': token }, + body: JSON.stringify({ + paneKey: PANE_KEY, + tabId: 'tab-1', + worktreeId: 'wt-1', + agentProcess: { pid: 4001, platform: 'linux', startTime: 'birth' }, + payload: { + hook_event_name: 'UserPromptSubmit', + session_id: OLD, + transcript_path: oldPath, + prompt: 'keep going' + } + }) + }) + expect(response.status).toBe(204) + const prompted = forward.mock.calls.at(-1)?.[0] + expect(prompted?.providerSession?.id).toBe(OLD) + return { server, forward, forkPath, prompted } + } + + it("forwards the pane's rebinding to the fork its transcript names", async () => { + const { server, forward, forkPath, prompted } = await promptForkedPane() + + await server.checkAgentPresence(PANE_KEY) + + const rebound = forward.mock.calls.at(-1)?.[0] + expect(rebound?.providerSession).toEqual({ + key: 'session_id', + id: FORK, + transcriptPath: forkPath + }) + expect(rebound?.isReplay).toBe(true) + expect(rebound?.payload.state).toBe(prompted?.payload.state) + }) + + it('forwards a suspension seen after the rebinding as fresh evidence', async () => { + const { server, forward } = await promptForkedPane() + probe.mockImplementation(async (_identity, _read, _platform, observe) => { + observe?.(true) + return 'unverifiable' + }) + + await server.checkAgentPresence(PANE_KEY) + + const suspended = forward.mock.calls.at(-1)?.[0] + expect(suspended?.providerSession?.id).toBe(FORK) + expect(suspended?.agentPresence?.suspended).toBe(true) + expect(suspended?.isReplay).toBeUndefined() + }) +}) diff --git a/src/wsl-guest/agent-hook-server.ts b/src/wsl-guest/agent-hook-server.ts index 78fc1199882..f928f225f88 100644 --- a/src/wsl-guest/agent-hook-server.ts +++ b/src/wsl-guest/agent-hook-server.ts @@ -71,7 +71,7 @@ export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus { private fixedToken: string | undefined private preferredPort: number private portFallbackApplied = false - private readonly presenceChecks = new RelayAgentPresence() + private readonly presenceChecks = new RelayAgentPresence(this.state) private retryScheduler: AgentHookResultRetryScheduler readonly claudeTerminalInterrupts = createRelayClaudeTerminalInterrupts(this.state, () => this.relayInterruptHost() @@ -243,7 +243,7 @@ export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus { () => this.state.lastStatusByPaneKey.get(paneKey), (event) => { if (meta) { - this.applyEvent(event, meta.source, meta.env, meta.version) + this.applyEvent(event, meta.source, meta.env, meta.version, { isReplay: event.isReplay }) } } ) diff --git a/src/wsl-guest/relay-agent-presence.ts b/src/wsl-guest/relay-agent-presence.ts index 1ee618e4e37..cebaffd9d93 100644 --- a/src/wsl-guest/relay-agent-presence.ts +++ b/src/wsl-guest/relay-agent-presence.ts @@ -1,45 +1,81 @@ +import { + ClaudeSessionContinuationTracker, + resolvePaneClaudeContinuation +} from '../shared/agent-hook-listener/claude-session-continuation' import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event' +import type { HookListenerState } from '../shared/agent-hook-listener/listener-state' import { probeAgentProcessPresence } from '../shared/agent-process-presence-probe' export class RelayAgentPresence { private readonly pending = new WeakMap>() + private readonly claudeContinuations = new ClaudeSessionContinuationTracker() + + constructor(private readonly state: Pick) {} check( row: AgentHookEventPayload | undefined, current: () => AgentHookEventPayload | undefined, publish: (event: AgentHookEventPayload) => void ): Promise { - if (!row?.agentPresence?.process || row.agentPresence.ended) { + if (row) { + this.followClaudeSessionContinuation(row, publish) + } + // Why re-read: a rebinding above replaced the cached row the probe must compare against. + const owner = row ? current() : undefined + if (!owner?.agentPresence?.process || owner.agentPresence.ended) { return Promise.resolve() } - const existing = this.pending.get(row) + const existing = this.pending.get(owner) if (existing) { return existing } - const presence = row.agentPresence + const presence = owner.agentPresence let suspended: boolean | undefined const check = probeAgentProcessPresence(presence.process, undefined, undefined, (observed) => { suspended = observed }) .then((verdict) => { - if (verdict === 'exited' && current() === row) { + if (verdict === 'exited' && current() === owner) { publish({ - ...row, + ...owner, + isReplay: undefined, hookEventName: 'AgentProcessExit', agentPresence: { ...presence, ended: true } }) - } else if (current() === row && (suspended === true) !== (presence.suspended === true)) { + } else if (current() === owner && (suspended === true) !== (presence.suspended === true)) { const { suspended: _suspended, ...unsuspended } = presence publish({ - ...row, + ...owner, + isReplay: undefined, hookEventName: undefined, hasExplicitPrompt: undefined, agentPresence: suspended ? { ...unsuspended, suspended: true } : unsuspended }) } }) - .finally(() => this.pending.delete(row)) - this.pending.set(row, check) + .finally(() => this.pending.delete(owner)) + this.pending.set(owner, check) return check } + + /** Same rebinding as the client's hook server (#24117); the relay owns this host's transcripts. */ + private followClaudeSessionContinuation( + row: AgentHookEventPayload, + publish: (event: AgentHookEventPayload) => void + ): void { + const next = resolvePaneClaudeContinuation( + this.claudeContinuations, + this.state.lastStatusByPaneKey, + row.paneKey + ) + if (next) { + publish({ + ...row, + providerSession: next, + isReplay: true, + hookEventName: undefined, + hasExplicitPrompt: undefined + }) + } + } }