diff --git a/src/main/claude/claude-at-rest-commands.ts b/src/main/claude/claude-at-rest-commands.ts new file mode 100644 index 000000000000..7eaa9c143ed3 --- /dev/null +++ b/src/main/claude/claude-at-rest-commands.ts @@ -0,0 +1,145 @@ +import { join } from 'node:path' +import type { AgentSessionRecord } from '../../shared/agent-session-record' +import type { AgentSessionSlashCommand } from '../../shared/agent-session-wire' +import { structuredSlashCommands } from '../../shared/structured-agent-session-composer' +import { discoverSkills } from '../skills/discovery' +import { scanClaudeCommandFolders } from './claude-command-folder-scan' +import { supportsClaudeStructuredLocation } from './claude-structured-location-support' + +/** How long one scan answers for a workspace and account before the next read scans again. */ +export const CLAUDE_AT_REST_COMMANDS_TTL_MS = 10_000 +/** A scan still unanswered this long no longer holds the next one off, so a hung folder can't + * freeze the menu. At most two run at once, so a folder that stays hung holds no more. */ +const SCAN_OVERDUE_MS = 30_000 +const MAX_SCANS_IN_FLIGHT = 2 + +export type ClaudeAtRestCommandsDeps = { + resolveWorkspacePath: (workspaceId: string) => Promise + now?: () => number + discover?: typeof discoverSkills +} + +type Entry = { + commands?: AgentSessionSlashCommand[] + /** When the kept answer, or the last failure, landed. */ + scannedAt: number + lastStartedAt: number + inFlight: number + /** Numbers each scan, so only an answer newer than the one kept replaces it. */ + started: number + kept: number +} + +/** + * The `/` surface of a Claude chat whose Claude is not running, read from the folders Claude reads + * its own from on the host that runs it: the account's and the workspace's command folders and + * skills, beside the commands the chat's menu always offers. Nothing until the first scan lands; + * `onChange` fires when a scan finds something different. + */ +export class ClaudeAtRestCommandCatalog { + private readonly entries = new Map() + private readonly listeners = new Set<() => void>() + + constructor(private readonly deps: ClaudeAtRestCommandsDeps) {} + + read = (record: AgentSessionRecord): AgentSessionSlashCommand[] | undefined => { + // Another host's folders are only readable there, so this host never answers for them. + if (record.provider !== 'claude' || !supportsClaudeStructuredLocation(record.location)) { + return undefined + } + const key = JSON.stringify([record.location.workspaceId, record.accountHome.path]) + let entry = this.entries.get(key) + if (!entry) { + entry = { + scannedAt: Number.NEGATIVE_INFINITY, + lastStartedAt: Number.NEGATIVE_INFINITY, + inFlight: 0, + started: 0, + kept: 0 + } + this.entries.set(key, entry) + } + const now = this.now() + const due = + entry.inFlight === 0 + ? now - entry.scannedAt >= CLAUDE_AT_REST_COMMANDS_TTL_MS + : entry.inFlight < MAX_SCANS_IN_FLIGHT && now - entry.lastStartedAt >= SCAN_OVERDUE_MS + if (due) { + this.scanInto(entry, record) + } + return entry.commands + } + + onChange = (listener: () => void): (() => void) => { + this.listeners.add(listener) + return () => this.listeners.delete(listener) + } + + private scanInto(entry: Entry, record: AgentSessionRecord): void { + const scan = ++entry.started + entry.lastStartedAt = this.now() + entry.inFlight += 1 + const newest = () => { + if (scan < entry.kept) { + return false + } + entry.kept = scan + entry.scannedAt = this.now() + return true + } + void this.scan(record) + .then( + (commands) => { + if (!newest() || JSON.stringify(commands) === JSON.stringify(entry.commands)) { + return + } + entry.commands = commands + for (const listener of this.listeners) { + listener() + } + }, + (error: unknown) => { + // Waits out the window before trying again, but leaves room for an older scan's answer. + entry.scannedAt = this.now() + console.warn('[claude] reading the at-rest `/` commands failed:', error) + } + ) + .finally(() => { + entry.inFlight -= 1 + }) + } + + private async scan(record: AgentSessionRecord): Promise { + const cwd = await this.deps.resolveWorkspacePath(record.location.workspaceId) + const account = record.accountHome.path + const [custom, discovered] = await Promise.all([ + scanClaudeCommandFolders([join(cwd, '.claude', 'commands'), join(account, 'commands')]), + (this.deps.discover ?? discoverSkills)({ + repos: [], + cwd, + providerRootOverrides: { claude: join(account, 'skills') } + }) + ]) + // What the menu offers a Claude chat at rest without a list, so reading one loses none of it. + const builtIn = structuredSlashCommands(['clear', 'compact'], 'claude').map( + (command): AgentSessionSlashCommand => ({ + name: command.name, + kind: 'command', + ...(command.description ? { description: command.description } : {}) + }) + ) + const names = new Set(builtIn.map((command) => command.name)) + const skills = discovered.skills + .filter((skill) => skill.providers.includes('claude')) + .map((skill): AgentSessionSlashCommand => ({ + name: skill.name, + kind: 'skill', + ...(skill.description ? { description: skill.description } : {}) + })) + return [...builtIn, ...custom.filter((command) => !names.has(command.name)), ...skills] + } + + private now(): number { + return this.deps.now?.() ?? Date.now() + } +} diff --git a/src/main/claude/claude-command-folder-scan.ts b/src/main/claude/claude-command-folder-scan.ts new file mode 100644 index 000000000000..574d73f7bba4 --- /dev/null +++ b/src/main/claude/claude-command-folder-scan.ts @@ -0,0 +1,88 @@ +import type { Dirent } from 'node:fs' +import { opendir, readFile } from 'node:fs/promises' +import { join, relative, sep } from 'node:path' +import type { AgentSessionSlashCommand } from '../../shared/agent-session-wire' +import { summarizeSkillMarkdown } from '../../shared/skill-metadata' + +const MARKDOWN_EXTENSION = '.md' +const MAX_SCAN_DEPTH = 24 +const MAX_SCAN_ENTRIES = 1_000 +const MAX_COMMAND_FILE_BYTES = 256 * 1024 + +type ScanBudget = { remainingEntries: number } + +/** A folder that cannot be read holds no commands we can offer, so it is skipped, not an error. */ +async function readEntries(path: string, budget: ScanBudget): Promise { + try { + const entries: Dirent[] = [] + for await (const entry of await opendir(path)) { + if (budget.remainingEntries === 0) { + break + } + budget.remainingEntries -= 1 + entries.push(entry) + } + return entries.sort((left, right) => left.name.localeCompare(right.name)) + } catch { + return [] + } +} + +async function walkMarkdown( + path: string, + depth: number, + budget: ScanBudget, + found: string[] +): Promise { + if (depth > MAX_SCAN_DEPTH || budget.remainingEntries === 0) { + return + } + for (const entry of await readEntries(path, budget)) { + // Links are not followed: a command folder is the user's own files, not a way out of it. + if (entry.isSymbolicLink()) { + continue + } + const entryPath = join(path, entry.name) + if (entry.isDirectory()) { + await walkMarkdown(entryPath, depth + 1, budget, found) + } else if (entry.isFile() && entry.name.endsWith(MARKDOWN_EXTENSION)) { + found.push(entryPath) + } + } +} + +async function describe(filePath: string): Promise { + try { + const markdown = await readFile(filePath, 'utf8') + return ( + summarizeSkillMarkdown(markdown.slice(0, MAX_COMMAND_FILE_BYTES)).description ?? undefined + ) + } catch { + return undefined + } +} + +/** + * Claude's custom commands in `roots`: every `*.md` below each folder, named by its path from + * the folder with `:` between segments (`frontend/test.md` is `/frontend:test`). The first root + * to name a command wins, so list the closest folder first. + */ +export async function scanClaudeCommandFolders( + roots: readonly string[] +): Promise { + const budget = { remainingEntries: MAX_SCAN_ENTRIES } + const commands = new Map() + for (const root of roots) { + const files: string[] = [] + await walkMarkdown(root, 0, budget, files) + for (const file of files) { + const name = relative(root, file).slice(0, -MARKDOWN_EXTENSION.length).split(sep).join(':') + if (commands.has(name)) { + continue + } + const description = await describe(file) + commands.set(name, { name, kind: 'command', ...(description ? { description } : {}) }) + } + } + return [...commands.values()] +} diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index cd241b5a33e2..83c49e5a77ae 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -63,6 +63,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda private readonly exitLifecycle: ClaudeExitLifecycle constructor(private readonly deps: ClaudeStructuredSessionAdapterDeps) { + this.atRestCommands = deps.atRestCommands this.exitLifecycle = { sessions: this.sessions, exits: this.exits, @@ -232,6 +233,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda this.sessions.has(sessionId) ? { supportsTaskStop: true, supportsStopAll: true } : undefined readCommands: NonNullable = (sessionId) => this.sessions.get(sessionId)?.commands.commands + readonly atRestCommands: ClaudeStructuredSessionAdapterDeps['atRestCommands'] holdsDispatch = (sessionId: string): boolean => { const session = this.sessions.get(sessionId) return session ? claudeHoldsDispatch(session) : false diff --git a/src/main/claude/claude-structured-session-state.ts b/src/main/claude/claude-structured-session-state.ts index 8ddbe7aed01c..2f0a262c3133 100644 --- a/src/main/claude/claude-structured-session-state.ts +++ b/src/main/claude/claude-structured-session-state.ts @@ -27,6 +27,7 @@ import type { AgentChildWorkEvidence } from '../../shared/agent-status-child-wor import type { ClaudeBackgroundTaskTracker } from './claude-background-task-tracker' import type { ClaudeChildWorkDecoder } from './claude-child-work-decoder' import type { ClaudeSlashCommandCatalog } from './claude-slash-command-catalog' +import type { ClaudeAtRestCommandCatalog } from './claude-at-rest-commands' import type { ClaudeSessionStartup } from './claude-structured-session-startup-state' export type ClaudeAuthDiagnostic = { @@ -93,6 +94,8 @@ export type ClaudeLateDispatchOutcome = | { clientMessageId: string; state: 'unknown'; reason: string } export type ClaudeStructuredSessionAdapterDeps = { + /** The `/` surface of a chat whose Claude is not running. */ + atRestCommands?: ClaudeAtRestCommandCatalog resolveLaunch: (input: { identity: AgentSessionJournalIdentity }) => Promise diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts index 9531adb0318a..ddf695630eb2 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts @@ -1,4 +1,5 @@ import type { SubmissionRejectionFact } from '../../../shared/agent-session-failure' +import type { StructuredAgentSessionAtRestCommands } from './structured-agent-session-at-rest-commands' import type { AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' import type { AgentSessionAccountHome, @@ -121,6 +122,16 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi readCommands: NonNullable = (sessionId) => this.liveOwnerOrNull(sessionId)?.readCommands?.(sessionId) + atRestCommands: StructuredAgentSessionAtRestCommands = { + read: (record) => this.adapters[record.provider].atRestCommands?.read(record), + onChange: (listener) => { + const stops = Object.values(this.adapters).flatMap((adapter) => + adapter.atRestCommands ? [adapter.atRestCommands.onChange(listener)] : [] + ) + return () => stops.forEach((stop) => stop()) + } + } + answerPrompt: StructuredAgentSessionAdapter['answerPrompt'] = (input) => this.owner(input.sessionId).answerPrompt(input) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index 129f363aa8cb..1d26ea427aea 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -25,6 +25,7 @@ import type { AgentSessionExecutionLocation, AgentSessionProcessIdentity } from '../../../shared/agent-session-record' +import type { StructuredAgentSessionAtRestCommands } from './structured-agent-session-at-rest-commands' import type { AgentSessionOptionsResult, AgentSessionSlashCommand, @@ -347,6 +348,8 @@ export type StructuredAgentSessionAdapter = { /** The `/` surface the running provider reports for itself. Undefined when the * provider never reports one, which is what keeps the client on its catalog. */ readCommands?(sessionId: string): AgentSessionSlashCommand[] | undefined + /** The `/` surface of a chat whose agent is not running; absent when the provider has none. */ + atRestCommands?: StructuredAgentSessionAtRestCommands /** Claims the live callback, builds the provider reply, commits the journal CAS while that claim is * held, then answers it. A reply that cannot be built throws `AgentSessionPromptAnswerRejectedError` * before the commit. A prompt cancel claims the same callback, so only one operation can commit. */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-at-rest-commands.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-at-rest-commands.test.ts new file mode 100644 index 000000000000..063cf84ebd79 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-at-rest-commands.test.ts @@ -0,0 +1,319 @@ +// A Claude chat whose Claude is not running offers the `/` commands and skills Claude would read +// from its folders on the host that runs it: after a /clear, after a relaunch, as files change. + +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import type { + AgentSessionSlashCommand, + AgentSessionSubscribeEvent +} from '../../../shared/agent-session-wire' +import { + CLAUDE_AT_REST_COMMANDS_TTL_MS, + ClaudeAtRestCommandCatalog +} from '../../claude/claude-at-rest-commands' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { + HOST_TEST_LOCATION, + HOST_TEST_NOW, + HOST_TEST_SESSION as SESSION, + hostTestAttachParams, + hostTestOperationId, + resetHostTestOperationIds +} from './structured-agent-session-host-test-data' +import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' + +const caller = { callerKey: 'desktop' } +const CLAUDE_SESSION = '819cf9f8-e43c-4ad7-b50f-54aa158a726a' + +let directory: string +let workspace: string +let account: string +let clock: number +let store: AgentSessionRecordStore +let host: StructuredAgentSessionHost + +/** Claude is never running here, so the only `/` surface is the one read from its folders. */ +function adapter(catalog: ClaudeAtRestCommandCatalog): StructuredAgentSessionAdapter { + return { + supportsLocation: () => true, + supportsCreate: () => true, + acquire: vi.fn(async (input) => ({ + process: { + hostId: 'local', + pid: 4001, + processStartTimeMs: HOST_TEST_NOW, + spawnToken: input.spawnToken + }, + link: { + linkId: `link-${input.fence}`, + mintedAtFence: input.fence, + observedAt: HOST_TEST_NOW, + origin: 'created' as const, + handle: { provider: 'claude' as const, sessionId: CLAUDE_SESSION, leafUuid: null } + } + })), + atRestCommands: catalog, + dispatch: vi.fn(async () => ({ state: 'unknown' as const, reason: 'test' })), + cancelTurn: async () => ({ cancelled: true }), + answerPrompt: async () => {}, + setOption: async () => {}, + releaseAcquisition: async () => true, + closeSession: async () => true + } +} + +function catalogFor(workspacePath: string): ClaudeAtRestCommandCatalog { + return new ClaudeAtRestCommandCatalog({ + resolveWorkspacePath: async () => workspacePath, + now: () => clock, + // Only the folders this test writes, never the machine's own home. + discover: async (args) => { + const { discoverSkills } = await import('../../skills/discovery') + return discoverSkills({ ...args, homeDir: join(directory, 'home'), refresh: true }) + } + }) +} + +async function openHost(catalog = catalogFor(workspace)): Promise { + store = await openTestAgentSessionRecordStore(directory) + host = new StructuredAgentSessionHost({ + store, + adapter: adapter(catalog), + journalDatabase: openTestJournalHostDatabase(directory), + claimKeyId: 'key', + now: () => clock, + probeOwner: async () => ({ outcome: 'pid-absent' }) + }) +} + +async function writeSkill(root: string, name: string): Promise { + await mkdir(join(root, name), { recursive: true }) + await writeFile(join(root, name, 'SKILL.md'), `---\nname: ${name}\ndescription: A skill\n---\n`) +} + +async function writeCommand(root: string, file: string): Promise { + await mkdir(join(root, file, '..'), { recursive: true }) + await writeFile(join(root, file), '---\ndescription: A command\n---\nDo it.\n') +} + +/** The `/` surface a pane subscribed to `sessionId` shows once the scan has landed. */ +async function menuOf(sessionId: string): Promise { + let commands: AgentSessionSlashCommand[] | null | undefined + const emit = (event: AgentSessionSubscribeEvent) => { + if ('commands' in event) { + commands = event.commands + } + } + const unsubscribe = await host.subscribe({ id: `pane-${sessionId}`, sessionId, emit }) + await vi.waitFor(() => expect(commands).toBeTruthy()) + unsubscribe() + return commands ?? null +} + +const namesOf = (menu: AgentSessionSlashCommand[] | null, kind: 'skill' | 'command') => + (menu ?? []).filter((entry) => entry.kind === kind).map((entry) => entry.name) + +beforeEach(async () => { + resetHostTestOperationIds() + clock = HOST_TEST_NOW + directory = await mkdtemp(join(tmpdir(), 'orca-at-rest-commands-')) + workspace = join(directory, 'workspace') + account = join(directory, 'claude-account') + await writeSkill(join(workspace, '.claude', 'skills'), 'review-pr') + await writeCommand(join(account, 'commands'), 'deploy.md') + await openHost() + const attached = await host.attach( + caller, + hostTestAttachParams(null, { + provider: 'claude', + agent: 'claude', + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: account }, + providerHandle: { kind: 'claude', sessionId: CLAUDE_SESSION, leafUuid: null } + }) + ) + expect(attached).toMatchObject({ ok: true }) + await host.setSessionTabVisibility(SESSION, true) +}) + +afterEach(async () => { + await host.flushAllStreamedEvents() + await rm(directory, { recursive: true, force: true }) +}) + +describe("a Claude chat whose Claude isn't running shows the `/` surface from its folders", () => { + it('after /clear, before the new chat has started', async () => { + const cleared = await host.conversationCommand(caller, { + command: 'clear', + envelope: { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: store.getRecord(SESSION)!.lease.runtimeFence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.conversationCommand', + sessionId: SESSION, + fields: { command: 'clear' } + }) + } + }) + const replacement = cleared.ok ? cleared.value.replacementSessionId! : '' + const menu = await menuOf(replacement) + expect(namesOf(menu, 'skill')).toEqual(['review-pr']) + // Everything the menu offered before a list existed, then what the folders add. + expect(namesOf(menu, 'command')).toEqual(['model', 'effort', 'clear', 'compact', 'deploy']) + }) + + it('keeps a slow scan once it answers, never a staler one after a newer, and runs at most two', async () => { + const pending: ((skills: string[]) => void)[] = [] + const catalog = new ClaudeAtRestCommandCatalog({ + resolveWorkspacePath: async () => workspace, + now: () => clock, + discover: () => + new Promise((resolve) => { + pending.push((names) => + resolve({ + skills: names.map((name) => ({ + id: name, + name, + description: null, + providers: ['claude'], + sourceKind: 'repo', + sourceLabel: 'repo', + rootPath: workspace, + directoryPath: workspace, + skillFilePath: workspace, + installed: true, + updatedAt: null + })), + sources: [], + scannedAt: clock + }) + ) + }) + }) + const record = store.getRecord(SESSION)! + const skills = () => namesOf(catalog.read(record) ?? null, 'skill') + const changed = vi.fn() + catalog.onChange(changed) + catalog.read(record) + await vi.waitFor(() => expect(pending).toHaveLength(1)) + // Overdue: a second scan starts beside the first; a third never does while both run. + clock += 30_000 + catalog.read(record) + await vi.waitFor(() => expect(pending).toHaveLength(2)) + clock += 30_000 + catalog.read(record) + await new Promise((resolve) => setTimeout(resolve, 20)) + expect(pending).toHaveLength(2) + // The second answers first; the first, older one landing after it changes nothing. + pending[1](['newer']) + await vi.waitFor(() => expect(skills()).toEqual(['newer'])) + pending[0](['older']) + await new Promise((resolve) => setTimeout(resolve, 20)) + expect(skills()).toEqual(['newer']) + expect(changed).toHaveBeenCalledTimes(1) + // A slow scan is not thrown away: the next one, however long it takes, is kept. + clock += CLAUDE_AT_REST_COMMANDS_TTL_MS + catalog.read(record) + await vi.waitFor(() => expect(pending).toHaveLength(3)) + clock += 35_000 + pending[2](['slow']) + await vi.waitFor(() => expect(skills()).toEqual(['slow'])) + }) + + it('waits out the window from when a scan lands, even one that failed', async () => { + let calls = 0 + let failNext = true + const catalog = new ClaudeAtRestCommandCatalog({ + resolveWorkspacePath: async () => workspace, + now: () => clock, + discover: async (args) => { + calls += 1 + // Each scan takes 15 s of the clock, longer than the window. + clock += 15_000 + if (failNext) { + failNext = false + throw new Error('unreadable') + } + const { discoverSkills } = await import('../../skills/discovery') + return discoverSkills({ ...args, homeDir: join(directory, 'home'), refresh: true }) + } + }) + const record = store.getRecord(SESSION)! + const warned = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + try { + catalog.read(record) + await vi.waitFor(() => expect(warned).toHaveBeenCalledOnce()) + catalog.read(record) + await new Promise((resolve) => setTimeout(resolve, 20)) + expect(calls).toBe(1) + clock += CLAUDE_AT_REST_COMMANDS_TTL_MS + catalog.read(record) + await vi.waitFor(() => + expect(namesOf(catalog.read(record) ?? null, 'skill')).toEqual(['review-pr']) + ) + clock += CLAUDE_AT_REST_COMMANDS_TTL_MS - 1 + catalog.read(record) + await new Promise((resolve) => setTimeout(resolve, 20)) + expect(calls).toBe(2) + } finally { + warned.mockRestore() + } + }) + + it('after a relaunch, with nothing remembered from before it', async () => { + await host.flushAllStreamedEvents() + await openHost() + const menu = await menuOf(SESSION) + expect(namesOf(menu, 'skill')).toEqual(['review-pr']) + expect(namesOf(menu, 'command')).toContain('deploy') + }) + + it('with a command added on disk, pushed to a pane already showing the chat', async () => { + let commands: AgentSessionSlashCommand[] | null | undefined + await host.subscribe({ + id: 'pane', + sessionId: SESSION, + emit: (event) => { + if ('commands' in event) { + commands = event.commands + } + } + }) + await vi.waitFor(() => expect(namesOf(commands ?? null, 'command')).toContain('deploy')) + await writeCommand(join(workspace, '.claude', 'commands'), 'frontend/test.md') + clock += CLAUDE_AT_REST_COMMANDS_TTL_MS + // Anything that reads the surface after the window scans again; the pane is told the answer. + host.readCommands(SESSION) + await vi.waitFor(() => + expect(namesOf(commands ?? null, 'command')).toEqual( + expect.arrayContaining(['deploy', 'frontend:test']) + ) + ) + }) + + it('from the folders of the host that runs it, never another host’s', async () => { + const remoteWorkspace = join(directory, 'remote-workspace') + await writeSkill(join(remoteWorkspace, '.claude', 'skills'), 'remote-only') + const remote = catalogFor(remoteWorkspace) + const record = store.getRecord(SESSION)! + remote.read(record) + await vi.waitFor(() => + expect(namesOf(remote.read(record) ?? null, 'skill')).toEqual(['remote-only']) + ) + // A chat this host does not run is answered by nothing here, not by this host's folders. + const local = catalogFor(workspace) + const elsewhere = { + ...record, + location: { ...HOST_TEST_LOCATION, executionHostId: 'ssh:box' as const } + } + local.read(elsewhere) + await new Promise((resolve) => setTimeout(resolve, 50)) + expect(local.read(elsewhere)).toBeUndefined() + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-at-rest-commands.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-at-rest-commands.ts new file mode 100644 index 000000000000..8eccca583916 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-at-rest-commands.ts @@ -0,0 +1,9 @@ +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentSessionSlashCommand } from '../../../shared/agent-session-wire' + +/** The `/` surface of a chat whose agent is not running, read where the provider reads its own on + * this host. `read` is undefined until known; `onChange` fires when it changes. */ +export type StructuredAgentSessionAtRestCommands = { + read(record: AgentSessionRecord): AgentSessionSlashCommand[] | undefined + onChange(listener: () => void): () => void +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts index 2efe7dd48f5c..8195e7c18f9b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts @@ -23,6 +23,7 @@ import { export class StructuredAgentSessionClientDelivery { readonly subscribers: AgentSessionSubscribers readonly waitForSendSettlement: StructuredAgentSessionSendSettlement['wait'] + private stopAtRestCommandUpdates: () => void = () => undefined private readonly statusFeed private readonly turnCompletionFeed private readonly sendSettlement @@ -30,7 +31,7 @@ export class StructuredAgentSessionClientDelivery { constructor( private readonly sessions: Map, now: () => number, - deps: () => StructuredAgentSessionHostDeps, + private readonly deps: () => StructuredAgentSessionHostDeps, private readonly onJournalActivity?: (sessionId: string) => void, onAgentStarted?: (sessionId: string) => void, /** A session's child records changed; the chat strip republishes from them. */ @@ -53,13 +54,31 @@ export class StructuredAgentSessionClientDelivery { ) this.waitForSendSettlement = this.sendSettlement.wait this.subscribers = new AgentSessionSubscribers({ - readCommands: (sessionId) => deps().adapter.readCommands?.(sessionId), + readCommands: (sessionId) => this.readCommands(sessionId), readQueuePublication: (sessionId) => tryReadQueuePublication(sessions.get(sessionId)?.journal), onJournalPublished: (sessionId, journal) => this.publishJournal(sessionId, journal) }) } + /** Re-sends the `/` surface whenever the provider's at-rest one changes. */ + watchAtRestCommands(adapter: StructuredAgentSessionHostDeps['adapter']): void { + this.stopAtRestCommandUpdates = + adapter.atRestCommands?.onChange(() => this.subscribers.republishCommands()) ?? + (() => undefined) + } + + /** What the running agent reports; with none running, what the provider would read at rest. */ + readCommands(sessionId: string) { + const { adapter, store } = this.deps() + const live = adapter.readCommands?.(sessionId) + if (live !== undefined) { + return live + } + const record = store.getRecord(sessionId) + return record ? adapter.atRestCommands?.read(record) : undefined + } + publishStatus = (sessionId: string): void => this.statusFeed.publish(sessionId) publishChildWork = (sessionId: string, evidence: AgentChildWorkEvidence[]): void => @@ -101,6 +120,7 @@ export class StructuredAgentSessionClientDelivery { } closeAll(): void { + this.stopAtRestCommandUpdates() this.sendSettlement.closeAll() } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts index 9740bac146ef..ad149dbf9472 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts @@ -115,7 +115,13 @@ export class StructuredAgentSessionDeliveryLoop { } if (!prepared.ok) { const { refusal, diagnostic } = prepared - const cause = { refusal, ...(diagnostic ? { diagnostic } : {}) } + // A conversation no agent ever ran, such as a cleared chat's, failed to start, not restart. + const newSession = this.deps.record(sessionId)?.providerHandleChain.length === 0 + const cause = { + refusal, + ...(diagnostic ? { diagnostic } : {}), + ...(newSession ? { newSession: true as const } : {}) + } await this.deps.serialize(sessionId, () => this.fail(sessionId, { startKey: null, cause }) ) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index c7e91ec8dde6..37f4a1966101 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -101,6 +101,7 @@ export class StructuredAgentSessionHost { readonly restartResume: StructuredAgentSessionRestartResume constructor(readonly deps: StructuredAgentSessionHostDeps) { + this.clientDelivery.watchAtRestCommands(deps.adapter) this.backgroundTasks = new StructuredAgentSessionBackgroundTaskChannel( deps, this.sessions, @@ -310,9 +311,7 @@ export class StructuredAgentSessionHost { this.conversationCommands.run(...args) conversationReplacements = () => this.conversationCommands.replacements() /** Undefined means unavailable; an empty array is an authoritative catalog. */ - readCommands = (sessionId: string): SessionWire.AgentSessionCommandsResult => ({ - commands: this.deps.adapter.readCommands?.(sessionId) - }) + readCommands = (sessionId: string) => ({ commands: this.clientDelivery.readCommands(sessionId) }) /** From the record store, never the session map: an idle-released chat has no map entry. */ handoffStatus = (sessionId: string): SessionWire.AgentSessionHandoffStatus => diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-command.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-command.test.ts index 9cc22f9547a0..2be4a8ea12b6 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-command.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-command.test.ts @@ -98,19 +98,22 @@ describe('a /compact in flight', () => { describe('/clear', () => { it('in flight, refuses a capable send as today: no card lands on the source it supersedes', async () => { - const attach = rig.host.attach.bind(rig.host) + const commit = rig.store.commitConversationClear let release: (() => void) | undefined const released = new Promise((resolve) => { release = resolve }) - const spy = vi.spyOn(rig.host, 'attach').mockImplementationOnce(async (...args) => { - await released - return attach(...args) - }) + const committing = vi.fn() + const spy = vi + .spyOn(rig.store, 'commitConversationClear') + .mockImplementationOnce(async (clear) => { + committing() + await released + return commit(clear) + }) try { const cleared = command('clear') - // Nothing is recorded before the clear commits; wait until it is starting the replacement. - await eventually(() => expect(spy).toHaveBeenCalled()) + await eventually(() => expect(committing).toHaveBeenCalledOnce()) expect(await rig.send('sent while clearing', 'queue-if-active').result).toEqual(WAIT_REFUSAL) release?.() const done = await cleared @@ -125,6 +128,58 @@ describe('/clear', () => { } }) + it('carries drafts to a replacement no agent has started, before it answers; the first send starts the agent ahead of them', async () => { + const working = await rig.workingSend() + const firstId = await queuedId(rig.send('first draft', 'queue-if-active').result) + const secondId = await queuedId(rig.send('second draft', 'queue-if-active').result) + await rig.stop() + await rig.settleAccepted(working, 'a') + const cleared = await command('clear') + const replacementId = cleared.ok ? cleared.value.replacementSessionId : undefined + if (!replacementId) { + throw new Error('expected a replacement session') + } + // The clear started nothing, yet the drafts had already moved when it answered. + expect(rig.store.getRecord(replacementId)).toMatchObject({ + providerHandleChain: [], + lease: { claimStatus: 'released' } + }) + expect(rig.host.collaboratorsForTests().sessions.get(replacementId)?.child ?? null).toBeNull() + expect(await rig.drafts()).toHaveLength(0) + expect(await rig.drafts(replacementId)).toEqual([ + { messageId: firstId, state: 'waiting' }, + { messageId: secondId, state: 'waiting' } + ]) + expect(await rig.queuePause(replacementId)).toEqual({ reason: 'cleared' }) + + const body = hostTestMessage('first in the new chat') + const fields = { body, delivery: 'queue-if-active' as const } + const sentId = hostTestOperationId() + expect( + await rig.host.send(CALLER, { + envelope: rig.envelope(fields, 'agentSession.send', sentId, replacementId), + ...fields, + userSend: true + }) + ).toMatchObject({ ok: true, value: { submission: expect.anything() } }) + await eventually(() => + expect(rig.store.getRecord(replacementId)?.providerHandleChain).toHaveLength(1) + ) + await rig.host.settleLateDispatch({ + sessionId: replacementId, + clientMessageId: sentId, + providerIdentity: { provider: 'codex', threadId: THREAD, turnId: 'turn-first', ordinal: 0 } + }) + await eventually(async () => + expect( + (await rig.host.journalSnapshot(replacementId)).submissions.map((entry) => + entry.clientMessageId === sentId ? 'sent' : entry.queuedMessageId + ) + ).toEqual(['sent', firstId]) + ) + expect(await rig.queuePause(replacementId)).toBeNull() + }) + it("a carried draft sent on the replacement: the provider's echo folds into its one bubble", async () => { const working = await rig.workingSend() const draftId = await queuedId(rig.send('carried text', 'queue-if-active').result) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.test.ts index e43897826a14..33de806ff3b7 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.test.ts @@ -728,6 +728,8 @@ describe('/clear', () => { if (!replacementId) { throw new Error('expected a replacement session') } + // Nothing to carry, so nothing opened the new conversation. + expect(host.hasSession(replacementId)).toBe(false) expect(await drafts(replacementId)).toHaveLength(0) }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts index 9da974a44a43..140573c5ff3f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts @@ -82,19 +82,19 @@ export async function withdrawQueuedMessagesForOperation( * discarded, so they wait for the user's next turn there, or Resume, rather than * sending into the fresh context unasked. Each card lands with the pause in one * transaction, so the drain never sees a carried card unpaused and no pause is - * left over an empty queue if an insert fails. Runs after the - * replacement's attach succeeded and before the clear commits. Each insert is - * idempotent on (session, message), so a retried clear replays it safely; the source rows are then tombstoned. Bookkeeping around the clear: - * a failure leaves the cards on the superseded source — whose supersession - * fence already blocks the drain — reported, never gating the clear. A crash - * between the copy and the tombstone leaves both, which the fence also makes - * harmless: nothing is lost and nothing runs. + * left over an empty queue if an insert fails. Runs after the clear commits, + * opening the replacement's conversation only when there are drafts to carry; + * the source rows are then tombstoned. Bookkeeping around the clear: a failure, + * or a crash before the carry, leaves the cards on the superseded source — whose + * supersession fence already blocks the drain — reported, never gating the + * clear. A crash between the copy and the tombstone leaves both, which the + * fence also makes harmless: nothing is lost and nothing runs. */ export async function carryQueuedMessagesToClearReplacement( ctx: AgentSessionTurnContext, input: { replacementSessionId: string - replacementJournal: AgentSessionJournal | undefined + openReplacementJournal: () => Promise callerKey: string operationId: string } @@ -104,7 +104,7 @@ export async function carryQueuedMessagesToClearReplacement( if (rows.length === 0) { return } - const replacement = input.replacementJournal + const replacement = await input.openReplacementJournal() if (!replacement) { throw new Error('the replacement journal is not open') } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind-at-rest.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind-at-rest.test.ts index b34de39b74f1..dd878ebe0d7c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind-at-rest.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind-at-rest.test.ts @@ -198,6 +198,29 @@ describe('an interrupted Codex rewind on a chat at rest (R16)', () => { expect(texts).toEqual(['verified history', 'after the rewind']) expect(snapshot.submissions.at(-1)?.dispatchState).toBe('accepted') }) + + it('is settled by the start a /clear makes, and the clear commits', async () => { + await interruptedRewindAtRest() + const before = acquires + + const cleared = await host.conversationCommand(caller, { + command: 'clear', + envelope: { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: fence(), + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.conversationCommand', + sessionId: SESSION, + fields: { command: 'clear' } + }) + } + }) + expect(cleared).toMatchObject({ ok: true, value: { phase: 'committed', state: 'completed' } }) + expect(acquires - before).toBe(1) + expect(recoverRewind).toHaveBeenCalledOnce() + expect(store.getRecord(SESSION)?.rewind?.phase).toBe('completed') + }) }) describe('a rewind asked of a chat at rest (P2-23)', () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts index ae7956745646..2560ff1172d2 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts @@ -206,6 +206,24 @@ export class AgentSessionSubscribers { } } + /** Re-sends the `/` surface to each subscriber whose view of it is out of date. */ + republishCommands(): void { + const hostNow = this.now() + for (const sessionId of this.bySession.keys()) { + for (const subscriber of this.subscribers(sessionId)) { + if ((this.hooks.readCommands?.(sessionId) ?? null) !== subscriber.commands) { + this.emit(subscriber, { + type: 'batch', + sessionId, + batch: emptyAgentSessionBatch(subscriber.cursor), + fence: subscriber.fence, + hostNow + }) + } + } + } + } + private subscribers(sessionId: string): Subscriber[] { return [...(this.bySession.get(sessionId)?.values() ?? [])] } diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command-controller.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command-controller.ts index 4e4e42c46e44..fae3c944ee26 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-command-controller.ts +++ b/src/main/native-chat/agent-session-wire/structured-conversation-command-controller.ts @@ -17,7 +17,7 @@ export class StructuredConversationCommandController { private readonly context: () => StructuredAgentSessionMutationContext, private readonly host: Pick< StructuredAgentSessionHost, - 'attach' | 'flushStreamedEvents' | 'waitForSendSettlement' + 'flushStreamedEvents' | 'waitForSendSettlement' > ) {} send = ( @@ -45,8 +45,8 @@ export class StructuredConversationCommandController { if (--entry.count === 0 && this.pending.get(params.envelope.sessionId) === entry) { this.pending.delete(params.envelope.sessionId) } - // A clear can settle with no journal commit (a failed attach), and drafts held behind - // its prepared phase would otherwise wait for an unrelated commit. + // A clear can settle with no journal commit (a refusal), and drafts held behind it + // would otherwise wait for an unrelated commit. this.context().wakeQueuedDrain?.(params.envelope.sessionId) } ) diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command.test.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command.test.ts index 89b07045b3d5..336e120d8baf 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-command.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-conversation-command.test.ts @@ -1,8 +1,8 @@ +import { createHash } from 'node:crypto' import { cp, mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS } from '../../../shared/agent-session-host-authority' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' import type { AgentSessionConversationCommand } from '../../../shared/agent-session-conversation-command' import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' @@ -113,7 +113,11 @@ beforeEach(async () => { linkId: `link-${acquisitions}`, mintedAtFence: input.fence, observedAt: HOST_TEST_NOW, - origin: input.fence > 1 ? ('resumed' as const) : ('created' as const), + // A start with no thread to resume creates one, whatever its fence. + origin: + input.fence > 1 && input.identity.providerHandle.kind === 'codex' + ? ('resumed' as const) + : ('created' as const), handle: { provider: 'codex' as const, threadId: @@ -147,141 +151,279 @@ afterEach(async () => { await rm(directory, { recursive: true, force: true }) }) -describe('host conversation commands', () => { - it('adopts the reported Fast preference into the replacement record', async () => { - adapter.readOptions = async () => ({ - models: [], - current: { model: 'test-model', effort: 'high', fastMode: false } - }) - const result = await host.conversationCommand(caller, commandParams('clear')) - expect(result.ok).toBe(true) - if (!result.ok) { - return +/** Every record but the first is one a committed /clear points at: nothing is left orphaned. */ +function expectEveryReplacementPointedAt(): void { + const records = store.listRecords() + const pointedAt = new Set( + records.map((record) => record.conversationCommand?.replacementSessionId) + ) + expect( + records.filter( + (record) => + record.sessionId !== HOST_TEST_SESSION && + !pointedAt.has(record.sessionId) && + !notFromAClear.has(record.sessionId) + ) + ).toEqual([]) +} +const notFromAClear = new Set() + +afterEach(() => { + expectEveryReplacementPointedAt() + notFromAClear.clear() +}) + +function startsFor(sessionId: string): number { + return vi + .mocked(adapter.acquire) + .mock.calls.filter(([input]) => input.identity.sessionId === sessionId).length +} + +function sendTo(sessionId: string, text: string) { + const body = hostTestMessage(text) + return host.send(caller, { + body, + envelope: { + sessionId, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: store.getRecord(sessionId)!.lease.runtimeFence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId, + fields: { body } + }) + } + }) +} + +async function clearCommits( + params = commandParams('clear'), + who: { callerKey: string } = caller +): Promise { + const result = await host.conversationCommand(who, params) + expect(result).toMatchObject({ + ok: true, + value: { + command: 'clear', + phase: 'committed', + state: 'completed', + replacementSessionId: expect.any(String) } - expect(store.getRecord(result.value.replacementSessionId!)).toMatchObject({ - options: { model: 'test-model', effort: 'high', fastMode: 'false' } - }) }) + expect(result.ok && 'error' in result.value).toBe(false) + return result.ok ? result.value.replacementSessionId! : '' +} + +/** The idle sweep stops the source's agent, leaving the chat at rest. */ +async function sourceAtRest(): Promise { + clock += STRUCTURED_AGENT_SESSION_IDLE_MS + 1 + await host.collaboratorsForTests().lifetime.idleSweep.tick() + expect(host.collaboratorsForTests().sessions.get(HOST_TEST_SESSION)?.child ?? null).toBeNull() + expect(store.getRecord(HOST_TEST_SESSION)?.lease.claimStatus).toBe('released') +} + +async function submissionOf(sessionId: string, clientMessageId: string) { + return (await host.journalSnapshot(sessionId)).submissions.find( + (entry) => entry.clientMessageId === clientMessageId + ) +} + +async function errorRows(sessionId: string): Promise { + return (await host.journalSnapshot(sessionId)).items.flatMap((item) => + item.body.kind === 'status' && item.body.tone === 'error' ? [item.body.text] : [] + ) +} - // What the child reports can be a value it fell back to, such as a model whose restore write it - // never answered; the replacement's start replays the choice, as the source's next start would. - it('starts the replacement from the options the user chose, not the values the child reports', async () => { +describe('/clear starts nothing', () => { + it('commits the marker and an at-rest replacement founded from the source, and starts no agent', async () => { await store.replaceSessionOptions({ sessionId: HOST_TEST_SESSION, fence: store.getRecord(HOST_TEST_SESSION)!.lease.runtimeFence, options: { model: 'test-model', effort: 'low' }, now: HOST_TEST_NOW }) - adapter.readOptions = async () => ({ - models: [], - current: { model: 'fallback-model', effort: 'high' } + const source = store.getRecord(HOST_TEST_SESSION)! + const params = commandParams('clear') + const replacement = await clearCommits(params) + expect(adapter.acquire).toHaveBeenCalledTimes(1) + expect(replacement).toMatch(/^clear-[0-9a-f]{40}$/) + expect(store.getRecord(replacement)).toMatchObject({ + sessionId: replacement, + location: source.location, + provider: source.provider, + accountHome: source.accountHome, + options: { model: 'test-model', effort: 'low' }, + // Never started: nothing to resume, so its first start is a fresh conversation. + providerHandleChain: [], + lease: { + runtimeFence: 1, + claimStatus: 'released', + handoffStage: null, + ownerProcess: null, + reservedSpawnToken: null, + unreconciled: false, + claimKeyId: 'key' + } }) - const attach = vi.spyOn(host, 'attach') - expect(await host.conversationCommand(caller, commandParams('clear'))).toMatchObject({ - ok: true + // Written at the fence the old agent's stop released the lease at. + expect(store.getRecord(HOST_TEST_SESSION)?.lease).toMatchObject({ + claimStatus: 'released', + runtimeFence: source.lease.runtimeFence + 1 }) - expect(attach.mock.calls[0]?.[1].options).toEqual({ model: 'test-model', effort: 'low' }) - expect(store.getRecord(HOST_TEST_SESSION)?.options).toEqual({ - model: 'test-model', - effort: 'low' + expect(store.getRecord(HOST_TEST_SESSION)?.conversationCommand).toEqual({ + command: 'clear', + phase: 'committed', + state: 'completed', + runtimeFence: source.lease.runtimeFence + 1, + operationId: params.envelope.clientOperationId, + callerKey: caller.callerKey, + replacementSessionId: replacement }) + expect(store.listVisibleSessionIds()).toEqual([replacement]) + expect(host.conversationReplacements().map((entry) => entry.sessionId)).toEqual([replacement]) + expect(host.collaboratorsForTests().sessions.get(replacement)?.child ?? null).toBeNull() + expect((await host.history({ sessionId: replacement, direction: 'tail' })).page.items).toEqual( + [] + ) + // The old conversation stays readable and takes no more messages. + expect(await host.send(caller, sendParams('into the old chat'))).toMatchObject({ + ok: false, + refusal: { details: { reason: 'conversationCleared' } } + }) + expect(adapter.dispatch).not.toHaveBeenCalled() }) - it('clears with a fresh record and effective options, retaining old history and idempotent mapping', async () => { - const before = store.getRecord(HOST_TEST_SESSION)! - const params = commandParams('clear') - const result = await host.conversationCommand(caller, params) - expect(result.ok).toBe(true) - if (!result.ok) { - return - } - const nextId = result.value.replacementSessionId! - expect(nextId).not.toBe(HOST_TEST_SESSION) - expect(store.getRecord(nextId)).toMatchObject({ - location: before.location, - accountHome: before.accountHome, - options: { model: 'test-model', effort: 'high' } - }) - expect(store.getRecord(HOST_TEST_SESSION)).not.toBeNull() - expect(store.listVisibleSessionIds()).toEqual([nextId]) - expect((await host.history({ sessionId: nextId, direction: 'tail' })).page.items).toEqual([]) - expect(await host.conversationCommand(caller, params)).toMatchObject({ - ok: true, - replayed: true, - value: { replacementSessionId: nextId } - }) - expect(acquisitions).toBe(2) - const body = hostTestMessage('late send') + it('copies the launch arguments the source was pinned to', async () => { + const pinned = 'session-pinned' expect( - await host.send(caller, { - body, - envelope: { - ...params.envelope, - clientOperationId: hostTestOperationId(), - payloadFingerprint: computeAgentSessionPayloadFingerprint({ - method: 'agentSession.send', - sessionId: HOST_TEST_SESSION, - fields: { body } - }) - } + await host.attach( + caller, + hostTestAttachParams(null, { + envelope: { + sessionId: pinned, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: null, + payloadFingerprint: '' + }, + launchArgs: ['--flag'] + }) + ) + ).toMatchObject({ ok: true }) + const clear = { ...commandParams('clear') } + clear.envelope = { + sessionId: pinned, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: store.getRecord(pinned)!.lease.runtimeFence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.conversationCommand', + sessionId: pinned, + fields: { command: 'clear' } }) - ).toMatchObject({ ok: false }) - expect(adapter.dispatch).not.toHaveBeenCalled() + } + notFromAClear.add(pinned) + const replacement = await clearCommits(clear) + expect(store.getRecord(replacement)?.launchArgs).toEqual(['--flag']) }) - it('leaves the source usable when replacement creation is definitely refused', async () => { - vi.spyOn(host, 'attach').mockResolvedValueOnce({ - ok: false, - refusal: { code: 'structured_agent_session_unsupported', message: 'Unavailable' } + it("does not start an at-rest source's agent", async () => { + await sourceAtRest() + const starts = vi.mocked(adapter.acquire).mock.calls.length + const replacement = await clearCommits() + expect(adapter.acquire).toHaveBeenCalledTimes(starts) + expect(startsFor(HOST_TEST_SESSION)).toBe(1) + expect(startsFor(replacement)).toBe(0) + }) + + it("stops a running source's agent before it writes the marker", async () => { + const commit = store.commitConversationClear + let atCommit: { child: unknown; claim: string | undefined } | undefined + vi.spyOn(store, 'commitConversationClear').mockImplementationOnce(async (clear) => { + atCommit = { + child: host.collaboratorsForTests().sessions.get(HOST_TEST_SESSION)?.child ?? null, + claim: store.getRecord(HOST_TEST_SESSION)?.lease.claimStatus + } + return commit(clear) }) - expect(await host.conversationCommand(caller, commandParams('clear'))).toMatchObject({ - ok: true, - value: { - state: 'completed', - replacementSessionId: undefined, - // The refusal's message is Orca's log text; the result words its situation. - error: "Codex couldn't start. Start a new chat to continue.", - failure: { kind: 'startFailed', refusal: { code: 'structured_agent_session_unsupported' } } + await clearCommits() + expect(atCommit).toEqual({ child: null, claim: 'released' }) + }) + + it('founds one record per /clear through a chain of clears, starting neither', async () => { + const first = await clearCommits() + const second = await host.conversationCommand(caller, { + command: 'clear', + envelope: { + sessionId: first, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: store.getRecord(first)!.lease.runtimeFence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.conversationCommand', + sessionId: first, + fields: { command: 'clear' } + }) } }) + expect(second).toMatchObject({ ok: true, value: { replacementSessionId: expect.any(String) } }) + const last = second.ok ? second.value.replacementSessionId! : '' + expect(last).not.toBe(first) + expect(store.listRecords().map((record) => record.sessionId)).toEqual( + expect.arrayContaining([HOST_TEST_SESSION, first, last]) + ) + expect(store.listRecords()).toHaveLength(3) + expect(startsFor(first) + startsFor(last)).toBe(0) + expect(store.listVisibleSessionIds()).toEqual([last]) + expect(host.conversationReplacements()).toEqual([ + expect.objectContaining({ sourceSessionId: HOST_TEST_SESSION, sessionId: last }), + expect.objectContaining({ sourceSessionId: first, sessionId: last }) + ]) + }) + + it('writes nothing when its one write fails, leaving the source usable', async () => { + adapter.rewindSupport = () => ({ supported: true }) + vi.spyOn(store, 'commitConversationClear').mockRejectedValueOnce(new Error('disk full')) + await expect(host.conversationCommand(caller, commandParams('clear'))).rejects.toThrow( + 'disk full' + ) + expect(store.listRecords().map((record) => record.sessionId)).toEqual([HOST_TEST_SESSION]) + expect(store.getRecord(HOST_TEST_SESSION)?.conversationCommand).toBeUndefined() expect(store.listVisibleSessionIds()).toEqual([HOST_TEST_SESSION]) - expect(acquisitions).toBe(1) + // Past every conversation check: only the stale epoch it names stops it. + expect( + await host.rewind(caller, { + envelope: envelope('agentSession.rewind', { + itemId: 'item-1', + expectedEpoch: 'stale-epoch' + }), + itemId: 'item-1', + expectedEpoch: 'stale-epoch' + }) + ).toMatchObject({ ok: false, refusal: { rewindReason: 'stale-epoch' } }) expect(await host.conversationCommand(caller, commandParams('compact'))).toMatchObject({ ok: true }) + expect(await host.send(caller, sendParams('still here'))).toMatchObject({ ok: true }) }) - it('tells the user to run /clear again when the replacement could not start', async () => { - vi.mocked(adapter.acquire).mockRejectedValueOnce(new Error('spawn codex ENOENT')) - expect(await host.conversationCommand(caller, commandParams('clear'))).toMatchObject({ - ok: true, - value: { state: 'completed', error: "Codex couldn't start. Run /clear again." } - }) + it('reruns under the same operation id after its write failed, and commits once', async () => { + vi.spyOn(store, 'commitConversationClear').mockRejectedValueOnce(new Error('disk full')) + const params = commandParams('clear') + await expect(host.conversationCommand(caller, params)).rejects.toThrow('disk full') + const replacement = await clearCommits(params) + expect(store.listRecords()).toHaveLength(2) + expect(store.listVisibleSessionIds()).toEqual([replacement]) }) - it('keeps the situation a refused replacement start named', async () => { - vi.mocked(adapter.acquire).mockRejectedValueOnce( - new AgentSessionAcquisitionRefusal('Codex is not signed in.', 'notSignedIn') - ) - expect(await host.conversationCommand(caller, commandParams('clear'))).toMatchObject({ + it('replays a same-id resend with the one replacement it committed', async () => { + const params = commandParams('clear') + const replacement = await clearCommits(params) + expect(await host.conversationCommand(caller, params)).toMatchObject({ ok: true, - value: { - state: 'completed', - replacementSessionId: undefined, - // The next step is the command the user ran, not a message into the old conversation. - error: 'Codex is not signed in for the selected account. Sign in, then run /clear again.', - failure: { kind: 'notSignedIn' } - } + replayed: true, + value: { replacementSessionId: replacement } }) - expect(store.listVisibleSessionIds()).toEqual([HOST_TEST_SESSION]) + expect(store.listRecords()).toHaveLength(2) }) - it('runs a command whose fence the client has not caught up to', async () => { - const params = commandParams('compact') - params.envelope.expectedRuntimeFence++ - expect(await host.conversationCommand(caller, params)).toMatchObject({ ok: true }) - await vi.waitFor(() => expect(compact).toHaveBeenCalledTimes(1)) - }) it('reconstructs a committed replacement after the ledger settlement is lost', async () => { const persist = store.recordOperationOutcome.bind(store) vi.spyOn(store, 'recordOperationOutcome').mockImplementation(async (input) => { @@ -292,95 +434,176 @@ describe('host conversation commands', () => { }) const params = commandParams('clear') await expect(host.conversationCommand(caller, params)).rejects.toThrow('crash') + const [replacement] = store + .listRecords() + .flatMap((record) => (record.sessionId === HOST_TEST_SESSION ? [] : [record.sessionId])) expect(await host.conversationCommand(caller, params)).toMatchObject({ ok: true, replayed: true, - value: { state: 'completed' } + value: { state: 'completed', replacementSessionId: replacement } }) - expect(acquisitions).toBe(2) + expect(store.listRecords()).toHaveLength(2) + expect(acquisitions).toBe(1) + }) + + it("answers another window's /clear after the commit with the cleared refusal, never an in-flight one", async () => { + await clearCommits() + expect( + await host.conversationCommand({ callerKey: 'mobile' }, commandParams('clear')) + ).toMatchObject({ ok: false, refusal: { details: { reason: 'conversationCleared' } } }) + expect(store.listRecords()).toHaveLength(2) }) it('keeps explicitly revealed history and closed replacement tabs out of automatic restoration', async () => { - const result = await host.conversationCommand(caller, commandParams('clear')) - if (!result.ok) { - throw new Error('clear failed') - } + const replacement = await clearCommits() expect(host.conversationReplacements()).toHaveLength(1) await host.setSessionTabVisibility(HOST_TEST_SESSION, true) expect(host.conversationReplacements()).toEqual([]) await host.setSessionTabVisibility(HOST_TEST_SESSION, false) - await host.setSessionTabVisibility(result.value.replacementSessionId!, false) + await host.setSessionTabVisibility(replacement, false) expect(host.conversationReplacements()).toEqual([]) }) + + it('runs a command whose fence the client has not caught up to', async () => { + const params = commandParams('compact') + params.envelope.expectedRuntimeFence++ + expect(await host.conversationCommand(caller, params)).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(compact).toHaveBeenCalledTimes(1)) + }) }) -describe('a clear that never committed', () => { - /** The replacement's start answers with a refusal that proves nothing either way. */ - function refuseReplacementStartOnce() { - vi.spyOn(host, 'attach').mockResolvedValueOnce({ - ok: false, - refusal: { code: 'agent_session_operation_capacity', message: 'Too many operations.' } - }) - } +describe("the replacement's first send", () => { + it('starts a fresh conversation and delivers the message', async () => { + const replacement = await clearCommits() + const sent = await sendTo(replacement, 'first message') + expect(sent).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(adapter.dispatch).toHaveBeenCalledTimes(1)) + expect(startsFor(replacement)).toBe(1) + const [start] = vi + .mocked(adapter.acquire) + .mock.calls.flatMap(([input]) => (input.identity.sessionId === replacement ? [input] : [])) + // No provider conversation to resume: the source's thread is not carried over. + expect(start?.identity.providerHandle).toEqual({ + kind: 'opaque', + agent: 'codex', + value: 'pending' + }) + expect(start?.fence).toBe(2) + expect(store.getRecord(replacement)?.lease).toMatchObject({ claimStatus: 'live' }) + }) - async function clearCommits(params = commandParams('clear')) { - const result = await host.conversationCommand(caller, params) - expect(result).toMatchObject({ - ok: true, - value: { state: 'completed', replacementSessionId: expect.any(String) } + it('starts with the options the user chose, and adopts what the started child reports', async () => { + await store.replaceSessionOptions({ + sessionId: HOST_TEST_SESSION, + fence: store.getRecord(HOST_TEST_SESSION)!.lease.runtimeFence, + options: { model: 'test-model', effort: 'low' }, + now: HOST_TEST_NOW }) - return result.ok ? result.value.replacementSessionId! : '' - } - - it('refuses nothing afterwards: a rewind, a compaction and a send all run', async () => { - adapter.rewindSupport = () => ({ supported: true }) - refuseReplacementStartOnce() - await expect(host.conversationCommand(caller, commandParams('clear'))).rejects.toThrow( - 'Too many operations.' - ) - // Past every conversation check: only the stale epoch it names stops it. - expect( - await host.rewind(caller, { - envelope: envelope('agentSession.rewind', { - itemId: 'item-1', - expectedEpoch: 'stale-epoch' - }), - itemId: 'item-1', - expectedEpoch: 'stale-epoch' - }) - ).toMatchObject({ ok: false, refusal: { rewindReason: 'stale-epoch' } }) - expect(await host.conversationCommand(caller, commandParams('compact'))).toMatchObject({ - ok: true + adapter.readOptions = async () => ({ + models: [], + current: { model: 'test-model', effort: 'low', fastMode: false } }) - expect(await host.send(caller, sendParams('still here'))).toMatchObject({ ok: true }) + const replacement = await clearCommits() + expect(await sendTo(replacement, 'first message')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(startsFor(replacement)).toBe(1)) + const [start] = vi + .mocked(adapter.acquire) + .mock.calls.flatMap(([input]) => (input.identity.sessionId === replacement ? [input] : [])) + expect(start?.options).toEqual({ model: 'test-model', effort: 'low' }) + await vi.waitFor(() => + expect(store.getRecord(replacement)?.options).toEqual({ + model: 'test-model', + effort: 'low', + fastMode: 'false' + }) + ) }) - it('lets a clear under a new operation id commit', async () => { - refuseReplacementStartOnce() - await expect(host.conversationCommand(caller, commandParams('clear'))).rejects.toThrow() + it('applies an option picked on the replacement before its first send', async () => { const replacement = await clearCommits() - expect(store.listVisibleSessionIds()).toEqual([replacement]) + const picked = await host.setOption(caller, { + envelope: { + sessionId: replacement, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: store.getRecord(replacement)!.lease.runtimeFence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.setOption', + sessionId: replacement, + fields: { key: 'model', value: 'picked-model' } + }) + }, + key: 'model', + value: 'picked-model' + }) + expect(picked).toMatchObject({ ok: true }) + expect(startsFor(replacement)).toBe(0) + expect(await sendTo(replacement, 'first message')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(startsFor(replacement)).toBe(1)) + const [start] = vi + .mocked(adapter.acquire) + .mock.calls.flatMap(([input]) => (input.identity.sessionId === replacement ? [input] : [])) + expect(start?.options).toMatchObject({ model: 'picked-model' }) }) - // Its id is too old to start anything now, and it started nothing to finish. - it('lets a clear under a new operation id commit a day after a try that started nothing', async () => { - refuseReplacementStartOnce() - await expect(host.conversationCommand(caller, commandParams('clear'))).rejects.toThrow() - clock += AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + 60_000 - const params = commandParams('clear') - params.envelope.clientOperationId = `${clock}-${'a'.repeat(32)}` - await clearCommits(params) - }) + it.each([ + { + name: 'not signed in', + error: () => new AgentSessionAcquisitionRefusal('Codex is not signed in.', 'notSignedIn'), + words: + 'Codex is not signed in for the selected account. Sign in, then send your message again.', + kind: 'notSignedIn' + }, + { + name: 'a generic refusal', + error: () => new Error('spawn codex ENOENT'), + words: "Codex couldn't start.", + kind: 'startFailed' + } + ])( + 'fails on the message when the start fails ($name): one row, the message kept as not sent, and Retry starts it', + async ({ error, words, kind }) => { + const replacement = await clearCommits() + vi.mocked(adapter.acquire).mockRejectedValueOnce(error()) + const sent = await sendTo(replacement, 'first message') + expect(sent).toMatchObject({ ok: true }) + const clientMessageId = sent.ok ? sent.value.clientMessageId : '' + await vi.waitFor(async () => + expect(await submissionOf(replacement, clientMessageId)).toMatchObject({ + dispatchState: 'rejected', + rejection: { kind } + }) + ) + const rows = await errorRows(replacement) + expect(rows).toHaveLength(1) + expect(rows[0]).toContain(words) + expect(rows.join(' ')).not.toContain('/clear') + expect((await submissionOf(replacement, clientMessageId))?.reason).not.toContain('/clear') + expect(adapter.dispatch).not.toHaveBeenCalled() + + // Retry resends the same words, which starts the agent and delivers them. + expect(await sendTo(replacement, 'first message')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(adapter.dispatch).toHaveBeenCalledTimes(1)) + expect(startsFor(replacement)).toBe(2) + } + ) - it('reruns under the same operation id and starts exactly one replacement', async () => { - refuseReplacementStartOnce() - const params = commandParams('clear') - await expect(host.conversationCommand(caller, params)).rejects.toThrow() - expect(acquisitions).toBe(1) - await clearCommits(params) - expect(acquisitions).toBe(2) + it('starts after a restart between the /clear and the first send', async () => { + const replacement = await clearCommits() + await restartHost() + await host.restoreReadableSessions(store.listVisibleSessionIds()) + expect(store.getRecord(replacement)?.lease).toMatchObject({ + claimStatus: 'released', + handoffStage: null, + runtimeFence: 1 + }) + expect(host.conversationReplacements().map((entry) => entry.sessionId)).toEqual([replacement]) + expect(await sendTo(replacement, 'first message')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(adapter.dispatch).toHaveBeenCalledTimes(1)) + expect(startsFor(replacement)).toBe(1) }) +}) +describe('what an older build left', () => { /** What an older build left when its clear's outcome was lost: gated every write until now. */ async function restartOverAnOlderBuildsUnconfirmedClear() { const record = store.getRecord(HOST_TEST_SESSION)! @@ -406,289 +629,43 @@ describe('a clear that never committed', () => { await clearCommits() }) - /** A clear whose replacement started but whose commit never landed. */ - async function clearThatDiesBeforeItsCommit(params = commandParams('clear')): Promise { - const commit = store.setConversationCommand.bind(store) - let crashed = false - vi.spyOn(store, 'setConversationCommand').mockImplementation(async (...args) => { - if (!crashed && args[2].phase === 'committed' && args[2].replacementSessionId) { - crashed = true - throw new Error('crash before the commit') - } - return commit(...args) - }) - await expect(host.conversationCommand(caller, params)).rejects.toThrow( - 'crash before the commit' - ) - const [orphan] = store - .listRecords() - .flatMap((record) => (record.sessionId === HOST_TEST_SESSION ? [] : [record.sessionId])) - expect(host.collaboratorsForTests().sessions.get(orphan!)?.child).toBeTruthy() - return orphan! - } - - function otherRecordIds(): string[] { - return store - .listRecords() - .flatMap((record) => (record.sessionId === HOST_TEST_SESSION ? [] : [record.sessionId])) - } - - function startsFor(sessionId: string): number { - return vi - .mocked(adapter.acquire) - .mock.calls.filter(([input]) => input.identity.sessionId === sessionId).length - } - - function sendTo(sessionId: string, text: string) { - const body = hostTestMessage(text) - return host.send(caller, { - body, - envelope: { - sessionId, - clientOperationId: hostTestOperationId(), - expectedRuntimeFence: store.getRecord(sessionId)!.lease.runtimeFence, - payloadFingerprint: computeAgentSessionPayloadFingerprint({ - method: 'agentSession.send', - sessionId, - fields: { body } - }) - } - }) - } - - /** The chat reads cleared onto `replacement`, which starts its agent for its first message. */ - async function expectClearedOnto(replacement: string): Promise { - expect(store.listVisibleSessionIds()).toEqual([replacement]) - expect(host.conversationReplacements().map((entry) => entry.sessionId)).toEqual([replacement]) - expect(await host.send(caller, sendParams('into the old chat'))).toMatchObject({ - ok: false, - refusal: { details: { reason: 'conversationCleared' } } - }) - const started = startsFor(replacement) - expect(await sendTo(replacement, 'first message')).toMatchObject({ ok: true }) - await vi.waitFor(() => expect(startsFor(replacement)).toBe(started + 1)) - } - - // A restart stopped the replacement the first try started, which leaves it at rest like any - // quiet chat, so the retry switches to it. - it('retried under the same operation id after a crash, finishes onto the replacement it started', async () => { + // The build before this one started the replacement before committing, so a crash between the + // two left a started record nothing points at, and its start in the ledger. + it('founds a fresh replacement beside the orphan an older build started for the same operation id', async () => { const params = commandParams('clear') - const orphan = await clearThatDiesBeforeItsCommit(params) - await restartHost() - await host.restoreReadableSessions(store.listVisibleSessionIds()) - const result = await host.conversationCommand(caller, params) - expect(result).toMatchObject({ - ok: true, - value: { phase: 'committed', state: 'completed', replacementSessionId: orphan } - }) - expect(result.ok && result.value.error).toBeFalsy() - expect(otherRecordIds()).toEqual([orphan]) - expect(startsFor(orphan)).toBe(1) - await expectClearedOnto(orphan) - }) - - it('retried in the same app session once the idle sweep stopped the replacement, finishes onto it', async () => { - const params = commandParams('clear') - const attach = host.attach.bind(host) - // The replacement starts, but the answer the clear gets proves nothing either way. - vi.spyOn(host, 'attach').mockImplementationOnce(async (...args) => { - await attach(...args) - return { - ok: false, - refusal: { code: 'agent_session_operation_capacity', message: 'Too many operations.' } - } - }) - await expect(host.conversationCommand(caller, params)).rejects.toThrow('Too many operations.') - const [orphan] = otherRecordIds() - expect(store.getRecord(orphan!)?.lease.claimStatus).not.toBe('released') - clock += STRUCTURED_AGENT_SESSION_IDLE_MS + 1 - await host.collaboratorsForTests().lifetime.idleSweep.tick() - expect(store.getRecord(orphan!)?.lease).toMatchObject({ claimStatus: 'released' }) - const result = await host.conversationCommand(caller, params) - expect(result).toMatchObject({ - ok: true, - value: { phase: 'committed', state: 'completed', replacementSessionId: orphan } - }) - expect(result.ok && result.value.error).toBeFalsy() - expect(startsFor(orphan!)).toBe(1) - await expectClearedOnto(orphan!) - }) - - it('retried after the replacement definitely failed to start, still reads that failure', async () => { - vi.mocked(adapter.acquire).mockRejectedValueOnce( - new AgentSessionAcquisitionRefusal('Codex is not signed in.', 'notSignedIn') - ) - vi.spyOn(store, 'setConversationCommand').mockRejectedValueOnce( - new Error('crash before the commit') - ) - const params = commandParams('clear') - await expect(host.conversationCommand(caller, params)).rejects.toThrow( - 'crash before the commit' - ) - const [replacement] = otherRecordIds() - expect(store.getRecord(replacement!)?.lease).toMatchObject({ claimStatus: 'released' }) - expect(await host.conversationCommand(caller, params)).toMatchObject({ - ok: true, - value: { - state: 'completed', - replacementSessionId: undefined, - error: 'Codex is not signed in for the selected account. Sign in, then run /clear again.', - failure: { kind: 'notSignedIn' } + const { clientOperationId } = params.envelope + // The older build derived both ids from the operation, then crashed between start and commit. + const digest = createHash('sha256') + .update(JSON.stringify([HOST_TEST_SESSION, caller.callerKey, clientOperationId])) + .digest('hex') + const orphanStart = hostTestAttachParams(null, { + envelope: { + sessionId: `clear-${digest.slice(0, 40)}`, + clientOperationId: `${HOST_TEST_NOW}-${digest.slice(0, 32)}`, + expectedRuntimeFence: null, + payloadFingerprint: '' } }) - expect(startsFor(replacement!)).toBe(1) - expect(store.listVisibleSessionIds()).toEqual([HOST_TEST_SESSION]) - }) - - // The client mints a fresh operation id per press; the host still finds that press's earlier try. - it('retried under a fresh operation id while its replacement runs, finishes onto it and starts no other', async () => { - const orphan = await clearThatDiesBeforeItsCommit() - const result = await host.conversationCommand(caller, commandParams('clear')) - expect(result).toMatchObject({ - ok: true, - value: { phase: 'committed', state: 'completed', replacementSessionId: orphan } - }) - expect(result.ok && result.value.error).toBeFalsy() - expect(otherRecordIds()).toEqual([orphan]) - expect(startsFor(orphan)).toBe(1) - expect(store.listVisibleSessionIds()).toEqual([orphan]) - expect(host.conversationReplacements().map((entry) => entry.sessionId)).toEqual([orphan]) - expect(await sendTo(orphan, 'first message')).toMatchObject({ ok: true }) - }) - - it('retried under a fresh operation id after a restart, finishes onto the replacement it started', async () => { - const orphan = await clearThatDiesBeforeItsCommit() + const orphan = orphanStart.envelope.sessionId + notFromAClear.add(orphan) + expect( + await store.admitMutationOperation({ + callerKey: caller.callerKey, + envelope: params.envelope, + hostFingerprint: params.envelope.payloadFingerprint, + now: clock + }) + ).toMatchObject({ admission: { decision: 'admit' } }) + expect(await host.attach(caller, orphanStart)).toMatchObject({ ok: true }) await restartHost() await host.restoreReadableSessions(store.listVisibleSessionIds()) - const result = await host.conversationCommand(caller, commandParams('clear')) - expect(result).toMatchObject({ - ok: true, - value: { phase: 'committed', state: 'completed', replacementSessionId: orphan } - }) - expect(otherRecordIds()).toEqual([orphan]) - expect(startsFor(orphan)).toBe(1) - expect(store.getRecord(orphan)?.lease).toMatchObject({ - claimStatus: 'released', - ownerProcess: null - }) - await expectClearedOnto(orphan) - }) - it("refuses another window's /clear while this one's replacement runs, and runs it once that stops", async () => { - const otherWindow = { callerKey: 'mobile' } - const orphan = await clearThatDiesBeforeItsCommit() - expect(await host.conversationCommand(otherWindow, commandParams('clear'))).toMatchObject({ - ok: false, - refusal: { details: { reason: 'conversationCommandInFlight' } } - }) - expect(otherRecordIds()).toEqual([orphan]) - clock += STRUCTURED_AGENT_SESSION_IDLE_MS + 1 - await host.collaboratorsForTests().lifetime.idleSweep.tick() - const result = await host.conversationCommand(otherWindow, commandParams('clear')) - expect(result).toMatchObject({ - ok: true, - value: { state: 'completed', replacementSessionId: expect.any(String) } - }) - const replacement = result.ok ? result.value.replacementSessionId! : '' + const replacement = await clearCommits(params) expect(replacement).not.toBe(orphan) - // Nothing points at the first window's replacement, so nothing lists, opens or starts it. expect(store.listVisibleSessionIds()).toEqual([replacement]) expect(host.conversationReplacements().map((entry) => entry.sessionId)).toEqual([replacement]) expect(host.collaboratorsForTests().sessions.has(orphan)).toBe(false) - expect(store.getRecord(orphan)?.lease).toMatchObject({ claimStatus: 'released' }) expect(startsFor(orphan)).toBe(1) - }) - - function failNextReplacementStart() { - vi.mocked(adapter.acquire).mockRejectedValueOnce( - new AgentSessionAcquisitionRefusal('Codex is not signed in.', 'notSignedIn') - ) - } - - // The other window's failure is the conversation's latest; this window's own still ended its try. - it('starts a new replacement for a /clear after one that committed its failure, in either window', async () => { - for (const each of [caller, { callerKey: 'mobile' }]) { - failNextReplacementStart() - expect(await host.conversationCommand(each, commandParams('clear'))).toMatchObject({ - ok: true, - value: { replacementSessionId: undefined, failure: { kind: 'notSignedIn' } } - }) - } - const failed = otherRecordIds() - expect(failed).toHaveLength(2) - const replacement = await clearCommits() - expect(failed).not.toContain(replacement) - expect(startsFor(replacement)).toBe(1) - expect(store.listVisibleSessionIds()).toEqual([replacement]) - }) - - it('starts a new replacement after a committed failure whose ledger settlement was lost', async () => { - failNextReplacementStart() - const persist = store.recordOperationOutcome.bind(store) - vi.spyOn(store, 'recordOperationOutcome').mockImplementation(async (input) => { - if (input.outcome.status === 'succeeded' && input.outcome.conversationCommand) { - throw new Error('crash') - } - return persist(input) - }) - await expect(host.conversationCommand(caller, commandParams('clear'))).rejects.toThrow('crash') - vi.mocked(store.recordOperationOutcome).mockRestore() - const [failed] = otherRecordIds() - const replacement = await clearCommits() - expect(replacement).not.toBe(failed) - expect(store.listVisibleSessionIds()).toEqual([replacement]) - }) - - it("does not gate another window's /clear on a replacement whose stop is only unproven", async () => { - const orphan = await clearThatDiesBeforeItsCommit() - ownerProbe = { outcome: 'indeterminate', reason: 'test' } - await restartHost() - await host.restoreReadableSessions(store.listVisibleSessionIds()) - expect(store.getRecord(orphan)?.lease.claimStatus).not.toBe('released') - const result = await host.conversationCommand({ callerKey: 'mobile' }, commandParams('clear')) - expect(result).toMatchObject({ - ok: true, - value: { state: 'completed', replacementSessionId: expect.any(String) } - }) - expect(result.ok && result.value.replacementSessionId).not.toBe(orphan) - }) - - it('starts afresh when a retry under a new operation id follows a start that definitely failed', async () => { - failNextReplacementStart() - vi.spyOn(store, 'setConversationCommand').mockRejectedValueOnce( - new Error('crash before the commit') - ) - await expect(host.conversationCommand(caller, commandParams('clear'))).rejects.toThrow( - 'crash before the commit' - ) - const [failed] = otherRecordIds() - const result = await host.conversationCommand(caller, commandParams('clear')) - expect(result).toMatchObject({ - ok: true, - value: { state: 'completed', replacementSessionId: expect.any(String) } - }) - expect(result.ok && result.value.error).toBeFalsy() - expect(result.ok && result.value.replacementSessionId).not.toBe(failed) - }) - - it('starts afresh when a retry after a restart follows a try whose start never answered', async () => { - let hostDies!: (error: Error) => void - vi.mocked(adapter.acquire).mockImplementationOnce( - () => new Promise((_, reject) => (hostDies = reject)) - ) - const dying = host.conversationCommand(caller, commandParams('clear')).catch(() => {}) - await vi.waitFor(() => expect(otherRecordIds()).toHaveLength(1)) - const [interrupted] = otherRecordIds() - await restartHost() - await host.restoreReadableSessions(store.listVisibleSessionIds()) - const result = await host.conversationCommand(caller, commandParams('clear')) - expect(result).toMatchObject({ - ok: true, - value: { state: 'completed', replacementSessionId: expect.any(String) } - }) - expect(result.ok && result.value.error).toBeFalsy() - expect(result.ok && result.value.replacementSessionId).not.toBe(interrupted) - hostDies(new Error('the host that started it is gone')) - await dying + expect(startsFor(replacement)).toBe(0) }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command.ts index bbc7aea52fb7..4a825d6081d5 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-command.ts +++ b/src/main/native-chat/agent-session-wire/structured-conversation-command.ts @@ -1,6 +1,4 @@ -import { createHash } from 'node:crypto' -import { isDefinitiveAgentSessionCreateRefusal } from '../../../shared/agent-session-definitive-refusal' -import { parseAgentSessionOperationTimestamp } from '../../../shared/agent-session-host-authority' +import { randomBytes } from 'node:crypto' import type { AgentSessionConversationCommand, AgentSessionConversationCommandResult @@ -9,34 +7,17 @@ import type { AgentSessionMutationEnvelope, AgentSessionMutationResult } from '../../../shared/agent-session-wire' -import { - agentSessionLeaseAdmitsWriter, - agentSessionLeaseIsReleased -} from '../../../shared/agent-session-lease-adjudication' -import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' -import { - attachFingerprintFields, - type AgentSessionAttachParams -} from './structured-agent-session-attach' import { admitAndRunAgentSessionMutation } from './structured-agent-session-mutation-admission' import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' -import { - openWithAgent, - structuredAgentSessionFailureWordsContext -} from './structured-agent-session-send-preparation' +import { sendPreparation } from './structured-agent-session-send-preparation' import type { StructuredAgentSessionCaller } from './structured-agent-session-host-types' import type { StructuredAgentSessionHost } from './structured-agent-session-host' -import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' -import { - conversationCommandInFlight, - conversationCommandBlocked -} from './structured-conversation-command-admission' +import { conversationCommandBlocked } from './structured-conversation-command-admission' import type { AgentSessionFailureFact } from '../../../shared/agent-session-failure' import { agentSessionFailureWords, type AgentSessionFailureWordsContext } from '../../../shared/agent-session-failure-words' -import { structuredAgentSessionStartFailureFact } from './structured-agent-session-failure-text' import { carryQueuedMessagesToClearReplacement } from './structured-agent-session-queued-mutations' /** A command's `error` is the sentence its row shows. */ @@ -62,79 +43,13 @@ export type ConversationReplacement = { agent: 'claude' | 'codex' } -/** The replacement a /clear try names, and the operation id of its start. */ -function clearReplacementIds(sessionId: string, callerKey: string, operationId: string) { - const digest = createHash('sha256') - .update(JSON.stringify([sessionId, callerKey, operationId])) - .digest('hex') - return { - sessionId: `clear-${digest.slice(0, 40)}`, - attachOperationId: `${parseAgentSessionOperationTimestamp(operationId)}-${digest.slice(0, 32)}` - } -} - /** - * The try this caller's /clear finishes: the oldest since its last commit whose replacement start - * reached the ledger, else this one. Read from the ledger in admission order, so a retry under a - * fresh operation id finds the replacement an earlier try started. + * `/clear`: one write that points this conversation at a new, at-rest one and moves its tab there. + * The new conversation's first send starts its agent. */ -function clearTryToFinish( - store: AgentSessionRecordStore, - sessionId: string, - callerKey: string, - clearFingerprint: string, - operationId: string -): string { - const committed = store.getRecord(sessionId)?.conversationCommand - let earliest: string | null = null - for (const row of store.listOperationRows()) { - if (row.callerKey !== callerKey || row.fingerprint !== clearFingerprint) { - continue - } - // The record also names a commit whose ledger settlement a crash lost. - if ( - row.outcome.status === 'succeeded' || - (committed?.phase === 'committed' && - committed.callerKey === callerKey && - committed.operationId === row.operationId) - ) { - earliest = null - } else if (earliest === null) { - const start = store.getOperationRow( - callerKey, - clearReplacementIds(sessionId, callerKey, row.operationId).attachOperationId - ) - // Only a start that succeeded left a conversation to finish; replaying any other repeats it. - if (start?.outcome.status === 'succeeded') { - earliest = row.operationId - } - } - } - return earliest ?? operationId -} - -/** Another caller's uncommitted /clear holds a replacement whose agent is running. One whose stop - * is merely unproven gates nothing: only that caller's own start would ever settle it. */ -function otherCallersClearIsLive( - store: AgentSessionRecordStore, - sessionId: string, - callerKey: string, - clearFingerprint: string -): boolean { - return store.listOperationRows().some((row) => { - if (row.callerKey === callerKey || row.fingerprint !== clearFingerprint) { - return false - } - const replacement = store.getRecord( - clearReplacementIds(sessionId, row.callerKey, row.operationId).sessionId - ) - return replacement !== null && agentSessionLeaseAdmitsWriter(replacement.lease) - }) -} - export function runStructuredConversationCommand( context: StructuredAgentSessionMutationContext, - host: Pick, + host: Pick, caller: StructuredAgentSessionCaller, params: ConversationCommandParams ): Promise> { @@ -153,8 +68,8 @@ export function runStructuredConversationCommand( adapter: context.deps.adapter, callerKey: caller.callerKey, envelope, - // Only the provider can do this, so an agent at rest is started first. - prepareSession: openWithAgent(context, params.envelope), + // Starts the agent only to settle a rewind in doubt, as a send does; a /clear itself starts nothing. + prepareSession: sendPreparation(context, envelope), journal: () => context.sessions.get(sessionId)?.journal, publish: (journal) => context.publish(sessionId, journal), flushStreamedEvents: context.flushStreamedEvents, @@ -162,6 +77,8 @@ export function runStructuredConversationCommand( plan: { method: 'agentSession.conversationCommand', fields: { command }, + // Written to the conversation, not the agent, so whoever owns the agent does not matter. + conversationWrite: true, recoverUnknownFromDurableState: true, settledOutcome: (value) => ({ status: 'succeeded', sessionId, conversationCommand: value }), replay: (_ctx, outcome) => { @@ -171,117 +88,52 @@ export function runStructuredConversationCommand( const prior = matching() return prior?.phase === 'committed' ? prior : null }, - // Nothing the chat reads is written before the commit, so a clear with no committed answer - // changed nothing and runs again, finishing the replacement its earliest try started. - rerunWhenReplayMissing: () => command === 'clear', + // The commit is the only write, so a clear with no committed answer changed nothing. + rerunWhenReplayMissing: () => true, run: async (ctx) => { await host.flushStreamedEvents(sessionId) const record = store.getRecord(sessionId)! - const blocked = conversationCommandBlocked(ctx, record, context.readChildWork(sessionId)) + const blocked = conversationCommandBlocked( + ctx, + record, + context.readChildWork(sessionId), + context.sessions.get(sessionId)?.child ? undefined : 'at-rest' + ) if (blocked) { return { ok: false, refusal: blocked } } - let ids: ReturnType | undefined - if (command === 'clear') { - const clearFingerprint = computeAgentSessionPayloadFingerprint({ - method: 'agentSession.conversationCommand', - sessionId, - fields: { command } - }) - if (otherCallersClearIsLive(store, sessionId, caller.callerKey, clearFingerprint)) { - return { ok: false, refusal: conversationCommandInFlight() } - } - ids = clearReplacementIds( - sessionId, - caller.callerKey, - clearTryToFinish( - store, - sessionId, - caller.callerKey, - clearFingerprint, - clientOperationId - ) - ) - } - const replacementSessionId = ids?.sessionId - const base = { + // Stopped before the marker, so nothing the old agent does can land after the clear. The + // stop releases the lease, which moves its fence: the marker is written at the new one. + await context.stopAgent(sessionId) + const fence = store.getRecord(sessionId)!.lease.runtimeFence + const completed = { command, - runtimeFence: ctx.fence, + runtimeFence: fence, operationId: clientOperationId, callerKey: caller.callerKey, - ...(replacementSessionId ? { replacementSessionId } : {}) - } - if (ids) { - const attach: AgentSessionAttachParams = { - envelope: { - sessionId: ids.sessionId, - clientOperationId: ids.attachOperationId, - expectedRuntimeFence: null, - payloadFingerprint: '' - }, - location: record.location, - accountHome: record.accountHome, - provider: record.provider, - agent: record.provider, - runtimeKind: 'native', - launchArgs: record.launchArgs, - // The options the user chose, which any restart of this chat would replay too. - options: record.options - } - attach.envelope.payloadFingerprint = computeAgentSessionPayloadFingerprint({ - method: 'agentSession.attach', - sessionId: ids.sessionId, - fields: attachFingerprintFields(attach) - }) - const acquired = await host.attach(caller, attach) - const replacement = store.getRecord(ids.sessionId) - // An earlier try started this replacement and a restart or the idle sweep has since - // stopped it: attach can't replay a settled start, but the new conversation is at rest. - const startedAndAtRest = - replacement !== null && - agentSessionLeaseIsReleased(replacement.lease) && - store.getOperationRow(caller.callerKey, attach.envelope.clientOperationId)?.outcome - .status === 'succeeded' - if (!acquired.ok && !startedAndAtRest) { - if ( - !isDefinitiveAgentSessionCreateRefusal(acquired.refusal.code) && - replacement?.lease.claimStatus !== 'released' - ) { - throw new Error(acquired.refusal.message) - } - // The refusal's message is Orca's log text; the result keeps its situation instead. - const failed = { - ...base, - replacementSessionId: undefined, - phase: 'committed' as const, - state: 'completed' as const, - ...conversationCommandFailure( - structuredAgentSessionStartFailureFact({ - refusal: acquired.refusal, - newSession: true - }), - { ...structuredAgentSessionFailureWordsContext(record), command: 'clear' } - ) - } - await store.setConversationCommand(sessionId, ctx.fence, failed) - return { ok: true, value: failed } - } - // Carry the source's drafts to the replacement, the same for every client version: - // the cards stay visible where the user now is, and no text rides the wire. - // Bookkeeping — a failure is reported and never fails the clear. - await carryQueuedMessagesToClearReplacement(ctx, { - replacementSessionId: ids.sessionId, - replacementJournal: context.sessions.get(ids.sessionId)?.journal, - callerKey: caller.callerKey, - operationId: clientOperationId - }) - } - const completed = { - ...base, + // Only has to be new: the marker is what points at it, and a same-id resend replays it. + replacementSessionId: `clear-${randomBytes(20).toString('hex')}`, phase: 'committed' as const, state: 'completed' as const } - await store.setConversationCommand(sessionId, ctx.fence, completed) + await store.commitConversationClear({ + sessionId, + fence, + command: completed, + claimKeyId: context.deps.claimKeyId, + now: context.now() + }) + // Carry the source's drafts to the replacement, the same for every client version: + // the cards stay visible where the user now is, and no text rides the wire. + // Bookkeeping — a failure is reported and never fails the clear. + await carryQueuedMessagesToClearReplacement(ctx, { + replacementSessionId: completed.replacementSessionId, + // Opened under its own lock, as every open is. + openReplacementJournal: async () => + (await context.conversation(completed.replacementSessionId)).journal, + callerKey: caller.callerKey, + operationId: clientOperationId + }) return { ok: true, value: completed } } } diff --git a/src/main/runtime/agent-session-conversation-clear-commit.test.ts b/src/main/runtime/agent-session-conversation-clear-commit.test.ts new file mode 100644 index 000000000000..3e7c63a41c34 --- /dev/null +++ b/src/main/runtime/agent-session-conversation-clear-commit.test.ts @@ -0,0 +1,162 @@ +// A /clear's one store write: the conversation it continues in, founded at rest, and the marker +// on the cleared record that points there. Either both land, on disk too, or neither does. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication' +import { isAgentSessionRefusalError } from '../../shared/agent-session-wire-refusals' +import type { AgentSessionRecordStore } from './agent-session-record-store' +import { openTestAgentSessionRecordStore } from './agent-session-record-store-test-harness' + +const NOW = 1_800_000_000_000 +const SOURCE = 'session-alpha' +const REPLACEMENT = `clear-${'a'.repeat(40)}` +const INDETERMINATE: AgentSessionOwnerProbe = { outcome: 'indeterminate', reason: 'no answer' } + +let directory: string + +beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), 'orca-agent-session-clear-commit-')) +}) + +afterEach(async () => { + await rm(directory, { recursive: true, force: true }) +}) + +const open = () => openTestAgentSessionRecordStore(directory) + +async function storeWithSource(): Promise { + const store = await open() + await store.reserveOwner({ + sessionId: SOURCE, + location: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'folder' + }, + provider: 'claude', + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/dev/.claude-work' }, + launchArgs: ['--flag'], + options: { model: 'sonnet' }, + expectedFence: null, + spawnToken: 'spawn-a', + claimKeyId: 'key-1', + handoffOperationId: null, + probe: INDETERMINATE, + operation: { + callerKey: 'client-1', + operationId: `${NOW}-${'1'.repeat(32)}`, + fingerprint: 'fp' + }, + now: NOW + }) + await store.setSessionTabVisibility(SOURCE, true, 'tab-alpha') + return store +} + +function marker(fence: number, replacementSessionId = REPLACEMENT) { + return { + command: 'clear' as const, + state: 'completed' as const, + phase: 'committed' as const, + runtimeFence: fence, + operationId: `${NOW}-${'2'.repeat(32)}`, + callerKey: 'client-1', + replacementSessionId + } +} + +describe("a /clear's commit", () => { + it('founds the replacement at rest and points the cleared record at it, on disk', async () => { + const store = await storeWithSource() + const fence = store.getRecord(SOURCE)!.lease.runtimeFence + await store.commitConversationClear({ + sessionId: SOURCE, + fence, + command: marker(fence), + claimKeyId: 'key-1', + now: NOW + 5 + }) + + const reopened = await open() + const source = reopened.getRecord(SOURCE)! + expect(source.conversationCommand).toEqual(marker(fence)) + expect(reopened.getRecord(REPLACEMENT)).toEqual({ + schemaVersion: source.schemaVersion, + sessionId: REPLACEMENT, + location: source.location, + provider: 'claude', + accountHome: source.accountHome, + options: { model: 'sonnet' }, + launchArgs: ['--flag'], + providerHandleChain: [], + createdAt: NOW + 5, + updatedAt: NOW + 5, + lease: { + sessionId: REPLACEMENT, + runtimeKind: 'native', + runtimeFence: 1, + handoffStage: null, + provenHandleLinkId: null, + ownerProcess: null, + reservedSpawnToken: null, + leaseDeadlineAt: NOW + 5, + lastRenewedAt: NOW + 5, + handoffOperationId: null, + journalCheckpoint: null, + claimKeyId: 'key-1', + claimStatus: 'released', + // Loading marks every lease for this host's adjudication, as it does any record's. + unreconciled: true, + deathEvidence: null + } + }) + expect(reopened.getSessionTabId(REPLACEMENT)).toBe('tab-alpha') + expect(reopened.getSessionTabId(SOURCE)).toBeNull() + }) + + it('writes neither the replacement nor the marker when the commit is refused', async () => { + const store = await storeWithSource() + const fence = store.getRecord(SOURCE)!.lease.runtimeFence + // The source's lease moved after the clear read it: the marker's fenced write refuses. + const refused = await store + .commitConversationClear({ + sessionId: SOURCE, + fence: fence + 1, + command: marker(fence + 1), + claimKeyId: 'key-1', + now: NOW + }) + .catch((error: unknown) => error) + expect(isAgentSessionRefusalError(refused)).toBe(true) + + for (const each of [store, await open()]) { + expect(each.getRecord(REPLACEMENT)).toBeNull() + expect(each.getRecord(SOURCE)?.conversationCommand).toBeUndefined() + expect(each.getSessionTabId(SOURCE)).toBe('tab-alpha') + expect(each.listRecords().map((record) => record.sessionId)).toEqual([SOURCE]) + } + }) + + it('never overwrites a record already under the replacement id', async () => { + const store = await storeWithSource() + const fence = store.getRecord(SOURCE)!.lease.runtimeFence + const before = store.getRecord(SOURCE)! + const refused = await store + .commitConversationClear({ + sessionId: SOURCE, + fence, + command: marker(fence, SOURCE), + claimKeyId: 'key-1', + now: NOW + }) + .catch((error: unknown) => error) + expect(refused).toMatchObject({ + refusal: { code: 'agent_session_conflict', details: { reason: 'sessionExists' } } + }) + expect(store.getRecord(SOURCE)).toEqual(before) + }) +}) diff --git a/src/main/runtime/agent-session-conversation-command-record.ts b/src/main/runtime/agent-session-conversation-command-record.ts index 297b9dd34e73..4aefce22d577 100644 --- a/src/main/runtime/agent-session-conversation-command-record.ts +++ b/src/main/runtime/agent-session-conversation-command-record.ts @@ -1,6 +1,7 @@ import { agentSessionRefusalError } from '../../shared/agent-session-wire-refusals' import type { AgentSessionStoreState } from './agent-session-record-store-file' import { AgentSessionTabTable } from './agent-session-tab-table' +import { foundAgentSessionRecord } from './agent-session-record-founding' import type { AgentSessionConversationCommandRecord } from '../../shared/agent-session-conversation-command' export function commitConversationCommandRecord( @@ -26,3 +27,32 @@ export function commitConversationCommandRecord( state.sessionTabs.move(sessionId, command.replacementSessionId) } } + +export type AgentSessionConversationClear = { + sessionId: string + fence: number + command: AgentSessionConversationCommandRecord & { replacementSessionId: string } + claimKeyId: string + now: number +} + +/** A /clear as one write: the conversation it continues in, founded from the source's identity, + * and the marker pointing there. Neither can land without the other. */ +export function commitConversationClearRecord( + state: AgentSessionStoreState, + clear: AgentSessionConversationClear +): void { + const source = state.records.get(clear.sessionId) + const replacementId = clear.command.replacementSessionId + if (!source) { + throw agentSessionRefusalError('agent_session_checkpoint_stale', { reason: 'leaseMoved' }) + } + if (state.records.has(replacementId) || state.unreadableRecords.has(replacementId)) { + throw agentSessionRefusalError('agent_session_conflict', { reason: 'sessionExists' }) + } + state.records.set( + replacementId, + foundAgentSessionRecord({ ...source, sessionId: replacementId }, clear) + ) + commitConversationCommandRecord(state, clear.sessionId, clear.fence, clear.command) +} diff --git a/src/main/runtime/agent-session-record-founding.ts b/src/main/runtime/agent-session-record-founding.ts new file mode 100644 index 000000000000..d759a2461e0c --- /dev/null +++ b/src/main/runtime/agent-session-record-founding.ts @@ -0,0 +1,59 @@ +import { + AGENT_SESSION_RECORD_SCHEMA_VERSION, + type AgentSessionRecord +} from '../../shared/agent-session-record' + +/** Who a conversation's agent is and how it launches: the same whether it is created or founded. */ +export type AgentSessionRecordIdentity = Pick< + AgentSessionRecord, + 'sessionId' | 'location' | 'provider' | 'accountHome' | 'options' | 'launchArgs' +> + +export function agentSessionRecordIdentityFields( + identity: AgentSessionRecordIdentity, + now: number +) { + return { + schemaVersion: AGENT_SESSION_RECORD_SCHEMA_VERSION, + sessionId: identity.sessionId, + location: identity.location, + provider: identity.provider, + accountHome: identity.accountHome, + ...(identity.options ? { options: { ...identity.options } } : {}), + ...(identity.launchArgs ? { launchArgs: [...identity.launchArgs] } : {}), + createdAt: now, + updatedAt: now + } +} + +/** + * A conversation no agent has run yet, at rest: its first send starts one. The empty handle chain + * is what makes that start a fresh conversation rather than a resume. + */ +export function foundAgentSessionRecord( + identity: AgentSessionRecordIdentity, + lease: { claimKeyId: string; now: number } +): AgentSessionRecord { + return { + ...agentSessionRecordIdentityFields(identity, lease.now), + providerHandleChain: [], + lease: { + sessionId: identity.sessionId, + runtimeKind: 'native', + // Not 0: clients echo the fence as their expected fence, which the wire requires positive. + runtimeFence: 1, + handoffStage: null, + provenHandleLinkId: null, + ownerProcess: null, + reservedSpawnToken: null, + leaseDeadlineAt: lease.now, + lastRenewedAt: lease.now, + handoffOperationId: null, + journalCheckpoint: null, + claimKeyId: lease.claimKeyId, + claimStatus: 'released', + unreconciled: false, + deathEvidence: null + } + } +} diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index 300e48d370d9..9f1c5bf6e7f9 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -2,7 +2,11 @@ * journal database. */ import { agentSessionRefusalError } from '../../shared/agent-session-wire-refusals' -import { commitConversationCommandRecord } from './agent-session-conversation-command-record' +import { + commitConversationClearRecord, + commitConversationCommandRecord, + type AgentSessionConversationClear +} from './agent-session-conversation-command-record' import { setAgentSessionRecordConversationName } from './agent-session-record-conversation-name' import { @@ -152,6 +156,10 @@ export class AgentSessionRecordStore { ) } + /** A committed /clear and the at-rest conversation it continues in, in one write. */ + commitConversationClear = (clear: AgentSessionConversationClear): Promise => + this.transact((draft) => commitConversationClearRecord(draft, clear)) + /** Unfenced on purpose: the name is a durable note, so writing it never contends with the * writer lease. `null` clears it. */ setConversationName = (sessionId: string, name: string | null): Promise => diff --git a/src/main/runtime/agent-session-reservation-admission.ts b/src/main/runtime/agent-session-reservation-admission.ts index d86383ddae72..db5707d3b4d8 100644 --- a/src/main/runtime/agent-session-reservation-admission.ts +++ b/src/main/runtime/agent-session-reservation-admission.ts @@ -25,7 +25,6 @@ import { type AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication' import { - AGENT_SESSION_RECORD_SCHEMA_VERSION, agentSessionExecutionLocationsEqual, isAgentSessionLaunchEnv, isAgentSessionOptions, @@ -47,6 +46,7 @@ import { type AgentSessionReservation } from './agent-session-lease-transitions' import type { AgentSessionStoreState } from './agent-session-record-store-file' +import { agentSessionRecordIdentityFields } from './agent-session-record-founding' export type AgentSessionReserveRequest = { sessionId: string @@ -278,18 +278,10 @@ function createAgentSessionRecord( reservation: AgentSessionReservation ): AgentSessionRecord { return { - schemaVersion: AGENT_SESSION_RECORD_SCHEMA_VERSION, - sessionId: request.sessionId, - location: request.location, - provider: request.provider, + ...agentSessionRecordIdentityFields(request, request.now), // Fence 1 below is this record's first, and the owner probe requires the head link to carry the // record's current fence — so an adopted link must be minted at that same fence. providerHandleChain: request.adoptedHandleLink ? [request.adoptedHandleLink] : [], - accountHome: request.accountHome, - ...(request.options ? { options: { ...request.options } } : {}), - ...(request.launchArgs ? { launchArgs: [...request.launchArgs] } : {}), - createdAt: request.now, - updatedAt: request.now, lease: { sessionId: request.sessionId, runtimeKind: 'native', diff --git a/src/main/runtime/claude-structured-startup-unanswered-control-request.test.ts b/src/main/runtime/claude-structured-startup-unanswered-control-request.test.ts index ce26647ae85f..56ba00618836 100644 --- a/src/main/runtime/claude-structured-startup-unanswered-control-request.test.ts +++ b/src/main/runtime/claude-structured-startup-unanswered-control-request.test.ts @@ -6,7 +6,6 @@ // the saved choice stays saved for the next start to retry. Against // the production runtime, adapter, record store and host, with only the CLI process scripted. -import { createHash } from 'node:crypto' import { afterEach, describe, expect, it, vi } from 'vitest' import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' import { claudeSessionIdForOrcaSession } from '../claude/claude-structured-launch-resolution' @@ -79,16 +78,24 @@ async function statusRows(host: StructuredAgentSessionHost): Promise { } async function send(host: StructuredAgentSessionHost, text: string): Promise { + await sendTo(host, SESSION, text) +} + +async function sendTo( + host: StructuredAgentSessionHost, + sessionId: string, + text: string +): Promise { const body = hostTestMessage(text) await expect( host.send(CALLER, { envelope: { - sessionId: SESSION, - clientOperationId: `${Date.now()}-${(++operations).toString(16).padStart(32, '0')}`, - expectedRuntimeFence: record(host)?.lease.runtimeFence ?? 0, + sessionId, + clientOperationId: operationId(), + expectedRuntimeFence: host.deps.store.getRecord(sessionId)?.lease.runtimeFence ?? 0, payloadFingerprint: computeAgentSessionPayloadFingerprint({ method: 'agentSession.send', - sessionId: SESSION, + sessionId, fields: { body } }) }, @@ -149,12 +156,8 @@ describe('a Claude start whose CLI answers initialize but not a control request' }) it('replays the unanswered saved model after a turn reports another model, another option changes, and the chat is cleared', async () => { - const clearOperation = operationId() - const replacement = `clear-${createHash('sha256') - .update(JSON.stringify([SESSION, CALLER.callerKey, clearOperation])) - .digest('hex') - .slice(0, 40)}` - claude = createScriptedClaudeRuntime([SESSION, replacement]) + const sessions = [SESSION] + claude = createScriptedClaudeRuntime(sessions) const behavior = { optionWritesHang: true, controlTimeoutMs: DEADLINE_MS } claude.behave(SESSION, behavior) vi.spyOn(console, 'warn').mockImplementation(() => {}) @@ -187,7 +190,7 @@ describe('a Claude start whose CLI answers initialize but not a control request' command: 'clear', envelope: { sessionId: SESSION, - clientOperationId: clearOperation, + clientOperationId: operationId(), expectedRuntimeFence: record(host)?.lease.runtimeFence ?? 0, payloadFingerprint: computeAgentSessionPayloadFingerprint({ method: 'agentSession.conversationCommand', @@ -198,10 +201,20 @@ describe('a Claude start whose CLI answers initialize but not a control request' }) expect(cleared, JSON.stringify(cleared)).toMatchObject({ ok: true, - value: { replacementSessionId: replacement } + value: { replacementSessionId: expect.any(String) } }) - // The cleared chat's start replays the saved model rather than the one the turn reported. + const replacement = cleared.ok ? cleared.value.replacementSessionId! : '' + sessions.push(replacement) + // The cleared chat starts nothing until its first message. + expect(claude.children(replacement)).toEqual([]) + await sendTo(host, replacement, 'first message') + // Started fresh under its own derived id, and it replays the saved model rather than the one + // the turn reported. await vi.waitFor(() => expect(claude.child(replacement).calls).toContain('set_model')) + expect(claude.child(replacement).launch.options).toMatchObject({ + sessionId: claudeSessionIdForOrcaSession(replacement) + }) + expect(claude.child(replacement).launch.options.resume).toBeUndefined() await vi.waitFor(() => expect(host.deps.store.getRecord(replacement)?.options).toEqual({ model: 'sonnet', diff --git a/src/main/runtime/rpc/methods/structured-chat-tab-table.test.ts b/src/main/runtime/rpc/methods/structured-chat-tab-table.test.ts index 42e7101e9484..e8f7108f9f99 100644 --- a/src/main/runtime/rpc/methods/structured-chat-tab-table.test.ts +++ b/src/main/runtime/rpc/methods/structured-chat-tab-table.test.ts @@ -8,6 +8,7 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { computeAgentSessionPayloadFingerprint } from '../../../../shared/agent-session-mutation-envelope' +import type { AgentSessionStatusSummary } from '../../../../shared/agent-session-wire' import { CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY, STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY @@ -261,9 +262,22 @@ describe('a chat tab across /clear', () => { expect(await send(replacement, 'still here')).toMatchObject({ ok: true, result: { ok: true } }) }) + // A worktree delete closes every chat in it; one that never started must not ask to be forced. + it('closes a replacement that never started as settled', async () => { + await createChat(HOST_TEST_SESSION) + const replacement = await clear(HOST_TEST_SESSION) + expect(await closeStructuredAgentSessionChild(replacement)).toEqual({ + stopped: true, + closeAttempted: true + }) + }) + it('puts a cleared chat back under the tab id it had when its close does not land', async () => { await createChat(HOST_TEST_SESSION) const replacement = await clear(HOST_TEST_SESSION) + // Its agent starts on its first message; only a running agent has a close that can fail. + expect(await send(replacement, 'first message')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(store.getRecord(replacement)?.lease.claimStatus).toBe('live')) closeSession.mockResolvedValue(false) const outcome = await closeStructuredAgentSessionChild(replacement) @@ -286,6 +300,86 @@ describe('a chat tab across /clear', () => { }) }) +describe('other clients after a /clear from this window', () => { + const STRUCTURED = [ + STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY, + CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY + ] + const clients = { + window: { clientKind: 'runtime' as const, clientCapabilities: STRUCTURED }, + phone: { clientKind: 'mobile' as const, clientCapabilities: STRUCTURED } + } + + /** The chat tabs a client lists, as that client is sent them. */ + async function listedChats(context: RpcDispatchStreamingOptions) { + const response = await dispatcher.dispatch( + { + id: 'list', + authToken: 'token', + method: 'session.tabs.list', + params: { worktree: WORKTREE } + }, + context + ) + const listed: { result?: { tabs?: { type: string }[] } } = JSON.parse(JSON.stringify(response)) + return (listed.result?.tabs ?? []).filter((tab) => tab.type === 'agent-session') + } + + /** The session list each client keeps: the latest status the host published per session. */ + function sessionList(id: string): Map { + const latest = new Map() + host.subscribeStatus({ + id, + emit: (event) => { + if (event.type === 'snapshot') { + event.sessions.forEach((session) => latest.set(session.sessionId, session)) + } else if (event.type === 'status') { + latest.set(event.session.sessionId, event.session) + } + } + }) + return latest + } + + it('shows another window and a phone the replacement the marker names, with nothing unread', async () => { + await createChat(HOST_TEST_SESSION) + const lists = { window: sessionList('window-b'), phone: sessionList('phone') } + const replacement = await clear(HOST_TEST_SESSION) + // The stored pointer is the only source of the id: nothing a client holds can derive it. + expect(store.getRecord(HOST_TEST_SESSION)?.conversationCommand).toMatchObject({ + phase: 'committed', + replacementSessionId: replacement + }) + expect(store.getSessionTabId(replacement)).toBe(SOURCE_TAB) + + for (const context of Object.values(clients)) { + expect(await listedChats(context)).toEqual([ + expect.objectContaining({ sessionId: replacement, replacesSessionId: HOST_TEST_SESSION }) + ]) + } + // Never started, so no client lists a turn for it to be unread. + for (const list of Object.values(lists)) { + expect(list.get(replacement)?.status ?? null).toBeNull() + expect(list.get(replacement)?.turnOutcome).toBeUndefined() + } + + expect(await send(replacement, 'first message')).toMatchObject({ + ok: true, + result: { ok: true } + }) + await vi.waitFor(() => { + for (const list of Object.values(lists)) { + expect(list.get(replacement)).toMatchObject({ sessionId: replacement, status: 'working' }) + } + }) + for (const context of Object.values(clients)) { + expect(await listedChats(context)).toEqual([ + expect.objectContaining({ sessionId: replacement }) + ]) + } + }) +}) + describe('session tab mutations from other clients, unchanged by the table', () => { it('reorders a group holding a chat for a paired client', async () => { await createChat(HOST_TEST_SESSION) diff --git a/src/main/runtime/structured-chat-coordinator-fake-codex-fixture.ts b/src/main/runtime/structured-chat-coordinator-fake-codex-fixture.ts index d5532dbc25c2..05f5c064ab50 100644 --- a/src/main/runtime/structured-chat-coordinator-fake-codex-fixture.ts +++ b/src/main/runtime/structured-chat-coordinator-fake-codex-fixture.ts @@ -17,6 +17,8 @@ export type FakeConnection = Omit & { closed: boolean handlers: CodexAppServerConnectionHandlers threadId: string | null + /** Every JSON-RPC method the adapter called, in order. */ + methods: string[] turns: { clientUserMessageId: string; text: string }[] } @@ -57,10 +59,12 @@ export function fakeCodex() { const connection: FakeConnection = { handlers, threadId: null, + methods: [], turns: [], pid: 4321, closed: false, request: async (method, params) => { + connection.methods.push(method) const input = isRecord(params) ? params : {} if (method === 'thread/start') { connection.threadId = `thread-${connections.length}` diff --git a/src/main/runtime/structured-chat-coordinator-mail.test.ts b/src/main/runtime/structured-chat-coordinator-mail.test.ts index 4ffbec11118e..e9606ea90cb1 100644 --- a/src/main/runtime/structured-chat-coordinator-mail.test.ts +++ b/src/main/runtime/structured-chat-coordinator-mail.test.ts @@ -20,6 +20,7 @@ import type { StructuredAgentSessionHost } from '../native-chat/agent-session-wi import type { AgentSessionJournal } from '../native-chat/agent-session-journal/journal-store' import { AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS } from '../../shared/agent-session-host-authority' import { refuse } from '../../shared/agent-session-wire-refusals' +import { agentSessionProviderHandleChainHead } from '../../shared/agent-session-provider-handle' import { OrcaRuntimeService } from './orca-runtime' import { OrchestrationDb } from './orchestration/db' import { localOrchestrationCliCommand } from './orchestration/cli-command' @@ -97,9 +98,13 @@ async function openChat(sessionId: string): Promise { const threadBySession = new Map() function connectionFor(sessionId: string): FakeConnection { - const connection = codex.connections.findLast( - (candidate) => candidate.threadId === threadBySession.get(sessionId) + // A cleared chat's successor starts on its first message; its record then names its thread. + const head = agentSessionProviderHandleChainHead( + host.deps.store.getRecord(sessionId)?.providerHandleChain ?? [] ) + const thread = + threadBySession.get(sessionId) ?? (head?.handle.provider === 'codex' && head.handle.threadId) + const connection = codex.connections.findLast((candidate) => candidate.threadId === thread) if (!connection) { throw new Error(`no app-server for ${sessionId}`) } @@ -233,10 +238,16 @@ async function clearChat(sessionId: string): Promise { // The surface swaps the tab over to the session that continues the chat. await host.setSessionTabVisibility(sessionId, false) await host.setSessionTabVisibility(successor, true) - threadBySession.set(successor, codex.connections.at(-1)!.threadId!) return successor } +/** A cleared chat's successor runs once the user writes to it; only then can its agent act. */ +async function startSuccessor(successor: string): Promise { + expect(await sendUserMessage(successor, 'hello')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(connectionFor(successor).turns).toHaveLength(1), WAIT) + await settleTurn(successor, 0) +} + beforeEach(async () => { resetProviderFaults() root = await mkdtemp(join(tmpdir(), 'orca-structured-coordinator-mail-')) @@ -743,15 +754,19 @@ describe('a /clear keeps the chat its orchestration address', () => { const { runId, taskId } = await coordinatorRunAndTask() const generation = db.getRunRaw(runId)!.consumer_generation const successor = await clearChat(COORDINATOR) - const next = connectionFor(successor) + const opened = codex.connections.length - await expect( - call('orchestration.runCurrent', {}, { sessionId: successor }) - ).resolves.toMatchObject({ run: { id: runId } }) + // The worker's result is the successor's first message, which starts it on a new thread. await finishWorker(taskId) - await vi.waitFor(() => expect(next.turns).toHaveLength(1), WAIT) + await vi.waitFor(() => expect(connectionFor(successor).turns).toHaveLength(1), WAIT) + const next = connectionFor(successor) + expect(codex.connections.slice(opened)).toEqual([next]) + expect(next.methods.filter((method) => method.startsWith('thread/'))).toEqual(['thread/start']) expect(next.turns[0]!.text).toMatch(POINTER) await settleTurn(successor, 0) + await expect( + call('orchestration.runCurrent', {}, { sessionId: successor }) + ).resolves.toMatchObject({ run: { id: runId } }) await expect(call('orchestration.check', {}, { sessionId: successor })).resolves.toMatchObject({ runId, count: 1, @@ -764,9 +779,31 @@ describe('a /clear keeps the chat its orchestration address', () => { }) }) + it('points mail the cleared chat never took at the successor, with no message from the user', async () => { + const chat = await openChat(COORDINATOR) + const { runId, taskId } = await coordinatorRunAndTask() + await finishWorker(taskId) + await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT) + // The provider dies before it runs the pointer, so the mail is still unpointed at the /clear. + chat.handlers.onExit?.(new Error('provider died before the echo')) + await vi.waitFor(() => + expect(host.deps.store.getRecord(COORDINATOR)?.lease.claimStatus).toBe('released') + ) + expect(db.getUndeliveredUnreadMessages(`run:${runId}`, undefined, {})).toHaveLength(1) + + const successor = await clearChat(COORDINATOR) + await vi.waitFor(() => expect(connectionFor(successor).turns[0]?.text).toMatch(POINTER), WAIT) + await settleTurn(successor, 0) + await vi.waitFor( + () => expect(db.getUndeliveredUnreadMessages(`run:${runId}`, undefined, {})).toEqual([]), + WAIT + ) + }) + it("stores the successor's own Run under the conversation's address, through a chain of clears", async () => { await openChat(COORDINATOR) const middle = await clearChat(COORDINATOR) + await startSuccessor(middle) const created = await call( 'orchestration.runCreate', { objective: 'next' }, @@ -775,7 +812,7 @@ describe('a /clear keeps the chat its orchestration address', () => { const runId = idOf(created.run) expect(db.getRunRaw(runId)!.coordinator_orca_session_id).toBe(COORDINATOR) const successor = await clearChat(middle) - runtime.onStructuredSessionStatusForMail({ sessionId: successor, status: 'idle' }) + await startSuccessor(successor) await expect( call('orchestration.runCurrent', {}, { sessionId: successor }) @@ -798,6 +835,7 @@ describe('a /clear keeps the chat its orchestration address', () => { currentRunCoordinatorOrcaSessionId(db.getRunRaw(runId)!) await openChat(COORDINATOR) const middle = await clearChat(COORDINATOR) + await startSuccessor(middle) const first = idOf( (await call('orchestration.runCreate', { objective: 'first' }, { sessionId: middle })).run ) @@ -812,6 +850,7 @@ describe('a /clear keeps the chat its orchestration address', () => { expect(boundOrcaSessionId(first)).toBeNull() const successor = await clearChat(middle) + await startSuccessor(successor) await call('orchestration.runUse', { id: first }, { sessionId: successor }) const rebound = db.getRunRaw(first)! expect(rebound.coordinator_orca_session_id).toBe(root) @@ -826,8 +865,7 @@ describe('a /clear keeps the chat its orchestration address', () => { await openChat(PEER_CHAT) const middle = await clearChat(PEER_CHAT) const successor = await clearChat(middle) - const next = connectionFor(successor) - + // The first ping starts the live session, which neither clear did. for (const [index, spelling] of [PEER_CHAT, middle, successor].entries()) { const sent = await call('orchestration.send', { from: 'term_worker', @@ -835,7 +873,7 @@ describe('a /clear keeps the chat its orchestration address', () => { subject: `ping ${index}` }) expect(sent).toMatchObject({ message: { to_handle: `session:${PEER_CHAT}` } }) - await vi.waitFor(() => expect(next.turns).toHaveLength(index + 1), WAIT) + await vi.waitFor(() => expect(connectionFor(successor).turns).toHaveLength(index + 1), WAIT) await settleTurn(successor, index) } await expect(call('orchestration.check', {}, { sessionId: successor })).resolves.toMatchObject({ diff --git a/src/main/runtime/structured-claude-runtime-adapter.ts b/src/main/runtime/structured-claude-runtime-adapter.ts index d6aa6a7278eb..9024c65b503d 100644 --- a/src/main/runtime/structured-claude-runtime-adapter.ts +++ b/src/main/runtime/structured-claude-runtime-adapter.ts @@ -16,6 +16,7 @@ import { } from './agent-session-provider-handle-transition' import type { ClaudeManagedAccountGateSettings } from '../native-chat/claude-structured-managed-account-support' import type { AgentSessionRecordStore } from './agent-session-record-store' +import { ClaudeAtRestCommandCatalog } from '../claude/claude-at-rest-commands' export type StructuredClaudeRuntimeAdapterDeps = { store: AgentSessionRecordStore @@ -73,6 +74,9 @@ export function createStructuredClaudeRuntimeAdapter( ): ClaudeStructuredSessionAdapter { const { store } = deps return new ClaudeStructuredSessionAdapter({ + atRestCommands: new ClaudeAtRestCommandCatalog({ + resolveWorkspacePath: deps.resolveWorkspacePath + }), resolveLaunch: createClaudeStructuredLaunchResolver({ store, resolveWorkspacePath: deps.resolveWorkspacePath, diff --git a/src/shared/agent-session-failure-words.test.ts b/src/shared/agent-session-failure-words.test.ts index 60f5504c956e..98273628214d 100644 --- a/src/shared/agent-session-failure-words.test.ts +++ b/src/shared/agent-session-failure-words.test.ts @@ -107,19 +107,6 @@ describe('the words written beside a failure fact', () => { } }) - it('names /clear as the next step for the start a /clear needed', () => { - const clear = (kind: AgentSessionFailureKind) => - agentSessionFailureSentence({ kind }, 'row', { agentName: 'Codex', command: 'clear' }) - expect(clear('notSignedIn')).toBe( - 'Codex is not signed in for the selected account. Sign in, then run /clear again.' - ) - expect(clear('startFailed')).toBe("Codex couldn't start. Run /clear again.") - expect(clear('restartFailed')).toBe("Codex couldn't restart. Run /clear again.") - expect(clear('providerStartFailed')).toBe( - 'Codex stopped before it finished starting. Run /clear again.' - ) - }) - it('names /compact as the next step for the start a /compact needed', () => { const compact = (kind: AgentSessionFailureKind) => agentSessionFailureSentence({ kind }, 'rejection', {