mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
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>
This commit is contained in:
co-authored by
mtsr404
dimakrest
parent
2f25699094
commit
2990d274c9
@@ -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<typeof probeAgentProcessPresence>(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<AgentHookServer> {
|
||||
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)
|
||||
})
|
||||
})
|
||||
@@ -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<AgentProcessVerdict | null>
|
||||
}
|
||||
>()
|
||||
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<AgentProcessVerdict | null> {
|
||||
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'
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<typeof ReverseFileRegionScan>()
|
||||
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, unknown>): 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<string, unknown>): 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('<command-name>/clear</command-name>') +
|
||||
userText([{ type: 'text', text: '[Request interrupted by user]' }]) +
|
||||
userText([{ type: 'text', text: '<system-reminder>x</system-reminder>' }]) +
|
||||
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()
|
||||
})
|
||||
})
|
||||
@@ -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<string, { size: number; pointer?: string }>()
|
||||
/** A session `/clear` started inside a job → that job's first session, which names its state dir. */
|
||||
private readonly jobSessionBySession = new Map<string, string>()
|
||||
|
||||
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<V>(map: Map<string, V>, 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<string, AgentHookEventPayload>,
|
||||
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<string, unknown> | 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<string, unknown>): 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 `<id>.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 }
|
||||
|
||||
/** <config>/projects/<cwd slug>/<id>.jsonl → <config>/jobs/<first 8 of id>/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<string, unknown> | 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}"`)
|
||||
}
|
||||
@@ -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<typeof probeAgentProcessPresence>(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()
|
||||
})
|
||||
})
|
||||
@@ -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 })
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
@@ -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<AgentHookEventPayload, Promise<void>>()
|
||||
private readonly claudeContinuations = new ClaudeSessionContinuationTracker()
|
||||
|
||||
constructor(private readonly state: Pick<HookListenerState, 'lastStatusByPaneKey'>) {}
|
||||
|
||||
check(
|
||||
row: AgentHookEventPayload | undefined,
|
||||
current: () => AgentHookEventPayload | undefined,
|
||||
publish: (event: AgentHookEventPayload) => void
|
||||
): Promise<void> {
|
||||
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
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user