Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions config/scripts/vitest-sqlite-runtime-files.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ export const SQLITE_RUNTIME_INCLUDE = [
'src/main/acp/acp-structured-prompt-failure.test.ts',
'src/main/acp/acp-timeline-turn-failures.test.ts',
'src/main/acp/acp-structured-session-adapter.test.ts',
'src/main/acp/acp-conversation-copy.test.ts',
'src/main/pi/rpc-turn-races.test.ts',
'src/main/acp/acp-structured-child-session-end.test.ts',
'src/main/acp/acp-structured-child-stop.test.ts',
Expand Down
37 changes: 37 additions & 0 deletions src/main/acp/acp-conversation-copies.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
import { randomUUID } from 'node:crypto'
import type { StructuredAgentSessionAdapter } from '../native-chat/agent-session-wire/structured-agent-session-adapter'
import { ProviderAcquisitionStarts } from '../provider-process/provider-acquisition-starts'
import { copyAcpConversation } from './acp-conversation-copy'
import type { AcpStructuredConnection } from './acp-structured-connection'
import type { AcpStructuredSessionAdapterDeps } from './acp-structured-session-adapter-deps'

/** Copy children have their own cleanup lifecycle; they never gate the original chat's start. */
export class AcpConversationCopies {
private readonly children = new ProviderAcquisitionStarts<AcpStructuredConnection>()
private readonly shutdown = new AbortController()
private readonly pending = new Set<Promise<unknown>>()

constructor(private readonly deps: AcpStructuredSessionAdapterDeps) {}

copy: NonNullable<StructuredAgentSessionAdapter['copyConversation']> = async (input) => {
const copy = copyAcpConversation(this.deps, input, {
signal: this.shutdown.signal,
track: (connection) => this.children.retainFailed(randomUUID(), connection)
})
this.pending.add(copy)
try {
return await copy
} finally {
this.pending.delete(copy)
}
}

async closeAll(): Promise<boolean> {
this.shutdown.abort()
await Promise.allSettled(this.pending)
const closed = await Promise.all(
this.children.failedSessionIds().map((id) => this.children.stopFailed(id))
)
return closed.every(Boolean)
}
}
137 changes: 137 additions & 0 deletions src/main/acp/acp-conversation-copy.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,137 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { acpLaunchSpecFor } from './acp-launch-specs'
import { copyAcpConversation } from './acp-conversation-copy'
import {
FakeAcpChild,
openAcpAdapterRig,
type AcpAdapterRig,
PROVIDER_SESSION
} from './acp-structured-adapter.test-support'

vi.mock('../native-chat/session-file-resolver', () => ({
resolveSessionFilePath: async () => '/account-a/sessions/copy-1.jsonl'
}))
const FORKS = { agentCapabilities: { loadSession: true, sessionCapabilities: { fork: {} } } }
const input = {
identity: {
sessionId: 'orca-chat-1',
workspaceId: 'workspace-1',
hostId: 'local',
agent: 'omp',
providerHandle: null
},
providerSessionId: PROVIDER_SESSION,
accountHome: { variable: 'PI_CODING_AGENT_DIR', path: '/account-a' }
}
const rigs: AcpAdapterRig[] = []
afterEach(async () => {
await Promise.all(rigs.splice(0).map((rig) => rig.adapter.closeAll()))
})

async function open(agent: 'omp' | 'opencode' = 'omp') {
const rig = await openAcpAdapterRig({
spec: acpLaunchSpecFor(agent)!,
initialize: FORKS,
script: (agent) =>
agent.on('session/fork', (frame) => agent.reply(frame, { sessionId: 'copy-1' }))
})
rigs.push(rig)
return rig
}

describe('copying a chat conversation for the CLI', () => {
it.each(['omp', 'opencode'] as const)(
'copies a stopped and running %s chat without loading or prompting its original',
async (agent) => {
const rig = await open(agent)
const request = {
...input,
identity: { ...input.identity, agent },
accountHome:
agent === 'omp'
? input.accountHome
: { kind: 'opencode' as const, locator: { kind: 'unmanaged' as const } }
}
const expected = {
sessionId: 'copy-1',
...(agent === 'omp' ? { filePath: '/account-a/sessions/copy-1.jsonl' } : {})
}
await expect(rig.adapter.copyConversation(request)).resolves.toEqual(expected)
expect(rig.child().agent.frames.map((frame) => frame.method)).toEqual([
'initialize',
'session/fork'
])
expect(rig.sent('session/fork')[0]?.params).toEqual({
sessionId: PROVIDER_SESSION,
cwd: '/workspace/project',
mcpServers: []
})
expect(rig.child().closes).toBe(1)
await rig.acquire()
const chat = rig.child()
const frames = [...chat.agent.frames]
await expect(rig.adapter.copyConversation(request)).resolves.toEqual(expected)
expect(chat.agent.frames).toEqual(frames)
expect(chat.closes).toBe(0)
expect(rig.child()).not.toBe(chat)
expect(rig.child().closes).toBe(1)
}
)

it('permits another copy after unproven cleanup and retries the retained child at shutdown', async () => {
const rig = await openAcpAdapterRig({ spec: acpLaunchSpecFor('omp')!, initialize: FORKS })
rigs.push(rig)
const copying = rig.adapter.copyConversation(input)
await vi.waitFor(() => expect(rig.sent('session/fork')).toHaveLength(1))
const child = rig.child()
child.proveClose = async () => false
child.agent.reply(rig.sent('session/fork')[0]!, { sessionId: 'copy-1' })
await expect(copying).resolves.toHaveProperty('sessionId', 'copy-1')
const next = rig.adapter.copyConversation(input)
await vi.waitFor(() => {
expect(rig.child()).not.toBe(child)
expect(rig.sent('session/fork')).toHaveLength(1)
})
rig.child().agent.reply(rig.sent('session/fork')[0]!, { sessionId: 'copy-2' })
await expect(next).resolves.toHaveProperty('sessionId', 'copy-2')
child.proveClose = async () => {
child.exit()
return true
}
await rig.adapter.closeAll()
expect(child.closes).toBe(2)
})

it('bounds an unanswered fork and closes its temporary process', async () => {
const children: FakeAcpChild[] = []
const spec = acpLaunchSpecFor('omp')!
await expect(
copyAcpConversation(
{
spec,
readLaunch: async () => ({
spec,
command: 'fake',
args: [],
cwd: '/repo',
env: {},
envToDelete: [],
fullAccess: false,
resume: null
}),
connect: (...args: ConstructorParameters<typeof FakeAcpChild>) => {
const child = new FakeAcpChild(...args)
child.agent.on('initialize', (frame) =>
child.agent.reply(frame, { protocolVersion: 1, ...FORKS })
)
children.push(child)
return child
}
},
input,
{ timeoutMs: 10, track: () => {} }
)
).rejects.toThrow('OMP could not copy this chat')
expect(children.map((child) => child.closes)).toEqual([1])
})
})
152 changes: 152 additions & 0 deletions src/main/acp/acp-conversation-copy.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,152 @@
import { normalizeAgentProviderSession } from '../../shared/agent-session-resume'
import {
assertOpenCodeCopyUsesLocalStore,
OpenCodeCopyServerRouteError
} from '../../shared/opencode-launch-command'
// The chat protocol saves an independent conversation; its ordinary CLI opens that copy.
import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types'
import {
agentSessionAccountHomesEqual,
type AgentSessionAccountHome
} from '../../shared/agent-session-account-home'
import { waitForPromiseWithSignal } from '../../shared/abort-signal-reason'
import type { AiVaultConversationCopy } from '../../shared/ai-vault-resume-preparation'
import { resolveSessionFilePath } from '../native-chat/session-file-resolver'
import { resolveOmpSessionsDir } from '../ai-vault/omp-session-root'
import { AcpRpcError } from './acp-errors'
import { acpAgentName } from './acp-structured-agent-definitions'
import type { AcpStructuredConnection } from './acp-structured-connection'
import type { AcpStructuredSessionAdapterDeps } from './acp-structured-session-adapter-deps'

class AcpCopyAccountMismatchError extends Error {
constructor() {
super('Select the account this chat uses before opening its copy in a CLI.')
}
}

export const ACP_CONVERSATION_COPY_TIMEOUT_MS = 60_000

export async function copyAcpConversation(
deps: Pick<
AcpStructuredSessionAdapterDeps,
'spec' | 'readLaunch' | 'resolveLaunchArgs' | 'connect' | 'logger'
>,
input: {
identity: AgentSessionJournalIdentity
providerSessionId: string
accountHome: AgentSessionAccountHome
},
options: {
timeoutMs?: number
signal?: AbortSignal
track: (connection: AcpStructuredConnection) => void
}
): Promise<AiVaultConversationCopy> {
const name = acpAgentName(deps.spec.agent)
const timeout = AbortSignal.timeout(options.timeoutMs ?? ACP_CONVERSATION_COPY_TIMEOUT_MS)
const signal = options.signal ? AbortSignal.any([timeout, options.signal]) : timeout
let connection: AcpStructuredConnection | null = null
try {
signal.throwIfAborted()
if (deps.spec.agent === 'opencode') {
const args = await waitForPromiseWithSignal(
Promise.resolve(deps.resolveLaunchArgs?.() ?? []),
signal
)
assertOpenCodeCopyUsesLocalStore(args)
}
signal.throwIfAborted()
const launch = await waitForPromiseWithSignal(deps.readLaunch(input.identity), signal)
signal.throwIfAborted()
await assertCopyAccountSelected(deps.spec, input.accountHome, launch.env, signal)
signal.throwIfAborted()
connection = deps.connect(
{
command: launch.command,
args: launch.args,
cwd: launch.cwd,
env: launch.env,
envToDelete: launch.envToDelete
},
{ clientInfo: { name: 'orca', version: '1' } }
)
options.track(connection)
const sessionId = await waitForPromiseWithSignal(
forkAdvertised(connection, input.providerSessionId, launch.cwd),
signal
)
await assertCopyAccountSelected(deps.spec, input.accountHome, launch.env, signal)
const providerSession = normalizeAgentProviderSession({ key: 'session_id', id: sessionId })
if (!providerSession || providerSession.id !== sessionId) {
throw new Error('The agent returned an invalid copied conversation identity.')
}
if (sessionId === input.providerSessionId) {
throw new Error('The agent returned the original conversation instead of a copy.')
}
const filePath =
deps.spec.agent === 'omp'
? await resolveSessionFilePath(
'omp',
sessionId,
{
ompSessionsDir: resolveOmpSessionsDir({ env: launch.env })
},
signal
)
: undefined
if (deps.spec.agent === 'omp' && !filePath) {
throw new Error('OMP did not save the copied conversation in its pinned session directory.')
}
return {
sessionId,
...(filePath ? { filePath } : {})
}
} catch (error) {
if (
error instanceof AcpCopyAccountMismatchError ||
error instanceof OpenCodeCopyServerRouteError
) {
throw error
}
throw new Error(
error instanceof AcpRpcError && error.code === -32601
? `This ${name} version cannot copy a chat. Update ${name} to resume this chat in a new CLI.`
: `${name} could not copy this chat.`,
{ cause: error }
)
} finally {
if (connection && !(await connection.close().catch(() => false))) {
deps.logger?.warn('ACP conversation copy child exit is unproven; retained for cleanup', {
scope: 'acp-conversation-copy',
sessionId: input.identity.sessionId
})
}
}
}

async function forkAdvertised(
connection: AcpStructuredConnection,
sessionId: string,
cwd: string
): Promise<string> {
const { agentCapabilities } = await connection.initialize()
if (agentCapabilities?.sessionCapabilities?.fork == null) {
throw new AcpRpcError(-32601, 'session/fork is not advertised')
}
return connection.forkSession(sessionId, cwd)
}

/** Rederive selection at the fork boundaries; ordinary terminals still own their normal routing. */
async function assertCopyAccountSelected(
spec: AcpStructuredSessionAdapterDeps['spec'],
home: AgentSessionAccountHome,
launchEnv: Record<string, string>,
signal: AbortSignal
): Promise<void> {
if ('kind' in home && home.kind === 'opencode') {
const selected = await waitForPromiseWithSignal(spec.account.resolve({ launchEnv }), signal)
if (!agentSessionAccountHomesEqual(home, selected)) {
throw new AcpCopyAccountMismatchError()
}
}
}
12 changes: 11 additions & 1 deletion src/main/acp/acp-session-runtime.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import type { Readable, Writable } from 'node:stream'
import type { z } from 'zod'
import { z } from 'zod'
import {
AcpAgentError,
AcpAuthRequiredError,
Expand Down Expand Up @@ -28,6 +28,7 @@ import {
SetSessionModeResponseSchema,
SetSessionModelResponseSchema,
SetSessionConfigOptionResponseSchema,
SessionIdSchema,
type InitializeRequest,
type InitializeResponse,
type AuthenticateResponse,
Expand All @@ -42,6 +43,8 @@ import {
type SetSessionModelResponse
} from './generated/acp-protocol.generated'

const ForkSessionResponseSchema = z.looseObject({ sessionId: SessionIdSchema })

type Meta = PromptRequest['_meta']
const withMeta = (meta: Meta): { _meta?: Meta } => (meta ? { _meta: meta } : {})

Expand Down Expand Up @@ -142,6 +145,13 @@ export class AcpSessionRuntime {
return this.peer.request(method, params, { timeoutMs: null })
}

/** Copies saved session `sessionId` into a new one the agent saves (`session/fork`) and answers
* its id; this connection starts no session of its own. Ask only an agent that advertises it. */
async forkSession(sessionId: string, cwd: string): Promise<string> {
const params = { sessionId, cwd, mcpServers: [] }
return (await this.call('session/fork', params, ForkSessionResponseSchema)).sessionId
}

start(options: AcpSessionStartOptions): Promise<AcpSessionStarted> {
if (this.started) {
return Promise.resolve(this.started)
Expand Down
1 change: 1 addition & 0 deletions src/main/acp/acp-structured-connection.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import type { ProviderProcessLaunch } from '../provider-process/provider-process
export type AcpStructuredConnection = Pick<
AcpAgentConnection,
| 'initialize'
| 'forkSession'
| 'start'
| 'prompt'
| 'cancel'
Expand Down
2 changes: 2 additions & 0 deletions src/main/acp/acp-structured-session-adapter-deps.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ export type AcpStructuredSessionAdapterDeps = {
}) => Promise<AcpStructuredLaunch>
/** The same launch for a read that starts no child (restart recovery): the agent's env, unsealed. */
readLaunch: (identity: AgentSessionJournalIdentity) => Promise<AcpStructuredLaunch>
/** Current host arguments before protocol-specific filtering; copies open through the CLI. */
resolveLaunchArgs?: () => Promise<string[]> | string[]
/** The chat's journal as it stands, read without opening it; null when it has none. */
readJournal?: (sessionId: string) => JournalLoad | null
/** Starts the agent's process and owns its protocol: `createAcpAgentConnection` in production. */
Expand Down
Loading
Loading