diff --git a/cli/src/codex/conversationHistory.test.ts b/cli/src/codex/conversationHistory.test.ts index c3041afe..ab85b5fe 100644 --- a/cli/src/codex/conversationHistory.test.ts +++ b/cli/src/codex/conversationHistory.test.ts @@ -24,6 +24,61 @@ function createClient(overrides?: { } describe('CodexConversationHistory', () => { + it.each([ + { name: 'steered message', localId: 'steer', turns: [ + { id: 'a', status: 'completed', items: [{ type: 'userMessage', clientId: 'first' }, { type: 'userMessage', clientId: 'steer' }] } + ] }, + { name: 'unknown turn status', localId: 'first', turns: [ + { id: 'a', items: [{ type: 'userMessage', clientId: 'first' }] } + ] }, + { name: 'incomplete prefix', localId: 'second', turns: [ + { id: 'a', status: 'inProgress', items: [{ type: 'userMessage', clientId: 'first' }] }, + { id: 'b', status: 'completed', items: [{ type: 'userMessage', clientId: 'second' }] } + ] }, + { name: 'duplicate client id', localId: 'first', turns: [ + { id: 'a', status: 'completed', items: [{ type: 'userMessage', clientId: 'first' }] }, + { id: 'b', status: 'completed', items: [{ type: 'userMessage', clientId: 'first' }] } + ] } + ])('rejects $name before native mutation while busy', async ({ localId, turns }) => { + const fork = vi.fn(async () => ({ thread: { id: 'child' } })) + const history = new CodexConversationHistory(() => createClient({ fork, + read: async () => ({ thread: { id: 'thread-1', turns } }) }) as never) + history.setThreadId('thread-1') + history.setBusy(true) + await expect(history.fork(localId)).rejects.toThrow() + expect(fork).not.toHaveBeenCalled() + }) + + it('uses an exclusive boundary for the first completed turn while busy', async () => { + const fork = vi.fn(async () => ({ thread: { id: 'child' } })) + const history = new CodexConversationHistory(() => createClient({ fork, + read: async () => ({ thread: { id: 'thread-1', turns: [ + { id: 'first', status: 'completed', items: [{ type: 'userMessage', clientId: 'first' }] } + ] } }) }) as never) + history.setThreadId('thread-1') + history.setBusy(true) + await history.fork('first') + expect(fork).toHaveBeenCalledWith({ threadId: 'thread-1', beforeTurnId: 'first' }) + }) + it('forks completed history while a later native turn is running', async () => { + const fork = vi.fn(async () => ({ thread: { id: 'child' } })) + const history = new CodexConversationHistory(() => createClient({ fork, read: async () => ({ thread: { + id: 'thread-1', turns: [ + { id: 'turn-a', status: 'completed', items: [{ type: 'userMessage', clientId: 'local-a' }] }, + { id: 'turn-b', status: 'completed', items: [{ type: 'userMessage', clientId: 'local-b' }] }, + { id: 'turn-c', status: 'inProgress', items: [{ type: 'userMessage', clientId: 'local-c' }] } + ] + } }) }) as never) + history.setThreadId('thread-1') + history.setBusy(true) + await expect(history.fork('local-b')).resolves.toEqual({ nativeSessionId: 'child' }) + expect(fork).toHaveBeenCalledWith({ threadId: 'thread-1', lastTurnId: 'turn-a' }) + await expect(history.fork()).rejects.toThrow('busy') + await expect(history.rewind('local-b')).rejects.toThrow('busy') + await expect(history.fork('local-c')).rejects.toThrow('completed') + expect(fork).toHaveBeenCalledTimes(1) + }) + it('only publishes methods confirmed by the app server', async () => { const supportsMethod = vi.fn(async (method: string) => method === 'thread/fork') const history = new CodexConversationHistory(() => ({ diff --git a/cli/src/codex/conversationHistory.ts b/cli/src/codex/conversationHistory.ts index 72d2b9b0..bbb2fab1 100644 --- a/cli/src/codex/conversationHistory.ts +++ b/cli/src/codex/conversationHistory.ts @@ -112,7 +112,7 @@ export class CodexConversationHistory { } async fork(messageLocalId?: string): Promise { - if (this.busy) throw new Error('Session is busy') + if (this.busy && !messageLocalId) throw new Error('Session is busy') const client = this.getClient() const threadId = this.threadId if (!client || !threadId) throw new Error('Codex thread is not ready') @@ -130,6 +130,23 @@ export class CodexConversationHistory { if (selectedIndex < 0) { throw new Error('Selected turn not found') } + const selected = turns[selectedIndex]! + const running = this.busy || turns.some(turn => turn.status === 'inProgress' || turn.status === 'in_progress') + if (selected.status === 'inProgress' || selected.status === 'in_progress' + || (running && !['completed', 'interrupted', 'failed'].includes(selected.status ?? ''))) { + throw new Error('Historical fork requires a completed turn') + } + // Excluding a native turn must also exclude its first HAPI user + // message. A steered message inside that turn is not a safe cutoff. + if (running && (selected.clientIds[0] !== messageLocalId || !selected.hasCompleteItems + || selected.userMessageCount !== selected.clientIds.length + || turns.flatMap(turn => turn.clientIds).filter(id => id === messageLocalId).length !== 1)) { + throw new Error('Cannot cut inside an ambiguous native turn. Select its first message.') + } + if (running && turns.slice(0, selectedIndex).some(turn => + !['completed', 'interrupted', 'failed'].includes(turn.status ?? ''))) { + throw new Error('Historical fork requires a completed history prefix') + } // Prefer stable inclusive lastTurnId of the previous turn. The first // turn has no predecessor, so fall back to experimental beforeTurnId // (exclusive) for that single boundary. diff --git a/cli/src/codex/shared/root.test.ts b/cli/src/codex/shared/root.test.ts index 711e3c87..7e88d0f9 100644 --- a/cli/src/codex/shared/root.test.ts +++ b/cli/src/codex/shared/root.test.ts @@ -53,7 +53,7 @@ afterEach(async () => { finally { vi.useRealTimers(); } }); -async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end'] }) { +async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end']; create?: RootHost['create'] }) { const directory = await mkdtemp('/tmp/hapi-shared-root-'); let state: AgentState = { steeringActive: true }; let metadata: Metadata = { path: directory, host: 'test', flavor: 'codex' }; @@ -82,7 +82,7 @@ async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end'] }) const end = opts?.end ?? (async () => { throw new Error('Unexpected root archive'); }); const root = new SharedCodexRoot({ session, workingDirectory: directory } as SessionBootstrapResult, { directory, generation: 'test', endpoint: 'mock', settingsFor: () => undefined, - create: async () => { throw new Error('Unexpected root creation'); }, + create: opts?.create ?? (async () => { throw new Error('Unexpected root creation'); }), end } satisfies RootHost); cleanups.push(async () => { await root.close(false); await rm(directory, { recursive: true, force: true }); }); @@ -119,6 +119,47 @@ async function startCompaction(f: Awaited>, id = 'inp return turn; } +describe('historical fork during generation', () => { + it('rejects steered and unknown-status boundaries before creating a child', async () => { + const create = vi.fn() + const f = await fixture({ create }) + f.native.thread.turns = [ + { id: 'earlier', status: 'completed', items: [ + { id: 'u1', type: 'userMessage', clientId: 'first' }, + { id: 'u2', type: 'userMessage', clientId: 'steer' } + ] }, + { id: 'unknown', status: 'unknown', items: [{ id: 'u3', type: 'userMessage', clientId: 'unknown' }] }, + { id: 'running', status: 'inProgress', items: [] } + ] + await f.root.refresh() + await f.root.activate() + await expect(f.rpc.get(RPC_METHODS.ForkConversation)!({ messageLocalId: 'steer' })).rejects.toThrow('first message') + await expect(f.rpc.get(RPC_METHODS.ForkConversation)!({ messageLocalId: 'unknown' })).rejects.toThrow('completed') + expect(create).not.toHaveBeenCalled() + }) + it('creates a child before a completed turn without interrupting the active source', async () => { + const initialSettings = vi.fn(async () => {}) + const create = vi.fn(async () => ({ threadId: 'child', + session: { sessionId: 'child-session' }, initialSettings }) as unknown as SharedCodexRoot) + const f = await fixture({ create }) + f.native.thread.turns = [ + { id: 'earlier', status: 'completed', items: [{ id: 'u1', type: 'userMessage', clientId: 'local-earlier' }] }, + { id: 'running', status: 'inProgress', items: [{ id: 'u2', type: 'userMessage', clientId: 'local-running' }] } + ] + await f.root.refresh() + await f.root.activate() + const request = vi.spyOn(f.root.client, 'request') + const fork = f.rpc.get(RPC_METHODS.ForkConversation)! + await expect(fork({ messageLocalId: 'local-earlier' })).resolves.toEqual({ nativeSessionId: 'child', sessionId: 'child-session' }) + expect(create).toHaveBeenCalledWith('thread/fork', expect.objectContaining({ threadId: 'thread', beforeTurnId: 'earlier' }), f.root) + expect(f.state().steeringActive).toBe(true) + expect(request.mock.calls.some(([method]) => method === 'turn/interrupt')).toBe(false) + await expect(fork({})).rejects.toThrow('busy') + await expect(fork({ messageLocalId: 'local-running' })).rejects.toThrow('completed') + expect(create).toHaveBeenCalledTimes(1) + }) +}) + describe('compaction delivery confirmation', () => { it('does not turn uncertainty discovered by a cold snapshot into live compaction evidence', async () => { const f = await fixture(); diff --git a/cli/src/codex/shared/root.ts b/cli/src/codex/shared/root.ts index ecd630fd..01c5afec 100644 --- a/cli/src/codex/shared/root.ts +++ b/cli/src/codex/shared/root.ts @@ -535,10 +535,18 @@ export class SharedCodexRoot { }); rpc.registerHandler(RPC_METHODS.ForkConversation, async raw => { const { messageLocalId } = z.object({ messageLocalId: z.string().optional() }).parse(raw); - if (this.currentTurn) throw new Error('Session is busy'); + if (this.currentTurn && !messageLocalId) throw new Error('Session is busy'); const beforeTurnId = messageLocalId ? this.projection.turnFor(messageLocalId) : undefined; if (messageLocalId && !beforeTurnId) throw new Error('No native history point for this message'); - if (messageLocalId) await this.assertBoundary(messageLocalId, beforeTurnId!); + if (messageLocalId) { + const thread = await this.assertBoundary(messageLocalId, beforeTurnId!); + const turns = (Array.isArray(thread.turns) ? thread.turns : []).map(record); + const index = turns.findIndex(turn => turn.id === beforeTurnId); + if (index < 0 || turns.slice(0, index + 1).some(turn => + !['completed', 'interrupted', 'failed'].includes(string(turn.status) ?? ''))) { + throw new Error('Historical fork requires a completed turn and history prefix'); + } + } const child = await this.host.create('thread/fork', { ...this.freshParams(), threadId: this.threadId, ...(beforeTurnId ? { beforeTurnId } : {}) }, this); await child.initialSettings({ collaborationMode: this.settings.collaborationMode }); diff --git a/docs/guide/codex-shared-sessions.md b/docs/guide/codex-shared-sessions.md index ddd44958..deb1e212 100644 --- a/docs/guide/codex-shared-sessions.md +++ b/docs/guide/codex-shared-sessions.md @@ -133,6 +133,11 @@ endpoints/tokens never go into HAPI session metadata or SSE. sessions: a native mutation plus hub transcript truncation needs a commit barrier across every sender. Use **Fork at message** instead. Cuts inside a steered turn are rejected; select its first user message. +- **Fork at message** can branch from a finished historical turn while a later + turn is still running. It excludes the selected user message's native turn + and all later turns; the source keeps running. The selected turn and retained + prefix must have finished, and queued messages or conflicting history actions + still block the operation. **Fork current** continues to require an idle session. - `--worktree` and unsupported launch flags fail explicitly; create/select the working directory before launch. `--oss` requires an explicit `--local-provider`. Bare resume pickers use `hapi resume` or the attached TUI. diff --git a/hub/src/sync/forkWhileRunning.test.ts b/hub/src/sync/forkWhileRunning.test.ts new file mode 100644 index 00000000..f87e3fc7 --- /dev/null +++ b/hub/src/sync/forkWhileRunning.test.ts @@ -0,0 +1,98 @@ +import { describe, expect, it } from 'bun:test' +import { Store } from '../store' +import { RpcRegistry } from '../socket/rpcRegistry' +import { SyncEngine } from './syncEngine' +import type { RpcGateway } from './rpcGateway' + +function fixture(flavor = 'codex') { + const store = new Store(':memory:') + const engine = new SyncEngine(store, {} as never, new RpcRegistry(), { broadcast() {} } as never) + const source = engine.getOrCreateSession('source', { + path: '/tmp/work', host: 'test', machineId: 'machine', flavor, + capabilities: { conversationHistory: { forkCurrent: true, forkAtMessage: true, rewindToMessage: true } }, + conversationHistoryPoints: { retained: true, boundary: true }, + conversationHistoryTurns: { retained: 'turn-1', boundary: 'turn-2' } + }, { controlledByUser: false }, 'default') + engine.handleSessionAlive({ sid: source.id, time: Date.now(), thinking: true, mode: 'remote' }) + for (const id of ['retained', 'boundary', 'running']) { + store.messages.addMessage(source.id, { role: 'user', content: id }, id) + store.messages.markMessagesInvoked(source.id, [id], Date.now()) + } + const rpc = (engine as unknown as { rpcGateway: RpcGateway }).rpcGateway + return { store, engine, source, rpc } +} + +describe('historical forks while the source runs', () => { + it('copies only the historical prefix even when the source receives events during native fork', async () => { + const { store, engine, source, rpc } = fixture() + try { + rpc.forkConversation = async (_id, payload) => { + expect(payload).toEqual({ messageLocalId: 'boundary' }) + store.messages.addMessage(source.id, { role: 'assistant', content: 'new streaming event' }, 'late') + store.messages.markMessagesInvoked(source.id, ['late'], Date.now()) + return { nativeSessionId: 'child-native' } + } + rpc.spawnSession = async (...args) => ({ type: 'success', sessionId: args[12]! }) + const result = await engine.forkConversation(source.id, 'default', 'boundary') + expect(result.type).toBe('success') + if (result.type !== 'success') throw new Error(result.message) + expect(store.messages.getAllMessages(result.sessionId).map(row => row.localId)).toEqual(['retained']) + expect(engine.getSession(result.sessionId)?.metadata?.conversationHistoryTurns).toEqual({ retained: 'turn-1' }) + expect(engine.getSession(source.id)?.thinking).toBe(true) + expect(store.messages.getAllMessages(source.id)).toHaveLength(4) + } finally { engine.stop() } + }) + + for (const flavor of ['claude', 'pi', 'grok', 'agy', 'dsh']) { + it(`retains the busy guard for ${flavor}`, async () => { + const { engine, source } = fixture(flavor) + try { + expect(await engine.forkConversation(source.id, 'default', 'boundary')) + .toEqual({ type: 'error', message: 'Session is busy' }) + } finally { engine.stop() } + }) + } + + it('rejects stale boundaries and concurrent history actions before another native mutation', async () => { + const { engine, source, rpc } = fixture() + try { + expect(await engine.forkConversation(source.id, 'default', 'missing')).toMatchObject({ type: 'error' }) + expect(await engine.forkConversation(source.id, 'other', 'boundary')).toMatchObject({ type: 'error' }) + let release!: () => void + rpc.forkConversation = async () => { + await new Promise(resolve => { release = resolve }) + throw new Error('native fork rejected') + } + const pending = engine.forkConversation(source.id, 'default', 'boundary') + expect(await engine.forkConversation(source.id, 'default', 'boundary')).toEqual({ + type: 'error', message: 'Conversation history action already in progress' + }) + release() + expect(await pending).toEqual({ type: 'error', message: 'native fork rejected' }) + expect(engine.getSession(source.id)?.thinking).toBe(true) + } finally { engine.stop() } + }) + + it('retains diverged-history and capability checks for a running Codex source', async () => { + const { store, engine, source, rpc } = fixture() + try { + let calls = 0 + rpc.forkConversation = async () => { calls++; return { nativeSessionId: 'unused' } } + const update = (more: Record) => { + const current = store.sessions.getSession(source.id)! + store.sessions.updateSessionMetadata(source.id, { ...current.metadata as Record, ...more }, + current.metadataVersion, 'default', { touchUpdatedAt: false }) + ;(engine as unknown as { sessionCache: { refreshSession(id: string): void } }).sessionCache.refreshSession(source.id) + } + update({ conversationHistoryDiverged: true }) + expect(await engine.forkConversation(source.id, 'default', 'boundary')).toEqual({ + type: 'error', message: 'Conversation history is diverged; refuse further fork/rewind' + }) + update({ conversationHistoryDiverged: false, capabilities: {} }) + expect(await engine.forkConversation(source.id, 'default', 'boundary')).toEqual({ + type: 'error', message: 'Historical fork is not supported for this session' + }) + expect(calls).toBe(0) + } finally { engine.stop() } + }) +}) diff --git a/hub/src/sync/sharedCodex.test.ts b/hub/src/sync/sharedCodex.test.ts index 390cd2ce..304ddd08 100644 --- a/hub/src/sync/sharedCodex.test.ts +++ b/hub/src/sync/sharedCodex.test.ts @@ -20,6 +20,25 @@ function fixture() { } describe('shared Codex hub binding', () => { + it('allows Codex historical forks while thinking but retains other history guards', async () => { + const { store, engine, create, rpc } = fixture() + try { + const source = create('source', { conversationHistoryPoints: { boundary: true } }) + const child = create('child', { forkedFrom: source.id }) + store.messages.addMessage(source.id, { role: 'user', content: 'boundary' }, 'boundary') + store.messages.markMessagesInvoked(source.id, ['boundary'], Date.now()) + engine.handleSessionAlive({ sid: source.id, time: Date.now(), thinking: true }) + let calls = 0 + rpc.forkConversation = async () => { calls++; return { nativeSessionId: 'native-child', sessionId: child.id } } + expect(await engine.forkConversation(source.id, 'default', 'boundary')).toEqual({ type: 'success', sessionId: child.id }) + expect(engine.getSession(source.id)?.thinking).toBe(true) + expect(await engine.forkConversation(source.id, 'default')).toMatchObject({ type: 'error', message: 'Session is busy' }) + expect(await engine.rewindConversation(source.id, 'default', 'boundary')).toMatchObject({ type: 'error', message: 'Session is busy' }) + store.messages.addMessage(source.id, { role: 'user', content: 'queued' }, 'queued') + expect(await engine.forkConversation(source.id, 'default', 'boundary')).toMatchObject({ type: 'error', message: 'Session has queued messages' }) + expect(calls).toBe(1) + } finally { engine.stop() } + }) it('guards plan actions by namespace and runtime, leaving stale-plan and retry validation to CLI', async () => { const { engine, create, rpc } = fixture() try { diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index 34265de8..a60997e0 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -1300,14 +1300,14 @@ export class SyncEngine { await this.rpcGateway.abortSession(sessionId) } - private assertConversationHistoryIdle(session: Session): void { + private assertConversationHistoryIdle(session: Session, allowRunningHistoricalFork = false): void { if (!session.active) { throw new Error('Session must be active') } if (session.agentState?.controlledByUser === true && !session.metadata?.capabilities?.concurrentClients) { throw new Error('Conversation history actions require a remote session') } - if (session.thinking) { + if (session.thinking && !allowRunningHistoricalFork) { throw new Error('Session is busy') } if (session.metadata?.conversationHistoryDiverged === true) { @@ -1530,7 +1530,7 @@ export class SyncEngine { } let source = access.session try { - this.assertConversationHistoryIdle(source) + this.assertConversationHistoryIdle(source, Boolean(messageLocalId) && this.resolveFlavor(source) === 'codex') } catch (error) { return { type: 'error', message: error instanceof Error ? error.message : String(error) } } diff --git a/web/e2e/codex-fork-while-running.spec.ts b/web/e2e/codex-fork-while-running.spec.ts new file mode 100644 index 00000000..e5993b6c --- /dev/null +++ b/web/e2e/codex-fork-while-running.spec.ts @@ -0,0 +1,149 @@ +import { expect, test } from '@playwright/test' +import { createServer, type ServerResponse } from 'node:http' +import { mkdir, writeFile } from 'node:fs/promises' +import { join } from 'node:path' +import { randomUUID } from 'node:crypto' +import { getHapiBaseUrl, installHapiAuth } from './helpers/hapi-live' + +// The task Runner must inherit CODEX_HOME=/codex-home. +// Uses the installed Codex binary and a deterministic local Responses server, +// with a real isolated Hub, Runner, machine and browser. Never use production. +const enabled = process.env.HAPI_LIVE === '1' && process.env.HAPI_FORK_LIVE === '1' + && Boolean(process.env.HAPI_FORK_TEST_DIR) && Boolean(process.env.HAPI_ACCESS_TOKEN) + +for (const mobile of [false, true]) { + test(`historical Codex fork preserves a running source (${mobile ? 'mobile' : 'desktop'})`, async ({ page, request }, testInfo) => { + test.skip(!enabled, 'Requires an isolated Full environment and explicit HAPI_FORK_LIVE opt-in') + test.setTimeout(180_000) + const base = process.env.HAPI_FORK_LOCAL_URL ?? getHapiBaseUrl() + const directory = process.env.HAPI_FORK_TEST_DIR! + const accessToken = process.env.HAPI_ACCESS_TOKEN! + const auth = await request.post(`${base}/api/auth`, { data: { accessToken } }) + expect(auth.ok()).toBe(true) + const headers = { Authorization: `Bearer ${(await auth.json()).token}` } + const api = async (path: string, data?: unknown) => { + const response = data === undefined + ? await request.get(`${base}/api${path}`, { headers }) + : await request.post(`${base}/api${path}`, { headers, data, timeout: 60_000 }) + expect(response.ok(), `${path}: ${await response.text()}`).toBe(true) + return response.json() + } + const run = randomUUID() + const retained = `RETAINED_${run}` + const excluded = `EXCLUDED_${run}` + const running = `RUNNING_${run}` + const childProbe = `CHILD_PROBE_${run}` + let held: { response: ServerResponse; id: string } | undefined + const modelInputs: string[] = [] + const event = (response: ServerResponse, body: { type: string; [key: string]: unknown }) => + response.write(`event: ${body.type}\ndata: ${JSON.stringify(body)}\n\n`) + const finish = (response: ServerResponse, id: string, text: string) => { + event(response, { type: 'response.output_item.done', item: { type: 'message', role: 'assistant', + id: `msg_${id}`, content: [{ type: 'output_text', text }] } }) + event(response, { type: 'response.completed', response: { id, + usage: { input_tokens: 12, output_tokens: 3, total_tokens: 15 } } }) + response.end() + } + const provider = createServer((incoming, response) => { + const chunks: Buffer[] = [] + incoming.on('data', chunk => chunks.push(Buffer.from(chunk))) + incoming.on('end', () => { + if (!incoming.url?.endsWith('/responses')) { response.writeHead(404).end(); return } + const input = Buffer.concat(chunks).toString() + modelInputs.push(input) + const id = randomUUID() + response.writeHead(200, { 'Content-Type': 'text/event-stream' }) + event(response, { type: 'response.created', response: { id } }) + if (input.includes(childProbe)) finish(response, id, 'CHILD_CONTEXT_OK') + else if (input.includes(running)) held = { response, id } + else finish(response, id, input.includes(excluded) ? 'SECOND_DONE' : 'FIRST_DONE') + }) + }) + const sessions: string[] = [] + try { + await new Promise(resolve => provider.listen(0, '127.0.0.1', resolve)) + const port = (provider.address() as { port: number }).port + const home = join(directory, 'codex-home') + await mkdir(home, { recursive: true }) + const cwd = join(directory, `work-${run}`) + await mkdir(cwd, { recursive: true }) + await writeFile(join(home, 'config.toml'), `model = "mock-model"\nmodel_provider = "mock_provider"\napproval_policy = "never"\nsandbox_mode = "read-only"\ncheck_for_update_on_startup = false\n[model_providers.mock_provider]\nname = "Isolated fork test"\nbase_url = "http://127.0.0.1:${port}/v1"\nwire_api = "responses"\nrequires_openai_auth = false\nsupports_websockets = false\nrequest_max_retries = 0\nstream_max_retries = 0\n[analytics]\nenabled = false\n[feedback]\nenabled = false\n`) + const { machines } = await api('/machines') + expect(machines).toHaveLength(1) + const spawn = await api(`/machines/${machines[0].id}/spawn`, { + directory: cwd, agent: 'codex', model: 'mock-model', permissionMode: 'read-only' + }) + expect(spawn.type, JSON.stringify(spawn)).toBe('success') + const source = spawn.sessionId as string + sessions.push(source) + const detail = async (id: string) => (await api(`/sessions/${id}`)).session + const messages = async (id: string) => JSON.stringify(await api(`/sessions/${id}/messages?limit=200`)) + await expect.poll(async () => (await detail(source)).metadata?.capabilities?.conversationHistory?.forkAtMessage, + { timeout: 30_000 }).toBe(true) + for (const [text, result] of [[retained, 'FIRST_DONE'], [excluded, 'SECOND_DONE']]) { + await api(`/sessions/${source}/messages`, { text, localId: text }) + await expect.poll(() => messages(source), { timeout: 30_000 }).toContain(result) + await expect.poll(async () => (await detail(source)).thinking).toBe(false) + } + await api(`/sessions/${source}/messages`, { text: running, localId: running }) + await expect.poll(() => Boolean(held), { timeout: 30_000 }).toBe(true) + await expect.poll(async () => (await detail(source)).thinking).toBe(true) + // Both API and UI must retain the busy guard for current forks. + const current = await request.post(`${base}/api/sessions/${source}/fork`, { headers, data: {} }) + expect(current.status()).toBe(409) + expect(await current.json()).toMatchObject({ error: 'Session is busy' }) + const activeBoundary = await request.post(`${base}/api/sessions/${source}/fork`, { + headers, data: { messageLocalId: running } + }) + expect(activeBoundary.status()).toBe(409) + await installHapiAuth(page, base, accessToken) + await page.setViewportSize(mobile ? { width: 390, height: 844 } : { width: 1440, height: 1000 }) + await page.goto(`${base}/sessions/${source}`) + const boundary = page.locator('[data-hapi-message-role="user"]').filter({ hasText: excluded }) + await expect(boundary).toBeVisible() + await boundary.hover() + await expect(boundary.getByRole('button', { name: 'Rewind', exact: true })).toHaveCount(0) + await expect(boundary.getByRole('button', { name: 'Fork', exact: true })).toBeVisible() + await boundary.getByRole('button', { name: 'Fork', exact: true }).click() + const forkResponse = page.waitForResponse(response => response.url().endsWith(`/sessions/${source}/fork`) && response.request().method() === 'POST') + await page.getByRole('dialog').getByRole('button', { name: 'Fork', exact: true }).click() + const response = await forkResponse + expect(response.ok(), await response.text()).toBe(true) + const child = (await response.json()).sessionId as string + sessions.push(child) + await expect(page).toHaveURL(new RegExp(`/sessions/${child}$`)) + expect((await detail(source)).thinking).toBe(true) + expect(held?.response.writableEnded).toBe(false) + const childHistory = await messages(child) + expect(childHistory).toContain(retained) + expect(childHistory).not.toContain(excluded) + expect(childHistory).not.toContain(running) + await expect(page.getByText(retained, { exact: true })).toBeVisible() + await expect(page.getByText(excluded, { exact: true })).toHaveCount(0) + await api(`/sessions/${child}/messages`, { text: childProbe, localId: childProbe }) + await expect.poll(() => messages(child), { timeout: 30_000 }).toContain('CHILD_CONTEXT_OK') + const childInput = modelInputs.find(input => input.includes(childProbe))! + expect(childInput).toContain(retained) + expect(childInput).not.toContain(excluded) + expect(childInput).not.toContain(running) + // The first historical boundary produces an empty native prefix. + const emptyFork = await api(`/sessions/${source}/fork`, { messageLocalId: retained }) + sessions.push(emptyFork.sessionId) + const emptyHistory = await messages(emptyFork.sessionId) + expect(emptyHistory).not.toContain(retained) + expect(emptyHistory).not.toContain(excluded) + expect((await detail(source)).thinking).toBe(true) + finish(held!.response, held!.id, 'SOURCE_COMPLETED_AFTER_FORK') + await expect.poll(() => messages(source), { timeout: 30_000 }).toContain('SOURCE_COMPLETED_AFTER_FORK') + await expect.poll(async () => (await detail(source)).thinking).toBe(false) + expect(await messages(child)).not.toContain('SOURCE_COMPLETED_AFTER_FORK') + await testInfo.attach('native-fork-context', { body: JSON.stringify({ source, child, modelRequests: modelInputs.length, + inheritedRetained: childInput.includes(retained), inheritedExcluded: childInput.includes(excluded), inheritedRunning: childInput.includes(running) }), contentType: 'application/json' }) + } finally { + if (held && !held.response.writableEnded) held.response.end() + for (const id of sessions.reverse()) await api(`/sessions/${id}/archive`, {}).catch(() => {}) + provider.closeAllConnections() + await new Promise(resolve => provider.close(() => resolve())) + } + }) +} diff --git a/web/playwright.live.config.ts b/web/playwright.live.config.ts index b0f17a5f..8c7efc24 100644 --- a/web/playwright.live.config.ts +++ b/web/playwright.live.config.ts @@ -5,7 +5,7 @@ const chromePath = process.env.PLAYWRIGHT_CHROME_PATH /** Live hub session tests — no Vite; hits HAPI_URL with HAPI_LIVE=1. */ export default defineConfig({ testDir: './e2e', - testMatch: ['mermaid-lightbox-session.spec.ts', 'latex-bracket-math-session.spec.ts'], + testMatch: ['mermaid-lightbox-session.spec.ts', 'latex-bracket-math-session.spec.ts', 'codex-fork-while-running.spec.ts'], timeout: 120_000, expect: { timeout: 20_000 }, fullyParallel: false, diff --git a/web/src/components/AssistantChat/messages/MessageActions.test.tsx b/web/src/components/AssistantChat/messages/MessageActions.test.tsx index 87a8bca3..c4ef6243 100644 --- a/web/src/components/AssistantChat/messages/MessageActions.test.tsx +++ b/web/src/components/AssistantChat/messages/MessageActions.test.tsx @@ -281,6 +281,17 @@ describe('MessageActions', () => { expect(screen.queryByRole('button', { name: 'Rewind' })).toBeNull() }) + it('allows an eligible historical fork while keeping rewind locked during generation', async () => { + auiState.thread.isRunning = true + const onFork = vi.fn(async () => {}) + renderActions({ align: 'end', showFork: true, allowForkWhileRunning: true, + showRewind: true, onFork, onRewind: async () => {} }) + expect(screen.queryByRole('button', { name: 'Rewind' })).toBeNull() + fireEvent.click(screen.getByRole('button', { name: 'Fork' })) + fireEvent.click(within(screen.getByRole('dialog')).getByRole('button', { name: 'Fork' })) + await waitFor(() => expect(onFork).toHaveBeenCalledTimes(1)) + }) + it('keeps the share action visible while the thread is running', () => { auiState.thread.isRunning = true diff --git a/web/src/components/AssistantChat/messages/MessageActions.tsx b/web/src/components/AssistantChat/messages/MessageActions.tsx index 8d750f61..6532e110 100644 --- a/web/src/components/AssistantChat/messages/MessageActions.tsx +++ b/web/src/components/AssistantChat/messages/MessageActions.tsx @@ -22,6 +22,7 @@ type MessageActionsProps = { metadata?: Omit messageElementId?: string showFork?: boolean + allowForkWhileRunning?: boolean showRewind?: boolean historyActionPending?: boolean onFork?: () => Promise @@ -51,6 +52,7 @@ export function MessageActions({ metadata, messageElementId, showFork = false, + allowForkWhileRunning = false, showRewind = false, historyActionPending = false, onFork, @@ -65,7 +67,7 @@ export function MessageActions({ const [rewindOpen, setRewindOpen] = useState(false) const [forkPending, setForkPending] = useState(false) const [rewindPending, setRewindPending] = useState(false) - const actionsLocked = historyActionPending || forkPending || rewindPending || threadIsRunning + const actionsLocked = historyActionPending || forkPending || rewindPending const shareButton = messageElementId ? ( - {showRewind && onRewind ? ( + {showRewind && onRewind && !threadIsRunning ? ( setRewindOpen(true)} @@ -84,7 +86,7 @@ export function MessageActions({ ) : null} - {showFork && onFork ? ( + {showFork && onFork && (!threadIsRunning || allowForkWhileRunning) ? ( setForkOpen(true)} diff --git a/web/src/components/AssistantChat/messages/UserMessage.test.tsx b/web/src/components/AssistantChat/messages/UserMessage.test.tsx new file mode 100644 index 00000000..5c5a9e85 --- /dev/null +++ b/web/src/components/AssistantChat/messages/UserMessage.test.tsx @@ -0,0 +1,95 @@ +import { fireEvent, render, screen, waitFor, within } from '@testing-library/react' +import type { PropsWithChildren } from 'react' +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { I18nProvider } from '@/lib/i18n-context' +import { findLatestCompletedBoundaryId } from '@/lib/assistant-runtime' +import { HappyUserMessage } from './UserMessage' + +const onFork = vi.fn(async (_localId?: string) => {}) +const state = { + message: { id: 'user-text:interrupted', role: 'user', createdAt: new Date(10), + content: [{ type: 'text', text: 'Interrupted prompt' }], + metadata: { custom: { kind: 'user', localId: 'local-interrupted' } } }, + thread: { isRunning: true, messages: [] } +} +const ctx = { + metadata: { flavor: 'codex', capabilities: { conversationHistory: { forkCurrent: true, forkAtMessage: true, rewindToMessage: true } }, + conversationHistoryPoints: { 'local-interrupted': true } as Record }, + disabled: false, historyActionPending: false, + isLatestCompletedBoundary: (_id: string) => true, + onForkConversation: onFork, onRewindConversation: async () => {} +} + +vi.mock('@assistant-ui/react', () => ({ + useAuiState: (selector: (value: typeof state) => unknown) => selector(state), + MessagePrimitive: { Root: ({ children, ...props }: PropsWithChildren) =>
{children}
} +})) +vi.mock('@/components/AssistantChat/context', () => ({ + useHappyChatContext: () => ctx, useOptionalHappyChatContext: () => null +})) +vi.mock('./MessageAttachments', () => ({ MessageAttachments: () => null })) +vi.mock('./MessageStatusIndicator', () => ({ MessageStatusIndicator: () => null })) +vi.mock('./user-bubble', () => ({ + UserBubbleContent: ({ text }: { text: string }) => {text}, + getUserBubbleClassName: () => '', shouldShowMessageStatus: () => false +})) +vi.mock('@/components/ui/ConfirmDialog', () => ({ + ConfirmDialog: ({ isOpen, confirmLabel, onConfirm }: { isOpen: boolean; confirmLabel: string; onConfirm: () => Promise }) => + isOpen ?
: null +})) + +function mount() { render() } +async function confirmFork() { + fireEvent.click(screen.getByRole('button', { name: 'Fork' })) + fireEvent.click(within(screen.getByRole('dialog')).getByRole('button', { name: 'Fork' })) + await waitFor(() => expect(onFork).toHaveBeenCalledTimes(1)) +} + +describe('user-only completed fork boundary', () => { + beforeEach(() => { + localStorage.clear() + onFork.mockClear() + state.thread.isRunning = true + ctx.metadata.flavor = 'codex' + ctx.metadata.capabilities.conversationHistory.forkAtMessage = true + ctx.metadata.conversationHistoryPoints = { 'local-interrupted': true } + ctx.disabled = false + ctx.historyActionPending = false + }) + + it('forks before an interrupted user-only boundary while the next turn runs', async () => { + // No assistant block was produced by the interrupted first turn. + const boundary = findLatestCompletedBoundaryId([ + { kind: 'user-text', id: 'interrupted', localId: 'local-interrupted', text: 'Interrupted prompt', createdAt: 10, invokedAt: 10 }, + { kind: 'user-text', id: 'running', localId: 'local-running', text: 'Next prompt', createdAt: 20, invokedAt: 20 } + ], true, 20) + expect(boundary).toBe(state.message.id) + mount() + expect(screen.queryByRole('button', { name: 'Rewind' })).toBeNull() + await confirmFork() + expect(onFork).toHaveBeenCalledWith('local-interrupted') + }) + + it('retains current-fork semantics for the same card when idle', async () => { + state.thread.isRunning = false + mount() + await confirmFork() + expect(onFork.mock.calls[0]).toEqual([]) + }) + + it.each(['claude', 'pi', 'agy', 'dsh'])('does not unlock a running %s current fork', flavor => { + ctx.metadata.flavor = flavor + mount() + expect(screen.queryByRole('button', { name: 'Fork' })).toBeNull() + }) + + it.each(['point', 'capability', 'disabled', 'pending'])('keeps the %s guard', guard => { + if (guard === 'point') ctx.metadata.conversationHistoryPoints = {} + if (guard === 'capability') ctx.metadata.capabilities.conversationHistory.forkAtMessage = false + if (guard === 'disabled') ctx.disabled = true + if (guard === 'pending') ctx.historyActionPending = true + mount() + expect(screen.queryByRole('button', { name: 'Fork' })).toBeNull() + expect(onFork).not.toHaveBeenCalled() + }) +}) diff --git a/web/src/components/AssistantChat/messages/UserMessage.tsx b/web/src/components/AssistantChat/messages/UserMessage.tsx index 537b54bd..aa0c1d2a 100644 --- a/web/src/components/AssistantChat/messages/UserMessage.tsx +++ b/web/src/components/AssistantChat/messages/UserMessage.tsx @@ -14,6 +14,7 @@ export function HappyUserMessage() { const { t } = useTranslation() const role = useAuiState((s) => s.message.role) const messageId = useAuiState((s) => s.message.id) + const threadIsRunning = useAuiState((s) => s.thread.isRunning) const elementId = getConversationMessageAnchorId(messageId) const text = useAuiState((s) => { if (s.message.role !== 'user') return '' @@ -56,16 +57,20 @@ export function HappyUserMessage() { && localId.length > 0 && ctx.metadata?.conversationHistoryPoints?.[localId] === true const isLatestBoundary = ctx.isLatestCompletedBoundary?.(messageId) === true + // A user-only finished turn can be the latest completed boundary while + // another turn runs. Branch before that user turn, never from the live tail. + const runningCodex = threadIsRunning && ctx.metadata?.flavor === 'codex' const showCurrentFork = Boolean( history?.forkCurrent && isLatestBoundary + && !runningCodex && !ctx.disabled && ctx.onForkConversation ) const showHistoricalFork = Boolean( history?.forkAtMessage && hasNativePoint - && !isLatestBoundary + && (!isLatestBoundary || runningCodex) && !ctx.disabled && ctx.onForkConversation ) @@ -127,6 +132,7 @@ export function HappyUserMessage() { copyText={hasText ? text : undefined} messageElementId={elementId} showFork={showFork} + allowForkWhileRunning={showHistoricalFork && ctx.metadata?.flavor === 'codex'} showRewind={showRewind} historyActionPending={ctx.historyActionPending} onFork={showCurrentFork